feat(materialize): advance a backlogged window by a bounded prefix

fluree-db-api/src/graph_source/r2rml.rs

@@ -244,6 +244,38 @@ impl ScanChoice {
     }
 }
 
+/// How many source snapshots one unpinned materialize pass may advance through.
+///
+/// `FLUREE_MATERIALIZE_MAX_SNAPSHOTS_PER_PASS`, default 64; `0` disables the cap
+/// and restores the previous "always read to head" behaviour.
+///
+/// **Why a cap exists at all.** The materialize watermark only advances when a
+/// whole pass succeeds. A pass that cannot finish its window writes no
+/// watermark, so the next pass re-reads a window one poll wider, and so on. Once
+/// that window outgrows the source's snapshot retention the stored watermark can
+/// no longer be resolved, and every subsequent poll falls back to a full table
+/// read — which is far more expensive, so it is even less likely to finish. The
+/// failure is self-reinforcing and has no exit: the entry condition for the
+/// incremental path is exactly what was lost.
+///
+/// Observed on a 17-table deployment: one affected table became 13 of 17 in
+/// about four hours, with every table then reading its whole self on every poll.
+///
+/// A cap bounds the work per pass, so the watermark advances every time and
+/// therefore stays inside retention. 64 is deliberately generous — a source
+/// committing 37-72 snapshots an hour stays under it at any sane poll interval,
+/// so a healthy job never notices the cap, while a backlogged one drains in
+/// bounded steps instead of never.
+fn materialize_max_snapshots_per_pass() -> usize {
+    static CACHED: std::sync::OnceLock<usize> = std::sync::OnceLock::new();
+    *CACHED.get_or_init(|| {
+        std::env::var("FLUREE_MATERIALIZE_MAX_SNAPSHOTS_PER_PASS")
+            .ok()
+            .and_then(|v| v.trim().parse::<usize>().ok())
+            .unwrap_or(64)
+    })
+}
+
 /// Field ids to project for `projection` against `schema`: every non-nested
 /// field when `projection` is empty, else the named columns that exist in the
 /// schema (unknown names are skipped — the consumer treats them as absent).
@@ -1266,7 +1298,39 @@ impl<'a> FlureeR2rmlProvider<'a> {
                         })?,
                 )
             }
-            None => metadata.current_snapshot(),
+            // Unpinned: read to the source's head, but advance through a long
+            // backlog in bounded steps rather than in one unbounded pass.
+            //
+            // A materialization whose watermark cannot advance re-reads a window
+            // that grows without bound, and once that window outgrows the source's
+            // snapshot retention the watermark can never be resolved again — every
+            // later poll degrades to a full table read, with no path back. Capping
+            // the pass keeps the watermark moving, which keeps it inside retention.
+            //
+            // Capping only ever moves `to` EARLIER, so it cannot skip a snapshot:
+            // the next poll resumes from the watermark this pass wrote. On any
+            // error (expired ancestor, non-ancestor, rollback) fall through to the
+            // head unchanged — the existing full-read fallback owns that case and
+            // must keep owning it.
+            None => {
+                let cap = materialize_max_snapshots_per_pass();
+                let chosen = metadata.capped_scan_end(from_snapshot_id, cap);
+                // Log only when the cap actually bit, so a healthy job stays quiet.
+                if let (Some(chosen), Some(head)) = (chosen, metadata.current_snapshot()) {
+                    if chosen.snapshot_id != head.snapshot_id {
+                        info!(
+                            graph_source_id = %graph_source_id,
+                            table = %table_name,
+                            from_snapshot_id = ?from_snapshot_id,
+                            head_snapshot_id = head.snapshot_id,
+                            bounded_to_snapshot_id = chosen.snapshot_id,
+                            max_snapshots_per_pass = cap,
+                            "materialize: backlog exceeds the per-pass cap, advancing by a prefix"
+                        );
+                    }
+                }
+                chosen
+            }
         };
         let Some(to_snapshot) = to_snapshot else {
             // Table has no snapshots: nothing to materialize.

fluree-db-iceberg/src/metadata/table.rs

@@ -170,6 +170,75 @@ impl TableMetadata {
         }
     }
 
+    /// The end of a window `(from_id, to_id]` capped to at most `max_snapshots`
+    /// snapshots, so a consumer can advance through a long backlog in bounded
+    /// steps instead of one unbounded pass.
+    ///
+    /// Returns `to_id` unchanged when the window already fits, when `from_id` is
+    /// `None` (an initial full read has no prefix to take), or when
+    /// `max_snapshots` is `0` (disabled).
+    ///
+    /// **Why a consumer wants this.** A materialization whose watermark cannot
+    /// advance re-reads a window that grows without bound, and once that window
+    /// outgrows the source's snapshot retention the watermark can never be
+    /// resolved again — every later poll degrades to a full table read, forever.
+    /// Advancing by a bounded prefix keeps the watermark moving, which keeps it
+    /// inside retention, which is what makes the incremental path recoverable
+    /// rather than a one-way door.
+    ///
+    /// Errors propagate from [`Self::snapshot_window`] (unknown/expired/
+    /// non-ancestor), where the caller must already fall back to a full re-read.
+    pub fn window_end_capped(
+        &self,
+        from_id: Option<i64>,
+        to_id: i64,
+        max_snapshots: usize,
+    ) -> crate::error::Result<i64> {
+        if max_snapshots == 0 || from_id.is_none() {
+            return Ok(to_id);
+        }
+        let window = self.snapshot_window(from_id, to_id)?;
+        if window.len() <= max_snapshots {
+            return Ok(to_id);
+        }
+        // `snapshot_window` is NEWEST-first, and we want the OLDEST
+        // `max_snapshots` of them — the prefix adjacent to `from_id`. Counting
+        // that many back from the old end lands on the last snapshot of the
+        // prefix, which becomes this pass's `to`.
+        Ok(window[window.len() - max_snapshots].snapshot_id)
+    }
+
+    /// Where an unpinned incremental consumer should end this pass: the head,
+    /// or an earlier snapshot when the backlog from `from_id` exceeds
+    /// `max_snapshots`.
+    ///
+    /// This is the whole decision in one place, so it is testable without a
+    /// storage backend — the caller does nothing but use the answer. Every way
+    /// of declining to cap returns the head unchanged:
+    ///
+    /// - no snapshots at all (`None`; there is nothing to read);
+    /// - `max_snapshots == 0`, the disable switch;
+    /// - `from_id` is `None` — an initial full read has no prefix to take, and a
+    ///   partial "full" read would be worse than an unbounded one;
+    /// - the backlog already fits;
+    /// - the window cannot be walked (expired ancestor, non-ancestor, rollback).
+    ///   That is the existing full-read fallback's case and it must keep it.
+    ///
+    /// Capping only ever returns an EARLIER snapshot than the head, never a
+    /// later one, so a consumer that records where it stopped cannot skip data.
+    pub fn capped_scan_end(
+        &self,
+        from_id: Option<i64>,
+        max_snapshots: usize,
+    ) -> Option<&super::Snapshot> {
+        let head = self.current_snapshot()?;
+        self.window_end_capped(from_id, head.snapshot_id, max_snapshots)
+            .ok()
+            .filter(|id| *id != head.snapshot_id)
+            .and_then(|id| self.snapshot(id))
+            .or(Some(head))
+    }
+
     /// Whether every snapshot in `(from_id, to_id]` was created by an `append`
     /// operation. Only then does an added-files incremental scan capture all
     /// changes (no `overwrite`/`delete`/`replace` => no updates or deletions to
@@ -551,6 +620,112 @@ mod tests {
         assert!(m.snapshot_window(Some(1), 42).is_err());
     }
 
+    /// A backlog longer than the cap is advanced in bounded steps, taking the
+    /// OLDEST snapshots first. Taking the newest instead would skip everything
+    /// between `from` and the chosen end — silent data loss, and the reason the
+    /// direction is asserted rather than assumed.
+    #[test]
+    fn window_end_capped_takes_the_oldest_prefix() {
+        // 1 <- 2 <- 3 <- 4 <- 5, watermark at 1, so the window is (1, 5] = 4 wide.
+        let meta = meta_with(vec![
+            snap(1, None, 1, Some("append")),
+            snap(2, Some(1), 2, Some("append")),
+            snap(3, Some(2), 3, Some("append")),
+            snap(4, Some(3), 4, Some("append")),
+            snap(5, Some(4), 5, Some("append")),
+        ]);
+
+        // Cap 2 -> advance to snapshot 3, NOT 5: the two oldest after the
+        // watermark. Snapshot 2 must not be skipped.
+        assert_eq!(meta.window_end_capped(Some(1), 5, 2).unwrap(), 3);
+        // Cap 1 -> one step at a time.
+        assert_eq!(meta.window_end_capped(Some(1), 5, 1).unwrap(), 2);
+        // Successive passes converge on the head rather than stalling short of it.
+        assert_eq!(meta.window_end_capped(Some(3), 5, 2).unwrap(), 5);
+    }
+
+    #[test]
+    fn window_end_capped_is_a_no_op_when_it_cannot_help() {
+        let meta = meta_with(vec![
+            snap(1, None, 1, Some("append")),
+            snap(2, Some(1), 2, Some("append")),
+            snap(3, Some(2), 3, Some("append")),
+        ]);
+
+        // Window already fits.
+        assert_eq!(meta.window_end_capped(Some(1), 3, 5).unwrap(), 3);
+        // Disabled.
+        assert_eq!(meta.window_end_capped(Some(1), 3, 0).unwrap(), 3);
+        // An initial full read has no prefix to take — capping it would produce a
+        // partial "full" read, which is worse than the unbounded one.
+        assert_eq!(meta.window_end_capped(None, 3, 1).unwrap(), 3);
+        // from == to: empty window, nothing to cap.
+        assert_eq!(meta.window_end_capped(Some(3), 3, 1).unwrap(), 3);
+    }
+
+    /// `capped_scan_end` is the decision a scan actually makes, so every way of
+    /// declining to cap is pinned here — a consumer calls this and nothing else.
+    #[test]
+    fn capped_scan_end_covers_every_decline_path() {
+        let meta = meta_with(vec![
+            snap(1, None, 1, Some("append")),
+            snap(2, Some(1), 2, Some("append")),
+            snap(3, Some(2), 3, Some("append")),
+            snap(4, Some(3), 4, Some("append")),
+        ]);
+        let head = 4;
+
+        // Caps when the backlog exceeds the limit.
+        assert_eq!(meta.capped_scan_end(Some(1), 2).unwrap().snapshot_id, 3);
+
+        // Declines: disabled, initial full read, backlog already fits.
+        assert_eq!(meta.capped_scan_end(Some(1), 0).unwrap().snapshot_id, head);
+        assert_eq!(meta.capped_scan_end(None, 1).unwrap().snapshot_id, head);
+        assert_eq!(meta.capped_scan_end(Some(1), 99).unwrap().snapshot_id, head);
+
+        // Declines: the window cannot be walked. `from` is not an ancestor here,
+        // which is the rollback/branch case — the full-read fallback owns it, so
+        // this must return the head rather than inventing a bound.
+        let orphan = meta_with(vec![
+            snap(7, None, 7, Some("append")),
+            snap(8, Some(7), 8, Some("append")),
+        ]);
+        assert_eq!(orphan.capped_scan_end(Some(1), 1).unwrap().snapshot_id, 8);
+
+        // No snapshots: nothing to read, and no panic.
+        assert!(meta_with(vec![]).capped_scan_end(Some(1), 1).is_none());
+    }
+
+    /// The cap must never hand back a snapshot NEWER than the head — that would
+    /// read past what the caller asked for.
+    #[test]
+    fn capped_scan_end_never_exceeds_the_head() {
+        let meta = meta_with(vec![
+            snap(1, None, 1, Some("append")),
+            snap(2, Some(1), 2, Some("append")),
+            snap(3, Some(2), 3, Some("append")),
+        ]);
+        for cap in 0..6 {
+            let chosen = meta.capped_scan_end(Some(1), cap).unwrap().sequence_number;
+            assert!(
+                chosen <= 3,
+                "cap {cap} chose sequence {chosen}, past the head"
+            );
+        }
+    }
+
+    /// An expired ancestor must still ERROR rather than silently capping to
+    /// something arbitrary — the caller falls back to a full re-read, and that
+    /// decision has to stay with the caller.
+    #[test]
+    fn window_end_capped_propagates_an_expired_ancestor() {
+        let meta = meta_with(vec![
+            snap(3, Some(2), 3, Some("append")),
+            snap(4, Some(3), 4, Some("append")),
+        ]);
+        assert!(meta.window_end_capped(Some(1), 4, 2).is_err());
+    }
+
     #[test]
     fn window_is_append_only_detects_non_append() {
         let all_append = meta_with(vec![