feat(worker): handle L1 reorgs in ASM block sync

bin/asm-runner/src/storage.rs

@@ -3,17 +3,17 @@
 use std::sync::Arc;
 
 use anyhow::Result;
-use asm_storage::{AsmStateDb, ExportEntriesDb, MmrDb};
+use asm_storage::{AsmManifestMmrDb, AsmStateDb, ExportEntriesDb};
 
 use crate::config::DatabaseConfig;
 
 /// Create storage backends for the ASM runner.
 pub(crate) fn create_storage(
     config: &DatabaseConfig,
-) -> Result<(Arc<AsmStateDb>, Arc<MmrDb>, ExportEntriesDb)> {
+) -> Result<(Arc<AsmStateDb>, Arc<AsmManifestMmrDb>, ExportEntriesDb)> {
     let db = sled::open(&config.path)?;
     let state_db = Arc::new(AsmStateDb::open(&db)?);
-    let mmr_db = Arc::new(MmrDb::open(&db)?);
+    let mmr_db = Arc::new(AsmManifestMmrDb::open(&db)?);
     let export_entries_db = ExportEntriesDb::open(&db)?;
     Ok((state_db, mmr_db, export_entries_db))
 }

bin/asm-runner/src/worker_context.rs

@@ -17,8 +17,8 @@
 
 use std::sync::Arc;
 
-use asm_storage::{AsmStateDb, ExportEntriesDb, MmrDb};
-use bitcoin::{Block, BlockHash, Network};
+use asm_storage::{AsmManifestMmrDb, AsmStateDb, ExportEntriesDb};
+use bitcoin::{Block, BlockHash, Network, block::Header};
 use bitcoind_async_client::{Client, error::ClientError, traits::Reader};
 use moho_runtime_interface::MohoProgram;
 use moho_types::{ExportState, MohoState};
@@ -61,7 +61,7 @@ pub(crate) struct AsmWorkerContext {
     /// Maximum retry attempts per Bitcoin RPC call.
     rpc_max_retries: u16,
     state_db: Arc<AsmStateDb>,
-    mmr_db: Arc<MmrDb>,
+    mmr_db: Arc<AsmManifestMmrDb>,
     export_entries_db: Option<ExportEntriesDb>,
     moho_storage: Option<MohoStorage>,
     /// L1 height of the chain genesis (anchor) block.
@@ -78,7 +78,7 @@ impl AsmWorkerContext {
         bitcoin_client: Arc<Client>,
         retry: &RetryConfig,
         state_db: Arc<AsmStateDb>,
-        mmr_db: Arc<MmrDb>,
+        mmr_db: Arc<AsmManifestMmrDb>,
         export_entries_db: Option<ExportEntriesDb>,
         moho_storage: Option<MohoStorage>,
         genesis_height: u64,
@@ -152,6 +152,21 @@ impl L1DataProvider for AsmWorkerContext {
             .map_err(|e: ClientError| WorkerError::BtcRpc(format!("get_block({block_hash}): {e}")))
     }
 
+    fn get_l1_block_header(&self, blockid: &L1BlockId) -> WorkerResult<Header> {
+        let block_hash: BlockHash = blockid.to_block_hash();
+        let client = &self.bitcoin_client;
+        self.runtime_handle
+            .block_on(retry_with_backoff_async(
+                "btc_get_block_header",
+                self.rpc_max_retries,
+                &self.rpc_backoff,
+                || async { client.get_block_header(&block_hash).await },
+            ))
+            .map_err(|e: ClientError| {
+                WorkerError::BtcRpc(format!("get_block_header({block_hash}): {e}"))
+            })
+    }
+
     fn get_network(&self) -> WorkerResult<Network> {
         let client = &self.bitcoin_client;
         self.runtime_handle
@@ -242,14 +257,9 @@ impl ManifestMmrStore for AsmWorkerContext {
     }
 
     fn put_manifest_hash(&self, height: u64, hash: AsmManifestHash) -> WorkerResult<()> {
-        let index = self
-            .mmr_db
-            .append_leaf(hash.into())
-            .map_err(|_| WorkerError::DbError)?;
-        if index != height {
-            return Err(WorkerError::ManifestMmrMisaligned { height, index });
-        }
-        Ok(())
+        self.mmr_db
+            .put_leaf(height, hash)
+            .map_err(|_| WorkerError::DbError)
     }
 
     fn manifest_mmr_leaf_count(&self) -> WorkerResult<u64> {
@@ -270,7 +280,6 @@ impl ManifestMmrStore for AsmWorkerContext {
         self.mmr_db
             .get_leaf(index)
             .map_err(|_| WorkerError::DbError)?
-            .map(AsmManifestHash::from)
             .ok_or(WorkerError::ManifestHashNotFound { index })
     }
 }

crates/stf/src/transition.rs

@@ -85,6 +85,12 @@ pub fn compute_asm_transition<S: AsmSpec>(
     // 8. Append the manifest to the history accumulator
     history_accumulator.add_manifest(&manifest)?;
 
+    // Sanity check
+    assert_eq!(
+        history_accumulator.last_inserted_height(),
+        pow_state.last_verified_block.height as u64
+    );
+
     // 9. Construct the final `AnchorState` and output.
     let chain_view = ChainViewState {
         pow_state,

crates/storage/src/lib.rs

@@ -5,13 +5,13 @@
 //!
 //! Three storage backends:
 //! - [`AsmStateDb`] — anchor states + aux data, keyed by L1 block commitment
-//! - [`MmrDb`] — manifest hash MMR (append, prove, query)
+//! - [`AsmManifestMmrDb`] — manifest hash MMR (append, prove, query)
 //! - [`ExportEntriesDb`] — per-container export entries, indexed for proof generation
 
 mod export_entries;
 mod mmr;
 mod state;
 
 pub use export_entries::ExportEntriesDb;
-pub use mmr::MmrDb;
+pub use mmr::AsmManifestMmrDb;
 pub use state::AsmStateDb;

crates/storage/src/mmr.rs

@@ -5,7 +5,7 @@
 //! walking the stored sibling path — no replay of the whole MMR from leaf 0.
 
 use anyhow::Result;
-use strata_identifiers::Buf32;
+use strata_asm_common::AsmManifestHash;
 use strata_merkle::{MerkleProofB32, Sha256Hasher};
 use strata_merkle_node_store::{MmrNodeStore, NodePos, StoredMmr};
 
@@ -25,11 +25,11 @@ fn decode_node(value: sled::IVec) -> [u8; 32] {
 /// One MMR per store, so nodes are keyed directly by [`NodePos::to_key`] with
 /// no namespacing.
 #[derive(Debug, Clone)]
-struct ManifestNodes {
+struct AsmManifestMmrNodeStore {
     nodes: sled::Tree,
 }
 
-impl MmrNodeStore for ManifestNodes {
+impl MmrNodeStore for AsmManifestMmrNodeStore {
     type Hash = [u8; 32];
     type Error = sled::Error;
 
@@ -59,15 +59,15 @@ impl MmrNodeStore for ManifestNodes {
 /// from the stored sibling path and verify against the compact-peaks
 /// accumulators the rest of the system already holds.
 #[derive(Debug, Clone)]
-pub struct MmrDb {
-    inner: ManifestNodes,
+pub struct AsmManifestMmrDb {
+    inner: AsmManifestMmrNodeStore,
 }
 
-impl MmrDb {
+impl AsmManifestMmrDb {
     /// Opens or creates the MMR node tree in the given sled instance.
     pub fn open(db: &sled::Db) -> Result<Self> {
         Ok(Self {
-            inner: ManifestNodes {
+            inner: AsmManifestMmrNodeStore {
                 nodes: db.open_tree("mmr_nodes")?,
             },
         })
@@ -78,14 +78,20 @@ impl MmrDb {
         Ok(StoredMmr::<Sha256Hasher>::leaf_count(&self.inner)?)
     }
 
-    /// Appends a manifest hash as a new leaf. Returns the leaf index.
-    pub fn append_leaf(&self, hash: Buf32) -> Result<u64> {
-        Ok(StoredMmr::<Sha256Hasher>::append_leaf(&self.inner, hash.0)?)
+    /// Writes a manifest `hash` as the leaf at `height`.
+    ///
+    /// The MMR is height-indexed, so the leaf for the L1 block at `height`
+    /// lands at leaf index `height`. `height` must be the current end (an
+    /// append) or an existing index (an overwrite); a gap past the end is
+    /// rejected.
+    pub fn put_leaf(&self, height: u64, hash: AsmManifestHash) -> Result<()> {
+        StoredMmr::<Sha256Hasher>::put_leaf(&self.inner, height, *hash.as_ref())?;
+        Ok(())
     }
 
     /// Retrieves a manifest hash by its leaf index.
-    pub fn get_leaf(&self, index: u64) -> Result<Option<Buf32>> {
-        Ok(StoredMmr::<Sha256Hasher>::get_leaf(&self.inner, index)?.map(Buf32::new))
+    pub fn get_leaf(&self, index: u64) -> Result<Option<AsmManifestHash>> {
+        Ok(StoredMmr::<Sha256Hasher>::get_leaf(&self.inner, index)?.map(AsmManifestHash::from))
     }
 
     /// Generates an MMR inclusion proof for the leaf at `index` against an MMR
@@ -104,7 +110,7 @@ impl MmrDb {
 
 #[cfg(test)]
 mod tests {
-    use strata_identifiers::Buf32;
+    use strata_asm_common::AsmManifestHash;
     use strata_merkle::{Mmr, Mmr64B32, MmrState, Sha256Hasher};
 
     use super::*;
@@ -117,52 +123,50 @@ mod tests {
     /// A distinct, non-zero leaf for `seed`. The non-zero marker matters: the
     /// compact-peaks MMR these proofs verify against treats an all-zero hash as
     /// an empty-peak sentinel, so `[0; 32]` is not a representable leaf.
-    fn make_leaf(seed: u8) -> Buf32 {
+    fn make_leaf(seed: u8) -> AsmManifestHash {
         let mut bytes = [seed; 32];
         bytes[31] = 0xAB;
-        Buf32::new(bytes)
+        AsmManifestHash::from(bytes)
     }
 
     /// Reference compact-peaks MMR built by replaying the first `size` leaves
     /// of `mmr_db`, matching the accumulators that proofs verify against.
-    fn rebuild_compact_mmr(mmr_db: &MmrDb, size: u64) -> Mmr64B32 {
+    fn rebuild_compact_mmr(mmr_db: &AsmManifestMmrDb, size: u64) -> Mmr64B32 {
         let mut compact = Mmr64B32::new_empty();
         for i in 0..size {
             let leaf = mmr_db.get_leaf(i).unwrap().unwrap();
-            Mmr::<Sha256Hasher>::add_leaf(&mut compact, leaf.0).unwrap();
+            Mmr::<Sha256Hasher>::add_leaf(&mut compact, *leaf.as_ref()).unwrap();
         }
         compact
     }
 
     #[test]
     fn empty_mmr_has_zero_leaves() {
         let db = test_db();
-        let mmr = MmrDb::open(&db).unwrap();
+        let mmr = AsmManifestMmrDb::open(&db).unwrap();
         assert_eq!(mmr.leaf_count().unwrap(), 0);
     }
 
     #[test]
-    fn append_and_retrieve_leaf() {
+    fn put_and_retrieve_leaf() {
         let db = test_db();
-        let mmr = MmrDb::open(&db).unwrap();
+        let mmr = AsmManifestMmrDb::open(&db).unwrap();
         let leaf = make_leaf(0xaa);
 
-        let idx = mmr.append_leaf(leaf).unwrap();
-        assert_eq!(idx, 0);
+        mmr.put_leaf(0, leaf).unwrap();
         assert_eq!(mmr.leaf_count().unwrap(), 1);
 
         let retrieved = mmr.get_leaf(0).unwrap().unwrap();
         assert_eq!(retrieved, leaf);
     }
 
     #[test]
-    fn append_multiple_leaves() {
+    fn put_multiple_leaves() {
         let db = test_db();
-        let mmr = MmrDb::open(&db).unwrap();
+        let mmr = AsmManifestMmrDb::open(&db).unwrap();
 
         for i in 0u8..5 {
-            let idx = mmr.append_leaf(make_leaf(i)).unwrap();
-            assert_eq!(idx, i as u64);
+            mmr.put_leaf(i as u64, make_leaf(i)).unwrap();
         }
 
         assert_eq!(mmr.leaf_count().unwrap(), 5);
@@ -173,61 +177,69 @@ mod tests {
         }
     }
 
+    #[test]
+    fn put_leaf_rejects_gap() {
+        let db = test_db();
+        let mmr = AsmManifestMmrDb::open(&db).unwrap();
+        // Leaf index 1 skips index 0, which would leave a hole.
+        assert!(mmr.put_leaf(1, make_leaf(0)).is_err());
+    }
+
     #[test]
     fn get_missing_leaf_returns_none() {
         let db = test_db();
-        let mmr = MmrDb::open(&db).unwrap();
+        let mmr = AsmManifestMmrDb::open(&db).unwrap();
         assert!(mmr.get_leaf(0).unwrap().is_none());
     }
 
     #[test]
     fn generate_and_verify_proof_single_leaf() {
         let db = test_db();
-        let mmr_db = MmrDb::open(&db).unwrap();
+        let mmr_db = AsmManifestMmrDb::open(&db).unwrap();
         let leaf = make_leaf(0x01);
-        mmr_db.append_leaf(leaf).unwrap();
+        mmr_db.put_leaf(0, leaf).unwrap();
 
         let proof = mmr_db.generate_proof(0, 1).unwrap();
         let compact = rebuild_compact_mmr(&mmr_db, 1);
-        assert!(compact.verify(&proof, &leaf.0));
+        assert!(compact.verify(&proof, leaf.as_ref()));
     }
 
     #[test]
     fn generate_proofs_for_all_leaves() {
         let db = test_db();
-        let mmr_db = MmrDb::open(&db).unwrap();
+        let mmr_db = AsmManifestMmrDb::open(&db).unwrap();
 
         for i in 0u8..8 {
-            mmr_db.append_leaf(make_leaf(i)).unwrap();
+            mmr_db.put_leaf(i as u64, make_leaf(i)).unwrap();
         }
 
         let compact = rebuild_compact_mmr(&mmr_db, 8);
         for i in 0u64..8 {
             let proof = mmr_db
                 .generate_proof(i, 8)
                 .unwrap_or_else(|e| panic!("proof generation failed for leaf {i}: {e}"));
-            assert!(compact.verify(&proof, &make_leaf(i as u8).0));
+            assert!(compact.verify(&proof, make_leaf(i as u8).as_ref()));
         }
     }
 
     #[test]
     fn proof_at_earlier_size_is_valid() {
         let db = test_db();
-        let mmr_db = MmrDb::open(&db).unwrap();
+        let mmr_db = AsmManifestMmrDb::open(&db).unwrap();
 
-        // Append 4 leaves, snapshot the compact state.
+        // Put 4 leaves, snapshot the compact state.
         for i in 0u8..4 {
-            mmr_db.append_leaf(make_leaf(i)).unwrap();
+            mmr_db.put_leaf(i as u64, make_leaf(i)).unwrap();
         }
         let compact_at_4 = rebuild_compact_mmr(&mmr_db, 4);
 
-        // Append 4 more.
+        // Put 4 more.
         for i in 4u8..8 {
-            mmr_db.append_leaf(make_leaf(i)).unwrap();
+            mmr_db.put_leaf(i as u64, make_leaf(i)).unwrap();
         }
 
         // Proof at size 4 should verify against the snapshot.
         let proof = mmr_db.generate_proof(2, 4).unwrap();
-        assert!(compact_at_4.verify(&proof, &make_leaf(2).0));
+        assert!(compact_at_4.verify(&proof, make_leaf(2).as_ref()));
     }
 }