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

iota_storage/
lib.rs
