Proposal bulding in separate thread (#898)

* Extended poposal transaction monitor with transaction building.

Building the blob sidecar (EIP-7584) takes more than 0.5 seconds. Moving
it to a separate thread keeps the main pre-confirmation heartbeat
running at consistent intervals.
Proposal builder passed as parameter for transaction monitor.

* Updated dep

* review, fixed wrong handling of error from submitted trasactions

* Corrected handling not submitted proposal which need to be submitted
again

* comment

* Update shasta/src/node/proposal_manager/batch_builder.rs

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* Update shasta/src/l1/proposal_tx_builder.rs

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* Build failed error instead of estimation error

* Renamed batch to proposal for shasta (#901)

---------

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Author: mskrzypkows

Cargo.lock

@@ -1,5 +1,6 @@
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub enum TransactionError {
+    BuildFailed,
     EstimationFailed,
     EstimationTooEarly,
     TransactionReverted,

common/src/l1/transaction_error.rs

@@ -80,6 +80,7 @@ impl L2Engine {
         }
     }
 
+    /// Batch (pacaya) is equivalent to the proposal (shasta).
     pub async fn get_pending_l2_tx_list(
         &self,
         base_fee: u64,

common/src/l2/engine.rs

@@ -13,12 +13,20 @@ use alloy::{
 };
 use alloy_json_rpc::RpcError;
 use anyhow::Error;
+use std::future::Future;
 use std::{sync::Arc, time::Duration};
 use tokio::sync::Mutex;
 use tokio::sync::mpsc::Sender;
 use tokio::task::JoinHandle;
 use tracing::{debug, error, info, warn};
 
+/// Trait for types that can asynchronously build a `TransactionRequest`.
+/// Implement this on protocol-specific builders (e.g. `ProposalTxBuilder`)
+/// to pass them into `monitor_new_transaction_with_builder`.
+pub trait TransactionRequestBuilder: Send + 'static {
+    fn build(self) -> impl Future<Output = Result<TransactionRequest, TransactionError>> + Send;
+}
+
 // Transaction status enum
 #[derive(Debug, Clone, PartialEq)]
 pub enum TxStatus {
@@ -118,6 +126,36 @@ impl TransactionMonitor {
         Ok(())
     }
 
+    /// Monitor a transaction built by a deferred builder.
+    /// The builder future is awaited inside the spawned task, so this method returns immediately.
+    /// If the builder fails, the error is sent via the error notification channel.
+    pub async fn monitor_new_transaction_with_builder(
+        &self,
+        tx_builder: impl TransactionRequestBuilder,
+        nonce: u64,
+    ) -> Result<(), Error> {
+        let mut guard = self.join_handle.lock().await;
+        if let Some(join_handle) = guard.as_ref()
+            && !join_handle.is_finished()
+        {
+            return Err(Error::msg(
+                "Cannot monitor new transaction, previous transaction is in progress",
+            ));
+        }
+
+        let monitor_thread = TransactionMonitorThread::new(
+            self.provider.clone(),
+            self.config.clone(),
+            nonce,
+            self.error_notification_channel.clone(),
+            self.metrics.clone(),
+            self.chain_id,
+        );
+        let join_handle = monitor_thread.spawn_monitoring_task_with_builder(tx_builder);
+        *guard = Some(join_handle);
+        Ok(())
+    }
+
     pub async fn is_transaction_in_progress(&self) -> Result<bool, Error> {
         let guard = self.join_handle.lock().await;
         if let Some(join_handle) = guard.as_ref() {
@@ -152,6 +190,23 @@ impl TransactionMonitorThread {
         })
     }
 
+    pub fn spawn_monitoring_task_with_builder(
+        mut self,
+        tx_builder: impl TransactionRequestBuilder,
+    ) -> JoinHandle<()> {
+        tokio::spawn(async move {
+            match tx_builder.build().await {
+                Ok(tx) => {
+                    self.monitor_transaction(tx).await;
+                }
+                Err(err) => {
+                    error!("Transaction builder failed: {}", err);
+                    self.send_error_signal(err).await;
+                }
+            }
+        })
+    }
+
     async fn monitor_transaction(&mut self, mut tx: TransactionRequest) {
         tx.set_nonce(self.nonce);
         if !matches!(tx.buildable_type(), Some(TxType::Eip1559 | TxType::Eip4844)) {

common/src/shared/transaction_monitor.rs

@@ -746,6 +746,10 @@ impl Node {
                 warn!("Propose batch transaction executed too late.");
                 return Ok(());
             }
+            TransactionError::BuildFailed => {
+                self.cancel_token.cancel_on_critical_error();
+                return Err(anyhow::anyhow!("Transaction build failed, exiting"));
+            }
         }
 
         Ok(())

pacaya/src/node/mod.rs

@@ -308,6 +308,10 @@ impl Node {
                 warn!("Proposal transaction executed too late.");
                 Ok(())
             }
+            TransactionError::BuildFailed => {
+                self.cancel_token.cancel_on_critical_error();
+                Err(anyhow::anyhow!("Transaction build failed, exiting"))
+            }
         }
     }
 }

permissionless/src/node/mod.rs

@@ -180,7 +180,7 @@ impl PreconfOperator for ExecutionLayer {
 }
 
 impl ExecutionLayer {
-    pub async fn send_batch_to_l1(
+    pub async fn send_proposal_to_l1(
         &self,
         l2_blocks: Vec<L2BlockV2>,
         num_forced_inclusion: u16,
@@ -191,28 +191,27 @@ impl ExecutionLayer {
             num_forced_inclusion,
         );
 
-        // Build propose transaction
-        let builder = ProposalTxBuilder::new(self.provider.clone(), self.extra_gas_percentage);
-        let tx = builder
-            .build_propose_tx(
-                l2_blocks,
-                self.common().preconfer_address(),
-                self.contract_addresses.shasta_inbox,
-                num_forced_inclusion,
-            )
-            .await
-            .map_err(|e| Error::msg(format!("build_propose_tx failed: {e}")))?;
-
         let pending_nonce = self.get_preconfer_nonce_pending().await.map_err(|e| {
             Error::msg(format!(
-                "get_preconfer_nonce_pending (send_batch_to_l1) failed: {e}"
+                "get_preconfer_nonce_pending (send_proposal_to_l1) failed: {e}"
             ))
         })?;
-        // Spawn a monitor for this transaction
+
+        // Build the transaction asynchronously inside the monitor's spawned task.
+        // This moves the ~650ms KZG sidecar computation off the hot path.
+        let tx_builder = ProposalTxBuilder::new(
+            self.provider.clone(),
+            self.extra_gas_percentage,
+            l2_blocks,
+            self.common().preconfer_address(),
+            self.contract_addresses.shasta_inbox,
+            num_forced_inclusion,
+        );
+
         self.transaction_monitor
-            .monitor_new_transaction(tx, pending_nonce)
+            .monitor_new_transaction_with_builder(tx_builder, pending_nonce)
             .await
-            .map_err(|e| Error::msg(format!("Sending batch to L1 failed: {e}")))?;
+            .map_err(|e| Error::msg(format!("Sending proposal to L1 failed: {e}")))?;
 
         Ok(())
     }

shasta/src/l1/execution_layer.rs

@@ -1,5 +1,6 @@
 use alloy::{
     consensus::SidecarBuilder,
+    eips::eip7594::BlobTransactionSidecarEip7594,
     network::{TransactionBuilder, TransactionBuilder7594},
     primitives::{
         Address, Bytes,
@@ -12,36 +13,91 @@ use alloy_json_rpc::RpcError;
 use anyhow::{Context, Error};
 use common::l1::{fees_per_gas::FeesPerGas, tools, transaction_error::TransactionError};
 use common::shared::l2_block_v2::L2BlockV2;
+use common::shared::transaction_monitor::TransactionRequestBuilder;
 use taiko_bindings::inbox::{IInbox::ProposeInput, Inbox, LibBlobs::BlobReference};
 use taiko_protocol::shasta::{
     BlobCoder,
     manifest::{BlockManifest, DerivationSourceManifest},
 };
-use tracing::warn;
+use tracing::{info, warn};
+
+/// Build the EIP-7594 blob sidecar from L2 blocks. This is a CPU-intensive operation
+/// (KZG commitment + cell proof computation).
+fn build_sidecar_from_l2_blocks(
+    l2_blocks: &[L2BlockV2],
+) -> Result<BlobTransactionSidecarEip7594, Error> {
+    let start = std::time::Instant::now();
+
+    let mut block_manifests = Vec::with_capacity(l2_blocks.len());
+    for l2_block in l2_blocks {
+        block_manifests.push(BlockManifest {
+            timestamp: l2_block.timestamp_sec,
+            coinbase: l2_block.coinbase,
+            anchor_block_number: l2_block.anchor_block_number,
+            gas_limit: l2_block.gas_limit_without_anchor,
+            transactions: l2_block
+                .prebuilt_tx_list
+                .tx_list
+                .iter()
+                .map(|tx| tx.clone().into())
+                .collect(),
+        });
+    }
+
+    let manifest = DerivationSourceManifest {
+        blocks: block_manifests,
+    };
+
+    let manifest_data = manifest
+        .encode_and_compress()
+        .map_err(|e| Error::msg(format!("Can't encode and compress manifest: {e}")))?;
+
+    let sidecar_builder: SidecarBuilder<BlobCoder> = SidecarBuilder::from_slice(&manifest_data);
+    let sidecar = sidecar_builder
+        .build_7594()
+        .map_err(|e| Error::msg(format!("sidecar builder build_7594 failed: {e}")))?;
+
+    info!(
+        "⏱️ build_sidecar_from_l2_blocks ({} blocks, {} bytes compressed) took {:?}",
+        l2_blocks.len(),
+        manifest_data.len(),
+        start.elapsed()
+    );
+
+    Ok(sidecar)
+}
 
 pub struct ProposalTxBuilder {
     provider: DynProvider,
     extra_gas_percentage: u64,
+    l2_blocks: Vec<L2BlockV2>,
+    from: Address,
+    to: Address,
+    num_forced_inclusion: u16,
 }
 
 impl ProposalTxBuilder {
-    pub fn new(provider: DynProvider, extra_gas_percentage: u64) -> Self {
+    pub fn new(
+        provider: DynProvider,
+        extra_gas_percentage: u64,
+        l2_blocks: Vec<L2BlockV2>,
+        from: Address,
+        to: Address,
+        num_forced_inclusion: u16,
+    ) -> Self {
         Self {
             provider,
             extra_gas_percentage,
+            l2_blocks,
+            from,
+            to,
+            num_forced_inclusion,
         }
     }
 
-    #[allow(clippy::too_many_arguments)]
-    pub async fn build_propose_tx(
-        &self,
-        l2_blocks: Vec<L2BlockV2>,
-        from: Address,
-        to: Address,
-        num_forced_inclusion: u16,
-    ) -> Result<TransactionRequest, Error> {
+    async fn build_propose_tx(&self) -> Result<TransactionRequest, Error> {
         let tx_blob = self
-            .build_propose_blob(l2_blocks, from, to, num_forced_inclusion)
+            .build_propose_blob()
             .await
             .map_err(|e| Error::msg(format!("build_propose_blob failed: {e}")))?;
         let tx_blob_gas = match self.provider.estimate_gas(tx_blob.clone()).await {
@@ -80,44 +136,8 @@ impl ProposalTxBuilder {
         Ok(tx_blob)
     }
 
-    #[allow(clippy::too_many_arguments)]
-    pub async fn build_propose_blob(
-        &self,
-        l2_blocks: Vec<L2BlockV2>,
-        from: Address,
-        to: Address,
-        num_forced_inclusion: u16,
-    ) -> Result<TransactionRequest, Error> {
-        let mut block_manifests = <Vec<BlockManifest>>::with_capacity(l2_blocks.len());
-        for l2_block in &l2_blocks {
-            // Build the block manifests.
-            block_manifests.push(BlockManifest {
-                timestamp: l2_block.timestamp_sec,
-                coinbase: l2_block.coinbase,
-                anchor_block_number: l2_block.anchor_block_number,
-                gas_limit: l2_block.gas_limit_without_anchor,
-                transactions: l2_block
-                    .prebuilt_tx_list
-                    .tx_list
-                    .iter()
-                    .map(|tx| tx.clone().into())
-                    .collect(),
-            });
-        }
-
-        // Build the proposal manifest.
-        let manifest = DerivationSourceManifest {
-            blocks: block_manifests,
-        };
-
-        let manifest_data = manifest
-            .encode_and_compress()
-            .map_err(|e| Error::msg(format!("Can't encode and compress manifest: {e}")))?;
-
-        let sidecar_builder: SidecarBuilder<BlobCoder> = SidecarBuilder::from_slice(&manifest_data);
-        let sidecar = sidecar_builder
-            .build_7594()
-            .map_err(|e| Error::msg(format!("sidecar builder build_7594 failed: {e}")))?;
+    async fn build_propose_blob(&self) -> Result<TransactionRequest, Error> {
+        let sidecar = build_sidecar_from_l2_blocks(&self.l2_blocks)?;
 
         // Build the propose input.
         let input = ProposeInput {
@@ -131,19 +151,19 @@ impl ProposalTxBuilder {
                     .context("blobs len try_into")?,
                 offset: U24::ZERO,
             },
-            numForcedInclusions: num_forced_inclusion,
+            numForcedInclusions: self.num_forced_inclusion,
         };
 
-        let inbox = Inbox::new(to, self.provider.clone());
+        let inbox = Inbox::new(self.to, self.provider.clone());
         let encoded_proposal_input = inbox
             .encodeProposeInput(input)
             .call()
             .await
             .map_err(|e| Error::msg(format!("inbox encodeProposeInput failed: {e}")))?;
 
         let tx = TransactionRequest::default()
-            .with_from(from)
-            .with_to(to)
+            .with_from(self.from)
+            .with_to(self.to)
             .with_blob_sidecar(sidecar)
             .with_call(&Inbox::proposeCall {
                 _lookahead: Bytes::new(),
@@ -153,3 +173,12 @@ impl ProposalTxBuilder {
         Ok(tx)
     }
 }
+
+impl TransactionRequestBuilder for ProposalTxBuilder {
+    async fn build(self) -> Result<TransactionRequest, TransactionError> {
+        self.build_propose_tx().await.map_err(|e| {
+            e.downcast::<TransactionError>()
+                .unwrap_or(TransactionError::BuildFailed)
+        })
+    }
+}