diff --git a/fluree-db-api/src/graph_source/r2rml.rs b/fluree-db-api/src/graph_source/r2rml.rs index 24461d9486..cb46a4dd9a 100644 --- a/fluree-db-api/src/graph_source/r2rml.rs +++ b/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 = std::sync::OnceLock::new(); + *CACHED.get_or_init(|| { + std::env::var("FLUREE_MATERIALIZE_MAX_SNAPSHOTS_PER_PASS") + .ok() + .and_then(|v| v.trim().parse::().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. diff --git a/fluree-db-iceberg/src/metadata/table.rs b/fluree-db-iceberg/src/metadata/table.rs index e6ac648a3f..44694873d7 100644 --- a/fluree-db-iceberg/src/metadata/table.rs +++ b/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, + to_id: i64, + max_snapshots: usize, + ) -> crate::error::Result { + 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, + 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![