mirrord protocol

[aviramha]

We should make progress with the protocol (#1353 #1354 #1355 #1356), as we have already few stuff pending protocol breakage and we don't want to do it without atleast featuring forward-proof changes. i.e if we break now let's make this the last break.
I'd say a good start would add a framing (size of each message before message) to our codec, so even if parsing fails because version is broken, we can skip it in the operator or even agent.
This also might improve performance since we can know how much data needs to be read before parsing, as now we have a slow cycle where we keep reading and going back to bincode parse.

Secondly, I'd add that the initial message would be each side sending their version, so the other side can adjust what it can send to it - so atleast for now we'll have a global variable saying "VERSION = 1" then we can decide if to send the new type of message or not, until we build a proper abstraction.

Lastly, I guess we should start with "tons of boiler plate" code that will just work, then try to figure out how to macro/do magic to make it short and ergonomic. Maybe introduce a new crate for defining network protocols (I suggest calling it nobuf).
In short, we'd want the protocol to be multiplex (reqid,responseid, response being optional so we don't block on it) and with nice Rust abstractions (similar to tower probably but different ofc).

I'd love to hear any thoughts on the mind dump I had here - this is a "tough" thing to build correctly, but we need to do it atleast once :)

[meowjesty]

add a framing (size of each message before message)

We should also add some form of id and version here, so we can skip the message in the agent if it doesn't match on version.

I'd add that the initial message would be each side sending their version

+1 to version negotiation.

Should the protocol support encryption?

this is a "tough" thing to build correctly

I'm still not sure we should roll our own here, were we not exploring using existing stuff? What happened?

[aviramha]

I'm still not sure we should roll our own here, were we not exploring using existing stuff? What happened?

We were exploring, but nothing fit like a glove or really solved the hard parts. I'm open to suggestions.

Should the protocol support encryption?

I don't think so, if anything it'd be layer before the actual application IMO.

[aviramha]

https://github.com/betwixt-labs/bebop/wiki

[t4lz]

Trying to understand the requirements:

1. Every session starts with version negotiation: both sides send the oldest and newest protocol version they support, they then use the newest version they both support, or break the connection if there is no common version. Correct?
2. It should be possible to "skip" unrecognised incoming messages (it should be possible to know where the next message starts).

• "Eavesdroppability" as a feature - we want a software component (operator) that is not a part of the version negotiation to be able to listen to the session on a best-effort basis. (Maybe we could instead let the operator in the middle change the version negotiation messages).
• I'm a bit worried about allowing to continue the session after an ignored message, because we would lose the ability to assume that if we successfully sent a message and did not get an error - the other side received it.

3. Adding a new field to a message does not break compatibility.

• Does that mean that software running the old version of the protocol (without the new field in some message type) can decode a message of the new version with the new field, ignoring the new field? Isn't that redundant if we have protocol version negotiation? Wouldn't both sides only send messages of the older version anyways?

[meowjesty]

It should be possible to "skip" unrecognised incoming messages (it should be possible to know where the next message starts).

I'm not sure how we could send an "unrecognised" message, the layer knows which version they're dealing with, so it should never send anything that's not available? Like, these messages should die on the ClientMessage handler.

I'm thinking that we could only send these types of messages if we had some sort of "soft" version handshake, where the version matching is done on a best-effort basis?

[aviramha]

I'm not sure how we could send an "unrecognised" message, the layer knows which version they're dealing with, so it should never send anything that's not available? Like, these messages should die on the ClientMessage handler.

not from the operator standpoint though

[meowjesty]

Not sure I follow.

1. layer handshakes with the operator;
2. operator handshakes with the agent;
3. layer and operator are synced, and operator passes to layer the agent's version range?

[aviramha]

There might be situations where layer = agent version but operator is older, and it wouldn't really care about the new unknown messages.

[eyalb181]

We can add a "meta" message of "unsupported message" to let know of the other side that it fouled.

Are skipped messages necessarily a foul? If so, why not error out with a dynamic version mismatch, rather than let undefined behaviour unfold?

[aviramha]

This change became real necessity after #1731 and we already have quite a lot of protocol changes we want to push and are unable at the moment.
Here is my proposal.

Protocol Negotation

• Protocol have to be same across the board layer - (operator) - agent
• The reason we want the operator to have same protocol as layer/agent is that the operator would act as a "firewall" in the end (enabling/disabling specific operations) and just sending arbitrary messages isn't something we want to happen.
• CLI would check operator feature, and see if it has new protocol.
• For non operator sessions, we would assume new protocol exists.
• On connection, first messages would be protocol negotiation where they set the highest-mutual-supported level

API Level

• We want the API to be comfortable for adding/removing/changing existing messages without breaking it.
• We also don't want the internal code to care about the versioning (as much as we can)

I'll try to draft pseudo code for how I'd imagine it to work.

[aviramha]

I agree on 4.

[t4lz]

How about this as an interface (I don't actually know yet if all of this can be implemented that way, it's rather optimistic about what we can achieve with macros):

you can define protocol messages kind of like you did up until now:

#[derive(Encode, Decode, Debug, PartialEq, Eq, Clone)]
#[message]
struct MyMessage {
    field1: String,
    field2: MyCustomStruct,
}

(the message macro would generate a struct called MyMessageV0)

But you can limit the range of protocol versions in which messages or message fields are defined. For example, if we want to introduce a new field to MyMessage, we change its code to:

#[derive(Encode, Decode, Debug, PartialEq, Eq, Clone)]
#[message]
struct MyMessage {
    field1: String,
    field2: MyCustomStruct,
    #[field(from_version="2.0.0")]
    field3: u8
}

For this definition 2 structs would be generated: MyMessageV0 (same one as before) and MyMessageV1 (includes the new field3).

For removing fields in newer versions we would have an analogous attribute argument until_version.

The using code would get a MyMessage object that would be an enum. We would generate getter functions for all fields. For fields that are not limited by from_version or until_version, the generated getter function's return type would be the field's type. For version-limited fields the generated getter function would return an Option of that type.
Additionally the user could also get the instance of the concrete type (e.g. MyMessageV0) for when that's more convenient.
Similar interfaces would be offered for constructing message objects.

So we could offer multiple ways to handle the message object:

fn by_getting_underlying_message(m: MyMessage) {
    match m {
        V0(MyMessageV0{
            field1,
            field2,
            ..
        }) => todo!(),
        V1(MyMessageV1{
            field1,
            field2,
            field3,
            ..
        }) => todo!(),
    }
}

fn by_getter_funcs(m: MyMessage) {
    let f1: &String = m.get_field1();
    if let Some(f3) = m.get_field3() {
        todo!()
    }
}

// "Overlayed" is probably not the right word, sorry for that.
// convert the message object to an object that has all fields from all versions.
// fields that are only present in some versions are options.
fn with_overlayed_object(m: MyMessage) {
    let m: MyMessageOverlayed{
        field 1,
        field 2,
        field 3, // Option
        ..
    } = m.into();
    todo!()
}

The generated structs would also have a message ID field, and would implement some common trait for getting the message ID and possibly other methods as well.

Ideally instead of manually defining the enums that contain variants containing the messages, we could have a special attribute for enums, and we would specify the enum to which a message belongs in its message attribute.
For example we could have:

#[message_enum]
enum ClientMessage { }

and then in above MyMessage's definition:

#[message(enum=ClientMessage)]

It seems to me like there's a good chance that's not really possible with macros in Rust.

Challenges

1. If we delete a message (we had it for a long time just for backwards compatibility with #[message(until_version=4.0.0)] and now we want to not support versions older than 4 anymore, so we want to delete that message from our code), the discriminants of all later messages in the enum will be different in our new binary than in the old one.
   • One possible solution is that when defining a message you would have to assign it a discriminant (with enough distance from the preceding message's discriminant for many versions). For example we could assign "message numbers" 10, 20, 30 etc. that would be used as discriminants for the enum. So the attributes would become more like #[message(message_num=0, enum=ClientMessage)], or #[message(message_num=0, enum=ClientMessage, version_until="3.0.0")] with version limits.
   • Alternatively, we could live a placeholder there to occupy the deleted message's place in the enum. That's ugly because old messages never completely disappear from the code.

[t4lz]

It definitely would not be easy to implement these macros (if it's even possible), but it would make the interface even nicer than in my last comment:

Maybe we could also have attribute proc macros for the handlers, to mark handler methods as handlers for different messages and protocol versions, like #[handle_message] and then #[handle_message(from_version="3.0.0")] once we have a new version of that message.

That way the handling code does not have to check the version or the individual fields, but each handling method already gets a specific version of the message on the type level (then we could also group the messages by version meaning we would generate a different enum for each protocol message, and the variants of that enum would be message structs of that specific version, not enums, so we wouldn't have to check the version of each incoming message - this makes sense, because they all always have the same version for the whole run. At least on the layer's side.)

[meowjesty]

4. Should be defined in rust, without additional languages, tools or compilers.

If we drop this, then most of the work would be converting our messages to the new protocol structs, right?

We could use RPC like @DmitryDodzin suggested, would tonic be good enough, or are there better stuff?

Or we could use protobuf-like stuff directly, like capnproto.

My suggestion for a REST-like protocol is basically an RPC, but we don't have to define types, we just shove them into a json object, and let the implementation handle the details.

I don't think mirrord has any special requirements for a protocol, that would only be solvable by a custom implementation.

[eyalb181]

We'll go with @t4lz suggestion, as it A. fulfils the requirements and B. introduces the smallest change into the product. We're not aware of any third-party library that does what Tal suggests without a ton of accompanying functionality that we don't need, but if anyone does know of something like that let us know (see Aviram's objections to capnproto below).
@aviramha @meowjesty @DmitryDodzin

[meowjesty]

What about attaching an HTTP engine to each part of mirrord, and communicating REST style? We could make the messages be more dynamic, use only what serde already supports + extractors from whatever HTTP framework is chosen.

I'm thinking something like:

#[post("/v1/layer_read/{file}")]
fn layer_read(file: FileId, message: Json<JsonObject>) -> Response {
    let bytes = message.get("bytes");
    let amount_read = message.get("amount_read");
    ....
}

#[post("/v100/layer_read/{file}")]
fn layer_read(file: FileId, message: Json<JsonObject>) -> Response {
    // we have changed the name `bytes` to `buffer`
    let buffer = message.get("buffer");

    // we removed the `amount_read` field, now we just rely on the length
    let amount_read = buffer.len();
    ...
}

#[post("/v1/agent_read/{file}")]
fn agent_read(file: FileId) -> Response {
    let read = FileManager::read(file);
    let read_response =   json!({ "bytes": read.bytes, "amount_read": read.bytes.len() });
    ...
    read_response
}

#[post("/v100/agent_read/{file}")]
fn agent_read(file: FileId) -> Response {
    let read = FileManager::read(file);
    let read_response =   json!({ "bytes": read.bytes });
    ...
    read_response
}

We could support version negotiation as a first message exchange between layer/operator/agent.

How awful does this sound?

[DmitryDodzin]

One Option, though a bit large at first, will allow a greater deal of control is creating a completely RPC-based API for the agent. With Cap'n Proto I created a small example of how it can look like for our stealing of tcp

interface ByteStream {
	write @1 (bytes: Data);

	close @2 ();

	expectSize @3 (lower: UInt64, upper: UInt64);
}

interface HttpRequest {
	struct Header {
		name @0 :Text;
		value @1 :Text;
	}

	struct Head {
		method @0 :Text;
		uri @1 :Text;
		headers @2 :List(Header);
		version: @3 :Text;
	}

	head @0 (header: Head);
	body @1 (bytes: ByteStream);
}

interface HttpResponse {
}

interface Tcp {
	struct ConnectionInfo {
		connectionId @0 :UInt64;
		remoteAddress @1 :Text; # IP?
		destinationPort @2 :UInt16;
		sourcePort @3 :UInt16
		localAddress @4 :Text; # IP?
	}

	struct HttpConnectionInfo {
		connectionId @0 :UInt64;
		requestId @1 :UInt64;
		port: @2 :UInt16;
	}

	connection @0 (connection_info: ConnectionInfo, inbound: ByteStream) -> (outbound: ByteStream);
	http @1 (connection_info: HttpConnectionInfo, request: HttpRequest) -> (response: HttpResponse);
}

interface AgentTcp {
	struct Steal {
		port @0 :UInt16;

		type union {
			full @1 :Void;
			header @2 :Text;
			path @3 :Text;
		}
	}

	mirror @0 (port: UInt16) -> (tcp: Tcp);
	steal @0 (type: Steal) -> (tcp: Tcp);
}

Or it can be split it to separate full port or HTTP only steals where the value

interface ByteStream {
	write @1 (bytes: Data);

	close @2 ();

	expectSize @3 (lower: UInt64, upper: UInt64);
}

interface HttpRequest {
	struct Header {
		name @0 :Text;
		value @1 :Text;
	}

	struct Head {
		method @0 :Text;
		uri @1 :Text;
		headers @2 :List(Header);
		version: @3 :Text;
	}

	head @0 (header: Head);
	body @1 (bytes: ByteStream);
}

interface HttpResponse {
}

interface Tcp {
	struct ConnectionInfo {
		connectionId @0 :UInt64;
		remoteAddress @1 :Text; # IP?
		destinationPort @2 :UInt16;
		sourcePort @3 :UInt16
		localAddress @4 :Text; # IP?
	}

	connection @0 (connection_info: ConnectionInfo, inbound: ByteStream) -> (outbound: ByteStream);
}

interface Http {
	struct HttpConnectionInfo {
		connectionId @0 :UInt64;
		requestId @1 :UInt64;
		port: @2 :UInt16;
	}

	request @0 (connection_info: HttpConnectionInfo, request: HttpRequest) -> (response: HttpResponse);
}

interface AgentTcp {
	struct HttpSteal {
		port @0 :UInt16;

		type union {
			header @2 :Text;
			path @3 :Text;
		}
	}

	mirror @0 (port: UInt16) -> (tcp: Tcp);
	steal @1 (port: UInt16) -> (tcp: Tcp);
	httpSteal @2 (type: HttpSteal) -> (http: Http);
}

The move can be partial with a large union but it is also a lot of work. But the main idea is I want to reuse the request handlers like the Tcp One for both steal and mirror.

[aviramha]

My takes against CapnProto:

1. Introduces a new language we need to understand
2. Introduces a compiler (neglible)
3. Isn't widely used, especially the rust bindings
4. The key thing you want to use capnproto is for the more advanced API, which relies on more Rust/CapnProto magic that if we'll have issues using we'll probably end up fixing up stream.

[eyalb181]

@DmitryDodzin Assuming our goal here is decoupling agent/operator/layer and nothing else, what's the merit of migrating to RPC?

[DmitryDodzin]

Encapsulating our codec to separate APIs will allow a more flexible approach to changes where we can only worry about the current packing of messages and not how it affects the entire protocol like the de-nesting is the main advantage. I don't stand on the CapnProto per se, I like the idea of sending the apis back and forth and having an abstraction that can use the lifetimes to keep certain connections alive (like in operator to have port-locks drop when the connection proxy is dropped)