Skip to main content

fynd_core/
solver.rs

1//! High-level solver setup via [`FyndBuilder`].
2//!
3//! [`FyndBuilder`] assembles the full Tycho feed + gas fetcher + computation
4//! manager + one or more worker pools + encoder + router pipeline with sensible
5//! defaults. For simple cases a single call chain is all that's needed:
6//!
7//! ```ignore
8//! let solver = FyndBuilder::new(chain, tycho_url, rpc_url, protocols, min_tvl)
9//!     .tycho_api_key(key)
10//!     .algorithm("most_liquid")
11//!     .build()?;
12//! ```
13use 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
55/// Default values for [`FyndBuilder`] configuration and [`PoolConfig`] deserialization.
56///
57/// These are the single source of truth for all tunable defaults. Downstream
58/// crates (e.g. `fynd-rpc`) should re-export or reference these rather than
59/// redeclaring their own copies.
60pub mod defaults {
61    use std::time::Duration;
62
63    /// Minimum token quality score required for a token to be included in routing.
64    pub const MIN_TOKEN_QUALITY: i32 = 100;
65    /// Maximum age (in days) of trading history required for a token to be considered liquid.
66    pub const TRADED_N_DAYS_AGO: u64 = 3;
67    /// Multiplier applied to a pool's TVL when estimating available liquidity.
68    pub const TVL_BUFFER_RATIO: f64 = 1.1;
69    /// How often the gas price is refreshed from the RPC node.
70    pub const GAS_REFRESH_INTERVAL: Duration = Duration::from_secs(30);
71    /// How often per-protocol market metrics are sampled and exported.
72    pub const METRICS_SAMPLE_INTERVAL: Duration = Duration::from_secs(10);
73    /// How often router fees are refreshed from the on-chain FeeCalculator contract.
74    pub const ROUTER_FEE_REFRESH_INTERVAL: Duration = Duration::from_secs(300);
75    /// Delay before reconnecting to the Tycho feed after a disconnect.
76    pub const RECONNECT_DELAY: Duration = Duration::from_secs(5);
77    /// Minimum number of solver pool responses required before returning a quote (`0` = wait for
78    /// all).
79    pub const ROUTER_MIN_RESPONSES: usize = 0;
80    /// Capacity of the task queue for each worker pool.
81    pub const POOL_TASK_QUEUE_CAPACITY: usize = 1000;
82    /// Minimum number of hops allowed in a route.
83    pub const POOL_MIN_HOPS: usize = 1;
84    /// Maximum number of hops allowed in a route.
85    pub const POOL_MAX_HOPS: usize = 3;
86    /// Per-pool solve timeout in milliseconds.
87    pub const POOL_TIMEOUT_MS: u64 = 100;
88}
89
90// Internal-only defaults not shared with downstream crates.
91const DEFAULT_TYCHO_USE_TLS: bool = true;
92const DEFAULT_DEPTH_SLIPPAGE_THRESHOLD: f64 = 0.01;
93/// Generous router timeout for standalone (non-server) use. HTTP services should
94/// override this to a tighter value appropriate for their SLA.
95const DEFAULT_ROUTER_TIMEOUT: Duration = Duration::from_secs(10);
96
97// serde requires free functions for `#[serde(default = "...")]` — these delegate to the
98// defaults module so both deserialization and the builder stay in sync.
99fn 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/// Per-pool configuration for [`FyndBuilder::add_pool`].
132#[must_use]
133#[derive(Debug, Clone, Serialize, Deserialize)]
134pub struct PoolConfig {
135    /// Algorithm name for this pool (e.g., `"most_liquid"`).
136    algorithm: String,
137    /// Number of worker threads for this pool.
138    #[serde(default = "num_cpus::get")]
139    num_workers: usize,
140    /// Task queue capacity for this pool.
141    #[serde(default = "default_task_queue_capacity")]
142    task_queue_capacity: usize,
143    /// Minimum hops to search (must be >= 1).
144    #[serde(default = "default_min_hops")]
145    min_hops: usize,
146    /// Maximum hops to search.
147    #[serde(default = "default_max_hops")]
148    max_hops: usize,
149    /// Timeout for solving in milliseconds.
150    #[serde(default = "default_algo_timeout_ms")]
151    timeout_ms: u64,
152    /// Maximum number of paths to simulate per solve. `None` simulates all scored paths.
153    #[serde(default)]
154    max_routes: Option<usize>,
155    /// Lowercase hex addresses (e.g. `"0xc02aaa…"`) allowed as intermediate routing hops.
156    /// Absent = no restriction. Typically 3–10 entries (e.g. WETH, USDC, USDT, DAI).
157    #[serde(default)]
158    connector_tokens: Option<Vec<String>>,
159}
160
161impl PoolConfig {
162    /// Creates a new pool config with the given algorithm name and defaults for all other fields.
163    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    /// Returns the algorithm name.
177    pub fn algorithm(&self) -> &str {
178        &self.algorithm
179    }
180
181    /// Returns the number of worker threads.
182    pub fn num_workers(&self) -> usize {
183        self.num_workers
184    }
185
186    /// Sets the number of worker threads.
187    pub fn with_num_workers(mut self, num_workers: usize) -> Self {
188        self.num_workers = num_workers;
189        self
190    }
191
192    /// Sets the task queue capacity.
193    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    /// Sets the minimum hops.
199    pub fn with_min_hops(mut self, min_hops: usize) -> Self {
200        self.min_hops = min_hops;
201        self
202    }
203
204    /// Sets the maximum hops.
205    pub fn with_max_hops(mut self, max_hops: usize) -> Self {
206        self.max_hops = max_hops;
207        self
208    }
209
210    /// Sets the timeout in milliseconds.
211    pub fn with_timeout_ms(mut self, timeout_ms: u64) -> Self {
212        self.timeout_ms = timeout_ms;
213        self
214    }
215
216    /// Sets the maximum number of routes to simulate.
217    pub fn with_max_routes(mut self, max_routes: Option<usize>) -> Self {
218        self.max_routes = max_routes;
219        self
220    }
221
222    /// Returns the task queue capacity.
223    pub fn task_queue_capacity(&self) -> usize {
224        self.task_queue_capacity
225    }
226
227    /// Returns the minimum hops.
228    pub fn min_hops(&self) -> usize {
229        self.min_hops
230    }
231
232    /// Returns the maximum hops.
233    pub fn max_hops(&self) -> usize {
234        self.max_hops
235    }
236
237    /// Returns the timeout in milliseconds.
238    pub fn timeout_ms(&self) -> u64 {
239        self.timeout_ms
240    }
241
242    /// Returns the maximum number of routes to simulate.
243    pub fn max_routes(&self) -> Option<usize> {
244        self.max_routes
245    }
246
247    /// Restricts intermediate hops to the given token addresses (hex strings with or without `0x`
248    /// prefix). Absent = no restriction.
249    pub fn with_connector_tokens(mut self, tokens: Vec<String>) -> Self {
250        self.connector_tokens = Some(tokens);
251        self
252    }
253
254    /// Returns the raw connector token address strings, or `None` if unrestricted.
255    pub fn connector_tokens(&self) -> Option<&[String]> {
256        self.connector_tokens.as_deref()
257    }
258}
259
260/// Error returned by [`Solver::wait_until_ready`].
261#[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/// Error returned by [`FyndBuilder::build`] and [`FyndBuilder::build_with_pending`].
268#[non_exhaustive]
269#[derive(Debug, thiserror::Error)]
270pub enum SolverBuildError {
271    /// The Ethereum RPC client could not be created (e.g. malformed URL).
272    #[error("failed to create ethereum RPC client: {0}")]
273    RpcClient(String),
274    /// An invalid algorithm configuration was supplied.
275    #[error(transparent)]
276    AlgorithmConfig(#[from] AlgorithmError),
277    /// The [`ComputationManager`] failed to initialise.
278    #[error("failed to create computation manager: {0}")]
279    ComputationManager(String),
280    /// The swap encoder could not be created for the target chain.
281    #[error("failed to create encoder: {0}")]
282    Encoder(String),
283    /// The router fee fetcher could not be created (e.g. malformed RPC URL).
284    #[error("failed to create router fee fetcher: {0}")]
285    RouterFeeFetcher(String),
286    /// A pool referenced an algorithm name that is not registered.
287    #[error(transparent)]
288    UnknownAlgorithm(#[from] UnknownAlgorithmError),
289    /// No native gas token is defined for the requested chain.
290    #[error("gas token not configured for chain")]
291    GasToken,
292    /// [`FyndBuilder::build`] was called without configuring any worker pools.
293    #[error("no worker pools configured")]
294    NoPools,
295    /// A recorded update failed to replay through the feed.
296    #[cfg(feature = "test-utils")]
297    #[error("replay failed: {0}")]
298    Replay(String),
299    /// The feed task failed before delivering the [`PendingBlockProcessor`].
300    ///
301    /// The inner string is the `DataFeedError` message from `TychoFeed::run_with_pending`
302    /// (e.g. "failed to load tokens: connection refused").
303    #[error("feed setup failed before delivering pending processor: {0}")]
304    FeedSetup(String),
305    /// The pending-processor oneshot closed without delivering a value, meaning the feed task
306    /// panicked rather than returning an error through the channel.
307    #[error("pending processor channel closed before processor was delivered")]
308    PendingChannelClosed,
309    /// The step-controller oneshot closed without delivering a value, meaning the feed task
310    /// panicked rather than returning an error through the channel.
311    #[cfg(feature = "experimental")]
312    #[error("step controller channel closed before controller was delivered")]
313    StepControllerChannelClosed,
314}
315
316/// Internal pool entry — either a built-in algorithm (by name) or a custom one.
317enum 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
332/// Pool entry backed by a custom [`Algorithm`] implementation.
333struct 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    /// Applies the custom algorithm to a `WorkerPoolBuilder`.
342    configure: Box<dyn FnOnce(WorkerPoolBuilder) -> WorkerPoolBuilder + Send>,
343}
344
345/// All components produced by [`FyndBuilder::assemble_components`], consumed by
346/// [`FyndBuilder::build`] and [`FyndBuilder::build_with_pending`].
347struct 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/// Builder for assembling the full solver pipeline.
367///
368/// Configures the Tycho market-data feed, gas price fetcher, derived-data
369/// computation manager, one or more worker pools, encoder, and router.
370#[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    /// Creates a new builder with the required parameters.
397    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    /// The blockchain this builder is configured for.
430    pub fn chain(&self) -> Chain {
431        self.chain
432    }
433
434    /// Sets the Tycho API key.
435    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    /// Overrides the minimum TVL filter set in [`FyndBuilder::new`].
441    pub fn min_tvl(mut self, min_tvl: f64) -> Self {
442        self.min_tvl = min_tvl;
443        self
444    }
445
446    /// Enables or disables TLS for the Tycho WebSocket connection (default: `true`).
447    pub fn tycho_use_tls(mut self, use_tls: bool) -> Self {
448        self.tycho_use_tls = use_tls;
449        self
450    }
451
452    /// Sets the minimum token quality score; tokens below this threshold are excluded (default:
453    /// 100).
454    pub fn min_token_quality(mut self, quality: i32) -> Self {
455        self.min_token_quality = quality;
456        self
457    }
458
459    /// Filters out pools whose last trade is older than `days` days (default: 3).
460    pub fn traded_n_days_ago(mut self, days: u64) -> Self {
461        self.traded_n_days_ago = days;
462        self
463    }
464
465    /// Multiplies reported TVL by `ratio` before applying the `min_tvl` filter (default: 1.1).
466    pub fn tvl_buffer_ratio(mut self, ratio: f64) -> Self {
467        self.tvl_buffer_ratio = ratio;
468        self
469    }
470
471    /// Sets how often the gas price is refreshed from the RPC node (default: 30 s).
472    pub fn gas_refresh_interval(mut self, interval: Duration) -> Self {
473        self.gas_refresh_interval = interval;
474        self
475    }
476
477    /// Sets the delay before reconnecting to Tycho after a disconnection (default: 5 s).
478    pub fn reconnect_delay(mut self, delay: Duration) -> Self {
479        self.reconnect_delay = delay;
480        self
481    }
482
483    /// Sets component IDs to exclude from the Tycho stream.
484    pub fn blocklisted_components(mut self, components: HashSet<String>) -> Self {
485        self.blocklisted_components = components;
486        self
487    }
488
489    /// Enables partial block (flashblock) updates from the Tycho stream (default: `false`).
490    ///
491    /// When enabled, the stream delivers pool state updates mid-block rather than only at
492    /// finalization, reducing latency. Only supported for on-chain protocols; RFQ streams are
493    /// unaffected.
494    pub fn partial_blocks(mut self, enabled: bool) -> Self {
495        self.partial_blocks = enabled;
496        self
497    }
498
499    /// Sets the worker router timeout (default: 10s).
500    pub fn worker_router_timeout(mut self, timeout: Duration) -> Self {
501        self.router_timeout = timeout;
502        self
503    }
504
505    /// Sets the minimum number of solver responses before early return (default: 0).
506    pub fn worker_router_min_responses(mut self, min: usize) -> Self {
507        self.router_min_responses = min;
508        self
509    }
510
511    /// Overrides the default encoder.
512    pub fn encoder(mut self, encoder: Encoder) -> Self {
513        self.encoder = Some(encoder);
514        self
515    }
516
517    /// Shorthand: adds a single pool named `"default"` using a built-in algorithm by name.
518    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    /// Shorthand: adds a single pool with a custom [`Algorithm`] implementation.
534    ///
535    /// The `factory` closure is called once per worker thread.
536    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    /// Registers the built-in price providers (Hyperliquid + Binance).
561    ///
562    /// Called automatically during [`build`](Self::build) if no providers have been
563    /// registered and the price guard is not disabled. To use only custom
564    /// providers, call [`register_price_provider`](Self::register_price_provider)
565    /// before `build()` and the defaults will be skipped.
566    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    /// Registers a custom price provider for the price guard.
576    ///
577    /// The provider's [`start`](PriceProvider::start) method is called during
578    /// [`build`](Self::build) with the shared market data.
579    pub fn register_price_provider(mut self, provider: Box<dyn PriceProvider>) -> Self {
580        self.price_providers.push(provider);
581        self
582    }
583
584    /// Registers a [`TxDeltaIndexer`] for ephemeral pending-block simulation.
585    ///
586    /// `extractor` is the protocol synchronizer name (e.g. `"uniswap_v3"`). Only has effect
587    /// when calling [`build_with_pending`](Self::build_with_pending). VM protocols (prefix
588    /// `"vm:"`) are rejected by the underlying stream builder at build time.
589    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    /// Enables or disables the price guard.
600    ///
601    /// When enabled, providers are started and caches stay warm. Validation
602    /// only runs for requests where the client sets `enabled: true` in
603    /// `PriceGuardConfig`. When disabled, no providers are started and
604    /// per-request attempts to use the guard return an error.
605    pub fn price_guard_enabled(mut self, enabled: bool) -> Self {
606        self.price_guard_enabled = enabled;
607        self
608    }
609
610    /// Adds a named pool using the given [`PoolConfig`].
611    ///
612    /// # Errors
613    ///
614    /// Returns [`SolverBuildError::AlgorithmConfig`] if any address in `connector_tokens` is not
615    /// valid hex.
616    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    /// Constructs all components shared between [`build`](Self::build) and
637    /// [`build_with_pending`](Self::build_with_pending).
638    fn assemble_components(mut self) -> Result<BuiltComponents, SolverBuildError> {
639        if self.pools.is_empty() {
640            return Err(SolverBuildError::NoPools);
641        }
642
643        // Add built-in providers if none were explicitly registered.
644        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        // ComputationManager::new returns a broadcast receiver that we don't need here —
679        // workers subscribe via computation_manager.event_sender() below.
680        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        // Subscribe event channels before spawning (one for computation manager + one per pool)
688        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        // Only start price providers when the guard is enabled.
793        // When disabled, per-request attempts to enable the guard return an error.
794        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    /// Assembles and starts all solver components.
831    ///
832    /// # Errors
833    ///
834    /// Returns [`SolverBuildError`] if any component fails to initialize.
835    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    /// Assembles and starts all solver components, also returning a [`PendingBlockProcessor`]
879    /// for ephemeral bundle simulation against the live Tycho market state.
880    ///
881    /// Identical to [`build`](Self::build) except the feed task runs via
882    /// `TychoFeed::run_with_pending`. The `PendingBlockProcessor` is delivered after
883    /// token loading, before the first block is processed.
884    ///
885    /// # Errors
886    ///
887    /// Returns [`SolverBuildError`] if any component fails to initialize or the pending
888    /// channel closes before delivering the processor (e.g. token loading failed).
889    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    /// Assembles and starts all solver components, also returning a [`BlockStepController`]
951    /// that lets the caller control when each buffered block is released for processing.
952    ///
953    /// Intended for deterministic testing: call [`BlockStepController::trigger_next_block`] to
954    /// step through blocks one at a time, and [`BlockStepController::peek_next_block`] to inspect
955    /// a block before it is decoded. Dropping the controller ungates the stream so it runs to its
956    /// natural end.
957    ///
958    /// Only valid when at least one non-RFQ protocol is configured.
959    ///
960    /// # Errors
961    ///
962    /// Returns [`SolverBuildError`] if any component fails to initialize, all protocols are RFQ,
963    /// or the step-controller channel closes before the controller is delivered.
964    #[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} // impl FyndBuilder
1024
1025/// A running solver assembled by [`FyndBuilder`].
1026pub 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    /// Returns a clone of the shared market data reference.
1045    pub fn market_data(&self) -> MarketData {
1046        self.market_data.clone()
1047    }
1048
1049    /// Returns the Tycho Router contract address, or `None` on a quote-only chain.
1050    pub fn router_address(&self) -> Option<&Bytes> {
1051        self.router_address.as_ref()
1052    }
1053
1054    /// Returns a clone of the shared derived data reference.
1055    pub fn derived_data(&self) -> SharedDerivedDataRef {
1056        Arc::clone(&self.derived_data)
1057    }
1058
1059    /// Returns a new receiver for [`MarketEvent`]s broadcast by the Tycho feed.
1060    ///
1061    /// Each call returns an independent receiver. Events are broadcast on every block update.
1062    /// Receivers created after a block has been processed will miss that block's event.
1063    pub fn subscribe_market_events(&self) -> broadcast::Receiver<crate::feed::events::MarketEvent> {
1064        self.market_event_tx.subscribe()
1065    }
1066
1067    /// Submits a [`QuoteRequest`] to the worker pools and returns the best [`Quote`].
1068    ///
1069    /// # Errors
1070    ///
1071    /// Returns [`SolveError`] if all pools fail or the router timeout elapses.
1072    pub async fn quote(&self, request: QuoteRequest) -> Result<Quote, SolveError> {
1073        self.router.quote(request).await
1074    }
1075
1076    /// Waits until the solver is ready to answer quotes.
1077    ///
1078    /// Ready means:
1079    /// - The Tycho feed has delivered at least one market snapshot.
1080    /// - The computation manager has completed at least one derived-data cycle (spot prices, pool
1081    ///   depths, token gas prices).
1082    /// - Router fees have been loaded from the on-chain FeeCalculator at least once.
1083    ///
1084    /// The method polls every 500 ms and returns as soon as all conditions are
1085    /// met, or returns [`WaitReadyError`] if `timeout` elapses first.
1086    ///
1087    /// # Example
1088    ///
1089    /// ```ignore
1090    /// solver.wait_until_ready(Duration::from_secs(180)).await?;
1091    /// ```
1092    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    /// Build a Solver by replaying recorded market updates.
1123    ///
1124    /// Creates the full pipeline (feed -> derived data -> worker pools -> router)
1125    /// from pre-recorded data instead of a live Tycho connection. The returned
1126    /// Solver behaves identically to a live one — call [`wait_until_ready`](Self::wait_until_ready)
1127    /// then [`quote`](Self::quote).
1128    ///
1129    /// VM-backed protocol states that couldn't be serialized will be absent from
1130    /// the recording. Pools without states will be registered as components but
1131    /// won't contribute to routing.
1132    ///
1133    /// Requires the `test-utils` feature.
1134    #[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        // Replay updates through TychoFeed (stays pub(crate))
1148        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        // Inject gas price (recorded value or default 10 gwei)
1161        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        // Computation manager
1186        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        // Build worker pools BEFORE sending MarketUpdated
1207        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        // Encoder + router
1236        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        // Replay mode has no FeeCalculator to read; seed a zero-fee config at the standard
1245        // 8-decimal scale so the recording-based solver reports ready (integration tests do
1246        // not exercise encoding).
1247        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        // Trigger derived data computation
1260        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        // Dummy handles for feed/gas/metrics/router-fees (not running in replay mode). The market
1276        // event channel stays alive through the `market_event_tx` field on `Solver`.
1277        let feed_handle = tokio::spawn(futures::future::pending::<()>());
1278        let gas_price_handle = tokio::spawn(async { /* no-op */ });
1279        let metrics_sampler_handle = tokio::spawn(async { /* no-op */ });
1280        let router_fee_handle = tokio::spawn(async { /* no-op */ });
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    /// Signals all worker pools and the computation manager to stop, then aborts background tasks.
1301    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    /// Consumes the solver into its raw parts for callers that add their own layer.
1313    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
1332/// Raw components of a [`Solver`], for callers adding their own layer (e.g., an HTTP server).
1333///
1334/// Obtained via [`Solver::into_parts`].
1335pub struct SolverParts {
1336    /// Routes quote requests across worker pools.
1337    router: WorkerPoolRouter,
1338    /// One [`WorkerPool`] per configured algorithm pool.
1339    worker_pools: Vec<WorkerPool>,
1340    /// Live market snapshot shared across all components.
1341    market_data: MarketData,
1342    /// Derived on-chain data (spot prices, depths, gas costs) shared across all components.
1343    derived_data: SharedDerivedDataRef,
1344    /// Router fee configuration, refreshed from chain by the router-fee fetcher.
1345    router_fees: SharedRouterFees,
1346    /// Background task running the Tycho market-data feed.
1347    feed_handle: JoinHandle<()>,
1348    /// Background task polling the RPC node for gas prices.
1349    gas_price_handle: JoinHandle<()>,
1350    /// Background task exporting per-protocol market metrics.
1351    metrics_sampler_handle: JoinHandle<()>,
1352    /// Background task refreshing router fees from the on-chain FeeCalculator.
1353    router_fee_handle: JoinHandle<()>,
1354    /// Background task running the computation manager.
1355    computation_handle: JoinHandle<()>,
1356    /// Send a unit value on this channel to trigger a graceful computation-manager shutdown.
1357    computation_shutdown_tx: broadcast::Sender<()>,
1358    /// Chain this solver is configured for.
1359    chain: Chain,
1360    /// Address of the Tycho Router contract on this chain, or `None` on a quote-only chain.
1361    router_address: Option<Bytes>,
1362}
1363
1364impl SolverParts {
1365    /// Returns the chain this solver is configured for.
1366    pub fn chain(&self) -> Chain {
1367        self.chain
1368    }
1369
1370    /// Returns the Tycho Router contract address for this chain, or `None` on a quote-only chain.
1371    pub fn router_address(&self) -> Option<&Bytes> {
1372        self.router_address.as_ref()
1373    }
1374
1375    /// Returns a reference to the worker pools.
1376    pub fn worker_pools(&self) -> &[WorkerPool] {
1377        &self.worker_pools
1378    }
1379
1380    /// Returns a reference to the shared market data.
1381    pub fn market_data(&self) -> &MarketData {
1382        &self.market_data
1383    }
1384
1385    /// Returns a reference to the shared derived data.
1386    pub fn derived_data(&self) -> &SharedDerivedDataRef {
1387        &self.derived_data
1388    }
1389
1390    /// Returns a reference to the shared router fee configuration.
1391    pub fn router_fees(&self) -> &SharedRouterFees {
1392        &self.router_fees
1393    }
1394
1395    /// Consumes the parts and returns the router.
1396    pub fn into_router(self) -> WorkerPoolRouter {
1397        self.router
1398    }
1399
1400    /// Consumes the parts, returning all owned components.
1401    #[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}