feat: block producer redesign

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

@@ -22,8 +22,7 @@ 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)]

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

@@ -1,11 +1,18 @@
-use std::{cmp::min, collections::BTreeSet, sync::Arc, time::Duration};
-
-use async_trait::async_trait;
-use miden_objects::{notes::NoteId, transaction::OutputNote};
-use tokio::time;
+use std::{cmp::min, collections::BTreeSet, num::NonZeroUsize, sync::Arc, time::Duration};
+
+use miden_objects::{
+    notes::NoteId,
+    transaction::{OutputNote, TransactionId},
+};
+use tokio::{sync::Mutex, time};
+use tonic::async_trait;
 use tracing::{debug, info, instrument, Span};
 
-use crate::{block_builder::BlockBuilder, ProvenTransaction, SharedRwVec, COMPONENT};
+use crate::{
+    block_builder::BlockBuilder,
+    mempool::{BatchJobId, Mempool},
+    ProvenTransaction, SharedRwVec, COMPONENT,
+};
 
 #[cfg(test)]
 mod tests;
@@ -206,3 +213,84 @@ where
         Ok(())
     }
 }
+
+pub struct BatchProducer {
+    pub batch_interval: Duration,
+    pub workers: NonZeroUsize,
+    pub mempool: Arc<Mutex<Mempool>>,
+    pub tx_per_batch: usize,
+}
+
+type BatchResult = Result<BatchJobId, (BatchJobId, BuildBatchError)>;
+
+/// Wrapper around tokio's JoinSet that remains pending if the set is empty,
+/// instead of returning None.
+struct WorkerPool(tokio::task::JoinSet<BatchResult>);
+
+impl WorkerPool {
+    async fn join_next(&mut self) -> Result<BatchResult, tokio::task::JoinError> {
+        if self.0.is_empty() {
+            std::future::pending().await
+        } else {
+            // Cannot be None as its not empty.
+            self.0.join_next().await.unwrap()
+        }
+    }
+
+    fn len(&self) -> usize {
+        self.0.len()
+    }
+
+    fn spawn(&mut self, id: BatchJobId, transactions: Vec<TransactionId>) {
+        self.0.spawn(async move {
+            todo!("Do actual work like aggregating transaction data");
+        });
+    }
+}
+
+impl BatchProducer {
+    pub async fn run(self) {
+        let mut interval = tokio::time::interval(self.batch_interval);
+        interval.set_missed_tick_behavior(time::MissedTickBehavior::Delay);
+
+        let mut inflight = WorkerPool(tokio::task::JoinSet::new());
+
+        loop {
+            tokio::select! {
+                _ = interval.tick() => {
+                    if inflight.len() >= self.workers.get() {
+                        tracing::info!("All batch workers occupied.");
+                        continue;
+                    }
+
+                    // Transactions available?
+                    let Some((batch_id, transactions)) =
+                        self.mempool.lock().await.select_batch()
+                    else {
+                        tracing::info!("No transactions available for batch.");
+                        continue;
+                    };
+
+                    inflight.spawn(batch_id, transactions);
+                },
+                result = inflight.join_next() => {
+                    let mut mempool = self.mempool.lock().await;
+                    match result {
+                        Err(err) => {
+                            tracing::warn!(%err, "Batch job panic'd.")
+                            // TODO: somehow embed the batch ID into the join error, though this doesn't seem possible?
+                            // mempool.batch_failed(batch_id);
+                        },
+                        Ok(Err((batch_id, err))) => {
+                            tracing::warn!(%batch_id, %err, "Batch job failed.");
+                            mempool.batch_failed(batch_id);
+                        },
+                        Ok(Ok(batch_id)) => {
+                            mempool.batch_proved(batch_id);
+                        }
+                    }
+                }
+            }
+        }
+    }
+}

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

@@ -8,12 +8,14 @@ use miden_objects::{
     notes::{NoteHeader, Nullifier},
     transaction::InputNoteCommitment,
 };
+use tokio::sync::Mutex;
 use tracing::{debug, info, instrument};
 
 use crate::{
     batch_builder::batch::TransactionBatch,
     errors::BuildBlockError,
-    store::{ApplyBlock, Store},
+    mempool::{BatchJobId, Mempool},
+    store::{ApplyBlock, DefaultStore, Store},
     COMPONENT,
 };
 
@@ -143,3 +145,33 @@ where
         Ok(())
     }
 }
+
+struct BlockProducer {
+    pub mempool: Arc<Mutex<Mempool>>,
+    pub block_interval: tokio::time::Duration,
+}
+
+impl BlockProducer {
+    pub async fn run(self) {
+        let mut interval = tokio::time::interval(self.block_interval);
+        interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
+
+        loop {
+            interval.tick().await;
+
+            let (block_number, batches) = self.mempool.lock().await.select_block();
+
+            let result = self.build_and_commit_block(batches).await;
+            let mut mempool = self.mempool.lock().await;
+
+            match result {
+                Ok(_) => mempool.block_committed(block_number),
+                Err(_) => mempool.block_failed(block_number),
+            }
+        }
+    }
+
+    async fn build_and_commit_block(&self, batches: BTreeSet<BatchJobId>) -> Result<(), ()> {
+        todo!("Aggregate, prove and commit block");
+    }
+}

crates/block-producer/src/errors.rs

@@ -1,3 +1,5 @@
+use std::collections::BTreeSet;
+
 use miden_node_proto::errors::ConversionError;
 use miden_node_utils::formatting::format_opt;
 use miden_objects::{
@@ -9,8 +11,11 @@ use miden_objects::{
     MAX_BATCHES_PER_BLOCK, MAX_INPUT_NOTES_PER_BATCH, MAX_OUTPUT_NOTES_PER_BATCH,
 };
 use miden_processor::ExecutionError;
+use miden_tx::TransactionVerifierError;
 use thiserror::Error;
 
+use crate::mempool::BlockNumber;
+
 // Transaction verification errors
 // =================================================================================================
 
@@ -58,6 +63,29 @@ pub enum AddTransactionError {
     VerificationFailed(#[from] VerifyTxError),
 }
 
+#[derive(thiserror::Error, Debug, PartialEq)]
+pub enum AddTransactionErrorRework {
+    #[error("Transaction's initial account state {expected} did not match the current account state {current}.")]
+    InvalidAccountState { current: Digest, expected: Digest },
+    #[error("Transaction input data is stale. Required data fresher than {stale_limit} but inputs are from {input_block}.")]
+    StaleInputs {
+        input_block: BlockNumber,
+        stale_limit: BlockNumber,
+    },
+    #[error("Authenticated note nullifier {0} not found.")]
+    AuthenticatedNoteNotFound(Nullifier),
+    #[error("Unauthenticated note {0} not found.")]
+    UnauthenticatedNoteNotFound(NoteId),
+    #[error("Note nullifiers already consumed: {0:?}")]
+    NotesAlreadyConsumed(BTreeSet<Nullifier>),
+    #[error(transparent)]
+    TxInputsError(#[from] TxInputsError),
+    #[error(transparent)]
+    ProofVerificationFailed(#[from] TransactionVerifierError),
+    #[error("Failed to deserialize transaction: {0}.")]
+    DeserializationError(String),
+}
+
 // Batch building errors
 // =================================================================================================
 

crates/block-producer/src/lib.rs

@@ -1,3 +1,6 @@
+// TODO: remove once block-producer rework is complete
+#![allow(unused)]
+
 use std::{sync::Arc, time::Duration};
 
 use batch_builder::batch::TransactionBatch;
@@ -10,6 +13,7 @@ pub mod test_utils;
 mod batch_builder;
 mod block_builder;
 mod errors;
+mod mempool;
 mod state_view;
 mod store;
 mod txqueue;