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

At interval 2 the aggregation worker runs up to MAX_AGGREGATION_JOBS
leanVM proofs. When interval 4 of the same slot builds the next slot's
block, that build runs its own proofs, so the two contend for the
prover — and the build is the one with a hard slot-boundary deadline.

Drop the session to a single job whenever one of our validators proposes
the next slot, mirroring the propose path's condition (proposer + duties
allowed) so a sync-suppressed slot keeps the full budget. The retained
job is the best-scoring candidate, so the highest-value coverage
survives the cap.

Applies to both entry points into start_aggregation_session: the
interval-2 tick and the early 2/3-threshold trigger.

Author: MegaRedHand

crates/blockchain/src/aggregation.rs

@@ -14,7 +14,9 @@
 //! 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
+//! [`PROPOSER_MAX_AGGREGATION_JOBS`] in the slot before one of our validators
+//! proposes.
 
 use std::collections::{HashMap, HashSet};
 use std::time::{Duration, Instant, SystemTime};
@@ -176,7 +178,14 @@ 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 = 3;
+
+/// Job cap for a session running in the slot before one of our validators
+/// proposes. The interval-4 build runs its own leanVM proofs for the block it
+/// is about to publish; keeping this slot's aggregation to a single job leaves
+/// the prover to that build instead of racing it. The one job we do run is the
+/// best-scoring candidate, so the highest-value coverage is retained.
+pub(crate) const PROPOSER_MAX_AGGREGATION_JOBS: usize = 1;
 
 /// Build a snapshot of everything needed to aggregate. Runs on the actor
 /// thread, touches the store, does no heavy cryptography. Returns `None` when
@@ -190,17 +199,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
+/// [`PROPOSER_MAX_AGGREGATION_JOBS`] 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();
@@ -269,9 +283,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,
@@ -1189,7 +1202,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
@@ -1218,7 +1231,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
@@ -1261,7 +1274,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"
         );
     }
@@ -1317,7 +1330,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!(
@@ -1327,23 +1340,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 {
@@ -1365,7 +1381,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 three 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);
 
@@ -1380,4 +1407,24 @@ mod tests {
             "the three highest target_slot groups win the new_voters tie"
         );
     }
+
+    /// The proposer cap 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, PROPOSER_MAX_AGGREGATION_JOBS)
+            .expect("should produce a job");
+        assert_eq!(snapshot.groups_considered, NUM_GROUPS);
+        assert_eq!(snapshot.jobs.len(), PROPOSER_MAX_AGGREGATION_JOBS);
+        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"
+        );
+    }
 }

crates/blockchain/src/lib.rs

@@ -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, PROPOSER_MAX_AGGREGATION_JOBS, run_aggregation_worker,
 };
 use crate::key_manager::ValidatorKeyPair;
 use crate::sync_status::SyncStatusTracker;
@@ -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 (see [`PROPOSER_MAX_AGGREGATION_JOBS`]).
     /// 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();
@@ -485,7 +490,29 @@ 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 {
+        // Throttle to a single job when interval 4 of this slot will build the
+        // next slot's block: that build runs its own leanVM proofs, and the two
+        // otherwise contend for the prover. Mirror the propose path's condition
+        // (`SlotInterval::EndOfSlot`) so a slot where duties are suppressed
+        // keeps the full job budget.
+        let next_proposer = self
+            .get_our_proposer(slot + 1)
+            .filter(|_| self.sync_status.duties_allowed());
+        let max_jobs = match next_proposer {
+            Some(validator_id) => {
+                info!(
+                    %slot,
+                    next_slot_proposer = validator_id,
+                    max_jobs = PROPOSER_MAX_AGGREGATION_JOBS,
+                    "Throttling aggregation ahead of our block proposal"
+                );
+                PROPOSER_MAX_AGGREGATION_JOBS
+            }
+            None => 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;
         };