diff --git a/zebra-network/src/zakura/handler.rs b/zebra-network/src/zakura/handler.rs index c5a68b24138..03bb4729498 100644 --- a/zebra-network/src/zakura/handler.rs +++ b/zebra-network/src/zakura/handler.rs @@ -910,17 +910,6 @@ struct ZakuraSupervisorState { next_registration_id: ZakuraConnId, } -#[derive(Debug)] -struct ZakuraPeerConnectionEntry { - /// Monotonic supervisor registration generation used by services to ignore - /// stale add/remove work from a superseded connection. - conn_id: ZakuraConnId, - outbound_handle: ZakuraPeerHandle, - disconnect_token: CancellationToken, - registered_at: Instant, - remote_ip: Option, -} - impl ZakuraSupervisorState { fn increment_ip(&mut self, remote_ip: Option) { if let Some(remote_ip) = remote_ip { @@ -956,6 +945,27 @@ impl ZakuraSupervisorState { fn debug_assert_accounting(&self) {} } +/// One authenticated peer's currently registered connection. +/// +/// The entry is bound to a per-supervisor monotonic connection generation, so +/// teardown ([`ZakuraSupervisorHandle::deregister`]) only removes state when the +/// exiting connection is still the registered one. Without this guard, a +/// displaced duplicate's late teardown would deregister the live winner. +#[derive(Debug)] +struct ZakuraPeerConnectionEntry { + /// Monotonic supervisor generation used by services to ignore stale work + /// from a superseded connection. + conn_id: ZakuraConnId, + outbound_handle: ZakuraPeerHandle, + disconnect_token: CancellationToken, + /// The remote IP this entry incremented in `active_by_ip`, released when + /// this exact entry is removed. + remote_ip: Option, + /// When this connection registered, used to decide whether a duplicate may + /// evict a stale incumbent (see [`ZAKURA_DUPLICATE_EVICT_MIN_AGE`]). + registered_at: Instant, +} + /// Queue-backed outbound send handle for one authenticated Zakura peer. #[derive(Clone, Debug)] pub struct ZakuraPeerHandle { @@ -1140,32 +1150,31 @@ impl ZakuraSupervisorHandle { ZakuraUpgradeOutcome::Upgraded { .. } => { let conn_id = state.next_registration_id; state.next_registration_id += 1; - let entry = ZakuraPeerConnectionEntry { - conn_id, - outbound_handle, - disconnect_token, - registered_at: Instant::now(), - remote_ip, - }; - if let Some(old_entry) = state.active_by_peer.insert(peer_id.clone(), entry) { + if let Some(old_entry) = state.active_by_peer.remove(&peer_id) { state.decrement_ip(old_entry.remote_ip); old_entry.disconnect_token.cancel(); - metrics::counter!("zakura.p2p.conn.duplicate.evicted_upgraded").increment(1); + metrics::counter!("zakura.p2p.conn.duplicate.displaced_incumbent").increment(1); } state.increment_ip(remote_ip); + state.active_by_peer.insert( + peer_id.clone(), + ZakuraPeerConnectionEntry { + conn_id, + outbound_handle, + disconnect_token: disconnect_token.clone(), + registered_at: Instant::now(), + remote_ip, + }, + ); state.debug_assert_accounting(); let registered_ids: Vec<_> = state.active_by_peer.keys().cloned().collect(); set_active_connection_gauge(registered_ids.len()); self.peer_set_tx.send_replace(registered_ids); - let disconnect_token = state - .active_by_peer - .get(&peer_id) - .map(|entry| entry.disconnect_token.clone()) - .expect("disconnect token exists because this peer was just registered"); ZakuraRegistration::Registered { conn_id, peer_id, remote_ip, + registration_id: conn_id, disconnect_token, } } @@ -1206,15 +1215,21 @@ impl ZakuraSupervisorHandle { } } - async fn deregister(&self, peer_id: &ZakuraPeerId, conn_id: ZakuraConnId) { + /// Removes a connection's registration, but only when `conn_id` still + /// identifies the currently registered connection for this peer. + /// + /// Returns true when the registration was removed. A displaced connection's + /// late teardown returns false and leaves the winning registration (and its + /// per-IP slot accounting) untouched. + async fn deregister(&self, peer_id: &ZakuraPeerId, conn_id: ZakuraConnId) -> bool { let mut state = self.inner.lock().await; let Some(entry) = state.active_by_peer.get(peer_id) else { state.debug_assert_accounting(); - return; + return false; }; if entry.conn_id != conn_id { state.debug_assert_accounting(); - return; + return false; } let entry = state .active_by_peer @@ -1226,6 +1241,7 @@ impl ZakuraSupervisorHandle { let registered_ids: Vec<_> = state.active_by_peer.keys().cloned().collect(); set_active_connection_gauge(registered_ids.len()); self.peer_set_tx.send_replace(registered_ids); + true } fn shutdown(&self) { @@ -1263,6 +1279,7 @@ enum ZakuraRegistration { conn_id: ZakuraConnId, peer_id: ZakuraPeerId, remote_ip: Option, + registration_id: u64, disconnect_token: CancellationToken, }, Duplicate { @@ -1427,6 +1444,10 @@ struct RegisteredConnectionServeContext { connection_token: CancellationToken, close_cause: CloseCause, accepted_capabilities: u64, + /// Supervisor registration id for this connection; teardown only + /// deregisters and fans out service removal while this id is still the + /// peer's registered connection. + registration_id: u64, /// Whether this side dialed the connection. The dialer (initiator) opens all /// of its demanded ordered streams as before; the responder additionally /// opens only block-sync (the sole symmetric service), keeping the legacy @@ -1966,8 +1987,8 @@ impl ZakuraProtocolHandler { should_run_freshness_reaper(queue_split_stream_count, request_response_stream_count); if negotiated_ordered_streams.is_empty() { - self.registry - .add_peer(Peer::new_with_conn_id_and_direction_and_close_cause( + self.registry.add_peer( + Peer::new_with_conn_id_and_direction_and_close_cause( conn_id, peer_id.clone(), remote_ip, @@ -1976,7 +1997,9 @@ impl ZakuraProtocolHandler { HashMap::new(), connection_token.clone(), close_cause.clone(), - )); + ) + .with_registration_id(context.registration_id), + ); cleanup_guard.add_admitted_capabilities(accepted_capabilities); } else if !connection_token.is_cancelled() { let mut opened_capabilities = 0; @@ -2025,18 +2048,19 @@ impl ZakuraProtocolHandler { // Escalation is already narrowed to opened ordered services. // Disconnect fanout still uses the registry's returned admitted // mask, not this peer context. - let admitted_capabilities = - self.registry - .add_escalated_peer(Peer::new_with_service_streams( - conn_id, - peer_id.clone(), - remote_ip, - opened_capabilities, - context.direction, - std::mem::take(&mut service_streams), - connection_token.clone(), - close_cause.clone(), - )); + let admitted_capabilities = self.registry.add_escalated_peer( + Peer::new_with_service_streams( + conn_id, + peer_id.clone(), + remote_ip, + opened_capabilities, + context.direction, + std::mem::take(&mut service_streams), + connection_token.clone(), + close_cause.clone(), + ) + .with_registration_id(context.registration_id), + ); cleanup_guard.add_admitted_capabilities(admitted_capabilities); } } @@ -2197,7 +2221,8 @@ impl ZakuraProtocolHandler { service_streams, connection_token.clone(), close_cause.clone(), - ), + ) + .with_registration_id(context.registration_id), ); cleanup_guard.add_admitted_capabilities(admitted_capabilities); } @@ -2611,6 +2636,7 @@ impl ZakuraProtocolHandler { conn_id, peer_id, remote_ip, + registration_id, disconnect_token, } => { metrics::counter!("zakura.p2p.conn.accepted", "role" => context.role).increment(1); @@ -2641,6 +2667,7 @@ impl ZakuraProtocolHandler { connection_token: disconnect_token, close_cause, accepted_capabilities: context.accepted_capabilities, + registration_id, is_initiator: context.role == "initiator", i_open_collision_winner: context.i_open_collision_winner, direction: context.direction, @@ -6450,6 +6477,222 @@ mod tests { Ok(()) } + async fn register_with_transcript( + supervisor: &ZakuraSupervisorHandle, + peer: &ZakuraPeerId, + ip: IpAddr, + transcript_hash: [u8; TRANSCRIPT_HASH_BYTES], + token: CancellationToken, + ) -> ZakuraRegistration { + let (outbound_tx, _outbound_rx) = mpsc::channel(1); + let outbound_handle = ZakuraPeerHandle::new_for_tests(peer.clone(), outbound_tx); + supervisor + .register( + test_conn_id(), + peer.clone(), + Some(ip), + transcript_hash, + outbound_handle, + token, + ZAKURA_CAP_LEGACY_GOSSIP | ZAKURA_CAP_HEADER_SYNC, + ) + .await + } + + // A duplicate connection whose transcript hash wins arbitration (`Upgraded`) + // replaces the incumbent registration. The displaced incumbent must be + // cancelled and release its per-IP slot immediately, and its later teardown + // (bound to the old registration id) must not deregister the winner. + #[tokio::test] + async fn winning_duplicate_displaces_incumbent_and_stale_teardown_is_ignored( + ) -> Result<(), BoxError> { + let supervisor = ZakuraSupervisorHandle::new(1); + let ip: IpAddr = "203.0.113.21".parse().expect("test ip parses"); + let peer = test_peer(21); + + // The incumbent registers with the larger transcript hash, so a + // cross-direction duplicate can win arbitration against it. + let incumbent_token = CancellationToken::new(); + let ZakuraRegistration::Registered { + registration_id: incumbent_id, + .. + } = register_with_transcript(&supervisor, &peer, ip, [2; 32], incumbent_token.clone()) + .await + else { + panic!("the incumbent registers"); + }; + + // The duplicate with the strictly smaller transcript hash upgrades and + // must displace the incumbent: cancel it and take over its per-IP slot. + let winner_token = CancellationToken::new(); + let ZakuraRegistration::Registered { + registration_id: winner_id, + .. + } = register_with_transcript(&supervisor, &peer, ip, [1; 32], winner_token.clone()).await + else { + panic!("the winning duplicate registers"); + }; + assert_ne!(incumbent_id, winner_id); + assert!( + incumbent_token.is_cancelled(), + "the displaced incumbent's connection must be cancelled so it stops serving", + ); + assert!(!winner_token.is_cancelled()); + assert_eq!(supervisor.registered_ids().await, vec![peer.clone()]); + assert!( + !supervisor.can_accept_remote_ip_with_in_flight(ip, 0).await, + "the winner holds exactly the one per-IP slot at cap 1", + ); + + // The displaced incumbent's late teardown carries the old registration + // id and must leave the winner's registration and per-IP slot intact. + assert!( + !supervisor.deregister(&peer, incumbent_id).await, + "a displaced connection's teardown must not deregister the winner", + ); + assert_eq!(supervisor.registered_ids().await, vec![peer.clone()]); + assert!( + !supervisor.can_accept_remote_ip_with_in_flight(ip, 0).await, + "a stale teardown must not release the winner's per-IP slot", + ); + + // The winner's own teardown deregisters normally and frees the slot. + assert!(supervisor.deregister(&peer, winner_id).await); + assert!(supervisor.registered_ids().await.is_empty()); + assert!( + supervisor.can_accept_remote_ip_with_in_flight(ip, 0).await, + "the per-IP slot is free once the current registration deregisters", + ); + + Ok(()) + } + + // A winning duplicate can arrive from a different IP than the incumbent + // (e.g. a redial after the peer's address changed). Displacement must move + // the per-IP slot: the incumbent's IP frees immediately, the winner's IP is + // charged, and the incumbent's stale teardown must not touch either count. + #[tokio::test] + async fn winning_duplicate_across_ips_moves_per_ip_slot() -> Result<(), BoxError> { + let supervisor = ZakuraSupervisorHandle::new(1); + let old_ip: IpAddr = "203.0.113.31".parse().expect("test ip parses"); + let new_ip: IpAddr = "203.0.113.32".parse().expect("test ip parses"); + let peer = test_peer(31); + + let incumbent_token = CancellationToken::new(); + let ZakuraRegistration::Registered { + registration_id: incumbent_id, + .. + } = register_with_transcript(&supervisor, &peer, old_ip, [2; 32], incumbent_token.clone()) + .await + else { + panic!("the incumbent registers"); + }; + assert!( + !supervisor + .can_accept_remote_ip_with_in_flight(old_ip, 0) + .await + ); + + let winner_token = CancellationToken::new(); + let ZakuraRegistration::Registered { + registration_id: winner_id, + .. + } = register_with_transcript(&supervisor, &peer, new_ip, [1; 32], winner_token.clone()) + .await + else { + panic!("the winning duplicate registers"); + }; + assert!( + supervisor + .can_accept_remote_ip_with_in_flight(old_ip, 0) + .await, + "displacement must free the incumbent's per-IP slot immediately", + ); + assert!( + !supervisor + .can_accept_remote_ip_with_in_flight(new_ip, 0) + .await, + "the winner is charged against its own IP", + ); + + // The stale teardown carries the incumbent's old IP implicitly through + // its registration id; it must not decrement the winner's IP count. + assert!(!supervisor.deregister(&peer, incumbent_id).await); + assert!( + supervisor + .can_accept_remote_ip_with_in_flight(old_ip, 0) + .await + ); + assert!( + !supervisor + .can_accept_remote_ip_with_in_flight(new_ip, 0) + .await, + "a stale teardown must not release the winner's per-IP slot", + ); + + assert!(supervisor.deregister(&peer, winner_id).await); + assert!( + supervisor + .can_accept_remote_ip_with_in_flight(new_ip, 0) + .await + ); + + Ok(()) + } + + // After displacement, supervisor-facing APIs must operate on the winner's + // entry: `disconnect_peer` cancels the winner's token, and the peer-set + // watch publishes no change for a stale teardown. + #[tokio::test] + async fn supervisor_apis_track_winner_after_displacement() -> Result<(), BoxError> { + let supervisor = ZakuraSupervisorHandle::new(1); + let ip: IpAddr = "203.0.113.41".parse().expect("test ip parses"); + let peer = test_peer(41); + let mut peer_set = supervisor.subscribe(); + + let incumbent_token = CancellationToken::new(); + let ZakuraRegistration::Registered { + registration_id: incumbent_id, + .. + } = register_with_transcript(&supervisor, &peer, ip, [2; 32], incumbent_token.clone()) + .await + else { + panic!("the incumbent registers"); + }; + let winner_token = CancellationToken::new(); + let ZakuraRegistration::Registered { + registration_id: winner_id, + .. + } = register_with_transcript(&supervisor, &peer, ip, [1; 32], winner_token.clone()).await + else { + panic!("the winning duplicate registers"); + }; + + // Mark the registration-time peer-set updates as seen, then verify a + // stale teardown publishes nothing: the registered set did not change. + peer_set.borrow_and_update(); + assert!(!supervisor.deregister(&peer, incumbent_id).await); + assert!( + !peer_set.has_changed()?, + "a stale teardown must not publish a peer-set change", + ); + + // `disconnect_peer` must target the winner's (current) connection. + assert!(!winner_token.is_cancelled()); + assert!(supervisor.disconnect_peer(&peer).await); + assert!( + winner_token.is_cancelled(), + "disconnect_peer cancels the currently registered connection", + ); + + // The winner's teardown publishes the now-empty peer set. + assert!(supervisor.deregister(&peer, winner_id).await); + assert!(peer_set.has_changed()?); + assert!(peer_set.borrow_and_update().is_empty()); + + Ok(()) + } + #[tokio::test] async fn header_sync_misbehavior_action_does_not_disconnect_peer() -> Result<(), BoxError> { let _guard = zebra_test::init(); diff --git a/zebra-network/src/zakura/header_sync/events.rs b/zebra-network/src/zakura/header_sync/events.rs index 33d624c39b2..49a5da785ad 100644 --- a/zebra-network/src/zakura/header_sync/events.rs +++ b/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, + }, /// 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", diff --git a/zebra-network/src/zakura/header_sync/pipe.rs b/zebra-network/src/zakura/header_sync/pipe.rs index 504ba9321fa..6192827a034 100644 --- a/zebra-network/src/zakura/header_sync/pipe.rs +++ b/zebra-network/src/zakura/header_sync/pipe.rs @@ -416,7 +416,10 @@ mod tests { fn saturated_events_handle() -> (HeaderSyncHandle, mpsc::Receiver) { 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]))); diff --git a/zebra-network/src/zakura/header_sync/reactor.rs b/zebra-network/src/zakura/header_sync/reactor.rs index c5d2b7ed8fa..bd5de4fd23c 100644 --- a/zebra-network/src/zakura/header_sync/reactor.rs +++ b/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) { + // 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); } diff --git a/zebra-network/src/zakura/header_sync/service.rs b/zebra-network/src/zakura/header_sync/service.rs index f8375a08646..504ac770be3 100644 --- a/zebra-network/src/zakura/header_sync/service.rs +++ b/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, } @@ -50,15 +56,25 @@ impl HeaderSyncPeerSession { fn new_with_commands( session: &PeerStreamSession, direction: ServicePeerDirection, + registration_id: u64, commands: mpsc::UnboundedSender, ) -> 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, + }); } } diff --git a/zebra-network/src/zakura/header_sync/tests.rs b/zebra-network/src/zakura/header_sync/tests.rs index b8697a1aeb9..cbb08c97bf4 100644 --- a/zebra-network/src/zakura/header_sync/tests.rs +++ b/zebra-network/src/zakura/header_sync/tests.rs @@ -578,7 +578,10 @@ async fn peer_caps_reject_full_without_status_or_misbehavior_and_free_on_remove( fixture .handle - .send(HeaderSyncEvent::PeerDisconnected(admitted)) + .send(HeaderSyncEvent::PeerDisconnected { + peer: admitted, + registration_id: None, + }) .await .unwrap(); tokio::time::sleep(std::time::Duration::from_millis(10)).await; @@ -586,6 +589,139 @@ async fn peer_caps_reject_full_without_status_or_misbehavior_and_free_on_remove( assert_eq!(fixture.handle.peer_snapshot().inbound_slots_free, 1); } +/// Duplicate arbitration can replace a peer's registered connection while the +/// displaced connection's header-sync session task is still tearing down. The +/// displaced session's `PeerDisconnected` (scoped to its registration id) must +/// not remove the winning connection's admitted session; the winner's own +/// teardown still must. +#[tokio::test] +async fn stale_disconnect_from_displaced_connection_keeps_live_session() { + let network = Network::Mainnet; + let anchor = (block::Height(0), network.genesis_hash()); + let mut startup = startup_for(network, anchor, Some(anchor)); + startup.range_state_actions_enabled = false; + let fixture = spawn_test_reactor(startup); + let peer_id = peer(33); + + // Session riding the peer's first registered connection. + connect_peer_with_registration(&fixture, peer_id.clone(), ServicePeerDirection::Inbound, 1) + .await; + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + assert_eq!(fixture.handle.peer_snapshot().inbound_peers, 1); + + // A winning duplicate connection replaces the registration and its session + // supersedes the first one in the reactor. + connect_peer_with_registration(&fixture, peer_id.clone(), ServicePeerDirection::Inbound, 2) + .await; + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + assert_eq!(fixture.handle.peer_snapshot().inbound_peers, 1); + + // The displaced connection's teardown arrives late and must be ignored. + fixture + .handle + .send(HeaderSyncEvent::PeerDisconnected { + peer: peer_id.clone(), + registration_id: Some(1), + }) + .await + .unwrap(); + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + assert_eq!( + fixture.handle.peer_snapshot().inbound_peers, + 1, + "a stale disconnect from a displaced connection must not remove the live session", + ); + + // The winning connection's own teardown still removes the peer. + fixture + .handle + .send(HeaderSyncEvent::PeerDisconnected { + peer: peer_id.clone(), + registration_id: Some(2), + }) + .await + .unwrap(); + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + assert_eq!(fixture.handle.peer_snapshot().inbound_peers, 0); +} + +/// The opposite event ordering to +/// [`stale_disconnect_from_displaced_connection_keeps_live_session`]: the +/// displaced session's scoped disconnect arrives while its own session is still +/// admitted (before the winner's `PeerConnected`). The disconnect matches and +/// removes the old session, and the winner's connect then re-admits the peer, +/// so the reactor converges to one live session in either ordering. +#[tokio::test] +async fn stale_disconnect_before_winner_connect_still_converges() { + let network = Network::Mainnet; + let anchor = (block::Height(0), network.genesis_hash()); + let mut startup = startup_for(network, anchor, Some(anchor)); + startup.range_state_actions_enabled = false; + let fixture = spawn_test_reactor(startup); + let peer_id = peer(34); + + connect_peer_with_registration(&fixture, peer_id.clone(), ServicePeerDirection::Inbound, 1) + .await; + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + assert_eq!(fixture.handle.peer_snapshot().inbound_peers, 1); + + // The displaced connection's teardown fires while its own session is still + // the admitted one, so the id matches and the session is removed. + fixture + .handle + .send(HeaderSyncEvent::PeerDisconnected { + peer: peer_id.clone(), + registration_id: Some(1), + }) + .await + .unwrap(); + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + assert_eq!(fixture.handle.peer_snapshot().inbound_peers, 0); + + // The winning connection's session arrives afterwards and is admitted. + connect_peer_with_registration(&fixture, peer_id.clone(), ServicePeerDirection::Inbound, 2) + .await; + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + assert_eq!( + fixture.handle.peer_snapshot().inbound_peers, + 1, + "the winner's session is admitted after the old session's removal", + ); +} + +/// An unscoped disconnect (`registration_id: None`, the registry fanout path) +/// removes the session regardless of the registration id it rides on, because +/// the fanout only runs for the currently registered connection. +#[tokio::test] +async fn unscoped_disconnect_removes_session_with_any_registration_id() { + let network = Network::Mainnet; + let anchor = (block::Height(0), network.genesis_hash()); + let mut startup = startup_for(network, anchor, Some(anchor)); + startup.range_state_actions_enabled = false; + let fixture = spawn_test_reactor(startup); + let peer_id = peer(35); + + connect_peer_with_registration(&fixture, peer_id.clone(), ServicePeerDirection::Inbound, 5) + .await; + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + assert_eq!(fixture.handle.peer_snapshot().inbound_peers, 1); + + fixture + .handle + .send(HeaderSyncEvent::PeerDisconnected { + peer: peer_id, + registration_id: None, + }) + .await + .unwrap(); + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + assert_eq!( + fixture.handle.peer_snapshot().inbound_peers, + 0, + "an unconditional disconnect removes the session whatever its registration id", + ); +} + #[tokio::test(flavor = "current_thread")] async fn advisory_summary_status_mismatch_uses_status_without_misbehavior_and_backs_off() { let network = regtest_network(); @@ -736,7 +872,10 @@ async fn advisory_backoff_is_pruned_on_peer_disconnected() { fixture .handle - .send(HeaderSyncEvent::PeerDisconnected(peer_id.clone())) + .send(HeaderSyncEvent::PeerDisconnected { + peer: peer_id.clone(), + registration_id: None, + }) .await .unwrap(); tokio::time::sleep(std::time::Duration::from_millis(10)).await; @@ -932,6 +1071,15 @@ async fn connect_peer_with_direction( fixture: &ReactorFixture, peer_id: ZakuraPeerId, direction: ServicePeerDirection, +) -> CancellationToken { + connect_peer_with_registration(fixture, peer_id, direction, 0).await +} + +async fn connect_peer_with_registration( + fixture: &ReactorFixture, + peer_id: ZakuraPeerId, + direction: ServicePeerDirection, + registration_id: u64, ) -> CancellationToken { let (send, recv) = crate::zakura::framed_channel(32); fixture @@ -941,7 +1089,8 @@ async fn connect_peer_with_direction( .push(recv); let cancel = CancellationToken::new(); let session = - HeaderSyncPeerSession::from_parts_with_direction(peer_id, direction, send, cancel.clone()); + HeaderSyncPeerSession::from_parts_with_direction(peer_id, direction, send, cancel.clone()) + .with_registration_id(registration_id); fixture .handle .send(HeaderSyncEvent::PeerConnected(session)) @@ -1030,7 +1179,7 @@ async fn stale_header_sync_teardown_keeps_replacement_session() { service.remove_peer(&peer_id, new_conn_id); assert!(matches!( lifecycle.recv().await, - Some(HeaderSyncEvent::PeerDisconnected(disconnected)) if disconnected == peer_id + Some(HeaderSyncEvent::PeerDisconnected { peer, registration_id: None }) if peer == peer_id )); } @@ -1967,11 +2116,13 @@ async fn rootless_non_empty_response_retries_without_misbehavior() { let (network, _) = checkpoint_testnet_with_hash(block::Height(3), checkpoint_hash); let first_checkpoint = block::Height(3); let start = block::Height(4); - let mut fixture = spawn_test_reactor(startup_for( + let mut startup = startup_for( network.clone(), (block::Height(0), network.genesis_hash()), Some((first_checkpoint, checkpoint_hash)), - )); + ); + startup.request_timeout = std::time::Duration::from_millis(20); + let mut fixture = spawn_test_reactor(startup); let peer_id = peer(8); connect_peer(&fixture, peer_id.clone()).await; @@ -2000,7 +2151,17 @@ async fn rootless_non_empty_response_retries_without_misbehavior() { .await .unwrap(); - // The range must be retried (a fresh `GetHeaders` for the same start height), with no + assert!( + tokio::time::timeout( + std::time::Duration::from_millis(5), + next_non_query_action(&mut fixture.actions) + ) + .await + .is_err(), + "rootless unusable responses must not retry immediately" + ); + + // The range must be retried after the short unusable-response backoff, with no // commit and no misbehavior along the way. loop { match next_non_query_action(&mut fixture.actions).await { @@ -2233,7 +2394,10 @@ async fn peer_disconnect_removes_outstanding_requests_for_that_peer() { fixture .handle - .send(HeaderSyncEvent::PeerDisconnected(peer_id.clone())) + .send(HeaderSyncEvent::PeerDisconnected { + peer: peer_id.clone(), + registration_id: None, + }) .await .unwrap(); fixture diff --git a/zebra-network/src/zakura/transport/registry.rs b/zebra-network/src/zakura/transport/registry.rs index d40505aa6b6..8143d1706b1 100644 --- a/zebra-network/src/zakura/transport/registry.rs +++ b/zebra-network/src/zakura/transport/registry.rs @@ -304,6 +304,7 @@ impl ServiceRegistry { /// Fan a newly connected peer out to every service enabled by its negotiated capabilities. pub fn add_peer(&self, peer: Peer) { + let registration_id = peer.registration_id; let ( peer_id, conn_id, @@ -331,17 +332,20 @@ impl ServiceRegistry { .map(|stream| stream.cancel_token.clone()) .unwrap_or_else(|| cancel_token.child_token()); - service.add_peer(Peer::new_with_service_cancel_token( - conn_id, - peer_id.clone(), - remote_ip, - negotiated, - direction, - service_streams, - cancel_token.clone(), - service_cancel_token, - close_cause.clone(), - )); + service.add_peer( + Peer::new_with_service_cancel_token( + conn_id, + peer_id.clone(), + remote_ip, + negotiated, + direction, + service_streams, + cancel_token.clone(), + service_cancel_token, + close_cause.clone(), + ) + .with_registration_id(registration_id), + ); } } @@ -350,6 +354,7 @@ impl ServiceRegistry { /// Returns the capability mask for services that received a peer session, so /// disconnect fanout can be limited to reactors that were actually reached. pub fn add_escalated_peer(&self, peer: Peer) -> u64 { + let registration_id = peer.registration_id; let ( peer_id, conn_id, @@ -384,17 +389,20 @@ impl ServiceRegistry { .map(|stream| stream.cancel_token.clone()) .unwrap_or_else(|| cancel_token.child_token()); - service.add_peer(Peer::new_with_service_cancel_token( - conn_id, - peer_id.clone(), - remote_ip, - negotiated, - direction, - service_streams, - cancel_token.clone(), - service_cancel_token, - close_cause.clone(), - )); + service.add_peer( + Peer::new_with_service_cancel_token( + conn_id, + peer_id.clone(), + remote_ip, + negotiated, + direction, + service_streams, + cancel_token.clone(), + service_cancel_token, + close_cause.clone(), + ) + .with_registration_id(registration_id), + ); } admitted_capabilities @@ -424,6 +432,7 @@ mod tests { wants: Mutex, added: Mutex>, added_streams: Mutex>>, + added_registration_ids: Mutex>, removed: Mutex>, } @@ -435,6 +444,7 @@ mod tests { wants: Mutex::new(true), added: Mutex::new(Vec::new()), added_streams: Mutex::new(Vec::new()), + added_registration_ids: Mutex::new(Vec::new()), removed: Mutex::new(Vec::new()), }) } @@ -469,6 +479,10 @@ mod tests { } fn add_peer(&self, peer: Peer) { + self.added_registration_ids + .lock() + .expect("test service registration id list should not be poisoned") + .push(peer.registration_id); let ( peer_id, _conn_id, @@ -800,4 +814,41 @@ mod tests { &[peer] ); } + + // Duplicate arbitration relies on every service seeing the supervisor + // registration id of the connection a session rides on, so both fanout + // paths must copy it onto the per-service `Peer`. + #[test] + fn add_peer_and_escalated_peer_propagate_registration_id() { + let header = TestService::new("header", vec![stream(5, 0b0001)]); + let registry = ServiceRegistry::new(vec![header.clone()]).expect("stream kinds are unique"); + let peer = ZakuraPeerId::new(vec![13; 32]).expect("32-byte test peer id is valid"); + + registry.add_peer( + Peer::new( + peer.clone(), + None, + 0b0001, + HashMap::new(), + CancellationToken::new(), + ) + .with_registration_id(7), + ); + + let (send_5, recv_5) = framed_channel(1); + let streams = HashMap::from([(5, (recv_5, send_5))]); + registry.add_escalated_peer( + Peer::new(peer, None, 0b0001, streams, CancellationToken::new()) + .with_registration_id(8), + ); + + assert_eq!( + header + .added_registration_ids + .lock() + .expect("test mutex should not be poisoned") + .as_slice(), + &[7, 8], + ); + } } diff --git a/zebra-network/src/zakura/transport/service.rs b/zebra-network/src/zakura/transport/service.rs index 943662adb6d..2ab394c638b 100644 --- a/zebra-network/src/zakura/transport/service.rs +++ b/zebra-network/src/zakura/transport/service.rs @@ -71,6 +71,13 @@ pub struct Peer { pub negotiated: u64, /// Direction of the underlying authenticated connection. pub direction: ServicePeerDirection, + /// Supervisor registration id of the connection this peer session rides on. + /// + /// Duplicate arbitration can replace a peer's registered connection while + /// the displaced one is still tearing down; services use this id to ignore + /// lifecycle events from sessions that no longer belong to the registered + /// connection. Zero for test peers built without a supervisor registration. + pub registration_id: u64, streams: HashMap, cancel_token: CancellationToken, service_cancel_token: CancellationToken, @@ -218,6 +225,7 @@ impl Peer { remote_ip, negotiated, direction, + registration_id: 0, streams, cancel_token, service_cancel_token, @@ -225,6 +233,12 @@ impl Peer { } } + /// Set the supervisor registration id of this peer's connection. + pub(crate) fn with_registration_id(mut self, registration_id: u64) -> Self { + self.registration_id = registration_id; + self + } + /// Take ownership of a stream pair for `kind`. pub fn take_stream(&mut self, kind: u16) -> Option<(FramedRecv, FramedSend)> { self.streams