Bound local DB sync RPC load when cycles keep failing

A cycle that fails before persisting keeps its old watermark, and the
schedulers retried every sync-interval-ms, each retry re-reading every
block from that watermark to the head. Nothing bounded it: st0x staging
walked Base in 50-block eth_getLogs slices at ~70 calls/s for weeks.

- Native and wasm schedulers double the wait after each consecutive
  failed cycle, up to 5 minutes; a good cycle resets it.
- SyncConfig.max_blocks_per_cycle caps one cycle's window. Client runners
  set it to 100 x batch-size; the CLI keeps no cap.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Author: kaisbaccour

crates/cli/src/commands/local_db/cli.rs

@@ -210,6 +210,7 @@ mod tests {
             fetch,
             finality: FinalityConfig { depth: 12 },
             window_overrides: WindowOverrides::default(),
+            max_blocks_per_cycle: None,
         };
         let inputs = SyncInputs {
             raindex_id: raindex_id.clone(),

crates/cli/src/commands/local_db/pipeline/runner/environment.rs

@@ -82,6 +82,7 @@ mod tests {
                     fetch,
                     finality: FinalityConfig { depth: 12 },
                     window_overrides: WindowOverrides::default(),
+                    max_blocks_per_cycle: None,
                 },
                 dump_str: None,
                 defer_analyze: false,

crates/cli/src/commands/local_db/pipeline/runner/export.rs

@@ -133,6 +133,7 @@ mod tests {
             fetch,
             finality: FinalityConfig { depth: 12 },
             window_overrides: WindowOverrides::default(),
+            max_blocks_per_cycle: None,
         };
 
         let raindex_id = RaindexIdentifier {
@@ -221,6 +222,7 @@ mod tests {
             fetch,
             finality: FinalityConfig { depth: 1 },
             window_overrides: WindowOverrides::default(),
+            max_blocks_per_cycle: None,
         };
 
         let raindex_id = RaindexIdentifier {

crates/cli/src/commands/local_db/pipeline/runner/manifest.rs

@@ -188,6 +188,7 @@ mod tests {
                     fetch,
                     finality: FinalityConfig { depth: 12 },
                     window_overrides: WindowOverrides::default(),
+                    max_blocks_per_cycle: None,
                 },
                 dump_str: None,
                 defer_analyze: false,

crates/common/src/local_db/pipeline/adapters/window.rs

@@ -73,7 +73,15 @@ impl WindowPipeline for DefaultWindowPipeline {
 
         // 5) Apply end override then clamp to safe head
         let requested_target = cfg.window_overrides.end_block.unwrap_or(safe_head);
-        let target = requested_target.min(safe_head);
+        let mut target = requested_target.min(safe_head);
+
+        // 6) Cap the span of one cycle; the next cycle resumes from the
+        //    watermark this one persists.
+        if let Some(max_blocks) = cfg.max_blocks_per_cycle.filter(|max| *max > 0) {
+            if target >= start {
+                target = target.min(start.saturating_add(max_blocks - 1));
+            }
+        }
 
         Ok((start, target))
     }
@@ -164,6 +172,7 @@ mod tests {
                 start_block: start,
                 end_block: end,
             },
+            max_blocks_per_cycle: None,
         }
     }
 
@@ -244,6 +253,71 @@ mod tests {
         assert_eq!(target, 980);
     }
 
+    #[tokio::test]
+    async fn max_blocks_per_cycle_caps_the_target() {
+        let db = MockDb {
+            last_synced: 499,
+            fail: false,
+        };
+        let pipe = DefaultWindowPipeline::new();
+        let mut config = cfg(100, 0, None, None);
+        config.max_blocks_per_cycle = Some(50);
+        // start = 500, safe head = 1000, cap = 50 blocks => target 549
+        let (start, target) = pipe
+            .compute(&db, &sample_raindex_id(), &config, 1000)
+            .await
+            .unwrap();
+        assert_eq!((start, target), (500, 549));
+    }
+
+    #[tokio::test]
+    async fn max_blocks_per_cycle_does_not_extend_a_short_window() {
+        let db = MockDb {
+            last_synced: 990,
+            fail: false,
+        };
+        let pipe = DefaultWindowPipeline::new();
+        let mut config = cfg(100, 0, None, None);
+        config.max_blocks_per_cycle = Some(50);
+        let (start, target) = pipe
+            .compute(&db, &sample_raindex_id(), &config, 1000)
+            .await
+            .unwrap();
+        assert_eq!((start, target), (991, 1000));
+    }
+
+    #[tokio::test]
+    async fn max_blocks_per_cycle_keeps_an_empty_window_empty() {
+        let db = MockDb {
+            last_synced: 1000,
+            fail: false,
+        };
+        let pipe = DefaultWindowPipeline::new();
+        let mut config = cfg(100, 0, None, None);
+        config.max_blocks_per_cycle = Some(50);
+        let (start, target) = pipe
+            .compute(&db, &sample_raindex_id(), &config, 1000)
+            .await
+            .unwrap();
+        assert!(start > target);
+    }
+
+    #[tokio::test]
+    async fn zero_max_blocks_per_cycle_means_no_cap() {
+        let db = MockDb {
+            last_synced: 0,
+            fail: false,
+        };
+        let pipe = DefaultWindowPipeline::new();
+        let mut config = cfg(100, 0, None, None);
+        config.max_blocks_per_cycle = Some(0);
+        let (start, target) = pipe
+            .compute(&db, &sample_raindex_id(), &config, 1000)
+            .await
+            .unwrap();
+        assert_eq!((start, target), (100, 1000));
+    }
+
     #[tokio::test]
     async fn end_override_is_clamped_to_safe_head() {
         let db = MockDb {

crates/common/src/local_db/pipeline/engine.rs

@@ -515,6 +515,7 @@ mod tests {
                 fetch: FetchConfig::default(),
                 finality: FinalityConfig { depth: 0 },
                 window_overrides: WindowOverrides::default(),
+                max_blocks_per_cycle: None,
             },
             dump_str: None,
             defer_analyze: false,

crates/common/src/local_db/pipeline/mod.rs

@@ -60,6 +60,11 @@ pub struct SyncConfig {
     pub finality: FinalityConfig,
     /// Optional manual window overrides (typically CLI only).
     pub window_overrides: WindowOverrides,
+    /// Most blocks a single cycle may cover (inclusive span); `None` means no
+    /// cap. Long-running clients set it so a cycle that keeps failing before
+    /// it persists cannot re-read an ever-growing range every time; a
+    /// successful cycle moves the watermark and the next one continues.
+    pub max_blocks_per_cycle: Option<u64>,
 }
 
 /// Coarse execution summary for a single sync cycle.

crates/common/src/local_db/pipeline/runner/environment.rs

@@ -192,6 +192,7 @@ mod tests {
                     fetch,
                     finality: FinalityConfig { depth: 0 },
                     window_overrides: WindowOverrides::default(),
+                    max_blocks_per_cycle: None,
                 },
                 dump_str: None,
                 defer_analyze: false,

crates/common/src/local_db/pipeline/runner/utils.rs

@@ -98,6 +98,7 @@ pub fn build_runner_targets(
                 fetch,
                 finality,
                 window_overrides: WindowOverrides::default(),
+                max_blocks_per_cycle: None,
             },
             dump_str: None,
             defer_analyze: false,

crates/common/src/raindex_client/local_db/pipeline/runner/config.rs

@@ -60,11 +60,26 @@ impl NetworkRunnerConfig {
         })
     }
 
+    /// Builds this network's targets with a per-cycle block cap of
+    /// [`MAX_CHUNKS_PER_CLIENT_CYCLE`] fetch chunks, so a client whose cycles
+    /// keep failing re-reads a bounded range instead of everything since its
+    /// last persisted block.
     pub fn build_targets(&self) -> Result<Vec<RunnerTarget>, LocalDbError> {
-        build_runner_targets(&self.settings.raindexes, &self.settings.syncs)
+        let mut targets = build_runner_targets(&self.settings.raindexes, &self.settings.syncs)?;
+        for target in &mut targets {
+            let chunk_size = target.inputs.cfg.fetch.chunk_size();
+            target.inputs.cfg.max_blocks_per_cycle =
+                Some(chunk_size.saturating_mul(MAX_CHUNKS_PER_CLIENT_CYCLE));
+        }
+        Ok(targets)
     }
 }
 
+/// Most fetch chunks (`batch-size` blocks each) one client sync cycle covers.
+/// A bootstrapped client is normally a few chunks behind, so this only bites
+/// when a client falls far behind or keeps failing.
+pub const MAX_CHUNKS_PER_CLIENT_CYCLE: u64 = 100;
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -178,6 +193,21 @@ raindexes:
         assert!(targets.iter().all(|t| t.network_key == "network-a"));
     }
 
+    #[test]
+    fn build_targets_caps_blocks_per_cycle_from_batch_size() {
+        let global = parse_runner_settings(&sample_settings_yaml()).expect("valid yaml");
+        let config =
+            NetworkRunnerConfig::from_global_settings(&global, "network-a").expect("config ok");
+
+        for target in config.build_targets().expect("targets ok") {
+            let chunk_size = target.inputs.cfg.fetch.chunk_size();
+            assert_eq!(
+                target.inputs.cfg.max_blocks_per_cycle,
+                Some(chunk_size * MAX_CHUNKS_PER_CLIENT_CYCLE)
+            );
+        }
+    }
+
     #[test]
     fn build_targets_single_raindex_network() {
         let global = parse_runner_settings(&sample_settings_yaml()).expect("valid yaml");