From 5b4ed6966e4384f21f9bb20823a26f8cc5987cb2 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:37:04 -0300 Subject: [PATCH] feat(p2p): metrics for whether gossip validation competes with import Dedicated validation workers, bounded queues or CPU budgets are only worth building if validation actually delays block import or epoch precompute. Nothing measured that: the validation permit pools had no metric, the existing arrival-to-verdict histogram cannot separate blocking-pool queueing from run time, and the precompute histogram gives a duration but no overlap with import latency. Add, instrumentation only (no pool size, scheduling or verdict changes): - lean_beacon_validation_permits_in_use{pool} and lean_beacon_validation_permit_exhausted_total{pool} - lean_beacon_validation_queue_seconds{kind} (spawn_blocking to closure start) and lean_beacon_validation_run_seconds{kind}, for gossip stateful checks and for the fetched column and envelope checks - lean_beacon_column_check_permit_wait_seconds{kind} - lean_beacon_epoch_precompute_running, 1 while a worker runs Block import already exposes the mailbox wait as lean_block_import_phase_seconds{phase="queue"}, so nothing is added there. docs/metrics.md documents each metric and the PromQL for the question. --- crates/blockchain/src/epoch_precompute.rs | 1 + crates/blockchain/src/metrics.rs | 29 +++ crates/net/p2p/src/beacon/column_checks.rs | 6 + crates/net/p2p/src/beacon/envelope_checks.rs | 8 +- crates/net/p2p/src/beacon/verdict.rs | 32 +++- crates/net/p2p/src/metrics.rs | 186 +++++++++++++++++++ docs/beacon_wire.md | 5 +- docs/metrics.md | 64 ++++++- 8 files changed, 316 insertions(+), 15 deletions(-) diff --git a/crates/blockchain/src/epoch_precompute.rs b/crates/blockchain/src/epoch_precompute.rs index 9c77e137a..be592baa5 100644 --- a/crates/blockchain/src/epoch_precompute.rs +++ b/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) diff --git a/crates/blockchain/src/metrics.rs b/crates/blockchain/src/metrics.rs index 34c67398f..b87362a82 100644 --- a/crates/blockchain/src/metrics.rs +++ b/crates/blockchain/src/metrics.rs @@ -118,6 +118,35 @@ impl SlotPosition { // --- Gauges --- +static LEAN_BEACON_EPOCH_PRECOMPUTE_RUNNING: std::sync::LazyLock = + 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 = std::sync::LazyLock::new(|| { register_int_gauge!("lean_head_slot", "Latest slot of the lean chain").unwrap() }); diff --git a/crates/net/p2p/src/beacon/column_checks.rs b/crates/net/p2p/src/beacon/column_checks.rs index 91fbd1efe..48025e040 100644 --- a/crates/net/p2p/src/beacon/column_checks.rs +++ b/crates/net/p2p/src/beacon/column_checks.rs @@ -46,12 +46,18 @@ pub(crate) fn check_and_forward(server: &P2PServer, sidecars: Vec 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) })); diff --git a/crates/net/p2p/src/beacon/verdict.rs b/crates/net/p2p/src/beacon/verdict.rs index ab5651c26..ad44a84c9 100644 --- a/crates/net/p2p/src/beacon/verdict.rs +++ b/crates/net/p2p/src/beacon/verdict.rs @@ -601,20 +601,26 @@ fn permits_for<'a>( server: &'a P2PServer, object: &Validated, ) -> &'a std::sync::Arc { + 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 diff --git a/crates/net/p2p/src/metrics.rs b/crates/net/p2p/src/metrics.rs index 643300967..ab74833c0 100644 --- a/crates/net/p2p/src/metrics.rs +++ b/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 = 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 = 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 = 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 = 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 = 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)); + } } diff --git a/docs/beacon_wire.md b/docs/beacon_wire.md index c2138785b..b1d3484ec 100644 --- a/docs/beacon_wire.md +++ b/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. diff --git a/docs/metrics.md b/docs/metrics.md index 1ea07953f..4071f8a2a 100644 --- a/docs/metrics.md +++ b/docs/metrics.md @@ -432,9 +432,67 @@ payload votes, and `builder_validation_permits` for bids and proposer preferences. They are deliberately separate: a mainnet slot's worth of aggregates and backbone attestations arrives every slot, not only during a range sync, and sharing one pool would let that burst answer `Ignore(Overloaded)` for a block or a column -instead. Neither pool has a metric of its own yet; a permit exhausted on -either shows up as `outcome="ignore",reason="overloaded"` on -`lean_beacon_gossip_validation_total`, for the topics that draw from it. +instead. `sync_validation_permits` (sync committee messages and contributions) +is a fourth, and `column_check_permits` bounds the checks on fetched columns and +envelopes, which await a permit instead of failing. A permit exhausted on a +gossip pool shows up as `outcome="ignore",reason="overloaded"` on +`lean_beacon_gossip_validation_total` and, per pool, on +`lean_beacon_validation_permit_exhausted_total` below. + +#### Validation contention + +These answer whether validation competes with block import and epoch precompute +for CPU. They only measure: no pool size, schedule or verdict depends on them. +ethlambda-specific, not part of the leanMetrics spec. + +| Name | Type | Usage | Sample collection event | Labels | Buckets | +|------|------|-------|-------------------------|--------|---------| +| `lean_beacon_validation_permits_in_use` | Gauge | Permits held per pool: blocking tasks queued or running | A permit is acquired / released | pool=gossip,attestation,builder,sync,column_check | | +| `lean_beacon_validation_permit_exhausted_total` | Counter | Objects reported `Ignore(Overloaded)` because their pool had no free permit | On each failed `try_acquire` | pool=gossip,attestation,builder,sync | | +| `lean_beacon_validation_queue_seconds` | Histogram | Time from `spawn_blocking` being called to the closure starting (tokio blocking-pool queueing) | When the closure starts | kind | 0.0001, 0.0005, 0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2, 4 | +| `lean_beacon_validation_run_seconds` | Histogram | Run time of the closure itself | When the closure ends (panic included) | kind | same as above | +| `lean_beacon_column_check_permit_wait_seconds` | Histogram | Time a fetched sidecar or envelope waited for a `column_check` permit | When the permit is granted | kind=column_check,envelope_check | same as above | +| `lean_beacon_epoch_precompute_running` | Gauge | 1 while an epoch precompute worker runs its advance | Worker start / end | | | + +`kind` on the queue and run histograms is the gossip topic kind above for gossip +validation. `column_check` and `envelope_check` are the tasks of +`check_and_forward` (fetched and replayed columns) and of the envelope checks: +their queue delay is measured from `spawn_blocking`, after the permit wait, which +has its own histogram, so the two add up. The in-use gauge counts a task from its +permit being taken, so one waiting in the blocking pool counts too; it is a count +of admitted work, not of busy threads. Only +`lean_beacon_validation_queue_seconds` isolates time spent waiting for a thread. + +`lean_beacon_epoch_precompute_running` is set when the worker's closure starts +and cleared when it ends, so the interval matches +`lean_beacon_epoch_precompute_seconds` (which gives a duration, not an overlap +with other series). At most one worker runs at a time. + +Block import needs no new series: `lean_block_import_phase_seconds{phase="queue"}` +is already the wait in the chain actor's mailbox before the import starts (see +Block Import Timing). + +**Does validation compete with import?** Plot these on one panel: + +```promql +# Validation work admitted, waiting for a thread, and refused +sum(lean_beacon_validation_permits_in_use) +histogram_quantile(0.99, sum by (le) (rate(lean_beacon_validation_queue_seconds_bucket[5m]))) +sum by (pool) (rate(lean_beacon_validation_permit_exhausted_total[5m])) + +# Against import's STF phase +histogram_quantile(0.99, sum by (le) (rate(lean_block_import_phase_seconds_bucket{phase="stf"}[5m]))) + +# Precompute overlap: fraction of the window a worker ran +avg_over_time(lean_beacon_epoch_precompute_running[5m]) +``` + +Reading it: if STF p99 rises in the windows where queue-delay p99 or in-use +rises, the two share threads (the tokio blocking pool, or cores). If queue delay +stays near zero and STF p99 still moves, suspect cores instead: validation's CPU +share is roughly `sum(rate(lean_beacon_validation_run_seconds_sum[5m]))`, in +cores. To see whether precompute hurts import, compare STF p99 in windows where +`lean_beacon_epoch_precompute_running` is 1 against windows where it is 0. ### Beacon Peer Scoring