iota_storage/
lib.rs

1// Copyright (c) Mysten Labs, Inc.
2// Modifications Copyright (c) 2024 IOTA Stiftung
3// SPDX-License-Identifier: Apache-2.0
4
5#![allow(dead_code)]
6
7use std::{
8    fs,
9    fs::File,
10    io,
11    io::{BufReader, Read, Write},
12    path::PathBuf,
13    sync::Arc,
14};
15
16use anyhow::{Result, anyhow};
17use byteorder::{BigEndian, ReadBytesExt, WriteBytesExt};
18use bytes::{Buf, Bytes};
19use fastcrypto::hash::{HashFunction, Sha3_256};
20use iota_types::{
21    committee::Committee,
22    messages_checkpoint::{CertifiedCheckpointSummary, VerifiedCheckpoint},
23    storage::WriteStore,
24};
25use num_enum::{IntoPrimitive, TryFromPrimitive};
26use serde::{Deserialize, Serialize, de::DeserializeOwned};
27use tracing::debug;
28
29use crate::blob::BlobIter;
30
31pub mod blob;
32pub mod http_key_value_store;
33pub mod key_value_store;
34pub mod key_value_store_metrics;
35pub mod mutex_table;
36pub mod object_store;
37pub mod package_object_cache;
38pub mod sharded_lru;
39pub mod write_path_pending_tx_log;
40
41pub const SHA3_BYTES: usize = 32;
42
43#[derive(
44    Copy, Clone, Debug, Eq, PartialEq, Serialize, Deserialize, TryFromPrimitive, IntoPrimitive,
45)]
46#[repr(u8)]
47pub enum StorageFormat {
48    Blob = 0,
49}
50
51#[derive(
52    Copy, Clone, Debug, Eq, PartialEq, Serialize, Deserialize, TryFromPrimitive, IntoPrimitive,
53)]
54#[repr(u8)]
55pub enum FileCompression {
56    None = 0,
57    Zstd,
58}
59
60impl FileCompression {
61    pub fn zstd_compress<R: Read, W: Write>(reader: &mut R, writer: &mut W) -> io::Result<()> {
62        // TODO: Add zstd compression level as function argument
63        let mut encoder = zstd::Encoder::new(writer, 1)?;
64        io::copy(reader, &mut encoder)?;
65        encoder.finish()?;
66        Ok(())
67    }
68    pub fn compress(&self, source: &std::path::Path) -> io::Result<()> {
69        match self {
70            FileCompression::Zstd => {
71                let mut input = File::open(source)?;
72                let tmp_file_name = source.with_extension("tmp");
73                let mut output = File::create(&tmp_file_name)?;
74                Self::zstd_compress(&mut input, &mut output)?;
75                fs::rename(tmp_file_name, source)?;
76            }
77            FileCompression::None => {}
78        }
79        Ok(())
80    }
81    pub fn decompress(&self, source: &PathBuf) -> Result<Box<dyn Read>> {
82        let file = File::open(source)?;
83        let res: Box<dyn Read> = match self {
84            FileCompression::Zstd => Box::new(zstd::stream::Decoder::new(file)?),
85            FileCompression::None => Box::new(BufReader::new(file)),
86        };
87        Ok(res)
88    }
89    pub fn bytes_decompress(&self, bytes: Bytes) -> Result<Box<dyn Read>> {
90        let res: Box<dyn Read> = match self {
91            FileCompression::Zstd => Box::new(zstd::stream::Decoder::new(bytes.reader())?),
92            FileCompression::None => Box::new(BufReader::new(bytes.reader())),
93        };
94        Ok(res)
95    }
96}
97
98pub fn compute_sha3_checksum_for_bytes(bytes: Bytes) -> Result<[u8; 32]> {
99    let mut hasher = Sha3_256::default();
100    io::copy(&mut bytes.reader(), &mut hasher)?;
101    Ok(hasher.finalize().digest)
102}
103
104pub fn compute_sha3_checksum_for_file(file: &mut File) -> Result<[u8; 32]> {
105    let mut hasher = Sha3_256::default();
106    io::copy(file, &mut hasher)?;
107    Ok(hasher.finalize().digest)
108}
109
110pub fn compute_sha3_checksum(source: &std::path::Path) -> Result<[u8; 32]> {
111    let mut file = fs::File::open(source)?;
112    compute_sha3_checksum_for_file(&mut file)
113}
114
115pub fn compress<R: Read, W: Write>(reader: &mut R, writer: &mut W) -> Result<()> {
116    let magic = reader.read_u32::<BigEndian>()?;
117    writer.write_u32::<BigEndian>(magic)?;
118    let storage_format = reader.read_u8()?;
119    writer.write_u8(storage_format)?;
120    let file_compression = FileCompression::try_from(reader.read_u8()?)?;
121    writer.write_u8(file_compression.into())?;
122    match file_compression {
123        FileCompression::Zstd => {
124            FileCompression::zstd_compress(reader, writer)?;
125        }
126        FileCompression::None => {}
127    }
128    Ok(())
129}
130
131pub fn read<R: Read + 'static>(
132    expected_magic: u32,
133    mut reader: R,
134) -> Result<(Box<dyn Read>, StorageFormat)> {
135    let magic = reader.read_u32::<BigEndian>()?;
136    if magic != expected_magic {
137        Err(anyhow!(
138            "Unexpected magic string in file: {magic:?}, expected: {expected_magic:?}"
139        ))
140    } else {
141        let storage_format = StorageFormat::try_from(reader.read_u8()?)?;
142        let file_compression = FileCompression::try_from(reader.read_u8()?)?;
143        let reader: Box<dyn Read> = match file_compression {
144            FileCompression::Zstd => Box::new(zstd::stream::Decoder::new(reader)?),
145            FileCompression::None => Box::new(BufReader::new(reader)),
146        };
147        Ok((reader, storage_format))
148    }
149}
150
151pub fn make_iterator<T: DeserializeOwned, R: Read + 'static>(
152    expected_magic: u32,
153    reader: R,
154) -> Result<impl Iterator<Item = T>> {
155    let (reader, storage_format) = read(expected_magic, reader)?;
156    match storage_format {
157        StorageFormat::Blob => Ok(BlobIter::new(reader)),
158    }
159}
160
161/// The chain-linkage part of checkpoint verification: checks that
162/// `checkpoint` directly extends `current` (matching previous digest and a
163/// valid epoch transition). Does not verify authority signatures; returns the
164/// summary unchanged on success so the caller can finish verification.
165///
166/// # Panics
167///
168/// Panics if `checkpoint`'s sequence number is not `current`'s plus one.
169#[expect(clippy::result_large_err)]
170pub fn verify_checkpoint_linkage(
171    current: &VerifiedCheckpoint,
172    checkpoint: CertifiedCheckpointSummary,
173) -> Result<CertifiedCheckpointSummary, CertifiedCheckpointSummary> {
174    assert_eq!(
175        checkpoint.sequence_number(),
176        current.sequence_number().checked_add(1).unwrap()
177    );
178
179    if Some(*current.digest()) != checkpoint.previous_digest {
180        debug!(
181            current_checkpoint_seq = current.sequence_number(),
182            current_digest =% current.digest(),
183            checkpoint_seq = checkpoint.sequence_number(),
184            checkpoint_digest =% checkpoint.digest(),
185            checkpoint_previous_digest =? checkpoint.previous_digest,
186            "checkpoint not on same chain"
187        );
188        return Err(checkpoint);
189    }
190
191    let current_epoch = current.epoch();
192    if checkpoint.epoch() != current_epoch
193        && checkpoint.epoch() != current_epoch.checked_add(1).unwrap()
194    {
195        debug!(
196            checkpoint_seq = checkpoint.sequence_number(),
197            checkpoint_epoch = checkpoint.epoch(),
198            current_checkpoint_seq = current.sequence_number(),
199            current_epoch = current_epoch,
200            "cannot verify checkpoint with too high of an epoch",
201        );
202        return Err(checkpoint);
203    }
204
205    if checkpoint.epoch() == current_epoch.checked_add(1).unwrap()
206        && current.next_epoch_committee().is_none()
207    {
208        debug!(
209            checkpoint_seq = checkpoint.sequence_number(),
210            checkpoint_epoch = checkpoint.epoch(),
211            current_checkpoint_seq = current.sequence_number(),
212            current_epoch = current_epoch,
213            "next checkpoint claims to be from the next epoch but the latest verified \
214            checkpoint does not indicate that it is the last checkpoint of an epoch"
215        );
216        return Err(checkpoint);
217    }
218
219    Ok(checkpoint)
220}
221
222#[expect(clippy::result_large_err)]
223pub fn verify_checkpoint_with_committee(
224    committee: Arc<Committee>,
225    current: &VerifiedCheckpoint,
226    checkpoint: CertifiedCheckpointSummary,
227) -> Result<VerifiedCheckpoint, CertifiedCheckpointSummary> {
228    let checkpoint = verify_checkpoint_linkage(current, checkpoint)?;
229
230    checkpoint
231        .verify_authority_signatures(&committee)
232        .map_err(|e| {
233            debug!("error verifying checkpoint: {e}");
234            checkpoint.clone()
235        })?;
236    Ok(VerifiedCheckpoint::new_unchecked(checkpoint))
237}
238
239#[expect(clippy::result_large_err)]
240pub fn verify_checkpoint<S>(
241    current: &VerifiedCheckpoint,
242    store: S,
243    checkpoint: CertifiedCheckpointSummary,
244) -> Result<VerifiedCheckpoint, CertifiedCheckpointSummary>
245where
246    S: WriteStore,
247{
248    let committee = store.get_committee(checkpoint.epoch()).unwrap_or_else(|| {
249        panic!(
250            "BUG: should have committee for epoch {} before we try to verify checkpoint {}",
251            checkpoint.epoch(),
252            checkpoint.sequence_number()
253        )
254    });
255
256    verify_checkpoint_with_committee(committee, current, checkpoint)
257}
iota_storage/lib.rs

iota_storage/
lib.rs
