diff --git a/Cargo.lock b/Cargo.lock index 0258ee806..0967c00f6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1592,7 +1592,7 @@ dependencies = [ "hex", "itertools 0.14.0", "pallas", - "rand 0.9.1", + "rand 0.9.4", "tokio", "tokio-stream", ] @@ -1612,7 +1612,7 @@ dependencies = [ "opentelemetry", "opentelemetry_sdk", "pallas", - "rand 0.9.1", + "rand 0.9.4", "serde", "serde_json", "thiserror 2.0.18", @@ -1786,7 +1786,7 @@ dependencies = [ "flume 0.12.0", "log", "lsm-tree", - "lz4_flex", + "lz4_flex 0.11.5", "tempfile", "xxhash-rust", ] @@ -2860,9 +2860,9 @@ checksum = "13dc2df351e3202783a1fe0d44375f7295ffb4049267b0f3018346dc122a1d94" [[package]] name = "lsm-tree" -version = "3.1.0" +version = "3.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fc5fa40c207eed45c811085aaa1b0a25fead22e298e286081cd4b98785fe759b" +checksum = "8ef86c3c797c10eefcc73407c43ae48c19d4df686131a8334b2895a513e91df4" dependencies = [ "byteorder-lite", "byteview", @@ -2870,7 +2870,7 @@ dependencies = [ "enum_dispatch", "interval-heap", "log", - "lz4_flex", + "lz4_flex 0.13.1", "quick_cache", "rustc-hash", "self_cell", @@ -2889,13 +2889,19 @@ dependencies = [ "twox-hash", ] +[[package]] +name = "lz4_flex" +version = "0.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7ef0d4ed8669f8f8826eb00dc878084aa8f253506c4fd5e8f58f5bce72ddb97e" + [[package]] name = "matchers" -version = "0.1.0" +version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8263075bb86c5a1b1427b5ae862e8889656f126e9f77c484496e8b47cf5c5558" +checksum = "d1525a2a28c7f4fa0fc98bb91ae755d1e2d1505079e05539e35bc876b5d65ae9" dependencies = [ - "regex-automata 0.1.10", + "regex-automata 0.4.9", ] [[package]] @@ -3203,12 +3209,11 @@ dependencies = [ [[package]] name = "nu-ansi-term" -version = "0.46.0" +version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "77a8165726e8236064dbb45459242600304b42a5ea24ee2948e18e023bf7ba84" +checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "overload", - "winapi", + "windows-sys 0.61.2", ] [[package]] @@ -3449,19 +3454,13 @@ dependencies = [ "futures-util", "opentelemetry", "percent-encoding", - "rand 0.9.1", + "rand 0.9.4", "serde_json", "thiserror 2.0.18", "tokio", "tokio-stream", ] -[[package]] -name = "overload" -version = "0.1.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b15813163c1d831bf4a13c3610c05c0d03b39feb07f7e09fa234dac9b15aaf39" - [[package]] name = "owo-colors" version = "4.2.0" @@ -3960,7 +3959,7 @@ dependencies = [ "hmac", "md-5", "memchr", - "rand 0.9.1", + "rand 0.9.4", "sha2", "stringprep", ] @@ -4021,7 +4020,7 @@ dependencies = [ "bitflags 2.9.0", "lazy_static", "num-traits", - "rand 0.9.1", + "rand 0.9.4", "rand_chacha 0.9.0", "rand_xorshift", "regex-syntax 0.8.5", @@ -4173,7 +4172,7 @@ checksum = "b820744eb4dc9b57a3398183639c511b5a26d2ed702cedd3febaa1393caa22cc" dependencies = [ "bytes", "getrandom 0.3.2", - "rand 0.9.1", + "rand 0.9.4", "ring", "rustc-hash", "rustls", @@ -4233,9 +4232,9 @@ dependencies = [ [[package]] name = "rand" -version = "0.9.1" +version = "0.9.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9fbfd9d094a40bf3ae768db9361049ace4c0e04a4fd6b359518bd7b73a73dd97" +checksum = "44c5af06bb1b7d3216d91932aed5265164bf384dc89cd6ba05cf59a35f5f76ea" dependencies = [ "rand_chacha 0.9.0", "rand_core 0.9.3", @@ -4402,15 +4401,6 @@ dependencies = [ "regex-syntax 0.8.5", ] -[[package]] -name = "regex-automata" -version = "0.1.10" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6c230d73fb8d8c1b9c0b3135c5142a8acee3a0558fb8db5cf1cb65f8d7862132" -dependencies = [ - "regex-syntax 0.6.29", -] - [[package]] name = "regex-automata" version = "0.3.9" @@ -4439,12 +4429,6 @@ version = "0.1.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cab834c73d247e67f4fae452806d17d3c7501756d98c8808d7c9c7aa7d18f973" -[[package]] -name = "regex-syntax" -version = "0.6.29" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f162c6dd7b008981e4d40210aca20b4bd0f9b60ca9271061b07f78537722f2e1" - [[package]] name = "regex-syntax" version = "0.7.5" @@ -5187,9 +5171,9 @@ checksum = "7da8b5736845d9f2fcb837ea5d9e2628564b3b043a70948a3f0b778838c5fb4f" [[package]] name = "strum" -version = "0.27.1" +version = "0.27.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f64def088c51c9510a8579e3c5d67c65349dcf755e5479ad3d010aa6454e2c32" +checksum = "af23d6f6c1a224baef9d3f61e287d2761385a5b88fdab4eb4c6f11aeb54c4bcf" dependencies = [ "strum_macros", ] @@ -5529,7 +5513,7 @@ dependencies = [ "pin-project-lite", "postgres-protocol", "postgres-types", - "rand 0.9.1", + "rand 0.9.4", "socket2 0.6.3", "tokio", "tokio-util", @@ -5908,14 +5892,14 @@ dependencies = [ [[package]] name = "tracing-subscriber" -version = "0.3.19" +version = "0.3.20" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e8189decb5ac0fa7bc8b96b7cb9b2701d60d48805aca84a238004d665fcc4008" +checksum = "2054a14f5307d601f88daf0553e1cbf472acc4f2c51afab632431cdcd72124d5" dependencies = [ "matchers", "nu-ansi-term", "once_cell", - "regex", + "regex-automata 0.4.9", "sharded-slab", "smallvec", "thread_local", diff --git a/crates/cardano/src/model/epochs.rs b/crates/cardano/src/model/epochs.rs index c7e547178..d6b6a85e7 100644 --- a/crates/cardano/src/model/epochs.rs +++ b/crates/cardano/src/model/epochs.rs @@ -1,4 +1,6 @@ -use std::{collections::HashSet, sync::Arc}; +// BTreeSet (not HashSet) so the CBOR encoding of `registered_pools` is in +// sorted key order and therefore byte-deterministic across independent syncs. +use std::{collections::BTreeSet, sync::Arc}; use dolos_core::{BlockSlot, EntityKey, Genesis, NsKey}; use pallas::{ @@ -105,7 +107,7 @@ pub struct RollingStats { pub withdrawals: Lovelace, #[n(8)] - pub registered_pools: HashSet, + pub registered_pools: BTreeSet, #[n(13)] pub blocks_minted: u32, @@ -430,7 +432,7 @@ pub struct EpochStatsUpdate { pub(crate) new_accounts: u64, pub(crate) removed_accounts: u64, pub(crate) withdrawals: u64, - pub(crate) registered_pools: HashSet, + pub(crate) registered_pools: BTreeSet, pub(crate) drep_deposits: Lovelace, pub(crate) proposal_deposits: Lovelace, pub(crate) drep_refunds: Lovelace, @@ -442,7 +444,7 @@ pub struct EpochStatsUpdate { // undo: did apply create rolling.live from default? Plus the pre-union pool set, which // can't be recovered by set subtraction (a pool in both prev and self would be removed). pub(crate) was_new: bool, - pub(crate) prev_registered_pools: HashSet, + pub(crate) prev_registered_pools: BTreeSet, } impl dolos_core::EntityDelta for EpochStatsUpdate { diff --git a/crates/cardano/src/rupd/work_unit.rs b/crates/cardano/src/rupd/work_unit.rs index 5eeaeee1a..53ba5d501 100644 --- a/crates/cardano/src/rupd/work_unit.rs +++ b/crates/cardano/src/rupd/work_unit.rs @@ -121,7 +121,6 @@ impl RupdWorkUnit { } self.rewards = None; } - } impl WorkUnit for RupdWorkUnit @@ -193,11 +192,7 @@ where work.merge_shard::(domain.state(), ranges)?; - info!( - epoch = work.current_epoch, - shard = shard_index, - "rupd" - ); + info!(epoch = work.current_epoch, shard = shard_index, "rupd"); Ok(()) } @@ -249,7 +244,8 @@ where for (credential, reward) in rewards.iter_pending() { let key = credential_to_key(credential); - let (as_leader, as_delegator) = match reward { + type PoolRewards = Vec<(PoolHash, u64)>; + let (mut as_leader, mut as_delegator): (PoolRewards, PoolRewards) = match reward { Reward::MultiPool(r) => ( r.leader_rewards().collect(), r.delegator_rewards().collect(), @@ -264,6 +260,10 @@ where } }; + // HashMap iteration order varies; sort so CBOR encoding is stable across syncs. + as_leader.sort_unstable_by_key(|(pool, _)| *pool); + as_delegator.sort_unstable_by_key(|(pool, _)| *pool); + let state = PendingRewardState { credential: credential.clone(), is_spendable: reward.is_spendable(), diff --git a/crates/fjall/src/index/mod.rs b/crates/fjall/src/index/mod.rs index b73ada492..43bc6511c 100644 --- a/crates/fjall/src/index/mod.rs +++ b/crates/fjall/src/index/mod.rs @@ -227,6 +227,54 @@ impl IndexStore { tracing::info!("index store: graceful shutdown complete"); Ok(()) } + + /// Copy all keyspaces into a fresh fjall database at `dest_path` and major-compact each one. + pub fn rebuild_canonical(&self, dest_path: impl AsRef) -> Result<(), Error> { + let snapshot = self.db.snapshot(); + + let fresh_db = Database::builder(dest_path.as_ref()).open()?; + let fresh_cursor = + fresh_db.keyspace(keyspace_names::CURSOR, KeyspaceCreateOptions::default)?; + let fresh_exact = + fresh_db.keyspace(keyspace_names::EXACT, KeyspaceCreateOptions::default)?; + let fresh_utxo_tags = + fresh_db.keyspace(keyspace_names::UTXO_TAGS, KeyspaceCreateOptions::default)?; + let fresh_block_tags = + fresh_db.keyspace(keyspace_names::BLOCK_TAGS, KeyspaceCreateOptions::default)?; + + let mut batch = fresh_db.batch(); + for guard in snapshot.iter(&self.cursor) { + let (k, v) = guard.into_inner()?; + batch.insert(&fresh_cursor, k.as_ref(), v.as_ref()); + } + for guard in snapshot.iter(&self.exact) { + let (k, v) = guard.into_inner()?; + batch.insert(&fresh_exact, k.as_ref(), v.as_ref()); + } + for guard in snapshot.iter(&self.utxo_tags) { + let (k, v) = guard.into_inner()?; + batch.insert(&fresh_utxo_tags, k.as_ref(), v.as_ref()); + } + for guard in snapshot.iter(&self.block_tags) { + let (k, v) = guard.into_inner()?; + batch.insert(&fresh_block_tags, k.as_ref(), v.as_ref()); + } + batch.commit()?; + + for ks in [ + &fresh_cursor, + &fresh_exact, + &fresh_utxo_tags, + &fresh_block_tags, + ] { + ks.rotate_memtable_and_wait()?; + ks.major_compact()?; + } + + fresh_db.persist(PersistMode::SyncAll)?; + + Ok(()) + } } /// Writer for batched index operations. diff --git a/crates/fjall/src/lib.rs b/crates/fjall/src/lib.rs index beb7fb10b..1ae150ad5 100644 --- a/crates/fjall/src/lib.rs +++ b/crates/fjall/src/lib.rs @@ -50,3 +50,4 @@ impl From for StateError { StateError::InternalStoreError(error.to_string()) } } + diff --git a/crates/fjall/src/state/mod.rs b/crates/fjall/src/state/mod.rs index 172f8f163..bb55f2ebe 100644 --- a/crates/fjall/src/state/mod.rs +++ b/crates/fjall/src/state/mod.rs @@ -195,6 +195,43 @@ impl StateStore { tracing::info!("state store: graceful shutdown complete"); Ok(()) } + + /// Copy all keyspaces into a fresh fjall database at `dest_path` and major-compact each one. + pub fn rebuild_canonical(&self, dest_path: impl AsRef) -> Result<(), Error> { + let snapshot = self.db.snapshot(); + + let fresh_db = Database::builder(dest_path.as_ref()).open()?; + let fresh_cursor = + fresh_db.keyspace(keyspace_names::CURSOR, KeyspaceCreateOptions::default)?; + let fresh_utxos = + fresh_db.keyspace(keyspace_names::UTXOS, KeyspaceCreateOptions::default)?; + let fresh_entities = + fresh_db.keyspace(keyspace_names::ENTITIES, KeyspaceCreateOptions::default)?; + + let mut batch = fresh_db.batch(); + for guard in snapshot.iter(&self.cursor) { + let (k, v) = guard.into_inner()?; + batch.insert(&fresh_cursor, k.as_ref(), v.as_ref()); + } + for guard in snapshot.iter(&self.utxos) { + let (k, v) = guard.into_inner()?; + batch.insert(&fresh_utxos, k.as_ref(), v.as_ref()); + } + for guard in snapshot.iter(&self.entities) { + let (k, v) = guard.into_inner()?; + batch.insert(&fresh_entities, k.as_ref(), v.as_ref()); + } + batch.commit()?; + + for ks in [&fresh_cursor, &fresh_utxos, &fresh_entities] { + ks.rotate_memtable_and_wait()?; + ks.major_compact()?; + } + + fresh_db.persist(PersistMode::SyncAll)?; + + Ok(()) + } } /// Writer for batched state operations. diff --git a/crates/redb3/src/archive/mod.rs b/crates/redb3/src/archive/mod.rs index bde547df9..fe8413ec5 100644 --- a/crates/redb3/src/archive/mod.rs +++ b/crates/redb3/src/archive/mod.rs @@ -1,4 +1,4 @@ -use ::redb::{Database, ReadableDatabase}; +use ::redb::{Database, ReadableDatabase, ReadableMultimapTable as _}; use redb::ReadTransaction; use std::{ collections::HashMap, @@ -109,6 +109,12 @@ impl From for RedbArchiveError { } } +impl From<::redb::CompactionError> for RedbArchiveError { + fn from(value: ::redb::CompactionError) -> Self { + Self(ArchiveError::InternalError(value.to_string())) + } +} + const DEFAULT_CACHE_SIZE_MB: usize = 500; #[derive(Clone)] @@ -764,6 +770,63 @@ impl ArchiveStore { Ok(done) } + /// Write all tables in key order into a fresh redb file at `dest_index_path`, then compact. + pub fn rebuild_index_to( + &self, + dest_index_path: impl AsRef, + ) -> Result<(), RedbArchiveError> { + let mut fresh = Database::builder().create(dest_index_path.as_ref())?; + + let src_rx = self.db().begin_read()?; + let dst_wx = fresh.begin_write()?; + + // BlocksTable: u64 → &[u8] + { + let src_t = src_rx.open_table(tables::BlocksTable::DEF)?; + let mut dst_t = dst_wx.open_table(tables::BlocksTable::DEF)?; + for entry in src_t.range(0u64..)? { + let (k, v) = entry?; + dst_t.insert(k.value(), v.value())?; + } + } + + // Entity log tables from schema: both &[u8] keyed + let mut sorted: Vec<_> = self.tables.iter().collect(); + sorted.sort_by_key(|(ns, _)| *ns); + for (_, table) in sorted { + match table { + Table::Value(def) => { + let src_t = src_rx.open_table(*def)?; + let mut dst_t = dst_wx.open_table(*def)?; + for entry in src_t.range::<&[u8]>(..)? { + let (k, v) = entry?; + dst_t.insert(k.value(), v.value())?; + } + } + Table::MultiValue(def) => { + let src_t = src_rx.open_multimap_table(*def)?; + let mut dst_t = dst_wx.open_multimap_table(*def)?; + for entry in src_t.iter()? { + let (k, vals) = entry?; + for val in vals { + let val = val?; + dst_t.insert(k.value(), val.value())?; + } + } + } + } + } + + indexes::Indexes::copy(&src_rx, &dst_wx)?; + + dst_wx.commit()?; + + // Loop-compact the fresh file until the B-tree is fully packed. + while fresh.compact()? {} + + Ok(()) + } + fn truncate_front(&self, after: &ChainPoint) -> Result<(), RedbArchiveError> { let mut wx = self.db().begin_write()?; wx.set_quick_repair(true); diff --git a/src/bin/dolos/data/export.rs b/src/bin/dolos/data/export.rs index 0a9abe962..5be8af5e0 100644 --- a/src/bin/dolos/data/export.rs +++ b/src/bin/dolos/data/export.rs @@ -1,9 +1,9 @@ use clap::Parser; -use dolos::storage::ArchiveStoreBackend; +use dolos::storage::{ArchiveStoreBackend, IndexStoreBackend, StateStoreBackend}; use dolos_core::config::RootConfig; use flate2::write::GzEncoder; use flate2::Compression; -use miette::{bail, IntoDiagnostic as _}; +use miette::{bail, Context as _, IntoDiagnostic as _}; use std::ffi::OsStr; use std::fs::File; use std::path::{Path, PathBuf}; @@ -30,6 +30,24 @@ pub struct Args { /// Skip the compact and integrity check of the archive database #[arg(long, action)] skip_sanitization: bool, + + /// Rebuild stores into canonical form before archiving. + /// + /// Each store is re-written from a snapshot into a fresh database so that + /// two exports of the same chain data produce identical exports. The live + /// stores are not modified. Mutually exclusive with --skip-sanitization. + /// + /// Byte-identity guarantee by backend: + /// archive/ (redb) — byte-for-byte identical across independent runs. + /// state/ (fjall) — content-identical (same keys/values), but NOT + /// bit-for-bit identical: Fjall SSTable files embed + /// wall-clock timestamps that differ between runs. + /// index/ (fjall) — same caveat as state/. + /// + /// Track the upstream Fjall fix at: + /// https://github.com/fjall-rs/lsm-tree/issues/296 + #[arg(long, action, conflicts_with = "skip_sanitization")] + canonical: bool, } fn is_macos_metadata(path: &Path) -> bool { @@ -130,6 +148,49 @@ fn append_dir_filtered( Ok(()) } +/// Tar `archive/`, replacing `archive/index` with `canonical_index`. +fn append_archive_with_canonical_index( + tar: &mut Builder>, + source_dir: &Path, + tar_name: &Path, + canonical_index: &Path, +) -> miette::Result<()> { + let header = deterministic_header(source_dir)?; + tar.append_data(&mut header.clone(), tar_name, &[] as &[u8]) + .into_diagnostic()?; + + let mut entries: Vec<_> = std::fs::read_dir(source_dir) + .into_diagnostic()? + .collect::, _>>() + .into_diagnostic()?; + + entries.sort_by_key(|e| e.file_name()); + + for entry in entries { + let path = entry.path(); + + if is_macos_metadata(&path) { + continue; + } + + let entry_name = tar_name.join(entry.file_name()); + + if entry.file_name() == OsStr::new("index") { + // Substitute the live index with the rebuilt canonical version. + let mut header = deterministic_header(canonical_index)?; + header.set_size(std::fs::metadata(canonical_index).into_diagnostic()?.len()); + header.set_cksum(); + let file = File::open(canonical_index).into_diagnostic()?; + tar.append_data(&mut header.clone(), &entry_name, file) + .into_diagnostic()?; + } else { + append_path_filtered(tar, &path, &entry_name)?; + } + } + + Ok(()) +} + fn prepare_archive( archive: &mut dolos_redb3::archive::ArchiveStore, pb: &crate::feedback::ProgressBar, @@ -144,6 +205,105 @@ fn prepare_archive( Ok(()) } +/// Rebuild each requested store into a fresh canonical copy. +/// +/// Temp paths are registered in `CanonicalPaths` as soon as they are created, so `Drop` +/// removes them even when a later rebuild step fails. +fn rebuild_stores_canonical( + stores: &crate::common::Stores, + root: &Path, + args: &Args, + pb: &crate::feedback::ProgressBar, +) -> miette::Result { + let mut paths = CanonicalPaths::default(); + + if args.include_archive { + match &stores.archive { + ArchiveStoreBackend::Redb(s) => { + // Keep temp file outside archive/ so it is never swept into the tar. + let dest = root.join("archive_index.canonical.tmp"); + let _ = std::fs::remove_file(&dest); // remove stale from a previous failed run + paths.archive_index = Some(dest); + pb.set_message("rebuilding archive index"); + s.rebuild_index_to(paths.archive_index.as_ref().unwrap()) + .into_diagnostic() + .context("rebuild archive index")?; + } + ArchiveStoreBackend::NoOp(_) => { + bail!("--canonical is not available for noop archive backend") + } + } + } + + if args.include_state { + match &stores.state { + StateStoreBackend::Fjall(s) => { + pb.println( + "warning: --canonical: state/ will be content-identical but NOT \ + bit-for-bit identical — Fjall SSTable files embed wall-clock timestamps \ + that differ between runs (https://github.com/fjall-rs/lsm-tree/issues/296)", + ); + let dest = root.join("state.canonical.tmp"); + let _ = std::fs::remove_dir_all(&dest); // remove stale from a previous failed run + paths.state = Some(dest); + pb.set_message("rebuilding state store"); + s.rebuild_canonical(paths.state.as_ref().unwrap()) + .into_diagnostic() + .context("rebuild state store")?; + } + StateStoreBackend::Redb(_) => { + bail!("--canonical is not yet supported for Redb state backend") + } + } + } + + if args.include_indexes { + match &stores.indexes { + IndexStoreBackend::Fjall(s) => { + pb.println( + "warning: --canonical: index/ will be content-identical but NOT \ + bit-for-bit identical — Fjall SSTable files embed wall-clock timestamps \ + that differ between runs (https://github.com/fjall-rs/lsm-tree/issues/296)", + ); + let dest = root.join("index.canonical.tmp"); + let _ = std::fs::remove_dir_all(&dest); // remove stale from a previous failed run + paths.index = Some(dest); + pb.set_message("rebuilding index store"); + s.rebuild_canonical(paths.index.as_ref().unwrap()) + .into_diagnostic() + .context("rebuild index store")?; + } + IndexStoreBackend::Redb(_) => { + bail!("--canonical is not yet supported for Redb index backend") + } + IndexStoreBackend::NoOp(_) => {} + } + } + + Ok(paths) +} + +#[derive(Default)] +struct CanonicalPaths { + archive_index: Option, + state: Option, + index: Option, +} + +impl Drop for CanonicalPaths { + fn drop(&mut self) { + if let Some(p) = self.archive_index.take() { + let _ = std::fs::remove_file(&p); + } + if let Some(p) = self.state.take() { + let _ = std::fs::remove_dir_all(&p); + } + if let Some(p) = self.index.take() { + let _ = std::fs::remove_dir_all(&p); + } + } +} + pub fn run( config: &RootConfig, args: &Args, @@ -153,53 +313,98 @@ pub fn run( let export_file = File::create(&args.output).into_diagnostic()?; let encoder = GzEncoder::new(export_file, Compression::default()); - let mut archive = Builder::new(encoder); + let mut tar = Builder::new(encoder); let mut stores = crate::common::open_data_stores(config)?; let root = crate::common::ensure_storage_path(config)?; - // prepare_archive requires direct redb access - match &mut stores.archive { - ArchiveStoreBackend::Redb(s) if !args.skip_sanitization => prepare_archive(s, &pb)?, - ArchiveStoreBackend::Redb(_) => {} - ArchiveStoreBackend::NoOp(_) => { - bail!("export command is not available for noop archive backend") + if args.canonical { + // Rebuild stores before shutdown so we can snapshot them. + let canonical = rebuild_stores_canonical(&stores, &root, args, &pb)?; + + // Shut down all live stores before accessing their files. + stores.wal.shutdown().into_diagnostic()?; + stores.state.shutdown().into_diagnostic()?; + stores.archive.shutdown().into_diagnostic()?; + stores.indexes.shutdown().into_diagnostic()?; + drop(stores); + + let result = (|| { + if args.include_archive { + pb.set_message("archiving (canonical)"); + let archive_dir = root.join("archive"); + match &canonical.archive_index { + Some(idx) => { + append_archive_with_canonical_index( + &mut tar, + &archive_dir, + Path::new("archive"), + idx, + )?; + } + None => { + append_dir_filtered(&mut tar, &archive_dir, Path::new("archive"))?; + } + } + } + + if args.include_state { + pb.set_message("archiving state (canonical)"); + let live_state = root.join("state"); + let state_src = canonical.state.as_deref().unwrap_or(&live_state); + append_dir_filtered(&mut tar, state_src, Path::new("state"))?; + } + + if args.include_indexes { + pb.set_message("archiving indexes (canonical)"); + let live_index = root.join("index"); + let index_src = canonical.index.as_deref().unwrap_or(&live_index); + append_dir_filtered(&mut tar, index_src, Path::new("index"))?; + } + + tar.finish().into_diagnostic() + })(); + + drop(canonical); // clean up temp dirs/files before propagating any error + result?; + } else { + // Non-canonical path: compact + integrity check archive, then tar live stores. + match &mut stores.archive { + ArchiveStoreBackend::Redb(s) if !args.skip_sanitization => prepare_archive(s, &pb)?, + ArchiveStoreBackend::Redb(_) => {} + ArchiveStoreBackend::NoOp(_) => { + bail!("export command is not available for noop archive backend") + } } - } - - // Ensure all stores flush pending work and release filesystem state before - // we archive their on-disk files. - stores.wal.shutdown().into_diagnostic()?; - stores.state.shutdown().into_diagnostic()?; - stores.archive.shutdown().into_diagnostic()?; - stores.indexes.shutdown().into_diagnostic()?; - drop(stores); - - if args.include_archive { - let path = root.join("archive"); - append_path_filtered(&mut archive, &path, Path::new("archive"))?; - - pb.set_message("creating archive"); - } - - if args.include_state { - let path = root.join("state"); - - append_path_filtered(&mut archive, &path, Path::new("state"))?; - - pb.set_message("creating archive"); - } + // Ensure all stores flush pending work and release filesystem state before + // we archive their on-disk files. + stores.wal.shutdown().into_diagnostic()?; + stores.state.shutdown().into_diagnostic()?; + stores.archive.shutdown().into_diagnostic()?; + stores.indexes.shutdown().into_diagnostic()?; + drop(stores); + + if args.include_archive { + let path = root.join("archive"); + append_path_filtered(&mut tar, &path, Path::new("archive"))?; + pb.set_message("creating archive"); + } - if args.include_indexes { - let path = root.join("index"); + if args.include_state { + let path = root.join("state"); + append_path_filtered(&mut tar, &path, Path::new("state"))?; + pb.set_message("creating archive"); + } - append_path_filtered(&mut archive, &path, Path::new("index"))?; + if args.include_indexes { + let path = root.join("index"); + append_path_filtered(&mut tar, &path, Path::new("index"))?; + pb.set_message("creating archive"); + } - pb.set_message("creating archive"); + tar.finish().into_diagnostic()?; } - archive.finish().into_diagnostic()?; - Ok(()) }