refactor(taiko-client-rs): remove proposal-id/canonical-tip wiring and use confirmed head-state sync (#21332)

Author: davidtaikocha

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

@@ -17,7 +17,7 @@ pub use config::DriverConfig;
 pub use driver::Driver;
 pub use error::{DriverError, map_driver_error};
 pub use production::PreconfPayload;
-pub use sync::{CanonicalTipState, SyncPipeline, SyncStage, event::EventSyncer};
+pub use sync::{ConfirmedSyncSnapshot, SyncPipeline, SyncStage, event::EventSyncer};
 
 // Re-export signer from protocol crate for backward compatibility
 pub use protocol::signer;

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

@@ -20,6 +20,9 @@ impl DriverMetrics {
     pub const EVENT_SCANNER_BATCHES_TOTAL: &'static str = "driver_event_scanner_batches_total";
     /// Counter tracking scanner stream errors.
     pub const EVENT_SCANNER_ERRORS_TOTAL: &'static str = "driver_event_scanner_errors_total";
+    /// Counter tracking failures while probing confirmed-sync readiness.
+    pub const EVENT_CONFIRMED_SYNC_PROBE_ERRORS_TOTAL: &'static str =
+        "driver_event_confirmed_sync_probe_errors_total";
     /// Counter tracking proposal logs processed by the driver.
     pub const EVENT_PROPOSALS_TOTAL: &'static str = "driver_event_proposals_total";
     /// Counter tracking skipped proposals (e.g. zero ID, below initial ID).
@@ -122,6 +125,11 @@ impl DriverMetrics {
             Unit::Count,
             "Errors emitted by the event scanner stream"
         );
+        metrics::describe_counter!(
+            Self::EVENT_CONFIRMED_SYNC_PROBE_ERRORS_TOTAL,
+            Unit::Count,
+            "Errors emitted while probing confirmed-sync readiness in event sync"
+        );
         metrics::describe_counter!(
             Self::EVENT_PROPOSALS_TOTAL,
             Unit::Count,
@@ -248,6 +256,7 @@ impl DriverMetrics {
         metrics::counter!(Self::BEACON_SYNC_REMOTE_SUBMISSIONS_TOTAL).absolute(0);
         metrics::counter!(Self::EVENT_SCANNER_BATCHES_TOTAL).absolute(0);
         metrics::counter!(Self::EVENT_SCANNER_ERRORS_TOTAL).absolute(0);
+        metrics::counter!(Self::EVENT_CONFIRMED_SYNC_PROBE_ERRORS_TOTAL).absolute(0);
         metrics::counter!(Self::EVENT_PROPOSALS_TOTAL).absolute(0);
         metrics::counter!(Self::EVENT_PROPOSALS_SKIPPED_TOTAL).absolute(0);
         metrics::counter!(Self::EVENT_DERIVED_BLOCKS_TOTAL).absolute(0);

packages/taiko-client-rs/crates/driver/src/production/path.rs

@@ -8,16 +8,13 @@ use crate::{
     derivation::DerivationPipeline,
     error::DriverError,
     metrics::DriverMetrics,
-    sync::{
-        AtomicCanonicalTip, CanonicalTipState, engine::PayloadApplier, error::SyncError,
-        is_stale_preconf,
-    },
+    sync::{engine::PayloadApplier, error::SyncError},
 };
 use alloy::{eips::BlockNumberOrTag, primitives::B256, providers::Provider};
 use async_trait::async_trait;
 use metrics::{counter, histogram};
 use rpc::{RpcClientError, client::Client};
-use tracing::{debug, error, warn};
+use tracing::{debug, error};
 
 /// A block-production path capable of materialising the provided input into execution blocks.
 ///
@@ -67,22 +64,15 @@ where
     A: PayloadApplier + BlockHashReader,
 {
     applier: A,
-    canonical_tip_state: Arc<AtomicCanonicalTip>,
 }
 
 impl<A> PreconfirmationPath<A>
 where
     A: PayloadApplier + BlockHashReader,
 {
-    /// Construct a preconfirmation path with a canonical block tip boundary.
-    ///
-    /// Blocks at or below this tip are considered event-synced canonical history and must never be
-    /// rewritten by preconfirmation payloads.
-    pub fn new_with_canonical_tip_state(
-        applier: A,
-        canonical_tip_state: Arc<AtomicCanonicalTip>,
-    ) -> Self {
-        Self { applier, canonical_tip_state }
+    /// Construct a preconfirmation path backed by a payload applier.
+    pub fn new(applier: A) -> Self {
+        Self { applier }
     }
 }
 
@@ -107,29 +97,6 @@ where
                 let block_number = preconf.block_number();
                 let parent_number = block_number.saturating_sub(1);
 
-                match self.canonical_tip_state.load(std::sync::atomic::Ordering::Relaxed) {
-                    CanonicalTipState::Unknown => {
-                        warn!(
-                            block_number,
-                            "rejecting preconfirmation production: canonical tip unknown"
-                        );
-                        return Err(DriverError::PreconfIngressNotReady);
-                    }
-                    CanonicalTipState::Known(canonical_block_tip) => {
-                        if is_stale_preconf(block_number, canonical_block_tip) {
-                            counter!(DriverMetrics::PRECONF_STALE_DROPPED_TOTAL).increment(1);
-                            counter!(DriverMetrics::PRECONF_STALE_DROPPED_PRODUCTION_TOTAL)
-                                .increment(1);
-                            warn!(
-                                block_number,
-                                canonical_block_tip,
-                                "dropping stale preconfirmation at or below canonical event-sync tip"
-                            );
-                            return Ok(Vec::new());
-                        }
-                    }
-                }
-
                 // Measure parent hash lookup duration.
                 let lookup_start = Instant::now();
                 let parent_hash_result = self.applier.block_hash_by_number(parent_number).await;
@@ -421,9 +388,7 @@ mod tests {
     fn preconfirmation_path_delegates_to_applier() {
         let parent_hash = B256::from([1u8; 32]);
         let applier = MockApplier::new(0, parent_hash);
-        let canonical_tip = Arc::new(AtomicCanonicalTip::new(CanonicalTipState::Known(0)));
-        let path =
-            PreconfirmationPath::new_with_canonical_tip_state(applier.clone(), canonical_tip);
+        let path = PreconfirmationPath::new(applier.clone());
         let payload = Arc::new(PreconfPayload::new(sample_payload(1)));
 
         let rt = Runtime::new().unwrap();
@@ -436,30 +401,10 @@ mod tests {
     }
 
     #[test]
-    fn preconfirmation_path_drops_block_at_or_below_canonical_tip() {
-        let parent_hash = B256::from([1u8; 32]);
-        let applier = MockApplier::new(0, parent_hash);
-        let canonical_tip = Arc::new(AtomicCanonicalTip::new(CanonicalTipState::Known(1)));
-        let path =
-            PreconfirmationPath::new_with_canonical_tip_state(applier.clone(), canonical_tip);
-        let payload = Arc::new(PreconfPayload::new(sample_payload(1)));
-
-        let rt = Runtime::new().unwrap();
-        let outcomes = rt
-            .block_on(path.produce(ProductionInput::Preconfirmation(payload)))
-            .expect("preconfirmation at canonical boundary should be dropped");
-
-        assert_eq!(applier.calls(), 0);
-        assert!(outcomes.is_empty());
-    }
-
-    #[test]
-    fn preconfirmation_path_allows_reorg_above_canonical_tip() {
+    fn preconfirmation_path_produces_above_parent() {
         let parent_hash = B256::from([1u8; 32]);
         let applier = MockApplier::new(1, parent_hash);
-        let canonical_tip = Arc::new(AtomicCanonicalTip::new(CanonicalTipState::Known(1)));
-        let path =
-            PreconfirmationPath::new_with_canonical_tip_state(applier.clone(), canonical_tip);
+        let path = PreconfirmationPath::new(applier.clone());
         let payload = Arc::new(PreconfPayload::new(sample_payload(2)));
 
         let rt = Runtime::new().unwrap();
@@ -470,22 +415,4 @@ mod tests {
         assert_eq!(applier.calls(), 1);
         assert_eq!(outcomes.len(), 1);
     }
-
-    #[test]
-    fn preconfirmation_path_rejects_when_canonical_tip_unknown() {
-        let parent_hash = B256::from([1u8; 32]);
-        let applier = MockApplier::new(0, parent_hash);
-        let canonical_tip = Arc::new(AtomicCanonicalTip::new(CanonicalTipState::Unknown));
-        let path =
-            PreconfirmationPath::new_with_canonical_tip_state(applier.clone(), canonical_tip);
-        let payload = Arc::new(PreconfPayload::new(sample_payload(1)));
-
-        let rt = Runtime::new().unwrap();
-        let err = rt
-            .block_on(path.produce(ProductionInput::Preconfirmation(payload)))
-            .expect_err("unknown canonical tip should reject preconfirmation production");
-
-        assert!(matches!(err, DriverError::PreconfIngressNotReady));
-        assert_eq!(applier.calls(), 0);
-    }
 }