feat(export): add --canonical flag for byte-reproducible snapshots

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<PoolHash>,
+    pub registered_pools: BTreeSet<PoolHash>,
 
     #[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<PoolHash>,
+    pub(crate) registered_pools: BTreeSet<PoolHash>,
     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<PoolHash>,
+    pub(crate) prev_registered_pools: BTreeSet<PoolHash>,
 }
 
 impl dolos_core::EntityDelta for EpochStatsUpdate {

crates/cardano/src/rupd/work_unit.rs

@@ -121,7 +121,6 @@ impl RupdWorkUnit {
         }
         self.rewards = None;
     }
-
 }
 
 impl<D> WorkUnit<D> for RupdWorkUnit
@@ -193,11 +192,7 @@ where
 
         work.merge_shard::<D>(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(),

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<Path>) -> 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.

crates/fjall/src/lib.rs

@@ -50,3 +50,4 @@ impl From<Error> for StateError {
         StateError::InternalStoreError(error.to_string())
     }
 }
+

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<Path>) -> 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.