wip: fast-forward buffered streaming + async task cancellation safety + lock free state

crates/flashblocks/builder/Cargo.toml

@@ -47,6 +47,7 @@ alloy-primitives.workspace = true
 alloy-eips.workspace = true
 alloy-eip7928.workspace = true
 op-alloy-consensus.workspace = true
+op-alloy-rpc-types-engine.workspace = true
 alloy-rpc-types-engine.workspace = true
 alloy-op-evm.workspace = true
 
@@ -59,6 +60,7 @@ revm-database.workspace = true
 
 # tokio
 tokio.workspace = true
+tokio-stream.workspace = true
 
 # misc
 crossbeam-channel.workspace = true

crates/flashblocks/builder/benches/coordinator.rs

@@ -2,7 +2,7 @@ use std::sync::Arc;
 
 use criterion::{BenchmarkId, Criterion, criterion_group, criterion_main};
 use flashblocks_builder::{
-    coordinator::{FlashblocksExecutionCoordinator, process_flashblock, run_flashblock_processor},
+    coordinator::{FlashblocksExecutionCoordinator, run_flashblock_processor},
     test_utils::{
         BenchProvider, CHAIN_SPEC, EVM_CONFIG, build_flashblock_fixture_eth_transfers,
         build_flashblock_fixture_fib, build_flashblock_fixture_world_id_like_bn254,
@@ -47,6 +47,8 @@ fn fresh_coordinator(
     (coordinator, handle, pending_tx)
 }
 
+/// Benchmarks a single flashblock by driving it through `run_flashblock_processor`
+/// with a one-item stream.
 fn bench_process_flashblock_case<F>(
     c: &mut Criterion,
     group_name: &str,
@@ -66,16 +68,28 @@ fn bench_process_flashblock_case<F>(
 
         group.bench_with_input(BenchmarkId::new("txs", tx_count), &tx_count, |b, &_n| {
             b.iter(|| {
-                let (coordinator, _handle, pending_tx) = fresh_coordinator(&rt);
-                process_flashblock(
+                let (coordinator, handle, pending_tx) = fresh_coordinator(&rt);
+                let coordinator = Arc::new(coordinator);
+                let pending_tx_clone = pending_tx.clone();
+
+                let stream = handle
+                    .event_stream(provider.clone(), move |event| {
+                        FlashblocksExecutionCoordinator::event_hook(event, &pending_tx_clone)
+                    })
+                    .take(2); // 1 Canon(genesis) + 1 Pending flashblock
+
+                handle.flashblocks_tx().send(flashblock.clone()).ok();
+
+                rt.block_on(run_flashblock_processor(
+                    coordinator.clone(),
+                    stream,
                     provider.clone(),
-                    &EVM_CONFIG,
-                    &coordinator,
+                    EVM_CONFIG.clone(),
                     CHAIN_SPEC.clone(),
-                    flashblock.clone(),
                     pending_tx,
-                )
-                .expect("process_flashblock failed");
+                ));
+
+                assert_eq!(coordinator.flashblocks().flashblocks().len(), 1);
             });
         });
     }

crates/flashblocks/builder/src/bal_executor.rs

@@ -5,14 +5,16 @@ use alloy_op_evm::{
     block::receipt_builder::OpReceiptBuilder,
 };
 use op_alloy_consensus::OpReceipt;
+use reth_chain_state::ExecutedBlock;
 use reth_evm::{
     Database, Evm, FromRecoveredTx, FromTxWithEncoded,
     block::{BlockExecutionError, BlockExecutor, InternalBlockExecutionError, StateDB},
     op_revm::{OpSpecId, OpTransaction},
 };
 use reth_optimism_chainspec::OpChainSpec;
 use reth_optimism_evm::OpRethReceiptBuilder;
-use reth_optimism_primitives::OpTransactionSigned;
+use reth_optimism_forks::OpHardforks;
+use reth_optimism_primitives::{OpPrimitives, OpTransactionSigned};
 use reth_payload_primitives::BuiltPayload;
 use revm::{
     DatabaseCommit,
@@ -34,6 +36,7 @@ use reth_evm::{
     },
     op_revm::OpHaltReason,
 };
+use reth_node_api::BuiltPayloadExecutedBlock;
 use reth_optimism_node::{OpBlockAssembler, OpBuiltPayload};
 use reth_primitives::{NodePrimitives, Recovered, RecoveredBlock, SealedHeader};
 use reth_provider::StateProvider;
@@ -314,6 +317,14 @@ where
             bundle,
         ))
     }
+
+    fn set_committed_state(&mut self, _state: &ExecutedBlock<Self::Primitives>) {
+        todo!("BalBlockBuilder::set_committed_state")
+    }
+
+    fn chain_spec(&self) -> impl OpHardforks {
+        self.inner.executor.spec.clone()
+    }
 }
 
 /// Simple counter to track sub pre-confirmation block access index bounds.
@@ -382,6 +393,55 @@ where
     }
 }
 
+impl<R> CommittedState<R>
+where
+    R: OpReceiptBuilder<Transaction = OpTransactionSigned, Receipt = OpReceipt> + Default,
+{
+    /// Constructs a [`CommittedState`] from an executed block and accumulated fees.
+    pub fn from_executed_block(
+        executed_block: Option<&BuiltPayloadExecutedBlock<OpPrimitives>>,
+        fees: U256,
+    ) -> Self {
+        let Some(executed_block) = executed_block else {
+            return Self {
+                transactions: vec![],
+                receipts: vec![],
+                gas_used: 0,
+                fees: U256::ZERO,
+                bundle: BundleState::default(),
+            };
+        };
+
+        let gas_used = executed_block.recovered_block.gas_used();
+        let bundle = executed_block.execution_output.state.clone();
+
+        let transactions: Vec<_> = executed_block
+            .recovered_block
+            .clone_transactions_recovered()
+            .enumerate()
+            .map(|(index, tx)| (index as BlockAccessIndex, tx))
+            .collect();
+
+        let receipts: Vec<_> = executed_block
+            .execution_output
+            .result
+            .receipts
+            .iter()
+            .cloned()
+            .enumerate()
+            .map(|(index, r)| (index as BlockAccessIndex, r))
+            .collect();
+
+        Self {
+            transactions,
+            receipts,
+            gas_used,
+            fees,
+            bundle,
+        }
+    }
+}
+
 impl<R> TryFrom<Option<&OpBuiltPayload>> for CommittedState<R>
 where
     R: OpReceiptBuilder<Transaction = OpTransactionSigned, Receipt = OpReceipt> + Default,
@@ -393,42 +453,10 @@ where
             let executed_block = value
                 .executed_block()
                 .ok_or(BalExecutorError::MissingExecutedBlock)?;
-
-            let gas_used = executed_block.recovered_block.gas_used();
-
-            let bundle = value
-                .executed_block()
-                .ok_or(BalExecutorError::MissingExecutedBlock)?
-                .execution_output
-                .state
-                .clone();
-
-            let fees = value.fees();
-
-            let transactions: Vec<_> = executed_block
-                .recovered_block
-                .clone_transactions_recovered()
-                .enumerate()
-                .map(|(index, tx)| (index as BlockAccessIndex, tx))
-                .collect();
-
-            let receipts: Vec<_> = executed_block
-                .execution_output
-                .result
-                .receipts
-                .iter()
-                .cloned()
-                .enumerate()
-                .map(|(index, r)| (index as BlockAccessIndex, r))
-                .collect();
-
-            Ok(Self {
-                transactions,
-                receipts,
-                gas_used,
-                fees,
-                bundle,
-            })
+            Ok(Self::from_executed_block(
+                Some(&executed_block),
+                value.fees(),
+            ))
         } else {
             Ok(Self {
                 transactions: vec![],