From a214048b176d5febaf95e68610fff77ff6bb0f97 Mon Sep 17 00:00:00 2001 From: Vladimir Logachev Date: Tue, 14 Jul 2026 20:11:49 +0400 Subject: [PATCH] fix(sync): clear connect state when the remote aborts a dial as already-syncing --- src/engine/live.rs | 5 ++ src/engine/state.rs | 116 ++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 121 insertions(+) diff --git a/src/engine/live.rs b/src/engine/live.rs index 7fe587fc..e58a65d9 100644 --- a/src/engine/live.rs +++ b/src/engine/live.rs @@ -497,6 +497,11 @@ impl LiveActor { match result { Err(ConnectError::RemoteAbort(AbortReason::AlreadySyncing)) => { debug!(?reason, "remote abort, already syncing"); + // The remote refused our dial because it sees an exchange with us already + // running. Nothing else will finish the dial recorded in our state, so clear + // it — otherwise this (namespace, peer) pair stays `Running` forever and every + // later sync trigger for it is silently dropped. + self.state.abort_connect(&namespace, peer, reason); } res => { self.on_sync_finished( diff --git a/src/engine/state.rs b/src/engine/state.rs index 6e040d26..166be5bd 100644 --- a/src/engine/state.rs +++ b/src/engine/state.rs @@ -84,6 +84,25 @@ impl NamespaceStates { } } + /// Clear the running outgoing exchange with `node` after the remote aborted it as + /// already-syncing. Nothing else will finish that exchange, so without this the pair would + /// stay running forever and every later sync trigger for it would be silently dropped. + /// Left untouched if a concurrent incoming exchange took the slot over (mutual-dial + /// tie-break): that exchange finishes through the accept path. + /// + /// Returns true if the running outgoing exchange was cleared. + pub fn abort_connect( + &mut self, + namespace: &NamespaceId, + node: EndpointId, + reason: SyncReason, + ) -> bool { + match self.entry(namespace, node) { + None => false, + Some(state) => state.abort_connect(reason), + } + } + /// Accept a sync request. /// /// Returns the [`AcceptOutcome`] to be performed. @@ -212,6 +231,19 @@ impl PeerState { } } + fn abort_connect(&mut self, reason: SyncReason) -> bool { + match &self.state { + SyncState::Running { + origin: Origin::Connect(running_reason), + .. + } if *running_reason == reason => { + self.state = SyncState::Idle; + true + } + _ => false, + } + } + fn accept_request(&mut self, me: &EndpointId, node: &EndpointId) -> AcceptOutcome { let outcome = match &self.state { SyncState::Idle => AcceptOutcome::Allow, @@ -254,3 +286,87 @@ fn expected_sync_direction(self_node_id: &EndpointId, other_node_id: &EndpointId SyncDirection::Connect } } + +#[cfg(test)] +mod tests { + use iroh::SecretKey; + + use super::*; + + fn namespace() -> NamespaceId { + NamespaceId::from(&[7u8; 32]) + } + + /// Two deterministic endpoint ids, returned as (lower, higher) by key bytes. + fn node_pair() -> (EndpointId, EndpointId) { + let a = SecretKey::from_bytes(&[1u8; 32]).public(); + let b = SecretKey::from_bytes(&[2u8; 32]).public(); + if a.as_bytes() < b.as_bytes() { + (a, b) + } else { + (b, a) + } + } + + /// A dial the remote rejected as already-syncing leaves nothing running that could + /// finish the exchange. `abort_connect` must return the pair to idle so the next + /// trigger is not silently dropped — without it the pair stays `Running` forever + /// and every later sync trigger for it is lost. + #[test] + fn abort_connect_returns_rejected_dial_to_idle() { + let namespace = namespace(); + let (peer, _) = node_pair(); + let mut states = NamespaceStates::default(); + states.insert(namespace); + + assert!(states.start_connect(&namespace, peer, SyncReason::DirectJoin)); + // While the dial is in flight the pair is busy: triggers are dropped. + assert!(!states.start_connect(&namespace, peer, SyncReason::NewNeighbor)); + + // Remote answered AlreadySyncing: the dial is dead, clear it. + assert!(states.abort_connect(&namespace, peer, SyncReason::DirectJoin)); + + // The pair accepts sync triggers again. + assert!(states.start_connect(&namespace, peer, SyncReason::NewNeighbor)); + } + + /// An abort whose reason does not match the running dial belongs to an older + /// exchange; it must not clear the current one. + #[test] + fn stale_abort_does_not_clear_newer_dial() { + let namespace = namespace(); + let (peer, _) = node_pair(); + let mut states = NamespaceStates::default(); + states.insert(namespace); + + assert!(states.start_connect(&namespace, peer, SyncReason::DirectJoin)); + assert!(states.abort_connect(&namespace, peer, SyncReason::DirectJoin)); + assert!(states.start_connect(&namespace, peer, SyncReason::NewNeighbor)); + + // Late abort for the first dial: reason mismatch, nothing cleared. + assert!(!states.abort_connect(&namespace, peer, SyncReason::DirectJoin)); + // The newer dial is still running. + assert!(!states.start_connect(&namespace, peer, SyncReason::NewNeighbor)); + } + + /// Mutual dial where the incoming exchange won the tie-break and took the slot + /// over: the remote abort of our own dial must leave the accept exchange + /// untouched — it finishes through the accept path. + #[test] + fn abort_connect_spares_accept_takeover() { + let namespace = namespace(); + // `me` compares higher, so the incoming request from `node` wins the tie-break. + let (node, me) = node_pair(); + let mut states = NamespaceStates::default(); + states.insert(namespace); + + assert!(states.start_connect(&namespace, node, SyncReason::DirectJoin)); + let outcome = states.accept_request(&me, &namespace, node); + assert!(matches!(outcome, AcceptOutcome::Allow)); + + // Our dial comes back rejected; the slot now belongs to the accept exchange. + assert!(!states.abort_connect(&namespace, node, SyncReason::DirectJoin)); + // Still busy until the accept exchange finishes. + assert!(!states.start_connect(&namespace, node, SyncReason::DirectJoin)); + } +}