Use correct DA transaction compression

Cargo.toml

@@ -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"] }

crates/op-rbuilder/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/src/args/op.rs

@@ -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/builders/context.rs

@@ -1,7 +1,7 @@
 use alloy_consensus::{Eip658Value, Transaction, TxEip1559};
-use alloy_eips::Typed2718;
+use alloy_eips::{Encodable2718, Typed2718};
 use alloy_op_evm::block::receipt_builder::OpReceiptBuilder;
-use alloy_primitives::{private::alloy_rlp::Encodable, Address, Bytes, TxKind, U256};
+use alloy_primitives::{Address, Bytes, TxKind, U256};
 use alloy_rpc_types_eth::Withdrawals;
 use core::fmt::Debug;
 use op_alloy_consensus::{OpDepositReceipt, OpTypedTransaction};
@@ -19,6 +19,7 @@ use reth_optimism_forks::OpHardforks;
 use reth_optimism_node::OpPayloadBuilderAttributes;
 use reth_optimism_payload_builder::{config::OpDAConfig, error::OpPayloadBuilderError};
 use reth_optimism_primitives::{OpReceipt, OpTransactionSigned};
+use reth_optimism_txpool::estimated_da_size::DataAvailabilitySized;
 use reth_payload_builder::PayloadId;
 use reth_primitives::{Recovered, SealedHeader};
 use reth_primitives_traits::{InMemorySize, SignedTransaction};
@@ -329,11 +330,18 @@ impl OpPayloadBuilderCtx {
 
         while let Some(tx) = best_txs.next(()) {
             let exclude_reverting_txs = tx.exclude_reverting_txs();
+            let tx_da_size = tx.estimated_da_size();
 
             let tx = tx.into_consensus();
             num_txs_considered += 1;
             // ensure we still have capacity for this transaction
-            if info.is_tx_over_limits(tx.inner(), block_gas_limit, tx_da_limit, block_da_limit) {
+            if info.is_tx_over_limits(
+                tx_da_size,
+                block_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
@@ -394,6 +402,8 @@ impl OpPayloadBuilderCtx {
             // 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;
 
             // Push transaction changeset and calculate header bloom filter for receipt.
             let ctx = ReceiptBuilderCtx {
@@ -498,7 +508,7 @@ impl OpPayloadBuilderCtx {
         db: &mut State<DB>,
         builder_tx_gas: u64,
         message: Vec<u8>,
-    ) -> Option<usize>
+    ) -> Option<u64>
     where
         DB: Database<Error = ProviderError>,
     {
@@ -509,7 +519,9 @@ impl OpPayloadBuilderCtx {
                 // Create and sign the transaction
                 let builder_tx =
                     signed_builder_tx(db, builder_tx_gas, message, signer, base_fee, chain_id)?;
-                Ok(builder_tx.length())
+                Ok(op_alloy_flz::tx_estimated_size_fjord(
+                    builder_tx.encoded_2718().as_slice(),
+                ))
             })
             .transpose()
             .unwrap_or_else(|err: PayloadBuilderError| {

crates/op-rbuilder/src/builders/flashblocks/payload.rs

@@ -222,10 +222,11 @@ where
 
         let gas_per_batch = ctx.block_gas_limit() / self.config.flashblocks_per_block();
         let mut total_gas_per_batch = gas_per_batch;
-        let total_da_bytes_per_batch = ctx
+        let da_per_batch = ctx
             .da_config
             .max_da_block_size()
-            .map(|limit| limit / self.config.flashblocks_per_block());
+            .map(|da_limit| da_limit / self.config.flashblocks_per_block());
+        let mut total_da_per_batch = da_per_batch;
 
         let mut flashblock_count = 0;
         // Create a channel to coordinate flashblock building
@@ -295,11 +296,11 @@ 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();
                     let state = StateProviderDatabase::new(&state_provider);
 
@@ -324,7 +325,7 @@ where
                         &mut db,
                         best_txs,
                         total_gas_per_batch.min(ctx.block_gas_limit()),
-                        total_da_bytes_per_batch,
+                        total_da_per_batch,
                     )?;
                     ctx.metrics
                         .payload_tx_simulation_duration
@@ -377,6 +378,13 @@ 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 {
+                                if let Some(da) = total_da_per_batch.as_mut() {
+                                    *da += da_limit;
+                                } else {
+                                    error!("Builder end up in faulty invariant, if da_per_batch is set then total_da_per_batch must be set");
+                                }
+                            }
                             flashblock_count += 1;
                             tracing::info!(target: "payload_builder", "Flashblock {} built", flashblock_count);
                         }

crates/op-rbuilder/src/builders/standard/payload.rs

@@ -355,15 +355,7 @@ impl<Txs: PayloadTxsBounds> OpBuilder<'_, Txs> {
         let block_da_limit = ctx
             .da_config
             .max_da_block_size()
-            .map(|da_size| da_size - builder_tx_da_size as u64);
-        // Check that it's possible to create builder tx, considering max_da_tx_size, otherwise panic
-        if let Some(tx_da_limit) = ctx.da_config.max_da_tx_size() {
-            // Panic indicate max_da_tx_size misconfiguration
-            assert!(
-                tx_da_limit >= builder_tx_da_size as u64,
-                "The configured da_config.max_da_tx_size is too small to accommodate builder tx."
-            );
-        }
+            .map(|da_size| da_size.saturating_sub(builder_tx_da_size));
 
         if !ctx.attributes().no_tx_pool {
             let best_txs_start_time = Instant::now();

crates/op-rbuilder/src/main.rs

@@ -68,7 +68,7 @@ where
     cli.run(|builder, builder_args| async move {
         let builder_config = BuilderConfig::<B::Config>::try_from(builder_args.clone())
             .expect("Failed to convert rollup args to builder config");
-
+        let da_config = builder_config.da_config.clone();
         let rollup_args = builder_args.rollup_args;
         let op_node = OpNode::new(rollup_args.clone());
         let handle = builder
@@ -95,6 +95,7 @@ where
                 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/primitives/reth/execution.rs

@@ -1,6 +1,5 @@
 //! Heavily influenced by [reth](https://github.com/paradigmxyz/reth/blob/1e965caf5fa176f244a31c0d2662ba1b590938db/crates/optimism/payload/src/builder.rs#L570)
-use alloy_consensus::Transaction;
-use alloy_primitives::{private::alloy_rlp::Encodable, Address, U256};
+use alloy_primitives::{Address, U256};
 use core::fmt::Debug;
 use reth_optimism_primitives::{OpReceipt, OpTransactionSigned};
 
@@ -44,21 +43,22 @@ impl<T: Debug + Default> ExecutionInfo<T> {
     ///   maximum allowed DA limit per block.
     pub fn is_tx_over_limits(
         &self,
-        tx: &OpTransactionSigned,
+        tx_da_size: u64,
         block_gas_limit: u64,
         tx_data_limit: Option<u64>,
         block_data_limit: Option<u64>,
+        tx_gas_limit: u64,
     ) -> bool {
-        if tx_data_limit.is_some_and(|da_limit| tx.length() as u64 > da_limit) {
+        if tx_data_limit.is_some_and(|da_limit| tx_da_size > da_limit) {
             return true;
         }
 
         if block_data_limit
-            .is_some_and(|da_limit| self.cumulative_da_bytes_used + (tx.length() as u64) > da_limit)
+            .is_some_and(|da_limit| self.cumulative_da_bytes_used + tx_da_size > da_limit)
         {
             return true;
         }
 
-        self.cumulative_gas_used + tx.gas_limit() > block_gas_limit
+        self.cumulative_gas_used + tx_gas_limit > block_gas_limit
     }
 }

crates/op-rbuilder/src/tests/framework/blocks.rs

@@ -372,4 +372,12 @@ impl BlockGenerated {
     pub fn includes(&self, tx_hash: B256) -> bool {
         self.block.transactions.hashes().any(|hash| hash == tx_hash)
     }
+
+    pub fn includes_vec(&self, tx_hashes: Vec<B256>) -> bool {
+        tx_hashes.iter().all(|hash| self.includes(*hash))
+    }
+
+    pub fn not_includes_vec(&self, tx_hashes: Vec<B256>) -> bool {
+        tx_hashes.iter().all(|hash| self.not_includes(*hash))
+    }
 }

crates/op-rbuilder/src/tests/framework/harness.rs

@@ -25,6 +25,8 @@ pub struct TestHarnessBuilder {
     flashblocks_ws_url: Option<String>,
     chain_block_time: Option<u64>,
     flashbots_block_time: Option<u64>,
+    namespaces: Option<String>,
+    extra_params: Option<String>,
 }
 
 impl TestHarnessBuilder {
@@ -35,6 +37,8 @@ impl TestHarnessBuilder {
             flashblocks_ws_url: None,
             chain_block_time: None,
             flashbots_block_time: None,
+            namespaces: None,
+            extra_params: None,
         }
     }
 
@@ -58,6 +62,16 @@ impl TestHarnessBuilder {
         self
     }
 
+    pub fn with_namespaces(mut self, namespaces: &str) -> Self {
+        self.namespaces = Some(namespaces.to_string());
+        self
+    }
+
+    pub fn with_extra_params(mut self, extra_params: &str) -> Self {
+        self.extra_params = Some(extra_params.to_string());
+        self
+    }
+
     pub async fn build(self) -> eyre::Result<TestHarness> {
         let mut framework = IntegrationFramework::new(&self.name).unwrap();
 
@@ -69,7 +83,7 @@ impl TestHarnessBuilder {
         std::fs::write(&genesis_path, genesis)?;
 
         // create the builder
-        let builder_data_dir = std::env::temp_dir().join(Uuid::new_v4().to_string());
+        let builder_data_dir: PathBuf = std::env::temp_dir().join(Uuid::new_v4().to_string());
         let builder_auth_rpc_port = get_available_port();
         let builder_http_port = get_available_port();
         let mut op_rbuilder_config = OpRbuilderConfig::new()
@@ -79,8 +93,9 @@ impl TestHarnessBuilder {
             .network_port(get_available_port())
             .http_port(builder_http_port)
             .with_builder_private_key(BUILDER_PRIVATE_KEY)
-            .with_revert_protection(self.use_revert_protection);
-
+            .with_revert_protection(self.use_revert_protection)
+            .with_namespaces(self.namespaces)
+            .with_extra_params(self.extra_params);
         if let Some(flashblocks_ws_url) = self.flashblocks_ws_url {
             op_rbuilder_config = op_rbuilder_config.with_flashblocks_ws_url(&flashblocks_ws_url);
         }
@@ -113,7 +128,7 @@ impl TestHarnessBuilder {
         let builder_log_path = builder.log_path.clone();
 
         Ok(TestHarness {
-            _framework: framework,
+            framework: framework,
             builder_auth_rpc_port,
             builder_http_port,
             validator_auth_rpc_port,
@@ -123,7 +138,7 @@ impl TestHarnessBuilder {
 }
 
 pub struct TestHarness {
-    _framework: IntegrationFramework,
+    framework: IntegrationFramework,
     builder_auth_rpc_port: u16,
     builder_http_port: u16,
     validator_auth_rpc_port: u16,
@@ -237,6 +252,16 @@ impl TestHarness {
     }
 }
 
+impl Drop for TestHarness {
+    fn drop(&mut self) {
+        for service in &mut self.framework.services {
+            let res = service.stop();
+            if let Err(e) = res {
+                println!("Failed to stop service: {}", e);
+            }
+        }
+    }
+}
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum TransactionStatus {
     NotFound,

crates/op-rbuilder/src/tests/framework/op.rs

@@ -27,6 +27,8 @@ pub struct OpRbuilderConfig {
     chain_block_time: Option<u64>,
     flashbots_block_time: Option<u64>,
     with_revert_protection: Option<bool>,
+    namespaces: Option<String>,
+    extra_params: Option<String>,
 }
 
 impl OpRbuilderConfig {
@@ -83,6 +85,16 @@ impl OpRbuilderConfig {
         self.flashbots_block_time = Some(time);
         self
     }
+
+    pub fn with_namespaces(mut self, namespaces: Option<String>) -> Self {
+        self.namespaces = namespaces;
+        self
+    }
+
+    pub fn with_extra_params(mut self, extra_params: Option<String>) -> Self {
+        self.extra_params = extra_params;
+        self
+    }
 }
 
 impl Service for OpRbuilderConfig {
@@ -137,9 +149,7 @@ impl Service for OpRbuilderConfig {
         if let Some(http_port) = self.http_port {
             cmd.arg("--http")
                 .arg("--http.port")
-                .arg(http_port.to_string())
-                .arg("--http.api")
-                .arg("eth,web3,txpool");
+                .arg(http_port.to_string());
         }
 
         if let Some(flashblocks_ws_url) = &self.flashblocks_ws_url {
@@ -159,6 +169,14 @@ impl Service for OpRbuilderConfig {
                 .arg(flashbots_block_time.to_string());
         }
 
+        if let Some(namespaces) = &self.namespaces {
+            cmd.arg("--http.api").arg(namespaces);
+        }
+
+        if let Some(extra_params) = &self.extra_params {
+            cmd.args(extra_params.split_ascii_whitespace());
+        }
+
         cmd
     }