perf(beacon): verify attestations against committee key sums

bin/ethlambda/src/benchmark/import/mod.rs

@@ -91,6 +91,12 @@ pub(crate) struct ReplayOptions {
     /// sleep is outside every measured span. 0 feeds blocks back to back.
     #[arg(long, value_name = "MILLISECONDS", default_value_t = 0)]
     pub block_delay: u64,
+    /// Warm each epoch's committee key sums before its first block is
+    /// imported, outside every measured span. A live node does this on a
+    /// worker between slots (the epoch precompute); a replay has no tick
+    /// loop, so without this flag part of the import never sees warm sums.
+    #[arg(long)]
+    pub warm_committee_sums: bool,
     /// Report format printed to stdout. Logs go to stderr, so JSON output can
     /// be piped directly (e.g. into jq).
     #[arg(long, value_enum, default_value_t = OutputFormat::Human)]
@@ -194,6 +200,12 @@ mod tests {
             .replay
     }
 
+    #[test]
+    fn warm_committee_sums_is_off_unless_asked_for() {
+        assert!(!parse(&[]).warm_committee_sums);
+        assert!(parse(&["--warm-committee-sums"]).warm_committee_sums);
+    }
+
     #[test]
     fn block_delay_defaults_to_zero_and_parses_milliseconds() {
         assert_eq!(parse(&[]).block_delay, 0);

bin/ethlambda/src/benchmark/import/replay.rs

@@ -24,6 +24,7 @@ use ethlambda_blockchain::metrics::{BLOCK_ARRIVAL_PHASES, BLOCK_IMPORT_PHASES};
 use ethlambda_blockchain::{BlockChainServer, ImportOutcome};
 use ethlambda_p2p::beacon::decode;
 use ethlambda_state_transition::beacon::fork_choice;
+use ethlambda_state_transition::beacon::helpers::accessors::CommitteeCacheExt as _;
 use ethlambda_storage::Store;
 use ethlambda_storage::backend::RocksDBBackend;
 use ethlambda_types::ShortRoot;
@@ -123,10 +124,17 @@ pub(crate) async fn replay_corpus(dir: &Path, options: &ReplayOptions) -> eyre::
         BlockChainServer::for_replay(store, None, options.safe_slots_to_import_optimistically);
 
     let mut samples = Vec::with_capacity(manifest.slots.len());
+    // The epoch of the block imported before the one in hand, for
+    // `--warm-committee-sums`.
+    let mut previous_epoch: Option<u64> = None;
 
     let warmup_total = manifest.warmup_slots.len();
     for (position, &slot) in manifest.warmup_slots.iter().enumerate() {
         let (block, block_root) = load_block(dir, &config, &mut clock, slot)?;
+        if options.warm_committee_sums {
+            warm_committee_sums(&clock, &block, previous_epoch)?;
+        }
+        previous_epoch = Some(slot / preset::SLOTS_PER_EPOCH);
         let wall_seconds = import(&mut server, block, slot, block_root).await?;
         eprintln!(
             "warm-up {}/{warmup_total}: slot {slot} imported in {}",
@@ -139,6 +147,12 @@ pub(crate) async fn replay_corpus(dir: &Path, options: &ReplayOptions) -> eyre::
     let total = manifest.slots.len();
     for (iteration, &slot) in manifest.slots.iter().enumerate() {
         let (block, block_root) = load_block(dir, &config, &mut clock, slot)?;
+        // Before the timer starts, so the warm-up is in no phase and no
+        // `wall_seconds`.
+        if options.warm_committee_sums {
+            warm_committee_sums(&clock, &block, previous_epoch)?;
+        }
+        previous_epoch = Some(slot / preset::SLOTS_PER_EPOCH);
 
         let timer = PhaseTimer::start(IMPORT_PHASE_HISTOGRAM);
         let wall_seconds = import(&mut server, block, slot, block_root).await?;
@@ -164,7 +178,12 @@ pub(crate) async fn replay_corpus(dir: &Path, options: &ReplayOptions) -> eyre::
 
     Ok(Report::new(
         Environment::collect(),
-        params_from(&manifest, dir, options.block_delay),
+        params_from(
+            &manifest,
+            dir,
+            options.block_delay,
+            options.warm_committee_sums,
+        ),
         samples,
     ))
 }
@@ -198,6 +217,48 @@ fn load_block(
     Ok((block, block_root))
 }
 
+/// Warms the committee key sums a live node's epoch precompute worker would
+/// have warmed before `block` arrives: its epoch's, when that epoch is later
+/// than `previous_epoch`'s, and, before the first block of the replay
+/// (`previous_epoch` is `None`), the epoch before as well.
+///
+/// The worker advances the head's post-state to the epoch start and warms
+/// from that state; this does the same from the block's parent post-state, so
+/// the shuffling key it files the sums under is the one the import looks up.
+/// Nothing else the worker does is reproduced: in a replay the import
+/// already produces the epoch's state itself, inside the measured span.
+fn warm_committee_sums(
+    store: &Store,
+    block: &SignedBeaconBlock,
+    previous_epoch: Option<u64>,
+) -> eyre::Result<()> {
+    let epoch = block.slot() / preset::SLOTS_PER_EPOCH;
+    let epochs: Vec<u64> = match previous_epoch {
+        None => epoch.checked_sub(1).into_iter().chain([epoch]).collect(),
+        Some(previous) if epoch > previous => vec![epoch],
+        Some(_) => return Ok(()),
+    };
+    let parent_root = block.parent_root();
+    let parent_state = store.get_state(&parent_root)?.ok_or_else(|| {
+        eyre::eyre!(
+            "no post-state for parent {} of the block at slot {}",
+            ShortRoot(&parent_root.0),
+            block.slot()
+        )
+    })?;
+    let config = store.config();
+    let committee_cache = store.committee_cache();
+    for epoch in epochs {
+        if parent_state.slot() < epoch * preset::SLOTS_PER_EPOCH {
+            let advanced = fork_choice::advance_to_epoch_start(&parent_state, epoch, &config)?;
+            committee_cache.warm_key_sums(&advanced, epoch);
+        } else {
+            committee_cache.warm_key_sums(&parent_state, epoch);
+        }
+    }
+    Ok(())
+}
+
 /// Import one block, returning the wall time of the `import_block` call
 /// alone, and fail unless it imported.
 async fn import(
@@ -240,7 +301,12 @@ fn read_anchor(dir: &Path, config: &Config) -> eyre::Result<(BeaconState, Signed
 
 /// The report parameters a manifest already carries, so the loop above never
 /// has to reconstruct them from samples.
-fn params_from(manifest: &Manifest, dir: &Path, block_delay_ms: u64) -> Params {
+fn params_from(
+    manifest: &Manifest,
+    dir: &Path,
+    block_delay_ms: u64,
+    warm_committee_sums: bool,
+) -> Params {
     Params {
         mode: "import",
         corpus: dir.display().to_string(),
@@ -252,6 +318,7 @@ fn params_from(manifest: &Manifest, dir: &Path, block_delay_ms: u64) -> Params {
         range_end: manifest.range_end,
         blocks: manifest.slots.len(),
         block_delay_ms,
+        warm_committee_sums,
     }
 }
 

bin/ethlambda/src/benchmark/report/import.rs

@@ -24,6 +24,9 @@ pub(crate) struct Params {
     /// Milliseconds slept after each block's import, outside every measured
     /// span. 0 means blocks were fed back to back.
     pub block_delay_ms: u64,
+    /// Whether each epoch's committee key sums were warmed before its first
+    /// block, outside every measured span (`--warm-committee-sums`).
+    pub warm_committee_sums: bool,
 }
 
 #[derive(Debug, Serialize)]
@@ -107,15 +110,16 @@ impl Report {
         let _ = writeln!(out, "Block-import benchmark — {} workload", params.mode);
         let _ = writeln!(
             out,
-            "  corpus={} network={} anchor_slot={} warmup_blocks={} range=[{}, {}] blocks={} block_delay_ms={}",
+            "  corpus={} network={} anchor_slot={} warmup_blocks={} range=[{}, {}] blocks={} block_delay_ms={} warm_committee_sums={}",
             params.corpus,
             params.network,
             params.anchor_slot,
             params.warmup_blocks,
             params.range_start,
             params.range_end,
             params.blocks,
-            params.block_delay_ms
+            params.block_delay_ms,
+            params.warm_committee_sums
         );
         let _ = writeln!(
             out,
@@ -217,6 +221,7 @@ mod tests {
             range_end: 3,
             blocks: 2,
             block_delay_ms: 0,
+            warm_committee_sums: false,
         }
     }
 
@@ -231,6 +236,17 @@ mod tests {
         assert!(report.human_table().contains("block_delay_ms=750"));
     }
 
+    #[test]
+    fn warm_committee_sums_is_recorded_in_the_json_and_the_header() {
+        let mut params = params();
+        params.warm_committee_sums = true;
+        let report = Report::new(Environment::collect(), params, Vec::new());
+
+        let json: serde_json::Value = serde_json::from_str(&report.to_json().unwrap()).unwrap();
+        assert_eq!(json["params"]["warm_committee_sums"], true);
+        assert!(report.human_table().contains("warm_committee_sums=true"));
+    }
+
     #[test]
     fn the_range_prints_inclusive_as_fetch_takes_it() {
         let report = Report::new(Environment::collect(), params(), Vec::new());

crates/blockchain/src/epoch_precompute.rs

@@ -25,6 +25,7 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use ethlambda_state_transition::beacon::fork_choice;
+use ethlambda_state_transition::beacon::helpers::accessors::CommitteeCacheExt as _;
 use ethlambda_state_transition::beacon::helpers::misc::compute_start_slot_at_epoch;
 use ethlambda_state_transition::metrics as stf_metrics;
 use ethlambda_storage::{CacheKey, Chain};
@@ -182,23 +183,35 @@ impl BlockChainServer {
             Trigger::Head => head_slot / preset::SLOTS_PER_EPOCH + 1,
             Trigger::Timer { slot } => slot / preset::SLOTS_PER_EPOCH + 1,
         };
-        let already_cached = self
-            .store
-            .cached_state(CacheKey::CheckpointState {
-                epoch: candidate_epoch,
-                root: head_root,
-            })
-            .is_some();
+        let cached = self.store.cached_state(CacheKey::CheckpointState {
+            epoch: candidate_epoch,
+            root: head_root,
+        });
         let inputs = DecisionInputs {
             trigger,
             syncing: self.sync_status.is_syncing(),
             head_slot,
             head_root,
             current_slot: self.wall_clock_slot(),
             in_flight: self.epoch_precompute_in_flight,
-            already_cached,
+            already_cached: cached.is_some(),
         };
         let Some(key) = decide(&inputs) else {
+            // Someone else advanced the state already (a gloas bid for the
+            // epoch's first slot does), so no worker runs: warm the sums from
+            // that state instead, or the epoch would import without them.
+            if let Some(state) = cached
+                && decide(&DecisionInputs {
+                    already_cached: false,
+                    ..inputs
+                })
+                .is_some()
+            {
+                let committee_cache = self.store.committee_cache();
+                tokio::task::spawn_blocking(move || {
+                    warm_epoch_key_sums(&committee_cache, &state, candidate_epoch)
+                });
+            }
             return;
         };
         // The head's post-state is resident (it was just imported); a read that
@@ -220,17 +233,38 @@ impl BlockChainServer {
 
         let config = self.store.config();
         let actor = ctx.actor_ref();
+        let committee_cache = self.store.committee_cache();
         tokio::task::spawn_blocking(move || {
             let result = {
                 let _timing = stf_metrics::time_epoch_precompute();
                 fork_choice::advance_to_epoch_start(&head_state, key.epoch, &config)
             };
             let result = result.map(Arc::new).map_err(|err| err.to_string());
+            let advanced = result.as_ref().ok().map(Arc::clone);
+            // The state goes to the actor first: the epoch's first block must
+            // not wait for key sums, which only make its attestations cheaper.
             let _ = actor.send(EpochPrecomputed { key, result });
+            if let Some(state) = advanced {
+                warm_epoch_key_sums(&committee_cache, &state, key.epoch);
+            }
         });
     }
 }
 
+/// Warms the key sums `epoch`'s blocks read: `epoch`'s own, and the previous
+/// epoch's for the attestations those blocks carry from it. The previous
+/// epoch's are normally warm already, from the last boundary; after a restart
+/// mid-epoch, or a boundary whose worker did not run, they are not. Both calls
+/// return at once when the sums are resident.
+fn warm_epoch_key_sums(
+    committee_cache: &ethlambda_storage::CommitteeCache,
+    state: &BeaconState,
+    epoch: u64,
+) {
+    committee_cache.warm_key_sums(state, epoch);
+    committee_cache.warm_key_sums(state, epoch.saturating_sub(1));
+}
+
 impl Handler<EpochPrecomputeCheck> for BlockChainServer {
     async fn handle(&mut self, msg: EpochPrecomputeCheck, ctx: &Context<Self>) {
         self.maybe_start_epoch_precompute(Trigger::Timer { slot: msg.slot }, ctx);