jetstreamer_firehose/
epochs.rs

1use reqwest::Client;
2use ripget::{WindowedDownload, WindowedDownloadOptions, download_url_windowed};
3use rseek::Seekable;
4use serde::Deserialize;
5use std::{fmt, io, pin::Pin};
6use tokio::io::{AsyncRead, AsyncSeek, BufReader, ReadBuf, SeekFrom};
7
8use crate::archive;
9use crate::node_reader::Len;
10
11#[cfg(feature = "s3-backend")]
12use {
13    crate::archive::S3Location,
14    aws_sdk_s3::Client as S3Client,
15    std::{future::Future, sync::Arc},
16};
17
18/// Default base URL used to fetch compact epoch CAR archives hosted by Old Faithful.
19pub const BASE_URL: &str = "https://files.old-faithful.net";
20
21#[inline(always)]
22/// Returns the inclusive slot range covered by a Solana epoch.
23///
24/// The tuple contains the first slot and the final slot of the epoch.
25pub const fn epoch_to_slot_range(epoch: u64) -> (u64, u64) {
26    let first = epoch * 432000;
27    (first, first + 431999)
28}
29
30#[inline(always)]
31/// Converts a slot back into the epoch that contains it.
32pub const fn slot_to_epoch(slot: u64) -> u64 {
33    slot / 432000
34}
35
36trait EpochReader: AsyncRead + AsyncSeek + Send + Unpin {}
37impl<T> EpochReader for T where T: AsyncRead + AsyncSeek + Send + Unpin {}
38type DynEpochReader = dyn EpochReader;
39
40/// Seekable reader returned by [`fetch_epoch_stream`] regardless of backend.
41pub struct EpochStream {
42    inner: Pin<Box<DynEpochReader>>,
43    total_len: u64,
44}
45
46impl EpochStream {
47    fn new<R>(reader: R) -> Self
48    where
49        R: AsyncRead + AsyncSeek + Len + Send + Unpin + 'static,
50    {
51        let len = reader.len();
52        Self {
53            inner: Box::pin(reader),
54            total_len: len,
55        }
56    }
57}
58
59impl AsyncRead for EpochStream {
60    fn poll_read(
61        self: Pin<&mut Self>,
62        cx: &mut std::task::Context<'_>,
63        buf: &mut ReadBuf<'_>,
64    ) -> std::task::Poll<io::Result<()>> {
65        let this = unsafe { self.get_unchecked_mut() };
66        this.inner.as_mut().poll_read(cx, buf)
67    }
68}
69
70impl AsyncSeek for EpochStream {
71    fn start_seek(self: Pin<&mut Self>, position: SeekFrom) -> io::Result<()> {
72        let this = unsafe { self.get_unchecked_mut() };
73        this.inner.as_mut().start_seek(position)
74    }
75
76    fn poll_complete(
77        self: Pin<&mut Self>,
78        cx: &mut std::task::Context<'_>,
79    ) -> std::task::Poll<io::Result<u64>> {
80        let this = unsafe { self.get_unchecked_mut() };
81        this.inner.as_mut().poll_complete(cx)
82    }
83}
84
85impl Len for EpochStream {
86    fn len(&self) -> u64 {
87        self.total_len
88    }
89}
90
91impl Unpin for EpochStream {}
92
93/* ────────────────────────────────────────────────────────────────────────── */
94/*  Blanket Len impl so BufReader<T> keeps the .len() we rely on            */
95/* ────────────────────────────────────────────────────────────────────────── */
96impl<T: Len + AsyncRead> Len for BufReader<T> {
97    #[inline]
98    fn len(&self) -> u64 {
99        self.get_ref().len()
100    }
101}
102
103/// Controls how epoch CAR streams are opened for [`fetch_epoch_stream_with_options`].
104#[derive(Clone, Debug)]
105pub struct FetchEpochStreamOptions {
106    /// When `true`, stream bytes sequentially through ripget's windowed downloader.
107    pub sequential: bool,
108    /// Parallel range request count used by ripget when `sequential` is enabled.
109    pub ripget_threads: usize,
110    /// Total hot/cold window size in bytes for ripget windowed streaming.
111    pub buffer_window_bytes: u64,
112    /// Pre-built ripget HTTP client for connection reuse across epoch downloads.
113    pub ripget_client: Option<ripget::Client>,
114}
115
116impl FetchEpochStreamOptions {
117    /// Returns default options that preserve the legacy seekable behavior.
118    pub fn parallel_default() -> Self {
119        Self {
120            sequential: false,
121            ripget_threads: 1,
122            buffer_window_bytes: 2,
123            ripget_client: None,
124        }
125    }
126}
127
128struct RipgetEpochReader {
129    inner: WindowedDownload,
130    len: u64,
131    position: u64,
132}
133
134impl RipgetEpochReader {
135    async fn new(
136        url: impl AsRef<str>,
137        threads: usize,
138        buffer_window_bytes: u64,
139        ripget_client: Option<ripget::Client>,
140    ) -> Result<Self, ripget::RipgetError> {
141        let mut options = WindowedDownloadOptions::new(buffer_window_bytes.max(2))
142            .threads(std::cmp::max(1, threads))
143            .user_agent(format!(
144                "jetstreamer-firehose/{}",
145                env!("CARGO_PKG_VERSION")
146            ));
147        if let Some(c) = ripget_client {
148            options = options.client(c);
149        }
150        let inner = download_url_windowed(url.as_ref(), options).await?;
151        let len = inner.expected_len();
152        Ok(Self {
153            inner,
154            len,
155            position: 0,
156        })
157    }
158}
159
160impl AsyncRead for RipgetEpochReader {
161    fn poll_read(
162        self: Pin<&mut Self>,
163        cx: &mut std::task::Context<'_>,
164        buf: &mut ReadBuf<'_>,
165    ) -> std::task::Poll<io::Result<()>> {
166        let this = self.get_mut();
167        let before = buf.filled().len();
168        let result = Pin::new(&mut this.inner).poll_read(cx, buf);
169        if let std::task::Poll::Ready(Ok(())) = &result {
170            let after = buf.filled().len();
171            let delta = after.saturating_sub(before) as u64;
172            this.position = this.position.saturating_add(delta);
173        }
174        result
175    }
176}
177
178impl AsyncSeek for RipgetEpochReader {
179    fn start_seek(self: Pin<&mut Self>, position: SeekFrom) -> io::Result<()> {
180        if matches!(position, SeekFrom::Current(0)) {
181            return Ok(());
182        }
183        Err(io::Error::new(
184            io::ErrorKind::Unsupported,
185            "seek is not supported for ripget windowed streams",
186        ))
187    }
188
189    fn poll_complete(
190        self: Pin<&mut Self>,
191        _cx: &mut std::task::Context<'_>,
192    ) -> std::task::Poll<io::Result<u64>> {
193        let this = self.get_mut();
194        std::task::Poll::Ready(Ok(this.position))
195    }
196}
197
198impl Len for RipgetEpochReader {
199    fn len(&self) -> u64 {
200        self.len
201    }
202}
203
204/// Checks the configured archive backend to determine whether an epoch CAR exists.
205pub async fn epoch_exists(epoch: u64, client: &Client) -> bool {
206    epoch_exists_at(epoch, client, archive::car_location()).await
207}
208
209async fn epoch_exists_at(epoch: u64, client: &Client, location: &archive::Location) -> bool {
210    let path = format!("{epoch}/epoch-{epoch}.car");
211
212    if location.is_http() {
213        let url = location
214            .url()
215            .join(&path)
216            .unwrap_or_else(|err| panic!("invalid CAR URL for epoch {epoch}: {err}"));
217        let response = client.head(url).send().await;
218        return response
219            .map(|res| res.status().is_success())
220            .unwrap_or(false);
221    }
222
223    #[cfg(feature = "s3-backend")]
224    if let Some(cfg) = location.as_s3() {
225        let key = cfg.key_for(&path);
226        return cfg
227            .client
228            .head_object()
229            .bucket(cfg.bucket.as_ref())
230            .key(key)
231            .send()
232            .await
233            .is_ok();
234    }
235
236    panic!(
237        "unsupported archive backend for CAR location {}",
238        location.url()
239    );
240}
241
242/// Fetches an epoch’s CAR file from the configured archive backend as a buffered, seekable stream.
243pub async fn fetch_epoch_stream(epoch: u64, client: &Client) -> EpochStream {
244    fetch_epoch_stream_with_options(epoch, client, None).await
245}
246
247/// Fetches an epoch’s CAR file with explicit stream options.
248///
249/// In sequential mode, arbitrary seeking is not supported and seek requests other than
250/// `SeekFrom::Current(0)` return `io::ErrorKind::Unsupported`.
251pub async fn fetch_epoch_stream_with_options(
252    epoch: u64,
253    client: &Client,
254    options: Option<FetchEpochStreamOptions>,
255) -> EpochStream {
256    fetch_epoch_stream_at(epoch, client, options, archive::car_location()).await
257}
258
259async fn fetch_epoch_stream_at(
260    epoch: u64,
261    client: &Client,
262    options: Option<FetchEpochStreamOptions>,
263    location: &archive::Location,
264) -> EpochStream {
265    let options = options.unwrap_or_else(FetchEpochStreamOptions::parallel_default);
266    let path = format!("{epoch}/epoch-{epoch}.car");
267
268    if location.is_http() {
269        let url = location
270            .url()
271            .join(&path)
272            .unwrap_or_else(|err| panic!("invalid CAR URL for epoch {epoch}: {err}"));
273        let request_url = url.to_string();
274        if options.sequential {
275            match RipgetEpochReader::new(
276                request_url.clone(),
277                options.ripget_threads,
278                options.buffer_window_bytes,
279                options.ripget_client,
280            )
281            .await
282            {
283                Ok(reader) => {
284                    return EpochStream::new(BufReader::with_capacity(8 * 1024 * 1024, reader));
285                }
286                Err(err) => {
287                    log::warn!(
288                        target: crate::LOG_MODULE,
289                        "ripget windowed stream failed to initialize for epoch {} ({}), falling back to seekable stream",
290                        epoch,
291                        err
292                    );
293                }
294            }
295        }
296        let http_client = client.clone();
297        let seekable = Seekable::new(move || http_client.get(request_url.clone())).await;
298        let reader = BufReader::with_capacity(8 * 1024 * 1024, seekable);
299        return EpochStream::new(reader);
300    }
301
302    #[cfg(feature = "s3-backend")]
303    if let Some(cfg) = location.as_s3() {
304        let s3_reader = S3SeekableReader::new(cfg, path)
305            .await
306            .unwrap_or_else(|err| panic!("failed to open epoch {epoch} CAR via S3: {err}"));
307        let reader = BufReader::with_capacity(8 * 1024 * 1024, s3_reader);
308        return EpochStream::new(reader);
309    }
310
311    panic!(
312        "unsupported archive backend for CAR location {}",
313        location.url()
314    );
315}
316
317#[cfg(feature = "s3-backend")]
318type ReaderFuture =
319    Pin<Box<dyn Future<Output = io::Result<Pin<Box<dyn AsyncRead + Send>>>> + Send>>;
320
321#[cfg(feature = "s3-backend")]
322struct S3SeekableReader {
323    client: Arc<S3Client>,
324    bucket: Arc<str>,
325    key: String,
326    len: u64,
327    position: u64,
328    reader: Option<Pin<Box<dyn AsyncRead + Send>>>,
329    init_fetch: Option<ReaderFuture>,
330}
331
332#[cfg(feature = "s3-backend")]
333impl S3SeekableReader {
334    async fn new(location: Arc<S3Location>, path: String) -> io::Result<Self> {
335        let key = location.key_for(&path);
336        let head = location
337            .client
338            .head_object()
339            .bucket(location.bucket.as_ref())
340            .key(&key)
341            .send()
342            .await
343            .map_err(|err| io::Error::other(err.to_string()))?;
344        let len = head
345            .content_length()
346            .ok_or_else(|| io::Error::other("S3 object missing length"))? as u64;
347
348        let mut reader = Self {
349            client: Arc::clone(&location.client),
350            bucket: Arc::clone(&location.bucket),
351            key,
352            len,
353            position: 0,
354            reader: None,
355            init_fetch: None,
356        };
357        reader.schedule_fetch(0);
358        Ok(reader)
359    }
360
361    fn schedule_fetch(&mut self, start: u64) {
362        let client = Arc::clone(&self.client);
363        let bucket = Arc::clone(&self.bucket);
364        let key = self.key.clone();
365        let len = self.len;
366        self.reader = None;
367        self.init_fetch = Some(Box::pin(async move {
368            let end = len.saturating_sub(1);
369            let range = if start >= len {
370                format!("bytes={end}-{end}")
371            } else {
372                format!("bytes={start}-{end}")
373            };
374            let resp = client
375                .get_object()
376                .bucket(bucket.as_ref())
377                .key(&key)
378                .range(range)
379                .send()
380                .await
381                .map_err(|err| io::Error::other(err.to_string()))?;
382            let reader = tokio::io::BufReader::new(resp.body.into_async_read());
383            Ok(Box::pin(reader) as Pin<Box<dyn AsyncRead + Send>>)
384        }));
385    }
386}
387
388#[cfg(feature = "s3-backend")]
389impl AsyncRead for S3SeekableReader {
390    fn poll_read(
391        self: Pin<&mut Self>,
392        cx: &mut std::task::Context<'_>,
393        buf: &mut ReadBuf<'_>,
394    ) -> std::task::Poll<io::Result<()>> {
395        let this = unsafe { self.get_unchecked_mut() };
396
397        if this.position >= this.len {
398            return std::task::Poll::Ready(Ok(()));
399        }
400
401        loop {
402            if let Some(reader) = this.reader.as_mut() {
403                let before = buf.filled().len();
404                match reader.as_mut().poll_read(cx, buf) {
405                    std::task::Poll::Ready(Ok(())) => {
406                        let after = buf.filled().len();
407                        let delta = after.saturating_sub(before);
408                        this.position = this.position.saturating_add(delta as u64);
409                        return std::task::Poll::Ready(Ok(()));
410                    }
411                    other => return other,
412                }
413            }
414
415            if let Some(fut) = this.init_fetch.as_mut() {
416                match fut.as_mut().poll(cx) {
417                    std::task::Poll::Ready(Ok(reader)) => {
418                        this.reader = Some(reader);
419                        this.init_fetch = None;
420                        continue;
421                    }
422                    std::task::Poll::Ready(Err(err)) => {
423                        this.init_fetch = None;
424                        return std::task::Poll::Ready(Err(err));
425                    }
426                    std::task::Poll::Pending => return std::task::Poll::Pending,
427                }
428            } else {
429                this.schedule_fetch(this.position);
430            }
431        }
432    }
433}
434
435#[cfg(feature = "s3-backend")]
436impl AsyncSeek for S3SeekableReader {
437    fn start_seek(self: Pin<&mut Self>, position: SeekFrom) -> io::Result<()> {
438        let this = unsafe { self.get_unchecked_mut() };
439        let target = match position {
440            SeekFrom::Start(offset) => offset,
441            SeekFrom::Current(delta) => {
442                let tmp = this.position as i64 + delta;
443                if tmp < 0 {
444                    return Err(io::Error::new(io::ErrorKind::InvalidInput, "negative seek"));
445                }
446                tmp as u64
447            }
448            SeekFrom::End(delta) => {
449                let tmp = this.len as i64 + delta;
450                if tmp < 0 {
451                    return Err(io::Error::new(io::ErrorKind::InvalidInput, "negative seek"));
452                }
453                tmp as u64
454            }
455        };
456        this.position = target.min(this.len);
457        this.reader = None;
458        this.init_fetch = None;
459        this.schedule_fetch(this.position);
460        Ok(())
461    }
462
463    fn poll_complete(
464        self: Pin<&mut Self>,
465        _cx: &mut std::task::Context<'_>,
466    ) -> std::task::Poll<io::Result<u64>> {
467        let this = unsafe { self.get_unchecked_mut() };
468        std::task::Poll::Ready(Ok(this.position))
469    }
470}
471
472#[cfg(feature = "s3-backend")]
473impl Len for S3SeekableReader {
474    fn len(&self) -> u64 {
475        self.len
476    }
477}
478
479#[cfg(feature = "s3-backend")]
480impl Unpin for S3SeekableReader {}
481
482/// Errors that can occur when calling [`get_slot_timestamp`].
483#[derive(Debug)]
484pub enum SlotTimestampError {
485    /// Network request failed while contacting the RPC endpoint.
486    Transport(reqwest::Error),
487    /// JSON payload could not be decoded.
488    Decode(serde_json::Error),
489    /// RPC returned an error object instead of a result.
490    Rpc(Option<serde_json::Value>),
491    /// The RPC response did not include a block time.
492    NoBlockTime,
493}
494impl fmt::Display for SlotTimestampError {
495    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
496        match self {
497            SlotTimestampError::Transport(e) => write!(f, "RPC transport error: {e}"),
498            SlotTimestampError::Decode(e) => write!(f, "RPC decode error: {e}"),
499            SlotTimestampError::Rpc(e) => write!(f, "RPC error: {:?}", e),
500            SlotTimestampError::NoBlockTime => write!(f, "No blockTime found in getBlock result"),
501        }
502    }
503}
504impl std::error::Error for SlotTimestampError {}
505
506/// Get the true Unix timestamp (seconds since epoch, UTC) for a Solana slot.
507/// Uses the validator RPC getBlock method, returns Ok(timestamp) or Err(reason).
508pub async fn get_slot_timestamp(
509    slot: u64,
510    rpc_url: &str,
511    client: &Client,
512) -> Result<u64, SlotTimestampError> {
513    #[derive(Deserialize)]
514    struct BlockResult {
515        #[serde(rename = "blockTime")]
516        block_time: Option<u64>,
517    }
518    #[derive(Deserialize)]
519    struct RpcResponse {
520        result: Option<BlockResult>,
521        error: Option<serde_json::Value>,
522    }
523
524    let req = serde_json::json!({
525        "jsonrpc": "2.0",
526        "id": 1,
527        "method": "getBlock",
528        "params": [slot, { "maxSupportedTransactionVersion": 0 }],
529    });
530
531    let resp = client
532        .post(rpc_url)
533        .json(&req)
534        .send()
535        .await
536        .map_err(SlotTimestampError::Transport)?;
537
538    let text = resp.text().await.map_err(SlotTimestampError::Transport)?;
539    let resp_val: RpcResponse = serde_json::from_str(&text).map_err(SlotTimestampError::Decode)?;
540
541    if resp_val.error.is_some() {
542        return Err(SlotTimestampError::Rpc(resp_val.error));
543    }
544    resp_val
545        .result
546        .and_then(|r| r.block_time)
547        .ok_or(SlotTimestampError::NoBlockTime)
548}
549
550/* ── Tests ──────────────────────────────────────────────────────────────── */
551#[cfg(test)]
552mod tests;
553
554#[test]
555fn test_epoch_to_slot() {
556    assert_eq!(epoch_to_slot_range(0), (0, 431999));
557    assert_eq!(epoch_to_slot_range(770), (332640000, 333071999));
558}
559
560#[test]
561fn test_slot_to_epoch() {
562    assert_eq!(slot_to_epoch(0), 0);
563    assert_eq!(slot_to_epoch(431999), 0);
564    assert_eq!(slot_to_epoch(432000), 1);
565    assert_eq!(slot_to_epoch(332640000), 770);
566    assert_eq!(slot_to_epoch(333071999), 770);
567}
568
569#[test]
570fn test_epoch_to_slot_range() {
571    assert_eq!(epoch_to_slot_range(0), (0, 431999));
572    assert_eq!(epoch_to_slot_range(1), (432000, 863999));
573    assert_eq!(epoch_to_slot_range(2), (864000, 1295999));
574    assert_eq!(epoch_to_slot_range(3), (1296000, 1727999));
575}
jetstreamer_firehose/epochs.rs

jetstreamer_firehose/
epochs.rs
