diff options
| -rw-r--r-- | src/transport/libp2p/mod.rs | 5 | ||||
| -rw-r--r-- | src/transport/mycelium/mod.rs | 18 | ||||
| -rw-r--r-- | src/transport/tor.rs | 6 | ||||
| -rw-r--r-- | src/transport/wire.rs | 46 |
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() { |