chore(mempool): user batch followups

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

@@ -98,7 +98,7 @@ impl BatchGraph {
     ///
     /// Batches are returned in reverse-chronological order.
     pub fn revert_expired(&mut self, chain_tip: BlockNumber) -> Vec<SelectedBatch> {
-        // We only revert transactions which are _not_ included in batches.
+        // We only revert batches which are _not_ included in blocks.
         let mut to_revert = self.inner.expired(chain_tip);
         to_revert.retain(|batch| !self.inner.is_selected(batch));
 

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

@@ -77,12 +77,8 @@ pub struct TransactionGraph {
     /// used to identify potentially buggy transactions that should be evicted.
     failures: HashMap<TransactionId, u32>,
 
-    /// Defines the transactions that belong to a user batch.
-    user_batch_txs: HashMap<BatchId, Vec<TransactionId>>,
-    /// A mapping of transactions to their user batch (if any).
-    ///
-    /// Inverse map of `user_batch_txs`.
-    txs_user_batch: HashMap<TransactionId, BatchId>,
+    /// Bijective mapping of user batches and their transactions.
+    user_batches: BatchTxMap,
 }
 
 impl TransactionGraph {
@@ -93,6 +89,9 @@ impl TransactionGraph {
         self.inner.append(tx)
     }
 
+    /// Appends the transactions into 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>],
@@ -115,67 +114,66 @@ impl TransactionGraph {
         }
 
         let txs = batch.iter().map(GraphNode::id).collect::<Vec<_>>();
-        for tx in &txs {
-            self.txs_user_batch.insert(*tx, batch_id);
-        }
-        self.user_batch_txs.insert(batch_id, txs);
+        self.user_batches.insert(batch_id, txs);
 
         Ok(())
     }
 
     pub fn select_batch(&mut self, budget: BatchBudget) -> Option<SelectedBatch> {
-        self.select_user_batch().or_else(|| self.select_conventional_batch(budget))
+        self.select_user_batch().or_else(|| self.select_internal_batch(budget))
     }
 
     fn select_user_batch(&mut self) -> Option<SelectedBatch> {
-        // Comb through all user batch candidates.
-        let candidate_batches = self
-            .inner
-            .selection_candidates()
-            .values()
-            .filter_map(|tx| self.txs_user_batch.get(&tx.id()))
-            .copied()
-            .collect::<HashSet<_>>();
-
-        'outer: for candidate in candidate_batches {
-            let mut selected = SelectedBatch::builder();
-
-            let txs = self
-                .user_batch_txs
-                .get(&candidate)
-                .cloned()
-                .expect("bi-directional mapping should be coherent");
-
-            for tx in txs {
-                let Some(tx) = self.inner.selection_candidates().get(&tx).copied() else {
-                    // Rollback this batch selection since it cannot complete.
-                    for tx in selected.txs.into_iter().rev() {
-                        self.inner.deselect(tx.id());
-                    }
-
-                    continue 'outer;
-                };
-                let tx = Arc::clone(tx);
-
-                self.inner.select_candidate(tx.id());
-                selected.push(tx);
+        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) {
+                return Some(batch);
             }
-
-            assert!(!selected.is_empty(), "User batch should not be empty");
-            return Some(selected.build());
         }
 
         None
     }
 
-    fn select_conventional_batch(&mut self, mut budget: BatchBudget) -> Option<SelectedBatch> {
+    /// Attempts to select all transactions from the candidate user batch.
+    ///
+    /// This action succeeds if all transactions in the batch can be sequentially selected.
+    /// If any transaction cannot be selected, then all previous selections are rolled back,
+    /// and `None` is returned. This makes this function atomic.
+    ///
+    /// 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 mut selected = SelectedBatch::builder();
+
+        let txs = self.user_batches.get_txs_contained_in_batch(&candidate)?;
+
+        for tx in txs {
+            let Some(tx) = self.inner.selection_candidates().get(&tx).copied() else {
+                // Rollback this batch selection since it cannot complete.
+                for tx in selected.txs.into_iter().rev() {
+                    self.inner.deselect(tx.id());
+                }
+
+                return None;
+            };
+            let tx = Arc::clone(tx);
+
+            self.inner.select_candidate(tx.id());
+            selected.push(tx);
+        }
+
+        assert!(!selected.is_empty(), "User batch should not be empty");
+        Some(selected.build())
+    }
+
+    fn select_internal_batch(&mut self, mut budget: BatchBudget) -> Option<SelectedBatch> {
         let mut selected = SelectedBatch::builder();
 
         loop {
             // Select arbitrary candidate which is _not_ part of a user batch.
             let candidates = self.inner.selection_candidates();
             let Some(candidate) =
-                candidates.values().find(|tx| !self.txs_user_batch.contains_key(&tx.id()))
+                candidates.values().find(|tx| !self.user_batches.contains_tx(&tx.id()))
             else {
                 break;
             };
@@ -213,7 +211,6 @@ impl TransactionGraph {
     /// _before_ calling this function. i.e. first revert expired batches and deselect their
     /// transactions, then call this.
     pub fn revert_expired(&mut self, chain_tip: BlockNumber) -> HashSet<TransactionId> {
-        // We only revert transactions which are _not_ included in batches.
         let mut to_revert = self.inner.expired(chain_tip);
         to_revert.retain(|tx| !self.inner.is_selected(tx));
 
@@ -256,10 +253,10 @@ impl TransactionGraph {
                 // transactions in, which will result in at least the current
                 // transaction being duplicated in `to_revert`. This isn't a concern
                 // though since we skip already processed transactions at the top of the loop.
-                if let Some(batch) = self.txs_user_batch.remove(&tx.id()) {
-                    if let Some(batch) = self.user_batch_txs.remove(&batch) {
-                        to_revert.extend(batch);
-                    }
+                if let Some(batch_id) = self.user_batches.get_batch_containing_tx(&tx.id()).copied()
+                {
+                    let batch_txs = self.user_batches.remove(&batch_id);
+                    to_revert.extend_from_slice(&batch_txs);
                 }
             }
 
@@ -322,9 +319,8 @@ impl TransactionGraph {
         for tx in batch.transactions() {
             self.inner.prune(tx.id());
             self.failures.remove(&tx.id());
-            self.txs_user_batch.remove(&tx.id());
         }
-        self.user_batch_txs.remove(&batch.id());
+        self.user_batches.remove(&batch.id());
     }
 
     /// Number of transactions which have not been selected for inclusion in a batch.
@@ -349,3 +345,47 @@ impl TransactionGraph {
         self.inner.output_note_count()
     }
 }
+
+// BIJECTIVE <USER BATCH, TRANSACTION> MAP
+// ================================================================================================
+
+/// A bijective mapping of batches and their transactions.
+#[derive(Clone, Debug, PartialEq, Default)]
+struct BatchTxMap {
+    by_batch: HashMap<BatchId, Vec<TransactionId>>,
+    by_tx: HashMap<TransactionId, BatchId>,
+}
+
+impl BatchTxMap {
+    fn insert(&mut self, batch: BatchId, txs: Vec<TransactionId>) {
+        for tx in &txs {
+            assert!(self.by_tx.insert(*tx, batch).is_none());
+        }
+        assert!(self.by_batch.insert(batch, txs).is_none());
+    }
+
+    fn remove(&mut self, batch: &BatchId) -> Vec<TransactionId> {
+        let txs = self.by_batch.remove(batch).unwrap_or_default();
+        for tx in &txs {
+            self.by_tx.remove(tx);
+        }
+        txs
+    }
+
+    fn batches(&self) -> impl Iterator<Item = &BatchId> {
+        self.by_batch.keys()
+    }
+
+    /// Returns the [`BatchId`] mapped to this transaction, if any.
+    fn get_batch_containing_tx(&self, tx: &TransactionId) -> Option<&BatchId> {
+        self.by_tx.get(tx)
+    }
+
+    fn get_txs_contained_in_batch(&self, batch: &BatchId) -> Option<&[TransactionId]> {
+        self.by_batch.get(batch).map(Vec::as_slice)
+    }
+
+    fn contains_tx(&self, tx: &TransactionId) -> bool {
+        self.by_tx.contains_key(tx)
+    }
+}

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

@@ -347,7 +347,7 @@ impl BlockProducerRpcServer {
      )]
     async fn submit_proven_batch(
         &self,
-        request: proto::transaction::TransactionBatch,
+        request: proto::transaction::ProvenTransactionBatch,
     ) -> Result<proto::blockchain::BlockNumber, MempoolSubmissionError> {
         let proposed = request
             .proposed_batch
@@ -395,7 +395,7 @@ impl api_server::Api for BlockProducerRpcServer {
 
     async fn submit_proven_batch(
         &self,
-        request: tonic::Request<proto::transaction::TransactionBatch>,
+        request: tonic::Request<proto::transaction::ProvenTransactionBatch>,
     ) -> Result<tonic::Response<proto::blockchain::BlockNumber>, Status> {
         self.submit_proven_batch(request.into_inner())
              .await

crates/rpc/src/server/api.rs

@@ -543,7 +543,7 @@ impl api_server::Api for RpcService {
     /// the batch, then forwards it to the block producer.
     async fn submit_proven_batch(
         &self,
-        request: tonic::Request<proto::transaction::TransactionBatch>,
+        request: tonic::Request<proto::transaction::ProvenTransactionBatch>,
     ) -> Result<tonic::Response<proto::blockchain::BlockNumber>, Status> {
         let Some(block_producer) = &self.block_producer else {
             return Err(Status::unavailable("Batch submission not available in read-only mode"));

docs/external/src/operator/architecture.md

@@ -77,7 +77,7 @@ The builder also exposes an internal gRPC server that the RPC component uses to
 
 ## Validator
 
-The validator is responsible for verifying the integrity of the blockchain by signing new blocks before they can be commited.
+The validator is responsible for verifying the integrity of the blockchain by signing new blocks before they can be committed.
 
 At the moment this is implemented by having all transactions sent here to be re-executed to double-check their integrity. This
 also guards against bugs in the proving or execution systems, by backing up the transactions and their private inputs. This 

docs/internal/src/SUMMARY.md

@@ -1,4 +1,4 @@
-<!-- This file is used to represent agregate documentation for the Miden book -->
+<!-- This file is used to represent aggregate documentation for the Miden book -->
 
 # Summary
 

docs/internal/src/validator.md

@@ -3,7 +3,7 @@
 The validator is responsible for verifying each new block and signing it if correct.
 
 This signature is required _before_ a block may be committed on chain, and thus acts as an
-indepedent safe guard.
+independent safe guard.
 
 The validator is therefore run completely separate from the main node operations, and is operated
 by a separate entity. The validator's public key is published (or at least will be for `mainnet`).

proto/proto/internal/block_producer.proto

@@ -24,7 +24,7 @@ service Api {
     // All transactions in this batch will be considered atomic, and be committed together or not all.
     //
     // Returns the node's current block height.
-    rpc SubmitProvenBatch(transaction.TransactionBatch) returns (blockchain.BlockNumber) {}
+    rpc SubmitProvenBatch(transaction.ProvenTransactionBatch) returns (blockchain.BlockNumber) {}
 
     // Subscribe to mempool events.
     //

proto/proto/rpc.proto

@@ -64,7 +64,7 @@ service Api {
     // All transactions in this batch will be considered atomic, and be committed together or not all.
     //
     // Returns the node's current block height.
-    rpc SubmitProvenBatch(transaction.TransactionBatch) returns (blockchain.BlockNumber) {}
+    rpc SubmitProvenBatch(transaction.ProvenTransactionBatch) returns (blockchain.BlockNumber) {}
 
     // STATE SYNCHRONIZATION ENDPOINTS
     // --------------------------------------------------------------------------------------------

proto/proto/types/transaction.proto

@@ -8,32 +8,57 @@ import "types/primitives.proto";
 // TRANSACTION
 // ================================================================================================
 
-// Submits proven transaction to the Miden network.
+// A proven transaction.
+//
+// Note that we currently require full transaction transparency for the network operator.
+// This is a temporary measure while Miden stabilizes its protocol and proof systems. To
+// this end, a transaction submission includes its **private** inputs which the operator
+// can use to verify the transaction execution and proofs are correct.
+//
+// This means the transaction is _not_ private wrt the network operator (but it is still
+// private onchain). This requirement will be lifted as Miden matures.
 message ProvenTransaction {
   // The transaction proof.
   //
   // Encoded using [miden_protocol::transaction::ProvenTransaction::to_bytes].
   bytes transaction = 1;
-  // The inputs used for the transaction proof.
+  // The private inputs used for the transaction proof.
   //
   // Encoded using [miden_protocol::transaction::TransactionInputs::to_bytes].
+  //
+  // Transactions missing this field will be rejected as per the message description.
   optional bytes transaction_inputs = 2;
 }
 
-message TransactionBatch {
+// A proven batch of transactions.
+//
+// Note that we currently require full transaction transparency for the network operator.
+// This is a temporary measure while Miden stabilizes its protocol and proof systems. To
+// this end, each transaction includes its **private** inputs which the operator can use
+// to verify the transaction execution and proofs are correct.
+//
+// This means the transaction is _not_ private wrt the network operator (but it is still
+// private onchain). This requirement will be lifted as Miden matures.
+//
+// In addition, in order to verify the batch itself, we also require the proposed batch.
+message ProvenTransactionBatch {
   // The batch proof.
   //
   // Encoded using [miden_protocol::batch::ProvenBatch::to_bytes].
   bytes batch_proof = 1;
   // The batch contents of the given proof.
   //
   // Encoded using [miden_protocol::batch::ProposedBatch::to_bytes].
+  //
+  // Batches missing this field will be rejected as per the message description.
   optional bytes proposed_batch = 2;
   // The transaction inputs for each transaction in the batch.
   //
   // Must match the transaction ordering in the batch.
   //
   // Encoded using [miden_protocol::transaction::TransactionInputs::to_bytes].
+  //
+  // Batch will be rejected if any transaction's input is missing as per the method description.
   repeated bytes transaction_inputs = 3;
 }