From 380fd29ed2b82594d018f35210276ce1ad876ccd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Fri, 30 Oct 2026 11:57:44 -0300 Subject: [PATCH 1/2] fix(p2p): disconnect peers whose gossipsub send queue never drains A gossipsub connection handler disables itself for good after a few substream negotiations on one connection, successful ones included, and then never takes another message off its send queue. The behaviour is not told, so it keeps queueing forwards for that peer until the queue is full, and a queued message is only dropped when it is taken off. The copies stay in memory until the connection closes, which other protocols can keep open for hours: on the mainnet follower these queues held about 490 MB. Gossipsub reports such a peer as SlowPeer every heartbeat, and the node ignored the event. The swarm task now collects those reports and, on the 10 s check that already drops graylisted peers, disconnects a peer that was reported in each of the last six checks. The follower's three stuck peers failed without a break for 5 to 63 minutes, so each would have been dropped within about a minute, while a peer with a few isolated failures keeps its connection. Beacon only, gated like the graylist check on peer scoring being on. --- CLAUDE.md | 2 +- crates/net/p2p/src/metrics.rs | 13 +++ crates/net/p2p/src/swarm_adapter.rs | 162 +++++++++++++++++++++++++++- docs/beacon_wire.md | 15 +++ docs/metrics.md | 1 + 5 files changed, 190 insertions(+), 3 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index a171704a8..769bdf2b1 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -336,7 +336,7 @@ actual_slot = finalized_slot + 1 + relative_index - `fork_digest` is a 4-byte hex string (no `0x` prefix); currently the dummy `12345678` agreed across clients - Mesh size: 8 (6-12 bounds), heartbeat: 700ms - Beacon wire: `validate_messages()` is on, so every beacon message waits for a verdict (~4.2s before gossipsub's cache evicts it). Rules in `state_transition::beacon::gossip` (cheap half inline, stateful half on a bounded `spawn_blocking` task); plumbing in `p2p/src/beacon/verdict.rs`. Lean gossip still auto-forwards - - Beacon peer scoring: lighthouse's parameters ported in `p2p/src/beacon/scoring.rs` (block, aggregate, 64 attestation subnets, exits, slashings; columns and sync contributions unscored), refreshed every slot from the head's active validator count. Mesh-delivery scoring (P3) is gated off while the head lags more than 4 slots and for an epoch after, because a catching-up node `Ignore`s every aggregate and would otherwise graylist its own mesh. Every held fork digest is parametrized before `transition::apply` joins it (gossipsub never scores a mesh grafted before its topic had parameters); the previous digest's weights are zeroed at the switch, and the switch restarts the P3 warmup. The swarm disconnects peers below the graylist (-16000) every 10 s. Lean runs unscored. See [`docs/beacon_wire.md`](docs/beacon_wire.md#peer-scoring) + - Beacon peer scoring: lighthouse's parameters ported in `p2p/src/beacon/scoring.rs` (block, aggregate, 64 attestation subnets, exits, slashings; columns and sync contributions unscored), refreshed every slot from the head's active validator count. Mesh-delivery scoring (P3) is gated off while the head lags more than 4 slots and for an epoch after, because a catching-up node `Ignore`s every aggregate and would otherwise graylist its own mesh. Every held fork digest is parametrized before `transition::apply` joins it (gossipsub never scores a mesh grafted before its topic had parameters); the previous digest's weights are zeroed at the switch, and the switch restarts the P3 warmup. The swarm disconnects peers below the graylist (-16000) every 10 s, and peers gossipsub reported as `SlowPeer` in every check for a minute (a disabled gossipsub handler never drains its send queue, so the queue's message copies stay in memory until the connection closes). Lean runs unscored. See [`docs/beacon_wire.md`](docs/beacon_wire.md#peer-scoring) - Data columns: every check runs in p2p. A column gossip did not accept (`Queue`/`Overloaded`), every fetched column, and parked columns replayed after their parent imports go through `column::chain_checks` in `p2p/src/beacon/column_checks.rs`. The chain actor stores what it gets unchecked; only debug builds re-run `chain_checks` there - Beacon subscribes seven global topics plus two node-id-derived subnet families: custody columns and backbone attestation subnets. One of the seven is diff --git a/crates/net/p2p/src/metrics.rs b/crates/net/p2p/src/metrics.rs index 643300967..c8f5f5cba 100644 --- a/crates/net/p2p/src/metrics.rs +++ b/crates/net/p2p/src/metrics.rs @@ -622,6 +622,19 @@ pub fn inc_gossipsub_score_disconnects() { LEAN_GOSSIPSUB_SCORE_DISCONNECTS_TOTAL.inc(); } +/// Count a peer disconnected for persistently failing to take gossipsub +/// messages. +pub fn inc_gossipsub_slow_peer_disconnects() { + static LEAN_GOSSIPSUB_SLOW_PEER_DISCONNECTS_TOTAL: LazyLock = LazyLock::new(|| { + register_int_counter!( + "lean_gossipsub_slow_peer_disconnects_total", + "Peers disconnected for persistently failing to take gossipsub messages" + ) + .unwrap() + }); + LEAN_GOSSIPSUB_SLOW_PEER_DISCONNECTS_TOTAL.inc(); +} + /// Whether gossipsub's mesh-delivery term (P3) is being scored: 0 while the /// head lags or is still warming up after catching up. See /// `beacon::scoring::MeshDeliveryGate`. diff --git a/crates/net/p2p/src/swarm_adapter.rs b/crates/net/p2p/src/swarm_adapter.rs index 83094c4bb..52425fbf9 100644 --- a/crates/net/p2p/src/swarm_adapter.rs +++ b/crates/net/p2p/src/swarm_adapter.rs @@ -4,7 +4,7 @@ use std::time::Duration; use libp2p::{ PeerId, futures::StreamExt, - gossipsub::{IdentTopic, TopicScoreParams}, + gossipsub::{FailedMessages, IdentTopic, TopicScoreParams}, request_response, swarm::{SwarmEvent, dial_opts::DialOpts}, }; @@ -17,9 +17,13 @@ use crate::{ }; /// Interval between gossipsub mesh peer metric refreshes, and between the -/// checks that disconnect peers scored below the graylist. +/// checks that disconnect peers scored below the graylist or persistently slow. const MESH_METRIC_REFRESH_INTERVAL: Duration = Duration::from_secs(10); +/// How many consecutive [`MESH_METRIC_REFRESH_INTERVAL`] checks a peer must +/// have failed gossipsub sends in before it is disconnected; see [`SlowPeers`]. +const SLOW_PEER_CHECK_LIMIT: u32 = 6; + pub enum SwarmCommand { Publish { topic: libp2p::gossipsub::IdentTopic, @@ -269,10 +273,17 @@ async fn swarm_loop( ) { let mut mesh_metric_tick = tokio::time::interval(MESH_METRIC_REFRESH_INTERVAL); mesh_metric_tick.set_missed_tick_behavior(MissedTickBehavior::Skip); + let mut slow_peers = SlowPeers::default(); let exit = loop { tokio::select! { event = swarm.next() => { let Some(event) = event else { break LoopExit::SwarmEnded }; + if let SwarmEvent::Behaviour(BehaviourEvent::Gossipsub( + libp2p::gossipsub::Event::SlowPeer { peer_id, failed_messages }, + )) = &event + { + slow_peers.record(*peer_id, failed_messages); + } if event_tx.send(event).is_err() { break LoopExit::ActorStopped } } cmd = cmd_rx.recv() => { @@ -299,6 +310,7 @@ async fn swarm_loop( counters.num_established_outgoing(), ); disconnect_graylisted_peers(&mut swarm); + disconnect_slow_peers(&mut swarm, &mut slow_peers); } } }; @@ -324,6 +336,92 @@ enum LoopExit { SwarmEnded, } +/// Peers gossipsub reported as failing to take messages, tracked across the +/// periodic checks so that only a persistent failure costs a peer its +/// connection. +/// +/// The failure this exists for is a dead gossipsub stream. The connection +/// handler disables itself for good after a handful of substream +/// negotiations on one connection, successful ones included, and from then +/// on never takes another message off the peer's send queue. The behaviour is +/// not told: it keeps queueing for the peer up to the queue's capacity, and +/// queued messages are only dropped when the handler takes them off, so a +/// full queue of message copies stays in memory for as long as the connection +/// lives, which other protocols keep open. The one trace this leaves is +/// gossipsub's `SlowPeer` report, emitted every heartbeat for each peer whose +/// queue refused a message (or that let a message expire in it, which the +/// report does not tell apart). Disconnecting frees the queue; a peer that +/// reconnects gets a fresh handler. +/// +/// A peer reported in fewer than [`SLOW_PEER_CHECK_LIMIT`] consecutive checks +/// keeps its connection, so a burst of expiries from a peer that catches up is +/// not enough. +#[derive(Default)] +struct SlowPeers { + /// Failed messages per peer reported since the last check. + reported: HashMap, + /// Per peer, the consecutive checks it has been reported in and the + /// failed messages across them. + streaks: HashMap, +} + +impl SlowPeers { + fn record(&mut self, peer_id: PeerId, failed_messages: &FailedMessages) { + let failed = failed_messages.priority + failed_messages.non_priority; + *self.reported.entry(peer_id).or_default() += failed; + } + + /// Closes one check: a peer reported since the last one extends its + /// streak, every other peer loses it. Returns each peer whose streak + /// reached [`SLOW_PEER_CHECK_LIMIT`], with its failed messages across the + /// streak, and forgets it. + fn check(&mut self) -> Vec<(PeerId, usize)> { + let reported = std::mem::take(&mut self.reported); + self.streaks + .retain(|peer_id, _| reported.contains_key(peer_id)); + let mut persistent = Vec::new(); + for (peer_id, failed) in reported { + let (checks, total) = self.streaks.entry(peer_id).or_default(); + *checks += 1; + *total += failed; + if *checks >= SLOW_PEER_CHECK_LIMIT { + persistent.push((peer_id, *total)); + } + } + for (peer_id, _) in &persistent { + self.streaks.remove(peer_id); + } + persistent + } +} + +/// Disconnect every peer [`SlowPeers`] finds persistently failing to take +/// gossipsub messages. +/// +/// Beacon only, like [`disconnect_graylisted_peers`]: the check still runs on +/// lean, so the reports do not pile up, but nothing is disconnected. +fn disconnect_slow_peers(swarm: &mut libp2p::Swarm, slow_peers: &mut SlowPeers) { + let persistent = slow_peers.check(); + if swarm + .behaviour() + .gossipsub + .peer_score(swarm.local_peer_id()) + .is_none() + { + return; + } + for (peer_id, failed) in persistent { + if swarm.disconnect_peer_id(peer_id).is_ok() { + metrics::inc_gossipsub_slow_peer_disconnects(); + info!( + %peer_id, + failed, + "Disconnecting peer that kept failing to take gossipsub messages" + ); + } + } +} + /// Disconnect every peer gossipsub scores below the graylist, and publish how /// the rest are spread across the score bands. /// @@ -497,3 +595,63 @@ pub(crate) fn execute_command(swarm: &mut libp2p::Swarm, cmd: SwarmCo } } } + +#[cfg(test)] +mod tests { + use super::*; + + fn failed(non_priority: usize) -> FailedMessages { + FailedMessages { + priority: 0, + non_priority, + } + } + + #[test] + fn a_peer_failing_in_every_check_is_returned_at_the_limit() { + let peer = PeerId::random(); + let mut slow_peers = SlowPeers::default(); + for _ in 1..SLOW_PEER_CHECK_LIMIT { + slow_peers.record(peer, &failed(2)); + assert!(slow_peers.check().is_empty()); + } + slow_peers.record(peer, &failed(2)); + let expected = 2 * usize::try_from(SLOW_PEER_CHECK_LIMIT).unwrap(); + assert_eq!(slow_peers.check(), vec![(peer, expected)]); + } + + #[test] + fn a_check_without_a_report_resets_the_streak() { + let peer = PeerId::random(); + let mut slow_peers = SlowPeers::default(); + for _ in 1..SLOW_PEER_CHECK_LIMIT { + slow_peers.record(peer, &failed(1)); + assert!(slow_peers.check().is_empty()); + } + assert!(slow_peers.check().is_empty()); + slow_peers.record(peer, &failed(1)); + assert!(slow_peers.check().is_empty()); + } + + #[test] + fn several_reports_within_one_check_count_once_towards_the_streak() { + let peer = PeerId::random(); + let mut slow_peers = SlowPeers::default(); + for _ in 0..usize::try_from(SLOW_PEER_CHECK_LIMIT).unwrap() { + slow_peers.record(peer, &failed(1)); + } + assert!(slow_peers.check().is_empty()); + } + + #[test] + fn a_returned_peer_starts_a_new_streak() { + let peer = PeerId::random(); + let mut slow_peers = SlowPeers::default(); + for _ in 0..SLOW_PEER_CHECK_LIMIT { + slow_peers.record(peer, &failed(1)); + slow_peers.check(); + } + slow_peers.record(peer, &failed(1)); + assert!(slow_peers.check().is_empty()); + } +} diff --git a/docs/beacon_wire.md b/docs/beacon_wire.md index c2138785b..76030089d 100644 --- a/docs/beacon_wire.md +++ b/docs/beacon_wire.md @@ -320,6 +320,20 @@ s the swarm task disconnects peers below the graylist. There is no ban list: a peer's negative score is kept for `retain_score` (100 epochs) after it leaves, so a peer that reconnects comes back graylisted and is dropped again. +The same check also disconnects a peer gossipsub reported as slow (`SlowPeer`: +its send queue refused a message, or a message expired in it) in each of the +last `SLOW_PEER_CHECK_LIMIT` checks. This is for a dead gossipsub stream, not +for slowness in general. The connection handler disables itself for good after +a few substream negotiations on one connection, successful ones included, and +then never takes another message off the queue. Gossipsub keeps queueing for +the peer until the queue is full, and a queued message is only dropped when +the handler takes it off, so the queue's message copies stay in memory for as +long as other protocols keep the connection open. On the mainnet follower, +such queues held about 490 MB. Gossipsub's own penalty for these failures +(`slow_peer_weight`) left the peer in the `negative` band, never near the +graylist. Lighthouse penalizes `SlowPeer` reports through its peer manager +instead. + Around a fork-digest boundary the node holds two digests' topics, and every held digest gets the whole parameter set (exits and slashings included), scored by its role: @@ -816,6 +830,7 @@ as a query filter, so a `quic`-only record is invisible to it. | `lean_beacon_aggregates_deferred` | Aggregates held until their own slot has passed | | `lean_gossipsub_peers_by_score{band}` | Gossipsub peers by score band; see [Peer scoring](#peer-scoring) | | `lean_gossipsub_score_disconnects_total` | Peers disconnected for a score below the graylist | +| `lean_gossipsub_slow_peer_disconnects_total` | Peers disconnected for persistently failing to take gossipsub messages; see [Peer scoring](#peer-scoring) | | `lean_gossipsub_mesh_delivery_scoring` | 1 while P3 is scored, 0 while the head lags or is warming up | | `lean_beacon_publish_data_columns_seconds` | Time to compute and verify a published block's cells and build its sidecars; see [metrics.md](./metrics.md) | diff --git a/docs/metrics.md b/docs/metrics.md index 1ea07953f..7bcc2212d 100644 --- a/docs/metrics.md +++ b/docs/metrics.md @@ -447,6 +447,7 @@ spec. |------|------|-------|-------------------------|--------|---------| | `lean_gossipsub_peers_by_score` | Gauge | Gossipsub peers by score band | Every 10 s, in the swarm task | band=non_negative,negative,below_gossip,below_publish,below_graylist | | | `lean_gossipsub_score_disconnects_total` | Counter | Peers disconnected for a gossipsub score below the graylist (-16000) | When the 10 s check disconnects one | | | +| `lean_gossipsub_slow_peer_disconnects_total` | Counter | Peers disconnected for failing to take gossipsub messages in every 10 s check for a minute (a dead gossipsub stream pins its send queue in memory) | When the 10 s check disconnects one | | | | `lean_gossipsub_mesh_delivery_scoring` | Gauge | 1 while mesh message deliveries (P3) are scored, 0 while the head lags the wall clock or is warming up after catching up | Every slot, on the scoring refresh | | | The bands are mutually exclusive and split at the gossip (-4000), publish From 83067707b9494afbaddc0c4cf9ad0339b36c2013 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Mon, 2 Nov 2026 15:21:11 -0300 Subject: [PATCH 2/2] fix(p2p): gate slow-peer disconnects on Lighthouse's threshold, cap them per check Gossipsub counts a forward that expired in a peer's send queue as the same non-priority failure as a full queue. A saturated uplink or a stalled runtime on our side therefore makes every peer look slow at once, and the check would disconnect the whole peer set in one tick. - A SlowPeer report now counts only when its priority or non-priority failures exceed SLOW_PEER_REPORT_THRESHOLD, the cut Lighthouse applies before penalizing. A dead stream fails every message queued for the peer; a few failures per heartbeat are more likely late forwards. - One check disconnects at most MAX_SLOW_PEER_DISCONNECTS_PER_CHECK peers, the most failed first. The rest keep their streak and are logged as a warning, since many peers failing together points at our side. --- CLAUDE.md | 2 +- crates/net/p2p/src/swarm_adapter.rs | 239 ++++++++++++++++++++++++---- docs/beacon_wire.md | 19 +++ docs/metrics.md | 2 +- 4 files changed, 225 insertions(+), 37 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 769bdf2b1..0c7a4eb6e 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -336,7 +336,7 @@ actual_slot = finalized_slot + 1 + relative_index - `fork_digest` is a 4-byte hex string (no `0x` prefix); currently the dummy `12345678` agreed across clients - Mesh size: 8 (6-12 bounds), heartbeat: 700ms - Beacon wire: `validate_messages()` is on, so every beacon message waits for a verdict (~4.2s before gossipsub's cache evicts it). Rules in `state_transition::beacon::gossip` (cheap half inline, stateful half on a bounded `spawn_blocking` task); plumbing in `p2p/src/beacon/verdict.rs`. Lean gossip still auto-forwards - - Beacon peer scoring: lighthouse's parameters ported in `p2p/src/beacon/scoring.rs` (block, aggregate, 64 attestation subnets, exits, slashings; columns and sync contributions unscored), refreshed every slot from the head's active validator count. Mesh-delivery scoring (P3) is gated off while the head lags more than 4 slots and for an epoch after, because a catching-up node `Ignore`s every aggregate and would otherwise graylist its own mesh. Every held fork digest is parametrized before `transition::apply` joins it (gossipsub never scores a mesh grafted before its topic had parameters); the previous digest's weights are zeroed at the switch, and the switch restarts the P3 warmup. The swarm disconnects peers below the graylist (-16000) every 10 s, and peers gossipsub reported as `SlowPeer` in every check for a minute (a disabled gossipsub handler never drains its send queue, so the queue's message copies stay in memory until the connection closes). Lean runs unscored. See [`docs/beacon_wire.md`](docs/beacon_wire.md#peer-scoring) + - Beacon peer scoring: lighthouse's parameters ported in `p2p/src/beacon/scoring.rs` (block, aggregate, 64 attestation subnets, exits, slashings; columns and sync contributions unscored), refreshed every slot from the head's active validator count. Mesh-delivery scoring (P3) is gated off while the head lags more than 4 slots and for an epoch after, because a catching-up node `Ignore`s every aggregate and would otherwise graylist its own mesh. Every held fork digest is parametrized before `transition::apply` joins it (gossipsub never scores a mesh grafted before its topic had parameters); the previous digest's weights are zeroed at the switch, and the switch restarts the P3 warmup. The swarm disconnects peers below the graylist (-16000) every 10 s, and peers gossipsub reported as `SlowPeer` in every check for a minute (a disabled gossipsub handler never drains its send queue, so the queue's message copies stay in memory until the connection closes). Gossipsub reports an expired forward like a full queue, so our own saturated uplink makes every peer look slow: a report counts only above `SLOW_PEER_REPORT_THRESHOLD` failures (Lighthouse's cut), and one check disconnects at most `MAX_SLOW_PEER_DISCONNECTS_PER_CHECK` peers. Lean runs unscored. See [`docs/beacon_wire.md`](docs/beacon_wire.md#peer-scoring) - Data columns: every check runs in p2p. A column gossip did not accept (`Queue`/`Overloaded`), every fetched column, and parked columns replayed after their parent imports go through `column::chain_checks` in `p2p/src/beacon/column_checks.rs`. The chain actor stores what it gets unchecked; only debug builds re-run `chain_checks` there - Beacon subscribes seven global topics plus two node-id-derived subnet families: custody columns and backbone attestation subnets. One of the seven is diff --git a/crates/net/p2p/src/swarm_adapter.rs b/crates/net/p2p/src/swarm_adapter.rs index 52425fbf9..0993d7131 100644 --- a/crates/net/p2p/src/swarm_adapter.rs +++ b/crates/net/p2p/src/swarm_adapter.rs @@ -24,6 +24,19 @@ const MESH_METRIC_REFRESH_INTERVAL: Duration = Duration::from_secs(10); /// have failed gossipsub sends in before it is disconnected; see [`SlowPeers`]. const SLOW_PEER_CHECK_LIMIT: u32 = 6; +/// A `SlowPeer` report counts towards a peer's streak only when its priority +/// or its non-priority failures exceed this. Lighthouse uses the same cut +/// before it penalizes a slow peer. A few failures in one heartbeat are more +/// likely late forwards that expired in the queue than a dead stream. +const SLOW_PEER_REPORT_THRESHOLD: usize = 10; + +/// The most peers one check disconnects for being slow; see +/// [`SlowPeers::check`]. Gossipsub reports a forward that waited too long in a +/// peer's queue the same way it reports a full queue, so a saturated uplink or +/// a stalled runtime on our side makes every peer look slow at once. +/// Disconnecting them all in one check would leave the node without peers. +const MAX_SLOW_PEER_DISCONNECTS_PER_CHECK: usize = 10; + pub enum SwarmCommand { Publish { topic: libp2p::gossipsub::IdentTopic, @@ -353,29 +366,55 @@ enum LoopExit { /// report does not tell apart). Disconnecting frees the queue; a peer that /// reconnects gets a fresh handler. /// -/// A peer reported in fewer than [`SLOW_PEER_CHECK_LIMIT`] consecutive checks -/// keeps its connection, so a burst of expiries from a peer that catches up is -/// not enough. +/// Three rules keep a slow but working peer, or a slow node of our own, from +/// costing connections: +/// - A report counts only above [`SLOW_PEER_REPORT_THRESHOLD`] failures. +/// - A peer reported in fewer than [`SLOW_PEER_CHECK_LIMIT`] consecutive +/// checks keeps its connection, so a burst of expiries from a peer that +/// catches up is not enough. +/// - One check disconnects at most [`MAX_SLOW_PEER_DISCONNECTS_PER_CHECK`] +/// peers. #[derive(Default)] struct SlowPeers { - /// Failed messages per peer reported since the last check. + /// Failed messages per peer, from the reports since the last check that + /// passed [`SLOW_PEER_REPORT_THRESHOLD`]. reported: HashMap, /// Per peer, the consecutive checks it has been reported in and the /// failed messages across them. streaks: HashMap, } +/// What one [`SlowPeers::check`] found. +#[derive(Debug, PartialEq)] +struct SlowPeerCheck { + /// Peers to disconnect, the most failed messages first, each with its + /// failed messages across the streak. At most + /// [`MAX_SLOW_PEER_DISCONNECTS_PER_CHECK`] of them. + disconnect: Vec<(PeerId, usize)>, + /// How many more peers reached [`SLOW_PEER_CHECK_LIMIT`] but were left for + /// a later check by the cap. + deferred: usize, +} + impl SlowPeers { fn record(&mut self, peer_id: PeerId, failed_messages: &FailedMessages) { - let failed = failed_messages.priority + failed_messages.non_priority; - *self.reported.entry(peer_id).or_default() += failed; + let FailedMessages { + priority, + non_priority, + } = *failed_messages; + if priority <= SLOW_PEER_REPORT_THRESHOLD && non_priority <= SLOW_PEER_REPORT_THRESHOLD { + return; + } + *self.reported.entry(peer_id).or_default() += priority + non_priority; } /// Closes one check: a peer reported since the last one extends its - /// streak, every other peer loses it. Returns each peer whose streak - /// reached [`SLOW_PEER_CHECK_LIMIT`], with its failed messages across the - /// streak, and forgets it. - fn check(&mut self) -> Vec<(PeerId, usize)> { + /// streak, every other peer loses it. Of the peers whose streak reached + /// [`SLOW_PEER_CHECK_LIMIT`], returns the ones with the most failed + /// messages, up to [`MAX_SLOW_PEER_DISCONNECTS_PER_CHECK`], and forgets + /// them. The rest keep their streak, so a peer still reported at the next + /// check is a candidate again then. + fn check(&mut self) -> SlowPeerCheck { let reported = std::mem::take(&mut self.reported); self.streaks .retain(|peer_id, _| reported.contains_key(peer_id)); @@ -388,20 +427,31 @@ impl SlowPeers { persistent.push((peer_id, *total)); } } + persistent.sort_unstable_by(|(_, a), (_, b)| b.cmp(a)); + let deferred = persistent + .len() + .saturating_sub(MAX_SLOW_PEER_DISCONNECTS_PER_CHECK); + persistent.truncate(MAX_SLOW_PEER_DISCONNECTS_PER_CHECK); for (peer_id, _) in &persistent { self.streaks.remove(peer_id); } - persistent + SlowPeerCheck { + disconnect: persistent, + deferred, + } } } -/// Disconnect every peer [`SlowPeers`] finds persistently failing to take -/// gossipsub messages. +/// Disconnect the peers [`SlowPeers`] finds persistently failing to take +/// gossipsub messages, up to its per-check cap. /// /// Beacon only, like [`disconnect_graylisted_peers`]: the check still runs on /// lean, so the reports do not pile up, but nothing is disconnected. fn disconnect_slow_peers(swarm: &mut libp2p::Swarm, slow_peers: &mut SlowPeers) { - let persistent = slow_peers.check(); + let SlowPeerCheck { + disconnect, + deferred, + } = slow_peers.check(); if swarm .behaviour() .gossipsub @@ -410,7 +460,16 @@ fn disconnect_slow_peers(swarm: &mut libp2p::Swarm, slow_peers: &mut { return; } - for (peer_id, failed) in persistent { + if deferred > 0 { + let disconnecting = disconnect.len(); + warn!( + disconnecting, + deferred, + "More peers kept failing to take gossipsub messages than one check disconnects, \ + which points at our own uplink or runtime; the rest wait for a later check" + ); + } + for (peer_id, failed) in disconnect { if swarm.disconnect_peer_id(peer_id).is_ok() { metrics::inc_gossipsub_slow_peer_disconnects(); info!( @@ -600,6 +659,9 @@ pub(crate) fn execute_command(swarm: &mut libp2p::Swarm, cmd: SwarmCo mod tests { use super::*; + /// The fewest failures in one count that make a report count. + const COUNTED: usize = SLOW_PEER_REPORT_THRESHOLD + 1; + fn failed(non_priority: usize) -> FailedMessages { FailedMessages { priority: 0, @@ -607,40 +669,59 @@ mod tests { } } + /// Runs the checks before the last one of a streak, each with `peers` + /// reported once with their failed messages. + fn checks_before_the_limit(slow_peers: &mut SlowPeers, peers: &[(PeerId, usize)]) { + for _ in 1..SLOW_PEER_CHECK_LIMIT { + for (peer, non_priority) in peers { + slow_peers.record(*peer, &failed(*non_priority)); + } + assert_eq!(slow_peers.check(), nothing()); + } + } + + fn nothing() -> SlowPeerCheck { + SlowPeerCheck { + disconnect: Vec::new(), + deferred: 0, + } + } + + fn checks_in_a_streak() -> usize { + usize::try_from(SLOW_PEER_CHECK_LIMIT).unwrap() + } + #[test] fn a_peer_failing_in_every_check_is_returned_at_the_limit() { let peer = PeerId::random(); let mut slow_peers = SlowPeers::default(); - for _ in 1..SLOW_PEER_CHECK_LIMIT { - slow_peers.record(peer, &failed(2)); - assert!(slow_peers.check().is_empty()); - } - slow_peers.record(peer, &failed(2)); - let expected = 2 * usize::try_from(SLOW_PEER_CHECK_LIMIT).unwrap(); - assert_eq!(slow_peers.check(), vec![(peer, expected)]); + checks_before_the_limit(&mut slow_peers, &[(peer, COUNTED)]); + slow_peers.record(peer, &failed(COUNTED)); + let expected = SlowPeerCheck { + disconnect: vec![(peer, COUNTED * checks_in_a_streak())], + deferred: 0, + }; + assert_eq!(slow_peers.check(), expected); } #[test] fn a_check_without_a_report_resets_the_streak() { let peer = PeerId::random(); let mut slow_peers = SlowPeers::default(); - for _ in 1..SLOW_PEER_CHECK_LIMIT { - slow_peers.record(peer, &failed(1)); - assert!(slow_peers.check().is_empty()); - } - assert!(slow_peers.check().is_empty()); - slow_peers.record(peer, &failed(1)); - assert!(slow_peers.check().is_empty()); + checks_before_the_limit(&mut slow_peers, &[(peer, COUNTED)]); + assert_eq!(slow_peers.check(), nothing()); + slow_peers.record(peer, &failed(COUNTED)); + assert_eq!(slow_peers.check(), nothing()); } #[test] fn several_reports_within_one_check_count_once_towards_the_streak() { let peer = PeerId::random(); let mut slow_peers = SlowPeers::default(); - for _ in 0..usize::try_from(SLOW_PEER_CHECK_LIMIT).unwrap() { - slow_peers.record(peer, &failed(1)); + for _ in 0..checks_in_a_streak() { + slow_peers.record(peer, &failed(COUNTED)); } - assert!(slow_peers.check().is_empty()); + assert_eq!(slow_peers.check(), nothing()); } #[test] @@ -648,10 +729,98 @@ mod tests { let peer = PeerId::random(); let mut slow_peers = SlowPeers::default(); for _ in 0..SLOW_PEER_CHECK_LIMIT { - slow_peers.record(peer, &failed(1)); + slow_peers.record(peer, &failed(COUNTED)); slow_peers.check(); } - slow_peers.record(peer, &failed(1)); - assert!(slow_peers.check().is_empty()); + slow_peers.record(peer, &failed(COUNTED)); + assert_eq!(slow_peers.check(), nothing()); + } + + #[test] + fn a_report_at_the_threshold_does_not_count_and_resets_the_streak() { + let peer = PeerId::random(); + let mut slow_peers = SlowPeers::default(); + checks_before_the_limit(&mut slow_peers, &[(peer, COUNTED)]); + slow_peers.record(peer, &failed(SLOW_PEER_REPORT_THRESHOLD)); + assert_eq!(slow_peers.check(), nothing()); + slow_peers.record(peer, &failed(COUNTED)); + assert_eq!(slow_peers.check(), nothing()); + } + + #[test] + fn the_threshold_applies_to_each_count_on_its_own() { + let peer = PeerId::random(); + let mut slow_peers = SlowPeers::default(); + // Over the threshold only when the two counts are added up. + let split = FailedMessages { + priority: SLOW_PEER_REPORT_THRESHOLD, + non_priority: SLOW_PEER_REPORT_THRESHOLD, + }; + for _ in 0..SLOW_PEER_CHECK_LIMIT { + slow_peers.record(peer, &split); + assert_eq!(slow_peers.check(), nothing()); + } + // Priority failures alone are enough. + let priority = FailedMessages { + priority: COUNTED, + non_priority: 0, + }; + for _ in 1..SLOW_PEER_CHECK_LIMIT { + slow_peers.record(peer, &priority); + assert_eq!(slow_peers.check(), nothing()); + } + slow_peers.record(peer, &priority); + assert_eq!(slow_peers.check().disconnect.len(), 1); + } + + #[test] + fn one_check_returns_at_most_the_cap_most_failed_first() { + let mut slow_peers = SlowPeers::default(); + let peers: Vec<(PeerId, usize)> = (0..=MAX_SLOW_PEER_DISCONNECTS_PER_CHECK) + .map(|i| (PeerId::random(), COUNTED + i)) + .collect(); + checks_before_the_limit(&mut slow_peers, &peers); + for (peer, non_priority) in &peers { + slow_peers.record(*peer, &failed(*non_priority)); + } + let check = slow_peers.check(); + + let mut most_failed_first = peers.clone(); + most_failed_first.reverse(); + let expected: Vec<(PeerId, usize)> = most_failed_first + .iter() + .take(MAX_SLOW_PEER_DISCONNECTS_PER_CHECK) + .map(|(peer, non_priority)| (*peer, non_priority * checks_in_a_streak())) + .collect(); + assert_eq!(check.disconnect, expected); + assert_eq!(check.deferred, 1); + + // The deferred peer kept its streak: still failing, it is returned at + // the next check. + let (deferred_peer, non_priority) = peers[0]; + slow_peers.record(deferred_peer, &failed(non_priority)); + let expected = SlowPeerCheck { + disconnect: vec![(deferred_peer, non_priority * (checks_in_a_streak() + 1))], + deferred: 0, + }; + assert_eq!(slow_peers.check(), expected); + } + + #[test] + fn a_deferred_peer_that_stops_failing_loses_its_streak() { + let mut slow_peers = SlowPeers::default(); + let peers: Vec<(PeerId, usize)> = (0..=MAX_SLOW_PEER_DISCONNECTS_PER_CHECK) + .map(|i| (PeerId::random(), COUNTED + i)) + .collect(); + checks_before_the_limit(&mut slow_peers, &peers); + for (peer, non_priority) in &peers { + slow_peers.record(*peer, &failed(*non_priority)); + } + assert_eq!(slow_peers.check().deferred, 1); + + assert_eq!(slow_peers.check(), nothing()); + let (deferred_peer, non_priority) = peers[0]; + slow_peers.record(deferred_peer, &failed(non_priority)); + assert_eq!(slow_peers.check(), nothing()); } } diff --git a/docs/beacon_wire.md b/docs/beacon_wire.md index 76030089d..65a2ed7bf 100644 --- a/docs/beacon_wire.md +++ b/docs/beacon_wire.md @@ -334,6 +334,25 @@ such queues held about 490 MB. Gossipsub's own penalty for these failures graylist. Lighthouse penalizes `SlowPeer` reports through its peer manager instead. +Two limits keep a slow node of our own from costing it its peers. Gossipsub +counts a forward that waited too long in a peer's queue as the same failure as +a full queue, so a saturated uplink or a stalled runtime on our side makes +every peer look slow at once: + +- A report counts only when its priority or its non-priority failures exceed + `SLOW_PEER_REPORT_THRESHOLD`. Lighthouse applies the same cut. A dead stream + fails every message queued for the peer; a few failures in a heartbeat are + more likely late forwards. +- One check disconnects at most `MAX_SLOW_PEER_DISCONNECTS_PER_CHECK` peers, + the ones with the most failed messages first. The others keep their streak, + so a peer still failing at the next check is disconnected then. Peers held + back this way are logged as a warning, since many peers failing together + points at our side. + +A dead stream on a peer that gets little traffic, such as one gossipsub has +already pruned from the mesh, may stay under the threshold. Its queue then +stays pinned until the connection closes, as it did before this check existed. + Around a fork-digest boundary the node holds two digests' topics, and every held digest gets the whole parameter set (exits and slashings included), scored by its role: diff --git a/docs/metrics.md b/docs/metrics.md index d29ada89d..601ef32f5 100644 --- a/docs/metrics.md +++ b/docs/metrics.md @@ -454,7 +454,7 @@ spec. |------|------|-------|-------------------------|--------|---------| | `lean_gossipsub_peers_by_score` | Gauge | Gossipsub peers by score band | Every 10 s, in the swarm task | band=non_negative,negative,below_gossip,below_publish,below_graylist | | | `lean_gossipsub_score_disconnects_total` | Counter | Peers disconnected for a gossipsub score below the graylist (-16000) | When the 10 s check disconnects one | | | -| `lean_gossipsub_slow_peer_disconnects_total` | Counter | Peers disconnected for failing to take gossipsub messages in every 10 s check for a minute (a dead gossipsub stream pins its send queue in memory) | When the 10 s check disconnects one | | | +| `lean_gossipsub_slow_peer_disconnects_total` | Counter | Peers disconnected for failing to take gossipsub messages in every 10 s check for a minute (a dead gossipsub stream pins its send queue in memory). A report counts only above `SLOW_PEER_REPORT_THRESHOLD` failures, and one check disconnects at most `MAX_SLOW_PEER_DISCONNECTS_PER_CHECK` peers | When the 10 s check disconnects one | | | | `lean_gossipsub_mesh_delivery_scoring` | Gauge | 1 while mesh message deliveries (P3) are scored, 0 while the head lags the wall clock or is warming up after catching up | Every slot, on the scoring refresh | | | The bands are mutually exclusive and split at the gossip (-4000), publish