feat(runtime): add an opt-in per-subscription max_lookback for bounded resume

Author: mfw78

crates/nexum-runtime/src/manifest/types.rs

@@ -71,12 +71,18 @@ pub enum Subscription {
         event_signature: Option<String>,
         /// Resume across engine restarts. When `true` the host persists a
         /// durable per-subscription cursor and re-opens the log poller
-        /// from just after the last dispatched block (clamped), instead of
-        /// at the current head. Delivery is then at-least-once, so the
-        /// module must tolerate redelivery (the chassis idempotency
-        /// journal already dedups it).
+        /// from just after the last dispatched block, instead of at the
+        /// current head. Delivery is then at-least-once, so the module must
+        /// tolerate redelivery (the chassis idempotency journal already
+        /// dedups it).
         #[serde(default)]
         resume: bool,
+        /// Optional cap on how far back a `resume` subscription will
+        /// backfill, in blocks. `None` (the default) backfills the entire
+        /// gap with no loss; set it only for a consumer that explicitly
+        /// tolerates dropping the oldest missed blocks.
+        #[serde(default)]
+        max_lookback: Option<u64>,
     },
     /// Cron-scheduled tick. 0.2 parses but does not dispatch; the
     /// supervisor emits a warning so the operator knows the

crates/nexum-runtime/src/runtime/event_loop.rs

@@ -130,17 +130,15 @@ where
     for sub in subs {
         let (tx, rx) = mpsc::channel::<TaggedChainLog>(RECONNECT_CHANNEL_BUF);
         let pool = pool.clone();
-        // The cursor key is constant per subscription and cloned onto every
-        // log; `Arc` keeps that clone cheap.
-        let cursor_key: Option<Arc<str>> = sub.cursor_key.map(Arc::from);
+        let resume = ChainLogResume {
+            // The cursor key is constant per subscription and cloned onto every
+            // log; `Arc` keeps that clone cheap.
+            cursor_key: sub.cursor_key.map(Arc::from),
+            initial_cursor: sub.initial_cursor,
+            max_lookback: sub.max_lookback,
+        };
         tasks.push(executor.spawn(Box::pin(reconnecting_chain_log_task(
-            pool,
-            sub.module,
-            sub.chain,
-            sub.filter,
-            cursor_key,
-            sub.initial_cursor,
-            tx,
+            pool, sub.module, sub.chain, sub.filter, resume, tx,
         ))));
         let tagged: TaggedChainLogStream = Box::pin(receiver_stream(rx));
         streams.push(tagged);
@@ -245,6 +243,18 @@ where
     }
 }
 
+/// Per-subscription resume and backfill knobs for a chain-log task.
+struct ChainLogResume {
+    /// Durable cursor key, `Some` for a `resume` subscription; the block
+    /// under it seeds `initial_cursor`.
+    cursor_key: Option<Arc<str>>,
+    /// Persisted resume block read at boot; the first open starts here.
+    initial_cursor: Option<u64>,
+    /// Opt-in cap (in blocks) on how far back the poller backfills; `None`
+    /// backfills the whole gap.
+    max_lookback: Option<u64>,
+}
+
 /// Poller-backed loop for a single (module, chain) chain-log
 /// subscription. Instead of `eth_subscribe(logs)` - which silently
 /// drops events emitted during a WebSocket reconnect - it drives
@@ -254,19 +264,25 @@ where
 /// transport's own retries), or a reorg deeper than the poller's
 /// retained history, ends the poller stream; this loop then re-opens
 /// from the block after the last one it delivered and backfills the
-/// entire missed range (nothing skipped) with exponential backoff.
+/// entire missed range (nothing skipped) with exponential backoff -
+/// unless the subscription set `max_lookback`, which bounds how far back
+/// the backfill reaches.
 async fn reconnecting_chain_log_task<C>(
     pool: C,
     module: String,
     chain: Chain,
     filter: alloy_rpc_types_eth::Filter,
-    cursor_key: Option<Arc<str>>,
-    initial_cursor: Option<u64>,
+    resume: ChainLogResume,
     tx: mpsc::Sender<TaggedChainLog>,
 ) -> TaskExit
 where
     C: ChainProvider + Send + Sync + 'static,
 {
+    let ChainLogResume {
+        cursor_key,
+        initial_cursor,
+        max_lookback,
+    } = resume;
     let chain_id = chain.id();
     let mut attempt: u32 = 0;
     let mut last_event: Option<Instant> = None;
@@ -304,11 +320,28 @@ where
         // - otherwise a re-open resumes just after the last delivered block
         //   (within-process gap-free), or at head on the very first open.
         // Either way the whole gap is backfilled with no lower floor, so
-        // nothing is ever skipped.
-        let start_block = match boot_resume.take() {
+        // nothing is skipped unless the subscription set `max_lookback`.
+        let mut start_block = match boot_resume.take() {
             Some(resume) => resume.min(head),
             None => poller_resume_block(last_seen_block, head),
         };
+        // Opt-in bound: `max_lookback` caps how far back a resume
+        // subscription backfills. The default (`None`) backfills fully; a
+        // set cap clamps the start up to `head - cap` and surfaces the
+        // dropped oldest blocks.
+        if let Some(cap) = max_lookback {
+            let floor = head.saturating_sub(cap);
+            if start_block < floor {
+                warn!(
+                    module = %module,
+                    chain_id,
+                    skipped_from = start_block,
+                    skipped_to = floor,
+                    "chain-log gap exceeds max_lookback - skipping the oldest missed blocks",
+                );
+                start_block = floor;
+            }
+        }
         // A large gap is backfilled in full (never skipped); surface it so a long
         // catch-up is visible rather than looking like a stall.
         if head.saturating_sub(start_block) >= LARGE_GAP_LOG_THRESHOLD {

crates/nexum-runtime/src/supervisor.rs

@@ -549,6 +549,7 @@ impl<T: RuntimeTypes> Supervisor<T> {
                     address,
                     event_signature,
                     resume,
+                    max_lookback,
                 } = sub
                 {
                     match build_alloy_filter(address.as_deref(), event_signature.as_deref()) {
@@ -574,6 +575,7 @@ impl<T: RuntimeTypes> Supervisor<T> {
                                 filter,
                                 cursor_key,
                                 initial_cursor,
+                                max_lookback: *max_lookback,
                             });
                         }
                         Err(err) => warn!(
@@ -1111,6 +1113,10 @@ pub struct ChainLogSub {
     /// The persisted resume block, read at boot for a `resume`
     /// subscription; `None` on first run or when `resume` is off.
     pub initial_cursor: Option<u64>,
+    /// Opt-in cap on how far back the poller backfills, in blocks. `None`
+    /// backfills the whole gap; `Some(cap)` bounds the start to
+    /// `head - cap`, dropping the oldest missed blocks.
+    pub max_lookback: Option<u64>,
 }
 
 /// Durable resume-cursor key for a chain-log subscription. Derived from