diff --git a/CLAUDE.md b/CLAUDE.md index 05cf5048..5885fa3c 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -19,7 +19,7 @@ crates/ ├─ src/lib.rs # BlockChain actor, tick events, validator duties ├─ src/store.rs # Fork choice store, block/attestation processing ├─ src/block_builder.rs # Block assembly (pre-built at previous slot's interval 4) - ├─ src/aggregation.rs # Interval-2 signature aggregation worker + ├─ src/aggregation.rs # Always-on signature aggregation worker (own thread + Store clone) ├─ src/reaggregate.rs # Re-aggregation of block-borne votes on import ├─ src/sync_status.rs # Sync-gate tracker (suppresses duties while syncing) ├─ src/key_manager.rs # Validator key management and signing @@ -56,7 +56,7 @@ crates/ ``` Interval 0: Block published (at the slot boundary). The build+publish code path is merged into the previous slot's interval 4 (see below) and aligned to publish here; no attestation acceptance happens at interval 0. Interval 1: Attestation production (all validators, including proposer) -Interval 2: Aggregation (aggregators create proofs from gossip signatures) +Interval 2: Aggregate publication (aggregators gossip the aggregates their worker produced). Proving itself is NOT confined to this interval: the worker runs continuously, and the actor buffers each finished aggregate until this tick. One that finishes DURING interval 2 or 3 is gossiped on arrival instead, since the window is already open and buffering would hold it a full slot. Interval 3: Safe target update (fork choice) Interval 4: Accept accumulated attestations; build the NEXT slot's block and publish it aligned to that slot's interval 0 (build and publish merged into this tick) ``` @@ -72,6 +72,63 @@ Fork choice head update (Store buffer fields are `new_payloads`/`known_payloads`; the accessors are named `extract_latest_new_attestations`/`extract_latest_known_attestations`.) +### Always-On Aggregation Worker (`aggregation.rs`) +- One plain `std::thread` spawned in the actor's `#[started]` hook, joined in `#[stopped]`. + Not `spawn_blocking`: it lives for the process and awaits nothing, so a blocking-pool + thread would be parked permanently for nothing +- Holds its own `Store` clone (same backend, same in-memory buffers), so it both re-reads + the pool itself rather than being handed a per-slot snapshot AND writes back what it + produces: `select_best_job` → prove → `store_aggregate` → `AggregateProduced` message → + repeat. Nothing eligible means a `WORKER_IDLE_POLL` sleep +- The WORKER stores each aggregate (payload into the pool, gossip signatures out of it) + before announcing it, so the message carries no proof and the actor's mailbox never + holds one. The actor only **buffers** the gossip publication in `pending_aggregates` + until interval 2 +- That write races the actor, and is safe on two counts: `insert_new_aggregated_payload` + records fork-choice votes with a max-merge (`should_replace_vote`), which is + order-independent, and a concurrent promote MOVES votes new→known rather than dropping + them, so the worst interleaving defers a vote or its payload by one tick. + `delete_gossip_signatures` can only race a duplicate of a vote the proof already binds +- `pending_aggregates` keys by attestation data root and stores PARTICIPANT SETS, not + proofs. The proof is already in the payload pool, and `Store::proof_for_participants` + fetches the named one back at publish time, so the buffer costs a bitfield per aggregate + instead of up to 512 KiB. A miss means the pool let the proof go (subsumed, pruned, + evicted) and there is nothing left to send +- Naming a SPECIFIC proof is what makes that read-back safe. Do NOT change it to ask the + pool for the best proof under a data root: the pool also holds peers' proofs for the + same attestation data, `PayloadBuffer` keeps a peer's disjoint proof beside ours, and + any "pick the best" rule would publish theirs in place of ours and leave our subnet's + votes off the wire entirely +- `PendingAggregate::push` mirrors `PayloadBuffer`'s subsumption rule on the participant + sets, so a re-proved group replaces the narrower name, disjoint sets both survive, and + the names kept are the ones that still resolve. `MAX_PENDING_AGGREGATES` bounds the rest +- `JobPolicy` gates what the worker may take, by position in the slot: backlog + work early, a current-slot group once it holds `min_current_slot_group_sigs`, and inside + `EARLY_AGGREGATION_WINDOW` before interval 2 nothing but that group (a backlog job is a + recursive merge that would occupy the single prover across the boundary) +- *Which* slot comes from `store.current_slot()`, the same clock `on_tick`'s idempotency + guard keys on; only the sub-interval position inside it comes from the wall clock + (`ms_into_slot`, clamped to that slot). Two independent clocks would let the worker and + the actor disagree about the current slot, and `select_best_job` buckets on exactly that +- The actor parks the worker through a `PauseReason` bitset (`AtomicU8`), and the worker + takes no new job while any bit is set. `BlockBuild` is scoped, taken as an RAII + `PauseGuard` around `propose_block`, since both compete for the same single-threaded + leanVM prover. `Syncing` is level-driven from `duties_allowed()` on every tick via + `set_paused`, so a node that is behind does not prove a backlog against the prover its + block import needs, and `--disable-duty-sync-gate` keeps the worker running for free. + A proof already in flight is not interrupted — `aggregate_mixed` cannot be +- One reason, one owner: a bitset (rather than a bool or a depth counter) makes setting + idempotent for the level-driven owner and stops a guard's drop from releasing someone + else's reason +- Division of labor: the worker reads shared state that OTHER threads write (the + `AggregatorController`, which the RPC thread toggles); state the actor itself owns + reaches the worker as a `PauseReason`, so the policy is not re-derived in two places +- The worker keeps NO memory of what it has already proved. It does not need one: it + stores each aggregate before its next selection round, so the pool that round re-reads + already accounts for it. A FAILED proof does leave the pool untouched and will be picked + again, which is accepted; the failure branch sleeps `WORKER_IDLE_POLL` so a proof that + fails cheaply cannot spin the thread + ### State Transition Phases 1. **process_slots()**: Advance through empty slots, update historical roots 2. **process_block()**: Validate header → process attestations → update justifications/finality @@ -350,7 +407,7 @@ incremental, and line-tables-only debuginfo, so rebuilds are much faster than ### Aggregator Flag Required for Finalization - At least one node **must** be started with `--is-aggregator` to finalize blocks - Without this flag, attestations pass signature verification and are logged as "Attestation processed", but the signature is never stored for aggregation (the `is_aggregator` gate in `on_gossip_attestation`, `store.rs`), so blocks are always built with `attestation_count=0` -- The attestation pipeline: gossip → verify signature → store gossip signature (only if `is_aggregator`) → aggregate at interval 2 → promote to known → pack into blocks +- The attestation pipeline: gossip → verify signature → store gossip signature (only if `is_aggregator`) → aggregation worker picks it up on its next selection round → publish at interval 2 → promote to known → pack into blocks - **Symptom**: `justified_slot=0` and `finalized_slot=0` indefinitely despite healthy block production and attestation gossip ### Runtime Aggregator Toggle (Hot-Standby Model) diff --git a/bin/ethlambda/src/main.rs b/bin/ethlambda/src/main.rs index fb67972b..b33aebab 100644 --- a/bin/ethlambda/src/main.rs +++ b/bin/ethlambda/src/main.rs @@ -35,7 +35,6 @@ use tokio_util::sync::CancellationToken; use cli::NodeOptions; use command::Command; - use ethlambda_blockchain::block_builder::ProposerConfig; use ethlambda_blockchain::key_manager::ValidatorKeyPair; use ethlambda_crypto::signature::ValidatorSecretKey; @@ -256,10 +255,10 @@ async fn run_node(options: NodeOptions) -> eyre::Result<()> { // 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. + // (to size the aggregation worker's vote-propagation gate), 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, diff --git a/crates/blockchain/src/aggregation.rs b/crates/blockchain/src/aggregation.rs index e5efc37f..e2b42b0b 100644 --- a/crates/blockchain/src/aggregation.rs +++ b/crates/blockchain/src/aggregation.rs @@ -1,80 +1,150 @@ -//! Committee-signature aggregation: off-thread worker orchestration and the +//! Committee-signature aggregation: the always-on off-thread worker and the //! pure functions it runs. //! -//! 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. +//! One worker thread is spawned when the blockchain actor starts and lives as +//! long as it does. It holds its own [`Store`] handle (a clone sharing the same +//! backend and in-memory buffers), so it both re-reads the pool itself instead +//! of being handed a per-slot snapshot and writes what it produces straight +//! back: pick the single best job available right now, run its expensive XMSS +//! proof, [`store_aggregate`] it, tell the actor with an [`AggregateProduced`] +//! message, pick again. With nothing eligible it polls every +//! [`WORKER_IDLE_POLL`]. //! -//! [`snapshot_aggregation_inputs`] builds the session's job list with a tiered -//! greedy selector modeled on `block_builder::select_attestations`: an -//! up-front store pass resolves every candidate `AttestationData`'s -//! aggregation material once (raw-first + trim, see [`resolve_job`]), then a -//! pure in-memory loop scores and orders candidates by consensus value -//! (current-slot before stale, then Finalize > Justify > Build), emitting at -//! most `max_jobs` jobs — [`MAX_AGGREGATION_JOBS`] normally, dropping to a -//! single job in the slot before one of our validators proposes. +//! It is a plain `std::thread`, not a `spawn_blocking` task. The thread runs +//! for the process's life and spends it in leanVM proofs, so handing it to the +//! runtime's blocking pool would park one of those threads permanently while +//! buying nothing: the loop awaits nothing, and it reaches the actor through an +//! unbounded channel that needs no reactor. +//! +//! Storing on the worker keeps the proof off the actor's mailbox: the message +//! carries only the attestation data and the participant set naming the proof, +//! and the actor reads the bytes back out of the pool when it publishes. That +//! publication is what stays on the interval grid, held to the +//! vote-aggregation interval unless the aggregate finishes inside the window +//! (see `SlotInterval::publishes_aggregates_on_arrival`), so proving is free to +//! run whenever. +//! +//! It also means the worker needs no memory of what it has already proved: the +//! pool its next selection round re-reads already accounts for it. A failed +//! proof is the exception, since it leaves the pool untouched and gets picked +//! again; that path sleeps [`WORKER_IDLE_POLL`] so a proof failing cheaply, +//! before the prover even runs, cannot spin the thread. +//! +//! [`select_best_job`] builds the candidate pool with the same tiered scoring +//! as `block_builder::select_attestations`: a store pass resolves every +//! candidate `AttestationData`'s aggregation material once (raw-first + trim, +//! see [`resolve_job`]), then a pure in-memory pass ranks candidates by +//! consensus value (current-slot before stale, then Finalize > Justify > +//! Build) and returns the winner. +//! +//! What the worker may pick up depends on where the slot is: see +//! [`JobPolicy`]. In short, a current-slot group needs two thirds of the +//! signatures this node expects before the vote-aggregation boundary, and +//! inside the early window ahead of that boundary the worker takes nothing +//! else — it would rather idle than start a recursive merge that runs into the +//! slot's committee aggregation. +//! +//! The actor parks the worker outright for as long as it needs the prover to +//! itself, or the node has no business aggregating: around its own block +//! build, and while the sync gate suppresses duties, so a node that is behind +//! spends the prover on the block import that closes the gap rather than on a +//! backlog the network has stopped waiting for. Both are [`PauseReason`]s (see +//! [`AggregationWorker::pause`] and [`AggregationWorker::set_paused`]). use std::collections::{HashMap, HashSet}; -use std::time::{Duration, Instant, SystemTime}; +use std::sync::Arc; +use std::sync::atomic::{AtomicU8, Ordering}; +use std::time::{Duration, Instant}; use ethlambda_crypto::aggregate_mixed; use ethlambda_crypto::signature::{ValidatorPublicKey, ValidatorSignature}; use ethlambda_storage::Store; use ethlambda_types::{ ShortRoot, + aggregator::AggregatorController, attestation::{AggregationBits, AttestationData, HashedAttestationData}, block::{ByteList512KiB, SingleMessageAggregate}, + chain_config::ChainConfig, constants::{INTERVALS_PER_SLOT, MIN_MILLISECONDS_PER_SLOT}, primitives::H256, state::Validator, }; use spawned_concurrency::message::Message; -use spawned_concurrency::tasks::{ActorRef, Context, send_after}; +use spawned_concurrency::tasks::ActorRef; use tokio_util::sync::CancellationToken; use tracing::{info, trace, warn}; use crate::block_builder::{self, EntryScore}; -use crate::metrics; - -/// Soft deadline for committee-signature aggregation measured from session -/// start: one full interval. After this much wall time elapses, the actor -/// signals the worker to stop via its cancellation token. A session started -/// exactly at interval 2 therefore runs until interval 3; 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) fn aggregation_deadline(milliseconds_per_interval: u64) -> Duration { - Duration::from_millis(milliseconds_per_interval) +use crate::{SlotInterval, metrics}; + +/// How long the worker waits before re-reading the pool when it found nothing +/// to do — no eligible job, the pause flag raised, or no aggregation duty. +/// Short enough that a signature arriving mid-interval is picked up promptly, +/// long enough that an idle node is not re-scanning the pool in a spin loop. +pub(crate) const WORKER_IDLE_POLL: Duration = Duration::from_millis(100); + +/// Upper bound we wait for the worker to exit on shutdown. Reached only when a +/// proof is mid-flight (`aggregate_mixed` cannot be interrupted); we warn +/// before giving up on the join. +pub(crate) const WORKER_JOIN_TIMEOUT: Duration = Duration::from_secs(2); + +/// How often shutdown checks whether the worker thread has exited. Joining a +/// thread blocks, and the actor's `stopped()` hook runs on the runtime, so the +/// wait polls instead of blocking a runtime thread on a proof that may still +/// have a second to run. +const WORKER_SHUTDOWN_POLL: Duration = Duration::from_millis(20); + +/// Offset within the slot at which the vote-propagation gate lifts: the start +/// of the vote-aggregation interval. Before it, a current-slot group needs +/// [`min_current_slot_group_sigs`] signatures to be worth a proof; from it on, +/// whatever the group holds is aggregated. +/// +/// Derived from the configured slot duration, like every other interval +/// boundary. +fn vote_aggregation_offset_ms(config: &ChainConfig) -> u64 { + // Slot 0 reduces `to_ms_since_genesis` to the offset within a slot. + SlotInterval::Aggregation.to_ms_since_genesis(0, config) } -/// 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); +/// How far into `slot` the wall clock is, in milliseconds. +/// +/// The store clock decides which slot the worker is in, but it counts whole +/// intervals and [`EARLY_AGGREGATION_WINDOW`] is finer than one, so the +/// position *inside* the slot still comes from the wall clock. It is measured +/// from `slot`'s own start and clamped to that slot, so when the two clocks +/// disagree the answer degrades to an edge of the slot the store says we are +/// in: short of it the permissive [`JobPolicy::Backlog`] end, past it +/// [`JobPolicy::Open`]. It never describes a position inside some other slot. +fn ms_into_slot(now_ms: u64, slot: u64, config: &ChainConfig) -> u64 { + let slot_start_ms = config.genesis_time_ms() + slot * config.milliseconds_per_slot; + now_ms + .saturating_sub(slot_start_ms) + .min(config.milliseconds_per_slot) +} -/// 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`). +/// How long before the vote-aggregation boundary the worker stops taking +/// anything but the slot's committee signatures. +/// +/// A backlog job is a recursive proof merge that can run well past the +/// boundary, and the prover is single-threaded: starting one here would delay +/// the aggregate the whole slot is waiting on. Idling instead costs little, +/// since the backlog is not going anywhere, and this window is where the +/// committee's signatures typically cross the two-thirds mark. /// -/// Fixed rather than scaled with the slot duration. What the window buys is -/// wall time for the leanVM proof to land before the block that carries it, -/// and a proof costs the same however long the network's slot is. +/// Fixed rather than scaled with the slot duration. What the window protects +/// is wall time for one leanVM proof, and a proof costs the same however long +/// the network's slot is. 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`. The -// slot duration is configurable, so the binding case is the narrowest interval -// a config file can ask for. Keep this invariant self-enforcing so a future -// bump to the window, or a lowered floor, can't silently underflow either -// subtraction. +// The window must not reach past the start of the slot, so `job_policy`'s +// subtraction cannot underflow into the previous one. The slot duration is +// configurable, so the binding case is the narrowest grid a config file can +// ask for. Keep the invariant self-enforcing so a future bump to the window, +// or a lowered floor, can't silently underflow that subtraction. const _: () = assert!( EARLY_AGGREGATION_WINDOW.as_millis() - <= (MIN_MILLISECONDS_PER_SLOT / INTERVALS_PER_SLOT) as u128, - "EARLY_AGGREGATION_WINDOW must not exceed the shortest configurable interval" + <= (2 * MIN_MILLISECONDS_PER_SLOT / INTERVALS_PER_SLOT) as u128, + "EARLY_AGGREGATION_WINDOW must not reach past the slot boundary at the shortest cadence" ); /// A single pre-prepared aggregation group. @@ -111,12 +181,6 @@ impl AggregationJob { } } -/// All input needed to run a session of committee-signature aggregation off-thread. -pub struct AggregationSnapshot { - pub(crate) jobs: Vec, - pub(crate) groups_considered: usize, -} - /// Result of one successful aggregation group. Carried back to the actor thread /// as a message payload so the store can be updated and gossip publish fired. pub struct AggregatedGroupOutput { @@ -126,101 +190,216 @@ pub struct AggregatedGroupOutput { pub(crate) keys_to_delete: Vec<(u64, H256)>, } -/// Tracks an in-flight off-thread aggregation worker so the actor can cancel, -/// join, and correlate incoming result messages with the right session. -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, - /// Handle to the `spawn_blocking` worker. Held so `stopped()` / new-session - /// start can await completion. - pub(crate) worker: tokio::task::JoinHandle<()>, +/// Why the worker is parked. The actor owns every reason and sets them +/// independently; the worker takes no new job while any is set. +/// +/// A bitset rather than a single flag or a depth counter: each reason has +/// exactly one owner, so setting one twice is idempotent, clearing one cannot +/// clear another's, and nothing is left to leak when an owner is level-driven +/// rather than scoped. Values are distinct bits of the `AtomicU8` in +/// [`AggregationWorker`]. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[repr(u8)] +pub(crate) enum PauseReason { + /// The actor is building a block. Both it and the worker run leanVM + /// proofs on the same single-threaded prover, and the block is the one + /// with a deadline. Scoped to the build, so it is taken as a + /// [`PauseGuard`]. + BlockBuild = 1 << 0, + /// The sync gate is suppressing this node's duties. A node that is behind + /// would otherwise prove a backlog the network has moved past, against + /// the same prover its block import needs to close the gap. Level-driven + /// from the actor's tick, so it is set through + /// [`AggregationWorker::set_paused`] rather than held as a guard. + Syncing = 1 << 1, } -/// One successful aggregate streamed back from the worker. -pub(crate) struct AggregateProduced { - pub(crate) session_id: u64, - pub(crate) output: AggregatedGroupOutput, +/// Handle to the always-on aggregation worker, held by the actor for the +/// actor's whole lifetime. +pub(crate) struct AggregationWorker { + /// Cancelled by the actor's `stopped()` hook; the worker breaks out of its + /// loop at the next job boundary. + cancel: CancellationToken, + /// Set of [`PauseReason`]s currently holding the worker back, as a bitset. + /// Non-zero means "take no new job"; see [`Self::pause`]. + paused: Arc, + /// Handle to the worker thread, held so shutdown can join it. + handle: std::thread::JoinHandle<()>, } -impl Message for AggregateProduced { - type Result = (); + +/// Set or clear one reason's bit. Read-modify-write, so reasons are +/// independent: an owner only ever touches its own bit. +fn set_pause_reason(paused: &AtomicU8, reason: PauseReason, on: bool) { + if on { + paused.fetch_or(reason as u8, Ordering::Release); + } else { + paused.fetch_and(!(reason as u8), Ordering::Release); + } } -/// Emitted by the worker after its loop exits (completion or cancellation). -pub(crate) struct AggregationDone { - pub(crate) session_id: u64, - pub(crate) groups_considered: usize, - pub(crate) groups_aggregated: usize, - pub(crate) total_raw_sigs: usize, - pub(crate) total_children: usize, - pub(crate) total_elapsed: Duration, - pub(crate) cancelled: bool, +impl AggregationWorker { + /// Stop handing the worker new jobs for `reason` for as long as the + /// returned guard lives. A proof already in flight is not interrupted + /// (`aggregate_mixed` cannot be), so this bounds contention rather than + /// eliminating it. + /// + /// For a reason whose lifetime is a scope. A reason the actor tracks as + /// state instead, recomputing it each tick, belongs in + /// [`Self::set_paused`]. Either way one reason has one owner: two live + /// guards for the same reason would release it when the first drops. + pub(crate) fn pause(&self, reason: PauseReason) -> PauseGuard { + set_pause_reason(&self.paused, reason, true); + PauseGuard { + paused: self.paused.clone(), + reason, + } + } + + /// Level-triggered form of [`Self::pause`]: bring `reason` in line with + /// `paused`, whatever it was before. Idempotent, so the actor can drive it + /// straight off a predicate it recomputes every tick without tracking + /// whether it already set it. + pub(crate) fn set_paused(&self, reason: PauseReason, paused: bool) { + set_pause_reason(&self.paused, reason, paused); + } + + /// Cancel the worker and wait up to [`WORKER_JOIN_TIMEOUT`] for it to exit. + /// + /// Polls rather than joining straight away: the thread only notices + /// cancellation between jobs, so a join here would block a runtime thread + /// for as long as the proof in flight takes. Past the timeout the thread is + /// left detached — it exits on its own once the current proof returns, and + /// the process is on its way out regardless. + pub(crate) async fn shutdown(self) { + self.cancel.cancel(); + + let deadline = Instant::now() + WORKER_JOIN_TIMEOUT; + while !self.handle.is_finished() && Instant::now() < deadline { + tokio::time::sleep(WORKER_SHUTDOWN_POLL).await; + } + + if !self.handle.is_finished() { + warn!( + timeout_secs = WORKER_JOIN_TIMEOUT.as_secs(), + "Aggregation worker still proving at shutdown; leaving it detached" + ); + return; + } + match self.handle.join() { + Ok(()) => info!("Aggregation worker joined on shutdown"), + Err(_) => warn!("Aggregation worker panicked"), + } + } } -impl Message for AggregationDone { - type Result = (); + +/// Clears its own [`PauseReason`] on drop, so an early return on the paused +/// code path cannot leave the worker parked forever. Touches no other +/// reason's bit; see [`AggregationWorker::pause`]. +pub(crate) struct PauseGuard { + paused: Arc, + reason: PauseReason, } -/// 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, +impl Drop for PauseGuard { + fn drop(&mut self) { + set_pause_reason(&self.paused, self.reason, false); + } } -impl Message for AggregationDeadline { - type Result = (); + +/// Startup-fixed inputs the worker's vote-propagation gate needs. Both come +/// from the CLI and never change at runtime, so the worker owns a copy instead +/// of reaching back into the actor. +#[derive(Clone)] +pub(crate) struct WorkerConfig { + /// Number of attestation committees (= subnet count). + pub(crate) attestation_committee_count: u64, + /// Attestation subnets this node subscribes to. + pub(crate) subscribed_subnets: HashSet, } -/// 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 { +/// One successful aggregate announced to the actor, after the worker has +/// already stored it. +/// +/// Carries no proof: [`store_aggregate`] put it in the pending payload pool, +/// and `participants` names it there for `Store::proof_for_participants`. A +/// proof is up to [`ByteList512KiB`], so keeping it out of the mailbox keeps +/// the actor's queue small however far behind publication falls. +pub(crate) struct AggregateProduced { + pub(crate) hashed: HashedAttestationData, + /// Participant set of the stored proof, which is both what names it in the + /// pool and what the actor buffers until publication. + pub(crate) participants: AggregationBits, + /// Wall time the proof itself took, observed on the worker thread. + pub(crate) elapsed: Duration, +} +impl Message for AggregateProduced { type Result = (); } -/// Maximum number of aggregation jobs selected per interval-2 session. Caps -/// leanVM prover work against [`aggregation_deadline`]: the greedy loop in -/// [`snapshot_aggregation_inputs`] stops after this many rounds even if -/// scoring candidates remain. -pub(crate) const MAX_AGGREGATION_JOBS: usize = 2; +/// What the worker is allowed to pick up, given where the slot is. +/// +/// The prover is single-threaded and the slot's committee aggregate is the one +/// piece of work with a deadline, so the policy tightens as that deadline +/// approaches and opens up once it has passed. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum JobPolicy { + /// Early in the slot: backlog work is welcome — stale groups, merges of + /// proofs already held — and a current-slot group is eligible once it + /// holds `min_sigs` signatures. + Backlog { min_sigs: usize }, + /// Inside [`EARLY_AGGREGATION_WINDOW`]: only a current-slot group that + /// already holds `min_sigs`. Anything else would occupy the prover across + /// the boundary and delay the aggregate the slot is waiting on, so the + /// worker idles until either the threshold is met or the boundary arrives. + CommitteeOnly { min_sigs: usize }, + /// From the vote-aggregation boundary on: everything is eligible, however + /// few signatures back it. + Open, +} + +impl JobPolicy { + /// Whether a current-slot gossip group holding `sigs` signatures may be + /// proved now. + fn admits_current_slot(self, sigs: usize) -> bool { + match self { + Self::Open => true, + Self::Backlog { min_sigs } | Self::CommitteeOnly { min_sigs } => sigs >= min_sigs, + } + } -/// 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. + /// Whether work other than the current slot's committee signatures may be + /// started now: a stale group, or a merge of proofs already in the pool. + fn admits_backlog(self) -> bool { + !matches!(self, Self::CommitteeOnly { .. }) + } +} + +/// Pick the single most valuable aggregation job available right now, or +/// `None` when nothing is worth proving. Touches the store, does no heavy +/// cryptography. /// -/// A tiered greedy selector modeled on `block_builder::select_attestations`: +/// A tiered selector modeled on `block_builder::select_attestations`: /// -/// 1. **Up-front store pass**: resolves every candidate `AttestationData` -/// into a store-free [`AggregationJob`] once via [`resolve_job`] -/// (raw-first, then trim). Candidates come from gossip groups +/// 1. **Store pass**: resolves every candidate `AttestationData` into a +/// store-free [`AggregationJob`] via [`resolve_job`] (raw-first, then +/// trim). Candidates come from gossip groups /// (`store.iter_gossip_signatures()`) and payload-only groups /// (`store.new_payload_keys()` not already a gossip candidate, requiring -/// at least two existing proofs to merge). -/// 2. **Greedy loop**, at most `max_jobs` rounds: each round -/// scores every unselected candidate against the projected state and -/// keeps the lowest ordering key (current-slot before stale, then -/// Finalize > Justify > Build, mirroring the block builder). The winning -/// [`AggregationJob`] is emitted as-is; the projection is updated with its -/// realized coverage. -/// -/// Stops early when no remaining candidate scores (converged). -/// -/// `max_jobs` is [`MAX_AGGREGATION_JOBS`] for an ordinary session and `1` when -/// the caller is about to build a block at interval 4 (see -/// `BlockChainServer::start_aggregation_session`). -pub fn snapshot_aggregation_inputs( - store: &Store, - current_slot: u64, - max_jobs: usize, -) -> Option { +/// at least two existing proofs to merge), each admitted or held back by +/// `policy`. +/// 2. **Ranking**: scores every candidate against the head state and keeps the +/// lowest ordering key (current-slot before stale, then Finalize > Justify +/// > Build, mirroring the block builder). +fn select_best_job(store: &Store, current_slot: u64, policy: JobPolicy) -> Option { let gossip_groups = store.iter_gossip_signatures(); - let new_payload_keys = store.new_payload_keys(); + let new_payload_keys = if policy.admits_backlog() { + store.new_payload_keys() + } else { + // A payload-only candidate is a pure proof merge: the most expensive + // job there is, and the one with the least claim on the prover right + // before the boundary. + Vec::new() + }; if gossip_groups.is_empty() && new_payload_keys.is_empty() { return None; @@ -233,6 +412,26 @@ pub fn snapshot_aggregation_inputs( for (hashed, validator_sigs) in &gossip_groups { let data_root = hashed.root(); + let admitted = if hashed.data().slot == current_slot { + // A current-slot group still collecting signatures is worth more as + // one wide proof after the boundary than as several thin ones + // before it. + policy.admits_current_slot(validator_sigs.len()) + } else { + // Stale groups are backlog: no further signature is coming for + // them, so they are only held back to keep the prover free. + policy.admits_backlog() + }; + if !admitted { + trace!( + ?policy, + sigs = validator_sigs.len(), + group_slot = hashed.data().slot, + data_root = %ShortRoot(&data_root.0), + "holding aggregation candidate back" + ); + continue; + } let (new_proofs, known_proofs) = store.existing_proofs_for_data(&data_root); if let Some(job) = resolve_job( hashed.clone(), @@ -264,7 +463,6 @@ pub fn snapshot_aggregation_inputs( if candidates.is_empty() { return None; } - let groups_considered = candidates.len(); let validator_count = validators.len(); // Chain view covering [0, head_slot]. A state's `historical_block_hashes` @@ -284,58 +482,104 @@ pub fn snapshot_aggregation_inputs( head_state.historical_block_hashes.iter().copied().collect(); extended_historical_block_hashes.push(store.head().expect("head read works")); - let mut projected = block_builder::ProjectedState::from_head_state(&head_state); + let projected = block_builder::ProjectedState::from_head_state(&head_state); + + // One round: the store is re-read before the next job, so a same-target + // candidate re-tiers against the aggregate this one produced (once + // applied) rather than against an in-memory projection of it. + let (data_root, score) = pick_best_candidate( + &candidates, + &projected, + &known_block_roots, + &extended_historical_block_hashes, + current_slot, + validator_count, + ) + .or_else(|| { + trace!("aggregation selection converged: no scoring candidates"); + None + })?; + + let job = candidates + .remove(&data_root) + .expect("picked candidate exists in pool"); + let att_data = job.hashed.data(); - let mut jobs: Vec = Vec::with_capacity(max_jobs.min(groups_considered)); - for _round in 0..max_jobs { - let Some((data_root, score)) = pick_best_candidate( - &candidates, - &projected, - &known_block_roots, - &extended_historical_block_hashes, - current_slot, - validator_count, - ) else { - trace!( - jobs_selected = jobs.len(), - "aggregation selection converged: no scoring candidates" - ); - break; - }; - - let job = candidates - .remove(&data_root) - .expect("picked candidate exists in pool"); - let coverage = job.coverage(); - let att_data = job.hashed.data(); - let target_root = att_data.target.root; - let target_slot = att_data.target.slot; - - trace!( - tier = ?score.tier, - new_voters = score.new_voters, - target_slot, - target_root = %ShortRoot(&target_root.0), - data_root = %ShortRoot(&data_root.0), - "selected aggregation job" - ); + trace!( + tier = ?score.tier, + new_voters = score.new_voters, + target_slot = att_data.target.slot, + target_root = %ShortRoot(&att_data.target.root.0), + data_root = %ShortRoot(&data_root.0), + "selected aggregation job" + ); - // Fold the job's realized coverage into the shared projection so - // same-target candidates re-tier across rounds exactly as the block - // builder's post-state would. - projected.advance(score.tier, att_data, coverage.iter().copied()); + Some(job) +} - jobs.push(job); +/// Minimum gossip signatures a current-slot group must hold for the worker to +/// prove it before the vote-aggregation boundary: two thirds of the votes this +/// node expects 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`. +/// +/// Returns `None` when no such validator exists (no subscribed subnet is in +/// range, or the chain has no committees), which no group can ever clear: the +/// caller treats that as "wait for the boundary". +fn min_current_slot_group_sigs( + validator_count: u64, + committee_count: u64, + subscribed_subnets: &HashSet, +) -> Option { + if committee_count == 0 { + return None; } + let expected_votes: u64 = subscribed_subnets + .iter() + .filter(|&&subnet| subnet < committee_count) + .map(|&subnet| { + validator_count / committee_count + + u64::from(subnet < validator_count % committee_count) + }) + .sum(); + let min_sigs = (2 * expected_votes).div_ceil(3) as usize; + (min_sigs > 0).then_some(min_sigs) +} - if jobs.is_empty() { - return None; +/// The policy in force `ms_into_slot` into the slot. +fn job_policy( + ms_into_slot: u64, + time_config: &ChainConfig, + store: &Store, + config: &WorkerConfig, +) -> JobPolicy { + let vote_aggregation_offset_ms = vote_aggregation_offset_ms(time_config); + if ms_into_slot >= vote_aggregation_offset_ms { + return JobPolicy::Open; } - Some(AggregationSnapshot { - jobs, - groups_considered, - }) + let validator_count = store.head_state().validators.len() as u64; + // With no votes expected there is no quorum to wait for, so nothing + // justifies proving a current-slot group early: an unreachable floor holds + // every one of them to the boundary. + let min_sigs = min_current_slot_group_sigs( + validator_count, + config.attestation_committee_count, + &config.subscribed_subnets, + ) + .unwrap_or(usize::MAX); + + let window_opens_at = vote_aggregation_offset_ms - EARLY_AGGREGATION_WINDOW.as_millis() as u64; + if ms_into_slot >= window_opens_at { + JobPolicy::CommitteeOnly { min_sigs } + } else { + JobPolicy::Backlog { min_sigs } + } } /// Scan the candidate pool and pick the best-scoring, not-yet-selected entry. @@ -570,22 +814,62 @@ pub fn aggregate_job(job: AggregationJob) -> Option { }) } -/// Apply a worker-produced aggregate to the store. Called per message on the -/// actor thread; gauge metrics that depend on total counts are batched into -/// `finalize_aggregation_session` so we pay one lock per session instead of -/// one per aggregate. Idempotent wrt the gossip delete. -pub fn apply_aggregated_group(store: &mut Store, output: &AggregatedGroupOutput) { - store.insert_new_aggregated_payload(output.hashed.clone(), output.proof.clone()); - store.delete_gossip_signatures(&output.keys_to_delete); +/// Store one aggregate the worker just produced and build the announcement +/// for the actor: the proof into the pending payload pool, and the gossip +/// signatures it consumed out of the pool. +/// +/// Runs on the worker thread through its own `Store` handle. `Store`'s `&mut +/// self` does not mean exclusive access (its buffers are behind `Arc>` +/// and the actor holds a handle of its own), so this interleaves with the +/// actor, and both writes are safe under that: +/// +/// - `insert_new_aggregated_payload` records the fork-choice votes before it +/// pushes the payload, and it records them with a max-merge +/// (`should_replace_vote`) that gives the same map whatever order concurrent +/// writers arrive in. A promote landing in the middle moves votes from `new` +/// to `known` rather than dropping them, so the worst interleaving leaves the +/// vote or the payload to be promoted one tick later. Neither is lost. +/// - `delete_gossip_signatures` removes keys this proof consumed. A signature +/// arriving concurrently for one of them is the same validator's signature +/// over the same attestation data, i.e. a duplicate of a vote the proof +/// already binds, so deleting it loses nothing. +/// +/// Gauge metrics that depend on total counts are batched into +/// [`refresh_pool_gauges`] instead, so we pay one lock per slot rather than one +/// per aggregate. Idempotent wrt the gossip delete. +fn store_aggregate( + store: &mut Store, + output: AggregatedGroupOutput, + elapsed: Duration, +) -> AggregateProduced { + let AggregatedGroupOutput { + hashed, + proof, + participants, + keys_to_delete, + } = output; + // Named before the proof moves into the pool; the bitfield is a few + // hundred bytes against the proof's up-to-512 KiB. + let bits = proof.participants.clone(); + + store.insert_new_aggregated_payload(hashed.clone(), proof); + store.delete_gossip_signatures(&keys_to_delete); metrics::inc_pq_sig_aggregated_signatures(); - metrics::inc_pq_sig_attestations_in_aggregated_signatures(output.participants.len() as u64); + metrics::inc_pq_sig_attestations_in_aggregated_signatures(participants.len() as u64); + + AggregateProduced { + hashed, + participants: bits, + elapsed, + } } -/// End-of-session gauge refresh. Called once after the worker finishes so the -/// `lean_latest_new_aggregated_payloads` and `lean_gossip_signatures` gauges -/// settle on the final counts instead of being churned per aggregate. -pub fn finalize_aggregation_session(store: &Store) { +/// Refresh the pool-size gauges. Called from the vote-aggregation tick, once +/// the slot's buffered aggregates have gone out, so +/// `lean_latest_new_aggregated_payloads` and `lean_gossip_signatures` settle +/// on a per-slot reading instead of being churned per aggregate. +pub fn refresh_pool_gauges(store: &Store) { metrics::update_latest_new_aggregated_payloads(store.new_aggregated_payloads_count()); metrics::update_gossip_signatures(store.gossip_signatures_count()); } @@ -670,59 +954,94 @@ pub(crate) fn aggregation_bits_from_validator_indices(bits: &[u64]) -> Aggregati aggregation_bits } -/// Worker loop — runs on a `spawn_blocking` thread, no store access. +/// Spawn the always-on aggregation worker on its own thread. /// -/// 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). +/// The worker owns a [`Store`] clone — same backend, same in-memory buffers — +/// the shared aggregator-role flag (so a runtime toggle from the RPC thread +/// reaches it without a restart), and the startup-fixed gate inputs. State the +/// actor owns rather than shares, such as the sync verdict, reaches it as a +/// [`PauseReason`] instead. It runs until the returned handle's +/// [`AggregationWorker::shutdown`] cancels it. +pub(crate) fn spawn_aggregation_worker( + store: Store, + actor: ActorRef, + aggregator: AggregatorController, + config: WorkerConfig, +) -> AggregationWorker { + let cancel = CancellationToken::new(); + let paused = Arc::new(AtomicU8::new(0)); + let handle = { + let cancel = cancel.clone(); + let paused = paused.clone(); + std::thread::Builder::new() + .name("aggregation-worker".to_owned()) + .spawn(move || run_aggregation_worker(store, actor, aggregator, config, cancel, paused)) + .expect("spawning the aggregation worker thread") + }; + + AggregationWorker { + cancel, + paused, + handle, + } +} + +/// Worker loop — runs on its own thread for the actor's lifetime. /// -/// 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, +/// Each round re-reads the pool through the store handle, picks the best job +/// ([`select_best_job`]), proves it, and hands the result to the actor as an +/// [`AggregateProduced`] message. With nothing to do — nothing eligible, +/// parked for some [`PauseReason`], or no aggregation duty — it sleeps +/// [`WORKER_IDLE_POLL`] and looks again. +/// +/// `aggregate_mixed` cannot be interrupted, so both cancellation and the pause +/// reasons are only observed between jobs. +fn run_aggregation_worker( + mut store: Store, actor: ActorRef, + aggregator: AggregatorController, + config: WorkerConfig, cancel: CancellationToken, - session_id: u64, - publish_at: SystemTime, + paused: Arc, ) { - let start = Instant::now(); - let groups_considered = snapshot.groups_considered; - let mut groups_aggregated = 0usize; - let mut total_raw_sigs = 0usize; - let mut total_children = 0usize; - let jobs_total = snapshot.jobs.len(); - let mut jobs_attempted = 0usize; - - for job in snapshot.jobs { - if cancel.is_cancelled() { - break; - } - jobs_attempted += 1; + info!("Aggregation worker started"); + + // The chain's time grid never changes at runtime, so one read covers the + // worker's whole life. + let time_config = *store.config(); + + while !cancel.is_cancelled() { + let Some(job) = next_job(&store, &time_config, &aggregator, &paused, &config) else { + std::thread::sleep(WORKER_IDLE_POLL); + continue; + }; let slot = job.slot; let raw_sigs = job.raw_ids.len(); let children = job.children.len(); - let group_start = Instant::now(); - let Some(output) = aggregate_job(job) else { - let elapsed = group_start.elapsed(); + let job_start = Instant::now(); + let output = aggregate_job(job); + let elapsed = job_start.elapsed(); + + let Some(output) = output else { warn!( - session_id, slot, raw_sigs, children, ?elapsed, "Committee signature aggregation failed" ); + metrics::inc_aggregator_skipped_other(1); + // A failure leaves the store exactly as it found it, so the next + // round re-reads the same pool and picks the same job. Sleep before + // looping: a proof that fails cheaply, before the prover runs, + // would otherwise spin this thread at full speed. + std::thread::sleep(WORKER_IDLE_POLL); continue; }; - let elapsed = group_start.elapsed(); + info!( - session_id, slot, raw_sigs, children, @@ -731,52 +1050,63 @@ pub(crate) fn run_aggregation_worker( "Committee signature aggregated" ); - groups_aggregated += 1; - total_raw_sigs += raw_sigs; - total_children += children; - - // 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 }, - ); + // Store before announcing, so the pool the actor reads to publish, and + // the one the next selection round re-reads, both already account for + // this aggregate. + let produced = store_aggregate(&mut store, output, elapsed); + if actor.send(produced).is_err() { + // Actor is gone; nothing would consume further aggregates. + break; } } - // Jobs the loop never reached (deadline cancellation or actor gone) are - // skipped aggregation submissions per leanMetrics. - let jobs_dropped = jobs_total - jobs_attempted; - if jobs_dropped > 0 { - metrics::inc_aggregator_skipped_other(jobs_dropped as u64); + info!("Aggregation worker stopped"); +} + +/// One round of job selection: honor the role flag and the pause reasons, take +/// the slot from the store clock and the [`JobPolicy`] from where the wall +/// clock sits inside it, then ask [`select_best_job`] for the winner. `None` +/// means "nothing to do right now", which inside the early window is a +/// deliberate answer rather than an idle one. +fn next_job( + store: &Store, + time_config: &ChainConfig, + aggregator: &AggregatorController, + paused: &AtomicU8, + config: &WorkerConfig, +) -> Option { + // The role flag is read here because the RPC thread writes it. Everything + // the actor itself owns, the sync verdict included, reaches us as a + // [`PauseReason`] instead of being re-derived from shared state. + if !aggregator.is_enabled() || paused.load(Ordering::Acquire) != 0 { + return None; + } + + let now_ms = crate::unix_now_ms(); + // Before genesis there is no slot to aggregate for. A "has the chain + // started" test only; which slot we are in comes from the store clock. + if now_ms < time_config.genesis_time_ms() { + return None; } - let _ = actor.send(AggregationDone { - session_id, - groups_considered, - groups_aggregated, - total_raw_sigs, - total_children, - total_elapsed: start.elapsed(), - cancelled: cancel.is_cancelled(), - }); + // The slot comes from the store clock, the one authority the actor drives + // the interval grid off (`on_tick`'s idempotency guard keys on it). + // Derived independently from the wall clock the two could disagree — the + // wall clock drifts behind the monotonic tick cadence inside VMs, and a + // long block build leaves the store clock ahead of it — and + // `select_best_job` would then bucket the slot's real group as stale, + // proving it thin below `min_sigs`, or bucket a stale group as current and + // hold it back to a boundary that has already passed. + let slot = store.current_slot(); + + let policy = job_policy( + ms_into_slot(now_ms, slot, time_config), + time_config, + store, + config, + ); + + select_best_job(store, slot, policy) } #[cfg(test)] @@ -1173,7 +1503,8 @@ mod tests { assert_eq!(picked_root, root_a); assert_eq!(score.tier, block_builder::Tier::Build); - // Apply the selection to the projection, as `snapshot_aggregation_inputs` would. + // Fold the winner into the projection, standing in for the store + // update the actor applies before the worker's next round. let winner = candidates.remove(&picked_root).expect("A is in the pool"); projected .current_votes @@ -1201,21 +1532,21 @@ mod tests { ); } - // ---- snapshot_aggregation_inputs (full pipeline) ---- + // ---- select_best_job (full pipeline) ---- /// An empty store (no gossip signatures, no pending payloads) has nothing /// to aggregate. #[test] - fn snapshot_returns_none_for_empty_store() { + fn select_returns_none_for_empty_store() { let hashes = vec![H256([1u8; 32])]; let store = new_test_store(make_head_state(0, 4, &hashes)); - assert!(snapshot_aggregation_inputs(&store, 0, MAX_AGGREGATION_JOBS).is_none()); + assert!(select_best_job(&store, 0, JobPolicy::Open).is_none()); } /// A single gossip signature with no other material to merge is dropped /// as non-viable up front, leaving zero candidates. #[test] - fn snapshot_returns_none_for_lone_raw_signature() { + fn select_returns_none_for_lone_raw_signature() { let hashes = vec![H256([1u8; 32])]; let mut store = new_test_store(make_head_state(0, 4, &hashes)); insert_test_block(&mut store, hashes[0], 0, H256::ZERO); @@ -1238,7 +1569,7 @@ mod tests { let hashed = HashedAttestationData::new(att_data); store.insert_gossip_signature(hashed, 0, dummy_sig()); - assert!(snapshot_aggregation_inputs(&store, 0, MAX_AGGREGATION_JOBS).is_none()); + assert!(select_best_job(&store, 0, JobPolicy::Open).is_none()); } /// A group whose target is already justified (here: at or behind the @@ -1246,7 +1577,7 @@ mod tests { /// must never become a job, even with enough raw sigs to otherwise be /// viable. #[test] - fn snapshot_skips_group_whose_target_is_already_justified() { + fn select_skips_group_whose_target_is_already_justified() { const NUM_VALIDATORS: usize = 10; const HEAD_SLOT: u64 = 20; const FINALIZED_SLOT: u64 = 10; @@ -1281,7 +1612,7 @@ mod tests { store.insert_gossip_signature(hashed, 1, dummy_sig()); assert!( - snapshot_aggregation_inputs(&store, 999, MAX_AGGREGATION_JOBS).is_none(), + select_best_job(&store, 999, JobPolicy::Open).is_none(), "a group targeting an already-justified slot must never become a job" ); } @@ -1301,7 +1632,7 @@ mod tests { /// This test FAILS against the unextended (buggy) chain view and PASSES /// after the `store.head()` extension. #[test] - fn snapshot_aggregates_vote_for_current_head_on_non_genesis_chain() { + fn select_aggregates_vote_for_current_head_on_non_genesis_chain() { const NUM_VALIDATORS: usize = 10; const HEAD_SLOT: u64 = 4; @@ -1337,11 +1668,10 @@ mod tests { store.insert_gossip_signature(hashed.clone(), 0, dummy_sig()); store.insert_gossip_signature(hashed, 1, dummy_sig()); - let snapshot = snapshot_aggregation_inputs(&store, HEAD_SLOT, MAX_AGGREGATION_JOBS) + let job = select_best_job(&store, HEAD_SLOT, JobPolicy::Open) .expect("a vote for the current head must produce a job (chain view covers the tip)"); - assert_eq!(snapshot.jobs.len(), 1); assert_eq!( - snapshot.jobs[0].hashed.data().target.slot, + job.hashed.data().target.slot, HEAD_SLOT, "the job aggregates the vote targeting the current head" ); @@ -1391,46 +1721,229 @@ mod tests { store } - /// With more scoring candidates than `MAX_AGGREGATION_JOBS`, exactly that - /// many jobs are produced — the best `MAX_AGGREGATION_JOBS` by ordering - /// key, i.e. the top two by `target_slot`. + /// From a pool of competing candidates the selector emits the single + /// best-scoring one — here the highest `target_slot`, which wins the + /// Build-tier `new_voters` tie. #[test] - fn snapshot_caps_jobs_at_max_aggregation_jobs() { + fn select_picks_the_best_scoring_candidate() { let store = store_with_competing_build_tier_groups(); - let snapshot = snapshot_aggregation_inputs(&store, 999, MAX_AGGREGATION_JOBS) - .expect("should produce jobs"); - assert_eq!(snapshot.groups_considered, NUM_GROUPS); - assert_eq!(snapshot.jobs.len(), MAX_AGGREGATION_JOBS); + let job = select_best_job(&store, 999, JobPolicy::Open).expect("should produce a job"); + assert_eq!(job.hashed.data().target.slot, NUM_GROUPS as u64); + } - let selected_targets: HashSet = snapshot - .jobs - .iter() - .map(|job| job.hashed.data().target.slot) - .collect(); + /// Early in the slot a current-slot group short of the signature floor is + /// held back, but stale groups in the same pool are still fair game: the + /// worker keeps busy on the backlog. + #[test] + fn select_holds_current_slot_group_below_the_floor() { + let store = store_with_competing_build_tier_groups(); + + // Groups carry two signatures each and are keyed by `target_slot`, + // which doubles as their attestation slot in this fixture. + let job = select_best_job( + &store, + NUM_GROUPS as u64, + JobPolicy::Backlog { min_sigs: 3 }, + ) + .expect("stale groups stay eligible"); assert_eq!( - selected_targets, - HashSet::from([4, 5]), - "the two highest target_slot groups win the new_voters tie" + job.hashed.data().target.slot, + NUM_GROUPS as u64 - 1, + "the current-slot group is held; the best stale one is taken instead" ); } - /// The proposer cap (`max_jobs = 1`) yields exactly one job from the same - /// pool, and it is the single best-scoring candidate — the one the uncapped - /// selection also picks first (highest `target_slot`). Every other candidate - /// is still counted in `groups_considered`, so the cap is visibly a - /// selection bound rather than a narrower candidate pool. + /// Inside the early window the backlog is held back too: with no + /// current-slot group at the floor there is nothing to do, and idling is + /// the point — a recursive merge started here would run into the slot's + /// committee aggregation. #[test] - fn snapshot_caps_jobs_at_one_for_proposer() { + fn select_holds_the_backlog_inside_the_early_window() { let store = store_with_competing_build_tier_groups(); - let snapshot = snapshot_aggregation_inputs(&store, 999, 1).expect("should produce a job"); - assert_eq!(snapshot.groups_considered, NUM_GROUPS); - assert_eq!(snapshot.jobs.len(), 1); - assert_eq!( - snapshot.jobs[0].hashed.data().target.slot, + assert!( + select_best_job( + &store, + NUM_GROUPS as u64, + JobPolicy::CommitteeOnly { min_sigs: 3 }, + ) + .is_none(), + "no current-slot group meets the floor, so the worker waits" + ); + } + + /// The current-slot group is taken inside the window as soon as it meets + /// the floor: that is the work the window is kept free for. + #[test] + fn select_takes_the_current_slot_group_at_the_floor() { + let store = store_with_competing_build_tier_groups(); + + let job = select_best_job( + &store, NUM_GROUPS as u64, - "the single job is the best-scoring candidate, not an arbitrary one" + JobPolicy::CommitteeOnly { min_sigs: 2 }, + ) + .expect("the current-slot group meets the floor"); + assert_eq!(job.hashed.data().target.slot, NUM_GROUPS as u64); + } + + /// The same group the floor held back is taken once the boundary opens the + /// policy, however few signatures it holds. + #[test] + fn select_takes_current_slot_group_once_the_policy_opens() { + let store = store_with_competing_build_tier_groups(); + + let job = select_best_job(&store, NUM_GROUPS as u64, JobPolicy::Open) + .expect("should produce a job"); + assert_eq!(job.hashed.data().target.slot, NUM_GROUPS as u64); + } + + /// The floor is two thirds of the votes the node's own subnets are + /// expected to carry, not two thirds of the validator set: with 10 + /// validators over 4 committees, subnets 0 and 1 hold 3 each, so a group + /// gathering both needs 4 of those 6. + #[test] + fn min_current_slot_group_sigs_counts_subscribed_subnets_only() { + let subscribed = HashSet::from([0, 1]); + assert_eq!( + min_current_slot_group_sigs(10, 4, &subscribed), + Some(4), + "ceil(2/3 * (3 + 3))" + ); + + // A subnet past the committee count carries no validators. + assert_eq!( + min_current_slot_group_sigs(10, 4, &HashSet::from([9])), + None + ); + // No committees at all: nothing to expect. + assert_eq!(min_current_slot_group_sigs(10, 0, &subscribed), None); + } + + /// The policy is purely a function of where in the slot we are: backlog + /// work early, committee signatures only inside the early window, and + /// everything from the vote-aggregation boundary to the slot's end. The + /// boundaries follow the configured slot duration; the window ahead of + /// them does not, since it is sized against one leanVM proof. + #[test] + fn job_policy_tightens_into_the_window_and_opens_at_the_boundary() { + let hashes = vec![H256([1u8; 32])]; + let store = new_test_store(make_head_state(0, 10, &hashes)); + let config = WorkerConfig { + attestation_committee_count: 4, + subscribed_subnets: HashSet::from([0, 1]), + }; + // 10 validators over 4 committees: subnets 0 and 1 hold 3 each, so a + // group gathering both needs 4 of those 6. + let min_sigs = 4; + + for milliseconds_per_slot in [DEFAULT_MILLISECONDS_PER_SLOT, 8_000] { + let time_config = ChainConfig::new(1_000, milliseconds_per_slot); + let boundary = vote_aggregation_offset_ms(&time_config); + assert_eq!(boundary, 2 * milliseconds_per_slot / INTERVALS_PER_SLOT); + let window_opens_at = boundary - EARLY_AGGREGATION_WINDOW.as_millis() as u64; + + assert_eq!( + job_policy(0, &time_config, &store, &config), + JobPolicy::Backlog { min_sigs } + ); + assert_eq!( + job_policy(window_opens_at - 1, &time_config, &store, &config), + JobPolicy::Backlog { min_sigs } + ); + assert_eq!( + job_policy(window_opens_at, &time_config, &store, &config), + JobPolicy::CommitteeOnly { min_sigs } + ); + assert_eq!( + job_policy(boundary - 1, &time_config, &store, &config), + JobPolicy::CommitteeOnly { min_sigs } + ); + assert_eq!( + job_policy(boundary, &time_config, &store, &config), + JobPolicy::Open + ); + assert_eq!( + job_policy(milliseconds_per_slot - 1, &time_config, &store, &config), + JobPolicy::Open + ); + } + } + + /// The point of a bitset over a single flag: reasons are independent, so + /// setting one twice is idempotent (the level-driven owner sets its reason + /// on every tick) and releasing one leaves the others holding the worker. + #[test] + fn pause_reasons_are_independent() { + let paused = AtomicU8::new(0); + + set_pause_reason(&paused, PauseReason::Syncing, true); + set_pause_reason(&paused, PauseReason::BlockBuild, true); + set_pause_reason(&paused, PauseReason::Syncing, true); + + // The build finishing leaves the sync gate still holding the worker. + set_pause_reason(&paused, PauseReason::BlockBuild, false); + assert_eq!(paused.load(Ordering::Acquire), PauseReason::Syncing as u8); + + set_pause_reason(&paused, PauseReason::Syncing, false); + assert_eq!(paused.load(Ordering::Acquire), 0); + } + + /// A guard clears its own reason and nothing else. Under the plain bool + /// this replaced, dropping the block-build guard released every reason. + #[test] + fn pause_guard_drop_releases_only_its_own_reason() { + let paused = Arc::new(AtomicU8::new(0)); + set_pause_reason(&paused, PauseReason::Syncing, true); + + { + set_pause_reason(&paused, PauseReason::BlockBuild, true); + let _guard = PauseGuard { + paused: paused.clone(), + reason: PauseReason::BlockBuild, + }; + } + + assert_eq!(paused.load(Ordering::Acquire), PauseReason::Syncing as u8); + } + + /// The store clock owns which slot the worker is in, so the wall-clock + /// offset is always measured against *that* slot and clamped to it. A wall + /// clock lagging the store reads as the start of the store's slot, one + /// running ahead as its end; neither can describe a position inside a + /// different slot, which is what would mis-bucket the slot's own group. + #[test] + fn ms_into_slot_is_measured_against_the_store_slot() { + let time_config = ChainConfig::new(1_000, DEFAULT_MILLISECONDS_PER_SLOT); + let genesis_ms = time_config.genesis_time_ms(); + let slot = 7; + let slot_start_ms = genesis_ms + slot * DEFAULT_MILLISECONDS_PER_SLOT; + + assert_eq!(ms_into_slot(slot_start_ms, slot, &time_config), 0); + assert_eq!( + ms_into_slot(slot_start_ms + 1_234, slot, &time_config), + 1_234 + ); + + // Wall clock a slot and a half behind the store: clamped to the start + // of the store's slot, the permissive `Backlog` end. + let behind = slot_start_ms - DEFAULT_MILLISECONDS_PER_SLOT / 2; + assert_eq!(ms_into_slot(behind, slot, &time_config), 0); + assert_eq!(ms_into_slot(genesis_ms, slot, &time_config), 0); + + // Wall clock past the end of the store's slot: clamped to the slot's + // width, which is at or past the vote-aggregation boundary, so `Open`. + let ahead = slot_start_ms + 3 * DEFAULT_MILLISECONDS_PER_SLOT; + assert_eq!( + ms_into_slot(ahead, slot, &time_config), + DEFAULT_MILLISECONDS_PER_SLOT ); + assert!( + ms_into_slot(ahead, slot, &time_config) >= vote_aggregation_offset_ms(&time_config) + ); + + // Before genesis at all: still the start of slot 0, no underflow. + assert_eq!(ms_into_slot(0, 0, &time_config), 0); } } diff --git a/crates/blockchain/src/block_builder.rs b/crates/blockchain/src/block_builder.rs index f7f4f25e..4cb1128e 100644 --- a/crates/blockchain/src/block_builder.rs +++ b/crates/blockchain/src/block_builder.rs @@ -307,11 +307,11 @@ struct ChainContext<'a> { /// finalized, and the running per-target-root voter set. /// /// Shared by `select_attestations` (block proposal) and -/// `aggregation::snapshot_aggregation_inputs` (interval-2 aggregation) so the -/// two selectors project justification/finalization identically. The -/// aggregator's projection is optimistic (a produced proof is not a processed -/// block), but that only affects the ordering of prover work within the -/// deadline, never the correctness of any produced proof. +/// `aggregation::select_best_job` (the aggregation worker) so the two +/// selectors project justification/finalization identically. The aggregator's +/// projection is optimistic (a produced proof is not a processed block), but +/// that only affects which job the worker picks next, never the correctness of +/// any produced proof. pub(crate) struct ProjectedState { pub(crate) justified_slots: JustifiedSlots, pub(crate) finalized_slot: u64, diff --git a/crates/blockchain/src/lib.rs b/crates/blockchain/src/lib.rs index 8d678e22..e10ba854 100644 --- a/crates/blockchain/src/lib.rs +++ b/crates/blockchain/src/lib.rs @@ -8,24 +8,22 @@ use ethlambda_storage::{ALL_TABLES, Store}; use ethlambda_types::{ ShortRoot, aggregator::AggregatorController, - attestation::{SignedAggregatedAttestation, SignedAttestation}, + attestation::{ + AggregationBits, AttestationData, HashedAttestationData, SignedAggregatedAttestation, + SignedAttestation, bits_is_subset, validator_indices, + }, block::{ByteList512KiB, MultiMessageAggregate, SignedBlock}, chain_config::ChainConfig, primitives::{H256, HashTreeRoot as _}, }; -use crate::aggregation::{ - AggregateProduced, AggregationDeadline, AggregationDone, AggregationSession, - EARLY_AGGREGATION_WINDOW, EarlyAggregationCheck, MAX_AGGREGATION_JOBS, - PRIOR_WORKER_JOIN_TIMEOUT, aggregation_deadline, run_aggregation_worker, -}; +use crate::aggregation::{AggregateProduced, AggregationWorker, PauseReason, WorkerConfig}; use crate::key_manager::ValidatorKeyPair; use crate::sync_status::SyncStatusTracker; use spawned_concurrency::actor; use spawned_concurrency::error::ActorError; use spawned_concurrency::protocol; use spawned_concurrency::tasks::{Actor, ActorRef, ActorStart, Context, Handler, send_after}; -use tokio_util::sync::CancellationToken; use tracing::{debug, error, info, trace, warn}; use crate::block_builder::ProposerConfig; @@ -110,6 +108,23 @@ impl SlotInterval { } } + /// Whether an aggregate the worker finishes during this interval is + /// gossiped on arrival rather than buffered for the vote-aggregation tick. + /// + /// True for the vote-aggregation interval and the one after it. Before + /// them, holding an aggregate to the boundary is the point: publication + /// stays on the interval grid even though proving does not. From the + /// boundary on there is nothing left to wait for, and buffering would hold + /// the aggregate until the *next* slot's boundary, a full slot away. + /// + /// It stops after the safe-target interval because the end-of-slot tick + /// promotes the round's votes: an aggregate finishing at or past that has + /// missed the round it belongs to, and its votes travel in the buffer to + /// the next boundary along with everything else. + pub(crate) fn publishes_aggregates_on_arrival(self) -> bool { + matches!(self, Self::Aggregation | Self::SafeTargetUpdate) + } + /// Milliseconds from genesis to the start of this interval in `slot`. /// /// Inverse of [`Self::from_ms_since_genesis`]. @@ -142,8 +157,62 @@ fn ms_until_next_interval(now_ms: u64, config: &ChainConfig) -> u64 { ms_per_interval - (ms_since_genesis % ms_per_interval) } +/// Upper bound on aggregates named by [`BlockChainServer::pending_aggregates`]. +/// +/// The buffer drains every vote-aggregation interval and normally names a +/// handful: one per attestation data the worker proved, minus what subsumption +/// collapsed. The cap only binds when publication is missed for several slots +/// running, and it bounds both the buffer and the burst that eventually +/// drains it. +const MAX_PENDING_AGGREGATES: usize = 32; + +/// Aggregates the worker produced for one `AttestationData`, waiting for the +/// vote-aggregation interval to gossip them. +struct PendingAggregate { + data: AttestationData, + /// Participant sets naming the proofs to publish, not the proofs + /// themselves: those are in the payload pool already, put there by + /// the worker's own `store_aggregate`, and `Store::proof_for_participants` fetches + /// one back by name at publication. A participant set is a bitfield of at + /// most a few hundred bytes where a proof is up to [`ByteList512KiB`]. + /// + /// Naming a specific proof, rather than asking the pool for the best one + /// under this attestation data, is what keeps this safe: the pool also + /// holds peers' proofs for the same data, and `PayloadBuffer` keeps a + /// peer's disjoint proof alongside ours since neither subsumes the other. + /// Any "pick the best" rule could then publish theirs in place of ours and + /// leave our own subnet's votes off the wire, which for the only + /// aggregator on that subnet means they reach the network from nobody. + /// + /// Kept under `PayloadBuffer`'s own subsumption rule: see [`Self::push`]. + participants: Vec, +} + +impl PendingAggregate { + /// Name one more proof to publish, applying `PayloadBuffer`'s subsumption + /// rule to the participant sets: an incoming set already covered by one we + /// hold (equal included) names a proof the pool has since dropped as + /// redundant, and any set we hold that the incoming one covers is replaced + /// by it. Disjoint sets both stay, since neither carries the other's votes. + /// + /// This mirrors what the pool did to the proofs themselves on insert, so + /// the names we keep stay the ones that still resolve. + fn push(&mut self, participants: AggregationBits) { + let already_covered = self + .participants + .iter() + .any(|held| bits_is_subset(&participants, held)); + if already_covered { + return; + } + self.participants + .retain(|held| !bits_is_subset(held, &participants)); + self.participants.push(participants); + } +} + /// Current UNIX timestamp in milliseconds. -fn unix_now_ms() -> u64 { +pub(crate) fn unix_now_ms() -> u64 { SystemTime::UNIX_EPOCH .elapsed() .expect("already past the unix epoch") @@ -191,7 +260,8 @@ impl BlockChain { pending_blocks: HashMap::new(), aggregator, pending_block_parents: HashMap::new(), - current_aggregation: None, + aggregation_worker: None, + pending_aggregates: HashMap::new(), last_tick_instant: None, attestation_committee_count, subscribed_subnets, @@ -246,25 +316,36 @@ pub struct BlockChainServer { /// `--is-aggregator` flag at spawn. aggregator: AggregatorController, - /// 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, + /// Handle to the always-on aggregation worker. Installed by the actor's + /// `started()` hook and only `None` before it runs, so every read site can + /// treat `None` as "not up yet". + aggregation_worker: Option, + + /// Aggregates the worker produced and not yet gossiped, keyed by + /// attestation data root. The worker stores each one before announcing it, + /// so the pool and its own next selection round already account for it; + /// what waits here is only the gossip publication, held to the + /// vote-aggregation interval so proving runs off the interval grid while + /// publication stays on it. + /// + /// Only names the proofs; see [`PendingAggregate::participants`] for why + /// it names specific ones rather than letting publication pick. An + /// aggregate that finishes inside the publication window skips this buffer + /// entirely (see [`Self::publishes_immediately`]). + pending_aggregates: HashMap, /// Last tick instant for measuring interval duration. last_tick_instant: Option, /// Number of attestation committees (= subnet count). Used by the - /// attestation aggregate coverage emission and the early-aggregation - /// threshold. + /// attestation aggregate coverage emission and handed to the aggregation + /// worker to size its vote-propagation gate. 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. + /// Handed to the aggregation worker to scale its vote-propagation gate. subscribed_subnets: HashSet, /// Proposer-side block-building policy @@ -293,7 +374,7 @@ pub struct BlockChainServer { } impl BlockChainServer { - async fn on_tick(&mut self, timestamp_ms: u64, ctx: &Context) { + async fn on_tick(&mut self, timestamp_ms: u64, _ctx: &Context) { let time_config = *self.store.config(); // Calculate current slot and interval from milliseconds @@ -332,6 +413,20 @@ impl BlockChainServer { metrics::update_current_slot(slot); self.update_sync_status(slot); + // Park the aggregation worker while the sync gate suppresses duties, + // on the same tick that recomputes the verdict it follows. A node that + // is behind would otherwise prove a backlog the network has moved past + // against the same single-threaded prover its block import needs to + // close the gap. + // + // Level-triggered off `duties_allowed`, the one predicate that already + // gates attestations and proposals, so `--disable-duty-sync-gate` keeps + // the worker running for free and the policy stays in one place. + if let Some(worker) = self.aggregation_worker.as_ref() { + let syncing = !self.sync_status.duties_allowed(); + worker.set_paused(PauseReason::Syncing, syncing); + } + // Snapshot the aggregator flag once per tick so all read sites within // the tick see a consistent value even if the admin API toggles it // mid-tick. Mirror it to the gauge from the actor side so @@ -412,36 +507,38 @@ 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(time_config.milliseconds_per_interval()) - - EARLY_AGGREGATION_WINDOW, - ctx.clone(), - EarlyAggregationCheck, - ); - } } // ==== interval 2 ==== SlotInterval::Aggregation => { if is_aggregator { - // 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; + // Sampled at the interval boundary, as before. It now sees + // any aggregate the worker already produced this slot, + // which is the point of proving off the grid. + // + // Aggregator-only, as it was when the session started + // here: the series measures what our own aggregation + // covered by the boundary. On a non-aggregator the same + // call reads `new_payloads` filled by gossip instead, so + // emitting it everywhere would put two different + // populations in one series. + coverage::emit_agg_start_new_coverage( + &self.store, + self.attestation_committee_count, + ); + if !self.sync_status.duties_allowed() { + // We hold the duty but the sync gate is parking the + // worker, so no aggregate was produced this slot. + metrics::inc_aggregator_skipped_not_synced(); } } else { metrics::inc_aggregator_skipped_not_aggregator(); } + + // Proving runs continuously on the worker; this interval is + // only where what it produced reaches the network. + self.publish_pending_aggregates(slot, is_aggregator); + aggregation::refresh_pool_gauges(&self.store); } // ==== interval 3 ==== @@ -465,6 +562,14 @@ impl BlockChainServer { .filter(|_| self.sync_status.duties_allowed()); if let Some(validator_id) = next_proposer { + // Park the aggregation worker for the build: both run + // leanVM proofs, and the block is the one with a deadline. + // The guard clears its own reason on the way out, including + // on `propose_block`'s early returns. + let _pause = self + .aggregation_worker + .as_ref() + .map(|worker| worker.pause(PauseReason::BlockBuild)); self.propose_block(next_slot, validator_id).await; } } @@ -479,182 +584,146 @@ impl BlockChainServer { self.key_manager.advance_keys_to((slot + 1) as u32); } - /// Kick off a committee-signature aggregation session: - /// 1. If a prior session is still running (pathological), warn and join it. - /// 2. Snapshot the aggregation inputs from the store, capped at a single job - /// when we propose next slot. - /// 3. Spawn a `spawn_blocking` worker that streams results back as messages. - /// 4. Schedule the `AggregationDeadline` self-message at one interval out. + /// Whether an aggregate finishing right now should go straight to gossip + /// instead of into [`Self::pending_aggregates`]. /// - /// Both entry points land here — the interval-2 tick and the early - /// 2/3-threshold trigger — so the proposer cap applies to whichever one - /// starts the slot's session. - async fn start_aggregation_session(&mut self, slot: u64, ctx: &Context) { - if let Some(prior) = self.current_aggregation.take() { - prior.cancel.cancel(); - if !prior.worker.is_finished() { - warn!( - prior_session_id = prior.session_id, - new_session_id = slot, - "Prior aggregation worker still running at next session start; joining before proceeding" - ); - } - match tokio::time::timeout(PRIOR_WORKER_JOIN_TIMEOUT, prior.worker).await { - Ok(Ok(())) => {} - Ok(Err(err)) => warn!(?err, "Prior aggregation worker task ended abnormally"), - Err(_) => warn!( - timeout_secs = PRIOR_WORKER_JOIN_TIMEOUT.as_secs(), - "Timed out joining prior aggregation worker" - ), - } - } - - coverage::emit_agg_start_new_coverage(&self.store, self.attestation_committee_count); + /// Publication is normally held to the vote-aggregation interval so + /// aggregates reach peers in the window they expect them in. Inside that + /// interval, or the one after it, the window is now or has just passed: + /// buffering would hold the aggregate until the *next* slot's interval 2 + /// for no benefit, and a receiver can still fold it into this round, whose + /// votes are not promoted until the end-of-slot tick. + /// + /// Read off the store clock, the authority `on_tick` guards on, so this + /// agrees with the tick that would otherwise drain the buffer. During a + /// block build that clock is already at the next slot's interval 0, which + /// answers `false` here, which is what we want: an aggregate finishing + /// then has missed the round. + fn publishes_immediately(&self) -> bool { + let intervals_since_genesis = self.store.time().expect("store time exists"); + SlotInterval::from_intervals_since_genesis(intervals_since_genesis) + .publishes_aggregates_on_arrival() + } - // Limit ourselves to a single round of aggregation if we propose next round. - // This buys us time to build the block before the next slot's interval-0 tick. - let next_proposer = self - .get_our_proposer(slot + 1) - .filter(|_| self.sync_status.duties_allowed()); - let max_jobs = if next_proposer.is_some() { - 1 - } else { - MAX_AGGREGATION_JOBS + /// Gossip one aggregate, counting it in the same arrival series as + /// gossip-received ones. Reports whether it went out. + /// + /// Observed at publication rather than at production: publication is the + /// moment comparable to a peer's arrival, and it is what a receiver would + /// time us on. + fn publish_aggregate( + &self, + aggregate: SignedAggregatedAttestation, + is_aggregator: bool, + publish_ms: u64, + ) -> bool { + let Some(p2p) = self.p2p.as_ref().filter(|_| is_aggregator) else { + return false; }; + metrics::observe_gossip_aggregation_arrival(publish_ms, self.store.config()); + p2p.publish_aggregated_attestation(aggregate) + .inspect_err(|err| error!(%err, "Failed to publish aggregated attestation")) + .is_ok() + } - let Some(snapshot) = aggregation::snapshot_aggregation_inputs(&self.store, slot, max_jobs) - else { - // No current-slot gossip sigs — nothing to aggregate this slot. - return; - }; + /// Note one aggregate the worker produced for publication at the + /// vote-aggregation interval, under [`PendingAggregate::push`]'s + /// subsumption rule, so a straggler signature that re-proves a group + /// replaces the narrower aggregate instead of queueing a second, + /// near-identical one behind it. + /// + /// Past [`MAX_PENDING_AGGREGATES`] the oldest attestation data is dropped, + /// that being the aggregate the network is least waiting on. + fn buffer_aggregate(&mut self, hashed: &HashedAttestationData, participants: AggregationBits) { + self.pending_aggregates + .entry(hashed.root()) + .or_insert_with(|| PendingAggregate { + data: hashed.data().clone(), + participants: Vec::new(), + }) + .push(participants); - let session_id = slot; - let time_config = *self.store.config(); - let t2_ms = time_config.genesis_time_ms() - + SlotInterval::Aggregation.to_ms_since_genesis(slot, &time_config); - // 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" + while self.pending_aggregates_len() > MAX_PENDING_AGGREGATES { + let Some(oldest) = self + .pending_aggregates + .iter() + .min_by_key(|(data_root, entry)| (entry.data.slot, **data_root)) + .map(|(data_root, _)| *data_root) + else { + break; + }; + warn!( + data_root = %ShortRoot(&oldest.0), + "Dropping the oldest buffered aggregate: publication has fallen behind" ); + self.pending_aggregates.remove(&oldest); } + } - // 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`. - let cancel = CancellationToken::new(); - let actor_ref = ctx.actor_ref(); - - 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, - publish_at, - ); - }); - - let _deadline_timer = send_after( - aggregation_deadline(time_config.milliseconds_per_interval()), - ctx.clone(), - AggregationDeadline { session_id }, - ); - - self.current_aggregation = Some(AggregationSession { - session_id, - early, - cancel, - worker, - }); + /// Total aggregates named across every buffered attestation data. + fn pending_aggregates_len(&self) -> usize { + self.pending_aggregates + .values() + .map(|entry| entry.participants.len()) + .sum() } - /// 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 time_config = *self.store.config(); - let Some(ms_since_genesis) = unix_now_ms().checked_sub(time_config.genesis_time_ms()) - else { - return; - }; - let ms_per_interval = time_config.milliseconds_per_interval(); - let ms_into_slot = ms_since_genesis % time_config.milliseconds_per_slot; - let t2_offset = 2 * ms_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 { + /// Gossip every aggregate the worker produced since the last + /// vote-aggregation interval, then clear the buffer. + /// + /// The buffer is drained even when this node has since dropped the + /// aggregator role: those aggregates are already in our own pool, and + /// holding them would leak the buffer until the role came back. + fn publish_pending_aggregates(&mut self, slot: u64, is_aggregator: bool) { + let pending = std::mem::take(&mut self.pending_aggregates); + if pending.is_empty() { return; } - let slot = ms_since_genesis / time_config.milliseconds_per_slot; - if self - .current_aggregation - .as_ref() - .is_some_and(|session| session.session_id == slot) - { + + if !is_aggregator || self.p2p.is_none() { + debug!( + %slot, + count = pending.len(), + is_aggregator, + "Dropping buffered aggregates: nowhere to publish them" + ); 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; + + // Oldest data first, and deterministically: the buffer is a map, whose + // own iteration order is RandomState-seeded. + let mut pending: Vec<(H256, PendingAggregate)> = pending.into_iter().collect(); + pending.sort_unstable_by_key(|(data_root, entry)| (entry.data.slot, *data_root)); + + let publish_ms = unix_now_ms(); + let mut count = 0usize; + + for (data_root, PendingAggregate { data, participants }) in pending { + for participants in participants { + // The pool owns the proof; we kept only enough to name it. A + // miss means the pool let it go, and in every case there is + // nothing left to send: a wider proof subsumed it (a peer's + // wider proof already carries these votes, and they published + // it), or a prune or the buffer cap dropped it. + let Some(proof) = self.store.proof_for_participants(&data_root, &participants) + else { + debug!( + %slot, + data_root = %ShortRoot(&data_root.0), + "Buffered aggregate is no longer in the payload pool" + ); + continue; + }; + let aggregate = SignedAggregatedAttestation { + data: data.clone(), + proof, + }; + if self.publish_aggregate(aggregate, is_aggregator, publish_ms) { + count += 1; + } + } } - info!( - %slot, - max_group, - min_group_sigs, - "Early-aggregation threshold met" - ); - self.start_aggregation_session(slot, ctx).await; + info!(%slot, count, "Published buffered aggregates"); } /// Returns the validator ID if any of our validators is the proposer for this slot. @@ -1368,28 +1437,33 @@ impl BlockChainServer { ); } - /// Actor lifecycle hook: wait for any in-flight aggregation worker to exit - /// before the actor is fully stopped. We cancel the session's token and - /// wait up to PRIOR_WORKER_JOIN_TIMEOUT for the worker's current - /// `aggregate_job` call to finish (the proof itself cannot be interrupted). + /// Actor lifecycle hook: bring up the always-on aggregation worker. + /// + /// It gets its own `Store` clone (same backend, same in-memory buffers), + /// the shared aggregator-role flag so a runtime toggle from the RPC thread + /// reaches it, and the startup-fixed inputs its vote-propagation gate + /// needs. State this actor owns reaches it as a `PauseReason` instead. + #[started] + async fn on_started(&mut self, ctx: &Context) { + self.aggregation_worker = Some(aggregation::spawn_aggregation_worker( + self.store.clone(), + ctx.actor_ref(), + self.aggregator.clone(), + WorkerConfig { + attestation_committee_count: self.attestation_committee_count, + subscribed_subnets: self.subscribed_subnets.clone(), + }, + )); + } + + /// Actor lifecycle hook: wait for the aggregation worker thread to exit + /// before the actor is fully stopped. Cancellation is only observed between + /// jobs, so this waits out the proof in flight (`aggregate_mixed` cannot be + /// interrupted) up to `WORKER_JOIN_TIMEOUT`. #[stopped] async fn on_stopped(&mut self, _ctx: &Context) { - let Some(session) = self.current_aggregation.take() else { - return; - }; - session.cancel.cancel(); - match tokio::time::timeout(PRIOR_WORKER_JOIN_TIMEOUT, session.worker).await { - Ok(Ok(())) => { - info!( - session_id = session.session_id, - "Aggregation worker joined on shutdown" - ); - } - Ok(Err(err)) => warn!(?err, "Aggregation worker task ended abnormally on shutdown"), - Err(_) => warn!( - timeout_secs = PRIOR_WORKER_JOIN_TIMEOUT.as_secs(), - "Timed out joining aggregation worker on shutdown" - ), + if let Some(worker) = self.aggregation_worker.take() { + worker.shutdown().await; } } } @@ -1433,20 +1507,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) { let arrival_ms = unix_now_ms(); let data_slot = msg.attestation.data.slot; if self.is_arrival_observable(data_slot) { metrics::observe_gossip_attestation_arrival(arrival_ms, self.store.config(), data_slot); } + // The stored signature is picked up by the aggregation worker on its + // next selection round; nothing has to be triggered from here. 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.current_slot(); - if data_slot == current_slot { - self.maybe_start_early_aggregation(ctx).await; - } } } @@ -1464,107 +1533,150 @@ impl Handler for BlockChainServer { impl Handler for BlockChainServer { async fn handle(&mut self, msg: AggregateProduced, _ctx: &Context) { - let arrival_ms = unix_now_ms(); + metrics::observe_committee_signatures_aggregation(msg.elapsed); - // 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. - let current = self.current_aggregation.as_ref().map(|s| s.session_id); - if current != Some(msg.session_id) { - trace!( - incoming_session_id = msg.session_id, - current_session_id = ?current, - "Dropping stale aggregate produced for non-current session" - ); - 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. - metrics::observe_gossip_aggregation_arrival(arrival_ms, self.store.config()); - - // 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); + // The worker already stored the proof and consumed the gossip + // signatures behind it; this message only says what to publish. // Surface our own freshly produced aggregate, the counterpart of the // gossip-received path in `on_gossip_aggregated_attestation` (we never // receive our own aggregate back over gossip). Low-rate; proof omitted. self.events.emit(ChainEvent::Aggregate { - participants: msg.output.participants.clone(), - data: msg.output.hashed.data().clone(), + participants: validator_indices(&msg.participants).collect(), + data: msg.hashed.data().clone(), }); - if let Some(ref p2p) = self.p2p { + // Inside the publication window there is nothing to wait for, and + // waiting costs a whole slot; outside it, hold the aggregate for the + // vote-aggregation tick. + if self.publishes_immediately() { + let data_root = msg.hashed.root(); + let Some(proof) = self + .store + .proof_for_participants(&data_root, &msg.participants) + else { + debug!( + data_root = %ShortRoot(&data_root.0), + "Aggregate is no longer in the payload pool; nothing to publish" + ); + return; + }; let aggregate = SignedAggregatedAttestation { - data: msg.output.hashed.data().clone(), - proof: msg.output.proof, + data: msg.hashed.data().clone(), + proof, }; - let _ = p2p - .publish_aggregated_attestation(aggregate) - .inspect_err(|err| error!(%err, "Failed to publish aggregated attestation")); + let is_aggregator = self.aggregator.is_enabled(); + self.publish_aggregate(aggregate, is_aggregator, unix_now_ms()); + } else { + self.buffer_aggregate(&msg.hashed, msg.participants); } } } -impl Handler for BlockChainServer { - async fn handle(&mut self, _msg: EarlyAggregationCheck, ctx: &Context) { - self.maybe_start_early_aggregation(ctx).await; +#[cfg(test)] +mod tests { + use super::*; + + const GENESIS_TIME: u64 = 1_000; + + fn config(milliseconds_per_slot: u64) -> ChainConfig { + ChainConfig::new(GENESIS_TIME, milliseconds_per_slot) } -} -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); + fn bits_for(validators: &[usize]) -> AggregationBits { + let max = validators.iter().copied().max().unwrap_or(0); + let mut bits = AggregationBits::with_length(max + 1).unwrap(); + for &v in validators { + bits.set(v, true).unwrap(); + } + bits + } - 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, - groups_considered = msg.groups_considered, - groups_aggregated = msg.groups_aggregated, - total_raw_sigs = msg.total_raw_sigs, - total_children = msg.total_children, - cancelled = msg.cancelled, - early, - aggregation_deadline_ms = - aggregation_deadline(self.store.config().milliseconds_per_interval()).as_millis() - as u64, - "Committee signatures aggregated" - ); + fn held(entry: &PendingAggregate) -> Vec> { + use ethlambda_types::attestation::validator_indices; + let mut sets: Vec> = entry + .participants + .iter() + .map(|bits| validator_indices(bits).collect()) + .collect(); + sets.sort(); + sets } -} -impl Handler for BlockChainServer { - async fn handle(&mut self, msg: AggregationDeadline, _ctx: &Context) { - if let Some(session) = &self.current_aggregation - && session.session_id == msg.session_id - { - session.cancel.cancel(); + fn pending() -> PendingAggregate { + use ethlambda_types::checkpoint::Checkpoint; + PendingAggregate { + data: AttestationData { + slot: 1, + head: Checkpoint::default(), + target: Checkpoint::default(), + source: Checkpoint::default(), + }, + participants: Vec::new(), } } -} -#[cfg(test)] -mod tests { - use super::*; + /// Which intervals publish an aggregate on arrival instead of buffering + /// it. The window opens at the vote-aggregation boundary, since before it + /// holding the aggregate is the point, and closes after the safe-target + /// interval, since the end-of-slot tick promotes the round's votes. + #[test] + fn immediate_publication_spans_the_aggregation_window() { + let publishes: Vec = (0..INTERVALS_PER_SLOT) + .map(|interval| { + SlotInterval::from_intervals_since_genesis(interval) + .publishes_aggregates_on_arrival() + }) + .collect(); - const GENESIS_TIME: u64 = 1_000; + assert_eq!(publishes, vec![false, false, true, true, false]); - fn config(milliseconds_per_slot: u64) -> ChainConfig { - ChainConfig::new(GENESIS_TIME, milliseconds_per_slot) + // Indexed off the store clock, which counts intervals from genesis + // rather than from the slot, so the window recurs every slot. + assert!( + SlotInterval::from_intervals_since_genesis(7 * INTERVALS_PER_SLOT + 2) + .publishes_aggregates_on_arrival() + ); + } + + /// Re-proving a group after a straggler signature supersedes the narrower + /// aggregate, matching what the payload pool did to the proof itself, so + /// the name we keep is the one that still resolves. + #[test] + fn buffered_aggregate_replaces_the_one_it_covers() { + let mut entry = pending(); + + entry.push(bits_for(&[0, 1, 2])); + entry.push(bits_for(&[0, 1, 2, 3])); + + assert_eq!(held(&entry), vec![vec![0, 1, 2, 3]]); + } + + /// The reverse direction, and equality: neither adds coverage, so neither + /// earns a second publication. + #[test] + fn buffered_aggregate_drops_what_is_already_covered() { + let mut entry = pending(); + + entry.push(bits_for(&[0, 1, 2, 3])); + entry.push(bits_for(&[0, 1])); + entry.push(bits_for(&[0, 1, 2, 3])); + + assert_eq!(held(&entry), vec![vec![0, 1, 2, 3]]); + } + + /// Both survive: a disjoint aggregate carries votes the other does not, so + /// dropping either would keep those votes off the wire. This is why the + /// buffer names specific proofs instead of letting publication pick one + /// per attestation data, where a peer's disjoint proof sits beside ours. + #[test] + fn buffered_disjoint_aggregates_both_survive() { + let mut entry = pending(); + + entry.push(bits_for(&[0, 1, 2])); + entry.push(bits_for(&[5, 6, 7, 8])); + + assert_eq!(held(&entry), vec![vec![0, 1, 2], vec![5, 6, 7, 8]]); } #[test] @@ -1648,13 +1760,4 @@ mod tests { assert_eq!(ms_until_next_interval(genesis_ms - 500, &config), 500); } - - #[test] - fn aggregation_deadline_is_one_interval() { - let config = config(8_000); - assert_eq!( - aggregation_deadline(config.milliseconds_per_interval()), - Duration::from_millis(1_600) - ); - } } diff --git a/crates/blockchain/src/metrics.rs b/crates/blockchain/src/metrics.rs index 7cd8f5d9..31e533f3 100644 --- a/crates/blockchain/src/metrics.rs +++ b/crates/blockchain/src/metrics.rs @@ -297,15 +297,6 @@ 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 = @@ -417,16 +408,6 @@ 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() - }); - /// Buckets clustered just past 0.8 s, the interval width of the default /// 4-second cadence ([`crate::DEFAULT_MILLISECONDS_PER_SLOT`]). Prometheus /// fixes buckets at registration, so a network on a different slot duration @@ -764,11 +745,11 @@ static LEAN_NODE_SYNC_STATUS: std::sync::LazyLock = std::sync::Lazy /// Cross-client label set for `lean_aggregator_skipped_total` (leanMetrics). /// -/// `not_synced`, `missing_state` and `spawn_failed` never fire in ethlambda -/// today: aggregation is not gated on sync status, needs no per-target -/// pre-state resolution, and the `spawn_blocking` worker cannot fail to -/// start. They are seeded at zero so fleet-wide dashboards see the full -/// series. +/// `missing_state` and `spawn_failed` never fire in ethlambda today: +/// aggregation needs no per-target pre-state resolution, and a worker thread +/// that fails to spawn takes the actor's `started()` hook down with it rather +/// than reaching this counter. They are seeded at zero so fleet-wide +/// dashboards see the full series. const AGGREGATOR_SKIP_REASONS: &[&str] = &[ "not_aggregator", "not_synced", @@ -836,7 +817,6 @@ 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); @@ -848,7 +828,6 @@ 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); @@ -893,14 +872,6 @@ 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()); } @@ -1031,24 +1002,35 @@ pub fn observe_aggregated_proof_size(bytes: usize) { LEAN_AGGREGATED_PROOF_SIZE_BYTES.observe(bytes as f64); } -/// Observe committee-signature aggregation duration. Measured in the -/// off-thread worker and reported back via an `AggregationDone` message, so a -/// drop-guard that crosses the thread boundary is not appropriate here. +/// Observe committee-signature aggregation duration: the wall time one +/// aggregate's proof took. Measured on the off-thread worker and reported back +/// with the aggregate itself, so a drop-guard that crosses the thread boundary +/// is not appropriate here. pub fn observe_committee_signatures_aggregation(elapsed: std::time::Duration) { LEAN_COMMITTEE_SIGNATURES_AGGREGATION_TIME_SECONDS.observe(elapsed.as_secs_f64()); } -/// One aggregation cycle (interval-2 tick) skipped because this node has no -/// aggregation duty. Bookkeeping label that lets dashboards separate "no -/// duty" from genuine misses. +/// One vote-aggregation interval passed with this node holding no aggregation +/// duty. Bookkeeping label that lets dashboards separate "no duty" from +/// genuine misses. pub fn inc_aggregator_skipped_not_aggregator() { LEAN_AGGREGATOR_SKIPPED_TOTAL .with_label_values(&["not_aggregator"]) .inc(); } -/// Aggregation jobs dropped without being attempted, e.g. because the -/// session deadline cancelled the worker before it reached them. +/// One vote-aggregation interval passed with this node holding the +/// aggregation duty but the sync gate parking the worker, so nothing was +/// proved. Counted per interval, like `not_aggregator`, rather than per +/// polling round. +pub fn inc_aggregator_skipped_not_synced() { + LEAN_AGGREGATOR_SKIPPED_TOTAL + .with_label_values(&["not_synced"]) + .inc(); +} + +/// Aggregation jobs the worker attempted but could not turn into an +/// aggregate, i.e. the proof itself failed. pub fn inc_aggregator_skipped_other(count: u64) { LEAN_AGGREGATOR_SKIPPED_TOTAL .with_label_values(&["other"]) diff --git a/crates/blockchain/src/store.rs b/crates/blockchain/src/store.rs index 010d8ff6..2ee1bb46 100644 --- a/crates/blockchain/src/store.rs +++ b/crates/blockchain/src/store.rs @@ -354,10 +354,12 @@ pub fn on_tick(store: &mut Store, timestamp_ms: u64, has_proposal: bool) { let should_signal_proposal = has_proposal && is_final_tick; // NOTE: here we assume on_tick never skips intervals. - // Interval 2 (committee-signature aggregation) is no longer handled here: - // the blockchain actor orchestrates the aggregation worker directly so - // the actor's message loop stays unblocked during the expensive XMSS - // proofs. See `BlockChainServer::start_aggregation_session` in `lib.rs`. + // Interval 2 (committee-signature aggregation) is not handled here: + // proving runs continuously on the blockchain actor's off-thread + // aggregation worker, so the actor's message loop stays unblocked + // during the expensive XMSS work, and the interval only carries the + // publication of what the worker produced. See + // `BlockChainServer::publish_pending_aggregates` in `lib.rs`. match interval { SlotInterval::BlockPublication => { // Start of slot - process attestations if proposal exists @@ -369,7 +371,8 @@ pub fn on_tick(store: &mut Store, timestamp_ms: u64, has_proposal: bool) { // Vote propagation — no action } SlotInterval::Aggregation => { - // Aggregation is driven by the actor (off-thread); nothing to do here. + // Aggregation runs on the actor's off-thread worker; nothing to + // do here. } SlotInterval::SafeTargetUpdate => { // Update safe target for validators diff --git a/crates/common/types/src/attestation.rs b/crates/common/types/src/attestation.rs index 91d00105..c0b6a2f1 100644 --- a/crates/common/types/src/attestation.rs +++ b/crates/common/types/src/attestation.rs @@ -152,6 +152,15 @@ pub fn bits_is_subset(a: &AggregationBits, b: &AggregationBits) -> bool { true } +/// Returns `true` iff `a` and `b` mark exactly the same validators. +/// +/// Set equality, not `==`: the same validator set can be carried at different +/// bitlist lengths (one built from `max_id + 1`, another padded further), and +/// `==` compares the encoding rather than the set. +pub fn bits_same_set(a: &AggregationBits, b: &AggregationBits) -> bool { + bits_is_subset(a, b) && bits_is_subset(b, a) +} + /// Aggregated attestation with its signature proof, used for gossip on the aggregation topic. /// /// The `proof` carries a single-message multi-signer aggregate: the signed diff --git a/crates/storage/src/store.rs b/crates/storage/src/store.rs index 2c775c1c..1d4ba5e2 100644 --- a/crates/storage/src/store.rs +++ b/crates/storage/src/store.rs @@ -11,7 +11,7 @@ use ethlambda_crypto::signature::ValidatorSignature; use ethlambda_types::{ attestation::{ AggregatedAttestation, AggregationBits, AttestationData, HashedAttestationData, - bits_is_subset, validator_indices, + bits_is_subset, bits_same_set, validator_indices, }, block::{ Block, BlockBody, BlockHeader, MultiMessageAggregate, SignedBlock, SingleMessageAggregate, @@ -269,6 +269,29 @@ impl PayloadBuffer { self.data.get(data_root).map_or(0, |e| e.proofs.len()) } + /// The proof for `data_root` binding exactly `participants`, cloned. + /// `None` when the buffer holds no such proof, i.e. it was never inserted, + /// or [`Self::push`] dropped it as subsumed by a wider one, or it was + /// evicted or pruned since. + /// + /// Matches on the participant set rather than picking a "best" proof for + /// the root: the buffer holds peers' proofs for the same attestation data + /// alongside ours, and only an exact match identifies the one a caller + /// meant. Clones one proof where [`Self::proofs_for_root`] clones them all, + /// and a proof is up to 512 KiB. + fn proof_for_participants( + &self, + data_root: &H256, + participants: &AggregationBits, + ) -> Option { + self.data + .get(data_root)? + .proofs + .iter() + .find(|proof| bits_same_set(&proof.participants, participants)) + .cloned() + } + /// Return cloned proofs for a given data_root, or empty vec if none. fn proofs_for_root(&self, data_root: &H256) -> Vec { self.data @@ -462,16 +485,6 @@ 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. /// /// Votes are processed newest-first with an equal-slot tie broken toward the @@ -1596,6 +1609,36 @@ impl Store { (new, known) } + /// The proof for `data_root` binding exactly `participants`, from the new + /// buffer or, failing that, the known one. `None` when neither holds it. + /// + /// Lets a caller hold on to the identity of a proof it put in the pool and + /// fetch the proof itself back later, without keeping a copy of the bytes. + /// A `None` is meaningful rather than an error: the proof is gone because + /// a wider one subsumed it, because it was promoted past a prune, or + /// because the buffer's cap evicted it, and in each case there is nothing + /// left for the caller to do with it. + /// + /// Checks the known buffer too, since a proposer's interval-0 promote + /// moves the new buffer wholesale before the vote-aggregation interval. + pub fn proof_for_participants( + &self, + data_root: &H256, + participants: &AggregationBits, + ) -> Option { + let from_new = self + .new_payloads + .lock() + .unwrap() + .proof_for_participants(data_root, participants); + from_new.or_else(|| { + self.known_payloads + .lock() + .unwrap() + .proof_for_participants(data_root, participants) + }) + } + /// Return attestation data entries from the new (pending) payload buffer. /// /// Used to iterate over data that has pending proofs but may lack gossip @@ -1709,15 +1752,6 @@ 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) @@ -2398,6 +2432,82 @@ mod tests { assert_eq!(buf.data[&data_root].proofs.len(), 3); } + /// The lookup an aggregator's publication path depends on: our own proof + /// comes back even while a peer's disjoint proof for the same attestation + /// data sits next to it. Anything that picked a single "best" proof per + /// root could return the peer's and leave our subnet's votes off the wire. + #[test] + fn proof_for_participants_returns_ours_beside_a_disjoint_peer_proof() { + let mut store = Store::test_store(); + let data = make_att_data(1); + let data_root = data.hash_tree_root(); + let ours = make_proof_for_validators(&[0, 1, 2]); + let theirs = make_proof_for_validators(&[5, 6, 7, 8]); + + store.insert_new_aggregated_payload(HashedAttestationData::new(data.clone()), ours.clone()); + store.insert_new_aggregated_payload(HashedAttestationData::new(data), theirs.clone()); + + let found = store + .proof_for_participants(&data_root, &ours.participants) + .expect("our own proof is still in the pool"); + assert_eq!( + validator_indices(&found.participants).collect::>(), + vec![0, 1, 2] + ); + let found = store + .proof_for_participants(&data_root, &theirs.participants) + .expect("the peer's proof is addressable too"); + assert_eq!( + validator_indices(&found.participants).collect::>(), + vec![5, 6, 7, 8] + ); + } + + /// A `None` covers every way the pool can let go of a proof: subsumed by a + /// wider one, never inserted, or promoted into the known buffer, which the + /// lookup follows. + #[test] + fn proof_for_participants_reports_what_the_pool_no_longer_holds() { + let mut store = Store::test_store(); + let data = make_att_data(1); + let data_root = data.hash_tree_root(); + let narrow = make_proof_for_validators(&[0, 1]); + let wide = make_proof_for_validators(&[0, 1, 2]); + + store.insert_new_aggregated_payload( + HashedAttestationData::new(data.clone()), + narrow.clone(), + ); + store.insert_new_aggregated_payload(HashedAttestationData::new(data), wide.clone()); + + // `push` dropped the narrower proof the wider one subsumes. + assert!( + store + .proof_for_participants(&data_root, &narrow.participants) + .is_none() + ); + assert!( + store + .proof_for_participants(&data_root, &wide.participants) + .is_some() + ); + + // The lookup follows a promote out of the new buffer. + store.promote_new_aggregated_payloads(); + assert!( + store + .proof_for_participants(&data_root, &wide.participants) + .is_some() + ); + + let never_inserted = make_proof_for_validators(&[42]); + assert!( + store + .proof_for_participants(&data_root, &never_inserted.participants) + .is_none() + ); + } + #[test] fn payload_buffer_drain_empties_buffer() { let mut buf = PayloadBuffer::new(10); diff --git a/docs/architecture.md b/docs/architecture.md index e15802d4..ed59e183 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -68,8 +68,8 @@ order. That walk and the actor split the duties: | Interval | In `store::on_tick` | In the actor | | --- | --- | --- | | 0 | accept new attestations, if we propose this slot | nothing: the build ran at the previous interval 4 | -| 1 | nothing | produce attestations, arm the early-aggregation check | -| 2 | nothing | start the aggregation session | +| 1 | nothing | produce attestations | +| 2 | nothing | publish the aggregates the worker produced | | 3 | update the safe target | nothing | | 4 | accept accumulated attestations | build and publish the next slot's block | @@ -94,18 +94,29 @@ recomputes the head with [LMD GHOST](./lmd_ghost.md) (`crates/blockchain/fork_ch ### Aggregation off the message loop XMSS proving costs hundreds of milliseconds, so it cannot run on the actor loop: a blocked -actor stops importing blocks. The actor instead snapshots aggregation inputs from the store, -ranks candidates by consensus value, and hands a job list to a `spawn_blocking` worker -(`crates/blockchain/src/aggregation.rs`). The worker holds no store access, streams one -`AggregateProduced` message back per finished job, and ends with `AggregationDone`. The actor -publishes each result on gossip when it arrives. - -A soft deadline cancels the worker through a `CancellationToken`, so an overrunning slot -cannot eat the next one. The session can also start up to `EARLY_AGGREGATION_WINDOW` before -interval 2, once two thirds of the expected signatures are in. Either entry point counts as -the slot's one session, so a slot aggregates once. Starting early buys proving time, not an -earlier publication: the worker holds each finished aggregate until the interval-2 boundary -before delivering it to the actor. +actor stops importing blocks. One worker thread +(`crates/blockchain/src/aggregation.rs`) is spawned when the actor starts and runs for as +long as it does. A plain `std::thread`, not a runtime task: it awaits nothing, it reaches +the actor through an unbounded channel, and it would otherwise hold a blocking-pool thread +for the life of the process. It holds its own `Store` handle — the same backend and the same in-memory +buffers — and loops: rank the pool's candidates by consensus value, prove the best one, hand +it to the actor as an `AggregateProduced` message, rank again. Idle rounds sleep +`WORKER_IDLE_POLL` before re-reading the pool. + +The actor applies each aggregate to the store the moment it arrives, so the pool the worker +re-reads already accounts for it, but buffers the gossip publication until interval 2. That +splits the two concerns the old per-slot session conflated: proving runs whenever there is +work, publication stays on the interval grid. + +What the worker may take up is a function of where the slot is (`JobPolicy`). Early on it +works the backlog — stale groups, merges of proofs already in the pool — and takes a +current-slot group once two thirds of the signatures this node expects are in, so the +slot's votes go out as one wide aggregate rather than several thin ones. Inside the last +600 ms before interval 2 it takes nothing else at all: those backlog jobs are recursive +merges that can run past the boundary, and one of them occupying the single prover would +delay the aggregate the slot is waiting on. From interval 2 on, whatever is in hand is +aggregated. Separately, the actor raises a pause flag around its own block build, since +that competes for the same prover. Block import runs a second, smaller aggregation path. `reaggregate.rs` splits an imported block's merged proof back into per-attestation aggregates and folds them into the local pool, diff --git a/docs/metrics.md b/docs/metrics.md index 061d7da0..2e7ffe9d 100644 --- a/docs/metrics.md +++ b/docs/metrics.md @@ -68,9 +68,11 @@ The exposed metrics follow [the leanMetrics specification](https://github.com/le | `lean_gossip_signatures` | Gauge | Number of gossip signatures in fork-choice store | On gossip signatures update | | | ✅ | | `lean_latest_new_aggregated_payloads` | Gauge | Number of new aggregated payload items | On `latest_new_aggregated_payloads` update | | | ✅ | | `lean_latest_known_aggregated_payloads` | Gauge | Number of known aggregated payload items | On `latest_known_aggregated_payloads` update | | | ✅ | -| `lean_committee_signatures_aggregation_time_seconds` | Histogram | Time taken to aggregate committee signatures | On committee signatures aggregation | | 0.05, 0.1, 0.25, 0.5, 0.75, 1, 2, 3, 4 | ✅ | +| `lean_committee_signatures_aggregation_time_seconds` | Histogram | Wall time one committee-signature aggregate's proof took | On each aggregate the aggregation worker produces | | 0.05, 0.1, 0.25, 0.5, 0.75, 1, 2, 3, 4 | ✅ | | `lean_node_sync_status` | Gauge | Node sync status | On node sync status change | status=idle,syncing,synced | | ✅ | +Three of these changed quantity when aggregation moved to the always-on worker, so thresholds and alerts carried over from before that change are comparing against something else now. `lean_committee_signatures_aggregation_time_seconds` used to time a whole interval-2 session, covering every group that session proved, and now times a single proof, so readings drop accordingly. `lean_aggregator_skipped_total{reason="other"}` counted jobs a cancelled session dropped unattempted, and now counts proofs the worker attempted and failed. `reason="not_synced"`, previously always zero, now fires once per vote-aggregation interval on an aggregator whose worker the sync gate is parking. + ## State Transition Metrics | Name | Type | Usage | Sample collection event | Labels | Buckets | Supported | @@ -148,9 +150,9 @@ Blocks anchor to interval 0 of their own slot and attestations to interval 1 of 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. +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 the same measurement. A local aggregate is sampled when it is published, and since proving runs continuously off the interval grid, publication happens at one of two times: an aggregate finishing outside intervals 2 and 3 is held to the interval-2 boundary and lands in the lowest bucket by construction, while one finishing inside them goes out on arrival and carries its real offset from that boundary. Only the second kind says anything about when a proof finished. A received aggregate still 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. +On an aggregator the distribution therefore carries a structural component: a spike in the lowest bucket, one sample per buffered aggregate, on top of the genuine arrival profile of peers' aggregates and of our own on-arrival publications. Nothing separates them by label, so a slot that buffers several aggregates pulls the whole histogram down. A rising tail is still aggregation cost rather than a slow link, since a late aggregate is late for every node at once. For this node's own proving cost use `lean_pq_sig_aggregated_signatures_building_time_seconds` and `lean_committee_signatures_aggregation_time_seconds` instead. | Name | Type | Usage | Sample collection event | Labels | Buckets | |------|------|-------|-------------------------|--------|---------| @@ -172,7 +174,7 @@ In practice the distribution is bimodal and dominated by production rather than Observability into how many validators/subnets are covered by the attestations the node has aggregated, broken down by pipeline section (the `section` label). The slot is the X-axis. These are sampled roughly once per slot, but emission is gated by the section's source data, so a gauge can retain its previous value: - `timely`, `late`, `block`, `combined` and the `diff_validators` directions are emitted on block import, and **only when the canonical head block carries that round's votes** (otherwise the round is skipped and prior values are kept). -- `agg_start_new` is emitted at interval 2, right before fork-choice aggregation runs. +- `agg_start_new` is emitted at interval 2, and only on a node holding the aggregator role, so the series stays a measurement of what our own aggregation covered by the boundary. On a non-aggregator the same reading would come from `new_payloads` filled by gossip, a different population. - `proposal_combined` is emitted only when this node proposes a block. | Name | Type | Usage | Sample collection event | Labels | diff --git a/docs/slots_and_intervals.md b/docs/slots_and_intervals.md index e658b378..3b01e45f 100644 --- a/docs/slots_and_intervals.md +++ b/docs/slots_and_intervals.md @@ -62,7 +62,8 @@ simply points its parent root at an older block. > **In ethlambda:** block proposal is merged into the previous slot's head-update > interval: the proposer advances its store to the next slot, builds the block there, > and holds publication until the slot boundary. That buys the build one extra interval -> of headroom and leaves no actor work at the block-proposal tick itself. +> of headroom and leaves no actor work at the block-proposal tick itself. The aggregation +> worker is paused for the duration, so the build does not share the prover with it. ## Interval 1: Vote propagation @@ -89,11 +90,18 @@ into one proof is the heaviest recurring computation in the client. It is also w block affordable, since a block carrying raw votes would need one full XMSS signature per voter, quickly going over the network bandwidth limit. -> **In ethlambda:** the proofs run on an off-thread worker so the blockchain actor's -> message loop stays responsive, and a session may start up to `EARLY_AGGREGATION_WINDOW` -> before the interval boundary once two thirds of the signatures are in. At the session's -> soft deadline the actor stops handing out new jobs, but a proof already in flight -> finishes and publishes late rather than being discarded. +> **In ethlambda:** the proofs run on an off-thread worker that never stops: it keeps its +> own store handle and works through the pool a job at a time, so proving is not confined +> to this interval. What the interval still owns is publication — the actor buffers each +> finished aggregate and gossips the lot here. +> +> What the worker may pick up tightens as this boundary approaches. Early in the slot it +> works the backlog (stale groups, merges of proofs it already holds) and takes a +> current-slot group once two thirds of the signatures this node expects are in. In the +> last 600 ms before the boundary it takes nothing but that group: a backlog job is a +> recursive merge that can run well past the boundary, and the prover is single-threaded, +> so starting one there would delay the very aggregate the slot is waiting on. From the +> boundary on, everything is eligible. ## Interval 3: Safe target computation diff --git a/docs/spec_deviations.md b/docs/spec_deviations.md index 8c7c6fa0..48907f61 100644 --- a/docs/spec_deviations.md +++ b/docs/spec_deviations.md @@ -5,16 +5,18 @@ reference in a few places, mainly for performance reasons. This page lists those deviations; each will be fleshed out with rationale, implementation notes, and trade-offs over time. -## Asynchronous signature aggregation with an early start and an early stop +## Continuous signature aggregation, published on the interval grid -Aggregation runs off the main BlockChainServer actor loop, may start before its -interval, and stops early once it runs out of time. +Aggregation is not a per-slot duty in ethlambda: it runs continuously off the main +BlockChainServer actor loop, and only the publication of its results sits on the interval +grid. -- **ethlambda:** the actor snapshots everything aggregation needs (`snapshot_aggregation_inputs`, `crates/blockchain/src/aggregation.rs`) and spawns a `tokio::task::spawn_blocking` worker (`run_aggregation_worker`, `aggregation.rs`). Candidates are the store's gossip-signature groups plus payload-only groups (`new_payload_keys`, which need at least two existing proofs to merge). A tiered greedy selector orders them by consensus value (current-slot before stale, then `Finalize > Justify > Build`, mirroring the block builder) and emits at most `MAX_AGGREGATION_JOBS` jobs, dropping to a single job in the slot before one of our validators proposes. The worker streams each finished group back as an `AggregateProduced` message; the actor loop is never blocked on XMSS work. -- **Early start:** a session normally fires at interval 2, but may start up to `EARLY_AGGREGATION_WINDOW` earlier once the 2/3 signature threshold is already met (`maybe_start_early_aggregation`, `crates/blockchain/src/lib.rs`), so the proof lands earlier in the slot. -- **Early stop:** a `send_after(AGGREGATION_DEADLINE, ...)` timer cancels the session that long after **session start**, so a session that started early also ends early (`AGGREGATION_DEADLINE`, `aggregation.rs`). The worker checks `cancel.is_cancelled()` before each job (`aggregation.rs`); in-flight jobs finish, remaining jobs are dropped. -- **leanSpec:** `aggregate()` is called inline and synchronously from `tick_interval`, at interval 2 only. It walks every attestation data with fresh evidence, with no job cap, no time budget, no worker, and no cancellation. -- **Equivalence:** on cancellation the worker emits only the groups that finished, so a slot may pack fewer aggregates than the synchronous path would; any such subset still yields a valid block, affecting how many votes are included rather than signature validity. The job cap has the same character: it bounds prover work per slot, not what a block may carry. +- **ethlambda:** one worker thread (`spawn_aggregation_worker`, `crates/blockchain/src/aggregation.rs`) is started with the actor and lives as long as it does. It holds its own `Store` handle and loops: pick the single best job (`select_best_job`), prove it, send it to the actor as an `AggregateProduced` message, pick again; an idle round sleeps `WORKER_IDLE_POLL`. Candidates are the store's gossip-signature groups plus payload-only groups (`new_payload_keys`, which need at least two existing proofs to merge), ranked by consensus value (current-slot before stale, then `Finalize > Justify > Build`, mirroring the block builder). The actor loop is never blocked on XMSS work. +- **Deferred publication:** the actor applies each aggregate to its store on arrival — so the pool the worker re-reads accounts for it — but buffers the gossip publication until interval 2 (`publish_pending_aggregates`, `crates/blockchain/src/lib.rs`). Proving therefore happens whenever there is work; the network still only sees aggregates at the vote-aggregation interval. +- **What the worker may take up** is a function of where the slot is (`JobPolicy`, `aggregation.rs`). Early in the slot: backlog work, plus a current-slot group that already holds two thirds of the signatures this node expects (`min_current_slot_group_sigs`), so a slot's votes go out as one wide aggregate instead of several thin ones. Inside the last `EARLY_AGGREGATION_WINDOW` before interval 2: that group and nothing else, since a backlog job is a recursive merge that would occupy the single prover across the boundary and delay the aggregate the slot is waiting on. From interval 2 on: everything, however few signatures back it. +- **Yielding to the block build:** the actor raises the worker's pause flag around `propose_block` (`AggregationWorker::pause`), since both run leanVM proofs and only the block has a deadline. A proof already in flight is not interrupted. +- **leanSpec:** `aggregate()` is called inline and synchronously from `tick_interval`, at interval 2 only. It walks every attestation data with fresh evidence, with no worker, no gate, and no separation between producing an aggregate and publishing it. +- **Equivalence:** the worker produces the same aggregates over a slot, at different times; what a block may carry is unchanged. Where the two can differ is count: a slot whose proving overruns publishes fewer aggregates than the synchronous path would, which affects how many votes are included rather than signature validity. ## Attestation scoring on block building