fix(taiko-client-rs): use trusted resume head for event sync (checkpoint head or head_l1_origin) (#21331)

Author: davidtaikocha

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

@@ -79,3 +79,17 @@ pub enum DriverError {
     #[error(transparent)]
     Other(#[from] AnyhowError),
 }
+
+/// Map a [`DriverError`] into a target error type while preserving sync errors.
+///
+/// This ensures `DriverError::Sync` is converted through `From<SyncError>` instead of being
+/// wrapped as a generic driver error.
+pub fn map_driver_error<T>(err: DriverError) -> T
+where
+    T: From<SyncError> + From<DriverError>,
+{
+    match err {
+        DriverError::Sync(sync_err) => sync_err.into(),
+        other => other.into(),
+    }
+}

packages/taiko-client-rs/crates/driver/src/lib.rs

@@ -15,7 +15,7 @@ pub mod sync;
 
 pub use config::DriverConfig;
 pub use driver::Driver;
-pub use error::DriverError;
+pub use error::{DriverError, map_driver_error};
 pub use production::PreconfPayload;
 pub use sync::{CanonicalTipState, SyncPipeline, SyncStage, event::EventSyncer};
 

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

@@ -1,6 +1,6 @@
 //! Beacon sync logic.
 
-use std::{borrow::Cow, marker::PhantomData, time::Duration};
+use std::{borrow::Cow, marker::PhantomData, sync::Arc, time::Duration};
 
 use alethia_reth_primitives::engine::types::TaikoExecutionDataSidecar;
 use alloy::providers::Provider;
@@ -21,7 +21,7 @@ use rpc::{
 use tokio::time::{MissedTickBehavior, interval};
 use tracing::{debug, info, instrument, warn};
 
-use super::{SyncError, SyncStage};
+use super::{SyncError, SyncStage, checkpoint_resume_head::CheckpointResumeHead};
 use crate::{config::DriverConfig, error::DriverError, metrics::DriverMetrics};
 
 /// Default polling interval used when no retry interval is configured.
@@ -32,9 +32,15 @@ pub struct BeaconSyncer<P>
 where
     P: Provider + Clone,
 {
+    /// Interval between beacon sync retries.
     retry_interval: Duration,
+    /// RPC client used for local node engine and chain calls.
     rpc: Client<P>,
+    /// Optional checkpoint provider used for remote catch-up blocks.
     checkpoint: Option<RootProvider>,
+    /// Shared checkpoint head used to resume event sync after beacon sync.
+    checkpoint_resume_head: Arc<CheckpointResumeHead>,
+    /// Marker that ties this type to the generic provider parameter.
     _marker: PhantomData<P>,
 }
 
@@ -44,11 +50,21 @@ where
 {
     /// Construct a new beacon syncer from the provided configuration and RPC client.
     #[instrument(skip(config, rpc))]
-    pub fn new(config: &DriverConfig, rpc: Client<P>) -> Self {
+    pub fn new(
+        config: &DriverConfig,
+        rpc: Client<P>,
+        checkpoint_resume_head: Arc<CheckpointResumeHead>,
+    ) -> Self {
         let checkpoint =
             config.l2_checkpoint_url.as_ref().map(|url| connect_http_with_timeout(url.clone()));
 
-        Self { retry_interval: config.retry_interval, rpc, checkpoint, _marker: PhantomData }
+        Self {
+            retry_interval: config.retry_interval,
+            rpc,
+            checkpoint,
+            checkpoint_resume_head,
+            _marker: PhantomData,
+        }
     }
 
     /// Query the checkpoint node for its head L1 origin block number.
@@ -139,6 +155,10 @@ where
     /// missing blocks to the local execution engine.
     #[instrument(skip(self), name = "beacon_syncer_run")]
     async fn run(&self) -> Result<(), SyncError> {
+        // Always clear stale state from previous attempts so event sync cannot accidentally
+        // consume an old checkpoint head after a failed or skipped beacon sync run.
+        self.checkpoint_resume_head.clear();
+
         if self.checkpoint.is_none() {
             info!("no checkpoint endpoint configured; skipping beacon sync stage");
             return Ok(());
@@ -219,6 +239,10 @@ where
                 })?;
                 counter!(DriverMetrics::BEACON_SYNC_REMOTE_SUBMISSIONS_TOTAL).increment(1);
             } else {
+                // Persist the checkpoint head we have confirmed local execution is synced to.
+                // Event sync uses this exact value as its authoritative resume source when
+                // checkpoint mode is enabled.
+                self.checkpoint_resume_head.set(checkpoint_head);
                 info!(checkpoint_head, local_head, "local engine at or ahead of checkpoint; done");
                 break Ok(());
             }

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

@@ -0,0 +1,68 @@
+//! Shared state for carrying beacon-sync checkpoint progress into event sync.
+
+use std::sync::atomic::{AtomicU64, Ordering};
+
+/// Sentinel value used to represent an unset checkpoint head.
+const CHECKPOINT_HEAD_UNSET: u64 = u64::MAX;
+
+/// Stores the L2 block number that beacon sync has confirmed as its catch-up head.
+///
+/// `u64::MAX` is reserved as "not set", allowing a single-atomic representation while preserving
+/// `Option<u64>` semantics.
+#[derive(Debug)]
+pub struct CheckpointResumeHead {
+    /// Stored checkpoint head block number.
+    value: AtomicU64,
+}
+
+impl Default for CheckpointResumeHead {
+    /// Initializes with no checkpoint head set.
+    fn default() -> Self {
+        Self { value: AtomicU64::new(CHECKPOINT_HEAD_UNSET) }
+    }
+}
+
+impl CheckpointResumeHead {
+    /// Clears the stored checkpoint head.
+    pub fn clear(&self) {
+        self.value.store(CHECKPOINT_HEAD_UNSET, Ordering::Release);
+    }
+
+    /// Stores a checkpoint head block number.
+    pub fn set(&self, block_number: u64) {
+        debug_assert_ne!(
+            block_number, CHECKPOINT_HEAD_UNSET,
+            "u64::MAX is reserved as an unset checkpoint-head sentinel"
+        );
+        self.value.store(block_number, Ordering::Release);
+    }
+
+    /// Returns the stored checkpoint head when present.
+    pub fn get(&self) -> Option<u64> {
+        let value = self.value.load(Ordering::Acquire);
+        if value == CHECKPOINT_HEAD_UNSET { None } else { Some(value) }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::mem::size_of;
+
+    #[test]
+    fn checkpoint_resume_head_uses_single_atomic_storage() {
+        assert_eq!(size_of::<CheckpointResumeHead>(), size_of::<AtomicU64>());
+    }
+
+    #[test]
+    fn checkpoint_resume_head_round_trip() {
+        let head = CheckpointResumeHead::default();
+        assert_eq!(head.get(), None);
+
+        head.set(42);
+        assert_eq!(head.get(), Some(42));
+
+        head.clear();
+        assert_eq!(head.get(), None);
+    }
+}

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

@@ -38,10 +38,22 @@ pub enum SyncError {
     #[error("execution engine returned no latest block")]
     MissingLatestExecutionBlock,
 
+    /// Event sync: checkpoint mode enabled, but beacon sync did not publish a resume head.
+    #[error("checkpoint mode enabled but no checkpoint resume head is available")]
+    MissingCheckpointResumeHead,
+
+    /// Event sync: no-checkpoint mode requires local head L1 origin to choose a safe resume head.
+    #[error("head_l1_origin is missing; cannot derive event resume head without checkpoint")]
+    MissingHeadL1OriginResume,
+
     /// Event sync: execution engine missing a specific block.
     #[error("execution engine returned no block {number}")]
     MissingExecutionBlock { 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 { proposal_id: u64 },