include version in wire format

1 files changed, 21 insertions, 4 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,