Restore subquery shapes correctly after a server restart (materializer move replay)

[robacourt]

Summary

Optimized streaming subquery shapes could diverge from Postgres after a graceful server restart — permanently losing or retaining stale rows — including the nested case where the inner subquery is itself an optimized subquery. This fixes that. A representative shape: issues WHERE project_id IN (SELECT id FROM projects WHERE active = true).

Stacked on #4666 (a separate shutdown shape-removal fix) — that's this PR's base; review/merge it first. Also builds on the already-merged #4668 (consumer-level restart de-duplication).

Why it diverges

A subquery shape is served by a materializer that tracks which rows the inner query selects and drives moves (move-in / move-out) into the outer shape as that set changes. A move is asynchronous and multi-stage: a move-in fires a Postgres query and only splices its rows into the shape log once that query returns.

A graceful restart can interrupt a move at any stage — in flight, queued behind another move, sitting unprocessed in the consumer's mailbox, or computed by the materializer but not yet emitted. When it does, the materializer rebuilds its in-memory view from disk on restart (so it "knows" the move happened), but the outer shape's storage never received it and nothing re-drives it. The shape is left permanently out of sync.

The approach: replay missed moves from a durable position

Rather than making each of those stages individually crash-safe — a fragile, case-by-case effort — this makes the materializer the single point that re-drives moves, so every timing class is handled by one mechanism.

Each emitted move is tagged with its source LSN, and each outer consumer persists, per dependency, the position up to which it has applied moves. On restart the consumer re-subscribes and the materializer replays exactly the moves after that position; the consumer re-applies them. Because the consumer just replays whatever it is handed, in-flight / queued / mailbox / not-yet-emitted all reduce to "the persisted position is behind" — covered by construction rather than case by case.

For that to be safe across a crash the persisted position must never run ahead of durable storage, so it is coupled to the writer flush: a move's position is staged when the move pipeline drains and only committed once the writer confirms the flush has passed that move's spliced rows (and unconditionally at terminate, after the writer is flushed).

Making the same mechanism correct for nested subqueries

When the dependency is itself an optimized subquery, its log contains move-in/move-out control messages, not just data rows — and two properties of those messages break the naive replay:

• they carry no lsn/op_position in their JSON headers; and
• being dependency-driven, they have no WAL LSN of their own, so they all share a single tx_offset (only op_offset increments).

Replay originally reconstructed each item's offset from its JSON headers and grouped items into batches by tx_offset. Control messages therefore couldn't be positioned (no header offset) or separated (shared tx_offset), so the replayed seed collapsed to the pre-log snapshot state and the corrective move was never re-emitted. Two changes fix this while leaving the mechanism above intact:

• read each item with its authoritative storage offset (Storage.get_log_stream_with_offsets/3) instead of re-parsing headers, so control messages are positioned correctly; and
• chunk replay batches by the full (tx_offset, op_offset) offset — the space from_lsn/positions actually live in — so op-distinguished dependency moves are delimited per move.

The commits

Each fix lands with the failing test that motivates it:

1. Add failing subquery-restore oracle test — a deterministic restart reproduction.
2. Replay missed moves after restart — the core replay mechanism.
3. Couple move-position persistence to durable flush — make the persisted position crash-safe.
4. Add failing test for nested subquery replay dropping control messages — the nested reproduction.
5. Replay control-message moves with authoritative storage offsets — position every item by its storage offset.
6. Chunk move replay by full offset, not by transaction — restore per-move granularity for op-distinguished moves.
7. Test replay of a value toggled multiple times within one transaction — lock in that per-op replay is safe (never a move-in and move-out for the same value in one payload).
8. Drop redundant materializer replay-skip guard — rely on Skip already-processed transactions on restart #4668 for restart de-duplication.

Testing

• :oracle_restore_bug_1 and :oracle_restore_optimized_refetch (single-level) — deterministic across repeated runs.
• :oracle_restore_nested_subquery — was ~4% divergent, now deterministic (80/80).
• Deterministic materializer unit coverage for control-message replay and for multi-toggle-in-one-transaction ordering.
• materializer + consumer + subquery suites — 0 failures.

---

🤖 Generated with Claude Code

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 60.02%. Comparing base (6aef3c7) to head (2a1a1d1).
✅ All tests successful. No failed tests found.

Additional details and impacted files

@@                      Coverage Diff                       @@
##           rob/subquery-restore-bug-6    #4681      +/-   ##
==============================================================
- Coverage                       60.05%   60.02%   -0.03%     
==============================================================
  Files                             397      397              
  Lines                           43766    43766              
  Branches                        12590    12588       -2     
==============================================================
- Hits                            26282    26269      -13     
- Misses                          17403    17416      +13     
  Partials                           81       81              

Flag	Coverage Δ
packages/agents	72.64% <ø> (ø)
packages/agents-mcp	77.70% <ø> (ø)
packages/agents-mobile	80.67% <ø> (ø)
packages/agents-runtime	83.73% <ø> (ø)
packages/agents-server	75.47% <ø> (-0.18%)	⬇️
packages/agents-server-ui	8.32% <ø> (ø)
packages/electric-ax	51.06% <ø> (ø)
packages/experimental	87.73% <ø> (ø)
packages/react-hooks	86.48% <ø> (ø)
packages/start	82.83% <ø> (ø)
packages/typescript-client	91.89% <ø> (ø)
packages/y-electric	56.05% <ø> (ø)
typescript	60.02% <ø> (-0.03%)	⬇️
unit-tests	60.02% <ø> (-0.03%)	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[claude]

Claude Code Review

Summary

Iteration 8 is a rebase-only update: the PR was replayed onto the current base (rob/subquery-restore-bug-6 @ 6aef3c78) with no new production changes. The top commit 2a1a1d1ae ("Drop redundant materializer replay-skip guard") is byte-for-byte identical to the ce2d63dd4 reviewed in iteration 7, and the full diff against the base is unchanged (+1525/-68 across the same 14 files). I re-verified the core move-replay + position-persistence logic independently and it remains sound. The PR stays ready to merge.

What's Working Well

• Durable-flush coupling is correct by construction. maybe_stage_move_positions/1 stages {latest_offset, source_lsn} only once the move pipeline is fully drained, and commit_flushed_move_positions/2 advances move_positions only for staged entries whose threshold <= flushed_offset. Because latest_offset is monotonic, the per-handle staged lists are ascending, so the Enum.split_while/2 partition into committed/remaining is well-defined and the persisted position can never run ahead of durable storage.

• Restart replay is closed under all first-move cases. subscribe_to_materializers/1 baselines a fresh dependency with Map.put_new(move_positions, handle, applied_offset) and persists it, so even a restart before the first move still has a from_lsn to replay from. Existing positions are preserved (no-op put_new).

• terminate/2 degrades safely. commit_all_move_positions/1 runs after terminate_writer/1 (writer flushed, so all staged splices durable), matches a bare map since :writer has been popped, and wraps the persist in a rescue that only logs — a persistence failure there costs at most extra replay on restart rather than masking the shutdown reason.

• Guard removal remains provably safe. As established in iteration 7, the deleted new_changes skip clause is unreachable: the consumer skips already-applied transactions before apply_event, the materializer seeds applied_offset from the consumer's latest_offset, forwarded ranges are strictly beyond it, and new_changes/3 has a single production caller. Dropping the silent skip lets a genuine invariant violation surface loudly rather than be masked — consistent with the existing straddling-transaction raise.

• Offset-authoritative replay (get_log_stream_with_offsets/3 + chunk_by_offset/1) correctly delimits dependency-driven moves that share a tx_offset by chunking on the full {tx_offset, op_offset}, which is the space from_lsn/move_positions live in.

Issues Found

Critical (Must Fix)

None.

Important (Should Fix)

None.

Suggestions (Nice to Have)

• Carried forward from iteration 7 (not blocking): the no-re-notify-at/below-applied_offset invariant that the guard removal now relies on is covered only indirectly via the optimized_refetch oracle repro. A small deterministic consumer-level unit assertion would make a future regression fail fast in the unit suite rather than only in the integration repro.

Issue Conformance

• No linked issue (per convention, a warning) — unchanged; the series tracks the #4648 restart-oracle triage and is stacked on #4666, building on merged #4668.
• Changesets present for the two behaviour changes (restore-shutdown-shape-removal.md, subquery-move-replay-on-restart.md). The rebase introduced no new publishable-surface changes requiring a changeset.

Previous Review Status

• ✅ All items from iterations 1–7 remain resolved; nothing regressed across the rebase.
• ✅ The full diff is identical to what iteration 7 approved — this iteration adds no production code, only a new base.
• ⏳ The one open non-blocking suggestion (direct consumer-level dedup unit test) is carried forward.

Rebase-clean, no new changes, no regressions. The PR remains ready to merge.

---

Review iteration: 8 | 2026-07-16