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::{
52        config::WorkerPoolRouterConfig, LiquidityScope, SolverPoolHandle, WorkerPoolRouter,
53    },
54    Algorithm, Quote, QuoteRequest, SolveError,
55};
56
57/// Default values for [`FyndBuilder`] configuration and [`PoolConfig`] deserialization.
58///
59/// These are the single source of truth for all tunable defaults. Downstream
60/// crates (e.g. `fynd-rpc`) should re-export or reference these rather than
61/// redeclaring their own copies.
62pub mod defaults {
63    use std::time::Duration;
64
65    /// Minimum token quality score required for a token to be included in routing.
66    pub const MIN_TOKEN_QUALITY: i32 = 100;
67    /// Maximum age (in days) of trading history required for a token to be considered liquid.
68    pub const TRADED_N_DAYS_AGO: u64 = 3;
69    /// Multiplier applied to a component's (liquidity pool's) TVL when estimating available
70    /// liquidity.
71    pub const TVL_BUFFER_RATIO: f64 = 1.1;
72    /// How often the gas price is refreshed from the RPC node.
73    pub const GAS_REFRESH_INTERVAL: Duration = Duration::from_secs(30);
74    /// How often per-protocol market metrics are sampled and exported.
75    pub const METRICS_SAMPLE_INTERVAL: Duration = Duration::from_secs(10);
76    /// How often router fees are refreshed from the on-chain FeeCalculator contract.
77    pub const ROUTER_FEE_REFRESH_INTERVAL: Duration = Duration::from_secs(300);
78    /// Delay before reconnecting to the Tycho feed after a disconnect.
79    pub const RECONNECT_DELAY: Duration = Duration::from_secs(5);
80    /// Minimum number of solver pool responses required before returning a quote (`0` = wait for
81    /// all).
82    pub const ROUTER_MIN_RESPONSES: usize = 0;
83    /// Capacity of the task queue for each worker pool.
84    pub const POOL_TASK_QUEUE_CAPACITY: usize = 1000;
85    /// Minimum number of hops allowed in a route.
86    pub const POOL_MIN_HOPS: usize = 1;
87    /// Maximum number of hops allowed in a route.
88    pub const POOL_MAX_HOPS: usize = 3;
89    /// Per-worker-pool solve timeout in milliseconds.
90    pub const POOL_TIMEOUT_MS: u64 = 100;
91}
92
93// Internal-only defaults not shared with downstream crates.
94const DEFAULT_TYCHO_USE_TLS: bool = true;
95const DEFAULT_DEPTH_SLIPPAGE_THRESHOLD: f64 = 0.01;
96/// Generous router timeout for standalone (non-server) use. HTTP services should
97/// override this to a tighter value appropriate for their SLA.
98const DEFAULT_ROUTER_TIMEOUT: Duration = Duration::from_secs(10);
99
100// serde requires free functions for `#[serde(default = "...")]` — these delegate to the
101// defaults module so both deserialization and the builder stay in sync.
102fn default_task_queue_capacity() -> usize {
103    defaults::POOL_TASK_QUEUE_CAPACITY
104}
105
106fn default_min_hops() -> usize {
107    defaults::POOL_MIN_HOPS
108}
109
110fn default_max_hops() -> usize {
111    defaults::POOL_MAX_HOPS
112}
113
114fn default_algo_timeout_ms() -> u64 {
115    defaults::POOL_TIMEOUT_MS
116}
117
118fn parse_connector_tokens(
119    raw: Option<&[String]>,
120) -> Result<Option<HashSet<Address>>, SolverBuildError> {
121    let Some(strings) = raw else {
122        return Ok(None);
123    };
124    let mut set = HashSet::with_capacity(strings.len());
125    for s in strings {
126        let addr = Address::from_str(s).map_err(|e| AlgorithmError::InvalidConfiguration {
127            reason: format!("connector_tokens: invalid address {s:?}: {e}"),
128        })?;
129        set.insert(addr);
130    }
131    Ok(Some(set))
132}
133
134/// Configuration for one worker pool, used by [`FyndBuilder::add_pool`].
135#[must_use]
136#[derive(Debug, Clone, Serialize, Deserialize)]
137pub struct PoolConfig {
138    /// Algorithm name for this worker pool (e.g., `"most_liquid"`).
139    algorithm: String,
140    /// Number of worker threads for this worker pool.
141    #[serde(default = "num_cpus::get")]
142    num_workers: usize,
143    /// Task queue capacity for this worker pool.
144    #[serde(default = "default_task_queue_capacity")]
145    task_queue_capacity: usize,
146    /// Minimum hops to search (must be >= 1).
147    #[serde(default = "default_min_hops")]
148    min_hops: usize,
149    /// Maximum hops to search.
150    #[serde(default = "default_max_hops")]
151    max_hops: usize,
152    /// Timeout for solving in milliseconds.
153    #[serde(default = "default_algo_timeout_ms")]
154    timeout_ms: u64,
155    /// Maximum number of paths to simulate per solve. `None` simulates all scored paths.
156    #[serde(default)]
157    max_routes: Option<usize>,
158    /// Lowercase hex addresses (e.g. `"0xc02aaa…"`) allowed as intermediate routing hops.
159    /// Absent = no restriction. Typically 3–10 entries (e.g. WETH, USDC, USDT, DAI).
160    #[serde(default)]
161    connector_tokens: Option<Vec<String>>,
162    /// Which liquidity this worker pool routes through.
163    #[serde(default)]
164    liquidity_scope: Option<LiquidityScope>,
165}
166
167impl PoolConfig {
168    /// Creates a new worker pool config with the given algorithm name and defaults for all other
169    /// fields.
170    pub fn new(algorithm: impl Into<String>) -> Self {
171        Self {
172            algorithm: algorithm.into(),
173            num_workers: num_cpus::get(),
174            task_queue_capacity: defaults::POOL_TASK_QUEUE_CAPACITY,
175            min_hops: defaults::POOL_MIN_HOPS,
176            max_hops: defaults::POOL_MAX_HOPS,
177            timeout_ms: defaults::POOL_TIMEOUT_MS,
178            max_routes: None,
179            connector_tokens: None,
180            liquidity_scope: None,
181        }
182    }
183
184    /// Returns the algorithm name.
185    pub fn algorithm(&self) -> &str {
186        &self.algorithm
187    }
188
189    /// Returns the worker pool's liquidity scope.
190    pub fn liquidity_scope(&self) -> Option<LiquidityScope> {
191        self.liquidity_scope
192    }
193
194    /// Sets the worker pool's liquidity scope.
195    pub fn with_liquidity_scope(mut self, scope: LiquidityScope) -> Self {
196        self.liquidity_scope = Some(scope);
197        self
198    }
199
200    /// Returns the number of worker threads.
201    pub fn num_workers(&self) -> usize {
202        self.num_workers
203    }
204
205    /// Sets the number of worker threads.
206    pub fn with_num_workers(mut self, num_workers: usize) -> Self {
207        self.num_workers = num_workers;
208        self
209    }
210
211    /// Sets the task queue capacity.
212    pub fn with_task_queue_capacity(mut self, task_queue_capacity: usize) -> Self {
213        self.task_queue_capacity = task_queue_capacity;
214        self
215    }
216
217    /// Sets the minimum hops.
218    pub fn with_min_hops(mut self, min_hops: usize) -> Self {
219        self.min_hops = min_hops;
220        self
221    }
222
223    /// Sets the maximum hops.
224    pub fn with_max_hops(mut self, max_hops: usize) -> Self {
225        self.max_hops = max_hops;
226        self
227    }
228
229    /// Sets the timeout in milliseconds.
230    pub fn with_timeout_ms(mut self, timeout_ms: u64) -> Self {
231        self.timeout_ms = timeout_ms;
232        self
233    }
234
235    /// Sets the maximum number of routes to simulate.
236    pub fn with_max_routes(mut self, max_routes: Option<usize>) -> Self {
237        self.max_routes = max_routes;
238        self
239    }
240
241    /// Returns the task queue capacity.
242    pub fn task_queue_capacity(&self) -> usize {
243        self.task_queue_capacity
244    }
245
246    /// Returns the minimum hops.
247    pub fn min_hops(&self) -> usize {
248        self.min_hops
249    }
250
251    /// Returns the maximum hops.
252    pub fn max_hops(&self) -> usize {
253        self.max_hops
254    }
255
256    /// Returns the timeout in milliseconds.
257    pub fn timeout_ms(&self) -> u64 {
258        self.timeout_ms
259    }
260
261    /// Returns the maximum number of routes to simulate.
262    pub fn max_routes(&self) -> Option<usize> {
263        self.max_routes
264    }
265
266    /// Restricts intermediate hops to the given token addresses (hex strings with or without `0x`
267    /// prefix). Absent = no restriction.
268    pub fn with_connector_tokens(mut self, tokens: Vec<String>) -> Self {
269        self.connector_tokens = Some(tokens);
270        self
271    }
272
273    /// Returns the raw connector token address strings, or `None` if unrestricted.
274    pub fn connector_tokens(&self) -> Option<&[String]> {
275        self.connector_tokens.as_deref()
276    }
277}
278
279/// Error returned by [`Solver::wait_until_ready`].
280#[derive(Debug, thiserror::Error)]
281#[error("timed out after {timeout_ms}ms waiting for market data and derived computations")]
282pub struct WaitReadyError {
283    timeout_ms: u64,
284}
285
286/// Error returned by [`FyndBuilder::build`] and [`FyndBuilder::build_with_pending`].
287#[non_exhaustive]
288#[derive(Debug, thiserror::Error)]
289pub enum SolverBuildError {
290    /// The Ethereum RPC client could not be created (e.g. malformed URL).
291    #[error("failed to create ethereum RPC client: {0}")]
292    RpcClient(String),
293    /// An invalid algorithm configuration was supplied.
294    #[error(transparent)]
295    AlgorithmConfig(#[from] AlgorithmError),
296    /// The [`ComputationManager`] failed to initialise.
297    #[error("failed to create computation manager: {0}")]
298    ComputationManager(String),
299    /// The swap encoder could not be created for the target chain.
300    #[error("failed to create encoder: {0}")]
301    Encoder(String),
302    /// The router fee fetcher could not be created (e.g. malformed RPC URL).
303    #[error("failed to create router fee fetcher: {0}")]
304    RouterFeeFetcher(String),
305    /// A worker pool referenced an algorithm name that is not registered.
306    #[error(transparent)]
307    UnknownAlgorithm(#[from] UnknownAlgorithmError),
308    /// No native gas token is defined for the requested chain.
309    #[error("gas token not configured for chain")]
310    GasToken,
311    /// [`FyndBuilder::build`] was called without configuring any worker pools.
312    #[error("no worker pools configured")]
313    NoPools,
314    /// A recorded update failed to replay through the feed.
315    #[cfg(feature = "test-utils")]
316    #[error("replay failed: {0}")]
317    Replay(String),
318    /// The feed task failed before delivering the [`PendingBlockProcessor`].
319    ///
320    /// The inner string is the `DataFeedError` message from `TychoFeed::run_with_pending`
321    /// (e.g. "failed to load tokens: connection refused").
322    #[error("feed setup failed before delivering pending processor: {0}")]
323    FeedSetup(String),
324    /// The pending-processor oneshot closed without delivering a value, meaning the feed task
325    /// panicked rather than returning an error through the channel.
326    #[error("pending processor channel closed before processor was delivered")]
327    PendingChannelClosed,
328    /// The step-controller oneshot closed without delivering a value, meaning the feed task
329    /// panicked rather than returning an error through the channel.
330    #[cfg(feature = "experimental")]
331    #[error("step controller channel closed before controller was delivered")]
332    StepControllerChannelClosed,
333}
334
335/// Internal worker pool entry — either a built-in algorithm (by name) or a custom one.
336enum PoolEntry {
337    BuiltIn {
338        name: String,
339        algorithm: String,
340        num_workers: usize,
341        task_queue_capacity: usize,
342        min_hops: usize,
343        max_hops: usize,
344        timeout_ms: u64,
345        max_routes: Option<usize>,
346        connector_tokens: Option<HashSet<Address>>,
347        liquidity_scope: Option<LiquidityScope>,
348    },
349    Custom(CustomPoolEntry),
350}
351
352impl PoolEntry {
353    /// Returns the configured liquidity scope for this worker pool.
354    fn liquidity_scope(&self) -> Option<LiquidityScope> {
355        match self {
356            PoolEntry::BuiltIn { liquidity_scope, .. } => *liquidity_scope,
357            PoolEntry::Custom(custom) => custom.liquidity_scope,
358        }
359    }
360}
361
362/// Worker pool entry backed by a custom [`Algorithm`] implementation.
363struct CustomPoolEntry {
364    name: String,
365    num_workers: usize,
366    task_queue_capacity: usize,
367    min_hops: usize,
368    max_hops: usize,
369    timeout_ms: u64,
370    max_routes: Option<usize>,
371    liquidity_scope: Option<LiquidityScope>,
372    /// Applies the custom algorithm to a `WorkerPoolBuilder`.
373    configure: Box<dyn FnOnce(WorkerPoolBuilder) -> WorkerPoolBuilder + Send>,
374}
375
376/// All components produced by [`FyndBuilder::assemble_components`], consumed by
377/// [`FyndBuilder::build`] and [`FyndBuilder::build_with_pending`].
378struct BuiltComponents {
379    tycho_feed: TychoFeed,
380    gas_price_fetcher: GasPriceFetcher<EthereumRpcClient>,
381    router_fee_fetcher: Option<RouterFeeFetcher>,
382    computation_manager: ComputationManager,
383    computation_event_rx: broadcast::Receiver<MarketEvent>,
384    computation_shutdown_tx: broadcast::Sender<()>,
385    computation_shutdown_rx: broadcast::Receiver<()>,
386    router: WorkerPoolRouter,
387    worker_pools: Vec<WorkerPool>,
388    market_data: MarketData,
389    derived_data: SharedDerivedDataRef,
390    router_fees: SharedRouterFees,
391    chain: Chain,
392    router_address: Option<Bytes>,
393    pending_indexers: Vec<(String, Box<dyn TxDeltaIndexer>)>,
394    market_event_tx: broadcast::Sender<MarketEvent>,
395}
396
397/// Builder for assembling the full solver pipeline.
398///
399/// Configures the Tycho market-data feed, gas price fetcher, derived-data
400/// computation manager, one or more worker pools, encoder, and router.
401#[must_use = "a builder does nothing until .build() is called"]
402pub struct FyndBuilder {
403    chain: Chain,
404    tycho_url: String,
405    rpc_url: String,
406    protocols: Vec<String>,
407    min_tvl: f64,
408    tycho_api_key: Option<String>,
409    tycho_use_tls: bool,
410    min_token_quality: i32,
411    traded_n_days_ago: u64,
412    tvl_buffer_ratio: f64,
413    gas_refresh_interval: Duration,
414    reconnect_delay: Duration,
415    blocklisted_components: HashSet<String>,
416    partial_blocks: bool,
417    router_timeout: Duration,
418    router_min_responses: usize,
419    encoder: Option<Encoder>,
420    pools: Vec<PoolEntry>,
421    price_guard_enabled: bool,
422    price_providers: Vec<Box<dyn PriceProvider>>,
423    pending_indexers: Vec<(String, Box<dyn TxDeltaIndexer>)>,
424}
425
426impl FyndBuilder {
427    /// Creates a new builder with the required parameters.
428    pub fn new(
429        chain: Chain,
430        tycho_url: impl Into<String>,
431        rpc_url: impl Into<String>,
432        protocols: Vec<String>,
433        min_tvl: f64,
434    ) -> Self {
435        Self {
436            chain,
437            tycho_url: tycho_url.into(),
438            rpc_url: rpc_url.into(),
439            protocols,
440            min_tvl,
441            tycho_api_key: None,
442            tycho_use_tls: DEFAULT_TYCHO_USE_TLS,
443            min_token_quality: defaults::MIN_TOKEN_QUALITY,
444            traded_n_days_ago: defaults::TRADED_N_DAYS_AGO,
445            tvl_buffer_ratio: defaults::TVL_BUFFER_RATIO,
446            gas_refresh_interval: defaults::GAS_REFRESH_INTERVAL,
447            reconnect_delay: defaults::RECONNECT_DELAY,
448            blocklisted_components: HashSet::new(),
449            partial_blocks: false,
450            router_timeout: DEFAULT_ROUTER_TIMEOUT,
451            router_min_responses: defaults::ROUTER_MIN_RESPONSES,
452            encoder: None,
453            pools: Vec::new(),
454            price_guard_enabled: false,
455            price_providers: Vec::new(),
456            pending_indexers: Vec::new(),
457        }
458    }
459
460    /// The blockchain this builder is configured for.
461    pub fn chain(&self) -> Chain {
462        self.chain
463    }
464
465    /// Sets the Tycho API key.
466    pub fn tycho_api_key(mut self, key: impl Into<String>) -> Self {
467        self.tycho_api_key = Some(key.into());
468        self
469    }
470
471    /// Overrides the minimum TVL filter set in [`FyndBuilder::new`].
472    pub fn min_tvl(mut self, min_tvl: f64) -> Self {
473        self.min_tvl = min_tvl;
474        self
475    }
476
477    /// Enables or disables TLS for the Tycho WebSocket connection (default: `true`).
478    pub fn tycho_use_tls(mut self, use_tls: bool) -> Self {
479        self.tycho_use_tls = use_tls;
480        self
481    }
482
483    /// Sets the minimum token quality score; tokens below this threshold are excluded (default:
484    /// 100).
485    pub fn min_token_quality(mut self, quality: i32) -> Self {
486        self.min_token_quality = quality;
487        self
488    }
489
490    /// Filters out components whose last trade is older than `days` days (default: 3).
491    pub fn traded_n_days_ago(mut self, days: u64) -> Self {
492        self.traded_n_days_ago = days;
493        self
494    }
495
496    /// Multiplies reported TVL by `ratio` before applying the `min_tvl` filter (default: 1.1).
497    pub fn tvl_buffer_ratio(mut self, ratio: f64) -> Self {
498        self.tvl_buffer_ratio = ratio;
499        self
500    }
501
502    /// Sets how often the gas price is refreshed from the RPC node (default: 30 s).
503    pub fn gas_refresh_interval(mut self, interval: Duration) -> Self {
504        self.gas_refresh_interval = interval;
505        self
506    }
507
508    /// Sets the delay before reconnecting to Tycho after a disconnection (default: 5 s).
509    pub fn reconnect_delay(mut self, delay: Duration) -> Self {
510        self.reconnect_delay = delay;
511        self
512    }
513
514    /// Sets component IDs to exclude from the Tycho stream.
515    pub fn blocklisted_components(mut self, components: HashSet<String>) -> Self {
516        self.blocklisted_components = components;
517        self
518    }
519
520    /// Enables partial block (flashblock) updates from the Tycho stream (default: `false`).
521    ///
522    /// When enabled, the stream delivers component state updates mid-block rather than only at
523    /// finalization, reducing latency. Only supported for on-chain protocols; RFQ streams are
524    /// unaffected.
525    pub fn partial_blocks(mut self, enabled: bool) -> Self {
526        self.partial_blocks = enabled;
527        self
528    }
529
530    /// Sets the worker router timeout (default: 10s).
531    pub fn worker_router_timeout(mut self, timeout: Duration) -> Self {
532        self.router_timeout = timeout;
533        self
534    }
535
536    /// Sets the minimum number of solver responses before early return (default: 0).
537    pub fn worker_router_min_responses(mut self, min: usize) -> Self {
538        self.router_min_responses = min;
539        self
540    }
541
542    /// Overrides the default encoder.
543    pub fn encoder(mut self, encoder: Encoder) -> Self {
544        self.encoder = Some(encoder);
545        self
546    }
547
548    /// Shorthand: adds a single worker pool named `"default"` using a built-in algorithm by name.
549    pub fn algorithm(mut self, algorithm: impl Into<String>) -> Self {
550        self.pools.push(PoolEntry::BuiltIn {
551            name: "default".to_string(),
552            algorithm: algorithm.into(),
553            num_workers: num_cpus::get(),
554            task_queue_capacity: defaults::POOL_TASK_QUEUE_CAPACITY,
555            min_hops: defaults::POOL_MIN_HOPS,
556            max_hops: defaults::POOL_MAX_HOPS,
557            timeout_ms: defaults::POOL_TIMEOUT_MS,
558            max_routes: None,
559            connector_tokens: None,
560            liquidity_scope: None,
561        });
562        self
563    }
564
565    /// Shorthand: adds a single worker pool with a custom [`Algorithm`] implementation.
566    ///
567    /// The `factory` closure is called once per worker thread.
568    pub fn with_algorithm<A, F>(mut self, name: impl Into<String>, factory: F) -> Self
569    where
570        A: Algorithm + 'static,
571        A::GraphManager: MarketEventHandler + EdgeWeightUpdaterWithDerived + 'static,
572        F: Fn(AlgorithmConfig) -> A + Clone + Send + Sync + 'static,
573    {
574        let name = name.into();
575        let algo_name = name.clone();
576        let configure =
577            Box::new(move |builder: WorkerPoolBuilder| builder.with_algorithm(algo_name, factory));
578        self.pools
579            .push(PoolEntry::Custom(CustomPoolEntry {
580                name,
581                num_workers: num_cpus::get(),
582                task_queue_capacity: defaults::POOL_TASK_QUEUE_CAPACITY,
583                min_hops: defaults::POOL_MIN_HOPS,
584                max_hops: defaults::POOL_MAX_HOPS,
585                timeout_ms: defaults::POOL_TIMEOUT_MS,
586                max_routes: None,
587                liquidity_scope: None,
588                configure,
589            }));
590        self
591    }
592
593    /// Registers the built-in price providers (Hyperliquid + Binance).
594    ///
595    /// Called automatically during [`build`](Self::build) if no providers have been
596    /// registered and the price guard is not disabled. To use only custom
597    /// providers, call [`register_price_provider`](Self::register_price_provider)
598    /// before `build()` and the defaults will be skipped.
599    pub fn add_default_price_providers(self) -> Self {
600        self.register_price_provider(Box::new(
601            crate::price_guard::hyperliquid::HyperliquidProvider::default(),
602        ))
603        .register_price_provider(Box::new(
604            crate::price_guard::binance_ws::BinanceWsProvider::default(),
605        ))
606    }
607
608    /// Registers a custom price provider for the price guard.
609    ///
610    /// The provider's [`start`](PriceProvider::start) method is called during
611    /// [`build`](Self::build) with the shared market data.
612    pub fn register_price_provider(mut self, provider: Box<dyn PriceProvider>) -> Self {
613        self.price_providers.push(provider);
614        self
615    }
616
617    /// Registers a [`TxDeltaIndexer`] for ephemeral pending-block simulation.
618    ///
619    /// `extractor` is the protocol synchronizer name (e.g. `"uniswap_v3"`). Only has effect
620    /// when calling [`build_with_pending`](Self::build_with_pending). VM protocols (prefix
621    /// `"vm:"`) are rejected by the underlying stream builder at build time.
622    pub fn with_pending_indexer(
623        mut self,
624        extractor: impl Into<String>,
625        indexer: Box<dyn TxDeltaIndexer>,
626    ) -> Self {
627        self.pending_indexers
628            .push((extractor.into(), indexer));
629        self
630    }
631
632    /// Enables or disables the price guard.
633    ///
634    /// When enabled, providers are started and caches stay warm. Validation
635    /// only runs for requests where the client sets `enabled: true` in
636    /// `PriceGuardConfig`. When disabled, no providers are started and
637    /// per-request attempts to use the guard return an error.
638    pub fn price_guard_enabled(mut self, enabled: bool) -> Self {
639        self.price_guard_enabled = enabled;
640        self
641    }
642
643    /// Adds a named worker pool using the given [`PoolConfig`].
644    ///
645    /// # Errors
646    ///
647    /// Returns [`SolverBuildError::AlgorithmConfig`] if any address in `connector_tokens` is not
648    /// valid hex.
649    pub fn add_pool(
650        mut self,
651        name: impl Into<String>,
652        config: &PoolConfig,
653    ) -> Result<Self, SolverBuildError> {
654        let connector_tokens = parse_connector_tokens(config.connector_tokens())?;
655        self.pools.push(PoolEntry::BuiltIn {
656            name: name.into(),
657            algorithm: config.algorithm().to_string(),
658            num_workers: config.num_workers(),
659            task_queue_capacity: config.task_queue_capacity(),
660            min_hops: config.min_hops(),
661            max_hops: config.max_hops(),
662            timeout_ms: config.timeout_ms(),
663            max_routes: config.max_routes(),
664            connector_tokens,
665            liquidity_scope: config.liquidity_scope(),
666        });
667        Ok(self)
668    }
669
670    /// Constructs all components shared between [`build`](Self::build) and
671    /// [`build_with_pending`](Self::build_with_pending).
672    fn assemble_components(mut self) -> Result<BuiltComponents, SolverBuildError> {
673        if self.pools.is_empty() {
674            return Err(SolverBuildError::NoPools);
675        }
676
677        // Add built-in providers if none were explicitly registered.
678        if self.price_providers.is_empty() {
679            self = self.add_default_price_providers();
680        }
681
682        let market_data = MarketData::new_shared();
683
684        let tycho_feed_config = TychoFeedConfig::new(
685            self.tycho_url,
686            self.chain,
687            self.tycho_api_key,
688            self.tycho_use_tls,
689            self.protocols,
690            self.min_tvl,
691        )
692        .tvl_buffer_ratio(self.tvl_buffer_ratio)
693        .reconnect_delay(self.reconnect_delay)
694        .min_token_quality(self.min_token_quality)
695        .traded_n_days_ago(self.traded_n_days_ago)
696        .blocklisted_components(self.blocklisted_components)
697        .partial_blocks(self.partial_blocks);
698
699        let ethereum_client = EthereumRpcClient::new(self.rpc_url.as_str())
700            .map_err(|e| SolverBuildError::RpcClient(e.to_string()))?;
701
702        let gas_price_fetcher =
703            GasPriceFetcher::new(ethereum_client, market_data.clone(), self.gas_refresh_interval);
704
705        let tycho_feed = TychoFeed::new(tycho_feed_config, market_data.clone());
706        let market_event_tx = tycho_feed.event_sender();
707
708        let gas_token = native_token(&self.chain).map_err(|_| SolverBuildError::GasToken)?;
709        let computation_config = ComputationManagerConfig::new()
710            .with_gas_token(gas_token)
711            .with_depth_slippage_threshold(DEFAULT_DEPTH_SLIPPAGE_THRESHOLD);
712        // ComputationManager::new returns a broadcast receiver that we don't need here —
713        // workers subscribe via computation_manager.event_sender() below.
714        let (computation_manager, _) =
715            ComputationManager::new(computation_config, market_data.clone())
716                .map_err(|e| SolverBuildError::ComputationManager(e.to_string()))?;
717
718        let derived_data: SharedDerivedDataRef = computation_manager.store();
719        let derived_event_tx = computation_manager.event_sender();
720
721        // Subscribe event channels before spawning (one for computation manager + one per worker
722        // pool)
723        let computation_event_rx = tycho_feed.subscribe();
724        let (computation_shutdown_tx, computation_shutdown_rx) = broadcast::channel(1);
725
726        let mut solver_pool_handles: Vec<SolverPoolHandle> = Vec::new();
727        let mut worker_pools: Vec<WorkerPool> = Vec::new();
728
729        let pools = std::mem::take(&mut self.pools);
730
731        for pool_entry in pools {
732            let pool_event_rx = tycho_feed.subscribe();
733            let derived_rx = derived_event_tx.subscribe();
734
735            let pool_scope = pool_entry
736                .liquidity_scope()
737                .unwrap_or_default();
738
739            let (worker_pool, task_handle) = match pool_entry {
740                PoolEntry::BuiltIn {
741                    name,
742                    algorithm,
743                    num_workers,
744                    task_queue_capacity,
745                    min_hops,
746                    max_hops,
747                    timeout_ms,
748                    max_routes,
749                    connector_tokens,
750                    liquidity_scope: _,
751                } => {
752                    let mut algo_cfg = AlgorithmConfig::new(
753                        min_hops,
754                        max_hops,
755                        Duration::from_millis(timeout_ms),
756                        max_routes,
757                    )?;
758                    if let Some(tokens) = connector_tokens {
759                        algo_cfg = algo_cfg.with_connector_tokens(tokens);
760                    }
761                    let builder = WorkerPoolBuilder::new()
762                        .name(name)
763                        .algorithm(algorithm)
764                        .algorithm_config(algo_cfg)
765                        .num_workers(num_workers)
766                        .task_queue_capacity(task_queue_capacity)
767                        .liquidity_scope(pool_scope);
768                    builder.build(
769                        market_data.clone(),
770                        Arc::clone(&derived_data),
771                        pool_event_rx,
772                        derived_rx,
773                    )?
774                }
775                PoolEntry::Custom(custom) => {
776                    let algo_cfg = AlgorithmConfig::new(
777                        custom.min_hops,
778                        custom.max_hops,
779                        Duration::from_millis(custom.timeout_ms),
780                        custom.max_routes,
781                    )?;
782                    let builder = WorkerPoolBuilder::new()
783                        .name(custom.name)
784                        .algorithm_config(algo_cfg)
785                        .num_workers(custom.num_workers)
786                        .task_queue_capacity(custom.task_queue_capacity)
787                        .liquidity_scope(pool_scope);
788                    let builder = (custom.configure)(builder);
789                    builder.build(
790                        market_data.clone(),
791                        Arc::clone(&derived_data),
792                        pool_event_rx,
793                        derived_rx,
794                    )?
795                }
796            };
797
798            solver_pool_handles.push(
799                SolverPoolHandle::new(worker_pool.name(), task_handle)
800                    .with_liquidity_scope(pool_scope),
801            );
802            worker_pools.push(worker_pool);
803        }
804
805        let encoder = match self.encoder {
806            Some(enc) => enc,
807            None => {
808                let registry = SwapEncoderRegistry::new(self.chain)
809                    .add_default_encoders(None)
810                    .map_err(|e| SolverBuildError::Encoder(e.to_string()))?;
811                Encoder::new(self.chain, registry)
812                    .map_err(|e| SolverBuildError::Encoder(e.to_string()))?
813            }
814        };
815
816        let chain = self.chain;
817        let router_address = encoder.router_address().cloned();
818        let router_fees = encoder.router_fees();
819
820        let router_fee_fetcher = match &router_address {
821            Some(addr) => Some(
822                RouterFeeFetcher::new(
823                    self.rpc_url.as_str(),
824                    addr,
825                    router_fees.clone(),
826                    defaults::ROUTER_FEE_REFRESH_INTERVAL,
827                )
828                .map_err(|e| SolverBuildError::RouterFeeFetcher(e.to_string()))?,
829            ),
830            None => {
831                tracing::warn!(
832                    %chain,
833                    "no Tycho router for this chain; running quote-only (encoding disabled)"
834                );
835                None
836            }
837        };
838
839        // Only start price providers when the guard is enabled.
840        // When disabled, per-request attempts to enable the guard return an error.
841        let router_config = WorkerPoolRouterConfig::default()
842            .with_timeout(self.router_timeout)
843            .with_min_responses(self.router_min_responses);
844        let mut router = WorkerPoolRouter::new(solver_pool_handles, router_config, encoder);
845
846        if self.price_guard_enabled {
847            let mut registry = PriceProviderRegistry::new();
848            let mut worker_handles = Vec::new();
849            for mut provider in self.price_providers {
850                worker_handles.push(provider.start(market_data.clone()));
851                registry = registry.register(provider);
852            }
853            let price_guard = PriceGuard::new(registry, worker_handles);
854            router = router.with_price_guard(price_guard);
855        }
856
857        Ok(BuiltComponents {
858            tycho_feed,
859            gas_price_fetcher,
860            router_fee_fetcher,
861            computation_manager,
862            computation_event_rx,
863            computation_shutdown_tx,
864            computation_shutdown_rx,
865            router,
866            worker_pools,
867            market_data,
868            derived_data,
869            router_fees,
870            chain,
871            router_address,
872            pending_indexers: self.pending_indexers,
873            market_event_tx,
874        })
875    }
876
877    /// Assembles and starts all solver components.
878    ///
879    /// # Errors
880    ///
881    /// Returns [`SolverBuildError`] if any component fails to initialize.
882    pub fn build(self) -> Result<Solver, SolverBuildError> {
883        let mut c = self.assemble_components()?;
884
885        let feed_handle = tokio::spawn(async move {
886            if let Err(e) = c.tycho_feed.run().await {
887                metrics::counter!("tycho_feed_failures_total").increment(1);
888                tracing::error!(error = %e, "tycho feed error");
889            }
890        });
891        let gas_price_handle = tokio::spawn(async move {
892            c.gas_price_fetcher.run().await;
893        });
894        let metrics_sampler =
895            MetricsSampler::new(c.market_data.clone(), defaults::METRICS_SAMPLE_INTERVAL);
896        let metrics_sampler_handle = tokio::spawn(async move { metrics_sampler.run().await });
897        let router_fee_handle = match c.router_fee_fetcher {
898            Some(fetcher) => tokio::spawn(async move { fetcher.run().await }),
899            None => tokio::spawn(async {}),
900        };
901        let computation_handle = tokio::spawn(async move {
902            c.computation_manager
903                .run(c.computation_event_rx, c.computation_shutdown_rx)
904                .await;
905        });
906
907        Ok(Solver {
908            router: c.router,
909            worker_pools: c.worker_pools,
910            market_data: c.market_data,
911            derived_data: c.derived_data,
912            router_fees: c.router_fees,
913            feed_handle,
914            gas_price_handle,
915            metrics_sampler_handle,
916            router_fee_handle,
917            computation_handle,
918            computation_shutdown_tx: c.computation_shutdown_tx,
919            chain: c.chain,
920            router_address: c.router_address,
921            market_event_tx: c.market_event_tx,
922        })
923    }
924
925    /// Assembles and starts all solver components, also returning a [`PendingBlockProcessor`]
926    /// for ephemeral bundle simulation against the live Tycho market state.
927    ///
928    /// Identical to [`build`](Self::build) except the feed task runs via
929    /// `TychoFeed::run_with_pending`. The `PendingBlockProcessor` is delivered after
930    /// token loading, before the first block is processed.
931    ///
932    /// # Errors
933    ///
934    /// Returns [`SolverBuildError`] if any component fails to initialize or the pending
935    /// channel closes before delivering the processor (e.g. token loading failed).
936    pub async fn build_with_pending(
937        self,
938    ) -> Result<(Solver, PendingBlockProcessor), SolverBuildError> {
939        let mut c = self.assemble_components()?;
940
941        let (pending_tx, pending_rx) =
942            tokio::sync::oneshot::channel::<Result<PendingBlockProcessor, String>>();
943
944        let pending_indexers = c.pending_indexers;
945        let feed_handle = tokio::spawn(async move {
946            if let Err(e) = c
947                .tycho_feed
948                .run_with_pending(pending_tx, pending_indexers)
949                .await
950            {
951                metrics::counter!("tycho_feed_failures_total").increment(1);
952                tracing::error!(error = %e, "tycho feed error");
953            }
954        });
955        let gas_price_handle = tokio::spawn(async move {
956            c.gas_price_fetcher.run().await;
957        });
958        let metrics_sampler =
959            MetricsSampler::new(c.market_data.clone(), defaults::METRICS_SAMPLE_INTERVAL);
960        let metrics_sampler_handle = tokio::spawn(async move { metrics_sampler.run().await });
961        let router_fee_handle = match c.router_fee_fetcher {
962            Some(fetcher) => tokio::spawn(async move { fetcher.run().await }),
963            None => tokio::spawn(async {}),
964        };
965        let computation_handle = tokio::spawn(async move {
966            c.computation_manager
967                .run(c.computation_event_rx, c.computation_shutdown_rx)
968                .await;
969        });
970
971        let pending = pending_rx
972            .await
973            .map_err(|_| SolverBuildError::PendingChannelClosed)?
974            .map_err(SolverBuildError::FeedSetup)?;
975
976        Ok((
977            Solver {
978                router: c.router,
979                worker_pools: c.worker_pools,
980                market_data: c.market_data,
981                derived_data: c.derived_data,
982                router_fees: c.router_fees,
983                feed_handle,
984                gas_price_handle,
985                metrics_sampler_handle,
986                router_fee_handle,
987                computation_handle,
988                computation_shutdown_tx: c.computation_shutdown_tx,
989                chain: c.chain,
990                router_address: c.router_address,
991                market_event_tx: c.market_event_tx,
992            },
993            pending,
994        ))
995    }
996
997    /// Assembles and starts all solver components, also returning a [`BlockStepController`]
998    /// that lets the caller control when each buffered block is released for processing.
999    ///
1000    /// Intended for deterministic testing: call [`BlockStepController::trigger_next_block`] to
1001    /// step through blocks one at a time, and [`BlockStepController::peek_next_block`] to inspect
1002    /// a block before it is decoded. Dropping the controller ungates the stream so it runs to its
1003    /// natural end.
1004    ///
1005    /// Only valid when at least one non-RFQ protocol is configured.
1006    ///
1007    /// # Errors
1008    ///
1009    /// Returns [`SolverBuildError`] if any component fails to initialize, all protocols are RFQ,
1010    /// or the step-controller channel closes before the controller is delivered.
1011    #[cfg(feature = "experimental")]
1012    pub async fn build_with_step_controller(
1013        self,
1014    ) -> Result<(Solver, BlockStepController), SolverBuildError> {
1015        let mut c = self.assemble_components()?;
1016
1017        let (controller_tx, controller_rx) =
1018            tokio::sync::oneshot::channel::<Result<BlockStepController, String>>();
1019
1020        let feed_handle = tokio::spawn(async move {
1021            if let Err(e) = c
1022                .tycho_feed
1023                .run_with_step_controller(controller_tx)
1024                .await
1025            {
1026                tracing::error!(error = %e, "tycho feed error");
1027            }
1028        });
1029        let gas_price_handle = tokio::spawn(async move {
1030            c.gas_price_fetcher.run().await;
1031        });
1032        let metrics_sampler =
1033            MetricsSampler::new(c.market_data.clone(), defaults::METRICS_SAMPLE_INTERVAL);
1034        let metrics_sampler_handle = tokio::spawn(async move { metrics_sampler.run().await });
1035        let router_fee_handle = match c.router_fee_fetcher {
1036            Some(fetcher) => tokio::spawn(async move { fetcher.run().await }),
1037            None => tokio::spawn(async {}),
1038        };
1039        let computation_handle = tokio::spawn(async move {
1040            c.computation_manager
1041                .run(c.computation_event_rx, c.computation_shutdown_rx)
1042                .await;
1043        });
1044
1045        let controller = controller_rx
1046            .await
1047            .map_err(|_| SolverBuildError::StepControllerChannelClosed)?
1048            .map_err(SolverBuildError::FeedSetup)?;
1049
1050        Ok((
1051            Solver {
1052                router: c.router,
1053                worker_pools: c.worker_pools,
1054                market_data: c.market_data,
1055                derived_data: c.derived_data,
1056                router_fees: c.router_fees,
1057                feed_handle,
1058                gas_price_handle,
1059                metrics_sampler_handle,
1060                router_fee_handle,
1061                computation_handle,
1062                computation_shutdown_tx: c.computation_shutdown_tx,
1063                chain: c.chain,
1064                router_address: c.router_address,
1065                market_event_tx: c.market_event_tx,
1066            },
1067            controller,
1068        ))
1069    }
1070} // impl FyndBuilder
1071
1072/// A running solver assembled by [`FyndBuilder`].
1073pub struct Solver {
1074    router: WorkerPoolRouter,
1075    worker_pools: Vec<WorkerPool>,
1076    market_data: MarketData,
1077    derived_data: SharedDerivedDataRef,
1078    router_fees: SharedRouterFees,
1079    feed_handle: JoinHandle<()>,
1080    gas_price_handle: JoinHandle<()>,
1081    metrics_sampler_handle: JoinHandle<()>,
1082    router_fee_handle: JoinHandle<()>,
1083    computation_handle: JoinHandle<()>,
1084    computation_shutdown_tx: broadcast::Sender<()>,
1085    chain: Chain,
1086    router_address: Option<Bytes>,
1087    market_event_tx: broadcast::Sender<MarketEvent>,
1088}
1089
1090impl Solver {
1091    /// Returns a clone of the shared market data reference.
1092    pub fn market_data(&self) -> MarketData {
1093        self.market_data.clone()
1094    }
1095
1096    /// Returns the Tycho Router contract address, or `None` on a quote-only chain.
1097    pub fn router_address(&self) -> Option<&Bytes> {
1098        self.router_address.as_ref()
1099    }
1100
1101    /// Returns a clone of the shared derived data reference.
1102    pub fn derived_data(&self) -> SharedDerivedDataRef {
1103        Arc::clone(&self.derived_data)
1104    }
1105
1106    /// Returns a new receiver for [`MarketEvent`]s broadcast by the Tycho feed.
1107    ///
1108    /// Each call returns an independent receiver. Events are broadcast on every block update.
1109    /// Receivers created after a block has been processed will miss that block's event.
1110    pub fn subscribe_market_events(&self) -> broadcast::Receiver<crate::feed::events::MarketEvent> {
1111        self.market_event_tx.subscribe()
1112    }
1113
1114    /// Submits a [`QuoteRequest`] to the worker pools and returns the best [`Quote`].
1115    ///
1116    /// # Errors
1117    ///
1118    /// Returns [`SolveError`] if all worker pools fail or the router timeout elapses.
1119    pub async fn quote(&self, request: QuoteRequest) -> Result<Quote, SolveError> {
1120        self.router.quote(request).await
1121    }
1122
1123    /// Waits until the solver is ready to answer quotes.
1124    ///
1125    /// Ready means:
1126    /// - The Tycho feed has delivered at least one market snapshot.
1127    /// - The computation manager has completed at least one derived-data cycle (spot prices,
1128    ///   component depths, token gas prices).
1129    /// - Router fees have been loaded from the on-chain FeeCalculator at least once.
1130    ///
1131    /// The method polls every 500 ms and returns as soon as all conditions are
1132    /// met, or returns [`WaitReadyError`] if `timeout` elapses first.
1133    ///
1134    /// # Example
1135    ///
1136    /// ```ignore
1137    /// solver.wait_until_ready(Duration::from_secs(180)).await?;
1138    /// ```
1139    pub async fn wait_until_ready(&self, timeout: Duration) -> Result<(), WaitReadyError> {
1140        const POLL_INTERVAL: Duration = Duration::from_millis(500);
1141
1142        let deadline = tokio::time::Instant::now() + timeout;
1143
1144        loop {
1145            let market_ready = self
1146                .market_data
1147                .read()
1148                .await
1149                .last_updated()
1150                .is_some();
1151            let derived_ready = self
1152                .derived_data
1153                .read()
1154                .await
1155                .derived_data_ready();
1156
1157            if market_ready && derived_ready {
1158                return Ok(());
1159            }
1160
1161            if tokio::time::Instant::now() >= deadline {
1162                return Err(WaitReadyError { timeout_ms: timeout.as_millis() as u64 });
1163            }
1164
1165            tokio::time::sleep(POLL_INTERVAL).await;
1166        }
1167    }
1168
1169    /// Build a Solver by replaying recorded market updates.
1170    ///
1171    /// Creates the full pipeline (feed -> derived data -> worker pools -> router)
1172    /// from pre-recorded data instead of a live Tycho connection. The returned
1173    /// Solver behaves identically to a live one — call [`wait_until_ready`](Self::wait_until_ready)
1174    /// then [`quote`](Self::quote).
1175    ///
1176    /// VM-backed protocol states that couldn't be serialized will be absent from
1177    /// the recording. Components without states will still be registered but
1178    /// won't contribute to routing.
1179    ///
1180    /// Requires the `test-utils` feature.
1181    #[cfg(feature = "test-utils")]
1182    pub async fn from_recording(
1183        chain: Chain,
1184        updates: Vec<tycho_simulation::protocol::models::Update>,
1185        pools: std::collections::HashMap<String, PoolConfig>,
1186        gas_price_wei: Option<num_bigint::BigUint>,
1187    ) -> Result<Self, SolverBuildError> {
1188        if pools.is_empty() {
1189            return Err(SolverBuildError::NoPools);
1190        }
1191
1192        let market_data = MarketData::new_shared();
1193
1194        // Replay updates through TychoFeed (stays pub(crate))
1195        let feed_config =
1196            TychoFeedConfig::new("ws://replay".to_string(), chain, None, false, vec![], 0.0);
1197        let feed = TychoFeed::new(feed_config, market_data.clone());
1198        let market_event_tx = feed.event_sender();
1199        let _feed_rx = feed.subscribe();
1200
1201        for update in updates {
1202            feed.handle_tycho_message(update)
1203                .await
1204                .map_err(|e| SolverBuildError::Replay(e.to_string()))?;
1205        }
1206
1207        // Inject gas price (recorded value or default 10 gwei)
1208        let gas_price = match gas_price_wei {
1209            Some(price) => price,
1210            None => {
1211                tracing::warn!("no recorded gas price, defaulting to 10 gwei");
1212                num_bigint::BigUint::from(10_000_000_000u64)
1213            }
1214        };
1215        let block_number = match market_data.read().await.last_updated() {
1216            Some(block) => block.number(),
1217            None => {
1218                tracing::warn!("no block number from replayed updates, defaulting to 0");
1219                0
1220            }
1221        };
1222        {
1223            let mut market = market_data.write().await;
1224            market.update_gas_price(BlockGasPrice {
1225                block_number,
1226                block_hash: Default::default(),
1227                block_timestamp: 0,
1228                pricing: GasPrice::Legacy { gas_price },
1229            });
1230        }
1231
1232        // Computation manager
1233        let gas_token = native_token(&chain).map_err(|_| SolverBuildError::GasToken)?;
1234        let computation_config = ComputationManagerConfig::new()
1235            .with_gas_token(gas_token)
1236            .with_depth_slippage_threshold(DEFAULT_DEPTH_SLIPPAGE_THRESHOLD);
1237        let (computation_manager, _) =
1238            ComputationManager::new(computation_config, market_data.clone())
1239                .map_err(|e| SolverBuildError::ComputationManager(e.to_string()))?;
1240
1241        let derived_data: SharedDerivedDataRef = computation_manager.store();
1242        let derived_event_tx = computation_manager.event_sender();
1243
1244        let computation_event_rx = feed.subscribe();
1245        let (computation_shutdown_tx, computation_shutdown_rx) = broadcast::channel(1);
1246
1247        let computation_handle = tokio::spawn(async move {
1248            computation_manager
1249                .run(computation_event_rx, computation_shutdown_rx)
1250                .await;
1251        });
1252
1253        // Build worker pools BEFORE sending MarketUpdated
1254        let mut solver_pool_handles: Vec<SolverPoolHandle> = Vec::new();
1255        let mut worker_pools: Vec<WorkerPool> = Vec::new();
1256        let mut max_timeout_ms = 0u64;
1257
1258        for (name, pool_cfg) in &pools {
1259            let algo_cfg = AlgorithmConfig::new(
1260                pool_cfg.min_hops(),
1261                pool_cfg.max_hops(),
1262                Duration::from_millis(pool_cfg.timeout_ms()),
1263                pool_cfg.max_routes(),
1264            )?;
1265
1266            let pool_event_rx = feed.subscribe();
1267            let derived_rx = derived_event_tx.subscribe();
1268
1269            let (worker_pool, task_handle) = WorkerPoolBuilder::new()
1270                .name(name.clone())
1271                .algorithm(pool_cfg.algorithm().to_string())
1272                .algorithm_config(algo_cfg)
1273                .num_workers(pool_cfg.num_workers())
1274                .task_queue_capacity(pool_cfg.task_queue_capacity())
1275                .build(market_data.clone(), Arc::clone(&derived_data), pool_event_rx, derived_rx)?;
1276
1277            solver_pool_handles.push(SolverPoolHandle::new(worker_pool.name(), task_handle));
1278            max_timeout_ms = max_timeout_ms.max(pool_cfg.timeout_ms());
1279            worker_pools.push(worker_pool);
1280        }
1281
1282        // Encoder + router
1283        let encoder = {
1284            let registry = SwapEncoderRegistry::new(chain)
1285                .add_default_encoders(None)
1286                .map_err(|e| SolverBuildError::Encoder(e.to_string()))?;
1287            Encoder::new(chain, registry).map_err(|e| SolverBuildError::Encoder(e.to_string()))?
1288        };
1289
1290        let router_address = encoder.router_address().cloned();
1291        // Replay mode has no FeeCalculator to read; seed a zero-fee config at the standard
1292        // 8-decimal scale so the recording-based solver reports ready (integration tests do
1293        // not exercise encoding).
1294        let router_fees = encoder.router_fees();
1295        router_fees.set(crate::encoding::router_fees::RouterFees::new(
1296            100_000_000,
1297            0,
1298            0,
1299            std::collections::HashMap::new(),
1300        ));
1301        let router_config = WorkerPoolRouterConfig::default()
1302            .with_timeout(Duration::from_millis(max_timeout_ms.max(5000)))
1303            .with_min_responses(defaults::ROUTER_MIN_RESPONSES);
1304        let router = WorkerPoolRouter::new(solver_pool_handles, router_config, encoder);
1305
1306        // Trigger derived data computation
1307        let market_read = market_data.read().await;
1308        let added = market_read.component_topology();
1309        drop(market_read);
1310
1311        if market_event_tx
1312            .send(MarketEvent::MarketUpdated {
1313                added_components: added,
1314                removed_components: vec![],
1315                updated_components: vec![],
1316            })
1317            .is_err()
1318        {
1319            tracing::warn!("no receivers for initial MarketUpdated broadcast");
1320        }
1321
1322        // Dummy handles for feed/gas/metrics/router-fees (not running in replay mode). The market
1323        // event channel stays alive through the `market_event_tx` field on `Solver`.
1324        let feed_handle = tokio::spawn(futures::future::pending::<()>());
1325        let gas_price_handle = tokio::spawn(async { /* no-op */ });
1326        let metrics_sampler_handle = tokio::spawn(async { /* no-op */ });
1327        let router_fee_handle = tokio::spawn(async { /* no-op */ });
1328
1329        Ok(Solver {
1330            router,
1331            worker_pools,
1332            market_data,
1333            derived_data,
1334            router_fees,
1335            feed_handle,
1336            gas_price_handle,
1337            metrics_sampler_handle,
1338            router_fee_handle,
1339            computation_handle,
1340            computation_shutdown_tx,
1341            chain,
1342            router_address,
1343            market_event_tx,
1344        })
1345    }
1346
1347    /// Signals all worker pools and the computation manager to stop, then aborts background tasks.
1348    pub fn shutdown(self) {
1349        let _ = self.computation_shutdown_tx.send(());
1350        for pool in self.worker_pools {
1351            pool.shutdown();
1352        }
1353        self.feed_handle.abort();
1354        self.gas_price_handle.abort();
1355        self.metrics_sampler_handle.abort();
1356        self.router_fee_handle.abort();
1357    }
1358
1359    /// Consumes the solver into its raw parts for callers that add their own layer.
1360    pub fn into_parts(self) -> SolverParts {
1361        SolverParts {
1362            router: self.router,
1363            worker_pools: self.worker_pools,
1364            market_data: self.market_data,
1365            derived_data: self.derived_data,
1366            router_fees: self.router_fees,
1367            feed_handle: self.feed_handle,
1368            gas_price_handle: self.gas_price_handle,
1369            metrics_sampler_handle: self.metrics_sampler_handle,
1370            router_fee_handle: self.router_fee_handle,
1371            computation_handle: self.computation_handle,
1372            computation_shutdown_tx: self.computation_shutdown_tx,
1373            chain: self.chain,
1374            router_address: self.router_address,
1375        }
1376    }
1377}
1378
1379/// Raw components of a [`Solver`], for callers adding their own layer (e.g., an HTTP server).
1380///
1381/// Obtained via [`Solver::into_parts`].
1382pub struct SolverParts {
1383    /// Routes quote requests across worker pools.
1384    router: WorkerPoolRouter,
1385    /// One [`WorkerPool`] per entry configured via [`FyndBuilder::add_pool`].
1386    worker_pools: Vec<WorkerPool>,
1387    /// Live market snapshot shared across all components.
1388    market_data: MarketData,
1389    /// Derived on-chain data (spot prices, depths, gas costs) shared across all components.
1390    derived_data: SharedDerivedDataRef,
1391    /// Router fee configuration, refreshed from chain by the router-fee fetcher.
1392    router_fees: SharedRouterFees,
1393    /// Background task running the Tycho market-data feed.
1394    feed_handle: JoinHandle<()>,
1395    /// Background task polling the RPC node for gas prices.
1396    gas_price_handle: JoinHandle<()>,
1397    /// Background task exporting per-protocol market metrics.
1398    metrics_sampler_handle: JoinHandle<()>,
1399    /// Background task refreshing router fees from the on-chain FeeCalculator.
1400    router_fee_handle: JoinHandle<()>,
1401    /// Background task running the computation manager.
1402    computation_handle: JoinHandle<()>,
1403    /// Send a unit value on this channel to trigger a graceful computation-manager shutdown.
1404    computation_shutdown_tx: broadcast::Sender<()>,
1405    /// Chain this solver is configured for.
1406    chain: Chain,
1407    /// Address of the Tycho Router contract on this chain, or `None` on a quote-only chain.
1408    router_address: Option<Bytes>,
1409}
1410
1411impl SolverParts {
1412    /// Returns the chain this solver is configured for.
1413    pub fn chain(&self) -> Chain {
1414        self.chain
1415    }
1416
1417    /// Returns the Tycho Router contract address for this chain, or `None` on a quote-only chain.
1418    pub fn router_address(&self) -> Option<&Bytes> {
1419        self.router_address.as_ref()
1420    }
1421
1422    /// Returns a reference to the worker pools.
1423    pub fn worker_pools(&self) -> &[WorkerPool] {
1424        &self.worker_pools
1425    }
1426
1427    /// Returns a reference to the shared market data.
1428    pub fn market_data(&self) -> &MarketData {
1429        &self.market_data
1430    }
1431
1432    /// Returns a reference to the shared derived data.
1433    pub fn derived_data(&self) -> &SharedDerivedDataRef {
1434        &self.derived_data
1435    }
1436
1437    /// Returns a reference to the shared router fee configuration.
1438    pub fn router_fees(&self) -> &SharedRouterFees {
1439        &self.router_fees
1440    }
1441
1442    /// Consumes the parts and returns the router.
1443    pub fn into_router(self) -> WorkerPoolRouter {
1444        self.router
1445    }
1446
1447    /// Consumes the parts, returning all owned components.
1448    #[allow(clippy::type_complexity)]
1449    pub fn into_components(
1450        self,
1451    ) -> (
1452        WorkerPoolRouter,
1453        Vec<WorkerPool>,
1454        MarketData,
1455        SharedDerivedDataRef,
1456        JoinHandle<()>,
1457        JoinHandle<()>,
1458        JoinHandle<()>,
1459        JoinHandle<()>,
1460        JoinHandle<()>,
1461        broadcast::Sender<()>,
1462    ) {
1463        (
1464            self.router,
1465            self.worker_pools,
1466            self.market_data,
1467            self.derived_data,
1468            self.feed_handle,
1469            self.gas_price_handle,
1470            self.metrics_sampler_handle,
1471            self.router_fee_handle,
1472            self.computation_handle,
1473            self.computation_shutdown_tx,
1474        )
1475    }
1476}
1477
1478#[cfg(test)]
1479mod tests {
1480    use super::*;
1481
1482    /// An unscoped pool resolves to `PublicOnly` — exclusive components are filtered out unless
1483    /// a pool explicitly opts in with `IncludeExclusive`.
1484    #[test]
1485    fn test_unscoped_pool_resolves_to_public_only() {
1486        let config = PoolConfig::new("most_liquid");
1487        assert_eq!(config.liquidity_scope(), None);
1488        assert_eq!(
1489            config
1490                .liquidity_scope()
1491                .unwrap_or_default(),
1492            LiquidityScope::PublicOnly
1493        );
1494    }
1495}