fix(network): ignore stale Zakura peer teardowns

zebra-network/src/zakura/header_sync/events.rs

@@ -190,7 +190,18 @@ pub enum HeaderSyncEvent {
     /// A peer became available for stream-5 header sync.
     PeerConnected(HeaderSyncPeerSession),
     /// A peer disconnected; all of its outstanding work is dropped.
-    PeerDisconnected(ZakuraPeerId),
+    PeerDisconnected {
+        /// The disconnected peer.
+        peer: ZakuraPeerId,
+        /// Registration id of the connection whose session is tearing down, or
+        /// `None` to disconnect unconditionally.
+        ///
+        /// A `Some` id is ignored when the reactor's admitted session rides on
+        /// a different (newer) connection: duplicate arbitration replaced the
+        /// sender's connection, and its late teardown must not remove the
+        /// replacement session.
+        registration_id: Option<u64>,
+    },
     /// First-party header-sync summary observed over the authenticated discovery stream.
     AdvisoryHeaderSummary {
         /// Peer that supplied its own summary.
@@ -337,7 +348,7 @@ impl HeaderSyncEvent {
     pub(super) fn metrics_label(&self) -> &'static str {
         match self {
             Self::PeerConnected(_) => "peer_connected",
-            Self::PeerDisconnected(_) => "peer_disconnected",
+            Self::PeerDisconnected { .. } => "peer_disconnected",
             Self::AdvisoryHeaderSummary { .. } => "advisory_header_summary",
             Self::FullBlockCommitted { .. } => "full_block_committed",
             Self::NewBlockAccepted { .. } => "new_block_accepted",
@@ -347,6 +358,7 @@ impl HeaderSyncEvent {
             Self::WireDecodeFailed { .. } => "wire_decode_failed",
             Self::WireProtocolFailure { .. } => "wire_protocol_failure",
             Self::StateFrontiersChanged(_) => "state_frontiers_changed",
+            Self::BestHeaderHistoryTreeLoaded { .. } => "best_header_history_tree_loaded",
             Self::HeaderRangeCommitted { .. } => "header_range_committed",
             Self::HeaderRangeCommitFailed { .. } => "header_range_commit_failed",
             Self::HeaderRangeResponseFinished { .. } => "header_range_response_finished",

zebra-network/src/zakura/header_sync/pipe.rs

@@ -416,7 +416,10 @@ mod tests {
     fn saturated_events_handle() -> (HeaderSyncHandle, mpsc::Receiver<HeaderSyncEvent>) {
         let (events, events_rx) = mpsc::channel(1);
         events
-            .try_send(HeaderSyncEvent::PeerDisconnected(peer()))
+            .try_send(HeaderSyncEvent::PeerDisconnected {
+                peer: peer(),
+                registration_id: None,
+            })
             .expect("the single events slot is free");
         let (lifecycle, _lifecycle_rx) = mpsc::unbounded_channel();
         let (_tip_tx, tip) = watch::channel((block::Height(0), block::Hash([0; 32])));

zebra-network/src/zakura/header_sync/reactor.rs

@@ -149,7 +149,10 @@ impl HeaderSyncReactor {
     async fn handle_event_inner(&mut self, event: HeaderSyncEvent) {
         match event {
             HeaderSyncEvent::PeerConnected(session) => self.handle_peer_connected(session).await,
-            HeaderSyncEvent::PeerDisconnected(peer) => self.handle_peer_disconnected(peer),
+            HeaderSyncEvent::PeerDisconnected {
+                peer,
+                registration_id,
+            } => self.handle_peer_disconnected(peer, registration_id),
             HeaderSyncEvent::AdvisoryHeaderSummary { peer, summary } => {
                 self.handle_advisory_header_summary(peer, summary)
             }
@@ -468,7 +471,24 @@ impl HeaderSyncReactor {
         self.schedule().await;
     }
 
-    fn handle_peer_disconnected(&mut self, peer: ZakuraPeerId) {
+    fn handle_peer_disconnected(&mut self, peer: ZakuraPeerId, registration_id: Option<u64>) {
+        // A session-scoped disconnect from a connection that duplicate
+        // arbitration already displaced must not remove the winning
+        // connection's admitted session.
+        if let (Some(event_registration_id), Some(peer_state)) =
+            (registration_id, self.state.peers.get(&peer))
+        {
+            if peer_state.session.registration_id() != event_registration_id {
+                tracing::debug!(
+                    ?peer,
+                    event_registration_id,
+                    current_registration_id = peer_state.session.registration_id(),
+                    "ignoring stale header-sync disconnect from a displaced connection"
+                );
+                return;
+            }
+        }
+
         self.state.peers.remove(&peer);
         self.state.parked_peers.remove(&peer);
         self.state.advisory.remove(&peer);
@@ -1127,8 +1147,8 @@ impl HeaderSyncReactor {
         // Tree-aux roots are an optional serving capability: a peer may legitimately have no
         // roots for a root-carrying request (it can never serve the root for its own tip, and
         // may not persist roots at all). A completely rootless non-empty response is therefore
-        // not misbehavior — the headers are unusable for this root-carrying range, so drop them,
-        // release the request, and retry the range without scoring the peer.
+        // not misbehavior — the headers are unusable for this root-carrying range, so drop them
+        // and retry the range after the same short backoff as an empty response.
         if outstanding.range.want_tree_aux_roots && !headers.is_empty() && tree_aux_roots.is_empty()
         {
             metrics::counter!("sync.header.response.rootless").increment(1);
@@ -1143,9 +1163,7 @@ impl HeaderSyncReactor {
                 outstanding.range.want_tree_aux_roots,
                 0,
             );
-            self.state.schedule.clear_assignment(outstanding.range);
-            self.state.schedule.retry(outstanding.range);
-            self.schedule().await;
+            self.delay_unusable_response_retry(&peer, outstanding);
             return;
         }
 
@@ -1171,7 +1189,6 @@ impl HeaderSyncReactor {
 
         if headers.is_empty() {
             self.record_advisory_unconfirmed(&peer);
-            let deadline = Instant::now() + self.empty_headers_retry_delay();
             self.trace_headers_received(
                 &peer,
                 outstanding.range.start_height,
@@ -1182,13 +1199,7 @@ impl HeaderSyncReactor {
                 outstanding.range.want_tree_aux_roots,
                 u32::try_from(tree_aux_roots.len()).unwrap_or(u32::MAX),
             );
-            if let Some(peer_state) = self.state.peers.get_mut(&peer) {
-                peer_state.outstanding.push(OutstandingRange {
-                    deadline,
-                    clear_assignment_on_timeout: true,
-                    ..outstanding
-                });
-            }
+            self.delay_unusable_response_retry(&peer, outstanding);
             return;
         }
 
@@ -1575,6 +1586,21 @@ impl HeaderSyncReactor {
         self.startup.request_timeout.min(EMPTY_HEADERS_RETRY_DELAY)
     }
 
+    fn delay_unusable_response_retry(
+        &mut self,
+        peer: &ZakuraPeerId,
+        outstanding: OutstandingRange,
+    ) {
+        let deadline = Instant::now() + self.empty_headers_retry_delay();
+        if let Some(peer_state) = self.state.peers.get_mut(peer) {
+            peer_state.outstanding.push(OutstandingRange {
+                deadline,
+                clear_assignment_on_timeout: true,
+                ..outstanding
+            });
+        }
+    }
+
     async fn schedule(&mut self) {
         if !self.startup.range_state_actions_enabled {
             return;
@@ -1930,7 +1956,7 @@ impl HeaderSyncReactor {
                 insert_optional_str(row, hs_trace::KIND, Some("peer_connected"));
                 insert_peer(row, hs_trace::PEER, session.peer_id());
             }
-            HeaderSyncEvent::PeerDisconnected(peer) => {
+            HeaderSyncEvent::PeerDisconnected { peer, .. } => {
                 insert_optional_str(row, hs_trace::KIND, Some("peer_disconnected"));
                 insert_peer(row, hs_trace::PEER, peer);
             }

zebra-network/src/zakura/header_sync/service.rs

@@ -36,6 +36,12 @@ pub(crate) fn header_sync_streams() -> &'static [Stream] {
 pub struct HeaderSyncPeerSession {
     peer_id: ZakuraPeerId,
     direction: ServicePeerDirection,
+    /// Supervisor registration id of the connection this session rides on.
+    ///
+    /// Used to drop stale `PeerDisconnected` teardown events from a session
+    /// whose connection was displaced by a winning duplicate. Zero for test
+    /// sessions built without a supervisor registration.
+    registration_id: u64,
     inner: Arc<HeaderSyncPeerSessionInner>,
 }
 
@@ -50,15 +56,25 @@ impl HeaderSyncPeerSession {
     fn new_with_commands(
         session: &PeerStreamSession,
         direction: ServicePeerDirection,
+        registration_id: u64,
         commands: mpsc::UnboundedSender<HeaderSyncPeerCommand>,
     ) -> Self {
-        Self::from_parts_with_direction_and_commands(
+        let mut this = Self::from_parts_with_direction_and_commands(
             session.peer_id().clone(),
             direction,
             session.sender(),
             session.cancel_token(),
             Some(commands),
-        )
+        );
+        this.registration_id = registration_id;
+        this
+    }
+
+    /// Set the supervisor registration id of this session's connection.
+    #[cfg(test)]
+    pub(crate) fn with_registration_id(mut self, registration_id: u64) -> Self {
+        self.registration_id = registration_id;
+        self
     }
 
     #[cfg(test)]
@@ -91,6 +107,7 @@ impl HeaderSyncPeerSession {
         Self {
             peer_id,
             direction,
+            registration_id: 0,
             inner: Arc::new(HeaderSyncPeerSessionInner {
                 send,
                 cancel_token,
@@ -110,6 +127,7 @@ impl HeaderSyncPeerSession {
         Self {
             peer_id,
             direction,
+            registration_id: 0,
             inner: Arc::new(HeaderSyncPeerSessionInner {
                 send,
                 cancel_token,
@@ -128,6 +146,11 @@ impl HeaderSyncPeerSession {
         self.direction
     }
 
+    /// Supervisor registration id of the connection this session rides on.
+    pub(crate) fn registration_id(&self) -> u64 {
+        self.registration_id
+    }
+
     /// Peer disconnect/local shutdown cancellation token.
     pub fn cancel_token(&self) -> CancellationToken {
         self.inner.cancel_token.clone()
@@ -344,6 +367,7 @@ impl Service for HeaderSyncService {
         };
 
         let peer_id = peer.id.clone();
+        let registration_id = peer.registration_id;
         let session = PeerStreamSession::new(
             peer_id.clone(),
             ZAKURA_STREAM_HEADER_SYNC,
@@ -361,8 +385,12 @@ impl Service for HeaderSyncService {
         let close_cause = peer.close_cause();
         let conn_id = peer.conn_id;
         let (commands_tx, commands_rx) = mpsc::unbounded_channel();
-        let header_sync_session =
-            HeaderSyncPeerSession::new_with_commands(&session, peer.direction, commands_tx);
+        let header_sync_session = HeaderSyncPeerSession::new_with_commands(
+            &session,
+            peer.direction,
+            registration_id,
+            commands_tx,
+        );
 
         {
             let mut peers = self
@@ -447,8 +475,10 @@ impl Service for HeaderSyncService {
                 }
             };
             if should_notify {
-                let _ = teardown_handle
-                    .send_lifecycle(HeaderSyncEvent::PeerDisconnected(teardown_peer));
+                let _ = teardown_handle.send_lifecycle(HeaderSyncEvent::PeerDisconnected {
+                    peer: teardown_peer,
+                    registration_id: Some(registration_id),
+                });
             }
         };
         let panic_connection_cancel_token = connection_cancel_token.clone();
@@ -482,7 +512,10 @@ impl Service for HeaderSyncService {
             record.cancel_token.cancel();
             let _ = self
                 .header_sync
-                .send_lifecycle(HeaderSyncEvent::PeerDisconnected(peer.clone()));
+                .send_lifecycle(HeaderSyncEvent::PeerDisconnected {
+                    peer: peer.clone(),
+                    registration_id: None,
+                });
         }
     }