diff --git a/tests/common/mod.rs b/tests/common/mod.rs index 509a1f5..196f3d3 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -6,7 +6,7 @@ #![allow(dead_code)] use std::collections::HashMap; -use std::sync::Arc; +use std::sync::{Arc, Condvar, Mutex}; use alloy_eips::BlockId; use alloy_primitives::{Address, Bytes, U256, hex}; @@ -128,19 +128,64 @@ pub fn failing_fetcher() -> StorageBatchFetchFn { }) } -/// Build a stub [`StorageBatchFetchFn`] that reports chosen values *and* flips a -/// shared flag the first time it is called. +/// A one-shot synchronous gate: a holder blocks in [`wait`](Gate::wait) until +/// some other thread calls [`release`](Gate::release). Cloning shares the same +/// underlying state, and `release` is sticky — once released, every present and +/// future `wait` returns immediately. /// -/// Used by the Drop-abort test to prove the background validator was cancelled -/// before it ever fetched (so it could not have queued a correction). The -/// returned values otherwise behave exactly like [`stub_fetcher`]. -pub fn tracking_fetcher( +/// Used by the Drop-abort test to make the background validator's fetch +/// deterministically ordered *after* the drop. The fetcher (running on a worker +/// thread) cannot return — and therefore the validator cannot reach its +/// post-fetch checkpoint — until the test has dropped the `SpeculativeSim` and +/// released the gate, eliminating the spawn/poll race regardless of how the +/// multi-thread scheduler interleaves the two threads. +/// +/// Built on a `Mutex` + `Condvar` so the whole thing is `Send + Sync`, +/// which a [`StorageBatchFetchFn`] closure must be. +#[derive(Clone, Default)] +pub struct Gate { + inner: Arc<(Mutex, Condvar)>, +} + +impl Gate { + pub fn new() -> Self { + Self::default() + } + + /// Block until [`release`](Gate::release) has been called (returns + /// immediately if it already has). + pub fn wait(&self) { + let (lock, cv) = &*self.inner; + let mut released = lock.lock().unwrap_or_else(|e| e.into_inner()); + while !*released { + released = cv.wait(released).unwrap_or_else(|e| e.into_inner()); + } + } + + /// Wake any current waiter and let all future waiters pass. + pub fn release(&self) { + let (lock, cv) = &*self.inner; + *lock.lock().unwrap_or_else(|e| e.into_inner()) = true; + cv.notify_all(); + } +} + +/// Build a stub [`StorageBatchFetchFn`] that reports chosen values but blocks on +/// `gate` before returning. +/// +/// Used by the Drop-abort test: by releasing the gate only *after* dropping the +/// `SpeculativeSim`, the test guarantees the validator's fetch completes (and so +/// its post-fetch, correction-queuing checkpoint runs) strictly after the +/// cancel flag is set — so the dropped speculation can never queue a correction, +/// no matter how the scheduler races the two threads. The returned values +/// otherwise behave exactly like [`stub_fetcher`]. +pub fn gated_tracking_fetcher( values: HashMap<(Address, U256), U256>, - called: Arc, + gate: Gate, ) -> StorageBatchFetchFn { Arc::new( move |requests: Vec<(Address, U256)>, _block: Option| { - called.store(true, std::sync::atomic::Ordering::SeqCst); + gate.wait(); requests .into_iter() .map(|(addr, slot)| { diff --git a/tests/freshness.rs b/tests/freshness.rs index 353f827..73607d0 100644 --- a/tests/freshness.rs +++ b/tests/freshness.rs @@ -16,8 +16,8 @@ use alloy_sol_types::SolCall; use anyhow::Result; use common::{ - MOCK_ERC20_BALANCE_SLOT, MockERC20, failing_fetcher, install_default_account, - install_mock_erc20, panicking_fetcher, setup_cache, stub_fetcher, tracking_fetcher, + Gate, MOCK_ERC20_BALANCE_SLOT, MockERC20, failing_fetcher, gated_tracking_fetcher, + install_default_account, install_mock_erc20, panicking_fetcher, setup_cache, stub_fetcher, }; use evm_fork_cache::cache::{ EvmCache, EvmOverlay, SimStatus, SlotObservationTracker, StorageBatchFetchFn, @@ -1224,9 +1224,20 @@ async fn run_into_optimistic_aborts_validation() -> Result<()> { // T3 (part 1): dropping the SpeculativeSim (no validate/into_optimistic) aborts // the validation task before it can push a correction. The fetcher reports a -// CHANGED value and flips a "called" flag; after the drop + settle we assert the -// pending queue is empty (and, robustly, that the fetcher was never even -// reached) — proving the abort beat the push. +// CHANGED value, so an *uncancelled* validator would queue a correction and bump +// the re-run count; we assert neither happens after the drop. +// +// Determinism: the validator's only correction-queuing path runs *after* its +// fetch returns (the post-fetch cancel checkpoint in `run_validator` gates it). +// We make that ordering race-free with a gate the test controls — the fetcher +// blocks until `gate.release()`, and we release only *after* `drop(sim)` has set +// the cancel flag. So however the multi-thread scheduler interleaves the spawned +// task and this thread, the fetch (and thus the post-fetch checkpoint) can only +// complete once cancellation is already observable, and the correction is +// suppressed. We deliberately do NOT assert the fetcher was never reached: the +// product only guarantees a cancel seen at a checkpoint suppresses side effects, +// not that an in-flight fetch is skipped — asserting the latter was the original +// over-strict, racy condition. #[tokio::test(flavor = "multi_thread")] async fn dropping_speculative_sim_aborts_before_queueing_correction() -> Result<()> { let token = Address::repeat_byte(0x44); @@ -1234,10 +1245,10 @@ async fn dropping_speculative_sim_aborts_before_queueing_correction() -> Result< let recipient = Address::repeat_byte(0x66); let mut cache = cache_with_balance(token, owner, U256::from(1000)).await?; - let called = Arc::new(std::sync::atomic::AtomicBool::new(false)); - cache.set_storage_batch_fetcher(tracking_fetcher( + let gate = Gate::new(); + cache.set_storage_batch_fetcher(gated_tracking_fetcher( HashMap::from([((token, balance_slot_for(owner)), U256::from(50))]), - Arc::clone(&called), + gate.clone(), )); let mut controller = FreshnessController::new(FreshnessRegistry::new(), AlwaysVerify); @@ -1249,9 +1260,12 @@ async fn dropping_speculative_sim_aborts_before_queueing_correction() -> Result< transfer_calldata(recipient, U256::from(100)), )], )?; - // Drop immediately, with NO intervening await, so the abort flag is set - // before the spawned task is ever polled. + // Drop with NO intervening await, then release the gate. Releasing only after + // the drop guarantees the validator's fetch (if it even reaches it) returns + // strictly after the cancel flag is set, so its post-fetch checkpoint bails + // out before queuing anything. drop(sim); + gate.release(); settle().await; @@ -1260,10 +1274,6 @@ async fn dropping_speculative_sim_aborts_before_queueing_correction() -> Result< 0, "dropping the sim must abort validation before it queues a correction" ); - assert!( - !called.load(std::sync::atomic::Ordering::SeqCst), - "the aborted validator should never have reached the fetcher" - ); assert_eq!(controller.rerun_count(), 0, "no re-run after abort"); Ok(()) }