diff --git a/crates/blockchain/src/lib.rs b/crates/blockchain/src/lib.rs index 98b62f00..ce56083e 100644 --- a/crates/blockchain/src/lib.rs +++ b/crates/blockchain/src/lib.rs @@ -2,7 +2,7 @@ use std::collections::{HashMap, HashSet, VecDeque}; use std::time::{Duration, Instant, SystemTime}; use ethlambda_crypto::signature::{ValidatorPublicKey, ValidatorSignature}; -use ethlambda_network_api::{BlockChainToP2PRef, InitP2P}; +use ethlambda_network_api::{BlockChainToP2PRef, BlockSource, InitP2P}; use ethlambda_state_transition::is_proposer; use ethlambda_storage::{ALL_TABLES, Store}; use ethlambda_types::{ @@ -109,6 +109,20 @@ impl SlotInterval { _ => unreachable!("slots only have 5 intervals"), } } + + /// Milliseconds from genesis to the start of this interval in `slot`. + /// + /// Inverse of [`Self::from_ms_since_genesis`]. + pub(crate) fn to_ms_since_genesis(self, slot: u64) -> u64 { + let interval = match self { + Self::BlockPublication => 0, + Self::AttestationProduction => 1, + Self::Aggregation => 2, + Self::SafeTargetUpdate => 3, + Self::EndOfSlot => 4, + }; + slot * MILLISECONDS_PER_SLOT + interval * MILLISECONDS_PER_INTERVAL + } } /// Milliseconds until the next interval boundary, measured relative to genesis. @@ -1367,16 +1381,37 @@ impl Handler for BlockChainServer { impl Handler for BlockChainServer { async fn handle(&mut self, msg: NewBlock, _ctx: &Context) { - self.events.emit(ChainEvent::BlockGossip { - slot: msg.block.message.slot, - block: msg.block.message.hash_tree_root(), - }); + let arrival_ms = unix_now_ms(); + // Gate both the event and the arrival metric on BlockSource::Gossip for + // two reasons: `ChainEvent::BlockGossip` is documented (events.rs) as "a + // block seen on gossip, before import", yet without this gate it also + // fired for req/resp sync blocks; and sync backfill delivers blocks many + // slots after they were due, which would swamp the arrival histogram + // with stale deltas that reflect catch-up speed, not gossip timeliness. + // `self.on_block(msg.block)` still runs for every source below: it is + // the import path and must not be gated. + if msg.source == BlockSource::Gossip { + let slot = msg.block.message.slot; + self.events.emit(ChainEvent::BlockGossip { + slot, + block: msg.block.message.hash_tree_root(), + }); + let genesis_ms = self.store.config().expect("config exists").genesis_time * 1000; + metrics::observe_gossip_block_arrival(arrival_ms, genesis_ms, slot); + } self.on_block(msg.block); } } impl Handler for BlockChainServer { async fn handle(&mut self, msg: NewAttestation, ctx: &Context) { + let arrival_ms = unix_now_ms(); + let genesis_ms = self.store.config().expect("config exists").genesis_time * 1000; + metrics::observe_gossip_attestation_arrival( + arrival_ms, + genesis_ms, + msg.attestation.data.slot, + ); self.on_gossip_attestation(&msg.attestation); // Early aggregation only advances the current slot's group counts, so a // late- or future-slot attestation can never cross the threshold; skip @@ -1390,6 +1425,9 @@ impl Handler for BlockChainServer { impl Handler for BlockChainServer { async fn handle(&mut self, msg: NewAggregatedAttestation, _ctx: &Context) { + let arrival_ms = unix_now_ms(); + let genesis_ms = self.store.config().expect("config exists").genesis_time * 1000; + metrics::observe_gossip_aggregation_arrival(arrival_ms, genesis_ms); self.on_gossip_aggregated_attestation(msg.attestation); } } @@ -1400,6 +1438,8 @@ impl Handler for BlockChainServer { impl Handler for BlockChainServer { async fn handle(&mut self, msg: AggregateProduced, _ctx: &Context) { + let arrival_ms = unix_now_ms(); + // Drop results from a prior session (or from an unexpected late worker). // Current session may be None if the actor already cleaned it up; accept // the message only when ids match. @@ -1413,6 +1453,17 @@ impl Handler for BlockChainServer { return; } + // Count our own aggregate in the same series as gossip-received ones, + // so an aggregator does not report an empty aggregate arrival profile. + // Delivery of this message is held to the interval-2 boundary upstream, + // so a local aggregate lands near zero unless proving overran the + // interval. Sharing one series with received aggregates is deliberate + // and costs little in practice: a late aggregate is late for every node + // at once, so both populations are dominated by production time rather + // than propagation and their distributions look alike. + let genesis_ms = self.store.config().expect("config exists").genesis_time * 1000; + metrics::observe_gossip_aggregation_arrival(arrival_ms, genesis_ms); + // Publish alignment is enforced upstream: the worker delays delivery of // this message until the interval-2 boundary, so by the time it lands // the aggregate is safe to apply and gossip immediately. diff --git a/crates/blockchain/src/metrics.rs b/crates/blockchain/src/metrics.rs index 873605cf..41b70bd7 100644 --- a/crates/blockchain/src/metrics.rs +++ b/crates/blockchain/src/metrics.rs @@ -35,6 +35,39 @@ pub const ATTESTATION_AGGREGATE_COVERAGE_DIFF_DIRECTIONS: &[&str] = &["block_onl pub const BLOCK_PROPOSAL_ATTESTATION_BUILD_PHASES: &[&str] = &["select_payloads", "compact", "stf_simulate"]; +/// Where a gossip message landed relative to the interval it was due in. +/// +/// Kept private to the module: unlike [`SyncStatus`] (which the RPC layer +/// reads for `/lean/v0/node/syncing`), this label is only ever produced and +/// consumed inside the gossip arrival-timing helpers below. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum SlotPosition { + /// Arrived before the interval began. + Before, + /// Arrived within the interval. + Inside, + /// Arrived after the interval ended. + After, +} + +impl SlotPosition { + fn as_str(&self) -> &'static str { + match self { + SlotPosition::Before => "before", + SlotPosition::Inside => "inside", + SlotPosition::After => "after", + } + } + + /// Every position reachable from a signed delta. + const ALL: &[&str] = &["before", "inside", "after"]; + + /// Positions reachable when the anchor cannot follow the arrival, so the + /// delta is never negative. Used by the aggregate counter, which must not + /// export an unreachable `before` series. + const NON_NEGATIVE: &[&str] = &["inside", "after"]; +} + // --- Gauges --- static LEAN_HEAD_SLOT: std::sync::LazyLock = std::sync::LazyLock::new(|| { @@ -507,6 +540,174 @@ static LEAN_BLOCK_PROPOSAL_AGGREGATES_SELECTED: std::sync::LazyLock = .unwrap() }); +// --- Gossip Arrival Timing --- + +/// Bucket boundaries shared by the three gossip arrival-delay histograms, +/// aligned to the interval and slot durations (see +/// [`crate::MILLISECONDS_PER_INTERVAL`], [`crate::MILLISECONDS_PER_SLOT`]). +fn gossip_arrival_delay_buckets() -> Vec { + vec![0.05, 0.1, 0.2, 0.4, 0.8, 1.2, 1.6, 2.4, 4.0, 8.0, 16.0] +} + +static LEAN_GOSSIP_BLOCK_ARRIVAL_DELAY_SECONDS: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + register_histogram!( + "lean_gossip_block_arrival_delay_seconds", + "Absolute delay between a gossip block's arrival and the start of the interval it \ + was due in", + gossip_arrival_delay_buckets() + ) + .unwrap() + }); + +static LEAN_GOSSIP_ATTESTATION_ARRIVAL_DELAY_SECONDS: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + register_histogram!( + "lean_gossip_attestation_arrival_delay_seconds", + "Absolute delay between a gossip attestation's arrival and the start of the interval \ + it was due in", + gossip_arrival_delay_buckets() + ) + .unwrap() + }); + +static LEAN_GOSSIP_AGGREGATION_ARRIVAL_DELAY_SECONDS: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + register_histogram!( + "lean_gossip_aggregation_arrival_delay_seconds", + "Absolute delay between an aggregate becoming available, whether received on gossip \ + or produced locally, and the most recent aggregation-interval boundary at or before \ + it. A locally produced aggregate is held until that boundary, so it normally lands \ + near zero and only registers a delay when proving overran its interval", + gossip_arrival_delay_buckets() + ) + .unwrap() + }); + +static LEAN_GOSSIP_BLOCK_ARRIVAL_TOTAL: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + register_int_counter_vec!( + "lean_gossip_block_arrival_total", + "Gossip blocks by arrival position relative to the interval they were due in", + &["position"] + ) + .unwrap() + }); + +static LEAN_GOSSIP_ATTESTATION_ARRIVAL_TOTAL: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + register_int_counter_vec!( + "lean_gossip_attestation_arrival_total", + "Gossip attestations by arrival position relative to the interval they were due in", + &["position"] + ) + .unwrap() + }); + +static LEAN_GOSSIP_AGGREGATION_ARRIVAL_TOTAL: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + register_int_counter_vec!( + "lean_gossip_aggregation_arrival_total", + "Aggregates, received on gossip or produced locally, by arrival position relative to \ + the most recent aggregation-interval boundary. Anchored to the latest such boundary \ + rather than the aggregate's own data slot, so an arrival can never precede it: only \ + `inside` and `after` occur, never `before`.", + &["position"] + ) + .unwrap() + }); + +/// Signed milliseconds from the start of `interval` in `anchor_slot` to +/// `arrival_ms`. Negative means the message arrived before it was due. +fn interval_delta_ms( + arrival_ms: u64, + genesis_ms: u64, + anchor_slot: u64, + interval: crate::SlotInterval, +) -> i64 { + let expected_ms = genesis_ms + interval.to_ms_since_genesis(anchor_slot); + arrival_ms as i64 - expected_ms as i64 +} + +/// Milliseconds since the most recent `interval` boundary at or before +/// `arrival_ms`. Always in `[0, MILLISECONDS_PER_SLOT)`, so it never reports +/// a negative delta. +fn latest_interval_delta_ms( + arrival_ms: u64, + genesis_ms: u64, + interval: crate::SlotInterval, +) -> i64 { + let since_genesis = arrival_ms.saturating_sub(genesis_ms) as i64; + // Slot 0 makes `to_ms_since_genesis` yield just the offset within a slot. + let anchor_offset = interval.to_ms_since_genesis(0) as i64; + (since_genesis - anchor_offset).rem_euclid(crate::MILLISECONDS_PER_SLOT as i64) +} + +/// Classify a signed delta against the interval width: `inside` is the +/// half-open range from the interval's start up to its end. +fn position_from_delta(delta_ms: i64) -> SlotPosition { + if delta_ms < 0 { + SlotPosition::Before + } else if delta_ms < crate::MILLISECONDS_PER_INTERVAL as i64 { + SlotPosition::Inside + } else { + SlotPosition::After + } +} + +/// Observe a gossip block's arrival against the start of its own slot's +/// [`crate::SlotInterval::BlockPublication`] interval. Zero point: `block_slot`'s +/// slot boundary. +pub fn observe_gossip_block_arrival(arrival_ms: u64, genesis_ms: u64, block_slot: u64) { + let delta_ms = interval_delta_ms( + arrival_ms, + genesis_ms, + block_slot, + crate::SlotInterval::BlockPublication, + ); + LEAN_GOSSIP_BLOCK_ARRIVAL_DELAY_SECONDS + .observe(Duration::from_millis(delta_ms.unsigned_abs()).as_secs_f64()); + LEAN_GOSSIP_BLOCK_ARRIVAL_TOTAL + .with_label_values(&[position_from_delta(delta_ms).as_str()]) + .inc(); +} + +/// Observe a gossip attestation's arrival against its data slot's +/// [`crate::SlotInterval::AttestationProduction`] interval. Zero point: +/// `data_slot`'s interval-1 boundary. +pub fn observe_gossip_attestation_arrival(arrival_ms: u64, genesis_ms: u64, data_slot: u64) { + let delta_ms = interval_delta_ms( + arrival_ms, + genesis_ms, + data_slot, + crate::SlotInterval::AttestationProduction, + ); + LEAN_GOSSIP_ATTESTATION_ARRIVAL_DELAY_SECONDS + .observe(Duration::from_millis(delta_ms.unsigned_abs()).as_secs_f64()); + LEAN_GOSSIP_ATTESTATION_ARRIVAL_TOTAL + .with_label_values(&[position_from_delta(delta_ms).as_str()]) + .inc(); +} + +/// Observe a gossip aggregate's arrival against the most recent +/// [`crate::SlotInterval::Aggregation`] boundary at or before `arrival_ms`. +/// Zero point: that boundary — not the aggregate's own `data.slot`. +/// +/// Deliberately takes no slot argument. An aggregate published at interval 2 +/// of slot N can carry `data.slot < N` (the stale-group catch-up path in +/// `aggregation.rs`), so anchoring to `data.slot` would fill the histogram +/// with large values that are not a health problem. Assuming the latest +/// aggregation interval bounds the value to one slot. +pub fn observe_gossip_aggregation_arrival(arrival_ms: u64, genesis_ms: u64) { + let delta_ms = + latest_interval_delta_ms(arrival_ms, genesis_ms, crate::SlotInterval::Aggregation); + LEAN_GOSSIP_AGGREGATION_ARRIVAL_DELAY_SECONDS + .observe(Duration::from_millis(delta_ms.unsigned_abs()).as_secs_f64()); + LEAN_GOSSIP_AGGREGATION_ARRIVAL_TOTAL + .with_label_values(&[position_from_delta(delta_ms).as_str()]) + .inc(); +} + // --- Sync Status --- /// Node synchronization status. @@ -649,6 +850,24 @@ pub fn init() { std::sync::LazyLock::force(&LEAN_BLOCK_PROPOSAL_CHILD_PAYLOADS_CONSUMED_TOTAL); std::sync::LazyLock::force(&LEAN_BLOCK_PROPOSAL_ATTESTATION_DATA_SELECTED); std::sync::LazyLock::force(&LEAN_BLOCK_PROPOSAL_AGGREGATES_SELECTED); + // Gossip arrival timing + std::sync::LazyLock::force(&LEAN_GOSSIP_BLOCK_ARRIVAL_DELAY_SECONDS); + std::sync::LazyLock::force(&LEAN_GOSSIP_ATTESTATION_ARRIVAL_DELAY_SECONDS); + std::sync::LazyLock::force(&LEAN_GOSSIP_AGGREGATION_ARRIVAL_DELAY_SECONDS); + std::sync::LazyLock::force(&LEAN_GOSSIP_BLOCK_ARRIVAL_TOTAL); + std::sync::LazyLock::force(&LEAN_GOSSIP_ATTESTATION_ARRIVAL_TOTAL); + std::sync::LazyLock::force(&LEAN_GOSSIP_AGGREGATION_ARRIVAL_TOTAL); + // Seed every reachable position so the series exist at zero from startup, + // mirroring AGGREGATOR_SKIP_REASONS above. The aggregate counter seeds + // only the non-negative positions: its anchor never follows the arrival, + // so `before` is unreachable and must not appear as a zero series. + for &position in SlotPosition::ALL { + LEAN_GOSSIP_BLOCK_ARRIVAL_TOTAL.with_label_values(&[position]); + LEAN_GOSSIP_ATTESTATION_ARRIVAL_TOTAL.with_label_values(&[position]); + } + for &position in SlotPosition::NON_NEGATIVE { + LEAN_GOSSIP_AGGREGATION_ARRIVAL_TOTAL.with_label_values(&[position]); + } // Sync status std::sync::LazyLock::force(&LEAN_NODE_SYNC_STATUS); // Aggregator skip counter: instantiate every cross-client reason so the diff --git a/crates/net/api/src/lib.rs b/crates/net/api/src/lib.rs index 9cdbcdd0..d6ec647d 100644 --- a/crates/net/api/src/lib.rs +++ b/crates/net/api/src/lib.rs @@ -20,11 +20,24 @@ pub trait BlockChainToP2P: Send + Sync { fn fetch_block(&self, root: H256) -> Result<(), ActorError>; } +/// How a block reached this node. +/// +/// Distinguishes blocks announced on gossip from blocks pulled by req/resp +/// during sync: the two have very different arrival-time characteristics, so +/// consumers that care about timeliness must be able to tell them apart. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum BlockSource { + /// Received on the block gossip topic. + Gossip, + /// Fetched via req/resp (`BlocksByRoot` / `BlocksByRange`). + Sync, +} + // --- Protocol: P2P -> BlockChain --- #[protocol] pub trait P2PToBlockChain: Send + Sync { - fn new_block(&self, block: SignedBlock) -> Result<(), ActorError>; + fn new_block(&self, block: SignedBlock, source: BlockSource) -> Result<(), ActorError>; fn new_attestation(&self, attestation: SignedAttestation) -> Result<(), ActorError>; fn new_aggregated_attestation( &self, diff --git a/crates/net/p2p/src/gossipsub/handler.rs b/crates/net/p2p/src/gossipsub/handler.rs index c257006b..95ac7081 100644 --- a/crates/net/p2p/src/gossipsub/handler.rs +++ b/crates/net/p2p/src/gossipsub/handler.rs @@ -1,3 +1,4 @@ +use ethlambda_network_api::BlockSource; use ethlambda_types::{ ShortRoot, attestation::{SignedAggregatedAttestation, SignedAttestation}, @@ -59,7 +60,7 @@ pub async fn handle_gossipsub_message(server: &mut P2PServer, event: Event) { ); if let Some(ref blockchain) = server.blockchain { let _ = blockchain - .new_block(signed_block) + .new_block(signed_block, BlockSource::Gossip) .inspect_err(|err| error!(%err, "Failed to forward block to blockchain")); } } diff --git a/crates/net/p2p/src/req_resp/handlers.rs b/crates/net/p2p/src/req_resp/handlers.rs index 810e658b..0bc6c204 100644 --- a/crates/net/p2p/src/req_resp/handlers.rs +++ b/crates/net/p2p/src/req_resp/handlers.rs @@ -1,5 +1,6 @@ use std::collections::HashSet; +use ethlambda_network_api::BlockSource; use ethlambda_storage::Store; use libp2p::{PeerId, request_response}; use rand::seq::SliceRandom; @@ -321,7 +322,7 @@ async fn handle_blocks_by_root_response( if let Some(ref blockchain) = server.blockchain { let _ = blockchain - .new_block(block) + .new_block(block, BlockSource::Sync) .inspect_err(|err| error!(%err, "Failed to forward fetched block to blockchain")); } } @@ -357,7 +358,7 @@ async fn handle_blocks_by_range_response( } let block_root = block.message.hash_tree_root(); - if let Err(err) = blockchain.new_block(block) { + if let Err(err) = blockchain.new_block(block, BlockSource::Sync) { error!( %err, %slot, %peer, block_root = %ethlambda_types::ShortRoot(&block_root.0), diff --git a/docs/metrics.md b/docs/metrics.md index 93e8de33..017dc9ea 100644 --- a/docs/metrics.md +++ b/docs/metrics.md @@ -120,6 +120,27 @@ The metrics below are not part of the [leanMetrics specification](https://github | `lean_reqresp_request_size_bytes` | Histogram | Bytes size of a req/resp request (raw SSZ or snappy on-wire) | On req/resp request send/receive | protocol=status,blocks_by_root
compression=raw,snappy | 64, 128, 256, 512, 1024, 4096, 16384, 65536 | | `lean_reqresp_response_chunk_size_bytes` | Histogram | Bytes size of a single req/resp response chunk (raw SSZ or snappy on-wire) | On req/resp response chunk send/receive | protocol=status,blocks_by_root
compression=raw,snappy | 128, 1024, 10000, 100000, 500000, 1000000, 5000000, 10000000 | +### Gossip Arrival Timing + +These histograms record the absolute distance between a gossip message's arrival and the start of the interval it was due in, so an arrival that is early by some amount and one that is late by the same amount land in the same bucket; the counters' `position` label is what tells them apart. `inside` means the message arrived within the interval it was due in, not merely somewhere in the right slot: an attestation for slot 10 that lands during slot 10's interval 2 is `after`, not `inside`, since it missed the AttestationProduction interval it was actually due in. + +Blocks anchor to interval 0 of their own slot and attestations to interval 1 of their data slot; both are unbounded above, so a message that never arrives close to real time can be arbitrarily late. Aggregates anchor instead to the most recent aggregation-interval boundary rather than their own data slot, since a stale-group catch-up aggregate can carry a `data.slot` several slots in the past; anchoring to the latest boundary bounds the delay to one slot and rules out `before` entirely. + +Only gossip-received blocks are sampled here: blocks fetched via req/resp during sync are excluded, since sync backfill delivers blocks long after they were due and would swamp these histograms with catch-up noise rather than gossip-health signal. + +The aggregate metrics do include an aggregator's own freshly produced aggregates, which never come back over gossip; without them an aggregator would report an empty aggregate profile. The two populations are not quite the same measurement: delivery of a locally produced aggregate is held until the interval-2 boundary, so it lands near zero unless proving overran the interval, whereas a received one adds propagation on top of whenever the producer managed to publish it. + +In practice the distribution is bimodal and dominated by production rather than propagation: a mode in the lowest bucket for aggregates that made their interval, plus a tail for those whose proving overran it. A late aggregate is late for every node at once, so that tail shows up on receivers too and is not evidence of a slow network. Read a rising tail as aggregation cost, and cross-check `lean_pq_sig_aggregated_signatures_building_time_seconds` and `lean_committee_signatures_aggregation_time_seconds` to confirm. + +| Name | Type | Usage | Sample collection event | Labels | Buckets | +|------|------|-------|-------------------------|--------|---------| +| `lean_gossip_block_arrival_delay_seconds` | Histogram | Absolute delay between a gossip block's arrival and the start of the interval it was due in | On gossip block receipt, before import | | 0.05, 0.1, 0.2, 0.4, 0.8, 1.2, 1.6, 2.4, 4, 8, 16 | +| `lean_gossip_attestation_arrival_delay_seconds` | Histogram | Absolute delay between a gossip attestation's arrival and the start of the interval it was due in | On gossip attestation receipt | | 0.05, 0.1, 0.2, 0.4, 0.8, 1.2, 1.6, 2.4, 4, 8, 16 | +| `lean_gossip_aggregation_arrival_delay_seconds` | Histogram | Absolute delay between an aggregate becoming available (gossip receipt, or local production) and the most recent aggregation-interval boundary at or before it | On gossip aggregated-attestation receipt, or on local aggregate production | | 0.05, 0.1, 0.2, 0.4, 0.8, 1.2, 1.6, 2.4, 4, 8, 16 | +| `lean_gossip_block_arrival_total` | Counter | Gossip blocks by arrival position relative to the interval they were due in | On gossip block receipt, before import | position=before,inside,after | | +| `lean_gossip_attestation_arrival_total` | Counter | Gossip attestations by arrival position relative to the interval they were due in | On gossip attestation receipt | position=before,inside,after | | +| `lean_gossip_aggregation_arrival_total` | Counter | Aggregates by arrival position relative to the most recent aggregation-interval boundary | On gossip aggregated-attestation receipt, or on local aggregate production | position=inside,after | | + ### Storage | Name | Type | Usage | Sample collection event | Labels |