diff --git a/Cargo.lock b/Cargo.lock index d1c239b017..2f78a60c06 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9250,6 +9250,7 @@ dependencies = [ "prometheus", "prometheus-metric-storage", "reqwest 0.13.4", + "rustls", "serde", "serde-ext", "serde_json", diff --git a/Cargo.toml b/Cargo.toml index 65a3ea7667..e4b5e4958e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -111,6 +111,7 @@ reqwest = { version = "0.13.4", features = ["hickory-dns"] } rstest = "0.26" ruint = { version = "1.17.2", default-features = false } rust_decimal = { version = "1.35.0", default-features = false } +rustls = { version = "0.23.41", default-features = false } s3 = { path = "crates/s3" } schemars = { version = "1.2", features = ["chrono04"] } scopeguard = "1.2.0" diff --git a/crates/driver/src/run.rs b/crates/driver/src/run.rs index 87ba8c2334..dd152c47fb 100644 --- a/crates/driver/src/run.rs +++ b/crates/driver/src/run.rs @@ -162,7 +162,6 @@ fn simulator( eth: simulator::Ethereum, http_factory: &HttpClientFactory, ) -> Simulator { - let block_stream = eth.current_block().clone(); let mut simulator = match &config.simulator { configs::simulator::Config { kind: configs::simulator::SimulatorKind::Tenderly(config), @@ -181,8 +180,7 @@ fn simulator( simulator.disable_gas(gas); } if let Some(cfg) = &config.simulator.state_override_stream { - simulator - .set_simulation_overrides(simulator::state_override_stream::spawn(cfg, block_stream)); + simulator.set_simulation_overrides(simulator::state_override_stream::spawn(cfg)); } simulator diff --git a/crates/e2e/tests/e2e/state_override.rs b/crates/e2e/tests/e2e/state_override.rs index 23dff5c7e0..0797595d82 100644 --- a/crates/e2e/tests/e2e/state_override.rs +++ b/crates/e2e/tests/e2e/state_override.rs @@ -1,5 +1,6 @@ use { alloy::{ + eips::BlockId, primitives::{Address, B256, U256}, rpc::types::state::{AccountOverride, StateOverride}, sol, @@ -54,14 +55,20 @@ async fn estimate_gas_state_override(web3: Web3) { let eth = ethereum(web3).await; let gated_call = gated_contract.gate().into_transaction_request(); - let without = eth.estimate_gas(gated_call.clone(), None).await; + let without = eth + .estimate_gas(gated_call.clone(), None, BlockId::pending()) + .await; assert!( without.is_err(), "gate() should revert without the slot-0 override, got {without:?}", ); let with = eth - .estimate_gas(gated_call, Some(unlock_override(*gated_contract.address()))) + .estimate_gas( + gated_call, + Some(unlock_override(*gated_contract.address())), + BlockId::pending(), + ) .await; assert!( with.is_ok(), diff --git a/crates/price-estimation/src/factory.rs b/crates/price-estimation/src/factory.rs index 2a32db5645..188d6a0d4d 100644 --- a/crates/price-estimation/src/factory.rs +++ b/crates/price-estimation/src/factory.rs @@ -122,7 +122,7 @@ impl<'a> PriceEstimatorFactory<'a> { let simulation_overrides = args .state_override_stream .as_ref() - .map(|cfg| simulator::state_override_stream::spawn(cfg, network.block_stream.clone())); + .map(simulator::state_override_stream::spawn); let simulator = SettlementSimulator::new( settlement_contract, network.flash_loan_router, diff --git a/crates/simulator/Cargo.toml b/crates/simulator/Cargo.toml index bb5185bf9b..fbd111d2ec 100644 --- a/crates/simulator/Cargo.toml +++ b/crates/simulator/Cargo.toml @@ -48,6 +48,7 @@ url = { workspace = true } [dev-dependencies] mockall = { workspace = true } +rustls = { workspace = true, features = ["aws-lc-rs"] } testlib = { workspace = true } [features] diff --git a/crates/simulator/src/encoding.rs b/crates/simulator/src/encoding.rs index d515e25df5..2ce85d1d45 100644 --- a/crates/simulator/src/encoding.rs +++ b/crates/simulator/src/encoding.rs @@ -348,9 +348,12 @@ pub(crate) async fn finish_simulation_builder( return Err(BuildError::NoOrder); } - let block = match builder.block { - Block::Latest => builder.simulator.0.current_block.borrow().number, - Block::Number(n) => n, + let (block, block_timestamp) = match builder.block { + Block::Latest => { + let block = builder.simulator.0.current_block.borrow(); + (block.number, Some(block.timestamp)) + } + Block::Number(n) => (n, None), }; let executed_amounts = futures::future::try_join_all( @@ -498,10 +501,11 @@ pub(crate) async fn finish_simulation_builder( // propAMM state when simulated on historic blocks is slightly different // from what you would have gotten if you actually traded with the propAMM // on that block but we can't do anything about it and most likely the - // price is close enough to not affect the simulation. - if matches!(builder.block, Block::Latest) + // price is close enough to not affect the simulation. Only the tip has a + // known timestamp, which is what gates the overrides to it here. + if let Some(block_timestamp) = block_timestamp && let Some(stream) = builder.simulator.0.simulation_overrides.as_ref() - && let Some(state_overrides) = stream.current() + && let Some(state_overrides) = stream.overrides_for(block, block_timestamp) { for (account, state) in state_overrides { builder diff --git a/crates/simulator/src/ethereum/mod.rs b/crates/simulator/src/ethereum/mod.rs index 23b5b24b20..3a8b4d4f69 100644 --- a/crates/simulator/src/ethereum/mod.rs +++ b/crates/simulator/src/ethereum/mod.rs @@ -1,5 +1,6 @@ use { crate::ethereum::contracts::Contracts, + alloy_eips::BlockId, alloy_primitives::U256, alloy_provider::{Provider, network::TransactionBuilder}, alloy_rpc_types::{TransactionRequest, state::StateOverride}, @@ -124,6 +125,7 @@ impl Ethereum { &self, tx: T, overrides: Option, + block: BlockId, ) -> Result where T: Into, @@ -139,7 +141,7 @@ impl Ethereum { .provider .estimate_gas(tx) .overrides_opt(overrides) - .pending() + .block(block) .await .map_err(Error::Rpc)? .into(); diff --git a/crates/simulator/src/lib.rs b/crates/simulator/src/lib.rs index 97de625fe2..d8369a13ab 100644 --- a/crates/simulator/src/lib.rs +++ b/crates/simulator/src/lib.rs @@ -8,6 +8,7 @@ mod utils; use { crate::state_override_stream::SimulationOverrides, + alloy_eips::BlockId, alloy_primitives::Address, eth_domain_types::{self as eth, AccessList, Tx}, http_client::HttpClientFactory, @@ -119,11 +120,23 @@ impl Simulator { if let Some(gas) = self.disable_gas { return Ok(gas); } - let block: eth::BlockNo = self.eth.current_block().borrow().number.into(); + let (block_number, block_timestamp) = { + let block = self.eth.current_block().borrow(); + (block.number, block.timestamp) + }; + let block: eth::BlockNo = block_number.into(); let state_overrides = self .simulation_overrides .as_ref() - .and_then(|overrides| overrides.current()); + .and_then(|overrides| overrides.overrides_for(block_number, block_timestamp)); + // The overrides are stamped for `block_number`, so the estimate has to + // run against that block. `pending` can't carry them: its timestamp is + // the node's wall clock, which is unknowable when the request is built + // and moves between calls. Without overrides nothing changes. + let block_id = match state_overrides { + Some(_) => BlockId::number(block_number), + None => BlockId::pending(), + }; Ok(match &self.inner { Inner::Tenderly(tenderly) => { tenderly @@ -140,7 +153,7 @@ impl Simulator { } Inner::Ethereum => self .eth - .estimate_gas(tx.clone(), state_overrides) + .estimate_gas(tx.clone(), state_overrides, block_id) .await .map_err(with(tx, block))?, }) diff --git a/crates/simulator/src/state_override_stream.rs b/crates/simulator/src/state_override_stream.rs index c571522f42..d265b926c7 100644 --- a/crates/simulator/src/state_override_stream.rs +++ b/crates/simulator/src/state_override_stream.rs @@ -6,27 +6,40 @@ //! previous-block state. use { - alloy_primitives::Address, - alloy_rpc_types::state::StateOverride, + alloy_primitives::{Address, B256, map::B256Map}, + alloy_rpc_types::state::{AccountOverride, StateOverride}, configs::simulator::StateOverrideStream as Config, - ethrpc::block_stream::CurrentBlockWatcher, futures::{SinkExt, StreamExt}, prometheus::{IntCounter, IntCounterVec, IntGauge}, serde::Deserialize, std::{ collections::BTreeMap, sync::Arc, - time::{Duration, Instant}, + time::{Duration, Instant, SystemTime, UNIX_EPOCH}, }, tokio::sync::watch, tracing::{debug, warn}, }; +/// Number of leading bytes of a storage word holding the venue's freshness +/// stamp. The remaining bytes are the maker's price and must survive +/// restamping untouched. +const STAMP_LEN: usize = 4; + +/// How far a word's leading bytes may sit from the time its frame was +/// published and still be read as a freshness stamp. +const STAMP_TOLERANCE: u64 = 120; + /// State overrides delivered some point in time. #[derive(Clone)] struct Snapshot { overrides: StateOverride, + /// Block the frames describe. This is the block the builder is about to + /// build (chain head + 1), not the block a simulation runs against. block_number: u64, + /// Freshness stamp of the newest frame folded into `overrides`. Only words + /// carrying it belong to a lane the maker requoted for `block_number`. + stamp: Option, received_at: Option, } @@ -35,7 +48,6 @@ pub struct SimulationOverrides(Arc); struct Inner { snapshots: watch::Receiver, - current_block: CurrentBlockWatcher, max_age: Duration, } @@ -48,33 +60,85 @@ impl std::fmt::Debug for SimulationOverrides { } impl SimulationOverrides { - /// Returns the live state overrides, or `None` (callers omit the RPC - /// override param entirely) when the stream is stale or unconfigured. - pub fn current(&self) -> Option { + /// Returns the live state overrides adjusted for a simulation running at + /// `block` with `timestamp`, or `None` (callers omit the RPC override + /// param entirely) when the stream can't serve that context. + /// + /// Frames describe the block the builder is about to build, so a snapshot + /// ahead of the simulated block is the normal case: its freshness stamp is + /// rewritten to `timestamp` so the venue accepts the quote in the context + /// it is actually evaluated in. A snapshot *behind* the simulated block is + /// withheld instead, because the maker has already moved on from the price + /// it carries. + pub fn overrides_for(&self, block: u64, timestamp: u64) -> Option { + let metrics = Metrics::get(); let snapshot = self.0.snapshots.borrow(); - let received_at = snapshot.received_at?; + let Some(received_at) = snapshot.received_at else { + metrics.record_override_result(OverrideResult::Empty); + return None; + }; if received_at.elapsed() > self.0.max_age { - Metrics::get().record_override_result(OverrideResult::Stale); + metrics.record_override_result(OverrideResult::TooOld); return None; } - let current_block_number = self.0.current_block.borrow().number; - if snapshot.block_number != current_block_number { - Metrics::get().record_override_result(OverrideResult::Stale); + if snapshot.block_number < block { + metrics.record_override_result(OverrideResult::WrongBlock); return None; } if snapshot.overrides.is_empty() { - Metrics::get().record_override_result(OverrideResult::Stale); + metrics.record_override_result(OverrideResult::Empty); return None; } - Metrics::get().record_override_result(OverrideResult::Fresh); - Some(snapshot.overrides.clone()) + metrics.record_override_result(OverrideResult::Fresh); + Some(restamp(&snapshot.overrides, snapshot.stamp, timestamp)) + } +} + +/// Reads the freshness stamp out of a storage word published at +/// `published_at`, if it carries one. +/// +/// A word only counts as stamped when its leading bytes land near the time its +/// frame was published. Words that carry no stamp hold unrelated values in the +/// same position: mostly zero, but a sample of the live stream also shows a few +/// hundred words leading with values that read as plausible unix timestamps +/// years in the past. Anchoring to the publish time is what tells the two +/// apart; a bare plausible-range check cannot. +fn stamp_of(word: &B256, published_at: u64) -> Option { + let stamp = u32::from_be_bytes(word[..STAMP_LEN].try_into().expect("4 bytes")); + (u64::from(stamp).abs_diff(published_at) <= STAMP_TOLERANCE).then_some(stamp) +} + +/// Moves the words quoted by the newest frame into the simulated block by +/// rewriting their freshness stamp to `timestamp`. +/// +/// Only words carrying `stamp` are rewritten. A lane the maker did not requote +/// keeps its older stamp and stays dead, so the venue rejects it exactly as it +/// would on chain; rewriting it too would forge liveness for a price nobody is +/// quoting. Only the stamp bytes are touched, never the price bytes next to +/// them. +fn restamp(overrides: &StateOverride, stamp: Option, timestamp: u64) -> StateOverride { + let mut overrides = overrides.clone(); + let Some(stamp) = stamp else { + return overrides; + }; + let (stamp, timestamp) = (stamp.to_be_bytes(), (timestamp as u32).to_be_bytes()); + for account in overrides.values_mut() { + let words = [account.state.as_mut(), account.state_diff.as_mut()]; + for word in words.into_iter().flatten().flat_map(B256Map::values_mut) { + if word[..STAMP_LEN] == stamp { + word[..STAMP_LEN].copy_from_slice(×tamp); + } + } } + overrides } #[derive(Debug, Deserialize)] #[serde(rename_all = "camelCase")] struct Frame { block_number: Option, + /// Nanosecond wall clock of when the venue published the frame. + timestamp: Option, // Venue keys are addresses flattened alongside the metadata fields above; // unknown non-address keys (e.g. future schema additions) are skipped by // the address parse inside the deserializer. @@ -82,6 +146,23 @@ struct Frame { venues: BTreeMap, } +impl Frame { + /// Second-resolution wall clock the frame's stamps are expected to sit + /// near. Frames are consumed live, so the local clock stands in fine when + /// the venue doesn't say. + fn published_at(&self) -> u64 { + self.timestamp.map_or_else( + || { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_secs() + }, + |nanos| nanos / 1_000_000_000, + ) + } +} + #[derive(Debug, Deserialize)] #[serde(rename_all = "camelCase")] struct VenueUpdate { @@ -124,10 +205,11 @@ where /// Spawns a new background task that streams state override updates /// into the return [`SimulationOverrides`] instance. -pub fn spawn(cfg: &Config, current_block: CurrentBlockWatcher) -> SimulationOverrides { +pub fn spawn(cfg: &Config) -> SimulationOverrides { let (sender, receiver) = watch::channel(Snapshot { overrides: StateOverride::default(), block_number: 0, + stamp: None, received_at: None, }); @@ -138,7 +220,6 @@ pub fn spawn(cfg: &Config, current_block: CurrentBlockWatcher) -> SimulationOver SimulationOverrides(Arc::new(Inner { snapshots: receiver, - current_block, max_age: cfg.max_age, })) } @@ -147,6 +228,7 @@ async fn run_stream(ws_url: url::Url, sender: watch::Sender) { let mut backoff = Duration::from_millis(250); let mut overrides = StateOverride::default(); let mut last_block_number = 0u64; + let mut stamp = None; loop { match tokio_tungstenite::connect_async(ws_url.as_str()).await { @@ -174,8 +256,10 @@ async fn run_stream(ws_url: url::Url, sender: watch::Sender) { if let Some(block_number) = frame.block_number { last_block_number = block_number; } - apply_frame(&mut overrides, frame); - publish(&overrides, last_block_number, &sender); + if let Some(newest) = apply_frame(&mut overrides, frame) { + stamp = Some(newest); + } + publish(&overrides, last_block_number, stamp, &sender); } Err(err) => { Metrics::get().parse_failures.inc(); @@ -198,21 +282,63 @@ async fn run_stream(ws_url: url::Url, sender: watch::Sender) { } } -// Each frame's stateOverride is a full per-venue snapshot, not a delta, so -// accounts are inserted directly (latest frame wins per account). -fn apply_frame(overrides: &mut StateOverride, frame: Frame) { +/// Folds a frame into the merged overrides, returning the freshness stamp it +/// carries if it quoted any lane. +/// +/// Every venue keeps its lanes in the same shared registry account and a frame +/// only ever carries its own, so storage is merged word by word: inserting the +/// account wholesale would drop the other venues' lanes. Balance, nonce and +/// code stay last-write-wins. +fn apply_frame(overrides: &mut StateOverride, frame: Frame) -> Option { + let published_at = frame.published_at(); + let mut stamp = None; for update in frame.venues.into_values() { for (account, account_override) in update.state_override { - overrides.insert(account, account_override); + let words = [&account_override.state, &account_override.state_diff]; + for word in words.into_iter().flatten().flat_map(B256Map::values) { + stamp = stamp.max(stamp_of(word, published_at)); + } + merge_account(overrides.entry(account).or_default(), account_override); } } + stamp +} + +fn merge_account(target: &mut AccountOverride, update: AccountOverride) { + let AccountOverride { + balance, + nonce, + code, + state, + state_diff, + move_precompile_to, + } = update; + target.balance = balance; + target.nonce = nonce; + target.code = code; + target.move_precompile_to = move_precompile_to; + merge_words(&mut target.state, state); + merge_words(&mut target.state_diff, state_diff); +} + +fn merge_words(target: &mut Option>, update: Option>) { + let Some(update) = update else { + return; + }; + target.get_or_insert_default().extend(update); } -fn publish(overrides: &StateOverride, block_number: u64, sender: &watch::Sender) { +fn publish( + overrides: &StateOverride, + block_number: u64, + stamp: Option, + sender: &watch::Sender, +) { Metrics::get().venue_count.set(overrides.len() as i64); let snapshot = Snapshot { overrides: overrides.clone(), block_number, + stamp, received_at: Some(Instant::now()), }; if let Err(err) = sender.send(snapshot) { @@ -228,12 +354,13 @@ struct Metrics { parse_failures: IntCounter, /// Reconnect attempts. reconnects: IntCounter, - /// Cross-venue override conflicts (no longer incremented; latest frame - /// wins per account via insert, but kept for metric stability). + /// Cross-venue override conflicts (no longer incremented; frames are + /// merged word by word, but kept for metric stability). merge_conflicts: IntCounter, /// Accounts in the merged state-override snapshot. venue_count: IntGauge, - /// Gas simulations by whether overrides were applied. + /// Simulations by whether overrides were applied, and why not if they + /// weren't. #[metric(labels("result"))] simulations_with_overrides: IntCounterVec, } @@ -252,14 +379,21 @@ impl Metrics { enum OverrideResult { Fresh, - Stale, + /// No frame arrived within the configured `max_age`. + TooOld, + /// The stream fell behind the block being simulated. + WrongBlock, + /// The stream has not published any override yet. + Empty, } impl OverrideResult { const fn as_str(&self) -> &'static str { match self { Self::Fresh => "fresh", - Self::Stale => "stale", + Self::TooOld => "too_old", + Self::WrongBlock => "wrong_block", + Self::Empty => "empty", } } } @@ -268,23 +402,17 @@ impl OverrideResult { mod tests { use { super::*, - alloy_primitives::{B256, U256, address}, - alloy_rpc_types::state::AccountOverride, - ethrpc::block_stream::BlockInfo, + alloy_primitives::{U256, address}, futures::StreamExt, std::time::Duration, tokio::time::timeout, tokio_tungstenite::tungstenite::Message, }; - fn block(number: u64, timestamp: u64) -> (watch::Sender, CurrentBlockWatcher) { - let (tx, rx) = watch::channel(BlockInfo { - number, - timestamp, - ..Default::default() - }); - (tx, rx) - } + /// Shared `PrioUpdateRegistry` every venue writes its lanes into. + const REGISTRY: Address = address!("da7afeed01fe625cf15d187a19f94b45f00b8c5f"); + /// Wall clock of the captured frames below, in seconds. + const PUBLISHED_AT: u64 = 1_783_363_067; fn frame_with( venue: Address, @@ -297,20 +425,55 @@ mod tests { account_override.balance = Some(balance); } if let Some(slot) = slot { - let mut diff = alloy_primitives::map::B256Map::default(); + let mut diff = B256Map::default(); diff.insert(slot, B256::ZERO); account_override.state_diff = Some(diff); } let mut state_override = StateOverride::default(); state_override.insert(account, account_override); - let mut venues = BTreeMap::new(); - venues.insert(venue, VenueUpdate { state_override }); Frame { block_number: None, - venues, + timestamp: None, + venues: BTreeMap::from([(venue, VenueUpdate { state_override })]), + } + } + + /// A registry word: freshness stamp in the leading bytes, maker price in + /// the rest. + fn word(stamp: u32, price: u8) -> B256 { + let mut word = B256::from([price; 32]); + word[..STAMP_LEN].copy_from_slice(&stamp.to_be_bytes()); + word + } + + /// A frame writing `lanes` of the shared registry on behalf of `venue`. + fn registry_frame(venue: Address, published_at: u64, lanes: &[(B256, B256)]) -> Frame { + let account_override = AccountOverride { + state_diff: Some(lanes.iter().copied().collect()), + ..Default::default() + }; + let mut state_override = StateOverride::default(); + state_override.insert(REGISTRY, account_override); + Frame { + block_number: None, + timestamp: Some(published_at * 1_000_000_000), + venues: BTreeMap::from([(venue, VenueUpdate { state_override })]), } } + fn lane(index: u8) -> B256 { + B256::from([index; 32]) + } + + fn lanes_of(overrides: &StateOverride) -> &B256Map { + overrides + .get(®ISTRY) + .unwrap() + .state_diff + .as_ref() + .unwrap() + } + #[test] fn frame_parses_verbatim_titan_sample() { let sample = r#"{ @@ -344,7 +507,7 @@ mod tests { } #[test] - fn apply_frame_replaces_per_venue_state() { + fn apply_frame_keeps_storage_across_frames() { let venue = address!("1111111111111111111111111111111111111111"); let account = address!("2222222222222222222222222222222222222222"); @@ -356,12 +519,14 @@ mod tests { assert!(overrides.get(&account).unwrap().state_diff.is_some()); assert!(overrides.get(&account).unwrap().balance.is_none()); + // Balance and nonce are last-write-wins, but storage a later frame + // doesn't mention survives. apply_frame( &mut overrides, frame_with(venue, account, Some(U256::ZERO), None), ); assert_eq!(overrides.get(&account).unwrap().balance, Some(U256::ZERO)); - assert!(overrides.get(&account).unwrap().state_diff.is_none()); + assert!(overrides.get(&account).unwrap().state_diff.is_some()); } #[test] @@ -402,18 +567,31 @@ mod tests { assert_eq!(overrides.get(&shared).unwrap().balance, Some(U256::from(2))); } - fn handle( - receiver: watch::Receiver, - block_rx: CurrentBlockWatcher, - max_age: Duration, - ) -> SimulationOverrides { + fn handle(receiver: watch::Receiver, max_age: Duration) -> SimulationOverrides { SimulationOverrides(Arc::new(Inner { snapshots: receiver, - current_block: block_rx, max_age, })) } + /// Handle over a snapshot built by folding `frames` in order. + fn handle_for(frames: Vec, block_number: u64, max_age: Duration) -> SimulationOverrides { + let mut overrides = StateOverride::default(); + let mut stamp = None; + for frame in frames { + if let Some(newest) = apply_frame(&mut overrides, frame) { + stamp = Some(newest); + } + } + let (_sender, receiver) = watch::channel(Snapshot { + overrides, + block_number, + stamp, + received_at: Some(Instant::now()), + }); + handle(receiver, max_age) + } + fn non_empty_snapshot(block_number: u64, received_at: Instant) -> Snapshot { let mut overrides = StateOverride::default(); overrides.insert( @@ -423,49 +601,192 @@ mod tests { Snapshot { overrides, block_number, + stamp: None, received_at: Some(received_at), } } #[test] - fn staleness_gates_return_none() { - // Stale by age. + fn overrides_withheld_when_context_cannot_be_served() { + // No frame arrived recently enough. let (_sender, receiver) = watch::channel(non_empty_snapshot( 100, Instant::now() - Duration::from_millis(100), )); - let (_, block_rx) = block(100, 1000); assert!( - handle(receiver, block_rx, Duration::from_millis(50)) - .current() + handle(receiver, Duration::from_millis(50)) + .overrides_for(99, 1000) .is_none() ); - // Stale by block mismatch: snapshot block 100, head block 105. + // The stream fell behind: it still describes block 100 while block 105 + // is being simulated, so the maker has moved on from its price. let (_sender, receiver) = watch::channel(non_empty_snapshot(100, Instant::now())); - let (_, block_rx) = block(105, 1000); assert!( - handle(receiver, block_rx, Duration::from_secs(30)) - .current() + handle(receiver, Duration::from_secs(30)) + .overrides_for(105, 1000) .is_none() ); - // Stale because the merged overrides are empty. - let (sender, receiver) = watch::channel(Snapshot { + // Nothing published yet. + let (_sender, receiver) = watch::channel(Snapshot { overrides: StateOverride::default(), block_number: 100, + stamp: None, received_at: Some(Instant::now()), }); - let (_, block_rx) = block(100, 1000); - let h = handle(receiver, block_rx, Duration::from_secs(30)); - sender - .send(Snapshot { - overrides: StateOverride::default(), - block_number: 100, - received_at: Some(Instant::now()), - }) - .unwrap(); - assert!(h.current().is_none()); + assert!( + handle(receiver, Duration::from_secs(30)) + .overrides_for(99, 1000) + .is_none() + ); + } + + #[test] + fn snapshot_ahead_of_simulated_block_is_served() { + // Frames name the block the builder is about to build, so the snapshot + // being one ahead of the simulated block is the normal case. + let (_sender, receiver) = watch::channel(non_empty_snapshot(100, Instant::now())); + assert!( + handle(receiver, Duration::from_secs(30)) + .overrides_for(99, 1000) + .is_some() + ); + } + + #[test] + fn newest_lanes_are_restamped_to_the_simulated_block() { + let venue = address!("1111111111111111111111111111111111111111"); + let simulated_at = PUBLISHED_AT - 12; + + let handle = handle_for( + vec![registry_frame( + venue, + PUBLISHED_AT, + &[ + (lane(1), word(PUBLISHED_AT as u32, 0xaa)), + (lane(2), word(PUBLISHED_AT as u32, 0xbb)), + ], + )], + 100, + Duration::from_secs(30), + ); + + let overrides = handle.overrides_for(99, simulated_at).unwrap(); + let lanes = lanes_of(&overrides); + assert_eq!(lanes[&lane(1)], word(simulated_at as u32, 0xaa)); + assert_eq!(lanes[&lane(2)], word(simulated_at as u32, 0xbb)); + } + + #[test] + fn lane_not_requoted_keeps_its_stale_stamp() { + let venue = address!("1111111111111111111111111111111111111111"); + let previous = PUBLISHED_AT - 12; + let simulated_at = PUBLISHED_AT - 12; + + // Both lanes were quoted for the previous block, only lane 1 was + // requoted for this one. + let handle = handle_for( + vec![ + registry_frame( + venue, + previous, + &[ + (lane(1), word(previous as u32, 0xaa)), + (lane(2), word(previous as u32, 0xbb)), + ], + ), + registry_frame( + venue, + PUBLISHED_AT, + &[(lane(1), word(PUBLISHED_AT as u32, 0xcc))], + ), + ], + 100, + Duration::from_secs(30), + ); + + let overrides = handle.overrides_for(99, simulated_at).unwrap(); + let lanes = lanes_of(&overrides); + assert_eq!(lanes[&lane(1)], word(simulated_at as u32, 0xcc)); + // Lane 2 must stay dead: forging liveness for it would quote a price + // the maker is no longer offering. + assert_eq!(lanes[&lane(2)], word(previous as u32, 0xbb)); + } + + #[test] + fn restamping_leaves_every_other_byte_untouched() { + let mut overrides = StateOverride::default(); + let stamp = apply_frame(&mut overrides, serde_json::from_str(FERMI_FRAME).unwrap()); + assert_eq!(stamp, Some(PUBLISHED_AT as u32)); + + let simulated_at = PUBLISHED_AT - 12; + let restamped = restamp(&overrides, stamp, simulated_at); + + // Only the stamp bytes of the registry words moved; the maker's price + // bytes and every other account are byte-identical. + let (before, after) = (lanes_of(&overrides), lanes_of(&restamped)); + assert_eq!(before.len(), after.len()); + for (slot, before) in before { + let after = after[slot]; + assert_eq!(&after[..STAMP_LEN], &(simulated_at as u32).to_be_bytes()); + assert_eq!(after[STAMP_LEN..], before[STAMP_LEN..]); + } + for (account, before) in &overrides { + if *account != REGISTRY { + assert_eq!(&restamped[account], before); + } + } + } + + #[test] + fn unrelated_words_are_never_mistaken_for_a_stamp() { + // This venue's word leads with bytes that read as a perfectly plausible + // unix timestamp (2019-07-25) — just not one near the time the frame + // was published. + let mut overrides = StateOverride::default(); + assert_eq!( + apply_frame(&mut overrides, serde_json::from_str(OTHER_FRAME).unwrap()), + None + ); + + let venue = address!("28d9ccedf1b7ac9b3f090f4f0292837de87c1d39"); + let before = overrides.clone(); + assert_eq!( + restamp(&overrides, Some(PUBLISHED_AT as u32), PUBLISHED_AT - 12)[&venue], + before[&venue] + ); + } + + #[test] + fn venues_sharing_the_registry_keep_each_others_lanes() { + let venue_a = address!("1111111111111111111111111111111111111111"); + let venue_b = address!("3333333333333333333333333333333333333333"); + let simulated_at = PUBLISHED_AT - 12; + + // Each venue only ever sends its own lanes of the shared registry. + let handle = handle_for( + vec![ + registry_frame( + venue_a, + PUBLISHED_AT, + &[(lane(1), word(PUBLISHED_AT as u32, 0xaa))], + ), + registry_frame( + venue_b, + PUBLISHED_AT, + &[(lane(2), word(PUBLISHED_AT as u32, 0xbb))], + ), + ], + 100, + Duration::from_secs(30), + ); + + let overrides = handle.overrides_for(99, simulated_at).unwrap(); + let lanes = lanes_of(&overrides); + assert_eq!(lanes.len(), 2); + assert_eq!(lanes[&lane(1)], word(simulated_at as u32, 0xaa)); + assert_eq!(lanes[&lane(2)], word(simulated_at as u32, 0xbb)); } #[tokio::test] @@ -487,18 +808,19 @@ mod tests { write.close().await.unwrap(); }); - let (_, block_rx) = block(11, 1000); let cfg = Config { ws_url: server_url, max_age: Duration::from_secs(30), }; - let handle = spawn(&cfg, block_rx); + let handle = spawn(&cfg); let _ = server_handle.await; let got = timeout(Duration::from_secs(2), async { loop { - if let Some(overrides) = handle.current() + // The frames name block 11, the block the builder is about to + // build on top of head 10. + if let Some(overrides) = handle.overrides_for(10, 1000) && let Some(account_override) = overrides.get(&account) && account_override.balance == Some(U256::from(2)) { @@ -519,10 +841,14 @@ mod tests { fn parses_real_titan_frames() { let fermi: Frame = serde_json::from_str(FERMI_FRAME).unwrap(); assert_eq!(fermi.block_number, Some(25475333)); + assert_eq!(fermi.published_at(), PUBLISHED_AT); assert_eq!(fermi.venues.len(), 1); let mut overrides = StateOverride::default(); - apply_frame(&mut overrides, fermi); + assert_eq!( + apply_frame(&mut overrides, fermi), + Some(PUBLISHED_AT as u32) + ); assert_eq!(overrides.len(), 5); let venue_account = address!("da7afeed01fe625cf15d187a19f94b45f00b8c5f"); @@ -550,33 +876,271 @@ mod tests { } #[test] - fn current_yields_state_overrides_for_real_data() { - let (block_tx, block_rx) = block(25475333, 1783363000); - let (_sender, receiver) = watch::channel(Snapshot { - overrides: { - let mut m = StateOverride::default(); - apply_frame(&mut m, serde_json::from_str(FERMI_FRAME).unwrap()); - m - }, - block_number: 25475333, - received_at: Some(Instant::now()), - }); - let handle = SimulationOverrides(Arc::new(Inner { - snapshots: receiver, - current_block: block_rx, - max_age: Duration::from_secs(30), - })); + fn yields_state_overrides_for_real_data() { + let handle = handle_for( + vec![serde_json::from_str(FERMI_FRAME).unwrap()], + 25475333, + Duration::from_secs(30), + ); - let overrides = handle.current().unwrap(); + // The frame names 25475333, so it serves a simulation on head 25475332. + let overrides = handle.overrides_for(25475332, PUBLISHED_AT - 12).unwrap(); assert_eq!(overrides.len(), 5); - block_tx - .send(BlockInfo { - number: 25475335, - timestamp: 1783363000, - ..Default::default() - }) + // ...but not one on a block the stream has already fallen behind. + assert!(handle.overrides_for(25475335, PUBLISHED_AT + 24).is_none()); + } + + /// Titan's Fermi router. Its pAMM reverts `StaleUpdate()` unless the + /// registry word it reads carries the timestamp of the block the call runs + /// against, which is exactly what restamping provides. + const FERMI_ROUTER: Address = address!("b1076fe3ab5e28005c7c323bac5ac06a680d452e"); + const USDT: Address = address!("dac17f958d2ee523a2206206994597c13d831ec7"); + const WETH: Address = address!("c02aaa39b223fe8d0a0e5c4f27ead9083c756cc2"); + + /// `(address tokenIn, address tokenOut, uint256 amountIn) view` on the + /// Fermi router. The contract is unverified, so the quote is addressed by + /// selector rather than by name. + const QUOTE: [u8; 4] = hex_literal::hex!("300aa47f"); + + /// Quotes 1000 USDT into WETH, returning the amount out. + async fn quote_amounts( + provider: ðrpc::AlloyProvider, + block: u64, + overrides: Option, + ) -> Result { + use { + alloy_provider::{Provider, network::TransactionBuilder}, + alloy_sol_types::SolValue, + }; + + let args = (USDT, WETH, U256::from(1_000_000_000u64)); + let output = provider + .call( + alloy_rpc_types::TransactionRequest::default() + .with_to(FERMI_ROUTER) + .with_input([QUOTE.as_slice(), &args.abi_encode_params()].concat()), + ) + .overrides_opt(overrides) + .block(block.into()) + .await?; + let (_amount_in, amount_out) = <(U256, U256)>::abi_decode_params(&output).unwrap(); + Ok(amount_out) + } + + /// Exercises the gas path end to end: `Simulator::gas` must estimate a pAMM + /// call that only succeeds when the live overrides are applied in the + /// context they were stamped for. + /// + /// Also pins down why the estimate is no longer run against `pending`: the + /// very same overrides are rejected there, because `pending`'s timestamp is + /// the node's wall clock rather than the block they were stamped for. + /// + /// Needs `NODE_URL`, `NODE_WS_URL` and `PAMM_QUOTE_STREAM_URL`: + /// + /// ```text + /// NODE_URL=... NODE_WS_URL=... PAMM_QUOTE_STREAM_URL=wss://.../ws/pamm_quote_stream \ + /// cargo nextest run -p simulator --run-ignored ignored-only \ + /// estimates_gas_for_pamm_call + /// ``` + #[tokio::test] + #[ignore] + async fn estimates_gas_for_pamm_call() { + use { + alloy_provider::{Provider, network::TransactionBuilder}, + alloy_sol_types::SolValue, + }; + + let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); + + let web3 = ethrpc::Web3::new_from_env(); + let ws_url: url::Url = std::env::var("NODE_WS_URL").unwrap().parse().unwrap(); + let blocks = ethrpc::block_stream::current_block_ws_stream(web3.provider.clone(), ws_url) + .await + .unwrap(); + let cfg = Config { + ws_url: std::env::var("PAMM_QUOTE_STREAM_URL") + .unwrap() + .parse() + .unwrap(), + max_age: Duration::from_secs(30), + }; + let overrides = spawn(&cfg); + tokio::time::sleep(Duration::from_secs(5)).await; + + let eth = crate::Ethereum::new( + web3.clone(), + chain::Chain::Mainnet, + Default::default(), + Arc::new(gas_price_estimation::FakeGasPriceEstimator::default()), + blocks.clone(), + U256::from(30_000_000), + ); + let args = (USDT, WETH, U256::from(1_000_000_000u64)); + let tx = eth_domain_types::Tx { + from: Address::ZERO, + to: FERMI_ROUTER, + value: U256::ZERO.into(), + input: [QUOTE.as_slice(), &args.abi_encode_params()] + .concat() + .into(), + access_list: Default::default(), + }; + + // Without the stream the venue is stale and the estimate reverts. + let bare = crate::Simulator::ethereum(eth.clone()); + assert!( + bare.gas(tx.clone()).await.is_err(), + "estimate succeeded without overrides, the check proves nothing" + ); + + // With it, the same estimate goes through. + let mut simulator = crate::Simulator::ethereum(eth); + simulator.set_simulation_overrides(overrides.clone()); + let gas = simulator + .gas(tx.clone()) + .await + .expect("gas estimation reverted with overrides applied"); + assert!( + gas.0 > U256::from(21_000), + "implausible gas estimate {gas:?}" + ); + + // ...but not against `pending`, which is where it used to run. + let head = *blocks.borrow(); + let state = overrides + .overrides_for(head.number, head.timestamp) + .expect("stream served no overrides at head"); + let request = alloy_rpc_types::TransactionRequest::default() + .with_to(FERMI_ROUTER) + .with_input(tx.input.0.clone()); + assert!( + web3.provider + .estimate_gas(request) + .overrides(state) + .pending() + .await + .is_err(), + "overrides stamped for {} were accepted at pending", + head.number + ); + } + + /// The overrides have to be served continuously, not just in the sliver + /// right after a block lands. Samples the accessor at chain head across + /// several blocks and requires virtually all samples to be served; the + /// block-number gate this replaced scored about 5% here. + /// + /// Needs `NODE_URL`, `NODE_WS_URL` and `PAMM_QUOTE_STREAM_URL`: + /// + /// ```text + /// NODE_URL=... NODE_WS_URL=... PAMM_QUOTE_STREAM_URL=wss://.../ws/pamm_quote_stream \ + /// cargo nextest run -p simulator --run-ignored ignored-only \ + /// serves_overrides_across_blocks + /// ``` + #[tokio::test] + #[ignore] + async fn serves_overrides_across_blocks() { + let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); + + // A real block watcher, as the binaries build it. + let provider = ethrpc::Web3::new_from_env().provider; + let ws_url: url::Url = std::env::var("NODE_WS_URL").unwrap().parse().unwrap(); + let blocks = ethrpc::block_stream::current_block_ws_stream(provider, ws_url) + .await .unwrap(); - assert!(handle.current().is_none()); + + let cfg = Config { + ws_url: std::env::var("PAMM_QUOTE_STREAM_URL") + .unwrap() + .parse() + .unwrap(), + max_age: Duration::from_secs(30), + }; + let handle = spawn(&cfg); + tokio::time::sleep(Duration::from_secs(5)).await; + + // ~60s, several blocks' worth. + let (mut served, mut withheld) = (0u32, 0u32); + let mut blocks_seen = std::collections::BTreeSet::new(); + for _ in 0..600 { + let head = *blocks.borrow(); + blocks_seen.insert(head.number); + match handle.overrides_for(head.number, head.timestamp) { + Some(_) => served += 1, + None => withheld += 1, + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + + assert!( + blocks_seen.len() >= 2, + "block watcher never advanced, the sample spans no block boundary" + ); + assert!( + served >= (served + withheld) * 9 / 10, + "overrides served for only {served} of {} samples across {} blocks", + served + withheld, + blocks_seen.len() + ); + } + + /// Live conformance check: the overrides this module hands out must + /// actually make a Titan pAMM quote, in the exact context they are applied + /// in. Nothing short of a real stream against a real node catches a frame + /// that describes a different block than the one being simulated. + /// + /// Requires a mainnet node and the venue's quote stream: + /// + /// ```text + /// NODE_URL=... PAMM_QUOTE_STREAM_URL=wss://.../ws/pamm_quote_stream \ + /// cargo nextest run -p simulator --run-ignored ignored-only \ + /// quotes_pamm_against_live_stream + /// ``` + #[tokio::test] + #[ignore] + async fn quotes_pamm_against_live_stream() { + // The workspace links more than one `rustls` crypto provider, so one + // has to be chosen before any TLS handshake. The binaries get this for + // free: the alloy websocket transport installs a provider while opening + // the block stream, long before this stream connects. A test process + // does not, so it picks one itself. + let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); + + let provider = ethrpc::Web3::new_from_env().provider; + let cfg = Config { + ws_url: std::env::var("PAMM_QUOTE_STREAM_URL") + .unwrap() + .parse() + .unwrap(), + max_age: Duration::from_secs(30), + }; + let overrides = spawn(&cfg); + + let quoted = timeout(Duration::from_secs(60), async { + loop { + let head = + ethrpc::block_stream::get_block_at_id(&provider, alloy_eips::BlockId::latest()) + .await + .unwrap(); + if let Some(state) = overrides.overrides_for(head.number, head.timestamp) { + // Without the overrides the pool has nothing fresh to + // quote from and reverts `StaleUpdate()` (0x666a2814); + // with them it quotes the maker's live price. + assert!( + quote_amounts(&provider, head.number, None).await.is_err(), + "venue quoted without overrides, the check proves nothing" + ); + let amount_out = quote_amounts(&provider, head.number, Some(state)) + .await + .expect("pAMM quote reverted with overrides applied"); + assert!(amount_out > U256::ZERO, "venue quoted nothing"); + return; + } + tokio::time::sleep(Duration::from_millis(250)).await; + } + }) + .await; + assert!(quoted.is_ok(), "stream never served the chain head"); } }