Preserve user-submitted batch proofs

crates/block-producer/src/errors.rs

@@ -11,6 +11,7 @@ use miden_node_store::{
 use miden_protocol::Word;
 use miden_protocol::account::AccountId;
 use miden_protocol::asset::AssetId;
+use miden_protocol::batch::BatchId;
 use miden_protocol::block::BlockNumber;
 use miden_protocol::crypto::utils::DeserializationError;
 use miden_protocol::errors::{ProposedBatchError, ProposedBlockError, ProvenBatchError};
@@ -96,6 +97,9 @@ pub enum MempoolSubmissionError {
         transaction_id: TransactionId,
         fee_asset_id: AssetId,
     },
+
+    #[error("user batch proof ID {proof_id} does not match transaction batch ID {batch_id}")]
+    BatchIdMismatch { proof_id: BatchId, batch_id: BatchId },
 }
 
 // Mempool submission conflicts with current state

crates/block-producer/src/mempool/graph/batch.rs

@@ -50,11 +50,12 @@ impl GraphNode for SelectedBatch {
 // BATCH GRAPH
 // ================================================================================================
 
-/// Tracks [`SelectedBatch`] instances that are pending proof generation.
+/// Tracks [`SelectedBatch`] instances that are waiting for inclusion in a block.
 ///
 /// Batches form nodes in the underlying [`Graph`]. Edges between batches capture dependencies
 /// introduced by shared resources (nullifiers, notes, and account states). The graph remains a DAG
 /// by requiring that each batch builds on top of the state created by previously inserted batches.
+/// Sequencer-built batches wait for a proof. User-proven batches include their proof when inserted.
 #[derive(Clone, Debug, PartialEq, Default)]
 pub struct BatchGraph {
     inner: Graph<SelectedBatch>,

crates/block-producer/src/mempool/graph/transaction.rs

@@ -3,7 +3,7 @@ use std::sync::Arc;
 
 use miden_protocol::Word;
 use miden_protocol::account::AccountId;
-use miden_protocol::batch::BatchId;
+use miden_protocol::batch::{BatchId, ProvenBatch};
 use miden_protocol::block::BlockNumber;
 use miden_protocol::note::Nullifier;
 use miden_protocol::transaction::{OutputNote, TransactionId};
@@ -73,7 +73,7 @@ impl GraphNode for Arc<AuthenticatedTransaction> {
 // TRANSACTION GRAPH
 // ================================================================================================
 
-/// Tracks all [`AuthenticatedTransaction`]s that are waiting to be included in a batch.
+/// Tracks standalone transactions and transactions from user-proven batches.
 ///
 /// Each transaction is a node in the underlying [`Graph`]. A directed edge from transaction `P`
 /// to transaction `C` exists when `C` depends on state produced by `P` — for example, `C`
@@ -147,13 +147,14 @@ impl TransactionGraph {
         Some((creator.id(), output_note))
     }
 
-    /// Appends the transactions into the graph as an atomic unit.
+    /// Appends a user-proven batch to the graph as an atomic unit.
     ///
     /// These transactions can only be selected as a batch, and are reverted and pruned together.
     pub fn append_user_batch(
         &mut self,
         batch: &[Arc<AuthenticatedTransaction>],
         parameters: BatchParameters,
+        proof: Arc<ProvenBatch>,
     ) -> Result<(), StateConflict> {
         let batch_id =
             BatchId::from_transactions(batch.iter().map(|tx| tx.raw_proven_transaction()));
@@ -173,29 +174,24 @@ impl TransactionGraph {
         }
 
         let txs = batch.iter().map(GraphNode::id).collect::<Vec<_>>();
-        self.user_batches.insert(batch_id, txs, parameters);
+        self.user_batches.insert(batch_id, txs, parameters, proof);
 
         Ok(())
     }
 
-    pub fn select_any_batch(
+    pub fn select_any_internal_batch(
         &mut self,
         budget: BatchBudget,
         internal_parameters: BatchParameters,
     ) -> Option<SelectedBatch> {
-        self.select_user_batch()
-            .or_else(|| self.select_internal_batch(budget, internal_parameters).into_batch())
+        self.select_internal_batch(budget, internal_parameters).into_batch()
     }
 
-    pub fn select_full_batch(
+    pub fn select_full_internal_batch(
         &mut self,
         budget: BatchBudget,
         internal_parameters: BatchParameters,
     ) -> Option<SelectedBatch> {
-        if let Some(user_batch) = self.select_user_batch() {
-            return Some(user_batch);
-        }
-
         match self.select_internal_batch(budget, internal_parameters) {
             BatchSelection::Full(batch) => Some(batch),
             BatchSelection::Partial(batch) => {
@@ -213,7 +209,7 @@ impl TransactionGraph {
         }
     }
 
-    fn select_user_batch(&mut self) -> Option<SelectedBatch> {
+    pub fn select_user_batch(&mut self) -> Option<(SelectedBatch, Arc<ProvenBatch>)> {
         let candidate_batches = self.user_batches.batches().copied().collect::<HashSet<_>>();
         for candidate in candidate_batches {
             if let Some(batch) = self.try_select_user_batch_candidate(candidate) {
@@ -232,8 +228,12 @@ impl TransactionGraph {
     ///
     /// Transactions can fail selection if they depend on any external transactions that have
     /// not yet been selected.
-    fn try_select_user_batch_candidate(&mut self, candidate: BatchId) -> Option<SelectedBatch> {
-        let (txs, parameters) = self.user_batches.get(&candidate)?;
+    fn try_select_user_batch_candidate(
+        &mut self,
+        candidate: BatchId,
+    ) -> Option<(SelectedBatch, Arc<ProvenBatch>)> {
+        let (txs, parameters, proof) = self.user_batches.get(&candidate)?;
+        let proof = Arc::clone(proof);
         let mut selected = SelectedBatch::builder(parameters);
 
         for tx in txs {
@@ -252,7 +252,7 @@ impl TransactionGraph {
         }
 
         assert!(!selected.is_empty(), "User batch should not be empty");
-        Some(selected.build())
+        Some((selected.build(), proof))
     }
 
     fn select_internal_batch(
@@ -493,14 +493,21 @@ struct BatchTxMap {
 struct UserBatch {
     txs: Vec<TransactionId>,
     parameters: BatchParameters,
+    proof: Arc<ProvenBatch>,
 }
 
 impl BatchTxMap {
-    fn insert(&mut self, batch: BatchId, txs: Vec<TransactionId>, parameters: BatchParameters) {
+    fn insert(
+        &mut self,
+        batch: BatchId,
+        txs: Vec<TransactionId>,
+        parameters: BatchParameters,
+        proof: Arc<ProvenBatch>,
+    ) {
         for tx in &txs {
             assert!(self.by_tx.insert(*tx, batch).is_none());
         }
-        assert!(self.by_batch.insert(batch, UserBatch { txs, parameters }).is_none());
+        assert!(self.by_batch.insert(batch, UserBatch { txs, parameters, proof }).is_none());
     }
 
     fn remove(&mut self, batch: &BatchId) -> Vec<TransactionId> {
@@ -522,8 +529,13 @@ impl BatchTxMap {
         self.by_tx.get(tx)
     }
 
-    fn get(&self, batch: &BatchId) -> Option<(&[TransactionId], BatchParameters)> {
-        self.by_batch.get(batch).map(|batch| (batch.txs.as_slice(), batch.parameters))
+    fn get(
+        &self,
+        batch: &BatchId,
+    ) -> Option<(&[TransactionId], BatchParameters, &Arc<ProvenBatch>)> {
+        self.by_batch
+            .get(batch)
+            .map(|batch| (batch.txs.as_slice(), batch.parameters, &batch.proof))
     }
 
     fn contains_tx(&self, tx: &TransactionId) -> bool {

crates/block-producer/src/mempool/mod.rs

@@ -274,6 +274,10 @@ impl Mempool {
         Ok(self.committed_chain_tip)
     }
 
+    /// Adds a user-proven batch to the mempool.
+    ///
+    /// The batch becomes available for block selection when its transaction dependencies are
+    /// selected.
     #[miden_instrument(
         target = COMPONENT,
         name = "mempool.add_user_batch",
@@ -282,6 +286,7 @@ impl Mempool {
         &mut self,
         txs: &[Arc<AuthenticatedTransaction>],
         parameters: BatchParameters,
+        proof: Arc<ProvenBatch>,
     ) -> Result<BlockNumber, MempoolSubmissionError> {
         assert!(!txs.is_empty(), "Cannot have a batch with no transactions");
 
@@ -300,14 +305,20 @@ impl Mempool {
             }
         }
 
+        let batch_id = BatchId::from_transactions(txs.iter().map(|tx| tx.raw_proven_transaction()));
+        if proof.id() != batch_id {
+            return Err(MempoolSubmissionError::BatchIdMismatch { proof_id: proof.id(), batch_id });
+        }
+
         for tx in txs {
             self.authentication_staleness_check(tx.authentication_height())?;
             self.expiration_check(tx.expires_at())?;
         }
 
         self.transactions
-            .append_user_batch(txs, parameters)
+            .append_user_batch(txs, parameters, proof)
             .map_err(MempoolSubmissionError::StateConflict)?;
+        self.promote_user_batches();
 
         let telemetry = self.telemetry();
         miden_span_record!(
@@ -326,7 +337,7 @@ impl Mempool {
         Ok(self.committed_chain_tip)
     }
 
-    /// Returns a set of transactions for the next batch.
+    /// Returns a set of standalone transactions for the next sequencer-built batch.
     ///
     /// Transactions are returned in a valid execution ordering.
     ///
@@ -336,11 +347,15 @@ impl Mempool {
         name = "mempool.select_any_batch",
     )]
     pub fn select_any_batch(&mut self) -> Option<SelectedBatch> {
+        self.promote_user_batches();
         let parameters = BatchParameters {
             reference_block: self.committed_chain_tip,
         };
-        let batch = self.transactions.select_any_batch(self.config.batch_budget, parameters)?;
+        let batch = self
+            .transactions
+            .select_any_internal_batch(self.config.batch_budget, parameters)?;
         let batch = self.append_selected_batch(batch);
+        self.promote_user_batches();
         let telemetry = self.telemetry();
         miden_span_record!(
             mempool.transactions.uncommitted = telemetry.uncommitted_transactions,
@@ -354,21 +369,24 @@ impl Mempool {
         Some(batch)
     }
 
-    /// Returns a full set of transactions for the next batch.
+    /// Returns a full set of standalone transactions for the next sequencer-built batch.
     ///
-    /// User batches count as full because they are externally chosen atomic batches.
-    /// Non-user batches are only returned when the selected set saturates the batch budget or when
+    /// The transactions are only returned when the selected set saturates the batch budget or when
     /// another selectable transaction cannot fit into the remaining budget.
     #[miden_instrument(
         target = COMPONENT,
         name = "mempool.select_full_batch",
     )]
     pub fn select_full_batch(&mut self) -> Option<SelectedBatch> {
+        self.promote_user_batches();
         let parameters = BatchParameters {
             reference_block: self.committed_chain_tip,
         };
-        let batch = self.transactions.select_full_batch(self.config.batch_budget, parameters)?;
+        let batch = self
+            .transactions
+            .select_full_internal_batch(self.config.batch_budget, parameters)?;
         let batch = self.append_selected_batch(batch);
+        self.promote_user_batches();
         let telemetry = self.telemetry();
         miden_span_record!(
             mempool.transactions.uncommitted = telemetry.uncommitted_transactions,
@@ -389,6 +407,14 @@ impl Mempool {
         batch
     }
 
+    /// Moves selectable user-proven batches into the batch graph.
+    fn promote_user_batches(&mut self) {
+        while let Some((batch, proof)) = self.transactions.select_user_batch() {
+            self.append_selected_batch(batch);
+            self.batches.submit_proof(proof);
+        }
+    }
+
     /// Drops the proposed batch and all of its descendants.
     ///
     /// The transactions are re-queued for inclusion in a batch. Additionally, the batch's