3 files changed, 64 insertions, 9 deletions

diff --git a/src/swarm/peer/keepalive.rs b/src/swarm/peer/keepalive.rs
index 44f2879..3b78b85 100644
--- a/src/swarm/peer/keepalive.rs
+++ b/src/swarm/peer/keepalive.rs
@@ -238,11 +238,17 @@ pub(super) async fn note_status(state: &AppState, peer: &str, remote: &transport
         peer_state.reported_heads = remote.heads.heads().map(<[_]>::len);
         peer_state.last_heads = Some((remote.epoch, remote.heads.clone()));
         if let Some(reaches) = &remote.reaches {
-            peer_state.reaches = Some((Instant::now(), reaches.clone()));
-            peer_state.round_trips = remote
-                .round_trips
-                .clone()
-                .filter(|round_trips| round_trips.len() == reaches.len());
+            match crate::swarm::reach::said(reaches, remote.round_trips.as_deref()) {
+                Some((reaches, round_trips)) => {
+                    peer_state.reaches = Some((Instant::now(), reaches));
+                    peer_state.round_trips = round_trips;
+                }
+                None => debug!(
+                    peer = %peer,
+                    reaches = reaches.len(),
+                    "Ignoring a reach set larger than any node's"
+                ),
+            }
         }
         drop(peer_state);
         if left {
diff --git a/src/swarm/reach.rs b/src/swarm/reach.rs
index e6cee2f..d4949ef 100644
--- a/src/swarm/reach.rs
+++ b/src/swarm/reach.rs
@@ -35,6 +35,35 @@ use crate::state::{AppState, HelloState, Standing};
 use crate::swarm::peer::announce_heads;
 use crate::swarm::{members, rtt, topology};
 
+/// The most members a node may say it reaches (`said`): more than it could hold
+/// sessions with.
+pub(crate) const MAX_REACHES: usize = 4096;
+
+/// A reach set as a neighbour said it (`Status::reaches`), with its round trips if
+/// they match it: sorted by pair key (as we send ours: `measure`), so it can be looked
+/// up by halving (`swarm::tree`); None if it is larger than `MAX_REACHES`.
+pub(crate) fn said(
+    reaches: &[u64],
+    round_trips: Option<&[u8]>,
+) -> Option<(Vec<u64>, Option<Vec<u8>>)> {
+    if reaches.len() > MAX_REACHES {
+        return None;
+    }
+    let round_trips = round_trips.filter(|round_trips| round_trips.len() == reaches.len());
+    if reaches.windows(2).all(|pair| pair[0] < pair[1]) {
+        return Some((reaches.to_vec(), round_trips.map(<[u8]>::to_vec)));
+    }
+    let mut pairs: Vec<(u64, Option<u8>)> = reaches
+        .iter()
+        .enumerate()
+        .map(|(i, key)| (*key, round_trips.map(|round_trips| round_trips[i])))
+        .collect();
+    pairs.sort_unstable_by_key(|(key, _)| *key);
+    pairs.dedup_by_key(|(key, _)| *key);
+    let round_trips = round_trips.map(|_| pairs.iter().filter_map(|(_, rtt)| *rtt).collect());
+    Some((pairs.into_iter().map(|(key, _)| key).collect(), round_trips))
+}
+
 /// Whom we reach, as last told our neighbours, and our round trip to each.
 #[derive(Default)]
 pub(crate) struct Reach {
@@ -160,3 +189,23 @@ pub(crate) async fn names(state: &AppState) -> HashMap<u64, String> {
         .map(|member| (topology::pair_key(&member), member))
         .collect()
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// A reach set comes sorted, its round trips along; too large, it is ignored.
+    #[test]
+    fn reach_sets_are_sorted_and_bounded() {
+        assert_eq!(
+            said(&[1, 3, 5], Some(&[9, 8, 7])),
+            Some((vec![1, 3, 5], Some(vec![9, 8, 7])))
+        );
+        assert_eq!(
+            said(&[5, 1, 3, 1], Some(&[7, 9, 8, 9])),
+            Some((vec![1, 3, 5], Some(vec![9, 8, 7])))
+        );
+        assert_eq!(said(&[2, 1], Some(&[1])), Some((vec![1, 2], None)));
+        assert_eq!(said(&vec![0; MAX_REACHES + 1], None), None);
+    }
+}
diff --git a/src/swarm/tree.rs b/src/swarm/tree.rs
index 81004e8..3ed8a9b 100644
--- a/src/swarm/tree.rs
+++ b/src/swarm/tree.rs
@@ -202,8 +202,8 @@ pub(crate) fn whole(
     )
 }
 
-/// Whom a member reaches, by pair key, and its round trip to each (`rtt::bucket`), as
-/// it said.
+/// Whom a member reaches, by pair key, sorted (`reach::said`), and its round trip to
+/// each (`rtt::bucket`), as it said.
 struct Said {
     keys: Vec<u64>,
     round_trips: Option<Vec<u8>>,
@@ -219,7 +219,7 @@ impl Known<'_> {
     fn reached(&self, by: &str, member: &str) -> bool {
         self.0
             .get(by)
-            .is_none_or(|said| said.keys.contains(&topology::pair_key(member)))
+            .is_none_or(|said| said.keys.binary_search(&topology::pair_key(member)).is_ok())
     }
 
     /// `by`'s round trip to `member`, as a bucket; `rtt::UNKNOWN_BUCKET` if not said.
@@ -228,7 +228,7 @@ impl Known<'_> {
         self.0
             .get(by)
             .and_then(|said| {
-                let index = said.keys.iter().position(|said| *said == key)?;
+                let index = said.keys.binary_search(&key).ok()?;
                 said.round_trips.as_ref()?.get(index).copied()
             })
             .unwrap_or(rtt::UNKNOWN_BUCKET)