refactor(block-producer): order batches within a block (#572)

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

@@ -69,7 +69,7 @@ impl BlockBuilder {
             interval.tick().await;
 
             let (block_number, batches) = mempool.lock().await.select_block();
-            let batches = batches.into_values().collect::<Vec<_>>();
+            let batches = batches.into_iter().map(|(_, batch)| batch).collect::<Vec<_>>();
 
             let mut result = self.build_block(&batches).await;
             let proving_duration = rand::thread_rng().gen_range(self.simulated_proof_time.clone());

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

@@ -217,11 +217,11 @@ impl BatchGraph {
 
     /// Selects the next set of batches ready for inclusion in a block while adhering to the given
     /// budget.
-    pub fn select_block(
-        &mut self,
-        mut budget: BlockBudget,
-    ) -> BTreeMap<BatchJobId, TransactionBatch> {
-        let mut batches = BTreeMap::new();
+    ///
+    /// Note that batch order should be maintained to allow for inter-batch dependencies to be
+    /// correctly resolved.
+    pub fn select_block(&mut self, mut budget: BlockBudget) -> Vec<(BatchJobId, TransactionBatch)> {
+        let mut batches = Vec::with_capacity(budget.batches);
 
         while let Some(batch_id) = self.inner.roots().first().copied() {
             // SAFETY: Since it was a root batch, it must definitely have a processed batch
@@ -240,7 +240,7 @@ impl BatchGraph {
             // SAFETY: This is definitely a root since we just selected it from the set of roots.
             self.inner.process_root(batch_id).expect("root should be processed");
 
-            batches.insert(batch_id, batch);
+            batches.push((batch_id, batch));
         }
 
         batches

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

@@ -1,8 +1,4 @@
-use std::{
-    collections::{BTreeMap, BTreeSet},
-    fmt::Display,
-    sync::Arc,
-};
+use std::{collections::BTreeSet, fmt::Display, sync::Arc};
 
 use batch_graph::BatchGraph;
 use inflight_state::InflightState;
@@ -314,16 +310,18 @@ impl Mempool {
 
     /// Select batches for the next block.
     ///
-    /// May return an empty set if no batches are ready.
+    /// Note that the set of batches
+    /// - may be empty if none are available, and
+    /// - may contain dependencies and therefore the order must be maintained
     ///
     /// # Panics
     ///
     /// Panics if there is already a block in flight.
-    pub fn select_block(&mut self) -> (BlockNumber, BTreeMap<BatchJobId, TransactionBatch>) {
+    pub fn select_block(&mut self) -> (BlockNumber, Vec<(BatchJobId, TransactionBatch)>) {
         assert!(self.block_in_progress.is_none(), "Cannot have two blocks inflight.");
 
         let batches = self.batches.select_block(self.block_budget);
-        self.block_in_progress = Some(batches.keys().cloned().collect());
+        self.block_in_progress = Some(batches.iter().map(|(id, _)| *id).collect());
 
         (self.chain_tip.next(), batches)
     }