From 7073a8835b28af112bb637567c2a6334580d1514 Mon Sep 17 00:00:00 2001 From: kayibal Date: Wed, 5 Aug 2026 15:31:36 +0100 Subject: [PATCH 1/3] fix: keep inventory-limited pools in water-fill splits via get_limits MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An inventory-limited pool — e.g. a propAMM whose output balance is below the order — lost its entire allocation to worse-priced pools the moment the order exceeded its inventory, through three compounding defects: 1. A path failing simulation at the full order amount was ranked with gross 0, sorting it below every split candidate window (top-4 disjoint, top-8 full-amount / top-32 marginal fill-and-spill probes). Rank such paths by their output at the get_limits-capped input instead: each hop's sell limit, back-converted to path-input units through the preceding edges' spot prices (pessimistic under AMM concavity, so the capped simulation succeeds). One retry at 99% covers marginally stale limits; a failure the limits don't explain keeps gross 0. 2. Even ranked honestly, a capped path's output sinks below every order-absorbing pool once the order is a small multiple of its inventory, dropping out of the probe window again. The fill-and- spill marginal probe now always includes the best limit-capped paths (CAPPED_PROBE_PATHS) wherever they rank; the first-chunk probe then judges them by price. 3. Chunked allocation over-filled such pools: propAMM balance overrides reject only a single swap larger than the balance, so per-chunk simulations all succeed past the real capacity, and the rebuilt single leg at the summed amount reverts — discarding the whole split candidate. Both chunk loops now stop allocating to a path at its get_limits cap. The extra get_limits/simulation calls run only for paths that fail at the full amount plus the small allocation subsets. get_limits was verified against live states: 0 errors across 1,868 pool directions on 15 protocols, and get_amount_out succeeds at exactly max_in on all but one broken pool. Live result (USDC->WETH, 1-hop water_fill, fermiswap inventory ~\$278k): the pool's absolute allocation now pins at its inventory from \$400k to \$2M order size instead of dropping to zero, with the remainder spilling to on-chain pools. Adds get_limits_guarded to GuardedProtocolSim for panic containment. Co-Authored-By: Claude Fable 5 --- fynd-core/src/algorithm/sim_guard.rs | 56 +++- fynd-core/src/algorithm/water_fill.rs | 420 ++++++++++++++++++++++++-- 2 files changed, 444 insertions(+), 32 deletions(-) diff --git a/fynd-core/src/algorithm/sim_guard.rs b/fynd-core/src/algorithm/sim_guard.rs index 08cb263e3..96290db95 100644 --- a/fynd-core/src/algorithm/sim_guard.rs +++ b/fynd-core/src/algorithm/sim_guard.rs @@ -16,6 +16,7 @@ use tycho_simulation::tycho_common::{ errors::SimulationError, protocol_sim::{GetAmountOutResult, ProtocolSim}, }, + Bytes, }; /// Extension trait adding panic-guarded simulation calls to every [`ProtocolSim`]. @@ -33,6 +34,29 @@ pub(crate) trait GuardedProtocolSim { token_in: &Token, token_out: &Token, ) -> Result; + + /// Calls `get_limits`, converting a panic into a `SimulationError::FatalError`. + /// + /// On a contained panic, logs the token pair so the offending component/quote can be + /// tracked down from the logs. + fn get_limits_guarded( + &self, + sell_token: Bytes, + buy_token: Bytes, + ) -> Result<(BigUint, BigUint), SimulationError>; +} + +/// Best-effort extraction of a human-readable message from a contained panic payload. +fn panic_message(panic_payload: &(dyn std::any::Any + Send)) -> &str { + panic_payload + .downcast_ref::<&str>() + .copied() + .or_else(|| { + panic_payload + .downcast_ref::() + .map(String::as_str) + }) + .unwrap_or("") } impl GuardedProtocolSim for T { @@ -48,15 +72,7 @@ impl GuardedProtocolSim for T { })); outcome.unwrap_or_else(|panic_payload| { - let message = panic_payload - .downcast_ref::<&str>() - .copied() - .or_else(|| { - panic_payload - .downcast_ref::() - .map(String::as_str) - }) - .unwrap_or(""); + let message = panic_message(panic_payload.as_ref()); warn!( %amount_in, token_in = %token_in.address, @@ -69,6 +85,28 @@ impl GuardedProtocolSim for T { Err(SimulationError::FatalError(format!("get_amount_out panicked: {message}"))) }) } + + fn get_limits_guarded( + &self, + sell_token: Bytes, + buy_token: Bytes, + ) -> Result<(BigUint, BigUint), SimulationError> { + // Tokens are cloned into the call so the originals stay available for the panic log. + let outcome = catch_unwind(AssertUnwindSafe(|| { + self.get_limits(sell_token.clone(), buy_token.clone()) + })); + + outcome.unwrap_or_else(|panic_payload| { + let message = panic_message(panic_payload.as_ref()); + warn!( + %sell_token, + %buy_token, + panic = message, + "component get_limits panicked; skipping component" + ); + Err(SimulationError::FatalError(format!("get_limits panicked: {message}"))) + }) + } } #[cfg(test)] diff --git a/fynd-core/src/algorithm/water_fill.rs b/fynd-core/src/algorithm/water_fill.rs index e01e383c0..14a0186a1 100644 --- a/fynd-core/src/algorithm/water_fill.rs +++ b/fynd-core/src/algorithm/water_fill.rs @@ -35,7 +35,7 @@ use std::{ }; use num_bigint::{BigInt, BigUint}; -use num_traits::Zero; +use num_traits::{FromPrimitive, ToPrimitive, Zero}; use petgraph::{graph::NodeIndex, prelude::EdgeRef}; use tracing::{debug, instrument}; use tycho_simulation::{ @@ -73,6 +73,11 @@ const SHARED_FULL_PATHS: usize = 8; 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; +/// Number of limit-capped paths (best capped output first) always added to the fill-and-spill +/// marginal probe, wherever they rank. An inventory-limited path's capped output sinks below +/// every pool deep enough to absorb the whole order once the order is a small multiple of its +/// inventory, but its price can still win the probe. +const CAPPED_PROBE_PATHS: usize = 16; /// Upper bound on fill-and-spill candidate paths. const SHARED_MAX_CANDIDATES: usize = 12; /// Candidate states retained per intermediate token during bounded discovery expansion. @@ -151,6 +156,9 @@ struct StepResult { /// allocation methods take one context instead of the same six references each. struct SplitContext<'a, 'g> { ordered: &'a [Path<'g, DepthAndPrice>], + /// Positions in `ordered` (best capped output first) of paths ranked by a limit-capped + /// simulation because they failed at the full order amount. + capped_positions: &'a [usize], market: &'a MarketState, gas_price: &'a BigUint, token_prices: Option<&'a TokenGasPrices>, @@ -163,6 +171,9 @@ struct SplitContext<'a, 'g> { /// token gas prices for gas-aware ranking. struct SetupResult<'a> { ordered: Vec>, + /// Positions in `ordered` of paths ranked by a limit-capped simulation, capped at + /// [`CAPPED_PROBE_PATHS`], best capped output first. + capped_positions: Vec, market: MarketState, gas_price: BigUint, best_single: Option, @@ -292,8 +303,87 @@ impl WaterFillAlgorithm { selected } + /// Largest input the path can accept according to each hop's `get_limits` sell limit, in + /// path-input token units. + /// + /// A downstream hop's limit is back-converted to path-input units through the product of the + /// preceding edges' spot prices. Output is concave in input for every pool type + /// (`out(x) <= spot * x`), so that conversion under-estimates the true bound and a simulation + /// at the returned amount stays within every hop's limit. The first hop's limit is already in + /// path-input units and needs no conversion. + /// + /// Returns `None` when a hop's simulation state, limit, or edge spot price is unavailable, + /// or when the resulting cap is zero. + fn max_fillable_input(path: &Path, market: &MarketState) -> Option { + let mut cap: Option = None; + // Path-input units per current hop-input unit: the product of preceding edges' spot + // prices (each token_out per token_in). + let mut path_input_price = 1.0_f64; + let hop_count = path.len(); + for (hop, (address_in, edge, address_out)) in path.iter().enumerate() { + let state = market.get_simulation_state(&edge.component_id)?; + let (max_in, _max_out) = state + .get_limits_guarded(address_in.clone(), address_out.clone()) + .ok()?; + let hop_cap = if hop == 0 { + max_in + } else { + BigUint::from_f64((max_in.to_f64()? / path_input_price).floor())? + }; + if cap + .as_ref() + .is_none_or(|current| hop_cap < *current) + { + cap = Some(hop_cap); + } + if hop + 1 < hop_count { + let spot = edge.data.as_ref()?.spot_price; + if spot <= 0.0 || !spot.is_finite() { + return None; + } + path_input_price *= spot; + } + } + cap.filter(|cap| !cap.is_zero()) + } + + /// Gross output used to rank a path that failed simulation at the full order amount: its + /// output at the limit-capped input. + /// + /// Returns `None` when the limits don't explain the failure (the cap covers the full order, + /// so the failure has another cause) or when the capped simulation fails as well. The limit + /// itself is a valid `get_amount_out` input per the `get_limits` contract; a single retry at + /// 99% covers a marginally stale limit. No further search is done. + fn capped_rank_output( + path: &Path, + market: &MarketState, + token_prices: Option<&TokenGasPrices>, + order_amount: &BigUint, + ) -> Option { + let cap = Self::max_fillable_input(path, market)?; + if cap >= *order_amount { + return None; + } + for amount in [cap.clone(), &cap * 99u32 / 100u32] { + if amount.is_zero() { + continue; + } + if let Ok(result) = + MostLiquidAlgorithm::simulate_path(path, market, token_prices, amount) + { + return result + .route() + .swaps() + .last() + .map(|swap| swap.amount_out().clone()); + } + } + None + } + /// 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). + /// path if any (a path that fails at the full amount is ranked by its limit-capped output + /// and kept as a split-only candidate). #[instrument(level = "debug", skip_all)] async fn setup<'a>( &self, @@ -419,6 +509,7 @@ impl WaterFillAlgorithm { let amount_in = order.amount().clone(); let mut best_single: Option = None; let mut full_outputs: Vec<(usize, BigUint)> = Vec::new(); + let mut capped_paths: HashSet = HashSet::new(); for (idx, path) in paths.iter().enumerate() { if start.elapsed().as_millis() as u64 > timeout_ms { @@ -446,11 +537,23 @@ impl WaterFillAlgorithm { 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())), + // A path that can't take the whole order (e.g. a propAMM whose inventory is + // below the order, or concentrated liquidity exhausted at the full amount) may + // still be worth a fraction in a split. Rank it by its output at the + // limit-capped amount so it stays inside the split candidate windows + // (`select_disjoint`, `select_shared_candidates`); the allocation passes cap its + // share chunk by chunk. It never becomes the single-path baseline, and a path + // whose failure the limits don't explain keeps gross 0 (ranked last). + Err(_) => { + match Self::capped_rank_output(path, &market, token_prices.as_ref(), &amount_in) + { + Some(capped_gross) => { + capped_paths.insert(idx); + full_outputs.push((idx, capped_gross)); + } + None => full_outputs.push((idx, BigUint::zero())), + } + } } } @@ -458,17 +561,22 @@ impl WaterFillAlgorithm { // 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 - .into_iter() - .map(|(idx, _)| paths[idx].clone()) - .collect(); + let mut capped_positions = Vec::new(); + let mut ordered: Vec> = Vec::with_capacity(full_outputs.len()); + for (position, (idx, _)) in full_outputs.into_iter().enumerate() { + if capped_paths.contains(&idx) && capped_positions.len() < CAPPED_PROBE_PATHS { + capped_positions.push(position); + } + ordered.push(paths[idx].clone()); + } debug!( candidate_paths = ordered.len(), + capped_paths = capped_positions.len(), elapsed_ms = start.elapsed().as_millis(), "water-fill discovery + full-amount ranking" ); - Ok(SetupResult { ordered, market, gas_price, best_single, token_prices }) + Ok(SetupResult { ordered, capped_positions, market, gas_price, best_single, token_prices }) } } @@ -494,12 +602,13 @@ impl Algorithm for WaterFillAlgorithm { return Err(AlgorithmError::ExactOutNotSupported); } - let SetupResult { ordered, market, gas_price, best_single, token_prices } = self - .setup(graph, market, label, derived, order, start) - .await?; + let SetupResult { ordered, capped_positions, market, gas_price, best_single, token_prices } = + self.setup(graph, market, label, derived, order, start) + .await?; let token_out = order.token_out(); let ctx = SplitContext { ordered: &ordered, + capped_positions: &capped_positions, market: &market, gas_price: &gas_price, token_prices: token_prices.as_ref(), @@ -837,6 +946,21 @@ impl WaterFillAlgorithm { Self::candidate_from_allocations(ctx, &allocations) } + /// Per-path allocation caps from `get_limits`, aligned with `subset`. + /// + /// Chunked simulations can overshoot a component's real capacity: some states (propAMM + /// balance overrides in particular) only reject a single swap larger than the balance, so + /// per-chunk swaps all succeed while the rebuilt single leg at the summed amount reverts and + /// the whole candidate is lost. The chunk loops therefore stop allocating to a path at its + /// limit cap. `None` means no cap could be computed (missing state or limits) — the path + /// stays bounded by chunk simulation failures alone, as before. + fn allocation_caps(ctx: &SplitContext, subset: &[usize]) -> Vec> { + subset + .iter() + .map(|&path_idx| Self::max_fillable_input(&ctx.ordered[path_idx], ctx.market)) + .collect() + } + /// 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). @@ -857,6 +981,7 @@ impl WaterFillAlgorithm { let timeout_ms = self.timeout.as_millis() as u64; let path_count = subset.len(); + let caps = Self::allocation_caps(ctx, subset); let mut committed: Vec>> = (0..path_count) .map(|_| HashMap::new()) .collect(); @@ -871,6 +996,12 @@ impl WaterFillAlgorithm { let mut best: Option<(usize, BigInt, StepResult)> = None; for (i, &path_idx) in subset.iter().enumerate() { + if caps[i] + .as_ref() + .is_some_and(|cap| &cum_in[i] + &chunk > *cap) + { + continue; + } let Some(step) = Self::simulate_step( &ctx.ordered[path_idx], ctx.market, @@ -909,7 +1040,10 @@ impl WaterFillAlgorithm { /// 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. + /// the extra path often ranks poorly at full size but wins on the margin. Limit-capped paths + /// are probed wherever they rank, for the same reason: an inventory-limited path's capped + /// output sinks below every order-absorbing pool as the order grows, but its price can still + /// win the probe. 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; @@ -918,16 +1052,23 @@ impl WaterFillAlgorithm { } let timeout_ms = self.timeout.as_millis() as u64; let empty: HashMap> = HashMap::new(); - let mut marginal: Vec<(usize, BigInt)> = Vec::new(); - for (idx, path) in ctx + let mut probe_indices: Vec = (0..ctx .ordered - .iter() - .enumerate() - .take(SHARED_MARGIN_PROBE_PATHS) - { + .len() + .min(SHARED_MARGIN_PROBE_PATHS)) + .collect(); + probe_indices.extend( + ctx.capped_positions + .iter() + .copied() + .filter(|position| *position >= SHARED_MARGIN_PROBE_PATHS), + ); + let mut marginal: Vec<(usize, BigInt)> = Vec::new(); + for idx in probe_indices { if ctx.start.elapsed().as_millis() as u64 > timeout_ms { break; } + let path = &ctx.ordered[idx]; let Some(step) = Self::simulate_step(path, ctx.market, &empty, first_chunk.clone()) else { continue; @@ -1001,10 +1142,12 @@ impl WaterFillAlgorithm { let remainder = &amount_in - &base_chunk * num_chunks; let timeout_ms = self.timeout.as_millis() as u64; + let caps = Self::allocation_caps(ctx, subset); let mut overlay: HashMap> = HashMap::new(); 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 cum_in: Vec = vec![BigUint::zero(); subset.len()]; let mut schedule: Vec<(usize, BigUint)> = Vec::with_capacity(num_chunks); for chunk_idx in 0..num_chunks { @@ -1018,6 +1161,12 @@ impl WaterFillAlgorithm { if !activated[i] && active_count >= self.max_paths { continue; } + if caps[i] + .as_ref() + .is_some_and(|cap| &cum_in[i] + &chunk > *cap) + { + continue; + } let Some(step) = Self::simulate_step( &ctx.ordered[path_idx], ctx.market, @@ -1053,6 +1202,7 @@ impl WaterFillAlgorithm { active_count += 1; } counts[best_i] += 1; + cum_in[best_i] += &chunk; schedule.push((best_i, chunk)); } Some((counts, schedule)) @@ -1498,7 +1648,7 @@ mod tests { }, test_utils::{ addr, setup_market_unweighted, setup_market_weighted_boxed, token_with_decimals, - ConstantProductSim, DivByZeroSim, + ConstantProductSim, DivByZeroSim, MockProtocolSim, }, }, graph::GraphManager, @@ -1625,6 +1775,230 @@ mod tests { assert!(split.route().swaps().len() >= 2, "expected a split across both components"); } + /// An inventory-limited component with the best price must stay in the split when the order + /// exceeds its inventory. Reproduces the fermiswap dropout: the limited component fails + /// simulation at the full order amount, and without limit-capped ranking it sorts to the + /// bottom of the candidate order — below the split candidate windows (top-4 disjoint, + /// top-8 full-amount / top-32 marginal fill-and-spill probes) once enough healthy + /// competitors rank above it — so its entire allocation was lost to worse-priced pools. + #[tokio::test] + async fn test_inventory_limited_component_kept_in_split() { + let a = token_with_decimals(0x01, "A", 18); + let b = token_with_decimals(0x02, "B", 18); + let unit = BigUint::from(10u64).pow(18); + + // Best rate (1000 B/A) but only 800 A of input-side reserve: the 1000 A order fails at + // the full amount; get_limits caps it at 400 A. + let limited = Box::new(ConstantProductSim { + reserve_0: BigUint::from(800u64) * &unit, + reserve_1: BigUint::from(800_000u64) * &unit, + gas: 50_000, + }) as Box; + + // 38 components absorb the full order (3 deep at rate 500, 35 at rate 400), so every one + // of their full-amount outputs (455k / 333k B) beats the limited component's capped + // output (267k B at 400 A). The limited component therefore ranks below the top-4 + // disjoint, top-8 full-amount, and top-32 marginal probe windows — only the explicit + // capped-path probe reaches it, and its 1000 B/A price then wins the probe. + let deep = || { + Box::new(ConstantProductSim { + reserve_0: BigUint::from(10_000u64) * &unit, + reserve_1: BigUint::from(5_000_000u64) * &unit, + gas: 50_000, + }) as Box + }; + let shallow = || { + Box::new(ConstantProductSim { + reserve_0: BigUint::from(5_000u64) * &unit, + reserve_1: BigUint::from(2_000_000u64) * &unit, + gas: 50_000, + }) as Box + }; + + let deep_names: Vec = (0..3) + .map(|i| format!("deep_{i}")) + .collect(); + let shallow_names: Vec = (0..35) + .map(|i| format!("shallow_{i}")) + .collect(); + let mut components = vec![("limited", &a, &b, limited)]; + for name in &deep_names { + components.push((name.as_str(), &a, &b, deep())); + } + for name in &shallow_names { + components.push((name.as_str(), &a, &b, shallow())); + } + let (market, gm) = setup_market_weighted_boxed(components); + + let order = Order::new( + a.address.clone(), + b.address.clone(), + BigUint::from(1_000u64) * &unit, + OrderSide::Sell, + addr(0xFF), + ); + let result = WaterFillAlgorithm::with_config(config()) + .unwrap() + .find_best_route(gm.graph(), market.clone(), None, None, &order) + .await + .expect("order fills across the deep components"); + + let limited_amount_in: BigUint = result + .route() + .swaps() + .iter() + .filter(|swap| swap.component_id() == "limited") + .map(|swap| swap.amount_in().clone()) + .sum(); + assert!( + !limited_amount_in.is_zero(), + "the limited component must be part of the split, not dropped", + ); + assert!( + limited_amount_in >= BigUint::from(100u64) * &unit, + "the limited component's share must be substantial, got {limited_amount_in}", + ); + assert!( + limited_amount_in < BigUint::from(800u64) * &unit, + "the limited component cannot take more than its input reserve", + ); + assert!(result.route().validate().is_ok(), "route validation failed"); + } + + /// A component that accepts every chunk-sized swap but rejects the rebuilt single leg must be + /// capped at its `get_limits` amount during allocation. `MockProtocolSim` reproduces the + /// propAMM balance-override semantics: the liquidity check is per swap and `new_state` never + /// depletes, so chunked simulation happily over-allocates; without the allocation cap the + /// rebuilt over-limit leg fails and the whole split candidate is discarded, dropping the + /// component from the route entirely. + #[tokio::test] + async fn test_per_swap_limited_component_capped_at_its_limit() { + let a = token_with_decimals(0x01, "A", 18); + let b = token_with_decimals(0x02, "B", 18); + let unit = BigUint::from(10u64).pow(18); + // Best rate (1000 B/A) but only 55,000 B of per-swap liquidity: the sell limit is 55 A, + // while every chunk-sized swap (50 A coarse and below) stays under the per-swap check. + let limited = Box::new( + MockProtocolSim::new(1000.0) + .with_liquidity( + (BigUint::from(55_000u64) * &unit) + .to_u128() + .unwrap(), + ) + .with_gas(50_000) + .with_tokens(&[a.clone(), b.clone()]), + ) as Box; + let deep = Box::new(ConstantProductSim { + reserve_0: BigUint::from(10_000u64) * &unit, + reserve_1: BigUint::from(5_000_000u64) * &unit, + gas: 50_000, + }) as Box; + let (market, gm) = + setup_market_weighted_boxed(vec![("limited", &a, &b, limited), ("deep", &a, &b, deep)]); + + let order = Order::new( + a.address.clone(), + b.address.clone(), + BigUint::from(1_000u64) * &unit, + OrderSide::Sell, + addr(0xFF), + ); + let result = WaterFillAlgorithm::with_config(config()) + .unwrap() + .find_best_route(gm.graph(), market.clone(), None, None, &order) + .await + .expect("order fills across both components"); + + let limited_amount_in: BigUint = result + .route() + .swaps() + .iter() + .filter(|swap| swap.component_id() == "limited") + .map(|swap| swap.amount_in().clone()) + .sum(); + assert!( + !limited_amount_in.is_zero(), + "the limited component must contribute up to its limit, not be dropped", + ); + assert!( + limited_amount_in <= BigUint::from(55u64) * &unit, + "allocation must respect the 55 A sell limit, got {limited_amount_in}", + ); + assert!(result.route().validate().is_ok(), "route validation failed"); + } + + /// The single-hop cap is the first hop's `get_limits` sell limit, exact and unconverted. + #[tokio::test] + async fn test_max_fillable_input_single_hop_uses_exact_limit() { + let a = token_with_decimals(0x01, "A", 18); + let b = token_with_decimals(0x02, "B", 18); + let unit = BigUint::from(10u64).pow(18); + let limited = Box::new(ConstantProductSim { + reserve_0: BigUint::from(800u64) * &unit, + reserve_1: BigUint::from(800_000u64) * &unit, + gas: 50_000, + }) as Box; + let (market, gm) = setup_market_weighted_boxed(vec![("limited", &a, &b, limited)]); + + let paths = MostLiquidAlgorithm::find_paths(gm.graph(), &a.address, &b.address, 1, 1, None) + .unwrap(); + assert_eq!(paths.len(), 1); + let subset = market + .read() + .await + .extract_subset_with_overlay(&HashSet::from(["limited".to_string()])); + + // ConstantProductSim's sell limit is half the input reserve: 400 A. + assert_eq!( + WaterFillAlgorithm::max_fillable_input(&paths[0], &subset), + Some(BigUint::from(400u64) * &unit), + ); + } + + /// A downstream hop's limit is back-converted to path-input units through the preceding + /// edge's spot price: a 400 B limit behind a 2.0 B/A hop caps the path at ~200 A. + #[tokio::test] + async fn test_max_fillable_input_two_hop_converts_via_spot() { + let a = token_with_decimals(0x01, "A", 18); + let b = token_with_decimals(0x02, "B", 18); + let c = token_with_decimals(0x03, "C", 18); + let unit = BigUint::from(10u64).pow(18); + let first = Box::new(ConstantProductSim { + reserve_0: BigUint::from(10_000u64) * &unit, + reserve_1: BigUint::from(20_000u64) * &unit, + gas: 50_000, + }) as Box; + let second = Box::new(ConstantProductSim { + reserve_0: BigUint::from(800u64) * &unit, + reserve_1: BigUint::from(800_000u64) * &unit, + gas: 50_000, + }) as Box; + let (market, gm) = + setup_market_weighted_boxed(vec![("first", &a, &b, first), ("second", &b, &c, second)]); + + let paths = MostLiquidAlgorithm::find_paths(gm.graph(), &a.address, &c.address, 2, 2, None) + .unwrap(); + assert_eq!(paths.len(), 1); + let subset = market + .read() + .await + .extract_subset_with_overlay(&HashSet::from([ + "first".to_string(), + "second".to_string(), + ])); + + // Hop 2's sell limit is 400 B; through hop 1's 2.0 spot that is 200 A of path input, + // tighter than hop 1's own 5,000 A limit. f64 conversion may be off by a rounding ulp. + let cap = WaterFillAlgorithm::max_fillable_input(&paths[0], &subset) + .expect("cap must be computable"); + let expected = BigUint::from(200u64) * &unit; + let diff = if cap > expected { &cap - &expected } else { &expected - &cap }; + assert!( + diff <= BigUint::from(1_000_000u64), + "cap {cap} should be within rounding error of {expected}", + ); + } + /// A component whose math panics mid-simulation must be skipped like any failing component, /// not unwind through the solver worker thread. The panicking component advertises huge depth /// so discovery ranks it first and the bulk fill loops actually simulate it. From 658bc43ad32239ec2e0204f42f0fc1dad4b1ee17 Mon Sep 17 00:00:00 2001 From: kayibal Date: Wed, 5 Aug 2026 19:34:02 +0100 Subject: [PATCH 2/3] perf(water-fill): bound capped ranking work and budget shared components The capped ranking pass simulated every full-amount-failing path at its limit cap with uncached per-hop get_limits calls. At three hops most of the ~4.5k candidate paths fail once the order is large, which added 1-2s to setup on large orders. Now sell limits are memoized per solve (LimitsCache), failing paths are deduplicated by bottleneck component and ranked by spot price, and only the top CAPPED_PROBE_PATHS get the honest capped simulation; the rest keep gross 0 as before the capped ranking existed. The allocation passes replace per-path caps with a shared per-component budget (ComponentBudget) fed by exact hop inputs from simulate_step: fill-and-spill paths share components, and their combined allocation must stay a valid single swap or the rebuilt merged leg reverts and the whole candidate is discarded. The gated coarse pass also anchors the top-ranked candidate in the active set, so inventory-limited probe winners cannot claim every max_paths slot, saturate together, and leave the order unfillable. Regression test: 20 two-hop paths bottlenecked by one limited pool must not crowd a distinct best-price limited pool out of the split (test_capped_probe_slots_deduped_by_bottleneck, fails on the previous commit). Co-Authored-By: Claude Fable 5 --- fynd-core/src/algorithm/water_fill.rs | 534 +++++++++++++++++++++----- 1 file changed, 437 insertions(+), 97 deletions(-) diff --git a/fynd-core/src/algorithm/water_fill.rs b/fynd-core/src/algorithm/water_fill.rs index 14a0186a1..ba9fd8043 100644 --- a/fynd-core/src/algorithm/water_fill.rs +++ b/fynd-core/src/algorithm/water_fill.rs @@ -73,10 +73,11 @@ const SHARED_FULL_PATHS: usize = 8; 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; -/// Number of limit-capped paths (best capped output first) always added to the fill-and-spill -/// marginal probe, wherever they rank. An inventory-limited path's capped output sinks below -/// every pool deep enough to absorb the whole order once the order is a small multiple of its -/// inventory, but its price can still win the probe. +/// Number of full-amount-failing paths given an honest limit-capped ranking simulation and +/// always added to the fill-and-spill marginal probe, wherever they rank. An inventory-limited +/// path's capped output sinks below every pool deep enough to absorb the whole order once the +/// order is a small multiple of its inventory, but its price can still win the probe. Selection +/// deduplicates by bottleneck component and ranks by spot price (`select_capped_probes`). const CAPPED_PROBE_PATHS: usize = 16; /// Upper bound on fill-and-spill candidate paths. const SHARED_MAX_CANDIDATES: usize = 12; @@ -149,6 +150,112 @@ struct StepResult { amount_out: BigUint, gas: BigUint, new_states: Vec<(ComponentId, Box)>, + /// Exact input each hop swapped, keyed like [`LimitsCache`], for component-budget tracking. + hop_inputs: Vec<((ComponentId, Address, Address), BigUint)>, +} + +/// Sell-limit lookups memoized for one solve, keyed by component and swap direction. `get_limits` +/// on VM-backed pools executes EVM code, and at three hops the candidate paths cross the same +/// component pair thousands of times — the cache collapses that to one call per direction. +/// A `None` value records that the state or its limits are unavailable, so failures are not +/// retried either. +type LimitsCache = HashMap<(ComponentId, Address, Address), Option>; + +/// Cumulative component usage across every path of one allocation pass, checked against each +/// component's `get_limits` sell limit. +/// +/// Chunked simulations can overshoot a component's real capacity: some states (propAMM balance +/// overrides in particular) only reject a single swap larger than the balance, so per-chunk swaps +/// all succeed while the rebuilt single leg at the summed amount reverts and the whole candidate +/// is lost. Per-path caps are not enough — fill-and-spill paths share components, and their +/// combined allocation must also stay a valid single swap. The chunk loops therefore refuse any +/// chunk that would push a component's summed input past its sell limit. A component without an +/// available limit stays bounded by chunk simulation failures alone. +struct ComponentBudget { + limits: LimitsCache, + used: HashMap<(ComponentId, Address, Address), BigUint>, +} + +impl ComponentBudget { + fn new() -> Self { + Self { limits: LimitsCache::new(), used: HashMap::new() } + } + + /// Whether every component the step swapped through stays within its sell limit after + /// adding the step's hop inputs. + fn allows(&mut self, market: &MarketState, step: &StepResult) -> bool { + // A path may traverse a component twice, so tentative additions accumulate per key + // before comparing against the limit. + let mut additions: HashMap<&(ComponentId, Address, Address), BigUint> = HashMap::new(); + for (key, amount_in) in &step.hop_inputs { + let Some(limit) = WaterFillAlgorithm::cached_sell_limit( + &mut self.limits, + market, + &key.0, + &key.1, + &key.2, + ) else { + continue; + }; + let added = additions.entry(key).or_default(); + *added += amount_in; + let used = self + .used + .get(key) + .cloned() + .unwrap_or_default(); + if used + &*added > limit { + return false; + } + } + true + } + + /// Whether any component of `path` has already consumed its whole sell limit — a cheap + /// pre-simulation skip for saturated paths; `allows` stays the exact check. + fn path_saturated(&mut self, market: &MarketState, path: &Path) -> bool { + for (address_in, edge, address_out) in path.iter() { + let Some(limit) = WaterFillAlgorithm::cached_sell_limit( + &mut self.limits, + market, + &edge.component_id, + address_in, + address_out, + ) else { + continue; + }; + let key = (edge.component_id.clone(), address_in.clone(), address_out.clone()); + if self + .used + .get(&key) + .is_some_and(|used| *used >= limit) + { + return true; + } + } + false + } + + /// Records the step's hop inputs as consumed. + fn commit(&mut self, step: &StepResult) { + for (key, amount_in) in &step.hop_inputs { + *self + .used + .entry(key.clone()) + .or_default() += amount_in; + } + } +} + +/// A path's input cap from `max_fillable_input`, with the data capped-probe selection needs. +struct PathCap { + /// Largest path input within every hop's sell limit, in path-input token units. + input_cap: BigUint, + /// Component whose sell limit set the cap. + bottleneck: ComponentId, + /// Product of every edge's spot price (path output per input at zero size); `None` when any + /// edge's derived data is missing. + spot_product: Option, } /// Shared inputs threaded through every split-allocation pass: the ranked candidate paths, the @@ -221,6 +328,8 @@ impl WaterFillAlgorithm { let mut intra_path_states: HashMap> = HashMap::new(); let mut new_states: Vec<(ComponentId, Box)> = Vec::with_capacity(path.len()); + let mut hop_inputs: Vec<((ComponentId, Address, Address), BigUint)> = + Vec::with_capacity(path.len()); for (address_in, edge, address_out) in path.iter() { let token_in = market.get_token(address_in)?; @@ -243,9 +352,13 @@ impl WaterFillAlgorithm { intra_path_states.insert(component_id.clone(), result.new_state.clone_box()); } new_states.push((component_id.clone(), result.new_state)); + hop_inputs.push(( + (component_id.clone(), address_in.clone(), address_out.clone()), + current.clone(), + )); current = result.amount; } - Some(StepResult { amount_out: current, gas: total_gas, new_states }) + Some(StepResult { amount_out: current, gas: total_gas, new_states, hop_inputs }) } /// Converts a gas amount to output-token terms. Returns `None` if no price is available. @@ -303,8 +416,32 @@ impl WaterFillAlgorithm { selected } + /// One hop's `get_limits` sell limit through the solve-scoped cache. + fn cached_sell_limit( + limits: &mut LimitsCache, + market: &MarketState, + component_id: &ComponentId, + address_in: &Address, + address_out: &Address, + ) -> Option { + let key = (component_id.clone(), address_in.clone(), address_out.clone()); + if let Some(cached) = limits.get(&key) { + return cached.clone(); + } + let limit = market + .get_simulation_state(component_id) + .and_then(|state| { + state + .get_limits_guarded(address_in.clone(), address_out.clone()) + .ok() + }) + .map(|(max_in, _max_out)| max_in); + limits.insert(key, limit.clone()); + limit + } + /// Largest input the path can accept according to each hop's `get_limits` sell limit, in - /// path-input token units. + /// path-input token units, with the component that set it and the path's spot price. /// /// A downstream hop's limit is back-converted to path-input units through the product of the /// preceding edges' spot prices. Output is concave in input for every pool type @@ -312,19 +449,27 @@ impl WaterFillAlgorithm { /// at the returned amount stays within every hop's limit. The first hop's limit is already in /// path-input units and needs no conversion. /// - /// Returns `None` when a hop's simulation state, limit, or edge spot price is unavailable, - /// or when the resulting cap is zero. - fn max_fillable_input(path: &Path, market: &MarketState) -> Option { - let mut cap: Option = None; + /// Returns `None` when a hop's simulation state, limit, or a non-final edge spot price is + /// unavailable, or when the resulting cap is zero. + fn max_fillable_input( + path: &Path, + market: &MarketState, + limits: &mut LimitsCache, + ) -> Option { + let mut cap: Option<(BigUint, ComponentId)> = None; // Path-input units per current hop-input unit: the product of preceding edges' spot // prices (each token_out per token_in). let mut path_input_price = 1.0_f64; + let mut spot_product = Some(1.0_f64); let hop_count = path.len(); for (hop, (address_in, edge, address_out)) in path.iter().enumerate() { - let state = market.get_simulation_state(&edge.component_id)?; - let (max_in, _max_out) = state - .get_limits_guarded(address_in.clone(), address_out.clone()) - .ok()?; + let max_in = Self::cached_sell_limit( + limits, + market, + &edge.component_id, + address_in, + address_out, + )?; let hop_cap = if hop == 0 { max_in } else { @@ -332,39 +477,95 @@ impl WaterFillAlgorithm { }; if cap .as_ref() - .is_none_or(|current| hop_cap < *current) + .is_none_or(|(current, _)| hop_cap < *current) { - cap = Some(hop_cap); + cap = Some((hop_cap, edge.component_id.clone())); } - if hop + 1 < hop_count { - let spot = edge.data.as_ref()?.spot_price; - if spot <= 0.0 || !spot.is_finite() { - return None; + match edge + .data + .as_ref() + .map(|data| data.spot_price) + { + Some(spot) if spot > 0.0 && spot.is_finite() => { + if hop + 1 < hop_count { + path_input_price *= spot; + } + spot_product = spot_product.map(|product| product * spot); } - path_input_price *= spot; + // A final hop's missing spot price does not affect the cap, only the proxy. + _ if hop + 1 == hop_count => spot_product = None, + _ => return None, } } - cap.filter(|cap| !cap.is_zero()) + let (input_cap, bottleneck) = cap?; + if input_cap.is_zero() { + return None; + } + Some(PathCap { input_cap, bottleneck, spot_product }) } - /// Gross output used to rank a path that failed simulation at the full order amount: its - /// output at the limit-capped input. + /// Chooses which full-amount-failing paths get an honest limit-capped ranking simulation, + /// returning `(path index, cap)` pairs. /// - /// Returns `None` when the limits don't explain the failure (the cap covers the full order, - /// so the failure has another cause) or when the capped simulation fails as well. The limit - /// itself is a valid `get_amount_out` input per the `get_limits` contract; a single retry at - /// 99% covers a marginally stale limit. No further search is done. - fn capped_rank_output( + /// Computing a cap is cheap (cached `get_limits` lookups), but the capped simulation is not, + /// and at three hops most candidate paths fail once the order is large — simulating every + /// one dominated solve time. All failing paths keep gross 0 except the strongest probes: + /// paths are deduplicated by bottleneck component (paths sharing a bottleneck cannot + /// contribute its inventory twice), then the best [`CAPPED_PROBE_PATHS`] by spot price are + /// kept — the fill-and-spill marginal probe these paths are retained for is itself + /// price-driven. Paths whose cap covers the full order (the failure has another cause) or + /// cannot absorb one fine chunk are dropped. + fn select_capped_probes( + paths: &[Path], + failing: &[usize], + market: &MarketState, + order_amount: &BigUint, + limits: &mut LimitsCache, + ) -> Vec<(usize, BigUint)> { + let chunk_floor = order_amount / FINE_CHUNKS; + let mut best_per_bottleneck: HashMap = HashMap::new(); + for &idx in failing { + let Some(path_cap) = Self::max_fillable_input(&paths[idx], market, limits) else { + continue; + }; + if path_cap.input_cap >= *order_amount || path_cap.input_cap < chunk_floor { + continue; + } + let Some(spot) = path_cap.spot_product else { + continue; + }; + let keep = best_per_bottleneck + .get(&path_cap.bottleneck) + .is_none_or(|(_, _, existing)| spot > *existing); + if keep { + best_per_bottleneck.insert(path_cap.bottleneck, (idx, path_cap.input_cap, spot)); + } + } + let mut probes: Vec<(usize, BigUint, f64)> = best_per_bottleneck + .into_values() + .collect(); + probes.sort_by(|(_, _, a), (_, _, b)| { + b.partial_cmp(a) + .unwrap_or(Ordering::Equal) + }); + probes.truncate(CAPPED_PROBE_PATHS); + probes + .into_iter() + .map(|(idx, cap, _)| (idx, cap)) + .collect() + } + + /// Gross output of a path simulated at its limit cap, used to rank it honestly. + /// + /// The cap itself is a valid `get_amount_out` input per the `get_limits` contract; a single + /// retry at 99% covers a marginally stale limit. No further search is done. + fn simulate_at_cap( path: &Path, market: &MarketState, token_prices: Option<&TokenGasPrices>, - order_amount: &BigUint, + cap: &BigUint, ) -> Option { - let cap = Self::max_fillable_input(path, market)?; - if cap >= *order_amount { - return None; - } - for amount in [cap.clone(), &cap * 99u32 / 100u32] { + for amount in [cap.clone(), cap * 99u32 / 100u32] { if amount.is_zero() { continue; } @@ -509,7 +710,7 @@ impl WaterFillAlgorithm { let amount_in = order.amount().clone(); let mut best_single: Option = None; let mut full_outputs: Vec<(usize, BigUint)> = Vec::new(); - let mut capped_paths: HashSet = HashSet::new(); + let mut failing: Vec = Vec::new(); for (idx, path) in paths.iter().enumerate() { if start.elapsed().as_millis() as u64 > timeout_ms { @@ -537,23 +738,39 @@ impl WaterFillAlgorithm { best_single = Some(result); } } - // A path that can't take the whole order (e.g. a propAMM whose inventory is - // below the order, or concentrated liquidity exhausted at the full amount) may - // still be worth a fraction in a split. Rank it by its output at the - // limit-capped amount so it stays inside the split candidate windows - // (`select_disjoint`, `select_shared_candidates`); the allocation passes cap its - // share chunk by chunk. It never becomes the single-path baseline, and a path - // whose failure the limits don't explain keeps gross 0 (ranked last). - Err(_) => { - match Self::capped_rank_output(path, &market, token_prices.as_ref(), &amount_in) - { - Some(capped_gross) => { - capped_paths.insert(idx); - full_outputs.push((idx, capped_gross)); - } - None => full_outputs.push((idx, BigUint::zero())), - } + Err(_) => failing.push(idx), + } + } + + // A path that can't take the whole order (e.g. a propAMM whose inventory is below the + // order, or concentrated liquidity exhausted at the full amount) may still be worth a + // fraction in a split. The selected probes are ranked by their output at the + // limit-capped amount so they stay inside the split candidate windows + // (`select_disjoint`, `select_shared_candidates`); the allocation passes cap their share + // chunk by chunk. They never become the single-path baseline. Every other failing path + // keeps gross 0 (ranked last). + let mut limits = LimitsCache::new(); + let mut capped_paths: HashSet = HashSet::new(); + let mut capped_grosses: HashMap = HashMap::new(); + for (idx, cap) in + Self::select_capped_probes(&paths, &failing, &market, &amount_in, &mut limits) + { + if start.elapsed().as_millis() as u64 > timeout_ms { + break; + } + if let Some(gross) = + Self::simulate_at_cap(&paths[idx], &market, token_prices.as_ref(), &cap) + { + capped_grosses.insert(idx, gross); + } + } + for idx in failing { + match capped_grosses.remove(&idx) { + Some(gross) => { + capped_paths.insert(idx); + full_outputs.push((idx, gross)); } + None => full_outputs.push((idx, BigUint::zero())), } } @@ -946,21 +1163,6 @@ impl WaterFillAlgorithm { Self::candidate_from_allocations(ctx, &allocations) } - /// Per-path allocation caps from `get_limits`, aligned with `subset`. - /// - /// Chunked simulations can overshoot a component's real capacity: some states (propAMM - /// balance overrides in particular) only reject a single swap larger than the balance, so - /// per-chunk swaps all succeed while the rebuilt single leg at the summed amount reverts and - /// the whole candidate is lost. The chunk loops therefore stop allocating to a path at its - /// limit cap. `None` means no cap could be computed (missing state or limits) — the path - /// stays bounded by chunk simulation failures alone, as before. - fn allocation_caps(ctx: &SplitContext, subset: &[usize]) -> Vec> { - subset - .iter() - .map(|&path_idx| Self::max_fillable_input(&ctx.ordered[path_idx], ctx.market)) - .collect() - } - /// 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). @@ -981,7 +1183,7 @@ impl WaterFillAlgorithm { let timeout_ms = self.timeout.as_millis() as u64; let path_count = subset.len(); - let caps = Self::allocation_caps(ctx, subset); + let mut budget = ComponentBudget::new(); let mut committed: Vec>> = (0..path_count) .map(|_| HashMap::new()) .collect(); @@ -996,10 +1198,7 @@ impl WaterFillAlgorithm { let mut best: Option<(usize, BigInt, StepResult)> = None; for (i, &path_idx) in subset.iter().enumerate() { - if caps[i] - .as_ref() - .is_some_and(|cap| &cum_in[i] + &chunk > *cap) - { + if budget.path_saturated(ctx.market, &ctx.ordered[path_idx]) { continue; } let Some(step) = Self::simulate_step( @@ -1010,6 +1209,9 @@ impl WaterFillAlgorithm { ) else { continue; }; + if !budget.allows(ctx.market, &step) { + continue; + } let gross_marginal = BigInt::from(step.amount_out.clone()); let net_marginal = if activated[i] { gross_marginal @@ -1029,6 +1231,7 @@ impl WaterFillAlgorithm { let Some((best_i, _, step)) = best else { break; }; + budget.commit(&step); for (id, state) in step.new_states { committed[best_i].insert(id, state); } @@ -1044,12 +1247,30 @@ impl WaterFillAlgorithm { /// are probed wherever they rank, for the same reason: an inventory-limited path's capped /// output sinks below every order-absorbing pool as the order grows, but its price can still /// win the probe. + /// + /// Probe winners that cannot take the whole order are deduplicated by bottleneck component: + /// near-identical variants of one limited pool would win several probe slots on the same + /// price, burn the gated activation slots (`max_paths`), and all saturate together — leaving + /// the order under-filled. Paths that can absorb the full order are never deduplicated, so + /// tree routes sharing a deep component are unaffected. 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; if first_chunk.is_zero() { return candidates; } + let mut limits = LimitsCache::new(); + let mut limited_bottlenecks: HashSet = HashSet::new(); + let bottleneck_of_limited = |idx: usize, limits: &mut LimitsCache| { + Self::max_fillable_input(&ctx.ordered[idx], ctx.market, limits) + .filter(|path_cap| path_cap.input_cap < *ctx.order.amount()) + .map(|path_cap| path_cap.bottleneck) + }; + for &idx in &candidates { + if let Some(bottleneck) = bottleneck_of_limited(idx, &mut limits) { + limited_bottlenecks.insert(bottleneck); + } + } let timeout_ms = self.timeout.as_millis() as u64; let empty: HashMap> = HashMap::new(); let mut probe_indices: Vec = (0..ctx @@ -1077,19 +1298,21 @@ impl WaterFillAlgorithm { marginal.push((idx, BigInt::from(step.amount_out) - activation)); } marginal.sort_by(|(_, a), (_, b)| b.cmp(a)); - for (idx, net) in marginal - .into_iter() - .take(SHARED_MARGIN_PATHS) - { - if net <= BigInt::zero() { - continue; + let mut winners = 0usize; + for (idx, net) in marginal { + if winners >= SHARED_MARGIN_PATHS || candidates.len() >= SHARED_MAX_CANDIDATES { + break; } - if !candidates.contains(&idx) { - candidates.push(idx); + if net <= BigInt::zero() || candidates.contains(&idx) { + continue; } - if candidates.len() >= SHARED_MAX_CANDIDATES { - break; + if let Some(bottleneck) = bottleneck_of_limited(idx, &mut limits) { + if !limited_bottlenecks.insert(bottleneck) { + continue; + } } + candidates.push(idx); + winners += 1; } candidates } @@ -1142,12 +1365,20 @@ impl WaterFillAlgorithm { let remainder = &amount_in - &base_chunk * num_chunks; let timeout_ms = self.timeout.as_millis() as u64; - let caps = Self::allocation_caps(ctx, subset); + let mut budget = ComponentBudget::new(); let mut overlay: HashMap> = HashMap::new(); let mut activated: Vec = vec![!gate; subset.len()]; let mut active_count = if gate { 0 } else { subset.len() }; + // The gated pass anchors the top-ranked candidate in the active set: probe winners are + // picked by marginal price, so several inventory-limited paths can win every `max_paths` + // activation slot, saturate their shared budgets together, and leave the order + // unfillable — the rebuilt route then scales the deficit onto the over-committed legs + // and is discarded. The best-ranked path can absorb the remainder, so it keeps a slot. + if gate && !activated.is_empty() { + activated[0] = true; + active_count = 1; + } let mut counts: Vec = vec![0; subset.len()]; - let mut cum_in: Vec = vec![BigUint::zero(); subset.len()]; let mut schedule: Vec<(usize, BigUint)> = Vec::with_capacity(num_chunks); for chunk_idx in 0..num_chunks { @@ -1161,10 +1392,7 @@ impl WaterFillAlgorithm { if !activated[i] && active_count >= self.max_paths { continue; } - if caps[i] - .as_ref() - .is_some_and(|cap| &cum_in[i] + &chunk > *cap) - { + if budget.path_saturated(ctx.market, &ctx.ordered[path_idx]) { continue; } let Some(step) = Self::simulate_step( @@ -1175,6 +1403,9 @@ impl WaterFillAlgorithm { ) else { continue; }; + if !budget.allows(ctx.market, &step) { + continue; + } let gross_marginal = BigInt::from(step.amount_out.clone()); let net_marginal = if activated[i] { gross_marginal @@ -1194,6 +1425,7 @@ impl WaterFillAlgorithm { let Some((best_i, _, step)) = best else { break; }; + budget.commit(&step); for (id, state) in step.new_states { overlay.insert(id, state); } @@ -1202,7 +1434,6 @@ impl WaterFillAlgorithm { active_count += 1; } counts[best_i] += 1; - cum_in[best_i] += &chunk; schedule.push((best_i, chunk)); } Some((counts, schedule)) @@ -1927,6 +2158,107 @@ mod tests { assert!(result.route().validate().is_ok(), "route validation failed"); } + /// Capped-probe slots are deduplicated by bottleneck component: many failing paths sharing + /// one limited pool must not crowd a distinct limited pool out of the probe set. 20 two-hop + /// paths all bottlenecked by the same B->C pool rank above the direct limited pool by capped + /// output; without dedup they fill all [`CAPPED_PROBE_PATHS`] slots and the direct pool is + /// dropped from the split even though its price (the market's best) wins a probe. + #[tokio::test] + async fn test_capped_probe_slots_deduped_by_bottleneck() { + let a = token_with_decimals(0x01, "A", 18); + let b = token_with_decimals(0x02, "B", 18); + let c = token_with_decimals(0x03, "C", 18); + let unit = BigUint::from(10u64).pow(18); + + // Direct A->C, best price in the market (1000 C/A) but only 800 A of input reserve: + // fails the 2000 A order, capped at 400 A (~267k C). + let limited_direct = Box::new(ConstantProductSim { + reserve_0: BigUint::from(800u64) * &unit, + reserve_1: BigUint::from(800_000u64) * &unit, + gas: 50_000, + }) as Box; + // 20 deep A->B bridges (rate 1.0) into one limited B->C pool (rate 950, 1,200 B input + // reserve): 20 failing two-hop paths (the bridges pass ~1,666 B at the full order), + // every one bottlenecked by the same component, each with capped output ~365k C — + // above the direct limited pool's 267k, but at a worse spot price than its 1000. + let bridge = || { + Box::new(ConstantProductSim { + reserve_0: BigUint::from(10_000u64) * &unit, + reserve_1: BigUint::from(10_000u64) * &unit, + gas: 50_000, + }) as Box + }; + let bottleneck_bc = Box::new(ConstantProductSim { + reserve_0: BigUint::from(1_200u64) * &unit, + reserve_1: BigUint::from(1_140_000u64) * &unit, + gas: 50_000, + }) as Box; + // Direct pools that absorb the full order, ranking between the two-hop capped outputs + // and the direct limited pool: one deep (833k C) and 15 shallow (571k C each), pushing + // the direct limited pool below the top-32 marginal-probe window. + let deep = Box::new(ConstantProductSim { + reserve_0: BigUint::from(10_000u64) * &unit, + reserve_1: BigUint::from(5_000_000u64) * &unit, + gas: 50_000, + }) as Box; + let shallow = || { + Box::new(ConstantProductSim { + reserve_0: BigUint::from(5_000u64) * &unit, + reserve_1: BigUint::from(2_000_000u64) * &unit, + gas: 50_000, + }) as Box + }; + + let bridge_names: Vec = (0..20) + .map(|i| format!("bridge_{i}")) + .collect(); + let shallow_names: Vec = (0..15) + .map(|i| format!("shallow_{i}")) + .collect(); + let mut components = vec![ + ("limited_direct", &a, &c, limited_direct), + ("bottleneck_bc", &b, &c, bottleneck_bc), + ("deep", &a, &c, deep), + ]; + for name in &bridge_names { + components.push((name.as_str(), &a, &b, bridge())); + } + for name in &shallow_names { + components.push((name.as_str(), &a, &c, shallow())); + } + let (market, gm) = setup_market_weighted_boxed(components); + + let order = Order::new( + a.address.clone(), + c.address.clone(), + BigUint::from(2_000u64) * &unit, + OrderSide::Sell, + addr(0xFF), + ); + let result = WaterFillAlgorithm::with_config(config()) + .unwrap() + .find_best_route(gm.graph(), market.clone(), None, None, &order) + .await + .expect("order fills across the direct components"); + + let limited_amount_in: BigUint = result + .route() + .swaps() + .iter() + .filter(|swap| swap.component_id() == "limited_direct") + .map(|swap| swap.amount_in().clone()) + .sum(); + assert!( + !limited_amount_in.is_zero(), + "the direct limited component must keep a probe slot despite 20 same-bottleneck paths", + ); + assert!( + limited_amount_in <= BigUint::from(400u64) * &unit, + "allocation must respect the 400 A cap, got {limited_amount_in}", + ); + assert!(result.route().validate().is_ok(), "route validation failed"); + } + /// The single-hop cap is the first hop's `get_limits` sell limit, exact and unconverted. #[tokio::test] async fn test_max_fillable_input_single_hop_uses_exact_limit() { @@ -1949,10 +2281,11 @@ mod tests { .extract_subset_with_overlay(&HashSet::from(["limited".to_string()])); // ConstantProductSim's sell limit is half the input reserve: 400 A. - assert_eq!( - WaterFillAlgorithm::max_fillable_input(&paths[0], &subset), - Some(BigUint::from(400u64) * &unit), - ); + let cap = + WaterFillAlgorithm::max_fillable_input(&paths[0], &subset, &mut LimitsCache::new()) + .expect("cap must be computable"); + assert_eq!(cap.input_cap, BigUint::from(400u64) * &unit); + assert_eq!(cap.bottleneck, "limited".to_string()); } /// A downstream hop's limit is back-converted to path-input units through the preceding @@ -1989,13 +2322,20 @@ mod tests { // Hop 2's sell limit is 400 B; through hop 1's 2.0 spot that is 200 A of path input, // tighter than hop 1's own 5,000 A limit. f64 conversion may be off by a rounding ulp. - let cap = WaterFillAlgorithm::max_fillable_input(&paths[0], &subset) - .expect("cap must be computable"); + let cap = + WaterFillAlgorithm::max_fillable_input(&paths[0], &subset, &mut LimitsCache::new()) + .expect("cap must be computable"); + assert_eq!(cap.bottleneck, "second".to_string()); let expected = BigUint::from(200u64) * &unit; - let diff = if cap > expected { &cap - &expected } else { &expected - &cap }; + let diff = if cap.input_cap > expected { + &cap.input_cap - &expected + } else { + &expected - &cap.input_cap + }; assert!( diff <= BigUint::from(1_000_000u64), - "cap {cap} should be within rounding error of {expected}", + "cap {} should be within rounding error of {expected}", + cap.input_cap, ); } From 3af5651e8ac6cbbdb95f96a201b3b94bf8021cbd Mon Sep 17 00:00:00 2001 From: kayibal Date: Wed, 5 Aug 2026 22:16:21 +0100 Subject: [PATCH 3/3] fix(water-fill): rank capped probes by input cap, not spot price MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Live validation showed the capped-probe slots being crowded out on blocks with many failing paths: broken pools advertise fantasy spot prices (multiples of the market rate) that outrank honest inventory, and VM components can lack derived edge data entirely, which excluded their paths from selection. Probes are now deduplicated keeping the largest cap per bottleneck and ranked by input cap — the path's real fillable notional from the same get_limits oracle the allocation budgets trust, comparable across paths because every cap is in path-input token units. A failed capped simulation no longer consumes one of the CAPPED_PROBE_PATHS slots; attempts continue down the ranked list, bounded by CAPPED_PROBE_ATTEMPTS. Adds debug-level logs for capped-probe selection, fill-and-spill candidate selection, and per-candidate nets with legs. Live result (paired same-block probes, USDC->WETH, water_fill 3-hop): fermiswap retained at ~its inventory in 8/9 probes across $600k-$2M orders (previously 0/9 at these sizes), +0.9 to +6.9 bps over main. Co-Authored-By: Claude Fable 5 --- fynd-core/src/algorithm/water_fill.rs | 108 ++++++++++++++++---------- 1 file changed, 65 insertions(+), 43 deletions(-) diff --git a/fynd-core/src/algorithm/water_fill.rs b/fynd-core/src/algorithm/water_fill.rs index ba9fd8043..83929fd0b 100644 --- a/fynd-core/src/algorithm/water_fill.rs +++ b/fynd-core/src/algorithm/water_fill.rs @@ -77,8 +77,12 @@ const SHARED_MARGIN_PATHS: usize = 8; /// always added to the fill-and-spill marginal probe, wherever they rank. An inventory-limited /// path's capped output sinks below every pool deep enough to absorb the whole order once the /// order is a small multiple of its inventory, but its price can still win the probe. Selection -/// deduplicates by bottleneck component and ranks by spot price (`select_capped_probes`). +/// deduplicates by bottleneck component and ranks by input cap (`select_capped_probes`). const CAPPED_PROBE_PATHS: usize = 16; +/// Upper bound on capped ranking simulation attempts while filling the [`CAPPED_PROBE_PATHS`] +/// slots. A failed capped simulation (broken pool, stale limit) does not consume a slot, so the +/// attempt bound keeps a pathological block from burning solve time on junk paths. +const CAPPED_PROBE_ATTEMPTS: usize = 48; /// Upper bound on fill-and-spill candidate paths. const SHARED_MAX_CANDIDATES: usize = 12; /// Candidate states retained per intermediate token during bounded discovery expansion. @@ -253,9 +257,6 @@ struct PathCap { input_cap: BigUint, /// Component whose sell limit set the cap. bottleneck: ComponentId, - /// Product of every edge's spot price (path output per input at zero size); `None` when any - /// edge's derived data is missing. - spot_product: Option, } /// Shared inputs threaded through every split-allocation pass: the ranked candidate paths, the @@ -460,7 +461,6 @@ impl WaterFillAlgorithm { // Path-input units per current hop-input unit: the product of preceding edges' spot // prices (each token_out per token_in). let mut path_input_price = 1.0_f64; - let mut spot_product = Some(1.0_f64); let hop_count = path.len(); for (hop, (address_in, edge, address_out)) in path.iter().enumerate() { let max_in = Self::cached_sell_limit( @@ -481,40 +481,37 @@ impl WaterFillAlgorithm { { cap = Some((hop_cap, edge.component_id.clone())); } - match edge - .data - .as_ref() - .map(|data| data.spot_price) - { - Some(spot) if spot > 0.0 && spot.is_finite() => { - if hop + 1 < hop_count { - path_input_price *= spot; - } - spot_product = spot_product.map(|product| product * spot); + if hop + 1 < hop_count { + let spot = edge.data.as_ref()?.spot_price; + if spot <= 0.0 || !spot.is_finite() { + return None; } - // A final hop's missing spot price does not affect the cap, only the proxy. - _ if hop + 1 == hop_count => spot_product = None, - _ => return None, + path_input_price *= spot; } } let (input_cap, bottleneck) = cap?; if input_cap.is_zero() { return None; } - Some(PathCap { input_cap, bottleneck, spot_product }) + Some(PathCap { input_cap, bottleneck }) } - /// Chooses which full-amount-failing paths get an honest limit-capped ranking simulation, - /// returning `(path index, cap)` pairs. + /// Orders the full-amount-failing paths for the honest limit-capped ranking simulations, + /// returning `(path index, cap)` attempt pairs, largest cap first. /// /// Computing a cap is cheap (cached `get_limits` lookups), but the capped simulation is not, /// and at three hops most candidate paths fail once the order is large — simulating every - /// one dominated solve time. All failing paths keep gross 0 except the strongest probes: - /// paths are deduplicated by bottleneck component (paths sharing a bottleneck cannot - /// contribute its inventory twice), then the best [`CAPPED_PROBE_PATHS`] by spot price are - /// kept — the fill-and-spill marginal probe these paths are retained for is itself - /// price-driven. Paths whose cap covers the full order (the failure has another cause) or - /// cannot absorb one fine chunk are dropped. + /// one dominated solve time. All failing paths keep gross 0 except the winners of the + /// [`CAPPED_PROBE_PATHS`] slots: paths are deduplicated by bottleneck component (paths + /// sharing a bottleneck cannot contribute its inventory twice) and ranked by input cap — + /// the path's real fillable notional, from the same `get_limits` oracle the allocation + /// budgets trust, comparable across paths because every cap is in path-input token units. + /// Spot prices are deliberately not the ranking key: broken pools advertise fantasy spots + /// and crowd out honest inventory. The list is bounded by [`CAPPED_PROBE_ATTEMPTS`] since a + /// failed capped simulation does not consume a probe slot. + /// + /// Paths whose cap covers the full order (the failure has another cause) or cannot absorb + /// one fine chunk are dropped. fn select_capped_probes( paths: &[Path], failing: &[usize], @@ -523,7 +520,7 @@ impl WaterFillAlgorithm { limits: &mut LimitsCache, ) -> Vec<(usize, BigUint)> { let chunk_floor = order_amount / FINE_CHUNKS; - let mut best_per_bottleneck: HashMap = HashMap::new(); + let mut best_per_bottleneck: HashMap = HashMap::new(); for &idx in failing { let Some(path_cap) = Self::max_fillable_input(&paths[idx], market, limits) else { continue; @@ -531,28 +528,36 @@ impl WaterFillAlgorithm { if path_cap.input_cap >= *order_amount || path_cap.input_cap < chunk_floor { continue; } - let Some(spot) = path_cap.spot_product else { - continue; - }; let keep = best_per_bottleneck .get(&path_cap.bottleneck) - .is_none_or(|(_, _, existing)| spot > *existing); + .is_none_or(|(_, existing)| path_cap.input_cap > *existing); if keep { - best_per_bottleneck.insert(path_cap.bottleneck, (idx, path_cap.input_cap, spot)); + best_per_bottleneck.insert(path_cap.bottleneck, (idx, path_cap.input_cap)); } } - let mut probes: Vec<(usize, BigUint, f64)> = best_per_bottleneck + let mut probes: Vec<(usize, BigUint)> = best_per_bottleneck .into_values() .collect(); - probes.sort_by(|(_, _, a), (_, _, b)| { - b.partial_cmp(a) - .unwrap_or(Ordering::Equal) - }); - probes.truncate(CAPPED_PROBE_PATHS); + probes.sort_by(|(_, a), (_, b)| b.cmp(a)); + probes.truncate(CAPPED_PROBE_ATTEMPTS); + debug!( + failing = failing.len(), + attempts = probes.len(), + probes = ?probes + .iter() + .take(CAPPED_PROBE_PATHS) + .map(|(idx, cap)| { + let components: Vec = paths[*idx] + .edge_iter() + .iter() + .map(|edge| edge.component_id.clone()) + .collect(); + (components, cap.to_string()) + }) + .collect::>(), + "water-fill capped probe selection" + ); probes - .into_iter() - .map(|(idx, cap, _)| (idx, cap)) - .collect() } /// Gross output of a path simulated at its limit cap, used to rank it honestly. @@ -755,7 +760,9 @@ impl WaterFillAlgorithm { for (idx, cap) in Self::select_capped_probes(&paths, &failing, &market, &amount_in, &mut limits) { - if start.elapsed().as_millis() as u64 > timeout_ms { + if capped_grosses.len() >= CAPPED_PROBE_PATHS || + start.elapsed().as_millis() as u64 > timeout_ms + { break; } if let Some(gross) = @@ -869,6 +876,16 @@ impl Algorithm for WaterFillAlgorithm { let mut best: Option<(BigInt, SplitCandidate)> = None; for cand in candidates { let net = cand.net(&gas_price, token_prices.as_ref(), token_out); + debug!( + %net, + legs = ?cand + .route + .swaps() + .iter() + .map(|swap| (swap.protocol(), swap.amount_in().to_string())) + .collect::>(), + "water-fill split candidate" + ); let beats = match (&best, &baseline_net) { (Some((current, _)), _) => net > *current, (None, Some(base)) => net > *base, @@ -1314,6 +1331,11 @@ impl WaterFillAlgorithm { candidates.push(idx); winners += 1; } + debug!( + ?candidates, + capped_positions = ?ctx.capped_positions, + "water-fill fill-and-spill candidate selection" + ); candidates }