fix(taiko-client-rs): align /status and fee range checks with go (#21335)

Co-authored-by: David <david@taiko.xyz>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: 3 people

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

@@ -332,6 +332,11 @@ impl BeaconClient {
         Ok((timestamp - self.genesis_time) / self.seconds_per_slot)
     }
 
+    /// Convert a timestamp to a beacon epoch.
+    pub fn timestamp_to_epoch(&self, timestamp: u64) -> Result<u64, BlobDataError> {
+        Ok(self.timestamp_to_slot(timestamp)? / self.slots_per_epoch)
+    }
+
     /// Return the current beacon slot based on local wall-clock time.
     ///
     /// Mirrors the Go driver implementation:

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

@@ -19,22 +19,6 @@ pub enum WhitelistPreconfirmationDriverError {
     /// Whitelist preconfirmation node task failed.
     #[error("whitelist preconfirmation node task failed: {0}")]
     NodeTaskFailed(String),
-    /// Driver preconfirmation ingress is not ready.
-    #[error("driver preconfirmation ingress not ready")]
-    PreconfIngressNotReady,
-    /// Unsupported chain for whitelist preconfirmation mode.
-    #[error(
-        "whitelist preconfirmation mode currently supports only chain_id={expected}, got {actual}"
-    )]
-    UnsupportedChain {
-        /// Expected chain id.
-        expected: u64,
-        /// Actual chain id.
-        actual: u64,
-    },
-    /// Missing execution payload in envelope.
-    #[error("missing execution payload")]
-    MissingExecutionPayload,
     /// Invalid or unsupported payload format.
     #[error("invalid payload format: {0}")]
     InvalidPayload(String),

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

@@ -16,7 +16,7 @@ where
     P: alloy_provider::Provider + Clone + Send + Sync + 'static,
 {
     /// Attempt to import cached envelopes if sync is ready.
-    pub(super) async fn maybe_import_from_cache(&mut self) -> Result<()> {
+    pub(crate) async fn maybe_import_from_cache(&mut self) -> Result<()> {
         let _ = self.refresh_sync_ready().await?;
         if !self.sync_ready || self.cache.is_empty() {
             return Ok(());

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

@@ -2,6 +2,7 @@ use std::sync::Arc;
 
 use alloy_primitives::B256;
 use alloy_provider::Provider;
+use tracing::debug;
 
 use crate::{
     codec::{
@@ -61,9 +62,31 @@ where
         }
         let envelope = Arc::new(envelope);
         self.cache.insert(envelope.clone());
-        self.recent_cache.insert_recent(envelope);
+        self.recent_cache.insert_recent(envelope.clone());
         self.update_cache_gauges();
 
+        if envelope.end_of_sequencing.unwrap_or(false) {
+            match self.beacon_client.timestamp_to_epoch(envelope.execution_payload.timestamp) {
+                Ok(epoch) => {
+                    debug!(
+                        epoch,
+                        hash = %envelope.execution_payload.block_hash,
+                        "recording end-of-sequencing envelope for epoch on payload ingress"
+                    );
+                    self.cache_state
+                        .record_end_of_sequencing(epoch, envelope.execution_payload.block_hash)
+                        .await;
+                }
+                Err(err) => {
+                    tracing::warn!(
+                        timestamp = envelope.execution_payload.timestamp,
+                        error = %err,
+                        "failed to derive epoch from payload timestamp for EOS recording"
+                    );
+                }
+            }
+        }
+
         Ok(())
     }
 
@@ -120,8 +143,31 @@ where
 
         let envelope = Arc::new(envelope);
         self.cache.insert(envelope.clone());
-        self.recent_cache.insert_recent(envelope);
+        self.recent_cache.insert_recent(envelope.clone());
         self.update_cache_gauges();
+
+        if envelope.end_of_sequencing.unwrap_or(false) {
+            match self.beacon_client.timestamp_to_epoch(envelope.execution_payload.timestamp) {
+                Ok(epoch) => {
+                    debug!(
+                        epoch,
+                        hash = %envelope.execution_payload.block_hash,
+                        "recording end-of-sequencing envelope for epoch on response ingress"
+                    );
+                    self.cache_state
+                        .record_end_of_sequencing(epoch, envelope.execution_payload.block_hash)
+                        .await;
+                }
+                Err(err) => {
+                    tracing::warn!(
+                        timestamp = envelope.execution_payload.timestamp,
+                        error = %err,
+                        "failed to derive epoch from response timestamp for EOS recording"
+                    );
+                }
+            }
+        }
+
         Ok(())
     }
 

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

@@ -2,23 +2,19 @@
 
 use std::sync::Arc;
 
-use alloy_eips::BlockNumberOrTag;
 use alloy_primitives::{Address, B256};
 use alloy_provider::Provider;
-use bindings::preconf_whitelist::PreconfWhitelist::PreconfWhitelistInstance;
 use driver::sync::event::EventSyncer;
-use rpc::client::Client;
+use rpc::{beacon::BeaconClient, client::Client};
 use tokio::sync::{Mutex, mpsc};
 use tracing::{debug, info, warn};
 
 use crate::{
-    cache::{
-        EnvelopeCache, L1_EPOCH_DURATION_SECS, RecentEnvelopeCache, RequestThrottle,
-        WhitelistSequencerCache,
-    },
+    cache::{EnvelopeCache, RecentEnvelopeCache, RequestThrottle, SharedPreconfCacheState},
     error::{Result, WhitelistPreconfirmationDriverError},
     metrics::WhitelistPreconfirmationDriverMetrics,
     network::{NetworkCommand, NetworkEvent},
+    whitelist_fetcher::WhitelistSequencerFetcher,
 };
 
 /// Cache re-import flow for out-of-order envelopes once parents arrive.
@@ -45,6 +41,29 @@ pub(crate) const MAX_COMPRESSED_TX_LIST_BYTES: usize = 131_072 * 6;
 ///
 /// Align with the preconfirmation tx-list cap to avoid zlib bomb expansion on untrusted payloads.
 pub(crate) const MAX_DECOMPRESSED_TX_LIST_BYTES: usize = 8 * 1024 * 1024;
+/// Dependency bundle for constructing [`WhitelistPreconfirmationImporter`].
+pub(crate) struct WhitelistPreconfirmationImporterParams<P>
+where
+    P: Provider + Clone + Send + Sync + 'static,
+{
+    /// Event syncer used to submit validated preconfirmation payloads.
+    pub(crate) event_syncer: Arc<EventSyncer<P>>,
+    /// RPC client used for L1/L2 reads and head-origin updates.
+    pub(crate) rpc: Client<P>,
+    /// Whitelist contract address used for signer validation.
+    pub(crate) whitelist_address: Address,
+    /// Chain id used for preconfirmation signature domain separation.
+    pub(crate) chain_id: u64,
+    /// Command channel used to publish P2P requests/responses.
+    pub(crate) network_command_tx: mpsc::Sender<NetworkCommand>,
+    /// Shared cache state used by status and EOS signaling.
+    pub(crate) cache_state: SharedPreconfCacheState,
+    /// Beacon client used for EOS epoch validation.
+    pub(crate) beacon_client: Arc<BeaconClient>,
+    /// Shared highest unsafe L2 payload block ID (updated on P2P import when REST server enabled).
+    pub(crate) highest_unsafe_l2_payload_block_id: Option<Arc<Mutex<u64>>>,
+}
+
 /// Imports whitelist preconfirmation payloads into the driver after event sync catches up.
 pub(crate) struct WhitelistPreconfirmationImporter<P>
 where
@@ -54,18 +73,20 @@ where
     event_syncer: Arc<EventSyncer<P>>,
     /// RPC client used for L1/L2 reads and head-origin updates.
     rpc: Client<P>,
-    /// On-chain whitelist contract instance used to validate sequencer signers.
-    whitelist: PreconfWhitelistInstance<P>,
     /// Chain id used for preconfirmation signature domain separation.
     chain_id: u64,
+    /// Shared cache state used by status and EOS signaling.
+    cache_state: SharedPreconfCacheState,
+    /// Beacon client used for EOS epoch validation.
+    beacon_client: Arc<BeaconClient>,
     /// Out-of-order payload cache waiting for parent availability.
     cache: EnvelopeCache,
     /// Recently accepted envelopes that can be served over response topic requests.
     recent_cache: RecentEnvelopeCache,
     /// Cooldown gate for repeated missing-parent requests.
     request_throttle: RequestThrottle,
-    /// TTL cache for current/next whitelist sequencer addresses.
-    sequencer_cache: WhitelistSequencerCache,
+    /// Shared sequencer fetcher for whitelist validation and epoch cache management.
+    sequencer_fetcher: WhitelistSequencerFetcher<P>,
     /// 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).
@@ -81,26 +102,31 @@ where
     P: Provider + Clone + Send + Sync + 'static,
 {
     /// Build an importer.
-    pub(crate) fn new(
-        event_syncer: Arc<EventSyncer<P>>,
-        rpc: Client<P>,
-        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());
+    pub(crate) fn new(params: WhitelistPreconfirmationImporterParams<P>) -> Self {
+        let WhitelistPreconfirmationImporterParams {
+            event_syncer,
+            rpc,
+            whitelist_address,
+            chain_id,
+            network_command_tx,
+            cache_state,
+            beacon_client,
+            highest_unsafe_l2_payload_block_id,
+        } = params;
+        let sequencer_fetcher =
+            WhitelistSequencerFetcher::new(whitelist_address, rpc.l1_provider.clone());
         let anchor_address = *rpc.shasta.anchor.address();
 
         let importer = Self {
             event_syncer,
             rpc,
-            whitelist,
             chain_id,
+            cache_state,
+            beacon_client,
             cache: EnvelopeCache::default(),
             recent_cache: RecentEnvelopeCache::default(),
             request_throttle: RequestThrottle::default(),
-            sequencer_cache: WhitelistSequencerCache::default(),
+            sequencer_fetcher,
             network_command_tx,
             highest_unsafe_l2_payload_block_id,
             sync_ready: false,
@@ -175,22 +201,60 @@ where
                 }
             }
             NetworkEvent::EndOfSequencingRequest { from, epoch } => {
-                if let Some(envelope) = self.recent_cache.latest_end_of_sequencing() {
-                    debug!(
-                        peer = %from,
-                        epoch,
-                        hash = %envelope.execution_payload.block_hash,
-                        "serving end-of-sequencing whitelist preconfirmation response from recent cache"
-                    );
-                    self.publish_unsafe_response(envelope).await;
-                    metrics::counter!(
-                        WhitelistPreconfirmationDriverMetrics::IMPORTER_EVENTS_TOTAL,
-                        "event_type" => "end_of_sequencing_request",
-                        "result" => "served",
-                    )
-                    .increment(1);
+                if let Some(hash) = self.cache_state.end_of_sequencing_for_epoch(epoch).await {
+                    if let Some(envelope) = self.recent_cache.get_recent(&hash) {
+                        debug!(
+                            peer = %from,
+                            epoch,
+                            hash = %envelope.execution_payload.block_hash,
+                            "serving end-of-sequencing whitelist preconfirmation response from recent cache"
+                        );
+                        self.publish_unsafe_response(envelope).await;
+                        metrics::counter!(
+                            WhitelistPreconfirmationDriverMetrics::IMPORTER_EVENTS_TOTAL,
+                            "event_type" => "end_of_sequencing_request",
+                            "result" => "served",
+                        )
+                        .increment(1);
+                    } else {
+                        debug!(
+                            peer = %from,
+                            epoch,
+                            hash = %hash,
+                            "end-of-sequencing hash known for epoch but envelope not in recent cache; rebuilding from L2"
+                        );
+
+                        if let Some(mut envelope) =
+                            self.build_response_envelope_from_l2(hash).await?
+                        {
+                            envelope.end_of_sequencing = Some(true);
+                            let envelope = Arc::new(envelope);
+                            self.recent_cache.insert_recent(envelope.clone());
+                            self.update_cache_gauges();
+                            self.publish_unsafe_response(envelope).await;
+                            metrics::counter!(
+                                WhitelistPreconfirmationDriverMetrics::IMPORTER_EVENTS_TOTAL,
+                                "event_type" => "end_of_sequencing_request",
+                                "result" => "served_l2_fallback",
+                            )
+                            .increment(1);
+                        } else {
+                            debug!(
+                                peer = %from,
+                                epoch,
+                                hash = %hash,
+                                "end-of-sequencing hash known for epoch but unavailable from recent cache and L2"
+                            );
+                            metrics::counter!(
+                                WhitelistPreconfirmationDriverMetrics::IMPORTER_EVENTS_TOTAL,
+                                "event_type" => "end_of_sequencing_request",
+                                "result" => "miss",
+                            )
+                            .increment(1);
+                        }
+                    }
                 } else {
-                    debug!(peer = %from, epoch, "no recent end-of-sequencing envelope to serve");
+                    debug!(peer = %from, epoch, "no end-of-sequencing block found for epoch");
                     metrics::counter!(
                         WhitelistPreconfirmationDriverMetrics::IMPORTER_EVENTS_TOTAL,
                         "event_type" => "end_of_sequencing_request",
@@ -204,35 +268,14 @@ where
         self.maybe_import_from_cache().await
     }
 
-    /// Event-sync progress signal.
-    ///
-    /// Drop cached sequencers only after crossing an epoch boundary to avoid unnecessary
-    /// re-fetches while still preventing stale signer acceptance across epochs.
-    pub(crate) async fn on_sync_ready_signal(&mut self) -> Result<()> {
-        if !self.refresh_sync_ready().await? {
-            return Ok(());
-        }
-
-        if self.sequencer_cache.current_epoch_start_timestamp().is_some() {
-            let latest_l1_timestamp = self.latest_l1_block_timestamp().await?;
-            if self
-                .sequencer_cache
-                .should_invalidate_for_l1_timestamp(latest_l1_timestamp, L1_EPOCH_DURATION_SECS)
-            {
-                self.sequencer_cache.invalidate();
-            }
-        }
-
-        if self.cache.is_empty() {
-            return Ok(());
+    /// Invalidate the sequencer cache if the L1 head has crossed an epoch boundary.
+    pub(crate) async fn maybe_invalidate_sequencer_cache_for_epoch(&mut self) {
+        if let Err(err) = self.sequencer_fetcher.maybe_invalidate_for_epoch_advance().await {
+            warn!(
+                error = %err,
+                "failed to check epoch boundary for sequencer cache invalidation"
+            );
         }
-
-        self.import_from_cache().await.inspect_err(|_err| {
-            metrics::counter!(
-                WhitelistPreconfirmationDriverMetrics::SYNC_READY_IMPORT_FAILURES_TOTAL
-            )
-            .increment(1);
-        })
     }
 
     /// Refresh whether sync is ready.
@@ -255,26 +298,11 @@ where
         Ok(self.rpc.head_l1_origin().await?.map(|head| head.block_id.to::<u64>()))
     }
 
-    /// Read latest L1 block timestamp.
-    pub(super) async fn latest_l1_block_timestamp(&self) -> Result<u64> {
-        self.rpc
-            .l1_provider
-            .get_block_by_number(BlockNumberOrTag::Latest)
-            .await
-            .map_err(provider_err)?
-            .map(|block| block.header.timestamp)
-            .ok_or_else(|| {
-                WhitelistPreconfirmationDriverError::WhitelistLookup(
-                    "missing latest L1 block while updating sequencer cache".to_string(),
-                )
-            })
-    }
-
     /// Get the block hash by block number.
     pub(super) async fn block_hash_by_number(&self, block_number: u64) -> Result<Option<B256>> {
         self.rpc
             .l2_provider
-            .get_block_by_number(BlockNumberOrTag::Number(block_number))
+            .get_block_by_number(alloy_eips::BlockNumberOrTag::Number(block_number))
             .await
             .map(|opt| opt.map(|block| block.hash()))
             .map_err(provider_err)