fix(sync): clear connect state when the remote aborts a dial as already-syncing

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(

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));
+    }
+}