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 max_page_size = StateRequestBody::effective_max_page_size(self.compression()) as usize;
803        let chunk_size = params
804            .chunk_size
805            .unwrap_or(CONTRACT_STATE_CHUNK_SIZE)
806            .min(max_page_size);
807
808        let mut tasks = Vec::new();
809        for chunk in sorted_ids.chunks(chunk_size) {
810            let sem = semaphore.clone();
811            let base_params =
812                ContractStateParams::new(params.chain, params.protocol_system.as_str())
813                    .with_contract_ids(chunk.to_vec())
814                    .with_version(params.version.clone())
815                    .with_pagination(0, chunk_size as i64);
816            tasks.push(async move {
817                let _permit = sem
818                    .acquire()
819                    .await
820                    .map_err(|_| RPCError::Fatal("Semaphore dropped".to_string()))?;
821                self.get_contract_state(base_params)
822                    .await
823            });
824        }
825
826        let pages = try_join_all(tasks).await?;
827
828        let accounts = pages
829            .into_iter()
830            .flat_map(|p| p.into_iter())
831            .collect();
832
833        Ok(accounts)
834    }
835
836    /// Retrieves protocol components matching the given filters.
837    ///
838    /// Pass `component_ids` to filter by specific IDs, or `tvl_gt` to filter by minimum TVL.
839    async fn get_protocol_components(
840        &self,
841        params: ProtocolComponentsParams,
842    ) -> Result<Page<Vec<ProtocolComponent>>, RPCError>;
843
844    /// Retrieves protocol components, fetching all pages automatically.
845    ///
846    /// If `chunk_size` is `None`, it defaults to the maximum page size.
847    async fn get_protocol_components_paginated(
848        &self,
849        params: ProtocolComponentsPaginatedParams,
850    ) -> Result<Vec<ProtocolComponent>, RPCError> {
851        let chain = params.chain;
852        let protocol_system = params.protocol_system;
853        let component_ids = params.component_ids;
854        let tvl_gt = params.tvl_gt;
855        let chunk_size = params.chunk_size;
856        let concurrency = params.concurrency;
857
858        let semaphore = Arc::new(Semaphore::new(concurrency));
859
860        let max_page_size =
861            ProtocolComponentsRequestBody::effective_max_page_size(self.compression()) as usize;
862        let chunk_size = chunk_size
863            .unwrap_or(max_page_size)
864            .min(max_page_size);
865
866        // If a set of component IDs is specified, the maximum return size is already known,
867        // allowing us to pre-compute the number of requests to be made.
868        match component_ids {
869            Some(ids) => {
870                // The server pages within the requested ids, so every chunk is page 0.
871                let tasks: Vec<_> =
872                    ids.chunks(chunk_size)
873                        .map(|chunk| {
874                            let sem = semaphore.clone();
875                            let mut base =
876                                ProtocolComponentsParams::new(chain, protocol_system.as_str())
877                                    .with_component_ids(chunk.to_vec())
878                                    .with_pagination(0, chunk_size as i64);
879                            if let Some(tvl) = tvl_gt {
880                                base = base.with_tvl_gt(tvl);
881                            }
882                            async move {
883                                let _permit = sem.acquire().await.map_err(|_| {
884                                    RPCError::Fatal("Semaphore dropped".to_string())
885                                })?;
886                                self.get_protocol_components(base).await
887                            }
888                        })
889                        .collect();
890
891                try_join_all(tasks)
892                    .await
893                    .map(|pages| pages.into_iter().flatten().collect())
894            }
895            None => {
896                // If no component ids are specified, we need to make requests based on the total
897                // number of results from the first response.
898                let mut base_params =
899                    ProtocolComponentsParams::new(chain, protocol_system.as_str())
900                        .with_pagination(0, chunk_size as i64);
901                if let Some(tvl) = tvl_gt {
902                    base_params = base_params.with_tvl_gt(tvl);
903                }
904
905                let first_page = self
906                    .get_protocol_components(base_params)
907                    .await?;
908
909                let total_items = first_page.total();
910                let total_pages = (total_items as f64 / chunk_size as f64).ceil() as i64;
911
912                let mut all: Vec<ProtocolComponent> = first_page.into_data();
913
914                let mut page = 1;
915                while page < total_pages {
916                    let requests_in_this_iteration = (total_pages - page).min(concurrency as i64);
917
918                    let tasks: Vec<_> = (0..requests_in_this_iteration)
919                        .map(|iter| {
920                            let sem = semaphore.clone();
921                            let mut p =
922                                ProtocolComponentsParams::new(chain, protocol_system.as_str())
923                                    .with_pagination(page + iter, chunk_size as i64);
924                            if let Some(tvl) = tvl_gt {
925                                p = p.with_tvl_gt(tvl);
926                            }
927                            async move {
928                                let _permit = sem.acquire().await.map_err(|_| {
929                                    RPCError::Fatal("Semaphore dropped".to_string())
930                                })?;
931                                self.get_protocol_components(p).await
932                            }
933                        })
934                        .collect();
935
936                    let responses = try_join_all(tasks).await?;
937
938                    for resp in responses {
939                        all.extend(resp);
940                    }
941
942                    page += requests_in_this_iteration;
943                }
944                Ok(all)
945            }
946        }
947    }
948
949    /// Retrieves a page of protocol component states.
950    ///
951    /// `block_number` pins the query to a specific block; pass `None` to use the latest state.
952    async fn get_protocol_states(
953        &self,
954        params: ProtocolStatesParams,
955    ) -> Result<Page<Vec<ProtocolComponentState>>, RPCError>;
956
957    /// Retrieves protocol states for a set of protocol IDs, fetching all pages automatically.
958    ///
959    /// If `chunk_size` is `None`, it defaults to the maximum page size.
960    async fn get_protocol_states_paginated(
961        &self,
962        params: ProtocolStatesPaginatedParams,
963    ) -> Result<Vec<ProtocolComponentState>, RPCError> {
964        let semaphore = Arc::new(Semaphore::new(params.concurrency));
965
966        let max_page_size =
967            ProtocolStateRequestBody::effective_max_page_size(self.compression()) as usize;
968        let chunk_size = params
969            .chunk_size
970            .unwrap_or(max_page_size)
971            .min(max_page_size);
972
973        let tasks: Vec<_> = params
974            .protocol_ids
975            .chunks(chunk_size)
976            .map(|c| {
977                let sem = semaphore.clone();
978                let p = ProtocolStatesParams::new(params.chain, params.protocol_system.as_str())
979                    .with_protocol_ids(c.to_vec())
980                    .with_include_balances(params.include_balances)
981                    .with_version(params.version.clone())
982                    .with_pagination(0, chunk_size as i64);
983                async move {
984                    let _permit = sem
985                        .acquire()
986                        .await
987                        .map_err(|_| RPCError::Fatal("Semaphore dropped".to_string()))?;
988                    self.get_protocol_states(p).await
989                }
990            })
991            .collect();
992
993        try_join_all(tasks)
994            .await
995            .map(|pages| pages.into_iter().flatten().collect())
996    }
997
998    /// Retrieves a page of tokens.
999    ///
1000    /// Use `get_all_tokens` to fetch all matching tokens automatically.
1001    async fn get_tokens(&self, params: TokensParams) -> Result<Page<Vec<Token>>, RPCError>;
1002
1003    /// Retrieves all tokens matching the given criteria, fetching all pages automatically.
1004    ///
1005    /// If `chunk_size` is `None`, it defaults to the maximum page size.
1006    async fn get_all_tokens(&self, params: AllTokensParams) -> Result<Vec<Token>, RPCError> {
1007        let max_page_size = TokensRequestBody::effective_max_page_size(self.compression()) as usize;
1008        let chunk_size = params
1009            .chunk_size
1010            .unwrap_or(max_page_size)
1011            .min(max_page_size);
1012
1013        let semaphore = Arc::new(Semaphore::new(params.concurrency));
1014
1015        let page_size: i64 = chunk_size.try_into().map_err(|_| {
1016            RPCError::FormatRequest("Failed to convert chunk_size into i64".to_string())
1017        })?;
1018
1019        let mut base_params = TokensParams::new(params.chain).with_pagination(0, page_size);
1020        if let Some(q) = params.min_quality {
1021            base_params = base_params.with_min_quality(q);
1022        }
1023        if let Some(d) = params.traded_n_days_ago {
1024            base_params = base_params.with_traded_n_days_ago(d);
1025        }
1026
1027        let first_page = self.get_tokens(base_params).await?;
1028        let total_pages = (first_page.total() as f64 / chunk_size as f64).ceil() as i64;
1029
1030        let mut all_tokens: Vec<Token> = first_page.into_data();
1031
1032        if total_pages <= 1 {
1033            return Ok(all_tokens);
1034        }
1035
1036        let tasks: Vec<_> = (1..total_pages)
1037            .map(|page| {
1038                let sem = semaphore.clone();
1039                let mut p = TokensParams::new(params.chain).with_pagination(page, page_size);
1040                if let Some(q) = params.min_quality {
1041                    p = p.with_min_quality(q);
1042                }
1043                if let Some(d) = params.traded_n_days_ago {
1044                    p = p.with_traded_n_days_ago(d);
1045                }
1046                async move {
1047                    let _permit = sem
1048                        .acquire()
1049                        .await
1050                        .map_err(|_| RPCError::Fatal("Semaphore dropped".to_string()))?;
1051                    self.get_tokens(p).await
1052                }
1053            })
1054            .collect();
1055
1056        let pages = try_join_all(tasks).await?;
1057        for page in pages {
1058            all_tokens.extend(page);
1059        }
1060
1061        Ok(all_tokens)
1062    }
1063
1064    /// Retrieves the protocol systems known to the server.
1065    async fn get_protocol_systems(
1066        &self,
1067        params: ProtocolSystemsParams,
1068    ) -> Result<Page<ProtocolSystems>, RPCError>;
1069
1070    /// Retrieves component TVL values.
1071    ///
1072    /// Filter by `component_ids` or by `protocol_system`; both are optional.
1073    async fn get_component_tvl(
1074        &self,
1075        params: ComponentTvlParams,
1076    ) -> Result<Page<HashMap<String, f64>>, RPCError>;
1077
1078    /// Retrieves component TVL values, fetching all pages automatically.
1079    ///
1080    /// If `chunk_size` is `None`, it defaults to the maximum page size.
1081    async fn get_component_tvl_paginated(
1082        &self,
1083        params: ComponentTvlPaginatedParams,
1084    ) -> Result<HashMap<String, f64>, RPCError> {
1085        let semaphore = Arc::new(Semaphore::new(params.concurrency));
1086
1087        let max_page_size =
1088            ComponentTvlRequestBody::effective_max_page_size(self.compression()) as usize;
1089        let chunk_size = params
1090            .chunk_size
1091            .unwrap_or(max_page_size)
1092            .min(max_page_size);
1093
1094        match params.component_ids {
1095            Some(ids) => {
1096                // The server pages within the requested ids, so every chunk is page 0.
1097                let tasks: Vec<_> =
1098                    ids.chunks(chunk_size)
1099                        .map(|chunk| {
1100                            let sem = semaphore.clone();
1101                            let mut p = ComponentTvlParams::new(params.chain)
1102                                .with_component_ids(chunk.to_vec())
1103                                .with_pagination(0, chunk_size as i64);
1104                            if let Some(ref ps) = params.protocol_system {
1105                                p = p.with_protocol_system(ps.as_str());
1106                            }
1107                            async move {
1108                                let _permit = sem.acquire().await.map_err(|_| {
1109                                    RPCError::Fatal("Semaphore dropped".to_string())
1110                                })?;
1111                                self.get_component_tvl(p).await
1112                            }
1113                        })
1114                        .collect();
1115
1116                let pages = try_join_all(tasks).await?;
1117
1118                let mut merged_tvl = HashMap::new();
1119                for page in pages {
1120                    for (key, value) in page {
1121                        *merged_tvl.entry(key).or_insert(0.0) = value;
1122                    }
1123                }
1124
1125                Ok(merged_tvl)
1126            }
1127            None => {
1128                let mut base =
1129                    ComponentTvlParams::new(params.chain).with_pagination(0, chunk_size as i64);
1130                if let Some(ref ps) = params.protocol_system {
1131                    base = base.with_protocol_system(ps.as_str());
1132                }
1133
1134                let first_page = self.get_component_tvl(base).await?;
1135                let total_items = first_page.total();
1136                let total_pages = (total_items as f64 / chunk_size as f64).ceil() as i64;
1137
1138                let mut merged_tvl: HashMap<String, f64> = first_page.into_data();
1139
1140                let mut page = 1;
1141                while page < total_pages {
1142                    let requests_in_this_iteration =
1143                        (total_pages - page).min(params.concurrency as i64);
1144
1145                    let tasks: Vec<_> = (0..requests_in_this_iteration)
1146                        .map(|i| {
1147                            let sem = semaphore.clone();
1148                            let mut p = ComponentTvlParams::new(params.chain)
1149                                .with_pagination(page + i, chunk_size as i64);
1150                            if let Some(ref ps) = params.protocol_system {
1151                                p = p.with_protocol_system(ps.as_str());
1152                            }
1153                            async move {
1154                                let _permit = sem.acquire().await.map_err(|_| {
1155                                    RPCError::Fatal("Semaphore dropped".to_string())
1156                                })?;
1157                                self.get_component_tvl(p).await
1158                            }
1159                        })
1160                        .collect();
1161
1162                    let responses = try_join_all(tasks).await?;
1163
1164                    for resp in responses {
1165                        for (key, value) in resp {
1166                            *merged_tvl.entry(key).or_insert(0.0) = value;
1167                        }
1168                    }
1169
1170                    page += requests_in_this_iteration;
1171                }
1172
1173                Ok(merged_tvl)
1174            }
1175        }
1176    }
1177
1178    /// Retrieves a page of traced entry points.
1179    ///
1180    /// Use `get_traced_entry_points_paginated` to fetch all pages automatically.
1181    async fn get_traced_entry_points(
1182        &self,
1183        params: TracedEntryPointsParams,
1184    ) -> Result<Page<TracedEntryPoints>, RPCError>;
1185
1186    /// Retrieves traced entry points for a set of component IDs, fetching all pages automatically.
1187    ///
1188    /// If `chunk_size` is `None`, it defaults to the maximum page size.
1189    async fn get_traced_entry_points_paginated(
1190        &self,
1191        params: TracedEntryPointsPaginatedParams,
1192    ) -> Result<TracedEntryPoints, RPCError> {
1193        let chain = params.chain;
1194        let protocol_system = params.protocol_system;
1195        let component_ids = params.component_ids;
1196        let chunk_size = params.chunk_size;
1197        let concurrency = params.concurrency;
1198
1199        let semaphore = Arc::new(Semaphore::new(concurrency));
1200
1201        let max_page_size =
1202            TracedEntryPointRequestBody::effective_max_page_size(self.compression()) as usize;
1203        let chunk_size = chunk_size
1204            .unwrap_or(max_page_size)
1205            .min(max_page_size);
1206
1207        let tasks: Vec<_> = component_ids
1208            .chunks(chunk_size)
1209            .map(|c| {
1210                let sem = semaphore.clone();
1211                let params = TracedEntryPointsParams::new(chain, protocol_system.as_str())
1212                    .with_component_ids(c.to_vec())
1213                    .with_pagination(0, chunk_size as i64);
1214                async move {
1215                    let _permit = sem
1216                        .acquire()
1217                        .await
1218                        .map_err(|_| RPCError::Fatal("Semaphore dropped".to_string()))?;
1219                    self.get_traced_entry_points(params)
1220                        .await
1221                }
1222            })
1223            .collect();
1224
1225        try_join_all(tasks)
1226            .await
1227            .map(|pages| pages.into_iter().flatten().collect())
1228    }
1229
1230    // clippy false positive: `'a` is required by the trait method signature and is
1231    // used in `SnapshotParameters<'a>`, but `async_trait` makes Clippy miss it.
1232    #[allow(clippy::extra_unused_lifetimes)]
1233    async fn get_snapshots<'a>(
1234        &self,
1235        request: &SnapshotParameters<'a>,
1236        chunk_size: Option<usize>,
1237        concurrency: usize,
1238    ) -> Result<Snapshot, RPCError>;
1239}
1240
1241/// Configuration options for HttpRPCClient
1242#[derive(Debug, Clone)]
1243pub struct HttpRPCClientOptions {
1244    /// Optional API key for authentication
1245    pub auth_key: Option<String>,
1246    /// Enable compression for requests (default: true)
1247    /// When enabled, adds Accept-Encoding: zstd header
1248    pub compression: bool,
1249    /// Pre-serialized `X-Tycho-Client-Metadata` header value. `None` sends no header.
1250    pub client_metadata_header: Option<String>,
1251}
1252
1253impl Default for HttpRPCClientOptions {
1254    fn default() -> Self {
1255        Self::new()
1256    }
1257}
1258
1259impl HttpRPCClientOptions {
1260    /// Create new options with default values (compression enabled)
1261    pub fn new() -> Self {
1262        Self { auth_key: None, compression: true, client_metadata_header: None }
1263    }
1264
1265    /// Set the authentication key
1266    pub fn with_auth_key(mut self, auth_key: Option<String>) -> Self {
1267        self.auth_key = auth_key;
1268        self
1269    }
1270
1271    /// Set whether to enable compression (default: true)
1272    pub fn with_compression(mut self, compression: bool) -> Self {
1273        self.compression = compression;
1274        self
1275    }
1276
1277    /// Set the pre-serialized client-metadata header value. `None` sends no header.
1278    pub fn with_client_metadata_header(mut self, header: Option<String>) -> Self {
1279        self.client_metadata_header = header;
1280        self
1281    }
1282}
1283
1284#[derive(Debug, Clone)]
1285pub struct HttpRPCClient {
1286    http_client: Client,
1287    url: Url,
1288    retry_after: Arc<RwLock<Option<SystemTime>>>,
1289    backoff_policy: ExponentialBackoff,
1290    server_restart_duration: Duration,
1291    compression: bool,
1292}
1293
1294impl HttpRPCClient {
1295    pub fn new(base_uri: &str, options: HttpRPCClientOptions) -> Result<Self, RPCError> {
1296        let uri = base_uri
1297            .parse::<Url>()
1298            .map_err(|e| RPCError::UrlParsing(base_uri.to_string(), e.to_string()))?;
1299
1300        // Add default headers
1301        let mut headers = header::HeaderMap::new();
1302        headers.insert(header::CONTENT_TYPE, header::HeaderValue::from_static("application/json"));
1303        let user_agent = format!("tycho-client-{version}", version = env!("CARGO_PKG_VERSION"));
1304        headers.insert(
1305            header::USER_AGENT,
1306            header::HeaderValue::from_str(&user_agent)
1307                .map_err(|e| RPCError::FormatRequest(format!("Invalid user agent format: {e}")))?,
1308        );
1309
1310        // Add Authorization if one is given
1311        if let Some(key) = options.auth_key.as_deref() {
1312            let mut auth_value = header::HeaderValue::from_str(key).map_err(|e| {
1313                RPCError::FormatRequest(format!("Invalid authorization key format: {e}"))
1314            })?;
1315            auth_value.set_sensitive(true);
1316            headers.insert(header::AUTHORIZATION, auth_value);
1317        }
1318
1319        // Add generic client metadata if one is given. Pre-validated by the serializer, but mirror
1320        // the auth pattern so any residual formatting error surfaces as a FormatRequest error.
1321        if let Some(metadata) = options
1322            .client_metadata_header
1323            .as_deref()
1324        {
1325            let value = header::HeaderValue::from_str(metadata).map_err(|e| {
1326                RPCError::FormatRequest(format!("Invalid client metadata format: {e}"))
1327            })?;
1328            headers.insert(header::HeaderName::from_static(CLIENT_METADATA_HEADER), value);
1329        }
1330
1331        let mut client_builder = ClientBuilder::new()
1332            .default_headers(headers)
1333            .http2_prior_knowledge();
1334
1335        // When compression is disabled, turn off all automatic compression
1336        if !options.compression {
1337            client_builder = client_builder.no_zstd();
1338        }
1339
1340        let client = client_builder
1341            .build()
1342            .map_err(|e| RPCError::HttpClient(e.to_string(), e))?;
1343
1344        Ok(Self {
1345            http_client: client,
1346            url: uri,
1347            retry_after: Arc::new(RwLock::new(None)),
1348            backoff_policy: ExponentialBackoffBuilder::new()
1349                .with_initial_interval(Duration::from_millis(250))
1350                // increase backoff time by 75% each failure
1351                .with_multiplier(1.75)
1352                // keep retrying every 30s
1353                .with_max_interval(Duration::from_secs(30))
1354                // if all retries take longer than 2m, give up
1355                .with_max_elapsed_time(Some(Duration::from_secs(125)))
1356                .build(),
1357            server_restart_duration: Duration::from_secs(120),
1358            compression: options.compression,
1359        })
1360    }
1361
1362    #[cfg(test)]
1363    pub fn with_test_backoff_policy(mut self) -> Self {
1364        // Extremely short intervals for very fast testing
1365        self.backoff_policy = ExponentialBackoffBuilder::new()
1366            .with_initial_interval(Duration::from_millis(1))
1367            .with_multiplier(1.1)
1368            .with_max_interval(Duration::from_millis(5))
1369            .with_max_elapsed_time(Some(Duration::from_millis(50)))
1370            .build();
1371        self.server_restart_duration = Duration::from_millis(50);
1372        self
1373    }
1374
1375    /// Converts a error response to a Result.
1376    ///
1377    /// Raises an error if the response status code id 429, 502, 503 or 504. In the 429
1378    /// case it will try to look for a retry-after header an parse it accordingly. The
1379    /// parsed value is then passed as part of the error.
1380    async fn error_for_response(
1381        &self,
1382        response: reqwest::Response,
1383    ) -> Result<reqwest::Response, RPCError> {
1384        match response.status() {
1385            StatusCode::TOO_MANY_REQUESTS => {
1386                let retry_after_raw = response
1387                    .headers()
1388                    .get(reqwest::header::RETRY_AFTER)
1389                    .and_then(|h| h.to_str().ok())
1390                    .and_then(parse_retry_value);
1391
1392                let reason = response
1393                    .text()
1394                    .await
1395                    .unwrap_or_default();
1396                warn!(reason, retry_after = ?retry_after_raw, "Rate limited by server");
1397
1398                Err(RPCError::RateLimited(retry_after_raw))
1399            }
1400            StatusCode::BAD_GATEWAY |
1401            StatusCode::SERVICE_UNAVAILABLE |
1402            StatusCode::GATEWAY_TIMEOUT => Err(RPCError::ServerUnreachable(
1403                response
1404                    .text()
1405                    .await
1406                    .unwrap_or_else(|_| "Server Unreachable".to_string()),
1407            )),
1408            _ => Ok(response),
1409        }
1410    }
1411
1412    /// Classifies errors into transient or permanent ones.
1413    ///
1414    /// Transient errors are retried with a potential backoff, permanent ones are not.
1415    /// If the error is RateLimited, this method will set the self.retry_after value so
1416    /// future requests wait until the rate limit has been reset.
1417    async fn handle_error_for_backoff(&self, e: RPCError) -> backoff::Error<RPCError> {
1418        match e {
1419            RPCError::ServerUnreachable(_) => {
1420                backoff::Error::retry_after(e, self.server_restart_duration)
1421            }
1422            RPCError::RateLimited(Some(until)) => {
1423                let mut retry_after_guard = self.retry_after.write().await;
1424                *retry_after_guard = Some(
1425                    retry_after_guard
1426                        .unwrap_or(until)
1427                        .max(until),
1428                );
1429
1430                if let Ok(duration) = until.duration_since(SystemTime::now()) {
1431                    backoff::Error::retry_after(e, duration)
1432                } else {
1433                    e.into()
1434                }
1435            }
1436            RPCError::RateLimited(None) => e.into(),
1437            _ => backoff::Error::permanent(e),
1438        }
1439    }
1440
1441    /// Waits until the current rate limit time has passed.
1442    ///
1443    /// Only waits if there is a time and that time is in the future, else return
1444    /// immediately.
1445    async fn wait_until_retry_after(&self) {
1446        if let Some(&until) = self.retry_after.read().await.as_ref() {
1447            let now = SystemTime::now();
1448            if until > now {
1449                if let Ok(duration) = until.duration_since(now) {
1450                    sleep(duration).await
1451                }
1452            }
1453        }
1454    }
1455
1456    /// Makes a post request handling transient failures.
1457    ///
1458    /// If a retry-after header is received it will be respected. Else the configured
1459    /// backoff policy is used to deal with transient network or server errors.
1460    async fn make_post_request<T: Serialize + ?Sized>(
1461        &self,
1462        request: &T,
1463        uri: &String,
1464    ) -> Result<Response, RPCError> {
1465        self.wait_until_retry_after().await;
1466        let response = backoff::future::retry(self.backoff_policy.clone(), || async {
1467            let server_response = self
1468                .http_client
1469                .post(uri)
1470                .json(request)
1471                .send()
1472                .await
1473                .map_err(|e| RPCError::HttpClient(e.to_string(), e))?;
1474
1475            match self
1476                .error_for_response(server_response)
1477                .await
1478            {
1479                Ok(response) => Ok(response),
1480                Err(e) => Err(self.handle_error_for_backoff(e).await),
1481            }
1482        })
1483        .await?;
1484        Ok(response)
1485    }
1486}
1487
1488fn parse_retry_value(val: &str) -> Option<SystemTime> {
1489    if let Ok(secs) = val.parse::<u64>() {
1490        return Some(SystemTime::now() + Duration::from_secs(secs));
1491    }
1492    if let Ok(date) = OffsetDateTime::parse(val, &Rfc2822) {
1493        return Some(date.into());
1494    }
1495    None
1496}
1497
1498#[async_trait]
1499impl RPCClient for HttpRPCClient {
1500    fn compression(&self) -> bool {
1501        self.compression
1502    }
1503
1504    #[instrument(skip(self))]
1505    async fn get_contract_state(
1506        &self,
1507        params: ContractStateParams,
1508    ) -> Result<Page<Vec<Account>>, RPCError> {
1509        if params
1510            .contract_ids
1511            .as_ref()
1512            .is_none_or(|ids| ids.is_empty())
1513        {
1514            warn!("No contract ids specified in request.");
1515        }
1516
1517        let request = StateRequestBody {
1518            contract_ids: params.contract_ids,
1519            protocol_system: params.protocol_system,
1520            chain: params.chain.into(),
1521            version: params.version,
1522            pagination: PaginationParams { page: params.page, page_size: params.page_size },
1523        };
1524
1525        let uri = format!(
1526            "{}/{}/contract_state",
1527            self.url
1528                .to_string()
1529                .trim_end_matches('/'),
1530            TYCHO_SERVER_VERSION
1531        );
1532        debug!(%uri, "Sending contract_state request to Tycho server");
1533        trace!(?request, "Sending request to Tycho server");
1534        let response = self
1535            .make_post_request(&request, &uri)
1536            .await?;
1537        trace!(?response, "Received response from Tycho server");
1538
1539        let body = response
1540            .text()
1541            .await
1542            .map_err(|e| RPCError::ParseResponse(e.to_string()))?;
1543        if body.is_empty() {
1544            // Pure native protocols will return empty contract states
1545            return Ok(Page::new(vec![], 0, 0, 0));
1546        }
1547
1548        // Decoding a large response takes seconds of CPU: run it on the blocking pool, so the
1549        // runtime stays responsive and concurrent requests decode in parallel.
1550        let (data, pagination) = tokio::task::spawn_blocking(move || {
1551            let dto_response = serde_json::from_str::<StateRequestResponse>(&body)
1552                .map_err(|err| RPCError::from_parse_error(err, &body))?;
1553            trace!(?dto_response, "Received contract_state response from Tycho server");
1554
1555            let data: Vec<Account> = dto_response
1556                .accounts
1557                .into_iter()
1558                .map(Account::from)
1559                .collect();
1560            Ok::<_, RPCError>((data, dto_response.pagination))
1561        })
1562        .await
1563        .map_err(|err| RPCError::Fatal(format!("Decoding contract_state failed: {err}")))??;
1564        Ok(Page::new(data, pagination.total, pagination.page, pagination.page_size))
1565    }
1566
1567    async fn get_protocol_components(
1568        &self,
1569        params: ProtocolComponentsParams,
1570    ) -> Result<Page<Vec<ProtocolComponent>>, RPCError> {
1571        let request = ProtocolComponentsRequestBody {
1572            protocol_system: params.protocol_system,
1573            component_ids: params.component_ids,
1574            tvl_gt: params.tvl_gt,
1575            chain: params.chain.into(),
1576            pagination: PaginationParams { page: params.page, page_size: params.page_size },
1577        };
1578
1579        let uri = format!(
1580            "{}/{}/protocol_components",
1581            self.url
1582                .to_string()
1583                .trim_end_matches('/'),
1584            TYCHO_SERVER_VERSION,
1585        );
1586        debug!(%uri, "Sending protocol_components request to Tycho server");
1587        trace!(?request, "Sending request to Tycho server");
1588
1589        let response = self
1590            .make_post_request(&request, &uri)
1591            .await?;
1592
1593        trace!(?response, "Received response from Tycho server");
1594
1595        let body = response
1596            .text()
1597            .await
1598            .map_err(|e| RPCError::ParseResponse(e.to_string()))?;
1599        let dto_response = serde_json::from_str::<ProtocolComponentRequestResponse>(&body)
1600            .map_err(|err| RPCError::from_parse_error(err, &body))?;
1601        trace!(?dto_response, "Received protocol_components response from Tycho server");
1602
1603        let data: Vec<ProtocolComponent> = dto_response
1604            .protocol_components
1605            .into_iter()
1606            .map(ProtocolComponent::from)
1607            .collect();
1608        Ok(Page::new(
1609            data,
1610            dto_response.pagination.total,
1611            dto_response.pagination.page,
1612            dto_response.pagination.page_size,
1613        ))
1614    }
1615
1616    async fn get_protocol_states(
1617        &self,
1618        params: ProtocolStatesParams,
1619    ) -> Result<Page<Vec<ProtocolComponentState>>, RPCError> {
1620        if params
1621            .protocol_ids
1622            .as_ref()
1623            .is_none_or(|ids| ids.is_empty())
1624        {
1625            warn!("No protocol ids specified in request.");
1626        }
1627
1628        let request = ProtocolStateRequestBody {
1629            protocol_ids: params.protocol_ids,
1630            protocol_system: params.protocol_system,
1631            chain: params.chain.into(),
1632            include_balances: params.include_balances,
1633            version: params.version,
1634            pagination: PaginationParams { page: params.page, page_size: params.page_size },
1635        };
1636
1637        let uri = format!(
1638            "{}/{}/protocol_state",
1639            self.url
1640                .to_string()
1641                .trim_end_matches('/'),
1642            TYCHO_SERVER_VERSION
1643        );
1644        debug!(%uri, "Sending protocol_states request to Tycho server");
1645        trace!(?request, "Sending request to Tycho server");
1646
1647        let response = self
1648            .make_post_request(&request, &uri)
1649            .await?;
1650        trace!(?response, "Received response from Tycho server");
1651
1652        let body = response
1653            .text()
1654            .await
1655            .map_err(|e| RPCError::ParseResponse(e.to_string()))?;
1656
1657        if body.is_empty() {
1658            // Pure VM protocols will return empty states
1659            return Ok(Page::new(vec![], 0, 0, 0));
1660        }
1661
1662        let dto_response = serde_json::from_str::<ProtocolStateRequestResponse>(&body)
1663            .map_err(|err| RPCError::from_parse_error(err, &body))?;
1664        trace!(?dto_response, "Received protocol_states response from Tycho server");
1665
1666        let data: Vec<ProtocolComponentState> = dto_response
1667            .states
1668            .into_iter()
1669            .map(ProtocolComponentState::from)
1670            .collect();
1671        Ok(Page::new(
1672            data,
1673            dto_response.pagination.total,
1674            dto_response.pagination.page,
1675            dto_response.pagination.page_size,
1676        ))
1677    }
1678
1679    async fn get_tokens(&self, params: TokensParams) -> Result<Page<Vec<Token>>, RPCError> {
1680        let request = TokensRequestBody {
1681            token_addresses: None,
1682            min_quality: params.min_quality,
1683            traded_n_days_ago: params.traded_n_days_ago,
1684            pagination: PaginationParams { page: params.page, page_size: params.page_size },
1685            chain: params.chain.into(),
1686        };
1687
1688        let uri = format!(
1689            "{}/{}/tokens",
1690            self.url
1691                .to_string()
1692                .trim_end_matches('/'),
1693            TYCHO_SERVER_VERSION
1694        );
1695        debug!(%uri, "Sending tokens request to Tycho server");
1696
1697        let response = self
1698            .make_post_request(&request, &uri)
1699            .await?;
1700
1701        let body = response
1702            .text()
1703            .await
1704            .map_err(|e| RPCError::ParseResponse(e.to_string()))?;
1705        let dto_response = serde_json::from_str::<TokensRequestResponse>(&body)
1706            .map_err(|err| RPCError::ParseResponse(format!("Error: {err}, Body: {body}")))?;
1707
1708        let data: Vec<Token> = dto_response
1709            .tokens
1710            .into_iter()
1711            .map(Token::from)
1712            .collect();
1713        Ok(Page::new(
1714            data,
1715            dto_response.pagination.total,
1716            dto_response.pagination.page,
1717            dto_response.pagination.page_size,
1718        ))
1719    }
1720
1721    async fn get_protocol_systems(
1722        &self,
1723        params: ProtocolSystemsParams,
1724    ) -> Result<Page<ProtocolSystems>, RPCError> {
1725        let request = ProtocolSystemsRequestBody {
1726            chain: params.chain.into(),
1727            pagination: PaginationParams { page: params.page, page_size: params.page_size },
1728        };
1729
1730        let uri = format!(
1731            "{}/{}/protocol_systems",
1732            self.url
1733                .to_string()
1734                .trim_end_matches('/'),
1735            TYCHO_SERVER_VERSION
1736        );
1737        debug!(%uri, "Sending protocol_systems request to Tycho server");
1738        trace!(?request, "Sending request to Tycho server");
1739        let response = self
1740            .make_post_request(&request, &uri)
1741            .await?;
1742        trace!(?response, "Received response from Tycho server");
1743        let body = response
1744            .text()
1745            .await
1746            .map_err(|e| RPCError::ParseResponse(e.to_string()))?;
1747        let dto = serde_json::from_str::<ProtocolSystemsRequestResponse>(&body)
1748            .map_err(|err| RPCError::ParseResponse(format!("Error: {err}, Body: {body}")))?;
1749        trace!(?dto, "Received protocol_systems response from Tycho server");
1750        Ok(Page::new(
1751            ProtocolSystems::new(dto.protocol_systems, dto.dci_protocols),
1752            dto.pagination.total,
1753            dto.pagination.page,
1754            dto.pagination.page_size,
1755        ))
1756    }
1757
1758    async fn get_component_tvl(
1759        &self,
1760        params: ComponentTvlParams,
1761    ) -> Result<Page<HashMap<String, f64>>, RPCError> {
1762        let request = ComponentTvlRequestBody {
1763            chain: params.chain.into(),
1764            protocol_system: params.protocol_system,
1765            component_ids: params.component_ids,
1766            pagination: PaginationParams { page: params.page, page_size: params.page_size },
1767        };
1768
1769        let uri = format!(
1770            "{}/{}/component_tvl",
1771            self.url
1772                .to_string()
1773                .trim_end_matches('/'),
1774            TYCHO_SERVER_VERSION
1775        );
1776        debug!(%uri, "Sending get_component_tvl request to Tycho server");
1777        trace!(?request, "Sending request to Tycho server");
1778        let response = self
1779            .make_post_request(&request, &uri)
1780            .await?;
1781        trace!(?response, "Received response from Tycho server");
1782        let body = response
1783            .text()
1784            .await
1785            .map_err(|e| RPCError::ParseResponse(e.to_string()))?;
1786        let dto_response =
1787            serde_json::from_str::<ComponentTvlRequestResponse>(&body).map_err(|err| {
1788                error!("Failed to parse component_tvl response: {:?}", &body);
1789                RPCError::ParseResponse(format!("Error: {err}, Body: {body}"))
1790            })?;
1791        trace!(?dto_response, "Received component_tvl response from Tycho server");
1792        Ok(Page::new(
1793            dto_response.tvl,
1794            dto_response.pagination.total,
1795            dto_response.pagination.page,
1796            dto_response.pagination.page_size,
1797        ))
1798    }
1799
1800    async fn get_traced_entry_points(
1801        &self,
1802        params: TracedEntryPointsParams,
1803    ) -> Result<Page<TracedEntryPoints>, RPCError> {
1804        let request = TracedEntryPointRequestBody {
1805            chain: params.chain.into(),
1806            protocol_system: params.protocol_system,
1807            component_ids: params.component_ids,
1808            pagination: PaginationParams { page: params.page, page_size: params.page_size },
1809        };
1810
1811        let uri = format!(
1812            "{}/{TYCHO_SERVER_VERSION}/traced_entry_points",
1813            self.url
1814                .to_string()
1815                .trim_end_matches('/')
1816        );
1817        debug!(%uri, "Sending traced_entry_points request to Tycho server");
1818        trace!(?request, "Sending request to Tycho server");
1819
1820        let response = self
1821            .make_post_request(&request, &uri)
1822            .await?;
1823
1824        trace!(?response, "Received response from Tycho server");
1825
1826        let body = response
1827            .text()
1828            .await
1829            .map_err(|e| RPCError::ParseResponse(e.to_string()))?;
1830        let dto_response =
1831            serde_json::from_str::<TracedEntryPointRequestResponse>(&body).map_err(|err| {
1832                error!("Failed to parse traced_entry_points response: {:?}", &body);
1833                RPCError::ParseResponse(format!("Error: {err}, Body: {body}"))
1834            })?;
1835        trace!(?dto_response, "Received traced_entry_points response from Tycho server");
1836        let data: TracedEntryPoints = dto_response
1837            .traced_entry_points
1838            .into_iter()
1839            .map(|(k, v)| {
1840                (
1841                    k,
1842                    v.into_iter()
1843                        .map(|(ep, tr)| {
1844                            (EntryPointWithTracingParams::from(ep), TracingResult::from(tr))
1845                        })
1846                        .collect(),
1847                )
1848            })
1849            .collect();
1850        Ok(Page::new(
1851            data,
1852            dto_response.pagination.total,
1853            dto_response.pagination.page,
1854            dto_response.pagination.page_size,
1855        ))
1856    }
1857
1858    async fn get_snapshots<'a>(
1859        &self,
1860        request: &SnapshotParameters<'a>,
1861        chunk_size: Option<usize>,
1862        concurrency: usize,
1863    ) -> Result<Snapshot, RPCError> {
1864        let component_ids: Vec<_> = request
1865            .components
1866            .keys()
1867            .cloned()
1868            .collect();
1869
1870        let component_tvl = if request.include_tvl && !component_ids.is_empty() {
1871            self.get_component_tvl_paginated(
1872                ComponentTvlPaginatedParams::new(request.chain, concurrency)
1873                    .with_component_ids(component_ids.clone()),
1874            )
1875            .await?
1876        } else {
1877            HashMap::new()
1878        };
1879
1880        let version = VersionParam::at_block(request.chain.into(), request.block_number);
1881
1882        let mut protocol_states = if !component_ids.is_empty() {
1883            self.get_protocol_states_paginated(
1884                ProtocolStatesPaginatedParams::new(
1885                    request.chain,
1886                    request.protocol_system,
1887                    concurrency,
1888                )
1889                .with_protocol_ids(component_ids.clone())
1890                .with_include_balances(request.include_balances)
1891                .with_version(version.clone()),
1892            )
1893            .await?
1894            .into_iter()
1895            .map(|state| (state.component_id.clone(), state))
1896            .collect()
1897        } else {
1898            HashMap::new()
1899        };
1900
1901        // Convert to ComponentWithState, which includes entrypoint information.
1902        let states = request
1903            .components
1904            .values()
1905            .filter_map(|component| {
1906                if let Some(state) = protocol_states.remove(&component.id) {
1907                    Some((
1908                        component.id.clone(),
1909                        ComponentWithState {
1910                            state,
1911                            component: component.clone(),
1912                            component_tvl: component_tvl
1913                                .get(&component.id)
1914                                .cloned(),
1915                            entrypoints: request
1916                                .entrypoints
1917                                .as_ref()
1918                                .and_then(|map| map.get(&component.id))
1919                                .cloned()
1920                                .unwrap_or_default(),
1921                        },
1922                    ))
1923                } else if component_ids.contains(&component.id) {
1924                    // only emit error event if we requested this component
1925                    let component_id = &component.id;
1926                    error!(?component_id, "Missing state for native component!");
1927                    None
1928                } else {
1929                    None
1930                }
1931            })
1932            .collect();
1933
1934        let vm_storage = if !request.contract_ids.is_empty() {
1935            let mut cp_params = ContractStatePaginatedParams::new(
1936                request.chain,
1937                request.protocol_system,
1938                concurrency,
1939            )
1940            .with_contract_ids(request.contract_ids.to_vec())
1941            .with_version(version.clone());
1942            if let Some(cs) = chunk_size {
1943                cp_params = cp_params.with_chunk_size(cs);
1944            }
1945            let contract_states = self
1946                .get_contract_state_paginated(cp_params)
1947                .await?
1948                .into_iter()
1949                .map(|acc| (acc.address.clone(), acc))
1950                .collect::<HashMap<_, _>>();
1951
1952            trace!(states=?&contract_states, "Retrieved ContractState");
1953
1954            let contract_address_to_components = request
1955                .components
1956                .iter()
1957                .filter_map(|(id, comp)| {
1958                    if component_ids.contains(id) {
1959                        Some(
1960                            comp.contract_addresses
1961                                .iter()
1962                                .map(|address| (address.clone(), comp.id.clone())),
1963                        )
1964                    } else {
1965                        None
1966                    }
1967                })
1968                .flatten()
1969                .fold(HashMap::<Bytes, Vec<String>>::new(), |mut acc, (addr, c_id)| {
1970                    acc.entry(addr).or_default().push(c_id);
1971                    acc
1972                });
1973
1974            request
1975                .contract_ids
1976                .iter()
1977                .filter_map(|address| {
1978                    if let Some(state) = contract_states.get(address) {
1979                        Some((address.clone(), state.clone()))
1980                    } else if let Some(ids) = contract_address_to_components.get(address) {
1981                        // only emit error even if we did actually request this address
1982                        error!(
1983                            ?address,
1984                            ?ids,
1985                            "Component with lacking contract storage encountered!"
1986                        );
1987                        None
1988                    } else {
1989                        None
1990                    }
1991                })
1992                .collect()
1993        } else {
1994            HashMap::new()
1995        };
1996
1997        Ok(Snapshot { states, vm_storage })
1998    }
1999}
2000
2001#[cfg(test)]
2002mod tests {
2003    use std::{
2004        collections::{HashMap, HashSet},
2005        str::FromStr,
2006    };
2007
2008    use mockito::Server;
2009    use rstest::rstest;
2010    use tycho_common::models::blockchain::AddressStorageLocation;
2011
2012    use super::*;
2013
2014    // Dummy implementation of `get_protocol_states_paginated` for backwards compatibility testing
2015    // purposes
2016    impl MockRPCClient {
2017        #[allow(clippy::too_many_arguments)]
2018        async fn test_get_protocol_states_paginated<T>(
2019            &self,
2020            chain: Chain,
2021            ids: &[T],
2022            protocol_system: &str,
2023            include_balances: bool,
2024            block_number: Option<u64>,
2025            chunk_size: usize,
2026            _concurrency: usize,
2027        ) -> Vec<(Chain, Vec<String>, String, bool, Option<u64>, PaginationParams)>
2028        where
2029            T: AsRef<str> + Clone + Send + Sync + 'static,
2030        {
2031            ids.chunks(chunk_size)
2032                .map(|chunk| {
2033                    (
2034                        chain,
2035                        chunk
2036                            .iter()
2037                            .map(|id| id.as_ref().to_string())
2038                            .collect(),
2039                        protocol_system.to_string(),
2040                        include_balances,
2041                        block_number,
2042                        PaginationParams { page: 0, page_size: chunk_size as i64 },
2043                    )
2044                })
2045                .collect()
2046        }
2047    }
2048
2049    const GET_CONTRACT_STATE_RESP: &str = r#"
2050        {
2051            "accounts": [
2052                {
2053                    "chain": "ethereum",
2054                    "address": "0x0000000000000000000000000000000000000000",
2055                    "title": "",
2056                    "slots": {},
2057                    "native_balance": "0x01f4",
2058                    "token_balances": {},
2059                    "code": "0x00",
2060                    "code_hash": "0x5c06b7c5b3d910fd33bc2229846f9ddaf91d584d9b196e16636901ac3a77077e",
2061                    "balance_modify_tx": "0x0000000000000000000000000000000000000000000000000000000000000000",
2062                    "code_modify_tx": "0x0000000000000000000000000000000000000000000000000000000000000000",
2063                    "creation_tx": null
2064                }
2065            ],
2066            "pagination": {
2067                "page": 0,
2068                "page_size": 20,
2069                "total": 10
2070            }
2071        }
2072        "#;
2073
2074    #[rstest]
2075    #[case::string_input(vec![
2076        "id1".to_string(),
2077        "id2".to_string()
2078    ])]
2079    #[tokio::test]
2080    async fn test_get_protocol_states_paginated<T>(#[case] ids: Vec<T>)
2081    where
2082        T: AsRef<str> + Clone + Send + Sync + 'static,
2083    {
2084        let mock_client = MockRPCClient::new();
2085
2086        let request_args = mock_client
2087            .test_get_protocol_states_paginated(
2088                Chain::Ethereum,
2089                &ids,
2090                "test_system",
2091                true,
2092                None,
2093                2,
2094                2,
2095            )
2096            .await;
2097
2098        // Verify that the request args have been split into chunks correctly
2099        assert_eq!(request_args.len(), 1);
2100        assert_eq!(request_args[0].1.len(), 2);
2101    }
2102
2103    #[tokio::test]
2104    async fn test_get_contract_state() {
2105        let mut server = Server::new_async().await;
2106        let server_resp = GET_CONTRACT_STATE_RESP;
2107        // test that the response is deserialized correctly
2108        serde_json::from_str::<StateRequestResponse>(server_resp).expect("deserialize");
2109
2110        let mocked_server = server
2111            .mock("POST", "/v1/contract_state")
2112            .expect(1)
2113            .with_body(server_resp)
2114            .create_async()
2115            .await;
2116
2117        let client = HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
2118            .expect("create client");
2119
2120        let accounts = client
2121            .get_contract_state(ContractStateParams::new(Chain::Ethereum, ""))
2122            .await
2123            .expect("get state");
2124
2125        mocked_server.assert();
2126        assert_eq!(accounts.data().len(), 1);
2127        assert_eq!(accounts.data()[0].slots, HashMap::new());
2128        assert_eq!(accounts.data()[0].native_balance, Bytes::from(500u16.to_be_bytes()));
2129        assert_eq!(accounts.data()[0].code, [0].to_vec());
2130        assert_eq!(
2131            accounts.data()[0].code_hash,
2132            hex::decode("5c06b7c5b3d910fd33bc2229846f9ddaf91d584d9b196e16636901ac3a77077e")
2133                .unwrap()
2134        );
2135    }
2136
2137    #[tokio::test]
2138    async fn test_get_contract_state_paginated_requests_default_sized_chunks() {
2139        let mut server = Server::new_async().await;
2140        let full_chunks = server
2141            .mock("POST", "/v1/contract_state")
2142            .match_body(mockito::Matcher::PartialJson(serde_json::json!({
2143                "pagination": {"page": 0, "page_size": 100}
2144            })))
2145            .expect(3)
2146            .with_body(GET_CONTRACT_STATE_RESP)
2147            .create_async()
2148            .await;
2149        let client = HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
2150            .expect("create client");
2151        let ids: Vec<Bytes> = (0..250u16)
2152            .map(|i| Bytes::from(i.to_be_bytes()))
2153            .collect();
2154
2155        let accounts = client
2156            .get_contract_state_paginated(
2157                ContractStatePaginatedParams::new(Chain::Ethereum, "", RPC_CLIENT_CONCURRENCY)
2158                    .with_contract_ids(ids),
2159            )
2160            .await
2161            .expect("get state");
2162
2163        full_chunks.assert();
2164        assert_eq!(accounts.len(), 3);
2165    }
2166
2167    #[tokio::test]
2168    async fn test_get_protocol_components_paginated_requests_every_id_chunk_as_page_zero() {
2169        let mut server = Server::new_async().await;
2170        let chunks = server
2171            .mock("POST", "/v1/protocol_components")
2172            .match_body(mockito::Matcher::PartialJson(serde_json::json!({
2173                "pagination": {"page": 0, "page_size": 100}
2174            })))
2175            .expect(3)
2176            .with_body(r#"{"protocol_components": [], "pagination": {"page": 0, "page_size": 100, "total": 0}}"#)
2177            .create_async()
2178            .await;
2179        let client = HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
2180            .expect("create client");
2181        let ids: Vec<String> = (0..250)
2182            .map(|i| format!("component_{i}"))
2183            .collect();
2184
2185        client
2186            .get_protocol_components_paginated(
2187                ProtocolComponentsPaginatedParams::new(
2188                    Chain::Ethereum,
2189                    "uniswap_v2",
2190                    RPC_CLIENT_CONCURRENCY,
2191                )
2192                .with_component_ids(ids)
2193                .with_chunk_size(100),
2194            )
2195            .await
2196            .expect("get components");
2197
2198        chunks.assert();
2199    }
2200
2201    #[tokio::test]
2202    async fn test_get_component_tvl_paginated_requests_every_id_chunk_as_page_zero() {
2203        let mut server = Server::new_async().await;
2204        let chunks = server
2205            .mock("POST", "/v1/component_tvl")
2206            .match_body(mockito::Matcher::PartialJson(serde_json::json!({
2207                "pagination": {"page": 0, "page_size": 100}
2208            })))
2209            .expect(3)
2210            .with_body(r#"{"tvl": {}, "pagination": {"page": 0, "page_size": 100, "total": 0}}"#)
2211            .create_async()
2212            .await;
2213        let client = HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
2214            .expect("create client");
2215        let ids: Vec<String> = (0..250)
2216            .map(|i| format!("component_{i}"))
2217            .collect();
2218
2219        client
2220            .get_component_tvl_paginated(
2221                ComponentTvlPaginatedParams::new(Chain::Ethereum, RPC_CLIENT_CONCURRENCY)
2222                    .with_component_ids(ids)
2223                    .with_chunk_size(100),
2224            )
2225            .await
2226            .expect("get component tvl");
2227
2228        chunks.assert();
2229    }
2230
2231    #[tokio::test]
2232    async fn test_get_protocol_components_paginated_clamps_chunk_size_to_max_page_size() {
2233        let mut server = Server::new_async().await;
2234        let max = ProtocolComponentsRequestBody::MAX_PAGE_SIZE_COMPRESSED;
2235        let chunks = server
2236            .mock("POST", "/v1/protocol_components")
2237            .match_body(mockito::Matcher::PartialJson(serde_json::json!({
2238                "pagination": {"page": 0, "page_size": max}
2239            })))
2240            .expect(2)
2241            .with_body(format!(
2242                r#"{{"protocol_components": [], "pagination": {{"page": 0, "page_size": {max}, "total": 0}}}}"#
2243            ))
2244            .create_async()
2245            .await;
2246        let client = HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
2247            .expect("create client");
2248        let ids: Vec<String> = (0..max + 1)
2249            .map(|i| format!("component_{i}"))
2250            .collect();
2251
2252        client
2253            .get_protocol_components_paginated(
2254                ProtocolComponentsPaginatedParams::new(
2255                    Chain::Ethereum,
2256                    "uniswap_v2",
2257                    RPC_CLIENT_CONCURRENCY,
2258                )
2259                .with_component_ids(ids)
2260                .with_chunk_size(max as usize * 2),
2261            )
2262            .await
2263            .expect("get components");
2264
2265        chunks.assert();
2266    }
2267
2268    #[tokio::test]
2269    async fn test_get_component_tvl_paginated_clamps_chunk_size_to_max_page_size() {
2270        let mut server = Server::new_async().await;
2271        let max = ComponentTvlRequestBody::MAX_PAGE_SIZE_COMPRESSED;
2272        let chunks = server
2273            .mock("POST", "/v1/component_tvl")
2274            .match_body(mockito::Matcher::PartialJson(serde_json::json!({
2275                "pagination": {"page": 0, "page_size": max}
2276            })))
2277            .expect(2)
2278            .with_body(format!(
2279                r#"{{"tvl": {{}}, "pagination": {{"page": 0, "page_size": {max}, "total": 0}}}}"#
2280            ))
2281            .create_async()
2282            .await;
2283        let client = HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
2284            .expect("create client");
2285        let ids: Vec<String> = (0..max + 1)
2286            .map(|i| format!("component_{i}"))
2287            .collect();
2288
2289        client
2290            .get_component_tvl_paginated(
2291                ComponentTvlPaginatedParams::new(Chain::Ethereum, RPC_CLIENT_CONCURRENCY)
2292                    .with_component_ids(ids)
2293                    .with_chunk_size(max as usize * 2),
2294            )
2295            .await
2296            .expect("get component tvl");
2297
2298        chunks.assert();
2299    }
2300
2301    #[tokio::test]
2302    async fn test_client_metadata_header_sent_when_set() {
2303        let mut server = Server::new_async().await;
2304        let expected_ua = format!("tycho-client-{}", env!("CARGO_PKG_VERSION"));
2305        let mock = server
2306            .mock("POST", "/v1/contract_state")
2307            .match_header(CLIENT_METADATA_HEADER, "fynd_version=0.57.0;preset=best")
2308            .match_header("user-agent", expected_ua.as_str())
2309            .with_body(GET_CONTRACT_STATE_RESP)
2310            .expect(1)
2311            .create_async()
2312            .await;
2313
2314        let client = HttpRPCClient::new(
2315            server.url().as_str(),
2316            HttpRPCClientOptions::default()
2317                .with_client_metadata_header(Some("fynd_version=0.57.0;preset=best".to_string())),
2318        )
2319        .expect("create client");
2320
2321        client
2322            .get_contract_state(ContractStateParams::new(Chain::Ethereum, ""))
2323            .await
2324            .expect("get state");
2325
2326        mock.assert_async().await;
2327    }
2328
2329    #[tokio::test]
2330    async fn test_no_client_metadata_header_when_unset() {
2331        let mut server = Server::new_async().await;
2332        let expected_ua = format!("tycho-client-{}", env!("CARGO_PKG_VERSION"));
2333        let mock = server
2334            .mock("POST", "/v1/contract_state")
2335            .match_header(CLIENT_METADATA_HEADER, mockito::Matcher::Missing)
2336            .match_header("user-agent", expected_ua.as_str())
2337            .with_body(GET_CONTRACT_STATE_RESP)
2338            .expect(1)
2339            .create_async()
2340            .await;
2341
2342        let client = HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
2343            .expect("create client");
2344
2345        client
2346            .get_contract_state(ContractStateParams::new(Chain::Ethereum, ""))
2347            .await
2348            .expect("get state");
2349
2350        mock.assert_async().await;
2351    }
2352
2353    #[tokio::test]
2354    async fn test_get_protocol_components() {
2355        let mut server = Server::new_async().await;
2356        let server_resp = r#"
2357        {
2358            "protocol_components": [
2359                {
2360                    "id": "State1",
2361                    "protocol_system": "ambient",
2362                    "protocol_type_name": "Pool",
2363                    "chain": "ethereum",
2364                    "tokens": [
2365                        "0x0000000000000000000000000000000000000000",
2366                        "0x0000000000000000000000000000000000000001"
2367                    ],
2368                    "contract_ids": [
2369                        "0x0000000000000000000000000000000000000000"
2370                    ],
2371                    "static_attributes": {
2372                        "attribute_1": "0x00000000000003e8"
2373                    },
2374                    "change": "Creation",
2375                    "creation_tx": "0x0000000000000000000000000000000000000000000000000000000000000000",
2376                    "created_at": "2022-01-01T00:00:00"
2377                }
2378            ],
2379            "pagination": {
2380                "page": 0,
2381                "page_size": 20,
2382                "total": 10
2383            }
2384        }
2385        "#;
2386        // test that the response is deserialized correctly
2387        serde_json::from_str::<ProtocolComponentRequestResponse>(server_resp).expect("deserialize");
2388
2389        let mocked_server = server
2390            .mock("POST", "/v1/protocol_components")
2391            .expect(1)
2392            .with_body(server_resp)
2393            .create_async()
2394            .await;
2395
2396        let client = HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
2397            .expect("create client");
2398
2399        let components = client
2400            .get_protocol_components(ProtocolComponentsParams::new(Chain::Ethereum, ""))
2401            .await
2402            .expect("get state");
2403
2404        mocked_server.assert();
2405        assert_eq!(components.data().len(), 1);
2406        assert_eq!(components.data()[0].id, "State1");
2407        assert_eq!(components.data()[0].protocol_system, "ambient");
2408        assert_eq!(components.data()[0].protocol_type_name, "Pool");
2409        assert_eq!(components.data()[0].tokens.len(), 2);
2410        let expected_attributes =
2411            [("attribute_1".to_string(), Bytes::from(1000_u64.to_be_bytes()))]
2412                .iter()
2413                .cloned()
2414                .collect::<HashMap<String, Bytes>>();
2415        assert_eq!(components.data()[0].static_attributes, expected_attributes);
2416    }
2417
2418    #[tokio::test]
2419    async fn test_get_protocol_states() {
2420        let mut server = Server::new_async().await;
2421        let server_resp = r#"
2422        {
2423            "states": [
2424                {
2425                    "component_id": "State1",
2426                    "attributes": {
2427                        "attribute_1": "0x00000000000003e8"
2428                    },
2429                    "balances": {
2430                        "0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2": "0x01f4"
2431                    }
2432                }
2433            ],
2434            "pagination": {
2435                "page": 0,
2436                "page_size": 20,
2437                "total": 10
2438            }
2439        }
2440        "#;
2441        // test that the response is deserialized correctly
2442        serde_json::from_str::<ProtocolStateRequestResponse>(server_resp).expect("deserialize");
2443
2444        let mocked_server = server
2445            .mock("POST", "/v1/protocol_state")
2446            .expect(1)
2447            .with_body(server_resp)
2448            .create_async()
2449            .await;
2450        let client = HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
2451            .expect("create client");
2452
2453        let states = client
2454            .get_protocol_states(
2455                ProtocolStatesParams::new(Chain::Ethereum, "").with_include_balances(true),
2456            )
2457            .await
2458            .expect("get state");
2459
2460        mocked_server.assert();
2461        assert_eq!(states.data().len(), 1);
2462        assert_eq!(states.data()[0].component_id, "State1");
2463        let expected_attributes =
2464            [("attribute_1".to_string(), Bytes::from(1000_u64.to_be_bytes()))]
2465                .iter()
2466                .cloned()
2467                .collect::<HashMap<String, Bytes>>();
2468        assert_eq!(states.data()[0].attributes, expected_attributes);
2469        let expected_balances = [(
2470            Bytes::from_str("0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2")
2471                .expect("Unsupported address format"),
2472            Bytes::from_str("0x01f4").unwrap(),
2473        )]
2474        .iter()
2475        .cloned()
2476        .collect::<HashMap<Bytes, Bytes>>();
2477        assert_eq!(states.data()[0].balances, expected_balances);
2478    }
2479
2480    #[tokio::test]
2481    async fn test_get_tokens() {
2482        let mut server = Server::new_async().await;
2483        let server_resp = r#"
2484        {
2485            "tokens": [
2486              {
2487                "chain": "ethereum",
2488                "address": "0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2",
2489                "symbol": "WETH",
2490                "decimals": 18,
2491                "tax": 0,
2492                "gas": [
2493                  29962
2494                ],
2495                "quality": 100
2496              },
2497              {
2498                "chain": "ethereum",
2499                "address": "0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48",
2500                "symbol": "USDC",
2501                "decimals": 6,
2502                "tax": 0,
2503                "gas": [
2504                  40652
2505                ],
2506                "quality": 100
2507              }
2508            ],
2509            "pagination": {
2510              "page": 0,
2511              "page_size": 20,
2512              "total": 10
2513            }
2514          }
2515        "#;
2516        // test that the response is deserialized correctly
2517        serde_json::from_str::<TokensRequestResponse>(server_resp).expect("deserialize");
2518
2519        let mocked_server = server
2520            .mock("POST", "/v1/tokens")
2521            .expect(1)
2522            .with_body(server_resp)
2523            .create_async()
2524            .await;
2525        let client = HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
2526            .expect("create client");
2527
2528        let tokens = client
2529            .get_tokens(TokensParams::new(Chain::Ethereum))
2530            .await
2531            .expect("get tokens");
2532
2533        let expected = vec![
2534            Token {
2535                chain: tycho_common::models::Chain::Ethereum,
2536                address: Bytes::from_str("0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2").unwrap(),
2537                symbol: "WETH".to_string(),
2538                decimals: 18,
2539                tax: 0,
2540                gas: vec![Some(29962)],
2541                quality: 100,
2542            },
2543            Token {
2544                chain: tycho_common::models::Chain::Ethereum,
2545                address: Bytes::from_str("0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48").unwrap(),
2546                symbol: "USDC".to_string(),
2547                decimals: 6,
2548                tax: 0,
2549                gas: vec![Some(40652)],
2550                quality: 100,
2551            },
2552        ];
2553
2554        mocked_server.assert();
2555        assert_eq!(*tokens.data(), expected);
2556    }
2557
2558    #[rstest]
2559    #[case::with_dci(Some(vec!["system2"]), vec!["system2"])]
2560    #[case::backward_compat(None, vec![])]
2561    #[tokio::test]
2562    async fn test_get_protocol_systems(
2563        #[case] dci_protocols: Option<Vec<&str>>,
2564        #[case] expected_dci: Vec<&str>,
2565    ) {
2566        use serde_json::json;
2567
2568        let mut json_value = json!({
2569            "protocol_systems": ["system1", "system2"],
2570            "pagination": { "page": 0, "page_size": 20, "total": 2 }
2571        });
2572        if let Some(dci) = dci_protocols {
2573            json_value["dci_protocols"] = json!(dci);
2574        }
2575        let server_resp = serde_json::to_string(&json_value).unwrap();
2576
2577        let mut server = Server::new_async().await;
2578        let mocked_server = server
2579            .mock("POST", "/v1/protocol_systems")
2580            .expect(1)
2581            .with_body(&server_resp)
2582            .create_async()
2583            .await;
2584        let client = HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
2585            .expect("create client");
2586
2587        let response = client
2588            .get_protocol_systems(ProtocolSystemsParams::new(Chain::Ethereum))
2589            .await
2590            .expect("get protocol systems");
2591
2592        mocked_server.assert();
2593        assert_eq!(response.data().protocol_systems(), ["system1", "system2"]);
2594        assert_eq!(response.data().dci_protocols(), expected_dci.as_slice());
2595    }
2596
2597    #[tokio::test]
2598    async fn test_get_component_tvl() {
2599        let mut server = Server::new_async().await;
2600        let server_resp = r#"
2601        {
2602            "tvl": {
2603                "component1": 100.0
2604            },
2605            "pagination": {
2606                "page": 0,
2607                "page_size": 20,
2608                "total": 10
2609            }
2610        }
2611        "#;
2612        // test that the response is deserialized correctly
2613        serde_json::from_str::<ComponentTvlRequestResponse>(server_resp).expect("deserialize");
2614
2615        let mocked_server = server
2616            .mock("POST", "/v1/component_tvl")
2617            .expect(1)
2618            .with_body(server_resp)
2619            .create_async()
2620            .await;
2621        let client = HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
2622            .expect("create client");
2623
2624        let component_tvl = client
2625            .get_component_tvl(ComponentTvlParams::new(Chain::Ethereum))
2626            .await
2627            .expect("get component tvl");
2628
2629        mocked_server.assert();
2630        assert_eq!(component_tvl.data().get("component1"), Some(&100.0));
2631    }
2632
2633    #[tokio::test]
2634    async fn test_get_traced_entry_points() {
2635        let mut server = Server::new_async().await;
2636        let server_resp = r#"
2637        {
2638            "traced_entry_points": {
2639                "component_1": [
2640                    [
2641                        {
2642                            "entry_point": {
2643                                "external_id": "entrypoint_a",
2644                                "target": "0x0000000000000000000000000000000000000001",
2645                                "signature": "sig()"
2646                            },
2647                            "params": {
2648                                "method": "rpctracer",
2649                                "caller": "0x000000000000000000000000000000000000000a",
2650                                "calldata": "0x000000000000000000000000000000000000000b"
2651                            }
2652                        },
2653                        {
2654                            "retriggers": [
2655                                [
2656                                    "0x00000000000000000000000000000000000000aa",
2657                                    {"key": "0x0000000000000000000000000000000000000aaa", "offset": 12}
2658                                ]
2659                            ],
2660                            "accessed_slots": {
2661                                "0x0000000000000000000000000000000000aaaa": [
2662                                    "0x0000000000000000000000000000000000aaaa"
2663                                ]
2664                            }
2665                        }
2666                    ]
2667                ]
2668            },
2669            "pagination": {
2670                "page": 0,
2671                "page_size": 20,
2672                "total": 1
2673            }
2674        }
2675        "#;
2676        // test that the response is deserialized correctly
2677        serde_json::from_str::<TracedEntryPointRequestResponse>(server_resp).expect("deserialize");
2678
2679        let mocked_server = server
2680            .mock("POST", "/v1/traced_entry_points")
2681            .expect(1)
2682            .with_body(server_resp)
2683            .create_async()
2684            .await;
2685        let client = HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
2686            .expect("create client");
2687
2688        let entrypoints = client
2689            .get_traced_entry_points(TracedEntryPointsParams::new(Chain::Ethereum, ""))
2690            .await
2691            .expect("get traced entry points");
2692
2693        mocked_server.assert();
2694        assert_eq!(entrypoints.data().len(), 1);
2695        let comp1_entrypoints = entrypoints
2696            .data()
2697            .get("component_1")
2698            .expect("component_1 entrypoints should exist");
2699        assert_eq!(comp1_entrypoints.len(), 1);
2700
2701        let (entrypoint, trace_result) = &comp1_entrypoints[0];
2702        assert_eq!(entrypoint.entry_point.external_id, "entrypoint_a");
2703        assert_eq!(
2704            entrypoint.entry_point.target,
2705            Bytes::from_str("0x0000000000000000000000000000000000000001").unwrap()
2706        );
2707        assert_eq!(entrypoint.entry_point.signature, "sig()");
2708        let tycho_common::models::blockchain::TracingParams::RPCTracer(rpc_params) =
2709            &entrypoint.params;
2710        assert_eq!(
2711            rpc_params.caller,
2712            Some(Bytes::from("0x000000000000000000000000000000000000000a"))
2713        );
2714        assert_eq!(rpc_params.calldata, Bytes::from("0x000000000000000000000000000000000000000b"));
2715
2716        assert_eq!(
2717            trace_result.retriggers,
2718            HashSet::from([(
2719                Bytes::from("0x00000000000000000000000000000000000000aa"),
2720                AddressStorageLocation::new(
2721                    Bytes::from("0x0000000000000000000000000000000000000aaa"),
2722                    12
2723                )
2724            )])
2725        );
2726        assert_eq!(trace_result.accessed_slots.len(), 1);
2727        assert_eq!(
2728            trace_result.accessed_slots,
2729            HashMap::from([(
2730                Bytes::from("0x0000000000000000000000000000000000aaaa"),
2731                HashSet::from([Bytes::from("0x0000000000000000000000000000000000aaaa")])
2732            )])
2733        );
2734    }
2735
2736    #[tokio::test]
2737    async fn test_parse_retry_value_numeric() {
2738        let result = parse_retry_value("60");
2739        assert!(result.is_some());
2740
2741        let expected_time = SystemTime::now() + Duration::from_secs(60);
2742        let actual_time = result.unwrap();
2743
2744        // Allow for small timing differences during test execution
2745        let diff = if actual_time > expected_time {
2746            actual_time
2747                .duration_since(expected_time)
2748                .unwrap()
2749        } else {
2750            expected_time
2751                .duration_since(actual_time)
2752                .unwrap()
2753        };
2754        assert!(diff < Duration::from_secs(1), "Time difference too large: {:?}", diff);
2755    }
2756
2757    #[tokio::test]
2758    async fn test_parse_retry_value_rfc2822() {
2759        // Use a fixed future date in RFC2822 format
2760        let rfc2822_date = "Sat, 01 Jan 2030 12:00:00 +0000";
2761        let result = parse_retry_value(rfc2822_date);
2762        assert!(result.is_some());
2763
2764        let parsed_time = result.unwrap();
2765        assert!(parsed_time > SystemTime::now());
2766    }
2767
2768    #[tokio::test]
2769    async fn test_parse_retry_value_invalid_formats() {
2770        // Test various invalid formats
2771        assert!(parse_retry_value("invalid").is_none());
2772        assert!(parse_retry_value("").is_none());
2773        assert!(parse_retry_value("not_a_number").is_none());
2774        assert!(parse_retry_value("Mon, 32 Jan 2030 25:00:00 +0000").is_none());
2775        // Invalid date
2776    }
2777
2778    #[tokio::test]
2779    async fn test_parse_retry_value_zero_seconds() {
2780        let result = parse_retry_value("0");
2781        assert!(result.is_some());
2782
2783        let expected_time = SystemTime::now();
2784        let actual_time = result.unwrap();
2785
2786        // Should be very close to current time
2787        let diff = if actual_time > expected_time {
2788            actual_time
2789                .duration_since(expected_time)
2790                .unwrap()
2791        } else {
2792            expected_time
2793                .duration_since(actual_time)
2794                .unwrap()
2795        };
2796        assert!(diff < Duration::from_secs(1));
2797    }
2798
2799    #[tokio::test]
2800    async fn test_error_for_response_rate_limited() {
2801        let mut server = Server::new_async().await;
2802        let mock = server
2803            .mock("GET", "/test")
2804            .with_status(429)
2805            .with_header("Retry-After", "60")
2806            .create_async()
2807            .await;
2808
2809        let client = reqwest::Client::new();
2810        let response = client
2811            .get(format!("{}/test", server.url()))
2812            .send()
2813            .await
2814            .unwrap();
2815
2816        let http_client =
2817            HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
2818                .unwrap()
2819                .with_test_backoff_policy();
2820        let result = http_client
2821            .error_for_response(response)
2822            .await;
2823
2824        mock.assert();
2825        assert!(matches!(result, Err(RPCError::RateLimited(_))));
2826        if let Err(RPCError::RateLimited(retry_after)) = result {
2827            assert!(retry_after.is_some());
2828        }
2829    }
2830
2831    #[tokio::test]
2832    async fn test_error_for_response_rate_limited_no_header() {
2833        let mut server = Server::new_async().await;
2834        let mock = server
2835            .mock("GET", "/test")
2836            .with_status(429)
2837            .create_async()
2838            .await;
2839
2840        let client = reqwest::Client::new();
2841        let response = client
2842            .get(format!("{}/test", server.url()))
2843            .send()
2844            .await
2845            .unwrap();
2846
2847        let http_client =
2848            HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
2849                .unwrap()
2850                .with_test_backoff_policy();
2851        let result = http_client
2852            .error_for_response(response)
2853            .await;
2854
2855        mock.assert();
2856        assert!(matches!(result, Err(RPCError::RateLimited(None))));
2857    }
2858
2859    #[tokio::test]
2860    async fn test_error_for_response_server_errors() {
2861        let test_cases =
2862            vec![(502, "Bad Gateway"), (503, "Service Unavailable"), (504, "Gateway Timeout")];
2863
2864        for (status_code, expected_body) in test_cases {
2865            let mut server = Server::new_async().await;
2866            let mock = server
2867                .mock("GET", "/test")
2868                .with_status(status_code)
2869                .with_body(expected_body)
2870                .create_async()
2871                .await;
2872
2873            let client = reqwest::Client::new();
2874            let response = client
2875                .get(format!("{}/test", server.url()))
2876                .send()
2877                .await
2878                .unwrap();
2879
2880            let http_client =
2881                HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
2882                    .unwrap()
2883                    .with_test_backoff_policy();
2884            let result = http_client
2885                .error_for_response(response)
2886                .await;
2887
2888            mock.assert();
2889            assert!(matches!(result, Err(RPCError::ServerUnreachable(_))));
2890            if let Err(RPCError::ServerUnreachable(body)) = result {
2891                assert_eq!(body, expected_body);
2892            }
2893        }
2894    }
2895
2896    #[tokio::test]
2897    async fn test_error_for_response_success() {
2898        let mut server = Server::new_async().await;
2899        let mock = server
2900            .mock("GET", "/test")
2901            .with_status(200)
2902            .with_body("success")
2903            .create_async()
2904            .await;
2905
2906        let client = reqwest::Client::new();
2907        let response = client
2908            .get(format!("{}/test", server.url()))
2909            .send()
2910            .await
2911            .unwrap();
2912
2913        let http_client =
2914            HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
2915                .unwrap()
2916                .with_test_backoff_policy();
2917        let result = http_client
2918            .error_for_response(response)
2919            .await;
2920
2921        mock.assert();
2922        assert!(result.is_ok());
2923
2924        let response = result.unwrap();
2925        assert_eq!(response.status(), 200);
2926    }
2927
2928    #[tokio::test]
2929    async fn test_handle_error_for_backoff_server_unreachable() {
2930        let http_client =
2931            HttpRPCClient::new("http://localhost:8080", HttpRPCClientOptions::default())
2932                .unwrap()
2933                .with_test_backoff_policy();
2934        let error = RPCError::ServerUnreachable("Service down".to_string());
2935
2936        let backoff_error = http_client
2937            .handle_error_for_backoff(error)
2938            .await;
2939
2940        match backoff_error {
2941            backoff::Error::Transient { err: RPCError::ServerUnreachable(msg), retry_after } => {
2942                assert_eq!(msg, "Service down");
2943                assert_eq!(retry_after, Some(Duration::from_millis(50))); // Fast test duration
2944            }
2945            _ => panic!("Expected transient error for ServerUnreachable"),
2946        }
2947    }
2948
2949    #[tokio::test]
2950    async fn test_handle_error_for_backoff_rate_limited_with_retry_after() {
2951        let http_client =
2952            HttpRPCClient::new("http://localhost:8080", HttpRPCClientOptions::default())
2953                .unwrap()
2954                .with_test_backoff_policy();
2955        let future_time = SystemTime::now() + Duration::from_secs(30);
2956        let error = RPCError::RateLimited(Some(future_time));
2957
2958        let backoff_error = http_client
2959            .handle_error_for_backoff(error)
2960            .await;
2961
2962        match backoff_error {
2963            backoff::Error::Transient { err: RPCError::RateLimited(retry_after), .. } => {
2964                assert_eq!(retry_after, Some(future_time));
2965            }
2966            _ => panic!("Expected transient error for RateLimited"),
2967        }
2968
2969        // Verify that retry_after was stored in the client state
2970        let stored_retry_after = http_client.retry_after.read().await;
2971        assert_eq!(*stored_retry_after, Some(future_time));
2972    }
2973
2974    #[tokio::test]
2975    async fn test_handle_error_for_backoff_rate_limited_no_retry_after() {
2976        let http_client =
2977            HttpRPCClient::new("http://localhost:8080", HttpRPCClientOptions::default())
2978                .unwrap()
2979                .with_test_backoff_policy();
2980        let error = RPCError::RateLimited(None);
2981
2982        let backoff_error = http_client
2983            .handle_error_for_backoff(error)
2984            .await;
2985
2986        match backoff_error {
2987            backoff::Error::Transient { err: RPCError::RateLimited(None), .. } => {
2988                // This is expected - no retry-after still allows retries with default policy
2989            }
2990            _ => panic!("Expected transient error for RateLimited without retry-after"),
2991        }
2992    }
2993
2994    #[tokio::test]
2995    async fn test_handle_error_for_backoff_other_errors() {
2996        let http_client =
2997            HttpRPCClient::new("http://localhost:8080", HttpRPCClientOptions::default())
2998                .unwrap()
2999                .with_test_backoff_policy();
3000        let error = RPCError::ParseResponse("Invalid JSON".to_string());
3001
3002        let backoff_error = http_client
3003            .handle_error_for_backoff(error)
3004            .await;
3005
3006        match backoff_error {
3007            backoff::Error::Permanent(RPCError::ParseResponse(msg)) => {
3008                assert_eq!(msg, "Invalid JSON");
3009            }
3010            _ => panic!("Expected permanent error for ParseResponse"),
3011        }
3012    }
3013
3014    #[tokio::test]
3015    async fn test_wait_until_retry_after_no_retry_time() {
3016        let http_client =
3017            HttpRPCClient::new("http://localhost:8080", HttpRPCClientOptions::default())
3018                .unwrap()
3019                .with_test_backoff_policy();
3020
3021        let start = std::time::Instant::now();
3022        http_client
3023            .wait_until_retry_after()
3024            .await;
3025        let elapsed = start.elapsed();
3026
3027        // Should return immediately if no retry time is set
3028        assert!(elapsed < Duration::from_millis(100));
3029    }
3030
3031    #[tokio::test]
3032    async fn test_wait_until_retry_after_past_time() {
3033        let http_client =
3034            HttpRPCClient::new("http://localhost:8080", HttpRPCClientOptions::default())
3035                .unwrap()
3036                .with_test_backoff_policy();
3037
3038        // Set a retry time in the past
3039        let past_time = SystemTime::now() - Duration::from_secs(10);
3040        *http_client.retry_after.write().await = Some(past_time);
3041
3042        let start = std::time::Instant::now();
3043        http_client
3044            .wait_until_retry_after()
3045            .await;
3046        let elapsed = start.elapsed();
3047
3048        // Should return immediately if retry time is in the past
3049        assert!(elapsed < Duration::from_millis(100));
3050    }
3051
3052    #[tokio::test]
3053    async fn test_wait_until_retry_after_future_time() {
3054        let http_client =
3055            HttpRPCClient::new("http://localhost:8080", HttpRPCClientOptions::default())
3056                .unwrap()
3057                .with_test_backoff_policy();
3058
3059        // Set a retry time 100ms in the future
3060        let future_time = SystemTime::now() + Duration::from_millis(100);
3061        *http_client.retry_after.write().await = Some(future_time);
3062
3063        let start = std::time::Instant::now();
3064        http_client
3065            .wait_until_retry_after()
3066            .await;
3067        let elapsed = start.elapsed();
3068
3069        // Should wait approximately the specified duration
3070        assert!(elapsed >= Duration::from_millis(80)); // Allow some tolerance
3071        assert!(elapsed <= Duration::from_millis(200)); // Upper bound for test stability
3072    }
3073
3074    #[tokio::test]
3075    async fn test_make_post_request_success() {
3076        let mut server = Server::new_async().await;
3077        let server_resp = r#"{"success": true}"#;
3078
3079        let mock = server
3080            .mock("POST", "/test")
3081            .with_status(200)
3082            .with_body(server_resp)
3083            .create_async()
3084            .await;
3085
3086        let http_client =
3087            HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
3088                .unwrap()
3089                .with_test_backoff_policy();
3090        let request_body = serde_json::json!({"test": "data"});
3091        let uri = format!("{}/test", server.url());
3092
3093        let result = http_client
3094            .make_post_request(&request_body, &uri)
3095            .await;
3096
3097        mock.assert();
3098        assert!(result.is_ok());
3099
3100        let response = result.unwrap();
3101        assert_eq!(response.status(), 200);
3102        assert_eq!(response.text().await.unwrap(), server_resp);
3103    }
3104
3105    #[tokio::test]
3106    async fn test_make_post_request_retry_on_server_error() {
3107        let mut server = Server::new_async().await;
3108        // First request fails with 503, second succeeds
3109        let error_mock = server
3110            .mock("POST", "/test")
3111            .with_status(503)
3112            .with_body("Service Unavailable")
3113            .expect(1)
3114            .create_async()
3115            .await;
3116
3117        let success_mock = server
3118            .mock("POST", "/test")
3119            .with_status(200)
3120            .with_body(r#"{"success": true}"#)
3121            .expect(1)
3122            .create_async()
3123            .await;
3124
3125        let http_client =
3126            HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
3127                .unwrap()
3128                .with_test_backoff_policy();
3129        let request_body = serde_json::json!({"test": "data"});
3130        let uri = format!("{}/test", server.url());
3131
3132        let result = http_client
3133            .make_post_request(&request_body, &uri)
3134            .await;
3135
3136        error_mock.assert();
3137        success_mock.assert();
3138        assert!(result.is_ok());
3139    }
3140
3141    #[tokio::test]
3142    async fn test_make_post_request_respect_retry_after_header() {
3143        let mut server = Server::new_async().await;
3144
3145        // First request returns 429 with retry-after, second succeeds
3146        let rate_limit_mock = server
3147            .mock("POST", "/test")
3148            .with_status(429)
3149            .with_header("Retry-After", "1") // 1 second
3150            .expect(1)
3151            .create_async()
3152            .await;
3153
3154        let success_mock = server
3155            .mock("POST", "/test")
3156            .with_status(200)
3157            .with_body(r#"{"success": true}"#)
3158            .expect(1)
3159            .create_async()
3160            .await;
3161
3162        let http_client =
3163            HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
3164                .unwrap()
3165                .with_test_backoff_policy();
3166        let request_body = serde_json::json!({"test": "data"});
3167        let uri = format!("{}/test", server.url());
3168
3169        let start = std::time::Instant::now();
3170        let result = http_client
3171            .make_post_request(&request_body, &uri)
3172            .await;
3173        let elapsed = start.elapsed();
3174
3175        rate_limit_mock.assert();
3176        success_mock.assert();
3177        assert!(result.is_ok());
3178
3179        // Should have waited at least 1 second due to retry-after header
3180        assert!(elapsed >= Duration::from_millis(900)); // Allow some tolerance
3181        assert!(elapsed <= Duration::from_millis(2000)); // Upper bound for test stability
3182    }
3183
3184    #[tokio::test]
3185    async fn test_make_post_request_permanent_error() {
3186        let mut server = Server::new_async().await;
3187
3188        let mock = server
3189            .mock("POST", "/test")
3190            .with_status(400) // Bad Request - should not be retried
3191            .with_body("Bad Request")
3192            .expect(1)
3193            .create_async()
3194            .await;
3195
3196        let http_client =
3197            HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
3198                .unwrap()
3199                .with_test_backoff_policy();
3200        let request_body = serde_json::json!({"test": "data"});
3201        let uri = format!("{}/test", server.url());
3202
3203        let result = http_client
3204            .make_post_request(&request_body, &uri)
3205            .await;
3206
3207        mock.assert();
3208        assert!(result.is_ok()); // 400 doesn't trigger retry logic, just returns the response
3209
3210        let response = result.unwrap();
3211        assert_eq!(response.status(), 400);
3212    }
3213
3214    #[tokio::test]
3215    async fn test_concurrent_requests_with_different_retry_after() {
3216        let mut server = Server::new_async().await;
3217
3218        // First request gets rate limited with 1 second retry-after
3219        let rate_limit_mock_1 = server
3220            .mock("POST", "/test1")
3221            .with_status(429)
3222            .with_header("Retry-After", "1")
3223            .expect(1)
3224            .create_async()
3225            .await;
3226
3227        // Second request gets rate limited with 2 second retry-after
3228        let rate_limit_mock_2 = server
3229            .mock("POST", "/test2")
3230            .with_status(429)
3231            .with_header("Retry-After", "2")
3232            .expect(1)
3233            .create_async()
3234            .await;
3235
3236        // Success mocks for retries
3237        let success_mock_1 = server
3238            .mock("POST", "/test1")
3239            .with_status(200)
3240            .with_body(r#"{"result": "success1"}"#)
3241            .expect(1)
3242            .create_async()
3243            .await;
3244
3245        let success_mock_2 = server
3246            .mock("POST", "/test2")
3247            .with_status(200)
3248            .with_body(r#"{"result": "success2"}"#)
3249            .expect(1)
3250            .create_async()
3251            .await;
3252
3253        let http_client =
3254            HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
3255                .unwrap()
3256                .with_test_backoff_policy();
3257        let request_body = serde_json::json!({"test": "data"});
3258
3259        let uri1 = format!("{}/test1", server.url());
3260        let uri2 = format!("{}/test2", server.url());
3261
3262        // Start both requests concurrently
3263        let start = std::time::Instant::now();
3264        let (result1, result2) = tokio::join!(
3265            http_client.make_post_request(&request_body, &uri1),
3266            http_client.make_post_request(&request_body, &uri2)
3267        );
3268        let elapsed = start.elapsed();
3269
3270        rate_limit_mock_1.assert();
3271        rate_limit_mock_2.assert();
3272        success_mock_1.assert();
3273        success_mock_2.assert();
3274
3275        assert!(result1.is_ok());
3276        assert!(result2.is_ok());
3277
3278        // Both requests should succeed, but the second should take longer due to the 2s retry-after
3279        // The total time should be at least 2 seconds since the shared retry_after state
3280        // gets updated by both requests
3281        assert!(elapsed >= Duration::from_millis(1800)); // Allow some tolerance
3282        assert!(elapsed <= Duration::from_millis(3000)); // Upper bound
3283
3284        // Check the final retry_after state - should be the latest (higher) value
3285        let final_retry_after = http_client.retry_after.read().await;
3286        assert!(final_retry_after.is_some());
3287
3288        // The retry_after should be set to the latest (higher) value from the two requests
3289        if let Some(retry_time) = *final_retry_after {
3290            // The retry_after time might be in the past now since we waited,
3291            // but it should be reasonable (not too far in past/future)
3292            let now = SystemTime::now();
3293            let diff = if retry_time > now {
3294                retry_time.duration_since(now).unwrap()
3295            } else {
3296                now.duration_since(retry_time).unwrap()
3297            };
3298
3299            // Should be within a reasonable range (the 2s retry-after plus some buffer)
3300            assert!(diff <= Duration::from_secs(3), "Retry time difference too large: {:?}", diff);
3301        }
3302    }
3303
3304    #[tokio::test]
3305    async fn test_get_snapshots() {
3306        let mut server = Server::new_async().await;
3307
3308        // Mock protocol states response
3309        let protocol_states_resp = r#"
3310        {
3311            "states": [
3312                {
3313                    "component_id": "component1",
3314                    "attributes": {
3315                        "attribute_1": "0x00000000000003e8"
3316                    },
3317                    "balances": {
3318                        "0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2": "0x01f4"
3319                    }
3320                }
3321            ],
3322            "pagination": {
3323                "page": 0,
3324                "page_size": 100,
3325                "total": 1
3326            }
3327        }
3328        "#;
3329
3330        // Mock contract state response
3331        let contract_state_resp = r#"
3332        {
3333            "accounts": [
3334                {
3335                    "chain": "ethereum",
3336                    "address": "0x1111111111111111111111111111111111111111",
3337                    "title": "",
3338                    "slots": {},
3339                    "native_balance": "0x01f4",
3340                    "token_balances": {},
3341                    "code": "0x00",
3342                    "code_hash": "0x5c06b7c5b3d910fd33bc2229846f9ddaf91d584d9b196e16636901ac3a77077e",
3343                    "balance_modify_tx": "0x0000000000000000000000000000000000000000000000000000000000000000",
3344                    "code_modify_tx": "0x0000000000000000000000000000000000000000000000000000000000000000",
3345                    "creation_tx": null
3346                }
3347            ],
3348            "pagination": {
3349                "page": 0,
3350                "page_size": 100,
3351                "total": 1
3352            }
3353        }
3354        "#;
3355
3356        // Mock component TVL response
3357        let tvl_resp = r#"
3358        {
3359            "tvl": {
3360                "component1": 1000000.0
3361            },
3362            "pagination": {
3363                "page": 0,
3364                "page_size": 100,
3365                "total": 1
3366            }
3367        }
3368        "#;
3369
3370        let protocol_states_mock = server
3371            .mock("POST", "/v1/protocol_state")
3372            .expect(1)
3373            .with_body(protocol_states_resp)
3374            .create_async()
3375            .await;
3376
3377        let contract_state_mock = server
3378            .mock("POST", "/v1/contract_state")
3379            .expect(1)
3380            .with_body(contract_state_resp)
3381            .create_async()
3382            .await;
3383
3384        let tvl_mock = server
3385            .mock("POST", "/v1/component_tvl")
3386            .expect(1)
3387            .with_body(tvl_resp)
3388            .create_async()
3389            .await;
3390
3391        let client = HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
3392            .expect("create client");
3393
3394        let component = tycho_common::models::protocol::ProtocolComponent {
3395            id: "component1".to_string(),
3396            protocol_system: "test_protocol".to_string(),
3397            protocol_type_name: "test_type".to_string(),
3398            chain: Chain::Ethereum,
3399            tokens: vec![],
3400            contract_addresses: vec![
3401                Bytes::from_str("0x1111111111111111111111111111111111111111").unwrap()
3402            ],
3403            static_attributes: HashMap::new(),
3404            change: tycho_common::models::ChangeType::Creation,
3405            creation_tx: Bytes::from_str(
3406                "0x0000000000000000000000000000000000000000000000000000000000000000",
3407            )
3408            .unwrap(),
3409            created_at: chrono::Utc::now().naive_utc(),
3410        };
3411
3412        let mut components = HashMap::new();
3413        components.insert("component1".to_string(), component);
3414
3415        let contract_ids =
3416            vec![Bytes::from_str("0x1111111111111111111111111111111111111111").unwrap()];
3417
3418        let request = SnapshotParameters::new(
3419            Chain::Ethereum,
3420            "test_protocol",
3421            &components,
3422            &contract_ids,
3423            12345,
3424        );
3425
3426        let response = client
3427            .get_snapshots(&request, None, RPC_CLIENT_CONCURRENCY)
3428            .await
3429            .expect("get snapshots");
3430
3431        // Verify all mocks were called
3432        protocol_states_mock.assert();
3433        contract_state_mock.assert();
3434        tvl_mock.assert();
3435
3436        // Assert states
3437        assert_eq!(response.states.len(), 1);
3438        assert!(response
3439            .states
3440            .contains_key("component1"));
3441
3442        // Check that the state has the expected TVL
3443        let component_state = response
3444            .states
3445            .get("component1")
3446            .unwrap();
3447        assert_eq!(component_state.component_tvl, Some(1000000.0));
3448
3449        // Assert VM storage
3450        assert_eq!(response.vm_storage.len(), 1);
3451        let contract_addr = Bytes::from_str("0x1111111111111111111111111111111111111111").unwrap();
3452        assert!(response
3453            .vm_storage
3454            .contains_key(&contract_addr));
3455    }
3456
3457    #[tokio::test]
3458    async fn test_get_snapshots_empty_components() {
3459        let server = Server::new_async().await;
3460        let client = HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
3461            .expect("create client");
3462
3463        let components = HashMap::new();
3464        let contract_ids = vec![];
3465
3466        let request = SnapshotParameters::new(
3467            Chain::Ethereum,
3468            "test_protocol",
3469            &components,
3470            &contract_ids,
3471            12345,
3472        );
3473
3474        let response = client
3475            .get_snapshots(&request, None, RPC_CLIENT_CONCURRENCY)
3476            .await
3477            .expect("get snapshots");
3478
3479        // Should return empty response without making any requests
3480        assert!(response.states.is_empty());
3481        assert!(response.vm_storage.is_empty());
3482    }
3483
3484    #[tokio::test]
3485    async fn test_get_snapshots_without_tvl() {
3486        let mut server = Server::new_async().await;
3487
3488        let protocol_states_resp = r#"
3489        {
3490            "states": [
3491                {
3492                    "component_id": "component1",
3493                    "attributes": {},
3494                    "balances": {}
3495                }
3496            ],
3497            "pagination": {
3498                "page": 0,
3499                "page_size": 100,
3500                "total": 1
3501            }
3502        }
3503        "#;
3504
3505        let protocol_states_mock = server
3506            .mock("POST", "/v1/protocol_state")
3507            .expect(1)
3508            .with_body(protocol_states_resp)
3509            .create_async()
3510            .await;
3511
3512        let client = HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
3513            .expect("create client");
3514
3515        // Create test component
3516        let component = tycho_common::models::protocol::ProtocolComponent {
3517            id: "component1".to_string(),
3518            protocol_system: "test_protocol".to_string(),
3519            protocol_type_name: "test_type".to_string(),
3520            chain: Chain::Ethereum,
3521            tokens: vec![],
3522            contract_addresses: vec![],
3523            static_attributes: HashMap::new(),
3524            change: tycho_common::models::ChangeType::Creation,
3525            creation_tx: Bytes::from_str(
3526                "0x0000000000000000000000000000000000000000000000000000000000000000",
3527            )
3528            .unwrap(),
3529            created_at: chrono::Utc::now().naive_utc(),
3530        };
3531
3532        let mut components = HashMap::new();
3533        components.insert("component1".to_string(), component);
3534        let contract_ids = vec![];
3535
3536        let request = SnapshotParameters::new(
3537            Chain::Ethereum,
3538            "test_protocol",
3539            &components,
3540            &contract_ids,
3541            12345,
3542        )
3543        .include_balances(false)
3544        .include_tvl(false);
3545
3546        let response = client
3547            .get_snapshots(&request, None, RPC_CLIENT_CONCURRENCY)
3548            .await
3549            .expect("get snapshots");
3550
3551        // Verify only necessary mocks were called
3552        protocol_states_mock.assert();
3553        // No contract_state_mock.assert() since contract_ids is empty
3554        // No tvl_mock.assert() since include_tvl is false
3555
3556        assert_eq!(response.states.len(), 1);
3557        // Check that TVL is None since we didn't request it
3558        let component_state = response
3559            .states
3560            .get("component1")
3561            .unwrap();
3562        assert_eq!(component_state.component_tvl, None);
3563    }
3564
3565    #[tokio::test]
3566    async fn test_compression_enabled() {
3567        let mut server = Server::new_async().await;
3568        let server_resp = GET_CONTRACT_STATE_RESP;
3569
3570        // Compress the response using zstd
3571        let compressed_body =
3572            zstd::encode_all(server_resp.as_bytes(), 0).expect("compression failed");
3573
3574        let mocked_server = server
3575            .mock("POST", "/v1/contract_state")
3576            .expect(1)
3577            .with_header("Content-Encoding", "zstd")
3578            .with_body(compressed_body)
3579            .create_async()
3580            .await;
3581
3582        // Create client with compression enabled
3583        let client = HttpRPCClient::new(
3584            server.url().as_str(),
3585            HttpRPCClientOptions::new().with_compression(true),
3586        )
3587        .expect("create client");
3588
3589        let response = client
3590            .get_contract_state(ContractStateParams::new(Chain::Ethereum, ""))
3591            .await
3592            .expect("get state");
3593        let accounts = response;
3594
3595        mocked_server.assert();
3596        assert_eq!(accounts.data().len(), 1);
3597        assert_eq!(accounts.data()[0].native_balance, Bytes::from(500u16.to_be_bytes()));
3598    }
3599
3600    #[tokio::test]
3601    async fn test_compression_disabled() {
3602        let mut server = Server::new_async().await;
3603        let server_resp = GET_CONTRACT_STATE_RESP;
3604
3605        // Verify client does NOT send Accept-Encoding: zstd when compression is disabled
3606        // Instead, server should receive request without compression headers
3607        let mocked_server = server
3608            .mock("POST", "/v1/contract_state")
3609            .expect(1)
3610            .match_header("Accept-Encoding", mockito::Matcher::Missing)
3611            .with_status(200)
3612            .with_body(server_resp)
3613            .create_async()
3614            .await;
3615
3616        // Create client with compression disabled
3617        let client = HttpRPCClient::new(
3618            server.url().as_str(),
3619            HttpRPCClientOptions::new().with_compression(false),
3620        )
3621        .expect("create client");
3622
3623        let response = client
3624            .get_contract_state(ContractStateParams::new(Chain::Ethereum, ""))
3625            .await
3626            .expect("get state");
3627        let accounts = response;
3628
3629        // Verify the mock was called (client sent request without Accept-Encoding header)
3630        mocked_server.assert();
3631        assert_eq!(accounts.data().len(), 1);
3632        assert_eq!(accounts.data()[0].native_balance, Bytes::from(500u16.to_be_bytes()));
3633    }
3634
3635    #[rstest]
3636    #[case::single_page(2, 1000)]
3637    #[case::multiple_pages_within_concurrency(10, 2)]
3638    #[case::exceeds_concurrency_limit(60, 2)]
3639    #[tokio::test]
3640    async fn test_get_all_tokens_pagination_and_concurrency(
3641        #[case] total_tokens: usize,
3642        #[case] page_size: usize,
3643    ) {
3644        use std::sync::atomic::{AtomicUsize, Ordering};
3645
3646        let allowed_concurrency = 10;
3647
3648        let concurrent_requests = Arc::new(AtomicUsize::new(0));
3649        let max_concurrent = Arc::new(AtomicUsize::new(0));
3650
3651        let mut server = Server::new_async().await;
3652
3653        let total_pages = (total_tokens as f64 / page_size as f64).ceil() as i64;
3654
3655        // Mock all required pages
3656        for page in 0..total_pages {
3657            let concurrent = concurrent_requests.clone();
3658            let max_conc = max_concurrent.clone();
3659
3660            let tokens_in_page = {
3661                let start_idx = (page as usize) * page_size;
3662                let end_idx = ((page as usize + 1) * page_size).min(total_tokens);
3663                (start_idx..end_idx)
3664                    .map(|i| {
3665                        format!(
3666                            r#"{{
3667                            "chain": "ethereum",
3668                            "address": "0x{i:040x}",
3669                            "symbol": "TOKEN_{i}",
3670                            "decimals": 18,
3671                            "tax": 0,
3672                            "gas": [30000],
3673                            "quality": 100
3674                        }}"#
3675                        )
3676                    })
3677                    .collect::<Vec<_>>()
3678            };
3679
3680            let tokens_json = tokens_in_page.join(",");
3681            let response = format!(
3682                r#"{{
3683                    "tokens": [{tokens_json}],
3684                    "pagination": {{
3685                        "page": {page},
3686                        "page_size": {page_size},
3687                        "total": {total_tokens}
3688                    }}
3689                }}"#,
3690            );
3691
3692            server
3693                .mock("POST", "/v1/tokens")
3694                .expect(1)
3695                .with_chunked_body(move |w| {
3696                    // Track concurrent requests
3697                    let current = concurrent.fetch_add(1, Ordering::SeqCst);
3698                    max_conc.fetch_max(current + 1, Ordering::SeqCst);
3699
3700                    // Simulate some work to increase likelihood of concurrent requests
3701                    std::thread::sleep(Duration::from_millis(10));
3702
3703                    concurrent.fetch_sub(1, Ordering::SeqCst);
3704
3705                    w.write_all(response.as_bytes())
3706                })
3707                .create_async()
3708                .await;
3709        }
3710
3711        let client = HttpRPCClient::new(server.url().as_str(), HttpRPCClientOptions::default())
3712            .expect("create client");
3713
3714        let tokens = client
3715            .get_all_tokens(
3716                AllTokensParams::new(Chain::Ethereum, allowed_concurrency)
3717                    .with_chunk_size(page_size),
3718            )
3719            .await
3720            .expect("get all tokens");
3721
3722        // Verify concurrency was respected
3723        let max = max_concurrent.load(Ordering::SeqCst);
3724        let expected_max_concurrency = (total_pages as usize)
3725            .saturating_sub(1)
3726            .min(allowed_concurrency);
3727        assert!(
3728            max <= allowed_concurrency,
3729            "Expected max concurrent requests <= {allowed_concurrency}, got {max}"
3730        );
3731
3732        // For cases with multiple pages, verify we actually used concurrency
3733        if total_pages > 1 && expected_max_concurrency > 1 {
3734            assert!(
3735                max > 0,
3736                "Expected some concurrent requests for multi-page response, got {max}"
3737            );
3738        }
3739
3740        // Verify we got all expected tokens
3741        assert_eq!(
3742            tokens.len(),
3743            total_tokens,
3744            "Expected {total_tokens} tokens, got {}",
3745            tokens.len()
3746        );
3747
3748        // Verify tokens are in the expected order
3749        for (i, token) in tokens.iter().enumerate() {
3750            assert_eq!(token.symbol, format!("TOKEN_{i}"), "Token at index {i} has wrong symbol");
3751        }
3752    }
3753}
tycho_client/rpc.rs

tycho_client/
rpc.rs
