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

[robacourt]

Summary

Fixes optimized streaming subquery shapes diverging from Postgres after a graceful server restart — bug 5 from the restart-aware oracle property-test triage (#4648 / bugs.md). A shape like issues WHERE project_id IN (SELECT id FROM projects WHERE active = true) could permanently lose (or retain stale) rows after a restart.

Problem

A dependency move (move-in / move-out) driven by the materializer is asynchronous: a move-in fires a PG query and only writes its rows to the shape log when that query completes and splices. If a graceful restart fires while a move is still in flight — or queued behind one, or sitting unprocessed in the mailbox, or not yet even emitted by the materializer — the move is lost. On restart the materializer restores its view (which reflects the move) but the outer shape's storage does not, and nothing re-drives the move, so the shape diverges from Postgres.

Solution (per commit)

• Add failing test — a deterministic test_against_oracle restart reproduction (:oracle_restore_optimized_refetch).
• Prevent shape removal and request crashes during restart — during a restart a source consumer's inline call to its dying materializer could exit with :noproc, crash the consumer, and cascade into removing the shape from disk (→ 409). The consumer now absorbs that exit and lets the monitored :DOWN drive a clean stop.
• Replay missed moves after restart — the materializer tags each emitted move with its source LSN and, on subscribe, replays the moves a consumer is missing from its persisted per-dependency position; the consumer catches up. Because the consumer simply re-applies whatever the materializer replays, every restart-timing class (in-flight, queued, mailbox, not-yet-emitted) is covered by construction.
• Couple move-position persistence to durable flush — the per-dependency moves-position is staged when the move pipeline drains and only committed/persisted once the writer confirms the flush has passed the move's splice (and at terminate, after the writer is flushed), so the persisted position can never run ahead of durable storage across a restart. Also raises the test's long_poll_timeout off an artificially short value that trips a separate post-restart long-poll readiness race (bug 3).

Test Plan

• oracle_restore_test.exs :oracle_restore_optimized_refetch — view-mismatch eliminated; deterministic across 100+ runs on healthy Postgres.
• --only oracle_restore_bug_1 still passes.
• test/electric/shapes/ + test/electric/replication/ — 0 failures.

---

🤖 Generated with Claude Code

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 60.03%. Comparing base (b8be884) to head (fdf08f4).
✅ All tests successful. No failed tests found.

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #4681      +/-   ##
==========================================
- Coverage   60.15%   60.03%   -0.13%     
==========================================
  Files         410      395      -15     
  Lines       44377    43747     -630     
  Branches    12580    12580              
==========================================
- Hits        26697    26264     -433     
+ Misses      17602    17405     -197     
  Partials       78       78              

Flag	Coverage Δ
electric-telemetry	?
elixir	?
packages/agents	72.64% <ø> (ø)
packages/agents-mcp	77.70% <ø> (ø)
packages/agents-mobile	80.67% <ø> (ø)
packages/agents-runtime	83.73% <ø> (ø)
packages/agents-server	75.65% <ø> (+0.20%)	⬆️
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.83% <ø> (ø)
packages/y-electric	56.05% <ø> (ø)
typescript	60.03% <ø> (+0.03%)	⬆️
unit-tests	60.03% <ø> (-0.13%)	⬇️

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

This PR fixes optimized streaming subquery shapes diverging from Postgres after a graceful restart (bug 5): it persists a per-dependency "moves-applied-up-to" source-LSN position, replays missed materializer moves on subscribe, and couples position persistence to durable writer flushes. The design is careful and well-tested — the staged→committed durability coupling is the right shape for the "never persist a position ahead of storage" invariant. Overall a solid, high-quality change; the notes below are mostly edge-cases and polish.

What's Working Well

• Durability invariant is enforced structurally. Positions move pending → staged (with a flush threshold) → committed and are only persisted once the writer confirms a flush at/after the move's splice offset (commit_flushed_move_positions/2), and unconditionally at terminate/2 after terminate_writer/1 flushes. This makes "persisted position never runs ahead of durable storage" hold by construction rather than by timing.
• Replay register-before-replay ordering in handle_call({:subscribe, from_lsn}, ...) is correct: the subscriber is added before replay, and because the call runs to completion inside the GenServer, replayed moves land in the subscriber mailbox strictly before any live move.
• The new_changes already-applied guard (when is_log_offset_lte(range_end, applied_offset)) is a clean fix for the slot-replay "Key already exists" crash, with a clear comment tying it to the persistent-slot re-delivery scenario.
• Changeset files present for both behavioural changes; storage round-trip, materializer replay, and consumer persistence/:noproc-survival are all covered by focused tests. :move_positions is correctly added to @stored_keys in PureFileStorage (I verified — without it, write_metadata!/read_metadata! would silently no-op via the catch-all clauses, which would have broken the whole feature on the production backend).

Issues Found

Critical (Must Fix)

None found.

Important (Should Fix)

1. Replay drops move-in/move-out control messages for nested subquery dependencies

File: packages/sync-service/lib/electric/shapes/consumer/materializer.ex:403-406

decode_offset/1 only reconstructs an offset from %{"lsn" => ..., "op_position" => ...} headers and otherwise falls back to LogOffset.before_all(). Regular insert/update/delete log items carry those headers (log_items.ex), but move-in/move-out control messages do not — they are Jason.encode!'d straight from %{headers: %{event:, patterns:, txids:}} in MoveBroadcast.make/7 via append_control_message!, with no lsn/op_position.

The existing decode_json_stream/1 handles these control messages fine because it never needs their offset. But the new replay path (apply_replay_stream/2 → chunk_by_txn/1) does depend on the offset: a control message decoded to before_all gets grouped into the first chunk and treated as <= from_lsn (emit? = is_log_offset_lt(from_lsn, before_all) is always false), so it is applied to build state but never re-emitted.

This only matters when a materializer's source (dependency) shape log itself contains move events — i.e. a nested subquery (a subquery shape that is a dependency of another subquery shape). Single-level shapes (the case this PR targets and tests) are unaffected, since the inner shape is a plain shape whose log has only regular changes. But the fact that decode_json_stream/1 bothers to decode "move-in"/"move-out" suggests the materializer can read such logs.

Please confirm whether nested optimized-subquery dependencies are reachable. If they are, this replay path silently drops their missed moves on restart — reintroducing the exact divergence class this PR fixes, for the nested case. If nesting is not supported, a guard/assert (or a comment noting the assumption) would make the limitation explicit.

Suggestions (Nice to Have)

2. Decoder duplication. decode_json_stream_with_offsets/1 + decode_change/1 (materializer.ex:391-452) re-implement the change-decoding already in decode_json_stream/1 (601-655), nearly line-for-line for the insert/update/delete/move cases. Consider a single decoder that optionally threads the offset/txids so the two paths can't drift (e.g. a move-event added to one decoder but not the other).

3. Broad rescue in commit_all_move_positions/1 (consumer.ex:1107-1111): rescue _ -> :ok swallows all errors from the terminate-time persist. A File.Error during shape removal is expected, but a genuine persistence failure here would be invisible and could leave the position stale after restart. Consider narrowing to the expected exception(s) or at least a Logger.warning so real failures are observable.

4. Duplicated comment block on maybe_stage_move_positions/1 (consumer.ex:1019-1030): there are two adjacent paragraphs both explaining "once the move pipeline is fully drained…". Looks like a merge artifact — worth collapsing to one.

5. Write amplification (minor). set_move_positions! rewrites the entire positions map (term-encode + temp-file + rename) on every committed flush that advances a position, plus once per startup in subscribe_to_materializers. Fine at realistic move frequencies; just noting it for very move-heavy subquery shapes.

Issue Conformance

• No linked issue. Per project convention this is a warning — the PR references the #4648 / bugs.md restart-oracle triage and "bug 5" but no tracking issue is linked. Recommend linking one so the bug-triage series (bugs 3/5/6 are all touched here) stays traceable.
• PR description is excellent — per-commit rationale, restart-timing taxonomy, and an explicit note that the "bug 6" removal cascade is only partially hardened. Scope is coherent and matches the diff.
• The oracle_restore_optimized_refetch test's own comment notes it currently fails during batch_1 replay before batch_2 is reached; the test is structured to still exercise a post-restart batch. Worth a follow-up to assert the post-restart move path directly once the cascade hardening lands.

---

Review iteration: 1 | 2026-07-02