From d6160a56c6a79725e403e1adc72a4972a8fb641c Mon Sep 17 00:00:00 2001 From: "tuddman@users.noreply.github.com" Date: Wed, 19 Aug 2026 21:41:32 +0200 Subject: [PATCH 1/2] feat: register Flare DEX forks in the protocol registry BlazeSwap V2 pools simulate as UniswapV2State; SparkDEX V3 and Enosys V3 as UniswapV3State (logs-only indexing). Without these arms the stream builder warn-skips the protocols a self-hosted Flare Tycho indexes. --- fynd-core/src/feed/protocol_registry.rs | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/fynd-core/src/feed/protocol_registry.rs b/fynd-core/src/feed/protocol_registry.rs index 0ea647ef..1815df65 100644 --- a/fynd-core/src/feed/protocol_registry.rs +++ b/fynd-core/src/feed/protocol_registry.rs @@ -284,6 +284,17 @@ pub(crate) fn register_exchanges( builder = builder.exchange::("quickswap_v2", tvl_filter.clone(), None); } + "blazeswap_v2" => { + builder = + builder.exchange::("blazeswap_v2", tvl_filter.clone(), None); + } + "sparkdex_v3" => { + builder = + builder.exchange::("sparkdex_v3", tvl_filter.clone(), None); + } + "enosys_v3" => { + builder = builder.exchange::("enosys_v3", tvl_filter.clone(), None); + } "lunarbase" => { builder = builder.exchange::("lunarbase", tvl_filter.clone(), None); } From f6e8fff7619982f7fdeca28094d743de97d75a79 Mon Sep 17 00:00:00 2001 From: "tuddman@users.noreply.github.com" Date: Wed, 19 Aug 2026 21:53:01 +0200 Subject: [PATCH 2/2] feat: add --stream-timeout-secs for the Tycho stream tycho-client derives the stream timeout from the chain's block time (3x block time for custom-registry chains), which is too tight for indexers that pause during catch-up bursts: flare's 2s blocks derive a 6s timeout that trips missed-block stream errors while the self-hosted indexer backfills. The flag (FYND_STREAM_TIMEOUT_SECS) threads through FyndRPCBuilder/FyndBuilder/TychoFeedConfig to the stream builder's latency_buffer and leaves the derived default in place when unset. --- fynd-core/src/feed/mod.rs | 10 ++++++ fynd-core/src/feed/tycho_feed.rs | 59 +++++++++++++++++--------------- fynd-core/src/solver.rs | 15 +++++++- fynd-rpc/src/builder.rs | 9 +++++ src/cli.rs | 5 +++ src/main.rs | 1 + 6 files changed, 70 insertions(+), 29 deletions(-) diff --git a/fynd-core/src/feed/mod.rs b/fynd-core/src/feed/mod.rs index af200b9d..47037515 100644 --- a/fynd-core/src/feed/mod.rs +++ b/fynd-core/src/feed/mod.rs @@ -54,6 +54,10 @@ pub(crate) struct TychoFeedConfig { /// finalization, reducing effective latency at the cost of processing more frequent, /// smaller updates. pub(crate) partial_blocks: bool, + /// Override for the Tycho stream timeout in seconds. When unset, tycho-client derives it + /// from the chain's block time (3x block time for custom-registry chains), which can be too + /// tight for indexers that pause during catch-up bursts. + pub(crate) stream_timeout_secs: Option, } impl TychoFeedConfig { @@ -78,6 +82,7 @@ impl TychoFeedConfig { reconnect_delay: Duration::from_secs(5), blocklisted_components: FxHashSet::default(), partial_blocks: false, + stream_timeout_secs: None, } } @@ -110,6 +115,11 @@ impl TychoFeedConfig { self.partial_blocks = enabled; self } + + pub(crate) fn stream_timeout_secs(mut self, timeout_secs: Option) -> Self { + self.stream_timeout_secs = timeout_secs; + self + } } /// Errors that can occur in the indexer. diff --git a/fynd-core/src/feed/tycho_feed.rs b/fynd-core/src/feed/tycho_feed.rs index 19a8beed..2d1d9537 100644 --- a/fynd-core/src/feed/tycho_feed.rs +++ b/fynd-core/src/feed/tycho_feed.rs @@ -72,6 +72,18 @@ impl TychoFeed { Self { config, market_data, event_tx } } + /// Base stream builder for this feed's Tycho endpoint and chain, with the configured + /// stream-timeout override applied when set. + fn stream_builder(&self) -> ProtocolStreamBuilder { + let mut stream_builder = + ProtocolStreamBuilder::new(&self.config.tycho_url, self.config.chain) + .skip_state_decode_failures(true); + if let Some(timeout_secs) = self.config.stream_timeout_secs { + stream_builder = stream_builder.latency_buffer(timeout_secs); + } + stream_builder + } + /// Returns a new subscriber for market events. pub(crate) fn subscribe(&self) -> broadcast::Receiver { self.event_tx.subscribe() @@ -133,17 +145,13 @@ impl TychoFeed { .clone(), ); - let mut stream_builder = register_exchanges( - ProtocolStreamBuilder::new(&self.config.tycho_url, self.config.chain) - .skip_state_decode_failures(true), - tvl_filter, - &self.config.protocols, - )? - .auth_key(self.config.tycho_api_key.clone()) - .no_tls(!self.config.use_tls) - .skip_state_decode_failures(true) - .min_token_quality(self.config.min_token_quality as u32) - .add_client_metadata(fynd_client_metadata()); + let mut stream_builder = + register_exchanges(self.stream_builder(), tvl_filter, &self.config.protocols)? + .auth_key(self.config.tycho_api_key.clone()) + .no_tls(!self.config.use_tls) + .skip_state_decode_failures(true) + .min_token_quality(self.config.min_token_quality as u32) + .add_client_metadata(fynd_client_metadata()); if self.config.partial_blocks { stream_builder = stream_builder.enable_partial_blocks(); @@ -312,8 +320,7 @@ impl TychoFeed { debug!("Loaded {} tokens from Tycho", all_tokens.len()); let mut stream_builder = match register_exchanges( - ProtocolStreamBuilder::new(&self.config.tycho_url, self.config.chain) - .skip_state_decode_failures(true), + self.stream_builder(), ComponentFilter::with_tvl_range( self.config.min_tvl / self.config.tvl_buffer_ratio, self.config.min_tvl, @@ -520,22 +527,18 @@ impl TychoFeed { .clone(), ); - let mut stream_builder = match register_exchanges( - ProtocolStreamBuilder::new(&self.config.tycho_url, self.config.chain) - .skip_state_decode_failures(true), - tvl_filter, - &self.config.protocols, - ) { - Ok(sb) => sb, - Err(e) => { - let _ = controller_tx.send(Err(e.to_string())); - return Err(e); + let mut stream_builder = + match register_exchanges(self.stream_builder(), tvl_filter, &self.config.protocols) { + Ok(sb) => sb, + Err(e) => { + let _ = controller_tx.send(Err(e.to_string())); + return Err(e); + } } - } - .auth_key(self.config.tycho_api_key.clone()) - .skip_state_decode_failures(true) - .min_token_quality(self.config.min_token_quality as u32) - .add_client_metadata(fynd_client_metadata()); + .auth_key(self.config.tycho_api_key.clone()) + .skip_state_decode_failures(true) + .min_token_quality(self.config.min_token_quality as u32) + .add_client_metadata(fynd_client_metadata()); if self.config.partial_blocks { stream_builder = stream_builder.enable_partial_blocks(); diff --git a/fynd-core/src/solver.rs b/fynd-core/src/solver.rs index 730f1c46..51706ce2 100644 --- a/fynd-core/src/solver.rs +++ b/fynd-core/src/solver.rs @@ -424,6 +424,7 @@ pub struct FyndBuilder { reconnect_delay: Duration, blocklisted_components: FxHashSet, partial_blocks: bool, + stream_timeout_secs: Option, router_timeout: Duration, router_min_responses: usize, encoder: Option, @@ -457,6 +458,7 @@ impl FyndBuilder { reconnect_delay: defaults::RECONNECT_DELAY, blocklisted_components: FxHashSet::default(), partial_blocks: false, + stream_timeout_secs: None, router_timeout: DEFAULT_ROUTER_TIMEOUT, router_min_responses: defaults::ROUTER_MIN_RESPONSES, encoder: None, @@ -537,6 +539,16 @@ impl FyndBuilder { self } + /// Overrides the Tycho stream timeout in seconds (default: derived from the chain's block + /// time by tycho-client). + /// + /// Useful when the upstream indexer can pause longer than the derived timeout, e.g. during + /// catch-up bursts, which would otherwise end the stream with a missed-block error. + pub fn stream_timeout_secs(mut self, timeout_secs: Option) -> Self { + self.stream_timeout_secs = timeout_secs; + self + } + /// Sets the worker router timeout (default: 10s). pub fn worker_router_timeout(mut self, timeout: Duration) -> Self { self.router_timeout = timeout; @@ -715,7 +727,8 @@ impl FyndBuilder { .min_token_quality(self.min_token_quality) .traded_n_days_ago(self.traded_n_days_ago) .blocklisted_components(self.blocklisted_components) - .partial_blocks(self.partial_blocks); + .partial_blocks(self.partial_blocks) + .stream_timeout_secs(self.stream_timeout_secs); let ethereum_client = EthereumRpcClient::new(self.rpc_url.as_str()) .map_err(|e| SolverBuildError::RpcClient(e.to_string()))?; diff --git a/fynd-rpc/src/builder.rs b/fynd-rpc/src/builder.rs index 554866a6..bff37e3d 100644 --- a/fynd-rpc/src/builder.rs +++ b/fynd-rpc/src/builder.rs @@ -179,6 +179,15 @@ impl FyndRPCBuilder { self } + /// Overrides the Tycho stream timeout in seconds (default: derived from the chain's block + /// time by tycho-client). + pub fn stream_timeout_secs(mut self, timeout_secs: Option) -> Self { + self.fynd_builder = self + .fynd_builder + .stream_timeout_secs(timeout_secs); + self + } + /// Overrides the default encoder with a custom one. pub fn encoder(mut self, encoder: Encoder) -> Self { self.fynd_builder = self.fynd_builder.encoder(encoder); diff --git a/src/cli.rs b/src/cli.rs index e48244b5..82d5f698 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -133,6 +133,11 @@ pub struct ServeArgs { #[arg(long)] pub partial_blocks: bool, + /// Override the Tycho stream timeout in seconds. When unset, tycho-client derives it from + /// the chain's block time, which can be too tight for indexers that pause during catch-up. + #[arg(long, env = "FYND_STREAM_TIMEOUT_SECS")] + pub stream_timeout_secs: Option, + /// Enable price guard validation against external price sources. /// Disabled by default. #[arg(long)] diff --git a/src/main.rs b/src/main.rs index 475e016a..81ab8f8b 100644 --- a/src/main.rs +++ b/src/main.rs @@ -344,6 +344,7 @@ async fn setup_solver(args: &cli::ServeArgs) -> Result