fix(driver): use trusted resume head for event sync (checkpoint head or head_l1_origin)

Author: davidtaikocha

.github/workflows/taiko-client-rs--docker.yml

@@ -4,7 +4,7 @@ permissions:
 
 on:
   push:
-    branches: [main]
+    branches: [main, event-resume]
     tags:
       - "taiko-alethia-client-rs-v*"
     paths:

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

@@ -203,7 +203,7 @@ impl PreconfirmationDriverRunner {
                 node_handle.abort();
                 match result {
                     Ok(Ok(())) => Err(RunnerError::EventSyncerExited),
-                    Ok(Err(err)) => Err(RunnerError::Sync(err)),
+                    Ok(Err(err)) => Err(map_driver_task_error(err)),
                     Err(err) => Err(RunnerError::EventSyncerFailed(err.to_string())),
                 }
             }
@@ -218,3 +218,11 @@ impl PreconfirmationDriverRunner {
         run_result
     }
 }
+
+/// Map a driver task error to a runner error, preserving sync errors but wrapping other driver errors.
+fn map_driver_task_error(err: driver::DriverError) -> RunnerError {
+    match err {
+        driver::DriverError::Sync(sync_err) => RunnerError::Sync(sync_err),
+        other => RunnerError::Driver(other),
+    }
+}

packages/taiko-client-rs/crates/preconfirmation-driver/src/runner/preconf_ingress_sync.rs

@@ -5,10 +5,7 @@ use std::{future::Future, sync::Arc};
 use alloy_provider::{
     Provider, RootProvider, fillers::FillProvider, utils::JoinedRecommendedFillers,
 };
-use driver::{
-    DriverConfig, SyncStage,
-    sync::{SyncError, event::EventSyncer},
-};
+use driver::{DriverConfig, SyncPipeline, sync::event::EventSyncer};
 use rpc::client::Client;
 use tokio::task::JoinHandle;
 
@@ -21,16 +18,16 @@ where
 {
     client: Client<P>,
     event_syncer: Arc<EventSyncer<P>>,
-    handle: JoinHandle<std::result::Result<(), SyncError>>,
+    handle: JoinHandle<std::result::Result<(), driver::DriverError>>,
 }
 
 impl PreconfIngressSync<FillProvider<JoinedRecommendedFillers, RootProvider>> {
-    /// Start the preconfirmation ingress syncer and its background task.
+    /// Start the preconfirmation ingress sync pipeline and its background task.
     pub(crate) async fn start(config: &DriverConfig) -> Result<Self, RunnerError> {
         let client = Client::new(config.client.clone()).await?;
-        let event_syncer = Arc::new(EventSyncer::new(config, client.clone()).await?);
-        let event_syncer_task = event_syncer.clone();
-        let handle = tokio::spawn(async move { event_syncer_task.run().await });
+        let pipeline = SyncPipeline::new(config.clone(), client.clone()).await?;
+        let event_syncer = pipeline.event_syncer();
+        let handle = tokio::spawn(async move { pipeline.run().await });
 
         Ok(Self { client, event_syncer, handle })
     }
@@ -50,8 +47,10 @@ where
         self.event_syncer.clone()
     }
 
-    /// Get a mutable handle to the ingress syncer task.
-    pub(crate) fn handle_mut(&mut self) -> &mut JoinHandle<std::result::Result<(), SyncError>> {
+    /// Get a mutable handle to the ingress sync pipeline task.
+    pub(crate) fn handle_mut(
+        &mut self,
+    ) -> &mut JoinHandle<std::result::Result<(), driver::DriverError>> {
         &mut self.handle
     }
 
@@ -65,10 +64,18 @@ where
     }
 }
 
+/// Wait for the preconfirmation ingress sync to signal readiness or fail.
+fn map_driver_error(err: driver::DriverError) -> RunnerError {
+    match err {
+        driver::DriverError::Sync(sync_err) => RunnerError::Sync(sync_err),
+        other => RunnerError::Driver(other),
+    }
+}
+
 /// Wait for preconfirmation ingress to be ready or the ingress syncer to exit.
 pub(crate) async fn wait_for_preconf_ingress_ready<F>(
     ready: F,
-    event_syncer_handle: &mut JoinHandle<std::result::Result<(), SyncError>>,
+    event_syncer_handle: &mut JoinHandle<std::result::Result<(), driver::DriverError>>,
 ) -> Result<(), RunnerError>
 where
     F: Future<Output = Option<()>> + Send,
@@ -78,7 +85,7 @@ where
         result = event_syncer_handle => {
             match result {
                 Ok(Ok(())) => Err(RunnerError::EventSyncerExited),
-                Ok(Err(err)) => Err(RunnerError::Sync(err)),
+                Ok(Err(err)) => Err(map_driver_error(err)),
                 Err(err) => Err(RunnerError::EventSyncerFailed(err.to_string())),
             }
         }
@@ -87,15 +94,38 @@ where
 
 #[cfg(test)]
 mod tests {
+    use driver::{DriverError, sync::SyncError};
+
     #[tokio::test]
     async fn wait_for_preconf_ingress_ready_errors_when_ready_none() {
-        use driver::sync::SyncError;
-
         let ready = async { None };
-        let mut handle = tokio::spawn(async { Ok::<(), SyncError>(()) });
+        let mut handle = tokio::spawn(async { Ok::<(), DriverError>(()) });
 
         let err = super::wait_for_preconf_ingress_ready(ready, &mut handle).await.unwrap_err();
 
         assert!(err.to_string().contains("preconfirmation ingress"));
     }
+
+    #[tokio::test]
+    async fn wait_for_preconf_ingress_ready_maps_sync_driver_error() {
+        let ready = std::future::pending::<Option<()>>();
+        let mut handle = tokio::spawn(async {
+            Err::<(), DriverError>(DriverError::Sync(SyncError::MissingCheckpointResumeHead))
+        });
+
+        let err = super::wait_for_preconf_ingress_ready(ready, &mut handle).await.unwrap_err();
+
+        assert!(matches!(err, super::RunnerError::Sync(SyncError::MissingCheckpointResumeHead)));
+    }
+
+    #[tokio::test]
+    async fn wait_for_preconf_ingress_ready_maps_non_sync_driver_error() {
+        let ready = std::future::pending::<Option<()>>();
+        let mut handle =
+            tokio::spawn(async { Err::<(), DriverError>(DriverError::PreconfirmationDisabled) });
+
+        let err = super::wait_for_preconf_ingress_ready(ready, &mut handle).await.unwrap_err();
+
+        assert!(matches!(err, super::RunnerError::Driver(DriverError::PreconfirmationDisabled)));
+    }
 }

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

@@ -5,10 +5,7 @@ use std::{future::Future, sync::Arc};
 use alloy_provider::{
     Provider, RootProvider, fillers::FillProvider, utils::JoinedRecommendedFillers,
 };
-use driver::{
-    DriverConfig,
-    sync::{SyncError, event::EventSyncer},
-};
+use driver::{DriverConfig, SyncPipeline, sync::event::EventSyncer};
 use rpc::client::Client;
 use tokio::task::JoinHandle;
 
@@ -23,17 +20,17 @@ where
     client: Client<P>,
     /// Shared event syncer that exposes ingress readiness and submit hooks.
     event_syncer: Arc<EventSyncer<P>>,
-    /// Background task running the event sync loop.
-    handle: JoinHandle<std::result::Result<(), SyncError>>,
+    /// Background task running the sync pipeline loop.
+    handle: JoinHandle<std::result::Result<(), driver::DriverError>>,
 }
 
 impl PreconfIngressSync<FillProvider<JoinedRecommendedFillers, RootProvider>> {
-    /// Start the event syncer and its background task.
+    /// Start the sync pipeline and expose its event syncer and background task.
     pub(crate) async fn start(config: &DriverConfig) -> Result<Self> {
         let client = Client::new(config.client.clone()).await?;
-        let event_syncer = Arc::new(EventSyncer::new(config, client.clone()).await?);
-        let event_syncer_task = event_syncer.clone();
-        let handle = tokio::spawn(async move { driver::SyncStage::run(&*event_syncer_task).await });
+        let pipeline = SyncPipeline::new(config.clone(), client.clone()).await?;
+        let event_syncer = pipeline.event_syncer();
+        let handle = tokio::spawn(async move { pipeline.run().await });
 
         Ok(Self { client, event_syncer, handle })
     }
@@ -54,7 +51,9 @@ where
     }
 
     /// Access the background event syncer task handle.
-    pub(crate) fn handle_mut(&mut self) -> &mut JoinHandle<std::result::Result<(), SyncError>> {
+    pub(crate) fn handle_mut(
+        &mut self,
+    ) -> &mut JoinHandle<std::result::Result<(), driver::DriverError>> {
         &mut self.handle
     }
 
@@ -68,10 +67,18 @@ where
     }
 }
 
+/// Map a driver error to a whitelist preconfirmation driver error, preserving sync errors but wrapping others.
+fn map_driver_error(err: driver::DriverError) -> WhitelistPreconfirmationDriverError {
+    match err {
+        driver::DriverError::Sync(sync_err) => WhitelistPreconfirmationDriverError::Sync(sync_err),
+        other => WhitelistPreconfirmationDriverError::Driver(other),
+    }
+}
+
 /// Wait for ingress readiness, or return if the event syncer exits first.
 pub(crate) async fn wait_for_preconf_ingress_ready<F>(
     ready: F,
-    event_syncer_handle: &mut JoinHandle<std::result::Result<(), SyncError>>,
+    event_syncer_handle: &mut JoinHandle<std::result::Result<(), driver::DriverError>>,
 ) -> Result<()>
 where
     F: Future<Output = Option<()>> + Send,
@@ -84,9 +91,57 @@ where
         result = event_syncer_handle => {
             match result {
                 Ok(Ok(())) => Err(WhitelistPreconfirmationDriverError::EventSyncerExited),
-                Ok(Err(err)) => Err(WhitelistPreconfirmationDriverError::Sync(err)),
+                Ok(Err(err)) => Err(map_driver_error(err)),
                 Err(err) => Err(WhitelistPreconfirmationDriverError::EventSyncerFailed(err.to_string())),
             }
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use driver::{DriverError, sync::SyncError};
+
+    #[tokio::test]
+    async fn wait_for_preconf_ingress_ready_errors_when_ready_none() {
+        let ready = async { None };
+        let mut handle = tokio::spawn(async { Ok::<(), DriverError>(()) });
+
+        let err = super::wait_for_preconf_ingress_ready(ready, &mut handle).await.unwrap_err();
+        assert!(matches!(
+            err,
+            super::WhitelistPreconfirmationDriverError::PreconfIngressNotEnabled
+        ));
+    }
+
+    #[tokio::test]
+    async fn wait_for_preconf_ingress_ready_maps_sync_driver_error() {
+        let ready = std::future::pending::<Option<()>>();
+        let mut handle = tokio::spawn(async {
+            Err::<(), DriverError>(DriverError::Sync(SyncError::MissingCheckpointResumeHead))
+        });
+
+        let err = super::wait_for_preconf_ingress_ready(ready, &mut handle).await.unwrap_err();
+        assert!(matches!(
+            err,
+            super::WhitelistPreconfirmationDriverError::Sync(
+                SyncError::MissingCheckpointResumeHead
+            )
+        ));
+    }
+
+    #[tokio::test]
+    async fn wait_for_preconf_ingress_ready_maps_non_sync_driver_error() {
+        let ready = std::future::pending::<Option<()>>();
+        let mut handle =
+            tokio::spawn(async { Err::<(), DriverError>(DriverError::PreconfirmationDisabled) });
+
+        let err = super::wait_for_preconf_ingress_ready(ready, &mut handle).await.unwrap_err();
+        assert!(matches!(
+            err,
+            super::WhitelistPreconfirmationDriverError::Driver(
+                DriverError::PreconfirmationDisabled
+            )
+        ));
+    }
+}

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

@@ -235,7 +235,7 @@ impl WhitelistPreconfirmationDriverRunner {
                                 "reason" => "event_syncer_error",
                             )
                             .increment(1);
-                            Err(WhitelistPreconfirmationDriverError::Sync(err))
+                            Err(map_driver_task_error(err))
                         }
                         Err(err) => {
                             metrics::counter!(
@@ -278,3 +278,11 @@ impl WhitelistPreconfirmationDriverRunner {
         }
     }
 }
+
+/// Runs event sync plus whitelist preconfirmation message ingestion, with optional REST/WS server for external access.
+fn map_driver_task_error(err: driver::DriverError) -> WhitelistPreconfirmationDriverError {
+    match err {
+        driver::DriverError::Sync(sync_err) => WhitelistPreconfirmationDriverError::Sync(sync_err),
+        other => WhitelistPreconfirmationDriverError::Driver(other),
+    }
+}