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

[christophediprima]

The failure this prevents

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 without bound.

That is not merely slow — it is a one-way door:

a pass defers  ->  the window is discarded, no watermark is written
               ->  the pinned snapshot ages out of the source's retention
               ->  the next poll cannot resolve `from_snapshot_id`
               ->  it falls back to a FULL TABLE READ
               ->  a full read is far more expensive, so it defers
               ->  repeat, permanently

Each turn makes the next more likely, and there is no exit: the entry condition for the incremental path — a resolvable from snapshot — is exactly what gets lost.

Measured on a 17-table deployment. One affected table became 13 of 17 in about four hours. 126 consecutive poll results, every one "partial window, remainder deferred", zero completions. All 13 tables showed an identical from_snapshot_id across nine consecutive polls: no watermark moved at all. Resident memory rose 19.5 → 24.7 GiB from the full-table reads alone.

The tables were not individually unhealthy. They were dragged there by a shared property of the watermark, and once there they could not get back.

The change

TableMetadata::capped_scan_end(from_id, max_snapshots) returns where an unpinned incremental consumer should end this pass: the head, or an earlier snapshot when the backlog exceeds the cap. An unpinned materialize scan reads to there instead of to head.

FLUREE_MATERIALIZE_MAX_SNAPSHOTS_PER_PASS, default 64; 0 disables it and restores the previous behaviour exactly.

64 is deliberately generous. A source committing 37–72 snapshots an hour — ours does — stays under it at any sane poll interval, so a healthy job never reaches the cap and nothing about its behaviour changes. Only a backlogged job reaches it, and it then drains in bounded steps instead of never.

Why it is safe

Three properties, each asserted by a test rather than argued:

• Capping only ever moves to earlier, never later, so no snapshot is skipped. The next pass resumes from the watermark this one wrote.
• An initial full read (from = None) is never capped. A partial "full" read is worse than an unbounded one — it would look complete while missing rows.
• Any window error falls through to the head unchanged — expired ancestor, non-ancestor, rollback. That is the existing full-read fallback's case and it keeps it.

The prefix is the oldest snapshots in the window, not the newest. That direction is the one thing here that is silent when wrong: taking the newest would skip everything between the watermark and the chosen end, which is data loss rather than a performance bug. window_end_capped_takes_the_oldest_prefix exists to fail if anyone inverts it, and it does.

On the second commit

The decision originally lived inline in scan_for_materialize_stream. Testing it there needs a storage backend and an Iceberg fixture, which in this repo means aws-testcontainers — a high price for asserting a branch, and the usual result is that the branch stays unasserted.

The second commit moves it to TableMetadata::capped_scan_end, beside the metadata fixtures, so the scan path calls it and has no logic of its own left to get wrong. Every decline path is then covered by an ordinary unit test: no snapshots, cap disabled, initial full read, backlog already fits, window unwalkable.

Three mutations, each killed by exactly one test: taking the newest prefix instead of the oldest, capping an initial full read, and dropping the fall-back-to-head on a window error.

Scope, stated plainly

This prevents a table from entering that state. It does not rescue one already in it. A table whose watermark has already expired takes the full-read path, and this deliberately does not cap a full read — so it stays on that path until one read completes. The fix for the already-stuck case is sub-snapshot progress (a resumable file cursor), which is a larger change to the scan API and the persisted watermark, and belongs in its own PR.

We are happy to bring that too, and would rather agree the shape first — particularly whether a per-(source, target, table) cursor is acceptable state growth, and whether it belongs in the state ledger or the target ledger. This PR stands on its own either way: it is the difference between a degradation that spreads and one that does not.