iota_snapshot/
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
7#[cfg(test)]
8mod tests;
9
10pub mod progress;
11pub mod reader;
12pub mod restore;
13pub mod uploader;
14mod writer;
15
16use std::{
17    collections::HashSet,
18    num::NonZeroUsize,
19    path::PathBuf,
20    sync::{
21        Arc,
22        atomic::{AtomicU64, Ordering},
23    },
24};
25
26use anyhow::Result;
27use fastcrypto::hash::MultisetHash;
28use indicatif::{MultiProgress, ProgressBar, ProgressStyle};
29use iota_core::{
30    authority::{
31        authority_store_tables::{AuthorityPerpetualTables, LiveObject},
32        epoch_start_configuration::{EpochFlag, EpochStartConfiguration},
33    },
34    checkpoints::CheckpointStore,
35    epoch::committee_store::CommitteeStore,
36    global_state_hasher::GlobalStateHasher,
37};
38use iota_sdk_types::{ObjectId, ObjectReference};
39use iota_storage::{
40    FileCompression, SHA3_BYTES, compute_sha3_checksum, object_store::util::path_to_filesystem,
41};
42use iota_types::{
43    IOTA_SYSTEM_STATE_OBJECT_ID,
44    base_types::ExecutionDigests,
45    committee::{Committee, CommitteeChainVerifier, EpochId},
46    digests::ChainIdentifier,
47    effects::{TransactionEffectsAPI, TransactionEffectsExt},
48    global_state_hash::GlobalStateHash,
49    iota_system_state::{
50        IotaSystemState, IotaSystemStateTrait,
51        epoch_start_iota_system_state::EpochStartSystemStateTrait, get_iota_system_state,
52    },
53    messages_checkpoint::{CheckpointSequenceNumber, ECMHLiveObjectSetDigest},
54    object::Object,
55    storage::{EpochInfoV1Entry, EpochInfoV2},
56};
57use num_enum::{IntoPrimitive, TryFromPrimitive};
58use object_store::path::Path;
59use serde::{Deserialize, Serialize};
60
61use crate::{
62    progress::{ProgressTicker, ProgressUnit},
63    restore::RestoreEpochInfo,
64};
65
66/// The following describes the format of an object file (*.obj) used for
67/// persisting live iota objects. The maximum size per .obj file is 128MB. State
68/// snapshot will be taken at the end of every epoch. Live object set is split
69/// into and stored across multiple hash buckets. The hashing function used
70/// for bucketing objects is the same as the one used to build the accumulator
71/// tree for computing state root hash. Buckets are further subdivided into
72/// partitions. A partition is a smallest storage unit which holds a subset of
73/// objects in one bucket. Each partition is a single *.obj file where
74/// objects are appended to in an append-only fashion. A new partition is
75/// created when the current one reaches its maximum size. i.e. 128MB.
76/// Partitions allow a single hash bucket to be consumed in parallel. Partition
77/// files are optionally compressed with the zstd compression format. Partition
78/// filenames follows the format <bucket_number>_<partition_number>.obj. Object
79/// references for hash. There is one single ref file per hash bucket. Object
80/// references are written in an append-only manner as well. Finally, the
81/// MANIFEST file contains per file metadata of every file in the snapshot
82/// directory.
83///
84/// Snapshot-format V2 additions over V1:
85/// - OBJECT file magic is `0x00B7EC76` (V1 was `0x00B7EC75`); a V2 reader fails
86///   fast on a V1 magic and vice versa. Encoded records are BCS-serialized
87///   `SnapshotLiveObject` carrying the per-object
88///   `previous_transaction_checkpoint` inline. The writer rejects rows whose
89///   checkpoint is `None` (lifted from pre-V2 store rows) at the publish
90///   boundary, so any record present in a published `.obj` file carries a
91///   concrete checkpoint sequence number.
92/// - REFERENCE file format is unchanged from V1.
93/// - A per-snapshot `EPOCH_INFO` file carries one [`EpochInfoV1Entry`] per
94///   epoch in `[0, snapshot_epoch]` from the CheckpointStore's `epoch_info`
95///   table. The writer refuses to publish unless that table's completeness
96///   watermark covers `snapshot_epoch`.
97/// - `MANIFEST` is now [`ManifestV2`], adding a `chain_id` field so a restore
98///   can reject a foreign-chain snapshot.
99///
100/// State Snapshot Directory Layout
101///  - snapshot/
102///     - epoch_0/
103///        - 1_1.obj
104///        - 1_2.obj
105///        - 1_3.obj
106///        - 2_1.obj
107///        - ...
108///        - 1000_1.obj
109///        - REFERENCE-1
110///        - REFERENCE-2
111///        - ...
112///        - REFERENCE-1000
113///        - EPOCH_INFO
114///        - MANIFEST
115///     - epoch_1/
116///       - 1_1.obj
117///       - ...
118///
119/// Object File Disk Format
120/// ┌──────────────────────────────┐
121/// │  magic(0x00B7EC76) <4 byte>  │
122/// ├──────────────────────────────┤
123/// │ ┌──────────────────────────┐ │
124/// │ │         Object 1         │ │
125/// │ ├──────────────────────────┤ │
126/// │ │          ...             │ │
127/// │ ├──────────────────────────┤ │
128/// │ │         Object N         │ │
129/// │ └──────────────────────────┘ │
130/// └──────────────────────────────┘
131/// Object
132/// ┌───────────────┬───────────────────┬──────────────┐
133/// │ len <uvarint> │ encoding <1 byte> │ data <bytes> │
134/// └───────────────┴───────────────────┴──────────────┘
135///
136/// REFERENCE File Disk Format
137/// ┌────────────────────────────────────┐
138/// │     magic(0xDEADBEEF) <4 byte>     │
139/// ├────────────────────────────────────┤
140/// │ ┌────────────────────────────────┐ │
141/// │ │       ObjectReference 1        │ │
142/// │ ├────────────────────────────────┤ │
143/// │ │              ...               │ │
144/// │ ├────────────────────────────────┤ │
145/// │ │       ObjectReference N        │ │
146/// │ └────────────────────────────────┘ │
147/// └────────────────────────────────────┘
148/// ObjectReference (ObjectId, Version, ObjectDigest)
149/// ┌───────────────┬───────────────────┬──────────────┐
150/// │         data (<(address_len + 8 + 32) bytes>)    │
151/// └───────────────┴───────────────────┴──────────────┘
152///
153/// EPOCH_INFO File Disk Format
154/// ┌──────────────────────────────┐
155/// │  magic(0x9000C001) <4 byte>  │
156/// ├──────────────────────────────┤
157/// │   bcs(EpochInfo)             │
158/// └──────────────────────────────┘
159/// See [`EpochInfo`] for the schema. `FileMetadata::sha3_digest` in the
160/// MANIFEST can be used to verify file integrity.
161///
162/// MANIFEST File Disk Format
163/// ┌──────────────────────────────┐
164/// │  magic(0x00C0FFEE) <4 byte>  │
165/// ├──────────────────────────────┤
166/// │   serialized manifest        │
167/// ├──────────────────────────────┤
168/// │      sha3 <32 bytes>         │
169/// └──────────────────────────────┘
170const OBJECT_FILE_MAGIC: u32 = 0x00B7EC76;
171const REFERENCE_FILE_MAGIC: u32 = 0xDEADBEEF;
172const EPOCH_INFO_FILE_MAGIC: u32 = 0x9000C001;
173const MANIFEST_FILE_MAGIC: u32 = 0x00C0FFEE;
174const MAGIC_BYTES: usize = 4;
175const SNAPSHOT_VERSION_BYTES: usize = 1;
176const ADDRESS_LENGTH_BYTES: usize = 8;
177const PADDING_BYTES: usize = 3;
178const MANIFEST_FILE_HEADER_BYTES: usize =
179    MAGIC_BYTES + SNAPSHOT_VERSION_BYTES + ADDRESS_LENGTH_BYTES + PADDING_BYTES;
180const FILE_MAX_BYTES: usize = 128 * 1024 * 1024;
181const OBJECT_ID_BYTES: usize = ObjectId::LENGTH;
182const SEQUENCE_NUM_BYTES: usize = 8;
183const OBJECT_DIGEST_BYTES: usize = 32;
184const OBJECT_REF_BYTES: usize = OBJECT_ID_BYTES + SEQUENCE_NUM_BYTES + OBJECT_DIGEST_BYTES;
185const FILE_TYPE_BYTES: usize = 1;
186const BUCKET_BYTES: usize = 4;
187const BUCKET_PARTITION_BYTES: usize = 4;
188const COMPRESSION_TYPE_BYTES: usize = 1;
189const FILE_METADATA_BYTES: usize =
190    FILE_TYPE_BYTES + BUCKET_BYTES + BUCKET_PARTITION_BYTES + COMPRESSION_TYPE_BYTES + SHA3_BYTES;
191
192pub fn default_download_concurrency() -> NonZeroUsize {
193    const MAX_PARALLEL_DOWNLOADS: usize = 8;
194    let cores = std::thread::available_parallelism().map_or(1, NonZeroUsize::get);
195    NonZeroUsize::new(cores.saturating_sub(1).clamp(1, MAX_PARALLEL_DOWNLOADS))
196        .unwrap_or(NonZeroUsize::MIN)
197}
198
199#[derive(
200    Copy, Clone, Debug, Eq, PartialEq, Serialize, Deserialize, TryFromPrimitive, IntoPrimitive,
201)]
202#[repr(u8)]
203pub enum FileType {
204    Object = 0,
205    Reference = 1,
206    /// per-epoch metadata file
207    EpochInfo = 2,
208}
209
210#[derive(Debug, Clone, Serialize, Deserialize, Eq, PartialEq)]
211/// FileMetadata holds either an object or a reference file metadata.
212pub struct FileMetadata {
213    pub file_type: FileType,
214    pub bucket_num: u32,
215    pub part_num: u32,
216    pub file_compression: FileCompression,
217    pub sha3_digest: [u8; 32],
218}
219
220impl FileMetadata {
221    pub fn file_path(&self, dir_path: &Path) -> Path {
222        match self.file_type {
223            FileType::Object => {
224                dir_path.child(&*format!("{}_{}.obj", self.bucket_num, self.part_num))
225            }
226            FileType::Reference => {
227                dir_path.child(&*format!("{}_{}.ref", self.bucket_num, self.part_num))
228            }
229            // EPOCH_INFO is a singleton per snapshot, so bucket/part numbers
230            // do not contribute to the filename.
231            FileType::EpochInfo => dir_path.child("EPOCH_INFO"),
232        }
233    }
234    pub fn local_file_path(&self, root_path: &std::path::Path, dir_path: &Path) -> Result<PathBuf> {
235        path_to_filesystem(root_path.to_path_buf(), &self.file_path(dir_path))
236    }
237}
238
239#[derive(Debug, Serialize, Deserialize, Eq, PartialEq)]
240pub struct ManifestV1 {
241    pub snapshot_version: u8,
242    pub address_length: u64,
243    pub file_metadata: Vec<FileMetadata>,
244    pub epoch: u64,
245}
246
247/// `ManifestV1` plus `chain_id`, letting a restore reject a foreign-chain
248/// snapshot.
249#[derive(Debug, Serialize, Deserialize, Eq, PartialEq)]
250pub struct ManifestV2 {
251    pub snapshot_version: u8,
252    pub address_length: u64,
253    pub file_metadata: Vec<FileMetadata>,
254    pub epoch: u64,
255    pub chain_id: ChainIdentifier,
256}
257
258#[derive(Debug, Serialize, Deserialize, Eq, PartialEq)]
259pub enum Manifest {
260    V1(ManifestV1),
261    V2(ManifestV2),
262}
263
264impl Manifest {
265    pub fn snapshot_version(&self) -> u8 {
266        match self {
267            Self::V1(manifest) => manifest.snapshot_version,
268            Self::V2(manifest) => manifest.snapshot_version,
269        }
270    }
271    pub fn address_length(&self) -> u64 {
272        match self {
273            Self::V1(manifest) => manifest.address_length,
274            Self::V2(manifest) => manifest.address_length,
275        }
276    }
277    pub fn file_metadata(&self) -> &Vec<FileMetadata> {
278        match self {
279            Self::V1(manifest) => &manifest.file_metadata,
280            Self::V2(manifest) => &manifest.file_metadata,
281        }
282    }
283    pub fn epoch(&self) -> u64 {
284        match self {
285            Self::V1(manifest) => manifest.epoch,
286            Self::V2(manifest) => manifest.epoch,
287        }
288    }
289    /// Producing chain's identifier; `None` for V1.
290    pub fn chain_id(&self) -> Option<ChainIdentifier> {
291        match self {
292            Self::V1(_) => None,
293            Self::V2(manifest) => Some(manifest.chain_id),
294        }
295    }
296}
297
298/// On-disk schema for the per-snapshot `EPOCH_INFO` file. Versioned for
299/// future schema evolution. `entries[i]` is the entry for epoch `i`; the length
300/// must be `snapshot_epoch + 1`, which `verify_epoch_info_chain` enforces.
301#[derive(Debug, Serialize, Deserialize)]
302pub enum EpochInfo {
303    V1(EpochInfoV1),
304}
305
306#[derive(Debug, Serialize, Deserialize)]
307pub struct EpochInfoV1 {
308    pub entries: Vec<EpochInfoV1Entry>,
309}
310
311impl EpochInfo {
312    pub fn entries(&self) -> &[EpochInfoV1Entry] {
313        match self {
314            Self::V1(info) => &info.entries,
315        }
316    }
317
318    pub fn into_entries(self) -> Vec<EpochInfoV1Entry> {
319        match self {
320            Self::V1(info) => info.entries,
321        }
322    }
323}
324
325/// Chain-verified `EPOCH_INFO`: the `chain_id` matched and every entry was
326/// anchored to its certified `last_checkpoint_summary` — the committee chain
327/// walked from the genesis committee, every byte hash-checked back to that
328/// signed summary. The only constructor is `verify_epoch_info_chain`, so
329/// holding one is proof the data is anchored to the operator-provided genesis.
330#[derive(Debug)]
331pub struct VerifiedEpochInfo {
332    epoch_info: EpochInfo,
333    committees: Vec<Committee>,
334    start_system_states: Vec<IotaSystemState>,
335}
336
337impl VerifiedEpochInfo {
338    /// Entries for epochs `[0, snapshot_epoch]`, in epoch order.
339    ///
340    /// These values represent data at the boundary of each epoch, including the
341    /// committee and start system state of the next epoch.
342    ///
343    /// The last entry represents the snapshot epoch boundary, which carries
344    /// information about the next epoch. Thus [`Self::committees`], and
345    /// [`Self::start_system_states`] have one more item for the committee and
346    /// the start system state of the epoch following the snapshot epoch.
347    pub fn entries(&self) -> &[EpochInfoV1Entry] {
348        self.epoch_info.entries()
349    }
350
351    /// The entry for the snapshot epoch, i.e. the last of [`Self::entries`].
352    pub fn snapshot_entry(&self) -> &EpochInfoV1Entry {
353        // `verify_epoch_info_chain` is the only constructor and rejects an
354        // entry count other than `snapshot_epoch + 1`, so there is one.
355        self.entries()
356            .last()
357            .expect("a verified EPOCH_INFO covers epochs [0, snapshot_epoch]")
358    }
359
360    /// Committees for epochs `[0, snapshot_epoch + 1]`: the genesis committee
361    /// plus one handed forward by each entry's `end_of_epoch_data`.
362    pub fn committees(&self) -> &[Committee] {
363        &self.committees
364    }
365
366    /// Digest-verified start system state per epochs `[0, snapshot_epoch + 1].
367    ///
368    /// Contains the genesis start system state, plus the ones derived by the
369    /// epoch boundaries represented by [`Self::entries`].
370    pub fn start_system_states(&self) -> &[IotaSystemState] {
371        &self.start_system_states
372    }
373
374    /// Consumes the verified info into its components: the epoch info entries,
375    /// the per-epoch committees, and the per-epoch start system states.
376    ///
377    /// See [`Self::entries`], [`Self::committees`], and
378    /// [`Self::start_system_states`].
379    pub fn into_parts(self) -> (EpochInfo, Vec<Committee>, Vec<IotaSystemState>) {
380        (self.epoch_info, self.committees, self.start_system_states)
381    }
382
383    /// Convert the verified entries `[0, snapshot_epoch]` into `EpochInfoV2`
384    /// index rows. Each row's `system_state` is the digest-verified start state
385    /// of its own epoch (`start_system_states[i]`).
386    pub(crate) fn into_epoch_info_v2_rows(self) -> Vec<EpochInfoV2> {
387        let VerifiedEpochInfo {
388            epoch_info,
389            start_system_states,
390            ..
391        } = self;
392        // `start_system_states[i]` is epoch `i`'s start state; `zip` drops the
393        // trailing one (the state the last boundary proves, which has no row).
394        // Each epoch's start checkpoint is the previous epoch's last + 1 (0 for
395        // epoch 0), derived inline from the signed summaries.
396        let mut previous_end_checkpoint: Option<u64> = None;
397        epoch_info
398            .into_entries()
399            .into_iter()
400            .zip(start_system_states)
401            .map(|(entry, start_system_state)| {
402                let start_checkpoint = previous_end_checkpoint.map_or(0, |seq| seq + 1);
403                previous_end_checkpoint =
404                    Some(entry.last_checkpoint_summary.data().sequence_number());
405                epoch_info_v2_row(entry, start_system_state, start_checkpoint)
406            })
407            .collect()
408    }
409
410    /// Restore the verified rows `[0, snapshot_epoch]` into the given
411    /// consumer's epoch store.
412    pub async fn restore_epoch_info(self, db: &impl RestoreEpochInfo) -> anyhow::Result<()> {
413        let rows = self.into_epoch_info_v2_rows();
414        db.restore_epoch_info(rows).await
415    }
416}
417
418/// Verify a snapshot's `EPOCH_INFO` against the operator's trust roots: the
419/// expected `chain_id`, the committee chain walked from `genesis_committee`,
420/// and `genesis_system_state` (epoch 0's start state, which no entry proves).
421/// The entries must be the contiguous certified closes of epochs
422/// `[0, snapshot_epoch]`, each signed by the committee the previous entry
423/// handed forward, with its proof bundle hashing back to the signed summary
424/// (see `verify_epoch_boundary_proof`). Nothing is written; the returned
425/// `VerifiedEpochInfo` is the witness consumers require.
426pub fn verify_epoch_info_chain(
427    epoch_info: EpochInfo,
428    snapshot_epoch: EpochId,
429    genesis_committee: Committee,
430    genesis_system_state: IotaSystemState,
431    snapshot_chain_id: ChainIdentifier,
432    expected_chain_id: ChainIdentifier,
433) -> anyhow::Result<VerifiedEpochInfo> {
434    anyhow::ensure!(
435        snapshot_chain_id == expected_chain_id,
436        "snapshot chain_id {snapshot_chain_id} does not match this node's chain \
437         {expected_chain_id} (snapshot from the wrong network's bucket?)"
438    );
439    anyhow::ensure!(
440        genesis_committee.epoch == 0,
441        "the trust root must be the genesis committee, got epoch {}",
442        genesis_committee.epoch
443    );
444    // With the contiguity check below, this binds the last entry to the
445    // requested epoch: a shorter chain would restore an earlier epoch, and an
446    // empty one leaves consumers with no snapshot boundary at all.
447    let entry_count = epoch_info.entries().len();
448    anyhow::ensure!(
449        entry_count as u64 == snapshot_epoch + 1,
450        "EPOCH_INFO carries {entry_count} entries, but a snapshot of epoch \
451         {snapshot_epoch} must carry one entry per epoch in [0, {snapshot_epoch}]"
452    );
453
454    let mut chain_verifier = CommitteeChainVerifier::new(genesis_committee);
455    let mut committees = vec![chain_verifier.committee().clone()];
456    // `start_system_states[i]` is epoch `i`'s start state; epoch 0's is the
457    // genesis root, every later one is derived from the previous boundary.
458    let mut start_system_states = vec![genesis_system_state];
459    for (index, entry) in epoch_info.entries().iter().enumerate() {
460        // EPOCH_INFO-specific: entries must be the contiguous epochs from 0.
461        // The epoch comes from the signed summary, not a stored field, so it
462        // can't disagree with the data it anchors.
463        let epoch = entry.last_checkpoint_summary.epoch();
464        anyhow::ensure!(
465            epoch == index as u64,
466            "EPOCH_INFO entry at index {index} carries a summary for epoch {epoch}",
467        );
468
469        // Defense in depth: the committee in epoch `index`'s start state must
470        // match the one the chain certified — catches a tampered validator set
471        // that left the (separately signed) committee handover intact.
472        let start_committee = start_system_states[index].get_current_epoch_committee();
473        anyhow::ensure!(
474            start_committee.committee() == chain_verifier.committee(),
475            "EPOCH_INFO entry for epoch {index}: the committee in its start system \
476             state does not match the certified committee",
477        );
478
479        chain_verifier
480            .verify_epoch_close(entry.last_checkpoint_summary.clone())
481            .map_err(|e| {
482                anyhow::anyhow!("EPOCH_INFO entry for epoch {index} failed verification: {e}")
483            })?;
484
485        // Anchor the rest of the entry to the now-verified summary and derive
486        // epoch `index + 1`'s start state from the boundary objects.
487        let next_start_state = verify_epoch_boundary_proof(entry)
488            .map_err(|e| anyhow::anyhow!("EPOCH_INFO entry for epoch {index}: {e}"))?;
489
490        committees.push(chain_verifier.committee().clone());
491        start_system_states.push(next_start_state);
492    }
493
494    Ok(VerifiedEpochInfo {
495        epoch_info,
496        committees,
497        start_system_states,
498    })
499}
500
501/// Anchor an entry's proof bundle to its (already signature-verified)
502/// `last_checkpoint_summary` and return the next epoch's digest-verified start
503/// state — the system-state objects this boundary wrote. Each link (contents,
504/// effects, events, start-state objects) is checked below.
505fn verify_epoch_boundary_proof(entry: &EpochInfoV1Entry) -> anyhow::Result<IotaSystemState> {
506    let summary = entry.last_checkpoint_summary.data();
507
508    // 1. Contents hash to the signed summary.
509    anyhow::ensure!(
510        entry.last_checkpoint_contents.digest() == summary.contents_digest,
511        "last_checkpoint_contents does not hash to the signed contents_digest",
512    );
513
514    // 2. The epoch-change effects are the last tx of the verified contents.
515    let expected_execution_digest = entry
516        .last_checkpoint_contents
517        .transactions()
518        .last()
519        .map(|info| ExecutionDigests::new(info.transaction, info.effects))
520        .ok_or_else(|| anyhow::anyhow!("the closing checkpoint has no transactions"))?;
521    let effects = &entry.end_of_epoch_tx_effects;
522    anyhow::ensure!(
523        effects.execution_digests() == expected_execution_digest,
524        "end_of_epoch_tx_effects digest pair does not match the closing checkpoint's last transaction",
525    );
526
527    // 3. Events hash to the effects' events_digest (`None` ⇒ events empty, the
528    // safe-mode boundary case).
529    match effects.events_digest() {
530        Some(events_digest) => anyhow::ensure!(
531            entry.end_of_epoch_tx_events.digest() == *events_digest,
532            "end_of_epoch_tx_events does not hash to the effects' events_digest",
533        ),
534        None => anyhow::ensure!(
535            entry.end_of_epoch_tx_events.is_empty(),
536            "the epoch-change effects carry no events_digest but \
537             end_of_epoch_tx_events is non-empty",
538        ),
539    }
540
541    // 4. Each start-state object's digest is one the effects wrote; `0x5` must
542    // be present. Decode `IotaSystemState` only from these verified bytes.
543    let written: HashSet<ObjectReference> = effects
544        .all_changed_objects()
545        .into_iter()
546        .map(|(changed, _)| *changed.reference())
547        .collect();
548    let mut objects = Vec::with_capacity(entry.next_epoch_start_system_state_objects.len());
549    for raw in &entry.next_epoch_start_system_state_objects {
550        let object: Object = bcs::from_bytes(raw)
551            .map_err(|e| anyhow::anyhow!("decoding a next-epoch start-state object: {e}"))?;
552        anyhow::ensure!(
553            written.contains(&object.object_ref()),
554            "a next-epoch start-state object is not written by the epoch-change tx",
555        );
556        objects.push(object);
557    }
558    anyhow::ensure!(
559        objects
560            .iter()
561            .any(|o| o.id() == IOTA_SYSTEM_STATE_OBJECT_ID),
562        "the next-epoch start-state objects do not include the system-state object 0x5",
563    );
564    get_iota_system_state(&objects.as_slice())
565        .map_err(|e| anyhow::anyhow!("decoding the next-epoch system state: {e}"))
566}
567
568/// Build an `EpochInfoV2` index row from a verified entry, its digest-verified
569/// start system state, and its start checkpoint. The `epoch` comes from the
570/// entry's signed summary; the `end_*` facts are derived from the embedded
571/// entry by `EpochInfoV2`'s methods.
572fn epoch_info_v2_row(
573    entry: EpochInfoV1Entry,
574    system_state: IotaSystemState,
575    start_checkpoint: CheckpointSequenceNumber,
576) -> EpochInfoV2 {
577    EpochInfoV2 {
578        epoch: entry.last_checkpoint_summary.epoch(),
579        start_checkpoint,
580        start_timestamp_ms: system_state.epoch_start_timestamp_ms(),
581        system_state,
582        epoch_close_proof: Some(entry),
583    }
584}
585
586/// Creates a FileMetadata of the provided file path, which is overwritten with
587/// compressed data of the original file.
588pub fn create_file_metadata(
589    file_path: &std::path::Path,
590    file_compression: FileCompression,
591    file_type: FileType,
592    bucket_num: u32,
593    part_num: u32,
594) -> Result<FileMetadata> {
595    // Overwrites the file with compressed data of the original file.
596    file_compression.compress(file_path)?;
597    // Computes the sha3 checksum of the compressed file.
598    let sha3_digest = compute_sha3_checksum(file_path)?;
599    let file_metadata = FileMetadata {
600        file_type,
601        bucket_num,
602        part_num,
603        file_compression,
604        sha3_digest,
605    };
606    Ok(file_metadata)
607}
608
609pub async fn setup_db_state(
610    epoch: u64,
611    state_hash: GlobalStateHash,
612    perpetual_db: Arc<AuthorityPerpetualTables>,
613    checkpoint_store: Arc<CheckpointStore>,
614    committee_store: Arc<CommitteeStore>,
615    verify: bool,
616    num_live_objects: u64,
617    m: MultiProgress,
618) -> Result<()> {
619    // This function should be called once state accumulator based hash verification
620    // is complete and live object set state is downloaded to local store
621    let system_state_object = get_iota_system_state(&perpetual_db)?;
622    let new_epoch_start_state = system_state_object.into_epoch_start_state();
623    let next_epoch_committee = new_epoch_start_state.get_iota_committee();
624    let root_digest: ECMHLiveObjectSetDigest = state_hash.digest().into();
625    let last_checkpoint = checkpoint_store
626        .get_epoch_last_checkpoint(epoch)
627        .expect("Error loading last checkpoint for current epoch")
628        .expect("Could not load last checkpoint for current epoch");
629    let flags = EpochFlag::default_for_no_config();
630    let epoch_start_configuration = EpochStartConfiguration::new(
631        new_epoch_start_state,
632        *last_checkpoint.digest(),
633        &perpetual_db,
634        flags,
635    )
636    .unwrap();
637    perpetual_db.set_epoch_start_configuration(&epoch_start_configuration)?;
638    perpetual_db.insert_root_state_hash(epoch, last_checkpoint.sequence_number, state_hash)?;
639    perpetual_db.set_highest_pruned_checkpoint_without_wb(last_checkpoint.sequence_number)?;
640    committee_store.insert_new_committee(&next_epoch_committee)?;
641    checkpoint_store.update_highest_executed_checkpoint(&last_checkpoint)?;
642
643    if verify {
644        let iter = perpetual_db.iter_live_object_set();
645        let local_digest = ECMHLiveObjectSetDigest::from(
646            accumulate_live_object_iter(Box::new(iter), m.clone(), num_live_objects)
647                .await
648                .digest(),
649        );
650        assert_eq!(
651            root_digest, local_digest,
652            "End of epoch {} root state digest {} does not match \
653                local root state hash {} after restoring db from formal snapshot",
654            epoch, root_digest.digest, local_digest.digest,
655        );
656        println!("DB live object state verification completed successfully!");
657    }
658
659    Ok(())
660}
661
662pub async fn accumulate_live_object_iter(
663    iter: Box<dyn Iterator<Item = LiveObject> + '_>,
664    m: MultiProgress,
665    num_live_objects: u64,
666) -> GlobalStateHash {
667    // Monitor progress of live object accumulation, whose position the ticker
668    // takes from the counter the loop below advances
669    let accum_counter = Arc::new(AtomicU64::new(0));
670    let accum_ticker = ProgressTicker::spawn(
671        m.add(
672            ProgressBar::new(num_live_objects).with_style(
673                ProgressStyle::with_template(
674                    "[{elapsed_precise}] {wide_bar} Accumulating DB live objects: {pos}/{len} \
675                     ({per_sec}, ETA {eta})",
676                )
677                .unwrap(),
678            ),
679        ),
680        "Accumulating DB live objects",
681        ProgressUnit::Count("objects"),
682        Some(accum_counter.clone()),
683    );
684
685    // Accumulate live objects
686    let mut acc = GlobalStateHash::default();
687    for live_object in iter {
688        GlobalStateHasher::accumulate_live_object(&mut acc, &live_object);
689        accum_counter.fetch_add(1, Ordering::Relaxed);
690    }
691    let num_accumulated = accum_counter.load(Ordering::Relaxed);
692    assert!(
693        num_accumulated <= num_live_objects,
694        "Accumulated more objects ({num_accumulated}) than expected ({num_live_objects})"
695    );
696    accum_ticker.finish_with_message("DB live object accumulation completed");
697    acc
698}
iota_snapshot/lib.rs

iota_snapshot/
lib.rs
