From 640ce3fdfc871d940fdcc55df8aa756d8f273ac3 Mon Sep 17 00:00:00 2001 From: Kai Aldag Date: Tue, 7 Jul 2026 16:06:00 +0100 Subject: [PATCH 1/2] feat(balancer): event-source Balancer V2 swaps (no RPC where warm) Extend the exact-write posture (V2/Solidly `Sync`, V3 `Mint`/`Burn`) to Balancer V2 `Swap`s: apply each swapped token's exact vault `cash` delta directly from the event's `amountIn`/`amountOut` with no RPC, falling back to a balance-slots resync only where a token's cash location is unknown. Mechanism - Vault balances pack as [lastChangeBlock:32][managed:112][cash:112]; a swap changes only the 112-bit `cash` field, by exactly +/-amount (protocol fees settle lazily on join/exit, so there is no per-swap skim -- verified live). - `BalancerTokenBalance` records each token's (slot, high_field). The discover cold-start probes it by matching getPoolTokens balances to warmed slot cash fields -- specialization-agnostic (TWO_TOKEN shared slot + GENERAL/MINIMAL per-token slots); ambiguous/zero/managed matches are skipped -> resync. - TWO_TOKEN's two fields share one word, so both deltas accumulate into a single write (two writes from one pre-image would clobber). - `PoolBalanceChanged` (join/exit) is now subscribed: it routes + resyncs the balances (event-sourcing its deltas is a follow-up). `token_cash` is built in discover mode and carried through verify-only cold starts, so event-sourcing works after either path; a slots-only pre-population safely falls back to resync. Tests - 7 inline unit tests: cash-field round-trips, RMW add/sub/overflow/underflow, probe (TWO_TOKEN + GENERAL + ambiguous-skip), event_source_swap (shared-slot accumulation, separate slots, unknown/cold -> None). - 4 reactive integration tests: warm TWO_TOKEN + GENERAL event-sourcing (Exact, no resync), unmapped-swap resync fallback, PoolBalanceChanged route+resync. - Live RPC parity (env-gated, #[ignore]): for a real TWO_TOKEN and GENERAL swap, trace_replayTransaction(stateDiff) confirms the event-sourced cash fields reproduce the on-chain per-tx values exactly. Offline gate green: tests (all-features/default/no-default), clippy -D warnings (all-features + no-default), doc -D warnings (missing_docs), fmt. Also fmt-normalizes 2 pre-existing nits in the Curve files inherited from the base commit (same rustfmt the repo uses). Co-Authored-By: Claude Opus 4.8 --- Cargo.toml | 13 +- src/adapters/balancer_v2.rs | 605 +++++++++++++++++++++++++++++--- src/adapters/mod.rs | 8 +- src/adapters/types.rs | 55 +++ tests/adapter_reactive.rs | 302 +++++++++++++++- tests/balancer_liquidity_rpc.rs | 282 +++++++++++++++ 6 files changed, 1199 insertions(+), 66 deletions(-) create mode 100644 tests/balancer_liquidity_rpc.rs diff --git a/Cargo.toml b/Cargo.toml index 8f0d5d8..fc8550b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -96,8 +96,9 @@ tokio = { version = "1.48.0", features = ["macros", "rt-multi-thread"] } # `plotters` deps — keeps the build lean and, since the crate deliberately ships # without rayon, avoids reintroducing it even in the dev/bench build. criterion = { version = "0.5", default-features = false } -# The live V3-liquidity parity test parses `trace_replayTransaction` stateDiff as -# raw JSON (per-tx storage before/after ground truth). Dev-only. +# The live V3-liquidity and Balancer event-sourcing parity tests parse +# `trace_replayTransaction` stateDiff as raw JSON (per-tx storage before/after +# ground truth). Dev-only. serde_json = "1.0" # Runnable adapters-path demo: register -> cold-start -> WS event subscribe -> @@ -204,6 +205,14 @@ required-features = ["uniswap-v3"] name = "v3_liquidity_rpc" required-features = ["uniswap-v3"] +# Live parity for event-sourced Balancer V2 swaps (env-gated, #[ignore]): for a +# real vault swap on both a TWO_TOKEN and a GENERAL pool, apply the event and +# assert the adapter's writes reproduce the on-chain per-tx cash deltas +# (trace_replayTransaction stateDiff). +[[test]] +name = "balancer_liquidity_rpc" +required-features = ["balancer-v2"] + [[test]] name = "reactive_ws_e2e" required-features = ["uniswap-v2"] diff --git a/src/adapters/balancer_v2.rs b/src/adapters/balancer_v2.rs index 8fc26ae..13b9088 100644 --- a/src/adapters/balancer_v2.rs +++ b/src/adapters/balancer_v2.rs @@ -8,15 +8,262 @@ use super::sim::{ }; use super::{ AdapterCache, AdapterEvent, AdapterEventError, AdapterEventKind, AdapterEventResult, - AmmAdapter, BalancerV2Metadata, ColdStartOutcome, ColdStartPolicy, ColdStartReport, - EventSource, PoolRegistration, PoolStatus, ProtocolId, ProtocolMetadata, RepairAction, - SlotChange, StateView, UnsupportedReason, UpdateQuality, + AmmAdapter, BalancerTokenBalance, BalancerV2Metadata, ColdStartOutcome, ColdStartPolicy, + ColdStartReport, EventSource, PoolRegistration, PoolStatus, ProtocolId, ProtocolMetadata, + RepairAction, SlotChange, StateUpdate, StateView, UnsupportedReason, UpdateQuality, }; use alloy_primitives::{Address, B256, Bytes, Log, U256}; use alloy_sol_types::{SolCall, SolEvent, sol}; sol! { event Swap(bytes32 indexed poolId, address indexed tokenIn, address indexed tokenOut, uint256 amountIn, uint256 amountOut); + /// Emitted by the vault on a join or exit — it changes the pool's vault + /// balances, so it is subscribed and resynced (event-sourcing its `deltas` / + /// `protocolFeeAmounts` is a follow-up). `topic0 = 0xe5ce2490…`. + event PoolBalanceChanged(bytes32 indexed poolId, address indexed liquidityProvider, address[] tokens, int256[] deltas, uint256[] protocolFeeAmounts); +} + +/// Width of the vault `BalanceAllocation` `cash` field (bits): a packed balance is +/// `[lastChangeBlock : top 32][managed : bits 112–223][cash : bits 0–111]`. +const CASH_BITS: usize = 112; + +/// Mask selecting a 112-bit `cash` field. +fn cash_mask() -> U256 { + (U256::from(1) << CASH_BITS) - U256::from(1) +} + +/// Extract the 112-bit `cash` field at the low (bits 0–111) or high (bits +/// 112–223) position of a packed vault balance word. +fn cash_field(word: U256, high: bool) -> U256 { + let shift = if high { CASH_BITS } else { 0 }; + (word >> shift) & cash_mask() +} + +/// Set the 112-bit `cash` field of `word`, preserving the other bits (`managed` / +/// `lastChangeBlock`, or the co-tenant token's `cash` in a TWO_TOKEN slot). +fn set_cash_field(word: U256, high: bool, cash: U256) -> U256 { + let shift = if high { CASH_BITS } else { 0 }; + let cleared = word & (U256::MAX ^ (cash_mask() << shift)); + cleared | ((cash & cash_mask()) << shift) +} + +/// Apply a `cash` delta (`+amount` on add, `-amount` on sub) to `word`'s field, +/// returning the new word — or `None` on 112-bit overflow / underflow (invalid +/// for real vault balances) so the caller can fall back to a resync. +fn apply_cash_delta(word: U256, high: bool, add: bool, amount: U256) -> Option { + let old = cash_field(word, high); + let new = if add { + old.checked_add(amount)? + } else { + old.checked_sub(amount)? + }; + if new > cash_mask() { + return None; + } + Some(set_cash_field(word, high, new)) +} + +/// Probe the warmed vault slots to locate each token's `cash` field by value: +/// match each token's balance (== `cash` for an unmanaged pool) to a discovered +/// slot's low (bits 0–111) or high (bits 112–223) 112-bit field. A token with no +/// **unique** match — a zero balance, a managed balance (`cash != balance`), or a +/// value collision with another field — is skipped, so the reactive `Swap` path +/// resyncs it rather than risk writing the wrong slot. Specialization-agnostic: +/// works for the TWO_TOKEN shared slot and per-token GENERAL/MINIMAL slots alike, +/// and naturally ignores `EnumerableMap` overhead slots (no balance matches them). +fn probe_token_cash( + tokens: &[Address], + balances: &[U256], + verified_slots: &[(Address, U256)], + vault: Address, + state: &dyn StateView, +) -> Vec { + let mut located = Vec::new(); + for (token, balance) in tokens.iter().zip(balances.iter()) { + if *balance == U256::ZERO { + continue; // a zero balance matches any empty field — ambiguous. + } + let matches: Vec<(U256, bool)> = verified_slots + .iter() + .filter(|(address, _)| *address == vault) + .filter_map(|(_, slot)| state.storage(vault, *slot).map(|word| (*slot, word))) + .flat_map(|(slot, word)| { + let mut found = Vec::new(); + if cash_field(word, false) == *balance { + found.push((slot, false)); + } + if cash_field(word, true) == *balance { + found.push((slot, true)); + } + found + }) + .collect(); + // Only a unique match is trustworthy; otherwise leave the token to resync. + if let [(slot, high)] = matches.as_slice() { + located.push(BalancerTokenBalance::new(*token, *slot, *high)); + } + } + located +} + +/// Locate a token's `cash` field in the metadata's probed map. +fn token_cash_location( + metadata: &BalancerV2Metadata, + token: Address, +) -> Option { + metadata + .token_cash + .iter() + .find(|balance| balance.token == token) + .copied() +} + +/// The resync repair + quality for the fallback path: re-verify the known +/// `balance_slots`, or a conservative no-op when none are known yet. +fn resync_repair(vault: Address, metadata: &BalancerV2Metadata) -> (RepairAction, UpdateQuality) { + if metadata.balance_slots.is_empty() { + (RepairAction::None, UpdateQuality::ConservativeInvalidation) + } else { + ( + RepairAction::VerifySlots( + metadata + .balance_slots + .iter() + .map(|slot| (vault, *slot)) + .collect(), + ), + UpdateQuality::RequiresRepair, + ) + } +} + +/// Event-source a swap: if both tokens' `cash` fields are located and warm, +/// return the exact vault-balance writes (`+amountIn` to `tokenIn`'s cash, +/// `-amountOut` from `tokenOut`'s). `None` on any gap (unknown token, cold slot, +/// or 112-bit overflow) so the caller falls back to a resync. +fn event_source_swap( + view: &dyn StateView, + vault: Address, + metadata: &BalancerV2Metadata, + token_in: Address, + amount_in: U256, + token_out: Address, + amount_out: U256, +) -> Option> { + let in_loc = token_cash_location(metadata, token_in)?; + let out_loc = token_cash_location(metadata, token_out)?; + + if in_loc.slot == out_loc.slot { + // TWO_TOKEN shared slot: both fields live in one word, so apply both + // deltas to a single write (two separate writes would clobber each other). + let word = view.storage(vault, in_loc.slot)?; + let word = apply_cash_delta(word, in_loc.high_field, true, amount_in)?; + let word = apply_cash_delta(word, out_loc.high_field, false, amount_out)?; + Some(vec![StateUpdate::slot(vault, in_loc.slot, word)]) + } else { + let word_in = view.storage(vault, in_loc.slot)?; + let word_in = apply_cash_delta(word_in, in_loc.high_field, true, amount_in)?; + let word_out = view.storage(vault, out_loc.slot)?; + let word_out = apply_cash_delta(word_out, out_loc.high_field, false, amount_out)?; + Some(vec![ + StateUpdate::slot(vault, in_loc.slot, word_in), + StateUpdate::slot(vault, out_loc.slot, word_out), + ]) + } +} + +/// Decode a vault `Swap` and **event-source** the two token cash balances +/// directly (no RPC) when possible, falling back to a `balance_slots` resync. +fn decode_swap(pool: &PoolRegistration, log: &Log, view: &dyn StateView) -> AdapterEventResult { + let decoded = match Swap::decode_log_data_validate(&log.data) { + Ok(decoded) => decoded, + Err(_) => { + return AdapterEventResult::error(AdapterEventError::MalformedLog( + "malformed Balancer V2 Swap log", + )); + } + }; + + let (updates, repair, quality) = match &pool.metadata { + ProtocolMetadata::BalancerV2(metadata) => match metadata.vault { + Some(vault) => match event_source_swap( + view, + vault, + metadata, + decoded.tokenIn, + decoded.amountIn, + decoded.tokenOut, + decoded.amountOut, + ) { + Some(updates) => (updates, RepairAction::None, UpdateQuality::Exact), + None => { + let (repair, quality) = resync_repair(vault, metadata); + (Vec::new(), repair, quality) + } + }, + None => ( + Vec::new(), + RepairAction::None, + UpdateQuality::ConservativeInvalidation, + ), + }, + _ => ( + Vec::new(), + RepairAction::None, + UpdateQuality::ConservativeInvalidation, + ), + }; + + AdapterEventResult::event( + AdapterEvent::new( + pool.key.clone(), + log.address, + Swap::SIGNATURE_HASH, + AdapterEventKind::Swap, + quality, + ) + .with_updates(updates) + .with_repair(repair), + ) +} + +/// Decode a vault `PoolBalanceChanged` (join/exit). It changes the vault +/// balances, so v1 resyncs the known read-set (event-sourcing its signed +/// `deltas`, net of `protocolFeeAmounts`, is a follow-up). The `kind` tag is +/// taken from the delta signs. +fn decode_liquidity_change(pool: &PoolRegistration, log: &Log) -> AdapterEventResult { + let decoded = match PoolBalanceChanged::decode_log_data_validate(&log.data) { + Ok(decoded) => decoded, + Err(_) => { + return AdapterEventResult::error(AdapterEventError::MalformedLog( + "malformed Balancer V2 PoolBalanceChanged log", + )); + } + }; + let kind = if decoded.deltas.iter().any(|delta| delta.is_positive()) { + AdapterEventKind::LiquidityAdded + } else { + AdapterEventKind::LiquidityRemoved + }; + + let (repair, quality) = match &pool.metadata { + ProtocolMetadata::BalancerV2(metadata) => match metadata.vault { + Some(vault) => resync_repair(vault, metadata), + None => (RepairAction::None, UpdateQuality::ConservativeInvalidation), + }, + _ => (RepairAction::None, UpdateQuality::ConservativeInvalidation), + }; + + AdapterEventResult::event( + AdapterEvent::new( + pool.key.clone(), + log.address, + PoolBalanceChanged::SIGNATURE_HASH, + kind, + quality, + ) + .with_repair(repair), + ) } sol! { @@ -46,7 +293,13 @@ impl AmmAdapter for BalancerV2Adapter { }; vault - .map(|vault| EventSource::indexed_bytes32(vault, vec![Swap::SIGNATURE_HASH], 1)) + .map(|vault| { + EventSource::indexed_bytes32( + vault, + vec![Swap::SIGNATURE_HASH, PoolBalanceChanged::SIGNATURE_HASH], + 1, + ) + }) .into_iter() .collect() } @@ -77,13 +330,16 @@ impl AmmAdapter for BalancerV2Adapter { // A non-empty `balance_slots` means the vault read-set is already known (a // prior discovery / trace), so the planner runs a verify-only fast path - // instead of rediscovering. `tokens` is preserved across a verify-only run - // (there is no `getPoolTokens` decode to repopulate it). - let (known_slots, tokens) = match &pool.metadata { - ProtocolMetadata::BalancerV2(metadata) => { - (metadata.balance_slots.clone(), metadata.tokens.clone()) - } - _ => (Vec::new(), Vec::new()), + // instead of rediscovering. `tokens` and the probed `token_cash` map are + // preserved across a verify-only run (there is no `getPoolTokens` decode / + // probe to repopulate them). + let (known_slots, tokens, known_token_cash) = match &pool.metadata { + ProtocolMetadata::BalancerV2(metadata) => ( + metadata.balance_slots.clone(), + metadata.tokens.clone(), + metadata.token_cash.clone(), + ), + _ => (Vec::new(), Vec::new(), Vec::new()), }; Ok(Box::new(BalancerV2ColdStartPlanner::new( @@ -91,6 +347,7 @@ impl AmmAdapter for BalancerV2Adapter { pool_id, known_slots, tokens, + known_token_cash, policy, ))) } @@ -99,51 +356,18 @@ impl AmmAdapter for BalancerV2Adapter { &self, pool: &PoolRegistration, log: &Log, - _view: &dyn StateView, + view: &dyn StateView, ) -> AdapterEventResult { - if log.topics().first() != Some(&Swap::SIGNATURE_HASH) { + let Some(topic0) = log.topics().first().copied() else { return AdapterEventResult::ignored(); - } - - if Swap::decode_log_data_validate(&log.data).is_err() { - return AdapterEventResult::error(AdapterEventError::MalformedLog( - "malformed Balancer V2 Swap log", - )); - } - - // The vault balances live behind a non-predictable storage mapping, so - // the Swap event payload cannot be turned into an exact masked write. - // Instead we keep the cached balances fresh by re-verifying the exact - // `(vault, slot)` pairs the cold-start `getPoolTokens` discovery found: - // a `VerifySlots` repair the reactive runtime lowers into a hash-pinned - // resync, re-reading the post-swap balances authoritatively. This stays - // consistent with the discover-based cold start and avoids lossy - // event-delta arithmetic. The discovered slots are persisted on - // `BalancerV2Metadata.balance_slots` by the cold-start `finish`. - let repair = match &pool.metadata { - ProtocolMetadata::BalancerV2(metadata) => { - match (metadata.vault, metadata.balance_slots.as_slice()) { - (Some(vault), slots) if !slots.is_empty() => { - RepairAction::VerifySlots(slots.iter().map(|slot| (vault, *slot)).collect()) - } - // Vault known but no discovered slots yet (cold-start has not - // run / found them): fall back to the conservative no-op so - // the routing/observability behavior is preserved. - _ => RepairAction::None, - } - } - _ => RepairAction::None, }; - - AdapterEventResult::event(AdapterEvent { - pool: pool.key.clone(), - emitter: log.address, - topic0: Swap::SIGNATURE_HASH, - kind: AdapterEventKind::Swap, - updates: Vec::new(), - quality: UpdateQuality::ConservativeInvalidation, - repair, - }) + if topic0 == Swap::SIGNATURE_HASH { + decode_swap(pool, log, view) + } else if topic0 == PoolBalanceChanged::SIGNATURE_HASH { + decode_liquidity_change(pool, log) + } else { + AdapterEventResult::ignored() + } } /// Quote via `Vault.queryBatchSwap(GIVEN_IN, [swap], assets, funds)`. @@ -263,6 +487,13 @@ struct BalancerV2ColdStartPlanner { /// Tokens decoded from `getPoolTokens` (discover mode) or carried from the /// config-supplied metadata (verify-only mode). tokens: Vec
, + /// Per-token balances decoded from `getPoolTokens` (discover mode only), used + /// by the Verify-phase probe to locate each token's cash field. Empty in + /// verify-only mode (no `getPoolTokens` call). + balances: Vec, + /// Per-token cash-field locations: rebuilt by the discover-phase probe, or + /// carried from config-supplied metadata in verify-only mode. + token_cash: Vec, /// The vault balance slots discovered in round 1 and verified in round 2. verified_slots: Vec<(Address, U256)>, /// Slots injected across the run (the refreshed balances). @@ -277,6 +508,7 @@ impl BalancerV2ColdStartPlanner { pool_id: B256, known_slots: Vec, tokens: Vec
, + known_token_cash: Vec, policy: ColdStartPolicy, ) -> Self { // A pre-populated balance read-set selects the verify-only fast path: @@ -304,6 +536,11 @@ impl BalancerV2ColdStartPlanner { // verify-only mode preserves the config-supplied tokens (they came // from the prior discovery that produced the known read-set). tokens, + // Discover mode fills these from the getPoolTokens decode + probe; + // verify-only mode has no decode, so `balances` stays empty (probe + // skipped) and the carried `token_cash` is preserved. + balances: Vec::new(), + token_cash: known_token_cash, verified_slots, changed_slots: Vec::new(), repair: None, @@ -346,7 +583,7 @@ impl AdapterColdStartPlanner for BalancerV2ColdStartPlanner { } } - fn on_results(&mut self, results: &ColdStartResults, _state: &dyn StateView) -> ColdStartStep { + fn on_results(&mut self, results: &ColdStartResults, state: &dyn StateView) -> ColdStartStep { // Record any slots injected this round (round 2's refreshed balances). self.changed_slots.extend(results.verified.iter().cloned()); @@ -374,7 +611,12 @@ impl AdapterColdStartPlanner for BalancerV2ColdStartPlanner { return ColdStartStep::Done; }; match getPoolTokensCall::abi_decode_returns_validate(output) { - Ok(decoded) => self.tokens = decoded.tokens, + Ok(decoded) => { + self.tokens = decoded.tokens; + // Keep the balances to probe each token's cash slot/offset + // once the verify round warms the discovered slots. + self.balances = decoded.balances; + } Err(_) => { self.repair = Some(BalancerRepair::DiscoverFailed); return ColdStartStep::Done; @@ -426,6 +668,19 @@ impl AdapterColdStartPlanner for BalancerV2ColdStartPlanner { if any_unfetched { self.repair = Some(BalancerRepair::BalancesUnfetched); } + // Discover-origin verify: the slots are now warm, so probe them to + // locate each token's `cash` field by value — this builds the map + // the reactive `Swap` path event-sources from. Verify-only mode has + // no fresh `balances`, so the carried `token_cash` is left intact. + if !self.balances.is_empty() { + self.token_cash = probe_token_cash( + &self.tokens, + &self.balances, + &self.verified_slots, + self.vault, + state, + ); + } ColdStartStep::Done } } @@ -493,6 +748,10 @@ impl AdapterColdStartPlanner for BalancerV2ColdStartPlanner { pool_address: Some(pool_address), tokens: self.tokens.clone(), balance_slots, + // The probed per-token cash locations (discover mode) or the + // carried map (verify-only mode) — drives reactive `Swap` + // event-sourcing. + token_cash: self.token_cash.clone(), }); pool.status = PoolStatus::Ready; report.status = PoolStatus::Ready; @@ -501,3 +760,239 @@ impl AdapterColdStartPlanner for BalancerV2ColdStartPlanner { } } } + +#[cfg(test)] +mod tests { + use super::*; + use std::collections::HashMap; + + /// Minimal `StateView` over an in-memory `(address, slot) -> value` map. + struct MapView(HashMap<(Address, U256), U256>); + impl StateView for MapView { + fn storage(&self, address: Address, slot: U256) -> Option { + self.0.get(&(address, slot)).copied() + } + } + + fn addr(byte: u8) -> Address { + Address::repeat_byte(byte) + } + + /// A single-token `BalanceAllocation`: `[block:32][managed:112][cash:112]`. + fn packed(block: u64, managed: u128, cash: u128) -> U256 { + (U256::from(block) << 224) | (U256::from(managed) << 112) | U256::from(cash) + } + + /// A TWO_TOKEN shared cash slot: `[block:32][cash_high:112][cash_low:112]`. + fn packed_two(block: u64, cash_high: u128, cash_low: u128) -> U256 { + (U256::from(block) << 224) | (U256::from(cash_high) << 112) | U256::from(cash_low) + } + + #[test] + fn cash_field_round_trips_both_offsets_preserving_others() { + let word = packed(0x1234, 0xAAAA, 0xBBBB); + assert_eq!(cash_field(word, false), U256::from(0xBBBB_u64)); + assert_eq!(cash_field(word, true), U256::from(0xAAAA_u64)); + + let low = set_cash_field(word, false, U256::from(0xCCCC_u64)); + assert_eq!(cash_field(low, false), U256::from(0xCCCC_u64)); + assert_eq!(cash_field(low, true), U256::from(0xAAAA_u64)); + assert_eq!(low >> 224, U256::from(0x1234_u64)); + + let high = set_cash_field(word, true, U256::from(0xDDDD_u64)); + assert_eq!(cash_field(high, true), U256::from(0xDDDD_u64)); + assert_eq!(cash_field(high, false), U256::from(0xBBBB_u64)); + assert_eq!(high >> 224, U256::from(0x1234_u64)); + } + + #[test] + fn apply_cash_delta_adds_subtracts_and_bounds() { + let word = packed(9, 0, 100); + assert_eq!( + cash_field( + apply_cash_delta(word, false, true, U256::from(5_u64)).unwrap(), + false + ), + U256::from(105_u64) + ); + assert_eq!( + cash_field( + apply_cash_delta(word, false, false, U256::from(40_u64)).unwrap(), + false + ), + U256::from(60_u64) + ); + // Underflow (burn more than cash) and 112-bit overflow both reject. + assert!(apply_cash_delta(word, false, false, U256::from(101_u64)).is_none()); + let full = set_cash_field(U256::ZERO, false, cash_mask()); + assert!(apply_cash_delta(full, false, true, U256::from(1_u64)).is_none()); + } + + #[test] + fn probe_locates_two_token_shared_slot() { + let vault = addr(0xba); + let (t0, t1) = (addr(0x01), addr(0x02)); + let slot = U256::from(0x77_u64); + let mut m = HashMap::new(); + m.insert((vault, slot), packed_two(5, 222, 111)); // low=t0=111, high=t1=222 + let view = MapView(m); + + let cash = probe_token_cash( + &[t0, t1], + &[U256::from(111_u64), U256::from(222_u64)], + &[(vault, slot)], + vault, + &view, + ); + assert_eq!(cash.len(), 2); + assert!(cash.contains(&BalancerTokenBalance::new(t0, slot, false))); + assert!(cash.contains(&BalancerTokenBalance::new(t1, slot, true))); + } + + #[test] + fn probe_locates_per_token_slots_and_skips_ambiguous_and_zero() { + let vault = addr(0xba); + let (t0, t1, t2, t3) = (addr(0x01), addr(0x02), addr(0x03), addr(0x04)); + let (s0, s1, s2, s3) = ( + U256::from(1_u64), + U256::from(2_u64), + U256::from(3_u64), + U256::from(4_u64), + ); + let mut m = HashMap::new(); + m.insert((vault, s0), packed(9, 0, 1000)); // t0 + m.insert((vault, s1), packed(9, 0, 2000)); // t1 (unique) + m.insert((vault, s2), packed(9, 0, 1000)); // t2 collides with t0 + m.insert((vault, s3), packed(9, 0, 0)); // t3 empty + let view = MapView(m); + + let cash = probe_token_cash( + &[t0, t1, t2, t3], + &[ + U256::from(1000_u64), + U256::from(2000_u64), + U256::from(1000_u64), + U256::ZERO, + ], + &[(vault, s0), (vault, s1), (vault, s2), (vault, s3)], + vault, + &view, + ); + // Only t1 is unambiguous: t0/t2 share value 1000 (2 matches each), t3 is zero. + assert_eq!(cash, vec![BalancerTokenBalance::new(t1, s1, false)]); + } + + #[test] + fn event_source_swap_shared_slot_accumulates_both_fields() { + let vault = addr(0xba); + let (t_in, t_out) = (addr(0x01), addr(0x02)); + let slot = U256::from(0x77_u64); + let meta = BalancerV2Metadata::default() + .with_vault(vault) + .with_token_cash([ + BalancerTokenBalance::new(t_in, slot, false), + BalancerTokenBalance::new(t_out, slot, true), + ]); + let mut m = HashMap::new(); + m.insert((vault, slot), packed_two(5, 1000, 500)); // high(t_out)=1000, low(t_in)=500 + let view = MapView(m); + + let updates = event_source_swap( + &view, + vault, + &meta, + t_in, + U256::from(30_u64), + t_out, + U256::from(20_u64), + ) + .unwrap(); + assert_eq!(updates.len(), 1, "shared slot -> one combined write"); + let StateUpdate::Slot { value, .. } = &updates[0] else { + panic!("expected a Slot write"); + }; + assert_eq!(cash_field(*value, false), U256::from(530_u64)); // t_in + 30 + assert_eq!(cash_field(*value, true), U256::from(980_u64)); // t_out - 20 + assert_eq!(*value >> 224, U256::from(5_u64), "block preserved"); + } + + #[test] + fn event_source_swap_separate_slots() { + let vault = addr(0xba); + let (t_in, t_out) = (addr(0x01), addr(0x02)); + let (s_in, s_out) = (U256::from(1_u64), U256::from(2_u64)); + let meta = BalancerV2Metadata::default() + .with_vault(vault) + .with_token_cash([ + BalancerTokenBalance::new(t_in, s_in, false), + BalancerTokenBalance::new(t_out, s_out, false), + ]); + let mut m = HashMap::new(); + m.insert((vault, s_in), packed(9, 0, 100)); + m.insert((vault, s_out), packed(9, 0, 200)); + let view = MapView(m); + + let updates = event_source_swap( + &view, + vault, + &meta, + t_in, + U256::from(10_u64), + t_out, + U256::from(20_u64), + ) + .unwrap(); + assert_eq!(updates.len(), 2); + for update in updates { + let StateUpdate::Slot { slot, value, .. } = update else { + panic!("expected Slot writes"); + }; + if slot == s_in { + assert_eq!(cash_field(value, false), U256::from(110_u64)); // +10 + } else { + assert_eq!(cash_field(value, false), U256::from(180_u64)); // -20 + } + } + } + + #[test] + fn event_source_swap_unknown_token_or_cold_slot_falls_back() { + let vault = addr(0xba); + let (t_in, t_out) = (addr(0x01), addr(0x02)); + let slot = U256::from(0x77_u64); + // Unknown token: empty token_cash -> None. + let bare = BalancerV2Metadata::default().with_vault(vault); + let view = MapView(HashMap::new()); + assert!( + event_source_swap( + &view, + vault, + &bare, + t_in, + U256::from(1_u64), + t_out, + U256::from(1_u64) + ) + .is_none() + ); + // Known token but cold slot (not in the view) -> None. + let mapped = BalancerV2Metadata::default() + .with_vault(vault) + .with_token_cash([ + BalancerTokenBalance::new(t_in, slot, false), + BalancerTokenBalance::new(t_out, slot, true), + ]); + assert!( + event_source_swap( + &view, + vault, + &mapped, + t_in, + U256::from(1_u64), + t_out, + U256::from(1_u64) + ) + .is_none() + ); + } +} diff --git a/src/adapters/mod.rs b/src/adapters/mod.rs index b94e01b..533f9cd 100644 --- a/src/adapters/mod.rs +++ b/src/adapters/mod.rs @@ -88,10 +88,10 @@ pub use sync_manager::{AmmSyncBatchReport, AmmSyncEngine, AmmSyncError}; pub use traits::AmmAdapter; pub use types::{ AdapterEvent, AdapterEventError, AdapterEventKind, AdapterEventReport, AdapterEventResult, - BalancerV2Metadata, ColdStartOutcome, ColdStartPolicy, ColdStartReport, CurveMetadata, - CurveVariant, CustomPoolKey, DeferredOutcome, DeferredWork, EventRoute, EventSource, PoolKey, - PoolRegistration, PoolStatus, ProtocolId, ProtocolMetadata, RepairAction, SolidlyV2Metadata, - UniswapV2Metadata, UnsupportedReason, UpdateQuality, V3Metadata, + BalancerTokenBalance, BalancerV2Metadata, ColdStartOutcome, ColdStartPolicy, ColdStartReport, + CurveMetadata, CurveVariant, CustomPoolKey, DeferredOutcome, DeferredWork, EventRoute, + EventSource, PoolKey, PoolRegistration, PoolStatus, ProtocolId, ProtocolMetadata, RepairAction, + SolidlyV2Metadata, UniswapV2Metadata, UnsupportedReason, UpdateQuality, V3Metadata, }; #[cfg(feature = "balancer-v2")] diff --git a/src/adapters/types.rs b/src/adapters/types.rs index 86799e8..c44adad 100644 --- a/src/adapters/types.rs +++ b/src/adapters/types.rs @@ -549,6 +549,18 @@ pub struct BalancerV2Metadata { /// balance-mapping layout or doing lossy event-delta arithmetic. Empty /// until the discover→verify cold-start runs. pub balance_slots: Vec, + /// Per-token vault `cash`-balance locations (see [`BalancerTokenBalance`]), + /// derived by the discover cold-start's probe. + /// + /// Lets the reactive `Swap` path **event-source** the exact `cash` delta with + /// no RPC — writing each swapped token's packed balance directly from the + /// event's `amountIn`/`amountOut` — falling back to a [`balance_slots`] resync + /// when a token is absent here (a slots-only pre-population, or a managed- + /// balance pool where `cash != getPoolTokens` balance). Empty until a discover + /// cold-start builds it. + /// + /// [`balance_slots`]: Self::balance_slots + pub token_cash: Vec, } impl BalancerV2Metadata { @@ -575,6 +587,49 @@ impl BalancerV2Metadata { self.balance_slots = balance_slots.into_iter().collect(); self } + + /// Set (replace) the per-token vault `cash`-balance locations. + pub fn with_token_cash( + mut self, + token_cash: impl IntoIterator, + ) -> Self { + self.token_cash = token_cash.into_iter().collect(); + self + } +} + +/// Location of one token's packed `cash` balance in the Balancer V2 vault storage. +/// +/// Vault balances are a packed `bytes32` +/// (`[lastChangeBlock : top 32][managed : bits 112–223][cash : bits 0–111]`); a +/// swap changes only the 112-bit `cash` field. This records where a given token's +/// `cash` lives so the reactive `Swap` path can write it directly. For a TWO_TOKEN +/// pool both tokens share one slot — one at the low field, one at the high field — +/// so `slot` can repeat across two entries with different [`high_field`]. +/// +/// [`high_field`]: Self::high_field +#[non_exhaustive] +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct BalancerTokenBalance { + /// The token whose vault `cash` balance this locates. + pub token: Address, + /// The vault storage slot holding the packed balance. + pub slot: U256, + /// Whether `cash` is the **high** 112-bit field (bits 112–223) of `slot` + /// rather than the low field (bits 0–111). `true` only for the second token of + /// a TWO_TOKEN pool's shared slot. + pub high_field: bool, +} + +impl BalancerTokenBalance { + /// Construct a token cash-balance location. + pub fn new(token: Address, slot: U256, high_field: bool) -> Self { + Self { + token, + slot, + high_field, + } + } } /// Which Curve pool dialect a pool speaks — selects the `get_dy` / `TokenExchange` diff --git a/tests/adapter_reactive.rs b/tests/adapter_reactive.rs index ed99cb8..c32c375 100644 --- a/tests/adapter_reactive.rs +++ b/tests/adapter_reactive.rs @@ -14,11 +14,11 @@ use evm_amm_state::adapters::storage::{ v3_tick_bitmap_storage_key_with_base, v3_tick_info_storage_keys_with_base, }; use evm_amm_state::adapters::{ - AdapterRegistry, AmmAdapter, AmmReactiveHandler, BalancerV2Adapter, BalancerV2Metadata, - ColdStartOutcome, ColdStartPolicy, ConcentratedLiquidityAdapter, CurveAdapter, CurveMetadata, - CurveVariant, DeferredWork, PoolKey, PoolRegistration, PoolStatus, ProtocolMetadata, - SolidlyV2Adapter, SolidlyV2Metadata, UniswapV2Adapter, UniswapV2Metadata, V3Metadata, - uniswap_v2_pair_runtime_code_hash, + AdapterRegistry, AmmAdapter, AmmReactiveHandler, BalancerTokenBalance, BalancerV2Adapter, + BalancerV2Metadata, ColdStartOutcome, ColdStartPolicy, ConcentratedLiquidityAdapter, + CurveAdapter, CurveMetadata, CurveVariant, DeferredWork, PoolKey, PoolRegistration, PoolStatus, + ProtocolMetadata, SolidlyV2Adapter, SolidlyV2Metadata, UniswapV2Adapter, UniswapV2Metadata, + V3Metadata, uniswap_v2_pair_runtime_code_hash, }; // The reactive-runtime seam these tests exercise (raw `EvmCache::apply_updates` // and the upstream `ResyncedReport`/`InvalidationRequest`) speaks the upstream @@ -812,6 +812,298 @@ async fn balancer_shared_emitter_routes_by_pool_id() -> Result<()> { Ok(()) } +/// Register a Balancer pool with a fully-probed `token_cash` map so the reactive +/// `Swap` path can event-source the vault balances directly. +fn balancer_registry_with_metadata( + vault: Address, + pool_id: B256, + metadata: BalancerV2Metadata, +) -> AdapterRegistry { + let adapter = Arc::new(BalancerV2Adapter::default()); + let mut registry = AdapterRegistry::new(); + registry.register_adapter(adapter.clone()).unwrap(); + let mut registration = PoolRegistration::new(PoolKey::BalancerV2(pool_id)) + .with_state_address(vault) + .with_metadata(ProtocolMetadata::BalancerV2(metadata)); + let sources = adapter.event_sources(®istration); + registration = registration.with_event_sources(sources); + registry.register_pool(registration).unwrap(); + registry +} + +/// A warm `TWO_TOKEN` swap event-sources both balances in the single shared vault +/// slot: `tokenIn`'s low field gains `amountIn`, `tokenOut`'s high field loses +/// `amountOut`, in one combined write — exactly, with no resync. +#[tokio::test] +async fn balancer_swap_event_sources_two_token_shared_slot() -> Result<()> { + let vault = Address::repeat_byte(0x52); + let pool_id = B256::repeat_byte(0xc3); + let token_in = Address::repeat_byte(0x01); + let token_out = Address::repeat_byte(0x02); + let slot = U256::from(0x77_u64); + // Shared word: [lastChangeBlock:32][cash_high(out)=1000][cash_low(in)=500]. + let block_stamp = U256::from(0xABCD_u64); + let warm = (block_stamp << 224) | (U256::from(1000_u64) << 112) | U256::from(500_u64); + + let metadata = BalancerV2Metadata::default() + .with_vault(vault) + .with_tokens([token_in, token_out]) + .with_balance_slots([slot]) + .with_token_cash([ + BalancerTokenBalance::new(token_in, slot, false), + BalancerTokenBalance::new(token_out, slot, true), + ]); + + let log = rpc_log( + vault, + vec![ + balancer_swap_topic(), + pool_id, + topic_address(token_in), + topic_address(token_out), + ], + abi_words([U256::from(30_u64), U256::from(20_u64)]), + 20, + 0, + 0, + ); + + let mut cache = setup_cache().await?; + cache.apply_updates(&[StateUpdate::slot(vault, slot, warm)]); + + let mut runtime = ReactiveRuntime::::new(ReactiveConfig::default()); + runtime.register_handler(Arc::new(AmmReactiveHandler::new( + balancer_registry_with_metadata(vault, pool_id, metadata), + )))?; + + let report = runtime.ingest_batch( + &mut cache, + batch(vec![(ReactiveInput::Log(log), included_context(20, 0))]), + )?; + + let raw = cache.cached_storage_value(vault, slot).unwrap(); + assert_eq!( + raw & low_mask(112), + U256::from(530_u64), + "tokenIn cash += amountIn" + ); + assert_eq!( + (raw >> 112) & low_mask(112), + U256::from(980_u64), + "tokenOut cash -= amountOut" + ); + assert_eq!(raw >> 224, block_stamp, "lastChangeBlock is preserved"); + assert_eq!(report.applied.len(), 1); + assert_eq!( + report.applied[0].quality, + StateEffectQuality::ExactFromInput + ); + assert!( + report.applied[0].resyncs.is_empty(), + "a warm event-source must not schedule a resync" + ); + Ok(()) +} + +/// A warm `GENERAL` swap event-sources two distinct per-token vault slots: one +/// `+amountIn` write and one `-amountOut` write, exactly, with no resync. +#[tokio::test] +async fn balancer_swap_event_sources_general_separate_slots() -> Result<()> { + let vault = Address::repeat_byte(0x53); + let pool_id = B256::repeat_byte(0xd4); + let token_in = Address::repeat_byte(0x03); + let token_out = Address::repeat_byte(0x04); + let slot_in = U256::from(0x11_u64); + let slot_out = U256::from(0x22_u64); + let stamp = U256::from(7_u64) << 224; + + let metadata = BalancerV2Metadata::default() + .with_vault(vault) + .with_tokens([token_in, token_out]) + .with_balance_slots([slot_in, slot_out]) + .with_token_cash([ + BalancerTokenBalance::new(token_in, slot_in, false), + BalancerTokenBalance::new(token_out, slot_out, false), + ]); + + let log = rpc_log( + vault, + vec![ + balancer_swap_topic(), + pool_id, + topic_address(token_in), + topic_address(token_out), + ], + abi_words([U256::from(10_u64), U256::from(20_u64)]), + 21, + 0, + 0, + ); + + let mut cache = setup_cache().await?; + cache.apply_updates(&[ + StateUpdate::slot(vault, slot_in, stamp | U256::from(100_u64)), + StateUpdate::slot(vault, slot_out, stamp | U256::from(200_u64)), + ]); + + let mut runtime = ReactiveRuntime::::new(ReactiveConfig::default()); + runtime.register_handler(Arc::new(AmmReactiveHandler::new( + balancer_registry_with_metadata(vault, pool_id, metadata), + )))?; + + let report = runtime.ingest_batch( + &mut cache, + batch(vec![(ReactiveInput::Log(log), included_context(21, 0))]), + )?; + + let raw_in = cache.cached_storage_value(vault, slot_in).unwrap(); + let raw_out = cache.cached_storage_value(vault, slot_out).unwrap(); + assert_eq!( + raw_in & low_mask(112), + U256::from(110_u64), + "tokenIn cash += amountIn" + ); + assert_eq!( + raw_out & low_mask(112), + U256::from(180_u64), + "tokenOut cash -= amountOut" + ); + assert_eq!(raw_in >> 224, U256::from(7_u64), "in-slot block preserved"); + assert_eq!(report.applied.len(), 1); + assert_eq!( + report.applied[0].quality, + StateEffectQuality::ExactFromInput + ); + assert!(report.applied[0].resyncs.is_empty()); + Ok(()) +} + +/// A `Swap` on a pool whose `token_cash` is unknown (never probed) cannot be +/// event-sourced, so it falls back to a `VerifySlots` resync of the known balance +/// slots rather than writing the wrong storage. +#[tokio::test] +async fn balancer_swap_without_token_cash_falls_back_to_resync() -> Result<()> { + let vault = Address::repeat_byte(0x54); + let pool_id = B256::repeat_byte(0xe5); + let slot = U256::from(0x33_u64); + let metadata = BalancerV2Metadata::default() + .with_vault(vault) + .with_balance_slots([slot]); // slots known, but no token_cash map. + + let log = rpc_log( + vault, + vec![ + balancer_swap_topic(), + pool_id, + topic_address(Address::repeat_byte(0x05)), + topic_address(Address::repeat_byte(0x06)), + ], + abi_words([U256::from(10_u64), U256::from(20_u64)]), + 22, + 0, + 0, + ); + + let mut cache = setup_cache().await?; + let mut runtime = ReactiveRuntime::::new(ReactiveConfig::default()); + runtime.register_handler(Arc::new(AmmReactiveHandler::new( + balancer_registry_with_metadata(vault, pool_id, metadata), + )))?; + + let report = runtime.ingest_batch( + &mut cache, + batch(vec![(ReactiveInput::Log(log), included_context(22, 0))]), + )?; + + assert_eq!(report.applied.len(), 1); + assert_eq!( + report.applied[0].resyncs.len(), + 1, + "unmapped swap must resync" + ); + assert!(matches!( + report.applied[0].resyncs[0].targets.as_slice(), + [ResyncTarget::StorageSlots { address, slots }] + if *address == vault && slots == &vec![slot] + )); + Ok(()) +} + +/// A `PoolBalanceChanged` (join/exit) is now subscribed: it routes to the adapter, +/// is classified by delta sign, and resyncs the vault balance slots (event-sourcing +/// its deltas is a follow-up). Validates the new subscription + routing. +#[tokio::test] +async fn balancer_pool_balance_changed_routes_and_resyncs() -> Result<()> { + let vault = Address::repeat_byte(0x55); + let pool_id = B256::repeat_byte(0xf6); + let token = Address::repeat_byte(0x07); + let slot = U256::from(0x44_u64); + let metadata = BalancerV2Metadata::default() + .with_vault(vault) + .with_tokens([token]) + .with_balance_slots([slot]); + + // ABI body for (address[1], int256[1] = [+500], uint256[1] = [0]): three + // dynamic arrays, so three head offsets then each array's [len, element]. + let body: Vec = [ + word(U256::from(0x60_u64)), + word(U256::from(0xa0_u64)), + word(U256::from(0xe0_u64)), + word(U256::from(1_u64)), + address_word(token), + word(U256::from(1_u64)), + word(U256::from(500_u64)), // positive delta -> LiquidityAdded + word(U256::from(1_u64)), + word(U256::ZERO), + ] + .concat(); + + let log = rpc_log( + vault, + vec![ + keccak256("PoolBalanceChanged(bytes32,address,address[],int256[],uint256[])"), + pool_id, + topic_address(Address::repeat_byte(0x99)), // liquidityProvider + ], + body, + 23, + 0, + 0, + ); + + let mut cache = setup_cache().await?; + let mut runtime = ReactiveRuntime::::new(ReactiveConfig::default()); + runtime.register_handler(Arc::new(AmmReactiveHandler::new( + balancer_registry_with_metadata(vault, pool_id, metadata), + )))?; + + let report = runtime.ingest_batch( + &mut cache, + batch(vec![(ReactiveInput::Log(log), included_context(23, 0))]), + )?; + + assert_eq!( + report.applied.len(), + 1, + "PoolBalanceChanged must route to the adapter" + ); + assert_eq!( + tag_value(&report.applied[0].tags, "protocol"), + Some("BalancerV2") + ); + assert_eq!( + report.applied[0].resyncs.len(), + 1, + "join/exit must resync the balances" + ); + assert!(matches!( + report.applied[0].resyncs[0].targets.as_slice(), + [ResyncTarget::StorageSlots { address, slots }] + if *address == vault && slots == &vec![slot] + )); + Ok(()) +} + #[tokio::test] async fn removed_log_rolls_back_previously_applied_update() -> Result<()> { let pool = Address::repeat_byte(0x61); diff --git a/tests/balancer_liquidity_rpc.rs b/tests/balancer_liquidity_rpc.rs new file mode 100644 index 0000000..3928227 --- /dev/null +++ b/tests/balancer_liquidity_rpc.rs @@ -0,0 +1,282 @@ +//! Live RPC parity for **event-sourced** Balancer V2 swaps (env-gated, +//! `#[ignore]`). +//! +//! For a real vault `Swap` on both a `TWO_TOKEN` pool (a single shared cash +//! slot, two 112-bit fields) and a `GENERAL` pool (per-token cash slots), this +//! fetches the transaction's *exact per-tx* storage diff via +//! `trace_replayTransaction(stateDiff)` — the on-chain ground truth, immune to +//! other transactions in the same block — derives each swapped token's cash +//! field location from that diff, warms the pre-tx (`from`) state into a cache, +//! applies the real `Swap` log through the adapter, and asserts the adapter's +//! event-sourced writes reproduce the on-chain post-tx (`to`) `cash` field for +//! both `tokenIn` (`+amountIn`) and `tokenOut` (`-amountOut`). +//! +//! The `lastChangeBlock` field (top 32 bits) is deliberately *not* compared: the +//! event-sourced write maintains only `cash`, leaving the block stamp as warmed, +//! whereas the on-chain word also bumps the block. Quotes read `cash`, not the +//! stamp, so field-level parity is the correctness bar. +//! +//! Run: `E2E_RPC_URL= cargo test --test balancer_liquidity_rpc -- --ignored --nocapture` + +use std::collections::HashMap; +use std::sync::Arc; + +use alloy_eips::BlockId; +use alloy_primitives::{Address, B256, U256, address}; +use alloy_provider::{Provider, RootProvider, network::AnyNetwork}; +use alloy_rpc_types_eth::Filter; +use anyhow::Result; +use evm_amm_state::adapters::driver::AdapterDriver; +use evm_amm_state::adapters::{ + AdapterCache, AdapterRegistry, AmmAdapter, BalancerTokenBalance, BalancerV2Adapter, + BalancerV2Metadata, PoolKey, PoolRegistration, ProtocolMetadata, StateUpdate, +}; +use evm_fork_cache::cache::EvmCache; + +const VAULT: Address = address!("BA12222222228d8Ba445958a75a0704d566BF2C8"); + +// A `TWO_TOKEN` pool (80BAL/20WETH): both token balances share one vault slot, +// packed `[block:32][cash_high:112][cash_low:112]`. +const TWO_TOKEN_POOL: &str = "0x5c6ee304399dbdb9c8ef030ab642b10820db8f56000200000000000000000014"; +const TWO_TOKEN_SWAP_TX: &str = + "0x13e54562bc37d6d86a1f8a2840ff28821ff63d258f4fe8ce90464756ca44804c"; + +// A `GENERAL` pool (Balancer stable USDT/DAI/USDC): each token's balance lives in +// its own vault slot (low 112-bit `cash` field). +const GENERAL_POOL: &str = "0x06df3b2bbb68adc8b0e302443692037ed9f91b42000000000000000000000063"; +const GENERAL_SWAP_TX: &str = "0xa109fec4cfc1c3336151594364afc71474f480f19d171eb031ce0ba47561f5e9"; + +const CASH_BITS: usize = 112; + +fn swap_topic() -> B256 { + alloy_primitives::keccak256("Swap(bytes32,address,address,uint256,uint256)") +} + +fn parse_u256(hex: &str) -> U256 { + U256::from_str_radix(hex.trim_start_matches("0x"), 16).unwrap_or(U256::ZERO) +} + +/// Extract a packed vault balance's 112-bit `cash` field (low = bits 0–111, high +/// = bits 112–223). Mirrors the adapter's crate-internal `cash_field`. +fn cash_field(word: U256, high: bool) -> U256 { + let mask = (U256::from(1) << CASH_BITS) - U256::from(1); + let shift = if high { CASH_BITS } else { 0 }; + (word >> shift) & mask +} + +/// Fetch the per-tx storage diff for `VAULT`: slot -> (from, to). +async fn vault_state_diff( + provider: &RootProvider, + tx: &str, +) -> Result> { + let value: serde_json::Value = provider + .raw_request( + "trace_replayTransaction".into(), + (tx.to_string(), vec!["stateDiff".to_string()]), + ) + .await?; + + let vault_lc = format!("{VAULT:?}").to_lowercase(); + let mut diff = HashMap::new(); + let Some(accounts) = value["stateDiff"].as_object() else { + anyhow::bail!("trace_replayTransaction returned no stateDiff (trace unsupported?)"); + }; + for (addr, account) in accounts { + if addr.to_lowercase() != vault_lc { + continue; + } + if let Some(storage) = account["storage"].as_object() { + for (slot_hex, change) in storage { + let slot = parse_u256(slot_hex); + if let Some(star) = change.get("*") { + diff.insert( + slot, + ( + parse_u256(star["from"].as_str().unwrap_or("0x0")), + parse_u256(star["to"].as_str().unwrap_or("0x0")), + ), + ); + } else if let Some(plus) = change.get("+") { + diff.insert( + slot, + (U256::ZERO, parse_u256(plus.as_str().unwrap_or("0x0"))), + ); + } + } + } + } + Ok(diff) +} + +/// Block number containing `tx` (via `eth_getTransactionByHash`). +async fn tx_block(provider: &RootProvider, tx: &str) -> Result { + let value: serde_json::Value = provider + .raw_request("eth_getTransactionByHash".into(), (tx.to_string(),)) + .await?; + let hex = value["blockNumber"] + .as_str() + .ok_or_else(|| anyhow::anyhow!("tx {tx} has no blockNumber (not mined?)"))?; + Ok(u64::from_str_radix(hex.trim_start_matches("0x"), 16)?) +} + +/// Locate `token`'s cash field: the `(slot, high)` in `diff` whose 112-bit field +/// changed by exactly `signed_delta` (`+amount` for `tokenIn`, `-amount` for +/// `tokenOut`). Returns `None` if no field matches — a mis-pin or a fee-skimming +/// pool would fail here rather than silently pass. +fn locate_field( + diff: &HashMap, + amount: U256, + is_in: bool, +) -> Option<(U256, bool)> { + for (&slot, &(from, to)) in diff { + for high in [false, true] { + let (before, after) = (cash_field(from, high), cash_field(to, high)); + let matches = if is_in { + after > before && after - before == amount + } else { + before > after && before - after == amount + }; + if matches { + return Some((slot, high)); + } + } + } + None +} + +async fn run_swap_parity(label: &str, pool_id_hex: &str, tx: &str) -> Result<()> { + let Ok(url) = std::env::var("E2E_RPC_URL") else { + eprintln!("E2E_RPC_URL unset — skipping Balancer swap RPC parity test."); + return Ok(()); + }; + + let provider = Arc::new(RootProvider::::connect(&url).await?); + let pool_id: B256 = pool_id_hex.parse()?; + let tx_hash: B256 = tx.parse()?; + let block = tx_block(&provider, tx).await?; + let diff = vault_state_diff(&provider, tx).await?; + anyhow::ensure!(!diff.is_empty(), "no vault storage diff for {label} swap"); + + // Fetch the real Swap log (its exact topics + data) rather than + // reconstructing it, so the test is immune to hand-copied hex errors. A + // multi-hop tx emits one Swap per hop; select ours by poolId (topic1). + let filter = Filter::new() + .address(VAULT) + .event_signature(swap_topic()) + .from_block(block) + .to_block(block); + let log = provider + .get_logs(&filter) + .await? + .into_iter() + .find(|entry| { + entry.transaction_hash == Some(tx_hash) + && entry.topic0() == Some(&swap_topic()) + && entry.inner.data.topics().get(1) == Some(&pool_id) + }) + .expect("Swap log for this pool not found in block") + .inner; + + let topics = log.data.topics(); + let token_in = Address::from_word(topics[2]); + let token_out = Address::from_word(topics[3]); + let amount_in = U256::from_be_slice(&log.data.data[0..32]); + let amount_out = U256::from_be_slice(&log.data.data[32..64]); + + // Derive each token's cash-field location from the ground-truth per-tx diff: + // tokenIn's field rose by +amountIn, tokenOut's fell by amountOut. + let (in_slot, in_high) = locate_field(&diff, amount_in, true) + .expect("tokenIn cash field (+amountIn) not found in the tx diff — mis-pin?"); + let (out_slot, out_high) = locate_field(&diff, amount_out, false) + .expect("tokenOut cash field (-amountOut) not found in the tx diff — mis-pin?"); + eprintln!( + "{label}: in={token_in} (+{amount_in}) @ {in_slot:#x}[{}] out={token_out} (-{amount_out}) @ {out_slot:#x}[{}]", + if in_high { "high" } else { "low" }, + if out_high { "high" } else { "low" } + ); + if pool_id_hex == TWO_TOKEN_POOL { + assert_eq!( + in_slot, out_slot, + "TWO_TOKEN pool: both tokens share one slot" + ); + } else { + assert_ne!(in_slot, out_slot, "GENERAL pool: tokens use distinct slots"); + } + + // Metadata a real cold-start would have recorded: the vault, the probed + // token->cash map, and the balance slots (for the resync fallback). + let mut balance_slots = vec![in_slot, out_slot]; + balance_slots.dedup(); + let metadata = BalancerV2Metadata::default() + .with_vault(VAULT) + .with_tokens([token_in, token_out]) + .with_balance_slots(balance_slots) + .with_token_cash([ + BalancerTokenBalance::new(token_in, in_slot, in_high), + BalancerTokenBalance::new(token_out, out_slot, out_high), + ]); + + // Warm every changed vault slot with its exact per-tx `from` value (pre-swap + // state), so the read-modify-write sees the true pre-image. + let mut cache = EvmCache::at_block(provider.clone(), BlockId::number(block - 1)).await; + let warm: Vec = diff + .iter() + .map(|(&slot, &(from, _))| StateUpdate::slot(VAULT, slot, from)) + .collect(); + AdapterCache::apply_updates(&mut cache, &warm); + + // Register the pool + adapter and apply the real Swap through the driver. + let adapter = Arc::new(BalancerV2Adapter::default()); + let mut registration = PoolRegistration::new(PoolKey::BalancerV2(pool_id)) + .with_state_address(VAULT) + .with_metadata(ProtocolMetadata::BalancerV2(metadata)); + let sources = adapter.event_sources(®istration); + registration = registration.with_event_sources(sources); + + let mut registry = AdapterRegistry::new(); + registry.register_adapter(adapter).unwrap(); + registry.register_pool(registration).unwrap(); + let driver = AdapterDriver::new(registry); + + driver + .apply_log(&mut cache, &log)? + .expect("Swap must route and apply"); + + // The event-sourced `cash` fields must equal the on-chain post-tx values. + let (_, in_to) = diff[&in_slot]; + let (_, out_to) = diff[&out_slot]; + let in_word = cache + .cached_storage_value(VAULT, in_slot) + .expect("in slot cached"); + let out_word = cache + .cached_storage_value(VAULT, out_slot) + .expect("out slot cached"); + assert_eq!( + cash_field(in_word, in_high), + cash_field(in_to, in_high), + "{label}: tokenIn cash != on-chain post-tx (expected +{amount_in})" + ); + assert_eq!( + cash_field(out_word, out_high), + cash_field(out_to, out_high), + "{label}: tokenOut cash != on-chain post-tx (expected -{amount_out})" + ); + eprintln!( + "{label}: OK in cash -> {} out cash -> {}", + cash_field(in_to, in_high), + cash_field(out_to, out_high) + ); + Ok(()) +} + +#[tokio::test] +#[ignore = "requires an archive+trace RPC via E2E_RPC_URL; run with --ignored"] +async fn two_token_swap_event_sourcing_matches_onchain() -> Result<()> { + run_swap_parity("TWO_TOKEN", TWO_TOKEN_POOL, TWO_TOKEN_SWAP_TX).await +} + +#[tokio::test] +#[ignore = "requires an archive+trace RPC via E2E_RPC_URL; run with --ignored"] +async fn general_swap_event_sourcing_matches_onchain() -> Result<()> { + run_swap_parity("GENERAL", GENERAL_POOL, GENERAL_SWAP_TX).await +} From 50cb65bec1e5f595f9e26423a0f589d14628ea34 Mon Sep 17 00:00:00 2001 From: Kai Aldag Date: Tue, 7 Jul 2026 17:31:10 +0100 Subject: [PATCH 2/2] fix(balancer): guard cash field event sourcing --- src/adapters/balancer_v2.rs | 186 ++++++++++++++++++++++++++++-------- tests/adapter_reactive.rs | 10 +- 2 files changed, 156 insertions(+), 40 deletions(-) diff --git a/src/adapters/balancer_v2.rs b/src/adapters/balancer_v2.rs index 13b9088..5e6857d 100644 --- a/src/adapters/balancer_v2.rs +++ b/src/adapters/balancer_v2.rs @@ -32,6 +32,16 @@ fn cash_mask() -> U256 { (U256::from(1) << CASH_BITS) - U256::from(1) } +/// Balancer V2 pool specialization encoded in the poolId after the 20-byte pool +/// address. `2` is the only specialization where the high 112-bit balance field +/// is another token's `cash`; GENERAL/MINIMAL pools use it for `managed`. +const TWO_TOKEN_SPECIALIZATION: u16 = 2; + +fn is_two_token_pool(pool_id: B256) -> bool { + let bytes = pool_id.as_slice(); + u16::from_be_bytes([bytes[20], bytes[21]]) == TWO_TOKEN_SPECIALIZATION +} + /// Extract the 112-bit `cash` field at the low (bits 0–111) or high (bits /// 112–223) position of a packed vault balance word. fn cash_field(word: U256, high: bool) -> U256 { @@ -69,14 +79,17 @@ fn apply_cash_delta(word: U256, high: bool, add: bool, amount: U256) -> Option Vec { let mut located = Vec::new(); for (token, balance) in tokens.iter().zip(balances.iter()) { @@ -92,7 +105,7 @@ fn probe_token_cash( if cash_field(word, false) == *balance { found.push((slot, false)); } - if cash_field(word, true) == *balance { + if high_field_can_be_cash && cash_field(word, true) == *balance { found.push((slot, true)); } found @@ -118,6 +131,14 @@ fn token_cash_location( .copied() } +#[derive(Clone, Copy)] +struct SwapCashDelta { + token_in: Address, + amount_in: U256, + token_out: Address, + amount_out: U256, +} + /// The resync repair + quality for the fallback path: re-verify the known /// `balance_slots`, or a conservative no-op when none are known yet. fn resync_repair(vault: Address, metadata: &BalancerV2Metadata) -> (RepairAction, UpdateQuality) { @@ -144,27 +165,32 @@ fn resync_repair(vault: Address, metadata: &BalancerV2Metadata) -> (RepairAction fn event_source_swap( view: &dyn StateView, vault: Address, + pool_id: B256, metadata: &BalancerV2Metadata, - token_in: Address, - amount_in: U256, - token_out: Address, - amount_out: U256, + swap: SwapCashDelta, ) -> Option> { - let in_loc = token_cash_location(metadata, token_in)?; - let out_loc = token_cash_location(metadata, token_out)?; + let in_loc = token_cash_location(metadata, swap.token_in)?; + let out_loc = token_cash_location(metadata, swap.token_out)?; + + if !is_two_token_pool(pool_id) && (in_loc.high_field || out_loc.high_field) { + return None; + } if in_loc.slot == out_loc.slot { + if in_loc.high_field == out_loc.high_field { + return None; + } // TWO_TOKEN shared slot: both fields live in one word, so apply both // deltas to a single write (two separate writes would clobber each other). let word = view.storage(vault, in_loc.slot)?; - let word = apply_cash_delta(word, in_loc.high_field, true, amount_in)?; - let word = apply_cash_delta(word, out_loc.high_field, false, amount_out)?; + let word = apply_cash_delta(word, in_loc.high_field, true, swap.amount_in)?; + let word = apply_cash_delta(word, out_loc.high_field, false, swap.amount_out)?; Some(vec![StateUpdate::slot(vault, in_loc.slot, word)]) } else { let word_in = view.storage(vault, in_loc.slot)?; - let word_in = apply_cash_delta(word_in, in_loc.high_field, true, amount_in)?; + let word_in = apply_cash_delta(word_in, in_loc.high_field, true, swap.amount_in)?; let word_out = view.storage(vault, out_loc.slot)?; - let word_out = apply_cash_delta(word_out, out_loc.high_field, false, amount_out)?; + let word_out = apply_cash_delta(word_out, out_loc.high_field, false, swap.amount_out)?; Some(vec![ StateUpdate::slot(vault, in_loc.slot, word_in), StateUpdate::slot(vault, out_loc.slot, word_out), @@ -186,15 +212,20 @@ fn decode_swap(pool: &PoolRegistration, log: &Log, view: &dyn StateView) -> Adap let (updates, repair, quality) = match &pool.metadata { ProtocolMetadata::BalancerV2(metadata) => match metadata.vault { - Some(vault) => match event_source_swap( - view, - vault, - metadata, - decoded.tokenIn, - decoded.amountIn, - decoded.tokenOut, - decoded.amountOut, - ) { + Some(vault) => match pool.key.bytes32().and_then(|pool_id| { + event_source_swap( + view, + vault, + pool_id, + metadata, + SwapCashDelta { + token_in: decoded.tokenIn, + amount_in: decoded.amountIn, + token_out: decoded.tokenOut, + amount_out: decoded.amountOut, + }, + ) + }) { Some(updates) => (updates, RepairAction::None, UpdateQuality::Exact), None => { let (repair, quality) = resync_repair(vault, metadata); @@ -679,6 +710,7 @@ impl AdapterColdStartPlanner for BalancerV2ColdStartPlanner { &self.verified_slots, self.vault, state, + is_two_token_pool(self.pool_id), ); } ColdStartStep::Done @@ -778,6 +810,26 @@ mod tests { Address::repeat_byte(byte) } + fn pool_id(specialization: u16) -> B256 { + let mut bytes = [0x11; 32]; + bytes[20..22].copy_from_slice(&specialization.to_be_bytes()); + B256::from(bytes) + } + + fn swap_delta( + token_in: Address, + amount_in: u64, + token_out: Address, + amount_out: u64, + ) -> SwapCashDelta { + SwapCashDelta { + token_in, + amount_in: U256::from(amount_in), + token_out, + amount_out: U256::from(amount_out), + } + } + /// A single-token `BalanceAllocation`: `[block:32][managed:112][cash:112]`. fn packed(block: u64, managed: u128, cash: u128) -> U256 { (U256::from(block) << 224) | (U256::from(managed) << 112) | U256::from(cash) @@ -843,6 +895,7 @@ mod tests { &[(vault, slot)], vault, &view, + true, ); assert_eq!(cash.len(), 2); assert!(cash.contains(&BalancerTokenBalance::new(t0, slot, false))); @@ -877,11 +930,32 @@ mod tests { &[(vault, s0), (vault, s1), (vault, s2), (vault, s3)], vault, &view, + false, ); // Only t1 is unambiguous: t0/t2 share value 1000 (2 matches each), t3 is zero. assert_eq!(cash, vec![BalancerTokenBalance::new(t1, s1, false)]); } + #[test] + fn probe_skips_high_managed_field_for_non_two_token_pools() { + let vault = addr(0xba); + let token = addr(0x01); + let slot = U256::from(0x77_u64); + let mut m = HashMap::new(); + m.insert((vault, slot), packed(9, 777, 111)); // high=managed, low=cash + let view = MapView(m); + + let cash = probe_token_cash( + &[token], + &[U256::from(777_u64)], + &[(vault, slot)], + vault, + &view, + false, + ); + assert!(cash.is_empty()); + } + #[test] fn event_source_swap_shared_slot_accumulates_both_fields() { let vault = addr(0xba); @@ -900,11 +974,9 @@ mod tests { let updates = event_source_swap( &view, vault, + pool_id(TWO_TOKEN_SPECIALIZATION), &meta, - t_in, - U256::from(30_u64), - t_out, - U256::from(20_u64), + swap_delta(t_in, 30, t_out, 20), ) .unwrap(); assert_eq!(updates.len(), 1, "shared slot -> one combined write"); @@ -935,11 +1007,9 @@ mod tests { let updates = event_source_swap( &view, vault, + pool_id(0), &meta, - t_in, - U256::from(10_u64), - t_out, - U256::from(20_u64), + swap_delta(t_in, 10, t_out, 20), ) .unwrap(); assert_eq!(updates.len(), 2); @@ -967,11 +1037,9 @@ mod tests { event_source_swap( &view, vault, + pool_id(TWO_TOKEN_SPECIALIZATION), &bare, - t_in, - U256::from(1_u64), - t_out, - U256::from(1_u64) + swap_delta(t_in, 1, t_out, 1) ) .is_none() ); @@ -986,11 +1054,53 @@ mod tests { event_source_swap( &view, vault, + pool_id(TWO_TOKEN_SPECIALIZATION), &mapped, - t_in, - U256::from(1_u64), - t_out, - U256::from(1_u64) + swap_delta(t_in, 1, t_out, 1) + ) + .is_none() + ); + } + + #[test] + fn event_source_swap_rejects_invalid_token_cash_metadata() { + let vault = addr(0xba); + let (t_in, t_out) = (addr(0x01), addr(0x02)); + let slot = U256::from(0x77_u64); + let mut m = HashMap::new(); + m.insert((vault, slot), packed_two(5, 1000, 500)); + let view = MapView(m); + + let duplicate_field = BalancerV2Metadata::default() + .with_vault(vault) + .with_token_cash([ + BalancerTokenBalance::new(t_in, slot, false), + BalancerTokenBalance::new(t_out, slot, false), + ]); + assert!( + event_source_swap( + &view, + vault, + pool_id(TWO_TOKEN_SPECIALIZATION), + &duplicate_field, + swap_delta(t_in, 30, t_out, 20), + ) + .is_none() + ); + + let high_field_on_general = BalancerV2Metadata::default() + .with_vault(vault) + .with_token_cash([ + BalancerTokenBalance::new(t_in, slot, false), + BalancerTokenBalance::new(t_out, slot, true), + ]); + assert!( + event_source_swap( + &view, + vault, + pool_id(0), + &high_field_on_general, + swap_delta(t_in, 30, t_out, 20), ) .is_none() ); diff --git a/tests/adapter_reactive.rs b/tests/adapter_reactive.rs index c32c375..14ceb04 100644 --- a/tests/adapter_reactive.rs +++ b/tests/adapter_reactive.rs @@ -67,6 +67,12 @@ fn block_ref(block_number: u64) -> BlockRef { } } +fn balancer_pool_id(specialization: u16, seed: u8) -> B256 { + let mut bytes = [seed; 32]; + bytes[20..22].copy_from_slice(&specialization.to_be_bytes()); + B256::from(bytes) +} + fn included_context( block_number: u64, log_index: u64, @@ -837,7 +843,7 @@ fn balancer_registry_with_metadata( #[tokio::test] async fn balancer_swap_event_sources_two_token_shared_slot() -> Result<()> { let vault = Address::repeat_byte(0x52); - let pool_id = B256::repeat_byte(0xc3); + let pool_id = balancer_pool_id(2, 0xc3); let token_in = Address::repeat_byte(0x01); let token_out = Address::repeat_byte(0x02); let slot = U256::from(0x77_u64); @@ -910,7 +916,7 @@ async fn balancer_swap_event_sources_two_token_shared_slot() -> Result<()> { #[tokio::test] async fn balancer_swap_event_sources_general_separate_slots() -> Result<()> { let vault = Address::repeat_byte(0x53); - let pool_id = B256::repeat_byte(0xd4); + let pool_id = balancer_pool_id(0, 0xd4); let token_in = Address::repeat_byte(0x03); let token_out = Address::repeat_byte(0x04); let slot_in = U256::from(0x11_u64);