Skip to content

feat(aggregation): cap aggregation at one job before our proposal #544

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
MegaRedHand merged 4 commits into from
Jul 29, 2026
Merged

feat(aggregation): cap aggregation at one job before our proposal #544

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
82abf6b
feat(aggregation): cap aggregation at one job before our proposal
MegaRedHand Jul 29, 2026
4e8027a
refactor(aggregation): lower default job cap to 2, inline the propose…
MegaRedHand Jul 29, 2026
1529530
refactor(aggregation): simplify the proposer job-cap branch
MegaRedHand Jul 29, 2026
6a0225b
Merge branch 'main' into feat/throttle-aggregation-before-proposal
MegaRedHand Jul 29, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
82 changes: 60 additions & 22 deletions crates/blockchain/src/aggregation.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,8 @@
//! 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.
//! most `max_jobs` jobs — [`MAX_AGGREGATION_JOBS`] normally, dropping to a
//! single job in the slot before one of our validators proposes.

use std::collections::{HashMap, HashSet};
use std::time::{Duration, Instant, SystemTime};
Expand Down Expand Up @@ -176,7 +177,7 @@ impl Message for EarlyAggregationCheck {
/// 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;
pub(crate) const MAX_AGGREGATION_JOBS: usize = 2;

/// Build a snapshot of everything needed to aggregate. Runs on the actor
/// thread, touches the store, does no heavy cryptography. Returns `None` when
Expand All @@ -190,17 +191,22 @@ const MAX_AGGREGATION_JOBS: usize = 3;
/// (`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
/// 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<AggregationSnapshot> {
let gossip_groups = store.iter_gossip_signatures();
let new_payload_keys = store.new_payload_keys();
Expand Down Expand Up @@ -269,9 +275,8 @@ pub fn snapshot_aggregation_inputs(

let mut projected = block_builder::ProjectedState::from_head_state(&head_state);

let mut jobs: Vec<AggregationJob> =
Vec::with_capacity(MAX_AGGREGATION_JOBS.min(groups_considered));
for _round in 0..MAX_AGGREGATION_JOBS {
let mut jobs: Vec<AggregationJob> = Vec::with_capacity(max_jobs.min(groups_considered));
for _round in 0..max_jobs {
let Some((data_root, score)) = pick_best_candidate(
&candidates,
&projected,
Expand Down Expand Up @@ -1192,7 +1197,7 @@ mod tests {
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());
assert!(snapshot_aggregation_inputs(&store, 0, MAX_AGGREGATION_JOBS).is_none());
}

/// A single gossip signature with no other material to merge is dropped
Expand Down Expand Up @@ -1221,7 +1226,7 @@ mod tests {
let hashed = HashedAttestationData::new(att_data);
store.insert_gossip_signature(hashed, 0, dummy_sig());

assert!(snapshot_aggregation_inputs(&store, 0).is_none());
assert!(snapshot_aggregation_inputs(&store, 0, MAX_AGGREGATION_JOBS).is_none());
}

/// A group whose target is already justified (here: at or behind the
Expand Down Expand Up @@ -1264,7 +1269,7 @@ mod tests {
store.insert_gossip_signature(hashed, 1, dummy_sig());

assert!(
snapshot_aggregation_inputs(&store, 999).is_none(),
snapshot_aggregation_inputs(&store, 999, MAX_AGGREGATION_JOBS).is_none(),
"a group targeting an already-justified slot must never become a job"
);
}
Expand Down Expand Up @@ -1320,7 +1325,7 @@ 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)
let snapshot = snapshot_aggregation_inputs(&store, HEAD_SLOT, MAX_AGGREGATION_JOBS)
.expect("a vote for the current head must produce a job (chain view covers the tip)");
assert_eq!(snapshot.jobs.len(), 1);
assert_eq!(
Expand All @@ -1330,23 +1335,26 @@ mod tests {
);
}

/// 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() {
/// Number of competing candidates built by
/// [`store_with_competing_build_tier_groups`]; more than either job cap so
/// both cap tests actually bind.
const NUM_GROUPS: usize = 5;

/// Store holding `NUM_GROUPS` competing Build-tier candidates (2 raw sigs
/// each, well under the 2/3 threshold) that differ only by `target_slot`
/// (`1..=NUM_GROUPS`, all justifiable at delta <= 5). Build-tier ordering
/// prefers larger `target_slot` on a new_voters tie, so selection takes
/// them highest-slot-first.
fn store_with_competing_build_tier_groups() -> Store {
const NUM_VALIDATORS: usize = 10;
const HEAD_SLOT: u64 = 10;
const NUM_GROUPS: usize = 5;

let hashes: Vec<H256> = (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 target_slot = i as u64 + 1;
let att_data = AttestationData {
slot: target_slot,
head: Checkpoint {
Expand All @@ -1368,7 +1376,18 @@ mod tests {
store.insert_gossip_signature(hashed, (2 * i + 1) as u64, dummy_sig());
}

let snapshot = snapshot_aggregation_inputs(&store, 999).expect("should produce jobs");
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`.
#[test]
fn snapshot_caps_jobs_at_max_aggregation_jobs() {
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);

Expand All @@ -1379,8 +1398,27 @@ mod tests {
.collect();
assert_eq!(
selected_targets,
HashSet::from([3, 4, 5]),
"the three highest target_slot groups win the new_voters tie"
HashSet::from([4, 5]),
"the two highest target_slot groups win the new_voters tie"
);
}

/// 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.
#[test]
fn snapshot_caps_jobs_at_one_for_proposer() {
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,
NUM_GROUPS as u64,
"the single job is the best-scoring candidate, not an arbitrary one"
);
}
}
25 changes: 21 additions & 4 deletions crates/blockchain/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,8 +15,8 @@ use ethlambda_types::{

use crate::aggregation::{
AGGREGATION_DEADLINE, AggregateProduced, AggregationDeadline, AggregationDone,
AggregationSession, EARLY_AGGREGATION_WINDOW, EarlyAggregationCheck, PRIOR_WORKER_JOIN_TIMEOUT,
run_aggregation_worker,
AggregationSession, EARLY_AGGREGATION_WINDOW, EarlyAggregationCheck, MAX_AGGREGATION_JOBS,
PRIOR_WORKER_JOIN_TIMEOUT, run_aggregation_worker,
};
use crate::key_manager::ValidatorKeyPair;
use crate::sync_status::SyncStatusTracker;
Expand Down Expand Up @@ -460,9 +460,14 @@ impl BlockChainServer {

/// 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.
/// 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 +`AGGREGATION_DEADLINE`.
///
/// 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<Self>) {
if let Some(prior) = self.current_aggregation.take() {
prior.cancel.cancel();
Expand All @@ -485,7 +490,19 @@ impl BlockChainServer {

coverage::emit_agg_start_new_coverage(&self.store, self.attestation_committee_count);

let Some(snapshot) = aggregation::snapshot_aggregation_inputs(&self.store, slot) else {
// 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());
Comment on lines +496 to +497

@greptile-apps greptile-apps Bot Jul 29, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Duty transition bypasses cap

When the node is syncing at the interval-2 tick but becomes synced before interval 4, aggregation starts with three jobs while the later duty check permits the proposal, causing the original prover contention and potentially delaying block publication.

Knowledge Base Used: Blockchain core: fork choice, state transition, block building, sync

Prompt To Fix With AI 
This is a comment left during a code review.
Path: crates/blockchain/src/lib.rs
Line: 499-500

Comment:
**Duty transition bypasses cap**

When the node is syncing at the interval-2 tick but becomes synced before interval 4, aggregation starts with three jobs while the later duty check permits the proposal, causing the original prover contention and potentially delaying block publication.

**Knowledge Base Used:** [Blockchain core: fork choice, state transition, block building, sync](https://app.greptile.com/lambdaclass/-/custom-context/knowledge-base/lambdaclass/ethlambda/-/docs/blockchain-core.md)

---

For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.

let max_jobs = if next_proposer.is_some() {
1
} else {
MAX_AGGREGATION_JOBS
};

let Some(snapshot) = aggregation::snapshot_aggregation_inputs(&self.store, slot, max_jobs)
else {
// No current-slot gossip sigs — nothing to aggregate this slot.
return;
};
Expand Down