From 607d19f5994e98b6a7f6f4582c51881c1172b2d8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Fri, 17 Jul 2026 15:38:02 -0300 Subject: [PATCH 1/5] refactor(blockchain): move projection and scoring onto ProjectedState Pull the round-by-round projection and candidate scoring out of select_attestations into ProjectedState methods: from_head_state() seeds it, advance() folds a selected entry's voters and moves justification/finalization forward, and score_entry() scores a candidate from its realized `coverage` against the projection (reading current_votes and finalized_slot from &self rather than as parameters). Widen ProjectedState with its methods, entry_passes_filters, Tier, EntryScore, OrderingKey, and EntryScore::ordering_key to pub(crate). Block-building behavior is unchanged: the method bodies are the former inline logic (advance()'s current_votes update simply moves after the trace it never fed), and pick_best_candidate now unions proof participants into `coverage` before scoring, the exact set the old score_entry derived internally. This exposes the projection and scoring primitives so interval-2 aggregation can reuse them instead of duplicating the logic. --- crates/blockchain/src/block_builder.rs | 296 ++++++++++++++----------- 1 file changed, 163 insertions(+), 133 deletions(-) diff --git a/crates/blockchain/src/block_builder.rs b/crates/blockchain/src/block_builder.rs index 9af5ff83..875c1412 100644 --- a/crates/blockchain/src/block_builder.rs +++ b/crates/blockchain/src/block_builder.rs @@ -202,11 +202,7 @@ fn select_attestations( // Running per-target-root voter set, seeded from state and updated // incrementally as entries are selected. Mirrors the role of Eth2 // participation flags in Prysm/Lighthouse-style packing. - let mut projected = ProjectedState { - justified_slots: head_state.justified_slots.clone(), - finalized_slot: head_state.latest_finalized.slot, - current_votes: build_running_votes(head_state), - }; + let mut projected = ProjectedState::from_head_state(head_state); let mut processed_data_roots: HashSet = HashSet::new(); // A block may carry at most `MAX_ATTESTATIONS_DATA` distinct entries @@ -231,12 +227,6 @@ fn select_attestations( extend_proofs_greedily(proofs, &mut selected, att_data); let target_root = att_data.target.root; - projected - .current_votes - .entry(target_root) - .or_default() - .extend(new_voters); - trace!( tier = ?score.tier, new_voters = score.new_voters, @@ -247,29 +237,7 @@ fn select_attestations( "selected" ); - // Project justification / finalization. Finalize implies Justify - // (target is justified, AND source is finalized). - if score.tier <= Tier::Justify { - justified_slots_ops::extend_to_slot( - &mut projected.justified_slots, - projected.finalized_slot, - att_data.target.slot, - ); - justified_slots_ops::set_justified( - &mut projected.justified_slots, - projected.finalized_slot, - att_data.target.slot, - ); - // Justified target's voter bucket is no longer relevant for - // scoring (no further entry can target it: filter rejects). - projected.current_votes.remove(&target_root); - } - if score.tier == Tier::Finalize { - let new_finalized = att_data.source.slot; - let delta = new_finalized.saturating_sub(projected.finalized_slot) as usize; - justified_slots_ops::shift_window(&mut projected.justified_slots, delta); - projected.finalized_slot = new_finalized; - } + projected.advance(score.tier, att_data, new_voters); } selected @@ -305,13 +273,13 @@ fn pick_best_candidate( continue; } - let Some((score, new_voters)) = score_entry( - att_data, - proofs, - &projected.current_votes, - projected.finalized_slot, - chain.validator_count, - ) else { + let coverage: HashSet = proofs + .iter() + .flat_map(|proof| proof.participant_indices()) + .collect(); + let Some((score, new_voters)) = + projected.score_entry(att_data, &coverage, chain.validator_count) + else { trace_skipped_attestation("zero_new_voters", att_data, data_root); continue; }; @@ -336,13 +304,141 @@ struct ChainContext<'a> { validator_count: usize, } -/// Mutable projection of the post-state that `select_attestations` maintains -/// across rounds: which slots are justified, which slot is finalized, and the -/// running per-target-root voter set. -struct ProjectedState { - justified_slots: JustifiedSlots, - finalized_slot: u64, - current_votes: HashMap>, +/// Mutable projection of the post-state that a tiered greedy selector +/// maintains across rounds: which slots are justified, which slot is +/// 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. +pub(crate) struct ProjectedState { + pub(crate) justified_slots: JustifiedSlots, + pub(crate) finalized_slot: u64, + pub(crate) current_votes: HashMap>, +} + +impl ProjectedState { + /// Seed the projection from the head state: justification/finalization as + /// of the head, and the running voter set derived from the state's + /// justification bitfields (see [`build_running_votes`]). + pub(crate) fn from_head_state(head_state: &State) -> Self { + Self { + justified_slots: head_state.justified_slots.clone(), + finalized_slot: head_state.latest_finalized.slot, + current_votes: build_running_votes(head_state), + } + } + + /// Fold a selected entry into the projection: record its voters under the + /// entry's `target.root`, then advance justification/finalization per + /// `tier` (Finalize implies Justify). `new_voters` is the entry's marginal + /// voter set (block builder) or realized coverage (aggregator); the + /// resulting per-target voter set is the same union either way. + pub(crate) fn advance( + &mut self, + tier: Tier, + att_data: &AttestationData, + new_voters: impl IntoIterator, + ) { + let target_root = att_data.target.root; + self.current_votes + .entry(target_root) + .or_default() + .extend(new_voters); + + // Finalize implies Justify (target is justified, AND source is + // finalized). + if tier <= Tier::Justify { + justified_slots_ops::extend_to_slot( + &mut self.justified_slots, + self.finalized_slot, + att_data.target.slot, + ); + justified_slots_ops::set_justified( + &mut self.justified_slots, + self.finalized_slot, + att_data.target.slot, + ); + // Justified target's voter bucket is no longer relevant for + // scoring (no further entry can target it: filter rejects). + self.current_votes.remove(&target_root); + } + if tier == Tier::Finalize { + let new_finalized = att_data.source.slot; + let delta = new_finalized.saturating_sub(self.finalized_slot) as usize; + justified_slots_ops::shift_window(&mut self.justified_slots, delta); + self.finalized_slot = new_finalized; + } + } + + /// Score a candidate entry from its realized validator `coverage` against + /// this projection. + /// + /// Returns `None` if `coverage` contributes zero validators relative to the + /// running voter set for `att_data.target.root` (no marginal value, drop). + /// On `Some`, the returned `HashSet` is the subset of `coverage` that is new + /// (caller uses it to `advance` the projection without re-scanning + /// `coverage`). A genesis self-vote cannot justify or finalize and is always + /// scored as tier 3. + /// + /// The caller resolves `coverage` and passes it in: block building unions a + /// data's proof participants (see `pick_best_candidate`); committee-signature + /// aggregation passes a job's realized raw + child participants. Keeping the + /// proof->coverage transform at the call site lets both share one scorer + /// without either duplicating the tiering. + pub(crate) fn score_entry( + &self, + att_data: &AttestationData, + coverage: &HashSet, + validator_count: usize, + ) -> Option<(EntryScore, HashSet)> { + let prior_voters = self.current_votes.get(&att_data.target.root); + let prior_count = prior_voters.map_or(0, HashSet::len); + + let new_voters: HashSet = coverage + .iter() + .copied() + .filter(|vid| prior_voters.is_none_or(|prior| !prior.contains(vid))) + .collect(); + if new_voters.is_empty() { + return None; + } + + let total = prior_count + new_voters.len(); + let crosses_2_3 = 3 * total >= 2 * validator_count; + + // 3SF-mini finalization requires the source to lie past the finalized + // boundary (a source at or behind it is already final and must not + // re-finalize) and no slot strictly between source.slot and target.slot + // to still be justifiable (so source and target are consecutive + // justified checkpoints in the projected post-state). Mirrors + // `try_finalize` in the state transition. + let finalizes = crosses_2_3 + && att_data.source.slot > self.finalized_slot + && (att_data.source.slot + 1..att_data.target.slot) + .all(|s| !slot_is_justifiable_after(s, self.finalized_slot)); + + let tier = if is_genesis_self_vote(att_data) || !crosses_2_3 { + Tier::Build + } else if finalizes { + Tier::Finalize + } else { + Tier::Justify + }; + + Some(( + EntryScore { + tier, + new_voters: new_voters.len(), + target_slot: att_data.target.slot, + att_slot: att_data.slot, + }, + new_voters, + )) + } } /// Validate a candidate entry against the projected chain view. @@ -355,7 +451,7 @@ struct ProjectedState { /// slot 0) is exempt from the `target.slot > source.slot` and /// `target_already_justified` checks since fork-choice bootstrapping needs /// it; STF will silently drop it, but it carries fork-choice signal. -fn entry_passes_filters( +pub(crate) fn entry_passes_filters( att_data: &AttestationData, known_block_roots: &HashSet, extended_historical_block_hashes: &[H256], @@ -396,74 +492,6 @@ fn entry_passes_filters( Ok(()) } -/// Score a single candidate entry under the current projected state. -/// -/// Returns `None` if the entry has zero new validators relative to the -/// running voter set for its `target.root` (no marginal value, drop). On -/// `Some`, the returned `HashSet` is the set of new voters contributed by -/// this entry (caller uses it to update the running voter map without -/// re-scanning aggregation bits). A genesis self-vote cannot justify or -/// finalize and is always scored as tier 3. -fn score_entry( - att_data: &AttestationData, - proofs: &[SingleMessageAggregate], - current_votes: &HashMap>, - projected_finalized_slot: u64, - validator_count: usize, -) -> Option<(EntryScore, HashSet)> { - let prior_voters = current_votes.get(&att_data.target.root); - let prior_count = prior_voters.map_or(0, HashSet::len); - - // Collect voters that this entry adds on top of prior_voters. Avoids - // cloning prior_voters; the inner contains() makes this O(participants) - // per candidate per round. `extend_proofs_greedily` selects proofs until - // none contribute new voters, so its final coverage equals this set - // unioned with prior_voters. - let mut new_voters: HashSet = HashSet::new(); - for proof in proofs { - for vid in proof.participant_indices() { - if prior_voters.is_none_or(|prior| !prior.contains(&vid)) { - new_voters.insert(vid); - } - } - } - if new_voters.is_empty() { - return None; - } - - let total = prior_count + new_voters.len(); - let crosses_2_3 = 3 * total >= 2 * validator_count; - - // 3SF-mini finalization requires the source to lie past the finalized - // boundary (a source at or behind it is already final and must not - // re-finalize) and no slot strictly between source.slot and target.slot to - // still be justifiable (so source and target are consecutive justified - // checkpoints in the projected post-state). Mirrors `try_finalize` in the - // state transition. - let finalizes = crosses_2_3 - && att_data.source.slot > projected_finalized_slot - && (att_data.source.slot + 1..att_data.target.slot) - .all(|s| !slot_is_justifiable_after(s, projected_finalized_slot)); - - let tier = if is_genesis_self_vote(att_data) || !crosses_2_3 { - Tier::Build - } else if finalizes { - Tier::Finalize - } else { - Tier::Justify - }; - - Some(( - EntryScore { - tier, - new_voters: new_voters.len(), - target_slot: att_data.target.slot, - att_slot: att_data.slot, - }, - new_voters, - )) -} - /// Selection tier for a candidate `AttestationData` entry. /// /// Declared in priority order: lower variant beats higher under derived @@ -471,7 +499,7 @@ fn score_entry( /// output (`tier = Finalize` is clearer than `tier = 1`). #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] #[repr(u8)] -enum Tier { +pub(crate) enum Tier { /// Applying the entry crosses 2/3 on target AND finalizes the source /// (no slot strictly between source.slot and target.slot is still /// justifiable given projected finalized_slot). @@ -499,23 +527,25 @@ enum Tier { /// /// In both tiers `data_root` (ascending) is the final deterministic tiebreak. #[derive(Debug, Clone, Copy, PartialEq, Eq)] -struct EntryScore { - tier: Tier, - new_voters: usize, +pub(crate) struct EntryScore { + pub(crate) tier: Tier, + pub(crate) new_voters: usize, + /// Read only inside [`EntryScore::ordering_key`]; kept private. target_slot: u64, + /// Read only inside [`EntryScore::ordering_key`]; kept private. att_slot: u64, } /// Total order over candidate entries; the smallest value is the best pick. /// `tier` leads, then three tier-dependent `Reverse`-encoded priorities, then /// `data_root` as the deterministic tiebreak. See [`EntryScore::ordering_key`]. -type OrderingKey = (Tier, Reverse, Reverse, Reverse, H256); +pub(crate) type OrderingKey = (Tier, Reverse, Reverse, Reverse, H256); impl EntryScore { /// Sort key where the smallest tuple is the best candidate. `tier` always /// leads; the remaining three slots carry tier-dependent priorities (see /// the type-level docs), all encoded as `Reverse` so "larger is better". - fn ordering_key(&self, data_root: H256) -> OrderingKey { + pub(crate) fn ordering_key(&self, data_root: H256) -> OrderingKey { let more_new_voters = Reverse(self.new_voters as u64); let newer_target = Reverse(self.target_slot); let newer_att = Reverse(self.att_slot); @@ -902,16 +932,16 @@ mod tests { }; // Supermajority (3 of 4) so the entry crosses 2/3. - let proofs = vec![SingleMessageAggregate::empty(make_bits(&[0, 1, 2]))]; - - let (score, _) = score_entry( - &att_data, - &proofs, - &HashMap::new(), - FINALIZED_SLOT, - NUM_VALIDATORS, - ) - .expect("entry contributes new voters"); + let coverage: HashSet = HashSet::from([0, 1, 2]); + let projected = ProjectedState { + justified_slots: JustifiedSlots::new(), + finalized_slot: FINALIZED_SLOT, + current_votes: HashMap::new(), + }; + + let (score, _) = projected + .score_entry(&att_data, &coverage, NUM_VALIDATORS) + .expect("entry contributes new voters"); assert_eq!( score.tier, From 94c96b420045dbceda67b0b0241825f381b9f652 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Fri, 17 Jul 2026 17:38:00 -0300 Subject: [PATCH 2/5] refactor: move struct to variable --- crates/blockchain/src/block_builder.rs | 16 +++++++--------- 1 file changed, 7 insertions(+), 9 deletions(-) diff --git a/crates/blockchain/src/block_builder.rs b/crates/blockchain/src/block_builder.rs index 875c1412..19bfc07f 100644 --- a/crates/blockchain/src/block_builder.rs +++ b/crates/blockchain/src/block_builder.rs @@ -429,15 +429,13 @@ impl ProjectedState { Tier::Justify }; - Some(( - EntryScore { - tier, - new_voters: new_voters.len(), - target_slot: att_data.target.slot, - att_slot: att_data.slot, - }, - new_voters, - )) + let score = EntryScore { + tier, + new_voters: new_voters.len(), + target_slot: att_data.target.slot, + att_slot: att_data.slot, + }; + Some((score, new_voters)) } } From 9d44c6be313524df21a8f794e6eccf1882a64d13 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Fri, 17 Jul 2026 17:45:34 -0300 Subject: [PATCH 3/5] refactor(blockchain): make entry_passes_filters a ProjectedState method Move entry_passes_filters onto ProjectedState alongside score_entry: it reads justified_slots and finalized_slot from &self and still takes the chain view (known_block_roots, extended_historical_block_hashes) as arguments. Callers become projected.entry_passes_filters(att_data, known_roots, hist). Behavior-preserving. --- crates/blockchain/src/block_builder.rs | 103 ++++++++++++------------- 1 file changed, 51 insertions(+), 52 deletions(-) diff --git a/crates/blockchain/src/block_builder.rs b/crates/blockchain/src/block_builder.rs index 19bfc07f..787c55b4 100644 --- a/crates/blockchain/src/block_builder.rs +++ b/crates/blockchain/src/block_builder.rs @@ -262,12 +262,10 @@ fn pick_best_candidate( if processed_data_roots.contains(data_root) { continue; } - if let Err(reason) = entry_passes_filters( + if let Err(reason) = projected.entry_passes_filters( att_data, chain.known_block_roots, chain.extended_historical_block_hashes, - &projected.justified_slots, - projected.finalized_slot, ) { trace_skipped_attestation(reason, att_data, data_root); continue; @@ -437,57 +435,58 @@ impl ProjectedState { }; Some((score, new_voters)) } -} -/// Validate a candidate entry against the projected chain view. -/// -/// Mirrors `state_transition::is_valid_vote`: the entry's head must be known, -/// its source must be justified, its (source, target) must match the -/// candidate-block chain view, `target.slot > source.slot`, target must not -/// already be justified, and target must be a justifiable slot relative to -/// the projected finalized slot. The genesis self-vote (source == target == -/// slot 0) is exempt from the `target.slot > source.slot` and -/// `target_already_justified` checks since fork-choice bootstrapping needs -/// it; STF will silently drop it, but it carries fork-choice signal. -pub(crate) fn entry_passes_filters( - att_data: &AttestationData, - known_block_roots: &HashSet, - extended_historical_block_hashes: &[H256], - projected_justified_slots: &JustifiedSlots, - projected_finalized_slot: u64, -) -> Result<(), &'static str> { - if !known_block_roots.contains(&att_data.head.root) { - return Err("head_root_unknown"); - } - if !justified_slots_ops::is_slot_justified( - projected_justified_slots, - projected_finalized_slot, - att_data.source.slot, - ) { - return Err("source_not_justified"); - } - if !attestation_data_matches_chain(extended_historical_block_hashes, att_data) { - return Err("chain_mismatch"); - } - let is_genesis_self_vote = is_genesis_self_vote(att_data); - if !is_genesis_self_vote && att_data.target.slot <= att_data.source.slot { - return Err("target_not_after_source"); - } - if !is_genesis_self_vote - && justified_slots_ops::is_slot_justified( - projected_justified_slots, - projected_finalized_slot, - att_data.target.slot, - ) - { - return Err("target_already_justified"); - } - if !is_genesis_self_vote - && !slot_is_justifiable_after(att_data.target.slot, projected_finalized_slot) - { - return Err("target_not_justifiable"); + /// Validate a candidate entry against the projection and the given chain + /// view. + /// + /// Mirrors `state_transition::is_valid_vote`: the entry's head must be + /// known, its source must be justified, its (source, target) must match + /// the candidate-block chain view, `target.slot > source.slot`, target + /// must not already be justified, and target must be a justifiable slot + /// relative to the projected finalized slot. The genesis self-vote + /// (source == target == slot 0) is exempt from the `target.slot > + /// source.slot` and `target_already_justified` checks since fork-choice + /// bootstrapping needs it; STF will silently drop it, but it carries + /// fork-choice signal. + pub(crate) fn entry_passes_filters( + &self, + att_data: &AttestationData, + known_block_roots: &HashSet, + extended_historical_block_hashes: &[H256], + ) -> Result<(), &'static str> { + if !known_block_roots.contains(&att_data.head.root) { + return Err("head_root_unknown"); + } + if !justified_slots_ops::is_slot_justified( + &self.justified_slots, + self.finalized_slot, + att_data.source.slot, + ) { + return Err("source_not_justified"); + } + if !attestation_data_matches_chain(extended_historical_block_hashes, att_data) { + return Err("chain_mismatch"); + } + let is_genesis_self_vote = is_genesis_self_vote(att_data); + if !is_genesis_self_vote && att_data.target.slot <= att_data.source.slot { + return Err("target_not_after_source"); + } + if !is_genesis_self_vote + && justified_slots_ops::is_slot_justified( + &self.justified_slots, + self.finalized_slot, + att_data.target.slot, + ) + { + return Err("target_already_justified"); + } + if !is_genesis_self_vote + && !slot_is_justifiable_after(att_data.target.slot, self.finalized_slot) + { + return Err("target_not_justifiable"); + } + Ok(()) } - Ok(()) } /// Selection tier for a candidate `AttestationData` entry. From cd015618c27d35d16e4a42964dbbc46c1560f2b9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Fri, 17 Jul 2026 17:51:03 -0300 Subject: [PATCH 4/5] feat(aggregation): score interval-2 jobs like the block builder The interval-2 session aggregated current-slot gossip groups in arbitrary order and ignored existing proofs, so it could not prioritize the groups whose aggregation most advances consensus. Rework snapshot_aggregation_inputs into a tiered greedy selector modeled on select_attestations: an up-front store pass resolves each candidate's material once (raw-first + trim), then a loop scores candidates by (current-slot-first, Finalize > Justify > Build) against an optimistically-projected state, emitting at most MAX_AGGREGATION_JOBS jobs. Candidates are filtered by the block builder's entry_passes_filters against a chain view covering [0, head_slot] (head root pushed onto the state's historical_block_hashes, which omits the head's own root), so prover time is spent only on aggregations a block could actually pack. Reuses the block builder's ProjectedState in full (from_head_state, advance, score_entry, and entry_passes_filters), so proposer and aggregator can never drift on the justify/finalize projection, candidate filtering, or tiering. The aggregator validates each candidate with projected.entry_passes_filters and scores its realized raw + child coverage with projected.score_entry. Deletes snapshot_current_slot_aggregation_inputs and build_raw_signature_job (subsumed). Builds on the block_builder refactor (ProjectedState owning projection + scoring). --- Cargo.lock | 2 + crates/blockchain/Cargo.toml | 2 + crates/blockchain/src/aggregation.rs | 1037 ++++++++++++++++++++++---- crates/blockchain/src/lib.rs | 4 +- 4 files changed, 907 insertions(+), 138 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index c57423fa..764020bf 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2048,8 +2048,10 @@ dependencies = [ "ethlambda-test-fixtures", "ethlambda-types", "hex", + "leansig", "libssz", "libssz-types", + "rand 0.10.1", "rayon", "serde", "serde_json", diff --git a/crates/blockchain/Cargo.toml b/crates/blockchain/Cargo.toml index 3a5725d6..3e615ed6 100644 --- a/crates/blockchain/Cargo.toml +++ b/crates/blockchain/Cargo.toml @@ -40,6 +40,8 @@ hex = { workspace = true } libssz.workspace = true libssz-types.workspace = true datatest-stable = "0.3.3" +leansig.workspace = true +rand.workspace = true [[test]] name = "forkchoice_spectests" diff --git a/crates/blockchain/src/aggregation.rs b/crates/blockchain/src/aggregation.rs index 6155946e..b957be7e 100644 --- a/crates/blockchain/src/aggregation.rs +++ b/crates/blockchain/src/aggregation.rs @@ -7,14 +7,23 @@ //! [`run_aggregation_worker`]. The actor stays on its message loop; the worker //! runs the expensive XMSS proofs on a `spawn_blocking` thread and streams //! results back as [`AggregateProduced`] / [`AggregationDone`] messages. - -use std::collections::HashSet; +//! +//! [`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_AGGREGATION_JOBS`] jobs. + +use std::collections::{HashMap, HashSet}; use std::time::{Duration, Instant, SystemTime}; use ethlambda_crypto::aggregate_mixed; use ethlambda_storage::Store; use ethlambda_types::{ - attestation::{AggregationBits, HashedAttestationData}, + ShortRoot, + attestation::{AggregationBits, AttestationData, HashedAttestationData}, block::{ByteList512KiB, SingleMessageAggregate}, primitives::H256, signature::{ValidatorPublicKey, ValidatorSignature}, @@ -23,8 +32,9 @@ use ethlambda_types::{ use spawned_concurrency::message::Message; use spawned_concurrency::tasks::{ActorRef, Context, send_after}; use tokio_util::sync::CancellationToken; -use tracing::{info, warn}; +use tracing::{info, trace, warn}; +use crate::block_builder::{self, EntryScore}; use crate::{MILLISECONDS_PER_INTERVAL, metrics}; /// Soft deadline for committee-signature aggregation measured from session @@ -147,10 +157,78 @@ impl Message for EarlyAggregationCheck { 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. +const MAX_AGGREGATION_JOBS: usize = 3; + +/// One candidate `AttestationData` group with its aggregation material +/// already resolved from the store (see [`resolve_job`]). Built once per +/// session by the up-front store pass in [`snapshot_aggregation_inputs`]; the +/// greedy loop scores and orders these without further store access. +struct Candidate { + hashed: HashedAttestationData, + resolved: ResolvedMaterial, +} + +/// Aggregation material resolved for one candidate, raw-first with redundant +/// raw signatures trimmed (see [`resolve_job`]). Store-free once built; a +/// selected candidate is converted into an [`AggregationJob`] as-is via +/// [`ResolvedMaterial::into_job`]. +struct ResolvedMaterial { + children: Vec<(Vec, ByteList512KiB)>, + accepted_child_ids: Vec, + raw_pubkeys: Vec, + raw_sigs: Vec, + raw_ids: Vec, + keys_to_delete: Vec<(u64, H256)>, + /// Realized coverage (`raw_ids ∪ accepted_child_ids`): the exact + /// validator set the produced proof will attest to. Used for scoring so + /// scores stay consistent with the job actually emitted, instead of the + /// full union of every proof considered. + coverage: HashSet, +} + +impl ResolvedMaterial { + fn into_job(self, hashed: HashedAttestationData) -> AggregationJob { + let slot = hashed.data().slot; + AggregationJob { + hashed, + slot, + children: self.children, + accepted_child_ids: self.accepted_child_ids, + raw_pubkeys: self.raw_pubkeys, + raw_sigs: self.raw_sigs, + raw_ids: self.raw_ids, + keys_to_delete: self.keys_to_delete, + } + } +} + /// 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. -pub fn snapshot_aggregation_inputs(store: &Store) -> Option { +/// +/// A tiered greedy selector modeled on `block_builder::select_attestations`: +/// +/// 1. **Up-front store pass**: resolves every candidate `AttestationData`'s +/// aggregation material once via [`resolve_job`] (raw-first + 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_AGGREGATION_JOBS`] rounds: each round +/// scores every unprocessed candidate against the projected state and +/// keeps the lowest ordering key (current-slot before stale, then +/// Finalize > Justify > Build, mirroring the block builder). The winner's +/// pre-resolved material becomes an [`AggregationJob`]; the projection is +/// updated with its realized coverage. +/// +/// Stops early when no remaining candidate scores (converged). +pub fn snapshot_aggregation_inputs( + store: &Store, + current_slot: u64, +) -> Option { let gossip_groups = store.iter_gossip_signatures(); let new_payload_keys = store.new_payload_keys(); @@ -161,81 +239,110 @@ pub fn snapshot_aggregation_inputs(store: &Store) -> Option let head_state = store.head_state(); let validators = &head_state.validators; - let gossip_roots: HashSet = gossip_groups - .iter() - .map(|(hashed, _)| hashed.root()) - .collect(); + let mut candidates: HashMap = HashMap::new(); - let groups_considered = gossip_groups.len() - + new_payload_keys - .iter() - .filter(|(root, _)| !gossip_roots.contains(root)) - .count(); - - let mut jobs = Vec::with_capacity(groups_considered); - - // Pass 1: attestation data with gossip signatures (may also reuse existing proofs as children). for (hashed, validator_sigs) in &gossip_groups { - if let Some(job) = build_job(store, validators, hashed.clone(), Some(validator_sigs)) { - jobs.push(job); + let data_root = hashed.root(); + let (new_proofs, known_proofs) = store.existing_proofs_for_data(&data_root); + if let Some(resolved) = resolve_job( + data_root, + validator_sigs, + &new_proofs, + &known_proofs, + validators, + ) { + candidates.insert( + data_root, + Candidate { + hashed: hashed.clone(), + resolved, + }, + ); } } - // Pass 2: attestation data with pending proofs but no gossip signatures — pure recursive merge. for (data_root, att_data) in &new_payload_keys { - if gossip_roots.contains(data_root) { + if candidates.contains_key(data_root) { continue; } // Cheap pre-check to skip the expensive `existing_proofs_for_data` clone when - // fewer than 2 proofs are present (merge needs at least 2). + // fewer than 2 proofs are present (a payload-only merge needs at least 2). if store.proof_count_for_data(data_root) < 2 { continue; } - let hashed = HashedAttestationData::new(att_data.clone()); - if let Some(job) = build_job(store, validators, hashed, None) { - jobs.push(job); + let (new_proofs, known_proofs) = store.existing_proofs_for_data(data_root); + if let Some(resolved) = resolve_job(*data_root, &[], &new_proofs, &known_proofs, validators) + { + let hashed = HashedAttestationData::new(att_data.clone()); + candidates.insert(*data_root, Candidate { hashed, resolved }); } } - Some(AggregationSnapshot { - jobs, - groups_considered, - }) -} - -/// Build a snapshot that aggregates *only* the current slot's raw gossip -/// signatures. -/// -/// Unlike [`snapshot_aggregation_inputs`], this deliberately ignores existing -/// proofs and payload-only groups: every job is a pure raw-signature -/// aggregation over the gossip signatures whose `data.slot` matches -/// `current_slot`. Stale-slot gossip groups are left untouched so the -/// interval-2 deadline is never spent re-aggregating past slots. -/// -/// Returns `None` when there is nothing to aggregate for `current_slot` so -/// callers can avoid spawning an empty worker. -pub fn snapshot_current_slot_aggregation_inputs( - store: &Store, - current_slot: u64, -) -> Option { - let gossip_groups = store.iter_gossip_signatures(); - if gossip_groups.is_empty() { + 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` + // only covers [0, head_slot - 1]: `process_block_header` pushes the + // *parent* root, never the block's own root, so the head root at index + // head_slot is absent. We push `store.head()` (the canonical tip, i.e. + // the block `head_state` is the state of) to land it at head_slot, so + // votes for the current head pass `attestation_data_matches_chain`. + // + // Unlike the block builder, which extends by parent_root + empty slots to + // model a *future* candidate block it is about to propose, we validate + // against the current chain: aggregated attestations only reference + // existing blocks (head.slot / target.slot <= head_slot), so no + // empty-slot padding beyond the tip is needed. + let known_block_roots = store.get_block_roots().expect("block roots read works"); + let mut extended_historical_block_hashes: Vec = + 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 mut jobs: Vec = + Vec::with_capacity(MAX_AGGREGATION_JOBS.min(groups_considered)); + for _round in 0..MAX_AGGREGATION_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 head_state = store.head_state(); - let validators = &head_state.validators; + let Candidate { hashed, resolved } = candidates + .remove(&data_root) + .expect("picked candidate exists in pool"); + let att_data = 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" + ); - let mut jobs = Vec::new(); - let mut groups_considered = 0usize; - for (hashed, validator_sigs) in &gossip_groups { - if hashed.data().slot != current_slot { - continue; - } - groups_considered += 1; - if let Some(job) = build_raw_signature_job(validators, hashed.clone(), validator_sigs) { - jobs.push(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, resolved.coverage.iter().copied()); + + jobs.push(resolved.into_job(hashed)); } if jobs.is_empty() { @@ -248,80 +355,104 @@ pub fn snapshot_current_slot_aggregation_inputs( }) } -/// Build one `AggregationJob` for a given attestation data. Returns `None` when -/// there is not enough material for a viable aggregation (no raw sigs and fewer -/// than two children). `validator_sigs` is `None` for Pass 2 (payload-only). -fn build_job( - store: &Store, - validators: &[Validator], - hashed: HashedAttestationData, - validator_sigs: Option<&[(u64, ValidatorSignature)]>, -) -> Option { - let data_root = hashed.root(); - let (new_proofs, known_proofs) = store.existing_proofs_for_data(&data_root); - let (child_proofs, covered) = select_proofs_greedily(&new_proofs, &known_proofs); - - let mut raw_sigs = Vec::new(); - let mut raw_pubkeys = Vec::new(); - let mut raw_ids = Vec::new(); - for (vid, sig) in validator_sigs.into_iter().flatten() { - if covered.contains(vid) { +/// Scan the candidate pool and pick the best-scoring, not-yet-selected entry. +/// +/// Mirrors `block_builder::pick_best_candidate`: skips entries failing +/// `entry_passes_filters` (logging the reason) and those scoring zero new +/// voters (relative to `candidate.resolved.coverage`, not the full proof +/// union — see [`resolve_job`]). Among the rest, returns `(data_root, score)` +/// for the entry with the lowest composite key: current-slot groups precede +/// stale ones, then `EntryScore::ordering_key` (tier, then tier-dependent +/// dims, then `data_root`) decides. +fn pick_best_candidate( + candidates: &HashMap, + projected: &block_builder::ProjectedState, + known_block_roots: &HashSet, + extended_historical_block_hashes: &[H256], + current_slot: u64, + validator_count: usize, +) -> Option<(H256, EntryScore)> { + let mut best: Option<(H256, EntryScore)> = None; + let mut best_key: Option<(u8, block_builder::OrderingKey)> = None; + + for (data_root, candidate) in candidates { + let att_data = candidate.hashed.data(); + if let Err(reason) = projected.entry_passes_filters( + att_data, + known_block_roots, + extended_historical_block_hashes, + ) { + trace_skipped_candidate(reason, att_data, data_root); continue; } - let Some(validator) = validators.get(*vid as usize) else { - continue; - }; - let Ok(pubkey) = validator.get_attestation_pubkey() else { + + let Some((score, _new_voters)) = + projected.score_entry(att_data, &candidate.resolved.coverage, validator_count) + else { + trace_skipped_candidate("zero_new_voters", att_data, data_root); continue; }; - raw_sigs.push(sig.clone()); - raw_pubkeys.push(pubkey); - raw_ids.push(*vid); - } - - let (children, accepted_child_ids) = resolve_child_pubkeys(&child_proofs, validators); - // Skip aggregation when there's nothing to aggregate - if raw_ids.is_empty() && children.len() < 2 { - return None; - } - // Skip aggregation when there's only a single raw signature to aggregate. - if children.is_empty() && raw_ids.len() <= 1 { - return None; + // Current-slot groups always precede stale ones (goal: consider + // current-slot signatures first); within a bucket, `EntryScore` + // decides. + let slot_bucket: u8 = if att_data.slot == current_slot { 0 } else { 1 }; + let candidate_key = candidate_ordering_key(slot_bucket, &score, *data_root); + if best_key.as_ref().is_none_or(|k| candidate_key < *k) { + best = Some((*data_root, score)); + best_key = Some(candidate_key); + } } - let keys_to_delete: Vec<(u64, H256)> = validator_sigs - .into_iter() - .flatten() - .map(|(vid, _)| (*vid, data_root)) - .collect(); + best +} - let slot = hashed.data().slot; - Some(AggregationJob { - hashed, - slot, - children, - accepted_child_ids, - raw_pubkeys, - raw_sigs, - raw_ids, - keys_to_delete, - }) +/// Composite ordering key (lower is better): current-slot groups (`0`) +/// precede stale ones (`1`); within a bucket, `EntryScore::ordering_key` +/// (tier, then tier-dependent dims, then `data_root`) decides. +fn candidate_ordering_key( + slot_bucket: u8, + score: &EntryScore, + data_root: H256, +) -> (u8, block_builder::OrderingKey) { + (slot_bucket, score.ordering_key(data_root)) } -/// Build a raw-signature-only `AggregationJob`: no existing-proof reuse and no -/// children. Every resolvable gossip signature in the group becomes a raw -/// participant. Returns `None` when no signature resolves to a pubkey. -fn build_raw_signature_job( - validators: &[Validator], - hashed: HashedAttestationData, - validator_sigs: &[(u64, ValidatorSignature)], -) -> Option { - let data_root = hashed.root(); +fn trace_skipped_candidate(reason: &'static str, att_data: &AttestationData, data_root: &H256) { + trace!( + reason, + attestation_slot = att_data.slot, + target_slot = att_data.target.slot, + target_root = %ShortRoot(&att_data.target.root.0), + data_root = %ShortRoot(&data_root.0), + "skipped aggregation candidate" + ); +} - let mut raw_sigs = Vec::with_capacity(validator_sigs.len()); - let mut raw_pubkeys = Vec::with_capacity(validator_sigs.len()); - let mut raw_ids = Vec::with_capacity(validator_sigs.len()); +/// Resolve one candidate's aggregation material, raw-first + trim. No store +/// access: the caller passes pre-resolved `(new_proofs, known_proofs)`. +/// +/// 1. Resolves every gossip sig to `(id, pubkey, sig)`; seeds `covered` with +/// their validator ids. +/// 2. Runs [`select_proofs_greedily`] seeded with that `covered` set so a +/// chosen child only adds coverage beyond the raw sigs; capped at +/// [`MAX_AGGREGATION_CHILDREN`]. +/// 3. Trims any raw sig whose validator id ended up in the chosen children's +/// participant union. This is not just an efficiency win: `aggregate_mixed` +/// must never receive a validator both as a raw participant and inside a +/// child (double inclusion corrupts the aggregate), and going raw-first +/// (instead of selecting children first) re-introduces that possibility. +/// +/// Returns `None` when the resulting material is non-viable: no raw sigs and +/// fewer than two children, or a lone raw sig with no children. +fn resolve_job( + data_root: H256, + validator_sigs: &[(u64, ValidatorSignature)], + new_proofs: &[SingleMessageAggregate], + known_proofs: &[SingleMessageAggregate], + validators: &[Validator], +) -> Option { + let mut raw_by_id: HashMap = HashMap::new(); for (vid, sig) in validator_sigs { let Some(validator) = validators.get(*vid as usize) else { continue; @@ -329,32 +460,54 @@ fn build_raw_signature_job( let Ok(pubkey) = validator.get_attestation_pubkey() else { continue; }; + raw_by_id.insert(*vid, (pubkey, sig.clone())); + } + let seed_covered: HashSet = raw_by_id.keys().copied().collect(); + + let (child_proofs, _) = select_proofs_greedily(new_proofs, known_proofs, seed_covered); + let (children, accepted_child_ids) = resolve_child_pubkeys(&child_proofs, validators); + let child_id_set: HashSet = accepted_child_ids.iter().copied().collect(); + + let mut raw_pubkeys = Vec::new(); + let mut raw_sigs = Vec::new(); + let mut raw_ids = Vec::new(); + for (vid, (pubkey, sig)) in &raw_by_id { + if child_id_set.contains(vid) { + continue; + } + raw_pubkeys.push(pubkey.clone()); raw_sigs.push(sig.clone()); - raw_pubkeys.push(pubkey); raw_ids.push(*vid); } - if raw_ids.is_empty() { + // Skip aggregation when there's nothing to aggregate. + if raw_ids.is_empty() && children.len() < 2 { + return None; + } + // Skip aggregation when there's only a single raw signature to aggregate. + if children.is_empty() && raw_ids.len() <= 1 { return None; } + let mut coverage: HashSet = raw_ids.iter().copied().collect(); + coverage.extend(accepted_child_ids.iter().copied()); + // Consume the whole group's gossip signatures on successful aggregation, - // mirroring `build_job`. + // including any trimmed in step 3: their vote is now represented via the + // child that covers them. let keys_to_delete: Vec<(u64, H256)> = validator_sigs .iter() .map(|(vid, _)| (*vid, data_root)) .collect(); - let slot = hashed.data().slot; - Some(AggregationJob { - hashed, - slot, - children: Vec::new(), - accepted_child_ids: Vec::new(), + Some(ResolvedMaterial { + children, + accepted_child_ids, raw_pubkeys, raw_sigs, raw_ids, keys_to_delete, + coverage, }) } @@ -460,15 +613,19 @@ const MAX_AGGREGATION_CHILDREN: usize = 2; /// /// Processes proof sets in priority order (new before known). Within each set, /// repeatedly picks the proof covering the most uncovered validators until no -/// proof adds new coverage. +/// proof adds new coverage. `seed_covered` primes the coverage set before +/// selection starts — [`resolve_job`] seeds it with raw-signature validator +/// ids so a chosen proof is only picked for coverage beyond what raw sigs +/// already provide. /// /// Caps the number of proofs selected at [`MAX_AGGREGATION_CHILDREN`]. fn select_proofs_greedily( new_proofs: &[SingleMessageAggregate], known_proofs: &[SingleMessageAggregate], + seed_covered: HashSet, ) -> (Vec, HashSet) { let mut selected: Vec = Vec::new(); - let mut covered: HashSet = HashSet::new(); + let mut covered: HashSet = seed_covered; for proof_set in [new_proofs, known_proofs] { let mut remaining: Vec<&SingleMessageAggregate> = proof_set.iter().collect(); @@ -635,3 +792,613 @@ pub(crate) fn run_aggregation_worker( cancelled: cancel.is_cancelled(), }); } + +#[cfg(test)] +mod tests { + use super::*; + use ethlambda_storage::backend::InMemoryBackend; + use ethlambda_types::{ + block::{Block, BlockBody, BlockHeader, MultiMessageAggregate, SignedBlock}, + checkpoint::Checkpoint, + state::{ChainConfig, JustificationValidators, JustifiedSlots, State}, + }; + use libssz_types::SszList; + use std::sync::Arc; + + fn make_bits(indices: &[usize]) -> AggregationBits { + let max = indices.iter().copied().max().unwrap_or(0); + let mut bits = AggregationBits::with_length(max + 1).unwrap(); + for &i in indices { + bits.set(i, true).unwrap(); + } + bits + } + + fn make_validators(n: usize) -> Vec { + (0..n) + .map(|i| Validator { + attestation_pubkey: [i as u8; 52], + proposal_pubkey: [i as u8; 52], + index: i as u64, + }) + .collect() + } + + /// A cheap-but-real XMSS signature (tiny lifetime, cached) for tests that + /// only need `ValidatorSignature::from_bytes` to succeed. `resolve_job` + /// never checks signature validity, only that it clones and carries a + /// resolvable id — mirrors `ethlambda_storage::store::tests::make_dummy_sig`. + fn dummy_sig() -> ValidatorSignature { + use ethlambda_types::signature::LeanSignatureScheme; + use leansig::{serialization::Serializable, signature::SignatureScheme}; + use rand::{SeedableRng, rngs::StdRng}; + + static CACHED_SIG: std::sync::LazyLock> = std::sync::LazyLock::new(|| { + let mut rng = StdRng::seed_from_u64(42); + let lifetime = 1 << 5; // small for speed + let (_pk, sk) = LeanSignatureScheme::key_gen(&mut rng, 0, lifetime); + let sig = LeanSignatureScheme::sign(&sk, 0, &[0u8; 32]).unwrap(); + sig.to_bytes() + }); + + ValidatorSignature::from_bytes(&CACHED_SIG).expect("cached test signature") + } + + fn make_head_state(head_slot: u64, num_validators: usize, hashes: &[H256]) -> State { + let head_header = BlockHeader { + slot: head_slot, + proposer_index: 0, + parent_root: H256::ZERO, + state_root: H256::ZERO, + body_root: H256::ZERO, + }; + State { + config: ChainConfig { genesis_time: 1000 }, + slot: head_slot, + latest_block_header: head_header, + latest_justified: Checkpoint::default(), + latest_finalized: Checkpoint::default(), + historical_block_hashes: SszList::try_from(hashes.to_vec()).unwrap(), + justified_slots: JustifiedSlots::new(), + validators: SszList::try_from(make_validators(num_validators)).unwrap(), + justifications_roots: Default::default(), + justifications_validators: JustificationValidators::new(), + } + } + + fn new_test_store(head_state: State) -> Store { + let backend: Arc = Arc::new(InMemoryBackend::new()); + Store::from_anchor_state(backend, head_state) + } + + /// Insert a header-only block at `root` so it shows up in + /// `store.get_block_roots()`. Mirrors the pattern used throughout the + /// blockchain crate's own store tests. + fn insert_test_block(store: &mut Store, root: H256, slot: u64, parent_root: H256) { + let signed_block = SignedBlock { + message: Block { + slot, + proposer_index: 0, + parent_root, + state_root: H256::ZERO, + body: BlockBody::default(), + }, + proof: MultiMessageAggregate::default(), + }; + store + .insert_signed_block(root, signed_block) + .expect("insert test block should succeed"); + } + + // ---- resolve_job ---- + + /// Given gossip sigs for {a,b} and a proof covering {c}, `resolve_job` + /// keeps {a,b} raw and reuses the proof for {c} as a child (c is not + /// covered by any raw sig, so nothing is trimmed). + #[test] + fn resolve_job_prefers_raw_then_fills_missing_coverage() { + let validators = make_validators(5); + let sig = dummy_sig(); + let validator_sigs = vec![(0u64, sig.clone()), (1u64, sig)]; + let proof_c = SingleMessageAggregate::empty(make_bits(&[2])); + let data_root = H256([9u8; 32]); + + let resolved = resolve_job(data_root, &validator_sigs, &[proof_c], &[], &validators) + .expect("raw {0,1} plus a filling child for {2} should be viable"); + + let raw_id_set: HashSet = resolved.raw_ids.iter().copied().collect(); + assert_eq!(raw_id_set, HashSet::from([0, 1]), "both raw sigs kept"); + assert_eq!( + resolved.children.len(), + 1, + "the proof for {{c}} is reused as a child" + ); + assert_eq!(resolved.accepted_child_ids, vec![2]); + assert_eq!(resolved.coverage, HashSet::from([0, 1, 2])); + } + + /// Given gossip sigs for {a,b,c} and a proof covering {c,d,e} (chosen for + /// its new coverage {d,e}), `resolve_job` trims the now-redundant raw sig + /// for c: `aggregate_mixed` must never see a validator both raw and + /// inside a child. Realized coverage still includes c via the child. + #[test] + fn resolve_job_trims_raw_covered_by_chosen_child() { + let validators = make_validators(5); + let sig = dummy_sig(); + let validator_sigs = vec![(0u64, sig.clone()), (1u64, sig.clone()), (2u64, sig)]; + let proof_cde = SingleMessageAggregate::empty(make_bits(&[2, 3, 4])); + let data_root = H256([9u8; 32]); + + let resolved = resolve_job(data_root, &validator_sigs, &[proof_cde], &[], &validators) + .expect("raw {0,1,2} plus a child for {2,3,4} should be viable"); + + let raw_id_set: HashSet = resolved.raw_ids.iter().copied().collect(); + assert_eq!( + raw_id_set, + HashSet::from([0, 1]), + "id 2 is trimmed: it is covered by the chosen child" + ); + assert_eq!(resolved.children.len(), 1); + assert_eq!(resolved.coverage, HashSet::from([0, 1, 2, 3, 4])); + // The whole gossip group (including the trimmed raw sig) is consumed. + assert_eq!(resolved.keys_to_delete.len(), 3); + } + + /// A lone raw signature with no children to merge is not a viable job: + /// aggregating a single signature carries no benefit over gossiping it. + #[test] + fn resolve_job_rejects_lone_raw_signature_with_no_children() { + let validators = make_validators(5); + let validator_sigs = vec![(0u64, dummy_sig())]; + let resolved = resolve_job(H256([9u8; 32]), &validator_sigs, &[], &[], &validators); + assert!(resolved.is_none()); + } + + /// Payload-only candidates (no raw gossip sigs) are viable once at least + /// two existing proofs can be merged. + #[test] + fn resolve_job_allows_payload_only_merge_with_two_children() { + let validators = make_validators(5); + let proof_a = SingleMessageAggregate::empty(make_bits(&[0])); + let proof_b = SingleMessageAggregate::empty(make_bits(&[1])); + + let resolved = resolve_job(H256([9u8; 32]), &[], &[proof_a, proof_b], &[], &validators) + .expect("two children with no raw sigs should be viable"); + + assert!(resolved.raw_ids.is_empty()); + assert_eq!(resolved.children.len(), 2); + assert_eq!(resolved.coverage, HashSet::from([0, 1])); + assert!( + resolved.keys_to_delete.is_empty(), + "nothing to delete from gossip: this candidate has no gossip sigs" + ); + } + + // ---- ordering ---- + + /// The slot bucket dominates the within-bucket score: a current-slot + /// candidate is picked ahead of a stale candidate that has *more* new + /// voters (which, absent the bucket, would win the Build-tier + /// `new_voters` dimension). Exercises `candidate_ordering_key` through the + /// real `pick_best_candidate` path rather than constructing an + /// `EntryScore` directly. + #[test] + fn pick_best_candidate_prefers_current_slot_over_higher_stale_score() { + const NUM_VALIDATORS: usize = 100; + const CURRENT_SLOT: u64 = 3; + + let genesis_root = H256([1u8; 32]); + let target_root = H256([7u8; 32]); + let source = Checkpoint { + root: genesis_root, + slot: 0, + }; + let target = Checkpoint { + root: target_root, + slot: 3, + }; + let head = Checkpoint { + root: genesis_root, + slot: 0, + }; + + // Current-slot candidate: only 1 new voter. + let att_current = AttestationData { + slot: CURRENT_SLOT, + head, + target, + source, + }; + // Stale candidate (different target root so their voter buckets are + // independent): 5 new voters — a strictly better within-Build score. + let stale_target_root = H256([9u8; 32]); + let att_stale = AttestationData { + slot: CURRENT_SLOT - 1, + head, + target: Checkpoint { + root: stale_target_root, + slot: 2, + }, + source, + }; + + let hashed_current = HashedAttestationData::new(att_current); + let hashed_stale = HashedAttestationData::new(att_stale); + let root_current = hashed_current.root(); + let root_stale = hashed_stale.root(); + + let make_resolved = |coverage: HashSet| ResolvedMaterial { + children: Vec::new(), + accepted_child_ids: Vec::new(), + raw_pubkeys: Vec::new(), + raw_sigs: Vec::new(), + raw_ids: coverage.iter().copied().collect(), + keys_to_delete: Vec::new(), + coverage, + }; + + let mut candidates: HashMap = HashMap::new(); + candidates.insert( + root_current, + Candidate { + hashed: hashed_current, + resolved: make_resolved(HashSet::from([0])), + }, + ); + candidates.insert( + root_stale, + Candidate { + hashed: hashed_stale, + resolved: make_resolved(HashSet::from([1, 2, 3, 4, 5])), + }, + ); + + let known_block_roots: HashSet = HashSet::from([genesis_root]); + // Index 0 = genesis (head/source), 2 = stale target, 3 = current target. + let historical_block_hashes = + vec![genesis_root, H256::ZERO, stale_target_root, target_root]; + + let projected = block_builder::ProjectedState { + justified_slots: JustifiedSlots::new(), + finalized_slot: 0, + current_votes: HashMap::new(), + }; + + let (picked_root, score) = pick_best_candidate( + &candidates, + &projected, + &known_block_roots, + &historical_block_hashes, + CURRENT_SLOT, + NUM_VALIDATORS, + ) + .expect("both candidates are viable Build-tier entries"); + + assert_eq!(score.tier, block_builder::Tier::Build); + assert_eq!( + picked_root, root_current, + "the current-slot group must be picked ahead of a stale group with more new voters" + ); + } + + // ---- projection ---- + + /// Two candidates targeting the same root accumulate coverage: the + /// second is re-tiered upward once the first candidate's realized + /// coverage is folded into `current_votes` for that target. + #[test] + fn pick_best_candidate_re_tiers_same_target_after_first_selection() { + const NUM_VALIDATORS: usize = 10; + + let genesis_root = H256([1u8; 32]); + let target_root = H256([7u8; 32]); + let source = Checkpoint { + root: genesis_root, + slot: 0, + }; + let target = Checkpoint { + root: target_root, + slot: 3, + }; + let head = Checkpoint { + root: genesis_root, + slot: 0, + }; + + // A covers 6 validators (Build tier alone: 6*3=18 < 2*10=20). + let att_a = AttestationData { + slot: 1, + head, + target, + source, + }; + // B covers 2 more validators on the SAME target. + let att_b = AttestationData { + slot: 2, + head, + target, + source, + }; + + let hashed_a = HashedAttestationData::new(att_a); + let hashed_b = HashedAttestationData::new(att_b); + let root_a = hashed_a.root(); + let root_b = hashed_b.root(); + + let make_resolved = |coverage: HashSet| ResolvedMaterial { + children: Vec::new(), + accepted_child_ids: Vec::new(), + raw_pubkeys: Vec::new(), + raw_sigs: Vec::new(), + raw_ids: coverage.iter().copied().collect(), + keys_to_delete: Vec::new(), + coverage, + }; + + let mut candidates: HashMap = HashMap::new(); + candidates.insert( + root_a, + Candidate { + hashed: hashed_a, + resolved: make_resolved(HashSet::from([0, 1, 2, 3, 4, 5])), + }, + ); + candidates.insert( + root_b, + Candidate { + hashed: hashed_b, + resolved: make_resolved(HashSet::from([6, 7])), + }, + ); + + let known_block_roots: HashSet = HashSet::from([genesis_root]); + // Index 0 = genesis (head/source), index 3 = target. + let historical_block_hashes = vec![genesis_root, H256::ZERO, H256::ZERO, target_root]; + + let mut projected = block_builder::ProjectedState { + justified_slots: JustifiedSlots::new(), + finalized_slot: 0, + current_votes: HashMap::new(), + }; + + // Round 1: A (6 new voters) outranks B (2 new voters); both Build tier. + let (picked_root, score) = pick_best_candidate( + &candidates, + &projected, + &known_block_roots, + &historical_block_hashes, + 999, + NUM_VALIDATORS, + ) + .expect("round 1 should find a candidate"); + 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. + let winner = candidates.remove(&picked_root).expect("A is in the pool"); + projected + .current_votes + .entry(target_root) + .or_default() + .extend(winner.resolved.coverage.iter().copied()); + + // Round 2: only B remains. Combined with A's now-recorded 6 voters, + // B's 2 new voters cross 2/3 of 10 — B is re-tiered from what would + // have been Build in isolation to Justify. + let (picked_root, score) = pick_best_candidate( + &candidates, + &projected, + &known_block_roots, + &historical_block_hashes, + 999, + NUM_VALIDATORS, + ) + .expect("round 2 should find B"); + assert_eq!(picked_root, root_b); + assert_eq!( + score.tier, + block_builder::Tier::Justify, + "B alone doesn't cross 2/3, but combined with A's prior coverage it does" + ); + } + + // ---- snapshot_aggregation_inputs (full pipeline) ---- + + /// An empty store (no gossip signatures, no pending payloads) has nothing + /// to aggregate. + #[test] + fn snapshot_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).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() { + 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); + + let att_data = AttestationData { + slot: 0, + head: Checkpoint { + root: hashes[0], + slot: 0, + }, + target: Checkpoint { + root: hashes[0], + slot: 0, + }, + source: Checkpoint { + root: hashes[0], + slot: 0, + }, + }; + let hashed = HashedAttestationData::new(att_data); + store.insert_gossip_signature(hashed, 0, dummy_sig()); + + assert!(snapshot_aggregation_inputs(&store, 0).is_none()); + } + + /// A group whose target is already justified (here: at or behind the + /// finalized boundary) can never justify or finalize anything further and + /// must never become a job, even with enough raw sigs to otherwise be + /// viable. + #[test] + fn snapshot_skips_group_whose_target_is_already_justified() { + const NUM_VALIDATORS: usize = 10; + const HEAD_SLOT: u64 = 20; + const FINALIZED_SLOT: u64 = 10; + const TARGET_SLOT: u64 = 5; // <= FINALIZED_SLOT: implicitly justified + + let hashes: Vec = (0..HEAD_SLOT).map(|i| H256([(i + 1) as u8; 32])).collect(); + let mut head_state = make_head_state(HEAD_SLOT, NUM_VALIDATORS, &hashes); + head_state.latest_finalized = Checkpoint { + root: hashes[FINALIZED_SLOT as usize], + slot: FINALIZED_SLOT, + }; + let mut store = new_test_store(head_state); + insert_test_block(&mut store, hashes[0], 0, H256::ZERO); + + let att_data = AttestationData { + slot: TARGET_SLOT, + head: Checkpoint { + root: hashes[0], + slot: 0, + }, + target: Checkpoint { + root: hashes[TARGET_SLOT as usize], + slot: TARGET_SLOT, + }, + source: Checkpoint { + root: hashes[0], + slot: 0, + }, + }; + let hashed = HashedAttestationData::new(att_data); + store.insert_gossip_signature(hashed.clone(), 0, dummy_sig()); + store.insert_gossip_signature(hashed, 1, dummy_sig()); + + assert!( + snapshot_aggregation_inputs(&store, 999).is_none(), + "a group targeting an already-justified slot must never become a job" + ); + } + + /// Regression: a vote for the *current head* (head.slot == target.slot == + /// head_slot) must pass the chain-match filter and become a job. + /// + /// A state's `historical_block_hashes` only covers [0, head_slot - 1] + /// (`process_block_header` pushes the parent root, never the block's own + /// root), so the chain view must be extended by `store.head()` to cover + /// the tip at index head_slot. Without that extension, + /// `attestation_data_matches_chain` rejects any vote whose head/target is + /// the current head (head_slot >= historical_block_hashes.len()), which on + /// a non-genesis chain is nearly every fresh vote: every candidate is + /// filtered as `chain_mismatch` and aggregation produces nothing. + /// + /// 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() { + const NUM_VALIDATORS: usize = 10; + const HEAD_SLOT: u64 = 4; + + // On-chain roots for slots [0, HEAD_SLOT - 1]; the head block's own + // root at HEAD_SLOT is NOT here (it is `store.head()`), mirroring how + // `process_block_header` builds the list. + let hashes: Vec = (0..HEAD_SLOT).map(|i| H256([(i + 1) as u8; 32])).collect(); + let mut store = new_test_store(make_head_state(HEAD_SLOT, NUM_VALIDATORS, &hashes)); + + // The canonical tip: `head_state` is the state at this block, and it + // sits at index HEAD_SLOT once the chain view is extended. + let head_root = store.head().expect("head read works"); + + // Vote whose head AND target are the current head at HEAD_SLOT, with a + // genesis (implicitly justified) source. Justifiable: delta 4 <= 5. + let att_data = AttestationData { + slot: HEAD_SLOT, + head: Checkpoint { + root: head_root, + slot: HEAD_SLOT, + }, + target: Checkpoint { + root: head_root, + slot: HEAD_SLOT, + }, + source: Checkpoint { + root: hashes[0], + slot: 0, + }, + }; + let hashed = HashedAttestationData::new(att_data); + // Two raw sigs so the group is viable (a lone raw sig is dropped). + 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) + .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, + HEAD_SLOT, + "the job aggregates the vote targeting the current head" + ); + } + + /// With more scoring candidates than `MAX_AGGREGATION_JOBS`, exactly that + /// many jobs are produced — the best `MAX_AGGREGATION_JOBS` by ordering + /// key. Five Build-tier candidates (2 raw sigs each, well under the 2/3 + /// threshold) differ only by `target_slot`; Build-tier ordering prefers + /// larger `target_slot` on a new_voters tie, so the top three by slot win. + #[test] + fn snapshot_caps_jobs_at_max_aggregation_jobs() { + const NUM_VALIDATORS: usize = 10; + const HEAD_SLOT: u64 = 10; + const NUM_GROUPS: usize = 5; + + let hashes: Vec = (0..HEAD_SLOT).map(|i| H256([(i + 1) as u8; 32])).collect(); + let mut store = new_test_store(make_head_state(HEAD_SLOT, NUM_VALIDATORS, &hashes)); + insert_test_block(&mut store, hashes[0], 0, H256::ZERO); + + for i in 0..NUM_GROUPS { + let target_slot = i as u64 + 1; // 1..=5, all justifiable (delta <= 5) + let att_data = AttestationData { + slot: target_slot, + head: Checkpoint { + root: hashes[0], + slot: 0, + }, + target: Checkpoint { + root: hashes[target_slot as usize], + slot: target_slot, + }, + source: Checkpoint { + root: hashes[0], + slot: 0, + }, + }; + let hashed = HashedAttestationData::new(att_data); + // Distinct validator pair per group so groups don't interact. + store.insert_gossip_signature(hashed.clone(), (2 * i) as u64, dummy_sig()); + store.insert_gossip_signature(hashed, (2 * i + 1) as u64, dummy_sig()); + } + + let snapshot = snapshot_aggregation_inputs(&store, 999).expect("should produce jobs"); + assert_eq!(snapshot.groups_considered, NUM_GROUPS); + assert_eq!(snapshot.jobs.len(), MAX_AGGREGATION_JOBS); + + let selected_targets: HashSet = snapshot + .jobs + .iter() + .map(|job| job.hashed.data().target.slot) + .collect(); + assert_eq!( + selected_targets, + HashSet::from([3, 4, 5]), + "the three highest target_slot groups win the new_voters tie" + ); + } +} diff --git a/crates/blockchain/src/lib.rs b/crates/blockchain/src/lib.rs index cdb4fb8c..d981d5f1 100644 --- a/crates/blockchain/src/lib.rs +++ b/crates/blockchain/src/lib.rs @@ -465,9 +465,7 @@ impl BlockChainServer { coverage::emit_agg_start_new_coverage(&self.store, self.attestation_committee_count); - let Some(snapshot) = - aggregation::snapshot_current_slot_aggregation_inputs(&self.store, slot) - else { + let Some(snapshot) = aggregation::snapshot_aggregation_inputs(&self.store, slot) else { // No current-slot gossip sigs — nothing to aggregate this slot. return; }; From 107ce90c24d205145845dfaeb25cc7c68fb81798 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Mon, 20 Jul 2026 16:17:06 -0300 Subject: [PATCH 5/5] refactor(aggregation): fold Candidate and ResolvedMaterial into AggregationJob MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The interval-2 selector kept a separate Candidate (hashed + ResolvedMaterial) during scoring and converted the winner into an AggregationJob only at the end. AggregationJob already carries `hashed` plus every material field, and the one extra ResolvedMaterial field, `coverage`, is exactly `raw_ids ∪ accepted_child_ids`. resolve_job now builds an AggregationJob directly and `coverage` is a derived method on it, so the selector scores and emits a single type end to end. --- crates/blockchain/src/aggregation.rs | 251 ++++++++++++--------------- 1 file changed, 115 insertions(+), 136 deletions(-) diff --git a/crates/blockchain/src/aggregation.rs b/crates/blockchain/src/aggregation.rs index b957be7e..2a62b2a6 100644 --- a/crates/blockchain/src/aggregation.rs +++ b/crates/blockchain/src/aggregation.rs @@ -84,6 +84,21 @@ pub struct AggregationJob { pub(crate) keys_to_delete: Vec<(u64, H256)>, } +impl AggregationJob { + /// Realized coverage (`raw_ids ∪ accepted_child_ids`): the exact validator + /// set the produced proof will attest to. Used for scoring during + /// selection so scores stay consistent with the job actually emitted, + /// instead of the full union of every proof considered. Derived on demand: + /// the fields it unions are already carried by the job. + fn coverage(&self) -> HashSet { + self.raw_ids + .iter() + .copied() + .chain(self.accepted_child_ids.iter().copied()) + .collect() + } +} + /// All input needed to run a session of committee-signature aggregation off-thread. pub struct AggregationSnapshot { pub(crate) jobs: Vec, @@ -163,66 +178,24 @@ impl Message for EarlyAggregationCheck { /// scoring candidates remain. const MAX_AGGREGATION_JOBS: usize = 3; -/// One candidate `AttestationData` group with its aggregation material -/// already resolved from the store (see [`resolve_job`]). Built once per -/// session by the up-front store pass in [`snapshot_aggregation_inputs`]; the -/// greedy loop scores and orders these without further store access. -struct Candidate { - hashed: HashedAttestationData, - resolved: ResolvedMaterial, -} - -/// Aggregation material resolved for one candidate, raw-first with redundant -/// raw signatures trimmed (see [`resolve_job`]). Store-free once built; a -/// selected candidate is converted into an [`AggregationJob`] as-is via -/// [`ResolvedMaterial::into_job`]. -struct ResolvedMaterial { - children: Vec<(Vec, ByteList512KiB)>, - accepted_child_ids: Vec, - raw_pubkeys: Vec, - raw_sigs: Vec, - raw_ids: Vec, - keys_to_delete: Vec<(u64, H256)>, - /// Realized coverage (`raw_ids ∪ accepted_child_ids`): the exact - /// validator set the produced proof will attest to. Used for scoring so - /// scores stay consistent with the job actually emitted, instead of the - /// full union of every proof considered. - coverage: HashSet, -} - -impl ResolvedMaterial { - fn into_job(self, hashed: HashedAttestationData) -> AggregationJob { - let slot = hashed.data().slot; - AggregationJob { - hashed, - slot, - children: self.children, - accepted_child_ids: self.accepted_child_ids, - raw_pubkeys: self.raw_pubkeys, - raw_sigs: self.raw_sigs, - raw_ids: self.raw_ids, - keys_to_delete: self.keys_to_delete, - } - } -} - /// 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. /// /// A tiered greedy selector modeled on `block_builder::select_attestations`: /// -/// 1. **Up-front store pass**: resolves every candidate `AttestationData`'s -/// aggregation material once via [`resolve_job`] (raw-first + 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). +/// 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 +/// (`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_AGGREGATION_JOBS`] rounds: each round -/// scores every unprocessed candidate against the projected state and +/// 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 winner's -/// pre-resolved material becomes an [`AggregationJob`]; the projection is -/// updated with its realized coverage. +/// 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). pub fn snapshot_aggregation_inputs( @@ -239,25 +212,19 @@ pub fn snapshot_aggregation_inputs( let head_state = store.head_state(); let validators = &head_state.validators; - let mut candidates: HashMap = HashMap::new(); + let mut candidates: HashMap = HashMap::new(); for (hashed, validator_sigs) in &gossip_groups { let data_root = hashed.root(); let (new_proofs, known_proofs) = store.existing_proofs_for_data(&data_root); - if let Some(resolved) = resolve_job( - data_root, + if let Some(job) = resolve_job( + hashed.clone(), validator_sigs, &new_proofs, &known_proofs, validators, ) { - candidates.insert( - data_root, - Candidate { - hashed: hashed.clone(), - resolved, - }, - ); + candidates.insert(data_root, job); } } @@ -271,10 +238,9 @@ pub fn snapshot_aggregation_inputs( continue; } let (new_proofs, known_proofs) = store.existing_proofs_for_data(data_root); - if let Some(resolved) = resolve_job(*data_root, &[], &new_proofs, &known_proofs, validators) - { - let hashed = HashedAttestationData::new(att_data.clone()); - candidates.insert(*data_root, Candidate { hashed, resolved }); + let hashed = HashedAttestationData::new(att_data.clone()); + if let Some(job) = resolve_job(hashed, &[], &new_proofs, &known_proofs, validators) { + candidates.insert(*data_root, job); } } @@ -321,10 +287,11 @@ pub fn snapshot_aggregation_inputs( break; }; - let Candidate { hashed, resolved } = candidates + let job = candidates .remove(&data_root) .expect("picked candidate exists in pool"); - let att_data = hashed.data(); + let coverage = job.coverage(); + let att_data = job.hashed.data(); let target_root = att_data.target.root; let target_slot = att_data.target.slot; @@ -340,9 +307,9 @@ pub fn snapshot_aggregation_inputs( // 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, resolved.coverage.iter().copied()); + projected.advance(score.tier, att_data, coverage.iter().copied()); - jobs.push(resolved.into_job(hashed)); + jobs.push(job); } if jobs.is_empty() { @@ -359,13 +326,13 @@ pub fn snapshot_aggregation_inputs( /// /// Mirrors `block_builder::pick_best_candidate`: skips entries failing /// `entry_passes_filters` (logging the reason) and those scoring zero new -/// voters (relative to `candidate.resolved.coverage`, not the full proof -/// union — see [`resolve_job`]). Among the rest, returns `(data_root, score)` -/// for the entry with the lowest composite key: current-slot groups precede -/// stale ones, then `EntryScore::ordering_key` (tier, then tier-dependent -/// dims, then `data_root`) decides. +/// voters (relative to the candidate's realized [`AggregationJob::coverage`], +/// not the full proof union — see [`resolve_job`]). Among the rest, returns +/// `(data_root, score)` for the entry with the lowest composite key: +/// current-slot groups precede stale ones, then `EntryScore::ordering_key` +/// (tier, then tier-dependent dims, then `data_root`) decides. fn pick_best_candidate( - candidates: &HashMap, + candidates: &HashMap, projected: &block_builder::ProjectedState, known_block_roots: &HashSet, extended_historical_block_hashes: &[H256], @@ -387,7 +354,7 @@ fn pick_best_candidate( } let Some((score, _new_voters)) = - projected.score_entry(att_data, &candidate.resolved.coverage, validator_count) + projected.score_entry(att_data, &candidate.coverage(), validator_count) else { trace_skipped_candidate("zero_new_voters", att_data, data_root); continue; @@ -446,12 +413,13 @@ fn trace_skipped_candidate(reason: &'static str, att_data: &AttestationData, dat /// Returns `None` when the resulting material is non-viable: no raw sigs and /// fewer than two children, or a lone raw sig with no children. fn resolve_job( - data_root: H256, + hashed: HashedAttestationData, validator_sigs: &[(u64, ValidatorSignature)], new_proofs: &[SingleMessageAggregate], known_proofs: &[SingleMessageAggregate], validators: &[Validator], -) -> Option { +) -> Option { + let data_root = hashed.root(); let mut raw_by_id: HashMap = HashMap::new(); for (vid, sig) in validator_sigs { let Some(validator) = validators.get(*vid as usize) else { @@ -489,9 +457,6 @@ fn resolve_job( return None; } - let mut coverage: HashSet = raw_ids.iter().copied().collect(); - coverage.extend(accepted_child_ids.iter().copied()); - // Consume the whole group's gossip signatures on successful aggregation, // including any trimmed in step 3: their vote is now represented via the // child that covers them. @@ -500,14 +465,16 @@ fn resolve_job( .map(|(vid, _)| (*vid, data_root)) .collect(); - Some(ResolvedMaterial { + let slot = hashed.data().slot; + Some(AggregationJob { + hashed, + slot, children, accepted_child_ids, raw_pubkeys, raw_sigs, raw_ids, keys_to_delete, - coverage, }) } @@ -844,6 +811,18 @@ mod tests { ValidatorSignature::from_bytes(&CACHED_SIG).expect("cached test signature") } + /// A `HashedAttestationData` over default (all-zero) data for `resolve_job` + /// tests, which never inspect the attestation data itself — only the raw + /// sigs / children / coverage the resulting job carries. + fn dummy_hashed() -> HashedAttestationData { + HashedAttestationData::new(AttestationData { + slot: 0, + head: Checkpoint::default(), + target: Checkpoint::default(), + source: Checkpoint::default(), + }) + } + fn make_head_state(head_slot: u64, num_validators: usize, hashes: &[H256]) -> State { let head_header = BlockHeader { slot: head_slot, @@ -901,10 +880,15 @@ mod tests { let sig = dummy_sig(); let validator_sigs = vec![(0u64, sig.clone()), (1u64, sig)]; let proof_c = SingleMessageAggregate::empty(make_bits(&[2])); - let data_root = H256([9u8; 32]); - let resolved = resolve_job(data_root, &validator_sigs, &[proof_c], &[], &validators) - .expect("raw {0,1} plus a filling child for {2} should be viable"); + let resolved = resolve_job( + dummy_hashed(), + &validator_sigs, + &[proof_c], + &[], + &validators, + ) + .expect("raw {0,1} plus a filling child for {2} should be viable"); let raw_id_set: HashSet = resolved.raw_ids.iter().copied().collect(); assert_eq!(raw_id_set, HashSet::from([0, 1]), "both raw sigs kept"); @@ -914,7 +898,7 @@ mod tests { "the proof for {{c}} is reused as a child" ); assert_eq!(resolved.accepted_child_ids, vec![2]); - assert_eq!(resolved.coverage, HashSet::from([0, 1, 2])); + assert_eq!(resolved.coverage(), HashSet::from([0, 1, 2])); } /// Given gossip sigs for {a,b,c} and a proof covering {c,d,e} (chosen for @@ -927,10 +911,15 @@ mod tests { let sig = dummy_sig(); let validator_sigs = vec![(0u64, sig.clone()), (1u64, sig.clone()), (2u64, sig)]; let proof_cde = SingleMessageAggregate::empty(make_bits(&[2, 3, 4])); - let data_root = H256([9u8; 32]); - let resolved = resolve_job(data_root, &validator_sigs, &[proof_cde], &[], &validators) - .expect("raw {0,1,2} plus a child for {2,3,4} should be viable"); + let resolved = resolve_job( + dummy_hashed(), + &validator_sigs, + &[proof_cde], + &[], + &validators, + ) + .expect("raw {0,1,2} plus a child for {2,3,4} should be viable"); let raw_id_set: HashSet = resolved.raw_ids.iter().copied().collect(); assert_eq!( @@ -939,7 +928,7 @@ mod tests { "id 2 is trimmed: it is covered by the chosen child" ); assert_eq!(resolved.children.len(), 1); - assert_eq!(resolved.coverage, HashSet::from([0, 1, 2, 3, 4])); + assert_eq!(resolved.coverage(), HashSet::from([0, 1, 2, 3, 4])); // The whole gossip group (including the trimmed raw sig) is consumed. assert_eq!(resolved.keys_to_delete.len(), 3); } @@ -950,7 +939,7 @@ mod tests { fn resolve_job_rejects_lone_raw_signature_with_no_children() { let validators = make_validators(5); let validator_sigs = vec![(0u64, dummy_sig())]; - let resolved = resolve_job(H256([9u8; 32]), &validator_sigs, &[], &[], &validators); + let resolved = resolve_job(dummy_hashed(), &validator_sigs, &[], &[], &validators); assert!(resolved.is_none()); } @@ -962,12 +951,12 @@ mod tests { let proof_a = SingleMessageAggregate::empty(make_bits(&[0])); let proof_b = SingleMessageAggregate::empty(make_bits(&[1])); - let resolved = resolve_job(H256([9u8; 32]), &[], &[proof_a, proof_b], &[], &validators) + let resolved = resolve_job(dummy_hashed(), &[], &[proof_a, proof_b], &[], &validators) .expect("two children with no raw sigs should be viable"); assert!(resolved.raw_ids.is_empty()); assert_eq!(resolved.children.len(), 2); - assert_eq!(resolved.coverage, HashSet::from([0, 1])); + assert_eq!(resolved.coverage(), HashSet::from([0, 1])); assert!( resolved.keys_to_delete.is_empty(), "nothing to delete from gossip: this candidate has no gossip sigs" @@ -1027,30 +1016,25 @@ mod tests { let root_current = hashed_current.root(); let root_stale = hashed_stale.root(); - let make_resolved = |coverage: HashSet| ResolvedMaterial { - children: Vec::new(), - accepted_child_ids: Vec::new(), - raw_pubkeys: Vec::new(), - raw_sigs: Vec::new(), - raw_ids: coverage.iter().copied().collect(), - keys_to_delete: Vec::new(), - coverage, + let make_job = |hashed: HashedAttestationData, coverage: HashSet| { + let slot = hashed.data().slot; + AggregationJob { + hashed, + slot, + children: Vec::new(), + accepted_child_ids: Vec::new(), + raw_pubkeys: Vec::new(), + raw_sigs: Vec::new(), + raw_ids: coverage.into_iter().collect(), + keys_to_delete: Vec::new(), + } }; - let mut candidates: HashMap = HashMap::new(); - candidates.insert( - root_current, - Candidate { - hashed: hashed_current, - resolved: make_resolved(HashSet::from([0])), - }, - ); + let mut candidates: HashMap = HashMap::new(); + candidates.insert(root_current, make_job(hashed_current, HashSet::from([0]))); candidates.insert( root_stale, - Candidate { - hashed: hashed_stale, - resolved: make_resolved(HashSet::from([1, 2, 3, 4, 5])), - }, + make_job(hashed_stale, HashSet::from([1, 2, 3, 4, 5])), ); let known_block_roots: HashSet = HashSet::from([genesis_root]); @@ -1125,31 +1109,26 @@ mod tests { let root_a = hashed_a.root(); let root_b = hashed_b.root(); - let make_resolved = |coverage: HashSet| ResolvedMaterial { - children: Vec::new(), - accepted_child_ids: Vec::new(), - raw_pubkeys: Vec::new(), - raw_sigs: Vec::new(), - raw_ids: coverage.iter().copied().collect(), - keys_to_delete: Vec::new(), - coverage, + let make_job = |hashed: HashedAttestationData, coverage: HashSet| { + let slot = hashed.data().slot; + AggregationJob { + hashed, + slot, + children: Vec::new(), + accepted_child_ids: Vec::new(), + raw_pubkeys: Vec::new(), + raw_sigs: Vec::new(), + raw_ids: coverage.into_iter().collect(), + keys_to_delete: Vec::new(), + } }; - let mut candidates: HashMap = HashMap::new(); + let mut candidates: HashMap = HashMap::new(); candidates.insert( root_a, - Candidate { - hashed: hashed_a, - resolved: make_resolved(HashSet::from([0, 1, 2, 3, 4, 5])), - }, - ); - candidates.insert( - root_b, - Candidate { - hashed: hashed_b, - resolved: make_resolved(HashSet::from([6, 7])), - }, + make_job(hashed_a, HashSet::from([0, 1, 2, 3, 4, 5])), ); + candidates.insert(root_b, make_job(hashed_b, HashSet::from([6, 7]))); let known_block_roots: HashSet = HashSet::from([genesis_root]); // Index 0 = genesis (head/source), index 3 = target. @@ -1180,7 +1159,7 @@ mod tests { .current_votes .entry(target_root) .or_default() - .extend(winner.resolved.coverage.iter().copied()); + .extend(winner.coverage()); // Round 2: only B remains. Combined with A's now-recorded 6 voters, // B's 2 new voters cross 2/3 of 10 — B is re-tiered from what would