fix(taiko-client-rs): make Shasta event sync resilient on genesis and pre-finality startup (#21348)

Co-authored-by: Maciej Skrzypkowski <mskr@gmx.com>

Author: MatusKysel

packages/taiko-client-rs/crates/driver/src/sync/engine.rs

@@ -275,7 +275,12 @@ fn envelope_into_submission(
 ) -> (ExecutionPayloadInputV2, B256, u64) {
     match envelope.execution_payload {
         ExecutionPayloadFieldV2::V1(payload) => (
-            ExecutionPayloadInputV2 { execution_payload: payload.clone(), withdrawals: None },
+            // Taiko chains are always post-Shanghai so withdrawals must be non-nil even
+            // when the engine returns a V1 envelope (which omits the withdrawals field).
+            ExecutionPayloadInputV2 {
+                execution_payload: payload.clone(),
+                withdrawals: Some(Vec::new()),
+            },
             payload.block_hash,
             payload.block_number,
         ),

packages/taiko-client-rs/crates/driver/src/sync/error.rs

@@ -56,10 +56,6 @@ pub enum SyncError {
         number: u64,
     },
 
-    /// Event sync: finalized L1 block is unavailable; resume must fail closed.
-    #[error("finalized l1 block is unavailable")]
-    MissingFinalizedL1Block,
-
     /// Event sync: execution engine missing batch-to-block mapping.
     #[error("no execution block found for batch {proposal_id}")]
     MissingExecutionBlockForBatch {

packages/taiko-client-rs/crates/driver/src/sync/event.rs

@@ -201,6 +201,15 @@ impl<P: Provider + Clone> Client<P> {
         let mut payload_value = serde_json::to_value(&payload.execution_payload)
             .map_err(|err| RpcClientError::Other(anyhow!(err)))?;
         if let serde_json::Value::Object(ref mut obj) = payload_value {
+            // Include the withdrawals list so taiko-geth can reconstruct the full block.
+            // The Go driver sends the full ExecutableData (with withdrawals); omitting
+            // this field causes a blockhash mismatch because geth cannot recompute the
+            // withdrawals root from the hash alone.
+            if let Some(ref withdrawals) = payload.withdrawals {
+                let withdrawals_value = serde_json::to_value(withdrawals)
+                    .map_err(|err| RpcClientError::Other(anyhow!(err)))?;
+                obj.insert("withdrawals".to_string(), withdrawals_value);
+            }
             obj.insert(
                 "txHash".to_string(),
                 serde_json::Value::String(format!("{:#066x}", sidecar.tx_hash)),
@@ -224,6 +233,17 @@ impl<P: Provider + Clone> Client<P> {
         forkchoice_state: ForkchoiceState,
         payload_attributes: Option<TaikoPayloadAttributes>,
     ) -> Result<ForkchoiceUpdated> {
+        let forkchoice_state = serde_json::to_value(forkchoice_state)
+            .map_err(|err| RpcClientError::Other(anyhow!(err)))?;
+
+        let payload_attributes = match payload_attributes {
+            Some(payload_attributes) => Some(
+                serde_json::to_value(&payload_attributes)
+                    .map_err(|err| RpcClientError::Other(anyhow!(err)))?,
+            ),
+            None => None,
+        };
+
         self.l2_auth_provider
             .raw_request(
                 Cow::Borrowed(TaikoEngineMethod::ForkchoiceUpdatedV2.as_str()),

packages/taiko-client-rs/crates/rpc/src/auth.rs

@@ -213,6 +213,11 @@ where
             "inserted whitelist preconfirmation block"
         );
 
+        if let Some(ref highest) = self.highest_unsafe_l2_payload_block_id {
+            let mut guard = highest.lock().await;
+            *guard = block_number.max(*guard);
+        }
+
         Ok(true)
     }
 }

packages/taiko-client-rs/crates/whitelist-preconfirmation-driver/src/importer/cache_import.rs

@@ -8,7 +8,7 @@ use alloy_provider::Provider;
 use bindings::preconf_whitelist::PreconfWhitelist::PreconfWhitelistInstance;
 use driver::sync::event::EventSyncer;
 use rpc::client::Client;
-use tokio::sync::mpsc;
+use tokio::sync::{Mutex, mpsc};
 use tracing::{debug, info, warn};
 
 use crate::{
@@ -68,6 +68,8 @@ where
     sequencer_cache: WhitelistSequencerCache,
     /// Command channel used to publish P2P requests/responses.
     network_command_tx: mpsc::Sender<NetworkCommand>,
+    /// Shared highest unsafe L2 payload block ID (updated on P2P import when REST server enabled).
+    highest_unsafe_l2_payload_block_id: Option<Arc<Mutex<u64>>>,
     /// Latched flag indicating event sync has exposed a head L1 origin.
     sync_ready: bool,
     /// Shasta anchor contract address used to validate the first transaction.
@@ -85,6 +87,7 @@ where
         whitelist_address: Address,
         chain_id: u64,
         network_command_tx: mpsc::Sender<NetworkCommand>,
+        highest_unsafe_l2_payload_block_id: Option<Arc<Mutex<u64>>>,
     ) -> Self {
         let whitelist = PreconfWhitelistInstance::new(whitelist_address, rpc.l1_provider.clone());
         let anchor_address = *rpc.shasta.anchor.address();
@@ -99,6 +102,7 @@ where
             request_throttle: RequestThrottle::default(),
             sequencer_cache: WhitelistSequencerCache::default(),
             network_command_tx,
+            highest_unsafe_l2_payload_block_id,
             sync_ready: false,
             anchor_address,
         };

packages/taiko-client-rs/crates/whitelist-preconfirmation-driver/src/importer/mod.rs

@@ -159,7 +159,7 @@ pub struct WhitelistStatus {
     pub highest_unsafe_block_number: u64,
     /// Local libp2p peer ID.
     pub peer_id: String,
-    /// Whether event sync has established a head L1 origin.
+    /// Whether preconfirmation ingress is ready.
     pub sync_ready: bool,
     /// Sequencing lookahead information.
     #[serde(skip_serializing_if = "Option::is_none")]

packages/taiko-client-rs/crates/whitelist-preconfirmation-driver/src/rest/types.rs

@@ -73,8 +73,8 @@ where
     build_preconf_lock: Mutex<()>,
     /// Preconf whitelist contract used for operator checks.
     whitelist: PreconfWhitelistInstance<P>,
-    /// Highest unsafe payload block ID tracked by this node.
-    highest_unsafe_l2_payload_block_id: Mutex<u64>,
+    /// Highest unsafe payload block ID tracked by this node (shared with importer).
+    highest_unsafe_l2_payload_block_id: Arc<Mutex<u64>>,
     /// End-of-sequencing hash cache keyed by epoch.
     end_of_sequencing_by_epoch: Mutex<HashMap<u64, B256>>,
     /// Broadcast channel for REST `/ws` end-of-sequencing notifications.
@@ -98,8 +98,8 @@ where
     pub beacon_client: Arc<BeaconClient>,
     /// Whitelist contract address used for operator checks.
     pub whitelist_address: Address,
-    /// Highest unsafe payload block ID tracked by this node at startup.
-    pub initial_highest_unsafe_l2_payload_block_id: u64,
+    /// Shared highest unsafe payload block ID (also updated by importer on P2P import).
+    pub highest_unsafe_l2_payload_block_id: Arc<Mutex<u64>>,
     /// Channel used to publish messages to the P2P network.
     pub network_command_tx: mpsc::Sender<NetworkCommand>,
     /// Local peer ID string.
@@ -119,7 +119,7 @@ where
             signer,
             beacon_client,
             whitelist_address,
-            initial_highest_unsafe_l2_payload_block_id,
+            highest_unsafe_l2_payload_block_id,
             network_command_tx,
             local_peer_id,
         } = params;
@@ -132,9 +132,7 @@ where
             signer,
             beacon_client,
             whitelist,
-            highest_unsafe_l2_payload_block_id: Mutex::new(
-                initial_highest_unsafe_l2_payload_block_id,
-            ),
+            highest_unsafe_l2_payload_block_id,
             end_of_sequencing_by_epoch: Mutex::new(HashMap::new()),
             eos_notification_tx,
             network_command_tx,
@@ -375,8 +373,14 @@ where
         let started_at = Instant::now();
         let _build_guard = self.build_preconf_lock.lock().await;
 
+        // Guard against building on a genuinely syncing node, but tolerate the false-
+        // positive that taiko-geth emits on genesis chains (currentBlock == highestBlock
+        // == 0, txIndexRemainingBlocks = 1).  When current == highest the node is not
+        // actually catching up to a remote peer, so we allow the build to proceed.
         let sync_status = self.rpc.l2_provider.syncing().await.map_err(provider_err)?;
-        if matches!(sync_status, SyncStatus::Info(_)) {
+        if let SyncStatus::Info(ref info) = sync_status &&
+            info.current_block < info.highest_block
+        {
             return Err(WhitelistPreconfirmationDriverError::Driver(
                 driver::DriverError::EngineSyncing(request.block_number),
             ));
@@ -520,7 +524,9 @@ where
             .get(&current_epoch)
             .copied()
             .map(|hash| hash.to_string());
-        let sync_ready = head_l1_origin_block_id.is_some();
+        // sync_ready reflects ingress readiness, which already includes the confirmed-sync
+        // and scanner-live checks required by the event syncer.
+        let sync_ready = self.event_syncer.is_preconf_ingress_ready();
 
         Ok(WhitelistStatus {
             head_l1_origin_block_id,

packages/taiko-client-rs/crates/whitelist-preconfirmation-driver/src/rest_handler.rs

@@ -14,7 +14,7 @@ use driver::{DriverConfig, map_driver_error};
 use preconfirmation_net::P2pConfig;
 use protocol::signer::FixedKSigner;
 use rpc::beacon::BeaconClient;
-use tokio::time;
+use tokio::{sync::Mutex, time};
 use tracing::{info, warn};
 
 use crate::{
@@ -227,62 +227,67 @@ impl WhitelistPreconfirmationDriverRunner {
         );
 
         // Optionally start the REST/WS server when both rpc_listen_addr and p2p_signer_key
-        // are configured.
-        let mut rest_ws_server = if let (Some(listen_addr), Some(signer_key)) =
-            (self.config.rpc_listen_addr, &self.config.p2p_signer_key)
-        {
-            let beacon_client = Arc::new(
-                BeaconClient::new(self.config.driver_config.l1_beacon_endpoint.clone())
-                    .await
-                    .map_err(|err| WhitelistPreconfirmationDriverError::RestWsServerBeaconInit {
+        // are configured. When enabled, create shared state for highestUnsafeL2PayloadBlockID
+        // so the importer can update it on P2P imports.
+        let (mut rest_ws_server, shared_highest_unsafe) =
+            if let (Some(listen_addr), Some(signer_key)) =
+                (self.config.rpc_listen_addr, &self.config.p2p_signer_key)
+            {
+                let beacon_client = Arc::new(
+                    BeaconClient::new(self.config.driver_config.l1_beacon_endpoint.clone())
+                        .await
+                        .map_err(|err| WhitelistPreconfirmationDriverError::RestWsServerBeaconInit {
                         reason: err.to_string(),
                     })?,
-            );
+                );
 
-            let signer = FixedKSigner::new(signer_key).map_err(|e| {
-                WhitelistPreconfirmationDriverError::Signing(format!(
-                    "failed to create P2P signer: {e}"
-                ))
-            })?;
-            let initial_highest_unsafe_l2_payload_block_id =
-                initial_highest_unsafe_l2_payload_block_id(&preconf_ingress_sync).await;
+                let signer = FixedKSigner::new(signer_key).map_err(|e| {
+                    WhitelistPreconfirmationDriverError::Signing(format!(
+                        "failed to create P2P signer: {e}"
+                    ))
+                })?;
+                let initial_highest_unsafe_l2_payload_block_id =
+                    initial_highest_unsafe_l2_payload_block_id(&preconf_ingress_sync).await;
+                let shared_highest =
+                    Arc::new(Mutex::new(initial_highest_unsafe_l2_payload_block_id));
 
-            let handler = WhitelistRestHandler::new(WhitelistRestHandlerParams {
-                event_syncer: preconf_ingress_sync.event_syncer(),
-                rpc: preconf_ingress_sync.client().clone(),
-                chain_id: self.config.p2p_config.chain_id,
-                signer,
-                beacon_client,
-                whitelist_address: self.config.whitelist_address,
-                initial_highest_unsafe_l2_payload_block_id,
-                network_command_tx: network.command_tx.clone(),
-                local_peer_id: network.local_peer_id.to_string(),
-            });
+                let handler = WhitelistRestHandler::new(WhitelistRestHandlerParams {
+                    event_syncer: preconf_ingress_sync.event_syncer(),
+                    rpc: preconf_ingress_sync.client().clone(),
+                    chain_id: self.config.p2p_config.chain_id,
+                    signer,
+                    beacon_client,
+                    whitelist_address: self.config.whitelist_address,
+                    highest_unsafe_l2_payload_block_id: shared_highest.clone(),
+                    network_command_tx: network.command_tx.clone(),
+                    local_peer_id: network.local_peer_id.to_string(),
+                });
 
-            let server_config = WhitelistRestWsServerConfig {
-                listen_addr,
-                jwt_secret: self.config.rpc_jwt_secret.clone(),
-                cors_origins: self.config.rpc_cors_origins.clone(),
-                ..Default::default()
+                let server_config = WhitelistRestWsServerConfig {
+                    listen_addr,
+                    jwt_secret: self.config.rpc_jwt_secret.clone(),
+                    cors_origins: self.config.rpc_cors_origins.clone(),
+                    ..Default::default()
+                };
+                let server = WhitelistRestWsServer::start(server_config, Arc::new(handler)).await?;
+                info!(
+                    addr = %server.local_addr(),
+                    http_url = %server.http_url(),
+                    ws_url = %server.ws_url(),
+                    "whitelist preconfirmation REST server started"
+                );
+                (Some(server), Some(shared_highest))
+            } else {
+                (None, None)
             };
-            let server = WhitelistRestWsServer::start(server_config, Arc::new(handler)).await?;
-            info!(
-                addr = %server.local_addr(),
-                http_url = %server.http_url(),
-                ws_url = %server.ws_url(),
-                "whitelist preconfirmation REST server started"
-            );
-            Some(server)
-        } else {
-            None
-        };
 
         let mut importer = WhitelistPreconfirmationImporter::new(
             preconf_ingress_sync.event_syncer(),
             preconf_ingress_sync.client().clone(),
             self.config.whitelist_address,
             self.config.p2p_config.chain_id,
             network.command_tx.clone(),
+            shared_highest_unsafe,
         );
         let mut epoch_tick = time::interval(Duration::from_secs(L1_EPOCH_DURATION_SECS));