Skip to content

fix(materialize): skip materialize targets that already applied the window - #1647

Merged
bplatz merged 6 commits into
fluree:mainfrom
christophediprima:fix/materialize-per-target-applied-marker
Aug 20, 2026
Merged

fix(materialize): skip materialize targets that already applied the window#1647
bplatz merged 6 commits into
fluree:mainfrom
christophediprima:fix/materialize-per-target-applied-marker

Conversation

@christophediprima

@christophediprima christophediprima commented Aug 13, 2026

Copy link
Copy Markdown
Contributor

A templated job's watermark is shared across every ledger it fans into, and it must not advance while any target is behind — otherwise the laggards skip that window permanently. That is correct, and this PR does not change it.

What it costs today is that the targets which did commit earn no credit. The next poll re-reads the same window and re-commits all of them, and it keeps doing that for as long as one target cannot fit.

This is the follow-through on a note I left in r2rml_materialize.rs when per-target failure isolation landed in #1422:

Note the loop already commits per target as it goes, so partial application was ALWAYS the reality. What was missing is recording which targets succeeded.

That PR made partial application survivable; it did not make it recorded. This records it. A resolved target that already holds the window is skipped instead of rewritten.

What it cost us

A 17-source job fanning out per (tenant, user) into 23 ledgers, one of which has a window larger than the novelty ceiling:

  • targets_ok=19 targets_deferred=4 on 22 consecutive polls, byte-identical remainder each time (items_deferred=421708)
  • the job's watermark was never written once in 13.7 h
  • so 19 ledgers were re-committed every ~4 minutes for 13.7 h

The re-commits are idempotent but they are still writes. They filled a 100 GiB volume: 479 No space left on device failures across 8 target ledgers in 13 minutes. That state does not self-heal — index GC has to write in order to reclaim anything, so at zero bytes free it cannot run.

After this change, on the same deployment and the same data:

before after
volume used 119 GiB, climbing 79 GiB, flat
CPU 2359m 851m
per-poll targets ok=19 deferred=4 ok=22 deferred=1 skipped=22

The deferral count dropping 4 → 1 was not predicted: removing the rewrite churn relieved enough novelty pressure that three of the four stuck targets got through on their own.

Design

Additive. A new urn:fluree:materialize#appliedSnapshotId marker per (source, RESOLVED target, table). The shared watermark still chooses the scan window, so the source read is byte-for-byte unchanged. A state ledger written before this simply has no markers, so the first poll behaves exactly as it did and every poll after is cheap.

Why not per-target scan watermarks, which is the obvious alternative: taking min() across targets as the window start is wrong the moment a target is created. A newly-created (tenant, user) ledger has no watermark, so min() over the existing ones starts the scan after rows that target still needs, and it silently never receives them. Detecting that mid-window and forcing a re-scan is real complexity, and it buys nothing here — per-target watermarks fix the rewrite, which this fixes more cheaply, not the volume.

Two orderings, both load-bearing:

  • Markers are written before the partial-window early returns. A job whose watermark cannot advance is precisely the job that needs them, and both returns sit between that point and the watermark write. Putting this after them would record progress only for jobs that never needed it.
  • Markers are written after the target's data commit. A crash in the gap re-applies the window, which is idempotent; a marker written first would skip rows that never landed. Same rule as the watermark, one level down.

Edges decided rather than left open:

  • A target must be caught up on every table in the window, not any. A target that applied one table of a two-table window still owes the other, and skipping it there would drop half a window silently — the only way this change could lose data rather than repeat work.
  • An empty window marks nothing caught up. all() over an empty slice is vacuously true, so without an explicit guard a source with no snapshots yet would skip every target forever and materialize nothing.
  • ==, never an inequality. Both sides are Iceberg snapshot ids, which the spec assigns randomly — there is no ordering between two of them, not even between a snapshot and its own parent. An earlier revision of this PR compared them with >=, which turns the marker into a one-way ratchet: it climbs the running maximum of a random sequence and after roughly ln(n) windows exceeds every later draw permanently. That ran on a 17-source deployment, where every marker ratcheted past every subsequent window — all 17 sources skipped on every poll, each skip counted as an ok target, nothing logged, the shared watermark advancing past data that was never applied, and about 80% of the source entities discarded. Equality is what the question actually asks, and it covers the amplification this PR exists to stop, because a job whose watermark cannot advance re-presents the same to every poll. If "at or beyond" is ever genuinely needed it requires a quantity that IS ordered — the Iceberg sequence number — stored alongside the snapshot id. Full account in 9820a68ad and on target_is_caught_up.
  • Unreadable markers are non-fatal and cost only the skip. Re-applying is idempotent, and failing here would turn a state-ledger hiccup into a stalled job.
  • force_full ignores the markers, for the same reason it ignores the watermark: the caller is asking for a rebuild, not a resume.

Testing

fmt and clippy --all-features --all-targets -- -D warnings clean. fluree-db-api --lib 1074 pass, 0 fail.

Six tests: five on the predicate and one driving the engine end to end over a re-presented window — same rows, same to_snapshot_id, which is exactly what a job whose watermark cannot advance sees on every poll. FakeSource drains its batches by design, so that needs an opt-in repeating() variant rather than a change to the default.

Four mutations: dropping the non-empty guard, allany, ==>=, and removing the skip branch entirely. Each is killed in isolation — the first, second and fourth by exactly one test, and ==>= by two (the equality predicate test and a_marker_numerically_above_a_later_snapshot_is_not_caught_up, which carries five real (marker, later-snapshot) pairs from the incident rather than a sequence-shaped fixture; an ordered fixture would ratify an ordering comparison). Removing the skip branch is the one worth dwelling on — the predicate tests all stayed green against it, because they only exercise a pure function. Only the engine test catches it, and it does so on the target ledger's t: assertion left == right failed: people_acme:main already applied this window; a second commit means the rewrite is back / left: 3 / right: 2. That mutation was re-run after the rebase onto a85e03682, with the same result.

Two assertions were wrong before this landed, and both looked authoritative:

  • subjects_upserted is live.len(), the subjects the accumulator prepared, read before the target loop runs. It reads 3 whether or not a single commit happened.
  • Reading the target data proves nothing either: re-application is idempotent, so it is identical whether the target was skipped or rewritten.

The target ledger's t is the one observable that separates them, because it only moves on a real commit. Removing the skip advances it 2 → 3 and fails the test.

Not in this PR

This stops the livelock consuming disk. It does not make an oversized target complete — four targets in our deployment still need ~45 MB in a single pass and still cannot get it, so that table remains without a watermark. Fixing that needs sub-window progress, which is a separate change and a separate discussion: subdividing the window at snapshot boundaries needs no new vocabulary at all, but does nothing when the volume sits in one snapshot, and the general case needs a per-target file cursor. Happy to follow up if you have a view on how progress should be represented.

@christophediprima

Copy link
Copy Markdown
Contributor Author

Pushed one more commit, and it fixes a data-loss bug in what I originally proposed here. Flagging it prominently rather than quietly amending, because the failure mode is nasty and the reasoning is worth having in the thread.

target_is_caught_up compared the applied marker against the window with applied >= to. Both are Iceberg snapshot ids — and snapshot ids are randomly assigned, with no ordering between them, not even between a snapshot and its own parent. So the check was comparing two unrelated 64-bit numbers.

That turns the marker into a one-way ratchet. It is only rewritten when a window's to happens to compare greater, so it climbs the running maximum of a random sequence and then exceeds every later draw permanently. After roughly ln(n) windows the target is skipped forever.

We hit this on a 17-source deployment. Every marker had ratcheted to between 8.19e18 and 9.22e18 against an i64::MAX of 9.223e18, so all 17 sources were skipped on every poll — each skip counted as an ok target, nothing logged, and the shared watermark advancing past data that was never applied. 3,505 rows in one source table against 102 in the target. No warning anywhere in three days of logs; the only observable that disagreed was the target ledger's t, which never moved.

The fix is ==. That is what the predicate's own doc comment already claimed — "Already has this whole window" — and it fully covers the amplification this PR exists to stop, because a job whose watermark cannot advance re-presents the same to on every poll, so equality matches and the skip still fires. The two cases my old comment cited as needing an inequality both survive it: a re-poll of an older snapshot now re-applies, which is idempotent and strictly safer than skipping data that was never applied; and a window moving backwards after a forced full is moot, since force_full discards the markers before the target loop runs.

If "at or beyond" is ever genuinely wanted, it needs a quantity that actually is ordered — the sequence number, which increases by one per commit — which means storing that alongside the snapshot id in the marker. I have not done that here; it is a schema change and this PR does not need it.

Two other changes came with it:

  • Skips are now logged on every pass, not only an incomplete one. An all-skip pass is is_complete(), so it was the single outcome that printed nothing at all — which is exactly the shape a wrongly-skipped target takes.
  • The test that asserted the old behaviour is replaced, not adjusted. It required a marker of 21 to count as caught up against a window ending at 20. Every fixture in it was sequence-shaped (10, 19, 20, 21), which is precisely why >= read as correct.

That last point is the part I would most want a reviewer to notice. My earlier description of this PR listed >=== as a killed mutation — so the suite was actively defending the defect, and the mutation testing gave me false confidence. A mutation suite can only ask whether a test notices an edit; it cannot ask whether the fixtures resemble production values. When a domain has no ordering, ordered fixtures will ratify an ordering comparison. The new regression test therefore carries five real (marker, later-snapshot) pairs from the incident and asserts up front that each marker really is numerically larger than the snapshot it wrongly skipped, so the fixture cannot quietly stop testing the trap.

fmt and clippy --all-features --all-targets -- -D warnings clean; fluree-db-api --lib 1075 pass, 0 fail. Reverting == to >= fails both new tests.

christophediprima and others added 3 commits August 18, 2026 14:49
A templated job's watermark is shared across every ledger it fans into, and it
must not advance while any target is behind or the laggards skip that window
permanently. Correct — but the targets that DID commit earn no credit, so the
next poll re-reads the window and rewrites all of them, for as long as one
target cannot fit. The existing comment above the loop already names the gap:
"What was missing is recording which targets succeeded."

Measured on our deployment: 19 of 23 targets committing and 4 deferring on
every poll, byte-identical across 22 consecutive polls, so 19 ledgers were
rewritten every ~4 minutes for 13.7 h while the job never once wrote a
watermark. The rewrites are idempotent but they are still writes, and they took
a 100 GiB volume to ENOSPC — 479 `No space left on device` failures across 8
target ledgers in 13 minutes. That state does not self-heal: GC has to write in
order to reclaim.

So record per resolved target what it has applied, and skip a target that is
already caught up. Deliberately ADDITIVE — the shared watermark still chooses
the scan window, so the source read is unchanged.

Why not per-target scan watermarks, which is the obvious version: taking
`min()` across targets as the window start is wrong the moment a target is
created. A new (tenant,user) ledger has no watermark, so `min()` over the
existing ones starts the scan AFTER rows it still needs, and it silently never
receives them. Detecting that mid-window and forcing a re-scan is real
complexity for no extra benefit here. Skipping needs no such thing.

Ordering that matters, both learned from the failure modes above:

- markers are written BEFORE the partial-window early returns. A job whose
  watermark cannot advance is precisely the job that needs them, and both
  returns sit between that point and the watermark write.
- markers are written AFTER the target's data commit. A crash in the gap
  re-applies the window, which is idempotent; a marker written first would skip
  rows that never landed.

Unreadable markers are non-fatal and just cost the skip, since re-applying is
idempotent and failing would turn a state-ledger hiccup into a stalled job. An
older state ledger simply has none, so the first poll behaves exactly as before.

Five tests. Three mutations each killed by one test in isolation: dropping the
non-empty guard (an empty window would otherwise mark every target caught up
vacuously and materialize nothing), `all` -> `any` (a target owing one table of
a two-table window would be skipped, dropping half a window — the only way this
change could lose data rather than repeat work), and `>=` -> `==`.
The five tests in the previous commit cover `target_is_caught_up` as a pure
function. They say nothing about whether the engine consults it, which is the
part that actually stops the rewrite — gutting the skip branch left all of them
green.

So drive the engine twice over a re-presented window: same rows, same
`to_snapshot_id`, which is exactly what a job whose shared watermark cannot
advance sees on every poll. `FakeSource` drains its batches by design (that
catches a fixture pulled twice when it should be pulled once), so this needs an
opt-in `repeating()` variant rather than a change to the default.

Two assertions were wrong before this landed, and both are worth naming because
each looked authoritative:

- `subjects_upserted` is `live.len()`, the subjects the accumulator PREPARED,
  read before the target loop runs. It is 3 whether or not a single commit
  happened, so asserting it went to 0 failed against working code.
- reading the target DATA proves nothing either: re-application is idempotent,
  so it is identical whether the target was skipped or rewritten.

The target ledger's `t` is the one observable that separates the two, because it
only moves on a real commit. Removing the skip branch advances it 2 -> 3 and
fails the test; that mutation was run.

Also asserts the first pass actually committed (`t > 0`), so the test cannot
pass by materializing nothing at all.
The applied-marker check this branch adds decided whether a target had already
applied a window with `applied >= to`. Both are Iceberg snapshot ids, and the
spec assigns those randomly — there is no ordering between two snapshot ids,
not even between a snapshot and its own parent. The comparison was between two
unrelated 64-bit numbers.

That makes the marker a one-way ratchet. It is only rewritten when a window's
`to` happens to compare greater, so it climbs the running maximum of a random
sequence and then exceeds every later draw permanently — after roughly ln(n)
windows the target stops being written to at all.

Found in production before this merged. On a 17-source deployment every marker
had ratcheted to between 8.19e18 and 9.22e18 against an i64::MAX of 9.223e18,
so all 17 sources were skipped on every poll. Each skip was counted as an ok
target and logged nothing, while the watermark advanced past data that was
never applied: 3,505 observations in the source against 102 in the target,
about 80% of all entities discarded, and no warning in three days of logs. The
only observable that disagreed was the target ledger's `t`, which never moved.

Equality is what the predicate's own doc comment already claimed ("Already has
this whole window"), and it fully covers the amplification this change exists
to stop: a job whose watermark cannot advance re-presents the SAME `to` on
every poll. The two cases the old comment cited both survive it — a re-poll of
an older snapshot now re-applies, which is idempotent and strictly safer than
skipping unapplied data, and a window moving backwards after a forced full is
moot because force_full discards the markers before the target loop.

Also report skips on every pass rather than only an incomplete one. An
all-skip pass is `is_complete()`, so it was the one outcome that logged
nothing at all, which is exactly the shape a wrongly-skipped target takes.

The test asserting the old behaviour is replaced rather than adjusted: it
required a marker of 21 to count as caught up against a window ending at 20.
Every fixture in it was sequence-shaped (10, 19, 20, 21), which is why `>=`
read as correct — and the earlier commit here recorded `>=` -> `==` as a
killed mutation, so the suite was defending the defect. A mutation suite
cannot ask whether a fixture is representative, so the new regression test
carries five real (marker, later-snapshot) pairs from the incident and asserts
up front that each marker is numerically larger than the snapshot it wrongly
skipped.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
@christophediprima
christophediprima force-pushed the fix/materialize-per-target-applied-marker branch from fc9fdb8 to 9820a68 Compare August 18, 2026 12:59

@aaj3f aaj3f left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I started reviewing this before your more recent updates and the PR has materially approved, @christophediprima. The == reversal in 9820a68ad is the right call for the right reason (two Iceberg snapshot ids simply have no ordering, so >= was comparing unrelated 64-bit numbers and ratcheting), and the commit body's incident narrative (17 markers at 8.19e18–9.22e18 against i64::MAX, every source silently skipped, ~80% of entities discarded with no warning in three days of logs) is a very good account of a silent-data-loss mechanism.

I verified the guard rather than trusting it: at head, a_re_presented_window_does_not_rewrite_targets_that_already_applied_it passes, and gutting target_is_caught_up fails it with exactly the assertion your body quotes (left: 3 / right: 2 on the target's t). The five-real-pairs regression test and the every-pass skip reporting both landed as described.

The one thing I'd ask before this merges is a more about PR body than code itself: the PR description still argues for >= in "Edges decided" and still lists >= --> == as a killed mutation. Both now contradicted by your own head commit. That prose is what the next person reads before deciding the comparison is safe to "fix", so it should tell the incident's story instead. Details inline.

Adherence to repo commitments:

  • Patterns/abstractions: ✔ extends the module's existing marker/watermark vocabulary; the repeating() opt-in on FakeSource preserves the drain-by-default trap-catching behavior rather than weakening it.
  • Performance (speed first, memory second): ✔ comparison-operator change plus one info! per pass on the skip path; no hot-path impact.
  • Testing: ✔ predicate unit tests + the incident-pair regression test + a mutation-verified engine test on the one observable that can't lie. (Note for CI-readers: the module is iceberg-feature-gated, so a plain --lib run filters these tests out.)

Verified locally at branch HEAD (9820a68ad): cargo test -p fluree-db-api --lib --features iceberg a_re_presented_window → 1 passed; skip-predicate mutation → red with the body's exact assertion; restored → green; git merge-tree against #1663's head → conflict-free, and neither branch contains the other.

Approving so you can merge when ready — just get the body brought up to date with 9820a68ad first, since right now it's the one thing still defending the bug you fixed.

!table_watermarks.is_empty()
&& table_watermarks.iter().all(|(table, _from, to)| {
applied
.get(&(target.to_string(), table.clone()))

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

fluree-db-api/src/graph_source/r2rml_materialize.rs:1190blocking (a body edit, not a code change). The code is right and the description is now arguing against it, in two places. 9820a68ad reverses the applied-marker comparison to == — with the production ratchet incident as the reason, and with a_marker_numerically_above_a_later_snapshot_is_not_caught_up pinning it on real (marker, later-snapshot) pairs — but the PR body's "Edges decided rather than left open" section still reads ">=, not ==: a marker ahead of this window still means the target has it. Equality would re-apply and rebuild the amplification", and the body's mutation list still cites ">===" as a killed mutation, which is the suite your own commit describes as defending the defect and replaces. The body is what the next maintainer reads before "fixing" == back to >=, so those two passages should tell the story the head commit tells — the incident narrative in 9820a68ad is the best prose in the PR and deserves to be in the body rather than only in the git log.

}

#[tokio::test]
async fn a_templated_target_fans_out_per_row() {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

fluree-db-api/src/graph_source/r2rml_materialize.rs:3666nit. a_templated_target_fans_out_per_row's doc comment ("A templated target fans out: one scan, N target ledgers, each its own commit domain…") is fused onto the top of the new skip test's doc block at :3577-3579, and the fan-out test itself is left with no doc at all. Just the three lines moved back above :3666. Minor and non-blocking — but if you agree it's right, I'd rather see it folded in now than lost in the backlog.

// wrongly-skipped target takes. `all_skipped` is the specific alarm: nothing was
// written this pass, yet the watermark is about to advance. That is legitimate
// for a genuinely idle window and a data-loss signature otherwise, so it has to
// be visible either way rather than inferred by diffing a target ledger's `t`.

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

fluree-db-api/src/graph_source/r2rml_materialize.rs:750praise. Reporting skips on every pass with the all_skipped flag — rather than only on an incomplete one — closes the exact observability hole the incident exposed ("each skip was counted as an ok target and logged nothing"). The one outcome that used to log nothing at all was precisely the shape a wrongly-skipped target takes, and now it can't be silent.

/// Small sequence-shaped fixtures cannot catch this — an ordered fixture will
/// ratify an ordering comparison — which is why these are the production numbers.
#[test]
fn a_marker_numerically_above_a_later_snapshot_is_not_caught_up() {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

fluree-db-api/src/graph_source/r2rml_materialize.rs:3036praise. Carrying five real (marker, later-snapshot) pairs from the incident, with the up-front assertion that each marker is numerically larger than the snapshot it wrongly skipped, and the stated reason — "an ordered fixture will ratify an ordering comparison" — is the fixture-representativeness lesson encoded where a mutation suite can't reach it. And the e2e test's choice of the target ledger's t as the only honest observable (with the doc block explaining why subjects_upserted and the target data are both vacuous) is exactly the kind of test that survives contact with a maintainer.

bplatz added 3 commits August 20, 2026 09:07
… test

Three lines describing `a_templated_target_fans_out_per_row` — one scan, N
target ledgers, and the tally counting targets rather than polls — had been
fused onto the head of the applied-marker skip test's doc block, leaving the
test they describe with no doc at all. Moved back, no code change.

Raised in review of fluree#1647.
…not the pass

The marker write propagated its error, which cost more than the write. It sits
between the target loop and the failed/deferred returns, so a state-ledger
failure surfaced as that error INSTEAD of MaterializePartial { tally, detail } —
discarding the per-target tally and the identity of the target that actually
failed. The condition most likely to fail it is the one this module most needs
to report: at ENOSPC, the incident these markers exist to prevent, every
state-ledger write fails, so an operator got a bare disk error where the useful
signal was "target X failed: No space left on device".

It is now non-fatal, matching the marker READ, which was already deliberately
non-fatal on identical reasoning: an unrecorded marker costs one redundant
re-apply on the next poll and nothing else, and re-application is idempotent.
The write moves into a helper so the two `?`s it needs stay inside it.

Also records what the marker set costs to hold. It is unbounded in a way the
watermark deliberately is not: the watermark is keyed on the target spec, so a
templated job keeps one row per source table however wide it fans, while a
marker is keyed on the resolved ledger, so the set grows with the fan-out, every
poll reads all of it, and nothing evicts a marker when a target is retired.
Immaterial at 23 targets; noted where the next person will look.

Both raised in review of fluree#1647.
@bplatz

bplatz commented Aug 20, 2026

Copy link
Copy Markdown
Contributor

Pushed four commits to this branch (maintainerCanModify), @christophediprima — two housekeeping, two from review. Happy to drop any of them if you disagree, particularly the third.

origin/main merged in (36 commits, conflict-free). Materialize tests, fmt and clippy re-verified after.

The fan-out test's doc block moved back to the fan-out test. Three lines describing a_templated_target_fans_out_per_row had been fused onto the head of the skip test's doc, leaving the test they describe undocumented. No code change.

The applied-marker write is now non-fatal. This is the one that is a judgement call rather than a tidy-up, so pushing back is reasonable.

The write propagated its error, and it sits between the target loop and the failed/deferred returns — so a state-ledger failure surfaced as that error instead of MaterializePartial { tally, detail }, discarding the per-target tally and the identity of the target that actually failed. What decided it for me is which condition makes that write fail: at ENOSPC — the incident these markers exist to prevent — every state-ledger write fails, so an operator would get a bare disk error exactly where the useful signal is "target X failed: No space left on device". It also read as asymmetric with the marker read 90 lines above, which you had already made deliberately non-fatal on what looks like identical reasoning ("an unreadable marker set costs the skip, not the pass"). An unwritten marker costs one redundant re-apply next poll, and re-application is idempotent. The write moved into a helper so its two ?s stay inside it.

A note on what the marker set costs to hold, in materialize_applied_markers. It is unbounded in a way the watermark deliberately is not: the watermark is keyed on the target spec, so a templated job keeps one row per source table however wide it fans, while a marker is keyed on the resolved ledger — so the set grows with the fan-out, every poll reads all of it, and nothing evicts a marker when a target ledger is retired. Immaterial at the 23 targets this was built for, and not something I think belongs in this PR; recorded where the next person will look.

I also updated the PR description, which was the blocking item in @aaj3f's review: "Edges decided" still argued for >= over ==, and the mutation list still credited >=== as killed. Both now tell the ratchet incident instead, and the mutation is recorded in the correct direction and as killed by two tests rather than one — I re-ran it to check that count rather than take it from the review.

Two things I verified rather than trusted while reading: mutating == back to >= fails exactly two tests (including a_marker_numerically_above_a_later_snapshot_is_not_caught_up), and allany on the every-table guard fails exactly one. The suite bites where it claims to.

@bplatz
bplatz merged commit 38a6fb3 into fluree:main Aug 20, 2026
15 checks passed
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants