feat(block-producer): switch to mempool version

CHANGELOG.md

@@ -6,6 +6,7 @@
 
 - Added `GetAccountProofs` endpoint (#506).
 - Support Https in endpoint configuration (#556).
+- Upgrade `block-producer` from FIFO queue to mempool dependency graph (#562).
 
 ### Changes
 

Makefile

@@ -13,7 +13,7 @@ BUILD_PROTO=BUILD_PROTO=1
 
 .PHONY: clippy
 clippy: ## Runs Clippy with configs
-	cargo clippy --locked --workspace --all-targets --all-features -- -D warnings --allow clippy::arc_with_non_send_sync
+	cargo clippy --locked --workspace --all-targets --all-features -- -D warnings
 
 
 .PHONY: fix

bin/node/src/commands/start.rs

@@ -1,3 +1,5 @@
+use std::collections::HashMap;
+
 use anyhow::{Context, Result};
 use miden_node_block_producer::server::BlockProducer;
 use miden_node_rpc::server::Rpc;
@@ -16,22 +18,41 @@ pub async fn start_node(config: NodeConfig) -> Result<()> {
 
     // Start store. The store endpoint is available after loading completes.
     let store = Store::init(store).await.context("Loading store")?;
-    join_set.spawn(async move { store.serve().await.context("Serving store") });
+    let store_id = join_set.spawn(async move { store.serve().await.context("Serving store") }).id();
 
     // Start block-producer. The block-producer's endpoint is available after loading completes.
     let block_producer =
         BlockProducer::init(block_producer).await.context("Loading block-producer")?;
-    join_set.spawn(async move { block_producer.serve().await.context("Serving block-producer") });
+    let block_producer_id = join_set
+        .spawn(async move { block_producer.serve().await.context("Serving block-producer") })
+        .id();
 
     // Start RPC component.
     let rpc = Rpc::init(rpc).await.context("Loading RPC")?;
-    join_set.spawn(async move { rpc.serve().await.context("Serving RPC") });
-
-    // block on all tasks
-    while let Some(res) = join_set.join_next().await {
-        // For now, if one of the components fails, crash the node
-        res??;
-    }
-
-    Ok(())
+    let rpc_id = join_set.spawn(async move { rpc.serve().await.context("Serving RPC") }).id();
+
+    // Lookup table so we can identify the failed component.
+    let component_ids = HashMap::from([
+        (store_id, "store"),
+        (block_producer_id, "block-producer"),
+        (rpc_id, "rpc"),
+    ]);
+
+    // SAFETY: The joinset is definitely not empty.
+    let component_result = join_set.join_next_with_id().await.unwrap();
+
+    // We expect components to run indefinitely, so we treat any return as fatal.
+    //
+    // Map all outcomes to an error, and provide component context.
+    let (id, err) = match component_result {
+        Ok((id, Ok(_))) => (id, Err(anyhow::anyhow!("Component completed unexpectedly"))),
+        Ok((id, Err(err))) => (id, Err(err)),
+        Err(join_err) => (join_err.id(), Err(join_err).context("Joining component task")),
+    };
+    let component = component_ids.get(&id).unwrap_or(&"unknown");
+
+    // We could abort and gracefully shutdown the other components, but since we're crashing the
+    // node there is no point.
+
+    err.context(format!("Component {component} failed"))
 }

crates/block-producer/Cargo.toml

@@ -24,6 +24,7 @@ miden-objects = { workspace = true }
 miden-processor = { workspace = true }
 miden-stdlib = { workspace = true }
 miden-tx = { workspace = true }
+rand = { version = "0.8" }
 serde = { version = "1.0", features = ["derive"] }
 thiserror = { workspace = true }
 tokio = { workspace = true, features = ["rt-multi-thread", "net", "macros", "sync", "time"] }
@@ -38,6 +39,7 @@ miden-lib = { workspace = true, features = ["testing"] }
 miden-node-test-macro = { path = "../test-macro" }
 miden-objects = { workspace = true, features = ["testing"] }
 miden-tx = { workspace = true, features = ["testing"] }
+pretty_assertions = "1.4"
 rand_chacha = { version = "0.3", default-features = false }
 tokio = { workspace = true, features = ["test-util"] }
 winterfell = { version = "0.10" }

crates/block-producer/src/batch_builder/batch.rs

@@ -9,21 +9,19 @@ use miden_objects::{
     batches::BatchNoteTree,
     crypto::hash::blake::{Blake3Digest, Blake3_256},
     notes::{NoteHeader, NoteId, Nullifier},
-    transaction::{InputNoteCommitment, OutputNote, TransactionId},
-    AccountDeltaError, Digest, MAX_ACCOUNTS_PER_BATCH, MAX_INPUT_NOTES_PER_BATCH,
-    MAX_OUTPUT_NOTES_PER_BATCH,
+    transaction::{InputNoteCommitment, OutputNote, ProvenTransaction, TransactionId},
+    AccountDeltaError, Digest,
 };
 use tracing::instrument;
 
-use crate::{errors::BuildBatchError, ProvenTransaction};
+use crate::errors::BuildBatchError;
 
 pub type BatchId = Blake3Digest<32>;
 
 // TRANSACTION BATCH
 // ================================================================================================
 
-/// A batch of transactions that share a common proof. For any given account, at most 1 transaction
-/// in the batch must be addressing that account (issue: #186).
+/// A batch of transactions that share a common proof.
 ///
 /// Note: Until recursive proofs are available in the Miden VM, we don't include the common proof.
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -77,13 +75,11 @@ impl TransactionBatch {
     /// for transforming unauthenticated notes into authenticated notes.
     ///
     /// # Errors
+    ///
     /// Returns an error if:
-    /// - The number of output notes across all transactions exceeds 4096.
     /// - There are duplicated output notes or unauthenticated notes found across all transactions
     ///   in the batch.
     /// - Hashes for corresponding input notes and output notes don't match.
-    ///
-    /// TODO: enforce limit on the number of created nullifiers.
     #[instrument(target = "miden-block-producer", name = "new_batch", skip_all, err)]
     pub fn new(
         txs: Vec<ProvenTransaction>,
@@ -102,27 +98,19 @@ impl TransactionBatch {
                     vacant.insert(AccountUpdate::new(tx));
                 },
                 Entry::Occupied(occupied) => occupied.into_mut().merge_tx(tx).map_err(|error| {
-                    BuildBatchError::AccountUpdateError {
-                        account_id: tx.account_id(),
-                        error,
-                        txs: txs.clone(),
-                    }
+                    BuildBatchError::AccountUpdateError { account_id: tx.account_id(), error }
                 })?,
             };
 
             // Check unauthenticated input notes for duplicates:
             for note in tx.get_unauthenticated_notes() {
                 let id = note.id();
                 if !unauthenticated_input_notes.insert(id) {
-                    return Err(BuildBatchError::DuplicateUnauthenticatedNote(id, txs.clone()));
+                    return Err(BuildBatchError::DuplicateUnauthenticatedNote(id));
                 }
             }
         }
 
-        if updated_accounts.len() > MAX_ACCOUNTS_PER_BATCH {
-            return Err(BuildBatchError::TooManyAccountsInBatch(txs));
-        }
-
         // Populate batch produced nullifiers and match output notes with corresponding
         // unauthenticated input notes in the same batch, which are removed from the unauthenticated
         // input notes set.
@@ -137,7 +125,7 @@ impl TransactionBatch {
             // Header is presented only for unauthenticated input notes.
             let input_note = match input_note.header() {
                 Some(input_note_header) => {
-                    if output_notes.remove_note(input_note_header, &txs)? {
+                    if output_notes.remove_note(input_note_header)? {
                         continue;
                     }
 
@@ -155,16 +143,8 @@ impl TransactionBatch {
             input_notes.push(input_note)
         }
 
-        if input_notes.len() > MAX_INPUT_NOTES_PER_BATCH {
-            return Err(BuildBatchError::TooManyInputNotes(input_notes.len(), txs));
-        }
-
         let output_notes = output_notes.into_notes();
 
-        if output_notes.len() > MAX_OUTPUT_NOTES_PER_BATCH {
-            return Err(BuildBatchError::TooManyNotesCreated(output_notes.len(), txs));
-        }
-
         // Build the output notes SMT.
         let output_notes_smt = BatchNoteTree::with_contiguous_leaves(
             output_notes.iter().map(|note| (note.id(), note.metadata())),
@@ -250,7 +230,7 @@ impl OutputNoteTracker {
         for tx in txs {
             for note in tx.output_notes().iter() {
                 if output_note_index.insert(note.id(), output_notes.len()).is_some() {
-                    return Err(BuildBatchError::DuplicateOutputNote(note.id(), txs.to_vec()));
+                    return Err(BuildBatchError::DuplicateOutputNote(note.id()));
                 }
                 output_notes.push(Some(note.clone()));
             }
@@ -259,11 +239,7 @@ impl OutputNoteTracker {
         Ok(Self { output_notes, output_note_index })
     }
 
-    pub fn remove_note(
-        &mut self,
-        input_note_header: &NoteHeader,
-        txs: &[ProvenTransaction],
-    ) -> Result<bool, BuildBatchError> {
+    pub fn remove_note(&mut self, input_note_header: &NoteHeader) -> Result<bool, BuildBatchError> {
         let id = input_note_header.id();
         if let Some(note_index) = self.output_note_index.remove(&id) {
             if let Some(output_note) = mem::take(&mut self.output_notes[note_index]) {
@@ -274,7 +250,6 @@ impl OutputNoteTracker {
                         id,
                         input_hash,
                         output_hash,
-                        txs: txs.to_vec(),
                     });
                 }
 
@@ -323,7 +298,7 @@ mod tests {
         ));
 
         match OutputNoteTracker::new(&txs) {
-            Err(BuildBatchError::DuplicateOutputNote(note_id, _)) => {
+            Err(BuildBatchError::DuplicateOutputNote(note_id)) => {
                 assert_eq!(note_id, duplicate_output_note.id())
             },
             res => panic!("Unexpected result: {res:?}"),
@@ -337,8 +312,8 @@ mod tests {
 
         let note_to_remove = mock_note(4);
 
-        assert!(tracker.remove_note(note_to_remove.header(), &txs).unwrap());
-        assert!(!tracker.remove_note(note_to_remove.header(), &txs).unwrap());
+        assert!(tracker.remove_note(note_to_remove.header()).unwrap());
+        assert!(!tracker.remove_note(note_to_remove.header()).unwrap());
 
         // Check that output notes are in the expected order and consumed note was removed
         assert_eq!(
@@ -359,7 +334,7 @@ mod tests {
         let duplicate_note = mock_note(5);
         txs.push(mock_proven_tx(4, vec![duplicate_note.clone()], vec![mock_output_note(9)]));
         match TransactionBatch::new(txs, Default::default()) {
-            Err(BuildBatchError::DuplicateUnauthenticatedNote(note_id, _)) => {
+            Err(BuildBatchError::DuplicateUnauthenticatedNote(note_id)) => {
                 assert_eq!(note_id, duplicate_note.id())
             },
             res => panic!("Unexpected result: {res:?}"),