1use std::{
32 cmp::{Ordering, Reverse},
33 collections::{HashMap, HashSet},
34 time::{Duration, Instant},
35};
36
37use num_bigint::{BigInt, BigUint};
38use num_traits::Zero;
39use petgraph::{graph::NodeIndex, prelude::EdgeRef};
40use tracing::{debug, instrument};
41use tycho_simulation::{
42 tycho_common::simulation::protocol_sim::ProtocolSim, tycho_core::models::Address,
43};
44
45use super::{
46 most_liquid::DepthAndPrice,
47 sim_guard::GuardedProtocolSim,
48 split_primitives::{build_split_route, HopDescriptor, PathAllocation, SimulatedHop},
49 Algorithm, AlgorithmConfig, MostLiquidAlgorithm, NoPathReason,
50};
51use crate::{
52 derived::{computation::ComputationRequirements, types::TokenGasPrices, SharedDerivedDataRef},
53 feed::market_data::{MarketData, MarketDataView, MarketState, StateLabel},
54 graph::{petgraph::StableDiGraph, EdgeData, Path, PetgraphStableDiGraphManager},
55 types::{ComponentId, Order, Route, RouteResult},
56 AlgorithmError,
57};
58
59const DEFAULT_MAX_CANDIDATES: usize = 5000;
61const BOUNDED_DISCOVERY_CANDIDATES: usize = 128;
64const DEFAULT_MAX_PATHS: usize = 4;
66const COARSE_CHUNKS: usize = 20;
68const FINE_CHUNKS: usize = 256;
70const SHARED_FULL_PATHS: usize = 8;
72const SHARED_MARGIN_PROBE_PATHS: usize = 32;
74const SHARED_MARGIN_PATHS: usize = 8;
76const SHARED_MAX_CANDIDATES: usize = 12;
78const CANDIDATE_STATES_PER_NODE: usize = 4;
80const CANDIDATE_EDGES_PER_STATE: usize = 16;
82const CANDIDATE_DIRECT_EDGES_PER_TOKEN: usize = 4;
84const CANDIDATE_CONNECTOR_EDGES_PER_TOKEN: usize = 2;
86const DERIVED_ANCHOR_COUNT: usize = 16;
89const EXCHANGE_DELTA_FLOOR: usize = 64;
92const EXCHANGE_MAX_SIMS: usize = 400;
94
95struct SplitCandidate {
97 route: Route,
98 gross: BigUint,
99 gas: BigUint,
100}
101
102impl SplitCandidate {
103 fn net(
105 &self,
106 gas_price: &BigUint,
107 token_prices: Option<&TokenGasPrices>,
108 token_out: &Address,
109 ) -> BigInt {
110 let cost =
111 WaterFillAlgorithm::gas_cost_in_token(&self.gas, gas_price, token_prices, token_out);
112 match cost {
113 Some(c) => BigInt::from(self.gross.clone()) - BigInt::from(c),
114 None => BigInt::from(self.gross.clone()),
115 }
116 }
117}
118
119pub struct WaterFillAlgorithm {
122 min_hops: usize,
123 max_hops: usize,
124 timeout: Duration,
125 max_candidates: usize,
126 max_paths: usize,
127 connector_tokens: Option<HashSet<Address>>,
128}
129
130struct ExchangeMove {
134 donor: usize,
135 recipient: usize,
136 donor_net: BigInt,
137 recip_net: BigInt,
138 gain: BigInt,
139}
140
141struct StepResult {
144 amount_out: BigUint,
145 gas: BigUint,
146 new_states: Vec<(ComponentId, Box<dyn ProtocolSim>)>,
147}
148
149struct SplitContext<'a, 'g> {
153 ordered: &'a [Path<'g, DepthAndPrice>],
154 market: &'a MarketState,
155 gas_price: &'a BigUint,
156 token_prices: Option<&'a TokenGasPrices>,
157 order: &'a Order,
158 start: Instant,
159}
160
161struct SetupResult<'a> {
165 ordered: Vec<Path<'a, DepthAndPrice>>,
166 market: MarketState,
167 gas_price: BigUint,
168 best_single: Option<RouteResult>,
169 token_prices: Option<TokenGasPrices>,
170}
171
172impl WaterFillAlgorithm {
173 pub(crate) fn with_config(config: AlgorithmConfig) -> Result<Self, AlgorithmError> {
175 Ok(Self {
176 min_hops: config.min_hops(),
177 max_hops: config.max_hops(),
178 timeout: config.timeout(),
179 max_candidates: config
180 .max_routes()
181 .unwrap_or(DEFAULT_MAX_CANDIDATES)
182 .max(DEFAULT_MAX_PATHS),
183 max_paths: DEFAULT_MAX_PATHS,
184 connector_tokens: config.connector_tokens().cloned(),
185 })
186 }
187
188 fn simulate_step(
192 path: &Path<DepthAndPrice>,
193 market: &MarketState,
194 overlay: &HashMap<ComponentId, Box<dyn ProtocolSim>>,
195 amount: BigUint,
196 ) -> Option<StepResult> {
197 let mut current = amount;
198 let mut total_gas = BigUint::zero();
199 let path_reuses_component = {
204 let mut seen: HashSet<&ComponentId> = HashSet::with_capacity(path.len());
205 !path
206 .edge_iter()
207 .iter()
208 .all(|e| seen.insert(&e.component_id))
209 };
210 let mut intra_path_states: HashMap<ComponentId, Box<dyn ProtocolSim>> = HashMap::new();
211 let mut new_states: Vec<(ComponentId, Box<dyn ProtocolSim>)> =
212 Vec::with_capacity(path.len());
213
214 for (address_in, edge, address_out) in path.iter() {
215 let token_in = market.get_token(address_in)?;
216 let token_out = market.get_token(address_out)?;
217 let component_id = &edge.component_id;
218 let state = intra_path_states
219 .get(component_id)
220 .map(Box::as_ref)
221 .or_else(|| {
222 overlay
223 .get(component_id)
224 .map(Box::as_ref)
225 })
226 .or_else(|| market.get_simulation_state(component_id))?;
227 let result = state
228 .get_amount_out_guarded(current.clone(), token_in, token_out)
229 .ok()?;
230 total_gas += &result.gas;
231 if path_reuses_component {
232 intra_path_states.insert(component_id.clone(), result.new_state.clone_box());
233 }
234 new_states.push((component_id.clone(), result.new_state));
235 current = result.amount;
236 }
237 Some(StepResult { amount_out: current, gas: total_gas, new_states })
238 }
239
240 fn gas_cost_in_token(
242 total_gas: &BigUint,
243 gas_price_wei: &BigUint,
244 token_prices: Option<&TokenGasPrices>,
245 token_out: &Address,
246 ) -> Option<BigUint> {
247 let price = token_prices?.get(token_out)?;
248 if price.denominator.is_zero() {
249 return None;
250 }
251 Some(total_gas * gas_price_wei * &price.numerator / &price.denominator)
252 }
253
254 fn activation_cost(ctx: &SplitContext, gas: &BigUint) -> BigInt {
258 Self::gas_cost_in_token(gas, ctx.gas_price, ctx.token_prices, ctx.order.token_out())
259 .map(BigInt::from)
260 .unwrap_or_else(BigInt::zero)
261 }
262
263 fn select_disjoint(ranked: &[Path<DepthAndPrice>], max_paths: usize) -> Vec<usize> {
270 let mut visited_components: HashSet<&ComponentId> = HashSet::new();
271 let mut selected = Vec::new();
272 for (idx, path) in ranked.iter().enumerate() {
273 let path_components: Vec<&ComponentId> = path
274 .edge_iter()
275 .iter()
276 .map(|e| &e.component_id)
277 .collect();
278 if path_components
279 .iter()
280 .any(|c| visited_components.contains(*c))
281 {
282 continue;
283 }
284 for c in path_components {
285 visited_components.insert(c);
286 }
287 selected.push(idx);
288 if selected.len() >= max_paths {
289 break;
290 }
291 }
292 selected
293 }
294
295 #[instrument(level = "debug", skip_all)]
298 async fn setup<'a>(
299 &self,
300 graph: &'a StableDiGraph<DepthAndPrice>,
301 market: MarketData,
302 label: Option<StateLabel>,
303 derived: Option<SharedDerivedDataRef>,
304 order: &Order,
305 start: Instant,
306 ) -> Result<SetupResult<'a>, AlgorithmError> {
307 let token_prices = if let Some(ref derived) = derived {
308 derived
309 .read()
310 .await
311 .token_prices()
312 .cloned()
313 } else {
314 None
315 };
316
317 let all_paths = MostLiquidAlgorithm::find_paths(
318 graph,
319 order.token_in(),
320 order.token_out(),
321 self.min_hops,
322 self.max_hops,
323 self.connector_tokens.as_ref(),
324 )?;
325 if all_paths.is_empty() {
326 return Err(AlgorithmError::NoPath {
327 from: order.token_in().clone(),
328 to: order.token_out().clone(),
329 reason: NoPathReason::NoGraphPath,
330 });
331 }
332
333 let mut scored: Vec<(Path<DepthAndPrice>, f64)> = all_paths
334 .into_iter()
335 .map(|p| {
336 let s = MostLiquidAlgorithm::try_score_path(&p).unwrap_or(f64::MIN);
337 (p, s)
338 })
339 .collect();
340 scored.sort_by(|(_, a), (_, b)| {
341 b.partial_cmp(a)
342 .unwrap_or(std::cmp::Ordering::Equal)
343 });
344 scored.truncate(self.max_candidates);
345 let mut paths: Vec<Path<DepthAndPrice>> = scored
346 .into_iter()
347 .map(|(p, _)| p)
348 .collect();
349
350 let timeout_ms = self.timeout.as_millis() as u64;
351 let market = {
352 let view = match label.as_ref() {
353 Some(l) => market
354 .read_labeled(l)
355 .await
356 .map_err(|e| AlgorithmError::Other(e.to_string()))?,
357 None => market.read().await,
358 };
359 if view.gas_price().is_none() {
360 return Err(AlgorithmError::DataNotFound { kind: "gas price", id: None });
361 }
362 let anchor_tokens = derive_anchor_tokens(graph);
367 let bounded = find_candidate_paths(
368 graph,
369 &view,
370 order,
371 CandidateSearchConfig {
372 min_hops: self.min_hops,
373 max_hops: self.max_hops,
374 max_candidates: BOUNDED_DISCOVERY_CANDIDATES,
375 connector_tokens: self.connector_tokens.as_ref(),
376 anchor_tokens: &anchor_tokens,
377 source_token: order.token_in(),
378 start: &start,
379 timeout_ms,
380 },
381 );
382 match bounded {
383 Ok((bounded_paths, _)) => {
384 let mut keys: HashSet<Vec<ComponentId>> = paths.iter().map(path_key).collect();
385 let mut union = Vec::with_capacity(bounded_paths.len() + paths.len());
386 for path in bounded_paths {
387 if keys.insert(path_key(&path)) {
388 union.push(path);
389 }
390 }
391 union.append(&mut paths);
392 paths = union;
393 }
394 Err(e) => {
398 debug!(error = %e, "water-fill bounded discovery failed; using exhaustive candidates only")
399 }
400 }
401 let component_ids: HashSet<ComponentId> = paths
402 .iter()
403 .flat_map(|p| {
404 p.edge_iter()
405 .iter()
406 .map(|e| e.component_id.clone())
407 })
408 .collect();
409 let subset = view.extract_subset_with_overlay(&component_ids);
410 drop(view);
411 subset
412 };
413 let gas_price = market
414 .gas_price()
415 .ok_or(AlgorithmError::DataNotFound { kind: "gas price", id: None })?
416 .effective_gas_price()
417 .clone();
418
419 let amount_in = order.amount().clone();
420 let mut best_single: Option<RouteResult> = None;
421 let mut full_outputs: Vec<(usize, BigUint)> = Vec::new();
422
423 for (idx, path) in paths.iter().enumerate() {
424 if start.elapsed().as_millis() as u64 > timeout_ms {
425 break;
426 }
427 match MostLiquidAlgorithm::simulate_path(
428 path,
429 &market,
430 token_prices.as_ref(),
431 amount_in.clone(),
432 ) {
433 Ok(result) => {
434 let gross = result
435 .route()
436 .swaps()
437 .last()
438 .map(|s| s.amount_out().clone())
439 .unwrap_or_else(BigUint::zero);
440 full_outputs.push((idx, gross));
441 if best_single
442 .as_ref()
443 .map(|b| result.net_amount_out() > b.net_amount_out())
444 .unwrap_or(true)
445 {
446 best_single = Some(result);
447 }
448 }
449 Err(_) => full_outputs.push((idx, BigUint::zero())),
454 }
455 }
456
457 full_outputs.sort_by(|(_, a), (_, b)| b.cmp(a));
461 let ordered: Vec<Path<DepthAndPrice>> = full_outputs
462 .into_iter()
463 .map(|(idx, _)| paths[idx].clone())
464 .collect();
465
466 debug!(
467 candidate_paths = ordered.len(),
468 elapsed_ms = start.elapsed().as_millis(),
469 "water-fill discovery + full-amount ranking"
470 );
471 Ok(SetupResult { ordered, market, gas_price, best_single, token_prices })
472 }
473}
474
475impl Algorithm for WaterFillAlgorithm {
476 type GraphType = StableDiGraph<DepthAndPrice>;
477 type GraphManager = PetgraphStableDiGraphManager<DepthAndPrice>;
478
479 fn name(&self) -> &str {
480 "water_fill"
481 }
482
483 #[instrument(level = "debug", skip_all, fields(order_id = %order.id()))]
484 async fn find_best_route(
485 &self,
486 graph: &Self::GraphType,
487 market: MarketData,
488 label: Option<StateLabel>,
489 derived: Option<SharedDerivedDataRef>,
490 order: &Order,
491 ) -> Result<RouteResult, AlgorithmError> {
492 let start = Instant::now();
493 if !order.is_sell() {
494 return Err(AlgorithmError::ExactOutNotSupported);
495 }
496
497 let SetupResult { ordered, market, gas_price, best_single, token_prices } = self
498 .setup(graph, market, label, derived, order, start)
499 .await?;
500 let token_out = order.token_out();
501 let ctx = SplitContext {
502 ordered: &ordered,
503 market: &market,
504 gas_price: &gas_price,
505 token_prices: token_prices.as_ref(),
506 order,
507 start,
508 };
509
510 let mut candidates: Vec<SplitCandidate> = Vec::new();
512 let disjoint = Self::select_disjoint(ctx.ordered, self.max_paths);
517 let coarse = (disjoint.len() >= 2)
518 .then(|| self.disjoint_waterfill(&ctx, &disjoint, COARSE_CHUNKS, true))
519 .flatten();
520 if let Some(coarse) = coarse.as_deref() {
521 if let Some(c) = self.build_disjoint_legs(&ctx, &disjoint, coarse) {
523 candidates.push(c);
524 }
525 if let Some(c) = self.disjoint_refine(&ctx, &disjoint, coarse, FINE_CHUNKS) {
528 candidates.push(c);
529 }
530 }
531 if let Some(c) = self.fillspill_alloc(&ctx, FINE_CHUNKS) {
534 candidates.push(c);
535 }
536
537 let candidate_count = candidates.len();
540 let baseline_net = best_single
541 .as_ref()
542 .map(|b| b.net_amount_out().clone());
543 let mut best: Option<(BigInt, SplitCandidate)> = None;
544 for cand in candidates {
545 let net = cand.net(&gas_price, token_prices.as_ref(), token_out);
546 let beats = match (&best, &baseline_net) {
547 (Some((current, _)), _) => net > *current,
548 (None, Some(base)) => net > *base,
549 (None, None) => true,
550 };
551 if beats {
552 best = Some((net, cand));
553 }
554 }
555 let split_won = best.is_some();
556 debug!(
557 candidate_count,
558 split_won,
559 elapsed_ms = start.elapsed().as_millis(),
560 "water-fill selected {}",
561 if split_won { "split candidate" } else { "single path" }
562 );
563 match best {
564 Some((net, cand)) => Ok(RouteResult::new(cand.route, net, gas_price)),
565 None => best_single.ok_or(AlgorithmError::InsufficientLiquidity),
567 }
568 }
569
570 fn computation_requirements(&self) -> ComputationRequirements {
571 ComputationRequirements::none()
572 .allow_stale("token_prices")
573 .expect("Conflicting Computation Requirements")
574 }
575
576 fn timeout(&self) -> Duration {
577 self.timeout
578 }
579}
580
581impl WaterFillAlgorithm {
582 fn disjoint_refine(
588 &self,
589 ctx: &SplitContext,
590 disjoint: &[usize],
591 coarse: &[BigUint],
592 fine_chunks: usize,
593 ) -> Option<SplitCandidate> {
594 let active: Vec<usize> = disjoint
596 .iter()
597 .copied()
598 .zip(coarse.iter())
599 .filter(|(_, amt)| !amt.is_zero())
600 .map(|(idx, _)| idx)
601 .collect();
602 if active.is_empty() {
603 return None;
604 }
605
606 let fine = self.disjoint_waterfill(ctx, &active, fine_chunks, false)?;
608
609 let refined = self.disjoint_exchange(ctx, &active, fine_chunks, fine);
613 self.build_disjoint_legs(ctx, &active, &refined)
614 }
615
616 fn path_net(
622 ctx: &SplitContext,
623 path: &Path<DepthAndPrice>,
624 amount: &BigUint,
625 ) -> Option<BigInt> {
626 if amount.is_zero() {
627 return Some(BigInt::zero());
628 }
629 let empty: HashMap<ComponentId, Box<dyn ProtocolSim>> = HashMap::new();
630 let step = Self::simulate_step(path, ctx.market, &empty, amount.clone())?;
631 let activation = Self::activation_cost(ctx, &step.gas);
632 Some(BigInt::from(step.amount_out) - activation)
633 }
634
635 fn disjoint_exchange(
644 &self,
645 ctx: &SplitContext,
646 active: &[usize],
647 fine_chunks: usize,
648 alloc: Vec<BigUint>,
649 ) -> Vec<BigUint> {
650 let path_count = active.len();
651 if path_count < 2 {
652 return alloc;
653 }
654 let amount_in = ctx.order.amount().clone();
655 let fine_chunks = fine_chunks.max(1);
656 let mut delta = &amount_in / fine_chunks;
657 let min_delta = &amount_in / (fine_chunks * EXCHANGE_DELTA_FLOOR);
658 if delta.is_zero() {
659 return alloc;
660 }
661 let timeout_ms = self.timeout.as_millis() as u64;
662
663 let mut cum = alloc;
664 let mut net_cache: Vec<BigInt> = Vec::with_capacity(path_count);
667 for (i, &path_idx) in active.iter().enumerate() {
668 let Some(net) = Self::path_net(ctx, &ctx.ordered[path_idx], &cum[i]) else {
669 return cum;
671 };
672 net_cache.push(net);
673 }
674
675 let mut sims = 0usize;
676 while delta >= min_delta && !delta.is_zero() {
677 if ctx.start.elapsed().as_millis() as u64 > timeout_ms || sims >= EXCHANGE_MAX_SIMS {
678 break;
679 }
680
681 let mut best: Option<ExchangeMove> = None;
682 for donor in 0..path_count {
683 if sims >= EXCHANGE_MAX_SIMS {
684 break;
685 }
686 if cum[donor] < delta {
687 continue;
688 }
689 let donor_amt = &cum[donor] - δ
690 let Some(donor_net) = Self::path_net(ctx, &ctx.ordered[active[donor]], &donor_amt)
691 else {
692 continue;
693 };
694 sims += 1;
695 for recipient in 0..path_count {
696 if recipient == donor || sims >= EXCHANGE_MAX_SIMS {
697 continue;
698 }
699 let recip_amt = &cum[recipient] + δ
700 let Some(recip_net) =
701 Self::path_net(ctx, &ctx.ordered[active[recipient]], &recip_amt)
702 else {
703 continue;
704 };
705 sims += 1;
706 let before = &net_cache[donor] + &net_cache[recipient];
707 let after = &donor_net + &recip_net;
708 if after <= before {
709 continue;
710 }
711 let gain = after - before;
712 if best
713 .as_ref()
714 .map(|m| gain > m.gain)
715 .unwrap_or(true)
716 {
717 best = Some(ExchangeMove {
718 donor,
719 recipient,
720 donor_net: donor_net.clone(),
721 recip_net,
722 gain,
723 });
724 }
725 }
726 }
727
728 let Some(mv) = best else {
729 delta = &delta / 2usize;
730 continue;
731 };
732 cum[mv.donor] = &cum[mv.donor] - δ
733 cum[mv.recipient] = &cum[mv.recipient] + δ
734 net_cache[mv.donor] = mv.donor_net;
735 net_cache[mv.recipient] = mv.recip_net;
736 }
737 cum
738 }
739
740 fn allocation_commit(
743 path: &Path<DepthAndPrice>,
744 market: &MarketState,
745 overrides: &mut HashMap<ComponentId, Box<dyn ProtocolSim>>,
746 amount: BigUint,
747 flow_fraction: f64,
748 ) -> Option<PathAllocation> {
749 let amount_in = amount;
750 let mut current = amount_in.clone();
751 let mut hops = Vec::with_capacity(path.len());
752
753 for (address_in, edge, address_out) in path.iter() {
754 let token_in = market.get_token(address_in)?;
755 let token_out = market.get_token(address_out)?;
756 let component_id = &edge.component_id;
757 let state = overrides
758 .get(component_id)
759 .map(Box::as_ref)
760 .or_else(|| market.get_simulation_state(component_id))?;
761 let result = state
762 .get_amount_out_guarded(current.clone(), token_in, token_out)
763 .ok()?;
764 hops.push(SimulatedHop {
765 descriptor: HopDescriptor::new(
766 component_id.clone(),
767 token_in.clone(),
768 token_out.clone(),
769 ),
770 amount_out: result.amount.clone(),
771 gas: result.gas.clone(),
772 });
773 overrides.insert(component_id.clone(), result.new_state);
774 current = result.amount;
775 }
776
777 Some(PathAllocation {
778 hops,
779 flow_fraction,
780 amount_in,
781 amount_out: current,
782 marginal_price_product: 0.0,
783 })
784 }
785
786 fn candidate_from_allocations(
790 ctx: &SplitContext,
791 allocations: &[PathAllocation],
792 ) -> Option<SplitCandidate> {
793 let route = build_split_route(allocations, ctx.market, ctx.order).ok()?;
794 let token_out = ctx.order.token_out();
795 let gross = route
796 .swaps()
797 .iter()
798 .filter(|s| s.token_out() == token_out)
799 .fold(BigUint::zero(), |acc, s| acc + s.amount_out());
800 if gross.is_zero() {
801 return None;
802 }
803 let gas = route.total_gas();
804 Some(SplitCandidate { route, gross, gas })
805 }
806
807 fn build_disjoint_legs(
811 &self,
812 ctx: &SplitContext,
813 subset: &[usize],
814 alloc: &[BigUint],
815 ) -> Option<SplitCandidate> {
816 let amount_in = ctx.order.amount().clone();
817 let mut allocations = Vec::new();
818 for (i, &path_idx) in subset.iter().enumerate() {
819 if alloc[i].is_zero() {
820 continue;
821 }
822 let mut overrides: HashMap<ComponentId, Box<dyn ProtocolSim>> = HashMap::new();
825 let allocation = Self::allocation_commit(
826 &ctx.ordered[path_idx],
827 ctx.market,
828 &mut overrides,
829 alloc[i].clone(),
830 ratio(&alloc[i], &amount_in),
831 )?;
832 allocations.push(allocation);
833 }
834 if allocations.is_empty() {
835 return None;
836 }
837 Self::candidate_from_allocations(ctx, &allocations)
838 }
839
840 fn disjoint_waterfill(
844 &self,
845 ctx: &SplitContext,
846 subset: &[usize],
847 num_chunks: usize,
848 gate: bool,
849 ) -> Option<Vec<BigUint>> {
850 let amount_in = ctx.order.amount().clone();
851 let num_chunks = num_chunks.max(1);
852 let base_chunk = &amount_in / num_chunks;
853 if base_chunk.is_zero() {
854 return None;
855 }
856 let remainder = &amount_in - &base_chunk * num_chunks;
857 let timeout_ms = self.timeout.as_millis() as u64;
858 let path_count = subset.len();
859
860 let mut committed: Vec<HashMap<ComponentId, Box<dyn ProtocolSim>>> = (0..path_count)
861 .map(|_| HashMap::new())
862 .collect();
863 let mut cum_in: Vec<BigUint> = vec![BigUint::zero(); path_count];
864 let mut activated: Vec<bool> = vec![!gate; path_count];
865
866 for chunk_idx in 0..num_chunks {
867 if ctx.start.elapsed().as_millis() as u64 > timeout_ms {
868 break;
869 }
870 let chunk = if chunk_idx == 0 { &base_chunk + &remainder } else { base_chunk.clone() };
871
872 let mut best: Option<(usize, BigInt, StepResult)> = None;
873 for (i, &path_idx) in subset.iter().enumerate() {
874 let Some(step) = Self::simulate_step(
875 &ctx.ordered[path_idx],
876 ctx.market,
877 &committed[i],
878 chunk.clone(),
879 ) else {
880 continue;
881 };
882 let gross_marginal = BigInt::from(step.amount_out.clone());
883 let net_marginal = if activated[i] {
884 gross_marginal
885 } else {
886 let activation = Self::activation_cost(ctx, &step.gas);
887 gross_marginal - activation
888 };
889 if best
890 .as_ref()
891 .map(|(_, m, _)| &net_marginal > m)
892 .unwrap_or(true)
893 {
894 best = Some((i, net_marginal, step));
895 }
896 }
897
898 let Some((best_i, _, step)) = best else {
899 break;
900 };
901 for (id, state) in step.new_states {
902 committed[best_i].insert(id, state);
903 }
904 cum_in[best_i] += &chunk;
905 activated[best_i] = true;
906 }
907 Some(cum_in)
908 }
909
910 fn select_shared_candidates(&self, ctx: &SplitContext) -> Vec<usize> {
914 let mut candidates: Vec<usize> = (0..ctx.ordered.len().min(SHARED_FULL_PATHS)).collect();
915 let first_chunk = ctx.order.amount() / COARSE_CHUNKS;
916 if first_chunk.is_zero() {
917 return candidates;
918 }
919 let timeout_ms = self.timeout.as_millis() as u64;
920 let empty: HashMap<ComponentId, Box<dyn ProtocolSim>> = HashMap::new();
921 let mut marginal: Vec<(usize, BigInt)> = Vec::new();
922 for (idx, path) in ctx
923 .ordered
924 .iter()
925 .enumerate()
926 .take(SHARED_MARGIN_PROBE_PATHS)
927 {
928 if ctx.start.elapsed().as_millis() as u64 > timeout_ms {
929 break;
930 }
931 let Some(step) = Self::simulate_step(path, ctx.market, &empty, first_chunk.clone())
932 else {
933 continue;
934 };
935 let activation = Self::activation_cost(ctx, &step.gas);
936 marginal.push((idx, BigInt::from(step.amount_out) - activation));
937 }
938 marginal.sort_by(|(_, a), (_, b)| b.cmp(a));
939 for (idx, net) in marginal
940 .into_iter()
941 .take(SHARED_MARGIN_PATHS)
942 {
943 if net <= BigInt::zero() {
944 continue;
945 }
946 if !candidates.contains(&idx) {
947 candidates.push(idx);
948 }
949 if candidates.len() >= SHARED_MAX_CANDIDATES {
950 break;
951 }
952 }
953 candidates
954 }
955
956 fn fillspill_alloc(&self, ctx: &SplitContext, fine_chunks: usize) -> Option<SplitCandidate> {
958 let candidates = self.select_shared_candidates(ctx);
959 if candidates.len() < 2 {
960 return None;
961 }
962
963 let (coarse_counts, _) = self.fillspill_waterfill(ctx, &candidates, COARSE_CHUNKS, true)?;
965 let active: Vec<usize> = candidates
966 .iter()
967 .copied()
968 .zip(coarse_counts.iter())
969 .filter(|(_, count)| **count > 0)
970 .map(|(idx, _)| idx)
971 .collect();
972 if active.len() < 2 {
973 return None;
974 }
975
976 let (_, schedule) = self.fillspill_waterfill(ctx, &active, fine_chunks, false)?;
978 if schedule.is_empty() {
979 return None;
980 }
981
982 self.build_fillspill_route(ctx, &active, &schedule)
983 }
984
985 #[allow(clippy::type_complexity)]
988 fn fillspill_waterfill(
989 &self,
990 ctx: &SplitContext,
991 subset: &[usize],
992 num_chunks: usize,
993 gate: bool,
994 ) -> Option<(Vec<usize>, Vec<(usize, BigUint)>)> {
995 let amount_in = ctx.order.amount().clone();
996 let num_chunks = num_chunks.max(1);
997 let base_chunk = &amount_in / num_chunks;
998 if base_chunk.is_zero() {
999 return None;
1000 }
1001 let remainder = &amount_in - &base_chunk * num_chunks;
1002 let timeout_ms = self.timeout.as_millis() as u64;
1003
1004 let mut overlay: HashMap<ComponentId, Box<dyn ProtocolSim>> = HashMap::new();
1005 let mut activated: Vec<bool> = vec![!gate; subset.len()];
1006 let mut active_count = if gate { 0 } else { subset.len() };
1007 let mut counts: Vec<usize> = vec![0; subset.len()];
1008 let mut schedule: Vec<(usize, BigUint)> = Vec::with_capacity(num_chunks);
1009
1010 for chunk_idx in 0..num_chunks {
1011 if ctx.start.elapsed().as_millis() as u64 > timeout_ms {
1012 break;
1013 }
1014 let chunk = if chunk_idx == 0 { &base_chunk + &remainder } else { base_chunk.clone() };
1015
1016 let mut best: Option<(usize, BigInt, StepResult)> = None;
1017 for (i, &path_idx) in subset.iter().enumerate() {
1018 if !activated[i] && active_count >= self.max_paths {
1019 continue;
1020 }
1021 let Some(step) = Self::simulate_step(
1022 &ctx.ordered[path_idx],
1023 ctx.market,
1024 &overlay,
1025 chunk.clone(),
1026 ) else {
1027 continue;
1028 };
1029 let gross_marginal = BigInt::from(step.amount_out.clone());
1030 let net_marginal = if activated[i] {
1031 gross_marginal
1032 } else {
1033 let activation = Self::activation_cost(ctx, &step.gas);
1034 gross_marginal - activation
1035 };
1036 if best
1037 .as_ref()
1038 .map(|(_, m, _)| &net_marginal > m)
1039 .unwrap_or(true)
1040 {
1041 best = Some((i, net_marginal, step));
1042 }
1043 }
1044
1045 let Some((best_i, _, step)) = best else {
1046 break;
1047 };
1048 for (id, state) in step.new_states {
1049 overlay.insert(id, state);
1050 }
1051 if !activated[best_i] {
1052 activated[best_i] = true;
1053 active_count += 1;
1054 }
1055 counts[best_i] += 1;
1056 schedule.push((best_i, chunk));
1057 }
1058 Some((counts, schedule))
1059 }
1060
1061 fn build_fillspill_route(
1065 &self,
1066 ctx: &SplitContext,
1067 active: &[usize],
1068 schedule: &[(usize, BigUint)],
1069 ) -> Option<SplitCandidate> {
1070 let amount_in = ctx.order.amount().clone();
1071 let mut cand_in: Vec<BigUint> = vec![BigUint::zero(); active.len()];
1072 for (i, chunk) in schedule {
1073 cand_in[*i] += chunk;
1074 }
1075 let mut execution_order: Vec<usize> = (0..active.len())
1076 .filter(|&i| !cand_in[i].is_zero())
1077 .collect();
1078 if execution_order.len() < 2 {
1079 return None;
1080 }
1081 execution_order.sort_by(|&a, &b| cand_in[b].cmp(&cand_in[a]));
1082
1083 let mut overrides: HashMap<ComponentId, Box<dyn ProtocolSim>> = HashMap::new();
1084 let mut allocations = Vec::new();
1085 for i in execution_order {
1086 let allocation = Self::allocation_commit(
1087 &ctx.ordered[active[i]],
1088 ctx.market,
1089 &mut overrides,
1090 cand_in[i].clone(),
1091 ratio(&cand_in[i], &amount_in),
1092 )?;
1093 allocations.push(allocation);
1094 }
1095 Self::candidate_from_allocations(ctx, &allocations)
1096 }
1097}
1098
1099fn path_key<W>(path: &Path<'_, W>) -> Vec<ComponentId> {
1101 path.edge_iter()
1102 .iter()
1103 .map(|e| e.component_id.clone())
1104 .collect()
1105}
1106
1107fn ratio(numerator: &BigUint, denominator: &BigUint) -> f64 {
1109 use num_traits::ToPrimitive;
1110 let n = numerator.to_f64().unwrap_or(0.0);
1111 let d = denominator.to_f64().unwrap_or(1.0);
1112 if d == 0.0 {
1113 0.0
1114 } else {
1115 n / d
1116 }
1117}
1118
1119type RankedPathScores = Vec<(usize, BigInt)>;
1130type CandidatePathSet<'a, W = ()> = (Vec<Path<'a, W>>, RankedPathScores);
1131
1132#[derive(Clone)]
1133struct CandidatePathState<'a, W> {
1134 node: NodeIndex,
1135 path: Path<'a, W>,
1136 amount_out: BigUint,
1137}
1138
1139struct ScoredEdge<'a, W> {
1140 target: NodeIndex,
1141 edge: &'a EdgeData<W>,
1142 amount_out: BigUint,
1143 priority: u8,
1144}
1145
1146#[derive(Clone, Copy)]
1148struct CandidateSearchConfig<'a> {
1149 min_hops: usize,
1150 max_hops: usize,
1151 max_candidates: usize,
1152 connector_tokens: Option<&'a HashSet<Address>>,
1153 anchor_tokens: &'a HashSet<Address>,
1154 source_token: &'a Address,
1155 start: &'a Instant,
1156 timeout_ms: u64,
1157}
1158
1159fn timed_out(start: &Instant, timeout_ms: u64) -> bool {
1160 start.elapsed().as_millis() as u64 > timeout_ms
1161}
1162
1163fn find_candidate_paths<'a, W>(
1166 graph: &'a StableDiGraph<W>,
1167 market: &MarketDataView<'_>,
1168 order: &Order,
1169 cfg: CandidateSearchConfig<'_>,
1170) -> Result<CandidatePathSet<'a, W>, AlgorithmError>
1171where
1172 W: Clone,
1173{
1174 if cfg.min_hops == 0 || cfg.min_hops > cfg.max_hops {
1175 return Err(AlgorithmError::InvalidConfiguration {
1176 reason: format!(
1177 "invalid hop configuration: min_hops={} max_hops={}",
1178 cfg.min_hops, cfg.max_hops,
1179 ),
1180 });
1181 }
1182 let from_idx =
1183 find_token_node(graph, order.token_in(), NoPathReason::SourceTokenNotInGraph, order)?;
1184 let to_idx =
1185 find_token_node(graph, order.token_out(), NoPathReason::DestinationTokenNotInGraph, order)?;
1186
1187 let mut found = Vec::new();
1188 let mut frontier = vec![CandidatePathState {
1189 node: from_idx,
1190 path: Path::new(),
1191 amount_out: order.amount().clone(),
1192 }];
1193
1194 for _depth in 0..cfg.max_hops {
1195 if timed_out(cfg.start, cfg.timeout_ms) || frontier.is_empty() {
1196 break;
1197 }
1198 let mut next_by_node: HashMap<NodeIndex, Vec<CandidatePathState<'a, W>>> = HashMap::new();
1199 for state in frontier {
1200 if state.node == to_idx && from_idx != to_idx {
1201 continue;
1202 }
1203 expand_candidate_state(
1204 graph,
1205 market,
1206 &cfg,
1207 to_idx,
1208 state,
1209 &mut found,
1210 &mut next_by_node,
1211 );
1212 }
1213 frontier = prune_candidate_frontier(next_by_node);
1214 }
1215
1216 rank_found_candidate_paths(found, cfg.max_candidates, order)
1217}
1218
1219fn find_token_node<W>(
1220 graph: &StableDiGraph<W>,
1221 token: &Address,
1222 reason: NoPathReason,
1223 order: &Order,
1224) -> Result<NodeIndex, AlgorithmError> {
1225 graph
1226 .node_indices()
1227 .find(|&node| &graph[node] == token)
1228 .ok_or(AlgorithmError::NoPath {
1229 from: order.token_in().clone(),
1230 to: order.token_out().clone(),
1231 reason,
1232 })
1233}
1234
1235fn expand_candidate_state<'a, W>(
1236 graph: &'a StableDiGraph<W>,
1237 market: &MarketDataView<'_>,
1238 cfg: &CandidateSearchConfig<'_>,
1239 target: NodeIndex,
1240 state: CandidatePathState<'a, W>,
1241 found: &mut Vec<(Path<'a, W>, BigUint)>,
1242 next_by_node: &mut HashMap<NodeIndex, Vec<CandidatePathState<'a, W>>>,
1243) where
1244 W: Clone,
1245{
1246 let edges = candidate_edges_for_state(graph, market, cfg, target, &state);
1247 for candidate in edges {
1248 if timed_out(cfg.start, cfg.timeout_ms) {
1249 break;
1250 }
1251 let mut path = state.path.clone();
1252 path.add_hop(&graph[state.node], candidate.edge, &graph[candidate.target]);
1253 let path_state = CandidatePathState {
1254 node: candidate.target,
1255 path: path.clone(),
1256 amount_out: candidate.amount_out,
1257 };
1258 if candidate.target == target && path.len() >= cfg.min_hops {
1259 found.push((path.clone(), path_state.amount_out.clone()));
1260 }
1261 if path.len() < cfg.max_hops {
1262 next_by_node
1263 .entry(candidate.target)
1264 .or_default()
1265 .push(path_state);
1266 }
1267 }
1268}
1269
1270fn candidate_edges_for_state<'a, W>(
1271 graph: &'a StableDiGraph<W>,
1272 market: &MarketDataView<'_>,
1273 cfg: &CandidateSearchConfig<'_>,
1274 target: NodeIndex,
1275 state: &CandidatePathState<'a, W>,
1276) -> Vec<ScoredEdge<'a, W>> {
1277 let mut preferred = score_candidate_edges(graph, market, cfg, target, state, true);
1278 if preferred.is_empty() {
1279 preferred = score_candidate_edges(graph, market, cfg, target, state, false);
1280 }
1281 select_candidate_edges(preferred, CANDIDATE_EDGES_PER_STATE)
1282}
1283
1284fn score_candidate_edges<'a, W>(
1285 graph: &'a StableDiGraph<W>,
1286 market: &MarketDataView<'_>,
1287 cfg: &CandidateSearchConfig<'_>,
1288 target: NodeIndex,
1289 state: &CandidatePathState<'a, W>,
1290 preferred_only: bool,
1291) -> Vec<ScoredEdge<'a, W>> {
1292 let mut scored = Vec::new();
1293 for edge in graph.edges(state.node) {
1294 let next_node = edge.target();
1295 if !can_extend_path(graph, state, next_node, target, edge.weight(), cfg) {
1296 continue;
1297 }
1298 let priority = match candidate_priority(graph, next_node, target, cfg) {
1299 Some(priority) => priority,
1300 None if preferred_only => continue,
1301 None => 3,
1302 };
1303 let Some(amount_out) = simulate_edge(
1304 market,
1305 &state.amount_out,
1306 &graph[state.node],
1307 edge.weight(),
1308 &graph[next_node],
1309 ) else {
1310 continue;
1311 };
1312 scored.push(ScoredEdge { target: next_node, edge: edge.weight(), amount_out, priority });
1313 }
1314 scored
1315}
1316
1317fn derive_anchor_tokens<W>(graph: &StableDiGraph<W>) -> HashSet<Address> {
1324 let mut by_degree: Vec<(NodeIndex, usize)> = graph
1325 .node_indices()
1326 .map(|node| (node, graph.edges(node).count()))
1327 .collect();
1328 by_degree.sort_unstable_by_key(|(_, degree)| Reverse(*degree));
1329 let mut anchors: HashSet<Address> = by_degree
1330 .into_iter()
1331 .take(DERIVED_ANCHOR_COUNT)
1332 .map(|(node, _)| graph[node].clone())
1333 .collect();
1334 anchors.insert(Address::from([0u8; 20]));
1335 anchors
1336}
1337
1338fn candidate_priority<W>(
1339 graph: &StableDiGraph<W>,
1340 node: NodeIndex,
1341 target: NodeIndex,
1342 cfg: &CandidateSearchConfig<'_>,
1343) -> Option<u8> {
1344 if node == target {
1345 return Some(0);
1346 }
1347 let token = &graph[node];
1348 match cfg.connector_tokens {
1349 Some(tokens) => tokens.contains(token).then_some(1),
1350 None => cfg
1351 .anchor_tokens
1352 .contains(token)
1353 .then_some(2),
1354 }
1355}
1356
1357fn can_extend_path<W>(
1358 graph: &StableDiGraph<W>,
1359 state: &CandidatePathState<'_, W>,
1360 next_node: NodeIndex,
1361 target: NodeIndex,
1362 edge: &EdgeData<W>,
1363 cfg: &CandidateSearchConfig<'_>,
1364) -> bool {
1365 let next_addr = &graph[next_node];
1366 if state
1367 .path
1368 .edge_iter()
1369 .iter()
1370 .any(|existing| existing.component_id == edge.component_id)
1371 {
1372 return false;
1373 }
1374 if state.path.tokens.contains(&next_addr) {
1375 return false;
1376 }
1377 if next_addr == cfg.source_token {
1378 return false;
1379 }
1380 if next_node == target {
1381 return true;
1382 }
1383 cfg.connector_tokens
1384 .map(|tokens| tokens.contains(next_addr))
1385 .unwrap_or(true)
1386}
1387
1388fn simulate_edge<W>(
1389 market: &MarketDataView<'_>,
1390 amount: &BigUint,
1391 token_in_addr: &Address,
1392 edge: &EdgeData<W>,
1393 token_out_addr: &Address,
1394) -> Option<BigUint> {
1395 let token_in = market.get_token(token_in_addr)?;
1396 let token_out = market.get_token(token_out_addr)?;
1397 let state = market.get_simulation_state(&edge.component_id)?;
1398 state
1399 .get_amount_out_guarded(amount.clone(), token_in, token_out)
1400 .ok()
1401 .map(|result| result.amount)
1402}
1403
1404fn select_candidate_edges<W>(
1405 mut scored: Vec<ScoredEdge<'_, W>>,
1406 max_edges: usize,
1407) -> Vec<ScoredEdge<'_, W>> {
1408 scored.sort_by(compare_scored_edges);
1409 let mut selected = Vec::new();
1410 let mut per_target: HashMap<NodeIndex, usize> = HashMap::new();
1411 for edge in scored {
1412 let limit = if edge.priority == 0 {
1413 CANDIDATE_DIRECT_EDGES_PER_TOKEN
1414 } else {
1415 CANDIDATE_CONNECTOR_EDGES_PER_TOKEN
1416 };
1417 let count = per_target
1418 .entry(edge.target)
1419 .or_default();
1420 if *count >= limit {
1421 continue;
1422 }
1423 *count += 1;
1424 selected.push(edge);
1425 if selected.len() >= max_edges {
1426 break;
1427 }
1428 }
1429 selected
1430}
1431
1432fn compare_scored_edges<W>(a: &ScoredEdge<'_, W>, b: &ScoredEdge<'_, W>) -> Ordering {
1433 a.priority
1434 .cmp(&b.priority)
1435 .then_with(|| b.amount_out.cmp(&a.amount_out))
1436}
1437
1438fn prune_candidate_frontier<W>(
1439 by_node: HashMap<NodeIndex, Vec<CandidatePathState<'_, W>>>,
1440) -> Vec<CandidatePathState<'_, W>> {
1441 by_node
1442 .into_values()
1443 .flat_map(|mut states| {
1444 states.sort_by(|a, b| b.amount_out.cmp(&a.amount_out));
1445 states.truncate(CANDIDATE_STATES_PER_NODE);
1446 states
1447 })
1448 .collect()
1449}
1450
1451fn rank_found_candidate_paths<'a, W>(
1452 mut found: Vec<(Path<'a, W>, BigUint)>,
1453 max_candidates: usize,
1454 order: &Order,
1455) -> Result<CandidatePathSet<'a, W>, AlgorithmError> {
1456 found.sort_by(|(_, a), (_, b)| b.cmp(a));
1457 let mut keys = HashSet::new();
1458 let mut paths = Vec::new();
1459 let mut scores = Vec::new();
1460
1461 for (path, amount_out) in found {
1462 if !keys.insert(path_key(&path)) {
1463 continue;
1464 }
1465 let idx = paths.len();
1466 paths.push(path);
1467 scores.push((idx, BigInt::from(amount_out)));
1468 if paths.len() >= max_candidates {
1469 break;
1470 }
1471 }
1472
1473 if paths.is_empty() {
1474 return Err(AlgorithmError::NoPath {
1475 from: order.token_in().clone(),
1476 to: order.token_out().clone(),
1477 reason: NoPathReason::NoGraphPath,
1478 });
1479 }
1480 Ok((paths, scores))
1481}
1482
1483#[cfg(test)]
1484mod tests {
1485 use std::time::Duration;
1486
1487 use alloy::primitives::U256;
1488 use num_bigint::BigUint;
1489 use num_traits::ToPrimitive;
1490 use tycho_simulation::evm::protocol::uniswap_v2::state::UniswapV2State;
1491
1492 use super::*;
1493 use crate::{
1494 algorithm::{
1495 split_test_harness::{
1496 evaluate_scenario, optimal_two_component_output, split_metrics, split_scenarios,
1497 two_equal_weth_usdc, TWO_EQUAL_USDC_RESERVE, TWO_EQUAL_WETH_RESERVE,
1498 },
1499 test_utils::{
1500 addr, setup_market_unweighted, setup_market_weighted_boxed, token_with_decimals,
1501 ConstantProductSim, DivByZeroSim,
1502 },
1503 },
1504 graph::GraphManager,
1505 types::quote::OrderSide,
1506 };
1507
1508 fn config() -> AlgorithmConfig {
1509 config_ms(2000)
1510 }
1511
1512 fn config_ms(ms: u64) -> AlgorithmConfig {
1513 AlgorithmConfig::new(1, 3, Duration::from_millis(ms), None).unwrap()
1514 }
1515
1516 fn whole_weth_order(token_in: &Address, token_out: &Address, weth: u64) -> Order {
1517 Order::new(
1518 token_in.clone(),
1519 token_out.clone(),
1520 BigUint::from(weth) * BigUint::from(10u64).pow(18),
1521 OrderSide::Sell,
1522 addr(0xFF),
1523 )
1524 }
1525
1526 fn v2_component(reserve_a: u128, reserve_b: u128) -> UniswapV2State {
1527 UniswapV2State::new(
1528 U256::from(reserve_a) * U256::from(10u64).pow(U256::from(18u64)),
1529 U256::from(reserve_b) * U256::from(10u64).pow(U256::from(18u64)),
1530 )
1531 }
1532
1533 fn water_fill_default() -> WaterFillAlgorithm {
1536 WaterFillAlgorithm::with_config(
1537 AlgorithmConfig::new(1, 4, Duration::from_millis(5000), None).unwrap(),
1538 )
1539 .unwrap()
1540 }
1541
1542 #[tokio::test]
1545 async fn test_water_fill_all_scenarios() {
1546 let algo = water_fill_default();
1547 for scenario in split_scenarios::all() {
1548 let name = scenario.name;
1549 let (market, gm) = scenario.build_market_weighted();
1550 let result = evaluate_scenario(&algo, &scenario, market, gm).await;
1551
1552 result.assert_passes_lower_bound();
1553 let tolerance_pct = if name == "DOUBLE_SPLIT" { 10 } else { 5 };
1559 assert!(
1560 result.within_pct_of_optimum(tolerance_pct),
1561 "'{name}': not within {tolerance_pct}% of optimum",
1562 );
1563 let route = result
1564 .route
1565 .as_ref()
1566 .unwrap_or_else(|| panic!("'{name}': expected a route"));
1567 assert!(route.validate().is_ok(), "'{name}': route validation failed");
1568 }
1569 }
1570
1571 #[tokio::test]
1574 async fn test_water_fill_gas_kills_split() {
1575 let scenario = split_scenarios::gas_kills_split();
1576 let (market, gm) = scenario.build_market_weighted();
1577 let result = evaluate_scenario(&water_fill_default(), &scenario, market, gm).await;
1578 assert_eq!(result.path_count, 1, "high gas should prevent splitting");
1579 }
1580
1581 #[tokio::test]
1587 async fn test_split_fills_when_no_single_path_can() {
1588 let a = token_with_decimals(0x01, "A", 18);
1589 let b = token_with_decimals(0x02, "B", 18);
1590 let component = || {
1594 Box::new(ConstantProductSim {
1595 reserve_0: BigUint::from(800u64) * BigUint::from(10u64).pow(18),
1596 reserve_1: BigUint::from(10_000u64) * BigUint::from(10u64).pow(18),
1597 gas: 50_000,
1598 }) as Box<dyn ProtocolSim>
1599 };
1600 let (market, gm) = setup_market_weighted_boxed(vec![
1601 ("component_x", &a, &b, component()),
1602 ("component_y", &a, &b, component()),
1603 ]);
1604 let order = Order::new(
1605 a.address.clone(),
1606 b.address.clone(),
1607 BigUint::from(1_000u64) * BigUint::from(10u64).pow(18),
1608 OrderSide::Sell,
1609 addr(0xFF),
1610 );
1611
1612 let single = MostLiquidAlgorithm::with_config(config())
1614 .unwrap()
1615 .find_best_route(gm.graph(), market.clone(), None, None, &order)
1616 .await;
1617 assert!(single.is_err(), "premise: no single path fills the full order");
1618
1619 let split = WaterFillAlgorithm::with_config(config())
1621 .unwrap()
1622 .find_best_route(gm.graph(), market.clone(), None, None, &order)
1623 .await
1624 .expect("water_fill returns a split when no single path fills");
1625 assert!(split.route().swaps().len() >= 2, "expected a split across both components");
1626 }
1627
1628 #[tokio::test]
1632 async fn test_water_fill_contains_simulation_panic() {
1633 let a = token_with_decimals(0x01, "A", 18);
1634 let b = token_with_decimals(0x02, "B", 18);
1635 let healthy = Box::new(ConstantProductSim {
1636 reserve_0: BigUint::from(10_000u64) * BigUint::from(10u64).pow(18),
1637 reserve_1: BigUint::from(10_000u64) * BigUint::from(10u64).pow(18),
1638 gas: 50_000,
1639 }) as Box<dyn ProtocolSim>;
1640 let (market, gm) = setup_market_weighted_boxed(vec![
1641 ("component_ok", &a, &b, healthy),
1642 ("component_panics", &a, &b, Box::new(DivByZeroSim::default())),
1643 ]);
1644 let order = Order::new(
1645 a.address.clone(),
1646 b.address.clone(),
1647 BigUint::from(100u64) * BigUint::from(10u64).pow(18),
1648 OrderSide::Sell,
1649 addr(0xFF),
1650 );
1651
1652 let result = WaterFillAlgorithm::with_config(config())
1653 .unwrap()
1654 .find_best_route(gm.graph(), market.clone(), None, None, &order)
1655 .await
1656 .expect("panicking component is skipped; the healthy component still fills the order");
1657
1658 assert!(
1659 result
1660 .route()
1661 .swaps()
1662 .iter()
1663 .all(|swap| swap.component_id() == "component_ok"),
1664 "route must only use the healthy component",
1665 );
1666 }
1667
1668 #[tokio::test]
1672 async fn test_water_fill_output_near_two_component_optimum() {
1673 let m = two_equal_weth_usdc(1);
1674 let trade = 500u64;
1675 let order = whole_weth_order(&m.weth, &m.usdc, trade);
1676
1677 let result = WaterFillAlgorithm::with_config(config())
1678 .unwrap()
1679 .find_best_route(
1680 m.weighted.graph(),
1681 m.market.clone(),
1682 None,
1683 Some(m.derived.clone()),
1684 &order,
1685 )
1686 .await
1687 .expect("split solves");
1688 let (_, path_count, gross) = split_metrics(&result, &m.weth, &m.usdc);
1689 assert_eq!(path_count, 2, "the optimum uses both components");
1690
1691 let reserve_in = TWO_EQUAL_WETH_RESERVE as f64 * 1e18;
1693 let reserve_out = TWO_EQUAL_USDC_RESERVE as f64 * 1e6;
1694 let trade_amount = trade as f64 * 1e18;
1695 let (_, optimum) = optimal_two_component_output(
1696 reserve_in,
1697 reserve_out,
1698 reserve_in,
1699 reserve_out,
1700 trade_amount,
1701 );
1702
1703 let gross = gross.to_f64().unwrap();
1704 assert!(
1705 gross >= optimum * 0.999 && gross <= optimum * 1.0001,
1706 "split gross {gross} should be within 0.1% of the two-component optimum {optimum}",
1707 );
1708 }
1709
1710 #[tokio::test]
1717 async fn test_water_fill_no_loss_under_tight_timeout() {
1718 for ms in [1u64, 5, 50] {
1719 let m = two_equal_weth_usdc(1_000_000_000);
1720 let order = whole_weth_order(&m.weth, &m.usdc, 500);
1721
1722 let split = WaterFillAlgorithm::with_config(config_ms(ms))
1723 .unwrap()
1724 .find_best_route(
1725 m.weighted.graph(),
1726 m.market.clone(),
1727 None,
1728 Some(m.derived.clone()),
1729 &order,
1730 )
1731 .await;
1732 let single = MostLiquidAlgorithm::with_config(config_ms(ms))
1733 .unwrap()
1734 .find_best_route(
1735 m.weighted.graph(),
1736 m.market.clone(),
1737 None,
1738 Some(m.derived.clone()),
1739 &order,
1740 )
1741 .await;
1742
1743 let (Ok(split), Ok(single)) = (split, single) else {
1744 continue;
1745 };
1746 let (split_net, _, _) = split_metrics(&split, &m.weth, &m.usdc);
1747 let (single_net, _, _) = split_metrics(&single, &m.weth, &m.usdc);
1748 assert!(
1749 split_net >= single_net,
1750 "split lost to single-path under {ms}ms timeout: split={split_net} single={single_net}",
1751 );
1752 }
1753 }
1754
1755 #[tokio::test]
1761 async fn test_discovery_finds_and_ranks_parallel_components() {
1762 let link = token_with_decimals(0x01, "LINK", 18);
1763 let weth = token_with_decimals(0x02, "WETH", 18);
1764 let (market, graph_manager) = setup_market_unweighted(vec![
1765 (
1766 "a_weak_link_weth",
1767 &link,
1768 &weth,
1769 Box::new(v2_component(2_000_000, 264)) as Box<dyn ProtocolSim>,
1770 ),
1771 (
1772 "z_strong_link_weth",
1773 &link,
1774 &weth,
1775 Box::new(v2_component(2_000_000, 5_700)) as Box<dyn ProtocolSim>,
1776 ),
1777 ]);
1778 let order = Order::new(
1779 link.address.clone(),
1780 weth.address.clone(),
1781 BigUint::from(1_000u64) * BigUint::from(10u64).pow(18),
1782 OrderSide::Sell,
1783 addr(0xFF),
1784 );
1785
1786 let start = Instant::now();
1787 let view = market.read().await;
1788 let (paths, scores) = find_candidate_paths(
1789 graph_manager.graph(),
1790 &view,
1791 &order,
1792 CandidateSearchConfig {
1793 min_hops: 1,
1794 max_hops: 3,
1795 max_candidates: 128,
1796 connector_tokens: None,
1797 anchor_tokens: &HashSet::new(),
1798 source_token: order.token_in(),
1799 start: &start,
1800 timeout_ms: 2000,
1801 },
1802 )
1803 .expect("discovery finds candidates");
1804
1805 assert_eq!(paths.len(), 2, "both parallel components should be discovered");
1806 let best_path = &paths[scores[0].0];
1808 assert_eq!(
1809 best_path.edge_iter()[0].component_id,
1810 "z_strong_link_weth",
1811 "discovery should rank by simulated output, not topology or edge weights",
1812 );
1813 }
1814
1815 #[tokio::test]
1817 async fn test_discovery_rejects_invalid_hop_configuration() {
1818 let link = token_with_decimals(0x01, "LINK", 18);
1819 let weth = token_with_decimals(0x02, "WETH", 18);
1820 let (market, graph_manager) = setup_market_unweighted(vec![(
1821 "link_weth",
1822 &link,
1823 &weth,
1824 Box::new(v2_component(2_000_000, 5_700)) as Box<dyn ProtocolSim>,
1825 )]);
1826 let order = Order::new(
1827 link.address.clone(),
1828 weth.address.clone(),
1829 BigUint::from(1_000u64) * BigUint::from(10u64).pow(18),
1830 OrderSide::Sell,
1831 addr(0xFF),
1832 );
1833
1834 let start = Instant::now();
1835 let view = market.read().await;
1836 let result = find_candidate_paths(
1837 graph_manager.graph(),
1838 &view,
1839 &order,
1840 CandidateSearchConfig {
1841 min_hops: 0,
1842 max_hops: 3,
1843 max_candidates: 128,
1844 connector_tokens: None,
1845 anchor_tokens: &HashSet::new(),
1846 source_token: order.token_in(),
1847 start: &start,
1848 timeout_ms: 2000,
1849 },
1850 );
1851 assert!(
1852 matches!(result, Err(AlgorithmError::InvalidConfiguration { .. })),
1853 "min_hops of 0 should be rejected",
1854 );
1855 }
1856
1857 #[test]
1860 fn test_derive_anchor_tokens_ranks_hub_and_includes_native_sentinel() {
1861 let hub = token_with_decimals(0x01, "HUB", 18);
1862 let a = token_with_decimals(0x02, "A", 18);
1863 let b = token_with_decimals(0x03, "B", 18);
1864 let c = token_with_decimals(0x04, "C", 18);
1865 let (_market, graph_manager) = setup_market_unweighted(vec![
1866 ("hub_a", &hub, &a, Box::new(v2_component(1, 1)) as Box<dyn ProtocolSim>),
1867 ("hub_b", &hub, &b, Box::new(v2_component(1, 1)) as Box<dyn ProtocolSim>),
1868 ("hub_c", &hub, &c, Box::new(v2_component(1, 1)) as Box<dyn ProtocolSim>),
1869 ]);
1870
1871 let anchors = derive_anchor_tokens(graph_manager.graph());
1872
1873 assert!(anchors.contains(&hub.address), "highest-degree token should be anchored");
1874 assert!(
1875 anchors.contains(&Address::from([0u8; 20])),
1876 "native-ETH sentinel should always be anchored",
1877 );
1878 }
1879}