Added drop impl to clean up after tests

Changed revert tests a bit

fmt

Add txpool test
Fix DA config, set it via miner
Extend tests

Add op-alloy-flz dependency and update related configurations

- Added `op-alloy-flz` as a dependency in `Cargo.toml` and `Cargo.lock`.
- Configured `op-alloy-flz` to be part of the workspace in `op-rbuilder`'s `Cargo.toml`.
- Updated the `payload_builder_vanilla.rs` to utilize `op-alloy-flz` for transaction size estimation.
- Enhanced test framework to include new data availability tests ensuring block size limits are respected.

Add max data availability transaction and block size configuration

- Introduced `max_da_tx_size` and `max_da_block_size` fields in `TestHarnessBuilder` and `OpRbuilderConfig`.
- Added builder methods `with_max_da_tx_size` and `with_max_da_block_size` for setting these values.
- Implemented a new test to ensure transaction size limits are respected in data availability scenarios.
- Updated test module to include the new data availability test.

Add cumulative_da_bytes_used accur

Add da config

Use correct DA transaction compression

Author: SozinM

Cargo.lock

@@ -132,6 +132,7 @@ op-alloy-rpc-types-engine = { version = "0.16.0", default-features = false }
 op-alloy-rpc-jsonrpsee = { version = "0.16.0", default-features = false }
 op-alloy-network = { version = "0.16.0", default-features = false }
 op-alloy-consensus = { version = "0.16.0", default-features = false }
+op-alloy-flz = { version = "0.13.0", default-features = false }
 
 async-trait = { version = "0.1.83" }
 clap = { version = "4.4.3", features = ["derive", "env"] }

Cargo.toml

@@ -64,6 +64,7 @@ op-alloy-consensus.workspace = true
 op-alloy-rpc-types-engine.workspace = true
 op-alloy-rpc-types.workspace = true
 op-alloy-network.workspace = true
+op-alloy-flz.workspace = true
 
 revm.workspace = true
 op-revm.workspace = true
@@ -110,6 +111,9 @@ tikv-jemallocator = { version = "0.6", optional = true }
 vergen = { workspace = true, features = ["build", "cargo", "emit_and_set"] }
 vergen-git2.workspace = true
 
+[dev-dependencies]
+alloy-provider = {workspace = true, default-features = true, features = ["txpool-api"]}
+
 [features]
 default = ["jemalloc"]
 

crates/op-rbuilder/Cargo.toml

@@ -5,9 +5,8 @@
 //! clap [Args](clap::Args) for optimism rollup configuration
 use std::path::PathBuf;
 
-use reth_optimism_node::args::RollupArgs;
-
 use crate::tx_signer::Signer;
+use reth_optimism_node::args::RollupArgs;
 
 /// Parameters for rollup configuration
 #[derive(Debug, Clone, Default, PartialEq, Eq, clap::Args)]

crates/op-rbuilder/src/args/op.rs

@@ -5,6 +5,7 @@ use reth_optimism_node::{
     node::{OpAddOnsBuilder, OpPoolBuilder},
     OpNode,
 };
+use reth_optimism_payload_builder::config::OpDAConfig;
 use reth_transaction_pool::TransactionPool;
 
 /// CLI argument parsing.
@@ -56,6 +57,7 @@ fn main() {
         .run(|builder, builder_args| async move {
             let rollup_args = builder_args.rollup_args;
 
+            let da_config = OpDAConfig::default();
             let op_node = OpNode::new(rollup_args.clone());
             let handle = builder
                 .with_types::<OpNode>()
@@ -78,6 +80,7 @@ fn main() {
                         .payload(CustomOpPayloadBuilder::new(
                             builder_args.builder_signer,
                             std::time::Duration::from_secs(builder_args.extra_block_deadline_secs),
+                            da_config.clone(),
                             builder_args.flashblocks_ws_url,
                             builder_args.chain_block_time,
                             builder_args.flashblock_block_time,
@@ -87,6 +90,7 @@ fn main() {
                     OpAddOnsBuilder::default()
                         .with_sequencer(rollup_args.sequencer.clone())
                         .with_enable_tx_conditional(rollup_args.enable_tx_conditional)
+                        .with_da_config(da_config)
                         .build(),
                 )
                 .extend_rpc_modules(move |ctx| {

crates/op-rbuilder/src/main.rs

@@ -38,11 +38,12 @@ use reth_optimism_evm::{OpEvmConfig, OpNextBlockEnvAttributes};
 use reth_optimism_forks::OpHardforks;
 use reth_optimism_node::OpEngineTypes;
 use reth_optimism_payload_builder::{
+    config::OpDAConfig,
     error::OpPayloadBuilderError,
     payload::{OpBuiltPayload, OpPayloadBuilderAttributes},
 };
 use reth_optimism_primitives::{OpPrimitives, OpReceipt, OpTransactionSigned};
-use reth_optimism_txpool::OpPooledTx;
+use reth_optimism_txpool::{estimated_da_size::DataAvailabilitySized, OpPooledTx};
 use reth_payload_builder::PayloadBuilderService;
 use reth_payload_builder_primitives::PayloadBuilderError;
 use reth_payload_primitives::PayloadBuilderAttributes;
@@ -96,6 +97,7 @@ struct FlashblocksMetadata<N: NodePrimitives> {
 pub struct CustomOpPayloadBuilder {
     #[expect(dead_code)]
     builder_signer: Option<Signer>,
+    da_config: OpDAConfig,
     flashblocks_ws_url: String,
     chain_block_time: u64,
     flashblock_block_time: u64,
@@ -106,12 +108,14 @@ impl CustomOpPayloadBuilder {
     pub fn new(
         builder_signer: Option<Signer>,
         extra_block_deadline: std::time::Duration,
+        da_config: OpDAConfig,
         flashblocks_ws_url: String,
         chain_block_time: u64,
         flashblock_block_time: u64,
     ) -> Self {
         Self {
             builder_signer,
+            da_config,
             flashblocks_ws_url,
             chain_block_time,
             flashblock_block_time,
@@ -132,6 +136,7 @@ where
     Pool: TransactionPool<Transaction: PoolTransaction<Consensus = TxTy<Node::Types>>>
         + Unpin
         + 'static,
+    <Pool as TransactionPool>::Transaction: OpPooledTx,
     Evm: ConfigureEvm<
             Primitives = PrimitivesTy<Node::Types>,
             NextBlockEnvCtx = OpNextBlockEnvAttributes,
@@ -149,6 +154,7 @@ where
             OpEvmConfig::optimism(ctx.chain_spec()),
             pool,
             ctx.provider().clone(),
+            self.da_config,
             self.flashblocks_ws_url.clone(),
             self.chain_block_time,
             self.flashblock_block_time,
@@ -241,6 +247,8 @@ pub struct OpPayloadBuilder<Pool, Client> {
     pub pool: Pool,
     /// Node client
     pub client: Client,
+    /// DA settings.
+    pub da_config: OpDAConfig,
     /// Channel sender for publishing messages
     pub tx: mpsc::UnboundedSender<String>,
     /// chain block time
@@ -259,6 +267,7 @@ impl<Pool, Client> OpPayloadBuilder<Pool, Client> {
         evm_config: OpEvmConfig,
         pool: Pool,
         client: Client,
+        da_config: OpDAConfig,
         flashblocks_ws_url: String,
         chain_block_time: u64,
         flashblock_block_time: u64,
@@ -276,6 +285,7 @@ impl<Pool, Client> OpPayloadBuilder<Pool, Client> {
             evm_config,
             pool,
             client,
+            da_config,
             tx,
             chain_block_time,
             flashblock_block_time,
@@ -337,7 +347,8 @@ impl<Pool, Client> OpPayloadBuilder<Pool, Client> {
 
 impl<Pool, Client> OpPayloadBuilder<Pool, Client>
 where
-    Pool: TransactionPool<Transaction: PoolTransaction<Consensus = OpTransactionSigned>>,
+    Pool:
+        TransactionPool<Transaction: PoolTransaction<Consensus = OpTransactionSigned> + OpPooledTx>,
     Client: StateProviderFactory + ChainSpecProvider<ChainSpec: EthChainSpec + OpHardforks>,
 {
     /// Send a message to be published
@@ -394,6 +405,7 @@ where
 
         let ctx = OpPayloadBuilderCtx {
             evm_config: self.evm_config.clone(),
+            da_config: self.da_config.clone(),
             chain_spec: self.client.chain_spec(),
             config,
             evm_env,
@@ -443,8 +455,17 @@ where
 
         let gas_per_batch = ctx.block_gas_limit() / self.flashblocks_per_block;
 
+        // We initially set total_gas_per_batch equal to 1 flashblocks portion and then
+        // add additional portions to the limit as we build more flashblocks
         let mut total_gas_per_batch = gas_per_batch;
 
+        let da_per_batch = if let Some(da_limit) = ctx.da_config.max_da_block_size() {
+            Some(da_limit / self.flashblocks_per_block)
+        } else {
+            None
+        };
+        let mut total_da_per_batch = da_per_batch;
+
         let mut flashblock_count = 0;
         // Create a channel to coordinate flashblock building
         let (build_tx, mut build_rx) = mpsc::channel(1);
@@ -513,9 +534,10 @@ where
                     // Continue with flashblock building
                     tracing::info!(
                         target: "payload_builder",
-                        "Building flashblock {} {}",
+                        "Building flashblock idx={} target_gas={} taget_da={}",
                         flashblock_count,
                         total_gas_per_batch,
+                        total_da_per_batch.unwrap_or(0),
                     );
 
                     let flashblock_build_start_time = Instant::now();
@@ -542,6 +564,7 @@ where
                         &mut db,
                         best_txs,
                         total_gas_per_batch.min(ctx.block_gas_limit()),
+                        total_da_per_batch,
                     )?;
                     ctx.metrics
                         .payload_tx_simulation_duration
@@ -596,6 +619,9 @@ where
                             // Update bundle_state for next iteration
                             bundle_state = new_bundle_state;
                             total_gas_per_batch += gas_per_batch;
+                            if let Some(da_limit) = da_per_batch {
+                                total_da_per_batch.as_mut().map(|da| *da += da_limit);
+                            }
                             flashblock_count += 1;
                             tracing::info!(target: "payload_builder", "Flashblock {} built", flashblock_count);
                         }
@@ -617,7 +643,8 @@ where
 impl<Pool, Client> PayloadBuilder for OpPayloadBuilder<Pool, Client>
 where
     Client: StateProviderFactory + ChainSpecProvider<ChainSpec: EthChainSpec + OpHardforks> + Clone,
-    Pool: TransactionPool<Transaction: PoolTransaction<Consensus = OpTransactionSigned>>,
+    Pool:
+        TransactionPool<Transaction: PoolTransaction<Consensus = OpTransactionSigned> + OpPooledTx>,
 {
     type Attributes = OpPayloadBuilderAttributes<OpTransactionSigned>;
     type BuiltPayload = OpBuiltPayload;
@@ -889,6 +916,8 @@ impl<T: PoolTransaction> OpPayloadTransactions<T> for () {
 pub struct OpPayloadBuilderCtx<ChainSpec> {
     /// The type that knows how to perform system calls and configure the evm.
     pub evm_config: OpEvmConfig,
+    /// The DA config for the payload builder
+    pub da_config: OpDAConfig,
     /// The chainspec
     pub chain_spec: Arc<ChainSpec>,
     /// How to build the payload.
@@ -1145,9 +1174,10 @@ where
         info: &mut ExecutionInfo<OpPrimitives>,
         db: &mut State<DB>,
         mut best_txs: impl PayloadTransactions<
-            Transaction: PoolTransaction<Consensus = OpTransactionSigned>,
+            Transaction: PoolTransaction<Consensus = OpTransactionSigned> + OpPooledTx,
         >,
         batch_gas_limit: u64,
+        block_da_limit: Option<u64>,
     ) -> Result<Option<()>, PayloadBuilderError>
     where
         DB: Database<Error = ProviderError>,
@@ -1157,10 +1187,11 @@ where
         let mut num_txs_simulated = 0;
         let mut num_txs_simulated_success = 0;
         let mut num_txs_simulated_fail = 0;
-
+        let tx_da_limit = self.da_config.max_da_tx_size();
         let mut evm = self.evm_config.evm_with_env(&mut *db, self.evm_env.clone());
 
         while let Some(tx) = best_txs.next(()) {
+            let tx_da_size = tx.estimated_da_size();
             let tx = tx.into_consensus();
             num_txs_considered += 1;
 
@@ -1170,7 +1201,13 @@ where
             }
 
             // ensure we still have capacity for this transaction
-            if info.is_tx_over_limits(tx.inner(), batch_gas_limit, None, None) {
+            if info.is_tx_over_limits(
+                tx_da_size,
+                batch_gas_limit,
+                tx_da_limit,
+                block_da_limit,
+                tx.gas_limit(),
+            ) {
                 // we can't fit this transaction into the block, so we need to mark it as
                 // invalid which also removes all dependent transaction from
                 // the iterator before we can continue
@@ -1227,6 +1264,8 @@ where
             // add gas used by the transaction to cumulative gas used, before creating the receipt
             let gas_used = result.gas_used();
             info.cumulative_gas_used += gas_used;
+            // record tx da size
+            info.cumulative_da_bytes_used += tx_da_size;
 
             let ctx = ReceiptBuilderCtx {
                 tx: tx.inner(),