Phase 3: state-update primitives (Pillar B.1) + relative read-modify-write

Cargo.toml

@@ -38,6 +38,7 @@ alloy-eips = "1.0.38"
 alloy-network = "1.0.38"
 alloy-primitives = { version = "1.4", features = ["map"] }
 alloy-provider = "1.0.38"
+alloy-rlp = "0.3"
 alloy-rpc-client = "1.0.38"
 alloy-rpc-types-eth = "1.0.38"
 alloy-sol-types = "1.4"
@@ -86,6 +87,14 @@ harness = false
 name = "freshness"
 harness = false
 
+[[bench]]
+name = "state_update"
+harness = false
+
+[[bench]]
+name = "event_pipeline"
+harness = false
+
 # RPC-gated real-contract benchmarks. Skipped (not failed) when RPC_URL is unset,
 # so `cargo bench` stays offline by default.
 [[bench]]

README.md

@@ -168,6 +168,8 @@ and inject all state directly:
 | `prefetch_registry` | Advanced | Record and persist storage touch sets for cross-cycle prefetch. |
 | `freshness_optimistic` | Advanced | Optimistic verify-and-rerun loop: a `Corrected` validation via a stub fetcher. |
 | `freshness_multi_sim` | Advanced | Many sims with selective re-run, plus classification and `ValidThrough` aging. |
+| `state_update_apply` | Advanced | Apply a mixed `StateUpdate` batch (`Slot`/`Account`/`Purge`) and inspect the returned `StateDiff`. |
+| `reactive_cache` | Advanced | Decode logs (ERC-20 `Transfer` + UniswapV3 `Swap`) into `StateUpdate`s, ingest a block, reconcile drift, and purge on a reorg. |
 
 **RPC examples** fork real mainnet state. Set `RPC_URL` to an Ethereum RPC
 endpoint (they print instructions and exit if it is unset):
@@ -215,15 +217,17 @@ println!("installed {} bytes at {}", etched.code_size, etched.target_address);
 
 ## Benchmarks
 
-Criterion benchmarks live in [`benches/`](benches). The offline benches are the
-baseline against which the planned copy-on-write snapshot rewrite (roadmap
-Pillar A) will be measured, so they exercise the real hot paths at a range of
-cache sizes:
+Criterion benchmarks live in [`benches/`](benches). The offline benches exercise
+the current hot paths at a range of cache sizes, including the Phase 5
+copy-on-write snapshot implementation and retained deep-clone baselines where
+useful for A/B comparison:
 
 | Bench | Measures |
 | --- | --- |
 | `simulation` | `create_snapshot` across cache sizes (100 → 10k accounts), overlay fan-out, `call_raw` throughput, sequential bundle execution, batched storage injection. |
 | `freshness` | The optimistic loop end-to-end (CPU and latency-hiding), `verify_slots` at scale (1 → 1000 slots), and multi-sim fan-out. |
+| `state_update` | `apply_updates` throughput across batch sizes (1 → 1000 `Slot`s) and per-variant apply cost (`Slot` vs `Account` vs `Purge`). |
+| `event_pipeline` | Per-event decode cost (ERC-20 `Transfer`, V3 `Swap`/`Mint`), `ingest_logs` decode+apply throughput (1 → 1000 logs), and `reorg_to` purge cost. |
 | `access_list` | Touch-set merge and EIP-2930 list construction. |
 | `revert_decoding` | Built-in and custom revert decoding, including decoder dispatch with many registered errors. |
 | `storage_keys` | Mapping/array storage-key derivation. |

benches/event_pipeline.rs

@@ -0,0 +1,240 @@
+//! Phase 4 benchmarks: the event → state pipeline (Pillar B.2).
+//!
+//! Measures three things, all offline (mocked provider, in-memory logs):
+//! - **decode** cost per event kind (ERC-20 `Transfer`, UniswapV3 `Swap`/`Mint`),
+//!   isolating the pure `EventDecoder::decode` work (no apply);
+//! - **ingest** throughput — [`EventPipeline::ingest_logs`] decoding **and**
+//!   applying a block of logs, across batch sizes (1 → 1000);
+//! - **reorg** purge cost — [`EventPipeline::reorg_to`] over a touched set of
+//!   1 → 1000 addresses.
+//!
+//! A current-thread runtime drives only the async cache constructor; the pipeline
+//! itself is synchronous and never touches the network.
+
+use std::collections::HashMap;
+use std::hint::black_box;
+use std::sync::Arc;
+
+use alloy_primitives::aliases::{I24, U160};
+use alloy_primitives::{Address, Bytes, I256, Log, U256, hex, keccak256};
+use alloy_provider::RootProvider;
+use alloy_provider::network::AnyNetwork;
+use alloy_rpc_client::RpcClient;
+use alloy_sol_types::{SolEvent, sol};
+use alloy_transport::mock::Asserter;
+use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main};
+use evm_fork_cache::cache::{EvmCache, V3_SLOT0_SLOT, v3_tick_info_storage_keys_with_base};
+use evm_fork_cache::events::{DecoderRegistry, EventDecoder, EventPipeline, StateView};
+use evm_fork_cache::{Erc20TransferDecoder, StateUpdate, UniswapV3Decoder, UniswapV3Layout};
+use revm::state::{AccountInfo, Bytecode};
+use tokio::runtime::{Builder, Runtime};
+
+const MOCK_ERC20_RUNTIME_HEX: &str = include_str!("../fixtures/mock_erc20_runtime.hex");
+const TOKEN: Address = Address::repeat_byte(0xAA);
+const POOL: Address = Address::repeat_byte(0xBB);
+
+fn current_thread_rt() -> Runtime {
+    Builder::new_current_thread().enable_all().build().unwrap()
+}
+
+/// A cache with `TOKEN` and `POOL` installed as storage-cleared accounts (so
+/// unseeded slots read as zero — no RPC fallthrough).
+fn seeded_cache(rt: &Runtime) -> EvmCache {
+    let provider = RootProvider::<AnyNetwork>::new(RpcClient::mocked(Asserter::new()));
+    let mut cache = rt.block_on(EvmCache::new(Arc::new(provider), None));
+    let runtime = Bytecode::new_raw(Bytes::from(
+        hex::decode(MOCK_ERC20_RUNTIME_HEX.trim()).unwrap(),
+    ));
+    let code_hash = runtime.hash_slow();
+    for addr in [TOKEN, POOL] {
+        cache.db_mut().insert_account_info(
+            addr,
+            AccountInfo {
+                balance: U256::ZERO,
+                nonce: 0,
+                code: Some(runtime.clone()),
+                code_hash,
+                account_id: None,
+            },
+        );
+        cache
+            .db_mut()
+            .replace_account_storage(addr, Default::default())
+            .unwrap();
+    }
+    cache
+}
+
+sol! {
+    event Swap(address indexed sender, address indexed recipient, int256 amount0, int256 amount1, uint160 sqrtPriceX96, uint128 liquidity, int24 tick);
+    event Mint(address sender, address indexed owner, int24 indexed tickLower, int24 indexed tickUpper, uint128 amount, uint256 amount0, uint256 amount1);
+}
+
+fn transfer_log(token: Address, from: Address, to: Address, value: U256) -> Log {
+    let sig = keccak256(b"Transfer(address,address,uint256)");
+    Log::new_unchecked(
+        token,
+        vec![sig, from.into_word(), to.into_word()],
+        Bytes::copy_from_slice(&value.to_be_bytes::<32>()),
+    )
+}
+
+fn swap_log(pool: Address, sqrt_price: u128, liquidity: u128, tick: i32) -> Log {
+    let ev = Swap {
+        sender: Address::repeat_byte(0x01),
+        recipient: Address::repeat_byte(0x02),
+        amount0: I256::try_from(-1i64).unwrap(),
+        amount1: I256::try_from(1i64).unwrap(),
+        sqrtPriceX96: U160::from(sqrt_price),
+        liquidity,
+        tick: I24::try_from(tick).unwrap(),
+    };
+    Log {
+        address: pool,
+        data: ev.encode_log_data(),
+    }
+}
+
+fn mint_log(pool: Address, lower: i32, upper: i32, amount: u128) -> Log {
+    let ev = Mint {
+        sender: Address::repeat_byte(0x03),
+        owner: Address::repeat_byte(0x04),
+        tickLower: I24::try_from(lower).unwrap(),
+        tickUpper: I24::try_from(upper).unwrap(),
+        amount,
+        amount0: U256::from(1),
+        amount1: U256::from(1),
+    };
+    Log {
+        address: pool,
+        data: ev.encode_log_data(),
+    }
+}
+
+/// A bench-local read-only [`StateView`] over a fixed map (for the V3 `Mint`
+/// decode, which reads the current tick word).
+struct MapView(HashMap<(Address, U256), U256>);
+impl StateView for MapView {
+    fn storage(&self, address: Address, slot: U256) -> Option<U256> {
+        self.0.get(&(address, slot)).copied()
+    }
+}
+
+/// A bench-local decoder that emits one absolute `Slot` write per log, keyed by
+/// the log's address — so repeated ingest is idempotent (stable across iters).
+struct AbsDecoder;
+impl EventDecoder for AbsDecoder {
+    fn decode(&self, log: &Log, _view: &dyn StateView) -> Vec<StateUpdate> {
+        vec![StateUpdate::slot(log.address, U256::from(0), U256::from(1))]
+    }
+}
+
+/// Pure `decode` cost per event kind (no apply).
+fn bench_decode(c: &mut Criterion) {
+    let mut group = c.benchmark_group("decode");
+
+    let erc20 = Erc20TransferDecoder::new(U256::from(3));
+    let tlog = transfer_log(
+        TOKEN,
+        Address::repeat_byte(0x21),
+        Address::repeat_byte(0x22),
+        U256::from(100),
+    );
+    let empty = MapView(HashMap::new());
+    group.bench_function("erc20_transfer", |b| {
+        b.iter(|| black_box(erc20.decode(black_box(&tlog), &empty)))
+    });
+
+    let v3 = UniswapV3Decoder::new().with_pool(POOL, UniswapV3Layout::uniswap(60));
+    let slog = swap_log(POOL, 2_000_000, 7_500, 120);
+    let mut slot0_view = HashMap::new();
+    slot0_view.insert(
+        (POOL, V3_SLOT0_SLOT),
+        (U256::from(1u64) << 240) | U256::from(1_000_000u64),
+    );
+    // Seed the tick words the Mint reads (lower/upper) so it computes (not skips).
+    let lo = v3_tick_info_storage_keys_with_base(60, evm_fork_cache::cache::V3_TICKS_BASE_SLOT)[0];
+    let hi = v3_tick_info_storage_keys_with_base(120, evm_fork_cache::cache::V3_TICKS_BASE_SLOT)[0];
+    slot0_view.insert((POOL, lo), U256::ZERO);
+    slot0_view.insert((POOL, hi), U256::ZERO);
+    let view = MapView(slot0_view);
+    group.bench_function("v3_swap", |b| {
+        b.iter(|| black_box(v3.decode(black_box(&slog), &view)))
+    });
+    let mlog = mint_log(POOL, 60, 120, 1_000);
+    group.bench_function("v3_mint", |b| {
+        b.iter(|| black_box(v3.decode(black_box(&mlog), &view)))
+    });
+
+    group.finish();
+}
+
+/// `ingest_logs` decode+apply throughput as the per-block log batch grows.
+fn bench_ingest_batch(c: &mut Criterion) {
+    let rt = current_thread_rt();
+    let mut cache = seeded_cache(&rt);
+
+    let mut group = c.benchmark_group("ingest_logs");
+    for &n in &[1usize, 10, 100, 1_000] {
+        let logs: Vec<Log> = (0..n)
+            .map(|i| {
+                Log::new_unchecked(
+                    Address::repeat_byte((i % 251 + 1) as u8),
+                    vec![],
+                    Bytes::new(),
+                )
+            })
+            .collect();
+        let mut registry = DecoderRegistry::new();
+        registry.register(Arc::new(AbsDecoder));
+        let mut pipeline = EventPipeline::new(registry);
+
+        group.throughput(Throughput::Elements(n as u64));
+        group.bench_with_input(BenchmarkId::from_parameter(n), &logs, |b, logs| {
+            let mut block = 0u64;
+            b.iter(|| {
+                block += 1;
+                black_box(pipeline.ingest_logs(&mut cache, block, black_box(logs)))
+            })
+        });
+    }
+    group.finish();
+}
+
+/// `reorg_to` purge cost over a touched set of N distinct addresses.
+fn bench_reorg(c: &mut Criterion) {
+    let rt = current_thread_rt();
+
+    let mut group = c.benchmark_group("reorg_to");
+    for &n in &[10usize, 100, 1_000] {
+        group.throughput(Throughput::Elements(n as u64));
+        group.bench_with_input(BenchmarkId::from_parameter(n), &n, |b, &n| {
+            b.iter_batched(
+                || {
+                    // Setup: a cache + pipeline with N addresses touched at block 1.
+                    let mut cache = seeded_cache(&rt);
+                    let mut registry = DecoderRegistry::new();
+                    registry.register(Arc::new(AbsDecoder));
+                    let mut pipeline = EventPipeline::new(registry);
+                    let logs: Vec<Log> = (0..n)
+                        .map(|i| {
+                            let mut bytes = [0u8; 20];
+                            bytes[0..8].copy_from_slice(&(i as u64).to_be_bytes());
+                            Log::new_unchecked(Address::from(bytes), vec![], Bytes::new())
+                        })
+                        .collect();
+                    pipeline.ingest_logs(&mut cache, 1, &logs);
+                    (pipeline, cache)
+                },
+                |(mut pipeline, mut cache)| {
+                    black_box(pipeline.reorg_to(&mut cache, 0));
+                },
+                criterion::BatchSize::SmallInput,
+            )
+        });
+    }
+    group.finish();
+}
+
+criterion_group!(benches, bench_decode, bench_ingest_batch, bench_reorg);
+criterion_main!(benches);