apollo_l1_events: bound catch-up commit-block backlog with a cap and metric

The catchupper's `commit_block_backlog` was an unbounded `Vec` populated by every
commit-block arriving above the provider's height during startup catch-up, and
drained only once L2 sync reached the target. A persistently slow or stalled sync
while the batcher keeps committing could grow it without limit (security finding
L-16).

Add a configurable `max_commit_block_backlog_len` (default 1,000,000) and a
`l1_message_provider_commit_block_backlog_len` gauge. On overflow,
`add_commit_block_to_backlog` returns a new `CatchUpBacklogOverflow` error rather
than dropping entries: the backlog must stay a gapless, strictly-sequential run,
so drop-oldest would corrupt the drain-time invariant and silently skip an
L1-handler commit. The error surfaces to the batcher (which already handles
commit_block errors) and to logs/alerts. The gauge is updated on each push and
reset to 0 when the backlog drains at catch-up completion.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: asaf-sw

Cargo.lock

@@ -42,10 +42,13 @@ alloy.workspace = true
 apollo_base_layer_tests.workspace = true
 apollo_infra_utils = { workspace = true, features = ["testing"] }
 apollo_l1_events_types = { workspace = true, features = ["testing"] }
+apollo_metrics = { workspace = true, features = ["testing"] }
 apollo_state_sync_types = { workspace = true, features = ["testing"] }
 apollo_time = { workspace = true, features = ["testing"] }
 assert_matches.workspace = true
 itertools.workspace = true
+metrics.workspace = true
+metrics-exporter-prometheus.workspace = true
 mockall.workspace = true
 papyrus_base_layer = { workspace = true, features = ["testing"] }
 pretty_assertions.workspace = true

crates/apollo_l1_events/Cargo.toml

@@ -2,13 +2,16 @@ use std::sync::atomic::{AtomicU8, Ordering};
 use std::sync::Arc;
 use std::time::Duration;
 
-use apollo_l1_events_types::SharedL1EventsProviderClient;
+use apollo_l1_events_types::errors::L1EventsProviderError;
+use apollo_l1_events_types::{L1EventsProviderResult, SharedL1EventsProviderClient};
 use apollo_state_sync_types::communication::SharedStateSyncClient;
 use indexmap::IndexSet;
 use starknet_api::block::BlockNumber;
 use starknet_api::transaction::TransactionHash;
 use tracing::{debug, warn};
 
+use crate::metrics::L1_MESSAGE_PROVIDER_COMMIT_BLOCK_BACKLOG_LEN;
+
 // When the Provider gets a commit_block that is too high, it starts catching up.
 // The commit is rejected by the provider, so it must use sync to catch up to the height of the
 // commit, including that height. The sync task continues until reaching the target height,
@@ -28,6 +31,9 @@ pub struct Catchupper {
     // Keep track of sync task for health checks and logging status.
     pub sync_task_handle: SyncTaskHandle,
     pub n_sync_health_check_failures: Arc<AtomicU8>,
+    /// Defensive cap on `commit_block_backlog` length; see L-16. Exceeding it is a hard error, not
+    /// a drop, to preserve the gapless-sequential invariant of the backlog.
+    pub max_commit_block_backlog_len: usize,
 }
 
 impl Catchupper {
@@ -39,6 +45,7 @@ impl Catchupper {
         l1_events_provider_client: SharedL1EventsProviderClient,
         sync_client: SharedStateSyncClient,
         sync_retry_interval: Duration,
+        max_commit_block_backlog_len: usize,
     ) -> Self {
         Self {
             sync_retry_interval,
@@ -47,6 +54,7 @@ impl Catchupper {
             sync_client,
             sync_task_handle: SyncTaskHandle::NotStartedYet,
             n_sync_health_check_failures: Default::default(),
+            max_commit_block_backlog_len,
             // This is overriden when starting the sync task (e.g., when provider starts
             // catching up).
             target_height: BlockNumber(0),
@@ -66,17 +74,35 @@ impl Catchupper {
         &mut self,
         committed_txs: IndexSet<TransactionHash>,
         height: BlockNumber,
-    ) {
+    ) -> L1EventsProviderResult<()> {
         assert!(
             self.commit_block_backlog
                 .last()
                 .is_none_or(|commit_block| commit_block.height.unchecked_next() == height),
             "Heights should be sequential."
         );
 
+        // Tripwire against unbounded growth on a stalled/lagging L2 sync (L-16). We must NOT drop
+        // entries: the backlog is a gapless, strictly-sequential run and a hole would corrupt the
+        // drain-time sequentiality assert and silently skip an L1-handler commit. Fail loudly
+        // instead, so the caller and alerts learn catch-up is pathologically behind.
+        if self.commit_block_backlog.len() >= self.max_commit_block_backlog_len {
+            warn!(
+                "Catch-up commit-block backlog reached its cap of {} entries at height {height}; \
+                 rejecting commit-block. L2 sync is likely stalled or lagging.",
+                self.max_commit_block_backlog_len
+            );
+            return Err(L1EventsProviderError::CatchUpBacklogOverflow {
+                height,
+                max: self.max_commit_block_backlog_len,
+            });
+        }
+
         debug!("Adding future commit-block to backlog at height: {height}");
         self.commit_block_backlog
             .push(CommitBlockBacklog { height, committed_txs: committed_txs.clone() });
+        L1_MESSAGE_PROVIDER_COMMIT_BLOCK_BACKLOG_LEN.set_lossy(self.commit_block_backlog.len());
+        Ok(())
     }
 
     /// Spawns async task that produces and sends commit block messages to the provider, according

crates/apollo_l1_events/src/catchupper.rs

@@ -22,7 +22,7 @@ use starknet_api::transaction::TransactionHash;
 use tracing::{debug, error, info, instrument, trace, warn};
 
 use crate::catchupper::Catchupper;
-use crate::metrics::register_provider_metrics;
+use crate::metrics::{register_provider_metrics, L1_MESSAGE_PROVIDER_COMMIT_BLOCK_BACKLOG_LEN};
 use crate::transaction_manager::TransactionManager;
 use crate::L1EventsProviderConfig;
 
@@ -71,6 +71,7 @@ impl L1EventsProvider {
             l1_events_provider_client,
             state_sync_client,
             config.startup_sync_sleep_retry_interval_seconds,
+            config.max_commit_block_backlog_len,
         );
         Self {
             config,
@@ -87,6 +88,7 @@ impl L1EventsProvider {
             self.catchupper.l1_events_provider_client.clone(),
             self.catchupper.sync_client.clone(),
             self.config.startup_sync_sleep_retry_interval_seconds,
+            self.config.max_commit_block_backlog_len,
         );
     }
     // Functions Called by the scraper.
@@ -422,7 +424,9 @@ impl L1EventsProvider {
             Equal => self.apply_commit_block(committed_txs, Default::default()),
             // We're still syncing, backlog it, it'll get applied later.
             Greater => {
-                self.catchupper.add_commit_block_to_backlog(committed_txs, new_height);
+                // Propagate a backlog-overflow error to the caller (the batcher) rather than
+                // growing memory unbounded; see L-16.
+                self.catchupper.add_commit_block_to_backlog(committed_txs, new_height)?;
                 // No need to check the backlog or catchup completion, since those are only
                 // applicable if we just increased the provider's height, like in the `Equal` case.
                 return Ok(());
@@ -438,6 +442,8 @@ impl L1EventsProvider {
                 self.current_height
             );
             let backlog = std::mem::take(&mut self.catchupper.commit_block_backlog);
+            // The backlog is fully consumed below; reset its observability gauge to 0.
+            L1_MESSAGE_PROVIDER_COMMIT_BLOCK_BACKLOG_LEN.set_lossy(0_usize);
             assert!(
                 backlog.is_empty()
                     || self.current_height == backlog.first().unwrap().height

crates/apollo_l1_events/src/l1_events_provider.rs

@@ -24,6 +24,7 @@ use apollo_time::time::Clock;
 use assert_matches::assert_matches;
 use indexmap::IndexSet;
 use itertools::Itertools;
+use metrics_exporter_prometheus::PrometheusBuilder;
 use pretty_assertions::assert_eq;
 use rstest::rstest;
 use starknet_api::block::{BlockNumber, BlockTimestamp};
@@ -33,6 +34,7 @@ use starknet_api::tx_hash;
 
 use crate::catchupper::{Catchupper, CommitBlockBacklog, SyncTaskHandle};
 use crate::l1_events_provider::L1EventsProvider;
+use crate::metrics::{register_provider_metrics, L1_MESSAGE_PROVIDER_COMMIT_BLOCK_BACKLOG_LEN};
 use crate::test_utils::{
     l1_handler,
     make_catchupper,
@@ -415,6 +417,118 @@ async fn commit_block_backlog() {
     expected_l1_events_provider.assert_eq(&l1_events_provider);
 }
 
+/// Drives `commit_block` with strictly-increasing (`Greater`) heights while catching up, so each
+/// one is appended to the backlog. Returns the provider so callers can inspect/continue.
+fn provider_catching_up_with_backlog_cap(max_commit_block_backlog_len: usize) -> L1EventsProvider {
+    const STARTUP_HEIGHT: BlockNumber = BlockNumber(0);
+    let catchupper = make_catchupper!(backlog: [], max: max_commit_block_backlog_len);
+    L1EventsProviderContentBuilder::new()
+        .with_catchupper(catchupper)
+        .with_height(STARTUP_HEIGHT)
+        .with_state(ProviderState::CatchingUp)
+        .build_into_l1_provider()
+}
+
+#[test]
+fn backlog_below_cap_accepts_all_commits() {
+    const CAP: usize = 3;
+    let mut l1_events_provider = provider_catching_up_with_backlog_cap(CAP);
+
+    // Two commits, both above current height (0) -> backlogged, staying below the cap of 3.
+    for height in [BlockNumber(1), BlockNumber(2)] {
+        commit_block_no_rejected(&mut l1_events_provider, &[], height);
+    }
+
+    assert_eq!(l1_events_provider.catchupper.commit_block_backlog.len(), 2);
+}
+
+#[test]
+fn backlog_fills_exactly_to_cap_without_error() {
+    const CAP: usize = 3;
+    let mut l1_events_provider = provider_catching_up_with_backlog_cap(CAP);
+
+    // Exactly CAP commits fit (heights 1..=3, all above current height 0).
+    for height in [BlockNumber(1), BlockNumber(2), BlockNumber(3)] {
+        commit_block_no_rejected(&mut l1_events_provider, &[], height);
+    }
+
+    let backlog = &l1_events_provider.catchupper.commit_block_backlog;
+    assert_eq!(backlog.len(), CAP);
+    // The backlog must remain a gapless, strictly-sequential run.
+    assert_eq!(
+        backlog.iter().map(|commit_block| commit_block.height).collect::<Vec<_>>(),
+        vec![BlockNumber(1), BlockNumber(2), BlockNumber(3)]
+    );
+}
+
+#[test]
+fn backlog_over_cap_returns_overflow_error_without_dropping() {
+    const CAP: usize = 3;
+    let mut l1_events_provider = provider_catching_up_with_backlog_cap(CAP);
+
+    // Fill the backlog exactly to the cap.
+    for height in [BlockNumber(1), BlockNumber(2), BlockNumber(3)] {
+        commit_block_no_rejected(&mut l1_events_provider, &[], height);
+    }
+
+    // The next commit overflows the cap and must be rejected with a descriptive error.
+    let overflow_height = BlockNumber(4);
+    let result = l1_events_provider.commit_block([].into(), [].into(), overflow_height);
+    assert_matches!(
+        result,
+        Err(L1EventsProviderError::CatchUpBacklogOverflow { height, max })
+            if height == overflow_height && max == CAP
+    );
+
+    // Nothing was silently dropped: the backlog is still exactly CAP entries and gapless.
+    let backlog = &l1_events_provider.catchupper.commit_block_backlog;
+    assert_eq!(backlog.len(), CAP);
+    assert_eq!(
+        backlog.iter().map(|commit_block| commit_block.height).collect::<Vec<_>>(),
+        vec![BlockNumber(1), BlockNumber(2), BlockNumber(3)]
+    );
+}
+
+#[tokio::test]
+async fn backlog_len_metric_tracks_push_and_drain() {
+    let recorder = PrometheusBuilder::new().build_recorder();
+    let _recorder_guard = metrics::set_default_local_recorder(&recorder);
+
+    // Start catching up with current height 0 and a target one above it, so a single `Equal`
+    // commit (height 1) completes catch-up and drains the backlog.
+    const TARGET_HEIGHT: BlockNumber = BlockNumber(0);
+    let mut catchupper = make_catchupper!(backlog: []);
+    catchupper.target_height = TARGET_HEIGHT;
+    let mut l1_events_provider = L1EventsProviderContentBuilder::new()
+        .with_catchupper(catchupper)
+        .with_height(BlockNumber(0))
+        .with_state(ProviderState::CatchingUp)
+        .build_into_l1_provider();
+    // Registers the gauge so it is rendered even before its first `set`.
+    register_provider_metrics();
+
+    // Two `Greater` commits (heights 1, 2) are backlogged; the gauge tracks the backlog length.
+    // They start sequentially one above the current height (0), as the drain-time invariant
+    // requires.
+    commit_block_no_rejected(&mut l1_events_provider, &[], BlockNumber(1));
+    commit_block_no_rejected(&mut l1_events_provider, &[], BlockNumber(2));
+    let metrics = recorder.handle().render();
+    assert_eq!(
+        L1_MESSAGE_PROVIDER_COMMIT_BLOCK_BACKLOG_LEN.parse_numeric_metric::<usize>(&metrics),
+        Some(2)
+    );
+
+    // The `Equal` commit at height 0 completes catch-up (current height becomes 1 > target 0),
+    // drains the backlog [1, 2], and resets the gauge to 0.
+    commit_block_no_rejected(&mut l1_events_provider, &[], BlockNumber(0));
+    assert_eq!(l1_events_provider.state, ProviderState::Pending);
+    let metrics = recorder.handle().render();
+    assert_eq!(
+        L1_MESSAGE_PROVIDER_COMMIT_BLOCK_BACKLOG_LEN.parse_numeric_metric::<usize>(&metrics),
+        Some(0)
+    );
+}
+
 #[test]
 fn commit_block_before_add_tx_stores_tx_in_committed() {
     // Setup

crates/apollo_l1_events/src/l1_events_provider_tests.rs

@@ -19,6 +19,7 @@ define_metrics!(
         MetricCounter { L1_MESSAGE_SCRAPER_REORG_DETECTED, "l1_message_scraper_reorg_detected", "Number of times the L1 message scraper detected a reorganization in the base layer", init=0},
         MetricGauge { L1_MESSAGE_SCRAPER_LAST_SUCCESS_TIMESTAMP_SECONDS, "l1_message_scraper_last_success_timestamp_seconds", "Unix timestamp (seconds) of the last successful L1 message scrape" },
         MetricGauge { L1_MESSAGE_PROVIDER_NUM_PENDING_TXS, "l1_message_provider_num_pending_txs", "The number of pending L1 handler transactions in the transaction manager" },
+        MetricGauge { L1_MESSAGE_PROVIDER_COMMIT_BLOCK_BACKLOG_LEN, "l1_message_provider_commit_block_backlog_len", "The number of commit-blocks buffered in the catch-up backlog while the provider syncs to the target height; abnormal sustained growth indicates a stalled or lagging L2 sync" },
     },
 );
 
@@ -33,4 +34,5 @@ pub(crate) fn register_scraper_metrics() {
 
 pub(crate) fn register_provider_metrics() {
     L1_MESSAGE_PROVIDER_NUM_PENDING_TXS.register();
+    L1_MESSAGE_PROVIDER_COMMIT_BLOCK_BACKLOG_LEN.register();
 }

crates/apollo_l1_events/src/metrics.rs

@@ -36,7 +36,10 @@ use crate::transaction_record::{TransactionPayload, TransactionRecord};
 use crate::L1EventsProviderConfig;
 
 macro_rules! make_catchupper {
-    (backlog: [$($height:literal => [$($tx:literal),* $(,)*]),* $(,)*]) => {{
+    (backlog: [$($height:literal => [$($tx:literal),* $(,)*]),* $(,)*]) => {
+        make_catchupper!(backlog: [$($height => [$($tx),*]),*], max: usize::MAX)
+    };
+    (backlog: [$($height:literal => [$($tx:literal),* $(,)*]),* $(,)*], max: $max:expr) => {{
         Catchupper {
             commit_block_backlog: vec![
                 $(CommitBlockBacklog {
@@ -49,7 +52,8 @@ macro_rules! make_catchupper {
             sync_client: Arc::new(MockStateSyncClient::default()),
             sync_task_handle: SyncTaskHandle::default(),
             n_sync_health_check_failures: Default::default(),
-            sync_retry_interval: Duration::from_millis(10)
+            sync_retry_interval: Duration::from_millis(10),
+            max_commit_block_backlog_len: $max
         }
     }};
 }

crates/apollo_l1_events/src/test_utils.rs

@@ -21,6 +21,12 @@ pub struct L1EventsProviderConfig {
     pub l1_handler_proposal_cooldown_seconds: Duration,
     /// When true, the L1 provider operates in dummy mode.
     pub dummy_mode: bool,
+    /// Maximum number of commit-blocks that may be buffered in the catch-up backlog while the
+    /// provider syncs to the target height. This is a defensive tripwire against unbounded memory
+    /// growth when L2 sync stalls or lags persistently during startup catch-up (see security
+    /// finding L-16). Hitting this bound is treated as a hard error rather than silently dropping
+    /// entries, because the backlog must remain a gapless, strictly-sequential run of heights.
+    pub max_commit_block_backlog_len: usize,
 }
 
 impl Default for L1EventsProviderConfig {
@@ -31,6 +37,10 @@ impl Default for L1EventsProviderConfig {
             l1_handler_consumption_timelock_seconds: Duration::from_secs(5 * 60),
             l1_handler_proposal_cooldown_seconds: Duration::from_secs(70),
             dummy_mode: false,
+            // ~1M empty-block entries is only ~tens of MB (per-entry overhead is dominated by the
+            // few 32-byte L1-handler tx hashes per block), which comfortably covers any legitimate
+            // startup sync gap while still bounding worst-case memory. See L-16.
+            max_commit_block_backlog_len: 1_000_000,
         }
     }
 }
@@ -72,6 +82,14 @@ impl SerializeConfig for L1EventsProviderConfig {
                  trivial truthy responses without connecting to actual L1.",
                 ParamPrivacyInput::Public,
             ),
+            ser_param(
+                "max_commit_block_backlog_len",
+                &self.max_commit_block_backlog_len,
+                "Maximum number of commit-blocks buffered in the catch-up backlog during startup \
+                 sync before commit_block fails; guards against unbounded memory growth on a \
+                 stalled or lagging L2 sync.",
+                ParamPrivacyInput::Public,
+            ),
         ])
     }
 }

crates/apollo_l1_events_config/src/config.rs

@@ -20,6 +20,14 @@ pub enum L1EventsProviderError {
     UnexpectedProviderState { expected: ProviderState, found: ProviderState },
     #[error("Cannot transition from {from} to {to}")]
     UnexpectedProviderStateTransition { from: String, to: String },
+    // The catch-up backlog is a gapless, strictly-sequential run of commit-blocks, so entries can
+    // never be dropped to make room. Hitting the cap means L2 sync is pathologically stalled or
+    // lagging; surface this to the caller (and logs/alerts) rather than growing memory unbounded.
+    #[error(
+        "Catch-up commit-block backlog overflow at height {height}: reached cap of {max} entries. \
+         L2 sync is likely stalled or lagging."
+    )]
+    CatchUpBacklogOverflow { height: BlockNumber, max: usize },
 }
 
 impl L1EventsProviderError {

crates/apollo_l1_events_types/src/errors.rs

@@ -3269,6 +3269,11 @@
     "privacy": "Public",
     "value": 70
   },
+  "l1_events_provider_config.max_commit_block_backlog_len": {
+    "description": "Maximum number of commit-blocks buffered in the catch-up backlog during startup sync before commit_block fails; guards against unbounded memory growth on a stalled or lagging L2 sync.",
+    "privacy": "Public",
+    "value": 1000000
+  },
   "l1_events_provider_config.startup_sync_sleep_retry_interval_seconds": {
     "description": "Interval in seconds between each retry of syncing with L2 during startup.",
     "privacy": "Public",