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