Keep recovery from closing newer connections

Author: myobie

now.md

@@ -121,6 +121,33 @@ peer dial and generation exchange. A deterministic red test proved that one
 slow peer open blocks an existing healthy peer. Branch
 `fix/peer-lock-head-of-line` narrows this serialization to one peer.
 
+PR #172 merged that isolation as `5dbe595`. Its focused red test and all three
+GitHub checks passed. The existing restart recovery test then exposed a real
+regression. It passed 4 of 4 times on parent `475807e`, 2 of 4 times on the PR
+tree, and 0 of 2 times in GitHub CI.
+
+The first race let a failed probe request recovery while the peer connection
+was still opening. Recovery waited for the peer gate and then closed the new
+successful connection. A second form let the connection enter the map between
+the failed probe and recovery. Recovery now closes only a connection that
+existed when that failed probe started.
+
+The peer gate also added two asynchronous lock operations to every cached
+connection use. A retained failed run kept one live canonical connection on
+both peers. The bus archive bytes arrived, but a concurrent older inbox entry
+returned before the local rename scan recorded its delete. The cached path now
+uses only the connection-map lock. The per-peer gate remains on replacement,
+dial, incoming registration, and stream-failure close operations.
+
+The corrected tree passed 4 of 4 serial restart recovery samples. The samples
+took 130.36, 129.42, 129.55, and 130.05 seconds. The focused recovery test, all
+15 active mux tests, and simultaneous-peer integration also passed.
+
+The original red test proved the lock fix removed cross-peer blocking. It did
+not prove the change preserved restart convergence. A focused test can prove
+its named behavior while the existing property suite finds a different loss.
+Keep both kinds of proof.
+
 The local library suite passed 513 tests with five measurements ignored. The
 binary suite passed 19 tests, and the update contract passed eight tests. The
 five-second latency property measured 502 records, seven scans, a 10.001166 ms

src/daemon.rs

@@ -1808,11 +1808,17 @@ impl DaemonState {
 
     /// Close only the failed peer's cached connection. The next probe opens a new
     /// connection and refreshes discovery without touching any other peer.
-    async fn recover_unreachable_peer(&self, peer_id: EndpointId, label: &str, attempt: usize) {
+    async fn recover_unreachable_peer(
+        &self,
+        peer_id: EndpointId,
+        label: &str,
+        attempt: usize,
+        failed_probe_started: Instant,
+    ) {
         let endpoint = self.endpoint_handle();
         let redialled = self
             .peer_connections
-            .redial(peer_id, b"peer health recovery")
+            .redial_opened_before(peer_id, b"peer health recovery", failed_probe_started)
             .await;
         warn!(
             target: VALIDATION_LOG_TARGET,
@@ -2760,9 +2766,10 @@ async fn run_peer_health_loop_with(
                                 "recent application traffic proved peer liveness"
                             ),
                         }
-                        round.push((peer_id, label, true, peer.roaming));
+                        round.push((peer_id, label, true, peer.roaming, Instant::now()));
                         continue;
                     }
+                    let probe_started = Instant::now();
                     let health = state.check_peer_reachability(&peer).await;
                     match peer_probe_log_action(
                         &mut roaming_away,
@@ -2852,14 +2859,20 @@ async fn run_peer_health_loop_with(
                             "persistent path degradation closed the shared peer connection"
                         );
                     }
-                    round.push((peer_id, label, health.reachable, peer.roaming));
+                    round.push((
+                        peer_id,
+                        label,
+                        health.reachable,
+                        peer.roaming,
+                        probe_started,
+                    ));
                 }
-                for (peer_id, label, reachable, roaming) in round {
+                for (peer_id, label, reachable, roaming, probe_started) in round {
                     if let PeerHealthAction::Recover { attempt } =
                         tracker.on_probe(peer_id, reachable, roaming, Instant::now())
                     {
                         state
-                            .recover_unreachable_peer(peer_id, &label, attempt)
+                            .recover_unreachable_peer(peer_id, &label, attempt, probe_started)
                             .await;
                     }
                 }
@@ -5914,14 +5927,18 @@ mod tests {
         let bluey = iroh::SecretKey::generate().public();
         let before = state.endpoint_handle().generation;
 
-        state.recover_unreachable_peer(bluey, "bluey", 3).await;
+        state
+            .recover_unreachable_peer(bluey, "bluey", 3, Instant::now())
+            .await;
         assert_eq!(
             state.endpoint_handle().generation,
             before,
             "one peer's absence must never replace the shared endpoint"
         );
 
-        state.recover_unreachable_peer(bluey, "bluey", 100).await;
+        state
+            .recover_unreachable_peer(bluey, "bluey", 100, Instant::now())
+            .await;
         assert_eq!(state.endpoint_handle().generation, before);
 
         node.shutdown().await?;
@@ -5956,7 +5973,7 @@ mod tests {
             };
             assert_eq!(attempt, expected_attempt);
             state
-                .recover_unreachable_peer(bluey, "bluey", attempt)
+                .recover_unreachable_peer(bluey, "bluey", attempt, now)
                 .await;
         }
 
@@ -5972,7 +5989,7 @@ mod tests {
         };
         assert_eq!(attempt, OLD_RECYCLE_ATTEMPT);
         state
-            .recover_unreachable_peer(bluey, "bluey", attempt)
+            .recover_unreachable_peer(bluey, "bluey", attempt, now)
             .await;
 
         assert_eq!(
@@ -6971,7 +6988,7 @@ mod tests {
         let absent = iroh::SecretKey::generate().public();
         let drops_before = *state.tunnel_drop_tx.borrow();
         state
-            .recover_unreachable_peer(absent, "absent-peer", 100)
+            .recover_unreachable_peer(absent, "absent-peer", 100, Instant::now())
             .await;
         let drops_after = *state.tunnel_drop_tx.borrow();
         assert_eq!(drops_after, drops_before);
@@ -7562,7 +7579,12 @@ mod tests {
             .await;
         // One peer away while the other answers: cheap recovery, no teardown.
         state
-            .recover_unreachable_peer(iroh::SecretKey::generate().public(), "absent", 3)
+            .recover_unreachable_peer(
+                iroh::SecretKey::generate().public(),
+                "absent",
+                3,
+                Instant::now(),
+            )
             .await;
         // And a recycle attempt, which is what a failed health poll ends in.
         let outcome = state

src/mux.rs

@@ -643,6 +643,24 @@ impl PeerConnections {
         protocol: &str,
         _activity: StreamActivity,
     ) -> Result<Option<Connection>> {
+        {
+            let conns = self.conns.lock().await;
+            if let Some(existing) = conns.get(&peer_addr.id)
+                && existing.generation == generation
+                && existing.connection.close_reason().is_none()
+            {
+                return Ok(Some(existing.connection.clone()));
+            }
+            if let Some(existing) = conns.get(&peer_addr.id)
+                && existing.generation > generation
+            {
+                bail!(
+                    "endpoint generation changed before mux open: requested {generation}, current {}",
+                    existing.generation
+                );
+            }
+        }
+
         let peer_gate = self.peer_gate(peer_addr.id).await;
         let _peer_guard = peer_gate.lock().await;
         let mut conns = self.conns.lock().await;
@@ -731,14 +749,29 @@ impl PeerConnections {
 
     /// Make one peer use a fresh multipath connection on its next stream.
     pub async fn redial(&self, peer: EndpointId, reason: &[u8]) -> bool {
-        let peer_gate = self.peer_gate(peer).await;
-        let _peer_guard = peer_gate.lock().await;
-        let removed = self.conns.lock().await.remove(&peer);
-        if let Some(removed) = removed {
-            removed.connection.close(0u32.into(), reason);
-            return true;
+        self.redial_opened_before(peer, reason, Instant::now()).await
+    }
+
+    /// Close only a connection that existed before the failed probe started.
+    pub async fn redial_opened_before(
+        &self,
+        peer: EndpointId,
+        reason: &[u8],
+        probe_started: Instant,
+    ) -> bool {
+        let mut connections = self.conns.lock().await;
+        let Some(current) = connections.get(&peer) else {
+            return false;
+        };
+        if current.opened_at > probe_started {
+            return false;
         }
-        false
+        let removed = connections
+            .remove(&peer)
+            .expect("the matched mux connection disappeared while locked");
+        drop(connections);
+        removed.connection.close(0u32.into(), reason);
+        true
     }
 
     /// Cache an accepted peer connection for traffic in the reverse direction.
@@ -1180,6 +1213,73 @@ mod tests {
         Ok(())
     }
 
+    /// Recovery must not close a connection that opened after the failed probe.
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    async fn recovery_does_not_close_a_connection_opened_after_the_probe() -> Result<()> {
+        let client = Endpoint::builder(presets::N0).bind().await?;
+        let (opened_tx, _opened_rx) = mpsc::unbounded_channel();
+        let manager = Arc::new(PeerConnections::new(client.id(), opened_tx));
+        let entered = Arc::new(tokio::sync::Barrier::new(2));
+        let release = Arc::new(tokio::sync::Notify::new());
+        let slow = Router::builder(
+            Endpoint::builder(presets::N0)
+                .alpns(vec![MUX_ALPN.to_vec()])
+                .bind()
+                .await?,
+        )
+        .accept(
+            MUX_ALPN,
+            BlockedGenerationMuxEcho {
+                entered: entered.clone(),
+                release: release.clone(),
+            },
+        )
+        .spawn();
+        slow.endpoint().online().await;
+
+        let peer = slow.endpoint().id();
+        let failed_probe_started = Instant::now();
+        let slow_manager = manager.clone();
+        let slow_client = client.clone();
+        let slow_addr = slow.endpoint().addr();
+        let slow_open = tokio::spawn(async move {
+            slow_manager
+                .open_mux_stream(&slow_client, 0, &slow_addr, "slow", StreamActivity::Probe)
+                .await
+        });
+        tokio::time::timeout(Duration::from_secs(3), entered.wait())
+            .await
+            .context("the slow peer did not reach generation exchange")?;
+
+        release.notify_one();
+        let stream = slow_open.await??;
+        let stable_id = stream.connection.stable_id();
+        drop(stream);
+        let recovery = manager
+            .redial_opened_before(peer, b"peer health recovery", failed_probe_started)
+            .await;
+
+        assert_eq!(
+            recovery,
+            false,
+            "recovery closed a connection that opened after the failed probe started"
+        );
+        assert_eq!(
+            manager.connection(peer).await.map(|conn| conn.stable_id()),
+            Some(stable_id),
+            "recovery closed the connection that completed after it started"
+        );
+        assert!(
+            manager.redial(peer, b"current connection recovery").await,
+            "recovery did not close the connection present when it started"
+        );
+        assert!(manager.connection(peer).await.is_none());
+
+        slow.shutdown().await?;
+        client.close().await;
+        Ok(())
+    }
+
     async fn assert_stream_failure_recovery(
         failures: usize,
         expect_same_connection: bool,