rob/restore-subqueries-bug-5 vs main
Executive summary
Branch purpose: persist subquery move positions so that after a restart the outer consumer can resume where it left off and replay any dependency-shape moves it missed — instead of the shape being invalidated and every client getting a 409 must-refetch.
The headline: the branch's core mechanism — replaying missed moves during Materializer.subscribe — runs synchronously and unboundedly inside the materializer's handle_call, and it triggers after virtually every restart, not just when moves were missed. Findings 1, 3, and 4 are three facets of that one design decision, and finding 1 in particular can reintroduce the exact 409 bug this branch fixes. A second cluster (findings 2 and 10) stems from duplicating the log-streaming stack instead of extending it: the copy already diverges from the original in error handling.
Root-cause map
flowchart TD
RC1["Design: synchronous unbounded
replay inside handle_call"]:::root
RC2["Design: _with_offsets duplicates
the log-streaming stack"]:::root
RC3["Shutdown / error-path
hardening gaps"]:::root
F1["F1 · subscribe 5s timeout
→ shape deleted, 409s"]:::crit
F3["F3 · Enum.chunk_by materializes
whole log → OOM risk"]:::high
F4["F4 · full rescan every restart
→ replication stalls"]:::high
F2["F2 · silent truncation
→ silent divergence"]:::crit
F10["F10 · ~8 parallel copies
across 4 modules"]:::clean
F5["F5 · exit-catch misses :killed
→ shape deleted"]:::high
F6["F6 · positions committed
after swallowed flush error"]:::high
RC1 --> F1
RC1 --> F3
RC1 --> F4
RC2 --> F2
RC2 --> F10
RC3 --> F5
RC3 --> F6
classDef root fill:#1b2540,stroke:#5ea2f7,stroke-width:1.5px,color:#dce6f5
classDef crit fill:#2b1520,stroke:#f87171,stroke-width:1.5px,color:#fecaca
classDef high fill:#2b2312,stroke:#fbbf24,stroke-width:1.5px,color:#fde68a
classDef clean fill:#0f2a20,stroke:#34d399,stroke-width:1.5px,color:#a7f3d0
Standalone findings not shown: F7 (compaction), F8 (fragment path), F9 (Storage behaviour). Fixing F10 (single streaming primitive) eliminates F2 as a side effect.
All findings at a glance
Default 5s timeout around unbounded replay — timeout deletes the shape
def subscribe(pid, from_lsn) when is_pid(pid),
do: GenServer.call(pid, {:subscribe, from_lsn})No timeout argument, so GenServer.call's 5-second default applies. Before this branch that was fine — subscribe just returned the in-memory view. Now handle_call({:subscribe, from_lsn}) runs maybe_replay_moves, which re-reads the dependency shape's entire snapshot plus log from LogOffset.before_all() before replying. Notably, the sibling calls new_changes (line 45) and wait_until_ready (line 49) both explicitly pass :infinity — subscribe is the odd one out.
consumer.ex:1406 exits with {:timeout, {GenServer, :call, ...}} → nothing catches it, so the consumer crashes during finish_initialization → terminate/2 hands the reason to ShapeCleaner.handle_writer_termination, whose exemption guard checks elem(reason, 0) == :shutdown — a :timeout tuple doesn't match, so it takes the catch-all branch and deletes the shape's data from disk. Every client of the outer shape gets a 409 must-refetch. That is precisely the failure mode this branch exists to eliminate, now triggered by shape size instead of shutdown.
:infinity (consistent with the siblings) — though the real fix is bounding the replay itself (F3, F4).
Offset-yielding log reader silently tolerates a truncated log file
stream_entries_with_offsets reads a whole chunk range in one pread and discards the parser's leftover:
with {:ok, data} <- :file.pread(file, start_position, end_position - start_position) do
{entries, _} = extract_entries_from_binary(data, exclusive_min_offset, nil)
entriesIf the log file is shorter than the chunk index claims (crash mid-write before trim, torn chunk index), :file.pread doesn't error — it returns {:ok, partial_data}. extract_entries_from_binary hits its catch-all clause and returns the entries it could parse plus an unconsumed remainder, which the {entries, _} pattern throws away. The truncated trailing entry is silently dropped.
The existing stream_jsons path handles this exact condition by raising "unexpected end of file" — deliberately fail-loud, because a shape reading a corrupt log must be invalidated, not served.
end_position - start_position) and raise the same error stream_jsons raises.
Replay eagerly materializes the entire decoded dependency log in memory
apply_replay_stream builds a lazy stream over the whole range [before_all, applied_offset], then immediately defeats the laziness:
defp chunk_by_offset(items) do
items
|> Enum.chunk_by(fn {offset, _txids, _change} -> {offset.tx_offset, offset.op_offset} end)
|> Enum.map(...)Enum.chunk_by is eager — every decoded JSON change struct for the entire dependency log is realized as one in-memory list inside the materializer process before the reduce even starts.
Stream.chunk_by + Stream.map (or Stream.transform) keeps the pipeline lazy with the same grouping semantics.
Full-history rescan fires after virtually every restart and blocks replication
replay_moves runs whenever from_lsn < applied_offset. But those two cursors advance at different rates: applied_offset moves on every dependency-shape write, while the consumer's persisted move position only advances when a move event is actually emitted. So after any restart where the dependency shape saw ordinary writes (i.e. almost always), the gap exists and the materializer performs a full log scan from before_all — even when zero moves were missed and there is nothing to replay.
Compounding it: the scan runs inside the materializer's handle_call, and the scans for multiple outer shapes queue serially in its mailbox. While they run, the dependency consumer's own Materializer.new_changes call (made with :infinity timeout) is blocked.
applied_offset alongside) so the no-missed-moves case is detectable and skipped entirely; (b) move the replay off the materializer's call path — reply-later with the scan in a spawned task, or replay from a checkpoint rather than before_all.
Exit-reason catch misses :killed and {:shutdown, term} shapes
The new catch in notify_materializer_of_new_changes handles:
:exit, {:noproc, _} -> :ok
:exit, :noproc -> :ok
:exit, {:normal, _} -> :ok
:exit, {:shutdown, _} -> :okTwo real exit shapes fall through:
{:killed, {GenServer, :call, ...}} — the materializer was brutally killed after exceeding its supervisor shutdown timeout (made more likely by the long replays in F1/F3/F4); and {{:shutdown, term}, {GenServer, :call, ...}} — the materializer stopped with a {:shutdown, reason} tuple. In that second case the outer element is the tuple {:shutdown, term}, which the pattern {:shutdown, _} does not match.
ShapeCleaner.handle_writer_termination's elem(reason, 0) == :shutdown guard doesn't cover them, so the shape is removed from disk → 409 after restart — again, the symptom the restore-shutdown-shape-removal changeset targets.
:exit, {:killed, _} and :exit, {{:shutdown, _}, _} clauses — and consider aligning the ShapeCleaner guard with the same set.
Move positions committed even when the writer flush failed
terminate/2 calls terminate_writer/1 then commit_all_move_positions/1 unconditionally. But terminate_writer rescues File.Error from Storage.terminate and returns normally — so if the flush fails (canonical case: ENOSPC during shutdown), the splice rows for a staged move never reach disk, yet the move position covering them still gets persisted (the small metadata write can succeed once space frees, or via its tmp-rename path).
fetch_move_positions reports the move as applied. Materializer.subscribe therefore skips replaying it, and the outer shape serves data permanently missing the moved-in rows, silently. This violates the branch's own invariant that the persisted position never runs ahead of durable storage.
commit_all_move_positions when terminate_writer rescued an error. Why Plausible, not Confirmed: the verifier couldn't fully establish that a File.Error escapes the specific flush path while leaving the metadata write able to succeed — a stricter variant of this finding was refuted on exactly that interleaving — but the rescue-then-commit ordering is real and the guard is cheap.
Compaction invalidates point-in-time replay reconstruction
replay_moves reconstructs "state as of from_lsn" by replaying the stored log from before_all and cutting at from_lsn. That's only sound if the log below from_lsn is still the log the outer consumer originally consumed. Compaction breaks that assumption: updates are collapsed forward and deletes dropped, so an entry that originally sat before from_lsn may now be merged into an entry after it.
from_lsn inside the compacted region. The reconstructed value_counts (seed view) differ from the true historical link values, and replayed move-in/move-out events are computed against the wrong baseline — spurious or missing moves, silent divergence.
from_lsn falls inside a compacted region (compare against the compaction boundary) and fall back to full re-initialization of the outer shape rather than replaying.
Fragment path never advances applied_offset — duplicate replays after restart
The range-based new_changes clause was updated to advance applied_offset (state = %{state | applied_offset: range_end} at line 502), which is what tags flushed move events with their :lsn (line 716). The list-based clause — handle_call({:new_changes, changes, xid, commit?}, ...) when is_list(changes) — used by the consumer's transaction-fragment write path (consumer.ex:785, flushed via the empty-list commit call at consumer.ex:856) was not updated and never touches applied_offset.
:lsn; the outer consumer's persisted move position never advances past it (LogOffset.max keeps the old baseline). On every subsequent restart those already-applied, already-durable moves are replayed again — duplicate move-in refetch inserts and move-out deletes appended to the outer shape log and delivered to clients after each restart.
applied_offset in the list clause on commit — the commit call carries the offset context, or thread it through explicitly.
New required Storage callbacks break third-party implementations
State.new now unconditionally runs {:ok, move_positions} = Storage.fetch_move_positions(storage) for every shape, and set_move_positions! for dep-having shapes — dispatching straight to the implementation module with no function_exported? guard, and the behaviour declares both as required (non-optional) callbacks.
Electric.ShapeCache.Storage is a documented public extension point (configurable via the storage option in Electric.Application.configuration/1). Any downstream embedder with a custom storage crashes every consumer at startup with UndefinedFunctionError after upgrading — the whole sync service serves nothing until they implement the new callbacks.
@optional_callbacks with a function_exported? fallback returning {:ok, %{}} / no-op — or, if a hard break is intended, make it a major-version changeset and document the migration.
The _with_offsets stack duplicates the entire log-streaming path
The offset-yielding variants are near-verbatim copies at every layer:
· read_range_from_ets_cache_with_offsets/5 (lines 1243–1263) is line-for-line read_range_from_ets_cache/5 (1212–1226), differing only in cons-ing {offset, item} vs item
· stream_from_disk_with_offsets duplicates stream_from_disk
· InMemoryStorage.get_offset_indexed_stream_with_offsets and get_log_stream_with_offsets duplicate their base versions
· LogFile grows three parallel functions
That's ~8 parallel copies across 4 modules.
Stream.map(fn {_offset, json} -> json end) — one code path, offsets dropped at the edge. Fixing this eliminates F2 as a side effect.
Review process & provenance
How these findings were produced (30 agents, 28 candidates, 1 refuted)
Scope: git diff main...HEAD on branch rob/restore-subqueries-bug-5 — 15 changed files, including two changesets (restore-shutdown-shape-removal, subquery-move-replay-on-restart), the materializer, consumer, consumer state, storage behaviour, both storage implementations, the log file reader, and tests.
Find: 3 independent correctness finder agents (different angles) plus 1 cleanup finder produced 28 candidate findings.
Verify: every distinct (file, line) location got an independent verifier agent with full code access — 24 verifier runs. 27 candidates survived; 1 was refuted: the claim that terminate/2's unconditional commit violates the durability invariant in the direct IO.binwrite path — refuted because a failing IO.binwrite raises ErlangError, not File.Error, so it is not swallowed by the rescue (the surviving, narrower version is F6).
Verdicts: Confirmed = the verifier traced the complete failure path in the code. Plausible = the mechanism is real but at least one link in the end-to-end scenario couldn't be fully established.
Synthesis: the 27 surviving findings were deduplicated by root cause and ranked; the top 10 are reported here.