diff --git a/fynd-core/Cargo.toml b/fynd-core/Cargo.toml index f6b78768..9b971b67 100644 --- a/fynd-core/Cargo.toml +++ b/fynd-core/Cargo.toml @@ -41,6 +41,10 @@ alloy = { workspace = true, features = ["sol-types", "signer-local"] } [features] test-utils = [] +# Per-pool simulation metering: counts and times every `get_amount_out` a solve makes, splits it +# by protocol and by solve pass, and reports it. Off by default — measuring a swap costs two clock +# reads on the hottest path in the solver. With it off, solving calls the panic guard directly. +swap-metrics = [] # Block-step-controller build path: lets a caller release buffered blocks one at # a time (deterministic stepping). Used by the hindsight monitor; the API is not # yet stable, so it is opt-in. diff --git a/fynd-core/src/algorithm/bellman_ford.rs b/fynd-core/src/algorithm/bellman_ford.rs index 27a42b2c..7be07c2e 100644 --- a/fynd-core/src/algorithm/bellman_ford.rs +++ b/fynd-core/src/algorithm/bellman_ford.rs @@ -43,7 +43,7 @@ use super::{ split_primitives::MarketOverrides, Algorithm, AlgorithmConfig, AlgorithmError, NoPathReason, }; use crate::{ - algorithm::sim_guard::GuardedProtocolSim, + algorithm::{paths, sim_guard::GuardedProtocolSim}, derived::{ computation::ComputationRequirements, types::{SpotPrices, TokenGasPrices}, @@ -205,13 +205,7 @@ impl BellmanFordAlgorithm { }, )?; - let market_view = match label.as_ref() { - Some(l) => market - .read_labeled(l) - .await - .map_err(|e| AlgorithmError::Other(e.to_string()))?, - None => market.read().await, - }; + let market_view = paths::read_market(&market, label).await?; let token_map: FxHashMap> = token_nodes .iter() .filter_map(|&node| { diff --git a/fynd-core/src/algorithm/mod.rs b/fynd-core/src/algorithm/mod.rs index 97943634..b5f2f66a 100644 --- a/fynd-core/src/algorithm/mod.rs +++ b/fynd-core/src/algorithm/mod.rs @@ -23,11 +23,13 @@ pub mod most_liquid; pub mod path_frank_wolfe; pub(crate) mod paths; pub(crate) mod sim_guard; +pub(crate) mod sim_meter; pub(crate) mod split_primitives; pub mod water_fill; #[cfg(test)] pub mod split_test_harness; +mod swap_cache; #[cfg(test)] pub mod test_utils; diff --git a/fynd-core/src/algorithm/most_liquid.rs b/fynd-core/src/algorithm/most_liquid.rs index f17be82d..ba2bc501 100644 --- a/fynd-core/src/algorithm/most_liquid.rs +++ b/fynd-core/src/algorithm/most_liquid.rs @@ -19,13 +19,17 @@ use rustc_hash::{FxHashMap, FxHashSet}; use smallvec::SmallVec; use tracing::{debug, instrument, trace}; use tycho_simulation::{ - tycho_common::simulation::protocol_sim::{GetAmountOutResult, Price, ProtocolSim}, + tycho_common::simulation::protocol_sim::{Price, ProtocolSim}, tycho_core::models::{token::Token, Address}, }; use super::{Algorithm, AlgorithmConfig, NoPathReason}; use crate::{ - algorithm::{paths, sim_guard::GuardedProtocolSim}, + algorithm::{ + paths, + sim_guard::GuardedProtocolSim, + swap_cache::{PoolDirection, Refusal, SwapCache, SwapResult}, + }, derived::{computation::ComputationRequirements, types::TokenGasPrices, SharedDerivedDataRef}, feed::market_data::{MarketData, MarketState, StateLabel}, graph::{ @@ -35,6 +39,10 @@ use crate::{ AlgorithmError, }; +/// What the simulation report calls the pass that picks a route. Most-liquid has one: every swap +/// it makes is ranking candidates, and the winner is re-simulated by `build_route` afterwards. +const SELECTION: &str = "selection"; + /// What can go wrong settling one token sequence. /// /// Never leaves most-liquid: the caller counts it and tries the next sequence. Variants carry node @@ -251,12 +259,6 @@ struct HopResult { gas: BigUint, } -impl HopResult { - fn new(pool_ix: usize, result: GetAmountOutResult) -> Self { - HopResult { pool_ix, amount_out: result.amount, gas: result.gas } - } -} - /// Algorithm that selects routes based on expected output after gas. pub struct MostLiquidAlgorithm { /// The hop bounds and connector tokens every route search runs under. Owned, so a solve hands @@ -588,13 +590,7 @@ impl MostLiquidAlgorithm { } } - let market = match label.as_ref() { - Some(l) => market - .read_labeled(l) - .await - .map_err(|e| AlgorithmError::Other(e.to_string()))?, - None => market.read().await, - }; + let market = paths::read_market(&market, label).await?; let market_subset = market.extract_subset_with_overlay(&component_ids); drop(market); Ok(market_subset) @@ -607,7 +603,8 @@ impl MostLiquidAlgorithm { ctx: &SolveContext, ) -> Result { let mut best_route: Option<(&TokenPath, SolvedRoute)> = None; - let mut cache = PoolSwapsCache::new(self.cache_pair_swaps); + let mut winners = PairWinners::new(self.cache_pair_swaps); + let mut swaps = SwapCache::new(); let timeout_ms = self.timeout.as_millis() as u64; for (token_path, _) in scored_paths { @@ -617,7 +614,7 @@ impl MostLiquidAlgorithm { break; } - let solved = match Self::solve_token_path(ctx, token_path, &mut cache) { + let solved = match Self::solve_token_path(ctx, token_path, &mut winners, &mut swaps) { Ok(solved) => solved, Err(e) => { trace!(error = %e, "could not solve path"); @@ -700,11 +697,13 @@ impl MostLiquidAlgorithm { /// both hops of A -> B -> C -- and a circular sequence crosses the same pair in both /// directions. Sequences whose hops all run out of pools this way are dropped, the same as any /// other sequence that cannot be settled. - fn solve_token_path( - ctx: &SolveContext<'_>, + fn solve_token_path<'g>( + ctx: &SolveContext<'g>, token_path: &[NodeIndex], - cache: &mut PoolSwapsCache, + winners: &mut PairWinners, + swaps: &mut SwapCache<'g>, ) -> Result { + let (graph, market, gas_price) = (ctx.graph, ctx.market, ctx.gas_price); let mut current_amount = ctx.amount_in.clone(); let mut hops: SmallVec<[HopResult; INLINE_EDGES]> = SmallVec::new(); let mut used_components: SmallVec<[&ComponentId; INLINE_EDGES]> = SmallVec::new(); @@ -716,27 +715,37 @@ impl MostLiquidAlgorithm { for pair in token_path.windows(2) { let (token_in, token_out, token_out_gas_price) = Self::get_pair_data(ctx, pair)?; - let simulate = |component_id: &ComponentId| { - let state = ctx - .market - .get_simulation_state(component_id)?; - let result = state - .get_amount_out_guarded(current_amount.clone(), token_in, token_out) - .ok()?; + let simulate = |component_id: &'g ComponentId| { + let paid = swaps.swap( + PoolDirection { + component_id, + address_in: &token_in.address, + address_out: &token_out.address, + }, + ¤t_amount, + SELECTION, + || { + let Some(state) = market.get_simulation_state(component_id) else { + return Err(Refusal::Failed); + }; + state + .get_amount_out_guarded(current_amount.clone(), token_in, token_out) + .map(|result| SwapResult { amount_out: result.amount, gas: result.gas }) + .map_err(|error| Refusal::of(&error)) + }, + true, + )?; let net = match token_out_gas_price { Some(price) => { - let cost = - &result.gas * ctx.gas_price * &price.numerator / &price.denominator; - BigInt::from(result.amount.clone()) - BigInt::from(cost) + let cost = &paid.gas * gas_price * &price.numerator / &price.denominator; + BigInt::from(paid.amount_out.clone()) - BigInt::from(cost) } - None => BigInt::from(result.amount.clone()), + None => BigInt::from(paid.amount_out.clone()), }; - Some((result, net)) + Some((paid, net)) }; - let pools = ctx - .graph - .pools_between(pair[0], pair[1]); + let pools = graph.pools_between(pair[0], pair[1]); // A pool this route already crossed cannot be offered again. Where that bites, the // pool is picked here and the cache is left alone in both directions: what it holds @@ -748,7 +757,7 @@ impl MostLiquidAlgorithm { let hop_result = if restricted { best_paying_pool(pools, |id| !used_components.contains(&id), simulate) } else { - cache.swap((pair[0], pair[1]), ¤t_amount, pools, simulate) + winners.hop((pair[0], pair[1]), pools, simulate) }; let Some(hop_result) = hop_result else { return Err(MostLiquidError::HopNotTradable { from: pair[0], to: pair[1] }) @@ -873,11 +882,7 @@ impl Algorithm for MostLiquidAlgorithm { // Step 3: Fetch all pools in scored_paths. let market = Self::snapshot_market_state(graph, market, label, &scored_paths).await?; - let gas_price = market - .gas_price() - .ok_or(AlgorithmError::DataNotFound { kind: "gas price", id: None })? - .effective_gas_price() - .clone(); + let gas_price = paths::fetch_gas_price(&market)?; // Step 4: Solve all paths in score order and return the best one let ctx = SolveContext { @@ -914,12 +919,12 @@ impl Algorithm for MostLiquidAlgorithm { /// /// `pool_ix` indexes `pools`, so a caller that skipped some still gets back the index the built /// route needs. -fn best_paying_pool( - pools: &[EdgeData], +fn best_paying_pool<'g, D>( + pools: &'g [EdgeData], mut usable: impl FnMut(&ComponentId) -> bool, - mut simulate: impl FnMut(&ComponentId) -> Option<(GetAmountOutResult, BigInt)>, + mut simulate: impl FnMut(&'g ComponentId) -> Option<(SwapResult, BigInt)>, ) -> Option { - let mut best: Option<(usize, GetAmountOutResult, BigInt)> = None; + let mut best: Option<(usize, SwapResult, BigInt)> = None; for (pool_ix, edge) in pools.iter().enumerate() { if !usable(&edge.component_id) { @@ -937,94 +942,63 @@ fn best_paying_pool( } } - let (pool_ix, result, _) = best?; - Some(HopResult::new(pool_ix, result)) + let (pool_ix, paid, _) = best?; + Some(HopResult { pool_ix, amount_out: paid.amount_out, gas: paid.gas }) } -/// What each token pair paid, created for every order and not persisted after it. +/// Which pool won each token pair, for this order only. +/// +/// Routes share pairs: every route through WBTC -> WETH asks the same pools the same question. +/// What each pool *paid* is not kept here -- that lives in the shared [`SwapCache`], keyed by pool +/// so two nearby amounts on one pool's curve can be read across. Keeping amounts per pair could +/// not do that: a later amount can be won by a different pool, and a line drawn between two +/// amounts won by two pools joins two unrelated curves. /// -/// Routes share pairs: every route through WBTC -> WETH asks the same pools the same question. This -/// answers it once. Which pool pays most is remembered for the pair; what it paid is remembered per -/// input amount, so routes arriving with the same amount get the same numbers. +/// What is left is the choice, which is the part worth reusing: try the pool that won this pair +/// last before asking every pool on it again. /// -/// Remembering nothing is a valid state: [`PoolSwapsCache::new`] takes a flag, and with it off -/// every hop asks every pool, which is the answer the cache is an approximation of. -struct PoolSwapsCache { - pairs: FxHashMap<(NodeIndex, NodeIndex), PairCacheEntry>, +/// Remembering nothing is a valid state: [`PairWinners::new`] takes a flag, and with it off every +/// hop asks every pool, which is the answer this is an approximation of. +struct PairWinners { + winner_by_pair: FxHashMap<(NodeIndex, NodeIndex), usize>, enabled: bool, } -/// The pool that won a pair, and what it paid at each amount seen so far. -struct PairCacheEntry { - /// Where the pool that last won this pair sits in its pool list. - pool_ix: usize, - /// Keyed by the amount that went in. Each outcome names the pool it came from, which is not - /// always `pool_ix`: a later amount can be won by a different pool, and the outcomes already - /// recorded still belong to whichever pool produced them. - outcomes_by_amount: FxHashMap, -} - -impl PoolSwapsCache { +impl PairWinners { fn new(enabled: bool) -> Self { - Self { pairs: FxHashMap::default(), enabled } + Self { winner_by_pair: FxHashMap::default(), enabled } } /// Swaps one pair: which pool to go through, and what it pays. /// - /// Tries in order: the amount already recorded for this pair, then the pool that won it, then - /// every pool. `simulate` runs one pool and reports what it pays and what that is worth after - /// its own gas. + /// Tries the pool that won this pair before falling back to every pool. `simulate` reports + /// what one pool pays and what that is worth after its own gas; whether it reaches the pool or + /// is answered from amounts already asked is the swap cache's business, not this one's. /// /// Returns `None` when no pool on the pair can trade the amount. - fn swap( + fn hop<'g, D>( &mut self, pair: (NodeIndex, NodeIndex), - amount_in: &BigUint, - pools: &[EdgeData], - mut simulate: impl FnMut(&ComponentId) -> Option<(GetAmountOutResult, BigInt)>, + pools: &'g [EdgeData], + mut simulate: impl FnMut(&'g ComponentId) -> Option<(SwapResult, BigInt)>, ) -> Option { - if let Some(choice) = self.pairs.get(&pair) { - if let Some(outcome) = choice.outcomes_by_amount.get(amount_in) { - return Some(outcome.clone()); - } - - // The pool was simulated but not at this amount. We assume that amounts dont move - // so much so we assume the same winner wins again at a slightly different amount. - if let Some((result, _)) = pools - .get(choice.pool_ix) + // The winner is assumed to hold at a nearby amount. Where it does not, it still pays what + // it pays, and the next full scan of this pair moves the choice on. + if let Some(&pool_ix) = self.winner_by_pair.get(&pair) { + if let Some((paid, _)) = pools + .get(pool_ix) .and_then(|edge| simulate(&edge.component_id)) { - let hop_result = HopResult::new(choice.pool_ix, result); - self.record(pair, amount_in, &hop_result); - return Some(hop_result); + return Some(HopResult { pool_ix, amount_out: paid.amount_out, gas: paid.gas }); } } let hop_result = best_paying_pool(pools, |_| true, simulate)?; - self.record(pair, amount_in, &hop_result); - Some(hop_result) - } - - /// Remembers what a pool paid for one amount on one pair, unless the cache is off. - /// - /// This is the only place anything is written, so an off cache stays empty and every lookup in - /// [`PoolSwapsCache::swap`] misses. - fn record(&mut self, pair: (NodeIndex, NodeIndex), amount_in: &BigUint, result: &HopResult) { - if !self.enabled { - return; + if self.enabled { + self.winner_by_pair + .insert(pair, hop_result.pool_ix); } - - let choice = self - .pairs - .entry(pair) - .or_insert_with(|| PairCacheEntry { - pool_ix: result.pool_ix, - outcomes_by_amount: FxHashMap::default(), - }); - choice.pool_ix = result.pool_ix; - choice - .outcomes_by_amount - .insert(amount_in.clone(), result.clone()); + Some(hop_result) } } @@ -2204,7 +2178,7 @@ mod tests { assert_eq!(MostLiquidAlgorithm::score_token_path(graph, &[node(&a)]), None); } - // ==================== PoolSwapsCache Tests ==================== + // ==================== PairWinners Tests ==================== /// `count` pools on one pair, named `pool0`, `pool1`, ... for [`simulator`] to answer as. fn pools(count: usize) -> Vec> { @@ -2217,126 +2191,115 @@ mod tests { fn simulator( multipliers: [u64; 2], amount_in: u64, - ) -> impl FnMut(&ComponentId) -> Option<(GetAmountOutResult, BigInt)> { + ) -> impl FnMut(&ComponentId) -> Option<(SwapResult, BigInt)> { move |component_id: &ComponentId| { let index: usize = component_id .trim_start_matches("pool") .parse() .ok()?; let amount = BigUint::from(amount_in * multipliers[index]); - let result = GetAmountOutResult::new( - amount.clone(), - BigUint::from(10u64), - Box::new(MockProtocolSim::new(1.0)), - ); - Some((result, BigInt::from(amount))) + let paid = SwapResult { amount_out: amount.clone(), gas: BigUint::from(10u64) }; + Some((paid, BigInt::from(amount))) } } - /// The one token pair every cache case works on. + /// The one token pair every case works on. fn pair() -> (NodeIndex, NodeIndex) { (NodeIndex::new(0), NodeIndex::new(1)) } #[test] - fn test_cache_takes_the_pool_that_pays_most() { - let mut cache = PoolSwapsCache::new(true); + fn test_winners_takes_the_pool_that_pays_most() { + let mut winners = PairWinners::new(true); let multipliers = [2u64, 5u64]; let pools = pools(multipliers.len()); - let outcome = cache - .swap(pair(), &BigUint::from(100u64), &pools, simulator(multipliers, 100)) + let outcome = winners + .hop(pair(), &pools, simulator(multipliers, 100)) .unwrap(); assert_eq!(outcome.pool_ix, 1, "pool1 pays 500 against pool0's 200"); assert_eq!(outcome.amount_out, BigUint::from(500u64)); } - /// The second ask at the same amount must not simulate anything. + /// Once a pair has a winner, only that pool is asked. Whether the ask reaches the pool or is + /// answered from an amount already asked is the swap cache's business, not this one's. #[test] - fn test_cache_answers_a_repeated_amount_without_simulating() { - let mut cache = PoolSwapsCache::new(true); + fn test_winners_asks_only_the_remembered_winner() { + let mut winners = PairWinners::new(true); let multipliers = [2u64, 5u64]; let pools = pools(multipliers.len()); - let amount = BigUint::from(100u64); - cache - .swap(pair(), &amount, &pools, simulator(multipliers, 100)) + winners + .hop(pair(), &pools, simulator(multipliers, 100)) .unwrap(); - let outcome = cache - .swap(pair(), &amount, &pools, |_| panic!("must not simulate on a hit")) + let mut asked = Vec::new(); + let outcome = winners + .hop(pair(), &pools, |component_id: &ComponentId| { + asked.push(component_id.clone()); + simulator(multipliers, 100)(component_id) + }) .unwrap(); + assert_eq!(asked, vec!["pool1".to_string()], "only the winner should be asked"); assert_eq!(outcome.pool_ix, 1); - assert_eq!(outcome.amount_out, BigUint::from(500u64)); } - /// A remembered outcome carries the pool that produced it. A later amount won by a different - /// pool must not rewrite what an earlier one was told. + /// A winner that cannot trade the amount does not end the hop: every pool is asked, and the + /// one that answers becomes the pair's winner. #[test] - fn test_cache_keeps_each_amount_with_the_pool_that_paid_it() { - let mut cache = PoolSwapsCache::new(true); + fn test_winners_falls_back_to_every_pool_when_the_winner_cannot_trade() { + let mut winners = PairWinners::new(true); let pools = pools(2); - - // At 100, pool1 wins. At 50 the simulator is rigged so only pool0 answers, which makes it - // the pair's remembered winner. - cache - .swap(pair(), &BigUint::from(100u64), &pools, simulator([2, 5], 100)) + winners + .hop(pair(), &pools, simulator([2, 5], 100)) .unwrap(); - cache - .swap(pair(), &BigUint::from(50u64), &pools, |component_id: &ComponentId| { + + let outcome = winners + .hop(pair(), &pools, |component_id: &ComponentId| { (component_id == "pool0").then(|| { let amount = BigUint::from(50u64); ( - GetAmountOutResult::new( - amount.clone(), - BigUint::from(10u64), - Box::new(MockProtocolSim::new(1.0)), - ), + SwapResult { amount_out: amount.clone(), gas: BigUint::from(10u64) }, BigInt::from(amount), ) }) }) .unwrap(); - let replay = cache - .swap(pair(), &BigUint::from(100u64), &pools, |_| panic!("must not simulate on a hit")) - .unwrap(); - - assert_eq!(replay.pool_ix, 1, "the 100 outcome still belongs to pool1"); - assert_eq!(replay.amount_out, BigUint::from(500u64)); + assert_eq!(outcome.pool_ix, 0, "pool1 refused, so pool0 takes the pair"); + assert_eq!(outcome.amount_out, BigUint::from(50u64)); } - /// Off, the cache stays empty, so every ask simulates again. + /// Off, no winner is remembered, so every pool is asked every time. #[test] - fn test_cache_disabled_remembers_nothing() { - let mut cache = PoolSwapsCache::new(false); + fn test_winners_disabled_scans_every_pool_each_time() { + let mut winners = PairWinners::new(false); let multipliers = [2u64, 5u64]; let pools = pools(multipliers.len()); - let amount = BigUint::from(100u64); - let first = cache - .swap(pair(), &amount, &pools, simulator(multipliers, 100)) + let first = winners + .hop(pair(), &pools, simulator(multipliers, 100)) .unwrap(); let mut asked = 0usize; - let second = cache - .swap(pair(), &amount, &pools, |component_id| { + let second = winners + .hop(pair(), &pools, |component_id: &ComponentId| { asked += 1; simulator(multipliers, 100)(component_id) }) .unwrap(); - assert_eq!(asked, 2, "both pools are asked again, so nothing was remembered"); + assert_eq!(asked, 2, "both pools are asked again, so no winner was remembered"); assert_eq!(first.amount_out, second.amount_out); } /// No pool on the pair can trade the amount. #[test] - fn test_cache_returns_none_when_no_pool_trades() { - let mut cache = PoolSwapsCache::new(true); + fn test_winners_returns_none_when_no_pool_trades() { + let mut winners = PairWinners::new(true); let pools = pools(2); - let outcome = cache.swap(pair(), &BigUint::from(100u64), &pools, |_| None); + let outcome = winners.hop(pair(), &pools, |_| None); assert!(outcome.is_none()); } diff --git a/fynd-core/src/algorithm/paths.rs b/fynd-core/src/algorithm/paths.rs index 7eddcb0a..a5695502 100644 --- a/fynd-core/src/algorithm/paths.rs +++ b/fynd-core/src/algorithm/paths.rs @@ -6,7 +6,7 @@ use num_bigint::{BigInt, BigUint}; use rustc_hash::FxHashMap; -use tracing::{debug, instrument, trace}; +use tracing::{instrument, trace}; use tycho_simulation::{ tycho_common::simulation::protocol_sim::ProtocolSim, tycho_core::models::{token::Token, Address}, @@ -16,10 +16,10 @@ use super::{most_liquid::DepthAndPrice, NoPathReason}; use crate::{ algorithm::sim_guard::GuardedProtocolSim, derived::types::TokenGasPrices, - feed::market_data::MarketState, + feed::market_data::{MarketData, MarketDataView, MarketState}, graph::{GraphError, GraphQueryFilter, Path, TokenPath, TopologyGraph}, types::{ComponentId, Route, RouteResult, Swap}, - AlgorithmError, + AlgorithmError, StateLabel, }; /// Every route between two tokens, one per combination of the pools serving its legs. @@ -97,7 +97,7 @@ pub(crate) fn try_score_path(path: &Path) -> Option { for edge in path.edge_iter() { let Some(data) = edge.data.as_ref() else { - debug!(component_id = %edge.component_id, "edge missing weight data, path cannot be scored"); + trace!(component_id = %edge.component_id, "edge missing weight data, path cannot be scored"); return None; }; @@ -236,6 +236,37 @@ pub(crate) fn get_token<'a>( }) } +/// A read view of the market, under `label` when one is given and of the live state otherwise. +/// +/// # Errors +/// +/// [`AlgorithmError::Other`] when `label` names no registered overlay. +pub(crate) async fn read_market( + market: &MarketData, + label: Option, +) -> Result, AlgorithmError> { + match label.as_ref() { + Some(l) => market + .read_labeled(l) + .await + .map_err(|e| AlgorithmError::Other(e.to_string())), + None => Ok(market.read().await), + } +} + +/// The view's effective gas price, for algorithms that cannot price a route without one. +/// +/// # Errors +/// +/// [`AlgorithmError::DataNotFound`] when the view carries no gas price. +pub(crate) fn fetch_gas_price(view: &MarketState) -> Result { + Ok(view + .gas_price() + .ok_or(AlgorithmError::DataNotFound { kind: "gas price", id: None })? + .effective_gas_price() + .clone()) +} + #[cfg(test)] mod tests { use num_bigint::BigUint; diff --git a/fynd-core/src/algorithm/sim_meter.rs b/fynd-core/src/algorithm/sim_meter.rs new file mode 100644 index 00000000..64f727af --- /dev/null +++ b/fynd-core/src/algorithm/sim_meter.rs @@ -0,0 +1,406 @@ +//! Measurement of the simulation work one solve does. +//! +//! Simulating a swap is the dominant cost of solving, and that cost varies by orders of magnitude +//! between protocols — a constant-product pool is arithmetic, a `vm:*` pool runs EVM bytecode. This +//! records what was asked of which component, so a solve can say where its time went. +//! +//! Callers go through [`MeteredProtocolSim::get_amount_out_metered`], which wraps the panic guard +//! in [`super::sim_guard`] rather than replacing it. Only `water_fill` calls it; the other +//! algorithms take the bare guard, so nothing outside a water-fill solve is counted. +//! +//! Counts live in a thread-local for the duration of a solve, which a worker runs start to finish +//! on one thread. Recording therefore needs nothing passed down to it, and the default build — +//! where `swap-metrics` is off — compiles every entry point here to nothing. + +#[cfg(feature = "swap-metrics")] +use std::{ + cell::RefCell, + cmp::Reverse, + time::{Duration, Instant}, +}; + +#[cfg(feature = "swap-metrics")] +use metrics::counter; +use num_bigint::BigUint; +#[cfg(feature = "swap-metrics")] +use rustc_hash::{FxHashMap, FxHashSet}; +#[cfg(feature = "swap-metrics")] +use tracing::{debug, enabled, Level}; +use tycho_simulation::tycho_common::{ + models::token::Token, + simulation::{ + errors::SimulationError, + protocol_sim::{GetAmountOutResult, ProtocolSim}, + }, +}; + +use super::sim_guard::GuardedProtocolSim; +use crate::{feed::market_data::MarketState, types::ComponentId}; + +/// What the report calls the stage a swap was asked for. +/// +/// A plain label rather than an algorithm's own stage type: an algorithm names its stages, this +/// module only groups by what it is told. +pub(crate) type StageLabel = &'static str; + +/// The swaps one component was asked for over a solve. Only some of them reached it — the rest +/// were answered from an amount it had already been asked, so they carry no call and no time. +#[cfg(feature = "swap-metrics")] +#[derive(Default, Clone, Copy)] +struct ComponentSwaps { + /// Calls made against this component. + calls: u64, + /// Of those, the ones that came back an error: the pool could not quote the swap, or its math + /// panicked and [`GuardedProtocolSim`] turned that into an error. + failed: u64, + /// Swaps the cache answered instead, so no call was made at all. + cache_hits: u64, + /// Swaps answered by reading across two nearby amounts the pool had already been asked, so + /// again no call was made. Only the stages that settle which paths get split do this, and the + /// amount they get back is slightly below what the pool would have paid. + interpolated: u64, + /// Swaps refused without calling, because the pool had already refused a smaller amount. + refused_without_calling: u64, + /// Time spent inside the calls that were made. + call_time: Duration, +} + +#[cfg(feature = "swap-metrics")] +impl ComponentSwaps { + fn add(&mut self, other: &ComponentSwaps) { + self.calls += other.calls; + self.failed += other.failed; + self.cache_hits += other.cache_hits; + self.interpolated += other.interpolated; + self.refused_without_calling += other.refused_without_calling; + self.call_time += other.call_time; + } +} + +#[cfg(feature = "swap-metrics")] +thread_local! { + /// Simulation work done while solving one order, on this thread. + /// + /// Counted per component, not per protocol: which protocol a component belongs to is known + /// only to the market, so it is resolved once in [`report`] rather than looked up on every + /// swap. Resolving it there also names every `vm:*` protocol individually, where reading it + /// off the concrete state type would collapse them into one. + /// + /// The component id is owned rather than borrowed, which costs a clone per recorded swap. + /// That only happens in the `swap-metrics` build; the default build records nothing. + static SOLVE_SWAPS: RefCell> = + RefCell::new(FxHashMap::default()); +} + +#[cfg(feature = "swap-metrics")] +fn with_counts( + component_id: &ComponentId, + stage: StageLabel, + edit: impl FnOnce(&mut ComponentSwaps), +) { + SOLVE_SWAPS.with_borrow_mut(|swaps| { + edit( + swaps + .entry((component_id.clone(), stage)) + .or_default(), + ); + }); +} + +/// Discards whatever the last solve on this thread left behind, so counts never run together. +#[cfg(feature = "swap-metrics")] +pub(crate) fn start_solve() { + SOLVE_SWAPS.with_borrow_mut(FxHashMap::clear); +} + +/// Records one `get_amount_out` call and how long it took. +#[cfg(feature = "swap-metrics")] +fn record_call(component_id: &ComponentId, stage: StageLabel, call_time: Duration, failed: bool) { + with_counts(component_id, stage, |counts| { + counts.calls += 1; + counts.call_time += call_time; + if failed { + counts.failed += 1; + } + }); +} + +/// Records a swap the cache answered, so no call was made. +#[cfg(feature = "swap-metrics")] +pub(crate) fn record_cache_hit(component_id: &ComponentId, stage: StageLabel) { + with_counts(component_id, stage, |counts| counts.cache_hits += 1); +} + +/// Records a swap answered by reading across two nearby amounts, so no call was made. +#[cfg(feature = "swap-metrics")] +pub(crate) fn record_interpolation(component_id: &ComponentId, stage: StageLabel) { + with_counts(component_id, stage, |counts| counts.interpolated += 1); +} + +/// Records a swap refused on the strength of a smaller amount the pool already refused. +#[cfg(feature = "swap-metrics")] +pub(crate) fn record_refusal_without_calling(component_id: &ComponentId, stage: StageLabel) { + with_counts(component_id, stage, |counts| counts.refused_without_calling += 1); +} + +/// Writes what the solve asked of which protocol and which stage, and feeds the same counts to the +/// metrics recorder. A component the market no longer holds is reported under `unknown` rather +/// than dropped, so the totals still add up. +/// +/// The counters always go out. The two lines are only built when debug logging is on, since +/// formatting them walks every protocol and every stage on a path that runs once per solve. +#[cfg(feature = "swap-metrics")] +pub(crate) fn report(market: &MarketState, solve_time_ms: impl FnOnce() -> u64) { + let solve_time_ms = solve_time_ms(); + SOLVE_SWAPS.with_borrow(|swaps| report_swaps(swaps, market, solve_time_ms)); +} + +#[cfg(feature = "swap-metrics")] +fn report_swaps( + swaps: &FxHashMap<(ComponentId, StageLabel), ComponentSwaps>, + market: &MarketState, + solve_time_ms: u64, +) { + let mut by_protocol: FxHashMap<&str, ComponentSwaps> = FxHashMap::default(); + for ((component_id, _), counts) in swaps { + let protocol = market + .get_component(component_id) + .map_or("unknown", |component| component.protocol_system.as_str()); + by_protocol + .entry(protocol) + .or_default() + .add(counts); + } + + let mut costliest_first: Vec<(&str, ComponentSwaps)> = by_protocol.into_iter().collect(); + costliest_first + .sort_unstable_by_key(|(protocol, counts)| (Reverse(counts.call_time), *protocol)); + + for (protocol, counts) in &costliest_first { + counter!("water_fill.get_amount_out_calls", "protocol" => protocol.to_string()) + .increment(counts.calls); + counter!("water_fill.failed_calls", "protocol" => protocol.to_string()) + .increment(counts.failed); + counter!("water_fill.cache_hits", "protocol" => protocol.to_string()) + .increment(counts.cache_hits); + counter!("water_fill.interpolated_swaps", "protocol" => protocol.to_string()) + .increment(counts.interpolated); + counter!("water_fill.refused_without_calling", "protocol" => protocol.to_string()) + .increment(counts.refused_without_calling); + } + + if !enabled!(Level::DEBUG) { + return; + } + + let mut by_stage: FxHashMap = FxHashMap::default(); + let mut components: FxHashSet<&ComponentId> = FxHashSet::default(); + let mut totals = ComponentSwaps::default(); + for ((component_id, stage), counts) in swaps { + by_stage + .entry(stage) + .or_default() + .add(counts); + components.insert(component_id); + totals.add(counts); + } + + let mut stages: Vec<(StageLabel, ComponentSwaps)> = by_stage.into_iter().collect(); + stages.sort_unstable_by_key(|(_, counts)| Reverse(counts.call_time)); + let per_stage = stages + .iter() + .map(|(stage, counts)| { + format!( + "{}: {} calls in {:.1}ms, {} answered without calling", + stage, + counts.calls, + counts.call_time.as_secs_f64() * 1000.0, + counts.cache_hits + counts.interpolated + counts.refused_without_calling, + ) + }) + .collect::>() + .join(" | "); + debug!(solve_time_ms, "water-fill simulation by stage: {per_stage}"); + + let per_protocol = costliest_first + .iter() + .map(|(protocol, counts)| { + format!( + "{protocol}: {} calls ({} failed) in {:.1}ms, {} cache hits, {} interpolated, \ + {} refused without calling", + counts.calls, + counts.failed, + counts.call_time.as_secs_f64() * 1000.0, + counts.cache_hits, + counts.interpolated, + counts.refused_without_calling, + ) + }) + .collect::>() + .join(" | "); + debug!( + solve_time_ms, + components = components.len(), + get_amount_out_calls = totals.calls, + failed_calls = totals.failed, + cache_hits = totals.cache_hits, + interpolated = totals.interpolated, + refused_without_calling = totals.refused_without_calling, + call_time_ms = totals.call_time.as_secs_f64() * 1000.0, + "water-fill simulation cost: {per_protocol}", + ); +} + +/// Recording is compiled out. The arguments are taken so the call sites read the same in either +/// build; the optimiser drops them. +#[cfg(not(feature = "swap-metrics"))] +pub(crate) fn start_solve() {} + +/// Recording is compiled out. See [`start_solve`]. +#[cfg(not(feature = "swap-metrics"))] +pub(crate) fn record_cache_hit(_component_id: &ComponentId, _stage: StageLabel) {} + +/// Recording is compiled out. See [`start_solve`]. +#[cfg(not(feature = "swap-metrics"))] +pub(crate) fn record_interpolation(_component_id: &ComponentId, _stage: StageLabel) {} + +/// Recording is compiled out. See [`start_solve`]. +#[cfg(not(feature = "swap-metrics"))] +pub(crate) fn record_refusal_without_calling(_component_id: &ComponentId, _stage: StageLabel) {} + +/// There is nothing to report without `swap-metrics`. +/// +/// The solve time is taken as a closure so the clock is never read in this build: an argument +/// would be evaluated at the call site even though nothing here uses it. +#[cfg(not(feature = "swap-metrics"))] +pub(crate) fn report(_market: &MarketState, _solve_time_ms: impl FnOnce() -> u64) {} + +/// Extension trait adding metered, panic-guarded simulation calls to every [`ProtocolSim`]. +/// +/// Wraps [`GuardedProtocolSim::get_amount_out_guarded`] and books the call against the component +/// that served it, which the guard alone cannot do — it sees only the state and the two tokens. +pub(crate) trait MeteredProtocolSim { + /// Calls the panic-guarded `get_amount_out` and records it against `component_id` and `stage`. + fn get_amount_out_metered( + &self, + component_id: &ComponentId, + stage: StageLabel, + amount_in: BigUint, + token_in: &Token, + token_out: &Token, + ) -> Result; +} + +impl MeteredProtocolSim for T { + fn get_amount_out_metered( + &self, + component_id: &ComponentId, + stage: StageLabel, + amount_in: BigUint, + token_in: &Token, + token_out: &Token, + ) -> Result { + #[cfg(not(feature = "swap-metrics"))] + { + let _ = (component_id, stage); + self.get_amount_out_guarded(amount_in, token_in, token_out) + } + #[cfg(feature = "swap-metrics")] + { + let started = Instant::now(); + let outcome = self.get_amount_out_guarded(amount_in, token_in, token_out); + record_call(component_id, stage, started.elapsed(), outcome.is_err()); + outcome + } + } +} + +#[cfg(all(test, feature = "swap-metrics"))] +mod tests { + use std::time::Duration; + + use super::*; + + fn component(id: &str) -> ComponentId { + ComponentId::from(id) + } + + fn counts_for(component_id: &ComponentId, stage: StageLabel) -> ComponentSwaps { + SOLVE_SWAPS.with_borrow(|swaps| { + swaps + .get(&(component_id.clone(), stage)) + .copied() + .unwrap_or_default() + }) + } + + /// The same component asked at two stages is counted separately, so the report can say which + /// stage the work sat in. + #[test] + fn test_records_each_stage_of_a_component_separately() { + start_solve(); + let pool = component("pool-a"); + + record_call(&pool, "ranking", Duration::from_millis(3), false); + record_call(&pool, "ranking", Duration::from_millis(2), true); + record_cache_hit(&pool, "chunking"); + + let ranking = counts_for(&pool, "ranking"); + assert_eq!(ranking.calls, 2); + assert_eq!(ranking.failed, 1); + assert_eq!(ranking.call_time, Duration::from_millis(5)); + assert_eq!(ranking.cache_hits, 0); + assert_eq!(counts_for(&pool, "chunking").cache_hits, 1); + } + + /// Swaps answered without calling the pool are counted apart from the calls, so a solve can + /// say how much the cache saved. + #[test] + fn test_counts_answers_that_never_reached_the_pool() { + start_solve(); + let pool = component("pool-b"); + + record_cache_hit(&pool, "ranking"); + record_interpolation(&pool, "ranking"); + record_refusal_without_calling(&pool, "ranking"); + + let counts = counts_for(&pool, "ranking"); + assert_eq!(counts.calls, 0); + assert_eq!(counts.cache_hits, 1); + assert_eq!(counts.interpolated, 1); + assert_eq!(counts.refused_without_calling, 1); + } + + /// A new solve starts from nothing, so one order's counts never land in the next one's report. + #[test] + fn test_start_solve_discards_the_previous_solve() { + start_solve(); + let pool = component("pool-c"); + record_call(&pool, "ranking", Duration::from_millis(1), false); + + start_solve(); + + assert_eq!(counts_for(&pool, "ranking").calls, 0); + } + + /// A component the market no longer holds still reaches the report, under `unknown`, so the + /// totals add up. + #[test] + fn test_unknown_component_is_reported_rather_than_dropped() { + start_solve(); + record_call(&component("gone"), "ranking", Duration::from_millis(1), false); + + // The market holds nothing, so every component resolves to "unknown". This asserts the + // report walks it without panicking and finds the fallback protocol name. + let market = MarketState::default(); + SOLVE_SWAPS.with_borrow(|swaps| { + let protocol = swaps.keys().map(|(component_id, _)| { + market + .get_component(component_id) + .map_or("unknown", |component| component.protocol_system.as_str()) + }); + assert!(protocol.eq(["unknown"])); + }); + report(&market, || 1); + } +} diff --git a/fynd-core/src/algorithm/split_primitives.rs b/fynd-core/src/algorithm/split_primitives.rs index 271fc5ba..90608085 100644 --- a/fynd-core/src/algorithm/split_primitives.rs +++ b/fynd-core/src/algorithm/split_primitives.rs @@ -160,6 +160,14 @@ impl MarketOverrides { pub(crate) fn get(&self, id: &ComponentId) -> Option<&dyn ProtocolSim> { self.0.get(id).map(|b| b.as_ref()) } + + /// Commits a post-swap component state, replacing whatever was there. + /// + /// The building counterpart to [`MarketOverrides::with_override`], for the passes that fill an + /// overlay chunk by chunk rather than declaring one up front. + pub(crate) fn insert(&mut self, id: ComponentId, sim: Box) { + self.0.insert(id, sim); + } } /// Wrapper that delegates all [`ProtocolSim`] calls unchanged except diff --git a/fynd-core/src/algorithm/swap_cache.rs b/fynd-core/src/algorithm/swap_cache.rs new file mode 100644 index 00000000..1a4f0929 --- /dev/null +++ b/fynd-core/src/algorithm/swap_cache.rs @@ -0,0 +1,557 @@ +use num_bigint::BigUint; +use rustc_hash::FxHashMap; +use tycho_simulation::tycho_common::{models::Address, simulation::errors::SimulationError}; + +use crate::{algorithm::sim_meter, ComponentId}; + +/// How far apart the two amounts either side of a requested one may sit, as a percentage of the +/// amount requested, before ranking stops reading across them and asks the pool instead. +const INTERPOLATION_GAP_PERCENT: u32 = 10; + +/// A pool and the direction taken through it. Both token addresses are part of it — a pool trading +/// three tokens answers `USDC -> DAI` and `USDT -> DAI` differently for the same amount. +#[derive(PartialEq, Eq, Hash)] +pub struct PoolDirection<'a> { + pub(crate) component_id: &'a ComponentId, + pub(crate) address_in: &'a Address, + pub(crate) address_out: &'a Address, +} + +/// Why a pool paid nothing for an amount. +/// +/// A pool that turns an amount down for being more than it can serve turns down every larger +/// amount too, and that is worth remembering. A pool that simply failed says nothing about any +/// other amount — the panic guard turns a component that blew up on one input into an error like +/// any other, and reading a size limit out of that would drop a working pool for the whole solve. +#[derive(Clone, Copy, PartialEq, Eq)] +pub enum Refusal { + /// The pool rejected the input itself, which is how it reports an amount beyond its limit. + OverLimit, + /// The pool failed for a reason that carries no information about size. + Failed, +} + +impl Refusal { + /// Reads a failed simulation. Only the pool rejecting the input says anything about size; + /// a fatal error is what a caught panic arrives as, and a recoverable one is transient. + pub(crate) fn of(error: &SimulationError) -> Self { + match error { + SimulationError::InvalidInput(_, _) => Refusal::OverLimit, + SimulationError::FatalError(_) | SimulationError::RecoverableError(_) => { + Refusal::Failed + } + } + } +} + +/// What a swap paid: what came out, and the gas it cost. +#[derive(Clone)] +pub struct SwapResult { + pub(crate) amount_out: BigUint, + pub(crate) gas: BigUint, +} + +/// What one pool paid at each amount it was asked, ascending by amount. A missing outcome is an +/// amount it refused. +/// +/// Short by nature — a handful of amounts per direction over one solve — so a sorted `Vec` searched +/// by bisection beats a map, and inserting in place keeps the neighbours of any amount adjacent. +#[derive(Default)] +pub struct SwappedAmounts { + /// Sorted ascending by amount! + amounts_and_results: Vec<(BigUint, Option)>, + /// The amount from which this pool has been taken to refuse everything. See + /// [`SwappedAmounts::record`] for what has to hold before an amount is recorded here. + failed_at: Option, +} + +impl SwappedAmounts { + /// Whether `amount_in` is at or above the point this pool started refusing. + fn refuses(&self, amount_in: &BigUint) -> bool { + self.failed_at + .as_ref() + .is_some_and(|refused_from| amount_in >= refused_from) + } + + /// Records what the pool paid for `amount_in`. + /// + /// A pool that turns an amount down for being over its limit turns down every larger amount + /// too, so `refused_from` is set to this amount (or lowered to it) and larger amounts are + /// refused without asking the pool. + /// + /// Three things must hold first, because a refusal is not always about size: + /// + /// * the pool rejected the input itself rather than failing ([`Refusal::OverLimit`]) — a pool + /// that blew up on one input still serves every other; + /// * the pool served some smaller amount, so this really is a limit — a swap can also fail for + /// being too small to quote, and that fails at the opposite end; + /// * the pool served no larger amount, which would prove it does not refuse by size at all. + fn record( + &mut self, + insert_at: usize, + amount_in: &BigUint, + outcome: Result, + ) { + let refusal = outcome.as_ref().err().copied(); + self.amounts_and_results + .insert(insert_at, (amount_in.clone(), outcome.ok())); + if refusal != Some(Refusal::OverLimit) { + return; + } + + let was_served = |(_, outcome): &(BigUint, Option)| outcome.is_some(); + let served_below = self.amounts_and_results[..insert_at] + .iter() + .any(was_served); + let served_above = self.amounts_and_results[insert_at + 1..] + .iter() + .any(was_served); + if !served_below || served_above { + return; + } + + let lowest_refused = match self.failed_at.take() { + Some(already_refused) if already_refused <= *amount_in => already_refused, + _ => amount_in.clone(), + }; + self.failed_at = Some(lowest_refused); + } +} + +/// Swaps already made, so a pool asked the same question twice is only simulated once. +/// +/// **Only for swaps that read untouched component state.** Every answer here is kept for the whole +/// solve, which is sound exactly while nothing commits a swap back into the state being read. The +/// chunked water-fills do commit — they ask one pool the same question repeatedly and depend on a +/// worse answer each time as it is drained — so they simulate against their own overlay and must +/// not come through here. +pub struct SwapCache<'a> { + by_direction: FxHashMap, SwappedAmounts>, +} + +impl<'a> SwapCache<'a> { + pub(crate) fn new() -> Self { + Self { by_direction: FxHashMap::default() } + } + + /// What `direction` pays for `amount_in`. + /// + /// Answers from the amounts already asked of that pool where it can, reading across two of them + /// when the asking pass allows it, and otherwise calls `simulate` and keeps the result. Every + /// route through here is booked against the component and the pass, so the report separates + /// what was simulated from what was reused and from what was read across. + pub(crate) fn swap( + &mut self, + direction: PoolDirection<'a>, + amount_in: &BigUint, + label: &'static str, + simulate: impl FnOnce() -> Result, + may_interpolate: bool, + ) -> Option { + let component_id = direction.component_id; + let amounts_swapped = self + .by_direction + .entry(direction) + .or_default(); + + let insert_at = match amounts_swapped + .amounts_and_results + .binary_search_by(|(amount, _)| amount.cmp(amount_in)) + { + Ok(asked_before) => { + sim_meter::record_cache_hit(component_id, label); + return amounts_swapped.amounts_and_results[asked_before] + .1 + .clone(); + } + Err(insert_at) => insert_at, + }; + + // Asking a pool for more than it has already turned down buys the same refusal again, and + // on a `vm:` pool that is as expensive as a swap it would have served. + if amounts_swapped.refuses(amount_in) { + sim_meter::record_refusal_without_calling(component_id, label); + return None; + } + + if may_interpolate { + if let Some(read_across) = Self::interpolate(amounts_swapped, insert_at, amount_in) { + sim_meter::record_interpolation(component_id, label); + return Some(read_across); + } + } + + // Only a simulated amount is kept. Keeping one that was itself read across would let the + // error compound, each reading drifting further from the pool's own curve. + let outcome = simulate(); + amounts_swapped.record(insert_at, amount_in, outcome.clone()); + outcome.ok() + } + + /// What the pool would pay for `amount_in`, read across the amounts either side of it. + /// + /// Output against input is concave for a pool — each further unit in buys less out — so the + /// straight line between two amounts runs below the pool's own curve. Reading across it + /// therefore comes out a little low, never high, and a path can only lose a ranking it + /// deserved rather than win one it did not. + /// + /// That only holds between two amounts. Past the largest one asked, the same line runs above + /// the curve, because it carries a price the pool no longer offers — so those are simulated. + /// + /// The gap between the two amounts must also be within [`INTERPOLATION_GAP_PERCENT`] of the + /// amount asked for. + /// A wider gap is where the line drifts furthest from the curve, and the gap narrows on its + /// own: a request this turns away is simulated, and that amount lands between the two, + /// leaving a closer pair behind for the next pass. + /// + /// Gas takes the larger amount's figure, not the nearer one's. Gas does not climb smoothly — a + /// crossed tick costs what it costs — and a swap never needs less gas than a smaller one, so + /// the larger amount's figure is never under the truth. Ranking subtracts gas from output, so + /// understating it here would overstate a path's net and let it take a place it had not + /// earned, which is the one thing this must not do. + fn interpolate( + amounts_swapped: &SwappedAmounts, + insert_at: usize, + amount_in: &BigUint, + ) -> Option { + let (lower_amount, lower) = amounts_swapped + .amounts_and_results + .get(insert_at.checked_sub(1)?)?; + let (upper_amount, upper) = amounts_swapped + .amounts_and_results + .get(insert_at)?; + let (lower, upper) = (lower.as_ref()?, upper.as_ref()?); + + let amount_gap = upper_amount - lower_amount; + if &amount_gap * 100u32 > amount_in * INTERPOLATION_GAP_PERCENT { + return None; + } + // A pool paying less for more is not the concave curve this reads across. + if upper.amount_out < lower.amount_out { + return None; + } + + let output_gap = &upper.amount_out - &lower.amount_out; + let amount_past_lower = amount_in - lower_amount; + let amount_out = &lower.amount_out + output_gap * amount_past_lower / amount_gap; + Some(SwapResult { amount_out, gas: upper.gas.clone() }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::algorithm::test_utils::addr; + + /// Stage labels the cache is asked under. The cache only groups by the label and obeys the + /// interpolation flag, so the tests name them here rather than reaching for a solve's stages. + const RANKING: &str = "ranking"; + const COMMITTING: &str = "chunking"; + const EXCHANGE: &str = "exchange"; + const INTERPOLATES: bool = true; + const NO_INTERPOLATION: bool = false; + + // ==================== Reading across known amounts ==================== + + fn hop(amount_out: u64, gas: u64) -> SwapResult { + SwapResult { amount_out: BigUint::from(amount_out), gas: BigUint::from(gas) } + } + + /// A cache holding `amounts` for one pool direction, with no refusal point recorded. + fn cache_holding(amounts: Vec<(u64, Option)>) -> SwappedAmounts { + SwappedAmounts { + amounts_and_results: amounts + .into_iter() + .map(|(amount, outcome)| (BigUint::from(amount), outcome)) + .collect(), + failed_at: None, + } + } + + /// Where `amount` would be inserted into a cache's ascending amounts. + fn insert_at(swapped: &SwappedAmounts, amount: &BigUint) -> usize { + swapped + .amounts_and_results + .binary_search_by(|(known, _)| known.cmp(amount)) + .expect_err("amount must not already be recorded") + } + + fn read_across(swapped: &SwappedAmounts, amount: u64) -> Option { + let amount = BigUint::from(amount); + SwapCache::interpolate(swapped, insert_at(swapped, &amount), &amount) + } + + /// Halfway between two amounts reads back halfway between their outputs. + #[test] + fn test_interpolate_reads_across_two_amounts() { + let swapped = cache_holding(vec![(1000, Some(hop(2000, 50))), (1040, Some(hop(2080, 90)))]); + + let across = read_across(&swapped, 1020).expect("bracketed and inside the gap"); + + assert_eq!(across.amount_out, BigUint::from(2040u64)); + } + + /// Gas comes from the larger amount, never the nearer one: understating gas would overstate a + /// path's output net of gas, which is the one direction reading across must not err in. + #[test] + fn test_interpolate_takes_gas_from_the_larger_amount() { + let swapped = cache_holding(vec![(1000, Some(hop(2000, 50))), (1040, Some(hop(2080, 90)))]); + + let nearer_the_lower = read_across(&swapped, 1001).expect("bracketed and inside the gap"); + + assert_eq!(nearer_the_lower.gas, BigUint::from(90u64)); + } + + /// Amounts further apart than `INTERPOLATION_GAP_PERCENT` are left to the pool. + #[test] + fn test_interpolate_declines_a_wide_gap() { + let swapped = cache_holding(vec![(1000, Some(hop(2000, 50))), (1500, Some(hop(2600, 90)))]); + + assert!(read_across(&swapped, 1200).is_none()); + } + + /// Above every amount asked, the straight line carries a price the pool no longer offers, so + /// there is nothing to read across. + #[test] + fn test_interpolate_declines_above_the_largest_amount() { + let swapped = cache_holding(vec![(1000, Some(hop(2000, 50))), (1040, Some(hop(2080, 90)))]); + + assert!(read_across(&swapped, 1050).is_none()); + } + + /// A pool paying less for more is not the curve this reads across. + #[test] + fn test_interpolate_declines_when_output_falls() { + let swapped = cache_holding(vec![(1000, Some(hop(2000, 50))), (1040, Some(hop(1900, 90)))]); + + assert!(read_across(&swapped, 1020).is_none()); + } + + /// A refused amount either side leaves nothing to read across. + #[test] + fn test_interpolate_declines_across_a_refusal() { + let swapped = cache_holding(vec![(1000, Some(hop(2000, 50))), (1040, None)]); + + assert!(read_across(&swapped, 1020).is_none()); + } + + // ==================== Refusals reaching upwards ==================== + + fn record(swapped: &mut SwappedAmounts, amount: u64, outcome: Result) { + let amount = BigUint::from(amount); + let at = insert_at(swapped, &amount); + swapped.record(at, &amount, outcome); + } + + /// A refusal above an amount the pool served is taken to refuse everything larger. + #[test] + fn test_refusal_above_a_served_amount_reaches_upwards() { + let mut swapped = cache_holding(vec![(1000, Some(hop(2000, 50)))]); + + record(&mut swapped, 2000, Err(Refusal::OverLimit)); + + assert!(swapped.refuses(&BigUint::from(2000u64))); + assert!(swapped.refuses(&BigUint::from(5000u64))); + assert!(!swapped.refuses(&BigUint::from(1500u64))); + } + + /// A refusal with nothing served below it may be an amount too small to quote rather than a + /// limit, so it stands only for itself. + #[test] + fn test_refusal_with_nothing_served_below_stands_alone() { + let mut swapped = cache_holding(vec![]); + + record(&mut swapped, 1000, Err(Refusal::OverLimit)); + + assert!(!swapped.refuses(&BigUint::from(5000u64))); + } + + /// A larger amount the pool did serve says it does not refuse upwards at all. + #[test] + fn test_refusal_below_a_served_amount_does_not_reach_upwards() { + let mut swapped = + cache_holding(vec![(1000, Some(hop(2000, 50))), (3000, Some(hop(5000, 50)))]); + + record(&mut swapped, 2000, Err(Refusal::OverLimit)); + + assert!(!swapped.refuses(&BigUint::from(4000u64))); + } + + // ==================== The cache deciding when a pool is called ==================== + + /// Drives `SwapCache::swap` with a closure that counts its calls, so a test can assert whether + /// the pool was reached at all — which is the whole point of the cache. + struct CountingPool { + component_id: ComponentId, + address_in: Address, + address_out: Address, + calls: std::cell::Cell, + answer: Option, + } + + impl CountingPool { + fn paying(amount_out: u64) -> Self { + Self { + component_id: ComponentId::from("pool"), + address_in: addr(0x01), + address_out: addr(0x02), + calls: std::cell::Cell::new(0), + answer: Some(hop(amount_out, 10)), + } + } + + fn refusing() -> Self { + Self { answer: None, ..Self::paying(0) } + } + + fn direction(&self) -> PoolDirection<'_> { + PoolDirection { + component_id: &self.component_id, + address_in: &self.address_in, + address_out: &self.address_out, + } + } + + fn ask<'a>( + &'a self, + cache: &mut SwapCache<'a>, + amount: u64, + label: &'static str, + may_interpolate: bool, + ) -> Option { + cache.swap( + self.direction(), + &BigUint::from(amount), + label, + || { + self.calls.set(self.calls.get() + 1); + self.answer + .clone() + .ok_or(Refusal::OverLimit) + }, + may_interpolate, + ) + } + } + + /// The same amount asked twice reaches the pool once; the second answer is the first one. + #[test] + fn test_swap_answers_a_repeated_amount_without_calling() { + let pool = CountingPool::paying(2000); + let mut cache = SwapCache::new(); + + let first = pool.ask(&mut cache, 1000, COMMITTING, NO_INTERPOLATION); + let second = pool.ask(&mut cache, 1000, COMMITTING, NO_INTERPOLATION); + + assert_eq!(pool.calls.get(), 1); + assert_eq!(first.map(|h| h.amount_out), Some(BigUint::from(2000u64))); + assert_eq!(second.map(|h| h.amount_out), Some(BigUint::from(2000u64))); + } + + /// Past an amount the pool refused for being over its limit, larger amounts are refused + /// without asking it again. + #[test] + fn test_swap_short_circuits_above_a_refusal() { + let pool = CountingPool::refusing(); + let mut cache = SwapCache::new(); + // Serve a smaller amount first: a refusal only reaches upwards once one below it worked. + cache.swap( + pool.direction(), + &BigUint::from(500u64), + COMMITTING, + || Ok(hop(1000, 10)), + NO_INTERPOLATION, + ); + pool.ask(&mut cache, 1000, COMMITTING, NO_INTERPOLATION); + let calls_after_refusal = pool.calls.get(); + + let larger = pool.ask(&mut cache, 5000, COMMITTING, NO_INTERPOLATION); + + assert!(larger.is_none()); + assert_eq!(pool.calls.get(), calls_after_refusal, "the pool was asked again"); + } + + /// A pass that may read across two nearby amounts gets an answer without a call; one that may + /// not is simulated for real. This is the gate that keeps approximated amounts out of the + /// passes that commit and report. + #[test] + fn test_swap_interpolates_only_for_a_pass_that_allows_it() { + let interpolating = CountingPool::paying(0); + let mut cache = SwapCache::new(); + cache.swap( + interpolating.direction(), + &BigUint::from(1000u64), + RANKING, + || Ok(hop(1000, 10)), + INTERPOLATES, + ); + cache.swap( + interpolating.direction(), + &BigUint::from(1100u64), + RANKING, + || Ok(hop(1100, 10)), + INTERPOLATES, + ); + let calls_before = interpolating.calls.get(); + + let read_across = interpolating.ask(&mut cache, 1050, RANKING, INTERPOLATES); + assert_eq!(interpolating.calls.get(), calls_before, "ranking should not have called"); + assert_eq!(read_across.map(|h| h.amount_out), Some(BigUint::from(1050u64))); + + let simulated = interpolating.ask(&mut cache, 1060, EXCHANGE, NO_INTERPOLATION); + assert_eq!(interpolating.calls.get(), calls_before + 1, "exchange must call the pool"); + assert_eq!(simulated.map(|h| h.amount_out), Some(BigUint::from(0u64))); + } + + /// An interpolated answer is not kept, so it can never be read across again and compound. The + /// same request a second time still reaches the pool. + #[test] + fn test_swap_does_not_store_an_interpolated_answer() { + let pool = CountingPool::paying(7777); + let mut cache = SwapCache::new(); + cache.swap( + pool.direction(), + &BigUint::from(1000u64), + RANKING, + || Ok(hop(1000, 10)), + INTERPOLATES, + ); + cache.swap( + pool.direction(), + &BigUint::from(1100u64), + RANKING, + || Ok(hop(1100, 10)), + INTERPOLATES, + ); + pool.ask(&mut cache, 1050, RANKING, INTERPOLATES); + + let asked_again = pool.ask(&mut cache, 1050, EXCHANGE, NO_INTERPOLATION); + + assert_eq!(pool.calls.get(), 1, "the interpolated answer should not have been stored"); + assert_eq!(asked_again.map(|h| h.amount_out), Some(BigUint::from(7777u64))); + } + + /// A pool that failed rather than rejected the amount says nothing about larger amounts: the + /// panic guard reports a component that blew up on one input as an error like any other, and + /// reading a limit out of that would drop a working pool for the rest of the solve. + #[test] + fn test_failure_that_is_not_a_limit_does_not_reach_upwards() { + let mut swapped = cache_holding(vec![(1000, Some(hop(2000, 50)))]); + + record(&mut swapped, 2000, Err(Refusal::Failed)); + + assert!(!swapped.refuses(&BigUint::from(2000u64))); + assert!(!swapped.refuses(&BigUint::from(5000u64))); + } + + /// A second, lower refusal moves the point everything above is refused from down to it. + #[test] + fn test_lower_refusal_moves_the_refusal_point_down() { + let mut swapped = cache_holding(vec![(1000, Some(hop(2000, 50)))]); + + record(&mut swapped, 3000, Err(Refusal::OverLimit)); + record(&mut swapped, 2000, Err(Refusal::OverLimit)); + + assert!(swapped.refuses(&BigUint::from(2000u64))); + } +} diff --git a/fynd-core/src/algorithm/water_fill/config.rs b/fynd-core/src/algorithm/water_fill/config.rs new file mode 100644 index 00000000..a21674dc --- /dev/null +++ b/fynd-core/src/algorithm/water_fill/config.rs @@ -0,0 +1,39 @@ +/// Maximum candidate paths simulated per order after heuristic ranking. +pub(crate) const DEFAULT_MAX_CANDIDATES: usize = 5000; +/// Cap on candidates from the bounded amount-aware discovery added to the candidate set; +/// see the discovery section below. +pub(crate) const MAX_DISCOVERY_CANDIDATES: usize = 128; +/// How many of the top-ranked paths are built and simulated to settle the single-path baseline. +/// Ranking can read amounts across two nearby ones, so the top place it reports is not always the +/// best path; the baseline is the bar splits must beat, so it is taken from exact figures. +pub(crate) const BASELINE_CANDIDATES: usize = 8; +/// Maximum number of parallel paths in a split. +pub(crate) const DEFAULT_MAX_PATHS: usize = 4; +/// Chunk grid for the coarse set-selection pass. +pub(crate) const COARSE_CHUNKS: usize = 20; +/// Chunk grid for the fine allocation pass over the fixed active set. +pub(crate) const FINE_CHUNKS: usize = 256; +/// Number of top full-amount paths always considered for shared-component fill-and-spill. +pub(crate) const SHARED_FULL_PATHS: usize = 8; +/// Number of full-amount-ranked paths probed with the first chunk for fill-and-spill. +pub(crate) const SHARED_MARGIN_PROBE_PATHS: usize = 32; +/// Number of marginal-probe winners added to the fill-and-spill candidate set. +pub(crate) const SHARED_MARGIN_PATHS: usize = 8; +/// Upper bound on fill-and-spill candidate paths. +pub(crate) const SHARED_MAX_CANDIDATES: usize = 12; +/// Candidate states retained per intermediate token during bounded discovery expansion. +pub(crate) const CANDIDATE_STATES_PER_NODE: usize = 4; +/// Candidate edge expansions from one path state during discovery. +pub(crate) const CANDIDATE_EDGES_PER_STATE: usize = 16; +/// Parallel components kept for a discovery edge directly into the target token. +pub(crate) const CANDIDATE_DIRECT_EDGES_PER_TOKEN: usize = 4; +/// Parallel components kept for a discovery edge into an anchor or configured connector token. +pub(crate) const CANDIDATE_CONNECTOR_EDGES_PER_TOKEN: usize = 2; +/// Number of highest-connectivity tokens taken as bounded-discovery anchors, derived per solve +/// from the graph (see `derive_anchor_tokens`). +pub(crate) const DERIVED_ANCHOR_COUNT: usize = 16; +/// Exchange-refinement step floor: the pass stops once `delta` falls below one fine chunk divided +/// by this factor, i.e. `amount_in / (fine_chunks * EXCHANGE_DELTA_FLOOR)`. +pub(crate) const EXCHANGE_DELTA_FLOOR: usize = 64; +/// Safety bound on trial simulations across the whole exchange-refinement pass. +pub(crate) const EXCHANGE_MAX_SIMS: usize = 400; diff --git a/fynd-core/src/algorithm/water_fill.rs b/fynd-core/src/algorithm/water_fill/mod.rs similarity index 61% rename from fynd-core/src/algorithm/water_fill.rs rename to fynd-core/src/algorithm/water_fill/mod.rs index b12b35e3..e1fa9e6b 100644 --- a/fynd-core/src/algorithm/water_fill.rs +++ b/fynd-core/src/algorithm/water_fill/mod.rs @@ -28,93 +28,55 @@ //! (including the native-ETH sentinel) survive the spot × depth cutoff. Every returned route is //! assembled through the shared split primitives and can be encoded on-chain. +mod config; +mod models; + use std::{ cmp::{Ordering, Reverse}, time::{Duration, Instant}, }; +use models::{ + CandidatePathState, CandidateSearchConfig, Deadline, Discovery, ExchangeMove, + FullAmountOutcome, FullAmountRanking, ScoredEdge, SetupResult, SolveInput, SolveStage, + SplitCandidate, StepResult, +}; use num_bigint::{BigInt, BigUint}; use num_traits::Zero; use petgraph::{graph::NodeIndex, prelude::EdgeRef}; use rustc_hash::{FxHashMap, FxHashSet}; -use tracing::{debug, instrument}; +use tracing::{debug, instrument, trace}; use tycho_simulation::tycho_common::{models::Address, simulation::protocol_sim::ProtocolSim}; use super::{ most_liquid::DepthAndPrice, paths, - sim_guard::GuardedProtocolSim, - split_primitives::{build_split_route, HopDescriptor, PathAllocation, SimulatedHop}, + sim_meter::{self, MeteredProtocolSim}, + split_primitives::{ + build_split_route, HopDescriptor, MarketOverrides, PathAllocation, SimulatedHop, + }, Algorithm, AlgorithmConfig, NoPathReason, }; use crate::{ + algorithm::{ + paths::read_market, + swap_cache::{PoolDirection, Refusal, SwapCache, SwapResult}, + water_fill::config::{ + BASELINE_CANDIDATES, CANDIDATE_CONNECTOR_EDGES_PER_TOKEN, + CANDIDATE_DIRECT_EDGES_PER_TOKEN, CANDIDATE_EDGES_PER_STATE, CANDIDATE_STATES_PER_NODE, + COARSE_CHUNKS, DEFAULT_MAX_CANDIDATES, DEFAULT_MAX_PATHS, DERIVED_ANCHOR_COUNT, + EXCHANGE_DELTA_FLOOR, EXCHANGE_MAX_SIMS, FINE_CHUNKS, MAX_DISCOVERY_CANDIDATES, + SHARED_FULL_PATHS, SHARED_MARGIN_PATHS, SHARED_MARGIN_PROBE_PATHS, + SHARED_MAX_CANDIDATES, + }, + }, derived::{computation::ComputationRequirements, types::TokenGasPrices, SharedDerivedDataRef}, feed::market_data::{MarketData, MarketDataView, MarketState, StateLabel}, graph::{EdgeData, GraphQueryFilter, Path, TopologyGraph, TopologyGraphManager}, - types::{ComponentId, Order, Route, RouteResult}, + types::{ComponentId, Order, RouteResult}, AlgorithmError, }; -/// Maximum candidate paths simulated per order after heuristic ranking. -const DEFAULT_MAX_CANDIDATES: usize = 5000; -/// Cap on candidates from the bounded amount-aware discovery added to the candidate set -/// (matches the bounded discovery's own cap; see the discovery section below). -const BOUNDED_DISCOVERY_CANDIDATES: usize = 128; -/// Maximum number of parallel paths in a split. -const DEFAULT_MAX_PATHS: usize = 4; -/// Chunk grid for the coarse set-selection pass. -const COARSE_CHUNKS: usize = 20; -/// Chunk grid for the fine allocation pass over the fixed active set. -const FINE_CHUNKS: usize = 256; -/// Number of top full-amount paths always considered for shared-component fill-and-spill. -const SHARED_FULL_PATHS: usize = 8; -/// Number of full-amount-ranked paths probed with the first chunk for fill-and-spill. -const SHARED_MARGIN_PROBE_PATHS: usize = 32; -/// Number of marginal-probe winners added to the fill-and-spill candidate set. -const SHARED_MARGIN_PATHS: usize = 8; -/// Upper bound on fill-and-spill candidate paths. -const SHARED_MAX_CANDIDATES: usize = 12; -/// Candidate states retained per intermediate token during bounded discovery expansion. -const CANDIDATE_STATES_PER_NODE: usize = 4; -/// Candidate edge expansions from one path state during discovery. -const CANDIDATE_EDGES_PER_STATE: usize = 16; -/// Parallel components kept for a discovery edge directly into the target token. -const CANDIDATE_DIRECT_EDGES_PER_TOKEN: usize = 4; -/// Parallel components kept for a discovery edge into an anchor or configured connector token. -const CANDIDATE_CONNECTOR_EDGES_PER_TOKEN: usize = 2; -/// Number of highest-connectivity tokens taken as bounded-discovery anchors, derived per solve -/// from the graph (see `derive_anchor_tokens`). -const DERIVED_ANCHOR_COUNT: usize = 16; -/// Exchange-refinement step floor: the pass stops once `delta` falls below one fine chunk divided -/// by this factor, i.e. `amount_in / (fine_chunks * EXCHANGE_DELTA_FLOOR)`. -const EXCHANGE_DELTA_FLOOR: usize = 64; -/// Safety bound on trial simulations across the whole exchange-refinement pass. -const EXCHANGE_MAX_SIMS: usize = 400; - -/// A fully-built split candidate: the assembled route plus its summed gross output and gas. -struct SplitCandidate { - route: Route, - gross: BigUint, - gas: BigUint, -} - -impl SplitCandidate { - /// Net output in output-token terms (gross minus gas cost). - fn net( - &self, - gas_price: &BigUint, - token_prices: Option<&TokenGasPrices>, - token_out: &Address, - ) -> BigInt { - let cost = - WaterFillAlgorithm::gas_cost_in_token(&self.gas, gas_price, token_prices, token_out); - match cost { - Some(c) => BigInt::from(self.gross.clone()) - BigInt::from(c), - None => BigInt::from(self.gross.clone()), - } - } -} - /// Splits an order across component-disjoint (and, via fill-and-spill, component-sharing) paths to /// reduce price impact, returning the best net of the single path and several split allocations. pub struct WaterFillAlgorithm { @@ -125,48 +87,6 @@ pub struct WaterFillAlgorithm { max_paths: usize, } -/// A candidate reallocation in the exchange-refinement pass: shift one `delta` of input from the -/// over-allocated `donor` to the under-allocated `recipient`, carrying the two paths' recomputed -/// net outputs and the resulting gain in summed net output. -struct ExchangeMove { - donor: usize, - recipient: usize, - donor_net: BigInt, - recip_net: BigInt, - gain: BigInt, -} - -/// One simulated traversal of a path, with the resulting per-component states so they can be -/// committed. -struct StepResult { - amount_out: BigUint, - gas: BigUint, - new_states: Vec<(ComponentId, Box)>, -} - -/// Shared inputs threaded through every split-allocation pass: the ranked candidate paths, the -/// market snapshot, gas pricing, and the order under a single solve clock. Bundled so the -/// allocation methods take one context instead of the same six references each. -struct SplitContext<'a, 'g> { - ordered: &'a [Path<'g, DepthAndPrice>], - market: &'a MarketState, - gas_price: &'a BigUint, - token_prices: Option<&'a TokenGasPrices>, - order: &'a Order, - start: Instant, -} - -/// Output of the shared setup pass: candidate paths ranked by full-amount output, the market -/// subset they touch, the effective gas price, the best single path (if one fills the order), and -/// token gas prices for gas-aware ranking. -struct SetupResult<'a> { - ordered: Vec>, - market: MarketState, - gas_price: BigUint, - best_single: Option, - token_prices: Option, -} - impl WaterFillAlgorithm { /// Creates a `WaterFillAlgorithm` from an `AlgorithmConfig`. pub(crate) fn with_config(config: AlgorithmConfig) -> Result { @@ -188,10 +108,10 @@ impl WaterFillAlgorithm { /// Simulates `amount` through `path`, reading each component from `overlay` if present else the /// base state. Returns the output, summed gas, and the resulting per-component states so /// the caller can commit them into an overlay. - fn simulate_step( - path: &Path, + fn simulate_step<'g>( + path: &Path<'g, DepthAndPrice>, market: &MarketState, - overlay: &FxHashMap>, + overlay: &MarketOverrides, amount: BigUint, ) -> Option { let mut current = amount; @@ -200,14 +120,7 @@ impl WaterFillAlgorithm { // intra-path overlay carried across hops. That reuse is rare, so only pay the // per-hop state clone when the path actually repeats a component; the common case // skips the clone entirely. - let path_reuses_component = { - let mut seen: FxHashSet<&ComponentId> = - FxHashSet::with_capacity_and_hasher(path.len(), Default::default()); - !path - .edge_iter() - .iter() - .all(|e| seen.insert(&e.component_id)) - }; + let path_reuses_component = path_reuses_component(path); let mut intra_path_states: FxHashMap> = FxHashMap::default(); let mut new_states: Vec<(ComponentId, Box)> = @@ -217,17 +130,15 @@ impl WaterFillAlgorithm { let token_in = market.get_token(address_in)?; let token_out = market.get_token(address_out)?; let component_id = &edge.component_id; - let state = intra_path_states - .get(component_id) - .map(Box::as_ref) - .or_else(|| { - overlay - .get(component_id) - .map(Box::as_ref) - }) - .or_else(|| market.get_simulation_state(component_id))?; + let state = hop_state(market, component_id, Some(&intra_path_states), Some(overlay))?; let result = state - .get_amount_out_guarded(current.clone(), token_in, token_out) + .get_amount_out_metered( + component_id, + SolveStage::Chunking.label(), + current.clone(), + token_in, + token_out, + ) .ok()?; total_gas += &result.gas; if path_reuses_component { @@ -256,10 +167,15 @@ impl WaterFillAlgorithm { /// A path's gas converted to output-token terms as a signed amount, or zero when no gas price /// is available. Subtracted from gross output to get net, so gas-blind solves fall back to /// gross ranking rather than erroring. - fn activation_cost(ctx: &SplitContext, gas: &BigUint) -> BigInt { - Self::gas_cost_in_token(gas, ctx.gas_price, ctx.token_prices, ctx.order.token_out()) - .map(BigInt::from) - .unwrap_or_else(BigInt::zero) + fn activation_cost(input: &SolveInput, gas: &BigUint) -> BigInt { + Self::gas_cost_in_token( + gas, + &input.gas_price, + input.token_prices.as_ref(), + input.order.token_out(), + ) + .map(BigInt::from) + .unwrap_or_else(BigInt::zero) } /// Picks up to `max_paths` paths that share no component, so their outputs can be summed @@ -297,15 +213,15 @@ impl WaterFillAlgorithm { /// Shared setup: enumerate + rank candidates, simulate at full amount, pick the best single /// path if any (a path that fails at the full amount is kept as a split-only candidate). #[instrument(level = "debug", skip_all)] - async fn setup<'a>( + async fn setup<'o, 'g>( &self, - graph: &'a TopologyGraph, + graph: &'g TopologyGraph, market: MarketData, label: Option, derived: Option, - order: &Order, - start: Instant, - ) -> Result, AlgorithmError> { + order: &'o Order, + deadline: Deadline, + ) -> Result, AlgorithmError> { let token_prices = if let Some(ref derived) = derived { derived .read() @@ -316,6 +232,182 @@ impl WaterFillAlgorithm { None }; + let mut scored_paths = self.top_scored_paths(graph, order)?; + let mut joined_paths = Vec::new(); + + let market_view = read_market(&market, label).await?; + + // Bounded amount-aware discovery (see the discovery section below): union its + // candidates ahead of the pre-ranked set, so connector/anchor routes (incl. the + // native-ETH sentinel) survive the spot×depth truncation. Discovery failure is not + // fatal — the pre-ranked set already guarantees a route. + let anchor_tokens = derive_anchor_tokens(graph); + + // Discovery and ranking both swap against untouched state, so they share one cache: every + // frontier edge discovery simulates is an answer ranking would otherwise pay for again. + // It outlives setup because the allocation passes that read untouched state reuse it too. + let mut cache = SwapCache::new(); + sim_meter::start_solve(); + + let discovered_paths = discover_paths( + graph, + &market_view, + order, + &mut cache, + CandidateSearchConfig { + query: &self.query, + max_candidates: MAX_DISCOVERY_CANDIDATES, + anchor_tokens: &anchor_tokens, + source_token: order.token_in(), + deadline, + }, + ) + .inspect_err(|e| { + debug!(error = %e, "water-fill bounded discovery failed; using exhaustive candidates only") + }) + .unwrap_or_default(); + + let mut keys: FxHashSet> = scored_paths + .iter() + .map(path_key) + .collect(); + for path in discovered_paths { + if keys.insert(path_key(&path)) { + joined_paths.push(path); + } + } + joined_paths.append(&mut scored_paths); + + let component_ids: FxHashSet<&ComponentId> = joined_paths + .iter() + .flat_map(|p| { + p.edge_iter() + .iter() + .map(|e| &e.component_id) + }) + .collect(); + let market_state = market_view.extract_subset_with_overlay(&component_ids); + let gas_price = paths::fetch_gas_price(&market_state)?; + drop(market_view); + + let amount_in = order.amount().clone(); + // Holds the candidates in enumeration order to start with; ranking reorders them below. + let mut input = SolveInput { + ordered: joined_paths, + market: market_state, + gas_price, + token_prices, + order, + deadline, + }; + let ranking = self.rank_at_full_amount(&input, &mut cache); + + // The baseline is the bar every split has to beat, so it is settled on exact figures. + // Ranking may have read a path's amounts across two nearby ones, which understates and can + // demote a path that deserved the top place; taking the best of the top few as simulated + // costs a handful of builds and keeps the bar at the best single path really on offer. + // Building is what copies a component and a pool state per leg, which is why it is a few + // and not all of them. + let build_baseline = |path_ix: usize| { + paths::simulate_pool_path( + &input.ordered[path_ix], + &input.market, + input.token_prices.as_ref(), + amount_in.clone(), + ) + .ok() + }; + let best_single = ranking + .by_output_net_gas + .iter() + .take(BASELINE_CANDIDATES) + .filter_map(|&path_ix| build_baseline(path_ix)) + .max_by(|a, b| { + a.net_amount_out() + .cmp(b.net_amount_out()) + }) + .or_else(|| { + // None of the top few assembled; fall down the list for any route at all. + ranking + .by_output_net_gas + .iter() + .skip(BASELINE_CANDIDATES) + .find_map(|&path_ix| build_baseline(path_ix)) + }); + + // No early exit on a missing single path: a split across thin components can fill an order + // that no single path can, so the caller decides — it only errors when neither a + // single path nor a split candidate fills the order. + input.ordered = ranking + .by_output + .iter() + .map(|&path_ix| input.ordered[path_ix].clone()) + .collect(); + + debug!( + candidate_paths = input.ordered.len(), + elapsed_ms = deadline.elapsed().as_millis(), + "water-fill discovery + full-amount ranking" + ); + Ok(SetupResult { input, best_single, cache }) + } + + /// Simulates every path at the full order amount and ranks them by what they pay. + /// + /// The paths overlap heavily — thousands open with the same pool, all carrying the whole order + /// into it — so every hop goes through `cache` and each distinct swap costs one simulation no + /// matter how many paths make it. + /// + /// A path that crosses one pool twice is dropped instead of ranked: its second crossing would + /// have to see its own earlier swap, and every swap here reads untouched state. + /// + /// Nothing here builds a route. Only the baseline `setup` picks out of the ranking is worth + /// copying a component and a pool state per leg. + fn rank_at_full_amount<'g>( + &self, + input: &SolveInput<'_, 'g>, + cache: &mut SwapCache<'g>, + ) -> FullAmountRanking { + let paths = &input.ordered; + let order_amount = input.order.amount(); + let mut outcomes_by_path: Vec> = vec![None; paths.len()]; + + for (path_ix, path) in paths.iter().enumerate() { + if input.deadline.expired() { + break; + } + if path_reuses_component(path) { + continue; + } + outcomes_by_path[path_ix] = Some( + match simulate_path( + path, + &input.market, + cache, + order_amount.clone(), + SolveStage::Ranking, + ) { + Some(paid) => FullAmountOutcome::Filled(paid), + None => FullAmountOutcome::Unfilled, + }, + ); + } + + rank_outcomes( + outcomes_by_path, + &input.gas_price, + input.token_prices.as_ref(), + input.order.token_out(), + ) + } + + /// Every path the graph holds between the order's tokens, best spot-price-times-depth score + /// first, cut to `max_candidates`. A path no edge weight covers sorts behind every scored one. + fn top_scored_paths<'a>( + &self, + graph: &'a TopologyGraph, + order: &Order, + ) -> Result>, AlgorithmError> { let all_paths = paths::find_paths(graph, order.token_in(), order.token_out(), &self.query, None)?; if all_paths.is_empty() { @@ -326,140 +418,38 @@ impl WaterFillAlgorithm { }); } + let path_count = all_paths.len(); + let mut scored_count = 0usize; + + // A path no edge weight covers sorts behind every scored one rather than level with a + // path scored at zero: the score is a spot price times a depth, so zero is a real score + // that a weightless path has not earned. let mut scored: Vec<(Path, f64)> = all_paths .into_iter() - .map(|p| { - let s = paths::try_score_path(&p).unwrap_or(f64::MIN); - (p, s) + .map(|path| { + let score = match paths::try_score_path(&path) { + Some(score) => { + scored_count += 1; + score + } + None => f64::MIN, + }; + (path, score) }) .collect(); scored.sort_by(|(_, a), (_, b)| { b.partial_cmp(a) .unwrap_or(std::cmp::Ordering::Equal) }); - scored.truncate(self.max_candidates); - let mut paths: Vec> = scored - .into_iter() - .map(|(p, _)| p) - .collect(); - - let timeout_ms = self.timeout.as_millis() as u64; - let market = { - let view = match label.as_ref() { - Some(l) => market - .read_labeled(l) - .await - .map_err(|e| AlgorithmError::Other(e.to_string()))?, - None => market.read().await, - }; - if view.gas_price().is_none() { - return Err(AlgorithmError::DataNotFound { kind: "gas price", id: None }); - } - // Bounded amount-aware discovery (see the discovery section below): union its - // candidates ahead of the pre-ranked set, so connector/anchor routes (incl. the - // native-ETH sentinel) survive the spot×depth truncation. Discovery failure is not - // fatal — the pre-ranked set already guarantees a route. - let anchor_tokens = derive_anchor_tokens(graph); - let bounded = find_candidate_paths( - graph, - &view, - order, - CandidateSearchConfig { - query: &self.query, - max_candidates: BOUNDED_DISCOVERY_CANDIDATES, - anchor_tokens: &anchor_tokens, - source_token: order.token_in(), - start: &start, - timeout_ms, - }, - ); - match bounded { - Ok((bounded_paths, _)) => { - let mut keys: FxHashSet> = - paths.iter().map(path_key).collect(); - let mut union = Vec::with_capacity(bounded_paths.len() + paths.len()); - for path in bounded_paths { - if keys.insert(path_key(&path)) { - union.push(path); - } - } - union.append(&mut paths); - paths = union; - } - // Not fatal — the pre-ranked exhaustive set already guarantees a route — but log it - // so a misconfiguration or systematic discovery failure is visible rather than - // silently narrowing the candidate set. - Err(e) => { - debug!(error = %e, "water-fill bounded discovery failed; using exhaustive candidates only") - } - } - let component_ids: FxHashSet<&ComponentId> = paths - .iter() - .flat_map(|p| { - p.edge_iter() - .iter() - .map(|e| &e.component_id) - }) - .collect(); - let subset = view.extract_subset_with_overlay(&component_ids); - drop(view); - subset - }; - let gas_price = market - .gas_price() - .ok_or(AlgorithmError::DataNotFound { kind: "gas price", id: None })? - .effective_gas_price() - .clone(); - let amount_in = order.amount().clone(); - let mut best_single: Option = None; - let mut full_outputs: Vec<(usize, BigUint)> = Vec::new(); + trace!(path_count, scored_count, limit = self.max_candidates, "water-fill path scoring"); - for (idx, path) in paths.iter().enumerate() { - if start.elapsed().as_millis() as u64 > timeout_ms { - break; - } - match paths::simulate_pool_path(path, &market, token_prices.as_ref(), amount_in.clone()) - { - Ok(result) => { - let gross = result - .route() - .swaps() - .last() - .map(|s| s.amount_out().clone()) - .unwrap_or_else(BigUint::zero); - full_outputs.push((idx, gross)); - if best_single - .as_ref() - .map(|b| result.net_amount_out() > b.net_amount_out()) - .unwrap_or(true) - { - best_single = Some(result); - } - } - // A path that can't take the whole order (e.g. concentrated liquidity exhausted at - // the full amount) may still be worth a fraction in a split, so keep it as a - // split-only candidate ranked last (gross 0). It never becomes the single-path - // baseline, and the allocation passes skip it at any chunk it still fails. - Err(_) => full_outputs.push((idx, BigUint::zero())), - } - } + scored.truncate(self.max_candidates); - // No early exit on a missing single path: a split across thin components can fill an order - // that no single path can, so the caller decides — it only errors when neither a - // single path nor a split candidate fills the order. - full_outputs.sort_by(|(_, a), (_, b)| b.cmp(a)); - let ordered: Vec> = full_outputs + Ok(scored .into_iter() - .map(|(idx, _)| paths[idx].clone()) - .collect(); - - debug!( - candidate_paths = ordered.len(), - elapsed_ms = start.elapsed().as_millis(), - "water-fill discovery + full-amount ranking" - ); - Ok(SetupResult { ordered, market, gas_price, best_single, token_prices }) + .map(|(p, _)| p) + .collect()) } } @@ -480,23 +470,14 @@ impl Algorithm for WaterFillAlgorithm { derived: Option, order: &Order, ) -> Result { - let start = Instant::now(); + let deadline = Deadline::new(Instant::now(), self.timeout); if !order.is_sell() { return Err(AlgorithmError::ExactOutNotSupported); } - let SetupResult { ordered, market, gas_price, best_single, token_prices } = self - .setup(graph, market, label, derived, order, start) + let SetupResult { input, best_single, mut cache } = self + .setup(graph, market, label, derived, order, deadline) .await?; - let token_out = order.token_out(); - let ctx = SplitContext { - ordered: &ordered, - market: &market, - gas_price: &gas_price, - token_prices: token_prices.as_ref(), - order, - start, - }; // Build the split candidates; the best net of them competes with the single path. let mut candidates: Vec = Vec::new(); @@ -504,24 +485,26 @@ impl Algorithm for WaterFillAlgorithm { // split and the refined split, so run it once. It is cheap and always finishes, so // a tight timeout cannot cut it off while leaving the single path — a winning split // is never lost to the clock. - let disjoint = Self::select_disjoint(ctx.ordered, self.max_paths); + let disjoint = Self::select_disjoint(&input.ordered, self.max_paths); let coarse = (disjoint.len() >= 2) - .then(|| self.disjoint_waterfill(&ctx, &disjoint, COARSE_CHUNKS, true)) + .then(|| self.disjoint_waterfill(&input, &disjoint, COARSE_CHUNKS, true)) .flatten(); if let Some(coarse) = coarse.as_deref() { // The 20-chunk floor split: exactly the coarse allocation. - if let Some(c) = self.build_disjoint_legs(&ctx, &disjoint, coarse) { + if let Some(c) = self.build_disjoint_legs(&input, &disjoint, coarse) { candidates.push(c); } // Finer allocation over the same active set (a bonus; a timeout may cut it off, and // then the floor stands). - if let Some(c) = self.disjoint_refine(&ctx, &disjoint, coarse, FINE_CHUNKS) { + if let Some(c) = + self.disjoint_refine(&input, &disjoint, coarse, FINE_CHUNKS, &mut cache) + { candidates.push(c); } } // Fill-and-spill: a split that lets paths share a component and branch at an intermediate // token (a tree route), which the component-disjoint splits cannot express. - if let Some(c) = self.fillspill_alloc(&ctx, FINE_CHUNKS) { + if let Some(c) = self.fillspill_alloc(&input, FINE_CHUNKS, &mut cache) { candidates.push(c); } @@ -533,7 +516,7 @@ impl Algorithm for WaterFillAlgorithm { .map(|b| b.net_amount_out().clone()); let mut best: Option<(BigInt, SplitCandidate)> = None; for cand in candidates { - let net = cand.net(&gas_price, token_prices.as_ref(), token_out); + let net = cand.net(&input); let beats = match (&best, &baseline_net) { (Some((current, _)), _) => net > *current, (None, Some(base)) => net > *base, @@ -544,15 +527,16 @@ impl Algorithm for WaterFillAlgorithm { } } let split_won = best.is_some(); + sim_meter::report(&input.market, || deadline.elapsed().as_millis() as u64); debug!( candidate_count, split_won, - elapsed_ms = start.elapsed().as_millis(), + elapsed_ms = deadline.elapsed().as_millis(), "water-fill selected {}", if split_won { "split candidate" } else { "single path" } ); match best { - Some((net, cand)) => Ok(RouteResult::new(cand.route, net, gas_price)), + Some((net, cand)) => Ok(RouteResult::new(cand.route, net, input.gas_price.clone())), // No split won: return the single path if there is one, else nothing fills the order. None => best_single.ok_or(AlgorithmError::InsufficientLiquidity), } @@ -575,12 +559,13 @@ impl WaterFillAlgorithm { /// amount), re-allocates over that set on a fine grid with the gate off (gas already /// justified), then runs the exchange-refinement pass. If a tight timeout cuts off either pass /// this returns `None` and the caller falls back to the 20-chunk floor candidate. - fn disjoint_refine( + fn disjoint_refine<'g>( &self, - ctx: &SplitContext, + input: &SolveInput<'_, 'g>, disjoint: &[usize], coarse: &[BigUint], fine_chunks: usize, + cache: &mut SwapCache<'g>, ) -> Option { // The active set is the coarse-gated paths that were allocated a nonzero amount. let active: Vec = disjoint @@ -595,13 +580,13 @@ impl WaterFillAlgorithm { } // Fine water-fill over the fixed active set, no gate (gas already justified). - let fine = self.disjoint_waterfill(ctx, &active, fine_chunks, false)?; + let fine = self.disjoint_waterfill(input, &active, fine_chunks, false)?; // Exchange refinement. The fine water-fill quantizes each path to a whole chunk, so it can // sit up to one chunk off the equal-marginal optimum. Nudge flow between paths at sub-chunk // resolution, accepting only strictly-improving moves (never-lose). - let refined = self.disjoint_exchange(ctx, &active, fine_chunks, fine); - self.build_disjoint_legs(ctx, &active, &refined) + let refined = self.disjoint_exchange(input, &active, fine_chunks, fine, cache); + self.build_disjoint_legs(input, &active, &refined) } /// Net output (gross output minus gas cost in output-token terms) of `path` simulated in @@ -609,18 +594,19 @@ impl WaterFillAlgorithm { /// re-simulation is exact. A zero amount means the path is dropped from the route: it /// yields no output and, since it is no longer swapped, no gas — so dropping a donor /// credits its saved gas automatically. - fn path_net( - ctx: &SplitContext, - path: &Path, + fn path_net<'g>( + input: &SolveInput<'_, 'g>, + path: &Path<'g, DepthAndPrice>, amount: &BigUint, + cache: &mut SwapCache<'g>, ) -> Option { if amount.is_zero() { return Some(BigInt::zero()); } - let empty: FxHashMap> = FxHashMap::default(); - let step = Self::simulate_step(path, ctx.market, &empty, amount.clone())?; - let activation = Self::activation_cost(ctx, &step.gas); - Some(BigInt::from(step.amount_out) - activation) + let result = + simulate_path(path, &input.market, cache, amount.clone(), SolveStage::Exchange)?; + let activation = Self::activation_cost(input, &result.gas); + Some(BigInt::from(result.amount_out) - activation) } /// Exchange-refinement pass over the fixed active set, starting from the fine water-fill split. @@ -631,41 +617,43 @@ impl WaterFillAlgorithm { /// component-disjoint, so a trial re-simulates only the two paths it touches (unchanged /// paths keep their cached net). Only strictly-improving moves are accepted, so the result /// never scores below the split it started from. - fn disjoint_exchange( + fn disjoint_exchange<'g>( &self, - ctx: &SplitContext, + input: &SolveInput<'_, 'g>, active: &[usize], fine_chunks: usize, alloc: Vec, + cache: &mut SwapCache<'g>, ) -> Vec { let path_count = active.len(); if path_count < 2 { return alloc; } - let amount_in = ctx.order.amount().clone(); + let amount_in = input.order.amount().clone(); let fine_chunks = fine_chunks.max(1); let mut delta = &amount_in / fine_chunks; let min_delta = &amount_in / (fine_chunks * EXCHANGE_DELTA_FLOOR); if delta.is_zero() { return alloc; } - let timeout_ms = self.timeout.as_millis() as u64; - let mut cum = alloc; - // Cache each active path's net at its current cumulative amount so a pair trial only - // re-simulates the two paths it moves flow between, not the whole active set. + let mut cumulative_amount_in = alloc; + // Cache each active path's net at its current cumulative_amount_in amount so a pair trial + // only re-simulates the two paths it moves flow between, not the whole active set. let mut net_cache: Vec = Vec::with_capacity(path_count); for (i, &path_idx) in active.iter().enumerate() { - let Some(net) = Self::path_net(ctx, &ctx.ordered[path_idx], &cum[i]) else { + let Some(net) = + Self::path_net(input, &input.ordered[path_idx], &cumulative_amount_in[i], cache) + else { // The starting split does not simulate cleanly; refining it is unsafe, so keep it. - return cum; + return cumulative_amount_in; }; net_cache.push(net); } let mut sims = 0usize; while delta >= min_delta && !delta.is_zero() { - if ctx.start.elapsed().as_millis() as u64 > timeout_ms || sims >= EXCHANGE_MAX_SIMS { + if input.deadline.expired() || sims >= EXCHANGE_MAX_SIMS { break; } @@ -674,22 +662,26 @@ impl WaterFillAlgorithm { if sims >= EXCHANGE_MAX_SIMS { break; } - if cum[donor] < delta { + if cumulative_amount_in[donor] < delta { continue; } - let donor_amt = &cum[donor] - δ - let Some(donor_net) = Self::path_net(ctx, &ctx.ordered[active[donor]], &donor_amt) + let donor_amt = &cumulative_amount_in[donor] - δ + let Some(donor_net) = + Self::path_net(input, &input.ordered[active[donor]], &donor_amt, cache) else { continue; }; sims += 1; for recipient in 0..path_count { - if recipient == donor || sims >= EXCHANGE_MAX_SIMS { + if sims >= EXCHANGE_MAX_SIMS { + break; + } + if recipient == donor { continue; } - let recip_amt = &cum[recipient] + δ + let recip_amt = &cumulative_amount_in[recipient] + δ let Some(recip_net) = - Self::path_net(ctx, &ctx.ordered[active[recipient]], &recip_amt) + Self::path_net(input, &input.ordered[active[recipient]], &recip_amt, cache) else { continue; }; @@ -720,20 +712,20 @@ impl WaterFillAlgorithm { delta = &delta / 2usize; continue; }; - cum[mv.donor] = &cum[mv.donor] - δ - cum[mv.recipient] = &cum[mv.recipient] + δ + cumulative_amount_in[mv.donor] = &cumulative_amount_in[mv.donor] - δ + cumulative_amount_in[mv.recipient] = &cumulative_amount_in[mv.recipient] + δ net_cache[mv.donor] = mv.donor_net; net_cache[mv.recipient] = mv.recip_net; } - cum + cumulative_amount_in } /// Simulates `amount` through `path`, reading and committing component states via `overrides`, /// and returns the allocation the route assembly consumes. - fn allocation_commit( - path: &Path, + fn allocation_commit<'g>( + path: &Path<'g, DepthAndPrice>, market: &MarketState, - overrides: &mut FxHashMap>, + overrides: &mut MarketOverrides, amount: BigUint, flow_fraction: f64, ) -> Option { @@ -745,12 +737,15 @@ impl WaterFillAlgorithm { let token_in = market.get_token(address_in)?; let token_out = market.get_token(address_out)?; let component_id = &edge.component_id; - let state = overrides - .get(component_id) - .map(Box::as_ref) - .or_else(|| market.get_simulation_state(component_id))?; + let state = hop_state(market, component_id, None, Some(overrides))?; let result = state - .get_amount_out_guarded(current.clone(), token_in, token_out) + .get_amount_out_metered( + component_id, + SolveStage::Assembly.label(), + current.clone(), + token_in, + token_out, + ) .ok()?; hops.push(SimulatedHop { descriptor: HopDescriptor::new( @@ -778,11 +773,11 @@ impl WaterFillAlgorithm { /// order, tycho-execution remainder-split convention, route token map) and derives the /// candidate's gross output and gas from the assembled route. fn candidate_from_allocations( - ctx: &SplitContext, + input: &SolveInput, allocations: &[PathAllocation], ) -> Option { - let route = build_split_route(allocations, ctx.market, ctx.order).ok()?; - let token_out = ctx.order.token_out(); + let route = build_split_route(allocations, &input.market, input.order).ok()?; + let token_out = input.order.token_out(); let gross = route .swaps() .iter() @@ -798,13 +793,13 @@ impl WaterFillAlgorithm { /// Builds one independent leg per path at its allocated amount. `subset` and `alloc` are /// aligned by index; component-disjoint paths never interfere, so each leg is a real /// independent simulation. - fn build_disjoint_legs( + fn build_disjoint_legs<'g>( &self, - ctx: &SplitContext, + input: &SolveInput<'_, 'g>, subset: &[usize], alloc: &[BigUint], ) -> Option { - let amount_in = ctx.order.amount().clone(); + let amount_in = input.order.amount().clone(); let mut allocations = Vec::new(); for (i, &path_idx) in subset.iter().enumerate() { if alloc[i].is_zero() { @@ -812,10 +807,10 @@ impl WaterFillAlgorithm { } // Fresh overrides per leg: legs are component-disjoint, but a component reused within // one path must still see its own first swap. - let mut overrides: FxHashMap> = FxHashMap::default(); + let mut overrides = MarketOverrides::empty(); let allocation = Self::allocation_commit( - &ctx.ordered[path_idx], - ctx.market, + &input.ordered[path_idx], + &input.market, &mut overrides, alloc[i].clone(), ratio(&alloc[i], &amount_in), @@ -825,107 +820,142 @@ impl WaterFillAlgorithm { if allocations.is_empty() { return None; } - Self::candidate_from_allocations(ctx, &allocations) + Self::candidate_from_allocations(input, &allocations) } /// Incremental water-fill over a set of component-disjoint paths. Returns the amount allocated /// to each path in `subset` order. With `gate`, a path only activates when its first chunk /// covers its gas; without it, every path is eligible (used once the active set is fixed). - fn disjoint_waterfill( + fn disjoint_waterfill<'g>( &self, - ctx: &SplitContext, + input: &SolveInput<'_, 'g>, subset: &[usize], num_chunks: usize, gate: bool, ) -> Option> { - let amount_in = ctx.order.amount().clone(); + let amount_in = input.order.amount().clone(); let num_chunks = num_chunks.max(1); let base_chunk = &amount_in / num_chunks; if base_chunk.is_zero() { return None; } let remainder = &amount_in - &base_chunk * num_chunks; - let timeout_ms = self.timeout.as_millis() as u64; let path_count = subset.len(); - let mut committed: Vec>> = (0..path_count) - .map(|_| FxHashMap::default()) + let mut committed: Vec = (0..path_count) + .map(|_| MarketOverrides::empty()) .collect(); - let mut cum_in: Vec = vec![BigUint::zero(); path_count]; + let mut cumulative_amount_in: Vec = vec![BigUint::zero(); path_count]; let mut activated: Vec = vec![!gate; path_count]; + // What each path last paid for a chunk. Only the path that wins a chunk commits anything, + // and these paths share no component, so every other path is asked the same question of the + // same untouched pools next chunk and pays the same. Its marginal is kept rather than + // simulated again. + let mut marginals: Vec> = (0..path_count).map(|_| None).collect(); + // The chunk every remembered marginal was priced at. The first one carries the remainder, + // so what the paths are asked changes once and nothing remembered still answers it. + let mut marginals_chunk: Option = None; + for chunk_idx in 0..num_chunks { - if ctx.start.elapsed().as_millis() as u64 > timeout_ms { + if input.deadline.expired() { break; } let chunk = if chunk_idx == 0 { &base_chunk + &remainder } else { base_chunk.clone() }; + if marginals_chunk.as_ref() != Some(&chunk) { + marginals + .iter_mut() + .for_each(|m| *m = None); + marginals_chunk = Some(chunk.clone()); + } - let mut best: Option<(usize, BigInt, StepResult)> = None; + let mut best: Option<(usize, BigInt)> = None; for (i, &path_idx) in subset.iter().enumerate() { - let Some(step) = Self::simulate_step( - &ctx.ordered[path_idx], - ctx.market, - &committed[i], - chunk.clone(), - ) else { + if marginals[i].is_none() { + marginals[i] = Self::simulate_step( + &input.ordered[path_idx], + &input.market, + &committed[i], + chunk.clone(), + ); + } + let Some(step) = marginals[i].as_ref() else { continue; }; let gross_marginal = BigInt::from(step.amount_out.clone()); let net_marginal = if activated[i] { gross_marginal } else { - let activation = Self::activation_cost(ctx, &step.gas); + let activation = Self::activation_cost(input, &step.gas); gross_marginal - activation }; if best .as_ref() - .map(|(_, m, _)| &net_marginal > m) + .map(|(_, m)| &net_marginal > m) .unwrap_or(true) { - best = Some((i, net_marginal, step)); + best = Some((i, net_marginal)); } } - let Some((best_i, _, step)) = best else { + let Some((best_i, _)) = best else { + break; + }; + // Take the winner's marginal rather than read it: its pools are about to move, so + // what it just paid no longer answers, and taking it is how that is forgotten. + let Some(step) = marginals[best_i].take() else { break; }; for (id, state) in step.new_states { committed[best_i].insert(id, state); } - cum_in[best_i] += &chunk; + cumulative_amount_in[best_i] += &chunk; activated[best_i] = true; } - Some(cum_in) + Some(cumulative_amount_in) } /// Selects fill-and-spill candidates: the top full-amount paths plus the best first-chunk /// marginal probes. The probe is what makes intermediate-token splits (tree routes) reachable: /// the extra path often ranks poorly at full size but wins on the margin. - fn select_shared_candidates(&self, ctx: &SplitContext) -> Vec { - let mut candidates: Vec = (0..ctx.ordered.len().min(SHARED_FULL_PATHS)).collect(); - let first_chunk = ctx.order.amount() / COARSE_CHUNKS; + fn select_shared_candidates<'g>( + &self, + input: &SolveInput<'_, 'g>, + cache: &mut SwapCache<'g>, + ) -> Vec { + let mut candidates: Vec = (0..input + .ordered + .len() + .min(SHARED_FULL_PATHS)) + .collect(); + let first_chunk = input.order.amount() / COARSE_CHUNKS; if first_chunk.is_zero() { return candidates; } - let timeout_ms = self.timeout.as_millis() as u64; - let empty: FxHashMap> = FxHashMap::default(); let mut marginal: Vec<(usize, BigInt)> = Vec::new(); - for (idx, path) in ctx + for (idx, path) in input .ordered .iter() .enumerate() .take(SHARED_MARGIN_PROBE_PATHS) { - if ctx.start.elapsed().as_millis() as u64 > timeout_ms { + if input.deadline.expired() { break; } - let Some(step) = Self::simulate_step(path, ctx.market, &empty, first_chunk.clone()) - else { + // Nothing is committed yet, so these probes read untouched state and go through the + // cache like every other swap that does. + let Some(probe) = simulate_path( + path, + &input.market, + cache, + first_chunk.clone(), + SolveStage::SetSelection, + ) else { continue; }; - let activation = Self::activation_cost(ctx, &step.gas); - marginal.push((idx, BigInt::from(step.amount_out) - activation)); + let activation = Self::activation_cost(input, &probe.gas); + marginal.push((idx, BigInt::from(probe.amount_out) - activation)); } marginal.sort_by(|(_, a), (_, b)| b.cmp(a)); for (idx, net) in marginal @@ -946,97 +976,146 @@ impl WaterFillAlgorithm { } /// Coarse set-selection then fine allocation with shared-component fill-and-spill. - fn fillspill_alloc(&self, ctx: &SplitContext, fine_chunks: usize) -> Option { - let candidates = self.select_shared_candidates(ctx); + /// + /// Only the probes that pick the candidate set can use `cache`: the two water-fill passes below + /// commit each chunk into an overlay and must see the pools they have already drained. + fn fillspill_alloc<'g>( + &self, + input: &SolveInput<'_, 'g>, + fine_chunks: usize, + cache: &mut SwapCache<'g>, + ) -> Option { + let candidates = self.select_shared_candidates(input, cache); if candidates.len() < 2 { return None; } - // Phase 1: coarse gated pass to choose the active candidate set. - let (coarse_counts, _) = self.fillspill_waterfill(ctx, &candidates, COARSE_CHUNKS, true)?; + // Phase 1: coarse gated pass to choose the active candidate set — the candidates that + // took at least one chunk. + let coarse = self.fillspill_waterfill(input, &candidates, COARSE_CHUNKS, true)?; + let took_a_chunk: FxHashSet = coarse.iter().map(|(i, _)| *i).collect(); let active: Vec = candidates .iter() .copied() - .zip(coarse_counts.iter()) - .filter(|(_, count)| **count > 0) - .map(|(idx, _)| idx) + .enumerate() + .filter(|(i, _)| took_a_chunk.contains(i)) + .map(|(_, idx)| idx) .collect(); if active.len() < 2 { return None; } // Phase 2: fine ungated pass over the active set, with the commit schedule for replay. - let (_, schedule) = self.fillspill_waterfill(ctx, &active, fine_chunks, false)?; + let schedule = self.fillspill_waterfill(input, &active, fine_chunks, false)?; if schedule.is_empty() { return None; } - self.build_fillspill_route(ctx, &active, &schedule) + self.build_fillspill_route(input, &active, &schedule) } - /// Incremental fill-and-spill water-fill over a single shared overlay. Returns the chunk count - /// each candidate received and the ordered commit schedule of `(active_index, chunk_amount)`. - #[allow(clippy::type_complexity)] - fn fillspill_waterfill( + /// Incremental fill-and-spill water-fill over a single shared overlay. Returns the ordered + /// commit schedule of `(subset_index, chunk_amount)`; which candidates took a chunk at all is + /// read off it. + fn fillspill_waterfill<'g>( &self, - ctx: &SplitContext, + input: &SolveInput<'_, 'g>, subset: &[usize], num_chunks: usize, gate: bool, - ) -> Option<(Vec, Vec<(usize, BigUint)>)> { - let amount_in = ctx.order.amount().clone(); + ) -> Option> { + let amount_in = input.order.amount().clone(); let num_chunks = num_chunks.max(1); let base_chunk = &amount_in / num_chunks; if base_chunk.is_zero() { return None; } let remainder = &amount_in - &base_chunk * num_chunks; - let timeout_ms = self.timeout.as_millis() as u64; - let mut overlay: FxHashMap> = FxHashMap::default(); + let mut overlay = MarketOverrides::empty(); let mut activated: Vec = vec![!gate; subset.len()]; let mut active_count = if gate { 0 } else { subset.len() }; - let mut counts: Vec = vec![0; subset.len()]; let mut schedule: Vec<(usize, BigUint)> = Vec::with_capacity(num_chunks); + // What each candidate last paid for a chunk, kept so a candidate the winning chunk did not + // touch is not asked the same question again. Candidates here may share pools — that is the + // point of fill-and-spill — so committing a chunk forgets every candidate crossing one of + // the pools it moved, not only the winner. + let mut marginals: Vec> = (0..subset.len()) + .map(|_| None) + .collect(); + // The chunk every remembered marginal was priced at. The first one carries the remainder, + // so what the candidates are asked changes once and nothing remembered still answers it. + let mut marginals_chunk: Option = None; for chunk_idx in 0..num_chunks { - if ctx.start.elapsed().as_millis() as u64 > timeout_ms { + if input.deadline.expired() { break; } let chunk = if chunk_idx == 0 { &base_chunk + &remainder } else { base_chunk.clone() }; + if marginals_chunk.as_ref() != Some(&chunk) { + marginals + .iter_mut() + .for_each(|m| *m = None); + marginals_chunk = Some(chunk.clone()); + } - let mut best: Option<(usize, BigInt, StepResult)> = None; + let mut best: Option<(usize, BigInt)> = None; for (i, &path_idx) in subset.iter().enumerate() { if !activated[i] && active_count >= self.max_paths { continue; } - let Some(step) = Self::simulate_step( - &ctx.ordered[path_idx], - ctx.market, - &overlay, - chunk.clone(), - ) else { + if marginals[i].is_none() { + marginals[i] = Self::simulate_step( + &input.ordered[path_idx], + &input.market, + &overlay, + chunk.clone(), + ); + } + let Some(step) = marginals[i].as_ref() else { continue; }; let gross_marginal = BigInt::from(step.amount_out.clone()); let net_marginal = if activated[i] { gross_marginal } else { - let activation = Self::activation_cost(ctx, &step.gas); + let activation = Self::activation_cost(input, &step.gas); gross_marginal - activation }; if best .as_ref() - .map(|(_, m, _)| &net_marginal > m) + .map(|(_, m)| &net_marginal > m) .unwrap_or(true) { - best = Some((i, net_marginal, step)); + best = Some((i, net_marginal)); } } - let Some((best_i, _, step)) = best else { + let Some((best_i, _)) = best else { break; }; + // Take the winner's marginal rather than read it: taking is how it is forgotten. + // `best` is only set while that entry holds a marginal, so the `else` cannot be + // reached; it stands in for an unwrap the crate's lints forbid. + let Some(step) = marginals[best_i].take() else { + break; + }; + + let pools_moved: FxHashSet<&ComponentId> = step + .new_states + .iter() + .map(|(id, _)| id) + .collect(); + for (i, &path_idx) in subset.iter().enumerate() { + if input.ordered[path_idx] + .edge_iter() + .iter() + .any(|e| pools_moved.contains(&e.component_id)) + { + marginals[i] = None; + } + } + for (id, state) in step.new_states { overlay.insert(id, state); } @@ -1044,22 +1123,22 @@ impl WaterFillAlgorithm { activated[best_i] = true; active_count += 1; } - counts[best_i] += 1; schedule.push((best_i, chunk)); } - Some((counts, schedule)) + + Some(schedule) } /// Rebuilds the fill-and-spill result as one leg per active path at its total allocated /// amount, committed sequentially (largest allocation first) against a shared overlay — the /// same execution model the router applies on-chain. - fn build_fillspill_route( + fn build_fillspill_route<'g>( &self, - ctx: &SplitContext, + input: &SolveInput<'_, 'g>, active: &[usize], schedule: &[(usize, BigUint)], ) -> Option { - let amount_in = ctx.order.amount().clone(); + let amount_in = input.order.amount().clone(); let mut cand_in: Vec = vec![BigUint::zero(); active.len()]; for (i, chunk) in schedule { cand_in[*i] += chunk; @@ -1072,19 +1151,173 @@ impl WaterFillAlgorithm { } execution_order.sort_by(|&a, &b| cand_in[b].cmp(&cand_in[a])); - let mut overrides: FxHashMap> = FxHashMap::default(); + let mut overrides = MarketOverrides::empty(); let mut allocations = Vec::new(); for i in execution_order { let allocation = Self::allocation_commit( - &ctx.ordered[active[i]], - ctx.market, + &input.ordered[active[i]], + &input.market, &mut overrides, cand_in[i].clone(), ratio(&cand_in[i], &amount_in), )?; allocations.push(allocation); } - Self::candidate_from_allocations(ctx, &allocations) + Self::candidate_from_allocations(input, &allocations) + } +} + +/// Whether `path` crosses the same component more than once. Such a path has to see its own +/// earlier swap, so it cannot be simulated against untouched component state. +fn path_reuses_component(path: &Path<'_, W>) -> bool { + let mut seen: FxHashSet<&ComponentId> = + FxHashSet::with_capacity_and_hasher(path.len(), Default::default()); + !path + .edge_iter() + .iter() + .all(|e| seen.insert(&e.component_id)) +} + +/// Swaps `amount_in` through one hop against the market's untouched state. `None` when the market +/// holds no token or state for the hop, or the pool refuses the swap. +fn simulate_hop( + market: &MarketState, + component_id: &ComponentId, + address_in: &Address, + address_out: &Address, + amount_in: &BigUint, + pass: SolveStage, +) -> Result { + let (Some(token_in), Some(token_out), Some(state)) = ( + market.get_token(address_in), + market.get_token(address_out), + market.get_simulation_state(component_id), + ) else { + return Err(Refusal::Failed); + }; + state + .get_amount_out_metered(component_id, pass.label(), amount_in.clone(), token_in, token_out) + .map(|result| SwapResult { amount_out: result.amount, gas: result.gas }) + .map_err(|error| Refusal::of(&error)) +} + +/// The state a hop swaps against: what this path has already moved, then what the pass has +/// committed, then the market as it stands. +/// +/// The precedence lives here rather than in each pass's own loop, so two passes cannot quietly +/// disagree about which state a hop sees. `intra_path` is `None` for a pass whose paths never +/// cross a pool twice, and `committed` is `None` for one that reads untouched state. +fn hop_state<'s>( + market: &'s MarketState, + component_id: &ComponentId, + intra_path: Option<&'s FxHashMap>>, + committed: Option<&'s MarketOverrides>, +) -> Option<&'s dyn ProtocolSim> { + intra_path + .and_then(|states| { + states + .get(component_id) + .map(Box::as_ref) + }) + .or_else(|| committed.and_then(|overrides| overrides.get(component_id))) + .or_else(|| market.get_simulation_state(component_id)) +} + +/// Swaps `amount_in` along `path` against untouched component state, going through `cache` so a +/// hop some other path already made is not made again. Returns what the path pays and its summed +/// gas, or `None` as soon as one hop refuses. +/// +/// The pass must not hand this a path that crosses one pool twice: the second crossing would +/// have to see the first one's swap, and every hop here reads untouched state. Both path sets this +/// runs on are already free of them — `rank_at_full_amount` drops them, and discovery never builds +/// one because `can_extend_path` refuses to repeat a component. +fn simulate_path<'a>( + path: &Path<'a, DepthAndPrice>, + market: &MarketState, + cache: &mut SwapCache<'a>, + amount_in: BigUint, + stage: SolveStage, +) -> Option { + // What the next hop swaps; once the last one is done, what the path pays out. + let mut hop_amount_in = amount_in; + let mut path_gas = BigUint::zero(); + + for (address_in, edge, address_out) in path.iter() { + let direction = PoolDirection { component_id: &edge.component_id, address_in, address_out }; + let hop = cache.swap( + direction, + &hop_amount_in, + stage.label(), + || { + simulate_hop( + market, + &edge.component_id, + address_in, + address_out, + &hop_amount_in, + stage, + ) + }, + stage.may_interpolate(), + )?; + hop_amount_in = hop.amount_out; + path_gas += hop.gas; + } + + Some(SwapResult { amount_out: hop_amount_in, gas: path_gas }) +} + +/// Turns the full-amount pass's per-path outcomes into the two orderings `setup` consumes. Both +/// are built in path order and sorted stably, so paths paying the same amount keep the order they +/// were enumerated in. A path with no outcome never ran — ranking timed out before reaching it, +/// or it crossed a pool twice — and appears in neither ordering. +fn rank_outcomes( + outcomes_by_path: Vec>, + gas_price: &BigUint, + token_prices: Option<&TokenGasPrices>, + token_out: &Address, +) -> FullAmountRanking { + // A path that took the whole order carries what it pays net of gas; one that could not takes + // no place among them, whatever its gas would have been. + let mut by_output: Vec<(usize, Option)> = Vec::with_capacity(outcomes_by_path.len()); + let mut by_output_net_gas: Vec<(usize, BigInt)> = Vec::new(); + + for (path_ix, outcome) in outcomes_by_path.into_iter().enumerate() { + let Some(outcome) = outcome else { + continue; + }; + let net_output = match outcome { + FullAmountOutcome::Filled(paid) => { + let gas_cost = WaterFillAlgorithm::gas_cost_in_token( + &paid.gas, + gas_price, + token_prices, + token_out, + ); + let net_output = BigInt::from(paid.amount_out.clone()) - + gas_cost.map_or_else(BigInt::zero, BigInt::from); + by_output_net_gas.push((path_ix, net_output.clone())); + Some(net_output) + } + FullAmountOutcome::Unfilled => None, + }; + by_output.push((path_ix, net_output)); + } + + // A path that could not take the whole order ranks below every path that did, however little + // that one is left with: net output can be negative once gas costs more than the path pays, + // and a path that filled for a loss still tells the split passes more than one that failed. + by_output.sort_by(|(_, a), (_, b)| b.cmp(a)); + by_output_net_gas.sort_by(|(_, a), (_, b)| b.cmp(a)); + FullAmountRanking { + by_output: by_output + .into_iter() + .map(|(path_ix, _)| path_ix) + .collect(), + by_output_net_gas: by_output_net_gas + .into_iter() + .map(|(path_ix, _)| path_ix) + .collect(), } } @@ -1118,62 +1351,19 @@ fn ratio(numerator: &BigUint, denominator: &BigUint) -> f64 { // (discovery only reads `component_id`s) so it runs on the production `DepthAndPrice` graph while // tests exercise it on a bare topology graph. -type RankedPathScores = Vec<(usize, BigInt)>; -type CandidatePathSet<'a, W = ()> = (Vec>, RankedPathScores); - -#[derive(Clone)] -struct CandidatePathState<'a, W> { - node: NodeIndex, - path: Path<'a, W>, - amount_out: BigUint, -} - -struct ScoredEdge<'a, W> { - target: NodeIndex, - edge: &'a EdgeData, - amount_out: BigUint, - priority: u8, -} - -/// Parameters for one bounded candidate-discovery run. -#[derive(Clone, Copy)] -struct CandidateSearchConfig<'a> { - /// The same hop bounds and connector set every other route search runs under. - query: &'a GraphQueryFilter, - max_candidates: usize, - anchor_tokens: &'a FxHashSet
, - source_token: &'a Address, - start: &'a Instant, - timeout_ms: u64, -} - -fn timed_out(start: &Instant, timeout_ms: u64) -> bool { - start.elapsed().as_millis() as u64 > timeout_ms -} - /// Runs the bounded discovery and returns the candidate paths plus their `(index, full-amount /// gross output)` ranking, best first. -fn find_candidate_paths<'a, W>( +fn discover_paths<'a, W>( graph: &'a TopologyGraph, market: &MarketDataView<'_>, order: &Order, + cache: &mut SwapCache<'a>, cfg: CandidateSearchConfig<'_>, -) -> Result, AlgorithmError> +) -> Result>, AlgorithmError> where W: Clone, { - if cfg.query.min_hops == 0 || cfg.query.min_hops > cfg.query.max_hops { - return Err(AlgorithmError::InvalidConfiguration { - reason: format!( - "invalid hop configuration: min_hops={} max_hops={}", - cfg.query.min_hops, cfg.query.max_hops, - ), - }); - } - let from_idx = - find_token_node(graph, order.token_in(), NoPathReason::SourceTokenNotInGraph, order)?; - let to_idx = - find_token_node(graph, order.token_out(), NoPathReason::DestinationTokenNotInGraph, order)?; + let (from_idx, to_idx) = get_token_ixs(graph, order)?; let mut found = Vec::new(); let mut frontier = vec![CandidatePathState { @@ -1183,7 +1373,7 @@ where }]; for _depth in 0..cfg.query.max_hops { - if timed_out(cfg.start, cfg.timeout_ms) || frontier.is_empty() { + if cfg.deadline.expired() || frontier.is_empty() { break; } let mut next_by_node: FxHashMap>> = @@ -1192,15 +1382,9 @@ where if state.node == to_idx && from_idx != to_idx { continue; } - expand_candidate_state( - graph, - market, - &cfg, - to_idx, - state, - &mut found, - &mut next_by_node, - ); + + let mut discovery = Discovery { graph, market, cfg: &cfg, cache }; + expand_candidate_state(&mut discovery, to_idx, state, &mut found, &mut next_by_node); } frontier = prune_candidate_frontier(next_by_node); } @@ -1208,25 +1392,31 @@ where rank_found_candidate_paths(found, cfg.max_candidates, order) } -fn find_token_node( +/// The graph nodes holding the order's sell and buy tokens. +/// +/// # Errors +/// +/// [`AlgorithmError::NoPath`] naming whichever of the two the graph does not hold. +fn get_token_ixs( graph: &TopologyGraph, - token: &Address, - reason: NoPathReason, order: &Order, -) -> Result { - graph - .get_token_ix(token) - .ok_or(AlgorithmError::NoPath { - from: order.token_in().clone(), - to: order.token_out().clone(), - reason, - }) +) -> Result<(NodeIndex, NodeIndex), AlgorithmError> { + let missing = |reason| AlgorithmError::NoPath { + from: order.token_in().clone(), + to: order.token_out().clone(), + reason, + }; + let from_idx = graph + .get_token_ix(order.token_in()) + .ok_or_else(|| missing(NoPathReason::SourceTokenNotInGraph))?; + let to_idx = graph + .get_token_ix(order.token_out()) + .ok_or_else(|| missing(NoPathReason::DestinationTokenNotInGraph))?; + Ok((from_idx, to_idx)) } fn expand_candidate_state<'a, W>( - graph: &'a TopologyGraph, - market: &MarketDataView<'_>, - cfg: &CandidateSearchConfig<'_>, + discovery: &mut Discovery<'a, '_, W>, target: NodeIndex, state: CandidatePathState<'a, W>, found: &mut Vec<(Path<'a, W>, BigUint)>, @@ -1234,9 +1424,11 @@ fn expand_candidate_state<'a, W>( ) where W: Clone, { - let edges = candidate_edges_for_state(graph, market, cfg, target, &state); + let cfg = discovery.cfg; + let graph = discovery.graph; + let edges = candidate_edges_for_state(discovery, target, &state); for candidate in edges { - if timed_out(cfg.start, cfg.timeout_ms) { + if cfg.deadline.expired() { break; } let mut path = state.path.clone(); @@ -1259,27 +1451,27 @@ fn expand_candidate_state<'a, W>( } fn candidate_edges_for_state<'a, W>( - graph: &'a TopologyGraph, - market: &MarketDataView<'_>, - cfg: &CandidateSearchConfig<'_>, + discovery: &mut Discovery<'a, '_, W>, target: NodeIndex, state: &CandidatePathState<'a, W>, ) -> Vec> { - let mut preferred = score_candidate_edges(graph, market, cfg, target, state, true); + let mut preferred = score_candidate_edges(discovery, target, state, true); if preferred.is_empty() { - preferred = score_candidate_edges(graph, market, cfg, target, state, false); + preferred = score_candidate_edges(discovery, target, state, false); } select_candidate_edges(preferred, CANDIDATE_EDGES_PER_STATE) } fn score_candidate_edges<'a, W>( - graph: &'a TopologyGraph, - market: &MarketDataView<'_>, - cfg: &CandidateSearchConfig<'_>, + discovery: &mut Discovery<'a, '_, W>, target: NodeIndex, state: &CandidatePathState<'a, W>, preferred_only: bool, ) -> Vec> { + let Discovery { graph, market, cfg, cache } = discovery; + let graph = *graph; + let market = *market; + let cfg = *cfg; let mut scored = Vec::new(); for edge in graph.edges(state.node) { let next_node = edge.target(); @@ -1294,16 +1486,25 @@ fn score_candidate_edges<'a, W>( if !can_extend_path(graph, state, next_node, target, pool, cfg) { continue; } - let Some(amount_out) = simulate_edge( - market, + let address_in = &graph[state.node]; + let address_out = &graph[next_node]; + let direction = + PoolDirection { component_id: &pool.component_id, address_in, address_out }; + let Some(hop) = cache.swap( + direction, &state.amount_out, - &graph[state.node], - pool, - &graph[next_node], + SolveStage::Discovery.label(), + || simulate_edge(market, &state.amount_out, address_in, pool, address_out), + SolveStage::Discovery.may_interpolate(), ) else { continue; }; - scored.push(ScoredEdge { target: next_node, edge: pool, amount_out, priority }); + scored.push(ScoredEdge { + target: next_node, + edge: pool, + amount_out: hop.amount_out, + priority, + }); } } scored @@ -1397,14 +1598,24 @@ fn simulate_edge( token_in_addr: &Address, edge: &EdgeData, token_out_addr: &Address, -) -> Option { - let token_in = market.get_token(token_in_addr)?; - let token_out = market.get_token(token_out_addr)?; - let state = market.get_simulation_state(&edge.component_id)?; +) -> Result { + let (Some(token_in), Some(token_out), Some(state)) = ( + market.get_token(token_in_addr), + market.get_token(token_out_addr), + market.get_simulation_state(&edge.component_id), + ) else { + return Err(Refusal::Failed); + }; state - .get_amount_out_guarded(amount.clone(), token_in, token_out) - .ok() - .map(|result| result.amount) + .get_amount_out_metered( + &edge.component_id, + SolveStage::Discovery.label(), + amount.clone(), + token_in, + token_out, + ) + .map(|result| SwapResult { amount_out: result.amount, gas: result.gas }) + .map_err(|error| Refusal::of(&error)) } fn select_candidate_edges( @@ -1458,7 +1669,7 @@ fn rank_found_candidate_paths<'a, W>( mut found: Vec<(Path<'a, W>, BigUint)>, max_candidates: usize, order: &Order, -) -> Result, AlgorithmError> { +) -> Result>, AlgorithmError> { found.sort_by(|(_, a), (_, b)| b.cmp(a)); let mut keys = FxHashSet::default(); let mut paths = Vec::new(); @@ -1483,7 +1694,7 @@ fn rank_found_candidate_paths<'a, W>( reason: NoPathReason::NoGraphPath, }); } - Ok((paths, scores)) + Ok(paths) } #[cfg(test)] @@ -1791,24 +2002,24 @@ mod tests { let start = Instant::now(); let view = market.read().await; - let (paths, scores) = find_candidate_paths( + let paths = discover_paths( graph_manager.graph(), &view, &order, + &mut SwapCache::new(), CandidateSearchConfig { query: &GraphQueryFilter { min_hops: 1, max_hops: 3, connector_tokens: None }, max_candidates: 128, anchor_tokens: &FxHashSet::default(), source_token: order.token_in(), - start: &start, - timeout_ms: 2000, + deadline: Deadline::new(start, Duration::from_millis(2000)), }, ) .expect("discovery finds candidates"); assert_eq!(paths.len(), 2, "both parallel components should be discovered"); // Scores are (path index, full-amount gross output), best first: the deeper component wins. - let best_path = &paths[scores[0].0]; + let best_path = &paths[0]; assert_eq!( best_path.edge_iter()[0].component_id, "z_strong_link_weth", @@ -1816,44 +2027,94 @@ mod tests { ); } - /// An invalid hop configuration is rejected before any graph work. - #[tokio::test] - async fn test_discovery_rejects_invalid_hop_configuration() { - let link = token_with_decimals(0x01, "LINK", 18); - let weth = token_with_decimals(0x02, "WETH", 18); - let (market, graph_manager) = setup_market_unweighted_topology(vec![( - "link_weth", - &link, - &weth, - Box::new(v2_component(2_000_000, 5_700)) as Box, - )]); - let order = Order::new( - link.address.clone(), - weth.address.clone(), - BigUint::from(1_000u64) * BigUint::from(10u64).pow(18), - OrderSide::Sell, - addr(0xFF), - ); + // ==================== Ranking the full-amount outcomes ==================== - let start = Instant::now(); - let view = market.read().await; - let result = find_candidate_paths( - graph_manager.graph(), - &view, - &order, - CandidateSearchConfig { - query: &GraphQueryFilter { min_hops: 0, max_hops: 3, connector_tokens: None }, - max_candidates: 128, - anchor_tokens: &FxHashSet::default(), - source_token: order.token_in(), - start: &start, - timeout_ms: 2000, - }, + fn hop(amount_out: u64, gas: u64) -> SwapResult { + SwapResult { amount_out: BigUint::from(amount_out), gas: BigUint::from(gas) } + } + + /// `Unfilled` paths rank last by output and are kept out of the net-of-gas ordering entirely, + /// so one can never become the single-path baseline. A path with no outcome is in neither. + #[test] + fn test_rank_outcomes_places_unfilled_and_missing_paths() { + let outcomes = vec![ + Some(FullAmountOutcome::Filled(hop(1000, 0))), + Some(FullAmountOutcome::Unfilled), + None, + Some(FullAmountOutcome::Filled(hop(3000, 0))), + ]; + + let ranking = rank_outcomes(outcomes, &BigUint::from(1u64), None, &addr(0x02)); + + assert_eq!(ranking.by_output, vec![3, 0, 1], "unfilled ranks last, missing is absent"); + assert_eq!(ranking.by_output_net_gas, vec![3, 0], "unfilled cannot be the baseline"); + } + + /// Both orderings take gas off, so a path paying more gross ranks below a cheaper one in each. + /// The candidate set the split passes are built from is sliced off `by_output`, and ranking + /// that gross put a long route above a short one on an output its extra swaps hand back. + #[test] + fn test_rank_outcomes_orders_by_output_net_of_gas() { + let token_out = addr(0x02); + let mut token_prices = TokenGasPrices::default(); + token_prices.insert( + token_out.clone(), + tycho_simulation::tycho_common::simulation::protocol_sim::Price::new( + BigUint::from(1u64), + BigUint::from(1u64), + ), ); - assert!( - matches!(result, Err(AlgorithmError::InvalidConfiguration { .. })), - "min_hops of 0 should be rejected", + let outcomes = vec![ + Some(FullAmountOutcome::Filled(hop(1000, 500))), + Some(FullAmountOutcome::Filled(hop(900, 10))), + ]; + + let ranking = + rank_outcomes(outcomes, &BigUint::from(1u64), Some(&token_prices), &token_out); + + assert_eq!(ranking.by_output, vec![1, 0], "the cheaper path ranks first"); + assert_eq!(ranking.by_output_net_gas, vec![1, 0], "and the baseline ordering agrees"); + } + + /// A path that filled but paid less than its gas ranks below every profitable path and above + /// every path that could not fill: net output goes negative, and an unfilled path has no net + /// at all rather than a zero that would float it over the loss-making ones. + #[test] + fn test_rank_outcomes_places_a_loss_making_path_above_an_unfilled_one() { + let token_out = addr(0x02); + let mut token_prices = TokenGasPrices::default(); + token_prices.insert( + token_out.clone(), + tycho_simulation::tycho_common::simulation::protocol_sim::Price::new( + BigUint::from(1u64), + BigUint::from(1u64), + ), ); + let outcomes = vec![ + Some(FullAmountOutcome::Unfilled), + // Gas costs more than it pays: net output is -400. + Some(FullAmountOutcome::Filled(hop(100, 500))), + Some(FullAmountOutcome::Filled(hop(900, 10))), + ]; + + let ranking = + rank_outcomes(outcomes, &BigUint::from(1u64), Some(&token_prices), &token_out); + + assert_eq!(ranking.by_output, vec![2, 1, 0], "unfilled ranks below a loss-making path"); + } + + /// With no price for the output token gas cannot be converted, so it is not taken off and the + /// two orderings agree. + #[test] + fn test_rank_outcomes_ignores_gas_it_cannot_price() { + let outcomes = vec![ + Some(FullAmountOutcome::Filled(hop(1000, 500))), + Some(FullAmountOutcome::Filled(hop(900, 10))), + ]; + + let ranking = rank_outcomes(outcomes, &BigUint::from(1u64), None, &addr(0x02)); + + assert_eq!(ranking.by_output_net_gas, vec![0, 1]); } /// Anchors are the most-connected tokens (highest component-edge degree) plus the native-ETH diff --git a/fynd-core/src/algorithm/water_fill/models.rs b/fynd-core/src/algorithm/water_fill/models.rs new file mode 100644 index 00000000..35e7e86b --- /dev/null +++ b/fynd-core/src/algorithm/water_fill/models.rs @@ -0,0 +1,233 @@ +use std::time::{Duration, Instant}; + +use num_bigint::{BigInt, BigUint}; +use petgraph::graph::NodeIndex; +use rustc_hash::FxHashSet; +use tycho_simulation::tycho_common::{models::Address, simulation::protocol_sim::ProtocolSim}; + +use crate::{ + algorithm::{ + most_liquid::DepthAndPrice, + sim_meter, + swap_cache::{SwapCache, SwapResult}, + WaterFillAlgorithm, + }, + derived::TokenGasPrices, + feed::market_data::{MarketDataView, MarketState}, + graph::{EdgeData, GraphQueryFilter, Path, TopologyGraph}, + types::RouteResult, + ComponentId, Order, Route, +}; + +/// A fully-built split candidate: the assembled route plus its summed gross output and gas. +pub struct SplitCandidate { + pub route: Route, + pub gross: BigUint, + pub gas: BigUint, +} + +impl SplitCandidate { + /// Net output in output-token terms (gross minus gas cost). + pub(crate) fn net(&self, input: &SolveInput<'_, '_>) -> BigInt { + let cost = WaterFillAlgorithm::gas_cost_in_token( + &self.gas, + &input.gas_price, + input.token_prices.as_ref(), + input.order.token_out(), + ); + match cost { + Some(c) => BigInt::from(self.gross.clone()) - BigInt::from(c), + None => BigInt::from(self.gross.clone()), + } + } +} + +/// A candidate reallocation in the exchange-refinement pass: shift one `delta` of input from the +/// over-allocated `donor` to the under-allocated `recipient`, carrying the two paths' recomputed +/// net outputs and the resulting gain in summed net output. +pub struct ExchangeMove { + pub donor: usize, + pub recipient: usize, + pub donor_net: BigInt, + pub recip_net: BigInt, + pub gain: BigInt, +} + +/// One simulated traversal of a path, with the resulting per-component states so they can be +/// committed. +pub struct StepResult { + pub amount_out: BigUint, + pub gas: BigUint, + pub new_states: Vec<(ComponentId, Box)>, +} + +/// What one path pays for the whole order. +#[derive(Clone)] +pub enum FullAmountOutcome { + /// The path took the whole order, paying what the swap reports over its hops. + Filled(SwapResult), + /// The path could not take the whole order. It still ranks by output, at zero, because a + /// fraction of the order may suit it; it cannot be the single-path baseline. + Unfilled, +} + +/// The two orderings the full-amount pass produces, both holding indices into the path list it +/// ranked. +pub struct FullAmountRanking { + /// Every path simulated, best output net of gas first — the same measure the baseline and the + /// finished split are judged on, so a long route does not rank above a short one on an output + /// its extra swaps hand straight back. One that could not take the whole order ranks last: it + /// may still be worth a fraction in a split. + pub by_output: Vec, + /// The paths that filled the order, best output net of gas first. The single-path baseline is + /// the first of these that builds into a route. + pub by_output_net_gas: Vec, +} + +/// What a solve reads: the ranked candidate paths, the market snapshot they touch, gas pricing, +/// and the order under a single solve clock. +/// +/// Owns the market and pricing rather than borrowing them, so setup can hand the whole thing back +/// and every allocation pass takes one argument instead of the same six. +pub struct SolveInput<'o, 'g> { + /// Candidate paths, best full-amount output net of gas first. + pub ordered: Vec>, + pub market: MarketState, + pub gas_price: BigUint, + pub token_prices: Option, + pub order: &'o Order, + pub deadline: Deadline, +} + +/// Output of the shared setup pass. +pub struct SetupResult<'o, 'g> { + /// Everything the allocation passes read. + pub input: SolveInput<'o, 'g>, + /// The best single path, when one fills the order — the bar every split has to beat. + pub best_single: Option, + /// Every untouched-state swap discovery and ranking already made, handed on so the allocation + /// passes that read untouched state do not repeat them. + pub cache: SwapCache<'g>, +} + +#[derive(Clone)] +pub struct CandidatePathState<'a, W> { + pub node: NodeIndex, + pub path: Path<'a, W>, + pub amount_out: BigUint, +} + +pub struct ScoredEdge<'a, W> { + pub target: NodeIndex, + pub edge: &'a EdgeData, + pub amount_out: BigUint, + pub priority: u8, +} + +/// Parameters for one bounded candidate-discovery run. +#[derive(Clone, Copy)] +pub struct CandidateSearchConfig<'a> { + /// The same hop bounds and connector set every other route search runs under. + pub query: &'a GraphQueryFilter, + pub max_candidates: usize, + pub anchor_tokens: &'a FxHashSet
, + pub source_token: &'a Address, + pub deadline: Deadline, +} + +/// What one bounded discovery run walks with: the graph, the market it prices against, the bounds +/// it runs under, and the cache its swaps go through. +/// +/// Held together because every step of the walk needs all four; passing them separately is what +/// pushed these signatures past what the reader can hold. +pub struct Discovery<'a, 'r, W> { + pub graph: &'a TopologyGraph, + pub market: &'r MarketDataView<'r>, + pub cfg: &'r CandidateSearchConfig<'r>, + pub cache: &'r mut SwapCache<'a>, +} + +/// The stage of a solve a swap was asked for. +/// +/// Recorded against every swap so the report can say how much of a solve sits where answers can be +/// reused, and how much is in the passes that read state they have committed to and can never come +/// through the cache. +#[derive(Clone, Copy, PartialEq, Eq, Hash)] +pub(crate) enum SolveStage { + /// Bounded discovery expanding its frontier. + Discovery, + /// Ranking every candidate path at the full order amount. + Ranking, + /// Choosing which candidates fill-and-spill will split across, by probing each with a first + /// chunk. + SetSelection, + /// Exchange refinement re-pricing a path on its own. + Exchange, + /// The chunked water-fills, which read what they have committed and never reach the cache. + Chunking, + /// Building the legs of a route that will be returned. + Assembly, +} + +impl SolveStage { + /// What the simulation report calls this pass. + pub(crate) fn label(self) -> sim_meter::StageLabel { + match self { + SolveStage::Discovery => "discovery", + SolveStage::Ranking => "ranking", + SolveStage::SetSelection => "set-selection", + SolveStage::Exchange => "exchange", + SolveStage::Chunking => "chunking", + SolveStage::Assembly => "assembly", + } + } + + /// Whether this pass can take an amount read across two nearby ones. + /// + /// The passes that settle which paths get split can: reading across errs low, so a path may + /// lose a place it deserved but never take one it did not, and every path they put forward is + /// simulated for real before anything is allocated to it. + /// + /// Discovery is held out even though it also only orders things. Its amounts feed the next hop + /// rather than staying with one path, so reading low compounds along the frontier and moves + /// which edges survive pruning — a different candidate set, not a differently ordered one. + /// + /// The rest cannot. Exchange refinement shifts flow on a strictly-improving comparison, where + /// a read-across amount could invent a gain that is not there; the chunked fills and the route + /// builders decide and report amounts that are handed back to the caller. + pub(crate) fn may_interpolate(self) -> bool { + match self { + SolveStage::Ranking | SolveStage::SetSelection => true, + SolveStage::Discovery | + SolveStage::Exchange | + SolveStage::Chunking | + SolveStage::Assembly => false, + } + } +} + +/// When a solve must stop, however many passes it has left. +/// +/// One value rather than a start instant and a budget carried side by side: every pass asks the +/// same question, and asking it of two fields is how they drift apart. +#[derive(Clone, Copy)] +pub struct Deadline { + pub start: Instant, + pub timeout: Duration, +} + +impl Deadline { + pub(crate) fn new(start: Instant, timeout: Duration) -> Self { + Self { start, timeout } + } + + /// Whether the solve has run past its budget. + pub(crate) fn expired(&self) -> bool { + self.start.elapsed() > self.timeout + } + + /// How long the solve has been running, for the lines that report it. + pub(crate) fn elapsed(&self) -> Duration { + self.start.elapsed() + } +}