From 27ae8762a70092fcc9c4aebf42ae4c8f34bb3526 Mon Sep 17 00:00:00 2001 From: Kai Aldag Date: Sun, 21 Jun 2026 16:26:52 +0100 Subject: [PATCH 1/3] add reactive reorg recovery --- CHANGELOG.md | 9 + README.md | 7 +- src/reactive/mod.rs | 488 +++++++++++++++++++++++++++++++- tests/reactive_reorg.rs | 612 ++++++++++++++++++++++++++++++++++++++++ 4 files changed, 1111 insertions(+), 5 deletions(-) create mode 100644 tests/reactive_reorg.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 330b1f3..372cefc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -148,6 +148,15 @@ pre-release development phases (see [`docs/ROADMAP.md`](docs/ROADMAP.md)). gives downstream retry policy and metrics a stable failure classification. Account-field resyncs remain explicitly unsupported until a provider-neutral account fetch callback exists. Generic core. +- **Reactive block journaling and reorg recovery** — canonical block inputs and + applied handler reports are retained in a depth-bounded runtime journal. + Removed logs, explicit `ChainStatus::Reorged` inputs, and parent-hash + discontinuities now emit `ReactiveReport::Reorg` with dropped blocks, dropped + inputs, rollback updates/diffs, purge updates/diffs, and canceled hash-pinned + resync requests. Reversible storage-slot changes are rolled back in reverse + apply order; account/code changes and prior purge effects conservatively fall + back to targeted purge updates because `StateDiff` does not carry enough data + to reconstruct those cache entries exactly. Generic core. - **`StateUpdate::SlotMasked`** (`state_update`, Phase 4) — a cold-aware read-modify-write *masked* slot write (`new = (old & !mask) | (value & mask)`) with the `StateUpdate::slot_masked` constructor, so a pure decoder can update diff --git a/README.md b/README.md index bc76e7c..12ec6f2 100644 --- a/README.md +++ b/README.md @@ -66,7 +66,12 @@ around three capabilities that target exactly this workload: inputs, deduplicates and orders canonical logs, validates pending semantics, applies canonical cache mutations through `EvmCache::apply_updates`, and can optionally execute storage resync requests through the cache's - provider-neutral storage batch fetcher before dispatching reports to hooks. The + provider-neutral storage batch fetcher before dispatching reports to hooks. + Canonical block effects are journaled for depth-bounded reorg recovery: + removed logs, explicit reorged inputs, and parent-hash discontinuities emit + `ReactiveReport::Reorg`, roll back reversible storage writes, fall back to + targeted purges for irreversible effects, and cancel stale hash-pinned + resyncs. The `ReactiveRegistry` exposes consolidated Alloy log filters for provider subscription setup and exact local log routing with optional route keys. The provider-agnostic `EventSubscriber` trait and `AlloySubscriber` scaffold are diff --git a/src/reactive/mod.rs b/src/reactive/mod.rs index 7ad9ddd..04e4bb2 100644 --- a/src/reactive/mod.rs +++ b/src/reactive/mod.rs @@ -14,7 +14,7 @@ use std::{ any::Any, borrow::Cow, - collections::{HashMap, HashSet}, + collections::{HashMap, HashSet, VecDeque}, fmt, future::Future, hash::Hash, @@ -956,12 +956,39 @@ pub struct BlockReport { /// Reorg reporting scaffold. #[derive(Clone, Debug)] pub struct ReorgReport { - /// Dropped block, when known. + /// First dropped block, when known. pub dropped: Option, + /// Blocks dropped from the journal, in ascending journal order. + pub dropped_blocks: Vec, + /// Input references that belonged to dropped blocks. + pub dropped_inputs: Vec, + /// Exact rollback updates applied for reversible dropped effects. + pub rollback_updates: Vec, + /// Diff returned by applying [`rollback_updates`](Self::rollback_updates). + pub rollback_diff: StateDiff, + /// Conservative purge updates applied for irreversible dropped effects. + pub purge_updates: Vec, + /// Diff returned by applying [`purge_updates`](Self::purge_updates). + pub purge_diff: StateDiff, + /// Hash-pinned pending resync requests canceled because their block was dropped. + pub canceled_resyncs: Vec, + /// Reorg trigger. + pub reason: ReorgReason, /// Network marker. pub _network: PhantomData, } +/// Reason reorg recovery ran. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] +pub enum ReorgReason { + /// A provider emitted an Alloy removed log. + RemovedLog, + /// The input context explicitly marked an input as reorged. + ReorgedInput, + /// A canonical block did not connect to the journaled head. + ParentMismatch, +} + /// Error report scaffold. #[derive(Clone, Debug)] pub struct ReactiveErrorReport { @@ -1131,6 +1158,16 @@ pub struct ReactiveRuntime { registry: ReactiveRegistry, hooks: Vec>>, config: ReactiveConfig, + journal: VecDeque>, + pending_resyncs: Vec, +} + +#[derive(Clone, Debug)] +struct BlockJournal { + block: BlockRef, + inputs: Vec, + applied: Vec>, + resynced: Vec, } /// Registry and router for provider-neutral reactive handlers. @@ -1243,6 +1280,8 @@ impl ReactiveRuntime { registry: ReactiveRegistry::new(), hooks: Vec::new(), config, + journal: VecDeque::new(), + pending_resyncs: Vec::new(), } } @@ -1286,7 +1325,7 @@ impl ReactiveRuntime { /// values are applied as [`StateUpdate::slot`] updates through /// [`EvmCache::apply_updates`], and unsupported or failed targets are reported /// in [`ResyncReport::failed`]. It does not start subscribers, background - /// workers, reorg journals, or network transport. + /// workers, or network transport. pub fn ingest_batch_with_resync( &mut self, cache: &mut EvmCache, @@ -1296,6 +1335,8 @@ impl ReactiveRuntime { if !batch_report.resyncs.is_empty() { let resync_report = execute_resync_requests(cache, &batch_report.resyncs); + self.remove_pending_resyncs(batch_report.resyncs.iter().map(|request| &request.id)); + self.record_journal_resync(&resync_report); batch_report .reports .push(Arc::new(ReactiveReport::Resynced(resync_report))); @@ -1307,7 +1348,7 @@ impl ReactiveRuntime { } fn ingest_batch_direct( - &self, + &mut self, cache: &mut EvmCache, batch: ReactiveInputBatch, ) -> Result, ReactiveError> { @@ -1324,6 +1365,27 @@ impl ReactiveRuntime { _network: PhantomData, }))); + if let Some(reorg_report) = self.recover_for_canonical_input(cache, &record) { + remove_canceled_resyncs_from_batch( + &mut batch_report.resyncs, + &reorg_report.canceled_resyncs, + ); + reports_to_dispatch.push(Arc::new(ReactiveReport::Reorg(reorg_report))); + } + + if let Some(reorg_report) = self.recover_for_reorged_input(cache, &record) { + remove_canceled_resyncs_from_batch( + &mut batch_report.resyncs, + &reorg_report.canceled_resyncs, + ); + reports_to_dispatch.push(Arc::new(ReactiveReport::Reorg(reorg_report))); + continue; + } + + if let Some(block) = canonical_record_block(&record) { + self.record_journal_input(block, input_ref); + } + let executions = self.execute_handlers(cache, &record, input_ref)?; if executions.is_empty() { continue; @@ -1350,6 +1412,8 @@ impl ReactiveRuntime { batch_report .resyncs .extend(execution.resyncs.iter().cloned()); + self.pending_resyncs + .extend(execution.resyncs.iter().cloned()); batch_report .speculative .extend(execution.speculative.iter().cloned()); @@ -1369,6 +1433,9 @@ impl ReactiveRuntime { }; let report = Arc::new(ReactiveReport::Applied(applied.clone())); reports_to_dispatch.push(report); + if let Some(block) = canonical_record_block(&record) { + self.record_journal_applied(block, applied.clone()); + } batch_report.applied.push(applied); } } @@ -1414,6 +1481,419 @@ impl ReactiveRuntime { } } } + + fn recover_for_canonical_input( + &mut self, + cache: &mut EvmCache, + record: &ReactiveInputRecord, + ) -> Option> { + let block = canonical_record_block(record)?; + let latest = self.journal.back()?.block.clone(); + + if self + .journal + .iter() + .any(|entry| entry.block.hash == block.hash && entry.block.number == block.number) + { + return None; + } + + if block.number == latest.number.saturating_add(1) && block.parent_hash == Some(latest.hash) + { + return None; + } + + if block.number > latest.number.saturating_add(1) { + return None; + } + + let dropped = if let Some(parent_hash) = block.parent_hash { + if let Some(parent_index) = self + .journal + .iter() + .rposition(|entry| entry.block.hash == parent_hash) + { + self.drain_journal_after(parent_index) + } else { + self.drain_journal_from_number(block.number) + } + } else { + self.drain_journal_from_number(block.number) + }; + + self.recover_dropped_journals(cache, dropped, ReorgReason::ParentMismatch) + } + + fn recover_for_reorged_input( + &mut self, + cache: &mut EvmCache, + record: &ReactiveInputRecord, + ) -> Option> { + let (dropped_block, reason) = reorg_signal_block(record)?; + let dropped = if let Some(index) = self + .journal + .iter() + .position(|entry| entry.block.hash == dropped_block.hash) + { + self.drain_journal_from(index) + } else { + self.drain_journal_from_number(dropped_block.number) + }; + + if dropped.is_empty() { + let canceled_resyncs = + self.cancel_resyncs_for_dropped_blocks(std::slice::from_ref(&dropped_block)); + if canceled_resyncs.is_empty() { + return None; + } + return Some(ReorgReport { + dropped: Some(dropped_block.clone()), + dropped_blocks: vec![dropped_block], + dropped_inputs: Vec::new(), + rollback_updates: Vec::new(), + rollback_diff: StateDiff::default(), + purge_updates: Vec::new(), + purge_diff: StateDiff::default(), + canceled_resyncs, + reason, + _network: PhantomData, + }); + } + + self.recover_dropped_journals(cache, dropped, reason) + } + + fn record_journal_input(&mut self, block: &BlockRef, input_ref: InputRef) { + let entry = self.journal_entry_mut(block); + if !entry.inputs.contains(&input_ref) { + entry.inputs.push(input_ref); + } + self.trim_journal(); + } + + fn record_journal_applied(&mut self, block: &BlockRef, applied: AppliedReport) { + self.journal_entry_mut(block).applied.push(applied); + self.trim_journal(); + } + + fn record_journal_resync(&mut self, report: &ResyncReport) { + if report.diff.is_empty() { + return; + } + let Some(block) = single_hash_pinned_resync_block(report) else { + return; + }; + self.journal_entry_mut(&block).resynced.push(report.clone()); + self.trim_journal(); + } + + fn journal_entry_mut(&mut self, block: &BlockRef) -> &mut BlockJournal { + if let Some(index) = self + .journal + .iter() + .position(|entry| entry.block.hash == block.hash && entry.block.number == block.number) + { + return &mut self.journal[index]; + } + + self.journal.push_back(BlockJournal { + block: block.clone(), + inputs: Vec::new(), + applied: Vec::new(), + resynced: Vec::new(), + }); + let index = self.journal.len() - 1; + &mut self.journal[index] + } + + fn trim_journal(&mut self) { + if self.config.journal_depth == 0 { + self.journal.clear(); + return; + } + while self.journal.len() > self.config.journal_depth { + self.journal.pop_front(); + } + } + + fn drain_journal_after(&mut self, index: usize) -> Vec> { + self.journal.drain((index + 1)..).collect() + } + + fn drain_journal_from(&mut self, index: usize) -> Vec> { + self.journal.drain(index..).collect() + } + + fn drain_journal_from_number(&mut self, number: u64) -> Vec> { + let Some(index) = self + .journal + .iter() + .position(|entry| entry.block.number >= number) + else { + return Vec::new(); + }; + self.drain_journal_from(index) + } + + fn recover_dropped_journals( + &mut self, + cache: &mut EvmCache, + dropped: Vec>, + reason: ReorgReason, + ) -> Option> { + if dropped.is_empty() { + return None; + } + + let dropped_blocks: Vec<_> = dropped.iter().map(|entry| entry.block.clone()).collect(); + let dropped_inputs: Vec<_> = dropped + .iter() + .flat_map(|entry| entry.inputs.iter().copied()) + .collect(); + let canceled_resyncs = self.cancel_resyncs_for_dropped_blocks(&dropped_blocks); + let purge_scopes = purge_scopes_for_dropped_journals(&dropped); + let rollback_updates = rollback_updates_for_dropped_journals(&dropped, &purge_scopes); + let purge_updates: Vec<_> = purge_scopes + .iter() + .map(|(address, scope)| StateUpdate::purge(*address, scope.clone())) + .collect(); + + let rollback_diff = if rollback_updates.is_empty() { + StateDiff::default() + } else { + cache.apply_updates(&rollback_updates) + }; + let purge_diff = if purge_updates.is_empty() { + StateDiff::default() + } else { + cache.apply_updates(&purge_updates) + }; + + Some(ReorgReport { + dropped: dropped_blocks.first().cloned(), + dropped_blocks, + dropped_inputs, + rollback_updates, + rollback_diff, + purge_updates, + purge_diff, + canceled_resyncs, + reason, + _network: PhantomData, + }) + } + + fn cancel_resyncs_for_dropped_blocks( + &mut self, + dropped_blocks: &[BlockRef], + ) -> Vec { + let mut canceled = Vec::new(); + self.pending_resyncs.retain(|request| { + let should_cancel = resync_request_targets_dropped_block(request, dropped_blocks); + if should_cancel { + canceled.push(request.clone()); + } + !should_cancel + }); + canceled + } + + fn remove_pending_resyncs<'a>(&mut self, ids: impl IntoIterator) { + let ids: HashSet<_> = ids.into_iter().cloned().collect(); + self.pending_resyncs + .retain(|request| !ids.contains(&request.id)); + } +} + +fn canonical_record_block(record: &ReactiveInputRecord) -> Option<&BlockRef> { + if matches!(&record.input, ReactiveInput::Log(log) if log.removed) { + return None; + } + if is_canonical_status(&record.context.chain_status) { + return context_block_ref(&record.context); + } + None +} + +fn context_block_ref(ctx: &ReactiveContext) -> Option<&BlockRef> { + match &ctx.chain_status { + ChainStatus::Included { block, .. } + | ChainStatus::Safe { block } + | ChainStatus::Finalized { block } => Some(block), + ChainStatus::Reorged { dropped_from } => Some(dropped_from), + ChainStatus::Pending => ctx.block.as_ref(), + } +} + +fn reorg_signal_block( + record: &ReactiveInputRecord, +) -> Option<(BlockRef, ReorgReason)> { + if matches!(&record.input, ReactiveInput::Log(log) if log.removed) { + return block_ref_from_record(record).map(|block| (block, ReorgReason::RemovedLog)); + } + + if let ChainStatus::Reorged { dropped_from } = &record.context.chain_status { + return Some((dropped_from.clone(), ReorgReason::ReorgedInput)); + } + + None +} + +fn block_ref_from_record(record: &ReactiveInputRecord) -> Option { + context_block_ref(&record.context) + .cloned() + .or_else(|| match &record.input { + ReactiveInput::Log(log) => Some(BlockRef { + number: log.block_number?, + hash: log.block_hash?, + parent_hash: None, + timestamp: log.block_timestamp, + }), + ReactiveInput::BlockHeader(header) => Some(BlockRef { + number: header.number(), + hash: header.hash(), + parent_hash: Some(header.parent_hash()), + timestamp: Some(header.timestamp()), + }), + ReactiveInput::FullBlock(block) => { + let header = block.header(); + Some(BlockRef { + number: header.number(), + hash: header.hash(), + parent_hash: Some(header.parent_hash()), + timestamp: Some(header.timestamp()), + }) + } + ReactiveInput::PendingTxHash(_) | ReactiveInput::PendingTx(_) => None, + }) +} + +fn remove_canceled_resyncs_from_batch( + resyncs: &mut Vec, + canceled: &[ResyncRequest], +) { + if canceled.is_empty() { + return; + } + let canceled_ids: HashSet<_> = canceled.iter().map(|request| request.id.clone()).collect(); + resyncs.retain(|request| !canceled_ids.contains(&request.id)); +} + +fn resync_request_targets_dropped_block( + request: &ResyncRequest, + dropped_blocks: &[BlockRef], +) -> bool { + let ResyncBlock::Hash { number, hash, .. } = &request.block else { + return false; + }; + dropped_blocks + .iter() + .any(|block| block.hash == *hash && block.number == *number) +} + +fn single_hash_pinned_resync_block(report: &ResyncReport) -> Option { + let first = report.requested.first()?.block.clone(); + if !report + .requested + .iter() + .all(|request| request.block == first) + { + return None; + } + + let ResyncBlock::Hash { number, hash, .. } = first else { + return None; + }; + + Some(BlockRef { + number, + hash, + parent_hash: None, + timestamp: None, + }) +} + +fn purge_scopes_for_dropped_journals( + dropped: &[BlockJournal], +) -> Vec<(Address, PurgeScope)> { + let mut scopes: Vec<(Address, PurgeScope)> = Vec::new(); + for entry in dropped.iter().rev() { + for resynced in entry.resynced.iter().rev() { + merge_purge_scopes_for_diff(&mut scopes, &resynced.diff); + } + for applied in entry.applied.iter().rev() { + merge_purge_scopes_for_diff(&mut scopes, &applied.diff); + } + } + scopes +} + +fn rollback_updates_for_dropped_journals( + dropped: &[BlockJournal], + purge_scopes: &[(Address, PurgeScope)], +) -> Vec { + let purge_addresses: HashSet<_> = purge_scopes + .iter() + .map(|(address, _scope)| *address) + .collect(); + let mut updates = Vec::new(); + for entry in dropped.iter().rev() { + for resynced in entry.resynced.iter().rev() { + push_rollback_updates_for_diff(&mut updates, &resynced.diff, &purge_addresses); + } + for applied in entry.applied.iter().rev() { + push_rollback_updates_for_diff(&mut updates, &applied.diff, &purge_addresses); + } + } + updates +} + +fn merge_purge_scopes_for_diff(scopes: &mut Vec<(Address, PurgeScope)>, diff: &StateDiff) { + for change in &diff.accounts { + merge_purge_scope(scopes, change.address, PurgeScope::Account); + } + for record in &diff.purged { + merge_purge_scope(scopes, record.address, record.scope.clone()); + } +} + +fn push_rollback_updates_for_diff( + updates: &mut Vec, + diff: &StateDiff, + purge_addresses: &HashSet
, +) { + for change in diff.slots.iter().rev() { + if purge_addresses.contains(&change.address) { + continue; + } + updates.push(StateUpdate::slot(change.address, change.slot, change.old)); + } +} + +fn merge_purge_scope(scopes: &mut Vec<(Address, PurgeScope)>, address: Address, scope: PurgeScope) { + if let Some((_existing_address, existing_scope)) = scopes + .iter_mut() + .find(|(existing_address, _scope)| *existing_address == address) + { + *existing_scope = merged_purge_scope(existing_scope.clone(), scope); + } else { + scopes.push((address, scope)); + } +} + +fn merged_purge_scope(left: PurgeScope, right: PurgeScope) -> PurgeScope { + match (left, right) { + (PurgeScope::Account, _) | (_, PurgeScope::Account) => PurgeScope::Account, + (PurgeScope::AllStorage, _) | (_, PurgeScope::AllStorage) => PurgeScope::AllStorage, + (PurgeScope::Slots(mut left), PurgeScope::Slots(right)) => { + for slot in right { + if !left.contains(&slot) { + left.push(slot); + } + } + PurgeScope::Slots(left) + } + } } #[derive(Clone, Debug)] diff --git a/tests/reactive_reorg.rs b/tests/reactive_reorg.rs new file mode 100644 index 0000000..47326bf --- /dev/null +++ b/tests/reactive_reorg.rs @@ -0,0 +1,612 @@ +//! Manager-authored acceptance tests for reactive block journaling and reorg recovery. +//! +//! These tests cover the runtime-owned machinery that downstream crates should not +//! need to rebuild: journaling canonical block effects, handling removed/reorged +//! inputs, rolling back reversible storage writes, falling back to targeted purges +//! for irreversible effects, and canceling stale hash-pinned resyncs. +#![cfg(feature = "reactive")] + +mod common; + +use std::sync::Arc; + +use alloy_network::Ethereum; +use alloy_primitives::{Address, B256, Bytes, Log as PrimitiveLog, U256, keccak256}; +use alloy_rpc_types_eth::{Filter, Log}; +use anyhow::Result; + +use common::{install_mock_erc20, setup_cache}; +use evm_fork_cache::events::StateView; +use evm_fork_cache::reactive::{ + BlockRef, ChainStatus, HandlerError, HandlerId, HandlerOutcome, InputSource, + InvalidationReason, InvalidationRequest, LogInterest, ReactiveConfig, ReactiveContext, + ReactiveEffect, ReactiveHandler, ReactiveInput, ReactiveInputBatch, ReactiveInputRecord, + ReactiveInterest, ReactiveReport, ReactiveRuntime, ResyncBlock, ResyncId, ResyncPriority, + ResyncReason, ResyncRequest, ResyncTarget, RouteKeySpec, StateEffectQuality, +}; +use evm_fork_cache::{PurgeScope, StateUpdate}; + +fn block(number: u64, hash: B256, parent_hash: B256) -> BlockRef { + BlockRef { + number, + hash, + parent_hash: Some(parent_hash), + timestamp: Some(1_700_000_000 + number), + } +} + +fn rpc_log( + address: Address, + topics: Vec, + block: &BlockRef, + tx_index: u64, + log_index: u64, + removed: bool, +) -> Log { + Log { + inner: PrimitiveLog::new_unchecked(address, topics, Bytes::new()), + block_hash: Some(block.hash), + block_number: Some(block.number), + block_timestamp: block.timestamp, + transaction_hash: Some(B256::repeat_byte((tx_index + 1) as u8)), + transaction_index: Some(tx_index), + log_index: Some(log_index), + removed, + } +} + +fn included_context(block: BlockRef, log_index: u64) -> ReactiveContext { + ReactiveContext { + chain_id: Some(1), + source: InputSource::Batch, + chain_status: ChainStatus::Included { + block: block.clone(), + confirmations: 0, + }, + block: Some(block), + transaction_index: Some(0), + log_index: Some(log_index), + } +} + +fn reorged_context(dropped_from: BlockRef, log_index: u64) -> ReactiveContext { + ReactiveContext { + chain_id: Some(1), + source: InputSource::Batch, + chain_status: ChainStatus::Reorged { + dropped_from: dropped_from.clone(), + }, + block: Some(dropped_from), + transaction_index: Some(0), + log_index: Some(log_index), + } +} + +fn batch(input: ReactiveInput, ctx: ReactiveContext) -> ReactiveInputBatch { + ReactiveInputBatch::new(vec![ReactiveInputRecord::new(input, ctx)]) +} + +struct SlotWriter { + id: HandlerId, + address: Address, + slot: U256, + value: U256, +} + +impl ReactiveHandler for SlotWriter { + fn id(&self) -> HandlerId { + self.id.clone() + } + + fn interests(&self) -> Vec { + vec![ReactiveInterest::Logs(LogInterest { + provider_filter: Filter::new().address(self.address), + local_matcher: None, + route_key: Some(RouteKeySpec::EmitterAddress), + })] + } + + fn handle( + &self, + _ctx: &ReactiveContext, + _input: &ReactiveInput, + _state: &dyn StateView, + ) -> Result { + Ok(HandlerOutcome { + effects: vec![ReactiveEffect::StateUpdate(StateUpdate::slot( + self.address, + self.slot, + self.value, + ))], + quality: StateEffectQuality::ExactFromInput, + tags: vec![], + }) + } +} + +struct LogIndexSlotWriter { + address: Address, + slot: U256, +} + +impl ReactiveHandler for LogIndexSlotWriter { + fn id(&self) -> HandlerId { + HandlerId::new("log-index-slot-writer") + } + + fn interests(&self) -> Vec { + vec![ReactiveInterest::Logs(LogInterest { + provider_filter: Filter::new().address(self.address), + local_matcher: None, + route_key: Some(RouteKeySpec::EmitterAddress), + })] + } + + fn handle( + &self, + ctx: &ReactiveContext, + _input: &ReactiveInput, + _state: &dyn StateView, + ) -> Result { + Ok(HandlerOutcome { + effects: vec![ReactiveEffect::StateUpdate(StateUpdate::slot( + self.address, + self.slot, + U256::from(ctx.log_index.expect("test log context carries index")), + ))], + quality: StateEffectQuality::ExactFromInput, + tags: vec![], + }) + } +} + +struct PurgeHandler { + address: Address, +} + +impl ReactiveHandler for PurgeHandler { + fn id(&self) -> HandlerId { + HandlerId::new("purge-handler") + } + + fn interests(&self) -> Vec { + vec![ReactiveInterest::Logs(LogInterest { + provider_filter: Filter::new().address(self.address), + local_matcher: None, + route_key: Some(RouteKeySpec::EmitterAddress), + })] + } + + fn handle( + &self, + _ctx: &ReactiveContext, + _input: &ReactiveInput, + _state: &dyn StateView, + ) -> Result { + Ok(HandlerOutcome { + effects: vec![ReactiveEffect::Invalidate(InvalidationRequest { + scope: PurgeScope::AllStorage, + address: self.address, + reason: InvalidationReason::HandlerRequested, + })], + quality: StateEffectQuality::RequiresRepair, + tags: vec![], + }) + } +} + +struct ResyncOnlyHandler { + address: Address, + slot: U256, + block: ResyncBlock, +} + +impl ReactiveHandler for ResyncOnlyHandler { + fn id(&self) -> HandlerId { + HandlerId::new("resync-only") + } + + fn interests(&self) -> Vec { + vec![ReactiveInterest::Logs(LogInterest { + provider_filter: Filter::new().address(self.address), + local_matcher: None, + route_key: Some(RouteKeySpec::EmitterAddress), + })] + } + + fn handle( + &self, + _ctx: &ReactiveContext, + _input: &ReactiveInput, + _state: &dyn StateView, + ) -> Result { + Ok(HandlerOutcome { + effects: vec![ReactiveEffect::Resync(ResyncRequest { + id: ResyncId::new("hash-pinned-repair"), + reason: ResyncReason::HandlerRequested, + block: self.block.clone(), + targets: vec![ResyncTarget::StorageSlot { + address: self.address, + slot: self.slot, + }], + priority: ResyncPriority::High, + })], + quality: StateEffectQuality::AppliedWithPendingResync, + tags: vec![], + }) + } +} + +#[tokio::test] +async fn reactive_runtime_rolls_back_hash_pinned_resync_effects_with_dropped_block() -> Result<()> { + let address = Address::repeat_byte(0xa5); + let slot = U256::from(10); + let dropped = block(110, B256::repeat_byte(0xbb), B256::repeat_byte(0xab)); + let mut cache = setup_cache().await?; + install_mock_erc20(&mut cache, address); + cache + .db_mut() + .insert_account_storage(address, slot, U256::from(10))?; + cache.set_storage_batch_fetcher(Arc::new(move |requests, _block| { + requests + .into_iter() + .map(|(address, slot)| (address, slot, Ok(U256::from(42)))) + .collect() + })); + + let mut runtime = ReactiveRuntime::::new(ReactiveConfig::default()); + runtime.register_handler(Arc::new(SlotWriter { + id: HandlerId::new("provisional-writer"), + address, + slot, + value: U256::from(20), + }))?; + runtime.register_handler(Arc::new(ResyncOnlyHandler { + address, + slot, + block: ResyncBlock::Hash { + number: dropped.number, + hash: dropped.hash, + require_canonical: true, + }, + }))?; + + runtime.ingest_batch_with_resync( + &mut cache, + batch( + ReactiveInput::Log(rpc_log( + address, + vec![keccak256(b"WriteThenRepair()")], + &dropped, + 0, + 0, + false, + )), + included_context(dropped.clone(), 0), + ), + )?; + assert_eq!( + cache.cached_storage_value(address, slot), + Some(U256::from(42)) + ); + + let report = runtime.ingest_batch( + &mut cache, + batch( + ReactiveInput::Log(rpc_log( + address, + vec![keccak256(b"WriteThenRepair()")], + &dropped, + 0, + 0, + true, + )), + reorged_context(dropped, 0), + ), + )?; + assert_eq!( + cache.cached_storage_value(address, slot), + Some(U256::from(10)), + "reorg rollback must unwind authoritative resync writes as well as direct effects" + ); + let reorg = report + .reports + .iter() + .find_map(|report| match report.as_ref() { + ReactiveReport::Reorg(report) => Some(report), + _ => None, + }) + .expect("removed log emits a reorg report"); + assert_eq!(reorg.rollback_updates.len(), 2); + assert_eq!(reorg.rollback_diff.slots[0].old, U256::from(42)); + assert_eq!(reorg.rollback_diff.slots[0].new, U256::from(20)); + assert_eq!(reorg.rollback_diff.slots[1].old, U256::from(20)); + assert_eq!(reorg.rollback_diff.slots[1].new, U256::from(10)); + + Ok(()) +} + +#[tokio::test] +async fn reactive_runtime_rolls_back_removed_log_storage_effects() -> Result<()> { + let address = Address::repeat_byte(0xa1); + let slot = U256::from(7); + let dropped = block(70, B256::repeat_byte(0x70), B256::repeat_byte(0x6f)); + let mut cache = setup_cache().await?; + install_mock_erc20(&mut cache, address); + cache + .db_mut() + .insert_account_storage(address, slot, U256::from(10))?; + + let mut runtime = ReactiveRuntime::::new(ReactiveConfig::default()); + runtime.register_handler(Arc::new(SlotWriter { + id: HandlerId::new("slot-writer"), + address, + slot, + value: U256::from(20), + }))?; + + runtime.ingest_batch( + &mut cache, + batch( + ReactiveInput::Log(rpc_log( + address, + vec![keccak256(b"Write(uint256)")], + &dropped, + 0, + 0, + false, + )), + included_context(dropped.clone(), 0), + ), + )?; + assert_eq!( + cache.cached_storage_value(address, slot), + Some(U256::from(20)) + ); + + let report = runtime.ingest_batch( + &mut cache, + batch( + ReactiveInput::Log(rpc_log( + address, + vec![keccak256(b"Write(uint256)")], + &dropped, + 0, + 0, + true, + )), + reorged_context(dropped.clone(), 0), + ), + )?; + + assert_eq!( + cache.cached_storage_value(address, slot), + Some(U256::from(10)), + "removed logs should roll back reversible storage writes" + ); + let reorg = report + .reports + .iter() + .find_map(|report| match report.as_ref() { + ReactiveReport::Reorg(report) => Some(report), + _ => None, + }) + .expect("removed log emits a reorg report"); + assert_eq!(reorg.dropped_blocks, vec![dropped]); + assert_eq!(reorg.rollback_updates.len(), 1); + assert!(reorg.purge_updates.is_empty()); + assert_eq!(reorg.rollback_diff.slots[0].old, U256::from(20)); + assert_eq!(reorg.rollback_diff.slots[0].new, U256::from(10)); + + Ok(()) +} + +#[tokio::test] +async fn reactive_runtime_reorgs_parent_mismatch_before_replacement_block() -> Result<()> { + let address = Address::repeat_byte(0xa2); + let slot = U256::from(8); + let parent = block(79, B256::repeat_byte(0x79), B256::repeat_byte(0x78)); + let dropped = block(80, B256::repeat_byte(0x80), parent.hash); + let replacement = block(80, B256::repeat_byte(0x81), parent.hash); + let mut cache = setup_cache().await?; + + let mut runtime = ReactiveRuntime::::new(ReactiveConfig::default()); + runtime.register_handler(Arc::new(LogIndexSlotWriter { address, slot }))?; + runtime.ingest_batch( + &mut cache, + batch( + ReactiveInput::Log(rpc_log( + address, + vec![keccak256(b"Base(uint256)")], + &parent, + 0, + 10, + false, + )), + included_context(parent.clone(), 10), + ), + )?; + runtime.ingest_batch( + &mut cache, + batch( + ReactiveInput::Log(rpc_log( + address, + vec![keccak256(b"Dropped(uint256)")], + &dropped, + 0, + 20, + false, + )), + included_context(dropped.clone(), 20), + ), + )?; + assert_eq!( + cache.cached_storage_value(address, slot), + Some(U256::from(20)) + ); + + let report = runtime.ingest_batch( + &mut cache, + batch( + ReactiveInput::Log(rpc_log( + address, + vec![keccak256(b"Replacement(uint256)")], + &replacement, + 0, + 30, + false, + )), + included_context(replacement.clone(), 30), + ), + )?; + + assert_eq!( + cache.cached_storage_value(address, slot), + Some(U256::from(30)), + "replacement block should apply after the dropped block is rolled back" + ); + let reorg = report + .reports + .iter() + .find_map(|report| match report.as_ref() { + ReactiveReport::Reorg(report) => Some(report), + _ => None, + }) + .expect("replacement block emits a parent-mismatch reorg report"); + assert_eq!(reorg.dropped_blocks, vec![dropped]); + assert_eq!(reorg.rollback_updates.len(), 1); + assert!(reorg + .dropped_inputs + .iter() + .any(|input| matches!(input, evm_fork_cache::reactive::InputRef::Log { block_hash, .. } if *block_hash == B256::repeat_byte(0x80)))); + + Ok(()) +} + +#[tokio::test] +async fn reactive_runtime_falls_back_to_purge_for_irreversible_dropped_effects() -> Result<()> { + let address = Address::repeat_byte(0xa3); + let slot = U256::from(9); + let dropped = block(90, B256::repeat_byte(0x90), B256::repeat_byte(0x8f)); + let replacement = block(90, B256::repeat_byte(0x91), B256::repeat_byte(0x8f)); + let unrelated = Address::repeat_byte(0xf0); + let mut cache = setup_cache().await?; + install_mock_erc20(&mut cache, address); + cache + .db_mut() + .insert_account_storage(address, slot, U256::from(99))?; + + let mut runtime = ReactiveRuntime::::new(ReactiveConfig::default()); + runtime.register_handler(Arc::new(PurgeHandler { address }))?; + runtime.ingest_batch( + &mut cache, + batch( + ReactiveInput::Log(rpc_log( + address, + vec![keccak256(b"Purge()")], + &dropped, + 0, + 0, + false, + )), + included_context(dropped.clone(), 0), + ), + )?; + assert_eq!(cache.cached_storage_value(address, slot), Some(U256::ZERO)); + + let report = runtime.ingest_batch( + &mut cache, + batch( + ReactiveInput::Log(rpc_log( + unrelated, + vec![keccak256(b"Replacement()")], + &replacement, + 0, + 0, + false, + )), + included_context(replacement, 0), + ), + )?; + + let reorg = report + .reports + .iter() + .find_map(|report| match report.as_ref() { + ReactiveReport::Reorg(report) => Some(report), + _ => None, + }) + .expect("replacement block emits a reorg report"); + assert_eq!(reorg.dropped_blocks, vec![dropped]); + assert!(reorg.rollback_updates.is_empty()); + assert_eq!( + reorg.purge_updates, + vec![StateUpdate::purge(address, PurgeScope::AllStorage)] + ); + + Ok(()) +} + +#[tokio::test] +async fn reactive_runtime_cancels_hash_pinned_resyncs_for_dropped_blocks() -> Result<()> { + let address = Address::repeat_byte(0xa4); + let dropped = block(100, B256::repeat_byte(0xaa), B256::repeat_byte(0x99)); + let mut cache = setup_cache().await?; + + let mut runtime = ReactiveRuntime::::new(ReactiveConfig::default()); + runtime.register_handler(Arc::new(ResyncOnlyHandler { + address, + slot: U256::from(1), + block: ResyncBlock::Hash { + number: dropped.number, + hash: dropped.hash, + require_canonical: true, + }, + }))?; + + let first = runtime.ingest_batch( + &mut cache, + batch( + ReactiveInput::Log(rpc_log( + address, + vec![keccak256(b"NeedsRepair()")], + &dropped, + 0, + 0, + false, + )), + included_context(dropped.clone(), 0), + ), + )?; + assert_eq!(first.resyncs.len(), 1); + + let second = runtime.ingest_batch( + &mut cache, + batch( + ReactiveInput::Log(rpc_log( + address, + vec![keccak256(b"NeedsRepair()")], + &dropped, + 0, + 0, + true, + )), + reorged_context(dropped, 0), + ), + )?; + let reorg = second + .reports + .iter() + .find_map(|report| match report.as_ref() { + ReactiveReport::Reorg(report) => Some(report), + _ => None, + }) + .expect("dropped block emits a reorg report"); + assert_eq!(reorg.canceled_resyncs.len(), 1); + assert_eq!( + reorg.canceled_resyncs[0].id, + ResyncId::new("hash-pinned-repair") + ); + + Ok(()) +} From 590bebc9bc52dcebed43f3c47fe7e727aec174c0 Mon Sep 17 00:00:00 2001 From: Kai Aldag Date: Tue, 23 Jun 2026 14:59:49 +0100 Subject: [PATCH 2/3] Add protocol-neutral cold-start sync (reactive feature) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds a cold-start sync mechanism gated under the existing `reactive` feature: a consumer declares the storage to warm on startup (verify / probe / accounts / discover) and drives a bounded multi-round loop via a pure `ColdStartPlanner`. Surfaces a per-slot `SlotFetch` (Value / Zero / FetchFailed / NotAttempted) that closes the archive-miss gap — distinguishing a genuine on-chain zero from a fetch failure, which `verify_slots` could not express. - src/cold_start/: ColdStartPlan/Call/Results/CallResult/RoundOutcome, ColdStartPlanner/Step, ColdStartConfig/Pin/RunReport, ColdStartError, and the driver methods EvmCache::{execute_cold_start_round, run_cold_start}. - src/freshness.rs: SlotOutcome / SlotFetch (ungated, beside SlotChange). - src/cache/mod.rs: verify_slots_core extraction (verify_slots / reconcile_slots behaviour unchanged), verify_slots_with_outcomes, ensure_account_blocking; classify and block_in_place_handle widened to pub(crate). - 19 offline acceptance tests (tests/cold_start.rs) + setup_cache_with_asserter. Composes existing primitives (verify_slots_inner, inject_storage_batch_fresh, call_raw_with_access_list, ensure_account, StorageBatchFetchFn). No AMM concepts enter evm-fork-cache. `cargo test --no-default-features` stays green. Co-Authored-By: Claude Opus 4.8 --- src/cache/mod.rs | 98 +++- src/cold_start/config.rs | 141 +++++ src/cold_start/driver.rs | 261 +++++++++ src/cold_start/error.rs | 23 + src/cold_start/mod.rs | 46 ++ src/cold_start/plan.rs | 49 ++ src/cold_start/planner.rs | 36 ++ src/cold_start/results.rs | 56 ++ src/freshness.rs | 47 ++ src/lib.rs | 10 +- tests/cold_start.rs | 1051 +++++++++++++++++++++++++++++++++++++ tests/common/mod.rs | 12 + 12 files changed, 1820 insertions(+), 10 deletions(-) create mode 100644 src/cold_start/config.rs create mode 100644 src/cold_start/driver.rs create mode 100644 src/cold_start/error.rs create mode 100644 src/cold_start/mod.rs create mode 100644 src/cold_start/plan.rs create mode 100644 src/cold_start/planner.rs create mode 100644 src/cold_start/results.rs create mode 100644 tests/cold_start.rs diff --git a/src/cache/mod.rs b/src/cache/mod.rs index 19af0e4..e58fbd3 100644 --- a/src/cache/mod.rs +++ b/src/cache/mod.rs @@ -47,7 +47,7 @@ use tracing::{debug, instrument, trace, warn}; use crate::access_set::StorageAccessList; use crate::errors::{SimError, SimulationError, SimulationResult}; -use crate::freshness::SlotChange; +use crate::freshness::{SlotChange, SlotFetch, SlotOutcome}; use crate::inspector::TransferInspector; use crate::state_update::{ AccountChange, AccountPatch, PurgeRecord, PurgeScope, SkippedAccountPatch, SkippedBalanceDelta, @@ -98,7 +98,7 @@ pub type StorageBatchFetchFn = Arc< /// degrade to a typed error instead. /// /// Requires a **multi-thread** tokio runtime. -fn block_in_place_handle() -> Result { +pub(crate) fn block_in_place_handle() -> Result { match tokio::runtime::Handle::try_current() { Ok(handle) => match handle.runtime_flavor() { tokio::runtime::RuntimeFlavor::CurrentThread => Err(anyhow!( @@ -2003,9 +2003,9 @@ impl EvmCache { /// /// For each slot whose freshly-fetched value differs from the cached value, /// the fresh value is written into the cache via - /// [`inject_storage_batch`](Self::inject_storage_batch) and a [`SlotChange`] - /// is recorded. Slots that are unchanged, or that the fetcher fails to - /// return, are left as-is. Returns the set of changed slots. + /// [`inject_storage_batch_fresh`](Self::inject_storage_batch_fresh) and a + /// [`SlotChange`] is recorded. Slots that are unchanged, or that the fetcher + /// fails to return, are left as-is. Returns the set of changed slots. /// /// Requires a batch fetcher (set at construction or via /// [`set_storage_batch_fetcher`](Self::set_storage_batch_fetcher)); errors if @@ -2024,8 +2024,60 @@ impl EvmCache { &mut self, slots: &[(Address, U256)], ) -> Result<(Vec, usize)> { + let (changed, outcomes) = self.verify_slots_core(slots)?; + let fetched_ok = outcomes + .iter() + .filter(|o| matches!(o.fetch, SlotFetch::Value(_) | SlotFetch::Zero)) + .count(); + Ok((changed, fetched_ok)) + } + + /// Classify a single fetched slot value into a [`SlotFetch`]. + /// + /// This is purely the *fetch* classification (`Value` / `Zero` / + /// `FetchFailed`); it is independent of change detection, which compares the + /// fetched value to the cached baseline separately. A non-zero `Ok` is + /// [`SlotFetch::Value`], a genuine `Ok(0)` is [`SlotFetch::Zero`], and an + /// `Err` is [`SlotFetch::FetchFailed`] carrying the error string. + /// + /// Shared with the cold-start probe phase + /// ([`execute_cold_start_round`](Self::execute_cold_start_round)) so the + /// single classification is reused rather than duplicated. + pub(crate) fn classify(fetched: Result) -> SlotFetch { + match fetched { + Ok(v) if v != U256::ZERO => SlotFetch::Value(v), + Ok(_) => SlotFetch::Zero, + Err(e) => SlotFetch::FetchFailed { + reason: e.to_string(), + }, + } + } + + /// Core slot-verification loop shared by [`verify_slots_inner`](Self::verify_slots_inner) + /// and [`verify_slots_with_outcomes`](Self::verify_slots_with_outcomes). + /// + /// Fetches every slot via the batch fetcher and, for each slot, performs two + /// **independent** reads of the same fetched value: + /// + /// 1. *Fetch classification* — every slot (including failed ones) produces one + /// [`SlotOutcome`] via [`classify`](Self::classify): `Value` / `Zero` / + /// `FetchFailed`. + /// 2. *Change detection* — a successfully-fetched value that differs from the + /// cached baseline (`old`, defaulting to `ZERO` for an unseen slot) is + /// injected via [`inject_storage_batch_fresh`](Self::inject_storage_batch_fresh) + /// and recorded as a [`SlotChange`]. + /// + /// These two reads are deliberately not collapsed: a genuine `Ok(0)` on a slot + /// whose cached value was also `0` yields [`SlotFetch::Zero`] **and** no + /// `SlotChange`. The returned `outcomes` vec has exactly one entry per + /// requested slot. An empty `slots` input short-circuits to empty results + /// without requiring a fetcher; otherwise a missing fetcher is an error. + fn verify_slots_core( + &mut self, + slots: &[(Address, U256)], + ) -> Result<(Vec, Vec)> { if slots.is_empty() { - return Ok((Vec::new(), 0)); + return Ok((Vec::new(), Vec::new())); } let fetcher = self .storage_batch_fetcher @@ -2043,9 +2095,21 @@ impl EvmCache { let results = (fetcher)(slots.to_vec(), Some(self.block)); let mut changed = Vec::new(); + let mut outcomes = Vec::with_capacity(results.len()); let mut to_inject = Vec::new(); - let mut fetched_ok = 0usize; for (addr, slot, fetched) in results { + // Read 1: classify the fetch outcome for every slot, failed or not. + let fetch = Self::classify(match &fetched { + Ok(v) => Ok(*v), + Err(e) => Err(anyhow!("{e}")), + }); + outcomes.push(SlotOutcome { + address: addr, + slot, + fetch, + }); + + // Read 2: change detection, independent of the classification above. let fresh = match fetched { Ok(value) => value, Err(e) => { @@ -2053,7 +2117,6 @@ impl EvmCache { continue; } }; - fetched_ok += 1; // A slot the cache never saw is treated as old = ZERO (the value a // sim would have read), so a non-zero fresh value counts as a change. let old = cached @@ -2075,7 +2138,24 @@ impl EvmCache { if !to_inject.is_empty() { self.inject_storage_batch_fresh(&to_inject); } - Ok((changed, fetched_ok)) + Ok((changed, outcomes)) + } + + /// Like [`verify_slots`](Self::verify_slots), but additionally returns one + /// [`SlotOutcome`] per requested slot (including slots the fetcher failed to + /// return), classified as `Value` / `Zero` / `FetchFailed`. + /// + /// This is the per-slot surface the cold-start driver consumes: it + /// distinguishes a genuine on-chain zero from a fetch failure for every slot, + /// closing the archive-miss gap. It is a pure alias of + /// [`verify_slots_core`](Self::verify_slots_core) and shares its injection + /// behaviour with [`verify_slots`](Self::verify_slots). + #[cfg(feature = "reactive")] + pub(crate) fn verify_slots_with_outcomes( + &mut self, + slots: &[(Address, U256)], + ) -> Result<(Vec, Vec)> { + self.verify_slots_core(slots) } /// Reconciliation re-read used by [`EventPipeline::reconcile`](crate::events::EventPipeline::reconcile). diff --git a/src/cold_start/config.rs b/src/cold_start/config.rs new file mode 100644 index 0000000..42241b8 --- /dev/null +++ b/src/cold_start/config.rs @@ -0,0 +1,141 @@ +//! Cold-start run configuration and the accumulated run report. + +use alloy_primitives::B256; + +use crate::cold_start::plan::ColdStartPlan; +use crate::cold_start::results::ColdStartResults; +use crate::freshness::SlotFetch; + +/// Configuration for a cold-start run. +/// +/// `Default` is hand-written (`max_rounds: 8`, `pin: ColdStartPin::CachePinned`) +/// rather than derived, because a derived `max_rounds: 0` would trip +/// [`RoundBudgetExceeded`](crate::cold_start::ColdStartError::RoundBudgetExceeded) +/// before any round ran. `max_rounds >= 1` is a precondition. +#[derive(Clone, Debug)] +pub struct ColdStartConfig { + /// Hard upper bound on **executed** rounds. A planner that returns `Continue` + /// past this bound yields + /// [`RoundBudgetExceeded`](crate::cold_start::ColdStartError::RoundBudgetExceeded). + /// Must be `>= 1`. + pub max_rounds: usize, + /// Block-pinning policy for the run's reads. + pub pin: ColdStartPin, +} + +impl Default for ColdStartConfig { + fn default() -> Self { + Self { + max_rounds: 8, + pin: ColdStartPin::CachePinned, + } + } +} + +/// How the driver pins the block for the run's reads. +#[derive(Clone, Debug)] +pub enum ColdStartPin { + /// Use the cache's currently-pinned block (`self.block`) for every round. + CachePinned, + /// Pin every round to an explicit hash for the whole run (reorg-stable cold + /// start), restoring the prior block on completion. With + /// `require_canonical: true`, the provider rejects a reorged hash so the run + /// fails fast. + Hash { + /// Block number associated with the hash (advisory). + number: u64, + /// Block hash to pin all reads to. + hash: B256, + /// Whether the provider must reject a non-canonical hash. + require_canonical: bool, + }, +} + +/// Summary of a completed (or partially-completed) cold-start run. +/// +/// Accumulated round-by-round as the run progresses; on a mid-round hard error +/// the partial round is absorbed before the error propagates, so the report +/// reflects work already done. +#[derive(Clone, Debug, Default)] +pub struct ColdStartRunReport { + /// Number of rounds executed. + pub rounds: usize, + /// Total verify slots requested across all rounds. + pub verified_slots: usize, + /// Total slots that changed and were injected. + pub changed_slots: usize, + /// Total distinct accounts touched by discover calls. + pub discovered_accounts: usize, + /// Total distinct slots touched by discover calls. + pub discovered_slots: usize, + /// Total verify + probe slots whose fetch failed. + pub failed_slots: usize, + /// One summary per round, in execution order. + pub per_round: Vec, +} + +/// Per-round breakdown recorded in [`ColdStartRunReport::per_round`]. +#[derive(Clone, Debug, Default)] +pub struct ColdStartRoundSummary { + /// Verify slots requested this round. + pub verify_requested: usize, + /// Verify slots that changed and were injected. + pub verify_changed: usize, + /// Verify slots whose fetch failed. + pub verify_failed: usize, + /// Probe slots requested this round. + pub probe_requested: usize, + /// Probe slots whose fetch failed. + pub probe_failed: usize, + /// Discover calls issued this round. + pub discover_calls: usize, + /// Slots touched by this round's discover calls. + pub discovered_slots: usize, +} + +impl ColdStartRunReport { + /// Fold one round's plan and results into the report. + /// + /// Plain field accumulation, no IO. `failed_slots` counts + /// [`FetchFailed`](crate::freshness::SlotFetch::FetchFailed) outcomes across + /// both the verify (`fetched`) and probe (`probed`) phases. + pub(crate) fn absorb_round(&mut self, plan: &ColdStartPlan, results: &ColdStartResults) { + self.rounds += 1; + let verify_failed = results + .fetched + .iter() + .filter(|o| matches!(o.fetch, SlotFetch::FetchFailed { .. })) + .count(); + let probe_failed = results + .probed + .iter() + .filter(|o| matches!(o.fetch, SlotFetch::FetchFailed { .. })) + .count(); + let discovered_slots: usize = results + .discovered + .iter() + .map(|d| d.access.slots.len()) + .sum(); + let discovered_accounts: usize = results + .discovered + .iter() + .map(|d| d.access.accounts.len()) + .sum(); + + self.verified_slots += plan.verify.len(); + self.changed_slots += results.verified.len(); + self.failed_slots += verify_failed + probe_failed; + self.discovered_slots += discovered_slots; + self.discovered_accounts += discovered_accounts; + + self.per_round.push(ColdStartRoundSummary { + verify_requested: plan.verify.len(), + verify_changed: results.verified.len(), + verify_failed, + probe_requested: plan.probe.len(), + probe_failed, + discover_calls: plan.discover.len(), + discovered_slots, + }); + } +} diff --git a/src/cold_start/driver.rs b/src/cold_start/driver.rs new file mode 100644 index 0000000..dbad821 --- /dev/null +++ b/src/cold_start/driver.rs @@ -0,0 +1,261 @@ +//! The cold-start driver: inherent [`EvmCache`] methods that execute rounds and +//! run the bounded multi-round loop. +//! +//! The driver performs every fetch and call; planners are pure and IO-free. The +//! whole module is gated behind the `reactive` feature. + +use std::collections::HashSet; + +use alloy_eips::BlockId; +use alloy_primitives::Address; +use anyhow::Result; + +use crate::cache::{EvmCache, block_in_place_handle}; +use crate::cold_start::config::{ColdStartConfig, ColdStartPin, ColdStartRunReport}; +use crate::cold_start::error::ColdStartError; +use crate::cold_start::plan::ColdStartPlan; +use crate::cold_start::planner::{ColdStartPlanner, ColdStartStep}; +use crate::cold_start::results::{ColdStartCallResult, ColdStartResults, RoundOutcome}; +use crate::events::StateView; +use crate::freshness::{SlotFetch, SlotOutcome}; + +impl EvmCache { + /// View `self` as a [`StateView`] for handing to a planner. + /// + /// The returned borrow must be **inlined** into each planner call rather than + /// held across [`execute_cold_start_round`](Self::execute_cold_start_round), + /// which needs `&mut self` (holding the shared borrow across it is a borrowck + /// error). + fn state_view(&self) -> &dyn StateView { + self + } + + /// Pre-seed an account synchronously, bridging [`ensure_account`] across the + /// async boundary. + /// + /// [`ensure_account`](EvmCache::ensure_account) early-returns for an + /// already-present account; for a missing one it issues a backend fetch that + /// can return `Err`. This is the only producer of a cold-start round hard error + /// in the accounts phase. Requires a **multi-thread** tokio runtime (the + /// cold-start runtime precondition); on a current-thread runtime or with no + /// runtime present it returns a typed error via + /// [`block_in_place_handle`](crate::cache::block_in_place_handle). + pub(crate) fn ensure_account_blocking(&mut self, address: Address) -> Result<()> { + let handle = block_in_place_handle()?; + tokio::task::block_in_place(|| handle.block_on(self.ensure_account(address))) + } + + /// Execute a single cold-start round and return its (possibly partial) outcome. + /// + /// Fixed phase order: **accounts → verify → probe → discover**. + /// + /// Per-round fetcher guard: if the plan declares any verify or probe slots and + /// the cache has no storage batch fetcher, the round short-circuits with + /// [`ColdStartError::NoBatchFetcher`] before issuing any read. A round + /// declaring only accounts/discover runs without a fetcher. + /// + /// - **accounts (first):** each `plan.accounts` address is pre-seeded via + /// `ensure_account_blocking`. A failure here is a hard error in the first + /// phase, so nothing after it ran: every declared verify/probe slot is marked + /// [`SlotFetch::NotAttempted`] and the round returns with `error: Some(..)`. + /// This is the only producer of `NotAttempted`. + /// - **verify:** each verify slot is re-fetched, classified into + /// `results.fetched`, and (when changed) injected and recorded in + /// `results.verified`. + /// - **probe:** each probe slot is re-fetched at the pinned block and + /// classified into `results.probed` via the same shared `Result` + /// classification verify uses. Unlike verify, a probe injects nothing and + /// records no [`SlotChange`](crate::freshness::SlotChange): it is the + /// archive-miss classification for slots a consumer does not want to warm. + /// - **discover (last):** each [`ColdStartCall`](crate::cold_start::ColdStartCall) + /// is executed via + /// [`call_raw_with_access_list`](EvmCache::call_raw_with_access_list), its + /// access list filtered by `restrict_to`, and the result pushed to + /// `results.discovered`. A discover failure preserves the verify/probe + /// outcomes already computed this round (they ran earlier, so they are *not* + /// `NotAttempted`); the failing call and all subsequent discover calls are + /// dropped, and the round returns with `error: Some(..)`. + pub fn execute_cold_start_round(&mut self, plan: &ColdStartPlan) -> RoundOutcome { + let mut results = ColdStartResults::default(); + + // Per-round fetcher guard: only fires for verify/probe-bearing rounds. + if (!plan.verify.is_empty() || !plan.probe.is_empty()) + && self.storage_batch_fetcher().is_none() + { + return RoundOutcome { + results, + error: Some(ColdStartError::NoBatchFetcher), + }; + } + + // Accounts phase (first): pre-seed each declared account. A failure here + // short-circuits the round before verify/probe/discover run, so every + // declared verify/probe slot is synthesized as NotAttempted. + for &address in &plan.accounts { + if let Err(e) = self.ensure_account_blocking(address) { + results.fetched = not_attempted_outcomes(&plan.verify); + results.probed = not_attempted_outcomes(&plan.probe); + return RoundOutcome { + results, + error: Some(ColdStartError::Fetch(e)), + }; + } + } + + // Verify phase: classify every slot, inject and record the changed ones. + if !plan.verify.is_empty() { + match self.verify_slots_with_outcomes(&plan.verify) { + Ok((changed, outcomes)) => { + results.verified = changed; + results.fetched = outcomes; + } + Err(e) => { + // The only error surface of verify_slots_with_outcomes is the + // missing-fetcher guard, already front-run above; surface any + // residual error explicitly rather than panicking. + return RoundOutcome { + results, + error: Some(ColdStartError::Fetch(e)), + }; + } + } + } + + // Probe phase: classify each declared slot at the pinned block WITHOUT + // injecting (the archive-miss classification for slots a consumer does + // not want to warm). It records into `results.probed` only — never + // `results.verified`, and never writes the cache. + if !plan.probe.is_empty() { + // The per-round guard already ensured a fetcher is present for a + // probe-bearing round. Read pinned to self.block (NOT read_storage_slot, + // which is unpinned), classify, and inject NOTHING. + let fetcher = self + .storage_batch_fetcher() + .cloned() + .expect("probe-bearing round guarded a fetcher above"); + let probed = (fetcher)(plan.probe.clone(), Some(self.block())); + results.probed = probed + .into_iter() + .map(|(address, slot, fetched)| SlotOutcome { + address, + slot, + fetch: EvmCache::classify(fetched), + }) + .collect(); + } + + // Discover phase (last): run each view-call, filter by restrict_to. A + // failure drops this and every subsequent call but preserves the + // verify/probe outcomes already computed above. + for call in &plan.discover { + match self.call_raw_with_access_list(call.from, call.to, call.calldata.clone()) { + Ok((result, mut access)) => { + if let Some(list) = &call.restrict_to { + let keep: HashSet
= list.iter().copied().collect(); + access.slots.retain(|(a, _)| keep.contains(a)); + access.accounts.retain(|a| keep.contains(a)); + } + results + .discovered + .push(ColdStartCallResult { result, access }); + } + Err(e) => { + return RoundOutcome { + results, + error: Some(ColdStartError::Fetch(e)), + }; + } + } + } + + RoundOutcome { + results, + error: None, + } + } + + /// Run a bounded multi-round cold start driven by `planner`. + /// + /// Pin handling per `config.pin`: [`ColdStartPin::CachePinned`] is a no-op; + /// [`ColdStartPin::Hash`] pins every round to + /// `BlockId::from((hash, Some(require_canonical)))`, capturing the prior block + /// and restoring it on **every** exit path (success, budget-exceeded, and + /// mid-round error). + /// + /// The loop checks the round budget at the top: with `max_rounds = N`, rounds + /// `0..N` execute and a planner still returning `Continue` after round `N` + /// yields [`RoundBudgetExceeded`](ColdStartError::RoundBudgetExceeded). Each + /// round's results are absorbed into the report **before** its `error` is + /// checked; a round error propagates after restoring the pin and **without** + /// calling `on_results`. + pub fn run_cold_start( + &mut self, + planner: &mut dyn ColdStartPlanner, + config: ColdStartConfig, + ) -> Result { + // Pin handling: capture the block to restore (None == no restore needed). + let restore: Option = match config.pin { + ColdStartPin::CachePinned => None, + ColdStartPin::Hash { + hash, + require_canonical, + .. + } => { + let prev = self.block(); + self.set_block(BlockId::from((hash, Some(require_canonical)))); + Some(prev) + } + }; + + let mut report = ColdStartRunReport::default(); + // Borrow inlined, not hoisted across the &mut self round call. + let mut plan = planner.initial_plan(self.state_view()); + + loop { + if report.rounds >= config.max_rounds { + if let Some(prev) = restore { + self.set_block(prev); + } + return Err(ColdStartError::RoundBudgetExceeded { + max_rounds: config.max_rounds, + }); + } + + let outcome = self.execute_cold_start_round(&plan); + report.absorb_round(&plan, &outcome.results); + + if let Some(err) = outcome.error { + if let Some(prev) = restore { + self.set_block(prev); + } + return Err(err); + } + + match planner.on_results(&outcome.results, self.state_view()) { + ColdStartStep::Done => break, + ColdStartStep::Continue(next) => plan = next, + } + } + + if let Some(prev) = restore { + self.set_block(prev); + } + Ok(report) + } +} + +/// Synthesize one [`SlotFetch::NotAttempted`] outcome per declared slot. +/// +/// Used when an accounts-phase hard error short-circuits a round before the +/// verify/probe phases run: every declared slot is reported as `NotAttempted` +/// rather than silently dropped. +fn not_attempted_outcomes(slots: &[(Address, alloy_primitives::U256)]) -> Vec { + slots + .iter() + .map(|&(address, slot)| SlotOutcome { + address, + slot, + fetch: SlotFetch::NotAttempted, + }) + .collect() +} diff --git a/src/cold_start/error.rs b/src/cold_start/error.rs new file mode 100644 index 0000000..ecf9f85 --- /dev/null +++ b/src/cold_start/error.rs @@ -0,0 +1,23 @@ +//! The typed error surface for cold-start runs. + +/// A hard error that aborts a cold-start round or run. +/// +/// Deliberately typed (no `#[from] anyhow::Error` blanket arm): composed-primitive +/// errors are converted explicitly at call sites via [`ColdStartError::Fetch`], so +/// a partial round's outcomes are never silently collapsed. +#[derive(Debug, thiserror::Error)] +pub enum ColdStartError { + /// A round declared verify/probe slots but the cache has no storage batch + /// fetcher configured. + #[error("cold-start requires a storage batch fetcher")] + NoBatchFetcher, + /// The planner kept returning `Continue` past `max_rounds` executed rounds. + #[error("cold-start round budget exceeded ({max_rounds})")] + RoundBudgetExceeded { + /// The configured maximum number of executed rounds. + max_rounds: usize, + }, + /// A composed fetch/call error, carrying the underlying cause explicitly. + #[error("cold-start fetch error: {0}")] + Fetch(anyhow::Error), +} diff --git a/src/cold_start/mod.rs b/src/cold_start/mod.rs new file mode 100644 index 0000000..6747fb5 --- /dev/null +++ b/src/cold_start/mod.rs @@ -0,0 +1,46 @@ +//! Protocol-neutral cold-start sync for [`EvmCache`](crate::cache::EvmCache). +//! +//! Cold start declares rounds of authoritative slot work — verify, probe, +//! accounts, discover — and drives a bounded multi-round loop via a pure +//! [`ColdStartPlanner`]. Every verify and probe slot yields a per-slot +//! [`SlotOutcome`] distinguishing a genuine on-chain zero ([`SlotFetch::Zero`]) +//! from a fetch failure ([`SlotFetch::FetchFailed`]) — closing the "archive-miss" +//! gap where a transient fetch failure was indistinguishable from absence. +//! +//! The module is gated behind the `reactive` feature. The per-slot +//! [`SlotOutcome`] / [`SlotFetch`] surface lives ungated in +//! [`crate::freshness`] and is re-exported here for consumer ergonomics. +//! +//! # Design +//! +//! - The driver performs every fetch and call; planners are pure and IO-free, +//! handed only a [`StateView`](crate::events::StateView) and a +//! [`ColdStartResults`]. +//! - Verify-phase changes are injected via the dual-layer +//! [`inject_storage_batch_fresh`](crate::cache::EvmCache::inject_storage_batch_fresh) +//! and are visible to the next `on_results` through the state view. +//! - A run can be pinned to a block hash via [`ColdStartPin::Hash`]; with +//! `require_canonical: true`, a reorged hash makes the run fail fast. This is +//! the cold-start reorg defense by design. +//! +//! # Runtime requirement +//! +//! Like the rest of the crate's RPC seams, cold-start fetching drives async work +//! synchronously and must run on a **multi-thread** tokio runtime. + +mod config; +mod driver; +mod error; +mod plan; +mod planner; +mod results; + +pub use config::{ColdStartConfig, ColdStartPin, ColdStartRoundSummary, ColdStartRunReport}; +pub use error::ColdStartError; +pub use plan::{ColdStartCall, ColdStartPlan}; +pub use planner::{ColdStartPlanner, ColdStartStep}; +pub use results::{ColdStartCallResult, ColdStartResults, RoundOutcome}; + +// The per-slot fetch surface is ungated in `freshness`; re-export so +// `evm_fork_cache::cold_start::SlotFetch` resolves for consumers. +pub use crate::freshness::{SlotFetch, SlotOutcome}; diff --git a/src/cold_start/plan.rs b/src/cold_start/plan.rs new file mode 100644 index 0000000..3120a38 --- /dev/null +++ b/src/cold_start/plan.rs @@ -0,0 +1,49 @@ +//! What a cold-start round declares it wants done. +//! +//! A [`ColdStartPlan`] is a pure, IO-free description handed to the driver. All +//! four phases are optional and an empty plan is a valid no-op round. + +use alloy_primitives::{Address, Bytes, U256}; + +/// A single round of cold-start work, declared by a +/// [`ColdStartPlanner`](crate::cold_start::ColdStartPlanner). +/// +/// All four phases are optional; an empty plan is a valid no-op round. The driver +/// executes the phases in a fixed order: **accounts → verify → probe → discover**. +/// +/// - `verify` slots are authoritatively re-fetched, classified, and (when changed) +/// injected into the cache via the dual-layer +/// [`inject_storage_batch_fresh`](crate::cache::EvmCache::inject_storage_batch_fresh). +/// - `probe` slots are classified at the pinned block **without** injecting. +/// - `accounts` are pre-seeded into the cache before discovery runs. +/// - `discover` view-calls capture the `(address, slot)` pairs and accounts they +/// touch. +#[derive(Clone, Debug, Default)] +pub struct ColdStartPlan { + /// Slots to authoritatively re-fetch, classify, and inject when changed. + pub verify: Vec<(Address, U256)>, + /// Slots to classify at the pinned block without injecting. + pub probe: Vec<(Address, U256)>, + /// Accounts to pre-seed into the cache before discovery. + pub accounts: Vec
, + /// View-calls whose touched slots and accounts are captured. + pub discover: Vec, +} + +/// A read-only view-call whose touched storage and accounts are captured during +/// the discover phase. +/// +/// When `restrict_to` is `Some`, the captured `slots` and `accounts` are filtered +/// to the named addresses; an empty restricted capture is observable as an empty +/// access list (distinct from a non-empty discovery). +#[derive(Clone, Debug)] +pub struct ColdStartCall { + /// Transaction sender. + pub from: Address, + /// Call target. + pub to: Address, + /// Calldata. + pub calldata: Bytes, + /// When set, filters captured slots and accounts to these addresses. + pub restrict_to: Option>, +} diff --git a/src/cold_start/planner.rs b/src/cold_start/planner.rs new file mode 100644 index 0000000..7cbd7f9 --- /dev/null +++ b/src/cold_start/planner.rs @@ -0,0 +1,36 @@ +//! The pure planner that drives a bounded multi-round cold start. +//! +//! A [`ColdStartPlanner`] performs no IO: the driver performs every fetch and +//! call and hands the planner only a [`StateView`] and the round's +//! [`ColdStartResults`]. The planner decides the next plan (or stops). + +use crate::cold_start::plan::ColdStartPlan; +use crate::cold_start::results::ColdStartResults; +use crate::events::StateView; + +/// The planner's decision after a round completes. +#[derive(Clone, Debug)] +pub enum ColdStartStep { + /// Stop the cold-start loop; the run succeeds. + Done, + /// Execute the carried plan as the next round. + Continue(ColdStartPlan), +} + +/// Drives a bounded multi-round cold start. +/// +/// The driver calls [`initial_plan`](Self::initial_plan) once, executes it, then +/// repeatedly calls [`on_results`](Self::on_results) with the round's results and +/// the post-injection [`StateView`]; a returned [`ColdStartStep::Continue`] is run +/// as the next round and [`ColdStartStep::Done`] ends the run. The loop is bounded +/// by [`ColdStartConfig::max_rounds`](crate::cold_start::ColdStartConfig::max_rounds). +/// +/// Implementations perform **no IO** — the trait hands them only borrowed, +/// read-only state, so all fetching is the driver's responsibility. +pub trait ColdStartPlanner { + /// The first plan to execute, derived from the current cached state. + fn initial_plan(&mut self, state: &dyn StateView) -> ColdStartPlan; + /// Decide whether to continue (with a next plan) or finish, given the just- + /// completed round's results and the post-injection state view. + fn on_results(&mut self, results: &ColdStartResults, state: &dyn StateView) -> ColdStartStep; +} diff --git a/src/cold_start/results.rs b/src/cold_start/results.rs new file mode 100644 index 0000000..db8f38c --- /dev/null +++ b/src/cold_start/results.rs @@ -0,0 +1,56 @@ +//! What a cold-start round produced. +//! +//! [`ColdStartResults`] carries a per-slot [`SlotOutcome`] for every declared +//! verify and probe slot — distinguishing a genuine on-chain zero from a fetch +//! failure — plus the injected [`SlotChange`]s and any discovered access lists. + +use crate::access_set::StorageAccessList; +use crate::cold_start::error::ColdStartError; +use crate::freshness::{SlotChange, SlotOutcome}; + +use revm::context::result::ExecutionResult; + +/// The outcome of executing one [`ColdStartPlan`](crate::cold_start::ColdStartPlan) +/// round. +/// +/// `fetched` and `probed` each carry exactly one [`SlotOutcome`] per declared +/// verify / probe slot, so a fetch failure surfaces as +/// [`SlotFetch::FetchFailed`](crate::freshness::SlotFetch::FetchFailed) rather +/// than as absence. `verified` carries only the slots whose value actually changed +/// (and were injected). `discovered` carries one [`ColdStartCallResult`] per +/// discover call. +#[derive(Clone, Debug, Default)] +pub struct ColdStartResults { + /// Slots whose value changed and were injected (one per change). + pub verified: Vec, + /// One outcome per declared verify slot (`Value` / `Zero` / `FetchFailed`). + pub fetched: Vec, + /// One outcome per declared probe slot (classified, not injected). + pub probed: Vec, + /// One result per discover call. + pub discovered: Vec, +} + +/// The result of one discover view-call: the raw EVM execution result and the +/// storage/account access list it touched (filtered by `restrict_to`). +#[derive(Clone, Debug)] +pub struct ColdStartCallResult { + /// The raw revm execution result of the view-call. + pub result: ExecutionResult, + /// The storage slots and accounts the call touched (after `restrict_to`). + pub access: StorageAccessList, +} + +/// The outcome of a single cold-start round, always carrying the +/// (possibly partial) [`ColdStartResults`]. +/// +/// `error` is `Some` only when a hard error short-circuited the round (an +/// accounts- or discover-phase failure in later slices). The verify-only path +/// never sets `error`. Carrying the results unconditionally lets the driver +/// absorb partial outcomes into the run report before propagating the error. +pub struct RoundOutcome { + /// The (possibly partial) results computed before any short-circuit. + pub results: ColdStartResults, + /// `Some` when a hard error aborted the round mid-way. + pub error: Option, +} diff --git a/src/freshness.rs b/src/freshness.rs index f741243..3241906 100644 --- a/src/freshness.rs +++ b/src/freshness.rs @@ -457,6 +457,53 @@ pub struct SlotChange { pub new: U256, } +/// The classified outcome of fetching a single storage slot. +/// +/// Where a [`SlotChange`] records only slots whose value *differed* from the +/// cache, a [`SlotOutcome`] is produced for **every** requested slot — including +/// ones that did not change and ones the fetcher could not return. This closes +/// the "archive-miss" gap: a transient fetch failure is surfaced explicitly as +/// [`SlotFetch::FetchFailed`] rather than collapsing into the same "no value" +/// signal as a genuine on-chain zero ([`SlotFetch::Zero`]). +/// +/// The fetch classification ([`SlotFetch`]) and change detection ([`SlotChange`]) +/// are independent reads of the same fetched value: a genuine `Ok(0)` on a slot +/// whose cached value was also `0` yields [`SlotFetch::Zero`] **and** no +/// `SlotChange`. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct SlotOutcome { + /// Contract whose storage slot was fetched. + pub address: Address, + /// Storage slot key. + pub slot: U256, + /// The classified result of fetching this slot. + pub fetch: SlotFetch, +} + +/// The classified result of an individual slot fetch. +/// +/// A fetcher `Ok(value)` is classified into [`Value`](SlotFetch::Value) (non-zero) +/// or [`Zero`](SlotFetch::Zero) (a genuine on-chain zero); a fetcher `Err` +/// becomes [`FetchFailed`](SlotFetch::FetchFailed), carrying the error string. +/// [`NotAttempted`](SlotFetch::NotAttempted) marks a declared slot that a +/// short-circuited round never reached (produced only by the accounts/discover +/// phases of a cold-start round, never by verify). +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum SlotFetch { + /// The slot was fetched and holds a non-zero value. + Value(U256), + /// The slot was fetched and holds a genuine on-chain zero. + Zero, + /// The fetcher returned an error for this slot; `reason` is its description. + FetchFailed { + /// Human-readable description of why the fetch failed. + reason: String, + }, + /// The slot was declared but never reached because the round + /// short-circuited on an earlier-phase hard error. + NotAttempted, +} + /// The deferred verdict on a [`SpeculativeSim`]'s optimistic results. pub enum Validation { /// Nothing the sims read had changed; the optimistic results are correct. diff --git a/src/lib.rs b/src/lib.rs index 0208e04..55000e6 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -106,6 +106,8 @@ pub mod access_list; pub mod access_set; pub mod cache; +#[cfg(feature = "reactive")] +pub mod cold_start; pub mod create3; pub mod deploy; pub mod errors; @@ -119,6 +121,12 @@ pub mod reactive; pub mod state_update; pub use access_set::StorageAccessList; +#[cfg(feature = "reactive")] +pub use cold_start::{ + ColdStartCall, ColdStartCallResult, ColdStartConfig, ColdStartError, ColdStartPin, + ColdStartPlan, ColdStartPlanner, ColdStartResults, ColdStartRoundSummary, ColdStartRunReport, + ColdStartStep, RoundOutcome, +}; pub use events::erc20::Erc20TransferDecoder; pub use events::{ BlockDigest, DecoderRegistry, EventDecoder, EventPipeline, ReconcileReport, ReorgConfig, @@ -127,7 +135,7 @@ pub use events::{ pub use freshness::{ AlwaysVerify, BlockClock, FreshnessClock, FreshnessController, FreshnessParams, FreshnessPolicy, FreshnessRegistry, NeverVerify, ObservationDriven, SimRequest, SlotChange, - SpeculativeSim, Validation, Validity, WallClock, + SlotFetch, SlotOutcome, SpeculativeSim, Validation, Validity, WallClock, }; pub use state_update::{ AccountChange, AccountPatch, PurgeRecord, PurgeScope, SkippedAccountPatch, SkippedBalanceDelta, diff --git a/tests/cold_start.rs b/tests/cold_start.rs new file mode 100644 index 0000000..2c39210 --- /dev/null +++ b/tests/cold_start.rs @@ -0,0 +1,1051 @@ +//! Offline tests for the cold-start sync driver. +//! +//! - S1: the verify-only, single- and multi-round path that closes the +//! archive-miss gap. +//! - S2: the accounts (ensure) and discover (view-call access-list capture) +//! phases, `restrict_to` filtering, and the mid-round partial-failure contract +//! (`NotAttempted`), including the Balancer-style two-round discover→verify. +//! - S3: the probe phase (classify at the pinned block without injecting). +//! - S4: `ColdStartPin::Hash` pins every round to the hash and restores the +//! prior block on completion and on the error path. +//! +//! Every test runs fully offline over a mocked provider; none reach the network +//! (an unexpected RPC fetch errors against the empty mock queue, failing the +//! test). These are manager-authored red-green acceptance tests. The +//! implementation agent must make them pass without weakening, skipping, or +//! rewriting them. Where they disagree with the original feature request, the +//! implementation spec (`...cold-start-implementation-spec.md`) and these tests win. +#![cfg(feature = "reactive")] + +mod common; + +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; + +use alloy_eips::BlockId; +use alloy_primitives::{Address, B256, Bytes, U256, keccak256}; +use alloy_sol_types::{SolCall, SolValue}; +use anyhow::{Result, anyhow}; +use revm::primitives::hardfork::SpecId; + +use evm_fork_cache::cache::{EvmCache, StorageBatchFetchFn}; +use evm_fork_cache::cold_start::{ + ColdStartCall, ColdStartConfig, ColdStartError, ColdStartPin, ColdStartPlan, ColdStartPlanner, + ColdStartResults, ColdStartStep, SlotFetch, +}; +use evm_fork_cache::events::StateView; + +use common::{ + MOCK_ERC20_BALANCE_SLOT, MockERC20, install_default_account, install_mock_erc20, setup_cache, + setup_cache_with_asserter, stub_fetcher, +}; + +/// The hashed storage slot for `MockERC20.balanceOf[owner]` (mapping at slot 3): +/// `keccak256(abi.encode(owner, 3))`. A `balanceOf(owner)` view-call SLOADs +/// exactly this slot, so the discover phase captures it. +fn balance_slot_hashed(owner: Address) -> U256 { + let key = keccak256((owner, U256::from(MOCK_ERC20_BALANCE_SLOT)).abi_encode()); + U256::from_be_bytes(key.0) +} + +/// Calldata for `MockERC20.balanceOf(owner)`. +fn balance_of_calldata(owner: Address) -> Bytes { + Bytes::from(MockERC20::balanceOfCall { account: owner }.abi_encode()) +} + +// --------------------------------------------------------------------------- +// Test fixtures: planners and fetchers +// --------------------------------------------------------------------------- + +/// A planner that emits a fixed `initial_plan` and immediately returns `Done`. +struct OneShotPlanner { + plan: ColdStartPlan, +} + +impl ColdStartPlanner for OneShotPlanner { + fn initial_plan(&mut self, _state: &dyn StateView) -> ColdStartPlan { + self.plan.clone() + } + fn on_results(&mut self, _results: &ColdStartResults, _state: &dyn StateView) -> ColdStartStep { + ColdStartStep::Done + } +} + +/// A planner that re-emits the same (empty) plan every round, returning `Done` +/// on the `done_on_call`-th `on_results` call (`None` = never, i.e. always +/// `Continue`). `on_results_calls` records how many rounds actually executed. +struct LoopPlanner { + on_results_calls: usize, + done_on_call: Option, +} + +impl LoopPlanner { + fn always_continue() -> Self { + Self { + on_results_calls: 0, + done_on_call: None, + } + } + fn done_after(n: usize) -> Self { + Self { + on_results_calls: 0, + done_on_call: Some(n), + } + } +} + +impl ColdStartPlanner for LoopPlanner { + fn initial_plan(&mut self, _state: &dyn StateView) -> ColdStartPlan { + ColdStartPlan::default() + } + fn on_results(&mut self, _results: &ColdStartResults, _state: &dyn StateView) -> ColdStartStep { + self.on_results_calls += 1; + match self.done_on_call { + Some(n) if self.on_results_calls >= n => ColdStartStep::Done, + _ => ColdStartStep::Continue(ColdStartPlan::default()), + } + } +} + +/// A two-round planner: round 1 verifies `slot_a`, then in `on_results` (which +/// runs after round 1's injection) it records what `slot_a` reads as via the +/// `StateView`, continues into a round that verifies `slot_b`, and finishes. +struct TwoRoundPlanner { + pool: Address, + slot_a: U256, + slot_b: U256, + phase: usize, + observed_slot_a_after_round1: Option, +} + +impl ColdStartPlanner for TwoRoundPlanner { + fn initial_plan(&mut self, _state: &dyn StateView) -> ColdStartPlan { + ColdStartPlan { + verify: vec![(self.pool, self.slot_a)], + ..Default::default() + } + } + fn on_results(&mut self, _results: &ColdStartResults, state: &dyn StateView) -> ColdStartStep { + self.phase += 1; + if self.phase == 1 { + // on_results sees post-injection state for round 1. + self.observed_slot_a_after_round1 = state.storage(self.pool, self.slot_a); + ColdStartStep::Continue(ColdStartPlan { + verify: vec![(self.pool, self.slot_b)], + ..Default::default() + }) + } else { + ColdStartStep::Done + } + } +} + +/// A fetcher that returns a chosen non-zero value for `value_slot`, a hard +/// `Err` for `fail_slot`, and a genuine `Ok(ZERO)` for everything else — so a +/// single round exercises all three `SlotFetch` arms. +fn mixed_fetcher( + value_slot: (Address, U256), + value: U256, + fail_slot: (Address, U256), +) -> StorageBatchFetchFn { + Arc::new( + move |requests: Vec<(Address, U256)>, _block: Option| { + requests + .into_iter() + .map(|(addr, slot)| { + if (addr, slot) == fail_slot { + (addr, slot, Err(anyhow!("archive miss"))) + } else if (addr, slot) == value_slot { + (addr, slot, Ok(value)) + } else { + (addr, slot, Ok(U256::ZERO)) + } + }) + .collect() + }, + ) +} + +/// Build a cache with NO storage batch fetcher (mirrors the canonical pattern in +/// `tests/freshness.rs`): `EvmCache::new` installs a default RPC fetcher, but a +/// `from_backend` cache does not capture one. +async fn no_fetcher_cache() -> Result { + let base = setup_cache().await?; + let cache = EvmCache::from_backend( + base.unchecked_backend().clone(), + base.unchecked_blockchain_db().clone(), + base.block(), + base.chain_id(), + None, + None, + SpecId::CANCUN, + ); + assert!( + cache.storage_batch_fetcher().is_none(), + "from_backend cache has no fetcher" + ); + Ok(cache) +} + +/// Find the `SlotFetch` recorded for a slot in `results.fetched`. +fn fetch_of(results: &ColdStartResults, addr: Address, slot: U256) -> SlotFetch { + results + .fetched + .iter() + .find(|o| o.address == addr && o.slot == slot) + .map(|o| o.fetch.clone()) + .unwrap_or_else(|| panic!("slot {slot} not present in results.fetched")) +} + +// --------------------------------------------------------------------------- +// S1 acceptance tests +// --------------------------------------------------------------------------- + +/// A single round classifies each verify slot as `Value` / `Zero` / +/// `FetchFailed` — closing the archive-miss gap (a fetch failure is `FetchFailed`, +/// NOT absence, and is distinct from a genuine on-chain `Zero`). +#[tokio::test(flavor = "multi_thread")] +async fn verify_classifies_value_zero_and_failed() -> Result<()> { + let mut cache = setup_cache().await?; + let pool = Address::repeat_byte(0x11); + install_mock_erc20(&mut cache, pool); // StorageCleared: unseen slots read ZERO + + let slot_zero = U256::from(1); + let slot_value = U256::from(2); + let slot_fail = U256::from(3); + + cache.set_storage_batch_fetcher(mixed_fetcher( + (pool, slot_value), + U256::from(7), + (pool, slot_fail), + )); + + let plan = ColdStartPlan { + verify: vec![(pool, slot_zero), (pool, slot_value), (pool, slot_fail)], + ..Default::default() + }; + let outcome = cache.execute_cold_start_round(&plan); + + assert!( + outcome.error.is_none(), + "verify-only round has no hard-error surface" + ); + let results = outcome.results; + + assert_eq!(results.fetched.len(), 3, "one outcome per verify slot"); + assert_eq!(fetch_of(&results, pool, slot_zero), SlotFetch::Zero); + assert_eq!( + fetch_of(&results, pool, slot_value), + SlotFetch::Value(U256::from(7)) + ); + assert!( + matches!( + fetch_of(&results, pool, slot_fail), + SlotFetch::FetchFailed { .. } + ), + "a fetcher Err must surface as FetchFailed, not absence" + ); + + // Only the non-zero, changed slot was injected and recorded as changed. + assert_eq!(results.verified.len(), 1, "only slot_value changed"); + assert_eq!(results.verified[0].slot, slot_value); + assert_eq!(results.verified[0].new, U256::from(7)); + + Ok(()) +} + +/// A changed slot appears in BOTH `verified` (as a `SlotChange`) and `fetched` +/// (as `Value`); an unchanged slot appears only in `fetched`. +#[tokio::test(flavor = "multi_thread")] +async fn changed_slot_in_verified_and_fetched_unchanged_only_fetched() -> Result<()> { + let mut cache = setup_cache().await?; + let pool = Address::repeat_byte(0x22); + install_mock_erc20(&mut cache, pool); + + let slot_changed = U256::from(8); + let slot_unchanged = U256::from(9); + // Seed EVM-visible cached baselines. + cache + .db_mut() + .insert_account_storage(pool, slot_changed, U256::from(100))?; + cache + .db_mut() + .insert_account_storage(pool, slot_unchanged, U256::from(200))?; + + cache.set_storage_batch_fetcher(stub_fetcher(HashMap::from([ + ((pool, slot_changed), U256::from(999)), + ((pool, slot_unchanged), U256::from(200)), + ]))); + + let plan = ColdStartPlan { + verify: vec![(pool, slot_changed), (pool, slot_unchanged)], + ..Default::default() + }; + let results = cache.execute_cold_start_round(&plan).results; + + // fetched has both (Value for each). + assert_eq!(results.fetched.len(), 2); + assert_eq!( + fetch_of(&results, pool, slot_changed), + SlotFetch::Value(U256::from(999)) + ); + assert_eq!( + fetch_of(&results, pool, slot_unchanged), + SlotFetch::Value(U256::from(200)) + ); + + // verified has only the changed slot. + assert_eq!(results.verified.len(), 1); + assert_eq!(results.verified[0].slot, slot_changed); + assert_eq!(results.verified[0].old, U256::from(100)); + assert_eq!(results.verified[0].new, U256::from(999)); + + // The change was injected; the unchanged slot is untouched. + assert_eq!( + cache.cached_storage_value(pool, slot_changed), + Some(U256::from(999)) + ); + assert_eq!( + cache.cached_storage_value(pool, slot_unchanged), + Some(U256::from(200)) + ); + + Ok(()) +} + +/// An empty plan is a valid no-op round. +#[tokio::test(flavor = "multi_thread")] +async fn empty_plan_is_noop_round() -> Result<()> { + let mut cache = setup_cache().await?; + let mut planner = OneShotPlanner { + plan: ColdStartPlan::default(), + }; + let report = cache.run_cold_start(&mut planner, ColdStartConfig::default())?; + assert_eq!( + report.rounds, 1, + "the initial empty plan still executes as one round" + ); + assert_eq!(report.changed_slots, 0); + assert_eq!(report.failed_slots, 0); + Ok(()) +} + +/// A verify-bearing round on a cache with no fetcher errors `NoBatchFetcher` +/// (the per-round guard), rather than silently no-opping. +#[tokio::test(flavor = "multi_thread")] +async fn verify_round_without_fetcher_errors_no_batch_fetcher() -> Result<()> { + let mut cache = no_fetcher_cache().await?; + let pool = Address::repeat_byte(0x33); + let mut planner = OneShotPlanner { + plan: ColdStartPlan { + verify: vec![(pool, U256::from(8))], + ..Default::default() + }, + }; + let err = cache + .run_cold_start(&mut planner, ColdStartConfig::default()) + .expect_err("verify round with no fetcher must error"); + assert!(matches!(err, ColdStartError::NoBatchFetcher), "got {err:?}"); + Ok(()) +} + +/// `max_rounds` is the maximum number of EXECUTED rounds: an always-`Continue` +/// planner trips `RoundBudgetExceeded` after exactly `max_rounds` rounds. +#[tokio::test(flavor = "multi_thread")] +async fn max_rounds_boundary_always_continue_exceeds() -> Result<()> { + let mut cache = setup_cache().await?; + let mut planner = LoopPlanner::always_continue(); + let cfg = ColdStartConfig { + max_rounds: 3, + ..Default::default() + }; + let err = cache + .run_cold_start(&mut planner, cfg) + .expect_err("always-continue must exceed the budget"); + assert!( + matches!(err, ColdStartError::RoundBudgetExceeded { max_rounds: 3 }), + "got {err:?}" + ); + assert_eq!( + planner.on_results_calls, 3, + "exactly max_rounds rounds executed before erroring" + ); + Ok(()) +} + +/// A planner that returns `Done` exactly on round `max_rounds` succeeds. +#[tokio::test(flavor = "multi_thread")] +async fn max_rounds_boundary_done_on_last_round_succeeds() -> Result<()> { + let mut cache = setup_cache().await?; + let mut planner = LoopPlanner::done_after(3); + let cfg = ColdStartConfig { + max_rounds: 3, + ..Default::default() + }; + let report = cache.run_cold_start(&mut planner, cfg)?; + assert_eq!(report.rounds, 3); + Ok(()) +} + +/// A multi-round cold start runs `initial_plan` then one `Continue` then `Done` +/// (exactly two rounds), and round 2's `on_results` sees round 1's injection via +/// the `StateView`. +#[tokio::test(flavor = "multi_thread")] +async fn multi_round_continuation_sees_injection() -> Result<()> { + let mut cache = setup_cache().await?; + let pool = Address::repeat_byte(0x44); + install_mock_erc20(&mut cache, pool); + + let slot_a = U256::from(8); + let slot_b = U256::from(9); + cache + .db_mut() + .insert_account_storage(pool, slot_a, U256::from(100))?; + + cache.set_storage_batch_fetcher(stub_fetcher(HashMap::from([ + ((pool, slot_a), U256::from(999)), + ((pool, slot_b), U256::from(7)), + ]))); + + let mut planner = TwoRoundPlanner { + pool, + slot_a, + slot_b, + phase: 0, + observed_slot_a_after_round1: None, + }; + let report = cache.run_cold_start(&mut planner, ColdStartConfig::default())?; + + assert_eq!(report.rounds, 2, "initial + one continue = two rounds"); + assert_eq!( + planner.observed_slot_a_after_round1, + Some(U256::from(999)), + "on_results must see round 1's dual-layer injection via StateView" + ); + Ok(()) +} + +// --------------------------------------------------------------------------- +// S2 acceptance tests: accounts + discover, restrict_to, mid-round failure +// --------------------------------------------------------------------------- + +/// A discover view-call captures the `(address, slot)` pairs and accounts it +/// touches. `balanceOf(owner)` SLOADs the token's balance mapping slot, so that +/// slot and the token account appear in the captured access list. +#[tokio::test(flavor = "multi_thread")] +async fn discover_captures_touched_slots() -> Result<()> { + let mut cache = setup_cache().await?; + let token = Address::repeat_byte(0x55); + let owner = Address::repeat_byte(0x56); + // The block beneficiary (default `Address::ZERO`) is credited gas during a + // discover call's transact; install it so the offline run does not fetch it + // (mirrors `overlay_call_raw_with_access_list_captures_read_set`). + install_default_account(&mut cache, Address::ZERO); + install_default_account(&mut cache, owner); + install_mock_erc20(&mut cache, token); + + let plan = ColdStartPlan { + discover: vec![ColdStartCall { + from: owner, + to: token, + calldata: balance_of_calldata(owner), + restrict_to: None, + }], + ..Default::default() + }; + let outcome = cache.execute_cold_start_round(&plan); + + assert!( + outcome.error.is_none(), + "discover on a local contract succeeds" + ); + let results = outcome.results; + assert_eq!(results.discovered.len(), 1, "one result per discover call"); + let call = &results.discovered[0]; + assert!( + call.result.is_success(), + "balanceOf succeeds: {:?}", + call.result + ); + assert!( + call.access.accounts.contains(&token), + "token account captured" + ); + assert!( + call.access + .slots + .contains(&(token, balance_slot_hashed(owner))), + "balance mapping slot captured in the access list" + ); + Ok(()) +} + +/// `restrict_to` filters the captured slots and accounts to the named addresses: +/// restricting to the token keeps only its entries; restricting to an address the +/// call never touched yields an empty (but observable) capture. +#[tokio::test(flavor = "multi_thread")] +async fn restrict_to_filters_captured_slots_and_accounts() -> Result<()> { + let mut cache = setup_cache().await?; + let token = Address::repeat_byte(0x57); + let owner = Address::repeat_byte(0x58); + let unrelated = Address::repeat_byte(0x59); + // The block beneficiary (default `Address::ZERO`) is credited gas during a + // discover call's transact; install it so the offline run does not fetch it + // (mirrors `overlay_call_raw_with_access_list_captures_read_set`). + install_default_account(&mut cache, Address::ZERO); + install_default_account(&mut cache, owner); + install_mock_erc20(&mut cache, token); + + // restrict_to = [token]: only token's slots/accounts survive. + let plan_keep = ColdStartPlan { + discover: vec![ColdStartCall { + from: owner, + to: token, + calldata: balance_of_calldata(owner), + restrict_to: Some(vec![token]), + }], + ..Default::default() + }; + let kept = cache.execute_cold_start_round(&plan_keep).results; + let access = &kept.discovered[0].access; + assert!( + access.slots.contains(&(token, balance_slot_hashed(owner))), + "token's slot survives restrict_to=[token]" + ); + assert!( + access.accounts.iter().all(|a| *a == token), + "only the token account survives restrict_to=[token]: {:?}", + access.accounts + ); + assert!( + access.slots.iter().all(|(a, _)| *a == token), + "only token slots survive restrict_to=[token]" + ); + + // restrict_to = [unrelated]: nothing the call touched matches → empty capture. + let plan_empty = ColdStartPlan { + discover: vec![ColdStartCall { + from: owner, + to: token, + calldata: balance_of_calldata(owner), + restrict_to: Some(vec![unrelated]), + }], + ..Default::default() + }; + let empty = cache.execute_cold_start_round(&plan_empty).results; + let access = &empty.discovered[0].access; + assert!( + access.slots.is_empty(), + "restrict_to an untouched address yields empty slots" + ); + assert!( + access.accounts.is_empty(), + "and empty accounts — distinct from a non-empty capture" + ); + Ok(()) +} + +/// A round declaring only `accounts`/`discover` (no verify/probe) runs even when +/// no batch fetcher is configured — the per-round `NoBatchFetcher` guard fires +/// only for verify/probe-bearing rounds. This is the Balancer round-1 case. +#[tokio::test(flavor = "multi_thread")] +async fn discover_only_round_runs_without_fetcher() -> Result<()> { + let mut cache = no_fetcher_cache().await?; + let token = Address::repeat_byte(0x5a); + let owner = Address::repeat_byte(0x5b); + // The block beneficiary (default `Address::ZERO`) is credited gas during a + // discover call's transact; install it so the offline run does not fetch it + // (mirrors `overlay_call_raw_with_access_list_captures_read_set`). + install_default_account(&mut cache, Address::ZERO); + install_default_account(&mut cache, owner); + install_mock_erc20(&mut cache, token); + + let mut planner = OneShotPlanner { + plan: ColdStartPlan { + accounts: vec![token], // already installed → ensure is a no-op + discover: vec![ColdStartCall { + from: owner, + to: token, + calldata: balance_of_calldata(owner), + restrict_to: Some(vec![token]), + }], + ..Default::default() + }, + }; + let report = cache.run_cold_start(&mut planner, ColdStartConfig::default())?; + assert_eq!(report.rounds, 1); + assert!( + report.discovered_slots >= 1, + "the discover-only round captured at least the balance slot" + ); + Ok(()) +} + +/// An accounts-phase hard error (the first phase) leaves every declared verify +/// slot `NotAttempted` and injects nothing — the partial results are still +/// returned. +#[tokio::test(flavor = "multi_thread")] +async fn accounts_failure_marks_verify_slots_not_attempted() -> Result<()> { + let (mut cache, asserter) = setup_cache_with_asserter().await?; + let pool = Address::repeat_byte(0x5c); + let uninstalled = Address::repeat_byte(0x5d); + install_mock_erc20(&mut cache, pool); + let slot = U256::from(8); + cache.set_storage_batch_fetcher(stub_fetcher(HashMap::from([( + (pool, slot), + U256::from(42), + )]))); + // Force the account fetch for `uninstalled` to fail deterministically. + for _ in 0..8 { + asserter.push_failure_msg("account fetch failed (offline test)"); + } + + let plan = ColdStartPlan { + accounts: vec![uninstalled], + verify: vec![(pool, slot)], + ..Default::default() + }; + let outcome = cache.execute_cold_start_round(&plan); + + assert!( + outcome.error.is_some(), + "an accounts-phase failure is a hard error" + ); + // verify never ran (it follows the accounts phase) → its slot is NotAttempted. + assert_eq!( + fetch_of(&outcome.results, pool, slot), + SlotFetch::NotAttempted, + "an unreached verify slot is NotAttempted, not silently dropped" + ); + assert!( + outcome.results.verified.is_empty(), + "nothing injected on accounts failure" + ); + assert_eq!( + cache.cached_storage_value(pool, slot), + Some(U256::ZERO), + "the slot was not warmed (StorageCleared reads 0, not the fetcher's 42)" + ); + Ok(()) +} + +/// A mid-round hard error propagates from `run_cold_start` and `on_results` is +/// NOT called for the errored round. +#[tokio::test(flavor = "multi_thread")] +async fn mid_round_failure_propagates_and_skips_on_results() -> Result<()> { + let (mut cache, asserter) = setup_cache_with_asserter().await?; + let uninstalled = Address::repeat_byte(0x5e); + for _ in 0..8 { + asserter.push_failure_msg("account fetch failed (offline test)"); + } + + struct FlagPlanner { + acct: Address, + on_results_called: bool, + } + impl ColdStartPlanner for FlagPlanner { + fn initial_plan(&mut self, _state: &dyn StateView) -> ColdStartPlan { + ColdStartPlan { + accounts: vec![self.acct], + ..Default::default() + } + } + fn on_results(&mut self, _r: &ColdStartResults, _s: &dyn StateView) -> ColdStartStep { + self.on_results_called = true; + ColdStartStep::Done + } + } + + let mut planner = FlagPlanner { + acct: uninstalled, + on_results_called: false, + }; + let err = cache + .run_cold_start(&mut planner, ColdStartConfig::default()) + .expect_err("an accounts-phase failure errors the run"); + assert!(matches!(err, ColdStartError::Fetch(_)), "got {err:?}"); + assert!( + !planner.on_results_called, + "on_results must not run for an errored round" + ); + Ok(()) +} + +/// A discover-phase hard error (the last phase) preserves the verify outcomes +/// already computed earlier in the round — they are classified, not +/// `NotAttempted` — while the failed discover call yields no result. +#[tokio::test(flavor = "multi_thread")] +async fn discover_failure_preserves_earlier_verify_outcomes() -> Result<()> { + let (mut cache, asserter) = setup_cache_with_asserter().await?; + let pool = Address::repeat_byte(0x5f); + let uninstalled_callee = Address::repeat_byte(0x60); + install_default_account(&mut cache, Address::ZERO); + install_mock_erc20(&mut cache, pool); + let slot = U256::from(8); + cache.set_storage_batch_fetcher(stub_fetcher(HashMap::from([( + (pool, slot), + U256::from(99), + )]))); + // The uninstalled callee's code load fails → the discover call errors. + for _ in 0..8 { + asserter.push_failure_msg("code fetch failed (offline test)"); + } + + let plan = ColdStartPlan { + verify: vec![(pool, slot)], + discover: vec![ColdStartCall { + from: Address::ZERO, + to: uninstalled_callee, + calldata: Bytes::new(), + restrict_to: None, + }], + ..Default::default() + }; + let outcome = cache.execute_cold_start_round(&plan); + + assert!( + outcome.error.is_some(), + "the discover call failure is a hard error" + ); + // verify ran BEFORE discover, so its outcome is classified, not NotAttempted. + assert_eq!( + fetch_of(&outcome.results, pool, slot), + SlotFetch::Value(U256::from(99)), + "verify outcomes computed before the discover failure are preserved" + ); + assert_eq!( + outcome.results.verified.len(), + 1, + "verify injected before discover failed" + ); + assert!( + outcome.results.discovered.is_empty(), + "the failed discover call yields no result" + ); + Ok(()) +} + +/// The canonical Balancer-style two-round cold start, fully offline: round 1 +/// discovers the touched slots via a view call (`restrict_to` the target), round +/// 2 verifies exactly those discovered slots, and the discovered slot ends up +/// warm — with no RPC issued. +#[tokio::test(flavor = "multi_thread")] +async fn two_round_discover_then_verify_offline() -> Result<()> { + let (mut cache, asserter) = setup_cache_with_asserter().await?; + let token = Address::repeat_byte(0x61); + let owner = Address::repeat_byte(0x62); + // The block beneficiary (default `Address::ZERO`) is credited gas during a + // discover call's transact; install it so the offline run does not fetch it + // (mirrors `overlay_call_raw_with_access_list_captures_read_set`). + install_default_account(&mut cache, Address::ZERO); + install_default_account(&mut cache, owner); + install_mock_erc20(&mut cache, token); + let hashed = balance_slot_hashed(owner); + // Round 2's verify fetcher returns a fresh value for the discovered slot. + cache.set_storage_batch_fetcher(stub_fetcher(HashMap::from([( + (token, hashed), + U256::from(1000), + )]))); + + struct BalancerLike { + token: Address, + owner: Address, + phase: usize, + } + impl ColdStartPlanner for BalancerLike { + fn initial_plan(&mut self, _state: &dyn StateView) -> ColdStartPlan { + ColdStartPlan { + accounts: vec![self.token], + discover: vec![ColdStartCall { + from: self.owner, + to: self.token, + calldata: Bytes::from( + MockERC20::balanceOfCall { + account: self.owner, + } + .abi_encode(), + ), + restrict_to: Some(vec![self.token]), + }], + ..Default::default() + } + } + fn on_results( + &mut self, + results: &ColdStartResults, + _state: &dyn StateView, + ) -> ColdStartStep { + self.phase += 1; + if self.phase == 1 { + // Verify exactly the slots discovered in round 1. + let verify: Vec<_> = results.discovered[0].access.slots.iter().copied().collect(); + assert!( + !verify.is_empty(), + "round 1 must discover at least one slot" + ); + ColdStartStep::Continue(ColdStartPlan { + verify, + ..Default::default() + }) + } else { + ColdStartStep::Done + } + } + } + + let mut planner = BalancerLike { + token, + owner, + phase: 0, + }; + let report = cache.run_cold_start(&mut planner, ColdStartConfig::default())?; + + assert_eq!(report.rounds, 2, "discover round + verify round"); + assert_eq!( + cache.cached_storage_value(token, hashed), + Some(U256::from(1000)), + "the discovered slot was warmed by round 2's verify" + ); + assert!( + asserter.read_q().is_empty(), + "no RPC was issued during the fully-offline cold start" + ); + Ok(()) +} + +// --------------------------------------------------------------------------- +// S3 acceptance tests: probe phase (classify at the pinned block, no inject) +// --------------------------------------------------------------------------- + +/// A probe reads and classifies a slot at the pinned block but does NOT inject +/// it: the fetched value is reported in `results.probed`, yet the cache keeps its +/// prior value and no `SlotChange` is recorded. +#[tokio::test(flavor = "multi_thread")] +async fn probe_classifies_without_injecting() -> Result<()> { + let mut cache = setup_cache().await?; + let pool = Address::repeat_byte(0x63); + install_mock_erc20(&mut cache, pool); + let slot = U256::from(8); + // Cache holds 100; the fetcher reports a different value the probe must NOT inject. + cache + .db_mut() + .insert_account_storage(pool, slot, U256::from(100))?; + cache.set_storage_batch_fetcher(stub_fetcher(HashMap::from([( + (pool, slot), + U256::from(777), + )]))); + + let plan = ColdStartPlan { + probe: vec![(pool, slot)], + ..Default::default() + }; + let outcome = cache.execute_cold_start_round(&plan); + assert!(outcome.error.is_none()); + let results = outcome.results; + + // The probe classified the freshly-fetched value... + assert_eq!(results.probed.len(), 1, "one outcome per probe slot"); + assert_eq!(results.probed[0].address, pool); + assert_eq!(results.probed[0].slot, slot); + assert_eq!(results.probed[0].fetch, SlotFetch::Value(U256::from(777))); + + // ...but did not inject it, did not record a change, and is not a verify slot. + assert!( + results.verified.is_empty(), + "probe never records a SlotChange" + ); + assert!( + results.fetched.is_empty(), + "probe slots are not verify slots" + ); + assert_eq!( + cache.cached_storage_value(pool, slot), + Some(U256::from(100)), + "probe must not write the fetched value into the cache" + ); + Ok(()) +} + +/// A probe classifies each slot as `Value` / `Zero` / `FetchFailed`, using the +/// same shared classification as verify, while injecting nothing. +#[tokio::test(flavor = "multi_thread")] +async fn probe_classifies_value_zero_and_failed() -> Result<()> { + let mut cache = setup_cache().await?; + let pool = Address::repeat_byte(0x64); + install_mock_erc20(&mut cache, pool); + let slot_zero = U256::from(1); + let slot_value = U256::from(2); + let slot_fail = U256::from(3); + cache.set_storage_batch_fetcher(mixed_fetcher( + (pool, slot_value), + U256::from(7), + (pool, slot_fail), + )); + + let plan = ColdStartPlan { + probe: vec![(pool, slot_zero), (pool, slot_value), (pool, slot_fail)], + ..Default::default() + }; + let results = cache.execute_cold_start_round(&plan).results; + + assert_eq!(results.probed.len(), 3); + let probe_of = |s: U256| { + results + .probed + .iter() + .find(|o| o.slot == s) + .map(|o| o.fetch.clone()) + .unwrap_or_else(|| panic!("slot {s} not present in results.probed")) + }; + assert_eq!(probe_of(slot_zero), SlotFetch::Zero); + assert_eq!(probe_of(slot_value), SlotFetch::Value(U256::from(7))); + assert!(matches!(probe_of(slot_fail), SlotFetch::FetchFailed { .. })); + assert!(results.verified.is_empty(), "probe never injects"); + Ok(()) +} + +/// Verify and probe coexist in one round independently: the verify slot is +/// injected, the probe slot is classified but left untouched in the cache. +#[tokio::test(flavor = "multi_thread")] +async fn probe_and_verify_in_one_round_are_independent() -> Result<()> { + let mut cache = setup_cache().await?; + let pool = Address::repeat_byte(0x65); + install_mock_erc20(&mut cache, pool); + let v_slot = U256::from(8); + let p_slot = U256::from(9); + cache + .db_mut() + .insert_account_storage(pool, p_slot, U256::from(100))?; + cache.set_storage_batch_fetcher(stub_fetcher(HashMap::from([ + ((pool, v_slot), U256::from(500)), + ((pool, p_slot), U256::from(600)), + ]))); + + let plan = ColdStartPlan { + verify: vec![(pool, v_slot)], + probe: vec![(pool, p_slot)], + ..Default::default() + }; + let results = cache.execute_cold_start_round(&plan).results; + + // verify injected v_slot. + assert_eq!(results.verified.len(), 1); + assert_eq!( + cache.cached_storage_value(pool, v_slot), + Some(U256::from(500)) + ); + // probe classified p_slot but did not inject it. + assert_eq!(results.probed.len(), 1); + assert_eq!(results.probed[0].fetch, SlotFetch::Value(U256::from(600))); + assert_eq!( + cache.cached_storage_value(pool, p_slot), + Some(U256::from(100)), + "probe leaves its slot untouched even alongside a verify" + ); + Ok(()) +} + +// --------------------------------------------------------------------------- +// S4 acceptance tests: ColdStartPin::Hash pins the run and restores afterward +// --------------------------------------------------------------------------- + +/// A fetcher that records the `Option` it is called with, so a test can +/// assert which block the run's reads were pinned to. +fn block_recording_fetcher(seen: Arc>>>) -> StorageBatchFetchFn { + Arc::new( + move |requests: Vec<(Address, U256)>, block: Option| { + seen.lock().unwrap().push(block); + requests + .into_iter() + .map(|(addr, slot)| (addr, slot, Ok(U256::ZERO))) + .collect() + }, + ) +} + +/// `ColdStartPin::Hash { require_canonical }` pins every round's reads to +/// `BlockId::from((hash, Some(require_canonical)))` and restores the cache's +/// prior block when the run completes. +#[tokio::test(flavor = "multi_thread")] +async fn hash_pin_reads_at_hash_and_restores_prior_block() -> Result<()> { + let mut cache = setup_cache().await?; + let pool = Address::repeat_byte(0x66); + install_mock_erc20(&mut cache, pool); + let slot = U256::from(8); + + let seen = Arc::new(Mutex::new(Vec::new())); + cache.set_storage_batch_fetcher(block_recording_fetcher(Arc::clone(&seen))); + + let prior_block = cache.block(); + let hash = B256::repeat_byte(0xab); + let expected = BlockId::from((hash, Some(true))); + + let mut planner = OneShotPlanner { + plan: ColdStartPlan { + verify: vec![(pool, slot)], + ..Default::default() + }, + }; + let report = cache.run_cold_start( + &mut planner, + ColdStartConfig { + max_rounds: 8, + pin: ColdStartPin::Hash { + number: 100, + hash, + require_canonical: true, + }, + }, + )?; + + assert_eq!(report.rounds, 1); + let seen = seen.lock().unwrap(); + assert!(!seen.is_empty(), "the verify phase issued a pinned read"); + assert!( + seen.iter().all(|b| *b == Some(expected)), + "every read was pinned to the hash (with require_canonical): {seen:?}" + ); + assert_eq!( + cache.block(), + prior_block, + "the prior block is restored after the run" + ); + Ok(()) +} + +/// The prior block is restored even when the run ends in an error +/// (`RoundBudgetExceeded`), so a failed hash-pinned run never leaves the cache +/// stuck on the pinned hash. +#[tokio::test(flavor = "multi_thread")] +async fn hash_pin_restores_prior_block_on_error() -> Result<()> { + let mut cache = setup_cache().await?; + let prior_block = cache.block(); + let hash = B256::repeat_byte(0xcd); + + // Empty plans → no verify/probe → no fetcher needed; the planner always + // continues, so the run trips RoundBudgetExceeded. + let mut planner = LoopPlanner::always_continue(); + let err = cache + .run_cold_start( + &mut planner, + ColdStartConfig { + max_rounds: 2, + pin: ColdStartPin::Hash { + number: 200, + hash, + require_canonical: true, + }, + }, + ) + .expect_err("an always-continue planner exceeds the budget"); + assert!( + matches!(err, ColdStartError::RoundBudgetExceeded { max_rounds: 2 }), + "got {err:?}" + ); + assert_eq!( + cache.block(), + prior_block, + "the prior block is restored even on the error path" + ); + Ok(()) +} diff --git a/tests/common/mod.rs b/tests/common/mod.rs index fab954f..c977e07 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -54,6 +54,18 @@ pub async fn setup_cache() -> Result { Ok(EvmCache::new(Arc::new(provider)).await) } +/// Build an `EvmCache` over a mocked provider, also returning the [`Asserter`] +/// so a test can (a) queue mock responses/failures to drive an offline fetch +/// down a specific path and (b) assert no RPC was issued via +/// `asserter.read_q().is_empty()` after the run. The asserter is cloned, so the +/// returned handle observes the same queue the cache's provider consumes. +pub async fn setup_cache_with_asserter() -> Result<(EvmCache, Asserter)> { + let asserter = Asserter::new(); + let client = RpcClient::mocked(asserter.clone()); + let provider = RootProvider::::new(client); + Ok((EvmCache::new(Arc::new(provider)).await, asserter)) +} + /// Insert a `MockERC20` account (with runtime bytecode) at `token`. /// /// The account's storage is marked as fully local, so any slot that is not From 903af3df887bf5718693d4bacf770552681e5273 Mon Sep 17 00:00:00 2001 From: Kai Aldag Date: Tue, 23 Jun 2026 15:11:16 +0100 Subject: [PATCH 3/3] Cold-start: doc-accuracy fixes from review MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - ColdStartRunReport / run_cold_start: clarify the report is returned only on the Ok path; on error it is dropped (use execute_cold_start_round to observe a failed round's partial outcomes). - discovered_accounts / discovered_slots: drop "distinct" — they are sums across calls and rounds, not de-duplicated. - ColdStartResults: document that fetched/probed order is unspecified (look up by (address, slot)). Doc-only; no behavior change. Co-Authored-By: Claude Opus 4.8 --- src/cold_start/config.rs | 19 +++++++++++++------ src/cold_start/driver.rs | 5 ++++- src/cold_start/results.rs | 3 +++ 3 files changed, 20 insertions(+), 7 deletions(-) diff --git a/src/cold_start/config.rs b/src/cold_start/config.rs index 42241b8..5617d08 100644 --- a/src/cold_start/config.rs +++ b/src/cold_start/config.rs @@ -51,11 +51,16 @@ pub enum ColdStartPin { }, } -/// Summary of a completed (or partially-completed) cold-start run. +/// Summary of a completed cold-start run, returned by +/// [`run_cold_start`](crate::cache::EvmCache::run_cold_start) on success. /// -/// Accumulated round-by-round as the run progresses; on a mid-round hard error -/// the partial round is absorbed before the error propagates, so the report -/// reflects work already done. +/// Accumulated round-by-round as the run progresses. Note `run_cold_start` +/// returns this report **only on the `Ok` path**: on a hard error the partial +/// round is folded in before the error is returned, but the report itself is +/// then dropped (the `Err` carries only the cause). To observe partial progress +/// on failure, drive rounds yourself via +/// [`execute_cold_start_round`](crate::cache::EvmCache::execute_cold_start_round), +/// which always returns a [`RoundOutcome`](crate::cold_start::RoundOutcome). #[derive(Clone, Debug, Default)] pub struct ColdStartRunReport { /// Number of rounds executed. @@ -64,9 +69,11 @@ pub struct ColdStartRunReport { pub verified_slots: usize, /// Total slots that changed and were injected. pub changed_slots: usize, - /// Total distinct accounts touched by discover calls. + /// Total accounts touched by discover calls, summed across calls and rounds + /// (not de-duplicated — the same account touched twice counts twice). pub discovered_accounts: usize, - /// Total distinct slots touched by discover calls. + /// Total slots touched by discover calls, summed across calls and rounds + /// (not de-duplicated — the same slot touched twice counts twice). pub discovered_slots: usize, /// Total verify + probe slots whose fetch failed. pub failed_slots: usize, diff --git a/src/cold_start/driver.rs b/src/cold_start/driver.rs index dbad821..32bedd7 100644 --- a/src/cold_start/driver.rs +++ b/src/cold_start/driver.rs @@ -187,7 +187,10 @@ impl EvmCache { /// yields [`RoundBudgetExceeded`](ColdStartError::RoundBudgetExceeded). Each /// round's results are absorbed into the report **before** its `error` is /// checked; a round error propagates after restoring the pin and **without** - /// calling `on_results`. + /// calling `on_results`. Note the absorbed report is returned only on the + /// `Ok` path — on error it is dropped and the `Err` carries only the cause; + /// use [`execute_cold_start_round`](Self::execute_cold_start_round) directly + /// to observe a failed round's partial outcomes. pub fn run_cold_start( &mut self, planner: &mut dyn ColdStartPlanner, diff --git a/src/cold_start/results.rs b/src/cold_start/results.rs index db8f38c..8c43e1b 100644 --- a/src/cold_start/results.rs +++ b/src/cold_start/results.rs @@ -19,6 +19,9 @@ use revm::context::result::ExecutionResult; /// than as absence. `verified` carries only the slots whose value actually changed /// (and were injected). `discovered` carries one [`ColdStartCallResult`] per /// discover call. +/// +/// The order of entries in `fetched` / `probed` is unspecified — look up a slot +/// by `(address, slot)` rather than relying on it matching the plan's order. #[derive(Clone, Debug, Default)] pub struct ColdStartResults { /// Slots whose value changed and were injected (one per change).