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::{config::WorkerPoolRouterConfig, SolverPoolHandle, WorkerPoolRouter},
52 Algorithm, Quote, QuoteRequest, SolveError,
53};
54
55pub mod defaults {
61 use std::time::Duration;
62
63 pub const MIN_TOKEN_QUALITY: i32 = 100;
65 pub const TRADED_N_DAYS_AGO: u64 = 3;
67 pub const TVL_BUFFER_RATIO: f64 = 1.1;
69 pub const GAS_REFRESH_INTERVAL: Duration = Duration::from_secs(30);
71 pub const METRICS_SAMPLE_INTERVAL: Duration = Duration::from_secs(10);
73 pub const ROUTER_FEE_REFRESH_INTERVAL: Duration = Duration::from_secs(300);
75 pub const RECONNECT_DELAY: Duration = Duration::from_secs(5);
77 pub const ROUTER_MIN_RESPONSES: usize = 0;
80 pub const POOL_TASK_QUEUE_CAPACITY: usize = 1000;
82 pub const POOL_MIN_HOPS: usize = 1;
84 pub const POOL_MAX_HOPS: usize = 3;
86 pub const POOL_TIMEOUT_MS: u64 = 100;
88}
89
90const DEFAULT_TYCHO_USE_TLS: bool = true;
92const DEFAULT_DEPTH_SLIPPAGE_THRESHOLD: f64 = 0.01;
93const DEFAULT_ROUTER_TIMEOUT: Duration = Duration::from_secs(10);
96
97fn default_task_queue_capacity() -> usize {
100 defaults::POOL_TASK_QUEUE_CAPACITY
101}
102
103fn default_min_hops() -> usize {
104 defaults::POOL_MIN_HOPS
105}
106
107fn default_max_hops() -> usize {
108 defaults::POOL_MAX_HOPS
109}
110
111fn default_algo_timeout_ms() -> u64 {
112 defaults::POOL_TIMEOUT_MS
113}
114
115fn parse_connector_tokens(
116 raw: Option<&[String]>,
117) -> Result<Option<HashSet<Address>>, SolverBuildError> {
118 let Some(strings) = raw else {
119 return Ok(None);
120 };
121 let mut set = HashSet::with_capacity(strings.len());
122 for s in strings {
123 let addr = Address::from_str(s).map_err(|e| AlgorithmError::InvalidConfiguration {
124 reason: format!("connector_tokens: invalid address {s:?}: {e}"),
125 })?;
126 set.insert(addr);
127 }
128 Ok(Some(set))
129}
130
131#[must_use]
133#[derive(Debug, Clone, Serialize, Deserialize)]
134pub struct PoolConfig {
135 algorithm: String,
137 #[serde(default = "num_cpus::get")]
139 num_workers: usize,
140 #[serde(default = "default_task_queue_capacity")]
142 task_queue_capacity: usize,
143 #[serde(default = "default_min_hops")]
145 min_hops: usize,
146 #[serde(default = "default_max_hops")]
148 max_hops: usize,
149 #[serde(default = "default_algo_timeout_ms")]
151 timeout_ms: u64,
152 #[serde(default)]
154 max_routes: Option<usize>,
155 #[serde(default)]
158 connector_tokens: Option<Vec<String>>,
159}
160
161impl PoolConfig {
162 pub fn new(algorithm: impl Into<String>) -> Self {
164 Self {
165 algorithm: algorithm.into(),
166 num_workers: num_cpus::get(),
167 task_queue_capacity: defaults::POOL_TASK_QUEUE_CAPACITY,
168 min_hops: defaults::POOL_MIN_HOPS,
169 max_hops: defaults::POOL_MAX_HOPS,
170 timeout_ms: defaults::POOL_TIMEOUT_MS,
171 max_routes: None,
172 connector_tokens: None,
173 }
174 }
175
176 pub fn algorithm(&self) -> &str {
178 &self.algorithm
179 }
180
181 pub fn num_workers(&self) -> usize {
183 self.num_workers
184 }
185
186 pub fn with_num_workers(mut self, num_workers: usize) -> Self {
188 self.num_workers = num_workers;
189 self
190 }
191
192 pub fn with_task_queue_capacity(mut self, task_queue_capacity: usize) -> Self {
194 self.task_queue_capacity = task_queue_capacity;
195 self
196 }
197
198 pub fn with_min_hops(mut self, min_hops: usize) -> Self {
200 self.min_hops = min_hops;
201 self
202 }
203
204 pub fn with_max_hops(mut self, max_hops: usize) -> Self {
206 self.max_hops = max_hops;
207 self
208 }
209
210 pub fn with_timeout_ms(mut self, timeout_ms: u64) -> Self {
212 self.timeout_ms = timeout_ms;
213 self
214 }
215
216 pub fn with_max_routes(mut self, max_routes: Option<usize>) -> Self {
218 self.max_routes = max_routes;
219 self
220 }
221
222 pub fn task_queue_capacity(&self) -> usize {
224 self.task_queue_capacity
225 }
226
227 pub fn min_hops(&self) -> usize {
229 self.min_hops
230 }
231
232 pub fn max_hops(&self) -> usize {
234 self.max_hops
235 }
236
237 pub fn timeout_ms(&self) -> u64 {
239 self.timeout_ms
240 }
241
242 pub fn max_routes(&self) -> Option<usize> {
244 self.max_routes
245 }
246
247 pub fn with_connector_tokens(mut self, tokens: Vec<String>) -> Self {
250 self.connector_tokens = Some(tokens);
251 self
252 }
253
254 pub fn connector_tokens(&self) -> Option<&[String]> {
256 self.connector_tokens.as_deref()
257 }
258}
259
260#[derive(Debug, thiserror::Error)]
262#[error("timed out after {timeout_ms}ms waiting for market data and derived computations")]
263pub struct WaitReadyError {
264 timeout_ms: u64,
265}
266
267#[non_exhaustive]
269#[derive(Debug, thiserror::Error)]
270pub enum SolverBuildError {
271 #[error("failed to create ethereum RPC client: {0}")]
273 RpcClient(String),
274 #[error(transparent)]
276 AlgorithmConfig(#[from] AlgorithmError),
277 #[error("failed to create computation manager: {0}")]
279 ComputationManager(String),
280 #[error("failed to create encoder: {0}")]
282 Encoder(String),
283 #[error("failed to create router fee fetcher: {0}")]
285 RouterFeeFetcher(String),
286 #[error(transparent)]
288 UnknownAlgorithm(#[from] UnknownAlgorithmError),
289 #[error("gas token not configured for chain")]
291 GasToken,
292 #[error("no worker pools configured")]
294 NoPools,
295 #[cfg(feature = "test-utils")]
297 #[error("replay failed: {0}")]
298 Replay(String),
299 #[error("feed setup failed before delivering pending processor: {0}")]
304 FeedSetup(String),
305 #[error("pending processor channel closed before processor was delivered")]
308 PendingChannelClosed,
309 #[cfg(feature = "experimental")]
312 #[error("step controller channel closed before controller was delivered")]
313 StepControllerChannelClosed,
314}
315
316enum PoolEntry {
318 BuiltIn {
319 name: String,
320 algorithm: String,
321 num_workers: usize,
322 task_queue_capacity: usize,
323 min_hops: usize,
324 max_hops: usize,
325 timeout_ms: u64,
326 max_routes: Option<usize>,
327 connector_tokens: Option<HashSet<Address>>,
328 },
329 Custom(CustomPoolEntry),
330}
331
332struct CustomPoolEntry {
334 name: String,
335 num_workers: usize,
336 task_queue_capacity: usize,
337 min_hops: usize,
338 max_hops: usize,
339 timeout_ms: u64,
340 max_routes: Option<usize>,
341 configure: Box<dyn FnOnce(WorkerPoolBuilder) -> WorkerPoolBuilder + Send>,
343}
344
345struct BuiltComponents {
348 tycho_feed: TychoFeed,
349 gas_price_fetcher: GasPriceFetcher<EthereumRpcClient>,
350 router_fee_fetcher: Option<RouterFeeFetcher>,
351 computation_manager: ComputationManager,
352 computation_event_rx: broadcast::Receiver<MarketEvent>,
353 computation_shutdown_tx: broadcast::Sender<()>,
354 computation_shutdown_rx: broadcast::Receiver<()>,
355 router: WorkerPoolRouter,
356 worker_pools: Vec<WorkerPool>,
357 market_data: MarketData,
358 derived_data: SharedDerivedDataRef,
359 router_fees: SharedRouterFees,
360 chain: Chain,
361 router_address: Option<Bytes>,
362 pending_indexers: Vec<(String, Box<dyn TxDeltaIndexer>)>,
363 market_event_tx: broadcast::Sender<MarketEvent>,
364}
365
366#[must_use = "a builder does nothing until .build() is called"]
371pub struct FyndBuilder {
372 chain: Chain,
373 tycho_url: String,
374 rpc_url: String,
375 protocols: Vec<String>,
376 min_tvl: f64,
377 tycho_api_key: Option<String>,
378 tycho_use_tls: bool,
379 min_token_quality: i32,
380 traded_n_days_ago: u64,
381 tvl_buffer_ratio: f64,
382 gas_refresh_interval: Duration,
383 reconnect_delay: Duration,
384 blocklisted_components: HashSet<String>,
385 partial_blocks: bool,
386 router_timeout: Duration,
387 router_min_responses: usize,
388 encoder: Option<Encoder>,
389 pools: Vec<PoolEntry>,
390 price_guard_enabled: bool,
391 price_providers: Vec<Box<dyn PriceProvider>>,
392 pending_indexers: Vec<(String, Box<dyn TxDeltaIndexer>)>,
393}
394
395impl FyndBuilder {
396 pub fn new(
398 chain: Chain,
399 tycho_url: impl Into<String>,
400 rpc_url: impl Into<String>,
401 protocols: Vec<String>,
402 min_tvl: f64,
403 ) -> Self {
404 Self {
405 chain,
406 tycho_url: tycho_url.into(),
407 rpc_url: rpc_url.into(),
408 protocols,
409 min_tvl,
410 tycho_api_key: None,
411 tycho_use_tls: DEFAULT_TYCHO_USE_TLS,
412 min_token_quality: defaults::MIN_TOKEN_QUALITY,
413 traded_n_days_ago: defaults::TRADED_N_DAYS_AGO,
414 tvl_buffer_ratio: defaults::TVL_BUFFER_RATIO,
415 gas_refresh_interval: defaults::GAS_REFRESH_INTERVAL,
416 reconnect_delay: defaults::RECONNECT_DELAY,
417 blocklisted_components: HashSet::new(),
418 partial_blocks: false,
419 router_timeout: DEFAULT_ROUTER_TIMEOUT,
420 router_min_responses: defaults::ROUTER_MIN_RESPONSES,
421 encoder: None,
422 pools: Vec::new(),
423 price_guard_enabled: false,
424 price_providers: Vec::new(),
425 pending_indexers: Vec::new(),
426 }
427 }
428
429 pub fn chain(&self) -> Chain {
431 self.chain
432 }
433
434 pub fn tycho_api_key(mut self, key: impl Into<String>) -> Self {
436 self.tycho_api_key = Some(key.into());
437 self
438 }
439
440 pub fn min_tvl(mut self, min_tvl: f64) -> Self {
442 self.min_tvl = min_tvl;
443 self
444 }
445
446 pub fn tycho_use_tls(mut self, use_tls: bool) -> Self {
448 self.tycho_use_tls = use_tls;
449 self
450 }
451
452 pub fn min_token_quality(mut self, quality: i32) -> Self {
455 self.min_token_quality = quality;
456 self
457 }
458
459 pub fn traded_n_days_ago(mut self, days: u64) -> Self {
461 self.traded_n_days_ago = days;
462 self
463 }
464
465 pub fn tvl_buffer_ratio(mut self, ratio: f64) -> Self {
467 self.tvl_buffer_ratio = ratio;
468 self
469 }
470
471 pub fn gas_refresh_interval(mut self, interval: Duration) -> Self {
473 self.gas_refresh_interval = interval;
474 self
475 }
476
477 pub fn reconnect_delay(mut self, delay: Duration) -> Self {
479 self.reconnect_delay = delay;
480 self
481 }
482
483 pub fn blocklisted_components(mut self, components: HashSet<String>) -> Self {
485 self.blocklisted_components = components;
486 self
487 }
488
489 pub fn partial_blocks(mut self, enabled: bool) -> Self {
495 self.partial_blocks = enabled;
496 self
497 }
498
499 pub fn worker_router_timeout(mut self, timeout: Duration) -> Self {
501 self.router_timeout = timeout;
502 self
503 }
504
505 pub fn worker_router_min_responses(mut self, min: usize) -> Self {
507 self.router_min_responses = min;
508 self
509 }
510
511 pub fn encoder(mut self, encoder: Encoder) -> Self {
513 self.encoder = Some(encoder);
514 self
515 }
516
517 pub fn algorithm(mut self, algorithm: impl Into<String>) -> Self {
519 self.pools.push(PoolEntry::BuiltIn {
520 name: "default".to_string(),
521 algorithm: algorithm.into(),
522 num_workers: num_cpus::get(),
523 task_queue_capacity: defaults::POOL_TASK_QUEUE_CAPACITY,
524 min_hops: defaults::POOL_MIN_HOPS,
525 max_hops: defaults::POOL_MAX_HOPS,
526 timeout_ms: defaults::POOL_TIMEOUT_MS,
527 max_routes: None,
528 connector_tokens: None,
529 });
530 self
531 }
532
533 pub fn with_algorithm<A, F>(mut self, name: impl Into<String>, factory: F) -> Self
537 where
538 A: Algorithm + 'static,
539 A::GraphManager: MarketEventHandler + EdgeWeightUpdaterWithDerived + 'static,
540 F: Fn(AlgorithmConfig) -> A + Clone + Send + Sync + 'static,
541 {
542 let name = name.into();
543 let algo_name = name.clone();
544 let configure =
545 Box::new(move |builder: WorkerPoolBuilder| builder.with_algorithm(algo_name, factory));
546 self.pools
547 .push(PoolEntry::Custom(CustomPoolEntry {
548 name,
549 num_workers: num_cpus::get(),
550 task_queue_capacity: defaults::POOL_TASK_QUEUE_CAPACITY,
551 min_hops: defaults::POOL_MIN_HOPS,
552 max_hops: defaults::POOL_MAX_HOPS,
553 timeout_ms: defaults::POOL_TIMEOUT_MS,
554 max_routes: None,
555 configure,
556 }));
557 self
558 }
559
560 pub fn add_default_price_providers(self) -> Self {
567 self.register_price_provider(Box::new(
568 crate::price_guard::hyperliquid::HyperliquidProvider::default(),
569 ))
570 .register_price_provider(Box::new(
571 crate::price_guard::binance_ws::BinanceWsProvider::default(),
572 ))
573 }
574
575 pub fn register_price_provider(mut self, provider: Box<dyn PriceProvider>) -> Self {
580 self.price_providers.push(provider);
581 self
582 }
583
584 pub fn with_pending_indexer(
590 mut self,
591 extractor: impl Into<String>,
592 indexer: Box<dyn TxDeltaIndexer>,
593 ) -> Self {
594 self.pending_indexers
595 .push((extractor.into(), indexer));
596 self
597 }
598
599 pub fn price_guard_enabled(mut self, enabled: bool) -> Self {
606 self.price_guard_enabled = enabled;
607 self
608 }
609
610 pub fn add_pool(
617 mut self,
618 name: impl Into<String>,
619 config: &PoolConfig,
620 ) -> Result<Self, SolverBuildError> {
621 let connector_tokens = parse_connector_tokens(config.connector_tokens())?;
622 self.pools.push(PoolEntry::BuiltIn {
623 name: name.into(),
624 algorithm: config.algorithm().to_string(),
625 num_workers: config.num_workers(),
626 task_queue_capacity: config.task_queue_capacity(),
627 min_hops: config.min_hops(),
628 max_hops: config.max_hops(),
629 timeout_ms: config.timeout_ms(),
630 max_routes: config.max_routes(),
631 connector_tokens,
632 });
633 Ok(self)
634 }
635
636 fn assemble_components(mut self) -> Result<BuiltComponents, SolverBuildError> {
639 if self.pools.is_empty() {
640 return Err(SolverBuildError::NoPools);
641 }
642
643 if self.price_providers.is_empty() {
645 self = self.add_default_price_providers();
646 }
647
648 let market_data = MarketData::new_shared();
649
650 let tycho_feed_config = TychoFeedConfig::new(
651 self.tycho_url,
652 self.chain,
653 self.tycho_api_key,
654 self.tycho_use_tls,
655 self.protocols,
656 self.min_tvl,
657 )
658 .tvl_buffer_ratio(self.tvl_buffer_ratio)
659 .reconnect_delay(self.reconnect_delay)
660 .min_token_quality(self.min_token_quality)
661 .traded_n_days_ago(self.traded_n_days_ago)
662 .blocklisted_components(self.blocklisted_components)
663 .partial_blocks(self.partial_blocks);
664
665 let ethereum_client = EthereumRpcClient::new(self.rpc_url.as_str())
666 .map_err(|e| SolverBuildError::RpcClient(e.to_string()))?;
667
668 let gas_price_fetcher =
669 GasPriceFetcher::new(ethereum_client, market_data.clone(), self.gas_refresh_interval);
670
671 let tycho_feed = TychoFeed::new(tycho_feed_config, market_data.clone());
672 let market_event_tx = tycho_feed.event_sender();
673
674 let gas_token = native_token(&self.chain).map_err(|_| SolverBuildError::GasToken)?;
675 let computation_config = ComputationManagerConfig::new()
676 .with_gas_token(gas_token)
677 .with_depth_slippage_threshold(DEFAULT_DEPTH_SLIPPAGE_THRESHOLD);
678 let (computation_manager, _) =
681 ComputationManager::new(computation_config, market_data.clone())
682 .map_err(|e| SolverBuildError::ComputationManager(e.to_string()))?;
683
684 let derived_data: SharedDerivedDataRef = computation_manager.store();
685 let derived_event_tx = computation_manager.event_sender();
686
687 let computation_event_rx = tycho_feed.subscribe();
689 let (computation_shutdown_tx, computation_shutdown_rx) = broadcast::channel(1);
690
691 let mut solver_pool_handles: Vec<SolverPoolHandle> = Vec::new();
692 let mut worker_pools: Vec<WorkerPool> = Vec::new();
693
694 for pool_entry in self.pools {
695 let pool_event_rx = tycho_feed.subscribe();
696 let derived_rx = derived_event_tx.subscribe();
697
698 let (worker_pool, task_handle) = match pool_entry {
699 PoolEntry::BuiltIn {
700 name,
701 algorithm,
702 num_workers,
703 task_queue_capacity,
704 min_hops,
705 max_hops,
706 timeout_ms,
707 max_routes,
708 connector_tokens,
709 } => {
710 let mut algo_cfg = AlgorithmConfig::new(
711 min_hops,
712 max_hops,
713 Duration::from_millis(timeout_ms),
714 max_routes,
715 )?;
716 if let Some(tokens) = connector_tokens {
717 algo_cfg = algo_cfg.with_connector_tokens(tokens);
718 }
719 WorkerPoolBuilder::new()
720 .name(name)
721 .algorithm(algorithm)
722 .algorithm_config(algo_cfg)
723 .num_workers(num_workers)
724 .task_queue_capacity(task_queue_capacity)
725 .build(
726 market_data.clone(),
727 Arc::clone(&derived_data),
728 pool_event_rx,
729 derived_rx,
730 )?
731 }
732 PoolEntry::Custom(custom) => {
733 let algo_cfg = AlgorithmConfig::new(
734 custom.min_hops,
735 custom.max_hops,
736 Duration::from_millis(custom.timeout_ms),
737 custom.max_routes,
738 )?;
739 let builder = WorkerPoolBuilder::new()
740 .name(custom.name)
741 .algorithm_config(algo_cfg)
742 .num_workers(custom.num_workers)
743 .task_queue_capacity(custom.task_queue_capacity);
744 let builder = (custom.configure)(builder);
745 builder.build(
746 market_data.clone(),
747 Arc::clone(&derived_data),
748 pool_event_rx,
749 derived_rx,
750 )?
751 }
752 };
753
754 solver_pool_handles.push(SolverPoolHandle::new(worker_pool.name(), task_handle));
755 worker_pools.push(worker_pool);
756 }
757
758 let encoder = match self.encoder {
759 Some(enc) => enc,
760 None => {
761 let registry = SwapEncoderRegistry::new(self.chain)
762 .add_default_encoders(None)
763 .map_err(|e| SolverBuildError::Encoder(e.to_string()))?;
764 Encoder::new(self.chain, registry)
765 .map_err(|e| SolverBuildError::Encoder(e.to_string()))?
766 }
767 };
768
769 let chain = self.chain;
770 let router_address = encoder.router_address().cloned();
771 let router_fees = encoder.router_fees();
772
773 let router_fee_fetcher = match &router_address {
774 Some(addr) => Some(
775 RouterFeeFetcher::new(
776 self.rpc_url.as_str(),
777 addr,
778 router_fees.clone(),
779 defaults::ROUTER_FEE_REFRESH_INTERVAL,
780 )
781 .map_err(|e| SolverBuildError::RouterFeeFetcher(e.to_string()))?,
782 ),
783 None => {
784 tracing::warn!(
785 %chain,
786 "no Tycho router for this chain; running quote-only (encoding disabled)"
787 );
788 None
789 }
790 };
791
792 let router_config = WorkerPoolRouterConfig::default()
795 .with_timeout(self.router_timeout)
796 .with_min_responses(self.router_min_responses);
797 let mut router = WorkerPoolRouter::new(solver_pool_handles, router_config, encoder);
798
799 if self.price_guard_enabled {
800 let mut registry = PriceProviderRegistry::new();
801 let mut worker_handles = Vec::new();
802 for mut provider in self.price_providers {
803 worker_handles.push(provider.start(market_data.clone()));
804 registry = registry.register(provider);
805 }
806 let price_guard = PriceGuard::new(registry, worker_handles);
807 router = router.with_price_guard(price_guard);
808 }
809
810 Ok(BuiltComponents {
811 tycho_feed,
812 gas_price_fetcher,
813 router_fee_fetcher,
814 computation_manager,
815 computation_event_rx,
816 computation_shutdown_tx,
817 computation_shutdown_rx,
818 router,
819 worker_pools,
820 market_data,
821 derived_data,
822 router_fees,
823 chain,
824 router_address,
825 pending_indexers: self.pending_indexers,
826 market_event_tx,
827 })
828 }
829
830 pub fn build(self) -> Result<Solver, SolverBuildError> {
836 let mut c = self.assemble_components()?;
837
838 let feed_handle = tokio::spawn(async move {
839 if let Err(e) = c.tycho_feed.run().await {
840 metrics::counter!("tycho_feed_failures_total").increment(1);
841 tracing::error!(error = %e, "tycho feed error");
842 }
843 });
844 let gas_price_handle = tokio::spawn(async move {
845 c.gas_price_fetcher.run().await;
846 });
847 let metrics_sampler =
848 MetricsSampler::new(c.market_data.clone(), defaults::METRICS_SAMPLE_INTERVAL);
849 let metrics_sampler_handle = tokio::spawn(async move { metrics_sampler.run().await });
850 let router_fee_handle = match c.router_fee_fetcher {
851 Some(fetcher) => tokio::spawn(async move { fetcher.run().await }),
852 None => tokio::spawn(async {}),
853 };
854 let computation_handle = tokio::spawn(async move {
855 c.computation_manager
856 .run(c.computation_event_rx, c.computation_shutdown_rx)
857 .await;
858 });
859
860 Ok(Solver {
861 router: c.router,
862 worker_pools: c.worker_pools,
863 market_data: c.market_data,
864 derived_data: c.derived_data,
865 router_fees: c.router_fees,
866 feed_handle,
867 gas_price_handle,
868 metrics_sampler_handle,
869 router_fee_handle,
870 computation_handle,
871 computation_shutdown_tx: c.computation_shutdown_tx,
872 chain: c.chain,
873 router_address: c.router_address,
874 market_event_tx: c.market_event_tx,
875 })
876 }
877
878 pub async fn build_with_pending(
890 self,
891 ) -> Result<(Solver, PendingBlockProcessor), SolverBuildError> {
892 let mut c = self.assemble_components()?;
893
894 let (pending_tx, pending_rx) =
895 tokio::sync::oneshot::channel::<Result<PendingBlockProcessor, String>>();
896
897 let pending_indexers = c.pending_indexers;
898 let feed_handle = tokio::spawn(async move {
899 if let Err(e) = c
900 .tycho_feed
901 .run_with_pending(pending_tx, pending_indexers)
902 .await
903 {
904 metrics::counter!("tycho_feed_failures_total").increment(1);
905 tracing::error!(error = %e, "tycho feed error");
906 }
907 });
908 let gas_price_handle = tokio::spawn(async move {
909 c.gas_price_fetcher.run().await;
910 });
911 let metrics_sampler =
912 MetricsSampler::new(c.market_data.clone(), defaults::METRICS_SAMPLE_INTERVAL);
913 let metrics_sampler_handle = tokio::spawn(async move { metrics_sampler.run().await });
914 let router_fee_handle = match c.router_fee_fetcher {
915 Some(fetcher) => tokio::spawn(async move { fetcher.run().await }),
916 None => tokio::spawn(async {}),
917 };
918 let computation_handle = tokio::spawn(async move {
919 c.computation_manager
920 .run(c.computation_event_rx, c.computation_shutdown_rx)
921 .await;
922 });
923
924 let pending = pending_rx
925 .await
926 .map_err(|_| SolverBuildError::PendingChannelClosed)?
927 .map_err(SolverBuildError::FeedSetup)?;
928
929 Ok((
930 Solver {
931 router: c.router,
932 worker_pools: c.worker_pools,
933 market_data: c.market_data,
934 derived_data: c.derived_data,
935 router_fees: c.router_fees,
936 feed_handle,
937 gas_price_handle,
938 metrics_sampler_handle,
939 router_fee_handle,
940 computation_handle,
941 computation_shutdown_tx: c.computation_shutdown_tx,
942 chain: c.chain,
943 router_address: c.router_address,
944 market_event_tx: c.market_event_tx,
945 },
946 pending,
947 ))
948 }
949
950 #[cfg(feature = "experimental")]
965 pub async fn build_with_step_controller(
966 self,
967 ) -> Result<(Solver, BlockStepController), SolverBuildError> {
968 let mut c = self.assemble_components()?;
969
970 let (controller_tx, controller_rx) =
971 tokio::sync::oneshot::channel::<Result<BlockStepController, String>>();
972
973 let feed_handle = tokio::spawn(async move {
974 if let Err(e) = c
975 .tycho_feed
976 .run_with_step_controller(controller_tx)
977 .await
978 {
979 tracing::error!(error = %e, "tycho feed error");
980 }
981 });
982 let gas_price_handle = tokio::spawn(async move {
983 c.gas_price_fetcher.run().await;
984 });
985 let metrics_sampler =
986 MetricsSampler::new(c.market_data.clone(), defaults::METRICS_SAMPLE_INTERVAL);
987 let metrics_sampler_handle = tokio::spawn(async move { metrics_sampler.run().await });
988 let router_fee_handle = match c.router_fee_fetcher {
989 Some(fetcher) => tokio::spawn(async move { fetcher.run().await }),
990 None => tokio::spawn(async {}),
991 };
992 let computation_handle = tokio::spawn(async move {
993 c.computation_manager
994 .run(c.computation_event_rx, c.computation_shutdown_rx)
995 .await;
996 });
997
998 let controller = controller_rx
999 .await
1000 .map_err(|_| SolverBuildError::StepControllerChannelClosed)?
1001 .map_err(SolverBuildError::FeedSetup)?;
1002
1003 Ok((
1004 Solver {
1005 router: c.router,
1006 worker_pools: c.worker_pools,
1007 market_data: c.market_data,
1008 derived_data: c.derived_data,
1009 router_fees: c.router_fees,
1010 feed_handle,
1011 gas_price_handle,
1012 metrics_sampler_handle,
1013 router_fee_handle,
1014 computation_handle,
1015 computation_shutdown_tx: c.computation_shutdown_tx,
1016 chain: c.chain,
1017 router_address: c.router_address,
1018 market_event_tx: c.market_event_tx,
1019 },
1020 controller,
1021 ))
1022 }
1023} pub struct Solver {
1027 router: WorkerPoolRouter,
1028 worker_pools: Vec<WorkerPool>,
1029 market_data: MarketData,
1030 derived_data: SharedDerivedDataRef,
1031 router_fees: SharedRouterFees,
1032 feed_handle: JoinHandle<()>,
1033 gas_price_handle: JoinHandle<()>,
1034 metrics_sampler_handle: JoinHandle<()>,
1035 router_fee_handle: JoinHandle<()>,
1036 computation_handle: JoinHandle<()>,
1037 computation_shutdown_tx: broadcast::Sender<()>,
1038 chain: Chain,
1039 router_address: Option<Bytes>,
1040 market_event_tx: broadcast::Sender<MarketEvent>,
1041}
1042
1043impl Solver {
1044 pub fn market_data(&self) -> MarketData {
1046 self.market_data.clone()
1047 }
1048
1049 pub fn router_address(&self) -> Option<&Bytes> {
1051 self.router_address.as_ref()
1052 }
1053
1054 pub fn derived_data(&self) -> SharedDerivedDataRef {
1056 Arc::clone(&self.derived_data)
1057 }
1058
1059 pub fn subscribe_market_events(&self) -> broadcast::Receiver<crate::feed::events::MarketEvent> {
1064 self.market_event_tx.subscribe()
1065 }
1066
1067 pub async fn quote(&self, request: QuoteRequest) -> Result<Quote, SolveError> {
1073 self.router.quote(request).await
1074 }
1075
1076 pub async fn wait_until_ready(&self, timeout: Duration) -> Result<(), WaitReadyError> {
1093 const POLL_INTERVAL: Duration = Duration::from_millis(500);
1094
1095 let deadline = tokio::time::Instant::now() + timeout;
1096
1097 loop {
1098 let market_ready = self
1099 .market_data
1100 .read()
1101 .await
1102 .last_updated()
1103 .is_some();
1104 let derived_ready = self
1105 .derived_data
1106 .read()
1107 .await
1108 .derived_data_ready();
1109
1110 if market_ready && derived_ready {
1111 return Ok(());
1112 }
1113
1114 if tokio::time::Instant::now() >= deadline {
1115 return Err(WaitReadyError { timeout_ms: timeout.as_millis() as u64 });
1116 }
1117
1118 tokio::time::sleep(POLL_INTERVAL).await;
1119 }
1120 }
1121
1122 #[cfg(feature = "test-utils")]
1135 pub async fn from_recording(
1136 chain: Chain,
1137 updates: Vec<tycho_simulation::protocol::models::Update>,
1138 pools: std::collections::HashMap<String, PoolConfig>,
1139 gas_price_wei: Option<num_bigint::BigUint>,
1140 ) -> Result<Self, SolverBuildError> {
1141 if pools.is_empty() {
1142 return Err(SolverBuildError::NoPools);
1143 }
1144
1145 let market_data = MarketData::new_shared();
1146
1147 let feed_config =
1149 TychoFeedConfig::new("ws://replay".to_string(), chain, None, false, vec![], 0.0);
1150 let feed = TychoFeed::new(feed_config, market_data.clone());
1151 let market_event_tx = feed.event_sender();
1152 let _feed_rx = feed.subscribe();
1153
1154 for update in updates {
1155 feed.handle_tycho_message(update)
1156 .await
1157 .map_err(|e| SolverBuildError::Replay(e.to_string()))?;
1158 }
1159
1160 let gas_price = match gas_price_wei {
1162 Some(price) => price,
1163 None => {
1164 tracing::warn!("no recorded gas price, defaulting to 10 gwei");
1165 num_bigint::BigUint::from(10_000_000_000u64)
1166 }
1167 };
1168 let block_number = match market_data.read().await.last_updated() {
1169 Some(block) => block.number(),
1170 None => {
1171 tracing::warn!("no block number from replayed updates, defaulting to 0");
1172 0
1173 }
1174 };
1175 {
1176 let mut market = market_data.write().await;
1177 market.update_gas_price(BlockGasPrice {
1178 block_number,
1179 block_hash: Default::default(),
1180 block_timestamp: 0,
1181 pricing: GasPrice::Legacy { gas_price },
1182 });
1183 }
1184
1185 let gas_token = native_token(&chain).map_err(|_| SolverBuildError::GasToken)?;
1187 let computation_config = ComputationManagerConfig::new()
1188 .with_gas_token(gas_token)
1189 .with_depth_slippage_threshold(DEFAULT_DEPTH_SLIPPAGE_THRESHOLD);
1190 let (computation_manager, _) =
1191 ComputationManager::new(computation_config, market_data.clone())
1192 .map_err(|e| SolverBuildError::ComputationManager(e.to_string()))?;
1193
1194 let derived_data: SharedDerivedDataRef = computation_manager.store();
1195 let derived_event_tx = computation_manager.event_sender();
1196
1197 let computation_event_rx = feed.subscribe();
1198 let (computation_shutdown_tx, computation_shutdown_rx) = broadcast::channel(1);
1199
1200 let computation_handle = tokio::spawn(async move {
1201 computation_manager
1202 .run(computation_event_rx, computation_shutdown_rx)
1203 .await;
1204 });
1205
1206 let mut solver_pool_handles: Vec<SolverPoolHandle> = Vec::new();
1208 let mut worker_pools: Vec<WorkerPool> = Vec::new();
1209 let mut max_timeout_ms = 0u64;
1210
1211 for (name, pool_cfg) in &pools {
1212 let algo_cfg = AlgorithmConfig::new(
1213 pool_cfg.min_hops(),
1214 pool_cfg.max_hops(),
1215 Duration::from_millis(pool_cfg.timeout_ms()),
1216 pool_cfg.max_routes(),
1217 )?;
1218
1219 let pool_event_rx = feed.subscribe();
1220 let derived_rx = derived_event_tx.subscribe();
1221
1222 let (worker_pool, task_handle) = WorkerPoolBuilder::new()
1223 .name(name.clone())
1224 .algorithm(pool_cfg.algorithm().to_string())
1225 .algorithm_config(algo_cfg)
1226 .num_workers(pool_cfg.num_workers())
1227 .task_queue_capacity(pool_cfg.task_queue_capacity())
1228 .build(market_data.clone(), Arc::clone(&derived_data), pool_event_rx, derived_rx)?;
1229
1230 solver_pool_handles.push(SolverPoolHandle::new(worker_pool.name(), task_handle));
1231 max_timeout_ms = max_timeout_ms.max(pool_cfg.timeout_ms());
1232 worker_pools.push(worker_pool);
1233 }
1234
1235 let encoder = {
1237 let registry = SwapEncoderRegistry::new(chain)
1238 .add_default_encoders(None)
1239 .map_err(|e| SolverBuildError::Encoder(e.to_string()))?;
1240 Encoder::new(chain, registry).map_err(|e| SolverBuildError::Encoder(e.to_string()))?
1241 };
1242
1243 let router_address = encoder.router_address().cloned();
1244 let router_fees = encoder.router_fees();
1248 router_fees.set(crate::encoding::router_fees::RouterFees::new(
1249 100_000_000,
1250 0,
1251 0,
1252 std::collections::HashMap::new(),
1253 ));
1254 let router_config = WorkerPoolRouterConfig::default()
1255 .with_timeout(Duration::from_millis(max_timeout_ms.max(5000)))
1256 .with_min_responses(defaults::ROUTER_MIN_RESPONSES);
1257 let router = WorkerPoolRouter::new(solver_pool_handles, router_config, encoder);
1258
1259 let market_read = market_data.read().await;
1261 let added = market_read.component_topology();
1262 drop(market_read);
1263
1264 if market_event_tx
1265 .send(MarketEvent::MarketUpdated {
1266 added_components: added,
1267 removed_components: vec![],
1268 updated_components: vec![],
1269 })
1270 .is_err()
1271 {
1272 tracing::warn!("no receivers for initial MarketUpdated broadcast");
1273 }
1274
1275 let feed_handle = tokio::spawn(futures::future::pending::<()>());
1278 let gas_price_handle = tokio::spawn(async { });
1279 let metrics_sampler_handle = tokio::spawn(async { });
1280 let router_fee_handle = tokio::spawn(async { });
1281
1282 Ok(Solver {
1283 router,
1284 worker_pools,
1285 market_data,
1286 derived_data,
1287 router_fees,
1288 feed_handle,
1289 gas_price_handle,
1290 metrics_sampler_handle,
1291 router_fee_handle,
1292 computation_handle,
1293 computation_shutdown_tx,
1294 chain,
1295 router_address,
1296 market_event_tx,
1297 })
1298 }
1299
1300 pub fn shutdown(self) {
1302 let _ = self.computation_shutdown_tx.send(());
1303 for pool in self.worker_pools {
1304 pool.shutdown();
1305 }
1306 self.feed_handle.abort();
1307 self.gas_price_handle.abort();
1308 self.metrics_sampler_handle.abort();
1309 self.router_fee_handle.abort();
1310 }
1311
1312 pub fn into_parts(self) -> SolverParts {
1314 SolverParts {
1315 router: self.router,
1316 worker_pools: self.worker_pools,
1317 market_data: self.market_data,
1318 derived_data: self.derived_data,
1319 router_fees: self.router_fees,
1320 feed_handle: self.feed_handle,
1321 gas_price_handle: self.gas_price_handle,
1322 metrics_sampler_handle: self.metrics_sampler_handle,
1323 router_fee_handle: self.router_fee_handle,
1324 computation_handle: self.computation_handle,
1325 computation_shutdown_tx: self.computation_shutdown_tx,
1326 chain: self.chain,
1327 router_address: self.router_address,
1328 }
1329 }
1330}
1331
1332pub struct SolverParts {
1336 router: WorkerPoolRouter,
1338 worker_pools: Vec<WorkerPool>,
1340 market_data: MarketData,
1342 derived_data: SharedDerivedDataRef,
1344 router_fees: SharedRouterFees,
1346 feed_handle: JoinHandle<()>,
1348 gas_price_handle: JoinHandle<()>,
1350 metrics_sampler_handle: JoinHandle<()>,
1352 router_fee_handle: JoinHandle<()>,
1354 computation_handle: JoinHandle<()>,
1356 computation_shutdown_tx: broadcast::Sender<()>,
1358 chain: Chain,
1360 router_address: Option<Bytes>,
1362}
1363
1364impl SolverParts {
1365 pub fn chain(&self) -> Chain {
1367 self.chain
1368 }
1369
1370 pub fn router_address(&self) -> Option<&Bytes> {
1372 self.router_address.as_ref()
1373 }
1374
1375 pub fn worker_pools(&self) -> &[WorkerPool] {
1377 &self.worker_pools
1378 }
1379
1380 pub fn market_data(&self) -> &MarketData {
1382 &self.market_data
1383 }
1384
1385 pub fn derived_data(&self) -> &SharedDerivedDataRef {
1387 &self.derived_data
1388 }
1389
1390 pub fn router_fees(&self) -> &SharedRouterFees {
1392 &self.router_fees
1393 }
1394
1395 pub fn into_router(self) -> WorkerPoolRouter {
1397 self.router
1398 }
1399
1400 #[allow(clippy::type_complexity)]
1402 pub fn into_components(
1403 self,
1404 ) -> (
1405 WorkerPoolRouter,
1406 Vec<WorkerPool>,
1407 MarketData,
1408 SharedDerivedDataRef,
1409 JoinHandle<()>,
1410 JoinHandle<()>,
1411 JoinHandle<()>,
1412 JoinHandle<()>,
1413 JoinHandle<()>,
1414 broadcast::Sender<()>,
1415 ) {
1416 (
1417 self.router,
1418 self.worker_pools,
1419 self.market_data,
1420 self.derived_data,
1421 self.feed_handle,
1422 self.gas_price_handle,
1423 self.metrics_sampler_handle,
1424 self.router_fee_handle,
1425 self.computation_handle,
1426 self.computation_shutdown_tx,
1427 )
1428 }
1429}