fix(netmon): re-derive a dropped change instead of losing the peers it named

The detector advanced its baseline immediately before offering the change
to the channel, and the channel holds one slot: when it is full the newer
change is discarded with a debug line. That was sound while the reaction
was node-wide, because any later change would rebind everything anyway,
and the doc on the spawning function said so in as many words.

Scoping the reaction to the peers a change names broke it. The handler
now reads only that peer set, so a discarded change loses those peers,
and because the baseline had already moved the ordinary diff compares the
post-change sample against itself and never names them again. Nothing
else repairs them: the session activation early-returns while a connected
socket is present, and the send path prefers that socket, so the peer
transmits from the abandoned source until the liveness timeout reaps the
peering.

The baseline now advances only on a send that was accepted. Holding it
where it was makes the next sample re-derive the move, and repairs are
idempotent, so re-reporting a peer that was already handled costs nothing
and stops as soon as the queue drains. The alternative, restoring the
previous baseline on a full channel, needs the map cloned on every
reported change to serve the rare case, so it was not taken.

The coalescing test could not see any of this: all three of its scripted
fingerprints named the same peer, so it passed under either policy. It is
rescripted over disjoint peer sets, and it now fails on the old ordering
with a message that names the defect.

Three prose corrections in the same module, all of them consequences of
the same scoping change. The claim that the handler still re-evaluates
every peer is now false and is replaced by what the handler does. The
first-sight residual said the ordinary diff recovers the peer on the next
sample, which it does not, because the socket's pinned source is read
only on the first-sight arm; the bound is the liveness timeout, and the
node-wide reaction is what used to repair it as collateral. And the
sampling cost is five syscalls per target, read off the sampling function
rather than measured; the earlier correction reached one site of three.

src/node/netmon/mod.rs

@@ -51,9 +51,15 @@
 //! One local source address per peer: for every peer whose transport address is
 //! a numeric IP endpoint, the address the kernel would pick to reach *that
 //! peer*. A connected-but-never-sending UDP socket makes the kernel run its
-//! route lookup and bind the source address it would use; three syscalls, no
-//! packets, no name resolution, and it works identically on every platform std
-//! supports.
+//! route lookup and bind the source address it would use; five syscalls, read
+//! off [`NetFingerprint::sample`] rather than measured, no packets, no name
+//! resolution, and it works identically on every platform std supports.
+//!
+//! Keying on peers bounds the *reaction* — only the peers a change names are
+//! acted on — and does not bound the *sampling*. One roaming peer still makes
+//! the detector re-probe every target, up to [`MAX_DEBOUNCE_ROUNDS`] extra
+//! times per wake, and the only limit on how often that can start is
+//! `node.netmon.poll_interval_secs`.
 //!
 //! That set is exactly the quantity the reaction cares about. The stale
 //! `connect(2)` this whole subsystem exists to repair pinned a *local source
@@ -122,7 +128,7 @@
 //! is reported when some peer present in both samples is now reached from a
 //! different local address. Peers joining and leaving is ordinary node
 //! behaviour and says nothing about the medium, so on its own it must not fire
-//! a reaction that drops every connected socket. A peer whose probe stops
+//! a reaction that tears down a working send path. A peer whose probe stops
 //! answering entirely — the route to it is gone — is a move to "no source
 //! address" and does count, because that peer is exactly the one now stranded.
 //!
@@ -292,9 +298,9 @@ impl NetFingerprint {
     /// The comparison is over the *intersection* of the two peer sets, never
     /// the union: a peer that has only just been authenticated, or one that has
     /// just been reaped, differs between the samples for reasons that have
-    /// nothing to do with the host's attachment, and the reaction — drop every
-    /// connected socket, heartbeat every peer — is far too blunt to fire on
-    /// ordinary peer churn.
+    /// nothing to do with the host's attachment, and the reaction — drop the
+    /// peer's connected socket and heartbeat it — is pure churn on a peer whose
+    /// send path was never stale.
     ///
     /// **A peer only in the newer sample** has no previous probe answer to be
     /// compared against, and skipping it outright leaves a hole this detector
@@ -332,10 +338,16 @@ impl NetFingerprint {
     /// on tick N is first visible to this detector in the snapshot published on
     /// tick N+1. A peer that joins and has its socket installed, and whose path
     /// then moves before that next publish, is first seen with `bound` still
-    /// `None` and is skipped — and it *does* hold a pinned socket. About one
-    /// `tick_interval_secs` per join against a `poll_interval_secs` five times
-    /// longer, and the peer is recovered by the ordinary diff on the sample
-    /// after, so it is bounded rather than permanent.
+    /// `None` and is skipped — and it *does* hold a pinned socket. The window
+    /// in which it can be missed is about one `tick_interval_secs` per join,
+    /// against a `poll_interval_secs` five times longer, but the *consequence*
+    /// is not so short: the ordinary diff does not recover the peer. Once it is
+    /// in both samples it is judged on its probe answer alone, and `bound` is
+    /// consulted only on the first-sight arm, so the socket stays pinned where
+    /// it was. While the reaction was node-wide such a peer was repaired as
+    /// collateral the next time any other peer moved; scoping the reaction to
+    /// the peers a change names removed that. What bounds it now is
+    /// `node.link_dead_timeout_secs` reaping the peering.
     ///
     /// A non-wildcard `transports.udp.bind_addr` is not part of the residual;
     /// see [`ProbeTarget::bind`], which keeps both sides answering the same
@@ -386,6 +398,16 @@ pub(in crate::node) struct ProbeTarget {
     /// unconstrained probe would answer with the kernel's choice instead, and
     /// the two would disagree permanently — reporting a first-sight move, on
     /// every peer, forever, with nothing having moved.
+    ///
+    /// It does not preserve per-peer detection under such a bind, and should
+    /// not be read as if it did. [`preferred_source`] returns the configured
+    /// address for every destination, which is the same value the pinned socket
+    /// already holds, so a route moving under one peer while that address stays
+    /// configured on the host is invisible here. What is still reported is the
+    /// bind address itself going away, which moves every peer at once. That is
+    /// a trade rather than a loss: the send path is pinned to the configured
+    /// address whatever the routing table says, so there is no per-peer pinning
+    /// left to repair.
     pub bind: Option<IpAddr>,
 }
 
@@ -403,9 +425,9 @@ pub(crate) struct PeerSourceMove {
 /// What moved between two fingerprints.
 ///
 /// Operator-facing, and — unlike the host-wide summary this replaced — it now
-/// names the peers affected, because the fingerprint is keyed on them. The
-/// handler still re-evaluates every peer regardless; narrowing the reaction to
-/// exactly [`Self::moved`] is a separate change.
+/// names the peers affected, because the fingerprint is keyed on them. It is
+/// also the reaction's whole input: the handler acts on exactly the peers in
+/// [`Self::moved`] and touches no others.
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub(crate) struct NetChangeSummary {
     /// Every peer whose local source address changed, in `NodeAddr` order.
@@ -640,11 +662,18 @@ impl WakeSource {
 /// Spawn the medium-change detector, using the best backend this platform has.
 ///
 /// Returns the receiver the rx loop drains and the task handle the supervisor
-/// aborts at teardown. The channel holds a single slot: a change already queued
-/// and not yet handled makes a newer one redundant, because the handler's
-/// reaction is "re-evaluate every peer and every backoff", which subsumes any
-/// number of coalesced changes. A full channel therefore drops rather than
-/// queues, and never applies backpressure to the detector.
+/// aborts at teardown. The channel holds a single slot, and a full channel
+/// drops rather than queues: the detector must never apply backpressure to
+/// itself, and the node must never work through a backlog of network states
+/// the host has already left.
+///
+/// Dropping is only safe because the dropped change is not the last word on
+/// the peers it named. The handler acts on exactly those peers, so discarding
+/// one would strand them if the detector had already adopted the sample it was
+/// derived from. It has not: the baseline in [`run_detector`] advances only on
+/// a successful send, so the next sample re-derives the move against the same
+/// baseline and reports it again once the queue drains. Repairing a peer is
+/// idempotent, which is what makes re-reporting cheap rather than a loop.
 pub(crate) fn spawn_detector(
     cfg: NetmonConfig,
     peers: Arc<arc_swap::ArcSwap<EntitySnapshot>>,
@@ -807,10 +836,14 @@ where
                 probed: candidate.sources.len(),
             },
         };
-        last = candidate;
-
         match tx.try_send(change) {
-            Ok(()) => {}
+            // The baseline advances only here. A dropped change is a peer set
+            // nobody will act on, and the reaction is scoped to the peers a
+            // change names, so adopting `candidate` regardless would leave
+            // those peers unrepaired for good: the next diff would compare the
+            // post-change sample against itself and report nothing. Holding
+            // `last` where it was makes the next sample re-derive the move.
+            Ok(()) => last = candidate,
             Err(mpsc::error::TrySendError::Full(dropped)) => {
                 debug!(
                     generation = dropped.generation,

src/node/netmon/tests.rs

@@ -48,6 +48,11 @@ fn all_from(peers: &[NodeAddr], source: Option<IpAddr>) -> NetFingerprint {
     NetFingerprint::for_test(&sources)
 }
 
+/// The peers a change names, in the order its summary carries them.
+fn moved_peers(change: &NetChange) -> Vec<NodeAddr> {
+    change.summary.moved.iter().map(|m| m.peer).collect()
+}
+
 /// A timer-only wake source at the config's poll period — the portable
 /// backend's behaviour, and the baseline the netlink tests compare against.
 fn timer_wake(poll_secs: u64) -> WakeSource {
@@ -129,9 +134,9 @@ async fn a_peer_losing_its_route_is_reported() {
 #[tokio::test(start_paused = true)]
 async fn a_peer_joining_is_absorbed_rather_than_reported() {
     // Peer churn is ordinary node behaviour and says nothing about the medium.
-    // The reaction is to drop every connected socket and heartbeat every peer,
-    // so firing it every time a peer authenticates would make a busy node
-    // continuously tear down its own send fast path.
+    // The reaction drops the named peers' connected sockets and heartbeats
+    // them, so firing it every time a peer authenticates would make a busy node
+    // continuously tear down a send fast path that was never stale.
     let one = all_from(&[peer(1)], Some(v4(192, 168, 1, 10)));
     let two = all_from(&[peer(1), peer(2)], Some(v4(192, 168, 1, 10)));
     let (sampler, _) = scripted(vec![one, two]);
@@ -312,14 +317,25 @@ async fn a_flap_that_settles_back_reports_nothing() {
 
 #[tokio::test(start_paused = true)]
 async fn an_unread_change_coalesces_rather_than_queues() {
-    // The handler's reaction is "re-evaluate every peer and every backoff",
-    // which subsumes any number of changes. A second change arriving before the
-    // first is drained must therefore drop, not queue: the node must never work
-    // through a backlog of stale network states.
-    let a = all_from(&[peer(1)], Some(v4(192, 168, 1, 10)));
-    let b = all_from(&[peer(1)], Some(v4(10, 40, 0, 7)));
-    let c = all_from(&[peer(1)], Some(v4(172, 16, 3, 2)));
-    let (sampler, _) = scripted(vec![a, b, c]);
+    // Two changes over disjoint peer sets: the first moves peer(1) alone, the
+    // second moves peer(2) alone. A second change arriving before the first is
+    // drained must not queue — the node must never work through a backlog of
+    // stale network states — and it must not be silently lost either, because
+    // the reaction is scoped to the peers a change names and nothing else
+    // would ever repair peer(2).
+    let start = NetFingerprint::for_test(&[
+        (peer(1), Some(v4(192, 168, 1, 10))),
+        (peer(2), Some(v4(192, 168, 1, 11))),
+    ]);
+    let first = NetFingerprint::for_test(&[
+        (peer(1), Some(v4(10, 40, 0, 7))),
+        (peer(2), Some(v4(192, 168, 1, 11))),
+    ]);
+    let second = NetFingerprint::for_test(&[
+        (peer(1), Some(v4(10, 40, 0, 7))),
+        (peer(2), Some(v4(10, 40, 0, 8))),
+    ]);
+    let (sampler, _) = scripted(vec![start, first, second]);
     let (tx, mut rx) = mpsc::channel(1);
 
     tokio::spawn(run_detector(tx, cfg(1, 0), sampler, timer_wake(1)));
@@ -328,11 +344,29 @@ async fn an_unread_change_coalesces_rather_than_queues() {
     // is draining, so the second meets a full channel.
     tokio::time::sleep(Duration::from_secs(10)).await;
 
-    assert!(rx.try_recv().is_ok(), "the first change is delivered");
+    let queued = rx.try_recv().expect("the first change is delivered");
+    assert_eq!(
+        moved_peers(&queued),
+        vec![peer(1)],
+        "the queued change is the first one, which named only peer(1)"
+    );
     assert!(
         rx.try_recv().is_err(),
         "the second must have coalesced into the undrained first, not queued behind it"
     );
+
+    // Draining frees the slot. The dropped change named peer(2) and nobody
+    // acted on it, so the detector's baseline must not have advanced past it:
+    // the next sample has to re-derive that move and deliver it.
+    let recovered = tokio::time::timeout(QUIET_WINDOW, rx.recv())
+        .await
+        .expect("a dropped change must be re-derived once the slot frees, not lost with its peers")
+        .expect("the channel must stay open");
+    assert!(
+        moved_peers(&recovered).contains(&peer(2)),
+        "a dropped change must not lose the peers it named; got {:?}",
+        recovered.summary.moved
+    );
 }
 
 #[tokio::test(start_paused = true)]