feat(beacon): score gossipsub peers with lighthouse's parameters

CLAUDE.md

@@ -335,6 +335,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)
   - 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/blockchain/state_transition/src/beacon/helpers/accessors.rs

@@ -131,8 +131,10 @@ pub fn get_seed(state: &BeaconState, epoch: Epoch, domain_type: DomainType) -> B
 /// split out so [`build_epoch_committees`] can share one
 /// [`get_active_validator_indices`] scan between this and its own committee
 /// derivation, rather than [`get_committee_count_per_slot`] repeating the scan
-/// the caller already did to get `active_count` in the first place.
-fn committee_count_per_slot(active_count: u64) -> u64 {
+/// the caller already did to get `active_count` in the first place. Public
+/// for p2p's gossipsub scoring, which sizes expected aggregate traffic from
+/// a validator count rather than from a state.
+pub fn committee_count_per_slot(active_count: u64) -> u64 {
     let ideal = active_count / preset::SLOTS_PER_EPOCH / preset::TARGET_COMMITTEE_SIZE;
     ideal.clamp(1, preset::MAX_COMMITTEES_PER_SLOT as u64)
 }

crates/common/types/src/beacon/committees.rs

@@ -104,6 +104,13 @@ impl EpochCommittees {
         self.committees_per_slot
     }
 
+    /// How many validators are active in this epoch: the shuffle is a
+    /// permutation of exactly that set, so its length is the count without
+    /// another registry scan.
+    pub fn active_validator_count(&self) -> u64 {
+        self.shuffled.len() as u64
+    }
+
     /// The committee at `slot` with `index`.
     ///
     /// The same members `ethlambda-state-transition`'s `get_beacon_committee`

crates/net/p2p/src/beacon/mod.rs

@@ -22,6 +22,7 @@ pub mod fork_schedule;
 pub mod handler;
 pub mod messages;
 pub mod protocols;
+pub mod scoring;
 pub mod subnets;
 pub mod swarm;
 pub mod sync_committee;

crates/net/p2p/src/beacon/swarm.rs

@@ -1,10 +1,11 @@
 //! The beacon half of the swarm configuration.
 //!
-//! Five things differ from lean at the swarm level: the topic set, the protocol
-//! set, the `seen_ttl`, the identify protocol version and the connection limits.
-//! [`crate::build_swarm`] resolves all five from the variant it is handed, and
-//! every one of them that is a beacon *value* rather than a lean one lives here,
-//! so tuning the mainnet numbers never means editing the crate root.
+//! Six things differ from lean at the swarm level: the topic set, the protocol
+//! set, the `seen_ttl`, the identify protocol version, the connection limits and
+//! peer scoring. [`crate::build_swarm`] resolves all six from the variant it is
+//! handed, and every one of them that is a beacon *value* rather than a lean one
+//! lives here or, for scoring, in [`crate::beacon::scoring`], so tuning the
+//! mainnet numbers never means editing the crate root.
 
 use std::time::Duration;
 

crates/net/p2p/src/beacon/transition.rs

@@ -107,13 +107,16 @@ pub(crate) fn advance(server: &mut P2PServer) -> Option<Duration> {
 
 /// Move the node to the state the schedule prescribes at `epoch`.
 ///
-/// 1. Join the topics of every digest the window now covers, including the
+/// 1. Hand the swarm the peer-scoring parameters of every digest the window
+///    now covers, each by its role (beacon scoring only; see
+///    [`super::scoring::PeerScoring::hold`]).
+/// 2. Join the topics of every digest the window now covers, including the
 ///    aggregator-subnet topics already held for the current slot.
-/// 2. Leave the topics of every digest the window no longer covers.
-/// 3. If the current digest changed, switch what this node publishes and
+/// 3. Leave the topics of every digest the window no longer covers.
+/// 4. If the current digest changed, switch what this node publishes and
 ///    advertises: digest, fork, metadata sequence number, metric, and the
 ///    `eth2` entry.
-/// 4. Point the admission filter at the new entry, admitting every held digest.
+/// 5. Point the admission filter at the new entry, admitting every held digest.
 pub(crate) fn apply(server: &mut P2PServer, epoch: Epoch) -> Changes {
     let mut changes = Changes::default();
     let Wire::Beacon(wire) = &mut server.wire else {
@@ -132,6 +135,19 @@ pub(crate) fn apply(server: &mut P2PServer, epoch: Epoch) -> Changes {
     }
     let column_subnets = topics::column_subnets(&wire.custody_columns);
 
+    // Before any join: gossipsub tracks a mesh peer on a topic only if the
+    // topic had parameters when the peer was grafted, and joining grafts at
+    // once.
+    if let Some(scoring) = &mut server.peer_scoring {
+        let digests = held.iter().map(|entry| entry.digest).collect();
+        scoring.hold(
+            digests,
+            current.digest,
+            wire.wall_slot(),
+            &server.swarm_handle,
+        );
+    }
+
     let mut existing = std::mem::take(&mut wire.window_topics);
     existing.push(wire.topics.clone());
 

crates/net/p2p/src/gossipsub/handler.rs

@@ -1143,7 +1143,7 @@ pub async fn publish_payload_attestation_message(
 }
 
 /// The beacon wall-clock slot, from the wire's genesis and slot duration.
-fn beacon_wall_slot(wire: &BeaconWire) -> u64 {
+pub(crate) fn beacon_wall_slot(wire: &BeaconWire) -> u64 {
     wire.wall_slot()
 }
 

crates/net/p2p/src/gossipsub/mod.rs

@@ -3,6 +3,7 @@ mod handler;
 mod messages;
 
 pub use encoding::{compress_message, decompress_message};
+pub(crate) use handler::beacon_wall_slot;
 pub(crate) use handler::operation_kind;
 pub use handler::{
     handle_gossip_message, join_aggregator_subnets, leave_expired_aggregator_subnets,

crates/net/p2p/src/lib.rs

@@ -657,9 +657,9 @@ pub struct SwarmConfig {
     /// Ignored on lean, which runs [`unlimited_connections`].
     pub target_peers: usize,
     /// Which network's wire to build. Decides the gossip topics, the req/resp
-    /// protocol set, the gossipsub `seen_ttl`, the identify protocol version and
-    /// the connection limits, and so decides the [`Wire`] the built swarm
-    /// carries.
+    /// protocol set, the gossipsub `seen_ttl`, the identify protocol version,
+    /// the connection limits and whether peers are scored, and so decides the
+    /// [`Wire`] the built swarm carries.
     pub wire: WireConfig,
     /// The identify `agentVersion`, which crawlers read to name a peer's
     /// client. Left unset, rust-libp2p reports its own crate version instead.
@@ -812,6 +812,11 @@ pub enum SwarmBuildError {
     Subscription(#[from] libp2p::gossipsub::SubscriptionError),
 }
 
+/// Gossipsub's target mesh size, the spec's `D`. Named because beacon peer
+/// scoring sizes its first-delivery caps by it; see
+/// [`beacon::scoring::ScoreSettings`].
+pub(crate) const MESH_N: usize = 8;
+
 /// The gossipsub parameters both wires share.
 ///
 /// `mesh_n` 8, low 6, high 12, the 700ms heartbeat, and the 6/3 history already
@@ -830,7 +835,7 @@ pub(crate) fn gossipsub_config(
     let mut builder = libp2p::gossipsub::ConfigBuilder::default();
     builder
         // d
-        .mesh_n(8)
+        .mesh_n(MESH_N)
         // d_low
         .mesh_n_low(6)
         // d_high
@@ -858,9 +863,10 @@ pub(crate) fn gossipsub_config(
 /// Build and configure the libp2p swarm, dial bootnodes, subscribe to topics.
 ///
 /// One builder for both networks. Four things differ at the behaviour level and
-/// are resolved in the first match below; the topic set differs and is resolved
-/// in the last one. Everything in between, the transport, the QUIC and TCP
-/// listeners and the static bootnode dialing, is the same on either wire.
+/// are resolved in the first match below, peer scoring (beacon only) just after
+/// it; the topic set differs and is resolved in the last one. Everything in
+/// between, the transport, the QUIC and TCP listeners and the static bootnode
+/// dialing, is the same on either wire.
 pub fn build_swarm(config: SwarmConfig) -> Result<BuiltSwarm, SwarmBuildError> {
     let SwarmConfig {
         node_key,
@@ -895,11 +901,27 @@ pub fn build_swarm(config: SwarmConfig) -> Result<BuiltSwarm, SwarmBuildError> {
     };
 
     let validate_messages = matches!(wire, WireConfig::Beacon(_));
-    let gossipsub = libp2p::gossipsub::Behaviour::new(
+    let mut gossipsub = libp2p::gossipsub::Behaviour::new(
         MessageAuthenticity::Anonymous,
         gossipsub_config(seen_ttl, validate_messages),
     )
     .expect("failed to initiate behaviour");
+    // Beacon only: the parameters are sized from the beacon chain's topics
+    // and message rates, and lean has no set of its own. Without a verdict
+    // on each message (lean's are auto-accepted) the invalid-message term
+    // would have nothing to count either. Built for a placeholder validator
+    // count; `P2PServer`'s first scoring refresh, at startup, replaces the
+    // topic parameters with the head's.
+    if let WireConfig::Beacon(beacon) = &wire {
+        let settings = beacon::scoring::ScoreSettings::new(&beacon.config, MESH_N);
+        let params = settings.peer_score_params(
+            beacon.fork_digest,
+            beacon::scoring::ScoreInputs::placeholder(),
+        );
+        gossipsub
+            .with_peer_score(params, beacon::scoring::thresholds())
+            .expect("valid peer score parameters");
+    }
 
     let req_resp = ReqResp::new(codec, &wire);
 
@@ -921,8 +943,6 @@ pub fn build_swarm(config: SwarmConfig) -> Result<BuiltSwarm, SwarmBuildError> {
         req_resp,
     };
 
-    // TODO: set peer scoring params
-
     let mut swarm = libp2p::SwarmBuilder::with_existing_identity(identity)
         .with_tokio()
         .with_tcp(
@@ -1165,6 +1185,10 @@ impl P2P {
             .wire
             .beacon()
             .map_or(0, |beacon| beacon.attestation_subnets.len());
+        let peer_scoring = built
+            .wire
+            .beacon()
+            .map(|beacon| beacon::scoring::PeerScoring::new(&beacon.config, beacon.fork_digest));
 
         let server = P2PServer {
             swarm_handle,
@@ -1217,9 +1241,20 @@ impl P2P {
             )),
             sync_committee_pool,
             aggregator_subnets: HashMap::new(),
+            peer_scoring,
         };
+        let scores_peers = server.peer_scoring.is_some();
         let discovery_enabled = server.discovery.is_some();
         let handle = server.start();
+        // At once rather than a slot in: the swarm starts on placeholder
+        // topic scores, and the head's are already known.
+        if scores_peers {
+            send_after(
+                Duration::ZERO,
+                handle.context(),
+                p2p_protocol::RefreshPeerScoring,
+            );
+        }
         send_after(
             AGGREGATOR_SUBNET_SWEEP_INTERVAL,
             handle.context(),
@@ -1394,6 +1429,10 @@ pub struct P2PServer {
     /// advertised in `attnets`, and left once the slot has passed. The
     /// backbone subnets are separate and never left.
     pub(crate) aggregator_subnets: HashMap<u64, u64>,
+
+    /// Gossipsub scoring's inputs and gate. `None` on lean, which runs
+    /// without peer scoring. See [`beacon::scoring`].
+    pub(crate) peer_scoring: Option<beacon::scoring::PeerScoring>,
 }
 
 impl P2PServer {
@@ -1487,6 +1526,8 @@ pub(crate) trait P2PProtocol: Send + Sync {
     /// then by its own timer for the next join, switch or leave.
     #[allow(dead_code)] // invoked via send_after, not called directly
     fn advance_fork_schedule(&self) -> Result<(), ActorError>;
+    #[allow(dead_code)] // invoked via send_after, not called directly
+    fn refresh_peer_scoring(&self) -> Result<(), ActorError>;
 }
 
 #[actor(protocol = P2PProtocol)]
@@ -1645,6 +1686,21 @@ impl P2PServer {
         send_after(interval, ctx.clone(), p2p_protocol::DiscoverPeers);
     }
 
+    /// Rebuild the gossipsub topic scores from the head, once a slot. Beacon
+    /// only: never scheduled on lean. See [`beacon::scoring::refresh`].
+    #[send_handler]
+    async fn handle_refresh_peer_scoring(
+        &mut self,
+        _msg: p2p_protocol::RefreshPeerScoring,
+        ctx: &Context<Self>,
+    ) {
+        if let Some(wire) = self.wire.beacon() {
+            let slot = Duration::from_millis(wire.config.slot_duration_ms);
+            send_after(slot, ctx.clone(), p2p_protocol::RefreshPeerScoring);
+        }
+        beacon::scoring::refresh(self);
+    }
+
     /// The deadline of a range batch held back for custody. Scheduled once,
     /// when the batch starts waiting; see [`RANGE_BATCH_CUSTODY_WAIT`].
     #[send_handler]
@@ -3013,6 +3069,7 @@ pub(crate) mod test_support {
             )),
             sync_committee_pool: Default::default(),
             aggregator_subnets: HashMap::new(),
+            peer_scoring: None,
         }
     }
 

crates/net/p2p/src/metrics.rs

@@ -591,6 +591,51 @@ pub fn set_swarm_established_connections(inbound: u32, outbound: u32) {
         .set(i64::from(outbound));
 }
 
+/// Set how many gossipsub peers fall in each score band. `bands` names every
+/// band, empty ones as zero, so a band that empties does not keep its last
+/// count. See `beacon::scoring::score_band`.
+pub fn set_gossipsub_peers_by_score(bands: &HashMap<&'static str, i64>) {
+    static LEAN_GOSSIPSUB_PEERS_BY_SCORE: LazyLock<IntGaugeVec> = LazyLock::new(|| {
+        register_int_gauge_vec!(
+            "lean_gossipsub_peers_by_score",
+            "Gossipsub peers by score band: non_negative, negative, below_gossip, below_publish, below_graylist",
+            &["band"]
+        )
+        .unwrap()
+    });
+    for (band, count) in bands {
+        LEAN_GOSSIPSUB_PEERS_BY_SCORE
+            .with_label_values(&[band])
+            .set(*count);
+    }
+}
+
+/// Count a peer disconnected for a gossipsub score below the graylist.
+pub fn inc_gossipsub_score_disconnects() {
+    static LEAN_GOSSIPSUB_SCORE_DISCONNECTS_TOTAL: LazyLock<IntCounter> = LazyLock::new(|| {
+        register_int_counter!(
+            "lean_gossipsub_score_disconnects_total",
+            "Peers disconnected for a gossipsub score below the graylist threshold"
+        )
+        .unwrap()
+    });
+    LEAN_GOSSIPSUB_SCORE_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`.
+pub fn set_gossipsub_mesh_delivery_scoring(scored: bool) {
+    static LEAN_GOSSIPSUB_MESH_DELIVERY_SCORING: LazyLock<IntGauge> = LazyLock::new(|| {
+        register_int_gauge!(
+            "lean_gossipsub_mesh_delivery_scoring",
+            "1 while gossipsub scores mesh message deliveries, 0 while the head lags or is warming up"
+        )
+        .unwrap()
+    });
+    LEAN_GOSSIPSUB_MESH_DELIVERY_SCORING.set(i64::from(scored));
+}
+
 /// Set how many connected peers are known to custody `column`.
 ///
 /// A peer counts only once it has answered `metadata/3` or arrived with a

crates/net/p2p/src/req_resp/handlers.rs

@@ -2840,6 +2840,7 @@ pub(crate) mod tests {
             )),
             sync_committee_pool: Default::default(),
             aggregator_subnets: HashMap::new(),
+            peer_scoring: None,
         }
     }