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

tycho_client/
rpc.rs
