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 32c2ac15..83ae1941 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; @@ -183,16 +184,19 @@ 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, /// Proposer-side block-building policy @@ -332,16 +336,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(); } @@ -386,7 +406,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(); @@ -407,6 +427,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 { @@ -415,6 +437,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. @@ -424,7 +465,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( @@ -435,11 +482,70 @@ 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 one + // committee's expected votes, rounded up: `ceil(2 * N / (3 * C))` + // (0 only when there are no 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; + (2 * validator_count).div_ceil(3 * self.attestation_committee_count) 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(); @@ -1056,8 +1162,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; + } } } @@ -1086,6 +1199,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 { @@ -1100,12 +1216,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, @@ -1114,6 +1240,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 28e510f5..5ae6e9e0 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!( @@ -590,6 +609,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); @@ -601,6 +621,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); @@ -627,6 +648,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/storage/src/store.rs b/crates/storage/src/store.rs index c916bfca..8e344f9b 100644 --- a/crates/storage/src/store.rs +++ b/crates/storage/src/store.rs @@ -464,6 +464,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`: iterate data_roots @@ -1453,6 +1463,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)