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    pub fn verify_and_get_manifest_files(&self, manifest: Manifest) -> Result<Vec<FileMetadata>> {
78        let mut files = manifest.to_files();
79        if files.is_empty() {
80            return Err(IngestionError::HistoryRead(
81                "unexpected empty remote store of historical data".to_string(),
82            ));
83        }
84
85        files.sort_by_key(|f| f.checkpoint_seq_range.start);
86
87        assert!(
88            files
89                .windows(2)
90                .all(|w| w[1].checkpoint_seq_range.start == w[0].checkpoint_seq_range.end)
91        );
92
93        assert_eq!(files.first().map(|f| f.checkpoint_seq_range.start), Some(0));
94
95        Ok(files)
96    }
97
98    /// This function downloads checkpoint data files and ensures their
99    /// computed checksum matches the one in manifest.
100    pub async fn verify_file_consistency(&self, files: Vec<FileMetadata>) -> Result<()> {
101        let remote_object_store = self.remote_object_store.clone();
102        futures::stream::iter(files.iter())
103            .map(|metadata| {
104                let remote_object_store = remote_object_store.clone();
105                async move {
106                    let checkpoint_data = get(&remote_object_store, &metadata.file_path()).await?;
107                    Ok::<(Bytes, &FileMetadata), IngestionError>((checkpoint_data, metadata))
108                }
109            })
110            .boxed()
111            .buffer_unordered(self.concurrency)
112            .try_for_each(|(checkpoint_data, metadata)| {
113                let checksum = compute_sha3_checksum_for_bytes(checkpoint_data).map_err(Into::into);
114                let result = checksum.and_then(|checksum| {
115                    if checksum == metadata.sha3_digest {
116                        return Ok(());
117                    };
118                    Err(IngestionError::HistoryRead(format!(
119                        "checksum doesn't match for file: {:?}",
120                        metadata.file_path()
121                    )))
122                });
123                futures::future::ready(result)
124            })
125            .await
126    }
127
128    /// Stream blobs of [`Bytes`] that include checkpoint data for the specified
129    /// range.
130    ///
131    /// This method retrieves files with batches of serialized checkpoint
132    /// data from the remote store, and streams the respective contents
133    /// as blobs.
134    ///
135    /// # Errors
136    ///
137    /// Returns an error if resolving the files that need to be fetched from the
138    /// remote store fails.
139    ///
140    /// Additionally the stream may fail if fetching the file from the remote
141    /// store fails.
142    ///
143    /// # Examples
144    ///
145    /// ```ignore
146    /// use futures::StreamExt;
147    ///
148    /// let range = 100..200;
149    /// let mut stream = historical_reader.stream_blobs_for_range(range.clone()).await?;
150    /// while let Some(Ok(blob)) = stream.next().await {
151    ///     // we can now iterate over the checkpoint data
152    ///     for data in make_blob_iterator_for_range(blob, range.clone())? {
153    ///         println!("Received checkpoint data: {data:?}");
154    ///     }
155    /// }
156    /// ```
157    pub async fn stream_blobs_for_range(
158        &self,
159        checkpoint_range: Range<CheckpointSequenceNumber>,
160    ) -> Result<impl Stream<Item = Result<Bytes>> + Send + use<'_>> {
161        let files = self.get_files_for_range(checkpoint_range).await?;
162        Ok(futures::stream::iter(files)
163            .map(move |metadata| async move {
164                let remote_object_store = Arc::clone(&self.remote_object_store);
165                let file_path = metadata.file_path();
166                Ok(get(&remote_object_store, &file_path).await?)
167            })
168            .buffered(self.concurrency))
169    }
170
171    /// Construct an [`Iterator`] over [`CheckpointData`] for the specified
172    /// range.
173    ///
174    /// This method eagerly consumes the stream of blobs returned from
175    /// [`Self::stream_blobs_for_range`] and holds the data in memory until
176    /// the iterator is consumed.
177    ///
178    /// For lazy processing of the blobs use directly
179    /// [`Self::stream_blobs_for_range`] along with
180    /// [`make_blob_iterator_for_range`].
181    pub async fn iter_for_range(
182        &self,
183        checkpoint_range: Range<CheckpointSequenceNumber>,
184    ) -> Result<impl Iterator<Item = CheckpointData>> {
185        let blobs = self
186            .stream_blobs_for_range(checkpoint_range.clone())
187            .await?
188            .try_collect::<Vec<_>>()
189            .await?;
190        let data_iterators = blobs
191            .into_iter()
192            .map(|blob| {
193                let range = checkpoint_range.clone();
194                make_blob_iterator_for_range(blob, range)
195            })
196            .collect::<Result<Vec<_>>>()?;
197        Ok(data_iterators.into_iter().flatten())
198    }
199
200    /// Iterate [`CheckpointData`] from the given remote file.
201    ///
202    /// This method retrieves the file with batches of serialized checkpoint
203    /// data from the remote store, decodes the raw data, and streams the
204    /// deserialized values.
205    ///
206    /// # Errors
207    ///
208    /// Returns an error in the following cases:
209    ///
210    /// * If fetching the file from the remote store fails.
211    /// * If the file is corrupted and fails to decode.
212    pub async fn iter_for_file(
213        &self,
214        file_path: Path,
215    ) -> Result<impl Iterator<Item = CheckpointData>> {
216        let raw_data_batch = get(&self.remote_object_store, &file_path).await?;
217        make_blob_iterator(raw_data_batch)
218    }
219
220    /// Return latest available checkpoint in archive.
221    pub async fn latest_available_checkpoint(&self) -> Result<CheckpointSequenceNumber> {
222        self.manifest
223            .lock()
224            .await
225            .next_checkpoint_seq_num()
226            .checked_sub(1)
227            .ok_or_else(|| {
228                IngestionError::HistoryRead("no checkpoint data in the remote store".into())
229            })
230    }
231
232    pub fn remote_store_identifier(&self) -> String {
233        self.remote_object_store.to_string()
234    }
235
236    /// Returns the last checkpoint of each epoch, indexed by epoch.
237    ///
238    /// Read from the epoch boundaries file maintained alongside the manifest.
239    /// Callers slice the boundaries by epoch range as needed.
240    ///
241    /// # Errors
242    ///
243    /// Fails if the epoch boundaries file cannot be read or if it fails to
244    /// decode.
245    pub async fn epoch_boundaries(&self) -> Result<EpochBoundaries> {
246        read_epoch_boundaries(self.remote_object_store.clone()).await
247    }
248
249    /// Syncs the Manifest from remote store.
250    pub async fn sync_manifest_once(&self) -> Result<()> {
251        Self::sync_manifest(self.remote_object_store.clone(), self.manifest.clone()).await?;
252        Ok(())
253    }
254
255    pub async fn get_manifest(&self) -> Manifest {
256        self.manifest.lock().await.clone()
257    }
258
259    /// Copies Manifest from remote store to the given Manifest.
260    async fn sync_manifest(
261        remote_store: Arc<dyn ObjectStoreGetExt>,
262        manifest: Arc<Mutex<Manifest>>,
263    ) -> Result<()> {
264        let new_manifest = read_manifest(remote_store.clone()).await?;
265        let mut locked = manifest.lock().await;
266        *locked = new_manifest;
267        Ok(())
268    }
269
270    /// Resolve the files to fetch for the specified range.
271    ///
272    /// The method retrieves the manifest from the remote store and
273    /// searches for the files that cover the given range of checkpoint
274    /// data.
275    ///
276    /// # Errors
277    ///
278    /// The method fails if the remote store has no data, or if the
279    /// manifest fails to verify.
280    async fn get_files_for_range(
281        &self,
282        checkpoint_range: Range<CheckpointSequenceNumber>,
283    ) -> Result<impl Iterator<Item = FileMetadata>> {
284        let manifest = self.get_manifest().await;
285
286        let latest_available_checkpoint = manifest
287            .next_checkpoint_seq_num()
288            .checked_sub(1)
289            .ok_or_else(|| {
290                IngestionError::HistoryRead("no checkpoint data in the remote store".into())
291            })?;
292
293        if checkpoint_range.start > latest_available_checkpoint {
294            return Err(IngestionError::HistoryRead(format!(
295                "latest available checkpoint is: {latest_available_checkpoint}",
296            )));
297        }
298
299        let files = self.verify_and_get_manifest_files(manifest)?;
300
301        let start_index = match files
302            .binary_search_by_key(&checkpoint_range.start, |s| s.checkpoint_seq_range.start)
303        {
304            Ok(index) => index,
305            Err(index) => index - 1,
306        };
307
308        let end_index = match files
309            .binary_search_by_key(&checkpoint_range.end, |s| s.checkpoint_seq_range.start)
310        {
311            Ok(index) => index,
312            Err(index) => index,
313        };
314
315        Ok(files
316            .into_iter()
317            .enumerate()
318            .filter_map(move |(index, metadata)| {
319                (index >= start_index && index < end_index).then_some(metadata)
320            }))
321    }
322
323    fn spawn_manifest_sync_task(
324        remote_store: Arc<dyn ObjectStoreGetExt>,
325        manifest: Arc<Mutex<Manifest>>,
326        mut recv: oneshot::Receiver<()>,
327    ) {
328        tokio::task::spawn(async move {
329            let mut interval = tokio::time::interval(Duration::from_secs(60));
330            loop {
331                tokio::select! {
332                    _ = interval.tick() => {
333                        Self::sync_manifest(remote_store.clone(), manifest.clone()).await?;
334                    }
335                    _ = &mut recv => break,
336                }
337            }
338            info!("terminating the manifest sync loop");
339            Ok::<(), IngestionError>(())
340        });
341    }
342}
343
344fn make_blob_iterator(blob: Bytes) -> Result<impl Iterator<Item = CheckpointData>> {
345    Ok(make_iterator::<CheckpointData, Reader<Bytes>>(
346        CHECKPOINT_FILE_MAGIC,
347        blob.reader(),
348    )?)
349}
350
351/// Construct an iterator over a blob of checkpoint data.
352///
353/// The iterator filters checkpoints that belong to the specified range.
354///
355/// # Errors
356///
357/// The function fails if the blob is corrupted and fails to decode.
358pub fn make_blob_iterator_for_range(
359    blob: Bytes,
360    range: Range<CheckpointSequenceNumber>,
361) -> Result<impl Iterator<Item = CheckpointData>> {
362    Ok(make_blob_iterator(blob)?
363        .filter(move |data| range.contains(&data.checkpoint_summary.sequence_number)))
364}
iota_data_ingestion_core/history/reader.rs

iota_data_ingestion_core/history/
reader.rs
