diff --git a/fynd-core/src/algorithm/sim_guard.rs b/fynd-core/src/algorithm/sim_guard.rs index 08cb263e..96290db9 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 e01e383c..83929fd0 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,16 @@ 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 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 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. @@ -144,6 +154,109 @@ 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, } /// Shared inputs threaded through every split-allocation pass: the ranked candidate paths, the @@ -151,6 +264,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 +279,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, @@ -210,6 +329,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)?; @@ -232,9 +353,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. @@ -292,8 +417,179 @@ 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, 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 + /// (`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 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 hop_count = path.len(); + for (hop, (address_in, edge, address_out)) in path.iter().enumerate() { + 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 { + 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, 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; + } + path_input_price *= spot; + } + } + let (input_cap, bottleneck) = cap?; + if input_cap.is_zero() { + return None; + } + Some(PathCap { input_cap, bottleneck }) + } + + /// 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 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], + 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 keep = best_per_bottleneck + .get(&path_cap.bottleneck) + .is_none_or(|(_, existing)| path_cap.input_cap > *existing); + if keep { + best_per_bottleneck.insert(path_cap.bottleneck, (idx, path_cap.input_cap)); + } + } + let mut probes: Vec<(usize, BigUint)> = best_per_bottleneck + .into_values() + .collect(); + 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 + } + + /// 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>, + cap: &BigUint, + ) -> Option { + 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 +715,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 failing: Vec = Vec::new(); for (idx, path) in paths.iter().enumerate() { if start.elapsed().as_millis() as u64 > timeout_ms { @@ -446,11 +743,41 @@ 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())), + 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 capped_grosses.len() >= CAPPED_PROBE_PATHS || + 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())), } } @@ -458,17 +785,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 +826,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(), @@ -543,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, @@ -857,6 +1200,7 @@ impl WaterFillAlgorithm { let timeout_ms = self.timeout.as_millis() as u64; let path_count = subset.len(); + let mut budget = ComponentBudget::new(); let mut committed: Vec>> = (0..path_count) .map(|_| HashMap::new()) .collect(); @@ -871,6 +1215,9 @@ impl WaterFillAlgorithm { let mut best: Option<(usize, BigInt, StepResult)> = None; for (i, &path_idx) in subset.iter().enumerate() { + if budget.path_saturated(ctx.market, &ctx.ordered[path_idx]) { + continue; + } let Some(step) = Self::simulate_step( &ctx.ordered[path_idx], ctx.market, @@ -879,6 +1226,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 @@ -898,6 +1248,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); } @@ -909,25 +1260,53 @@ 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. + /// + /// 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 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; @@ -936,20 +1315,27 @@ 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; } + debug!( + ?candidates, + capped_positions = ?ctx.capped_positions, + "water-fill fill-and-spill candidate selection" + ); candidates } @@ -1001,9 +1387,19 @@ impl WaterFillAlgorithm { let remainder = &amount_in - &base_chunk * num_chunks; let timeout_ms = self.timeout.as_millis() as u64; + 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 schedule: Vec<(usize, BigUint)> = Vec::with_capacity(num_chunks); @@ -1018,6 +1414,9 @@ impl WaterFillAlgorithm { if !activated[i] && active_count >= self.max_paths { continue; } + if budget.path_saturated(ctx.market, &ctx.ordered[path_idx]) { + continue; + } let Some(step) = Self::simulate_step( &ctx.ordered[path_idx], ctx.market, @@ -1026,6 +1425,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 @@ -1045,6 +1447,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); } @@ -1498,7 +1901,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 +2028,339 @@ 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"); + } + + /// 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() { + 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. + 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 + /// 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, &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.input_cap > expected { + &cap.input_cap - &expected + } else { + &expected - &cap.input_cap + }; + assert!( + diff <= BigUint::from(1_000_000u64), + "cap {} should be within rounding error of {expected}", + cap.input_cap, + ); + } + /// 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.