other wire versions warned

4 files changed, 59 insertions, 16 deletions

diff --git a/src/transport/libp2p/mod.rs b/src/transport/libp2p/mod.rs
index 3c4ddf5..2290dea 100644
--- a/src/transport/libp2p/mod.rs
+++ b/src/transport/libp2p/mod.rs
@@ -658,6 +658,7 @@ impl Transport for Libp2pTransport {
         let stranger_reads = Arc::new(Semaphore::new(STRANGER_READS));
         let probes = Arc::clone(&handler);
         let admission = Arc::clone(&self.admission);
+        let other_versions = Arc::new(wire::OtherVersions::default());
         tokio::spawn(async move {
             while let Some((peer, stream)) = incoming.next().await {
                 let limit = crate::transport::Admission::limit(
@@ -686,6 +687,7 @@ impl Transport for Libp2pTransport {
                 };
                 let sender = sender.clone();
                 let probes = Arc::clone(&probes);
+                let other_versions = Arc::clone(&other_versions);
                 tokio::spawn(async move {
                     match read_message(stream, limit).await {
                         Ok(body) => match wire::decode(&body) {
@@ -701,6 +703,9 @@ impl Transport for Libp2pTransport {
                             Ok(message) => {
                                 let _ = sender.send((peer, message, body.len())).await;
                             }
+                            Err(wire::WireError::Version(version)) => {
+                                other_versions.note(SCHEME, &peer.to_string(), version);
+                            }
                             Err(error) => {
                                 debug!(peer = %Addressed(SCHEME, &peer), error = %error, "Dropping a message that doesn't decode");
                             }
diff --git a/src/transport/mycelium/mod.rs b/src/transport/mycelium/mod.rs
index 369d683..f1dbe40 100644
--- a/src/transport/mycelium/mod.rs
+++ b/src/transport/mycelium/mod.rs
@@ -41,8 +41,6 @@ const POST_MIN_RATE: u64 = 1 << 20;
 /// The one Mycelium topic all of continuum's messages travel on; what a message is, its
 /// payload says (`wire`).
 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);
 
@@ -398,8 +396,7 @@ 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();
+        let other_versions = wire::OtherVersions::default();
         tracing::info!(
             endpoint = self.messages_endpoint.as_str(),
             "Starting the consumer"
@@ -451,18 +448,7 @@ impl Transport for MyceliumTransport {
                     }
                 }
                 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 = %Addressed(SCHEME, &inbound_message.src_pk),
-                            version,
-                            ours = wire::VERSION,
-                            "Peer speaks another wire version (another build); ignoring its messages"
-                        );
-                    }
+                    other_versions.note(SCHEME, &inbound_message.src_pk, version);
                 }
                 Err(transport_error) => {
                     // Who sent it and what it claims to be.
diff --git a/src/transport/tor.rs b/src/transport/tor.rs
index a08a5d0..c194330 100644
--- a/src/transport/tor.rs
+++ b/src/transport/tor.rs
@@ -192,6 +192,7 @@ struct Shared {
     probes: mpsc::Sender<Inbound>,
     /// Whose messages may be large (`transport::Admission`).
     admission: std::sync::OnceLock<crate::transport::Admission>,
+    other_versions: wire::OtherVersions,
 }
 
 fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
@@ -303,6 +304,7 @@ impl TorTransport {
             inbound,
             probes,
             admission: std::sync::OnceLock::new(),
+            other_versions: wire::OtherVersions::default(),
         });
         let (stopping, stopped) = watch::channel(false);
         {
@@ -566,6 +568,10 @@ impl Shared {
             }
             let message = match wire::decode(&body) {
                 Ok(message) => message,
+                Err(wire::WireError::Version(version)) => {
+                    self.other_versions.note(SCHEME, location, version);
+                    continue;
+                }
                 Err(error) => {
                     debug!(peer = %Addressed(SCHEME, &location), error = %error, "Dropping a message that doesn't decode");
                     continue;
diff --git a/src/transport/wire.rs b/src/transport/wire.rs
index 769545c..ef692b3 100644
--- a/src/transport/wire.rs
+++ b/src/transport/wire.rs
@@ -90,6 +90,38 @@ pub(crate) fn encode(message: &ParsedMessage) -> Result<Vec<u8>, WireError> {
 }
 
 /// The message in `payload` (as made by [`encode`]).
+/// How many senders on another wire version are remembered as already logged
+/// (`OtherVersions`).
+const MAX_OTHER_VERSIONS: usize = 1024;
+
+/// Senders whose messages are of another wire version (`WireError::Version`): each
+/// logged once, at warn level, so two builds that don't speak to each other say so,
+/// instead of silently retrying their HELLOs for good. Bounded: anyone can send us
+/// anything.
+#[derive(Default)]
+pub(crate) struct OtherVersions(std::sync::Mutex<std::collections::HashSet<String>>);
+
+impl OtherVersions {
+    /// A message from `sender` (on the transport `scheme`) was of `version`.
+    pub(crate) fn note(&self, scheme: &'static str, sender: &str, version: u8) {
+        let mut known = self
+            .0
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner());
+        if known.len() >= MAX_OTHER_VERSIONS {
+            known.clear();
+        }
+        if known.insert(sender.to_owned()) {
+            tracing::warn!(
+                from = %super::Addressed(scheme, &sender),
+                version,
+                ours = VERSION,
+                "Peer speaks another wire version (another build); ignoring its messages"
+            );
+        }
+    }
+}
+
 pub(crate) fn decode(payload: &[u8]) -> Result<ParsedMessage, WireError> {
     let (&version, payload) = payload.split_first().ok_or(WireError::Empty)?;
     if version != VERSION {
@@ -492,6 +524,20 @@ mod tests {
         ListedPeer, MAX_LISTED_HEADS, Status,
     };
 
+    /// Each sender on another wire version is remembered once, and no more than
+    /// `MAX_OTHER_VERSIONS` of them.
+    #[test]
+    fn other_versions_are_noted_once_each() {
+        let other = OtherVersions::default();
+        other.note("s", "a", VERSION + 1);
+        other.note("s", "a", VERSION + 1);
+        assert_eq!(other.0.lock().unwrap().len(), 1);
+        for i in 0..MAX_OTHER_VERSIONS + 1 {
+            other.note("s", &i.to_string(), VERSION + 1);
+        }
+        assert!(other.0.lock().unwrap().len() <= MAX_OTHER_VERSIONS);
+    }
+
     /// Every kind decodes to what was encoded: re-encoded, the same bytes.
     #[test]
     fn every_kind_round_trips() {