hypersync_client/
config.rs

1use anyhow::Result;
2use schemars::JsonSchema;
3use serde::{Deserialize, Serialize};
4use url::Url;
5
6use crate::ColumnMapping;
7
8/// Configuration for the hypersync client.
9#[derive(Debug, Clone, Deserialize, Serialize)]
10pub struct ClientConfig {
11    /// HyperSync server URL.
12    #[serde(default)]
13    pub url: String,
14    /// HyperSync server api token.
15    #[serde(default)]
16    pub api_token: String,
17    /// Milliseconds to wait for a response before timing out.
18    #[serde(default = "ClientConfig::default_http_req_timeout_millis")]
19    pub http_req_timeout_millis: u64,
20    /// Number of retries to attempt before returning error.
21    #[serde(default = "ClientConfig::default_max_num_retries")]
22    pub max_num_retries: usize,
23    /// Milliseconds that would be used for retry backoff increasing.
24    #[serde(default = "ClientConfig::default_retry_backoff_ms")]
25    pub retry_backoff_ms: u64,
26    /// Initial wait time for request backoff.
27    #[serde(default = "ClientConfig::default_retry_base_ms")]
28    pub retry_base_ms: u64,
29    /// Ceiling time for request backoff.
30    #[serde(default = "ClientConfig::default_retry_ceiling_ms")]
31    pub retry_ceiling_ms: u64,
32    /// Query serialization format to use for HTTP requests.
33    #[serde(default)]
34    pub serialization_format: SerializationFormat,
35    /// Whether to proactively sleep when the rate limit is exhausted instead of
36    /// sending requests that will be rejected with 429.
37    ///
38    /// Enabled by default. Set to `false` to opt out and handle rate limits yourself.
39    #[serde(default = "ClientConfig::default_proactive_rate_limit_sleep")]
40    pub proactive_rate_limit_sleep: bool,
41}
42
43impl Default for ClientConfig {
44    fn default() -> Self {
45        Self {
46            url: String::default(),
47            api_token: String::default(),
48            http_req_timeout_millis: Self::default_http_req_timeout_millis(),
49            max_num_retries: Self::default_max_num_retries(),
50            retry_backoff_ms: Self::default_retry_backoff_ms(),
51            retry_base_ms: Self::default_retry_base_ms(),
52            retry_ceiling_ms: Self::default_retry_ceiling_ms(),
53            serialization_format: SerializationFormat::default(),
54            proactive_rate_limit_sleep: Self::default_proactive_rate_limit_sleep(),
55        }
56    }
57}
58
59impl ClientConfig {
60    /// Default HTTP request timeout in milliseconds
61    pub const fn default_http_req_timeout_millis() -> u64 {
62        30_000
63    }
64
65    /// Default maximum number of retries
66    pub const fn default_max_num_retries() -> usize {
67        12
68    }
69
70    /// Default retry backoff in milliseconds
71    pub const fn default_retry_backoff_ms() -> u64 {
72        500
73    }
74
75    /// Default retry base time in milliseconds
76    pub const fn default_retry_base_ms() -> u64 {
77        200
78    }
79
80    /// Default retry ceiling time in milliseconds
81    pub const fn default_retry_ceiling_ms() -> u64 {
82        5_000
83    }
84
85    /// Default proactive rate limit sleep setting
86    pub const fn default_proactive_rate_limit_sleep() -> bool {
87        true
88    }
89    /// Validates the config
90    pub fn validate(&self) -> Result<()> {
91        if self.url.is_empty() {
92            anyhow::bail!("url is required");
93        }
94
95        // validate that url is a valid url
96        if Url::parse(&self.url).is_err() {
97            anyhow::bail!("url is malformed");
98        }
99
100        if self.api_token.is_empty() {
101            anyhow::bail!("api_token is required - get one from https://envio.dev/app/api-tokens");
102        }
103        // validate that api token is a uuid
104        if uuid::Uuid::parse_str(self.api_token.as_str()).is_err() {
105            anyhow::bail!("api_token is malformed - make sure its a token from https://envio.dev/app/api-tokens");
106        }
107
108        if self.http_req_timeout_millis == 0 {
109            anyhow::bail!("http_req_timeout_millis must be greater than 0");
110        }
111
112        Ok(())
113    }
114}
115
116/// Determines query serialization format for HTTP requests.
117#[derive(Clone, Copy, Debug, Serialize, Deserialize)]
118pub enum SerializationFormat {
119    /// Use JSON serialization
120    Json,
121    /// Use Cap'n Proto binary serialization (default, with query caching enabled)
122    CapnProto {
123        /// Whether to use query caching
124        should_cache_queries: bool,
125    },
126}
127
128impl Default for SerializationFormat {
129    fn default() -> Self {
130        Self::CapnProto {
131            should_cache_queries: true,
132        }
133    }
134}
135
136/// Config for hypersync event streaming.
137#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
138pub struct StreamConfig {
139    /// Column mapping for stream function output.
140    /// It lets you map columns you want into the DataTypes you want.
141    pub column_mapping: Option<ColumnMapping>,
142    /// Event signature used to populate decode logs. Decode logs would be empty if set to None.
143    pub event_signature: Option<String>,
144    /// Determines formatting of binary columns numbers into utf8 hex.
145    #[serde(default)]
146    pub hex_output: HexOutput,
147    /// Initial batch size. Size would be adjusted based on response size during execution.
148    #[serde(default = "StreamConfig::default_batch_size")]
149    pub batch_size: u64,
150    /// Maximum batch size that could be used during dynamic adjustment.
151    #[serde(default = "StreamConfig::default_max_batch_size")]
152    pub max_batch_size: u64,
153    /// Minimum batch size that could be used during dynamic adjustment.
154    #[serde(default = "StreamConfig::default_min_batch_size")]
155    pub min_batch_size: u64,
156    /// Number of async threads that would be spawned to execute different block ranges of queries.
157    #[serde(default = "StreamConfig::default_concurrency")]
158    pub concurrency: usize,
159    /// Max number of blocks to fetch in a single request.
160    #[serde(default)]
161    pub max_num_blocks: Option<usize>,
162    /// Max number of transactions to fetch in a single request.
163    #[serde(default)]
164    pub max_num_transactions: Option<usize>,
165    /// Max number of logs to fetch in a single request.
166    #[serde(default)]
167    pub max_num_logs: Option<usize>,
168    /// Max number of traces to fetch in a single request.
169    #[serde(default)]
170    pub max_num_traces: Option<usize>,
171    /// Size of a response in bytes from which step size will be lowered
172    #[serde(default = "StreamConfig::default_response_bytes_ceiling")]
173    pub response_bytes_ceiling: u64,
174    /// Size of a response in bytes from which step size will be increased
175    #[serde(default = "StreamConfig::default_response_bytes_floor")]
176    pub response_bytes_floor: u64,
177    /// Stream data in reverse order
178    #[serde(default = "StreamConfig::default_reverse")]
179    pub reverse: bool,
180}
181
182/// Determines format of Binary column
183#[derive(Default, Clone, Copy, Debug, Serialize, Deserialize, PartialEq, JsonSchema)]
184pub enum HexOutput {
185    /// Binary column won't be formatted as hex
186    #[default]
187    NoEncode,
188    /// Binary column would be formatted as prefixed hex i.e. 0xdeadbeef
189    Prefixed,
190    /// Binary column would be formatted as non prefixed hex i.e. deadbeef
191    NonPrefixed,
192}
193
194impl Default for StreamConfig {
195    fn default() -> Self {
196        Self {
197            column_mapping: None,
198            event_signature: None,
199            hex_output: HexOutput::default(),
200            batch_size: Self::default_batch_size(),
201            max_batch_size: Self::default_max_batch_size(),
202            min_batch_size: Self::default_min_batch_size(),
203            concurrency: Self::default_concurrency(),
204            max_num_blocks: None,
205            max_num_transactions: None,
206            max_num_logs: None,
207            max_num_traces: None,
208            response_bytes_ceiling: Self::default_response_bytes_ceiling(),
209            response_bytes_floor: Self::default_response_bytes_floor(),
210            reverse: Self::default_reverse(),
211        }
212    }
213}
214
215impl StreamConfig {
216    /// Default concurrency for stream processing
217    pub const fn default_concurrency() -> usize {
218        10
219    }
220
221    /// Default initial batch size
222    pub const fn default_batch_size() -> u64 {
223        1000
224    }
225
226    /// Default maximum batch size
227    pub const fn default_max_batch_size() -> u64 {
228        200_000
229    }
230
231    /// Default minimum batch size
232    pub const fn default_min_batch_size() -> u64 {
233        200
234    }
235
236    /// Default response bytes ceiling for dynamic batch adjustment
237    pub const fn default_response_bytes_ceiling() -> u64 {
238        500_000
239    }
240
241    /// Default response bytes floor for dynamic batch adjustment
242    pub const fn default_response_bytes_floor() -> u64 {
243        250_000
244    }
245
246    /// Default reverse streaming setting
247    pub const fn default_reverse() -> bool {
248        false
249    }
250}
251
252#[cfg(test)]
253mod tests {
254    use super::*;
255
256    #[test]
257    fn test_validate() {
258        let valid_cfg = ClientConfig {
259            url: "https://hypersync.xyz".into(),
260            api_token: "00000000-0000-0000-0000-000000000000".to_string(),
261            ..Default::default()
262        };
263
264        assert!(valid_cfg.validate().is_ok(), "valid config");
265
266        let cfg = ClientConfig {
267            url: "https://hypersync.xyz".to_string(),
268            api_token: "not a uuid".to_string(),
269            ..Default::default()
270        };
271
272        assert!(cfg.validate().is_err(), "invalid uuid");
273
274        let cfg = ClientConfig {
275            url: "https://hypersync.xyz".to_string(),
276            ..Default::default()
277        };
278
279        assert!(cfg.validate().is_err(), "missing api token");
280
281        let cfg = ClientConfig {
282            api_token: "00000000-0000-0000-0000-000000000000".to_string(),
283            ..Default::default()
284        };
285
286        assert!(cfg.validate().is_err(), "missing url");
287        let cfg = ClientConfig {
288            http_req_timeout_millis: 0,
289            ..valid_cfg
290        };
291        assert!(
292            cfg.validate().is_err(),
293            "http_req_timeout_millis must be greater than 0"
294        );
295    }
296
297    #[test]
298    fn test_stream_config_defaults() {
299        let default_config = StreamConfig::default();
300
301        // Check that all defaults are applied correctly
302        assert_eq!(default_config.concurrency, 10);
303        assert_eq!(default_config.batch_size, 1000);
304        assert_eq!(default_config.max_batch_size, 200_000);
305        assert_eq!(default_config.min_batch_size, 200);
306        assert_eq!(default_config.response_bytes_ceiling, 500_000);
307        assert_eq!(default_config.response_bytes_floor, 250_000);
308        assert!(!default_config.reverse);
309        assert_eq!(default_config.hex_output, HexOutput::NoEncode);
310        assert!(default_config.column_mapping.is_none());
311        assert!(default_config.event_signature.is_none());
312        assert!(default_config.max_num_blocks.is_none());
313        assert!(default_config.max_num_transactions.is_none());
314        assert!(default_config.max_num_logs.is_none());
315        assert!(default_config.max_num_traces.is_none());
316    }
317
318    #[test]
319    fn test_stream_config_serde() {
320        // Test serialization of default config
321        let default_config = StreamConfig::default();
322        let json = serde_json::to_string(&default_config).unwrap();
323        let deserialized: StreamConfig = serde_json::from_str(&json).unwrap();
324
325        // Verify round-trip works
326        assert_eq!(deserialized.concurrency, default_config.concurrency);
327        assert_eq!(deserialized.batch_size, default_config.batch_size);
328        assert_eq!(deserialized.reverse, default_config.reverse);
329
330        // Test partial JSON (missing some fields should use defaults)
331        let partial_json = r#"{"reverse": true, "batch_size": 500}"#;
332        let partial_config: StreamConfig = serde_json::from_str(partial_json).unwrap();
333
334        assert!(partial_config.reverse);
335        assert_eq!(partial_config.batch_size, 500);
336        assert_eq!(partial_config.concurrency, 10); // should use default
337        assert_eq!(partial_config.max_batch_size, 200_000); // should use default
338    }
339}
hypersync_client/config.rs

hypersync_client/
config.rs
