hypersync_client_solana/
lib.rs

1pub mod arrow_reader;
2pub mod config;
3pub mod decode;
4pub mod from_arrow;
5pub mod rate_limit;
6pub mod simple_types;
7pub mod stream;
8pub mod types;
9
10use std::sync::Arc;
11use std::time::{Duration, Instant};
12
13use anyhow::{Context, Result};
14use tokio::sync::mpsc;
15
16use config::{ClientConfig, StreamConfig};
17use hypersync_solana_net_types::query::SolanaQuery;
18pub use rate_limit::{QueryResponseWithRateLimit, RateLimitInfo};
19use simple_types::SolanaResponse;
20use types::QueryResponse;
21
22/// Solana HyperSync client.
23///
24/// Thread-safe and cheap to clone (wraps an `Arc`).
25#[derive(Clone)]
26pub struct Client {
27    inner: Arc<ClientInner>,
28}
29
30struct ClientInner {
31    http: reqwest::Client,
32    base_url: String,
33    config: ClientConfig,
34    /// Last observed rate limit headers and when they were captured, for the
35    /// proactive sleep in [`Client::wait_for_rate_limit`].
36    rate_limit_state: std::sync::Mutex<Option<(RateLimitInfo, Instant)>>,
37}
38
39/// Outcome of a single query POST, before retry logic.
40enum PostError {
41    /// Server responded with 429 Too Many Requests.
42    RateLimited(RateLimitInfo),
43    /// Any other request or server error.
44    Other(anyhow::Error),
45}
46
47impl Client {
48    /// Create a new client with the given configuration.
49    pub fn new(config: ClientConfig) -> Result<Self> {
50        // hscs stands for hypersync client solana.
51        let user_agent = format!("hscs/{}", env!("CARGO_PKG_VERSION"));
52        Self::new_with_agent(config, user_agent)
53    }
54
55    /// Create a new client with the given configuration and a custom user agent.
56    ///
57    /// This mirrors the EVM and Fuel HyperSync clients and is intended for use by
58    /// language bindings (Node.js) and downstream tools that want to identify
59    /// themselves to the server.
60    pub fn new_with_agent(config: ClientConfig, user_agent: impl Into<String>) -> Result<Self> {
61        anyhow::ensure!(!config.url.is_empty(), "url must not be empty");
62
63        let mut builder = reqwest::Client::builder()
64            .timeout(config.http_req_timeout)
65            .user_agent(user_agent.into());
66
67        if let Some(ref token) = config.bearer_token {
68            use reqwest::header;
69            let mut headers = header::HeaderMap::new();
70            let val = header::HeaderValue::from_str(&format!("Bearer {}", token))
71                .context("invalid bearer token")?;
72            headers.insert(header::AUTHORIZATION, val);
73            builder = builder.default_headers(headers);
74        }
75
76        let http = builder.build().context("build HTTP client")?;
77        let base_url = config.url.trim_end_matches('/').to_owned();
78
79        Ok(Self {
80            inner: Arc::new(ClientInner {
81                http,
82                base_url,
83                config,
84                rate_limit_state: std::sync::Mutex::new(None),
85            }),
86        })
87    }
88
89    /// Get the current chain height (latest slot).
90    pub async fn get_height(&self) -> Result<u64> {
91        let url = format!("{}/height", self.inner.base_url);
92        let resp = self
93            .request_with_retry(|| self.inner.http.get(&url))
94            .await
95            .context("get height")?;
96
97        let text = resp.text().await.context("read height response")?;
98        text.trim()
99            .parse()
100            .with_context(|| format!("parse height '{}'", text.trim()))
101    }
102
103    /// Execute a single query and return Arrow data.
104    pub async fn get_arrow(&self, query: &SolanaQuery) -> Result<QueryResponse> {
105        Ok(self.get_arrow_with_rate_limit(query).await?.response)
106    }
107
108    /// Executes query with retries and returns the response in Arrow format
109    /// along with rate limit information from the server.
110    ///
111    /// This is useful for consumers that want to inspect rate limit headers and
112    /// implement their own rate limiting logic in external systems. Retry and
113    /// back-off behaviour is identical to [`get_arrow`](Self::get_arrow):
114    /// a 429 is slept out against `x-ratelimit-reset` and retried.
115    pub async fn get_arrow_with_rate_limit(
116        &self,
117        query: &SolanaQuery,
118    ) -> Result<QueryResponseWithRateLimit<QueryResponse>> {
119        let (resp_bytes, rate_limit) = self.post_with_retry(query).await.context("query arrow")?;
120
121        let response =
122            arrow_reader::decode_response(&resp_bytes).context("decode arrow response")?;
123        Ok(QueryResponseWithRateLimit {
124            response,
125            rate_limit,
126        })
127    }
128
129    /// Executes query with retries and returns typed Rust structs along with
130    /// rate limit information from the server.
131    ///
132    /// This is useful for consumers that want to inspect rate limit headers and
133    /// implement their own rate limiting logic in external systems. Retry and
134    /// back-off behaviour is identical to [`get`](Self::get).
135    pub async fn get_with_rate_limit(
136        &self,
137        query: &SolanaQuery,
138    ) -> Result<QueryResponseWithRateLimit<SolanaResponse>> {
139        let result = self.get_arrow_with_rate_limit(query).await?;
140        Ok(QueryResponseWithRateLimit {
141            response: decode_response_tables(result.response)?,
142            rate_limit: result.rate_limit,
143        })
144    }
145
146    /// Execute a query that may span many server responses, paginating automatically.
147    /// Returns a single merged response.
148    pub async fn collect_arrow(
149        self: &Arc<Self>,
150        query: SolanaQuery,
151        config: StreamConfig,
152    ) -> Result<QueryResponse> {
153        let mut rx = self.stream_arrow(query, config);
154        let mut acc: Option<QueryResponse> = None;
155
156        while let Some(result) = rx.recv().await {
157            let resp = result?;
158            acc = Some(match acc {
159                None => resp,
160                Some(mut a) => {
161                    a.next_slot = resp.next_slot;
162                    a.response_bytes += resp.response_bytes;
163                    // Keep the most recent page's guard; a page without one
164                    // (head not in memory) must not erase an earlier guard
165                    // that still covers merged rows.
166                    if resp.rollback_guard.is_some() {
167                        a.rollback_guard = resp.rollback_guard;
168                    }
169                    for (name, batch) in resp.data.tables {
170                        if batch.num_rows() == 0 {
171                            continue;
172                        }
173                        if let Some(existing) = a.data.tables.get(name) {
174                            if existing.num_rows() > 0 {
175                                let merged = arrow::compute::concat_batches(
176                                    &existing.schema(),
177                                    &[existing.clone(), batch],
178                                )
179                                .with_context(|| format!("concat {} batches", name))?;
180                                a.data.tables.insert(name, merged);
181                            } else {
182                                a.data.tables.insert(name, batch);
183                            }
184                        } else {
185                            a.data.tables.insert(name, batch);
186                        }
187                    }
188                    a
189                }
190            });
191        }
192
193        acc.ok_or_else(|| anyhow::anyhow!("no data returned"))
194    }
195
196    /// Stream query results with concurrent fetching and adaptive batch sizing.
197    ///
198    /// Returns an mpsc receiver that yields `QueryResponse` items in slot order.
199    pub fn stream_arrow(
200        self: &Arc<Self>,
201        query: SolanaQuery,
202        config: StreamConfig,
203    ) -> mpsc::Receiver<Result<QueryResponse>> {
204        stream::stream_arrow(self.clone(), query, config)
205    }
206
207    /// Execute a single query and return typed Rust structs.
208    ///
209    /// Like [`Client::get_arrow`], but decodes the Arrow tables into the
210    /// `Vec<T>` shapes in [`crate::simple_types`].
211    pub async fn get(&self, query: &SolanaQuery) -> Result<SolanaResponse> {
212        let arrow = self.get_arrow(query).await?;
213        decode_response_tables(arrow)
214    }
215
216    /// Execute a query that may span many server responses, paginating
217    /// automatically, and return typed Rust structs.
218    ///
219    /// This is the typed counterpart of [`Client::collect_arrow`].
220    pub async fn collect(
221        self: &Arc<Self>,
222        query: SolanaQuery,
223        config: StreamConfig,
224    ) -> Result<SolanaResponse> {
225        let arrow = self.collect_arrow(query, config).await?;
226        decode_response_tables(arrow)
227    }
228
229    /// Executes the query POST once. 429 responses become
230    /// [`PostError::RateLimited`] with the parsed rate limit headers; every
231    /// successful response's headers are returned so callers can track their
232    /// quota.
233    async fn post_once(
234        &self,
235        query: &SolanaQuery,
236    ) -> std::result::Result<(Vec<u8>, RateLimitInfo), PostError> {
237        let url = format!("{}/query/arrow", self.inner.base_url);
238        let resp = self
239            .inner
240            .http
241            .post(&url)
242            .json(query)
243            .send()
244            .await
245            .map_err(|e| PostError::Other(anyhow::Error::from(e).context("execute http req")))?;
246
247        let status = resp.status();
248        let rate_limit = RateLimitInfo::from_response(&resp);
249
250        if status == reqwest::StatusCode::TOO_MANY_REQUESTS {
251            return Err(PostError::RateLimited(rate_limit));
252        }
253        if !status.is_success() {
254            let body = resp.text().await.unwrap_or_default();
255            return Err(PostError::Other(anyhow::anyhow!(
256                "query returned {}: {}",
257                status,
258                body
259            )));
260        }
261        let bytes = resp
262            .bytes()
263            .await
264            .map_err(|e| PostError::Other(anyhow::Error::from(e).context("read response body")))?;
265        Ok((bytes.to_vec(), rate_limit))
266    }
267
268    /// Executes the query with retries.
269    ///
270    /// A 429 sleeps until the window resets and retries; other transient errors
271    /// use the generic exponential back-off. Once the retries are spent, the
272    /// caller receives the accumulated attempt errors.
273    async fn post_with_retry(&self, query: &SolanaQuery) -> Result<(Vec<u8>, RateLimitInfo)> {
274        let cfg = &self.inner.config;
275        let mut err = anyhow::anyhow!("");
276
277        // Proactive throttling: if we know we're rate limited, wait before sending.
278        if cfg.proactive_rate_limit_sleep {
279            self.wait_for_rate_limit().await;
280        }
281
282        for attempt in 0..=cfg.max_num_retries {
283            match self.post_once(query).await {
284                Ok((bytes, rate_limit)) => {
285                    self.update_rate_limit_state(&rate_limit);
286                    return Ok((bytes, rate_limit));
287                }
288                Err(PostError::RateLimited(rate_limit)) => {
289                    self.update_rate_limit_state(&rate_limit);
290                    err = err.context(format!(
291                        "rate limited by server ({rate_limit}). To increase your rate limits, upgrade your plan at https://envio.dev/app/api-tokens"
292                    ));
293                    if attempt == cfg.max_num_retries {
294                        return Err(err);
295                    }
296
297                    let wait_secs = rate_limit.suggested_wait_secs().unwrap_or(1) + 1;
298                    tracing::warn!(
299                        attempt,
300                        %rate_limit,
301                        wait_secs,
302                        "rate limited by server, waiting before retry. To increase your rate limits, upgrade your plan at https://envio.dev/app/api-tokens. For more info: https://docs.envio.dev/docs/HyperSync/api-tokens"
303                    );
304                    tokio::time::sleep(Duration::from_secs(wait_secs)).await;
305                    continue;
306                }
307                Err(PostError::Other(e)) => {
308                    tracing::warn!(attempt, error = ?e, "Query failed");
309                    err = err.context(format!("{e:?}"));
310                    if attempt == cfg.max_num_retries {
311                        return Err(err);
312                    }
313
314                    let delay_ms = cfg.retry_base_ms * 2u64.pow((attempt + 1).min(5));
315                    let delay_ms = delay_ms.min(cfg.retry_ceiling_ms);
316                    tokio::time::sleep(Duration::from_millis(delay_ms)).await;
317                }
318            }
319        }
320
321        Err(err)
322    }
323
324    /// Locks the rate limit state, recovering from a poisoned mutex.
325    ///
326    /// The state is advisory metadata, so a stale value is always preferable to
327    /// bricking every later call on a client whose lock holder happened to panic.
328    fn lock_rate_limit_state(&self) -> std::sync::MutexGuard<'_, Option<(RateLimitInfo, Instant)>> {
329        self.inner
330            .rate_limit_state
331            .lock()
332            .unwrap_or_else(|poisoned| poisoned.into_inner())
333    }
334
335    /// Returns the most recently observed rate limit information, if any.
336    ///
337    /// Updated after successful and 429 query responses that contain rate limit
338    /// headers.
339    pub fn rate_limit_info(&self) -> Option<RateLimitInfo> {
340        self.lock_rate_limit_state()
341            .as_ref()
342            .map(|(info, _captured_at)| info.clone())
343    }
344
345    /// Waits until the current rate limit window resets, if the client is rate limited.
346    ///
347    /// Returns immediately if:
348    /// - No rate limit information has been observed yet
349    /// - There is remaining quota in the current window
350    ///
351    /// This method is useful for consumers who want to explicitly wait before making
352    /// requests, for example when coordinating rate limits across multiple systems.
353    pub async fn wait_for_rate_limit(&self) {
354        let wait_info = {
355            let state = self.lock_rate_limit_state();
356            match state.as_ref() {
357                Some((info, captured_at)) if info.is_rate_limited() => {
358                    info.suggested_wait_secs().map(|secs| {
359                        let elapsed = captured_at.elapsed().as_secs();
360                        let remaining_wait = secs.saturating_sub(elapsed);
361                        (remaining_wait, info.clone())
362                    })
363                }
364                _ => None,
365            }
366        };
367        if let Some((secs, info)) = wait_info {
368            if secs > 0 {
369                tracing::warn!(
370                    rate_limit = %info,
371                    wait_secs = secs,
372                    "rate limit exhausted, proactively waiting for window reset. To increase your rate limits, upgrade your plan at https://envio.dev/app/api-tokens. For more info: https://docs.envio.dev/docs/HyperSync/api-tokens"
373                );
374                tokio::time::sleep(Duration::from_secs(secs)).await;
375            }
376        }
377    }
378
379    /// Updates the internally tracked rate limit state with the current timestamp.
380    fn update_rate_limit_state(&self, rate_limit: &RateLimitInfo) {
381        // Only update if the response actually contained rate limit headers
382        if rate_limit.limit.is_some()
383            || rate_limit.remaining.is_some()
384            || rate_limit.reset_secs.is_some()
385        {
386            *self.lock_rate_limit_state() = Some((rate_limit.clone(), Instant::now()));
387        }
388    }
389
390    async fn request_with_retry<F>(&self, make_request: F) -> Result<reqwest::Response>
391    where
392        F: Fn() -> reqwest::RequestBuilder,
393    {
394        let cfg = &self.inner.config;
395        let mut last_err = None;
396
397        for attempt in 0..=cfg.max_num_retries {
398            if attempt > 0 {
399                let delay_ms = cfg.retry_base_ms * 2u64.pow(attempt.min(5));
400                let delay_ms = delay_ms.min(cfg.retry_ceiling_ms);
401                tokio::time::sleep(Duration::from_millis(delay_ms)).await;
402            }
403            match make_request().send().await {
404                Ok(resp) => {
405                    if resp.status().is_success() {
406                        return Ok(resp);
407                    }
408                    let status = resp.status();
409                    let body = resp.text().await.unwrap_or_default();
410                    tracing::warn!(attempt, status = %status, "Request failed");
411                    last_err = Some(anyhow::anyhow!("HTTP {}: {}", status, body));
412                }
413                Err(e) => {
414                    tracing::warn!(attempt, error = ?e, "Request error");
415                    last_err = Some(e.into());
416                }
417            }
418        }
419
420        Err(last_err.unwrap_or_else(|| anyhow::anyhow!("request failed after retries")))
421    }
422}
423
424fn decode_response_tables(arrow: QueryResponse) -> Result<SolanaResponse> {
425    let mut resp = SolanaResponse {
426        next_slot: arrow.next_slot,
427        rollback_guard: arrow.rollback_guard,
428        response_bytes: arrow.response_bytes,
429        ..Default::default()
430    };
431    for (name, batch) in arrow.data.tables {
432        match name {
433            "blocks" => {
434                resp.blocks = from_arrow::blocks_from_arrow(&batch).context("decode blocks")?
435            }
436            "transactions" => {
437                resp.transactions =
438                    from_arrow::transactions_from_arrow(&batch).context("decode transactions")?
439            }
440            "instruction_calls" => {
441                resp.instruction_calls = from_arrow::instruction_calls_from_arrow(&batch)
442                    .context("decode instruction_calls")?
443            }
444            "logs" => resp.logs = from_arrow::logs_from_arrow(&batch).context("decode logs")?,
445            "account_activity" => {
446                resp.account_activity = from_arrow::account_activity_from_arrow(&batch)
447                    .context("decode account_activity")?
448            }
449            "rewards" => {
450                resp.rewards = from_arrow::rewards_from_arrow(&batch).context("decode rewards")?
451            }
452            other => {
453                tracing::debug!(table = other, "ignoring unknown table in response");
454            }
455        }
456    }
457    Ok(resp)
458}
hypersync_client_solana/lib.rs

hypersync_client_solana/
lib.rs
