Skip to main content

evm_fork_cache/
freshness.rs

1//! Freshness control plane and the optimistic verify-and-rerun execution loop.
2//!
3//! This module is the generic core of the engine's "honest freshness" model: it
4//! knows which cached state it can trust, for how long, and how to keep the rest
5//! correct without blocking simulations on RPC. It is built from four layers:
6//!
7//! 1. **Classification** — [`Validity`] (`Pinned` / `Volatile` / `ValidThrough`)
8//!    and the [`FreshnessRegistry`] that resolves a validity per `(address, slot)`
9//!    with the precedence **slot ▸ account ▸ default**.
10//! 2. **Observation** — [`SlotObservationTracker`] records per-slot change
11//!    frequency (clock-agnostic) to drive adaptive re-verification, tuned by
12//!    [`FreshnessParams`].
13//! 3. **Policy** — the [`FreshnessPolicy`] trait decides *which* volatile slots to
14//!    verify this cycle; built-ins are [`AlwaysVerify`], [`NeverVerify`] and
15//!    [`ObservationDriven`].
16//! 4. **Mechanism** — `EvmCache::verify_slots` / `EvmCache::purge_account`, and
17//!    the freshness controller that runs the optimistic loop.
18//!
19//! The clock is configurable via [`FreshnessClock`]: [`BlockClock`] (the default,
20//! block-number based) or [`WallClock`] (unix seconds). The controller threads
21//! `clock.now()` as `now: u64` through the tracker, the policy, and
22//! [`FreshnessRegistry::is_volatile`].
23//!
24//! # Reconciliation scope
25//!
26//! The optimistic loop verifies only **volatile storage slots** in each sim's
27//! read set. Account-level state — native balance, nonce, and bytecode — is
28//! **not** re-fetched or diffed, so [`Validation::Confirmed`] means *"no volatile
29//! storage slot the sims read had changed"*, not *"no account state changed"*. A
30//! sim whose result depends on a `BALANCE`/`SELFBALANCE` (or nonce/code) that
31//! moved on-chain without a co-changing storage slot in its read set can still be
32//! reported `Confirmed`. If account-level state matters to a sim, mark the
33//! account [`Validity::Pinned`] and keep it fresh via event-driven writes, or
34//! reconcile it out of band. See [`Validation`] for the per-verdict note.
35//!
36//! # Example
37//!
38//! Classification + policy selection, no network required:
39//!
40//! ```
41//! use alloy_primitives::{Address, U256};
42//! use evm_fork_cache::freshness::{
43//!     AlwaysVerify, FreshnessPolicy, FreshnessRegistry, NeverVerify,
44//! };
45//! use evm_fork_cache::cache::SlotObservationTracker;
46//!
47//! let contract = Address::repeat_byte(0x01);
48//! let volatile_slot = U256::from(0);
49//! let immutable_slot = U256::from(6); // e.g. a constructor-set config value
50//!
51//! let mut registry = FreshnessRegistry::new(); // default: Volatile
52//! registry.pin_slot(contract, immutable_slot); // never re-verified
53//!
54//! // `now` is in clock units (block number for the default BlockClock).
55//! let now = 100;
56//! assert!(registry.is_volatile(contract, volatile_slot, now));
57//! assert!(!registry.is_volatile(contract, immutable_slot, now));
58//!
59//! // Policies pick which volatile candidates to verify this cycle.
60//! let obs = SlotObservationTracker::new();
61//! let candidates = [(contract, volatile_slot)];
62//! assert_eq!(
63//!     AlwaysVerify.select(&candidates, &obs, now),
64//!     vec![(contract, volatile_slot)]
65//! );
66//! assert!(NeverVerify.select(&candidates, &obs, now).is_empty());
67//! ```
68
69use std::collections::{HashMap, HashSet};
70use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
71use std::sync::{Arc, Mutex};
72use std::time::{SystemTime, UNIX_EPOCH};
73
74use alloy_eips::BlockId;
75use alloy_eips::eip2930::AccessList;
76use alloy_primitives::{Address, Bytes, U256};
77use revm::context::result::ExecutionResult;
78use tokio::task::JoinHandle;
79
80use crate::cache::{
81    CallSimulationResult, EvmCache, EvmOverlay, EvmSnapshot, SimStatus, SlotObservationTracker,
82    StorageBatchFetchFn, TxConfig,
83};
84use crate::state_update::StateUpdate;
85
86/// Default minimum observations before the change-frequency data is trusted.
87pub const DEFAULT_MIN_OBSERVATIONS: u32 = 10;
88
89/// Default maximum reuse window, in clock units, before a slot is rechecked.
90///
91/// Block-based default (≈300 blocks). Wall-clock users typically set this to
92/// `7 * 86400` (one week) to reproduce the original behavior.
93pub const DEFAULT_MAX_REUSE: u64 = 300;
94
95/// Default refetch threshold on expected probability of change.
96pub const DEFAULT_STALENESS_THRESHOLD: f64 = 0.05;
97
98/// Default change-rate above which a slot is always refetched.
99pub const DEFAULT_ALWAYS_REFETCH_RATE: f64 = 0.9;
100
101/// Default clock units per "cycle" used by the probabilistic model.
102pub const DEFAULT_CYCLE_INTERVAL: u64 = 1;
103
104/// Tunable thresholds for the adaptive freshness model.
105///
106/// All time-like fields are expressed in **clock units** (`FreshnessClock`):
107/// block numbers for a block clock, unix seconds for a wall clock. The defaults
108/// are block-oriented; wall-clock users should raise [`max_reuse`](Self::max_reuse)
109/// and [`cycle_interval`](Self::cycle_interval) accordingly.
110#[derive(Clone, Debug, PartialEq)]
111pub struct FreshnessParams {
112    /// Minimum observations before the change frequency is trusted (else refetch).
113    pub min_observations: u32,
114    /// Maximum reuse window (clock units) before a slot is force-rechecked.
115    pub max_reuse: u64,
116    /// Refetch when the expected probability of change exceeds this threshold.
117    pub staleness_threshold: f64,
118    /// Slots changing more often than this rate are always refetched.
119    pub always_refetch_rate: f64,
120    /// Clock units per "cycle" for the probabilistic expected-change estimate.
121    /// Must be non-zero; a zero is treated as one to avoid division by zero.
122    pub cycle_interval: u64,
123}
124
125impl Default for FreshnessParams {
126    fn default() -> Self {
127        Self {
128            min_observations: DEFAULT_MIN_OBSERVATIONS,
129            max_reuse: DEFAULT_MAX_REUSE,
130            staleness_threshold: DEFAULT_STALENESS_THRESHOLD,
131            always_refetch_rate: DEFAULT_ALWAYS_REFETCH_RATE,
132            cycle_interval: DEFAULT_CYCLE_INTERVAL,
133        }
134    }
135}
136
137impl FreshnessParams {
138    /// Block-oriented defaults (`max_reuse ≈ 300` blocks, one cycle per block).
139    pub fn for_block_clock() -> Self {
140        Self::default()
141    }
142
143    /// Wall-clock defaults: reuse up to one week, ~60s cycles, matching the
144    /// original (pre-Phase-2) hardcoded behavior of the observation tracker.
145    pub fn for_wall_clock() -> Self {
146        Self {
147            max_reuse: 7 * 86400,
148            cycle_interval: 60,
149            ..Self::default()
150        }
151    }
152}
153
154// ---------------------------------------------------------------------------
155// 1. Classification
156// ---------------------------------------------------------------------------
157
158/// How long a cached account or storage slot can be trusted.
159///
160/// Resolution precedence is **slot ▸ account ▸ default** (see
161/// [`FreshnessRegistry::validity`]).
162#[derive(Clone, Copy, Debug, PartialEq, Eq)]
163pub enum Validity {
164    /// Caller-owned: immutable, or kept fresh out-of-band (e.g. via event
165    /// writes). The freshness system never re-verifies or purges it.
166    Pinned,
167    /// Governed by the active [`FreshnessPolicy`]; may be re-verified each cycle.
168    Volatile,
169    /// Pinned until clock value `N` (inclusive), then treated as [`Volatile`].
170    ///
171    /// [`Volatile`]: Validity::Volatile
172    ValidThrough(u64),
173}
174
175/// Per-address / per-slot validity classification.
176///
177/// A slot's validity is resolved with the precedence **slot ▸ account ▸
178/// default**: an explicit `(address, slot)` entry wins, else the account-level
179/// entry for `address`, else the registry default ([`Validity::Volatile`] unless
180/// changed via [`with_default`](Self::with_default)).
181///
182/// The setters are builder-style (`&mut Self`) so they can be chained.
183#[derive(Clone, Debug)]
184pub struct FreshnessRegistry {
185    default: Validity,
186    accounts: HashMap<Address, Validity>,
187    slots: HashMap<(Address, U256), Validity>,
188}
189
190impl Default for FreshnessRegistry {
191    fn default() -> Self {
192        Self::new()
193    }
194}
195
196impl FreshnessRegistry {
197    /// A registry whose default validity is [`Validity::Volatile`].
198    pub fn new() -> Self {
199        Self {
200            default: Validity::Volatile,
201            accounts: HashMap::new(),
202            slots: HashMap::new(),
203        }
204    }
205
206    /// A registry with a custom default validity for unclassified state.
207    pub fn with_default(default: Validity) -> Self {
208        Self {
209            default,
210            accounts: HashMap::new(),
211            slots: HashMap::new(),
212        }
213    }
214
215    /// The default validity applied when neither the slot nor the account is set.
216    pub fn default_validity(&self) -> Validity {
217        self.default
218    }
219
220    /// Pin an account ([`Validity::Pinned`]).
221    pub fn pin(&mut self, addr: Address) -> &mut Self {
222        self.set_account(addr, Validity::Pinned)
223    }
224
225    /// Pin a single slot ([`Validity::Pinned`]).
226    pub fn pin_slot(&mut self, addr: Address, slot: U256) -> &mut Self {
227        self.set_slot(addr, slot, Validity::Pinned)
228    }
229
230    /// Mark an account [`Validity::Volatile`].
231    pub fn mark_volatile(&mut self, addr: Address) -> &mut Self {
232        self.set_account(addr, Validity::Volatile)
233    }
234
235    /// Mark a single slot [`Validity::Volatile`].
236    pub fn mark_volatile_slot(&mut self, addr: Address, slot: U256) -> &mut Self {
237        self.set_slot(addr, slot, Validity::Volatile)
238    }
239
240    /// Mark an account [`Validity::ValidThrough`] block/clock `n`.
241    pub fn valid_through(&mut self, addr: Address, n: u64) -> &mut Self {
242        self.set_account(addr, Validity::ValidThrough(n))
243    }
244
245    /// Mark a single slot [`Validity::ValidThrough`] block/clock `n`.
246    pub fn valid_through_slot(&mut self, addr: Address, slot: U256, n: u64) -> &mut Self {
247        self.set_slot(addr, slot, Validity::ValidThrough(n))
248    }
249
250    /// Set the account-level validity for `addr`.
251    pub fn set_account(&mut self, addr: Address, validity: Validity) -> &mut Self {
252        self.accounts.insert(addr, validity);
253        self
254    }
255
256    /// Set the slot-level validity for `(addr, slot)`.
257    pub fn set_slot(&mut self, addr: Address, slot: U256, validity: Validity) -> &mut Self {
258        self.slots.insert((addr, slot), validity);
259        self
260    }
261
262    /// Resolve the validity of `(addr, slot)` with **slot ▸ account ▸ default**.
263    pub fn validity(&self, addr: Address, slot: U256) -> Validity {
264        if let Some(v) = self.slots.get(&(addr, slot)) {
265            return *v;
266        }
267        if let Some(v) = self.accounts.get(&addr) {
268            return *v;
269        }
270        self.default
271    }
272
273    /// Whether `(addr, slot)` is currently volatile (subject to verification).
274    ///
275    /// `true` for [`Validity::Volatile`], and for [`Validity::ValidThrough`]`(m)`
276    /// once `now > m`. `false` for [`Validity::Pinned`] and a still-valid
277    /// `ValidThrough` (`now <= m`).
278    pub fn is_volatile(&self, addr: Address, slot: U256, now: u64) -> bool {
279        match self.validity(addr, slot) {
280            Validity::Pinned => false,
281            Validity::Volatile => true,
282            Validity::ValidThrough(m) => now > m,
283        }
284    }
285}
286
287// ---------------------------------------------------------------------------
288// 2. Clock
289// ---------------------------------------------------------------------------
290
291/// Source of the current clock value used throughout the freshness model.
292///
293/// Implementations return a monotone-ish `u64` in their own units. The two
294/// built-ins are [`BlockClock`] (block number, the default) and [`WallClock`]
295/// (unix seconds).
296pub trait FreshnessClock: Send + Sync {
297    /// The current clock value (block number or unix seconds).
298    fn now(&self) -> u64;
299
300    /// Advance the clock to `now`.
301    ///
302    /// Called by [`FreshnessController::on_new_block`] so the natural API drives
303    /// the clock forward. The default is a no-op (for clocks like [`WallClock`]
304    /// that advance on their own); [`BlockClock`] overrides it to set the block.
305    fn advance(&self, _now: u64) {}
306}
307
308/// Block-number clock (the default). Cloning shares the underlying counter, so a
309/// clone observed by a background task sees [`set_block`](Self::set_block)
310/// updates made on the main thread.
311#[derive(Clone, Debug, Default)]
312pub struct BlockClock(Arc<AtomicU64>);
313
314impl BlockClock {
315    /// A block clock starting at block 0.
316    pub fn new() -> Self {
317        Self(Arc::new(AtomicU64::new(0)))
318    }
319
320    /// A block clock starting at `block`.
321    pub fn at(block: u64) -> Self {
322        Self(Arc::new(AtomicU64::new(block)))
323    }
324
325    /// Set the current block number. Shared across clones.
326    pub fn set_block(&self, block: u64) {
327        self.0.store(block, Ordering::Relaxed);
328    }
329}
330
331impl FreshnessClock for BlockClock {
332    fn now(&self) -> u64 {
333        self.0.load(Ordering::Relaxed)
334    }
335
336    /// Set the current block to `now` (shared across clones).
337    fn advance(&self, now: u64) {
338        self.set_block(now);
339    }
340}
341
342/// Wall-clock clock: [`now`](FreshnessClock::now) returns unix seconds.
343///
344/// A zero-sized unit struct: unlike [`BlockClock`] it holds no `Arc`/`AtomicU64`,
345/// since the value is read straight from the system clock on each call. It
346/// advances on its own, so [`advance`](FreshnessClock::advance) is the trait
347/// default no-op and has no effect.
348#[derive(Clone, Copy, Debug, Default)]
349pub struct WallClock;
350
351impl FreshnessClock for WallClock {
352    fn now(&self) -> u64 {
353        SystemTime::now()
354            .duration_since(UNIX_EPOCH)
355            .map(|d| d.as_secs())
356            .unwrap_or(0)
357    }
358}
359
360// ---------------------------------------------------------------------------
361// 3. Policy
362// ---------------------------------------------------------------------------
363
364/// Decides which volatile candidate slots must be verified this cycle.
365///
366/// The controller passes the volatile candidates (predicted read set) plus the
367/// current observation stats and `now`; the policy returns the subset to
368/// re-fetch. Correctness does not depend on the policy being complete — the
369/// background validator always re-checks each sim's *actual* volatile read set
370/// before trusting results — so a policy only trades RPC cost against how often a
371/// `Corrected` verdict is needed.
372pub trait FreshnessPolicy: Send {
373    /// Of these volatile candidate slots, which must be verified this cycle?
374    fn select(
375        &mut self,
376        candidates: &[(Address, U256)],
377        obs: &SlotObservationTracker,
378        now: u64,
379    ) -> Vec<(Address, U256)>;
380
381    /// Hook called when the controller advances to a new block.
382    fn on_new_block(&mut self, _block: u64) {}
383}
384
385/// Verifies every volatile candidate (safe / eager). Always correct, most RPC.
386#[derive(Clone, Copy, Debug, Default)]
387pub struct AlwaysVerify;
388
389impl FreshnessPolicy for AlwaysVerify {
390    fn select(
391        &mut self,
392        candidates: &[(Address, U256)],
393        _obs: &SlotObservationTracker,
394        _now: u64,
395    ) -> Vec<(Address, U256)> {
396        candidates.to_vec()
397    }
398}
399
400/// Verifies nothing (trust-all). Selects no slots from the predicted set, though
401/// the actual-read-set reconcile in the background validator can still surface
402/// changes.
403#[derive(Clone, Copy, Debug, Default)]
404pub struct NeverVerify;
405
406impl FreshnessPolicy for NeverVerify {
407    fn select(
408        &mut self,
409        _candidates: &[(Address, U256)],
410        _obs: &SlotObservationTracker,
411        _now: u64,
412    ) -> Vec<(Address, U256)> {
413        Vec::new()
414    }
415}
416
417/// Adaptive policy: verifies candidates the observation tracker flags via
418/// [`SlotObservationTracker::should_refetch`](crate::cache::SlotObservationTracker::should_refetch),
419/// driven by the thresholds in [`FreshnessParams`].
420#[derive(Clone, Debug, Default)]
421pub struct ObservationDriven {
422    /// Thresholds for the underlying [`SlotObservationTracker::should_refetch`](crate::cache::SlotObservationTracker::should_refetch)
423    /// heuristic.
424    pub params: FreshnessParams,
425}
426
427impl ObservationDriven {
428    /// An observation-driven policy with the given parameters.
429    pub fn new(params: FreshnessParams) -> Self {
430        Self { params }
431    }
432}
433
434impl FreshnessPolicy for ObservationDriven {
435    fn select(
436        &mut self,
437        candidates: &[(Address, U256)],
438        obs: &SlotObservationTracker,
439        now: u64,
440    ) -> Vec<(Address, U256)> {
441        candidates
442            .iter()
443            .copied()
444            .filter(|(addr, slot)| obs.should_refetch(*addr, *slot, now, &self.params))
445            .collect()
446    }
447}
448
449// ---------------------------------------------------------------------------
450// 4. Results
451// ---------------------------------------------------------------------------
452
453/// A storage slot whose value changed: `old` is the prior cached/snapshot value
454/// (`ZERO` if previously uncached), `new` is the resulting value.
455///
456/// Produced by two paths: the freshness verifier
457/// ([`EvmCache::verify_slots`](crate::cache::EvmCache::verify_slots) and the
458/// background validator), where `new` is a freshly-fetched value that differed
459/// from the cache; and the state-update writer
460/// ([`EvmCache::apply_update`](crate::cache::EvmCache::apply_update) /
461/// [`apply_updates`](crate::cache::EvmCache::apply_updates)), where `new` is the
462/// value just written.
463#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
464pub struct SlotChange {
465    /// Contract whose storage changed.
466    pub address: Address,
467    /// Storage slot key.
468    pub slot: U256,
469    /// Value previously held in the cache/snapshot.
470    pub old: U256,
471    /// Freshly-fetched value.
472    pub new: U256,
473}
474
475/// The classified outcome of fetching a single storage slot.
476///
477/// Where a [`SlotChange`] records only slots whose value *differed* from the
478/// cache, a [`SlotOutcome`] is produced for **every** requested slot — including
479/// ones that did not change and ones the fetcher could not return. This closes
480/// the "archive-miss" gap: a transient fetch failure is surfaced explicitly as
481/// [`SlotFetch::FetchFailed`] rather than collapsing into the same "no value"
482/// signal as a genuine on-chain zero ([`SlotFetch::Zero`]).
483///
484/// The fetch classification ([`SlotFetch`]) and change detection ([`SlotChange`])
485/// are independent reads of the same fetched value: a genuine `Ok(0)` on a slot
486/// whose cached value was also `0` yields [`SlotFetch::Zero`] **and** no
487/// `SlotChange`.
488#[derive(Clone, Debug, PartialEq, Eq)]
489pub struct SlotOutcome {
490    /// Contract whose storage slot was fetched.
491    pub address: Address,
492    /// Storage slot key.
493    pub slot: U256,
494    /// The classified result of fetching this slot.
495    pub fetch: SlotFetch,
496}
497
498/// The classified result of an individual slot fetch.
499///
500/// A fetcher `Ok(value)` is classified into [`Value`](SlotFetch::Value) (non-zero)
501/// or [`Zero`](SlotFetch::Zero) (a genuine on-chain zero); a fetcher `Err`
502/// becomes [`FetchFailed`](SlotFetch::FetchFailed), carrying the error string.
503/// [`NotAttempted`](SlotFetch::NotAttempted) marks a declared slot that a
504/// short-circuited round never reached (produced only by the accounts/discover
505/// phases of a cold-start round, never by verify).
506#[derive(Clone, Debug, PartialEq, Eq)]
507pub enum SlotFetch {
508    /// The slot was fetched and holds a non-zero value.
509    Value(U256),
510    /// The slot was fetched and holds a genuine on-chain zero.
511    Zero,
512    /// The fetcher returned an error for this slot; `reason` is its description.
513    FetchFailed {
514        /// Human-readable description of why the fetch failed.
515        reason: String,
516    },
517    /// The slot was declared but never reached because the round
518    /// short-circuited on an earlier-phase hard error.
519    NotAttempted,
520}
521
522/// The deferred verdict on a [`SpeculativeSim`]'s optimistic results.
523///
524/// # Reconciliation scope
525///
526/// The verdict reflects only **volatile storage slots** in each sim's read set.
527/// Account-level state — native balance, nonce, and bytecode — is **not**
528/// re-verified, so a sim whose result depends on a `BALANCE`/`SELFBALANCE` (or
529/// nonce/code) that changed on-chain *without* a co-changing storage slot in its
530/// read set can still be reported [`Confirmed`](Validation::Confirmed). Classify
531/// such accounts as [`Validity::Pinned`] and keep them fresh via event-driven
532/// writes if their account-level state matters to your sims. See the module-level
533/// docs for the full freshness contract.
534pub enum Validation {
535    /// No **volatile storage slot** in the sims' read sets had changed, so the
536    /// optimistic results are trusted. Note this does *not* cover account-level
537    /// balance/nonce/code changes — see the [type-level scope](Validation).
538    Confirmed,
539    /// At least one read storage slot changed. `results` is the optimistic set
540    /// with the affected sims re-run against the fresh values; `changed` lists the
541    /// slots that differed (also queued for flow-back into the cache). Only
542    /// storage slots are reconciled — account-level state is not (see the
543    /// [type-level scope](Validation)).
544    Corrected {
545        /// Optimistic results with the affected sims replaced by re-runs.
546        results: Vec<CallSimulationResult>,
547        /// Slots whose fresh value differed from the snapshot.
548        changed: Vec<SlotChange>,
549    },
550    /// The fetcher failed, so the results could not be validated. The optimistic
551    /// results are *not* trusted.
552    Unverified {
553        /// Human-readable description of why validation could not complete.
554        reason: String,
555    },
556}
557
558impl std::fmt::Debug for Validation {
559    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
560        match self {
561            Validation::Confirmed => write!(f, "Confirmed"),
562            Validation::Corrected { changed, .. } => f
563                .debug_struct("Corrected")
564                .field("changed", changed)
565                .finish_non_exhaustive(),
566            Validation::Unverified { reason } => f
567                .debug_struct("Unverified")
568                .field("reason", reason)
569                .finish(),
570        }
571    }
572}
573
574/// A single non-committing simulation request for the optimistic loop.
575///
576/// `tx.access_list` is the *predicted* read set (a performance hint that seeds
577/// the verify candidates); correctness does not depend on it because the
578/// background validator re-checks each sim's actual volatile read set.
579#[derive(Clone, Debug)]
580pub struct SimRequest {
581    /// Transaction sender.
582    pub from: Address,
583    /// Call target.
584    pub to: Address,
585    /// Calldata.
586    pub calldata: Bytes,
587    /// Per-call tx environment; `tx.access_list` is the predicted read set.
588    pub tx: TxConfig,
589}
590
591impl SimRequest {
592    /// A zero-value request with default tx environment.
593    pub fn new(from: Address, to: Address, calldata: Bytes) -> Self {
594        Self {
595            from,
596            to,
597            calldata,
598            tx: TxConfig::default(),
599        }
600    }
601
602    /// Set the predicted read set (EIP-2930 access list hint).
603    pub fn with_access_list(mut self, access_list: AccessList) -> Self {
604        self.tx.access_list = Some(access_list);
605        self
606    }
607
608    /// Set the native value (wei) sent with the call (e.g. for a payable call).
609    pub fn with_value(mut self, value: U256) -> Self {
610        self.tx.value = value;
611        self
612    }
613
614    /// Set the gas limit for the call (e.g. to model out-of-gas behavior).
615    pub fn with_gas_limit(mut self, gas_limit: u64) -> Self {
616        self.tx.gas_limit = Some(gas_limit);
617        self
618    }
619
620    /// Set the gas price (wei) for the call.
621    pub fn with_gas_price(mut self, gas_price: u128) -> Self {
622        self.tx.gas_price = Some(gas_price);
623        self
624    }
625}
626
627/// Optimistic simulation results plus a handle to their deferred validation.
628///
629/// Returned by [`FreshnessController::run`] as soon as the optimistic sims
630/// finish (without awaiting RPC). Read [`optimistic`](Self::optimistic)
631/// immediately, then `await` [`validate`](Self::validate) for the verdict.
632///
633/// # Cancellation (best-effort)
634/// Dropping this — or calling [`into_optimistic`](Self::into_optimistic) — sets a
635/// cancel flag and aborts the background task. Cancellation is **cooperative and
636/// best-effort, not instantaneous**: `run_validator` is synchronous, so an abort
637/// cannot preempt it once it is running. Instead the validator checks the flag at
638/// several checkpoints — before each fetch, and (on the first pass) after a fetch
639/// returns but before it records observations or queues corrections — so a cancel
640/// observed at a checkpoint skips the side effects downstream of it. A validator
641/// already executing a synchronous step (e.g. mid-fetch) completes that step
642/// before reaching the next checkpoint, and corrections accumulated up to a fetch
643/// that completed in the *final* loop iteration may still be queued for flow-back
644/// (the post-loop queue is not guarded by an immediately-preceding checkpoint).
645/// The intent is that a dropped speculation stops doing further work and, in the
646/// common case, does not flow corrections back into the cache; it does not
647/// guarantee that an in-flight or just-completed fetch's correction is withheld.
648pub struct SpeculativeSim {
649    optimistic: Vec<CallSimulationResult>,
650    /// `Option` so `validate`/`into_optimistic` can take the handle and skip the
651    /// abort-on-drop; `Drop` only aborts a handle still left in place.
652    validation: Option<JoinHandle<Validation>>,
653    /// Set when the caller drops or [`into_optimistic`](Self::into_optimistic)s
654    /// this handle; the validator polls it at its checkpoints to bail out before
655    /// causing side effects (fetching, observing, queuing corrections).
656    cancelled: Arc<AtomicBool>,
657}
658
659impl SpeculativeSim {
660    /// The optimistic results, readable before validation completes.
661    ///
662    /// These (and the re-run results in [`Validation::Corrected`]) carry an
663    /// **empty `token_deltas`** map: the optimistic loop does not run transfer
664    /// tracking, so the signed per-token balance deltas populated by the
665    /// committing `simulate_call_with_balance_deltas` path are not available here.
666    pub fn optimistic(&self) -> &[CallSimulationResult] {
667        &self.optimistic
668    }
669
670    /// Consume the handle and return the optimistic results, aborting the
671    /// background validation task.
672    ///
673    /// # Panics
674    /// The validation [`JoinHandle`] is single-consumption. Because this takes
675    /// `self` by value, it and [`validate`](Self::validate) are mutually
676    /// exclusive: only one of them can ever run for a given `SpeculativeSim`, and
677    /// each takes the handle. `into_optimistic` takes the handle defensively (it
678    /// does not panic if the handle is already gone), whereas `validate` panics
679    /// with `"validation handle taken twice"` if it is invoked once the handle has
680    /// been consumed.
681    pub fn into_optimistic(mut self) -> Vec<CallSimulationResult> {
682        self.cancelled.store(true, Ordering::Relaxed);
683        if let Some(handle) = self.validation.take() {
684            handle.abort();
685        }
686        std::mem::take(&mut self.optimistic)
687    }
688
689    /// Await the deferred validation verdict.
690    ///
691    /// If the background task failed to complete (e.g. it panicked), returns
692    /// [`Validation::Unverified`]. This consumes `self`, so it is mutually
693    /// exclusive with the cancel paths ([`into_optimistic`](Self::into_optimistic)
694    /// / drop) — a handle that is awaited here is never cancelled.
695    ///
696    /// # Panics
697    /// The validation [`JoinHandle`] is single-consumption: it is taken by the
698    /// first of `validate` or [`into_optimistic`](Self::into_optimistic) to run.
699    /// `validate` panics with `"validation handle taken twice"` if the handle has
700    /// already been consumed. Both take `self` by value, so under normal ownership
701    /// this is unreachable.
702    pub async fn validate(mut self) -> Validation {
703        let handle = self
704            .validation
705            .take()
706            .expect("validation handle taken twice");
707        match handle.await {
708            Ok(v) => v,
709            Err(e) => Validation::Unverified {
710                reason: format!("validation task failed: {e}"),
711            },
712        }
713    }
714}
715
716impl Drop for SpeculativeSim {
717    fn drop(&mut self) {
718        self.cancelled.store(true, Ordering::Relaxed);
719        if let Some(handle) = self.validation.take() {
720            handle.abort();
721        }
722    }
723}
724
725// ---------------------------------------------------------------------------
726// Controller
727// ---------------------------------------------------------------------------
728
729/// Drives the optimistic verify-and-rerun loop over an [`EvmCache`].
730///
731/// Holds the freshness [`FreshnessRegistry`], the shared
732/// [`SlotObservationTracker`], a [`FreshnessPolicy`], a [`FreshnessClock`], and
733/// the pending-corrections queue. The tracker and the pending queue are
734/// `Arc<Mutex<…>>` so the background validator can update them without touching
735/// the `!Send` cache. Adaptive thresholds ([`FreshnessParams`]) live on the
736/// policy that uses them ([`ObservationDriven`]), not on the controller.
737///
738/// # Runtime requirement
739/// [`run`](Self::run) spawns a background task and the (synchronous) fetcher uses
740/// `block_in_place` internally, so a **multi-thread** tokio runtime is required
741/// (`#[tokio::main(flavor = "multi_thread")]` or
742/// `Builder::new_multi_thread()`), mirroring the [`EvmCache`] constructor note.
743pub struct FreshnessController<P: FreshnessPolicy, C: FreshnessClock> {
744    registry: FreshnessRegistry,
745    tracker: Arc<Mutex<SlotObservationTracker>>,
746    policy: P,
747    clock: C,
748    pending: Arc<Mutex<Vec<SlotChange>>>,
749    /// Cumulative count of background re-runs performed by the validator across
750    /// all `run` calls. Shared with the spawned task; incremented once per
751    /// re-executed sim. Lets callers observe that selective re-run actually
752    /// skipped the unaffected sims rather than re-running every one.
753    rerun_count: Arc<AtomicUsize>,
754}
755
756impl<P: FreshnessPolicy> FreshnessController<P, BlockClock> {
757    /// Build a controller with the default [`BlockClock`] (starting at block 0).
758    ///
759    /// Starts with a fresh, empty [`SlotObservationTracker`] and an empty
760    /// pending-corrections queue. Use [`with_tracker`](Self::with_tracker) to share
761    /// a persisted tracker, or [`with_clock`](Self::with_clock) for a non-default
762    /// clock such as [`WallClock`].
763    pub fn new(registry: FreshnessRegistry, policy: P) -> Self {
764        Self::with_clock(registry, policy, BlockClock::new())
765    }
766}
767
768impl<P: FreshnessPolicy, C: FreshnessClock> FreshnessController<P, C> {
769    /// Build a controller with an explicit clock.
770    ///
771    /// Starts with a fresh, empty [`SlotObservationTracker`] and an empty
772    /// pending-corrections queue. The clock's units must match those the
773    /// `policy`'s [`FreshnessParams`] were tuned for (block numbers for
774    /// [`BlockClock`], unix seconds for [`WallClock`]).
775    pub fn with_clock(registry: FreshnessRegistry, policy: P, clock: C) -> Self {
776        Self {
777            registry,
778            tracker: Arc::new(Mutex::new(SlotObservationTracker::new())),
779            policy,
780            clock,
781            pending: Arc::new(Mutex::new(Vec::new())),
782            rerun_count: Arc::new(AtomicUsize::new(0)),
783        }
784    }
785
786    /// Use an existing shared observation tracker (e.g. a persisted one).
787    ///
788    /// Builder-style override that replaces the fresh tracker installed by
789    /// [`new`](Self::new) / [`with_clock`](Self::with_clock) with the given shared
790    /// handle, so change-frequency history survives across runs or is shared with
791    /// other components. The background validator updates this same tracker under
792    /// its `Mutex`.
793    pub fn with_tracker(mut self, tracker: Arc<Mutex<SlotObservationTracker>>) -> Self {
794        self.tracker = tracker;
795        self
796    }
797
798    /// The shared observation tracker.
799    pub fn tracker(&self) -> &Arc<Mutex<SlotObservationTracker>> {
800        &self.tracker
801    }
802
803    /// The freshness registry.
804    pub fn registry(&self) -> &FreshnessRegistry {
805        &self.registry
806    }
807
808    /// Mutable access to the freshness registry.
809    pub fn registry_mut(&mut self) -> &mut FreshnessRegistry {
810        &mut self.registry
811    }
812
813    /// Number of corrections waiting to be drained into the cache on the next
814    /// [`run`](Self::run).
815    pub fn pending_len(&self) -> usize {
816        self.pending.lock().unwrap_or_else(|e| e.into_inner()).len()
817    }
818
819    /// Cumulative number of background re-runs performed by the validator across
820    /// all [`run`](Self::run) calls so far.
821    ///
822    /// Incremented once per sim that the reconcile step actually re-executes
823    /// (i.e. whose read set intersected a changed slot). A `Corrected` verdict
824    /// over `n` requests where only one slot changed advances this by the number
825    /// of *affected* sims, not by `n` — making the selective-re-run behavior
826    /// directly observable.
827    pub fn rerun_count(&self) -> usize {
828        self.rerun_count.load(Ordering::Relaxed)
829    }
830
831    /// Advance to a new block.
832    ///
833    /// Advances the clock to `block` (a no-op for [`WallClock`], a `set_block`
834    /// for [`BlockClock`]) and then notifies the policy. Advancing the clock is
835    /// what ages [`Validity::ValidThrough`] slots into [`Validity::Volatile`] and
836    /// progresses the observation-tracker reuse window through the natural API.
837    pub fn on_new_block(&mut self, block: u64) {
838        self.clock.advance(block);
839        self.policy.on_new_block(block);
840    }
841
842    /// Run the optimistic loop for a batch of requests.
843    ///
844    /// 1. Drain queued corrections from prior cycles into the cache.
845    /// 2. Snapshot the cache and grab the batch fetcher.
846    /// 3. Run each request optimistically against the snapshot, capturing its
847    ///    actual volatile read set.
848    /// 4. Compute the predicted volatile candidates and ask the policy which to
849    ///    verify.
850    /// 5. Spawn the background validator (Send data only) and return a
851    ///    [`SpeculativeSim`] immediately.
852    ///
853    /// # Panics
854    /// Must be called from within a tokio runtime: it calls `tokio::spawn` to
855    /// launch the background validator, which panics (`there is no reactor
856    /// running`) if no runtime is active. The spawned (synchronous) fetcher
857    /// additionally uses `tokio::task::block_in_place` internally, so the runtime
858    /// must be **multi-thread** (`#[tokio::main(flavor = "multi_thread")]` or
859    /// `Builder::new_multi_thread()`); on a current-thread runtime `block_in_place`
860    /// panics, mirroring the [`EvmCache`] constructor note.
861    ///
862    /// # Errors
863    /// Returns an error if any optimistic simulation fails to execute against the
864    /// freshly-created snapshot (propagated from
865    /// `EvmOverlay::call_raw_with_access_list`).
866    pub fn run(
867        &mut self,
868        cache: &mut EvmCache,
869        requests: Vec<SimRequest>,
870    ) -> anyhow::Result<SpeculativeSim> {
871        let now = self.clock.now();
872
873        // 1. Drain pending corrections into the cache before snapshotting.
874        //    Routed through the unified write primitive (`apply_updates` of
875        //    write-through `Slot`s); behavior-identical to the old
876        //    `inject_storage_batch_fresh`, demonstrating the one write path.
877        {
878            let mut pending = self.pending.lock().unwrap_or_else(|e| e.into_inner());
879            if !pending.is_empty() {
880                let injects: Vec<StateUpdate> = pending
881                    .iter()
882                    .map(|c| StateUpdate::slot(c.address, c.slot, c.new))
883                    .collect();
884                cache.apply_updates(&injects);
885                pending.clear();
886            }
887        }
888
889        // 2. Snapshot + fetcher (Arc clones, both Send). Capture the cache's
890        //    pinned block now, so the deferred validator fetches at the block the
891        //    snapshot was built from even if the cache is re-pinned meanwhile.
892        let snapshot = cache.create_snapshot();
893        let fetcher = cache.storage_batch_fetcher().cloned();
894        let validation_block = cache.block();
895
896        // 3. Optimistic sims + per-sim actual volatile read sets.
897        let mut optimistic = Vec::with_capacity(requests.len());
898        let mut read_sets: Vec<Vec<(Address, U256)>> = Vec::with_capacity(requests.len());
899        for req in &requests {
900            let mut overlay = EvmOverlay::new(Arc::clone(&snapshot), None);
901            let (result, access) = overlay.call_raw_with_access_list_with(
902                req.from,
903                req.to,
904                req.calldata.clone(),
905                &req.tx,
906            )?;
907            optimistic.push(result_to_sim(result, &access.to_eip2930()));
908
909            let volatile: Vec<(Address, U256)> = access
910                .slots
911                .iter()
912                .copied()
913                .filter(|(addr, slot)| self.registry.is_volatile(*addr, *slot, now))
914                .collect();
915            read_sets.push(volatile);
916        }
917
918        // 4. Predicted candidates (union of request access lists, volatile only).
919        let mut candidate_set: HashSet<(Address, U256)> = HashSet::new();
920        for req in &requests {
921            if let Some(al) = &req.tx.access_list {
922                for item in &al.0 {
923                    for key in &item.storage_keys {
924                        let slot = U256::from_be_bytes(key.0);
925                        if self.registry.is_volatile(item.address, slot, now) {
926                            candidate_set.insert((item.address, slot));
927                        }
928                    }
929                }
930            }
931        }
932        let candidates: Vec<(Address, U256)> = candidate_set.into_iter().collect();
933        let verify_set = {
934            let tracker = self.tracker.lock().unwrap_or_else(|e| e.into_inner());
935            self.policy.select(&candidates, &tracker, now)
936        };
937
938        // 5. Spawn the validator with Send-only data.
939        let registry = self.registry.clone();
940        let tracker = Arc::clone(&self.tracker);
941        let pending = Arc::clone(&self.pending);
942        let rerun_count = Arc::clone(&self.rerun_count);
943        let optimistic_for_task = optimistic.clone();
944        let cancelled = Arc::new(AtomicBool::new(false));
945        let cancelled_for_task = Arc::clone(&cancelled);
946        let validation = tokio::spawn(async move {
947            // Yield once before doing any work, so a prompt drop/into_optimistic
948            // can cancel before the validator is first polled. `run_validator` is
949            // otherwise synchronous, so cancellation past this point is
950            // cooperative: it observes the cancel flag at checkpoints.
951            tokio::task::yield_now().await;
952            run_validator(ValidatorInput {
953                snapshot,
954                fetcher,
955                requests,
956                read_sets,
957                registry,
958                tracker,
959                pending,
960                rerun_count,
961                now,
962                verify_set,
963                optimistic: optimistic_for_task,
964                cancelled: cancelled_for_task,
965                validation_block,
966            })
967        });
968
969        Ok(SpeculativeSim {
970            optimistic,
971            validation: Some(validation),
972            cancelled,
973        })
974    }
975}
976
977/// Owned inputs handed to the background validator (all `Send`).
978struct ValidatorInput {
979    snapshot: Arc<EvmSnapshot>,
980    fetcher: Option<StorageBatchFetchFn>,
981    requests: Vec<SimRequest>,
982    read_sets: Vec<Vec<(Address, U256)>>,
983    registry: FreshnessRegistry,
984    tracker: Arc<Mutex<SlotObservationTracker>>,
985    pending: Arc<Mutex<Vec<SlotChange>>>,
986    rerun_count: Arc<AtomicUsize>,
987    now: u64,
988    verify_set: Vec<(Address, U256)>,
989    optimistic: Vec<CallSimulationResult>,
990    cancelled: Arc<AtomicBool>,
991    /// Block the snapshot was built from; passed to the fetcher so the deferred
992    /// fetch reads the same block the snapshot represents.
993    validation_block: BlockId,
994}
995
996/// Maximum fixed-point iterations the background validator performs while a
997/// correction keeps expanding a sim's volatile read set. A backstop against
998/// pathological contracts that read an unbounded chain of new volatile slots;
999/// reaching it yields [`Validation::Unverified`] (the results have not reached a
1000/// verified fixed point, so they must not be trusted), logged via `tracing::warn!`.
1001const MAX_VALIDATION_ROUNDS: u32 = 8;
1002
1003/// Collect batch-fetcher results into a lookup map, requiring **every** requested
1004/// `(address, slot)` to be present and `Ok`.
1005///
1006/// The validator must never silently trust a gap: a fetch error *or* a slot the
1007/// fetcher omitted from its response yields `Err(reason)` (mapped to
1008/// [`Validation::Unverified`] by the caller) rather than defaulting the missing
1009/// value to zero — a custom fetcher that drops a slot would otherwise produce a
1010/// false confirmation or correction.
1011fn collect_fetch_results(
1012    requested: &[(Address, U256)],
1013    results: Vec<(Address, U256, anyhow::Result<U256>)>,
1014) -> Result<HashMap<(Address, U256), U256>, String> {
1015    let mut map: HashMap<(Address, U256), U256> = HashMap::new();
1016    for (addr, slot, value) in results {
1017        match value {
1018            Ok(v) => {
1019                map.insert((addr, slot), v);
1020            }
1021            Err(e) => return Err(format!("fetch failed for {addr}:{slot}: {e}")),
1022        }
1023    }
1024    for &key in requested {
1025        if !map.contains_key(&key) {
1026            return Err(format!(
1027                "fetcher omitted requested slot {}:{}",
1028                key.0, key.1
1029            ));
1030        }
1031    }
1032    Ok(map)
1033}
1034
1035/// The background validation routine. Touches only `Send` data — never the cache.
1036fn run_validator(input: ValidatorInput) -> Validation {
1037    let ValidatorInput {
1038        snapshot,
1039        fetcher,
1040        requests,
1041        read_sets,
1042        registry,
1043        tracker,
1044        pending,
1045        rerun_count,
1046        now,
1047        verify_set,
1048        optimistic,
1049        cancelled,
1050        validation_block,
1051    } = input;
1052
1053    // Checkpoint: cancelled before we even begin (the caller dropped or
1054    // `into_optimistic`d the handle while we were parked at the initial yield).
1055    if cancelled.load(Ordering::Relaxed) {
1056        return Validation::Confirmed;
1057    }
1058
1059    let Some(fetcher) = fetcher else {
1060        return Validation::Unverified {
1061            reason: "no storage batch fetcher available".to_string(),
1062        };
1063    };
1064
1065    // verify = policy-selected set ∪ each sim's actual volatile read set,
1066    // re-filtered through the registry clone so only currently-volatile slots
1067    // are checked (defensive: read sets and the policy selection are already
1068    // volatile-filtered on the main thread).
1069    let mut verify: HashSet<(Address, U256)> = verify_set.into_iter().collect();
1070    for set in &read_sets {
1071        verify.extend(set.iter().copied());
1072    }
1073    verify.retain(|(addr, slot)| registry.is_volatile(*addr, *slot, now));
1074    if verify.is_empty() {
1075        return Validation::Confirmed;
1076    }
1077    let verify: Vec<(Address, U256)> = verify.into_iter().collect();
1078
1079    // Checkpoint: cancelled before issuing the (costly, side-effecting) fetch.
1080    // This is what makes the "dropped before fetching" guarantee hold.
1081    if cancelled.load(Ordering::Relaxed) {
1082        return Validation::Confirmed;
1083    }
1084
1085    // Fetch fresh values. Any error OR any omitted slot → Unverified (never trust
1086    // silently: a missing result must not default to zero).
1087    let results = (fetcher)(verify.clone(), Some(validation_block));
1088    let fresh = match collect_fetch_results(&verify, results) {
1089        Ok(map) => map,
1090        Err(reason) => return Validation::Unverified { reason },
1091    };
1092
1093    // Checkpoint: cancelled after the fetch returned but before we record any
1094    // observations or queue a correction. A cancel seen here discards the
1095    // verdict's side effects entirely.
1096    if cancelled.load(Ordering::Relaxed) {
1097        return Validation::Confirmed;
1098    }
1099
1100    // Compare the initial verify set against the snapshot, observe each checked
1101    // slot, and seed the changed set (deduped by `(address, slot)`).
1102    let mut changed_map: HashMap<(Address, U256), SlotChange> = HashMap::new();
1103    let mut verified: HashSet<(Address, U256)> = verify.iter().copied().collect();
1104    {
1105        let mut tracker = tracker.lock().unwrap_or_else(|e| e.into_inner());
1106        for &(addr, slot) in &verify {
1107            // `collect_fetch_results` guarantees every requested slot is present.
1108            let new = fresh[&(addr, slot)];
1109            let old = snapshot.storage_value(addr, slot).unwrap_or(U256::ZERO);
1110            tracker.observe(addr, slot, new, now);
1111            if new != old {
1112                changed_map.insert(
1113                    (addr, slot),
1114                    SlotChange {
1115                        address: addr,
1116                        slot,
1117                        old,
1118                        new,
1119                    },
1120                );
1121            }
1122        }
1123    }
1124
1125    if changed_map.is_empty() {
1126        return Validation::Confirmed;
1127    }
1128
1129    // Re-run affected sims to a fixed point. A correction can flip control flow
1130    // so a re-run reads a *new* volatile slot the optimistic read set never
1131    // touched; that slot must itself be verified, or the "corrected" result
1132    // would still rest on stale snapshot state. Each round re-runs every sim
1133    // whose (possibly expanded) read set intersects a changed slot — applying
1134    // the full accumulated override set — collects newly-read volatile slots,
1135    // fetches and diffs them, and repeats until no new volatile slot appears,
1136    // none of the newly fetched slots differ, or the iteration cap is reached.
1137    let mut results = optimistic;
1138    // Per-sim current volatile read set; starts at the optimistic read set and
1139    // expands as corrections open new branches.
1140    let mut sim_reads = read_sets;
1141    let mut rerun_indices: HashSet<usize> = HashSet::new();
1142    let mut round: u32 = 0;
1143    loop {
1144        let changed_keys: HashSet<(Address, U256)> = changed_map.keys().copied().collect();
1145        let overrides: Vec<(Address, U256, U256)> = changed_map
1146            .values()
1147            .map(|c| (c.address, c.slot, c.new))
1148            .collect();
1149
1150        // Re-run sims whose current read set intersects a changed slot, applying
1151        // every accumulated override, and gather newly-read volatile candidates.
1152        let mut any_rerun = false;
1153        let mut new_candidates: HashSet<(Address, U256)> = HashSet::new();
1154        for (i, req) in requests.iter().enumerate() {
1155            if !sim_reads[i].iter().any(|k| changed_keys.contains(k)) {
1156                continue;
1157            }
1158            any_rerun = true;
1159            rerun_indices.insert(i);
1160            let mut overlay = EvmOverlay::new(Arc::clone(&snapshot), None);
1161            for &(addr, slot, value) in &overrides {
1162                overlay.override_slot(addr, slot, value);
1163            }
1164            // A host/transact error means the corrected re-run could not execute;
1165            // we must not keep the stale optimistic result and call it "Corrected".
1166            // (A revert/halt is `Ok(..)`, not an `Err`.) → Unverified.
1167            let (result, access) = match overlay.call_raw_with_access_list_with(
1168                req.from,
1169                req.to,
1170                req.calldata.clone(),
1171                &req.tx,
1172            ) {
1173                Ok(v) => v,
1174                Err(e) => {
1175                    return Validation::Unverified {
1176                        reason: format!("corrected re-run failed for request {i}: {e}"),
1177                    };
1178                }
1179            };
1180            results[i] = result_to_sim(result, &access.to_eip2930());
1181            let new_volatile: Vec<(Address, U256)> = access
1182                .slots
1183                .iter()
1184                .copied()
1185                .filter(|(a, s)| registry.is_volatile(*a, *s, now))
1186                .collect();
1187            for &key in &new_volatile {
1188                if !verified.contains(&key) {
1189                    new_candidates.insert(key);
1190                }
1191            }
1192            sim_reads[i] = new_volatile;
1193        }
1194
1195        // No sim read a changed slot (the change came from the predicted
1196        // candidate set, not an actual read), or no new volatile slot surfaced:
1197        // the current results already reflect every override, so we are done.
1198        if !any_rerun || new_candidates.is_empty() {
1199            break;
1200        }
1201        // The fixed point was not reached within the cap: corrections kept opening
1202        // new volatile slots. The results still rest on un-verified state, so we
1203        // must NOT return a trusted `Corrected`. Return `Unverified` without
1204        // queuing any pending corrections (matching the fetch-error paths); the
1205        // still-volatile slots are re-discovered and re-fetched on the next run.
1206        if round >= MAX_VALIDATION_ROUNDS {
1207            tracing::warn!(
1208                rounds = round,
1209                "freshness validator exceeded fixed-point round cap; returning Unverified"
1210            );
1211            return Validation::Unverified {
1212                reason: format!(
1213                    "freshness validation exceeded fixed-point round cap ({MAX_VALIDATION_ROUNDS})"
1214                ),
1215            };
1216        }
1217
1218        // Checkpoint: cancelled mid-loop. Results so far reflect the applied
1219        // overrides; do not fetch further or queue corrections.
1220        if cancelled.load(Ordering::Relaxed) {
1221            return Validation::Confirmed;
1222        }
1223
1224        // Fetch the newly-discovered candidates; any error OR omitted slot →
1225        // Unverified (a missing result must not default to zero).
1226        let new_vec: Vec<(Address, U256)> = new_candidates.into_iter().collect();
1227        let fetched = (fetcher)(new_vec.clone(), Some(validation_block));
1228        let new_fresh = match collect_fetch_results(&new_vec, fetched) {
1229            Ok(map) => map,
1230            Err(reason) => return Validation::Unverified { reason },
1231        };
1232
1233        // Diff + observe the newly fetched slots, growing the changed set.
1234        let mut grew = false;
1235        {
1236            let mut tracker = tracker.lock().unwrap_or_else(|e| e.into_inner());
1237            for &(addr, slot) in &new_vec {
1238                verified.insert((addr, slot));
1239                // `collect_fetch_results` guarantees every requested slot is present.
1240                let new = new_fresh[&(addr, slot)];
1241                let old = snapshot.storage_value(addr, slot).unwrap_or(U256::ZERO);
1242                tracker.observe(addr, slot, new, now);
1243                if new != old {
1244                    changed_map.insert(
1245                        (addr, slot),
1246                        SlotChange {
1247                            address: addr,
1248                            slot,
1249                            old,
1250                            new,
1251                        },
1252                    );
1253                    grew = true;
1254                }
1255            }
1256        }
1257
1258        // The newly fetched slots were all unchanged → another round would not
1259        // alter any result; current results are final.
1260        if !grew {
1261            break;
1262        }
1263        round += 1;
1264    }
1265
1266    // Count distinct affected sims once: a sim re-run across multiple rounds is
1267    // still one affected sim, preserving the "once per re-executed sim" contract.
1268    rerun_count.fetch_add(rerun_indices.len(), Ordering::Relaxed);
1269
1270    // Queue every accumulated correction for flow-back into the cache next run.
1271    let changed: Vec<SlotChange> = changed_map.into_values().collect();
1272    {
1273        let mut pending = pending.lock().unwrap_or_else(|e| e.into_inner());
1274        pending.extend(changed.iter().cloned());
1275    }
1276
1277    Validation::Corrected { results, changed }
1278}
1279
1280/// Build a [`CallSimulationResult`] from a non-committing execution result and
1281/// its captured access list. `token_deltas` is empty (the optimistic path does
1282/// not run transfer tracking); gas, logs, and return data come from the
1283/// execution result. `status` records whether the call succeeded, reverted, or
1284/// halted; `output` carries the `Success`/`Revert` payload (empty on `Halt`),
1285/// so a corrected view-call's new return value is observable here.
1286fn result_to_sim(result: ExecutionResult, access_list: &AccessList) -> CallSimulationResult {
1287    let (status, gas_used, logs, output) = match result {
1288        ExecutionResult::Success {
1289            gas_used,
1290            logs,
1291            output,
1292            ..
1293        } => (SimStatus::Success, gas_used, logs, output.into_data()),
1294        ExecutionResult::Revert { gas_used, output } => {
1295            (SimStatus::Revert, gas_used, Vec::new(), output)
1296        }
1297        ExecutionResult::Halt { gas_used, reason } => (
1298            SimStatus::Halt {
1299                reason: format!("{reason:?}"),
1300            },
1301            gas_used,
1302            Vec::new(),
1303            Bytes::new(),
1304        ),
1305    };
1306    CallSimulationResult {
1307        status,
1308        gas_used,
1309        token_deltas: HashMap::new(),
1310        logs,
1311        access_list: access_list.clone(),
1312        output,
1313    }
1314}
1315
1316#[cfg(test)]
1317mod tests {
1318    use super::*;
1319
1320    fn addr(n: u8) -> Address {
1321        Address::repeat_byte(n)
1322    }
1323
1324    // --- Classification ----------------------------------------------------
1325
1326    #[test]
1327    fn registry_default_is_volatile() {
1328        let reg = FreshnessRegistry::new();
1329        assert_eq!(reg.default_validity(), Validity::Volatile);
1330        assert_eq!(reg.validity(addr(1), U256::from(0)), Validity::Volatile);
1331    }
1332
1333    #[test]
1334    fn registry_with_default_overrides_unclassified() {
1335        let reg = FreshnessRegistry::with_default(Validity::Pinned);
1336        assert_eq!(reg.validity(addr(1), U256::from(0)), Validity::Pinned);
1337        assert!(!reg.is_volatile(addr(1), U256::from(0), 100));
1338    }
1339
1340    #[test]
1341    fn registry_resolution_order_slot_account_default() {
1342        let a = addr(1);
1343        let mut reg = FreshnessRegistry::new(); // default Volatile
1344        reg.pin(a); // account-level Pinned
1345        reg.mark_volatile_slot(a, U256::from(7)); // slot-level Volatile
1346
1347        // slot-level wins over account-level
1348        assert_eq!(reg.validity(a, U256::from(7)), Validity::Volatile);
1349        // account-level wins over default for a non-overridden slot
1350        assert_eq!(reg.validity(a, U256::from(8)), Validity::Pinned);
1351        // default for an unrelated account
1352        assert_eq!(reg.validity(addr(2), U256::from(7)), Validity::Volatile);
1353    }
1354
1355    #[test]
1356    fn is_volatile_per_variant() {
1357        let a = addr(1);
1358        let mut reg = FreshnessRegistry::new();
1359        reg.pin_slot(a, U256::from(1));
1360        reg.mark_volatile_slot(a, U256::from(2));
1361        reg.valid_through_slot(a, U256::from(3), 50);
1362
1363        assert!(!reg.is_volatile(a, U256::from(1), 100)); // Pinned
1364        assert!(reg.is_volatile(a, U256::from(2), 100)); // Volatile
1365    }
1366
1367    #[test]
1368    fn valid_through_boundary() {
1369        let a = addr(1);
1370        let slot = U256::from(3);
1371        let mut reg = FreshnessRegistry::new();
1372        reg.valid_through_slot(a, slot, 50);
1373
1374        assert!(!reg.is_volatile(a, slot, 49)); // before
1375        assert!(!reg.is_volatile(a, slot, 50)); // at boundary: still valid (now == m)
1376        assert!(reg.is_volatile(a, slot, 51)); // after: now > m
1377    }
1378
1379    #[test]
1380    fn registry_is_clone() {
1381        let mut reg = FreshnessRegistry::new();
1382        reg.pin(addr(1));
1383        let clone = reg.clone();
1384        assert_eq!(clone.validity(addr(1), U256::from(0)), Validity::Pinned);
1385    }
1386
1387    // --- Clock -------------------------------------------------------------
1388
1389    #[test]
1390    fn block_clock_default_and_set() {
1391        let clock = BlockClock::new();
1392        assert_eq!(clock.now(), 0);
1393        clock.set_block(123);
1394        assert_eq!(clock.now(), 123);
1395    }
1396
1397    #[test]
1398    fn block_clock_clone_shares_counter() {
1399        let clock = BlockClock::at(10);
1400        let clone = clock.clone();
1401        clock.set_block(42);
1402        // The clone observes the update through the shared Arc.
1403        assert_eq!(clone.now(), 42);
1404    }
1405
1406    #[test]
1407    fn wall_clock_is_unix_seconds() {
1408        let now = WallClock.now();
1409        // Sanity: after 2021-01-01.
1410        assert!(now > 1_600_000_000);
1411    }
1412
1413    // --- Policy ------------------------------------------------------------
1414
1415    #[test]
1416    fn always_verify_selects_all() {
1417        let obs = SlotObservationTracker::new();
1418        let candidates = [(addr(1), U256::from(0)), (addr(2), U256::from(1))];
1419        let mut policy = AlwaysVerify;
1420        assert_eq!(policy.select(&candidates, &obs, 0), candidates.to_vec());
1421    }
1422
1423    #[test]
1424    fn never_verify_selects_none() {
1425        let obs = SlotObservationTracker::new();
1426        let candidates = [(addr(1), U256::from(0))];
1427        let mut policy = NeverVerify;
1428        assert!(policy.select(&candidates, &obs, 0).is_empty());
1429    }
1430
1431    #[test]
1432    fn observation_driven_selects_only_should_refetch() {
1433        let mut obs = SlotObservationTracker::new();
1434        let params = FreshnessParams::default();
1435        let stable = (addr(1), U256::from(0));
1436        let unknown = (addr(2), U256::from(0));
1437
1438        // Build a stable (never-changed) slot with enough observations so
1439        // `should_refetch` returns false for it.
1440        for now in 0..params.min_observations {
1441            obs.observe(stable.0, stable.1, U256::from(42), now as u64);
1442        }
1443        let now = params.min_observations as u64 - 1;
1444        assert!(!obs.should_refetch(stable.0, stable.1, now, &params));
1445        assert!(obs.should_refetch(unknown.0, unknown.1, now, &params));
1446
1447        let mut policy = ObservationDriven::new(params);
1448        let selected = policy.select(&[stable, unknown], &obs, now);
1449        assert_eq!(selected, vec![unknown]);
1450    }
1451}