From 5cadfd94b7c51ce0dea095c20a4c156afde791f2 Mon Sep 17 00:00:00 2001 From: Carlos Zanella Date: Tue, 18 Aug 2026 10:53:03 +0100 Subject: [PATCH 1/5] feat: Waterfill - Slim Shady version I treated the slow performance of this algorithm as if it was an offense to my mother, so I dig deeply in how to improve it. Compared to current main, it is like a 3x~5x speed. We will see once this is in dev for sure. The code was quite the AI spaghetti, with a lot of bad names, weird design and separation, questionable abstractions, etc. I had to work on these to even understand what the code was doing, before I could work on performance. What you see in this commit is a whole mix, I decided not to separate the AI clean-up from actual logic changes because it would take me another week to organize that in proper commits. This is far from ideal, so in an effort to make you focus on the important things to review I will describe the actual FEAT changes, and point at the code for you to look at. Here we go: swap_cache.rs - new file with a pool swap cache for repeated swap amounts can also interpolate. Interpolating was key here, for very small changes in input amount in relation to the cached value interpolation - caching can also interpolate amounts smartly. If amounts are close to the cached value and a boolean flag is used, a linear interpolation will be used to save from a pool simulation. errors and are also handed smartly. this also lives in swap_cache.rs. waterfill has solve "stages" and each of them decides whether interpolation can be used or not. the idea is to use interpolation when you dont need exact precision. we skip paths that use the same pool twice (at rank_at_full_amount). This is an edge case (very few >=3 token pools) and I dont think it ever makes sense to swap twice on the same pool, when you could just go to the final token directly. waterfill's single-path baseline is now decided based on this cache+interpolation ranking. results might change slightly but from running the benchmark there was no change at all in WF for 10k orders. Reuse some swap results at the chunking stages of waterfill. Here it is harder to use caching because amounts are quite different, we cannot interpolate, and state changes with each iteration. But still could take away some repetition from it. Timeout refactoring allows for finer resolution on solve tiems New swap-metrics feature, disabled by default, adds a MeteredProtocolSim which collects metrics in a thread-local cache for measuring specific protocol usage and solve time. Useful for local debugging. By default it is not activated. --- fynd-core/Cargo.toml | 4 + fynd-core/src/algorithm/bellman_ford.rs | 10 +- fynd-core/src/algorithm/mod.rs | 2 + fynd-core/src/algorithm/most_liquid.rs | 14 +- fynd-core/src/algorithm/paths.rs | 39 +- fynd-core/src/algorithm/sim_meter.rs | 406 ++++++ fynd-core/src/algorithm/split_primitives.rs | 8 + fynd-core/src/algorithm/swap_cache.rs | 557 +++++++ fynd-core/src/algorithm/water_fill/config.rs | 39 + .../{water_fill.rs => water_fill/mod.rs} | 1290 ++++++++++------- fynd-core/src/algorithm/water_fill/models.rs | 234 +++ 11 files changed, 2048 insertions(+), 555 deletions(-) create mode 100644 fynd-core/src/algorithm/sim_meter.rs create mode 100644 fynd-core/src/algorithm/swap_cache.rs create mode 100644 fynd-core/src/algorithm/water_fill/config.rs rename fynd-core/src/algorithm/{water_fill.rs => water_fill/mod.rs} (62%) create mode 100644 fynd-core/src/algorithm/water_fill/models.rs 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..9258858c 100644 --- a/fynd-core/src/algorithm/most_liquid.rs +++ b/fynd-core/src/algorithm/most_liquid.rs @@ -588,13 +588,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) @@ -873,11 +867,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 { 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 62% rename from fynd-core/src/algorithm/water_fill.rs rename to fynd-core/src/algorithm/water_fill/mod.rs index b12b35e3..e8602cbd 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::{ + CandidatePathSet, 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,168 @@ 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 { + let mut by_output: Vec<(usize, BigUint)> = 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 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)); + paid.amount_out + } + FullAmountOutcome::Unfilled => BigUint::zero(), + }; + by_output.push((path_ix, output)); + } + + 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 +1346,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> 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 +1368,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 +1377,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 +1387,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 +1419,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 +1446,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 +1481,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 +1593,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( @@ -1791,17 +1997,17 @@ mod tests { let start = Instant::now(); let view = market.read().await; - let (paths, scores) = find_candidate_paths( + let (paths, scores) = 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"); @@ -1816,44 +2022,66 @@ 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, - }, - ); - assert!( - matches!(result, Err(AlgorithmError::InvalidConfiguration { .. })), - "min_hops of 0 should be rejected", + 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"); + } + + /// Gas is taken off before the net-of-gas ordering, so a path paying more gross can rank below + /// a cheaper one — which is the whole reason for keeping a second ordering. + #[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), + ), ); + 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![0, 1], "gross output still puts the larger first"); + assert_eq!(ranking.by_output_net_gas, vec![1, 0], "net of gas reverses them"); + } + + /// 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..d2267208 --- /dev/null +++ b/fynd-core/src/algorithm/water_fill/models.rs @@ -0,0 +1,234 @@ +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 first. One that could not take the whole order ranks last + /// at zero: 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>, +} + +type RankedPathScores = Vec<(usize, BigInt)>; +pub type CandidatePathSet<'a, W = ()> = (Vec>, RankedPathScores); + +#[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() + } +} From 1b75b6891af29785319b8f3e95e4b981a47881f0 Mon Sep 17 00:00:00 2001 From: Carlos Zanella Date: Tue, 18 Aug 2026 15:22:21 +0100 Subject: [PATCH 2/5] fix: rank water-fill candidate paths net of gas MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The candidate ordering the split passes read was gross output, gas never subtracted, while the single-path baseline and the finished split were both judged net of gas. A three-hop route pays more gross than a one-hop route and costs three swaps; on net the short one wins, on gross it does not. Enough long routes above it and the short one falls out of the windows `select_disjoint` and `select_shared_candidates` slice off `by_output` — so the splits were built without the route that was about to beat them. `by_output` now holds `Option`: the net output of a path that filled, and nothing for a path that could not. Sorting descending keeps the filled paths in front, ordered by what they really pay, and leaves the unfilled ones last as before. A plain zero would not do that — net output goes negative once gas costs more than a path pays, and a path that filled for a loss still tells the split passes more than one that failed outright. What still separates the two orderings is membership: only paths that took the whole order can become the single-path baseline. Co-Authored-By: Claude Opus 5 (1M context) --- fynd-core/src/algorithm/water_fill/mod.rs | 53 ++++++++++++++++---- fynd-core/src/algorithm/water_fill/models.rs | 6 ++- 2 files changed, 47 insertions(+), 12 deletions(-) diff --git a/fynd-core/src/algorithm/water_fill/mod.rs b/fynd-core/src/algorithm/water_fill/mod.rs index e8602cbd..ce4c7d22 100644 --- a/fynd-core/src/algorithm/water_fill/mod.rs +++ b/fynd-core/src/algorithm/water_fill/mod.rs @@ -1277,14 +1277,16 @@ fn rank_outcomes( token_prices: Option<&TokenGasPrices>, token_out: &Address, ) -> FullAmountRanking { - let mut by_output: Vec<(usize, BigUint)> = Vec::with_capacity(outcomes_by_path.len()); + // 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 output = match outcome { + let net_output = match outcome { FullAmountOutcome::Filled(paid) => { let gas_cost = WaterFillAlgorithm::gas_cost_in_token( &paid.gas, @@ -1294,14 +1296,17 @@ fn rank_outcomes( ); 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)); - paid.amount_out + by_output_net_gas.push((path_ix, net_output.clone())); + Some(net_output) } - FullAmountOutcome::Unfilled => BigUint::zero(), + FullAmountOutcome::Unfilled => None, }; - by_output.push((path_ix, output)); + 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 { @@ -2045,8 +2050,9 @@ mod tests { assert_eq!(ranking.by_output_net_gas, vec![3, 0], "unfilled cannot be the baseline"); } - /// Gas is taken off before the net-of-gas ordering, so a path paying more gross can rank below - /// a cheaper one — which is the whole reason for keeping a second ordering. + /// 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); @@ -2066,8 +2072,35 @@ mod tests { let ranking = rank_outcomes(outcomes, &BigUint::from(1u64), Some(&token_prices), &token_out); - assert_eq!(ranking.by_output, vec![0, 1], "gross output still puts the larger first"); - assert_eq!(ranking.by_output_net_gas, vec![1, 0], "net of gas reverses them"); + 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 diff --git a/fynd-core/src/algorithm/water_fill/models.rs b/fynd-core/src/algorithm/water_fill/models.rs index d2267208..0fcd8016 100644 --- a/fynd-core/src/algorithm/water_fill/models.rs +++ b/fynd-core/src/algorithm/water_fill/models.rs @@ -74,8 +74,10 @@ pub enum FullAmountOutcome { /// 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 first. One that could not take the whole order ranks last - /// at zero: it may still be worth a fraction in a split. + /// 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. From d5e3a91c48b901c3b9d94a23bb5a1fa98fe61dd9 Mon Sep 17 00:00:00 2001 From: Carlos Zanella Date: Tue, 18 Aug 2026 13:53:56 +0100 Subject: [PATCH 3/5] feat: Reuse caching developed for WF in ML Interpolating and caching all swaps (not only the winners) will save us some pool simulations on ML as well. --- fynd-core/src/algorithm/most_liquid.rs | 279 +++++++++++-------------- 1 file changed, 126 insertions(+), 153 deletions(-) diff --git a/fynd-core/src/algorithm/most_liquid.rs b/fynd-core/src/algorithm/most_liquid.rs index 9258858c..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 @@ -601,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 { @@ -611,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"); @@ -694,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(); @@ -710,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 @@ -742,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] }) @@ -904,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) { @@ -927,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) } } @@ -2194,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> { @@ -2207,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()); } From 45588cb4fcae5e19be809c7450d79100d0605f1a Mon Sep 17 00:00:00 2001 From: Carlos Zanella Date: Thu, 20 Aug 2026 11:54:56 +0100 Subject: [PATCH 4/5] refactor: Simplify unused type --- fynd-core/src/algorithm/water_fill/mod.rs | 14 +++++++------- fynd-core/src/algorithm/water_fill/models.rs | 1 - 2 files changed, 7 insertions(+), 8 deletions(-) diff --git a/fynd-core/src/algorithm/water_fill/mod.rs b/fynd-core/src/algorithm/water_fill/mod.rs index ce4c7d22..e1fa9e6b 100644 --- a/fynd-core/src/algorithm/water_fill/mod.rs +++ b/fynd-core/src/algorithm/water_fill/mod.rs @@ -37,7 +37,7 @@ use std::{ }; use models::{ - CandidatePathSet, CandidatePathState, CandidateSearchConfig, Deadline, Discovery, ExchangeMove, + CandidatePathState, CandidateSearchConfig, Deadline, Discovery, ExchangeMove, FullAmountOutcome, FullAmountRanking, ScoredEdge, SetupResult, SolveInput, SolveStage, SplitCandidate, StepResult, }; @@ -249,7 +249,7 @@ impl WaterFillAlgorithm { let mut cache = SwapCache::new(); sim_meter::start_solve(); - let (discovered_paths, _) = discover_paths( + let discovered_paths = discover_paths( graph, &market_view, order, @@ -1359,7 +1359,7 @@ fn discover_paths<'a, W>( order: &Order, cache: &mut SwapCache<'a>, cfg: CandidateSearchConfig<'_>, -) -> Result, AlgorithmError> +) -> Result>, AlgorithmError> where W: Clone, { @@ -1669,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(); @@ -1694,7 +1694,7 @@ fn rank_found_candidate_paths<'a, W>( reason: NoPathReason::NoGraphPath, }); } - Ok((paths, scores)) + Ok(paths) } #[cfg(test)] @@ -2002,7 +2002,7 @@ mod tests { let start = Instant::now(); let view = market.read().await; - let (paths, scores) = discover_paths( + let paths = discover_paths( graph_manager.graph(), &view, &order, @@ -2019,7 +2019,7 @@ mod tests { 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", diff --git a/fynd-core/src/algorithm/water_fill/models.rs b/fynd-core/src/algorithm/water_fill/models.rs index 0fcd8016..98b7b9d7 100644 --- a/fynd-core/src/algorithm/water_fill/models.rs +++ b/fynd-core/src/algorithm/water_fill/models.rs @@ -111,7 +111,6 @@ pub struct SetupResult<'o, 'g> { } type RankedPathScores = Vec<(usize, BigInt)>; -pub type CandidatePathSet<'a, W = ()> = (Vec>, RankedPathScores); #[derive(Clone)] pub struct CandidatePathState<'a, W> { From c0e7b4736652c9a8d87bb479fb77711bcaece448 Mon Sep 17 00:00:00 2001 From: Carlos Zanella Date: Thu, 20 Aug 2026 11:57:11 +0100 Subject: [PATCH 5/5] chore: REmove dead code --- fynd-core/src/algorithm/water_fill/models.rs | 2 -- 1 file changed, 2 deletions(-) diff --git a/fynd-core/src/algorithm/water_fill/models.rs b/fynd-core/src/algorithm/water_fill/models.rs index 98b7b9d7..35e7e86b 100644 --- a/fynd-core/src/algorithm/water_fill/models.rs +++ b/fynd-core/src/algorithm/water_fill/models.rs @@ -110,8 +110,6 @@ pub struct SetupResult<'o, 'g> { pub cache: SwapCache<'g>, } -type RankedPathScores = Vec<(usize, BigInt)>; - #[derive(Clone)] pub struct CandidatePathState<'a, W> { pub node: NodeIndex,