Incremental sink reads: per-row ingest-seq watermark for forward + blob sinks (LLP 0039/0040)

[philcunliffe]

Implements the incremental sink reads change set — sinks read only rows added since their last successful export instead of re-reading the whole partition every tick.

• Request: LLP 0039 (escalated from central forward sink has no cursor — re-reads & re-sends the whole dataset every tick #122) · Design: LLP 0040 · Plan: LLP 0042
• T1 stamp a monotonic _hyp_ingest_seq at the decorateRow write chokepoint (crash-safe allocator)
• T2 extend storage.readRows with cursor-aware since/continuation + null-seq migration
• T3 persist a per-(sink instance, partition) watermark keyed by logical partition path
• T4 wire the central forward sink to incremental read
• T5 wire the core blob sink (local-fs + s3)
• T6 exactly-once tests across retention prune + compaction generation swap

Each task landed as its own verified --no-ff merge with green CI. Server idempotency ledger retained as the in-flight retry net.

Change-Set: incremental-sink-reads

[philcunliffe]

🤖 neutral dual-review — verdict: block

Blast radius / risk: HIGH — kernel storage read API (QueryStorageService.readRows/readRowsSince) + the cache write chokepoint (decorateRow seq stamp) + both production sinks. A skip in the seq/watermark contract silently drops rows forever; a dup breaks the spec's exactly-once guarantee. Reviewed at 35d77bd (pre-fix head).

Both reviewers ran. Both verdicts: block.

The exactly-once skeleton is sound where it's been proven: the T6 acceptance suite is non-vacuous — it drives the real cache, real retention enforcer, real maintainCache compaction, both real sinks, and the real driver outbox, and genuinely asserts no skip/no dup across a retention front-prune and a compaction generation swap. I empirically confirmed the load-bearing claims hold: retention deletes only sub-watermark rows; the seq rides compaction verbatim; the logical-partition watermark key reads straight through the tableDir swap; the allocator is crash-safe and never-regressing (reserve-before-stamp). Exactly-once across retention+compaction: HELD. The blockers below are in the retry/redelivery and migration edges the T6 suite does not exercise.

Codex key points

Codex (gpt-5.5, xhigh) returned 3 blockers + 3 majors + 1 nit. Reconciled:

• Batch-id keyed on per-tick chunk index (blocker) — confirmed & FIXED (see below).
• Null-seq legacy rows are "new forever" (blocker) — confirmed. Not a one-time migration.
• Cross-process duplicate seqs → skip (blocker) — confirmed, with caveat: design-documented out-of-scope; concurrent same-partition flush already crashes on a cursor.json race, bounding it.
• Spool progress advances after append (major) — pre-existing spool behaviour, not introduced here; compaction dedupes by _hyp_cache_row_id. Downgraded to minor.
• appendRowsToPartition bypasses seq stamping (major) — confirmed; same root cause as the null-seq blocker (permanent null-seq rows).
• Retention uncoupled from watermark (major) — explicitly documented out-of-scope in LLP 0040 §6 risk [codex] clarify first-run picker and npx daemon bin #5; not a defect of this PR. Note only.
• Inline import() types in test files (nit) — valid style nit; left as-is.

My key points (independent Claude pass)

• I independently reproduced the batch-id instability and the null-seq re-export against the real kernel cache.
• Null-seq re-export is a true exactly-once duplication, not just waste: scanRowsFromTable unconditionally yields null-seq rows for any since, and they never advance the monotonic after. So after the first export advances the watermark to a real seq, every later tick re-yields the legacy rows — the blob sink writes them into a second object (legacy rows appear in both the 0-maxSeq blob and the later maxSeq-maxSeq blob → duplicated), and the forward sink re-sends them under a different body+batch-id (server stores them again). Affects every deployment with a pre-upgrade cache until retention prunes the legacy rows (≥30 days), and permanently for any dataset written via appendRowsToPartition. Violates spec acceptance "≈0 bytes on a no-new-rows tick" and "no row duplicated".
• Latent fragility (minor, not currently reachable): the forward sink's per-chunk watermark advance to the running-max after is only skip-safe because the physical read order happens to be seq-monotonic in every reachable path (sequential flush, post-compaction — both I verified empirically; the one out-of-order trigger, concurrent same-partition flush, crashes first). It is not a guaranteed icebird contract — icebergDataSource even documents "record-count order" for file iteration. If file read order ever ceases to be seq-ascending, a mid-tick respool would skip the lower-seq suffix. The blob sink is robust (single end-of-scan advance); the forward sink is not. Worth a guard or an explicit ordered-read contract.

Fix pushed (round 2 will re-review)

Batch-id stability (hypaware-core/plugins-workspace/central/src/sink.js): batchIdForChunk now keys on the chunk's start seq (the watermark it resumes from) instead of the per-tick chunkIndex. A respooled suffix reproduces the same [startSeq, body] → same id → server ledger dedupes the redelivery; distinct chunks still differ (each row's _hyp_ingest_seq is unique). Added a cross-tick regression test (a respooled suffix re-sends the un-acked chunk with the SAME batch-id). Full suite green (1448 tests).

The two remaining blockers (null-seq migration; cross-process seq safety) are design-level — LLP 0040 §6 risk #1 itself flags the null-seq default as "must be nailed down (the wrong default … silently skips null-seq rows (data loss))". Resolving them is an author decision (stamp legacy rows during compaction / a one-time backfill migration, or track a legacyExported flag in the watermark; assert/guard single-writer for the allocator), so I did not guess at them in review — a wrong choice here risks silent data loss.

Findings

• Severity: blocker — Forward sink batch-id keyed on per-tick chunkIndex: after a watermark advance, a respooled (committed-but-unacked) chunk gets a new id and the server ledger can't dedupe it → duplicate rows. Violates spec "ledger covers mid-batch retries". FIXED in 1141f3d.
• Severity: blocker — Null-seq legacy rows yielded every tick and never captured by the watermark → blob sink duplicates them across objects; forward sink re-sends them (server re-stores). Not a one-time migration; violates exactly-once + "≈0 bytes on no-new". src/core/cache/iceberg/store.js:330, src/core/cache/storage.js:227. Escalated (design decision).
• Severity: blocker/major — appendRowsToPartition writes committed rows without stamping _hyp_ingest_seq → permanent null-seq rows; same family as above; such datasets never reach a no-new-rows steady state. src/core/cache/partition.js. Escalated.
• Severity: major (caveat) — Allocator mutex is in-process only; two processes over one cache root (e.g. daemon + hyp backfill) can reserve the same block → duplicate seqs → a same-partition duplicate split across an export boundary is skipped forever. Documented out-of-scope in LLP 0040 §7; concurrent same-partition flush crashes on a cursor race, bounding reachability. Escalated.
• Severity: minor — Forward sink per-chunk watermark advance is skip-safe only under seq-monotonic read order, which icebird does not contract (vs. the blob sink's robust single advance). Not currently reachable; latent. hypaware-core/plugins-workspace/central/src/sink.js.
• Severity: minor — Spool advances writeProgress after appendChunk; a crash in between re-appends rows with new seqs (dup content until compaction dedup). Pre-existing spool behaviour, not introduced here. src/core/cache/spool.js:136.
• Severity: note — Retention is intentionally uncoupled from the watermark (LLP 0040 §6 risk [codex] clarify first-run picker and npx daemon bin #5): a lagging sink's un-exported front rows can still be pruned. Documented, out-of-scope.
• Severity: nit — Inline import('...') types in new test/smoke files violate the repo JSDoc rule. test/core/sink-watermarks.test.js:21, test/core/sink-incremental-acceptance.test.js:214, hypaware-core/smoke/flows/incremental_sink_compaction.js:209.

[philcunliffe]

🤖 neutral: held for a design decision (review verdict = block)

The dual-review found real exactly-once defects. One is fixed; three are design-level and need a human decision — neutral will not guess at choices that risk silent data loss, so this PR is neutral:stuck (won't auto-advance).

Fixed in this PR (1141f3d, regression test added, suite green):

• Forward-sink batch-id was keyed on the per-tick chunk ordinal, so a respooled committed-but-unacked chunk got a new id and the server ledger couldn't dedupe it → duplicate rows. Now keyed stably.

Needs a design decision (escalated, not fixed):

1. null-seq migration is not one-time. Legacy (pre-upgrade) rows have no _hyp_ingest_seq, are yielded every tick, and never advance the watermark → the blob sink duplicates them across objects and the forward sink re-sends them every tick. The design's own risk register flagged the null-seq default as "must be nailed down — wrong default risks silent data loss."
2. appendRowsToPartition doesn't stamp _hyp_ingest_seq → it writes permanent null-seq rows (same failure family as [codex] Add durable cache spool #1).
3. Allocator mutex is in-process only → the daemon and a concurrent hyp backfill can hand out the same seq block → duplicate/skew.

What holds: the core watermark design is sound — exactly-once across a retention front-prune and a compaction generation swap was verified, and the T6 acceptance suite is non-vacuous (drives real cache/retention/compaction/both sinks).

Recommended path: revise the design (LLP 0040) to nail down (a) the null-seq backfill/stamping strategy for legacy + appendRowsToPartition rows, and (b) cross-process allocator safety; then re-implement the affected tasks. Remove neutral:stuck once the design direction is set and neutral will resume.

[philcunliffe]

Dual-agent review — block

• Verdict: block
• Risk class: high (cache/sinks data-integrity; the exactly-once guarantee is the whole point of this PR)
• Reviewers: Codex (gpt-5.5) + 5 Claude reviewers, run in an isolated git worktree at the PR head.
• Advisory only: no merge was attempted.

The headline guarantee — exactly-once incremental reads — does not hold under the failure modes the design itself anticipates. One blocker (3 reviewers agree) plus four exactly-once-violating majors.

🛑 Blocker — watermark advances by max-seq over an unordered scan → silent permanent row skip

readRowsSince yields rows in physical scan order while emitting after = the running max seq seen (src/core/cache/storage.js:227-234); the forward sink advances its durable watermark per acked chunk to that max (central/src/sink.js:288-292). When physical order ≠ seq order — which the design itself anticipates (LLP 0040 risk #3 / §7: interleaved live+backfill spool; and post-compaction sortOrder re-sort, maintenance.js:517-518) — an early acked chunk advances the watermark past lower-seq rows sitting in a later, un-acked chunk. A crash or single chunk failure (poison 4xx / backpressure) between chunks then leaves since above those rows and the next tick filters them out (seq <= since) forever — silent permanent data loss, the opposite of the exactly-once claim. Every test appends rows in strictly increasing seq order, so this exact scenario is unproven. (Codex + 2 Claude reviewers.)
Fix: either scan ordered by _hyp_ingest_seq (so a shipped prefix truly means "all seqs ≤ last"), or advance the forward-sink watermark only once at end-of-partition (as the blob sink does). Add a forward-sink test with seqs out of physical order across a chunk boundary, fail the later chunk, re-tick, and assert the lower-seq rows are still delivered.

Major — legacy null-seq rows re-export every tick (never reach a no-new-rows steady state)

Null-seq rows are always yielded and never advance the watermark (iceberg/store.js:328-331, storage.js:229-230), so a cache upgraded from before this PR (its entire corpus is null-seq) re-exports its whole legacy backlog on every tick — violating the design's "one-time full re-export" (§6 risk #1 / plan T2) and the "≈0 bytes on a no-new-rows tick" acceptance. For a mixed/central partition this is duplication, not just cost (re-POST under a new batch-id after compaction reorders the body; blob S.0-10 then S.10-10). (Codex + 3 Claude reviewers.)
Fix: on first post-upgrade export of a partition, stamp the watermark to its max real seq (or stamp seqs onto legacy rows at next compaction) so re-export is genuinely one-time. Add 2nd-tick tests for pure-legacy and mixed partitions.

Major — mid-retry exactly-once breaks when new rows arrive before the retry (both sinks)

"Ship/PUT first, advance watermark second" is only exactly-once if the in-flight unit is fixed. If a chunk/blob commits server-side, the watermark write is lost, and new rows append before retry, the resumed export reads since=oldWatermark and the suffix grows: the forward chunk body changes → new batch-id → server stores it again; the blob lastSeq changes → proxy.0-50.json and proxy.0-60.json both land, duplicating rows 0–50. _hyp_cache_row_id is stripped from the wire so the server can only dedup by batch-id → at-least-once, not exactly-once. T6 tests only cover the no-new-rows-before-retry case. (2 Claude reviewers.)
Fix: make the in-flight unit boundary independent of later arrivals (cap the resumed chunk/blob at the in-flight seq range, or persist a pre-commit intent record so the retry reproduces the identical body/key), or correct §5 to claim at-least-once.

Major — central forward sink scopes watermarks per-plugin, not per-instance

central/index.js:69 uses createSinkWatermarkStore({ stateDir: sinkCtx.paths.stateDir }) (per-plugin) while local-fs/s3 use createInstanceWatermarkStore({ paths, instanceName: sinkCtx.name }). Two @hypaware/central instances (dual-homing) share one <stateDir>/watermarks/<dataset>/<partition>.json and clobber each other → silent skip/data loss. The @ref comment at central/index.js:68 ("scoped to the sink instance via PluginPaths.stateDir") is factually wrong — PluginPaths.stateDir is per-plugin. (Codex + 1 Claude reviewer.)
Fix: use the public createInstanceWatermarkStore({ paths: sinkCtx.paths, instanceName: sinkCtx.name }) for central too; drop the deep import; correct the @ref.

Major — bounded-reads constraint (C) is not delivered

Design §1 makes "a tick reads ≈N rows for N new rows, independent of total partition size" the deciding constraint that picks Candidate B, and §5 claims file-level max(seq) ≤ watermark pruning. As built there's no pushdown/file-skip (iceberg/store.js:309-331 full-scans the snapshot and applies seq <= since as a yielded-row filter), so reads are O(surviving rows) per tick and the stated advantage isn't realized or tested. (1 Claude reviewer.)
Fix: land the null-aware seq > since file/row-group skip, or correct §1(C)/§5/acceptance to state reads are bounded by surviving-partition size pending the icebird pushdown.

Minor — batchIdForChunk @ref gloss contradicts design §4

Code correctly keys the batch-id by chunkStartSeq (the "keep chunk batch-id stable across watermark advance" fix), but LLP 0040 §4 still describes the old chunkIndex model, so the cited section asserts the opposite of the code and §5's retry argument only holds under the code's keying (central/src/sink.js:251). Update §4 (keep refs honest / living docs).

Nit — inline import('...') types in new test/smoke files

sink-watermarks.test.js:21, sink-incremental-acceptance.test.js:214, incremental_sink_compaction.js:209 use inline import('...') types against the house rule (the PR's production files follow it). Low priority — loosely enforced elsewhere.

---

🤖 neutral-reconcile dual review (worktree-isolated). Held for a human — neutral never merges.

[philcunliffe]

neutral:stuck — one design decision remains

The block verdict was driven by the watermark-ordering blocker, which is now fixed and CI-green (commit 7819992): the forward sink advances the watermark only at end-of-partition, with a regression test proving an unordered scan no longer skips a lower-seq row when a later chunk fails. Also fixed this round and verified green (npm test 1453 pass, typecheck/lint clean): legacy null-seq one-time export (includeLegacy), central per-instance watermark scoping, and the bounded-reads (C) + §4 @ref doc corrections. LLP 0040/0042 were updated in the same commit.

Remaining — needs a human design decision (not auto-fixable)

Mid-retry duplication when new rows arrive before the retry (review finding #3). If a chunk/blob commits server-side, the watermark write is lost, and new rows append before the retry, the resumed in-flight unit grows → the forward batch-id changes (server re-stores) / the blob lastSeq changes (overlapping duplicate blob). The fix worker declined a half-fix and documented it as a known gap (LLP 0040 §6 risk #7 / §5), scoping the exactly-once claim to exclude this narrow window. The common retry (no new arrivals) stays covered by the server ledger / seq-range filename.

Choose one:

• (a) Ratify the documented at-least-once edge for this narrow window as an accepted limitation (LLP 0040 §5/§6 already corrected to say so), or
• (b) Implement a pre-commit intent mechanism — a read until upper bound + a persisted intent record so a retry caps its read at the committed lastSeq and reproduces the identical body/key, restoring full exactly-once. This is a watermark-contract change affecting both sinks.

Either way the resolution lands as additive commits on this same branch (mint a small decision LLP + update LLP 0040). Full review: the dual-review comment above on this PR.

Held as a draft — neutral will not re-review or churn this while neutral:stuck.