iota_data_ingestion_core/history/
manifest.rs

1// Copyright (c) Mysten Labs, Inc.
2// Modifications Copyright (c) 2024 IOTA Stiftung
3// SPDX-License-Identifier: Apache-2.0
4
5//! Handle the manifest for historical checkpoint data.
6//!
7//! MANIFEST File Disk Format
8//! ┌──────────────────────────────┐
9//! │        magic<4 byte>         │
10//! ├──────────────────────────────┤
11//! │   serialized manifest        │
12//! ├──────────────────────────────┤
13//! │      sha3 <32 bytes>         │
14//! └──────────────────────────────┘
15
16use std::{num::NonZeroUsize, ops::Range};
17
18use bytes::Bytes;
19use iota_config::object_storage_config::ObjectStoreConfig;
20use iota_storage::{
21    compute_sha3_checksum, compute_sha3_checksum_for_bytes,
22    object_store::{
23        ObjectStoreGetExt, ObjectStorePutExt,
24        util::{get, put},
25    },
26};
27use object_store::path::Path;
28use serde::{Deserialize, Serialize};
29use tracing::info;
30
31use crate::{
32    errors::IngestionResult as Result,
33    history::{
34        CHECKPOINT_FILE_SUFFIX, MANIFEST_FILE_MAGIC, MANIFEST_FILENAME, finalize_magic_blob,
35        read_magic_blob,
36        reader::{HistoricalReader, HistoricalReaderConfig},
37    },
38};
39
40#[derive(Debug, Clone, Serialize, Deserialize, Eq, PartialEq)]
41pub struct FileMetadata {
42    pub checkpoint_seq_range: Range<u64>,
43    pub sha3_digest: [u8; 32],
44}
45
46impl FileMetadata {
47    pub fn file_path(&self) -> Path {
48        Path::from(format!(
49            "{}.{CHECKPOINT_FILE_SUFFIX}",
50            self.checkpoint_seq_range.start
51        ))
52    }
53}
54
55#[derive(Debug, Clone, Serialize, Deserialize, Eq, PartialEq)]
56pub struct ManifestV1 {
57    pub archive_version: u8,
58    pub next_checkpoint_seq_num: u64,
59    pub file_metadata: Vec<FileMetadata>,
60}
61
62#[derive(Debug, Serialize, Deserialize, Clone, Eq, PartialEq)]
63#[non_exhaustive]
64pub enum Manifest {
65    V1(ManifestV1),
66}
67
68impl Manifest {
69    pub fn new(next_checkpoint_seq_num: u64) -> Self {
70        Manifest::V1(ManifestV1 {
71            archive_version: 1,
72            next_checkpoint_seq_num,
73            file_metadata: vec![],
74        })
75    }
76
77    pub fn to_files(&self) -> Vec<FileMetadata> {
78        match self {
79            Manifest::V1(manifest) => manifest.file_metadata.clone(),
80        }
81    }
82
83    pub fn next_checkpoint_seq_num(&self) -> u64 {
84        match self {
85            Manifest::V1(manifest) => manifest.next_checkpoint_seq_num,
86        }
87    }
88
89    pub fn update(&mut self, checkpoint_sequence_number: u64, file_metadata: FileMetadata) {
90        match self {
91            Manifest::V1(manifest) => {
92                manifest.file_metadata.push(file_metadata);
93                manifest.next_checkpoint_seq_num = checkpoint_sequence_number;
94            }
95        }
96    }
97
98    pub fn file_path() -> Path {
99        Path::from(MANIFEST_FILENAME)
100    }
101}
102
103#[derive(Debug, Clone, Serialize, Deserialize, Eq, PartialEq)]
104pub struct CheckpointUpdates {
105    file_metadata: FileMetadata,
106    manifest: Manifest,
107}
108
109impl CheckpointUpdates {
110    pub fn new(
111        checkpoint_sequence_number: u64,
112        file_metadata: FileMetadata,
113        manifest: &mut Manifest,
114    ) -> Self {
115        manifest.update(checkpoint_sequence_number, file_metadata.clone());
116        CheckpointUpdates {
117            file_metadata,
118            manifest: manifest.clone(),
119        }
120    }
121
122    pub fn file_path(&self) -> Path {
123        self.file_metadata.file_path()
124    }
125
126    pub fn manifest_file_path(&self) -> Path {
127        Path::from(MANIFEST_FILENAME)
128    }
129}
130
131pub fn create_file_metadata(
132    file_path: &std::path::Path,
133    checkpoint_seq_range: Range<u64>,
134) -> Result<FileMetadata> {
135    let sha3_digest = compute_sha3_checksum(file_path)?;
136    let file_metadata = FileMetadata {
137        checkpoint_seq_range,
138        sha3_digest,
139    };
140    Ok(file_metadata)
141}
142
143pub fn create_file_metadata_from_bytes(
144    contents: Bytes,
145    checkpoint_seq_range: Range<u64>,
146) -> Result<FileMetadata> {
147    let sha3_digest = compute_sha3_checksum_for_bytes(contents)?;
148    let file_metadata = FileMetadata {
149        checkpoint_seq_range,
150        sha3_digest,
151    };
152    Ok(file_metadata)
153}
154
155/// Reads the manifest file from the store.
156pub async fn read_manifest<S: ObjectStoreGetExt>(remote_store: S) -> Result<Manifest> {
157    let vec = get(&remote_store, &Manifest::file_path()).await?.to_vec();
158    read_manifest_from_bytes(vec)
159}
160
161/// Reads the manifest file from the given byte vector and verifies the
162/// integrity of the file.
163pub fn read_manifest_from_bytes(vec: Vec<u8>) -> Result<Manifest> {
164    read_magic_blob(vec, MANIFEST_FILE_MAGIC, MANIFEST_FILENAME)
165}
166
167/// Computes the SHA3 checksum of the Manifest and writes it to a byte vector.
168pub fn finalize_manifest(manifest: Manifest) -> Result<Bytes> {
169    finalize_magic_blob(&manifest, MANIFEST_FILE_MAGIC)
170}
171
172/// Writes the Manifest to the remote store.
173pub async fn write_manifest<S: ObjectStorePutExt>(
174    manifest: Manifest,
175    remote_store: S,
176) -> Result<()> {
177    let bytes = finalize_manifest(manifest)?;
178    put(&remote_store, &Manifest::file_path(), bytes).await?;
179    Ok(())
180}
181
182pub async fn verify_historical_checkpoints_with_checksums(
183    remote_store_config: ObjectStoreConfig,
184    concurrency: usize,
185) -> Result<()> {
186    let config = HistoricalReaderConfig {
187        remote_store_config,
188        download_concurrency: NonZeroUsize::new(concurrency).unwrap(),
189    };
190    // Gets the Manifest from the remote store.
191    let reader = HistoricalReader::new(config)?;
192    reader.sync_manifest_once().await?;
193    let manifest = reader.get_manifest().await;
194    info!(
195        "next checkpoint in archive store: {}",
196        manifest.next_checkpoint_seq_num()
197    );
198
199    let file_metadata = reader.verify_and_get_manifest_files(manifest)?;
200
201    // Account for both summary and content files
202    let num_files = file_metadata.len() * 2;
203    reader.verify_file_consistency(file_metadata).await?;
204    info!("all {num_files} files are valid");
205    Ok(())
206}
iota_data_ingestion_core/history/manifest.rs

iota_data_ingestion_core/history/
manifest.rs
