feat: Deferred block proving

CHANGELOG.md

@@ -11,6 +11,7 @@
 - Introduce `SyncChainMmr` RPC endpoint to sync chain MMR deltas within specified block ranges ([#1591](https://github.com/0xMiden/node/issues/1591)).
 - Fixed `TransactionHeader` serialization for row insertion on database & fixed transaction cursor on retrievals ([#1701](https://github.com/0xMiden/node/issues/1701)).
 - Added KMS signing support in validator ([#1677](https://github.com/0xMiden/node/pull/1677)).
+- Restructured block proving to be asynchronous and added finality field for `SyncChainMmr` requests ([#1725](https://github.com/0xMiden/miden-node/pull/1725)).
 
 ### Changes
 

bin/remote-prover/src/server/tests.rs

@@ -3,7 +3,6 @@ use std::num::NonZeroUsize;
 use std::sync::Arc;
 use std::time::Duration;
 
-use assert_matches::assert_matches;
 use miden_protocol::MIN_PROOF_SECURITY_LEVEL;
 use miden_protocol::account::auth::AuthScheme;
 use miden_protocol::asset::{Asset, FungibleAsset};
@@ -239,7 +238,7 @@ async fn capacity_is_respected() {
     result.sort_unstable();
     assert_eq!(expected, result);
 
-    assert_matches!(first.err().or(second.err()).or(third.err()), Some(err) => {
+    assert_matches::assert_matches!(first.err().or(second.err()).or(third.err()), Some(err) => {
         assert_eq!(err.code(), tonic::Code::ResourceExhausted);
     });
 

bin/stress-test/src/seeding/mod.rs

@@ -92,7 +92,7 @@ pub async fn seed_store(
     let faucet = create_faucet();
     let fee_params = FeeParameters::new(faucet.id(), 0).unwrap();
     let signer = EcdsaSecretKey::new();
-    let genesis_state = GenesisState::new(vec![faucet.clone()], fee_params, 1, 1, signer);
+    let genesis_state = GenesisState::new(vec![faucet.clone()], fee_params, 1, 1, signer.clone());
     Store::bootstrap(genesis_state.clone(), &data_directory)
         .await
         .expect("store should bootstrap");
@@ -114,6 +114,7 @@ pub async fn seed_store(
         &store_client,
         data_directory,
         accounts_filepath,
+        &signer,
     )
     .await;
 
@@ -125,6 +126,7 @@ pub async fn seed_store(
 ///
 /// The first transaction in each batch sends assets from the faucet to 255 accounts.
 /// The rest of the transactions consume the notes created by the faucet in the previous block.
+#[expect(clippy::too_many_arguments)]
 async fn generate_blocks(
     num_accounts: usize,
     public_accounts_percentage: u8,
@@ -133,6 +135,7 @@ async fn generate_blocks(
     store_client: &StoreClient,
     data_directory: DataDirectory,
     accounts_filepath: PathBuf,
+    signer: &EcdsaSecretKey,
 ) -> SeedingMetrics {
     // Each block is composed of [`BATCHES_PER_BLOCK`] batches, and each batch is composed of
     // [`TRANSACTIONS_PER_BATCH`] txs. The first note of the block is always a send assets tx
@@ -211,7 +214,8 @@ async fn generate_blocks(
         let block_inputs = get_block_inputs(store_client, &batches, &mut metrics).await;
 
         // update blocks
-        prev_block_header = apply_block(batches, block_inputs, store_client, &mut metrics).await;
+        prev_block_header =
+            apply_block(batches, block_inputs, store_client, &mut metrics, signer).await;
         if current_anchor_header.block_epoch() != prev_block_header.block_epoch() {
             current_anchor_header = prev_block_header.clone();
         }
@@ -246,11 +250,12 @@ async fn apply_block(
     block_inputs: BlockInputs,
     store_client: &StoreClient,
     metrics: &mut SeedingMetrics,
+    signer: &EcdsaSecretKey,
 ) -> BlockHeader {
     let proposed_block = ProposedBlock::new(block_inputs, batches).unwrap();
     let (header, body) = proposed_block.clone().into_header_and_body().unwrap();
     let block_size: usize = header.to_bytes().len() + body.to_bytes().len();
-    let signature = EcdsaSecretKey::new().sign(header.commitment());
+    let signature = signer.sign(header.commitment());
     // SAFETY: The header, body, and signature are known to correspond to each other.
     let signed_block = SignedBlock::new_unchecked(header, body, signature);
     let ordered_batches = proposed_block.batches().clone();

bin/stress-test/src/store/mod.rs

@@ -469,6 +469,7 @@ async fn sync_chain_mmr(
 ) -> SyncChainMmrRun {
     let sync_request = proto::rpc::SyncChainMmrRequest {
         block_range: Some(proto::rpc::BlockRange { block_from, block_to: Some(block_to) }),
+        finality: proto::rpc::Finality::Committed.into(),
     };
 
     let start = Instant::now();

crates/rpc/src/tests.rs

@@ -547,6 +547,7 @@ async fn sync_chain_mmr_returns_delta() {
 
     let request = proto::rpc::SyncChainMmrRequest {
         block_range: Some(proto::rpc::BlockRange { block_from: 0, block_to: None }),
+        finality: proto::rpc::Finality::Committed.into(),
     };
     let response = rpc_client.sync_chain_mmr(request).await.expect("sync_chain_mmr should succeed");
     let response = response.into_inner();

crates/store/src/blocks.rs

@@ -107,6 +107,41 @@ impl BlockStore {
         fs_err::write(block_path, data)
     }
 
+    // PROOF STORAGE
+    // --------------------------------------------------------------------------------------------
+
+    #[expect(dead_code)]
+    pub async fn load_proof(
+        &self,
+        block_num: BlockNumber,
+    ) -> Result<Option<Vec<u8>>, std::io::Error> {
+        match tokio::fs::read(self.proof_path(block_num)).await {
+            Ok(data) => Ok(Some(data)),
+            Err(err) if err.kind() == std::io::ErrorKind::NotFound => Ok(None),
+            Err(err) => Err(err),
+        }
+    }
+
+    #[instrument(
+        target = COMPONENT,
+        name = "store.block_store.save_proof",
+        skip(self, data),
+        err,
+        fields(proof_size = data.len())
+    )]
+    pub async fn save_proof(
+        &self,
+        block_num: BlockNumber,
+        data: &[u8],
+    ) -> Result<(), std::io::Error> {
+        let (epoch_path, proof_path) = self.epoch_proof_path(block_num)?;
+        if !epoch_path.exists() {
+            tokio::fs::create_dir_all(epoch_path).await?;
+        }
+
+        tokio::fs::write(proof_path, data).await
+    }
+
     // HELPER FUNCTIONS
     // --------------------------------------------------------------------------------------------
 
@@ -117,6 +152,13 @@ impl BlockStore {
         epoch_dir.join(format!("block_{block_num:08x}.dat"))
     }
 
+    fn proof_path(&self, block_num: BlockNumber) -> PathBuf {
+        let block_num = block_num.as_u32();
+        let epoch = block_num >> 16;
+        let epoch_dir = self.store_dir.join(format!("{epoch:04x}"));
+        epoch_dir.join(format!("proof_{block_num:08x}.dat"))
+    }
+
     fn epoch_block_path(
         &self,
         block_num: BlockNumber,
@@ -127,6 +169,16 @@ impl BlockStore {
         Ok((epoch_path.to_path_buf(), block_path))
     }
 
+    fn epoch_proof_path(
+        &self,
+        block_num: BlockNumber,
+    ) -> Result<(PathBuf, PathBuf), std::io::Error> {
+        let proof_path = self.proof_path(block_num);
+        let epoch_path = proof_path.parent().ok_or(std::io::Error::from(ErrorKind::NotFound))?;
+
+        Ok((epoch_path.to_path_buf(), proof_path))
+    }
+
     pub fn display(&self) -> std::path::Display<'_> {
         self.store_dir.display()
     }

crates/store/src/db/migrations/2025062000000_setup/up.sql

@@ -1,13 +1,18 @@
 CREATE TABLE block_headers (
-    block_num    INTEGER NOT NULL,
-    block_header BLOB    NOT NULL,
-    signature    BLOB    NOT NULL,
-    commitment   BLOB    NOT NULL,
+    block_num      INTEGER NOT NULL,
+    block_header   BLOB    NOT NULL,
+    signature      BLOB    NOT NULL,
+    commitment     BLOB    NOT NULL,
+    proving_inputs BLOB,             -- Serialized BlockProofRequest needed for deferred proving. NULL for genesis block.
+    is_proven      BOOLEAN NOT NULL DEFAULT 0, -- Whether the block has been proven
 
     PRIMARY KEY (block_num),
     CONSTRAINT block_header_block_num_is_u32 CHECK (block_num BETWEEN 0 AND 0xFFFFFFFF)
 );
 
+CREATE INDEX block_headers_to_be_proven ON block_headers(block_num ASC) WHERE is_proven = 0;
+CREATE INDEX block_headers_proven_desc ON block_headers(block_num DESC) WHERE is_proven = 1;
+
 CREATE TABLE account_codes (
     code_commitment BLOB NOT NULL,
     code            BLOB NOT NULL,

crates/store/src/db/mod.rs

@@ -5,7 +5,7 @@ use std::path::PathBuf;
 use anyhow::Context;
 use diesel::{Connection, QueryableByName, RunQueryDsl, SqliteConnection};
 use miden_node_proto::domain::account::AccountInfo;
-use miden_node_proto::generated as proto;
+use miden_node_proto::{BlockProofRequest, generated as proto};
 use miden_node_utils::tracing::OpenTelemetrySpanExt;
 use miden_protocol::Word;
 use miden_protocol::account::{AccountHeader, AccountId, AccountStorageHeader};
@@ -270,12 +270,15 @@ impl Db {
         conn.transaction(move |conn| {
             models::queries::apply_block(
                 conn,
-                genesis.header(),
-                genesis.signature(),
-                &[],
-                &[],
-                genesis.body().updated_accounts(),
-                genesis.body().transactions(),
+                models::queries::ApplyBlockData {
+                    block_header: genesis.header(),
+                    signature: genesis.signature(),
+                    notes: &[],
+                    nullifiers: &[],
+                    accounts: genesis.body().updated_accounts(),
+                    transactions: genesis.body().transactions(),
+                    proving_inputs: None, // Genesis block is never proven.
+                },
             )
         })
         .context("failed to insert genesis block")?;
@@ -555,16 +558,20 @@ impl Db {
         acquire_done: oneshot::Receiver<()>,
         signed_block: SignedBlock,
         notes: Vec<(NoteRecord, Option<Nullifier>)>,
+        proving_inputs: Option<BlockProofRequest>,
     ) -> Result<()> {
         self.transact("apply block", move |conn| -> Result<()> {
             models::queries::apply_block(
                 conn,
-                signed_block.header(),
-                signed_block.signature(),
-                &notes,
-                signed_block.body().created_nullifiers(),
-                signed_block.body().updated_accounts(),
-                signed_block.body().transactions(),
+                models::queries::ApplyBlockData {
+                    block_header: signed_block.header(),
+                    signature: signed_block.signature(),
+                    notes: &notes,
+                    nullifiers: signed_block.body().created_nullifiers(),
+                    accounts: signed_block.body().updated_accounts(),
+                    transactions: signed_block.body().transactions(),
+                    proving_inputs,
+                },
             )?;
 
             // XXX FIXME TODO free floating mutex MUST NOT exist
@@ -582,6 +589,49 @@ impl Db {
         .await
     }
 
+    /// Marks a previously committed block as proven.
+    ///
+    /// Sets the `is_proven` flag to `true` for the given block number.
+    #[instrument(target = COMPONENT, skip_all, err)]
+    pub async fn mark_block_proven(&self, block_num: BlockNumber) -> Result<()> {
+        self.transact("mark block proven", move |conn| {
+            models::queries::mark_block_proven(conn, block_num)
+        })
+        .await?;
+        Ok(())
+    }
+
+    /// Returns block numbers for all blocks that have not yet been proven, ordered ascending.
+    #[instrument(level = "debug", target = COMPONENT, skip_all, ret(level = "debug"), err)]
+    pub async fn select_unproven_blocks(&self, limit: i64) -> Result<Vec<BlockNumber>> {
+        self.transact("select unproven blocks", move |conn| {
+            models::queries::select_unproven_blocks(conn, limit)
+        })
+        .await
+    }
+
+    /// Returns the proving inputs for a given block number, if stored.
+    #[instrument(level = "debug", target = COMPONENT, skip_all, err)]
+    pub async fn select_block_proving_inputs(
+        &self,
+        block_num: BlockNumber,
+    ) -> Result<Option<BlockProofRequest>> {
+        self.transact("select block proving inputs", move |conn| {
+            models::queries::select_block_proving_inputs(conn, block_num)
+        })
+        .await
+    }
+
+    /// Returns the highest block number that has been proven, or `None` if no blocks have been
+    /// proven yet.
+    #[instrument(level = "debug", target = COMPONENT, skip_all, ret(level = "debug"), err)]
+    pub async fn select_latest_proven_block_num(&self) -> Result<Option<BlockNumber>> {
+        self.transact("select latest proven block num", |conn| {
+            models::queries::select_latest_proven_block_num(conn)
+        })
+        .await
+    }
+
     /// Selects storage map values for syncing storage maps for a specific account ID.
     ///
     /// The returned values are the latest known values up to `block_range.end()`, and no values