tycho_client/
rpc.rs

1//! # Tycho RPC Client
2//!
3//! The objective of this module is to provide swift and simplified access to the Remote Procedure
4//! Call (RPC) endpoints of Tycho. These endpoints are chiefly responsible for facilitating data
5//! queries, especially querying snapshots of data.
6use std::{
7    collections::HashMap,
8    sync::Arc,
9    time::{Duration, SystemTime},
10};
11
12use async_trait::async_trait;
13use backoff::{exponential::ExponentialBackoffBuilder, ExponentialBackoff};
14use futures03::future::try_join_all;
15#[cfg(test)]
16use mockall::automock;
17use reqwest::{header, Client, ClientBuilder, Response, StatusCode, Url};
18use serde::Serialize;
19use thiserror::Error;
20use time::{format_description::well_known::Rfc2822, OffsetDateTime};
21use tokio::{
22    sync::{RwLock, Semaphore},
23    time::sleep,
24};
25use tracing::{debug, error, instrument, trace, warn};
26use tycho_common::{
27    dto::{
28        ComponentTvlRequestBody, ComponentTvlRequestResponse, PaginationLimits, PaginationParams,
29        ProtocolComponentRequestResponse, ProtocolComponentsRequestBody, ProtocolStateRequestBody,
30        ProtocolStateRequestResponse, ProtocolSystemsRequestBody, ProtocolSystemsRequestResponse,
31        StateRequestBody, StateRequestResponse, TokensRequestBody, TokensRequestResponse,
32        TracedEntryPointRequestBody, TracedEntryPointRequestResponse, VersionParam,
33    },
34    models::{
35        blockchain::{EntryPointWithTracingParams, TracedEntryPoints, TracingResult},
36        contract::Account,
37        protocol::{ProtocolComponent, ProtocolComponentState},
38        token::Token,
39        Chain, ComponentId,
40    },
41    Bytes,
42};
43
44/// Data payload returned by `RPCClient::get_protocol_systems`.
45#[derive(Debug, Clone, PartialEq, Eq)]
46pub struct ProtocolSystems {
47    protocol_systems: Vec<String>,
48    dci_protocols: Vec<String>,
49}
50
51impl ProtocolSystems {
52    pub(crate) fn new(protocol_systems: Vec<String>, dci_protocols: Vec<String>) -> Self {
53        Self { protocol_systems, dci_protocols }
54    }
55
56    pub fn protocol_systems(&self) -> &[String] {
57        &self.protocol_systems
58    }
59
60    pub fn dci_protocols(&self) -> &[String] {
61        &self.dci_protocols
62    }
63}
64
65/// An RPC response page, bundling data with the pagination metadata the server returned.
66#[derive(Debug, Clone, PartialEq)]
67pub struct Page<T> {
68    data: T,
69    total: i64,
70    page: i64,
71    page_size: i64,
72}
73
74impl<T> Page<T> {
75    pub fn new(data: T, total: i64, page: i64, page_size: i64) -> Self {
76        Page { data, total, page, page_size }
77    }
78
79    pub fn data(&self) -> &T {
80        &self.data
81    }
82
83    pub fn into_data(self) -> T {
84        self.data
85    }
86
87    pub fn total(&self) -> i64 {
88        self.total
89    }
90
91    pub fn page(&self) -> i64 {
92        self.page
93    }
94
95    pub fn page_size(&self) -> i64 {
96        self.page_size
97    }
98}
99
100impl<T> Page<Vec<T>> {
101    pub fn len(&self) -> usize {
102        self.data.len()
103    }
104
105    pub fn is_empty(&self) -> bool {
106        self.data.is_empty()
107    }
108}
109
110impl<K, V, S: std::hash::BuildHasher> Page<HashMap<K, V, S>> {
111    pub fn len(&self) -> usize {
112        self.data.len()
113    }
114
115    pub fn is_empty(&self) -> bool {
116        self.data.is_empty()
117    }
118}
119
120impl<T: IntoIterator> IntoIterator for Page<T> {
121    type Item = T::Item;
122    type IntoIter = T::IntoIter;
123
124    fn into_iter(self) -> Self::IntoIter {
125        self.data.into_iter()
126    }
127}
128
129impl<'a, T> IntoIterator for &'a Page<T>
130where
131    &'a T: IntoIterator,
132{
133    type Item = <&'a T as IntoIterator>::Item;
134    type IntoIter = <&'a T as IntoIterator>::IntoIter;
135
136    fn into_iter(self) -> Self::IntoIter {
137        (&self.data).into_iter()
138    }
139}
140
141use crate::{
142    client_metadata::CLIENT_METADATA_HEADER,
143    feed::synchronizer::{ComponentWithState, Snapshot},
144    TYCHO_SERVER_VERSION,
145};
146
147/// Suggested concurrency level for RPC clients.
148pub const RPC_CLIENT_CONCURRENCY: usize = 4;
149
150/// Default number of contracts per `/contract_state` request. A few large contracts can hold most
151/// of a protocol's storage, so smaller requests let the server build them and the client decode
152/// them in parallel.
153pub const CONTRACT_STATE_CHUNK_SIZE: usize = 100;
154
155/// Parameters for [`RPCClient::get_contract_state`].
156#[derive(Clone, PartialEq, Debug)]
157pub struct ContractStateParams {
158    chain: Chain,
159    protocol_system: String,
160    contract_ids: Option<Vec<Bytes>>,
161    version: VersionParam,
162    page: i64,
163    page_size: i64,
164}
165
166impl ContractStateParams {
167    pub fn new(chain: Chain, protocol_system: impl Into<String>) -> Self {
168        Self {
169            chain,
170            protocol_system: protocol_system.into(),
171            contract_ids: None,
172            version: VersionParam::default(),
173            page: 0,
174            page_size: StateRequestBody::MAX_PAGE_SIZE_COMPRESSED,
175        }
176    }
177
178    pub fn with_contract_ids(mut self, ids: Vec<Bytes>) -> Self {
179        self.contract_ids = Some(ids);
180        self
181    }
182
183    pub fn with_version(mut self, version: VersionParam) -> Self {
184        self.version = version;
185        self
186    }
187
188    pub fn with_block_number(mut self, block_number: u64) -> Self {
189        self.version = VersionParam::at_block(self.chain.into(), block_number);
190        self
191    }
192
193    pub(crate) fn with_pagination(mut self, page: i64, page_size: i64) -> Self {
194        self.page = page;
195        self.page_size = page_size;
196        self
197    }
198}
199
200/// Parameters for [`RPCClient::get_protocol_components`].
201#[derive(Clone, PartialEq, Debug)]
202pub struct ProtocolComponentsParams {
203    chain: Chain,
204    protocol_system: String,
205    component_ids: Option<Vec<ComponentId>>,
206    tvl_gt: Option<f64>,
207    page: i64,
208    page_size: i64,
209}
210
211impl ProtocolComponentsParams {
212    pub fn new(chain: Chain, protocol_system: impl Into<String>) -> Self {
213        Self {
214            chain,
215            protocol_system: protocol_system.into(),
216            component_ids: None,
217            tvl_gt: None,
218            page: 0,
219            page_size: ProtocolComponentsRequestBody::MAX_PAGE_SIZE_COMPRESSED,
220        }
221    }
222
223    pub fn with_component_ids(mut self, ids: Vec<ComponentId>) -> Self {
224        self.component_ids = Some(ids);
225        self
226    }
227
228    pub fn with_tvl_gt(mut self, tvl_gt: f64) -> Self {
229        self.tvl_gt = Some(tvl_gt);
230        self
231    }
232
233    pub(crate) fn with_pagination(mut self, page: i64, page_size: i64) -> Self {
234        self.page = page;
235        self.page_size = page_size;
236        self
237    }
238
239    #[cfg(test)]
240    pub(crate) fn component_ids(&self) -> Option<&Vec<ComponentId>> {
241        self.component_ids.as_ref()
242    }
243}
244
245/// Parameters for [`RPCClient::get_protocol_states`].
246#[derive(Clone, PartialEq, Debug)]
247pub struct ProtocolStatesParams {
248    chain: Chain,
249    protocol_system: String,
250    protocol_ids: Option<Vec<String>>,
251    include_balances: bool,
252    version: VersionParam,
253    page: i64,
254    page_size: i64,
255}
256
257impl ProtocolStatesParams {
258    pub fn new(chain: Chain, protocol_system: impl Into<String>) -> Self {
259        Self {
260            chain,
261            protocol_system: protocol_system.into(),
262            protocol_ids: None,
263            include_balances: false,
264            version: VersionParam::default(),
265            page: 0,
266            page_size: ProtocolStateRequestBody::MAX_PAGE_SIZE_COMPRESSED,
267        }
268    }
269
270    pub fn with_protocol_ids(mut self, ids: Vec<String>) -> Self {
271        self.protocol_ids = Some(ids);
272        self
273    }
274
275    pub fn with_include_balances(mut self, include_balances: bool) -> Self {
276        self.include_balances = include_balances;
277        self
278    }
279
280    pub fn with_version(mut self, version: VersionParam) -> Self {
281        self.version = version;
282        self
283    }
284
285    pub fn with_block_number(mut self, block_number: u64) -> Self {
286        self.version = VersionParam::at_block(self.chain.into(), block_number);
287        self
288    }
289
290    pub(crate) fn with_pagination(mut self, page: i64, page_size: i64) -> Self {
291        self.page = page;
292        self.page_size = page_size;
293        self
294    }
295}
296
297/// Parameters for [`RPCClient::get_tokens`].
298#[derive(Clone, PartialEq, Debug)]
299pub struct TokensParams {
300    chain: Chain,
301    min_quality: Option<i32>,
302    traded_n_days_ago: Option<u64>,
303    page: i64,
304    page_size: i64,
305}
306
307impl TokensParams {
308    pub fn new(chain: Chain) -> Self {
309        Self {
310            chain,
311            min_quality: None,
312            traded_n_days_ago: None,
313            page: 0,
314            page_size: TokensRequestBody::MAX_PAGE_SIZE_COMPRESSED,
315        }
316    }
317
318    pub fn with_min_quality(mut self, min_quality: i32) -> Self {
319        self.min_quality = Some(min_quality);
320        self
321    }
322
323    pub fn with_traded_n_days_ago(mut self, days: u64) -> Self {
324        self.traded_n_days_ago = Some(days);
325        self
326    }
327
328    pub(crate) fn with_pagination(mut self, page: i64, page_size: i64) -> Self {
329        self.page = page;
330        self.page_size = page_size;
331        self
332    }
333}
334
335/// Parameters for [`RPCClient::get_protocol_systems`].
336#[derive(Clone, PartialEq, Debug)]
337pub struct ProtocolSystemsParams {
338    chain: Chain,
339    page: i64,
340    page_size: i64,
341}
342
343impl ProtocolSystemsParams {
344    pub fn new(chain: Chain) -> Self {
345        Self { chain, page: 0, page_size: ProtocolSystemsRequestBody::MAX_PAGE_SIZE_COMPRESSED }
346    }
347
348    pub(crate) fn with_pagination(mut self, page: i64, page_size: i64) -> Self {
349        self.page = page;
350        self.page_size = page_size;
351        self
352    }
353}
354
355/// Parameters for [`RPCClient::get_component_tvl`].
356#[derive(Clone, PartialEq, Debug)]
357pub struct ComponentTvlParams {
358    chain: Chain,
359    protocol_system: Option<String>,
360    component_ids: Option<Vec<String>>,
361    page: i64,
362    page_size: i64,
363}
364
365impl ComponentTvlParams {
366    pub fn new(chain: Chain) -> Self {
367        Self {
368            chain,
369            protocol_system: None,
370            component_ids: None,
371            page: 0,
372            page_size: ComponentTvlRequestBody::MAX_PAGE_SIZE_COMPRESSED,
373        }
374    }
375
376    pub fn with_protocol_system(mut self, protocol_system: impl Into<String>) -> Self {
377        self.protocol_system = Some(protocol_system.into());
378        self
379    }
380
381    pub fn with_component_ids(mut self, ids: Vec<String>) -> Self {
382        self.component_ids = Some(ids);
383        self
384    }
385
386    pub(crate) fn with_pagination(mut self, page: i64, page_size: i64) -> Self {
387        self.page = page;
388        self.page_size = page_size;
389        self
390    }
391}
392
393/// Parameters for [`RPCClient::get_traced_entry_points`].
394#[derive(Clone, PartialEq, Debug)]
395pub struct TracedEntryPointsParams {
396    chain: Chain,
397    protocol_system: String,
398    component_ids: Option<Vec<String>>,
399    page: i64,
400    page_size: i64,
401}
402
403impl TracedEntryPointsParams {
404    pub fn new(chain: Chain, protocol_system: impl Into<String>) -> Self {
405        Self {
406            chain,
407            protocol_system: protocol_system.into(),
408            component_ids: None,
409            page: 0,
410            page_size: TracedEntryPointRequestBody::MAX_PAGE_SIZE_COMPRESSED,
411        }
412    }
413
414    pub fn with_component_ids(mut self, ids: Vec<String>) -> Self {
415        self.component_ids = Some(ids);
416        self
417    }
418
419    pub(crate) fn with_pagination(mut self, page: i64, page_size: i64) -> Self {
420        self.page = page;
421        self.page_size = page_size;
422        self
423    }
424}
425
426/// Parameters for [`RPCClient::get_protocol_components_paginated`].
427#[derive(Clone, PartialEq, Debug)]
428pub struct ProtocolComponentsPaginatedParams {
429    chain: Chain,
430    protocol_system: String,
431    component_ids: Option<Vec<ComponentId>>,
432    tvl_gt: Option<f64>,
433    chunk_size: Option<usize>,
434    concurrency: usize,
435}
436
437impl ProtocolComponentsPaginatedParams {
438    pub fn new(chain: Chain, protocol_system: impl Into<String>, concurrency: usize) -> Self {
439        Self {
440            chain,
441            protocol_system: protocol_system.into(),
442            component_ids: None,
443            tvl_gt: None,
444            chunk_size: None,
445            concurrency,
446        }
447    }
448
449    pub fn with_component_ids(mut self, ids: Vec<ComponentId>) -> Self {
450        self.component_ids = Some(ids);
451        self
452    }
453
454    pub fn with_tvl_gt(mut self, tvl_gt: f64) -> Self {
455        self.tvl_gt = Some(tvl_gt);
456        self
457    }
458
459    pub fn with_chunk_size(mut self, chunk_size: usize) -> Self {
460        self.chunk_size = Some(chunk_size);
461        self
462    }
463}
464
465/// Parameters for [`RPCClient::get_traced_entry_points_paginated`].
466#[derive(Clone, PartialEq, Debug)]
467pub struct TracedEntryPointsPaginatedParams {
468    chain: Chain,
469    protocol_system: String,
470    component_ids: Vec<String>,
471    chunk_size: Option<usize>,
472    concurrency: usize,
473}
474
475impl TracedEntryPointsPaginatedParams {
476    pub fn new(
477        chain: Chain,
478        protocol_system: impl Into<String>,
479        component_ids: Vec<String>,
480        concurrency: usize,
481    ) -> Self {
482        Self {
483            chain,
484            protocol_system: protocol_system.into(),
485            component_ids,
486            chunk_size: None,
487            concurrency,
488        }
489    }
490
491    pub fn with_chunk_size(mut self, chunk_size: usize) -> Self {
492        self.chunk_size = Some(chunk_size);
493        self
494    }
495}
496
497/// Parameters for [`RPCClient::get_protocol_states_paginated`].
498#[derive(Clone, PartialEq, Debug)]
499pub struct ProtocolStatesPaginatedParams {
500    chain: Chain,
501    protocol_system: String,
502    protocol_ids: Vec<String>,
503    include_balances: bool,
504    version: VersionParam,
505    chunk_size: Option<usize>,
506    concurrency: usize,
507}
508
509impl ProtocolStatesPaginatedParams {
510    pub fn new(chain: Chain, protocol_system: impl Into<String>, concurrency: usize) -> Self {
511        Self {
512            chain,
513            protocol_system: protocol_system.into(),
514            protocol_ids: Vec::new(),
515            include_balances: true,
516            version: VersionParam::default(),
517            chunk_size: None,
518            concurrency,
519        }
520    }
521
522    pub fn with_protocol_ids(mut self, ids: Vec<String>) -> Self {
523        self.protocol_ids = ids;
524        self
525    }
526
527    pub fn with_include_balances(mut self, include_balances: bool) -> Self {
528        self.include_balances = include_balances;
529        self
530    }
531
532    pub fn with_version(mut self, version: VersionParam) -> Self {
533        self.version = version;
534        self
535    }
536
537    pub fn with_block_number(mut self, block_number: u64) -> Self {
538        self.version = VersionParam::at_block(self.chain.into(), block_number);
539        self
540    }
541
542    pub fn with_chunk_size(mut self, chunk_size: usize) -> Self {
543        self.chunk_size = Some(chunk_size);
544        self
545    }
546}
547
548/// Parameters for [`RPCClient::get_all_tokens`].
549#[derive(Clone, PartialEq, Debug)]
550pub struct AllTokensParams {
551    chain: Chain,
552    min_quality: Option<i32>,
553    traded_n_days_ago: Option<u64>,
554    chunk_size: Option<usize>,
555    concurrency: usize,
556}
557
558impl AllTokensParams {
559    pub fn new(chain: Chain, concurrency: usize) -> Self {
560        Self { chain, min_quality: None, traded_n_days_ago: None, chunk_size: None, concurrency }
561    }
562
563    pub fn with_min_quality(mut self, min_quality: i32) -> Self {
564        self.min_quality = Some(min_quality);
565        self
566    }
567
568    pub fn with_traded_n_days_ago(mut self, days: u64) -> Self {
569        self.traded_n_days_ago = Some(days);
570        self
571    }
572
573    pub fn with_chunk_size(mut self, chunk_size: usize) -> Self {
574        self.chunk_size = Some(chunk_size);
575        self
576    }
577}
578
579/// Parameters for [`RPCClient::get_component_tvl_paginated`].
580#[derive(Clone, PartialEq, Debug)]
581pub struct ComponentTvlPaginatedParams {
582    chain: Chain,
583    protocol_system: Option<String>,
584    component_ids: Option<Vec<String>>,
585    chunk_size: Option<usize>,
586    concurrency: usize,
587}
588
589impl ComponentTvlPaginatedParams {
590    pub fn new(chain: Chain, concurrency: usize) -> Self {
591        Self { chain, protocol_system: None, component_ids: None, chunk_size: None, concurrency }
592    }
593
594    pub fn with_protocol_system(mut self, protocol_system: impl Into<String>) -> Self {
595        self.protocol_system = Some(protocol_system.into());
596        self
597    }
598
599    pub fn with_component_ids(mut self, ids: Vec<String>) -> Self {
600        self.component_ids = Some(ids);
601        self
602    }
603
604    pub fn with_chunk_size(mut self, chunk_size: usize) -> Self {
605        self.chunk_size = Some(chunk_size);
606        self
607    }
608}
609
610/// Parameters for [`RPCClient::get_contract_state_paginated`].
611#[derive(Clone, PartialEq, Debug)]
612pub struct ContractStatePaginatedParams {
613    chain: Chain,
614    protocol_system: String,
615    contract_ids: Vec<Bytes>,
616    version: VersionParam,
617    chunk_size: Option<usize>,
618    concurrency: usize,
619}
620
621impl ContractStatePaginatedParams {
622    pub fn new(chain: Chain, protocol_system: impl Into<String>, concurrency: usize) -> Self {
623        Self {
624            chain,
625            protocol_system: protocol_system.into(),
626            contract_ids: Vec::new(),
627            version: VersionParam::default(),
628            chunk_size: None,
629            concurrency,
630        }
631    }
632
633    pub fn with_contract_ids(mut self, ids: Vec<Bytes>) -> Self {
634        self.contract_ids = ids;
635        self
636    }
637
638    pub fn with_version(mut self, version: VersionParam) -> Self {
639        self.version = version;
640        self
641    }
642
643    pub fn with_block_number(mut self, block_number: u64) -> Self {
644        self.version = VersionParam::at_block(self.chain.into(), block_number);
645        self
646    }
647
648    pub fn with_chunk_size(mut self, chunk_size: usize) -> Self {
649        self.chunk_size = Some(chunk_size);
650        self
651    }
652}
653
654/// Request body for fetching a snapshot of protocol states and VM storage.
655///
656/// This struct helps to coordinate fetching  multiple pieces of related data
657/// (protocol states, contract storage, TVL, entry points).
658#[derive(Clone, Debug, PartialEq)]
659pub struct SnapshotParameters<'a> {
660    /// Which chain to fetch snapshots for
661    pub chain: Chain,
662    /// Protocol system name, required for correct state resolution
663    pub protocol_system: &'a str,
664    /// Components to fetch protocol states for
665    pub components: &'a HashMap<ComponentId, ProtocolComponent>,
666    /// Traced entry points data mapped by component id (model types)
667    pub entrypoints: Option<&'a TracedEntryPoints>,
668    /// Contract addresses to fetch VM storage for
669    pub contract_ids: &'a [Bytes],
670    /// Block number for versioning
671    pub block_number: u64,
672    /// Whether to include balance information
673    pub include_balances: bool,
674    /// Whether to fetch TVL data
675    pub include_tvl: bool,
676}
677
678impl<'a> SnapshotParameters<'a> {
679    pub fn new(
680        chain: Chain,
681        protocol_system: &'a str,
682        components: &'a HashMap<ComponentId, ProtocolComponent>,
683        contract_ids: &'a [Bytes],
684        block_number: u64,
685    ) -> Self {
686        Self {
687            chain,
688            protocol_system,
689            components,
690            entrypoints: None,
691            contract_ids,
692            block_number,
693            include_balances: true,
694            include_tvl: true,
695        }
696    }
697
698    /// Set whether to include balance information (default: true)
699    pub fn include_balances(mut self, include_balances: bool) -> Self {
700        self.include_balances = include_balances;
701        self
702    }
703
704    /// Set whether to fetch TVL data (default: true)
705    pub fn include_tvl(mut self, include_tvl: bool) -> Self {
706        self.include_tvl = include_tvl;
707        self
708    }
709
710    pub fn entrypoints(mut self, entrypoints: &'a TracedEntryPoints) -> Self {
711        self.entrypoints = Some(entrypoints);
712        self
713    }
714}
715
716#[derive(Error, Debug)]
717pub enum RPCError {
718    /// The passed tycho url failed to parse.
719    #[error("Failed to parse URL: {0}. Error: {1}")]
720    UrlParsing(String, String),
721
722    /// The request data is not correctly formed.
723    #[error("Failed to format request: {0}")]
724    FormatRequest(String),
725
726    /// Errors forwarded from the HTTP protocol.
727    #[error("Unexpected HTTP client error: {0}")]
728    HttpClient(String, #[source] reqwest::Error),
729
730    /// The response from the server could not be parsed correctly.
731    #[error("Failed to parse response: {0}")]
732    ParseResponse(String),
733
734    /// The requested block is outside the server's retention window.
735    #[error("Snapshot block is stale: {0}")]
736    StaleBlock(String),
737
738    /// The requested extractor does not exist on the server.
739    #[error("Unknown extractor: {0}")]
740    UnknownExtractor(String),
741
742    /// Other fatal errors.
743    #[error("Fatal error: {0}")]
744    Fatal(String),
745
746    #[error("Rate limited until {0:?}")]
747    RateLimited(Option<SystemTime>),
748
749    #[error("Server unreachable: {0}")]
750    ServerUnreachable(String),
751}
752
753impl RPCError {
754    /// Converts an HTTP response body parse failure into the correct `RPCError`.
755    ///
756    /// The tycho server returns plain-text error messages (not JSON) when a requested block falls
757    /// outside its retention window. Detecting these here gives callers a typed signal to retry
758    /// with a more recent block rather than treating it as an unrecoverable parse failure.
759    ///
760    /// NOTE: The string matching below is coupled to the server's error message text. If those
761    /// messages change server-side this silently regresses to `ParseResponse`. Replace with a
762    /// structured error code if the server ever returns typed error responses.
763    fn from_parse_error(err: serde_json::Error, body: &str) -> Self {
764        if body.contains("version is older than") || body.contains("Could not find Block") {
765            RPCError::StaleBlock(body.to_string())
766        } else if body.starts_with("Unknown extractor:") {
767            RPCError::UnknownExtractor(body.to_string())
768        } else {
769            RPCError::ParseResponse(format!("Error: {err}, Body: {body}"))
770        }
771    }
772}
773
774#[cfg_attr(test, automock)]
775#[async_trait]
776pub trait RPCClient: Send + Sync {
777    /// Returns whether compression is enabled for requests.
778    fn compression(&self) -> bool;
779
780    /// Retrieves a snapshot of contract state for the given contract addresses.
781    ///
782    /// `block_number` pins the query to a specific block; pass `None` to use the latest state.
783    async fn get_contract_state(
784        &self,
785        params: ContractStateParams,
786    ) -> Result<Page<Vec<Account>>, RPCError>;
787
788    /// Retrieves a snapshot of contract state for a set of contract IDs.
789    ///
790    /// If `chunk_size` is `None`, it defaults to [`CONTRACT_STATE_CHUNK_SIZE`], capped at the
791    /// maximum page size.
792    async fn get_contract_state_paginated(
793        &self,
794        params: ContractStatePaginatedParams,
795    ) -> Result<Vec<Account>, RPCError> {
796        let semaphore = Arc::new(Semaphore::new(params.concurrency));
797
798        // Sort the ids to maximize server-side cache hits
799        let mut sorted_ids = params.contract_ids;
800        sorted_ids.sort();
801
802        let chunk_size = params.chunk_size.unwrap_or_else(|| {
803            CONTRACT_STATE_CHUNK_SIZE
804                .min(StateRequestBody::effective_max_page_size(self.compression()) as usize)
805        });
806
807        let mut tasks = Vec::new();
808        for chunk in sorted_ids.chunks(chunk_size) {
809            let sem = semaphore.clone();
810            let base_params =
811                ContractStateParams::new(params.chain, params.protocol_system.as_str())
812                    .with_contract_ids(chunk.to_vec())
813                    .with_version(params.version.clone())
814                    .with_pagination(0, chunk_size as i64);
815            tasks.push(async move {
816                let _permit = sem
817                    .acquire()
818                    .await
819                    .map_err(|_| RPCError::Fatal("Semaphore dropped".to_string()))?;
820                self.get_contract_state(base_params)
821                    .await
822            });
823        }
824
825        let pages = try_join_all(tasks).await?;
826
827        let accounts = pages
828            .into_iter()
829            .flat_map(|p| p.into_iter())
830            .collect();
831
832        Ok(accounts)
833    }
834
835    /// Retrieves protocol components matching the given filters.
836    ///
837    /// Pass `component_ids` to filter by specific IDs, or `tvl_gt` to filter by minimum TVL.
838    async fn get_protocol_components(
839        &self,
840        params: ProtocolComponentsParams,
841    ) -> Result<Page<Vec<ProtocolComponent>>, RPCError>;
842
843    /// Retrieves protocol components, fetching all pages automatically.
844    ///
845    /// If `chunk_size` is `None`, it defaults to the maximum page size.
846    async fn get_protocol_components_paginated(
847        &self,
848        params: ProtocolComponentsPaginatedParams,
849    ) -> Result<Vec<ProtocolComponent>, RPCError> {
850        let chain = params.chain;
851        let protocol_system = params.protocol_system;
852        let component_ids = params.component_ids;
853        let tvl_gt = params.tvl_gt;
854        let chunk_size = params.chunk_size;
855        let concurrency = params.concurrency;
856
857        let semaphore = Arc::new(Semaphore::new(concurrency));
858
859        let chunk_size = chunk_size.unwrap_or(
860            ProtocolComponentsRequestBody::effective_max_page_size(self.compression()) as usize,
861        );
862
863        // If a set of component IDs is specified, the maximum return size is already known,
864        // allowing us to pre-compute the number of requests to be made.
865        match component_ids {
866            Some(ids) => {
867                let tasks: Vec<_> =
868                    ids.chunks(chunk_size)
869                        .enumerate()
870                        .map(|(index, chunk)| {
871                            let sem = semaphore.clone();
872                            let mut base =
873                                ProtocolComponentsParams::new(chain, protocol_system.as_str())
874                                    .with_component_ids(chunk.to_vec())
875                                    .with_pagination(index as i64, chunk_size as i64);
876                            if let Some(tvl) = tvl_gt {
877                                base = base.with_tvl_gt(tvl);
878                            }
879                            async move {
880                                let _permit = sem.acquire().await.map_err(|_| {
881                                    RPCError::Fatal("Semaphore dropped".to_string())
882                                })?;
883                                self.get_protocol_components(base).await
884                            }
885                        })
886                        .collect();
887
888                try_join_all(tasks)
889                    .await
890                    .map(|pages| pages.into_iter().flatten().collect())
891            }
892            None => {
893                // If no component ids are specified, we need to make requests based on the total
894                // number of results from the first response.
895                let mut base_params =
896                    ProtocolComponentsParams::new(chain, protocol_system.as_str())
897                        .with_pagination(0, chunk_size as i64);
898                if let Some(tvl) = tvl_gt {
899                    base_params = base_params.with_tvl_gt(tvl);
900                }
901
902                let first_page = self
903                    .get_protocol_components(base_params)
904                    .await?;
905
906                let total_items = first_page.total();
907                let total_pages = (total_items as f64 / chunk_size as f64).ceil() as i64;
908
909                let mut all: Vec<ProtocolComponent> = first_page.into_data();
910
911                let mut page = 1;
912                while page < total_pages {
913                    let requests_in_this_iteration = (total_pages - page).min(concurrency as i64);
914
915                    let tasks: Vec<_> = (0..requests_in_this_iteration)
916                        .map(|iter| {
917                            let sem = semaphore.clone();
918                            let mut p =
919                                ProtocolComponentsParams::new(chain, protocol_system.as_str())
920                                    .with_pagination(page + iter, chunk_size as i64);
921                            if let Some(tvl) = tvl_gt {
922                                p = p.with_tvl_gt(tvl);
923                            }
924                            async move {
925                                let _permit = sem.acquire().await.map_err(|_| {
926                                    RPCError::Fatal("Semaphore dropped".to_string())
927                                })?;
928                                self.get_protocol_components(p).await
929                            }
930                        })
931                        .collect();
932
933                    let responses = try_join_all(tasks).await?;
934
935                    for resp in responses {
936                        all.extend(resp);
937                    }
938
939                    page += requests_in_this_iteration;
940                }
941                Ok(all)
942            }
943        }
944    }
945
946    /// Retrieves a page of protocol component states.
947    ///
948    /// `block_number` pins the query to a specific block; pass `None` to use the latest state.
949    async fn get_protocol_states(
950        &self,
951        params: ProtocolStatesParams,
952    ) -> Result<Page<Vec<ProtocolComponentState>>, RPCError>;
953
954    /// Retrieves protocol states for a set of protocol IDs, fetching all pages automatically.
955    ///
956    /// If `chunk_size` is `None`, it defaults to the maximum page size.
957    async fn get_protocol_states_paginated(
958        &self,
959        params: ProtocolStatesPaginatedParams,
960    ) -> Result<Vec<ProtocolComponentState>, RPCError> {
961        let semaphore = Arc::new(Semaphore::new(params.concurrency));
962
963        let chunk_size =
964            params
965                .chunk_size
966                .unwrap_or(
967                    ProtocolStateRequestBody::effective_max_page_size(self.compression()) as usize
968                );
969
970        let tasks: Vec<_> = params
971            .protocol_ids
972            .chunks(chunk_size)
973            .map(|c| {
974                let sem = semaphore.clone();
975                let p = ProtocolStatesParams::new(params.chain, params.protocol_system.as_str())
976                    .with_protocol_ids(c.to_vec())
977                    .with_include_balances(params.include_balances)
978                    .with_version(params.version.clone())
979                    .with_pagination(0, chunk_size as i64);
980                async move {
981                    let _permit = sem
982                        .acquire()
983                        .await
984                        .map_err(|_| RPCError::Fatal("Semaphore dropped".to_string()))?;
985                    self.get_protocol_states(p).await
986                }
987            })
988            .collect();
989
990        try_join_all(tasks)
991            .await
992            .map(|pages| pages.into_iter().flatten().collect())
993    }
994
995    /// Retrieves a page of tokens.
996    ///
997    /// Use `get_all_tokens` to fetch all matching tokens automatically.
998    async fn get_tokens(&self, params: TokensParams) -> Result<Page<Vec<Token>>, RPCError>;
999
1000    /// Retrieves all tokens matching the given criteria, fetching all pages automatically.
1001    ///
1002    /// If `chunk_size` is `None`, it defaults to the maximum page size.
1003    async fn get_all_tokens(&self, params: AllTokensParams) -> Result<Vec<Token>, RPCError> {
1004        let chunk_size = params
1005            .chunk_size
1006            .unwrap_or(TokensRequestBody::effective_max_page_size(self.compression()) as usize);
1007
1008        let semaphore = Arc::new(Semaphore::new(params.concurrency));
1009
1010        let page_size: i64 = chunk_size.try_into().map_err(|_| {
1011            RPCError::FormatRequest("Failed to convert chunk_size into i64".to_string())
1012        })?;
1013
1014        let mut base_params = TokensParams::new(params.chain).with_pagination(0, page_size);
1015        if let Some(q) = params.min_quality {
1016            base_params = base_params.with_min_quality(q);
1017        }
1018        if let Some(d) = params.traded_n_days_ago {
1019            base_params = base_params.with_traded_n_days_ago(d);
1020        }
1021
1022        let first_page = self.get_tokens(base_params).await?;
1023        let total_pages = (first_page.total() as f64 / chunk_size as f64).ceil() as i64;
1024
1025        let mut all_tokens: Vec<Token> = first_page.into_data();
1026
1027        if total_pages <= 1 {
1028            return Ok(all_tokens);
1029        }
1030
1031        let tasks: Vec<_> = (1..total_pages)
1032            .map(|page| {
1033                let sem = semaphore.clone();
1034                let mut p = TokensParams::new(params.chain).with_pagination(page, page_size);
1035                if let Some(q) = params.min_quality {
1036                    p = p.with_min_quality(q);
1037                }
1038                if let Some(d) = params.traded_n_days_ago {
1039                    p = p.with_traded_n_days_ago(d);
1040                }
1041                async move {
1042                    let _permit = sem
1043                        .acquire()
1044                        .await
1045                        .map_err(|_| RPCError::Fatal("Semaphore dropped".to_string()))?;
1046                    self.get_tokens(p).await
1047                }
1048            })
1049            .collect();
1050
1051        let pages = try_join_all(tasks).await?;
1052        for page in pages {
1053            all_tokens.extend(page);
1054        }
1055
1056        Ok(all_tokens)
1057    }
1058
1059    /// Retrieves the protocol systems known to the server.
1060    async fn get_protocol_systems(
1061        &self,
1062        params: ProtocolSystemsParams,
1063    ) -> Result<Page<ProtocolSystems>, RPCError>;
1064
1065    /// Retrieves component TVL values.
1066    ///
1067    /// Filter by `component_ids` or by `protocol_system`; both are optional.
1068    async fn get_component_tvl(
1069        &self,
1070        params: ComponentTvlParams,
1071    ) -> Result<Page<HashMap<String, f64>>, RPCError>;
1072
1073    /// Retrieves component TVL values, fetching all pages automatically.
1074    ///
1075    /// If `chunk_size` is `None`, it defaults to the maximum page size.
1076    async fn get_component_tvl_paginated(
1077        &self,
1078        params: ComponentTvlPaginatedParams,
1079    ) -> Result<HashMap<String, f64>, RPCError> {
1080        let semaphore = Arc::new(Semaphore::new(params.concurrency));
1081
1082        let chunk_size =
1083            params
1084                .chunk_size
1085                .unwrap_or(
1086                    ComponentTvlRequestBody::effective_max_page_size(self.compression()) as usize
1087                );
1088
1089        match params.component_ids {
1090            Some(ids) => {
1091                let tasks: Vec<_> =
1092                    ids.chunks(chunk_size)
1093                        .enumerate()
1094                        .map(|(index, chunk)| {
1095                            let sem = semaphore.clone();
1096                            let mut p = ComponentTvlParams::new(params.chain)
1097                                .with_component_ids(chunk.to_vec())
1098                                .with_pagination(index as i64, chunk_size as i64);
1099                            if let Some(ref ps) = params.protocol_system {
1100                                p = p.with_protocol_system(ps.as_str());
1101                            }
1102                            async move {
1103                                let _permit = sem.acquire().await.map_err(|_| {
1104                                    RPCError::Fatal("Semaphore dropped".to_string())
1105                                })?;
1106                                self.get_component_tvl(p).await
1107                            }
1108                        })
1109                        .collect();
1110
1111                let pages = try_join_all(tasks).await?;
1112
1113                let mut merged_tvl = HashMap::new();
1114                for page in pages {
1115                    for (key, value) in page {
1116                        *merged_tvl.entry(key).or_insert(0.0) = value;
1117                    }
1118                }
1119
1120                Ok(merged_tvl)
1121            }
1122            None => {
1123                let mut base =
1124                    ComponentTvlParams::new(params.chain).with_pagination(0, chunk_size as i64);
1125                if let Some(ref ps) = params.protocol_system {
1126                    base = base.with_protocol_system(ps.as_str());
1127                }
1128
1129                let first_page = self.get_component_tvl(base).await?;
1130                let total_items = first_page.total();
1131                let total_pages = (total_items as f64 / chunk_size as f64).ceil() as i64;
1132
1133                let mut merged_tvl: HashMap<String, f64> = first_page.into_data();
1134
1135                let mut page = 1;
1136                while page < total_pages {
1137                    let requests_in_this_iteration =
1138                        (total_pages - page).min(params.concurrency as i64);
1139
1140                    let tasks: Vec<_> = (0..requests_in_this_iteration)
1141                        .map(|i| {
1142                            let sem = semaphore.clone();
1143                            let mut p = ComponentTvlParams::new(params.chain)
1144                                .with_pagination(page + i, chunk_size as i64);
1145                            if let Some(ref ps) = params.protocol_system {
1146                                p = p.with_protocol_system(ps.as_str());
1147                            }
1148                            async move {
1149                                let _permit = sem.acquire().await.map_err(|_| {
1150                                    RPCError::Fatal("Semaphore dropped".to_string())
1151                                })?;
1152                                self.get_component_tvl(p).await
1153                            }
1154                        })
1155                        .collect();
1156
1157                    let responses = try_join_all(tasks).await?;
1158
1159                    for resp in responses {
1160                        for (key, value) in resp {
1161                            *merged_tvl.entry(key).or_insert(0.0) = value;
1162                        }
1163                    }
1164
1165                    page += requests_in_this_iteration;
1166                }
1167
1168                Ok(merged_tvl)
1169            }
1170        }
1171    }
1172
1173    /// Retrieves a page of traced entry points.
1174    ///
1175    /// Use `get_traced_entry_points_paginated` to fetch all pages automatically.
1176    async fn get_traced_entry_points(
1177        &self,
1178        params: TracedEntryPointsParams,
1179    ) -> Result<Page<TracedEntryPoints>, RPCError>;
1180
1181    /// Retrieves traced entry points for a set of component IDs, fetching all pages automatically.
1182    ///
1183    /// If `chunk_size` is `None`, it defaults to the maximum page size.
1184    async fn get_traced_entry_points_paginated(
1185        &self,
1186        params: TracedEntryPointsPaginatedParams,
1187    ) -> Result<TracedEntryPoints, RPCError> {
1188        let chain = params.chain;
1189        let protocol_system = params.protocol_system;
1190        let component_ids = params.component_ids;
1191        let chunk_size = params.chunk_size;
1192        let concurrency = params.concurrency;
1193
1194        let semaphore = Arc::new(Semaphore::new(concurrency));
1195
1196        let chunk_size = chunk_size.unwrap_or(
1197            TracedEntryPointRequestBody::effective_max_page_size(self.compression()) as usize,
1198        );
1199
1200        let tasks: Vec<_> = component_ids
1201            .chunks(chunk_size)
1202            .map(|c| {
1203                let sem = semaphore.clone();
1204                let params = TracedEntryPointsParams::new(chain, protocol_system.as_str())
1205                    .with_component_ids(c.to_vec())
1206                    .with_pagination(0, chunk_size as i64);
1207                async move {
1208                    let _permit = sem
1209                        .acquire()
1210                        .await
1211                        .map_err(|_| RPCError::Fatal("Semaphore dropped".to_string()))?;
1212                    self.get_traced_entry_points(params)
1213                        .await
1214                }
1215            })
1216            .collect();
1217
1218        try_join_all(tasks)
1219            .await
1220            .map(|pages| pages.into_iter().flatten().collect())
1221    }
1222
1223    // clippy false positive: `'a` is required by the trait method signature and is
1224    // used in `SnapshotParameters<'a>`, but `async_trait` makes Clippy miss it.
1225    #[allow(clippy::extra_unused_lifetimes)]
1226    async fn get_snapshots<'a>(
1227        &self,
1228        request: &SnapshotParameters<'a>,
1229        chunk_size: Option<usize>,
1230        concurrency: usize,
1231    ) -> Result<Snapshot, RPCError>;
1232}
1233
1234/// Configuration options for HttpRPCClient
1235#[derive(Debug, Clone)]
1236pub struct HttpRPCClientOptions {
1237    /// Optional API key for authentication
1238    pub auth_key: Option<String>,
1239    /// Enable compression for requests (default: true)
1240    /// When enabled, adds Accept-Encoding: zstd header
1241    pub compression: bool,
1242    /// Pre-serialized `X-Tycho-Client-Metadata` header value. `None` sends no header.
1243    pub client_metadata_header: Option<String>,
1244}
1245
1246impl Default for HttpRPCClientOptions {
1247    fn default() -> Self {
1248        Self::new()
1249    }
1250}
1251
1252impl HttpRPCClientOptions {
1253    /// Create new options with default values (compression enabled)
1254    pub fn new() -> Self {
1255        Self { auth_key: None, compression: true, client_metadata_header: None }
1256    }
1257
1258    /// Set the authentication key
1259    pub fn with_auth_key(mut self, auth_key: Option<String>) -> Self {
1260        self.auth_key = auth_key;
1261        self
1262    }
1263
1264    /// Set whether to enable compression (default: true)
1265    pub fn with_compression(mut self, compression: bool) -> Self {
1266        self.compression = compression;
1267        self
1268    }
1269
1270    /// Set the pre-serialized client-metadata header value. `None` sends no header.
1271    pub fn with_client_metadata_header(mut self, header: Option<String>) -> Self {
1272        self.client_metadata_header = header;
1273        self
1274    }
1275}
1276
1277#[derive(Debug, Clone)]
1278pub struct HttpRPCClient {
1279    http_client: Client,
1280    url: Url,
1281    retry_after: Arc<RwLock<Option<SystemTime>>>,
1282    backoff_policy: ExponentialBackoff,
1283    server_restart_duration: Duration,
1284    compression: bool,
1285}
1286
1287impl HttpRPCClient {
1288    pub fn new(base_uri: &str, options: HttpRPCClientOptions) -> Result<Self, RPCError> {
1289        let uri = base_uri
1290            .parse::<Url>()
1291            .map_err(|e| RPCError::UrlParsing(base_uri.to_string(), e.to_string()))?;
1292
1293        // Add default headers
1294        let mut headers = header::HeaderMap::new();
1295        headers.insert(header::CONTENT_TYPE, header::HeaderValue::from_static("application/json"));
1296        let user_agent = format!("tycho-client-{version}", version = env!("CARGO_PKG_VERSION"));
1297        headers.insert(
1298            header::USER_AGENT,
1299            header::HeaderValue::from_str(&user_agent)
1300                .map_err(|e| RPCError::FormatRequest(format!("Invalid user agent format: {e}")))?,
1301        );
1302
1303        // Add Authorization if one is given
1304        if let Some(key) = options.auth_key.as_deref() {
1305            let mut auth_value = header::HeaderValue::from_str(key).map_err(|e| {
1306                RPCError::FormatRequest(format!("Invalid authorization key format: {e}"))
1307            })?;
1308            auth_value.set_sensitive(true);
1309            headers.insert(header::AUTHORIZATION, auth_value);
1310        }
1311
1312        // Add generic client metadata if one is given. Pre-validated by the serializer, but mirror
1313        // the auth pattern so any residual formatting error surfaces as a FormatRequest error.
1314        if let Some(metadata) = options
1315            .client_metadata_header
1316            .as_deref()
1317        {
1318            let value = header::HeaderValue::from_str(metadata).map_err(|e| {
1319                RPCError::FormatRequest(format!("Invalid client metadata format: {e}"))
1320            })?;
1321            headers.insert(header::HeaderName::from_static(CLIENT_METADATA_HEADER), value);
1322        }
1323
1324        let mut client_builder = ClientBuilder::new()
1325            .default_headers(headers)
1326            .http2_prior_knowledge();
1327
1328        // When compression is disabled, turn off all automatic compression
1329        if !options.compression {
1330            client_builder = client_builder.no_zstd();
1331        }
1332
1333        let client = client_builder
1334            .build()
1335            .map_err(|e| RPCError::HttpClient(e.to_string(), e))?;
1336
1337        Ok(Self {
1338            http_client: client,
1339            url: uri,
1340            retry_after: Arc::new(RwLock::new(None)),
1341            backoff_policy: ExponentialBackoffBuilder::new()
1342                .with_initial_interval(Duration::from_millis(250))
1343                // increase backoff time by 75% each failure
1344                .with_multiplier(1.75)
1345                // keep retrying every 30s
1346                .with_max_interval(Duration::from_secs(30))
1347                // if all retries take longer than 2m, give up
1348                .with_max_elapsed_time(Some(Duration::from_secs(125)))
1349                .build(),
1350            server_restart_duration: Duration::from_secs(120),
1351            compression: options.compression,
1352        })
1353    }
1354
1355    #[cfg(test)]
1356    pub fn with_test_backoff_policy(mut self) -> Self {
1357        // Extremely short intervals for very fast testing
1358        self.backoff_policy = ExponentialBackoffBuilder::new()
1359            .with_initial_interval(Duration::from_millis(1))
1360            .with_multiplier(1.1)
1361            .with_max_interval(Duration::from_millis(5))
1362            .with_max_elapsed_time(Some(Duration::from_millis(50)))
1363            .build();
1364        self.server_restart_duration = Duration::from_millis(50);
1365        self
1366    }
1367
1368    /// Converts a error response to a Result.
1369    ///
1370    /// Raises an error if the response status code id 429, 502, 503 or 504. In the 429
1371    /// case it will try to look for a retry-after header an parse it accordingly. The
1372    /// parsed value is then passed as part of the error.
1373    async fn error_for_response(
1374        &self,
1375        response: reqwest::Response,
1376    ) -> Result<reqwest::Response, RPCError> {
1377        match response.status() {
1378            StatusCode::TOO_MANY_REQUESTS => {
1379                let retry_after_raw = response
1380                    .headers()
1381                    .get(reqwest::header::RETRY_AFTER)
1382                    .and_then(|h| h.to_str().ok())
1383                    .and_then(parse_retry_value);
1384
1385                let reason = response
1386                    .text()
1387                    .await
1388                    .unwrap_or_default();
1389                warn!(reason, retry_after = ?retry_after_raw, "Rate limited by server");
1390
1391                Err(RPCError::RateLimited(retry_after_raw))
1392            }
1393            StatusCode::BAD_GATEWAY |
1394            StatusCode::SERVICE_UNAVAILABLE |
1395            StatusCode::GATEWAY_TIMEOUT => Err(RPCError::ServerUnreachable(
1396                response
1397                    .text()
1398                    .await
1399                    .unwrap_or_else(|_| "Server Unreachable".to_string()),
1400            )),
1401            _ => Ok(response),
1402        }
1403    }
1404
1405    /// Classifies errors into transient or permanent ones.
1406    ///
1407    /// Transient errors are retried with a potential backoff, permanent ones are not.
1408    /// If the error is RateLimited, this method will set the self.retry_after value so
1409    /// future requests wait until the rate limit has been reset.
1410    async fn handle_error_for_backoff(&self, e: RPCError) -> backoff::Error<RPCError> {
1411        match e {
1412            RPCError::ServerUnreachable(_) => {
1413                backoff::Error::retry_after(e, self.server_restart_duration)
1414            }
1415            RPCError::RateLimited(Some(until)) => {
1416                let mut retry_after_guard = self.retry_after.write().await;
1417                *retry_after_guard = Some(
1418                    retry_after_guard
1419                        .unwrap_or(until)
1420                        .max(until),
1421                );
1422
1423                if let Ok(duration) = until.duration_since(SystemTime::now()) {
1424                    backoff::Error::retry_after(e, duration)
1425                } else {
1426                    e.into()
1427                }
1428            }
1429            RPCError::RateLimited(None) => e.into(),
1430            _ => backoff::Error::permanent(e),
1431        }
1432    }
1433
1434    /// Waits until the current rate limit time has passed.
1435    ///
1436    /// Only waits if there is a time and that time is in the future, else return
1437    /// immediately.
1438    async fn wait_until_retry_after(&self) {
1439        if let Some(&until) = self.retry_after.read().await.as_ref() {
1440            let now = SystemTime::now();
1441            if until > now {
1442                if let Ok(duration) = until.duration_since(now) {
1443                    sleep(duration).await
1444                }
1445            }
1446        }
1447    }
1448
1449    /// Makes a post request handling transient failures.
1450    ///
1451    /// If a retry-after header is received it will be respected. Else the configured
1452    /// backoff policy is used to deal with transient network or server errors.
1453    async fn make_post_request<T: Serialize + ?Sized>(
1454        &self,
1455        request: &T,
1456        uri: &String,
1457    ) -> Result<Response, RPCError> {
1458        self.wait_until_retry_after().await;
1459        let response = backoff::future::retry(self.backoff_policy.clone(), || async {
1460            let server_response = self
1461                .http_client
1462                .post(uri)
1463                .json(request)
1464                .send()
1465                .await
1466                .map_err(|e| RPCError::HttpClient(e.to_string(), e))?;
1467
1468            match self
1469                .error_for_response(server_response)
1470                .await
1471            {
1472                Ok(response) => Ok(response),
1473                Err(e) => Err(self.handle_error_for_backoff(e).await),
1474            }
1475        })
1476        .await?;
1477        Ok(response)
1478    }
1479}
1480
1481fn parse_retry_value(val: &str) -> Option<SystemTime> {
1482    if let Ok(secs) = val.parse::<u64>() {
1483        return Some(SystemTime::now() + Duration::from_secs(secs));
1484    }
1485    if let Ok(date) = OffsetDateTime::parse(val, &Rfc2822) {
1486        return Some(date.into());
1487    }
1488    None
1489}
1490
1491#[async_trait]
1492impl RPCClient for HttpRPCClient {
1493    fn compression(&self) -> bool {
1494        self.compression
1495    }
1496
1497    #[instrument(skip(self))]
1498    async fn get_contract_state(
1499        &self,
1500        params: ContractStateParams,
1501    ) -> Result<Page<Vec<Account>>, RPCError> {
1502        if params
1503            .contract_ids
1504            .as_ref()
1505            .is_none_or(|ids| ids.is_empty())
1506        {
1507            warn!("No contract ids specified in request.");
1508        }
1509
1510        let request = StateRequestBody {
1511            contract_ids: params.contract_ids,
1512            protocol_system: params.protocol_system,
1513            chain: params.chain.into(),
1514            version: params.version,
1515            pagination: PaginationParams { page: params.page, page_size: params.page_size },
1516        };
1517
1518        let uri = format!(
1519            "{}/{}/contract_state",
1520            self.url
1521                .to_string()
1522                .trim_end_matches('/'),
1523            TYCHO_SERVER_VERSION
1524        );
1525        debug!(%uri, "Sending contract_state request to Tycho server");
1526        trace!(?request, "Sending request to Tycho server");
1527        let response = self
1528            .make_post_request(&request, &uri)
1529            .await?;
1530        trace!(?response, "Received response from Tycho server");
1531
1532        let body = response
1533            .text()
1534            .await
1535            .map_err(|e| RPCError::ParseResponse(e.to_string()))?;
1536        if body.is_empty() {
1537            // Pure native protocols will return empty contract states
1538            return Ok(Page::new(vec![], 0, 0, 0));
1539        }
1540
1541        // Decoding a large response takes seconds of CPU: run it on the blocking pool, so the
1542        // runtime stays responsive and concurrent requests decode in parallel.
1543        let (data, pagination) = tokio::task::spawn_blocking(move || {
1544            let dto_response = serde_json::from_str::<StateRequestResponse>(&body)
1545                .map_err(|err| RPCError::from_parse_error(err, &body))?;
1546            trace!(?dto_response, "Received contract_state response from Tycho server");
1547
1548            let data: Vec<Account> = dto_response
1549                .accounts
1550                .into_iter()
1551                .map(Account::from)
1552                .collect();
1553            Ok::<_, RPCError>((data, dto_response.pagination))
1554        })
1555        .await
1556        .map_err(|err| RPCError::Fatal(format!("Decoding contract_state failed: {err}")))??;
1557        Ok(Page::new(data, pagination.total, pagination.page, pagination.page_size))
1558    }
1559
1560    async fn get_protocol_components(
1561        &self,
1562        params: ProtocolComponentsParams,
1563    ) -> Result<Page<Vec<ProtocolComponent>>, RPCError> {
1564        let request = ProtocolComponentsRequestBody {
1565            protocol_system: params.protocol_system,
1566            component_ids: params.component_ids,
1567            tvl_gt: params.tvl_gt,
1568            chain: params.chain.into(),
1569            pagination: PaginationParams { page: params.page, page_size: params.page_size },
1570        };
1571
1572        let uri = format!(
1573            "{}/{}/protocol_components",
1574            self.url
1575                .to_string()
1576                .trim_end_matches('/'),
1577            TYCHO_SERVER_VERSION,
1578        );
1579        debug!(%uri, "Sending protocol_components request to Tycho server");
1580        trace!(?request, "Sending request to Tycho server");
1581
1582        let response = self
1583            .make_post_request(&request, &uri)
1584            .await?;
1585
1586        trace!(?response, "Received response from Tycho server");
1587
1588        let body = response
1589            .text()
1590            .await
1591            .map_err(|e| RPCError::ParseResponse(e.to_string()))?;
1592        let dto_response = serde_json::from_str::<ProtocolComponentRequestResponse>(&body)
1593            .map_err(|err| RPCError::from_parse_error(err, &body))?;
1594        trace!(?dto_response, "Received protocol_components response from Tycho server");
1595
1596        let data: Vec<ProtocolComponent> = dto_response
1597            .protocol_components
1598            .into_iter()
1599            .map(ProtocolComponent::from)
1600            .collect();
1601        Ok(Page::new(
1602            data,
1603            dto_response.pagination.total,
1604            dto_response.pagination.page,
1605            dto_response.pagination.page_size,
1606        ))
1607    }
1608
1609    async fn get_protocol_states(
1610        &self,
1611        params: ProtocolStatesParams,
1612    ) -> Result<Page<Vec<ProtocolComponentState>>, RPCError> {
1613        if params
1614            .protocol_ids
1615            .as_ref()
1616            .is_none_or(|ids| ids.is_empty())
1617        {
1618            warn!("No protocol ids specified in request.");
1619        }
1620
1621        let request = ProtocolStateRequestBody {
1622            protocol_ids: params.protocol_ids,
1623            protocol_system: params.protocol_system,
1624            chain: params.chain.into(),
1625            include_balances: params.include_balances,
1626            version: params.version,
1627            pagination: PaginationParams { page: params.page, page_size: params.page_size },
1628        };
1629
1630        let uri = format!(
1631            "{}/{}/protocol_state",
1632            self.url
1633                .to_string()
1634                .trim_end_matches('/'),
1635            TYCHO_SERVER_VERSION
1636        );
1637        debug!(%uri, "Sending protocol_states request to Tycho server");
1638        trace!(?request, "Sending request to Tycho server");
1639
1640        let response = self
1641            .make_post_request(&request, &uri)
1642            .await?;
1643        trace!(?response, "Received response from Tycho server");
1644
1645        let body = response
1646            .text()
1647            .await
1648            .map_err(|e| RPCError::ParseResponse(e.to_string()))?;
1649
1650        if body.is_empty() {
1651            // Pure VM protocols will return empty states
1652            return Ok(Page::new(vec![], 0, 0, 0));
1653        }
1654
1655        let dto_response = serde_json::from_str::<ProtocolStateRequestResponse>(&body)
1656            .map_err(|err| RPCError::from_parse_error(err, &body))?;
1657        trace!(?dto_response, "Received protocol_states response from Tycho server");
1658
1659        let data: Vec<ProtocolComponentState> = dto_response
1660            .states
1661            .into_iter()
1662            .map(ProtocolComponentState::from)
1663            .collect();
1664        Ok(Page::new(
1665            data,
1666            dto_response.pagination.total,
1667            dto_response.pagination.page,
1668            dto_response.pagination.page_size,
1669        ))
1670    }
1671
1672    async fn get_tokens(&self, params: TokensParams) -> Result<Page<Vec<Token>>, RPCError> {
1673        let request = TokensRequestBody {
1674            token_addresses: None,
1675            min_quality: params.min_quality,
1676            traded_n_days_ago: params.traded_n_days_ago,
1677            pagination: PaginationParams { page: params.page, page_size: params.page_size },
1678            chain: params.chain.into(),
1679        };
1680
1681        let uri = format!(
1682            "{}/{}/tokens",
1683            self.url
1684                .to_string()
1685                .trim_end_matches('/'),
1686            TYCHO_SERVER_VERSION
1687        );
1688        debug!(%uri, "Sending tokens request to Tycho server");
1689
1690        let response = self
1691            .make_post_request(&request, &uri)
1692            .await?;
1693
1694        let body = response
1695            .text()
1696            .await
1697            .map_err(|e| RPCError::ParseResponse(e.to_string()))?;
1698        let dto_response = serde_json::from_str::<TokensRequestResponse>(&body)
1699            .map_err(|err| RPCError::ParseResponse(format!("Error: {err}, Body: {body}")))?;
1700
1701        let data: Vec<Token> = dto_response
1702            .tokens
1703            .into_iter()
1704            .map(Token::from)
1705            .collect();
1706        Ok(Page::new(
1707            data,
1708            dto_response.pagination.total,
1709            dto_response.pagination.page,
1710            dto_response.pagination.page_size,
1711        ))
1712    }
1713
1714    async fn get_protocol_systems(
1715        &self,
1716        params: ProtocolSystemsParams,
1717    ) -> Result<Page<ProtocolSystems>, RPCError> {
1718        let request = ProtocolSystemsRequestBody {
1719            chain: params.chain.into(),
1720            pagination: PaginationParams { page: params.page, page_size: params.page_size },
1721        };
1722
1723        let uri = format!(
1724            "{}/{}/protocol_systems",
1725            self.url
1726                .to_string()
1727                .trim_end_matches('/'),
1728            TYCHO_SERVER_VERSION
1729        );
1730        debug!(%uri, "Sending protocol_systems request to Tycho server");
1731        trace!(?request, "Sending request to Tycho server");
1732        let response = self
1733            .make_post_request(&request, &uri)
1734            .await?;
1735        trace!(?response, "Received response from Tycho server");
1736        let body = response
1737            .text()
1738            .await
1739            .map_err(|e| RPCError::ParseResponse(e.to_string()))?;
1740        let dto = serde_json::from_str::<ProtocolSystemsRequestResponse>(&body)
1741            .map_err(|err| RPCError::ParseResponse(format!("Error: {err}, Body: {body}")))?;
1742        trace!(?dto, "Received protocol_systems response from Tycho server");
1743        Ok(Page::new(
1744            ProtocolSystems::new(dto.protocol_systems, dto.dci_protocols),
1745            dto.pagination.total,
1746            dto.pagination.page,
1747            dto.pagination.page_size,
1748        ))
1749    }
1750
1751    async fn get_component_tvl(
1752        &self,
1753        params: ComponentTvlParams,
1754    ) -> Result<Page<HashMap<String, f64>>, RPCError> {
1755        let request = ComponentTvlRequestBody {
1756            chain: params.chain.into(),
1757            protocol_system: params.protocol_system,
1758            component_ids: params.component_ids,
1759            pagination: PaginationParams { page: params.page, page_size: params.page_size },
1760        };
1761
1762        let uri = format!(
1763            "{}/{}/component_tvl",
1764            self.url
1765                .to_string()
1766                .trim_end_matches('/'),
1767            TYCHO_SERVER_VERSION
1768        );
1769        debug!(%uri, "Sending get_component_tvl request to Tycho server");
1770        trace!(?request, "Sending request to Tycho server");
1771        let response = self
1772            .make_post_request(&request, &uri)
1773            .await?;
1774        trace!(?response, "Received response from Tycho server");
1775        let body = response
1776            .text()
1777            .await
1778            .map_err(|e| RPCError::ParseResponse(e.to_string()))?;
1779        let dto_response =
1780            serde_json::from_str::<ComponentTvlRequestResponse>(&body).map_err(|err| {
1781                error!("Failed to parse component_tvl response: {:?}", &body);
1782                RPCError::ParseResponse(format!("Error: {err}, Body: {body}"))
1783            })?;
1784        trace!(?dto_response, "Received component_tvl response from Tycho server");
1785        Ok(Page::new(
1786            dto_response.tvl,
1787            dto_response.pagination.total,
1788            dto_response.pagination.page,
1789            dto_response.pagination.page_size,
1790        ))
1791    }
1792
1793    async fn get_traced_entry_points(
1794        &self,
1795        params: TracedEntryPointsParams,
1796    ) -> Result<Page<TracedEntryPoints>, RPCError> {
1797        let request = TracedEntryPointRequestBody {
1798            chain: params.chain.into(),
1799            protocol_system: params.protocol_system,
1800            component_ids: params.component_ids,
1801            pagination: PaginationParams { page: params.page, page_size: params.page_size },
1802        };
1803
1804        let uri = format!(
1805            "{}/{TYCHO_SERVER_VERSION}/traced_entry_points",
1806            self.url
1807                .to_string()
1808                .trim_end_matches('/')
1809        );
1810        debug!(%uri, "Sending traced_entry_points request to Tycho server");
1811        trace!(?request, "Sending request to Tycho server");
1812
1813        let response = self
1814            .make_post_request(&request, &uri)
1815            .await?;
1816
1817        trace!(?response, "Received response from Tycho server");
1818
1819        let body = response
1820            .text()
1821            .await
1822            .map_err(|e| RPCError::ParseResponse(e.to_string()))?;
1823        let dto_response =
1824            serde_json::from_str::<TracedEntryPointRequestResponse>(&body).map_err(|err| {
1825                error!("Failed to parse traced_entry_points response: {:?}", &body);
1826                RPCError::ParseResponse(format!("Error: {err}, Body: {body}"))
1827            })?;
1828        trace!(?dto_response, "Received traced_entry_points response from Tycho server");
1829        let data: TracedEntryPoints = dto_response
1830            .traced_entry_points
1831            .into_iter()
1832            .map(|(k, v)| {
1833                (
1834                    k,
1835                    v.into_iter()
1836                        .map(|(ep, tr)| {
1837                            (EntryPointWithTracingParams::from(ep), TracingResult::from(tr))
1838                        })
1839                        .collect(),
1840                )
1841            })
1842            .collect();
1843        Ok(Page::new(
1844            data,
1845            dto_response.pagination.total,
1846            dto_response.pagination.page,
1847            dto_response.pagination.page_size,
1848        ))
1849    }
1850
1851    async fn get_snapshots<'a>(
1852        &self,
1853        request: &SnapshotParameters<'a>,
1854        chunk_size: Option<usize>,
1855        concurrency: usize,
1856    ) -> Result<Snapshot, RPCError> {
1857        let component_ids: Vec<_> = request
1858            .components
1859            .keys()
1860            .cloned()
1861            .collect();
1862
1863        let component_tvl = if request.include_tvl && !component_ids.is_empty() {
1864            self.get_component_tvl_paginated(
1865                ComponentTvlPaginatedParams::new(request.chain, concurrency)
1866                    .with_component_ids(component_ids.clone()),
1867            )
1868            .await?
1869        } else {
1870            HashMap::new()
1871        };
1872
1873        let version = VersionParam::at_block(request.chain.into(), request.block_number);
1874
1875        let mut protocol_states = if !component_ids.is_empty() {
1876            self.get_protocol_states_paginated(
1877                ProtocolStatesPaginatedParams::new(
1878                    request.chain,
1879                    request.protocol_system,
1880                    concurrency,
1881                )
1882                .with_protocol_ids(component_ids.clone())
1883                .with_include_balances(request.include_balances)
1884                .with_version(version.clone()),
1885            )
1886            .await?
1887            .into_iter()
1888            .map(|state| (state.component_id.clone(), state))
1889            .collect()
1890        } else {
1891            HashMap::new()
1892        };
1893
1894        // Convert to ComponentWithState, which includes entrypoint information.
1895        let states = request
1896            .components
1897            .values()
1898            .filter_map(|component| {
1899                if let Some(state) = protocol_states.remove(&component.id) {
1900                    Some((
1901                        component.id.clone(),
1902                        ComponentWithState {
1903                            state,
1904                            component: component.clone(),
1905                            component_tvl: component_tvl
1906                                .get(&component.id)
1907                                .cloned(),
1908                            entrypoints: request
1909                                .entrypoints
1910                                .as_ref()
1911                                .and_then(|map| map.get(&component.id))
1912                                .cloned()
1913                                .unwrap_or_default(),
1914                        },
1915                    ))
1916                } else if component_ids.contains(&component.id) {
1917                    // only emit error event if we requested this component
1918                    let component_id = &component.id;
1919                    error!(?component_id, "Missing state for native component!");
1920                    None
1921                } else {
1922                    None
1923                }
1924            })
1925            .collect();
1926
1927        let vm_storage = if !request.contract_ids.is_empty() {
1928            let mut cp_params = ContractStatePaginatedParams::new(
1929                request.chain,
1930                request.protocol_system,
1931                concurrency,
1932            )
1933            .with_contract_ids(request.contract_ids.to_vec())
1934            .with_version(version.clone());
1935            if let Some(cs) = chunk_size {
1936                cp_params = cp_params.with_chunk_size(cs);
1937            }
1938            let contract_states = self
1939                .get_contract_state_paginated(cp_params)
1940                .await?
1941                .into_iter()
1942                .map(|acc| (acc.address.clone(), acc))
1943                .collect::<HashMap<_, _>>();
1944
1945            trace!(states=?&contract_states, "Retrieved ContractState");
1946
1947            let contract_address_to_components = request
1948                .components
1949                .iter()
1950                .filter_map(|(id, comp)| {
1951                    if component_ids.contains(id) {
1952                        Some(
1953                            comp.contract_addresses
1954                                .iter()
1955                                .map(|address| (address.clone(), comp.id.clone())),
1956                        )
1957                    } else {
1958                        None
1959                    }
1960                })
1961                .flatten()
1962                .fold(HashMap::<Bytes, Vec<String>>::new(), |mut acc, (addr, c_id)| {
1963                    acc.entry(addr).or_default().push(c_id);
1964                    acc
1965                });
1966
1967            request
1968                .contract_ids
1969                .iter()
1970                .filter_map(|address| {
1971                    if let Some(state) = contract_states.get(address) {
1972                        Some((address.clone(), state.clone()))
1973                    } else if let Some(ids) = contract_address_to_components.get(address) {
1974                        // only emit error even if we did actually request this address
1975                        error!(
1976                            ?address,
1977                            ?ids,
1978                            "Component with lacking contract storage encountered!"
1979                        );
1980                        None
1981                    } else {
1982                        None
1983                    }
1984                })
1985                .collect()
1986        } else {
1987            HashMap::new()
1988        };
1989
1990        Ok(Snapshot { states, vm_storage })
1991    }
1992}
1993
1994#[cfg(test)]
1995mod tests {
1996    use std::{
1997        collections::{HashMap, HashSet},
1998        str::FromStr,
1999    };
2000
2001    use mockito::Server;
2002    use rstest::rstest;
2003    use tycho_common::models::blockchain::AddressStorageLocation;
2004
2005    use super::*;
2006
2007    // Dummy implementation of `get_protocol_states_paginated` for backwards compatibility testing
2008    // purposes
2009    impl MockRPCClient {
2010        #[allow(clippy::too_many_arguments)]
2011        async fn test_get_protocol_states_paginated<T>(
2012            &self,
2013            chain: Chain,
2014            ids: &[T],
2015            protocol_system: &str,
2016            include_balances: bool,
2017            block_number: Option<u64>,
2018            chunk_size: usize,
2019            _concurrency: usize,
2020        ) -> Vec<(Chain, Vec<String>, String, bool, Option<u64>, PaginationParams)>
2021        where
2022            T: AsRef<str> + Clone + Send + Sync + 'static,
2023        {
2024            ids.chunks(chunk_size)
2025                .map(|chunk| {
2026                    (
2027                        chain,
2028                        chunk
2029                            .iter()
2030                            .map(|id| id.as_ref().to_string())
2031                            .collect(),
2032                        protocol_system.to_string(),
2033                        include_balances,
2034                        block_number,
2035                        PaginationParams { page: 0, page_size: chunk_size as i64 },
2036                    )
2037                })
2038                .collect()
2039        }
2040    }
2041
2042    const GET_CONTRACT_STATE_RESP: &str = r#"
2043        {
2044            "accounts": [
2045                {
2046                    "chain": "ethereum",
2047                    "address": "0x0000000000000000000000000000000000000000",
2048                    "title": "",
2049                    "slots": {},
2050                    "native_balance": "0x01f4",
2051                    "token_balances": {},
2052                    "code": "0x00",
2053                    "code_hash": "0x5c06b7c5b3d910fd33bc2229846f9ddaf91d584d9b196e16636901ac3a77077e",
2054                    "balance_modify_tx": "0x0000000000000000000000000000000000000000000000000000000000000000",
2055                    "code_modify_tx": "0x0000000000000000000000000000000000000000000000000000000000000000",
2056                    "creation_tx": null
2057                }
2058            ],
2059            "pagination": {
2060                "page": 0,
2061                "page_size": 20,
2062                "total": 10
2063            }
2064        }
2065        "#;
2066
2067    #[rstest]
2068    #[case::string_input(vec![
2069        "id1".to_string(),
2070        "id2".to_string()
2071    ])]
2072    #[tokio::test]
2073    async fn test_get_protocol_states_paginated<T>(#[case] ids: Vec<T>)
2074    where
2075        T: AsRef<str> + Clone + Send + Sync + 'static,
2076    {
2077        let mock_client = MockRPCClient::new();
2078
2079        let request_args = mock_client
2080            .test_get_protocol_states_paginated(
2081                Chain::Ethereum,
2082                &ids,
2083                "test_system",
2084                true,
2085                None,
2086                2,
2087                2,
2088            )
2089            .await;
2090
2091        // Verify that the request args have been split into chunks correctly
2092        assert_eq!(request_args.len(), 1);
2093        assert_eq!(request_args[0].1.len(), 2);
2094    }
2095
2096    #[tokio::test]
2097    async fn test_get_contract_state() {
2098        let mut server = Server::new_async().await;
2099        let server_resp = GET_CONTRACT_STATE_RESP;
2100        // test that the response is deserialized correctly
2101        serde_json::from_str::<StateRequestResponse>(server_resp).expect("deserialize");
2102
2103        let mocked_server = server
2104            .mock("POST", "/v1/contract_state")
2105            .expect(1)
2106            .with_body(server_resp)
2107            .create_async()
2108            .await;
2109
2110        let client = HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
2111            .expect("create client");
2112
2113        let accounts = client
2114            .get_contract_state(ContractStateParams::new(Chain::Ethereum, ""))
2115            .await
2116            .expect("get state");
2117
2118        mocked_server.assert();
2119        assert_eq!(accounts.data().len(), 1);
2120        assert_eq!(accounts.data()[0].slots, HashMap::new());
2121        assert_eq!(accounts.data()[0].native_balance, Bytes::from(500u16.to_be_bytes()));
2122        assert_eq!(accounts.data()[0].code, [0].to_vec());
2123        assert_eq!(
2124            accounts.data()[0].code_hash,
2125            hex::decode("5c06b7c5b3d910fd33bc2229846f9ddaf91d584d9b196e16636901ac3a77077e")
2126                .unwrap()
2127        );
2128    }
2129
2130    #[tokio::test]
2131    async fn test_get_contract_state_paginated_requests_default_sized_chunks() {
2132        let mut server = Server::new_async().await;
2133        let full_chunks = server
2134            .mock("POST", "/v1/contract_state")
2135            .match_body(mockito::Matcher::PartialJson(serde_json::json!({
2136                "pagination": {"page": 0, "page_size": 100}
2137            })))
2138            .expect(3)
2139            .with_body(GET_CONTRACT_STATE_RESP)
2140            .create_async()
2141            .await;
2142        let client = HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
2143            .expect("create client");
2144        let ids: Vec<Bytes> = (0..250u16)
2145            .map(|i| Bytes::from(i.to_be_bytes()))
2146            .collect();
2147
2148        let accounts = client
2149            .get_contract_state_paginated(
2150                ContractStatePaginatedParams::new(Chain::Ethereum, "", RPC_CLIENT_CONCURRENCY)
2151                    .with_contract_ids(ids),
2152            )
2153            .await
2154            .expect("get state");
2155
2156        full_chunks.assert();
2157        assert_eq!(accounts.len(), 3);
2158    }
2159
2160    #[tokio::test]
2161    async fn test_client_metadata_header_sent_when_set() {
2162        let mut server = Server::new_async().await;
2163        let expected_ua = format!("tycho-client-{}", env!("CARGO_PKG_VERSION"));
2164        let mock = server
2165            .mock("POST", "/v1/contract_state")
2166            .match_header(CLIENT_METADATA_HEADER, "fynd_version=0.57.0;preset=best")
2167            .match_header("user-agent", expected_ua.as_str())
2168            .with_body(GET_CONTRACT_STATE_RESP)
2169            .expect(1)
2170            .create_async()
2171            .await;
2172
2173        let client = HttpRPCClient::new(
2174            server.url().as_str(),
2175            HttpRPCClientOptions::default()
2176                .with_client_metadata_header(Some("fynd_version=0.57.0;preset=best".to_string())),
2177        )
2178        .expect("create client");
2179
2180        client
2181            .get_contract_state(ContractStateParams::new(Chain::Ethereum, ""))
2182            .await
2183            .expect("get state");
2184
2185        mock.assert_async().await;
2186    }
2187
2188    #[tokio::test]
2189    async fn test_no_client_metadata_header_when_unset() {
2190        let mut server = Server::new_async().await;
2191        let expected_ua = format!("tycho-client-{}", env!("CARGO_PKG_VERSION"));
2192        let mock = server
2193            .mock("POST", "/v1/contract_state")
2194            .match_header(CLIENT_METADATA_HEADER, mockito::Matcher::Missing)
2195            .match_header("user-agent", expected_ua.as_str())
2196            .with_body(GET_CONTRACT_STATE_RESP)
2197            .expect(1)
2198            .create_async()
2199            .await;
2200
2201        let client = HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
2202            .expect("create client");
2203
2204        client
2205            .get_contract_state(ContractStateParams::new(Chain::Ethereum, ""))
2206            .await
2207            .expect("get state");
2208
2209        mock.assert_async().await;
2210    }
2211
2212    #[tokio::test]
2213    async fn test_get_protocol_components() {
2214        let mut server = Server::new_async().await;
2215        let server_resp = r#"
2216        {
2217            "protocol_components": [
2218                {
2219                    "id": "State1",
2220                    "protocol_system": "ambient",
2221                    "protocol_type_name": "Pool",
2222                    "chain": "ethereum",
2223                    "tokens": [
2224                        "0x0000000000000000000000000000000000000000",
2225                        "0x0000000000000000000000000000000000000001"
2226                    ],
2227                    "contract_ids": [
2228                        "0x0000000000000000000000000000000000000000"
2229                    ],
2230                    "static_attributes": {
2231                        "attribute_1": "0x00000000000003e8"
2232                    },
2233                    "change": "Creation",
2234                    "creation_tx": "0x0000000000000000000000000000000000000000000000000000000000000000",
2235                    "created_at": "2022-01-01T00:00:00"
2236                }
2237            ],
2238            "pagination": {
2239                "page": 0,
2240                "page_size": 20,
2241                "total": 10
2242            }
2243        }
2244        "#;
2245        // test that the response is deserialized correctly
2246        serde_json::from_str::<ProtocolComponentRequestResponse>(server_resp).expect("deserialize");
2247
2248        let mocked_server = server
2249            .mock("POST", "/v1/protocol_components")
2250            .expect(1)
2251            .with_body(server_resp)
2252            .create_async()
2253            .await;
2254
2255        let client = HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
2256            .expect("create client");
2257
2258        let components = client
2259            .get_protocol_components(ProtocolComponentsParams::new(Chain::Ethereum, ""))
2260            .await
2261            .expect("get state");
2262
2263        mocked_server.assert();
2264        assert_eq!(components.data().len(), 1);
2265        assert_eq!(components.data()[0].id, "State1");
2266        assert_eq!(components.data()[0].protocol_system, "ambient");
2267        assert_eq!(components.data()[0].protocol_type_name, "Pool");
2268        assert_eq!(components.data()[0].tokens.len(), 2);
2269        let expected_attributes =
2270            [("attribute_1".to_string(), Bytes::from(1000_u64.to_be_bytes()))]
2271                .iter()
2272                .cloned()
2273                .collect::<HashMap<String, Bytes>>();
2274        assert_eq!(components.data()[0].static_attributes, expected_attributes);
2275    }
2276
2277    #[tokio::test]
2278    async fn test_get_protocol_states() {
2279        let mut server = Server::new_async().await;
2280        let server_resp = r#"
2281        {
2282            "states": [
2283                {
2284                    "component_id": "State1",
2285                    "attributes": {
2286                        "attribute_1": "0x00000000000003e8"
2287                    },
2288                    "balances": {
2289                        "0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2": "0x01f4"
2290                    }
2291                }
2292            ],
2293            "pagination": {
2294                "page": 0,
2295                "page_size": 20,
2296                "total": 10
2297            }
2298        }
2299        "#;
2300        // test that the response is deserialized correctly
2301        serde_json::from_str::<ProtocolStateRequestResponse>(server_resp).expect("deserialize");
2302
2303        let mocked_server = server
2304            .mock("POST", "/v1/protocol_state")
2305            .expect(1)
2306            .with_body(server_resp)
2307            .create_async()
2308            .await;
2309        let client = HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
2310            .expect("create client");
2311
2312        let states = client
2313            .get_protocol_states(
2314                ProtocolStatesParams::new(Chain::Ethereum, "").with_include_balances(true),
2315            )
2316            .await
2317            .expect("get state");
2318
2319        mocked_server.assert();
2320        assert_eq!(states.data().len(), 1);
2321        assert_eq!(states.data()[0].component_id, "State1");
2322        let expected_attributes =
2323            [("attribute_1".to_string(), Bytes::from(1000_u64.to_be_bytes()))]
2324                .iter()
2325                .cloned()
2326                .collect::<HashMap<String, Bytes>>();
2327        assert_eq!(states.data()[0].attributes, expected_attributes);
2328        let expected_balances = [(
2329            Bytes::from_str("0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2")
2330                .expect("Unsupported address format"),
2331            Bytes::from_str("0x01f4").unwrap(),
2332        )]
2333        .iter()
2334        .cloned()
2335        .collect::<HashMap<Bytes, Bytes>>();
2336        assert_eq!(states.data()[0].balances, expected_balances);
2337    }
2338
2339    #[tokio::test]
2340    async fn test_get_tokens() {
2341        let mut server = Server::new_async().await;
2342        let server_resp = r#"
2343        {
2344            "tokens": [
2345              {
2346                "chain": "ethereum",
2347                "address": "0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2",
2348                "symbol": "WETH",
2349                "decimals": 18,
2350                "tax": 0,
2351                "gas": [
2352                  29962
2353                ],
2354                "quality": 100
2355              },
2356              {
2357                "chain": "ethereum",
2358                "address": "0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48",
2359                "symbol": "USDC",
2360                "decimals": 6,
2361                "tax": 0,
2362                "gas": [
2363                  40652
2364                ],
2365                "quality": 100
2366              }
2367            ],
2368            "pagination": {
2369              "page": 0,
2370              "page_size": 20,
2371              "total": 10
2372            }
2373          }
2374        "#;
2375        // test that the response is deserialized correctly
2376        serde_json::from_str::<TokensRequestResponse>(server_resp).expect("deserialize");
2377
2378        let mocked_server = server
2379            .mock("POST", "/v1/tokens")
2380            .expect(1)
2381            .with_body(server_resp)
2382            .create_async()
2383            .await;
2384        let client = HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
2385            .expect("create client");
2386
2387        let tokens = client
2388            .get_tokens(TokensParams::new(Chain::Ethereum))
2389            .await
2390            .expect("get tokens");
2391
2392        let expected = vec![
2393            Token {
2394                chain: tycho_common::models::Chain::Ethereum,
2395                address: Bytes::from_str("0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2").unwrap(),
2396                symbol: "WETH".to_string(),
2397                decimals: 18,
2398                tax: 0,
2399                gas: vec![Some(29962)],
2400                quality: 100,
2401            },
2402            Token {
2403                chain: tycho_common::models::Chain::Ethereum,
2404                address: Bytes::from_str("0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48").unwrap(),
2405                symbol: "USDC".to_string(),
2406                decimals: 6,
2407                tax: 0,
2408                gas: vec![Some(40652)],
2409                quality: 100,
2410            },
2411        ];
2412
2413        mocked_server.assert();
2414        assert_eq!(*tokens.data(), expected);
2415    }
2416
2417    #[rstest]
2418    #[case::with_dci(Some(vec!["system2"]), vec!["system2"])]
2419    #[case::backward_compat(None, vec![])]
2420    #[tokio::test]
2421    async fn test_get_protocol_systems(
2422        #[case] dci_protocols: Option<Vec<&str>>,
2423        #[case] expected_dci: Vec<&str>,
2424    ) {
2425        use serde_json::json;
2426
2427        let mut json_value = json!({
2428            "protocol_systems": ["system1", "system2"],
2429            "pagination": { "page": 0, "page_size": 20, "total": 2 }
2430        });
2431        if let Some(dci) = dci_protocols {
2432            json_value["dci_protocols"] = json!(dci);
2433        }
2434        let server_resp = serde_json::to_string(&json_value).unwrap();
2435
2436        let mut server = Server::new_async().await;
2437        let mocked_server = server
2438            .mock("POST", "/v1/protocol_systems")
2439            .expect(1)
2440            .with_body(&server_resp)
2441            .create_async()
2442            .await;
2443        let client = HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
2444            .expect("create client");
2445
2446        let response = client
2447            .get_protocol_systems(ProtocolSystemsParams::new(Chain::Ethereum))
2448            .await
2449            .expect("get protocol systems");
2450
2451        mocked_server.assert();
2452        assert_eq!(response.data().protocol_systems(), ["system1", "system2"]);
2453        assert_eq!(response.data().dci_protocols(), expected_dci.as_slice());
2454    }
2455
2456    #[tokio::test]
2457    async fn test_get_component_tvl() {
2458        let mut server = Server::new_async().await;
2459        let server_resp = r#"
2460        {
2461            "tvl": {
2462                "component1": 100.0
2463            },
2464            "pagination": {
2465                "page": 0,
2466                "page_size": 20,
2467                "total": 10
2468            }
2469        }
2470        "#;
2471        // test that the response is deserialized correctly
2472        serde_json::from_str::<ComponentTvlRequestResponse>(server_resp).expect("deserialize");
2473
2474        let mocked_server = server
2475            .mock("POST", "/v1/component_tvl")
2476            .expect(1)
2477            .with_body(server_resp)
2478            .create_async()
2479            .await;
2480        let client = HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
2481            .expect("create client");
2482
2483        let component_tvl = client
2484            .get_component_tvl(ComponentTvlParams::new(Chain::Ethereum))
2485            .await
2486            .expect("get component tvl");
2487
2488        mocked_server.assert();
2489        assert_eq!(component_tvl.data().get("component1"), Some(&100.0));
2490    }
2491
2492    #[tokio::test]
2493    async fn test_get_traced_entry_points() {
2494        let mut server = Server::new_async().await;
2495        let server_resp = r#"
2496        {
2497            "traced_entry_points": {
2498                "component_1": [
2499                    [
2500                        {
2501                            "entry_point": {
2502                                "external_id": "entrypoint_a",
2503                                "target": "0x0000000000000000000000000000000000000001",
2504                                "signature": "sig()"
2505                            },
2506                            "params": {
2507                                "method": "rpctracer",
2508                                "caller": "0x000000000000000000000000000000000000000a",
2509                                "calldata": "0x000000000000000000000000000000000000000b"
2510                            }
2511                        },
2512                        {
2513                            "retriggers": [
2514                                [
2515                                    "0x00000000000000000000000000000000000000aa",
2516                                    {"key": "0x0000000000000000000000000000000000000aaa", "offset": 12}
2517                                ]
2518                            ],
2519                            "accessed_slots": {
2520                                "0x0000000000000000000000000000000000aaaa": [
2521                                    "0x0000000000000000000000000000000000aaaa"
2522                                ]
2523                            }
2524                        }
2525                    ]
2526                ]
2527            },
2528            "pagination": {
2529                "page": 0,
2530                "page_size": 20,
2531                "total": 1
2532            }
2533        }
2534        "#;
2535        // test that the response is deserialized correctly
2536        serde_json::from_str::<TracedEntryPointRequestResponse>(server_resp).expect("deserialize");
2537
2538        let mocked_server = server
2539            .mock("POST", "/v1/traced_entry_points")
2540            .expect(1)
2541            .with_body(server_resp)
2542            .create_async()
2543            .await;
2544        let client = HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
2545            .expect("create client");
2546
2547        let entrypoints = client
2548            .get_traced_entry_points(TracedEntryPointsParams::new(Chain::Ethereum, ""))
2549            .await
2550            .expect("get traced entry points");
2551
2552        mocked_server.assert();
2553        assert_eq!(entrypoints.data().len(), 1);
2554        let comp1_entrypoints = entrypoints
2555            .data()
2556            .get("component_1")
2557            .expect("component_1 entrypoints should exist");
2558        assert_eq!(comp1_entrypoints.len(), 1);
2559
2560        let (entrypoint, trace_result) = &comp1_entrypoints[0];
2561        assert_eq!(entrypoint.entry_point.external_id, "entrypoint_a");
2562        assert_eq!(
2563            entrypoint.entry_point.target,
2564            Bytes::from_str("0x0000000000000000000000000000000000000001").unwrap()
2565        );
2566        assert_eq!(entrypoint.entry_point.signature, "sig()");
2567        let tycho_common::models::blockchain::TracingParams::RPCTracer(rpc_params) =
2568            &entrypoint.params;
2569        assert_eq!(
2570            rpc_params.caller,
2571            Some(Bytes::from("0x000000000000000000000000000000000000000a"))
2572        );
2573        assert_eq!(rpc_params.calldata, Bytes::from("0x000000000000000000000000000000000000000b"));
2574
2575        assert_eq!(
2576            trace_result.retriggers,
2577            HashSet::from([(
2578                Bytes::from("0x00000000000000000000000000000000000000aa"),
2579                AddressStorageLocation::new(
2580                    Bytes::from("0x0000000000000000000000000000000000000aaa"),
2581                    12
2582                )
2583            )])
2584        );
2585        assert_eq!(trace_result.accessed_slots.len(), 1);
2586        assert_eq!(
2587            trace_result.accessed_slots,
2588            HashMap::from([(
2589                Bytes::from("0x0000000000000000000000000000000000aaaa"),
2590                HashSet::from([Bytes::from("0x0000000000000000000000000000000000aaaa")])
2591            )])
2592        );
2593    }
2594
2595    #[tokio::test]
2596    async fn test_parse_retry_value_numeric() {
2597        let result = parse_retry_value("60");
2598        assert!(result.is_some());
2599
2600        let expected_time = SystemTime::now() + Duration::from_secs(60);
2601        let actual_time = result.unwrap();
2602
2603        // Allow for small timing differences during test execution
2604        let diff = if actual_time > expected_time {
2605            actual_time
2606                .duration_since(expected_time)
2607                .unwrap()
2608        } else {
2609            expected_time
2610                .duration_since(actual_time)
2611                .unwrap()
2612        };
2613        assert!(diff < Duration::from_secs(1), "Time difference too large: {:?}", diff);
2614    }
2615
2616    #[tokio::test]
2617    async fn test_parse_retry_value_rfc2822() {
2618        // Use a fixed future date in RFC2822 format
2619        let rfc2822_date = "Sat, 01 Jan 2030 12:00:00 +0000";
2620        let result = parse_retry_value(rfc2822_date);
2621        assert!(result.is_some());
2622
2623        let parsed_time = result.unwrap();
2624        assert!(parsed_time > SystemTime::now());
2625    }
2626
2627    #[tokio::test]
2628    async fn test_parse_retry_value_invalid_formats() {
2629        // Test various invalid formats
2630        assert!(parse_retry_value("invalid").is_none());
2631        assert!(parse_retry_value("").is_none());
2632        assert!(parse_retry_value("not_a_number").is_none());
2633        assert!(parse_retry_value("Mon, 32 Jan 2030 25:00:00 +0000").is_none());
2634        // Invalid date
2635    }
2636
2637    #[tokio::test]
2638    async fn test_parse_retry_value_zero_seconds() {
2639        let result = parse_retry_value("0");
2640        assert!(result.is_some());
2641
2642        let expected_time = SystemTime::now();
2643        let actual_time = result.unwrap();
2644
2645        // Should be very close to current time
2646        let diff = if actual_time > expected_time {
2647            actual_time
2648                .duration_since(expected_time)
2649                .unwrap()
2650        } else {
2651            expected_time
2652                .duration_since(actual_time)
2653                .unwrap()
2654        };
2655        assert!(diff < Duration::from_secs(1));
2656    }
2657
2658    #[tokio::test]
2659    async fn test_error_for_response_rate_limited() {
2660        let mut server = Server::new_async().await;
2661        let mock = server
2662            .mock("GET", "/test")
2663            .with_status(429)
2664            .with_header("Retry-After", "60")
2665            .create_async()
2666            .await;
2667
2668        let client = reqwest::Client::new();
2669        let response = client
2670            .get(format!("{}/test", server.url()))
2671            .send()
2672            .await
2673            .unwrap();
2674
2675        let http_client =
2676            HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
2677                .unwrap()
2678                .with_test_backoff_policy();
2679        let result = http_client
2680            .error_for_response(response)
2681            .await;
2682
2683        mock.assert();
2684        assert!(matches!(result, Err(RPCError::RateLimited(_))));
2685        if let Err(RPCError::RateLimited(retry_after)) = result {
2686            assert!(retry_after.is_some());
2687        }
2688    }
2689
2690    #[tokio::test]
2691    async fn test_error_for_response_rate_limited_no_header() {
2692        let mut server = Server::new_async().await;
2693        let mock = server
2694            .mock("GET", "/test")
2695            .with_status(429)
2696            .create_async()
2697            .await;
2698
2699        let client = reqwest::Client::new();
2700        let response = client
2701            .get(format!("{}/test", server.url()))
2702            .send()
2703            .await
2704            .unwrap();
2705
2706        let http_client =
2707            HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
2708                .unwrap()
2709                .with_test_backoff_policy();
2710        let result = http_client
2711            .error_for_response(response)
2712            .await;
2713
2714        mock.assert();
2715        assert!(matches!(result, Err(RPCError::RateLimited(None))));
2716    }
2717
2718    #[tokio::test]
2719    async fn test_error_for_response_server_errors() {
2720        let test_cases =
2721            vec![(502, "Bad Gateway"), (503, "Service Unavailable"), (504, "Gateway Timeout")];
2722
2723        for (status_code, expected_body) in test_cases {
2724            let mut server = Server::new_async().await;
2725            let mock = server
2726                .mock("GET", "/test")
2727                .with_status(status_code)
2728                .with_body(expected_body)
2729                .create_async()
2730                .await;
2731
2732            let client = reqwest::Client::new();
2733            let response = client
2734                .get(format!("{}/test", server.url()))
2735                .send()
2736                .await
2737                .unwrap();
2738
2739            let http_client =
2740                HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
2741                    .unwrap()
2742                    .with_test_backoff_policy();
2743            let result = http_client
2744                .error_for_response(response)
2745                .await;
2746
2747            mock.assert();
2748            assert!(matches!(result, Err(RPCError::ServerUnreachable(_))));
2749            if let Err(RPCError::ServerUnreachable(body)) = result {
2750                assert_eq!(body, expected_body);
2751            }
2752        }
2753    }
2754
2755    #[tokio::test]
2756    async fn test_error_for_response_success() {
2757        let mut server = Server::new_async().await;
2758        let mock = server
2759            .mock("GET", "/test")
2760            .with_status(200)
2761            .with_body("success")
2762            .create_async()
2763            .await;
2764
2765        let client = reqwest::Client::new();
2766        let response = client
2767            .get(format!("{}/test", server.url()))
2768            .send()
2769            .await
2770            .unwrap();
2771
2772        let http_client =
2773            HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
2774                .unwrap()
2775                .with_test_backoff_policy();
2776        let result = http_client
2777            .error_for_response(response)
2778            .await;
2779
2780        mock.assert();
2781        assert!(result.is_ok());
2782
2783        let response = result.unwrap();
2784        assert_eq!(response.status(), 200);
2785    }
2786
2787    #[tokio::test]
2788    async fn test_handle_error_for_backoff_server_unreachable() {
2789        let http_client =
2790            HttpRPCClient::new("http://localhost:8080", HttpRPCClientOptions::default())
2791                .unwrap()
2792                .with_test_backoff_policy();
2793        let error = RPCError::ServerUnreachable("Service down".to_string());
2794
2795        let backoff_error = http_client
2796            .handle_error_for_backoff(error)
2797            .await;
2798
2799        match backoff_error {
2800            backoff::Error::Transient { err: RPCError::ServerUnreachable(msg), retry_after } => {
2801                assert_eq!(msg, "Service down");
2802                assert_eq!(retry_after, Some(Duration::from_millis(50))); // Fast test duration
2803            }
2804            _ => panic!("Expected transient error for ServerUnreachable"),
2805        }
2806    }
2807
2808    #[tokio::test]
2809    async fn test_handle_error_for_backoff_rate_limited_with_retry_after() {
2810        let http_client =
2811            HttpRPCClient::new("http://localhost:8080", HttpRPCClientOptions::default())
2812                .unwrap()
2813                .with_test_backoff_policy();
2814        let future_time = SystemTime::now() + Duration::from_secs(30);
2815        let error = RPCError::RateLimited(Some(future_time));
2816
2817        let backoff_error = http_client
2818            .handle_error_for_backoff(error)
2819            .await;
2820
2821        match backoff_error {
2822            backoff::Error::Transient { err: RPCError::RateLimited(retry_after), .. } => {
2823                assert_eq!(retry_after, Some(future_time));
2824            }
2825            _ => panic!("Expected transient error for RateLimited"),
2826        }
2827
2828        // Verify that retry_after was stored in the client state
2829        let stored_retry_after = http_client.retry_after.read().await;
2830        assert_eq!(*stored_retry_after, Some(future_time));
2831    }
2832
2833    #[tokio::test]
2834    async fn test_handle_error_for_backoff_rate_limited_no_retry_after() {
2835        let http_client =
2836            HttpRPCClient::new("http://localhost:8080", HttpRPCClientOptions::default())
2837                .unwrap()
2838                .with_test_backoff_policy();
2839        let error = RPCError::RateLimited(None);
2840
2841        let backoff_error = http_client
2842            .handle_error_for_backoff(error)
2843            .await;
2844
2845        match backoff_error {
2846            backoff::Error::Transient { err: RPCError::RateLimited(None), .. } => {
2847                // This is expected - no retry-after still allows retries with default policy
2848            }
2849            _ => panic!("Expected transient error for RateLimited without retry-after"),
2850        }
2851    }
2852
2853    #[tokio::test]
2854    async fn test_handle_error_for_backoff_other_errors() {
2855        let http_client =
2856            HttpRPCClient::new("http://localhost:8080", HttpRPCClientOptions::default())
2857                .unwrap()
2858                .with_test_backoff_policy();
2859        let error = RPCError::ParseResponse("Invalid JSON".to_string());
2860
2861        let backoff_error = http_client
2862            .handle_error_for_backoff(error)
2863            .await;
2864
2865        match backoff_error {
2866            backoff::Error::Permanent(RPCError::ParseResponse(msg)) => {
2867                assert_eq!(msg, "Invalid JSON");
2868            }
2869            _ => panic!("Expected permanent error for ParseResponse"),
2870        }
2871    }
2872
2873    #[tokio::test]
2874    async fn test_wait_until_retry_after_no_retry_time() {
2875        let http_client =
2876            HttpRPCClient::new("http://localhost:8080", HttpRPCClientOptions::default())
2877                .unwrap()
2878                .with_test_backoff_policy();
2879
2880        let start = std::time::Instant::now();
2881        http_client
2882            .wait_until_retry_after()
2883            .await;
2884        let elapsed = start.elapsed();
2885
2886        // Should return immediately if no retry time is set
2887        assert!(elapsed < Duration::from_millis(100));
2888    }
2889
2890    #[tokio::test]
2891    async fn test_wait_until_retry_after_past_time() {
2892        let http_client =
2893            HttpRPCClient::new("http://localhost:8080", HttpRPCClientOptions::default())
2894                .unwrap()
2895                .with_test_backoff_policy();
2896
2897        // Set a retry time in the past
2898        let past_time = SystemTime::now() - Duration::from_secs(10);
2899        *http_client.retry_after.write().await = Some(past_time);
2900
2901        let start = std::time::Instant::now();
2902        http_client
2903            .wait_until_retry_after()
2904            .await;
2905        let elapsed = start.elapsed();
2906
2907        // Should return immediately if retry time is in the past
2908        assert!(elapsed < Duration::from_millis(100));
2909    }
2910
2911    #[tokio::test]
2912    async fn test_wait_until_retry_after_future_time() {
2913        let http_client =
2914            HttpRPCClient::new("http://localhost:8080", HttpRPCClientOptions::default())
2915                .unwrap()
2916                .with_test_backoff_policy();
2917
2918        // Set a retry time 100ms in the future
2919        let future_time = SystemTime::now() + Duration::from_millis(100);
2920        *http_client.retry_after.write().await = Some(future_time);
2921
2922        let start = std::time::Instant::now();
2923        http_client
2924            .wait_until_retry_after()
2925            .await;
2926        let elapsed = start.elapsed();
2927
2928        // Should wait approximately the specified duration
2929        assert!(elapsed >= Duration::from_millis(80)); // Allow some tolerance
2930        assert!(elapsed <= Duration::from_millis(200)); // Upper bound for test stability
2931    }
2932
2933    #[tokio::test]
2934    async fn test_make_post_request_success() {
2935        let mut server = Server::new_async().await;
2936        let server_resp = r#"{"success": true}"#;
2937
2938        let mock = server
2939            .mock("POST", "/test")
2940            .with_status(200)
2941            .with_body(server_resp)
2942            .create_async()
2943            .await;
2944
2945        let http_client =
2946            HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
2947                .unwrap()
2948                .with_test_backoff_policy();
2949        let request_body = serde_json::json!({"test": "data"});
2950        let uri = format!("{}/test", server.url());
2951
2952        let result = http_client
2953            .make_post_request(&request_body, &uri)
2954            .await;
2955
2956        mock.assert();
2957        assert!(result.is_ok());
2958
2959        let response = result.unwrap();
2960        assert_eq!(response.status(), 200);
2961        assert_eq!(response.text().await.unwrap(), server_resp);
2962    }
2963
2964    #[tokio::test]
2965    async fn test_make_post_request_retry_on_server_error() {
2966        let mut server = Server::new_async().await;
2967        // First request fails with 503, second succeeds
2968        let error_mock = server
2969            .mock("POST", "/test")
2970            .with_status(503)
2971            .with_body("Service Unavailable")
2972            .expect(1)
2973            .create_async()
2974            .await;
2975
2976        let success_mock = server
2977            .mock("POST", "/test")
2978            .with_status(200)
2979            .with_body(r#"{"success": true}"#)
2980            .expect(1)
2981            .create_async()
2982            .await;
2983
2984        let http_client =
2985            HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
2986                .unwrap()
2987                .with_test_backoff_policy();
2988        let request_body = serde_json::json!({"test": "data"});
2989        let uri = format!("{}/test", server.url());
2990
2991        let result = http_client
2992            .make_post_request(&request_body, &uri)
2993            .await;
2994
2995        error_mock.assert();
2996        success_mock.assert();
2997        assert!(result.is_ok());
2998    }
2999
3000    #[tokio::test]
3001    async fn test_make_post_request_respect_retry_after_header() {
3002        let mut server = Server::new_async().await;
3003
3004        // First request returns 429 with retry-after, second succeeds
3005        let rate_limit_mock = server
3006            .mock("POST", "/test")
3007            .with_status(429)
3008            .with_header("Retry-After", "1") // 1 second
3009            .expect(1)
3010            .create_async()
3011            .await;
3012
3013        let success_mock = server
3014            .mock("POST", "/test")
3015            .with_status(200)
3016            .with_body(r#"{"success": true}"#)
3017            .expect(1)
3018            .create_async()
3019            .await;
3020
3021        let http_client =
3022            HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
3023                .unwrap()
3024                .with_test_backoff_policy();
3025        let request_body = serde_json::json!({"test": "data"});
3026        let uri = format!("{}/test", server.url());
3027
3028        let start = std::time::Instant::now();
3029        let result = http_client
3030            .make_post_request(&request_body, &uri)
3031            .await;
3032        let elapsed = start.elapsed();
3033
3034        rate_limit_mock.assert();
3035        success_mock.assert();
3036        assert!(result.is_ok());
3037
3038        // Should have waited at least 1 second due to retry-after header
3039        assert!(elapsed >= Duration::from_millis(900)); // Allow some tolerance
3040        assert!(elapsed <= Duration::from_millis(2000)); // Upper bound for test stability
3041    }
3042
3043    #[tokio::test]
3044    async fn test_make_post_request_permanent_error() {
3045        let mut server = Server::new_async().await;
3046
3047        let mock = server
3048            .mock("POST", "/test")
3049            .with_status(400) // Bad Request - should not be retried
3050            .with_body("Bad Request")
3051            .expect(1)
3052            .create_async()
3053            .await;
3054
3055        let http_client =
3056            HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
3057                .unwrap()
3058                .with_test_backoff_policy();
3059        let request_body = serde_json::json!({"test": "data"});
3060        let uri = format!("{}/test", server.url());
3061
3062        let result = http_client
3063            .make_post_request(&request_body, &uri)
3064            .await;
3065
3066        mock.assert();
3067        assert!(result.is_ok()); // 400 doesn't trigger retry logic, just returns the response
3068
3069        let response = result.unwrap();
3070        assert_eq!(response.status(), 400);
3071    }
3072
3073    #[tokio::test]
3074    async fn test_concurrent_requests_with_different_retry_after() {
3075        let mut server = Server::new_async().await;
3076
3077        // First request gets rate limited with 1 second retry-after
3078        let rate_limit_mock_1 = server
3079            .mock("POST", "/test1")
3080            .with_status(429)
3081            .with_header("Retry-After", "1")
3082            .expect(1)
3083            .create_async()
3084            .await;
3085
3086        // Second request gets rate limited with 2 second retry-after
3087        let rate_limit_mock_2 = server
3088            .mock("POST", "/test2")
3089            .with_status(429)
3090            .with_header("Retry-After", "2")
3091            .expect(1)
3092            .create_async()
3093            .await;
3094
3095        // Success mocks for retries
3096        let success_mock_1 = server
3097            .mock("POST", "/test1")
3098            .with_status(200)
3099            .with_body(r#"{"result": "success1"}"#)
3100            .expect(1)
3101            .create_async()
3102            .await;
3103
3104        let success_mock_2 = server
3105            .mock("POST", "/test2")
3106            .with_status(200)
3107            .with_body(r#"{"result": "success2"}"#)
3108            .expect(1)
3109            .create_async()
3110            .await;
3111
3112        let http_client =
3113            HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
3114                .unwrap()
3115                .with_test_backoff_policy();
3116        let request_body = serde_json::json!({"test": "data"});
3117
3118        let uri1 = format!("{}/test1", server.url());
3119        let uri2 = format!("{}/test2", server.url());
3120
3121        // Start both requests concurrently
3122        let start = std::time::Instant::now();
3123        let (result1, result2) = tokio::join!(
3124            http_client.make_post_request(&request_body, &uri1),
3125            http_client.make_post_request(&request_body, &uri2)
3126        );
3127        let elapsed = start.elapsed();
3128
3129        rate_limit_mock_1.assert();
3130        rate_limit_mock_2.assert();
3131        success_mock_1.assert();
3132        success_mock_2.assert();
3133
3134        assert!(result1.is_ok());
3135        assert!(result2.is_ok());
3136
3137        // Both requests should succeed, but the second should take longer due to the 2s retry-after
3138        // The total time should be at least 2 seconds since the shared retry_after state
3139        // gets updated by both requests
3140        assert!(elapsed >= Duration::from_millis(1800)); // Allow some tolerance
3141        assert!(elapsed <= Duration::from_millis(3000)); // Upper bound
3142
3143        // Check the final retry_after state - should be the latest (higher) value
3144        let final_retry_after = http_client.retry_after.read().await;
3145        assert!(final_retry_after.is_some());
3146
3147        // The retry_after should be set to the latest (higher) value from the two requests
3148        if let Some(retry_time) = *final_retry_after {
3149            // The retry_after time might be in the past now since we waited,
3150            // but it should be reasonable (not too far in past/future)
3151            let now = SystemTime::now();
3152            let diff = if retry_time > now {
3153                retry_time.duration_since(now).unwrap()
3154            } else {
3155                now.duration_since(retry_time).unwrap()
3156            };
3157
3158            // Should be within a reasonable range (the 2s retry-after plus some buffer)
3159            assert!(diff <= Duration::from_secs(3), "Retry time difference too large: {:?}", diff);
3160        }
3161    }
3162
3163    #[tokio::test]
3164    async fn test_get_snapshots() {
3165        let mut server = Server::new_async().await;
3166
3167        // Mock protocol states response
3168        let protocol_states_resp = r#"
3169        {
3170            "states": [
3171                {
3172                    "component_id": "component1",
3173                    "attributes": {
3174                        "attribute_1": "0x00000000000003e8"
3175                    },
3176                    "balances": {
3177                        "0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2": "0x01f4"
3178                    }
3179                }
3180            ],
3181            "pagination": {
3182                "page": 0,
3183                "page_size": 100,
3184                "total": 1
3185            }
3186        }
3187        "#;
3188
3189        // Mock contract state response
3190        let contract_state_resp = r#"
3191        {
3192            "accounts": [
3193                {
3194                    "chain": "ethereum",
3195                    "address": "0x1111111111111111111111111111111111111111",
3196                    "title": "",
3197                    "slots": {},
3198                    "native_balance": "0x01f4",
3199                    "token_balances": {},
3200                    "code": "0x00",
3201                    "code_hash": "0x5c06b7c5b3d910fd33bc2229846f9ddaf91d584d9b196e16636901ac3a77077e",
3202                    "balance_modify_tx": "0x0000000000000000000000000000000000000000000000000000000000000000",
3203                    "code_modify_tx": "0x0000000000000000000000000000000000000000000000000000000000000000",
3204                    "creation_tx": null
3205                }
3206            ],
3207            "pagination": {
3208                "page": 0,
3209                "page_size": 100,
3210                "total": 1
3211            }
3212        }
3213        "#;
3214
3215        // Mock component TVL response
3216        let tvl_resp = r#"
3217        {
3218            "tvl": {
3219                "component1": 1000000.0
3220            },
3221            "pagination": {
3222                "page": 0,
3223                "page_size": 100,
3224                "total": 1
3225            }
3226        }
3227        "#;
3228
3229        let protocol_states_mock = server
3230            .mock("POST", "/v1/protocol_state")
3231            .expect(1)
3232            .with_body(protocol_states_resp)
3233            .create_async()
3234            .await;
3235
3236        let contract_state_mock = server
3237            .mock("POST", "/v1/contract_state")
3238            .expect(1)
3239            .with_body(contract_state_resp)
3240            .create_async()
3241            .await;
3242
3243        let tvl_mock = server
3244            .mock("POST", "/v1/component_tvl")
3245            .expect(1)
3246            .with_body(tvl_resp)
3247            .create_async()
3248            .await;
3249
3250        let client = HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
3251            .expect("create client");
3252
3253        let component = tycho_common::models::protocol::ProtocolComponent {
3254            id: "component1".to_string(),
3255            protocol_system: "test_protocol".to_string(),
3256            protocol_type_name: "test_type".to_string(),
3257            chain: Chain::Ethereum,
3258            tokens: vec![],
3259            contract_addresses: vec![
3260                Bytes::from_str("0x1111111111111111111111111111111111111111").unwrap()
3261            ],
3262            static_attributes: HashMap::new(),
3263            change: tycho_common::models::ChangeType::Creation,
3264            creation_tx: Bytes::from_str(
3265                "0x0000000000000000000000000000000000000000000000000000000000000000",
3266            )
3267            .unwrap(),
3268            created_at: chrono::Utc::now().naive_utc(),
3269        };
3270
3271        let mut components = HashMap::new();
3272        components.insert("component1".to_string(), component);
3273
3274        let contract_ids =
3275            vec![Bytes::from_str("0x1111111111111111111111111111111111111111").unwrap()];
3276
3277        let request = SnapshotParameters::new(
3278            Chain::Ethereum,
3279            "test_protocol",
3280            &components,
3281            &contract_ids,
3282            12345,
3283        );
3284
3285        let response = client
3286            .get_snapshots(&request, None, RPC_CLIENT_CONCURRENCY)
3287            .await
3288            .expect("get snapshots");
3289
3290        // Verify all mocks were called
3291        protocol_states_mock.assert();
3292        contract_state_mock.assert();
3293        tvl_mock.assert();
3294
3295        // Assert states
3296        assert_eq!(response.states.len(), 1);
3297        assert!(response
3298            .states
3299            .contains_key("component1"));
3300
3301        // Check that the state has the expected TVL
3302        let component_state = response
3303            .states
3304            .get("component1")
3305            .unwrap();
3306        assert_eq!(component_state.component_tvl, Some(1000000.0));
3307
3308        // Assert VM storage
3309        assert_eq!(response.vm_storage.len(), 1);
3310        let contract_addr = Bytes::from_str("0x1111111111111111111111111111111111111111").unwrap();
3311        assert!(response
3312            .vm_storage
3313            .contains_key(&contract_addr));
3314    }
3315
3316    #[tokio::test]
3317    async fn test_get_snapshots_empty_components() {
3318        let server = Server::new_async().await;
3319        let client = HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
3320            .expect("create client");
3321
3322        let components = HashMap::new();
3323        let contract_ids = vec![];
3324
3325        let request = SnapshotParameters::new(
3326            Chain::Ethereum,
3327            "test_protocol",
3328            &components,
3329            &contract_ids,
3330            12345,
3331        );
3332
3333        let response = client
3334            .get_snapshots(&request, None, RPC_CLIENT_CONCURRENCY)
3335            .await
3336            .expect("get snapshots");
3337
3338        // Should return empty response without making any requests
3339        assert!(response.states.is_empty());
3340        assert!(response.vm_storage.is_empty());
3341    }
3342
3343    #[tokio::test]
3344    async fn test_get_snapshots_without_tvl() {
3345        let mut server = Server::new_async().await;
3346
3347        let protocol_states_resp = r#"
3348        {
3349            "states": [
3350                {
3351                    "component_id": "component1",
3352                    "attributes": {},
3353                    "balances": {}
3354                }
3355            ],
3356            "pagination": {
3357                "page": 0,
3358                "page_size": 100,
3359                "total": 1
3360            }
3361        }
3362        "#;
3363
3364        let protocol_states_mock = server
3365            .mock("POST", "/v1/protocol_state")
3366            .expect(1)
3367            .with_body(protocol_states_resp)
3368            .create_async()
3369            .await;
3370
3371        let client = HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
3372            .expect("create client");
3373
3374        // Create test component
3375        let component = tycho_common::models::protocol::ProtocolComponent {
3376            id: "component1".to_string(),
3377            protocol_system: "test_protocol".to_string(),
3378            protocol_type_name: "test_type".to_string(),
3379            chain: Chain::Ethereum,
3380            tokens: vec![],
3381            contract_addresses: vec![],
3382            static_attributes: HashMap::new(),
3383            change: tycho_common::models::ChangeType::Creation,
3384            creation_tx: Bytes::from_str(
3385                "0x0000000000000000000000000000000000000000000000000000000000000000",
3386            )
3387            .unwrap(),
3388            created_at: chrono::Utc::now().naive_utc(),
3389        };
3390
3391        let mut components = HashMap::new();
3392        components.insert("component1".to_string(), component);
3393        let contract_ids = vec![];
3394
3395        let request = SnapshotParameters::new(
3396            Chain::Ethereum,
3397            "test_protocol",
3398            &components,
3399            &contract_ids,
3400            12345,
3401        )
3402        .include_balances(false)
3403        .include_tvl(false);
3404
3405        let response = client
3406            .get_snapshots(&request, None, RPC_CLIENT_CONCURRENCY)
3407            .await
3408            .expect("get snapshots");
3409
3410        // Verify only necessary mocks were called
3411        protocol_states_mock.assert();
3412        // No contract_state_mock.assert() since contract_ids is empty
3413        // No tvl_mock.assert() since include_tvl is false
3414
3415        assert_eq!(response.states.len(), 1);
3416        // Check that TVL is None since we didn't request it
3417        let component_state = response
3418            .states
3419            .get("component1")
3420            .unwrap();
3421        assert_eq!(component_state.component_tvl, None);
3422    }
3423
3424    #[tokio::test]
3425    async fn test_compression_enabled() {
3426        let mut server = Server::new_async().await;
3427        let server_resp = GET_CONTRACT_STATE_RESP;
3428
3429        // Compress the response using zstd
3430        let compressed_body =
3431            zstd::encode_all(server_resp.as_bytes(), 0).expect("compression failed");
3432
3433        let mocked_server = server
3434            .mock("POST", "/v1/contract_state")
3435            .expect(1)
3436            .with_header("Content-Encoding", "zstd")
3437            .with_body(compressed_body)
3438            .create_async()
3439            .await;
3440
3441        // Create client with compression enabled
3442        let client = HttpRPCClient::new(
3443            server.url().as_str(),
3444            HttpRPCClientOptions::new().with_compression(true),
3445        )
3446        .expect("create client");
3447
3448        let response = client
3449            .get_contract_state(ContractStateParams::new(Chain::Ethereum, ""))
3450            .await
3451            .expect("get state");
3452        let accounts = response;
3453
3454        mocked_server.assert();
3455        assert_eq!(accounts.data().len(), 1);
3456        assert_eq!(accounts.data()[0].native_balance, Bytes::from(500u16.to_be_bytes()));
3457    }
3458
3459    #[tokio::test]
3460    async fn test_compression_disabled() {
3461        let mut server = Server::new_async().await;
3462        let server_resp = GET_CONTRACT_STATE_RESP;
3463
3464        // Verify client does NOT send Accept-Encoding: zstd when compression is disabled
3465        // Instead, server should receive request without compression headers
3466        let mocked_server = server
3467            .mock("POST", "/v1/contract_state")
3468            .expect(1)
3469            .match_header("Accept-Encoding", mockito::Matcher::Missing)
3470            .with_status(200)
3471            .with_body(server_resp)
3472            .create_async()
3473            .await;
3474
3475        // Create client with compression disabled
3476        let client = HttpRPCClient::new(
3477            server.url().as_str(),
3478            HttpRPCClientOptions::new().with_compression(false),
3479        )
3480        .expect("create client");
3481
3482        let response = client
3483            .get_contract_state(ContractStateParams::new(Chain::Ethereum, ""))
3484            .await
3485            .expect("get state");
3486        let accounts = response;
3487
3488        // Verify the mock was called (client sent request without Accept-Encoding header)
3489        mocked_server.assert();
3490        assert_eq!(accounts.data().len(), 1);
3491        assert_eq!(accounts.data()[0].native_balance, Bytes::from(500u16.to_be_bytes()));
3492    }
3493
3494    #[rstest]
3495    #[case::single_page(2, 1000)]
3496    #[case::multiple_pages_within_concurrency(10, 2)]
3497    #[case::exceeds_concurrency_limit(60, 2)]
3498    #[tokio::test]
3499    async fn test_get_all_tokens_pagination_and_concurrency(
3500        #[case] total_tokens: usize,
3501        #[case] page_size: usize,
3502    ) {
3503        use std::sync::atomic::{AtomicUsize, Ordering};
3504
3505        let allowed_concurrency = 10;
3506
3507        let concurrent_requests = Arc::new(AtomicUsize::new(0));
3508        let max_concurrent = Arc::new(AtomicUsize::new(0));
3509
3510        let mut server = Server::new_async().await;
3511
3512        let total_pages = (total_tokens as f64 / page_size as f64).ceil() as i64;
3513
3514        // Mock all required pages
3515        for page in 0..total_pages {
3516            let concurrent = concurrent_requests.clone();
3517            let max_conc = max_concurrent.clone();
3518
3519            let tokens_in_page = {
3520                let start_idx = (page as usize) * page_size;
3521                let end_idx = ((page as usize + 1) * page_size).min(total_tokens);
3522                (start_idx..end_idx)
3523                    .map(|i| {
3524                        format!(
3525                            r#"{{
3526                            "chain": "ethereum",
3527                            "address": "0x{i:040x}",
3528                            "symbol": "TOKEN_{i}",
3529                            "decimals": 18,
3530                            "tax": 0,
3531                            "gas": [30000],
3532                            "quality": 100
3533                        }}"#
3534                        )
3535                    })
3536                    .collect::<Vec<_>>()
3537            };
3538
3539            let tokens_json = tokens_in_page.join(",");
3540            let response = format!(
3541                r#"{{
3542                    "tokens": [{tokens_json}],
3543                    "pagination": {{
3544                        "page": {page},
3545                        "page_size": {page_size},
3546                        "total": {total_tokens}
3547                    }}
3548                }}"#,
3549            );
3550
3551            server
3552                .mock("POST", "/v1/tokens")
3553                .expect(1)
3554                .with_chunked_body(move |w| {
3555                    // Track concurrent requests
3556                    let current = concurrent.fetch_add(1, Ordering::SeqCst);
3557                    max_conc.fetch_max(current + 1, Ordering::SeqCst);
3558
3559                    // Simulate some work to increase likelihood of concurrent requests
3560                    std::thread::sleep(Duration::from_millis(10));
3561
3562                    concurrent.fetch_sub(1, Ordering::SeqCst);
3563
3564                    w.write_all(response.as_bytes())
3565                })
3566                .create_async()
3567                .await;
3568        }
3569
3570        let client = HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
3571            .expect("create client");
3572
3573        let tokens = client
3574            .get_all_tokens(
3575                AllTokensParams::new(Chain::Ethereum, allowed_concurrency)
3576                    .with_chunk_size(page_size),
3577            )
3578            .await
3579            .expect("get all tokens");
3580
3581        // Verify concurrency was respected
3582        let max = max_concurrent.load(Ordering::SeqCst);
3583        let expected_max_concurrency = (total_pages as usize)
3584            .saturating_sub(1)
3585            .min(allowed_concurrency);
3586        assert!(
3587            max <= allowed_concurrency,
3588            "Expected max concurrent requests <= {allowed_concurrency}, got {max}"
3589        );
3590
3591        // For cases with multiple pages, verify we actually used concurrency
3592        if total_pages > 1 && expected_max_concurrency > 1 {
3593            assert!(
3594                max > 0,
3595                "Expected some concurrent requests for multi-page response, got {max}"
3596            );
3597        }
3598
3599        // Verify we got all expected tokens
3600        assert_eq!(
3601            tokens.len(),
3602            total_tokens,
3603            "Expected {total_tokens} tokens, got {}",
3604            tokens.len()
3605        );
3606
3607        // Verify tokens are in the expected order
3608        for (i, token) in tokens.iter().enumerate() {
3609            assert_eq!(token.symbol, format!("TOKEN_{i}"), "Token at index {i} has wrong symbol");
3610        }
3611    }
3612}
tycho_client/rpc.rs

tycho_client/
rpc.rs
