include version in wire format

2 files changed, 111 insertions, 11 deletions

diff --git a/src/transport/mycelium/mod.rs b/src/transport/mycelium/mod.rs
index f8569c1..f20f6a0 100644
--- a/src/transport/mycelium/mod.rs
+++ b/src/transport/mycelium/mod.rs
@@ -18,7 +18,9 @@ const MYCELIUM_POLL_TIMEOUT_SECS: u64 = 300;
 
 /// The one Mycelium topic all of continuum's messages travel on; what a message is, its
 /// payload says (`wire`).
-const TOPIC: &str = "continuum.v1";
+const TOPIC: &str = "continuum";
+/// How many senders on another wire version are remembered as already logged.
+const MAX_OTHER_VERSIONS: usize = 1024;
 
 const EMPTY_QUEUE_DELAY: Duration = Duration::from_millis(25);
 
@@ -247,6 +249,8 @@ impl Transport for MyceliumTransport {
     }
 
     async fn start_consumer(&self, handler: MessageHandler) -> Result<()> {
+        // Senders on another wire version, logged once each.
+        let mut other_versions = std::collections::HashSet::new();
         tracing::info!(
             endpoint = self.messages_endpoint.as_str(),
             "Starting Mycelium message consumer"
@@ -321,13 +325,26 @@ impl Transport for MyceliumTransport {
                         tracing::warn!(error = %format!("{error:#}"), "Failed to process inbound message");
                     }
                 }
+                Err(TransportError::Wire(wire::WireError::Version(version))) => {
+                    // Bounded: anyone can send us anything.
+                    if other_versions.len() >= MAX_OTHER_VERSIONS {
+                        other_versions.clear();
+                    }
+                    if other_versions.insert(inbound_message.src_pk.clone()) {
+                        tracing::warn!(
+                            from = %inbound_message.src_pk,
+                            version,
+                            ours = wire::VERSION,
+                            "Peer speaks another wire version (another build); ignoring its messages"
+                        );
+                    }
+                }
                 Err(transport_error) => {
-                    // Who sent it and what it claims to be: a peer on another build
-                    // shows up as one sender whose messages of one kind all fail.
+                    // Who sent it and what it claims to be.
                     let kind = BASE64
                         .decode(&inbound_message.payload)
                         .ok()
-                        .and_then(|payload| payload.first().copied());
+                        .and_then(|payload| payload.get(1).copied());
                     tracing::error!(
                         error = %transport_error,
                         from = %inbound_message.src_pk,
diff --git a/src/transport/wire.rs b/src/transport/wire.rs
index e055393..11d36cf 100644
--- a/src/transport/wire.rs
+++ b/src/transport/wire.rs
@@ -1,16 +1,22 @@
-//! The wire format of a message: one byte saying what it is (`MessageKind`), then the
-//! message itself in MessagePack, as derived from its type in `transport` (structs as
+//! The wire format of a message: one byte for the format's version (`VERSION`), one
+//! saying what the message is (`MessageKind`), then the message itself in MessagePack, as derived from its type in `transport` (structs as
 //! arrays, fields in declaration order). The codecs below make hashes, IDs and bytes
 //! compact (MessagePack binaries, not arrays of numbers) and check their lengths.
 //!
 //! The wire format may change freely until the first release: there is no compatibility
-//! with older nodes to keep.
+//! with older nodes to keep. But every change bumps `VERSION`, so nodes on different
+//! builds tell each other apart instead of misreading each other's messages. The
+//! transport's topic says which application a message is for, not which version.
 
 use thiserror::Error;
 
 use super::{MessageKind, ParsedMessage};
 
-/// What the first byte of a payload says.
+/// The wire format's version, the first byte of every payload. Bump it with every change
+/// to a message type or a codec (`golden_bytes` fails then).
+pub(crate) const VERSION: u8 = 1;
+
+/// What the second byte of a payload says.
 fn tag(kind: MessageKind) -> u8 {
     match kind {
         MessageKind::Hello => 0,
@@ -27,6 +33,9 @@ pub enum WireError {
     #[error("empty payload")]
     Empty,
 
+    #[error("wire format version {0} (we speak version {VERSION})")]
+    Version(u8),
+
     #[error("unknown message kind {0}")]
     UnknownKind(u8),
 
@@ -39,7 +48,7 @@ pub enum WireError {
 
 /// The payload of `message`.
 pub(crate) fn encode(message: &ParsedMessage) -> Result<Vec<u8>, WireError> {
-    let mut payload = vec![tag(message.kind())];
+    let mut payload = vec![VERSION, tag(message.kind())];
     match message {
         ParsedMessage::Hello(hello) => rmp_serde::encode::write(&mut payload, hello)?,
         ParsedMessage::HelloReply(reply) => rmp_serde::encode::write(&mut payload, reply)?,
@@ -54,6 +63,10 @@ pub(crate) fn encode(message: &ParsedMessage) -> Result<Vec<u8>, WireError> {
 
 /// The message in `payload` (as made by [`encode`]).
 pub(crate) fn decode(payload: &[u8]) -> Result<ParsedMessage, WireError> {
+    let (&version, payload) = payload.split_first().ok_or(WireError::Empty)?;
+    if version != VERSION {
+        return Err(WireError::Version(version));
+    }
     let (&kind, body) = payload.split_first().ok_or(WireError::Empty)?;
     let is = |wanted: MessageKind| kind == tag(wanted);
     Ok(if is(MessageKind::Hello) {
@@ -336,7 +349,15 @@ mod tests {
     #[test]
     fn unknown_kinds_and_bad_payloads_are_refused() {
         assert!(matches!(decode(&[]), Err(WireError::Empty)));
-        assert!(matches!(decode(&[99]), Err(WireError::UnknownKind(99))));
+        assert!(matches!(decode(&[VERSION]), Err(WireError::Empty)));
+        assert!(matches!(
+            decode(&[VERSION, 99]),
+            Err(WireError::UnknownKind(99))
+        ));
+        assert!(matches!(
+            decode(&[VERSION + 1, 0]),
+            Err(WireError::Version(version)) if version == VERSION + 1
+        ));
 
         let blob = ParsedMessage::Blob(BlobMessage {
             epoch: Epoch::GENESIS,
@@ -350,7 +371,69 @@ mod tests {
         assert!(decode(&payload[..payload.len() - 1]).is_err());
         // Nor one of another kind.
         let mut relabelled = payload.clone();
-        relabelled[0] = tag(MessageKind::Hello);
+        relabelled[1] = tag(MessageKind::Hello);
         assert!(decode(&relabelled).is_err());
     }
+
+    fn hex(bytes: &[u8]) -> String {
+        bytes.iter().map(|byte| format!("{byte:02x}")).collect()
+    }
+
+    /// The exact bytes of a few messages. If this fails after a change to a message type
+    /// or a codec, the change is a new wire format: bump `VERSION` and take the new bytes.
+    #[test]
+    fn golden_bytes() {
+        let epoch = Epoch {
+            generation: 2,
+            id: [3; 32],
+        };
+        let epoch_hex = format!("9202c420{}", "03".repeat(32));
+        let run = |byte: &str| format!("c410{}", byte.repeat(16));
+
+        let hello = ParsedMessage::Hello(HelloRequest {
+            swarm_id: Some("swarm".into()),
+            epoch,
+            epoch_seq: 1,
+            auth: vec![1, 2, 3],
+            ready: true,
+            run_id: [4; 16],
+            handshake: 7,
+        });
+        assert_eq!(
+            hex(&encode(&hello).unwrap()),
+            format!("010097a5737761726d{epoch_hex}01c403010203c3{}07", run("04"))
+        );
+
+        let reply = ParsedMessage::HelloReply(HelloReply {
+            swarm_id: "swarm".into(),
+            peers: vec!["p".into()],
+            epoch,
+            epoch_seq: 1,
+            auth: Vec::new(),
+            ready: false,
+            run_id: [4; 16],
+            away_ms: Some(5),
+            sender_run_id: [6; 16],
+        });
+        assert_eq!(
+            hex(&encode(&reply).unwrap()),
+            format!(
+                "010199a5737761726d91a170{epoch_hex}01c400c2{}05{}",
+                run("04"),
+                run("06")
+            )
+        );
+
+        let ping = ParsedMessage::Ping(Keepalive {
+            heads: HeadsSummary::Heads(Vec::new()),
+            epoch,
+            epoch_seq: 1,
+            ready: true,
+            syncs_with_you: false,
+        });
+        assert_eq!(
+            hex(&encode(&ping).unwrap()),
+            format!("0103959290c400{epoch_hex}01c3c2")
+        );
+    }
 }