include version in wire format

1 files changed, 90 insertions, 7 deletions

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")
+        );
+    }
 }