feat(p2p): metrics for whether gossip validation competes with import

crates/blockchain/src/epoch_precompute.rs

@@ -221,6 +221,7 @@ impl BlockChainServer {
         let config = self.store.config();
         let actor = ctx.actor_ref();
         tokio::task::spawn_blocking(move || {
+            let _running = crate::metrics::EpochPrecomputeRunning::start();
             let result = {
                 let _timing = stf_metrics::time_epoch_precompute();
                 fork_choice::advance_to_epoch_start(&head_state, key.epoch, &config)

crates/blockchain/src/metrics.rs

@@ -118,6 +118,35 @@ impl SlotPosition {
 
 // --- Gauges ---
 
+static LEAN_BEACON_EPOCH_PRECOMPUTE_RUNNING: std::sync::LazyLock<IntGauge> =
+    std::sync::LazyLock::new(|| {
+        register_int_gauge!(
+            "lean_beacon_epoch_precompute_running",
+            "1 while an epoch precompute worker is running its advance, 0 otherwise"
+        )
+        .unwrap()
+    });
+
+/// Marks an epoch precompute worker as running in
+/// `lean_beacon_epoch_precompute_running` until dropped. A guard rather than a
+/// set/unset pair so a panicking worker still clears the gauge. At most one
+/// worker runs at a time (`epoch_precompute_in_flight`), so a plain 0/1 gauge
+/// is exact.
+pub struct EpochPrecomputeRunning;
+
+impl EpochPrecomputeRunning {
+    pub fn start() -> Self {
+        LEAN_BEACON_EPOCH_PRECOMPUTE_RUNNING.set(1);
+        Self
+    }
+}
+
+impl Drop for EpochPrecomputeRunning {
+    fn drop(&mut self) {
+        LEAN_BEACON_EPOCH_PRECOMPUTE_RUNNING.set(0);
+    }
+}
+
 static LEAN_HEAD_SLOT: std::sync::LazyLock<IntGauge> = std::sync::LazyLock::new(|| {
     register_int_gauge!("lean_head_slot", "Latest slot of the lean chain").unwrap()
 });

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

@@ -46,12 +46,18 @@ pub(crate) fn check_and_forward(server: &P2PServer, sidecars: Vec<DataColumnSide
         for sidecar in sidecars {
             // The semaphore is never closed, so this only returns once a
             // permit is free.
+            let waiting_since = std::time::Instant::now();
             let Ok(permit) = permits.clone().acquire_owned().await else {
                 return;
             };
+            metrics::observe_column_check_permit_wait("column_check", waiting_since.elapsed());
+            let in_use = metrics::PoolInUse::new("column_check");
+            let task = metrics::BlockingTask::new("column_check");
             let store = store.clone();
             checks.push(tokio::task::spawn_blocking(move || {
                 let _permit = permit;
+                let _in_use = in_use;
+                let _run = task.start();
                 let verdict = column::chain_checks_for(&store, &sidecar, unix_now_ms());
                 (sidecar, verdict)
             }));

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

@@ -32,7 +32,7 @@ use ethlambda_storage::Store;
 use ethlambda_types::beacon::containers::gloas::SignedExecutionPayloadEnvelope;
 use tracing::{debug, error, warn};
 
-use crate::P2PServer;
+use crate::{P2PServer, metrics};
 
 /// Whether a fetched envelope goes to the chain actor.
 pub(crate) fn forwards(outcome: &Outcome) -> bool {
@@ -103,15 +103,21 @@ pub(crate) fn check_and_forward_then(
         for envelope in envelopes {
             // The semaphore is never closed, so this only returns once a
             // permit is free.
+            let waiting_since = std::time::Instant::now();
             let Ok(permit) = permits.clone().acquire_owned().await else {
                 if let Some(done) = done {
                     done();
                 }
                 return;
             };
+            metrics::observe_column_check_permit_wait("envelope_check", waiting_since.elapsed());
+            let in_use = metrics::PoolInUse::new("column_check");
+            let task = metrics::BlockingTask::new("envelope_check");
             let store = store.clone();
             checks.push(tokio::task::spawn_blocking(move || {
                 let _permit = permit;
+                let _in_use = in_use;
+                let _run = task.start();
                 let outcome = judge(&store, &envelope);
                 (envelope, outcome)
             }));

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

@@ -601,20 +601,26 @@ fn permits_for<'a>(
     server: &'a P2PServer,
     object: &Validated,
 ) -> &'a std::sync::Arc<tokio::sync::Semaphore> {
+    match pool_of(object) {
+        "gossip" => &server.gossip_validation_permits,
+        "attestation" => &server.attestation_validation_permits,
+        "builder" => &server.builder_validation_permits,
+        _ => &server.sync_validation_permits,
+    }
+}
+
+/// The label (see [`metrics::VALIDATION_POOLS`]) of the permit pool
+/// [`permits_for`] draws `object`'s permit from. The one `match` on the
+/// object kind, so the pool metrics and the pool actually used cannot differ.
+fn pool_of(object: &Validated) -> &'static str {
     match object {
-        Validated::Block { .. } | Validated::Column(_) | Validated::Envelope(_) => {
-            &server.gossip_validation_permits
-        }
+        Validated::Block { .. } | Validated::Column(_) | Validated::Envelope(_) => "gossip",
         Validated::Aggregate { .. }
         | Validated::Attestation { .. }
         | Validated::PayloadAttestation(_)
-        | Validated::Operation(_) => &server.attestation_validation_permits,
-        Validated::ExecutionPayloadBid { .. } | Validated::ProposerPreferences(_) => {
-            &server.builder_validation_permits
-        }
-        Validated::SyncCommitteeMessage { .. } | Validated::SyncContribution(_) => {
-            &server.sync_validation_permits
-        }
+        | Validated::Operation(_) => "attestation",
+        Validated::ExecutionPayloadBid { .. } | Validated::ProposerPreferences(_) => "builder",
+        Validated::SyncCommitteeMessage { .. } | Validated::SyncContribution(_) => "sync",
     }
 }
 
@@ -643,7 +649,9 @@ pub(crate) fn spawn_stateful_checks(
     id: GossipId,
     object: Validated,
 ) {
+    let pool = pool_of(&object);
     let Ok(permit) = permits_for(server, &object).clone().try_acquire_owned() else {
+        metrics::inc_validation_permit_exhausted(pool);
         let received_at = id.received_at;
         let outcome = Outcome::Ignore(IgnoreReason::Overloaded);
         report(server, id, outcome);
@@ -652,8 +660,12 @@ pub(crate) fn spawn_stateful_checks(
     };
     let store = server.store.clone();
     let actor = ctx.actor_ref();
+    let in_use = metrics::PoolInUse::new(pool);
+    let task = metrics::BlockingTask::new(id.kind);
     tokio::task::spawn_blocking(move || {
         let _permit = permit;
+        let _in_use = in_use;
+        let _run = task.start();
         let mut object = object;
         let outcome = guarded(|| object.stateful_checks(&store));
         let _ = actor

crates/net/p2p/src/metrics.rs

@@ -450,6 +450,10 @@ pub fn init() {
     for &reason in DATA_COLUMN_REJECT_REASONS {
         LEAN_DATA_COLUMNS_REJECTED_TOTAL.with_label_values(&[reason]);
     }
+    for &pool in VALIDATION_POOLS {
+        LEAN_BEACON_VALIDATION_PERMITS_IN_USE.with_label_values(&[pool]);
+        LEAN_BEACON_VALIDATION_PERMIT_EXHAUSTED_TOTAL.with_label_values(&[pool]);
+    }
 }
 
 /// Count one beacon gossip verdict and how long it took from arrival.
@@ -467,6 +471,151 @@ pub fn observe_beacon_gossip_verdict(
         .observe(elapsed.as_secs_f64());
 }
 
+/// Labels of the permit pools that bound blocking-thread validation work.
+/// `column_check` is [`P2PServer::column_check_permits`](crate::P2PServer),
+/// shared by the fetched-column and fetched-envelope checks; the rest are the
+/// four gossip pools.
+pub(crate) const VALIDATION_POOLS: &[&str] =
+    &["gossip", "attestation", "builder", "sync", "column_check"];
+
+static LEAN_BEACON_VALIDATION_PERMITS_IN_USE: LazyLock<IntGaugeVec> = LazyLock::new(|| {
+    register_int_gauge_vec!(
+        "lean_beacon_validation_permits_in_use",
+        "Permits held right now, per validation permit pool (one per blocking task queued or running)",
+        &["pool"]
+    )
+    .unwrap()
+});
+
+static LEAN_BEACON_VALIDATION_PERMIT_EXHAUSTED_TOTAL: LazyLock<IntCounterVec> = LazyLock::new(
+    || {
+        register_int_counter_vec!(
+            "lean_beacon_validation_permit_exhausted_total",
+            "Gossip objects reported Ignore(Overloaded) because their permit pool had no free permit",
+            &["pool"]
+        )
+        .unwrap()
+    },
+);
+
+/// Buckets for blocking-pool queue delay and closure run time: from a
+/// scheduler hop to a verdict that has eaten a whole gossip deadline.
+const VALIDATION_TASK_BUCKETS: &[f64] = &[
+    0.0001, 0.0005, 0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.0, 4.0,
+];
+
+static LEAN_BEACON_VALIDATION_QUEUE_SECONDS: LazyLock<HistogramVec> = LazyLock::new(|| {
+    register_histogram_vec!(
+        "lean_beacon_validation_queue_seconds",
+        "Time from spawn_blocking being called for a validation task to its closure starting",
+        &["kind"],
+        VALIDATION_TASK_BUCKETS.to_vec()
+    )
+    .unwrap()
+});
+
+static LEAN_BEACON_VALIDATION_RUN_SECONDS: LazyLock<HistogramVec> = LazyLock::new(|| {
+    register_histogram_vec!(
+        "lean_beacon_validation_run_seconds",
+        "Run time of a validation task's blocking closure",
+        &["kind"],
+        VALIDATION_TASK_BUCKETS.to_vec()
+    )
+    .unwrap()
+});
+
+static LEAN_BEACON_COLUMN_CHECK_PERMIT_WAIT_SECONDS: LazyLock<HistogramVec> = LazyLock::new(|| {
+    register_histogram_vec!(
+        "lean_beacon_column_check_permit_wait_seconds",
+        "Time a fetched sidecar or envelope waited for a column_check permit",
+        &["kind"],
+        VALIDATION_TASK_BUCKETS.to_vec()
+    )
+    .unwrap()
+});
+
+/// A held permit of the pool `pool`, counted in
+/// `lean_beacon_validation_permits_in_use` for as long as the guard lives.
+/// Move it into the blocking closure next to the permit itself, so the gauge
+/// falls when the permit is released (including on a panic's unwind).
+pub(crate) struct PoolInUse(&'static str);
+
+impl PoolInUse {
+    pub(crate) fn new(pool: &'static str) -> Self {
+        LEAN_BEACON_VALIDATION_PERMITS_IN_USE
+            .with_label_values(&[pool])
+            .inc();
+        Self(pool)
+    }
+}
+
+impl Drop for PoolInUse {
+    fn drop(&mut self) {
+        LEAN_BEACON_VALIDATION_PERMITS_IN_USE
+            .with_label_values(&[self.0])
+            .dec();
+    }
+}
+
+/// Count one object refused for want of a permit in `pool`.
+pub(crate) fn inc_validation_permit_exhausted(pool: &str) {
+    LEAN_BEACON_VALIDATION_PERMIT_EXHAUSTED_TOTAL
+        .with_label_values(&[pool])
+        .inc();
+}
+
+/// Record how long a fetched sidecar or envelope (`kind`) awaited a
+/// `column_check` permit.
+pub(crate) fn observe_column_check_permit_wait(kind: &str, waited: std::time::Duration) {
+    LEAN_BEACON_COLUMN_CHECK_PERMIT_WAIT_SECONDS
+        .with_label_values(&[kind])
+        .observe(waited.as_secs_f64());
+}
+
+/// Times one `spawn_blocking` validation task. Create it immediately before
+/// the `spawn_blocking` call, move it into the closure and call
+/// [`Self::start`] first thing there: the gap is the tokio blocking pool's
+/// queueing, and the returned guard observes the closure's run time when it
+/// drops, a panic's unwind included.
+pub(crate) struct BlockingTask {
+    kind: &'static str,
+    spawned_at: std::time::Instant,
+}
+
+impl BlockingTask {
+    pub(crate) fn new(kind: &'static str) -> Self {
+        Self {
+            kind,
+            spawned_at: std::time::Instant::now(),
+        }
+    }
+
+    pub(crate) fn start(self) -> BlockingRun {
+        let started_at = std::time::Instant::now();
+        LEAN_BEACON_VALIDATION_QUEUE_SECONDS
+            .with_label_values(&[self.kind])
+            .observe(started_at.duration_since(self.spawned_at).as_secs_f64());
+        BlockingRun {
+            kind: self.kind,
+            started_at,
+        }
+    }
+}
+
+/// Observes a blocking validation closure's run time on drop.
+pub(crate) struct BlockingRun {
+    kind: &'static str,
+    started_at: std::time::Instant,
+}
+
+impl Drop for BlockingRun {
+    fn drop(&mut self) {
+        LEAN_BEACON_VALIDATION_RUN_SECONDS
+            .with_label_values(&[self.kind])
+            .observe(self.started_at.elapsed().as_secs_f64());
+    }
+}
+
 /// Count one verdict gossipsub could no longer act on.
 pub fn inc_beacon_gossip_verdict_expired(kind: &str) {
     LEAN_BEACON_GOSSIP_VERDICT_EXPIRED_TOTAL
@@ -685,4 +834,41 @@ mod tests {
         assert_eq!(gauge.with_label_values(&["bbbbbbbb"]).get(), 1);
         assert_eq!(gauge.with_label_values(&["aaaaaaaa"]).get(), 0);
     }
+
+    #[test]
+    fn a_held_permit_is_counted_until_its_guard_drops() {
+        // A pool label no real pool uses, so parallel tests cannot move it.
+        let gauge = || {
+            LEAN_BEACON_VALIDATION_PERMITS_IN_USE
+                .with_label_values(&["test_pool"])
+                .get()
+        };
+        let first = PoolInUse::new("test_pool");
+        let second = PoolInUse::new("test_pool");
+        assert_eq!(gauge(), 2);
+        drop(first);
+        assert_eq!(gauge(), 1);
+        drop(second);
+        assert_eq!(gauge(), 0);
+    }
+
+    #[test]
+    fn a_blocking_task_observes_queue_delay_and_run_time_once() {
+        let queue = || {
+            LEAN_BEACON_VALIDATION_QUEUE_SECONDS
+                .with_label_values(&["test_kind"])
+                .get_sample_count()
+        };
+        let run = || {
+            LEAN_BEACON_VALIDATION_RUN_SECONDS
+                .with_label_values(&["test_kind"])
+                .get_sample_count()
+        };
+        let task = BlockingTask::new("test_kind");
+        assert_eq!((queue(), run()), (0, 0));
+        let running = task.start();
+        assert_eq!((queue(), run()), (1, 0));
+        drop(running);
+        assert_eq!((queue(), run()), (1, 1));
+    }
 }

docs/beacon_wire.md

@@ -183,7 +183,10 @@ operation topics. A
 mainnet slot carries up to `MAX_COMMITTEES_PER_SLOT *
 TARGET_AGGREGATORS_PER_COMMITTEE` aggregates alone, arriving every slot rather
 than only during a range sync, which is why that traffic needs a pool of its
-own rather than sharing the block and column one. Accept propagates the
+own rather than sharing the block and column one. Each pool's occupancy,
+refusals and blocking-pool queue delay are metrics, with the queries that tell
+whether validation competes with block import: see "Validation contention" in
+[`metrics.md`](metrics.md). Accept propagates the
 message; a message whose dependency is not ready yet is IGNOREd. See
 [Aggregate attestations](#aggregate-attestations) for the two topics with
 their own section.