fix(p2p): disconnect peers whose gossipsub send queue never drains

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

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<IntCounter> = 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`.