diff options
Diffstat (limited to 'src/transport/mycelium/mod.rs')
| -rw-r--r-- | src/transport/mycelium/mod.rs | 25 |
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, |
