L1 liveness probe (#479)

* introduce L1 liveness probe

* add L1 liveness mod

Author: frisitano

crates/node/src/args.rs

@@ -392,6 +392,8 @@ impl ScrollRollupNodeConfig {
                             l1_block_startup_info,
                             node_config,
                             self.l1_provider_args.logs_query_block_range,
+                            self.l1_provider_args.liveness_threshold,
+                            self.l1_provider_args.liveness_check_interval,
                         )
                         .await,
                     ),
@@ -668,7 +670,7 @@ impl RollupNodeNetworkArgs {
 }
 
 /// The arguments for the L1 provider.
-#[derive(Debug, Default, Clone, clap::Args)]
+#[derive(Debug, Clone, clap::Args)]
 pub struct L1ProviderArgs {
     /// The URL for the L1 RPC.
     #[arg(long = "l1.url", id = "l1_url", value_name = "L1_URL")]
@@ -688,6 +690,28 @@ pub struct L1ProviderArgs {
     /// The maximum number of items to be stored in the cache layer.
     #[arg(long = "l1.cache-max-items", id = "l1_cache_max_items", value_name = "L1_CACHE_MAX_ITEMS", default_value_t = constants::L1_PROVIDER_CACHE_MAX_ITEMS)]
     pub cache_max_items: u32,
+    /// The L1 liveness threshold in seconds. If no new L1 block is received within this duration,
+    /// an error is logged.
+    #[arg(long = "l1.liveness-threshold", id = "l1_liveness_threshold", value_name = "L1_LIVENESS_THRESHOLD", default_value_t = constants::L1_LIVENESS_THRESHOLD)]
+    pub liveness_threshold: u64,
+    /// The interval in seconds at which to check L1 liveness.
+    #[arg(long = "l1.liveness-check-interval", id = "l1_liveness_check_interval", value_name = "L1_LIVENESS_CHECK_INTERVAL", default_value_t = constants::L1_LIVENESS_CHECK_INTERVAL)]
+    pub liveness_check_interval: u64,
+}
+
+impl Default for L1ProviderArgs {
+    fn default() -> Self {
+        Self {
+            url: None,
+            compute_units_per_second: constants::PROVIDER_COMPUTE_UNITS_PER_SECOND,
+            max_retries: constants::L1_PROVIDER_MAX_RETRIES,
+            initial_backoff: constants::L1_PROVIDER_INITIAL_BACKOFF,
+            logs_query_block_range: constants::LOGS_QUERY_BLOCK_RANGE,
+            cache_max_items: constants::L1_PROVIDER_CACHE_MAX_ITEMS,
+            liveness_threshold: constants::L1_LIVENESS_THRESHOLD,
+            liveness_check_interval: constants::L1_LIVENESS_CHECK_INTERVAL,
+        }
+    }
 }
 
 /// The arguments for the Beacon provider.

crates/node/src/constants.rs

@@ -11,6 +11,12 @@ pub(crate) const L1_PROVIDER_INITIAL_BACKOFF: u64 = 100;
 /// The maximum number of items to store in L1 provider's cache layer.
 pub(crate) const L1_PROVIDER_CACHE_MAX_ITEMS: u32 = 100;
 
+/// The default L1 liveness threshold in seconds.
+pub(crate) const L1_LIVENESS_THRESHOLD: u64 = 60;
+
+/// The default L1 liveness check interval in seconds.
+pub(crate) const L1_LIVENESS_CHECK_INTERVAL: u64 = 12;
+
 /// The block range used to fetch L1 logs.
 pub(crate) const LOGS_QUERY_BLOCK_RANGE: u64 = 500;
 

crates/node/tests/sync.rs

@@ -55,6 +55,7 @@ async fn test_should_consolidate_to_block_15k() -> eyre::Result<()> {
             initial_backoff: 100,
             logs_query_block_range: 500,
             cache_max_items: 100,
+            ..Default::default()
         },
         engine_driver_args: EngineDriverArgs { sync_at_startup: false },
         sequencer_args: SequencerArgs {

crates/watcher/src/lib.rs

@@ -6,6 +6,9 @@ pub use error::{EthRequestError, FilterLogError, L1WatcherError};
 mod handle;
 pub use handle::{L1WatcherCommand, L1WatcherHandle};
 
+mod liveness;
+use liveness::LivenessProbe;
+
 mod metrics;
 pub use metrics::WatcherMetrics;
 
@@ -97,6 +100,8 @@ pub struct L1Watcher<EP> {
     is_synced: bool,
     /// The log query block range.
     log_query_block_range: u64,
+    /// The L1 liveness probe.
+    liveness_probe: LivenessProbe,
 }
 
 /// The L1 notification type yielded by the [`L1Watcher`].
@@ -206,6 +211,8 @@ where
         l1_block_startup_info: L1BlockStartupInfo,
         config: Arc<NodeConfig>,
         log_query_block_range: u64,
+        liveness_threshold: u64,
+        liveness_check_interval: u64,
     ) -> L1WatcherHandle {
         tracing::trace!(target: "scroll::watcher", ?l1_block_startup_info, ?config, "spawning L1 watcher");
 
@@ -271,6 +278,7 @@ where
             metrics: WatcherMetrics::default(),
             is_synced: false,
             log_query_block_range,
+            liveness_probe: LivenessProbe::new(liveness_threshold, liveness_check_interval),
         };
 
         // notify at spawn.
@@ -304,6 +312,11 @@ where
                 }
             }
 
+            // Check L1 liveness if due.
+            if self.liveness_probe.is_due() {
+                self.liveness_probe.check(self.unfinalized_blocks.last());
+            }
+
             // step the watcher.
             if let Err(L1WatcherError::SendError(_)) = self
                 .step()
@@ -923,6 +936,7 @@ mod tests {
                 metrics: WatcherMetrics::default(),
                 is_synced: false,
                 log_query_block_range: LOG_QUERY_BLOCK_RANGE,
+                liveness_probe: LivenessProbe::new(60, 12),
             },
             handle,
         )

crates/watcher/src/liveness.rs

@@ -0,0 +1,50 @@
+use super::Header;
+use std::time::Instant;
+
+/// A probe that checks L1 liveness by monitoring block timestamps.
+#[derive(Debug)]
+pub(crate) struct LivenessProbe {
+    /// The threshold in seconds after which to log an error if no new block is received.
+    threshold: u64,
+    /// The interval in seconds at which to perform the liveness check.
+    check_interval: u64,
+    /// The last time a liveness check was performed.
+    last_check: Instant,
+}
+
+impl LivenessProbe {
+    /// Creates a new liveness probe.
+    pub(crate) fn new(threshold: u64, check_interval: u64) -> Self {
+        Self { threshold, check_interval, last_check: Instant::now() }
+    }
+
+    /// Returns true if a liveness check is due based on the configured interval.
+    pub(crate) fn is_due(&self) -> bool {
+        self.last_check.elapsed().as_secs() >= self.check_interval
+    }
+
+    /// Checks L1 liveness based on the latest block header.
+    /// Logs an error if no new block has been received within the threshold.
+    pub(crate) fn check(&mut self, latest_block: Option<&Header>) {
+        self.last_check = Instant::now();
+
+        if let Some(block) = latest_block {
+            let now = std::time::SystemTime::now()
+                .duration_since(std::time::UNIX_EPOCH)
+                .expect("time went backwards")
+                .as_secs();
+
+            let elapsed = now.saturating_sub(block.timestamp);
+            if elapsed > self.threshold {
+                tracing::error!(
+                    target: "scroll::watcher",
+                    latest_block_number = block.number,
+                    latest_block_timestamp = block.timestamp,
+                    elapsed_secs = elapsed,
+                    threshold_secs = self.threshold,
+                    "L1 liveness check failed: no new L1 block received"
+                );
+            }
+        }
+    }
+}

crates/watcher/tests/indexing.rs

@@ -19,6 +19,8 @@ async fn test_should_not_index_latest_block_multiple_times() -> eyre::Result<()>
     const CHAIN_LEN: usize = 200;
     const HALF_CHAIN_LEN: usize = 100;
     const LOGS_QUERY_BLOCK_RANGE: u64 = 500;
+    const L1_LIVENESS_THRESHOLD: u64 = 60;
+    const L1_LIVENESS_CHECK_INTERVAL: u64 = 12;
 
     // Given
     let (finalized, latest, headers) = chain(CHAIN_LEN);
@@ -64,6 +66,8 @@ async fn test_should_not_index_latest_block_multiple_times() -> eyre::Result<()>
         L1BlockStartupInfo::None,
         Arc::new(config),
         LOGS_QUERY_BLOCK_RANGE,
+        L1_LIVENESS_THRESHOLD,
+        L1_LIVENESS_CHECK_INTERVAL,
     )
     .await;
     let mut prev_block_info = Default::default();

crates/watcher/tests/logs.rs

@@ -21,6 +21,8 @@ async fn test_should_not_miss_logs_on_reorg() -> eyre::Result<()> {
     const CHAIN_LEN: usize = 200;
     const HALF_CHAIN_LEN: usize = CHAIN_LEN / 2;
     const LOGS_QUERY_BLOCK_RANGE: u64 = 500;
+    const L1_LIVENESS_THRESHOLD: u64 = 60;
+    const L1_LIVENESS_CHECK_INTERVAL: u64 = 12;
 
     // Given
     let (finalized, _, headers) = chain(CHAIN_LEN);
@@ -69,6 +71,8 @@ async fn test_should_not_miss_logs_on_reorg() -> eyre::Result<()> {
         L1BlockStartupInfo::None,
         Arc::new(config),
         LOGS_QUERY_BLOCK_RANGE,
+        L1_LIVENESS_THRESHOLD,
+        L1_LIVENESS_CHECK_INTERVAL,
     )
     .await;
     let mut received_logs = Vec::new();

crates/watcher/tests/reorg.rs

@@ -11,6 +11,8 @@ use rollup_node_watcher::{
     random, test_utils::provider::MockProvider, Block, L1Notification, L1Watcher,
 };
 const LOGS_QUERY_BLOCK_RANGE: u64 = 500;
+const L1_LIVENESS_THRESHOLD: u64 = 60;
+const L1_LIVENESS_CHECK_INTERVAL: u64 = 12;
 
 // Generate a set blocks that will be fed to the l1 watcher.
 // Every fork_cycle blocks, generates a small reorg.
@@ -77,6 +79,8 @@ async fn test_should_detect_reorg() -> eyre::Result<()> {
         L1BlockStartupInfo::None,
         Arc::new(config),
         LOGS_QUERY_BLOCK_RANGE,
+        L1_LIVENESS_THRESHOLD,
+        L1_LIVENESS_CHECK_INTERVAL,
     )
     .await;
 
@@ -184,6 +188,8 @@ async fn test_should_fetch_gap_in_unfinalized_blocks() -> eyre::Result<()> {
         L1BlockStartupInfo::None,
         Arc::new(config),
         LOGS_QUERY_BLOCK_RANGE,
+        L1_LIVENESS_THRESHOLD,
+        L1_LIVENESS_CHECK_INTERVAL,
     )
     .await;