Add reactive runtime + cold-start sync (the state-engine substrate the ecosystem consumes)

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

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

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<tokio::runtime::Handle> {
+pub(crate) fn block_in_place_handle() -> Result<tokio::runtime::Handle> {
     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<SlotChange>, 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<U256>) -> 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<SlotChange>, Vec<SlotOutcome>)> {
         if slots.is_empty() {
-            return Ok((Vec::new(), 0));
+            return Ok((Vec::new(), Vec::new()));
         }
         let fetcher = self
             .storage_batch_fetcher
@@ -2043,17 +2095,28 @@ 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) => {
                     debug!(%addr, %slot, error = %e, "verify_slots: fetch failed, skipping slot");
                     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<SlotChange>, Vec<SlotOutcome>)> {
+        self.verify_slots_core(slots)
     }
 
     /// Reconciliation re-read used by [`EventPipeline::reconcile`](crate::events::EventPipeline::reconcile).

src/cold_start/config.rs

@@ -0,0 +1,148 @@
+//! 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 cold-start run, returned by
+/// [`run_cold_start`](crate::cache::EvmCache::run_cold_start) on success.
+///
+/// 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.
+    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 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 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,
+    /// One summary per round, in execution order.
+    pub per_round: Vec<ColdStartRoundSummary>,
+}
+
+/// 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,
+        });
+    }
+}