1use std::{collections::HashSet, str::FromStr, sync::Arc, time::Duration};
14
15use num_cpus;
16use serde::{Deserialize, Serialize};
17use tokio::{sync::broadcast, task::JoinHandle};
18use tycho_execution::encoding::evm::swap_encoder::swap_encoder_registry::SwapEncoderRegistry;
19#[cfg(feature = "experimental")]
20use tycho_simulation::evm::stream::BlockStepController;
21#[cfg(feature = "test-utils")]
22use tycho_simulation::tycho_ethereum::gas::{BlockGasPrice, GasPrice};
23use tycho_simulation::{
24 evm::pending::PendingBlockProcessor,
25 tycho_common::{models::Chain, traits::TxDeltaIndexer, Bytes},
26 tycho_core::models::Address,
27 tycho_ethereum::rpc::EthereumRpcClient,
28};
29
30use crate::{
31 algorithm::{AlgorithmConfig, AlgorithmError},
32 derived::{ComputationManager, ComputationManagerConfig, SharedDerivedDataRef},
33 encoding::{encoder::Encoder, fee_fetcher::RouterFeeFetcher, router_fees::SharedRouterFees},
34 feed::{
35 events::{MarketEvent, MarketEventHandler},
36 gas::GasPriceFetcher,
37 market_data::MarketData,
38 metrics_sampler::MetricsSampler,
39 tycho_feed::TychoFeed,
40 TychoFeedConfig,
41 },
42 graph::EdgeWeightUpdaterWithDerived,
43 price_guard::{
44 guard::PriceGuard, provider::PriceProvider, provider_registry::PriceProviderRegistry,
45 },
46 types::constants::native_token,
47 worker_pool::{
48 pool::{WorkerPool, WorkerPoolBuilder},
49 registry::UnknownAlgorithmError,
50 },
51 worker_pool_router::{
52 config::WorkerPoolRouterConfig, LiquidityScope, SolverPoolHandle, WorkerPoolRouter,
53 },
54 Algorithm, Quote, QuoteRequest, SolveError,
55};
56
57pub mod defaults {
63 use std::time::Duration;
64
65 pub const MIN_TOKEN_QUALITY: i32 = 100;
67 pub const TRADED_N_DAYS_AGO: u64 = 3;
69 pub const TVL_BUFFER_RATIO: f64 = 1.1;
72 pub const GAS_REFRESH_INTERVAL: Duration = Duration::from_secs(30);
74 pub const METRICS_SAMPLE_INTERVAL: Duration = Duration::from_secs(10);
76 pub const ROUTER_FEE_REFRESH_INTERVAL: Duration = Duration::from_secs(300);
78 pub const RECONNECT_DELAY: Duration = Duration::from_secs(5);
80 pub const ROUTER_MIN_RESPONSES: usize = 0;
83 pub const POOL_TASK_QUEUE_CAPACITY: usize = 1000;
85 pub const POOL_MIN_HOPS: usize = 1;
87 pub const POOL_MAX_HOPS: usize = 3;
89 pub const POOL_TIMEOUT_MS: u64 = 100;
91}
92
93const DEFAULT_TYCHO_USE_TLS: bool = true;
95const DEFAULT_DEPTH_SLIPPAGE_THRESHOLD: f64 = 0.01;
96const DEFAULT_ROUTER_TIMEOUT: Duration = Duration::from_secs(10);
99
100fn default_task_queue_capacity() -> usize {
103 defaults::POOL_TASK_QUEUE_CAPACITY
104}
105
106fn default_min_hops() -> usize {
107 defaults::POOL_MIN_HOPS
108}
109
110fn default_max_hops() -> usize {
111 defaults::POOL_MAX_HOPS
112}
113
114fn default_algo_timeout_ms() -> u64 {
115 defaults::POOL_TIMEOUT_MS
116}
117
118fn parse_connector_tokens(
119 raw: Option<&[String]>,
120) -> Result<Option<HashSet<Address>>, SolverBuildError> {
121 let Some(strings) = raw else {
122 return Ok(None);
123 };
124 let mut set = HashSet::with_capacity(strings.len());
125 for s in strings {
126 let addr = Address::from_str(s).map_err(|e| AlgorithmError::InvalidConfiguration {
127 reason: format!("connector_tokens: invalid address {s:?}: {e}"),
128 })?;
129 set.insert(addr);
130 }
131 Ok(Some(set))
132}
133
134#[must_use]
136#[derive(Debug, Clone, Serialize, Deserialize)]
137pub struct PoolConfig {
138 algorithm: String,
140 #[serde(default = "num_cpus::get")]
142 num_workers: usize,
143 #[serde(default = "default_task_queue_capacity")]
145 task_queue_capacity: usize,
146 #[serde(default = "default_min_hops")]
148 min_hops: usize,
149 #[serde(default = "default_max_hops")]
151 max_hops: usize,
152 #[serde(default = "default_algo_timeout_ms")]
154 timeout_ms: u64,
155 #[serde(default)]
157 max_routes: Option<usize>,
158 #[serde(default)]
161 connector_tokens: Option<Vec<String>>,
162 #[serde(default)]
164 liquidity_scope: Option<LiquidityScope>,
165}
166
167impl PoolConfig {
168 pub fn new(algorithm: impl Into<String>) -> Self {
171 Self {
172 algorithm: algorithm.into(),
173 num_workers: num_cpus::get(),
174 task_queue_capacity: defaults::POOL_TASK_QUEUE_CAPACITY,
175 min_hops: defaults::POOL_MIN_HOPS,
176 max_hops: defaults::POOL_MAX_HOPS,
177 timeout_ms: defaults::POOL_TIMEOUT_MS,
178 max_routes: None,
179 connector_tokens: None,
180 liquidity_scope: None,
181 }
182 }
183
184 pub fn algorithm(&self) -> &str {
186 &self.algorithm
187 }
188
189 pub fn liquidity_scope(&self) -> Option<LiquidityScope> {
191 self.liquidity_scope
192 }
193
194 pub fn with_liquidity_scope(mut self, scope: LiquidityScope) -> Self {
196 self.liquidity_scope = Some(scope);
197 self
198 }
199
200 pub fn num_workers(&self) -> usize {
202 self.num_workers
203 }
204
205 pub fn with_num_workers(mut self, num_workers: usize) -> Self {
207 self.num_workers = num_workers;
208 self
209 }
210
211 pub fn with_task_queue_capacity(mut self, task_queue_capacity: usize) -> Self {
213 self.task_queue_capacity = task_queue_capacity;
214 self
215 }
216
217 pub fn with_min_hops(mut self, min_hops: usize) -> Self {
219 self.min_hops = min_hops;
220 self
221 }
222
223 pub fn with_max_hops(mut self, max_hops: usize) -> Self {
225 self.max_hops = max_hops;
226 self
227 }
228
229 pub fn with_timeout_ms(mut self, timeout_ms: u64) -> Self {
231 self.timeout_ms = timeout_ms;
232 self
233 }
234
235 pub fn with_max_routes(mut self, max_routes: Option<usize>) -> Self {
237 self.max_routes = max_routes;
238 self
239 }
240
241 pub fn task_queue_capacity(&self) -> usize {
243 self.task_queue_capacity
244 }
245
246 pub fn min_hops(&self) -> usize {
248 self.min_hops
249 }
250
251 pub fn max_hops(&self) -> usize {
253 self.max_hops
254 }
255
256 pub fn timeout_ms(&self) -> u64 {
258 self.timeout_ms
259 }
260
261 pub fn max_routes(&self) -> Option<usize> {
263 self.max_routes
264 }
265
266 pub fn with_connector_tokens(mut self, tokens: Vec<String>) -> Self {
269 self.connector_tokens = Some(tokens);
270 self
271 }
272
273 pub fn connector_tokens(&self) -> Option<&[String]> {
275 self.connector_tokens.as_deref()
276 }
277}
278
279#[derive(Debug, thiserror::Error)]
281#[error("timed out after {timeout_ms}ms waiting for market data and derived computations")]
282pub struct WaitReadyError {
283 timeout_ms: u64,
284}
285
286#[non_exhaustive]
288#[derive(Debug, thiserror::Error)]
289pub enum SolverBuildError {
290 #[error("failed to create ethereum RPC client: {0}")]
292 RpcClient(String),
293 #[error(transparent)]
295 AlgorithmConfig(#[from] AlgorithmError),
296 #[error("failed to create computation manager: {0}")]
298 ComputationManager(String),
299 #[error("failed to create encoder: {0}")]
301 Encoder(String),
302 #[error("failed to create router fee fetcher: {0}")]
304 RouterFeeFetcher(String),
305 #[error(transparent)]
307 UnknownAlgorithm(#[from] UnknownAlgorithmError),
308 #[error("gas token not configured for chain")]
310 GasToken,
311 #[error("no worker pools configured")]
313 NoPools,
314 #[cfg(feature = "test-utils")]
316 #[error("replay failed: {0}")]
317 Replay(String),
318 #[error("feed setup failed before delivering pending processor: {0}")]
323 FeedSetup(String),
324 #[error("pending processor channel closed before processor was delivered")]
327 PendingChannelClosed,
328 #[cfg(feature = "experimental")]
331 #[error("step controller channel closed before controller was delivered")]
332 StepControllerChannelClosed,
333}
334
335enum PoolEntry {
337 BuiltIn {
338 name: String,
339 algorithm: String,
340 num_workers: usize,
341 task_queue_capacity: usize,
342 min_hops: usize,
343 max_hops: usize,
344 timeout_ms: u64,
345 max_routes: Option<usize>,
346 connector_tokens: Option<HashSet<Address>>,
347 liquidity_scope: Option<LiquidityScope>,
348 },
349 Custom(CustomPoolEntry),
350}
351
352impl PoolEntry {
353 fn liquidity_scope(&self) -> Option<LiquidityScope> {
355 match self {
356 PoolEntry::BuiltIn { liquidity_scope, .. } => *liquidity_scope,
357 PoolEntry::Custom(custom) => custom.liquidity_scope,
358 }
359 }
360}
361
362struct CustomPoolEntry {
364 name: String,
365 num_workers: usize,
366 task_queue_capacity: usize,
367 min_hops: usize,
368 max_hops: usize,
369 timeout_ms: u64,
370 max_routes: Option<usize>,
371 liquidity_scope: Option<LiquidityScope>,
372 configure: Box<dyn FnOnce(WorkerPoolBuilder) -> WorkerPoolBuilder + Send>,
374}
375
376struct BuiltComponents {
379 tycho_feed: TychoFeed,
380 gas_price_fetcher: GasPriceFetcher<EthereumRpcClient>,
381 router_fee_fetcher: Option<RouterFeeFetcher>,
382 computation_manager: ComputationManager,
383 computation_event_rx: broadcast::Receiver<MarketEvent>,
384 computation_shutdown_tx: broadcast::Sender<()>,
385 computation_shutdown_rx: broadcast::Receiver<()>,
386 router: WorkerPoolRouter,
387 worker_pools: Vec<WorkerPool>,
388 market_data: MarketData,
389 derived_data: SharedDerivedDataRef,
390 router_fees: SharedRouterFees,
391 chain: Chain,
392 router_address: Option<Bytes>,
393 pending_indexers: Vec<(String, Box<dyn TxDeltaIndexer>)>,
394 market_event_tx: broadcast::Sender<MarketEvent>,
395}
396
397#[must_use = "a builder does nothing until .build() is called"]
402pub struct FyndBuilder {
403 chain: Chain,
404 tycho_url: String,
405 rpc_url: String,
406 protocols: Vec<String>,
407 min_tvl: f64,
408 tycho_api_key: Option<String>,
409 tycho_use_tls: bool,
410 min_token_quality: i32,
411 traded_n_days_ago: u64,
412 tvl_buffer_ratio: f64,
413 gas_refresh_interval: Duration,
414 reconnect_delay: Duration,
415 blocklisted_components: HashSet<String>,
416 partial_blocks: bool,
417 router_timeout: Duration,
418 router_min_responses: usize,
419 encoder: Option<Encoder>,
420 pools: Vec<PoolEntry>,
421 price_guard_enabled: bool,
422 price_providers: Vec<Box<dyn PriceProvider>>,
423 pending_indexers: Vec<(String, Box<dyn TxDeltaIndexer>)>,
424}
425
426impl FyndBuilder {
427 pub fn new(
429 chain: Chain,
430 tycho_url: impl Into<String>,
431 rpc_url: impl Into<String>,
432 protocols: Vec<String>,
433 min_tvl: f64,
434 ) -> Self {
435 Self {
436 chain,
437 tycho_url: tycho_url.into(),
438 rpc_url: rpc_url.into(),
439 protocols,
440 min_tvl,
441 tycho_api_key: None,
442 tycho_use_tls: DEFAULT_TYCHO_USE_TLS,
443 min_token_quality: defaults::MIN_TOKEN_QUALITY,
444 traded_n_days_ago: defaults::TRADED_N_DAYS_AGO,
445 tvl_buffer_ratio: defaults::TVL_BUFFER_RATIO,
446 gas_refresh_interval: defaults::GAS_REFRESH_INTERVAL,
447 reconnect_delay: defaults::RECONNECT_DELAY,
448 blocklisted_components: HashSet::new(),
449 partial_blocks: false,
450 router_timeout: DEFAULT_ROUTER_TIMEOUT,
451 router_min_responses: defaults::ROUTER_MIN_RESPONSES,
452 encoder: None,
453 pools: Vec::new(),
454 price_guard_enabled: false,
455 price_providers: Vec::new(),
456 pending_indexers: Vec::new(),
457 }
458 }
459
460 pub fn chain(&self) -> Chain {
462 self.chain
463 }
464
465 pub fn tycho_api_key(mut self, key: impl Into<String>) -> Self {
467 self.tycho_api_key = Some(key.into());
468 self
469 }
470
471 pub fn min_tvl(mut self, min_tvl: f64) -> Self {
473 self.min_tvl = min_tvl;
474 self
475 }
476
477 pub fn tycho_use_tls(mut self, use_tls: bool) -> Self {
479 self.tycho_use_tls = use_tls;
480 self
481 }
482
483 pub fn min_token_quality(mut self, quality: i32) -> Self {
486 self.min_token_quality = quality;
487 self
488 }
489
490 pub fn traded_n_days_ago(mut self, days: u64) -> Self {
492 self.traded_n_days_ago = days;
493 self
494 }
495
496 pub fn tvl_buffer_ratio(mut self, ratio: f64) -> Self {
498 self.tvl_buffer_ratio = ratio;
499 self
500 }
501
502 pub fn gas_refresh_interval(mut self, interval: Duration) -> Self {
504 self.gas_refresh_interval = interval;
505 self
506 }
507
508 pub fn reconnect_delay(mut self, delay: Duration) -> Self {
510 self.reconnect_delay = delay;
511 self
512 }
513
514 pub fn blocklisted_components(mut self, components: HashSet<String>) -> Self {
516 self.blocklisted_components = components;
517 self
518 }
519
520 pub fn partial_blocks(mut self, enabled: bool) -> Self {
526 self.partial_blocks = enabled;
527 self
528 }
529
530 pub fn worker_router_timeout(mut self, timeout: Duration) -> Self {
532 self.router_timeout = timeout;
533 self
534 }
535
536 pub fn worker_router_min_responses(mut self, min: usize) -> Self {
538 self.router_min_responses = min;
539 self
540 }
541
542 pub fn encoder(mut self, encoder: Encoder) -> Self {
544 self.encoder = Some(encoder);
545 self
546 }
547
548 pub fn algorithm(mut self, algorithm: impl Into<String>) -> Self {
550 self.pools.push(PoolEntry::BuiltIn {
551 name: "default".to_string(),
552 algorithm: algorithm.into(),
553 num_workers: num_cpus::get(),
554 task_queue_capacity: defaults::POOL_TASK_QUEUE_CAPACITY,
555 min_hops: defaults::POOL_MIN_HOPS,
556 max_hops: defaults::POOL_MAX_HOPS,
557 timeout_ms: defaults::POOL_TIMEOUT_MS,
558 max_routes: None,
559 connector_tokens: None,
560 liquidity_scope: None,
561 });
562 self
563 }
564
565 pub fn with_algorithm<A, F>(mut self, name: impl Into<String>, factory: F) -> Self
569 where
570 A: Algorithm + 'static,
571 A::GraphManager: MarketEventHandler + EdgeWeightUpdaterWithDerived + 'static,
572 F: Fn(AlgorithmConfig) -> A + Clone + Send + Sync + 'static,
573 {
574 let name = name.into();
575 let algo_name = name.clone();
576 let configure =
577 Box::new(move |builder: WorkerPoolBuilder| builder.with_algorithm(algo_name, factory));
578 self.pools
579 .push(PoolEntry::Custom(CustomPoolEntry {
580 name,
581 num_workers: num_cpus::get(),
582 task_queue_capacity: defaults::POOL_TASK_QUEUE_CAPACITY,
583 min_hops: defaults::POOL_MIN_HOPS,
584 max_hops: defaults::POOL_MAX_HOPS,
585 timeout_ms: defaults::POOL_TIMEOUT_MS,
586 max_routes: None,
587 liquidity_scope: None,
588 configure,
589 }));
590 self
591 }
592
593 pub fn add_default_price_providers(self) -> Self {
600 self.register_price_provider(Box::new(
601 crate::price_guard::hyperliquid::HyperliquidProvider::default(),
602 ))
603 .register_price_provider(Box::new(
604 crate::price_guard::binance_ws::BinanceWsProvider::default(),
605 ))
606 }
607
608 pub fn register_price_provider(mut self, provider: Box<dyn PriceProvider>) -> Self {
613 self.price_providers.push(provider);
614 self
615 }
616
617 pub fn with_pending_indexer(
623 mut self,
624 extractor: impl Into<String>,
625 indexer: Box<dyn TxDeltaIndexer>,
626 ) -> Self {
627 self.pending_indexers
628 .push((extractor.into(), indexer));
629 self
630 }
631
632 pub fn price_guard_enabled(mut self, enabled: bool) -> Self {
639 self.price_guard_enabled = enabled;
640 self
641 }
642
643 pub fn add_pool(
650 mut self,
651 name: impl Into<String>,
652 config: &PoolConfig,
653 ) -> Result<Self, SolverBuildError> {
654 let connector_tokens = parse_connector_tokens(config.connector_tokens())?;
655 self.pools.push(PoolEntry::BuiltIn {
656 name: name.into(),
657 algorithm: config.algorithm().to_string(),
658 num_workers: config.num_workers(),
659 task_queue_capacity: config.task_queue_capacity(),
660 min_hops: config.min_hops(),
661 max_hops: config.max_hops(),
662 timeout_ms: config.timeout_ms(),
663 max_routes: config.max_routes(),
664 connector_tokens,
665 liquidity_scope: config.liquidity_scope(),
666 });
667 Ok(self)
668 }
669
670 fn assemble_components(mut self) -> Result<BuiltComponents, SolverBuildError> {
673 if self.pools.is_empty() {
674 return Err(SolverBuildError::NoPools);
675 }
676
677 if self.price_providers.is_empty() {
679 self = self.add_default_price_providers();
680 }
681
682 let market_data = MarketData::new_shared();
683
684 let tycho_feed_config = TychoFeedConfig::new(
685 self.tycho_url,
686 self.chain,
687 self.tycho_api_key,
688 self.tycho_use_tls,
689 self.protocols,
690 self.min_tvl,
691 )
692 .tvl_buffer_ratio(self.tvl_buffer_ratio)
693 .reconnect_delay(self.reconnect_delay)
694 .min_token_quality(self.min_token_quality)
695 .traded_n_days_ago(self.traded_n_days_ago)
696 .blocklisted_components(self.blocklisted_components)
697 .partial_blocks(self.partial_blocks);
698
699 let ethereum_client = EthereumRpcClient::new(self.rpc_url.as_str())
700 .map_err(|e| SolverBuildError::RpcClient(e.to_string()))?;
701
702 let gas_price_fetcher =
703 GasPriceFetcher::new(ethereum_client, market_data.clone(), self.gas_refresh_interval);
704
705 let tycho_feed = TychoFeed::new(tycho_feed_config, market_data.clone());
706 let market_event_tx = tycho_feed.event_sender();
707
708 let gas_token = native_token(&self.chain).map_err(|_| SolverBuildError::GasToken)?;
709 let computation_config = ComputationManagerConfig::new()
710 .with_gas_token(gas_token)
711 .with_depth_slippage_threshold(DEFAULT_DEPTH_SLIPPAGE_THRESHOLD);
712 let (computation_manager, _) =
715 ComputationManager::new(computation_config, market_data.clone())
716 .map_err(|e| SolverBuildError::ComputationManager(e.to_string()))?;
717
718 let derived_data: SharedDerivedDataRef = computation_manager.store();
719 let derived_event_tx = computation_manager.event_sender();
720
721 let computation_event_rx = tycho_feed.subscribe();
724 let (computation_shutdown_tx, computation_shutdown_rx) = broadcast::channel(1);
725
726 let mut solver_pool_handles: Vec<SolverPoolHandle> = Vec::new();
727 let mut worker_pools: Vec<WorkerPool> = Vec::new();
728
729 let pools = std::mem::take(&mut self.pools);
730
731 for pool_entry in pools {
732 let pool_event_rx = tycho_feed.subscribe();
733 let derived_rx = derived_event_tx.subscribe();
734
735 let pool_scope = pool_entry
736 .liquidity_scope()
737 .unwrap_or_default();
738
739 let (worker_pool, task_handle) = match pool_entry {
740 PoolEntry::BuiltIn {
741 name,
742 algorithm,
743 num_workers,
744 task_queue_capacity,
745 min_hops,
746 max_hops,
747 timeout_ms,
748 max_routes,
749 connector_tokens,
750 liquidity_scope: _,
751 } => {
752 let mut algo_cfg = AlgorithmConfig::new(
753 min_hops,
754 max_hops,
755 Duration::from_millis(timeout_ms),
756 max_routes,
757 )?;
758 if let Some(tokens) = connector_tokens {
759 algo_cfg = algo_cfg.with_connector_tokens(tokens);
760 }
761 let builder = WorkerPoolBuilder::new()
762 .name(name)
763 .algorithm(algorithm)
764 .algorithm_config(algo_cfg)
765 .num_workers(num_workers)
766 .task_queue_capacity(task_queue_capacity)
767 .liquidity_scope(pool_scope);
768 builder.build(
769 market_data.clone(),
770 Arc::clone(&derived_data),
771 pool_event_rx,
772 derived_rx,
773 )?
774 }
775 PoolEntry::Custom(custom) => {
776 let algo_cfg = AlgorithmConfig::new(
777 custom.min_hops,
778 custom.max_hops,
779 Duration::from_millis(custom.timeout_ms),
780 custom.max_routes,
781 )?;
782 let builder = WorkerPoolBuilder::new()
783 .name(custom.name)
784 .algorithm_config(algo_cfg)
785 .num_workers(custom.num_workers)
786 .task_queue_capacity(custom.task_queue_capacity)
787 .liquidity_scope(pool_scope);
788 let builder = (custom.configure)(builder);
789 builder.build(
790 market_data.clone(),
791 Arc::clone(&derived_data),
792 pool_event_rx,
793 derived_rx,
794 )?
795 }
796 };
797
798 solver_pool_handles.push(
799 SolverPoolHandle::new(worker_pool.name(), task_handle)
800 .with_liquidity_scope(pool_scope),
801 );
802 worker_pools.push(worker_pool);
803 }
804
805 let encoder = match self.encoder {
806 Some(enc) => enc,
807 None => {
808 let registry = SwapEncoderRegistry::new(self.chain)
809 .add_default_encoders(None)
810 .map_err(|e| SolverBuildError::Encoder(e.to_string()))?;
811 Encoder::new(self.chain, registry)
812 .map_err(|e| SolverBuildError::Encoder(e.to_string()))?
813 }
814 };
815
816 let chain = self.chain;
817 let router_address = encoder.router_address().cloned();
818 let router_fees = encoder.router_fees();
819
820 let router_fee_fetcher = match &router_address {
821 Some(addr) => Some(
822 RouterFeeFetcher::new(
823 self.rpc_url.as_str(),
824 addr,
825 router_fees.clone(),
826 defaults::ROUTER_FEE_REFRESH_INTERVAL,
827 )
828 .map_err(|e| SolverBuildError::RouterFeeFetcher(e.to_string()))?,
829 ),
830 None => {
831 tracing::warn!(
832 %chain,
833 "no Tycho router for this chain; running quote-only (encoding disabled)"
834 );
835 None
836 }
837 };
838
839 let router_config = WorkerPoolRouterConfig::default()
842 .with_timeout(self.router_timeout)
843 .with_min_responses(self.router_min_responses);
844 let mut router = WorkerPoolRouter::new(solver_pool_handles, router_config, encoder);
845
846 if self.price_guard_enabled {
847 let mut registry = PriceProviderRegistry::new();
848 let mut worker_handles = Vec::new();
849 for mut provider in self.price_providers {
850 worker_handles.push(provider.start(market_data.clone()));
851 registry = registry.register(provider);
852 }
853 let price_guard = PriceGuard::new(registry, worker_handles);
854 router = router.with_price_guard(price_guard);
855 }
856
857 Ok(BuiltComponents {
858 tycho_feed,
859 gas_price_fetcher,
860 router_fee_fetcher,
861 computation_manager,
862 computation_event_rx,
863 computation_shutdown_tx,
864 computation_shutdown_rx,
865 router,
866 worker_pools,
867 market_data,
868 derived_data,
869 router_fees,
870 chain,
871 router_address,
872 pending_indexers: self.pending_indexers,
873 market_event_tx,
874 })
875 }
876
877 pub fn build(self) -> Result<Solver, SolverBuildError> {
883 let mut c = self.assemble_components()?;
884
885 let feed_handle = tokio::spawn(async move {
886 if let Err(e) = c.tycho_feed.run().await {
887 metrics::counter!("tycho_feed_failures_total").increment(1);
888 tracing::error!(error = %e, "tycho feed error");
889 }
890 });
891 let gas_price_handle = tokio::spawn(async move {
892 c.gas_price_fetcher.run().await;
893 });
894 let metrics_sampler =
895 MetricsSampler::new(c.market_data.clone(), defaults::METRICS_SAMPLE_INTERVAL);
896 let metrics_sampler_handle = tokio::spawn(async move { metrics_sampler.run().await });
897 let router_fee_handle = match c.router_fee_fetcher {
898 Some(fetcher) => tokio::spawn(async move { fetcher.run().await }),
899 None => tokio::spawn(async {}),
900 };
901 let computation_handle = tokio::spawn(async move {
902 c.computation_manager
903 .run(c.computation_event_rx, c.computation_shutdown_rx)
904 .await;
905 });
906
907 Ok(Solver {
908 router: c.router,
909 worker_pools: c.worker_pools,
910 market_data: c.market_data,
911 derived_data: c.derived_data,
912 router_fees: c.router_fees,
913 feed_handle,
914 gas_price_handle,
915 metrics_sampler_handle,
916 router_fee_handle,
917 computation_handle,
918 computation_shutdown_tx: c.computation_shutdown_tx,
919 chain: c.chain,
920 router_address: c.router_address,
921 market_event_tx: c.market_event_tx,
922 })
923 }
924
925 pub async fn build_with_pending(
937 self,
938 ) -> Result<(Solver, PendingBlockProcessor), SolverBuildError> {
939 let mut c = self.assemble_components()?;
940
941 let (pending_tx, pending_rx) =
942 tokio::sync::oneshot::channel::<Result<PendingBlockProcessor, String>>();
943
944 let pending_indexers = c.pending_indexers;
945 let feed_handle = tokio::spawn(async move {
946 if let Err(e) = c
947 .tycho_feed
948 .run_with_pending(pending_tx, pending_indexers)
949 .await
950 {
951 metrics::counter!("tycho_feed_failures_total").increment(1);
952 tracing::error!(error = %e, "tycho feed error");
953 }
954 });
955 let gas_price_handle = tokio::spawn(async move {
956 c.gas_price_fetcher.run().await;
957 });
958 let metrics_sampler =
959 MetricsSampler::new(c.market_data.clone(), defaults::METRICS_SAMPLE_INTERVAL);
960 let metrics_sampler_handle = tokio::spawn(async move { metrics_sampler.run().await });
961 let router_fee_handle = match c.router_fee_fetcher {
962 Some(fetcher) => tokio::spawn(async move { fetcher.run().await }),
963 None => tokio::spawn(async {}),
964 };
965 let computation_handle = tokio::spawn(async move {
966 c.computation_manager
967 .run(c.computation_event_rx, c.computation_shutdown_rx)
968 .await;
969 });
970
971 let pending = pending_rx
972 .await
973 .map_err(|_| SolverBuildError::PendingChannelClosed)?
974 .map_err(SolverBuildError::FeedSetup)?;
975
976 Ok((
977 Solver {
978 router: c.router,
979 worker_pools: c.worker_pools,
980 market_data: c.market_data,
981 derived_data: c.derived_data,
982 router_fees: c.router_fees,
983 feed_handle,
984 gas_price_handle,
985 metrics_sampler_handle,
986 router_fee_handle,
987 computation_handle,
988 computation_shutdown_tx: c.computation_shutdown_tx,
989 chain: c.chain,
990 router_address: c.router_address,
991 market_event_tx: c.market_event_tx,
992 },
993 pending,
994 ))
995 }
996
997 #[cfg(feature = "experimental")]
1012 pub async fn build_with_step_controller(
1013 self,
1014 ) -> Result<(Solver, BlockStepController), SolverBuildError> {
1015 let mut c = self.assemble_components()?;
1016
1017 let (controller_tx, controller_rx) =
1018 tokio::sync::oneshot::channel::<Result<BlockStepController, String>>();
1019
1020 let feed_handle = tokio::spawn(async move {
1021 if let Err(e) = c
1022 .tycho_feed
1023 .run_with_step_controller(controller_tx)
1024 .await
1025 {
1026 tracing::error!(error = %e, "tycho feed error");
1027 }
1028 });
1029 let gas_price_handle = tokio::spawn(async move {
1030 c.gas_price_fetcher.run().await;
1031 });
1032 let metrics_sampler =
1033 MetricsSampler::new(c.market_data.clone(), defaults::METRICS_SAMPLE_INTERVAL);
1034 let metrics_sampler_handle = tokio::spawn(async move { metrics_sampler.run().await });
1035 let router_fee_handle = match c.router_fee_fetcher {
1036 Some(fetcher) => tokio::spawn(async move { fetcher.run().await }),
1037 None => tokio::spawn(async {}),
1038 };
1039 let computation_handle = tokio::spawn(async move {
1040 c.computation_manager
1041 .run(c.computation_event_rx, c.computation_shutdown_rx)
1042 .await;
1043 });
1044
1045 let controller = controller_rx
1046 .await
1047 .map_err(|_| SolverBuildError::StepControllerChannelClosed)?
1048 .map_err(SolverBuildError::FeedSetup)?;
1049
1050 Ok((
1051 Solver {
1052 router: c.router,
1053 worker_pools: c.worker_pools,
1054 market_data: c.market_data,
1055 derived_data: c.derived_data,
1056 router_fees: c.router_fees,
1057 feed_handle,
1058 gas_price_handle,
1059 metrics_sampler_handle,
1060 router_fee_handle,
1061 computation_handle,
1062 computation_shutdown_tx: c.computation_shutdown_tx,
1063 chain: c.chain,
1064 router_address: c.router_address,
1065 market_event_tx: c.market_event_tx,
1066 },
1067 controller,
1068 ))
1069 }
1070} pub struct Solver {
1074 router: WorkerPoolRouter,
1075 worker_pools: Vec<WorkerPool>,
1076 market_data: MarketData,
1077 derived_data: SharedDerivedDataRef,
1078 router_fees: SharedRouterFees,
1079 feed_handle: JoinHandle<()>,
1080 gas_price_handle: JoinHandle<()>,
1081 metrics_sampler_handle: JoinHandle<()>,
1082 router_fee_handle: JoinHandle<()>,
1083 computation_handle: JoinHandle<()>,
1084 computation_shutdown_tx: broadcast::Sender<()>,
1085 chain: Chain,
1086 router_address: Option<Bytes>,
1087 market_event_tx: broadcast::Sender<MarketEvent>,
1088}
1089
1090impl Solver {
1091 pub fn market_data(&self) -> MarketData {
1093 self.market_data.clone()
1094 }
1095
1096 pub fn router_address(&self) -> Option<&Bytes> {
1098 self.router_address.as_ref()
1099 }
1100
1101 pub fn derived_data(&self) -> SharedDerivedDataRef {
1103 Arc::clone(&self.derived_data)
1104 }
1105
1106 pub fn subscribe_market_events(&self) -> broadcast::Receiver<crate::feed::events::MarketEvent> {
1111 self.market_event_tx.subscribe()
1112 }
1113
1114 pub async fn quote(&self, request: QuoteRequest) -> Result<Quote, SolveError> {
1120 self.router.quote(request).await
1121 }
1122
1123 pub async fn wait_until_ready(&self, timeout: Duration) -> Result<(), WaitReadyError> {
1140 const POLL_INTERVAL: Duration = Duration::from_millis(500);
1141
1142 let deadline = tokio::time::Instant::now() + timeout;
1143
1144 loop {
1145 let market_ready = self
1146 .market_data
1147 .read()
1148 .await
1149 .last_updated()
1150 .is_some();
1151 let derived_ready = self
1152 .derived_data
1153 .read()
1154 .await
1155 .derived_data_ready();
1156
1157 if market_ready && derived_ready {
1158 return Ok(());
1159 }
1160
1161 if tokio::time::Instant::now() >= deadline {
1162 return Err(WaitReadyError { timeout_ms: timeout.as_millis() as u64 });
1163 }
1164
1165 tokio::time::sleep(POLL_INTERVAL).await;
1166 }
1167 }
1168
1169 #[cfg(feature = "test-utils")]
1182 pub async fn from_recording(
1183 chain: Chain,
1184 updates: Vec<tycho_simulation::protocol::models::Update>,
1185 pools: std::collections::HashMap<String, PoolConfig>,
1186 gas_price_wei: Option<num_bigint::BigUint>,
1187 ) -> Result<Self, SolverBuildError> {
1188 if pools.is_empty() {
1189 return Err(SolverBuildError::NoPools);
1190 }
1191
1192 let market_data = MarketData::new_shared();
1193
1194 let feed_config =
1196 TychoFeedConfig::new("ws://replay".to_string(), chain, None, false, vec![], 0.0);
1197 let feed = TychoFeed::new(feed_config, market_data.clone());
1198 let market_event_tx = feed.event_sender();
1199 let _feed_rx = feed.subscribe();
1200
1201 for update in updates {
1202 feed.handle_tycho_message(update)
1203 .await
1204 .map_err(|e| SolverBuildError::Replay(e.to_string()))?;
1205 }
1206
1207 let gas_price = match gas_price_wei {
1209 Some(price) => price,
1210 None => {
1211 tracing::warn!("no recorded gas price, defaulting to 10 gwei");
1212 num_bigint::BigUint::from(10_000_000_000u64)
1213 }
1214 };
1215 let block_number = match market_data.read().await.last_updated() {
1216 Some(block) => block.number(),
1217 None => {
1218 tracing::warn!("no block number from replayed updates, defaulting to 0");
1219 0
1220 }
1221 };
1222 {
1223 let mut market = market_data.write().await;
1224 market.update_gas_price(BlockGasPrice {
1225 block_number,
1226 block_hash: Default::default(),
1227 block_timestamp: 0,
1228 pricing: GasPrice::Legacy { gas_price },
1229 });
1230 }
1231
1232 let gas_token = native_token(&chain).map_err(|_| SolverBuildError::GasToken)?;
1234 let computation_config = ComputationManagerConfig::new()
1235 .with_gas_token(gas_token)
1236 .with_depth_slippage_threshold(DEFAULT_DEPTH_SLIPPAGE_THRESHOLD);
1237 let (computation_manager, _) =
1238 ComputationManager::new(computation_config, market_data.clone())
1239 .map_err(|e| SolverBuildError::ComputationManager(e.to_string()))?;
1240
1241 let derived_data: SharedDerivedDataRef = computation_manager.store();
1242 let derived_event_tx = computation_manager.event_sender();
1243
1244 let computation_event_rx = feed.subscribe();
1245 let (computation_shutdown_tx, computation_shutdown_rx) = broadcast::channel(1);
1246
1247 let computation_handle = tokio::spawn(async move {
1248 computation_manager
1249 .run(computation_event_rx, computation_shutdown_rx)
1250 .await;
1251 });
1252
1253 let mut solver_pool_handles: Vec<SolverPoolHandle> = Vec::new();
1255 let mut worker_pools: Vec<WorkerPool> = Vec::new();
1256 let mut max_timeout_ms = 0u64;
1257
1258 for (name, pool_cfg) in &pools {
1259 let algo_cfg = AlgorithmConfig::new(
1260 pool_cfg.min_hops(),
1261 pool_cfg.max_hops(),
1262 Duration::from_millis(pool_cfg.timeout_ms()),
1263 pool_cfg.max_routes(),
1264 )?;
1265
1266 let pool_event_rx = feed.subscribe();
1267 let derived_rx = derived_event_tx.subscribe();
1268
1269 let (worker_pool, task_handle) = WorkerPoolBuilder::new()
1270 .name(name.clone())
1271 .algorithm(pool_cfg.algorithm().to_string())
1272 .algorithm_config(algo_cfg)
1273 .num_workers(pool_cfg.num_workers())
1274 .task_queue_capacity(pool_cfg.task_queue_capacity())
1275 .build(market_data.clone(), Arc::clone(&derived_data), pool_event_rx, derived_rx)?;
1276
1277 solver_pool_handles.push(SolverPoolHandle::new(worker_pool.name(), task_handle));
1278 max_timeout_ms = max_timeout_ms.max(pool_cfg.timeout_ms());
1279 worker_pools.push(worker_pool);
1280 }
1281
1282 let encoder = {
1284 let registry = SwapEncoderRegistry::new(chain)
1285 .add_default_encoders(None)
1286 .map_err(|e| SolverBuildError::Encoder(e.to_string()))?;
1287 Encoder::new(chain, registry).map_err(|e| SolverBuildError::Encoder(e.to_string()))?
1288 };
1289
1290 let router_address = encoder.router_address().cloned();
1291 let router_fees = encoder.router_fees();
1295 router_fees.set(crate::encoding::router_fees::RouterFees::new(
1296 100_000_000,
1297 0,
1298 0,
1299 std::collections::HashMap::new(),
1300 ));
1301 let router_config = WorkerPoolRouterConfig::default()
1302 .with_timeout(Duration::from_millis(max_timeout_ms.max(5000)))
1303 .with_min_responses(defaults::ROUTER_MIN_RESPONSES);
1304 let router = WorkerPoolRouter::new(solver_pool_handles, router_config, encoder);
1305
1306 let market_read = market_data.read().await;
1308 let added = market_read.component_topology();
1309 drop(market_read);
1310
1311 if market_event_tx
1312 .send(MarketEvent::MarketUpdated {
1313 added_components: added,
1314 removed_components: vec![],
1315 updated_components: vec![],
1316 })
1317 .is_err()
1318 {
1319 tracing::warn!("no receivers for initial MarketUpdated broadcast");
1320 }
1321
1322 let feed_handle = tokio::spawn(futures::future::pending::<()>());
1325 let gas_price_handle = tokio::spawn(async { });
1326 let metrics_sampler_handle = tokio::spawn(async { });
1327 let router_fee_handle = tokio::spawn(async { });
1328
1329 Ok(Solver {
1330 router,
1331 worker_pools,
1332 market_data,
1333 derived_data,
1334 router_fees,
1335 feed_handle,
1336 gas_price_handle,
1337 metrics_sampler_handle,
1338 router_fee_handle,
1339 computation_handle,
1340 computation_shutdown_tx,
1341 chain,
1342 router_address,
1343 market_event_tx,
1344 })
1345 }
1346
1347 pub fn shutdown(self) {
1349 let _ = self.computation_shutdown_tx.send(());
1350 for pool in self.worker_pools {
1351 pool.shutdown();
1352 }
1353 self.feed_handle.abort();
1354 self.gas_price_handle.abort();
1355 self.metrics_sampler_handle.abort();
1356 self.router_fee_handle.abort();
1357 }
1358
1359 pub fn into_parts(self) -> SolverParts {
1361 SolverParts {
1362 router: self.router,
1363 worker_pools: self.worker_pools,
1364 market_data: self.market_data,
1365 derived_data: self.derived_data,
1366 router_fees: self.router_fees,
1367 feed_handle: self.feed_handle,
1368 gas_price_handle: self.gas_price_handle,
1369 metrics_sampler_handle: self.metrics_sampler_handle,
1370 router_fee_handle: self.router_fee_handle,
1371 computation_handle: self.computation_handle,
1372 computation_shutdown_tx: self.computation_shutdown_tx,
1373 chain: self.chain,
1374 router_address: self.router_address,
1375 }
1376 }
1377}
1378
1379pub struct SolverParts {
1383 router: WorkerPoolRouter,
1385 worker_pools: Vec<WorkerPool>,
1387 market_data: MarketData,
1389 derived_data: SharedDerivedDataRef,
1391 router_fees: SharedRouterFees,
1393 feed_handle: JoinHandle<()>,
1395 gas_price_handle: JoinHandle<()>,
1397 metrics_sampler_handle: JoinHandle<()>,
1399 router_fee_handle: JoinHandle<()>,
1401 computation_handle: JoinHandle<()>,
1403 computation_shutdown_tx: broadcast::Sender<()>,
1405 chain: Chain,
1407 router_address: Option<Bytes>,
1409}
1410
1411impl SolverParts {
1412 pub fn chain(&self) -> Chain {
1414 self.chain
1415 }
1416
1417 pub fn router_address(&self) -> Option<&Bytes> {
1419 self.router_address.as_ref()
1420 }
1421
1422 pub fn worker_pools(&self) -> &[WorkerPool] {
1424 &self.worker_pools
1425 }
1426
1427 pub fn market_data(&self) -> &MarketData {
1429 &self.market_data
1430 }
1431
1432 pub fn derived_data(&self) -> &SharedDerivedDataRef {
1434 &self.derived_data
1435 }
1436
1437 pub fn router_fees(&self) -> &SharedRouterFees {
1439 &self.router_fees
1440 }
1441
1442 pub fn into_router(self) -> WorkerPoolRouter {
1444 self.router
1445 }
1446
1447 #[allow(clippy::type_complexity)]
1449 pub fn into_components(
1450 self,
1451 ) -> (
1452 WorkerPoolRouter,
1453 Vec<WorkerPool>,
1454 MarketData,
1455 SharedDerivedDataRef,
1456 JoinHandle<()>,
1457 JoinHandle<()>,
1458 JoinHandle<()>,
1459 JoinHandle<()>,
1460 JoinHandle<()>,
1461 broadcast::Sender<()>,
1462 ) {
1463 (
1464 self.router,
1465 self.worker_pools,
1466 self.market_data,
1467 self.derived_data,
1468 self.feed_handle,
1469 self.gas_price_handle,
1470 self.metrics_sampler_handle,
1471 self.router_fee_handle,
1472 self.computation_handle,
1473 self.computation_shutdown_tx,
1474 )
1475 }
1476}
1477
1478#[cfg(test)]
1479mod tests {
1480 use super::*;
1481
1482 #[test]
1485 fn test_unscoped_pool_resolves_to_public_only() {
1486 let config = PoolConfig::new("most_liquid");
1487 assert_eq!(config.liquidity_scope(), None);
1488 assert_eq!(
1489 config
1490 .liquidity_scope()
1491 .unwrap_or_default(),
1492 LiquidityScope::PublicOnly
1493 );
1494 }
1495}