diff --git a/CLAUDE.md b/CLAUDE.md index a171704a8..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. 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/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..0993d7131 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,26 @@ 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; + +/// 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, @@ -269,10 +286,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 +323,7 @@ async fn swarm_loop( counters.num_established_outgoing(), ); disconnect_graylisted_peers(&mut swarm); + disconnect_slow_peers(&mut swarm, &mut slow_peers); } } }; @@ -324,6 +349,138 @@ 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. +/// +/// 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, 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 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. 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)); + 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)); + } + } + 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); + } + SlowPeerCheck { + disconnect: persistent, + deferred, + } + } +} + +/// 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 SlowPeerCheck { + disconnect, + deferred, + } = slow_peers.check(); + if swarm + .behaviour() + .gossipsub + .peer_score(swarm.local_peer_id()) + .is_none() + { + return; + } + 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!( + %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 +654,173 @@ pub(crate) fn execute_command(swarm: &mut libp2p::Swarm, cmd: SwarmCo } } } + +#[cfg(test)] +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, + non_priority, + } + } + + /// 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(); + 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(); + 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..checks_in_a_streak() { + slow_peers.record(peer, &failed(COUNTED)); + } + assert_eq!(slow_peers.check(), nothing()); + } + + #[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(COUNTED)); + slow_peers.check(); + } + 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 c2138785b..65a2ed7bf 100644 --- a/docs/beacon_wire.md +++ b/docs/beacon_wire.md @@ -320,6 +320,39 @@ 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. + +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: @@ -816,6 +849,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 7107778d1..601ef32f5 100644 --- a/docs/metrics.md +++ b/docs/metrics.md @@ -454,6 +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). 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