diff --git a/bin/ethlambda/src/main.rs b/bin/ethlambda/src/main.rs index 7909519a..0ca6d077 100644 --- a/bin/ethlambda/src/main.rs +++ b/bin/ethlambda/src/main.rs @@ -37,7 +37,9 @@ use ethlambda_blockchain::MILLISECONDS_PER_SLOT; use ethlambda_blockchain::block_builder::ProposerConfig; use ethlambda_blockchain::key_manager::ValidatorKeyPair; use ethlambda_network_api::{InitBlockChain, InitP2P, ToBlockChainToP2PRef, ToP2PToBlockChainRef}; -use ethlambda_p2p::{Bootnode, P2P, PeerId, SwarmConfig, build_swarm, parse_enrs}; +use ethlambda_p2p::{ + Bootnode, P2P, PeerId, SwarmConfig, attestation_subscription_subnets, build_swarm, parse_enrs, +}; use ethlambda_types::primitives::{H256, HashTreeRoot as _}; use ethlambda_types::{ aggregator::AggregatorController, @@ -50,7 +52,7 @@ use serde::Deserialize; use tracing::{error, info, warn}; use tracing_subscriber::{EnvFilter, Layer, Registry, layer::SubscriberExt}; -use ethlambda_blockchain::{BlockChain, SyncStatusController}; +use ethlambda_blockchain::{BlockChain, BlockChainConfig, SyncStatusController}; use ethlambda_rpc::RpcConfig; use ethlambda_storage::{ MAX_RESUMABLE_DB_STATE_AGE, StorageBackend, Store, backend::RocksDBBackend, @@ -214,38 +216,46 @@ async fn main() -> eyre::Result<()> { // and the API server (which exposes GET/POST admin endpoints). let aggregator = AggregatorController::new(options.is_aggregator); + // Attestation subnets this node subscribes to, computed once and shared by + // the P2P swarm (to open gossip subscriptions) and the blockchain actor + // (to size the early-aggregation threshold), so both agree on which subnets + // feed this node's gossip groups. Subscriptions are fixed at startup and + // are not re-evaluated when the aggregator role is toggled at runtime; see + // the hot-standby note on SwarmConfig. + let subscribed_subnets = attestation_subscription_subnets( + &validator_ids, + attestation_committee_count, + options.is_aggregator, + options.aggregate_subnet_ids.as_deref(), + ); + // Shared, runtime-readable sync status. The blockchain actor writes it each // tick (alongside the `lean_node_sync_status` metric); the RPC // `/lean/v0/node/syncing` endpoint reads it. Seeded to Idle, matching the // metric's startup value. let sync_status = SyncStatusController::default(); - let blockchain = BlockChain::spawn( - store.clone(), - validator_keys, - aggregator.clone(), - sync_status.clone(), + let blockchain_config = BlockChainConfig { + aggregator: aggregator.clone(), + sync_status_controller: sync_status.clone(), attestation_committee_count, - !options.disable_duty_sync_gate, - ProposerConfig { + gate_duties: !options.disable_duty_sync_gate, + subscribed_subnets: subscribed_subnets.clone(), + proposer_config: ProposerConfig { enable_proposer_aggregation: options.enable_proposer_aggregation, max_attestations_per_block: options.max_attestations_per_block, }, - ); + }; + + let blockchain = BlockChain::spawn(store.clone(), validator_keys, blockchain_config); - // Note: SwarmConfig.is_aggregator is intentionally a plain bool, not the - // AggregatorController — subnet subscriptions are decided once here and - // are not re-evaluated at runtime. Toggling via the admin API affects - // aggregation logic but not the gossip mesh. See crates/net/p2p/src/lib.rs - // for the invariant. let built = build_swarm(SwarmConfig { node_key: node_p2p_key, bootnodes, listening_socket: p2p_socket, validator_ids, attestation_committee_count, - is_aggregator: options.is_aggregator, - aggregate_subnet_ids: options.aggregate_subnet_ids, + subscription_subnets: subscribed_subnets, }) .wrap_err("failed to build swarm")?; diff --git a/crates/blockchain/src/aggregation.rs b/crates/blockchain/src/aggregation.rs index f8387fcb..6155946e 100644 --- a/crates/blockchain/src/aggregation.rs +++ b/crates/blockchain/src/aggregation.rs @@ -1,13 +1,15 @@ //! Committee-signature aggregation: off-thread worker orchestration and the //! pure functions it runs. //! -//! The blockchain actor fires one aggregation session per interval 2 via +//! The blockchain actor fires one aggregation session per slot — at interval 2, +//! or up to [`EARLY_AGGREGATION_WINDOW`] early when the 2/3 signature +//! threshold is met — via //! [`run_aggregation_worker`]. The actor stays on its message loop; the worker //! runs the expensive XMSS proofs on a `spawn_blocking` thread and streams //! results back as [`AggregateProduced`] / [`AggregationDone`] messages. use std::collections::HashSet; -use std::time::{Duration, Instant}; +use std::time::{Duration, Instant, SystemTime}; use ethlambda_crypto::aggregate_mixed; use ethlambda_storage::Store; @@ -19,23 +21,40 @@ use ethlambda_types::{ state::Validator, }; use spawned_concurrency::message::Message; -use spawned_concurrency::tasks::ActorRef; +use spawned_concurrency::tasks::{ActorRef, Context, send_after}; use tokio_util::sync::CancellationToken; use tracing::{info, warn}; -use crate::metrics; +use crate::{MILLISECONDS_PER_INTERVAL, metrics}; -/// Soft deadline for committee-signature aggregation measured from the -/// interval-2 tick. After this much wall time elapses, the actor signals the -/// worker to stop via its cancellation token. The 50 ms budget before the next -/// interval (interval 3 at +800 ms) is reserved for publishing any late-arriving -/// aggregates and for gossip propagation margin. -pub(crate) const AGGREGATION_DEADLINE: Duration = Duration::from_millis(750); +/// Soft deadline for committee-signature aggregation measured from session +/// start. After this much wall time elapses, the actor signals the worker to +/// stop via its cancellation token. A session started exactly at interval 2 +/// gets the full interval (interval 3 is one interval later); a session +/// started early (see `maybe_start_early_aggregation`) ends correspondingly +/// earlier. The deadline only stops new jobs from starting — a job mid-proof +/// finishes and publishes right after. +pub(crate) const AGGREGATION_DEADLINE: Duration = Duration::from_millis(800); /// Upper bound we wait for a prior worker to exit if it is still running when /// the next session is about to start. Reached only in pathological cases /// (mismatched timers, stuck proofs); we warn before blocking. pub(crate) const PRIOR_WORKER_JOIN_TIMEOUT: Duration = Duration::from_secs(2); +/// Width of the early-aggregation window: a session may start at most this +/// long before the interval-2 boundary, provided the signature threshold is +/// met (see the check in `maybe_start_early_aggregation`). +pub(crate) const EARLY_AGGREGATION_WINDOW: Duration = Duration::from_millis(600); + +// The window must fit within one interval: `maybe_start_early_aggregation` +// subtracts it from the interval-2 offset, and the interval-1 tick schedules +// the check at `MILLISECONDS_PER_INTERVAL - EARLY_AGGREGATION_WINDOW`. Keep +// this invariant self-enforcing so a future bump to the window can't silently +// underflow either subtraction. +const _: () = assert!( + EARLY_AGGREGATION_WINDOW.as_millis() <= MILLISECONDS_PER_INTERVAL as u128, + "EARLY_AGGREGATION_WINDOW must not exceed one interval" +); + /// A single pre-prepared aggregation group. /// /// Built on the actor thread from a store snapshot; consumed by an off-thread @@ -76,6 +95,9 @@ pub(crate) struct AggregationSession { /// Slot at which this session was started; used as a fencing id so we can /// drop late-arriving messages from a prior session. pub(crate) session_id: u64, + /// Whether the session started before the slot's interval-2 boundary via + /// the early-aggregation trigger. + pub(crate) early: bool, /// Child of the actor cancellation token; fires either at the deadline or /// when the actor itself is stopping. pub(crate) cancel: CancellationToken, @@ -107,7 +129,7 @@ impl Message for AggregationDone { type Result = (); } -/// Self-message scheduled via `send_after` at interval-2 start. Cancels the +/// Self-message scheduled via `send_after` at session start. Cancels the /// session's token so the worker stops starting new aggregations. pub(crate) struct AggregationDeadline { pub(crate) session_id: u64, @@ -116,6 +138,15 @@ impl Message for AggregationDeadline { type Result = (); } +/// One-shot self-message scheduled at the interval-1 tick; fires when the +/// early-aggregation window opens (T2 - EARLY_AGGREGATION_WINDOW) to run +/// the threshold check for signatures that all arrived before the window. +/// Arrivals inside the window are checked per insert instead. +pub(crate) struct EarlyAggregationCheck; +impl Message for EarlyAggregationCheck { + type Result = (); +} + /// Build a snapshot of everything needed to aggregate. Runs on the actor /// thread, touches the store, does no heavy cryptography. Returns `None` when /// there is nothing to aggregate so callers can avoid spawning an empty worker. @@ -501,11 +532,19 @@ pub(crate) fn aggregation_bits_from_validator_indices(bits: &[u64]) -> Aggregati /// Pulls jobs from the snapshot, runs [`aggregate_job`] for each, and streams /// successful aggregates back to the actor as [`AggregateProduced`] messages. /// Emits [`AggregationDone`] when the loop exits (completion or cancellation). +/// +/// Publish alignment: aggregates must not reach the actor (and thus gossip) +/// before the interval-2 boundary. `publish_at` is that boundary as a wall-clock +/// instant; a produced aggregate still ahead of it is delivered via +/// [`send_after`] timed to land at the boundary, otherwise it is sent +/// immediately. A normal interval-2 session starts at the boundary, so its +/// aggregates are always past it and sent without delay. pub(crate) fn run_aggregation_worker( snapshot: AggregationSnapshot, actor: ActorRef, cancel: CancellationToken, session_id: u64, + publish_at: SystemTime, ) { let start = Instant::now(); let groups_considered = snapshot.groups_considered; @@ -553,12 +592,29 @@ pub(crate) fn run_aggregation_worker( total_raw_sigs += raw_sigs; total_children += children; - if actor - .send(AggregateProduced { session_id, output }) - .is_err() - { - // Actor is gone; no point producing more. - break; + // Hold the aggregate until the interval-2 boundary (early session), or + // send now if already at/past it. `send_after` is fire-and-forget: it + // spawns a timer that delivers the message and is cancelled only if the + // actor stops, so the produced aggregate is not lost when the worker's + // own loop ends. `duration_since` errs once the boundary has passed, + // which collapses to a zero delay here. + let delay = publish_at + .duration_since(SystemTime::now()) + .unwrap_or(Duration::ZERO); + if delay.is_zero() { + if actor + .send(AggregateProduced { session_id, output }) + .is_err() + { + // Actor is gone; no point producing more. + break; + } + } else { + send_after( + delay, + Context::from_ref(&actor), + AggregateProduced { session_id, output }, + ); } } diff --git a/crates/blockchain/src/lib.rs b/crates/blockchain/src/lib.rs index d83271a8..675933f1 100644 --- a/crates/blockchain/src/lib.rs +++ b/crates/blockchain/src/lib.rs @@ -15,7 +15,8 @@ use ethlambda_types::{ use crate::aggregation::{ AGGREGATION_DEADLINE, AggregateProduced, AggregationDeadline, AggregationDone, - AggregationSession, PRIOR_WORKER_JOIN_TIMEOUT, run_aggregation_worker, + AggregationSession, EARLY_AGGREGATION_WINDOW, EarlyAggregationCheck, PRIOR_WORKER_JOIN_TIMEOUT, + run_aggregation_worker, }; use crate::key_manager::ValidatorKeyPair; use crate::sync_status::SyncStatusTracker; @@ -43,6 +44,25 @@ pub struct BlockChain { handle: ActorRef, } +/// Startup configuration for the [`BlockChain`] actor: the distinct +/// dependencies wired in once at spawn, grouped to keep the constructor's +/// signature small. +pub struct BlockChainConfig { + /// Committee-aggregator role, toggleable at runtime via the admin API. + pub aggregator: AggregatorController, + /// Runtime-readable sync status: written by the actor each tick and read + /// by the RPC `/lean/v0/node/syncing` endpoint. + pub sync_status_controller: SyncStatusController, + /// Number of attestation committees (= subnet count). + pub attestation_committee_count: u64, + /// Whether the sync-gate suppresses validator duties (vs observe-only). + pub gate_duties: bool, + /// Attestation subnets this node subscribes to. + pub subscribed_subnets: HashSet, + /// Proposer-side block-building policy. + pub proposer_config: ProposerConfig, +} + /// Milliseconds per interval (800ms ticks). pub const MILLISECONDS_PER_INTERVAL: u64 = 800; /// Number of intervals per slot (5 intervals of 800ms = 4 seconds). @@ -107,12 +127,17 @@ impl BlockChain { pub fn spawn( store: Store, validator_keys: HashMap, - aggregator: AggregatorController, - sync_status_controller: SyncStatusController, - attestation_committee_count: u64, - gate_duties: bool, - proposer_config: ProposerConfig, + config: BlockChainConfig, ) -> BlockChain { + let BlockChainConfig { + aggregator, + sync_status_controller, + attestation_committee_count, + gate_duties, + subscribed_subnets, + proposer_config, + } = config; + metrics::set_is_aggregator(aggregator.is_enabled()); metrics::set_node_sync_status(metrics::SyncStatus::Idle); let genesis_time = store.config().genesis_time; @@ -136,6 +161,7 @@ impl BlockChain { current_aggregation: None, last_tick_instant: None, attestation_committee_count, + subscribed_subnets, proposer_config, pre_merge_coverage: None, sync_status: SyncStatusTracker::new(gate_duties), @@ -186,18 +212,27 @@ pub struct BlockChainServer { /// `--is-aggregator` flag at spawn. aggregator: AggregatorController, - /// In-flight committee-signature aggregation, if any. Present only while a - /// worker started at the most recent interval 2 is still running or until - /// the next interval 2 takes over. + /// The slot's one committee-signature aggregation session (started at + /// interval 2, or early via the 2/3 trigger). Deliberately persists after + /// the worker finishes — that persistence is the once-per-slot latch the + /// early trigger and the interval-2 skip both check — until the next + /// session start replaces it. current_aggregation: Option, /// Last tick instant for measuring interval duration. last_tick_instant: Option, /// Number of attestation committees (= subnet count). Used by the - /// attestation aggregate coverage emission. + /// attestation aggregate coverage emission and the early-aggregation + /// threshold. attestation_committee_count: u64, + /// Attestation subnets this node subscribes to (its validators' own + /// subnets plus any aggregator-only subnets), computed once at startup and + /// shared with the P2P swarm via [`ethlambda_p2p::attestation_subscription_subnets`]. + /// Used to scale the early-aggregation threshold. + subscribed_subnets: HashSet, + /// Proposer-side block-building policy proposer_config: ProposerConfig, @@ -340,16 +375,32 @@ impl BlockChainServer { } else if !self.key_manager.validator_ids().is_empty() { info!(%slot, "Skipping attestations while syncing"); } + + // Schedule the early-aggregation window check. This tick is + // one interval before T2, so the timer fires right as the + // window opens at T2 - EARLY_AGGREGATION_WINDOW. + if is_aggregator { + send_after( + Duration::from_millis(MILLISECONDS_PER_INTERVAL) - EARLY_AGGREGATION_WINDOW, + ctx.clone(), + EarlyAggregationCheck, + ); + } } // ==== interval 2 ==== SlotInterval::Aggregation => { if is_aggregator { - coverage::emit_agg_start_new_coverage( - &self.store, - self.attestation_committee_count, - ); - self.start_aggregation_session(slot, ctx).await; + // The early trigger may have already started this slot's + // session (running or finished) — it IS the slot's session, + // so don't start a second one. + let already_started = self + .current_aggregation + .as_ref() + .is_some_and(|session| session.session_id == slot); + if !already_started { + self.start_aggregation_session(slot, ctx).await; + } } else { metrics::inc_aggregator_skipped_not_aggregator(); } @@ -394,7 +445,7 @@ impl BlockChainServer { /// 1. If a prior session is still running (pathological), warn and join it. /// 2. Snapshot the aggregation inputs from the store. /// 3. Spawn a `spawn_blocking` worker that streams results back as messages. - /// 4. Schedule the `AggregationDeadline` self-message at +750 ms. + /// 4. Schedule the `AggregationDeadline` self-message at +`AGGREGATION_DEADLINE`. async fn start_aggregation_session(&mut self, slot: u64, ctx: &Context) { if let Some(prior) = self.current_aggregation.take() { prior.cancel.cancel(); @@ -415,6 +466,8 @@ impl BlockChainServer { } } + coverage::emit_agg_start_new_coverage(&self.store, self.attestation_committee_count); + let Some(snapshot) = aggregation::snapshot_current_slot_aggregation_inputs(&self.store, slot) else { @@ -423,6 +476,25 @@ impl BlockChainServer { }; let session_id = slot; + let genesis_time_ms = self.store.config().genesis_time * 1000; + let t2_ms = genesis_time_ms + slot * MILLISECONDS_PER_SLOT + 2 * MILLISECONDS_PER_INTERVAL; + // Interval-2 boundary as a wall-clock instant; the worker holds each + // produced aggregate until this before sending it back, so nothing + // reaches gossip early. + let publish_at = SystemTime::UNIX_EPOCH + Duration::from_millis(t2_ms); + let now_ms = unix_now_ms(); + let early = now_ms < t2_ms; + if early { + let lead = Duration::from_millis(t2_ms - now_ms); + metrics::inc_aggregation_early_starts(); + metrics::observe_aggregation_early_start_lead(lead); + info!( + %slot, + lead_ms = lead.as_millis() as u64, + "Starting aggregation session early" + ); + } + // Independent token per session. Shutdown propagates via our // #[stopped] hook which cancels any current session; the deadline // timer cancels this specific session at +AGGREGATION_DEADLINE. @@ -432,7 +504,13 @@ impl BlockChainServer { let worker_cancel = cancel.clone(); let worker_actor = actor_ref.clone(); let worker = tokio::task::spawn_blocking(move || { - run_aggregation_worker(snapshot, worker_actor, worker_cancel, session_id); + run_aggregation_worker( + snapshot, + worker_actor, + worker_cancel, + session_id, + publish_at, + ); }); let _deadline_timer = send_after( @@ -443,11 +521,86 @@ impl BlockChainServer { self.current_aggregation = Some(AggregationSession { session_id, + early, cancel, worker, }); } + /// Early-aggregation trigger: start the slot's session ahead of the + /// interval-2 tick when, inside the window `[T2 - EARLY_AGGREGATION_WINDOW, T2)`, + /// a single attestation-data group already holds 2/3 of the signatures + /// expected from this node's aggregation subnets. Called after every + /// stored current-slot gossip signature and once at the window opening via + /// [`EarlyAggregationCheck`]. Fires at most once per slot: the started + /// session stays in `current_aggregation` (running or finished) until the + /// next session replaces it. The latch has one hole: if the snapshot + /// yields no jobs (possible only when no signer's pubkey resolves, i.e. a + /// corrupted validator registry), no session is installed and the check + /// retries on later inserts — each retry is a no-op session attempt. + async fn maybe_start_early_aggregation(&mut self, ctx: &Context) { + if !self.aggregator.is_enabled() { + return; + } + // Only fire inside the early-aggregation window + // `[T2 - EARLY_AGGREGATION_WINDOW, T2)`, where T2 is the current + // slot's interval-2 boundary; the slot is derived from the wall clock. + let genesis_time_ms = self.store.config().genesis_time * 1000; + let Some(ms_since_genesis) = unix_now_ms().checked_sub(genesis_time_ms) else { + return; + }; + let ms_into_slot = ms_since_genesis % MILLISECONDS_PER_SLOT; + let t2_offset = 2 * MILLISECONDS_PER_INTERVAL; + let window_ms = EARLY_AGGREGATION_WINDOW.as_millis() as u64; + if ms_into_slot < t2_offset - window_ms || ms_into_slot >= t2_offset { + return; + } + let slot = ms_since_genesis / MILLISECONDS_PER_SLOT; + if self + .current_aggregation + .as_ref() + .is_some_and(|session| session.session_id == slot) + { + return; + } + let max_group = self.store.max_gossip_group_count_for_slot(slot); + // Trigger once the largest current-slot group holds two-thirds of the + // votes we expect it to collect, rounded up. Groups are keyed by + // attestation data (not by subnet), so one group gathers signatures + // from every subnet we subscribe to; the expected count is therefore + // the number of network validators whose committee subnet is one of + // ours, not a single committee's worth. With `N` validators across `C` + // committees, subnet `s` holds `N / C` validators, plus one more when + // `s < N % C`. (0 only when there are no such validators, which never + // triggers.) + let min_group_sigs = if self.attestation_committee_count == 0 { + 0 + } else { + let validator_count = self.store.head_state().validators.len() as u64; + let committee_count = self.attestation_committee_count; + let expected_votes: u64 = self + .subscribed_subnets + .iter() + .filter(|&&subnet| subnet < committee_count) + .map(|&subnet| { + validator_count / committee_count + + u64::from(subnet < validator_count % committee_count) + }) + .sum(); + (2 * expected_votes).div_ceil(3) as usize + }; + if min_group_sigs == 0 || max_group < min_group_sigs { + return; + } + info!( + %slot, + max_group, + min_group_sigs, + "Early-aggregation threshold met" + ); + self.start_aggregation_session(slot, ctx).await; + } + /// Returns the validator ID if any of our validators is the proposer for this slot. fn get_our_proposer(&self, slot: u64) -> Option { let head_state = self.store.head_state(); @@ -1065,8 +1218,15 @@ impl Handler for BlockChainServer { } impl Handler for BlockChainServer { - async fn handle(&mut self, msg: NewAttestation, _ctx: &Context) { + async fn handle(&mut self, msg: NewAttestation, ctx: &Context) { 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 + // the check unless this attestation is for the store's current slot. + let current_slot = self.store.time() / INTERVALS_PER_SLOT; + if msg.attestation.data.slot == current_slot { + self.maybe_start_early_aggregation(ctx).await; + } } } @@ -1095,6 +1255,9 @@ impl Handler for BlockChainServer { return; } + // 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. aggregation::apply_aggregated_group(&mut self.store, &msg.output); if let Some(ref p2p) = self.p2p { @@ -1109,12 +1272,22 @@ impl Handler for BlockChainServer { } } +impl Handler for BlockChainServer { + async fn handle(&mut self, _msg: EarlyAggregationCheck, ctx: &Context) { + self.maybe_start_early_aggregation(ctx).await; + } +} + impl Handler for BlockChainServer { async fn handle(&mut self, msg: AggregationDone, _ctx: &Context) { aggregation::finalize_aggregation_session(&self.store); metrics::observe_committee_signatures_aggregation(msg.total_elapsed); let aggregation_elapsed = msg.total_elapsed; + let early = self + .current_aggregation + .as_ref() + .is_some_and(|s| s.session_id == msg.session_id && s.early); info!( ?aggregation_elapsed, session_id = msg.session_id, @@ -1123,6 +1296,7 @@ impl Handler for BlockChainServer { total_raw_sigs = msg.total_raw_sigs, total_children = msg.total_children, cancelled = msg.cancelled, + early, aggregation_deadline_ms = AGGREGATION_DEADLINE.as_millis() as u64, "Committee signatures aggregated" ); diff --git a/crates/blockchain/src/metrics.rs b/crates/blockchain/src/metrics.rs index dce62656..7c5cd0fc 100644 --- a/crates/blockchain/src/metrics.rs +++ b/crates/blockchain/src/metrics.rs @@ -263,6 +263,15 @@ static LEAN_PQ_SIG_ATTESTATION_SIGNATURES_INVALID_TOTAL: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + register_int_counter!( + "lean_aggregation_early_starts_total", + "Aggregation sessions started before the interval-2 boundary" + ) + .unwrap() + }); + // --- Histograms --- static LEAN_FORK_CHOICE_BLOCK_PROCESSING_TIME_SECONDS: std::sync::LazyLock = @@ -374,6 +383,16 @@ static LEAN_FORK_CHOICE_REORG_DEPTH: std::sync::LazyLock = .unwrap() }); +static LEAN_AGGREGATION_EARLY_START_LEAD_SECONDS: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + register_histogram!( + "lean_aggregation_early_start_lead_seconds", + "How far before the interval-2 boundary an early aggregation session started", + vec![0.075, 0.15, 0.225, 0.3, 0.375, 0.45, 0.525, 0.6] + ) + .unwrap() + }); + static LEAN_TICK_INTERVAL_DURATION_SECONDS: std::sync::LazyLock = std::sync::LazyLock::new(|| { register_histogram!( @@ -604,6 +623,7 @@ pub fn init() { std::sync::LazyLock::force(&LEAN_PQ_SIG_ATTESTATION_SIGNATURES_TOTAL); std::sync::LazyLock::force(&LEAN_PQ_SIG_ATTESTATION_SIGNATURES_VALID_TOTAL); std::sync::LazyLock::force(&LEAN_PQ_SIG_ATTESTATION_SIGNATURES_INVALID_TOTAL); + std::sync::LazyLock::force(&LEAN_AGGREGATION_EARLY_STARTS_TOTAL); // Histograms std::sync::LazyLock::force(&LEAN_FORK_CHOICE_BLOCK_PROCESSING_TIME_SECONDS); std::sync::LazyLock::force(&LEAN_ATTESTATION_VALIDATION_TIME_SECONDS); @@ -615,6 +635,7 @@ pub fn init() { std::sync::LazyLock::force(&LEAN_COMMITTEE_SIGNATURES_AGGREGATION_TIME_SECONDS); std::sync::LazyLock::force(&LEAN_AGGREGATED_PROOF_SIZE_BYTES); std::sync::LazyLock::force(&LEAN_FORK_CHOICE_REORG_DEPTH); + std::sync::LazyLock::force(&LEAN_AGGREGATION_EARLY_START_LEAD_SECONDS); std::sync::LazyLock::force(&LEAN_TICK_INTERVAL_DURATION_SECONDS); // Block production std::sync::LazyLock::force(&LEAN_BLOCK_AGGREGATED_PAYLOADS); @@ -641,6 +662,14 @@ pub fn init() { // --- Public API --- +pub fn inc_aggregation_early_starts() { + LEAN_AGGREGATION_EARLY_STARTS_TOTAL.inc(); +} + +pub fn observe_aggregation_early_start_lead(lead: Duration) { + LEAN_AGGREGATION_EARLY_START_LEAD_SECONDS.observe(lead.as_secs_f64()); +} + pub fn update_head_slot(slot: u64) { LEAN_HEAD_SLOT.set(slot.try_into().unwrap()); } diff --git a/crates/net/p2p/src/lib.rs b/crates/net/p2p/src/lib.rs index ba0318d3..4726ce25 100644 --- a/crates/net/p2p/src/lib.rs +++ b/crates/net/p2p/src/lib.rs @@ -154,22 +154,53 @@ pub(crate) struct Behaviour { /// Configuration for building the libp2p swarm. /// -/// INVARIANT: `is_aggregator` is consumed once during [`build_swarm`] to decide -/// subnet subscriptions and is NOT stored on [`P2PServer`]. Runtime toggles -/// of the aggregator role via the admin API (see -/// [`ethlambda_types::aggregator::AggregatorController`]) intentionally do -/// not resubscribe gossip subnets — this is the leanSpec PR #636 scope -/// limitation ("hot-standby model"). If a runtime reader is ever added on -/// the P2P side, it must consult the shared `AggregatorController` instead -/// of a bool captured here, or the runtime toggle will silently diverge. +/// INVARIANT: `subscription_subnets` is the fixed set of attestation subnets +/// this node subscribes to. It is computed once by the caller via +/// [`attestation_subscription_subnets`] and shared with the blockchain actor, +/// so both agree on exactly which subnets feed this node's gossip groups. The +/// set is consumed during [`build_swarm`] and NOT stored on [`P2PServer`]: +/// runtime toggles of the aggregator role via the admin API (see +/// [`ethlambda_types::aggregator::AggregatorController`]) intentionally do not +/// resubscribe gossip subnets; this is the leanSpec PR #636 "hot-standby model" +/// scope limitation. A node that may aggregate at runtime must include those +/// subnets here at startup. pub struct SwarmConfig { pub node_key: Vec, pub bootnodes: Vec, pub listening_socket: SocketAddr, pub validator_ids: Vec, pub attestation_committee_count: u64, - pub is_aggregator: bool, - pub aggregate_subnet_ids: Option>, + /// Attestation subnets to subscribe to, precomputed via + /// [`attestation_subscription_subnets`]. + pub subscription_subnets: HashSet, +} + +/// The attestation subnets a node subscribes to: every validator subscribes +/// to its own committee subnet (`validator_id % attestation_committee_count`) +/// for mesh health, and an aggregator additionally subscribes to any explicit +/// `aggregate_subnet_ids`, falling back to subnet 0 when it would otherwise +/// subscribe to none. +pub fn attestation_subscription_subnets( + validator_ids: &[u64], + attestation_committee_count: u64, + is_aggregator: bool, + aggregate_subnet_ids: Option<&[u64]>, +) -> HashSet { + let mut subnets: HashSet = validator_ids + .iter() + .map(|vid| vid % attestation_committee_count) + .collect(); + if is_aggregator { + if let Some(ids) = aggregate_subnet_ids { + subnets.extend(ids.iter().copied()); + } + // Fall back to subnet 0 only when the aggregator has no validators and + // no explicit subnets; otherwise leave the set as configured. + if subnets.is_empty() { + subnets.insert(0); + } + } + subnets } /// Result of building the swarm — contains all pieces needed to start the P2P actor. @@ -304,35 +335,18 @@ pub fn build_swarm( .subscribe(&aggregation_topic) .unwrap(); - // Subscribe to attestation subnets per leanSpec (`src/lean_spec/__main__.py`): - // every validator subscribes to its own subnet for mesh health; aggregators - // additionally subscribe to explicit `aggregate_subnet_ids` and fall back to - // subnet 0 when they have no validators of their own. - let validator_subnets: HashSet = config + // The committee metric should reflect validator membership only, not + // aggregator-only subscriptions. + let metric_subnet = config .validator_ids .iter() .map(|vid| vid % config.attestation_committee_count) - .collect(); - - // The committee metric should reflect validator membership only, not - // aggregator-only subscriptions. - let metric_subnet = validator_subnets.iter().copied().min().unwrap_or(0); + .min() + .unwrap_or(0); metrics::set_attestation_committee_subnet(metric_subnet); - let mut subscription_subnets = validator_subnets; - if config.is_aggregator { - if let Some(ref explicit_ids) = config.aggregate_subnet_ids { - subscription_subnets.extend(explicit_ids); - } - // Fall back to subnet 0 only when the aggregator has no validators - // and no explicit subnets — otherwise leave the set as configured. - if subscription_subnets.is_empty() { - subscription_subnets.insert(0); - } - } - let mut attestation_topics: HashMap = HashMap::new(); - for &subnet_id in &subscription_subnets { + for &subnet_id in &config.subscription_subnets { let topic = attestation_subnet_topic(subnet_id); swarm.behaviour_mut().gossipsub.subscribe(&topic)?; info!(subnet_id, "Subscribed to attestation subnet"); diff --git a/crates/storage/src/store.rs b/crates/storage/src/store.rs index c1bd412e..cd1a811b 100644 --- a/crates/storage/src/store.rs +++ b/crates/storage/src/store.rs @@ -463,6 +463,16 @@ impl GossipSignatureBuffer { .collect() } + /// Largest signature count among data groups whose attestation slot is `slot`. + fn max_group_count_for_slot(&self, slot: u64) -> usize { + self.data + .values() + .filter(|entry| entry.data.slot == slot) + .map(|entry| entry.signatures.len()) + .max() + .unwrap_or(0) + } + /// Extract per-validator latest attestations from the raw signature pool. /// /// Mirrors `PayloadBuffer::extract_latest_attestations`: votes are processed @@ -1449,6 +1459,15 @@ impl Store { gossip.total_signatures() } + /// Largest per-group signature count among gossip groups voting for `slot`. + /// + /// One lock, no signature clones — cheap enough to call per gossip insert. + /// Drives the early-aggregation threshold check. + pub fn max_gossip_group_count_for_slot(&self, slot: u64) -> usize { + let gossip = self.gossip_signatures.lock().unwrap(); + gossip.max_group_count_for_slot(slot) + } + /// Estimated live data size in bytes for a table, as reported by the backend. pub fn estimate_table_bytes(&self, table: Table) -> u64 { self.backend.estimate_table_bytes(table)