iota_data_ingestion_core/history/
reader.rs

1// Copyright (c) Mysten Labs, Inc.
2// Modifications Copyright (c) 2024 IOTA Stiftung
3// SPDX-License-Identifier: Apache-2.0
4
5use std::{num::NonZeroUsize, ops::Range, sync::Arc, time::Duration};
6
7use bytes::{Buf, Bytes, buf::Reader};
8use futures::{Stream, StreamExt, TryStreamExt};
9use iota_config::object_storage_config::ObjectStoreConfig;
10use iota_storage::{
11    compute_sha3_checksum_for_bytes, make_iterator,
12    object_store::{ObjectStoreGetExt, http::HttpDownloaderBuilder, util::get},
13};
14use iota_types::{
15    full_checkpoint_content::CheckpointData, messages_checkpoint::CheckpointSequenceNumber,
16};
17use object_store::path::Path;
18use tokio::sync::{
19    Mutex,
20    oneshot::{self, Sender},
21};
22use tracing::info;
23
24use crate::{
25    IngestionError,
26    errors::IngestionResult as Result,
27    history::{
28        CHECKPOINT_FILE_MAGIC,
29        epoch_boundaries::{EpochBoundaries, read_epoch_boundaries},
30        manifest::{FileMetadata, Manifest, read_manifest},
31    },
32};
33
34#[derive(Clone)]
35pub struct HistoricalReader {
36    concurrency: usize,
37    #[expect(dead_code)]
38    /// We store this to get dropped along with the
39    /// reader and hence terminate the manifest sync
40    /// process.
41    sender: Arc<Sender<()>>,
42    manifest: Arc<Mutex<Manifest>>,
43    remote_object_store: Arc<dyn ObjectStoreGetExt>,
44}
45
46#[derive(Debug, Clone)]
47pub struct HistoricalReaderConfig {
48    pub remote_store_config: ObjectStoreConfig,
49    pub download_concurrency: NonZeroUsize,
50}
51
52impl HistoricalReader {
53    pub fn new(config: HistoricalReaderConfig) -> Result<Self> {
54        let remote_object_store = if config.remote_store_config.no_sign_request {
55            config.remote_store_config.make_http()?
56        } else {
57            config.remote_store_config.make().map(Arc::new)?
58        };
59        let (sender, recv) = oneshot::channel();
60        let manifest = Arc::new(Mutex::new(Manifest::new(0)));
61        // Start a background tokio task to keep local manifest in sync with remote
62        Self::spawn_manifest_sync_task(remote_object_store.clone(), manifest.clone(), recv);
63        Ok(Self {
64            manifest,
65            sender: Arc::new(sender),
66            remote_object_store,
67            concurrency: config.download_concurrency.get(),
68        })
69    }
70
71    /// This function verifies the manifest and returns the file metadata
72    /// sorted by the starting sequence number.
73    ///
74    /// More specifically it verifies that the files in the remote store
75    /// cover the entire range of checkpoints from sequence number 0
76    /// until the latest available checkpoint with no missing checkpoint.
77    ///
78    /// # Errors
79    ///
80    /// Fails if the manifest lists no files, or if its files do not cover
81    /// every checkpoint from 0 without a gap or overlap.
82    pub fn verify_and_get_manifest_files(&self, manifest: Manifest) -> Result<Vec<FileMetadata>> {
83        let mut files = manifest.to_files();
84        if files.is_empty() {
85            return Err(IngestionError::HistoryRead(
86                "unexpected empty remote store of historical data".to_string(),
87            ));
88        }
89
90        files.sort_by_key(|f| f.checkpoint_seq_range.start);
91
92        let covers_every_checkpoint = files[0].checkpoint_seq_range.start == 0
93            && files
94                .windows(2)
95                .all(|w| w[1].checkpoint_seq_range.start == w[0].checkpoint_seq_range.end)
96            && files.last().map(|f| f.checkpoint_seq_range.end)
97                == Some(manifest.next_checkpoint_seq_num());
98        if !covers_every_checkpoint {
99            return Err(IngestionError::HistoryRead(
100                "manifest files do not cover every checkpoint from 0".to_string(),
101            ));
102        }
103
104        Ok(files)
105    }
106
107    /// This function downloads checkpoint data files and ensures their
108    /// computed checksum matches the one in manifest.
109    pub async fn verify_file_consistency(&self, files: Vec<FileMetadata>) -> Result<()> {
110        let remote_object_store = self.remote_object_store.clone();
111        futures::stream::iter(files.iter())
112            .map(|metadata| {
113                let remote_object_store = remote_object_store.clone();
114                async move {
115                    let checkpoint_data = get(&remote_object_store, &metadata.file_path()).await?;
116                    Ok::<(Bytes, &FileMetadata), IngestionError>((checkpoint_data, metadata))
117                }
118            })
119            .boxed()
120            .buffer_unordered(self.concurrency)
121            .try_for_each(|(checkpoint_data, metadata)| {
122                let checksum = compute_sha3_checksum_for_bytes(checkpoint_data).map_err(Into::into);
123                let result = checksum.and_then(|checksum| {
124                    if checksum == metadata.sha3_digest {
125                        return Ok(());
126                    };
127                    Err(IngestionError::HistoryRead(format!(
128                        "checksum doesn't match for file: {:?}",
129                        metadata.file_path()
130                    )))
131                });
132                futures::future::ready(result)
133            })
134            .await
135    }
136
137    /// Stream blobs of [`Bytes`] that include checkpoint data for the specified
138    /// range.
139    ///
140    /// This method retrieves files with batches of serialized checkpoint
141    /// data from the remote store, and streams the respective contents
142    /// as blobs.
143    ///
144    /// # Errors
145    ///
146    /// Returns an error if resolving the files that need to be fetched from the
147    /// remote store fails.
148    ///
149    /// Additionally the stream may fail if fetching the file from the remote
150    /// store fails.
151    ///
152    /// # Examples
153    ///
154    /// ```ignore
155    /// use futures::StreamExt;
156    ///
157    /// let range = 100..200;
158    /// let mut stream = historical_reader.stream_blobs_for_range(range.clone()).await?;
159    /// while let Some(Ok(blob)) = stream.next().await {
160    ///     // we can now iterate over the checkpoint data
161    ///     for data in make_blob_iterator_for_range(blob, range.clone())? {
162    ///         println!("Received checkpoint data: {data:?}");
163    ///     }
164    /// }
165    /// ```
166    pub async fn stream_blobs_for_range(
167        &self,
168        checkpoint_range: Range<CheckpointSequenceNumber>,
169    ) -> Result<impl Stream<Item = Result<Bytes>> + Send + use<'_>> {
170        let files = self.get_files_for_range(checkpoint_range).await?;
171        Ok(futures::stream::iter(files)
172            .map(move |metadata| async move {
173                let remote_object_store = Arc::clone(&self.remote_object_store);
174                let file_path = metadata.file_path();
175                Ok(get(&remote_object_store, &file_path).await?)
176            })
177            .buffered(self.concurrency))
178    }
179
180    /// Construct an [`Iterator`] over [`CheckpointData`] for the specified
181    /// range.
182    ///
183    /// This method eagerly consumes the stream of blobs returned from
184    /// [`Self::stream_blobs_for_range`] and holds the data in memory until
185    /// the iterator is consumed.
186    ///
187    /// For lazy processing of the blobs use directly
188    /// [`Self::stream_blobs_for_range`] along with
189    /// [`make_blob_iterator_for_range`].
190    pub async fn iter_for_range(
191        &self,
192        checkpoint_range: Range<CheckpointSequenceNumber>,
193    ) -> Result<impl Iterator<Item = CheckpointData>> {
194        let blobs = self
195            .stream_blobs_for_range(checkpoint_range.clone())
196            .await?
197            .try_collect::<Vec<_>>()
198            .await?;
199        let data_iterators = blobs
200            .into_iter()
201            .map(|blob| {
202                let range = checkpoint_range.clone();
203                make_blob_iterator_for_range(blob, range)
204            })
205            .collect::<Result<Vec<_>>>()?;
206        Ok(data_iterators.into_iter().flatten())
207    }
208
209    /// Iterate [`CheckpointData`] from the given remote file.
210    ///
211    /// This method retrieves the file with batches of serialized checkpoint
212    /// data from the remote store, decodes the raw data, and streams the
213    /// deserialized values.
214    ///
215    /// # Errors
216    ///
217    /// Returns an error in the following cases:
218    ///
219    /// * If fetching the file from the remote store fails.
220    /// * If the file is corrupted and fails to decode.
221    pub async fn iter_for_file(
222        &self,
223        file_path: Path,
224    ) -> Result<impl Iterator<Item = CheckpointData>> {
225        let raw_data_batch = get(&self.remote_object_store, &file_path).await?;
226        make_blob_iterator(raw_data_batch)
227    }
228
229    /// Return latest available checkpoint in archive.
230    pub async fn latest_available_checkpoint(&self) -> Result<CheckpointSequenceNumber> {
231        self.manifest
232            .lock()
233            .await
234            .next_checkpoint_seq_num()
235            .checked_sub(1)
236            .ok_or_else(|| {
237                IngestionError::HistoryRead("no checkpoint data in the remote store".into())
238            })
239    }
240
241    pub fn remote_store_identifier(&self) -> String {
242        self.remote_object_store.to_string()
243    }
244
245    /// Returns the last checkpoint of each epoch, indexed by epoch.
246    ///
247    /// Read from the epoch boundaries file maintained alongside the manifest.
248    /// Callers slice the boundaries by epoch range as needed.
249    ///
250    /// # Errors
251    ///
252    /// Fails if the epoch boundaries file cannot be read or if it fails to
253    /// decode.
254    pub async fn epoch_boundaries(&self) -> Result<EpochBoundaries> {
255        read_epoch_boundaries(self.remote_object_store.clone()).await
256    }
257
258    /// Syncs the Manifest from remote store.
259    pub async fn sync_manifest_once(&self) -> Result<()> {
260        Self::sync_manifest(self.remote_object_store.clone(), self.manifest.clone()).await?;
261        Ok(())
262    }
263
264    pub async fn get_manifest(&self) -> Manifest {
265        self.manifest.lock().await.clone()
266    }
267
268    /// Copies Manifest from remote store to the given Manifest.
269    async fn sync_manifest(
270        remote_store: Arc<dyn ObjectStoreGetExt>,
271        manifest: Arc<Mutex<Manifest>>,
272    ) -> Result<()> {
273        let new_manifest = read_manifest(remote_store.clone()).await?;
274        let mut locked = manifest.lock().await;
275        *locked = new_manifest;
276        Ok(())
277    }
278
279    /// Resolve the files to fetch for the specified range.
280    ///
281    /// The method retrieves the manifest from the remote store and
282    /// searches for the files that cover the given range of checkpoint
283    /// data.
284    ///
285    /// # Errors
286    ///
287    /// The method fails if the remote store has no data, or if the
288    /// manifest fails to verify.
289    async fn get_files_for_range(
290        &self,
291        checkpoint_range: Range<CheckpointSequenceNumber>,
292    ) -> Result<impl Iterator<Item = FileMetadata>> {
293        let manifest = self.get_manifest().await;
294
295        let latest_available_checkpoint = manifest
296            .next_checkpoint_seq_num()
297            .checked_sub(1)
298            .ok_or_else(|| {
299                IngestionError::HistoryRead("no checkpoint data in the remote store".into())
300            })?;
301
302        if checkpoint_range.start > latest_available_checkpoint {
303            return Err(IngestionError::HistoryRead(format!(
304                "latest available checkpoint is: {latest_available_checkpoint}",
305            )));
306        }
307
308        let files = self.verify_and_get_manifest_files(manifest)?;
309
310        let start_index = match files
311            .binary_search_by_key(&checkpoint_range.start, |s| s.checkpoint_seq_range.start)
312        {
313            Ok(index) => index,
314            Err(index) => index - 1,
315        };
316
317        let end_index = match files
318            .binary_search_by_key(&checkpoint_range.end, |s| s.checkpoint_seq_range.start)
319        {
320            Ok(index) => index,
321            Err(index) => index,
322        };
323
324        Ok(files
325            .into_iter()
326            .enumerate()
327            .filter_map(move |(index, metadata)| {
328                (index >= start_index && index < end_index).then_some(metadata)
329            }))
330    }
331
332    fn spawn_manifest_sync_task(
333        remote_store: Arc<dyn ObjectStoreGetExt>,
334        manifest: Arc<Mutex<Manifest>>,
335        mut recv: oneshot::Receiver<()>,
336    ) {
337        tokio::task::spawn(async move {
338            let mut interval = tokio::time::interval(Duration::from_secs(60));
339            loop {
340                tokio::select! {
341                    _ = interval.tick() => {
342                        Self::sync_manifest(remote_store.clone(), manifest.clone()).await?;
343                    }
344                    _ = &mut recv => break,
345                }
346            }
347            info!("terminating the manifest sync loop");
348            Ok::<(), IngestionError>(())
349        });
350    }
351}
352
353fn make_blob_iterator(blob: Bytes) -> Result<impl Iterator<Item = CheckpointData>> {
354    Ok(make_iterator::<CheckpointData, Reader<Bytes>>(
355        CHECKPOINT_FILE_MAGIC,
356        blob.reader(),
357    )?)
358}
359
360/// Construct an iterator over a blob of checkpoint data.
361///
362/// The iterator filters checkpoints that belong to the specified range.
363///
364/// # Errors
365///
366/// The function fails if the blob is corrupted and fails to decode.
367pub fn make_blob_iterator_for_range(
368    blob: Bytes,
369    range: Range<CheckpointSequenceNumber>,
370) -> Result<impl Iterator<Item = CheckpointData>> {
371    Ok(make_blob_iterator(blob)?
372        .filter(move |data| range.contains(&data.checkpoint_summary.sequence_number)))
373}
iota_data_ingestion_core/history/reader.rs

iota_data_ingestion_core/history/
reader.rs
