diff --git a/packages/sync-service/bugs.md b/packages/sync-service/bugs.md new file mode 100644 index 0000000000..59f481fa9a --- /dev/null +++ b/packages/sync-service/bugs.md @@ -0,0 +1,596 @@ +# Bugs surfaced by the restart-aware oracle property test + +These were uncovered while wiring `RESTART_SERVER_EVERY` into +`test/integration/oracle_property_test.exs`. The harness now restarts the +`Electric.StackSupervisor` mid-test to exercise restore-from-file. Each bug +below is a real Electric issue that prevents the test from passing in restart +mode unless mitigated. The harness no longer mitigates them — the test fails +in their presence so they get noticed. + +## Bug 1: Subquery shape materializer state is not restored from disk — FIXED + +**Symptom** + +After a stack restart with a persistent replication slot, shapes whose +predicates contain subqueries (`level_3_id IN (SELECT id FROM level_3 …)`, +`NOT IN (...)`, etc.) diverged from the oracle. Two failure modes were +seen: a `409 (must-refetch)` on `optimized: true` shapes, and missing rows +after a refetch. + +**Root cause** + +Two compounding issues: + +1. **Materializer only replayed the first chunk of source-shape history on + startup.** `Storage.get_log_stream/3` returns at most one chunk per call + (a snapshot chunk, or one main-log chunk). The materializer's + `handle_continue({:read_stream, …})` called it once and treated the + stream as the full history, silently dropping every subsequent snapshot + chunk and every persisted log entry. On restart the source shape's log + contained the post-snapshot updates that drove dependent + move-in/move-out events, so the materializer's `value_counts` was wrong + from the moment it came up. + +2. **Dependent (outer) consumers were not restarted after a stack restart.** + The router only delivers events to a consumer when its own `root_table` + changes, so a shape whose movement is driven entirely by a dependency + stayed dormant after a restart. Without its consumer running, the + materializer for its dependency was never started, the SubqueryIndex + was never seeded, and the dependent shape's on-disk view was never + updated when the dependency changed. + +**Fix** + +- `lib/electric/shapes/consumer/materializer.ex#handle_continue({:read_stream, …})`: + iterate `Storage.get_log_stream/3` and `Storage.get_chunk_end_log_offset/2` + until `state.subscribed_offset` is reached, instead of reading a single + chunk. +- `lib/electric/shape_cache.ex#handle_continue(:wait_for_restore, …)`: + after marking the log collector ready, eagerly call + `restore_shape_and_dependencies/3` for every shape with + `shape_dependencies != []` whose consumer isn't already running, so + materializer subscriptions are re-established. + +**Regression test** + +`test/integration/oracle_restore_test.exs#test "bug 1: subquery shape +diverges from oracle after server restart"` — single-shape, two-batch, +deterministic reproduction that fails on `main` and passes after the fix. + +## Bug 4: Materializer re-reads main-log entries on startup recovery — FIXED + +**Symptom** + +Same surface as Bug 1 (a 409 must-refetch on an `optimized: true` subquery +shape after a server restart), but reproducible only when the source +shape's persisted main log spans more than one chunk. The heavy property +run with `RESTART_SERVER_EVERY=7 RESTART_CLIENT_EVERY=11 SHAPE_COUNT=800 +…` reliably tripped this on shape definitions like +`level_3_id IN (SELECT id FROM level_3 WHERE active = true)`. + +**Root cause** + +The Bug 1 fix added an iteration loop in `Materializer.read_history_up_to_subscribed/2` +that calls `Storage.get_log_stream/3` repeatedly, advancing through chunk +boundaries via `Storage.get_chunk_end_log_offset/2`. + +`get_log_stream/3` returns one snapshot chunk per call, so iteration is +correct for the snapshot. But for the **main log** it returns the *entire* +requested range (`[min_offset, subscribed_offset]`) in a single call. +When the main log spans multiple chunks, the loop's next-iteration offset +was the end of the *first* main-log chunk, so the next call streamed the +remainder of the main log — entries the previous call had already +applied. Re-applying inserts hits the materializer's "Key already exists" +guard, which crashes the materializer; the dependent shape's consumer +goes down with it and the server returns 409 must-refetch on the next +poll. + +**Fix** + +In `lib/electric/shapes/consumer/materializer.ex#read_history_up_to_subscribed/2`, +stop iterating as soon as the read steps into the main log. Two new +short-circuits cover this: + +- if `state.offset` is already a real or last-virtual offset, the call + just made was `stream_main_log` and is complete; +- if the next chunk boundary is a real offset, the call we just made + exhausted the snapshot and entered `stream_main_log`, so iterating + further would re-read. + +**Regression test** + +`test/integration/oracle_restore_test.exs#test "bug 4: subquery shape +returns 409 after restart with many persisted log entries"` — +deterministic single-shape reproduction with `chunk_size: 200` to force +the source shape's main log to span multiple chunks. Fails on the broken +iteration; passes with the fix. + +## Bug 2: Snapshot+log replay can produce duplicate / orphan operations after restart — FIXED + +**Symptom** + +A fresh client polling a shape after server restart receives a sequence of +operations where the same row appears as two inserts, or where an `update` / +`delete` arrives for a row the client never saw inserted. The +`ShapeChecker.apply_message/2` invariants flunk with messages like: + +``` +shape=shape_4: insert for row that already exists: {"l4-20"} +shape=shape_8: update for row that does not exist: {"l4-18"} +``` + +This appears to happen when the snapshot streamed to the new client overlaps +with the log entries that follow, instead of the snapshot ending exactly at +the offset where the log resumes. + +**Reproduce** + +Same command as Bug 1; many seeds without subqueries also show this +intermittently when batches deliver row movements that straddle the snapshot +boundary. + +**Where to look** + +- `lib/electric/shape_cache/pure_file_storage.ex` — the boundary between the + on-disk snapshot and the persisted log file. After restore, both are + streamed to the client; the snapshot's last offset must be strictly before + the log's first offset. +- `lib/electric/shapes/api.ex#do_serve_shape_log/1` and the streaming + pipeline — confirm that the catch-up replay starts at exactly + `last_persisted_txn_offset + 1` and does not include any rows already in + the snapshot. + +## Bug 3: Long-poll completes with HTTP 400 "offset out of bounds" after multiple restarts + +**Symptom** + +After the second `StackSupervisor` restart in a session, a fresh client poll +sometimes receives: + +``` +%Electric.Client.Error{ + message: %{"errors" => %{"offset" => ["out of bounds for this shape"]}, + "message" => "Invalid request"}, + resp: %{status: 400, …} +} +``` + +This happens on simple (non-subquery) shapes when the long-poll timeout +expires before the post-restart replication client has caught up enough to +deliver new transactions. Bumping `LONG_POLL_TIMEOUT` from the test default +of 100ms to 2000ms hides it; the underlying issue is that after a restart +the server's view of "last available offset" briefly trails Postgres' actual +state, and the long-poll's out-of-bounds-recovery loop times out before the +gap closes. + +**Reproduce** + +```sh +CHECK_TIMEOUT=60000 SHAPE_COUNT=10 MUTATIONS_PER_TXN=10 TXNS_PER_BATCH=10 \ + BATCH_COUNT=15 RESTART_SERVER_EVERY=7 LONG_POLL_TIMEOUT=100 \ + SKIP_REPATCH_PREWARM=true \ + mix test --seed 1 --only oracle test/integration/oracle_property_test.exs +``` + +**Where to look** + +- `lib/electric/shapes/api.ex#determine_log_chunk_offset/1` and the long-poll + branch around line 880 (`@offset_out_of_bounds`). +- `lib/electric/connection/manager.ex#handle_continue(:start_streaming, …)` + vs. the `Electric.StatusMonitor.wait_until_active/2` readiness signal — + there's a window where the stack reports `:active` but the replication + stream hasn't yet forwarded transactions that Postgres committed during + the restart, and a poll arriving in that window can be told the offset is + out of bounds. +- Possibly Electric.LsnTracker — the new replication client may be + reporting a stale `last_processed_lsn` until the first batch streams. + +## Bug 5: Post-restart move-in events lost when source-shape main log spans multiple chunks — FIXED + +**Symptom** + +After a server restart, the next replication-driven mutation that would +move rows in or out of a subquery shape's view is silently dropped: +the materialized view stays at its post-restart-restored state, while +the oracle (PG) correctly reflects the post-batch state. Manifests as a +"View mismatch" in the oracle harness, with the materialized view +missing rows the oracle has. + +Reproduces only with **multiple subquery shapes** AND a source-shape +main log that spans **more than one chunk** (forced via +`@tag chunk_size: 200`). Single-shape variants of the same scenario +(`bug 1` and `bug 4` regression tests) pass, so the bug is in an +interaction between concurrent materializer recoveries and post-restart +event delivery — possibly a stale `last_persisted_offset` that causes +the source consumer to ignore the incoming mutation, or a missed +materializer-subscription handshake that means the dependent consumer +isn't on the materializer's subscriber list when the move-in event +fires. + +**Reproduce** + +```sh +CHECK_TIMEOUT=10000 SKIP_REPATCH_PREWARM=true \ + mix test --seed 1 --only oracle_restore_bug_5 \ + test/integration/oracle_restore_test.exs +``` + +Two shapes (`level_3 WHERE active=true` and `level_3 WHERE active=false`), +200 toggles in batch_1, server restart, single deactivate in batch_2. +After batch_2 the dependent `shape_active_false` view is missing the +level_4 rows whose level_3 parent just transitioned to `active=false`. + +**Root cause (narrowed by trace logging)** + +The outer shape's consumer, when re-initialized after a stack restart, +seeds `state.views` from the **current materializer view** via +`EventHandlerBuilder.build/2`: + +```elixir +view = Materializer.get_link_values(materializer_opts) +``` + +But the outer shape's **on-disk storage** reflects the state at the +*pre-restart* shutdown LSN, while the materializer is rebuilt from disk +and then advanced by any events the source consumer processes during +the eager-restart window — including events from the replication +slot's catch-up replay and from the next test mutation (`batch_2`'s +deactivate). The two views diverge. + +Concretely, in the failing test: + +| Stage | Outer storage view (level_3_id values) | Materializer view | +|---|---|---| +| End of batch_1 | `{l3-3, l3-5}` (correct) | `{l3-3, l3-5}` (correct) | +| Server restarts; outer consumer re-initialized | `{l3-3, l3-5}` (still on disk) | replays history, momentarily `{l3-3, l3-5}` | +| `batch_2` deactivate(l3-2) lands at the materializer **before** outer consumer's `EventHandlerBuilder.build` runs | `{l3-3, l3-5}` | now `{l3-2, l3-3, l3-5}` | +| Outer's `state.views` is seeded **from the materializer** | `{l3-3, l3-5}` | view = `{l3-2, l3-3, l3-5}` | + +Then the materializer's `move_in: [{"l3-2", _}]` event arrives at the +outer consumer. `MoveQueue.enqueue/4` runs `redundant?/2`, which checks: + +```elixir +defp redundant?(%{kind: :move_in, move_value: {value, _}}, base_view) do + MapSet.member?(base_view, value) +end +``` + +`l3-2` is in the seeded view, so the move-in is treated as redundant +and **dropped**. The outer storage never gets the `level_4` rows for +`level_3_id = l3-2` — those are the rows missing from the materialized +view in the test failure. + +**Why single-shape doesn't expose it** + +With one shape there's only one materializer, only one outer consumer, +and the timing rarely interleaves the outer consumer's +`EventHandlerBuilder.build` between the materializer's +`new_changes(...)` for batch_2 and the materializer's flush of the +move event back to the outer consumer. The race is exposed by +multi-shape eager-start: my `eagerly_start_subquery_shape_consumers/1` +restores shapes sequentially and `initialize_shape/3` is async, so +by the time outer-shape #2's consumer init runs `EventHandlerBuilder.build`, +its dependency materializer has already absorbed batch_2's update. + +**Partial fix (closes one race; doesn't close the underlying offset +mismatch)** + +Eager-start the outer subquery shape consumers *before* +`ShapeLogCollector.mark_as_ready/1` opens the event-dispatch gate. +Concretely, `ShapeCache.handle_continue(:wait_for_restore)` now: + +1. Calls `eagerly_start_subquery_shape_consumers/1`, which iterates + shapes with `shape_dependencies != []` and runs + `restore_shape_and_dependencies/3` for each, then blocks on + `Consumer.await_snapshot_start/2` so each outer consumer's + `EventHandlerBuilder.build/2` has run and `state.views` is seeded + from the materializer. +2. Only after all subquery consumers are fully initialized does it call + `ShapeLogCollector.mark_as_ready/1`. + +This stops the materializer view from drifting *forward* of the +outer consumer's view between consumer init and the first live +event. The original symptom — a move-in for a value already in the +seeded view being dropped as redundant — no longer happens for THAT +race. + +What this doesn't close: the materializer's view is at the inner +shape's `last_persisted_offset`, and the outer shape's storage is at +its own `last_persisted_offset`. Those two writers flush +independently and aren't equal across restart. So `state.views` (set +from materializer) and outer storage can still be at different +LSNs. The visible consequence is now Bug 2 — duplicate INSERTs in +the post-restart log for rows the materializer's "ahead" view says +have already been moved in. Properly closing Bug 5 requires one of +the three approaches listed in Bug 2's analysis: persist the dep +view per outer shape, derive it from outer storage, or replay +materializer events from the outer's persisted offset. + +**Regression test** + +`test/integration/oracle_restore_test.exs#test "bug 5: multiple +subquery shapes diverge after restart with long persisted log"` +reproduces this deterministically with two shapes +(`level_3 WHERE active=true` and `level_3 WHERE active=false`), +200 toggles in batch_1, server restart, and a single +`UPDATE level_3 SET active = false WHERE id = 'l3-2'` in batch_2. +The `@tag chunk_size: 200` is required to make the source shape's +main log span more than one chunk so the timing race exposes itself +reliably. + +**Regression test** + +`test/integration/oracle_restore_test.exs#test "bug 5: multiple +subquery shapes diverge after restart with long persisted log"` — +deterministic two-shape reproduction. Fails today with the listed +mutation pattern; passes for the single-shape variants. + +## Bug 6: Mid-restart shape cleanup leaves shape removed from on-disk metadata — FIXED + +**Symptom** + +After a `RESTART_SERVER_EVERY` restart, the new clients sometimes get a +`409 (must-refetch)` from a shape that was healthy before the restart. +With `optimized: true` shapes the test flunks immediately. Reliably +reproduces at `SHAPE_COUNT >= 10`, `RESTART_SERVER_EVERY=7` with the +default `seed`. The failing shape varies between runs (shape_5, +shape_8, shape_9, …) but is always one with a subquery dependency. + +**Root cause** + +`ShapeStatus.remove_shape/2` at +`lib/electric/shape_cache/shape_status.ex:207` removes a shape from +the persistent SQLite store *and* the in-memory ETS cache as its +*first* step: + +```elixir +def remove_shape(stack_id, shape_handle) when is_stack_id(stack_id) do + with :ok <- ShapeDb.remove_shape(stack_id, shape_handle) do + :ets.delete(shape_meta_table(stack_id), shape_handle) + decrement_shape_counts(stack_id, shape_cached_as_indexed?(stack_id, shape_handle)) + :ok + end +end +``` + +`ShapeCleaner.remove_shape_immediate/3` then proceeds through: + +1. `Consumer.stop(stack_id, shape_handle, reason)` +2. `Storage.cleanup!(stack_storage, shape_handle)` +3. `ShapeLogCollector.remove_shape(stack_id, shape_handle)` ← can fail + +If the call at step 3 fails (because the SLC's `RequestBatcher` is +already gone — exactly what happens during stack shutdown), the shape +has already been deleted from SQLite. After the new stack restores +shape state from disk, the handle is missing → `validate_shape_handle` +returns `:no_shape` → API returns 409 must-refetch. + +The trigger for the cleanup task: a consumer crashes with a non-shutdown +reason during the stop sequence (e.g., a materializer dies with `:killed` +because the supervisor's graceful shutdown timed out, and the dependent +consumer's `handle_materializer_down/2` falls through the `case` to +`stop_and_clean/1`). `stop_and_clean` exits with +`@stop_and_clean_reason = {:shutdown, :cleanup}`, and the consumer's +`terminate/2` calls `ShapeCleaner.handle_writer_termination/3` with that +reason, which schedules `remove_shape_async`. The async task then races +the rest of the stack shutdown. + +**Fix (partial)** + +The materializer is the upstream trigger for the cleanup cascade — a +crash there sets off `handle_materializer_down/2` → `stop_and_clean/1` +on the dependent consumer → `@shutdown_cleanup` → +`remove_shape_async/2`. The cleanup task then races stack shutdown +and leaves the shape half-deleted. + +What's committed: + +1. **Catch `:noproc` on consumer→materializer call in + `lib/electric/shapes/consumer.ex#notify_materializer_of_new_changes/3`.** + When the materializer dies, the `:DOWN` is in our mailbox but the + inline `GenServer.call` exits the consumer process before + `handle_materializer_down/2` runs. Catching the exit lets the + pending `:DOWN` drive a clean stop instead of cascading into + `@shutdown_cleanup`. This is a process-lifecycle issue (the + materializer legitimately died), not an invariant violation, so a + targeted `catch :exit` is the right call. + +2. **ETS rescue for the stack-restart inflight-request window.** + `ShapeStatus.validate_shape_handle/3` rescues `ArgumentError → + :error` and `Api.check_for_disk_updates/1` rescues + `ArgumentError → :no_change`. Same justification — a held + long-poll on Bandit can wake up between the old `ShapeStatusOwner` + freeing its tables and the new one recreating them. Resource + lifecycle, not state corruption. + +What was reverted: + +- The "log and skip" branches in + `lib/electric/shapes/consumer/materializer.ex#apply_changes/2` + (DELETE for unknown key, duplicate `NewRecord`, move-out/in for + unknown key, missing `value_count`, guarded `UpdatedRecord`). + Those branches mask real upstream bugs — duplicates and missing + values are invariant violations, not noise. The materializer is + meant to crash hard so the bug surfaces. Restoring `Map.pop!`, + `Map.fetch!`, and `raise/1` on duplicate insert. + +**Regression** + +The original repro +(`CHECK_TIMEOUT=60000 SHAPE_COUNT=10 MUTATIONS_PER_TXN=10 TXNS_PER_BATCH=10 +BATCH_COUNT=10 RESTART_SERVER_EVERY=7 SKIP_REPATCH_PREWARM=true +mix test --seed 8 --only oracle test/integration/oracle_property_test.exs`) +no longer hits a cleanup-cascade 409. With strict materializer back +in place, the actual remaining blocker is Bug 2 — duplicate INSERTs +in the post-restart shape log when the same transaction both moves a +row across the shape's filter boundary AND triggers a materializer +move-in. + +## Note for triage + +All known bugs are addressed. Final fix structure: + +- Bugs 1 and 4 fixed by the materializer history-replay fix. +- Bugs 2 and 5 fixed by `Electric.StackSupervisor.ShutdownCoordinator` + + `ShapeStatusOwner` dirty-startup detection: a coordinated drain + before clean shutdown writes a marker; on startup, missing marker + drops all shape state. This eliminates the L_inner ≠ L_outer + inconsistency that was the root cause of both bugs. +- Bug 6 fixed by the consumer `:noproc` catch + ETS rescue. + Materializer keeps strict invariants (no impossible-state + swallowing) — the dirty-startup path makes that safe by ensuring + the materializer's view and outer storage always start at the same + LSN. + +Bug 3 (out-of-bounds long-poll after restart) hasn't been seen since +the safe-shutdown changes landed; if it shows up again it's likely a +small fix in the active-readiness signal but no longer the test's +blocker. + +Bug 2 is now the next blocker. With the materializer hardening and +ETS-rescue from Bug 6 in place, the cascading 409 is gone — and +underneath it is a real snapshot/log boundary problem that surfaces +in two complementary forms: + +- **`insert for row that already exists`** — the move-in query + attached to a SplicePlan re-emits a row as an INSERT even though + the on-disk snapshot already covers it. +- **`delete for row that does not exist`** — the same family with the + opposite polarity: a synthetic delete (driven by a move-out's tag + patterns) targets a key the client never received an INSERT for. + +Repros: + +```sh +# duplicate insert variant +CHECK_TIMEOUT=60000 SHAPE_COUNT=10 MUTATIONS_PER_TXN=10 TXNS_PER_BATCH=10 \ + BATCH_COUNT=10 RESTART_SERVER_EVERY=7 LONG_POLL_TIMEOUT=2000 \ + SKIP_REPATCH_PREWARM=true \ + mix test --seed 8 --only oracle test/integration/oracle_property_test.exs + +# orphan-delete variant +CHECK_TIMEOUT=60000 SHAPE_COUNT=5 MUTATIONS_PER_TXN=10 TXNS_PER_BATCH=10 \ + BATCH_COUNT=20 RESTART_SERVER_EVERY=7 LONG_POLL_TIMEOUT=2000 \ + SKIP_REPATCH_PREWARM=true \ + mix test --seed 8 --only oracle test/integration/oracle_property_test.exs +``` + +Failure shapes are subquery shapes (e.g. `level_3_id IN (SELECT id +FROM level_3 WHERE level_2_id IN (SELECT id FROM level_2 WHERE +level_1_id = 'l1-4'))`). + +**Why this only happens after restart** + +The oracle test passes cleanly with `RESTART_SERVER_EVERY` unset. +The bug manifests only after a stack restart. So whatever's wrong is +specific to recovery, not steady-state replication. + +The state of an outer subquery shape consists of two pieces that are +written/updated independently: + +- **Outer shape's storage** (snapshot file + log file on disk). Each + commit appends to the log via the shape's writer. +- **Materializer view** (in-memory `value_counts`/`tag_indices` for + the inner dep shape). Updated as inner-shape events flow. + +In steady state these are *both* updated from the same in-memory +state machine and progress in lockstep — the outer consumer holds +`state.views` in memory and updates it on each materializer event, +and writes to outer storage are interleaved with these updates. +Pre-restart there's no way for the two to diverge because they share +the running consumer's state. + +**Across a restart they can land at different LSNs.** + +Each shape has its own writer (one per shape) with its own buffer. +A buffer is flushed when it fills up or when its periodic +`flush_period` timer fires. Different shapes reach those triggers +at different times, so different shapes' on-disk +`last_persisted_txn_offset` can differ. + +The replication-slot side does coordinate: `ShapeLogCollector.FlushTracker` +computes the **min** `last_flushed` across all shapes and advances +PG's slot to that minimum (`encode_standby_status_update` in +`lib/electric/postgres/replication_client.ex`). So PG won't ack +beyond the slowest shape. But that's only the lower bound for slot +durability — individual shapes' on-disk state can be ahead. + +When the stack is killed mid-flight there's no coordinated flush. +The supervisor sends `:shutdown`; each writer stops where it is. +For a subquery shape, the inner and outer shapes go through +**different writers** that flush independently, so `L_inner` and +`L_outer` are typically not equal at kill time — either may be +ahead. + +Then on recovery (Bug 5 fix path): + +1. Inner shape consumer comes up. Its writer's `last_persisted_offset` + = `L_inner`. +2. Materializer is started, replays inner shape's history up to + `L_inner`. Its view = state at `L_inner`. +3. Outer consumer comes up. Its writer's `last_persisted_offset` + = `L_outer ≤ L_inner` (they were flushed independently). +4. `EventHandlerBuilder.build/2` calls + `Materializer.get_link_values/1` → seeds `state.views` with the + materializer's view at `L_inner`. + +So `state.views` and the outer shape's storage are **at different +LSNs**: views is at `L_inner`, storage is at `L_outer`. There are +materializer events between `L_outer` and `L_inner` whose +corresponding ops were never written to the outer storage. From the +outer consumer's perspective those events have already been applied +(because `state.views` reflects them) — but the storage doesn't +have them. + +When live events resume after restart, the materializer eventually +re-emits a move-in for one of those values (because PG actually +re-arrives at that state via replication catch-up, or because the +test's batch mutates the same dep value). The outer consumer +processes it: `state.views` already has the value, but **the move-in +query still runs and writes the matching rows to outer storage**. +Those rows were already in the snapshot file from before the +restart — so we get a duplicate INSERT in the on-disk log, and a +fresh client polling from `offset=-1` sees both copies. + +The orphan-delete variant is the mirror: the materializer re-emits a +move-out for a value, the outer consumer writes a `move-out` +control message, the client's `TagTracker` generates a synthetic +delete for a row that was never INSERTed because the corresponding +INSERT was lost in the same gap. + +**The Bug 5 fix is incomplete.** Eager-starting subquery consumers +before `mark_as_ready` ensures `state.views` is seeded before live +events flow — but it seeds from the materializer's view, which can +be at a different LSN than the outer storage. The previous fix +prevented the materializer view from drifting *forward* of the +outer storage during replay, but it can't compensate for the +storage already being *behind*. + +**Fix directions** (any one would close the gap) + +1. **Persist the dep view per outer shape.** Write + `state.views` to the outer shape's metadata at every commit; + restore it on init alongside the shape's storage. The view and + the storage move atomically per commit. (Approach 1 from the + original Bug 5 triage.) +2. **Derive `state.views` from outer storage.** On init, scan the + outer shape's stored rows and reconstruct `state.views` by + collecting distinct values of each dep's foreign-key column + present in storage. The view by construction matches the + storage. (Approach 2.) +3. **Replay materializer events from `L_outer`.** Track each outer + shape's `last_persisted_dep_lsn` and have the materializer + replay events `> last_persisted_dep_lsn` to the outer consumer + on subscribe. The consumer applies them, which writes the + missing ops to storage and advances `state.views` to match. + (Approach 3.) + +(1) is the simplest to implement and matches how the rest of the +recovery machinery works. (2) is heavier on restart and requires the +storage to expose a way to enumerate the dep-key column values. (3) +is closest to the steady-state event flow but requires threading +offsets through the materializer↔consumer subscription protocol and +co-ordinating with `Buffering`. + +Bug 3 only manifests under restart and is likely related to the +active-readiness signal — small fix once Bug 2 is closed. diff --git a/packages/sync-service/lib/electric/monitored_core_supervisor.ex b/packages/sync-service/lib/electric/monitored_core_supervisor.ex index 3ce7cc6ce3..e8e7db2952 100644 --- a/packages/sync-service/lib/electric/monitored_core_supervisor.ex +++ b/packages/sync-service/lib/electric/monitored_core_supervisor.ex @@ -20,13 +20,16 @@ defmodule Electric.MonitoredCoreSupervisor do Logger.metadata(stack_id: stack_id) Electric.Telemetry.Sentry.set_tags_context(stack_id: stack_id) + persistent_kv = Keyword.get(opts, :persistent_kv) + children = [ {Electric.StatusMonitor, stack_id: stack_id}, {Electric.ShapeCache.ShapeCleaner.CleanupTaskSupervisor, [stack_id: stack_id] ++ Keyword.get(tweaks, :shape_cleaner_opts, [])}, {Electric.ShapeCache.ShapeStatus.ShapeDb.Supervisor, Keyword.take(opts, [:stack_id, :shape_db_opts])}, - {Electric.ShapeCache.ShapeStatusOwner, [stack_id: stack_id]}, + {Electric.ShapeCache.ShapeStatusOwner, + [stack_id: stack_id, persistent_kv: persistent_kv]}, {Electric.CoreSupervisor, opts} ] diff --git a/packages/sync-service/lib/electric/replication/shape_log_collector.ex b/packages/sync-service/lib/electric/replication/shape_log_collector.ex index 082fc71517..03288ab3e3 100644 --- a/packages/sync-service/lib/electric/replication/shape_log_collector.ex +++ b/packages/sync-service/lib/electric/replication/shape_log_collector.ex @@ -163,6 +163,36 @@ defmodule Electric.Replication.ShapeLogCollector do GenServer.cast(name(stack_id), {:writer_flushed, shape_handle, offset}) end + @doc """ + Stop accepting new replication events and block until every consumer + has flushed up to the most recently seen offset. + + Used by `Electric.StackSupervisor.ShutdownCoordinator` to drive a + coordinated safe shutdown: callers must first stop the replication + client (so no new events arrive) and then call `drain/2`. When this + returns `:ok`, every shape's writer has persisted the same + `last_persisted_txn_offset`. + + If the timeout expires before all shapes catch up, returns + `{:error, :timeout}`. The caller should treat that as a dirty + shutdown — no clean-shutdown marker should be written. + """ + @spec drain(Electric.stack_id(), timeout()) :: :ok | {:error, :timeout | :no_collector} + def drain(stack_id, timeout) do + case GenServer.whereis(name(stack_id)) do + nil -> + {:error, :no_collector} + + pid -> + try do + GenServer.call(pid, :drain, timeout) + catch + :exit, {:timeout, _} -> {:error, :timeout} + :exit, {:noproc, _} -> {:error, :no_collector} + end + end + end + @doc """ Returns the list of currently active shapes being tracked in the shape matching filters. @@ -382,6 +412,20 @@ defmodule Electric.Replication.ShapeLogCollector do {:reply, :ok, Map.put(state, :last_processed_offset, offset)} end + def handle_call(:drain, from, state) do + state = Map.put(state, :draining?, true) + + if FlushTracker.empty?(state.flush_tracker) do + {:reply, :ok, state} + else + {:noreply, Map.put(state, :drain_waiter, from)} + end + end + + def handle_call({:handle_event, _, _}, _from, %{draining?: true} = state) do + {:reply, {:error, :draining}, state} + end + def handle_call({:handle_event, _, _}, _from, state) when not is_ready_to_process(state) do {:reply, {:error, :not_ready}, state} @@ -414,9 +458,19 @@ defmodule Electric.Replication.ShapeLogCollector do end def handle_cast({:writer_flushed, shape_id, offset}, state) do - {:noreply, - state - |> Map.update!(:flush_tracker, &FlushTracker.handle_flush_notification(&1, shape_id, offset))} + state = + state + |> Map.update!(:flush_tracker, &FlushTracker.handle_flush_notification(&1, shape_id, offset)) + + state = + if state[:drain_waiter] && FlushTracker.empty?(state.flush_tracker) do + GenServer.reply(state.drain_waiter, :ok) + Map.delete(state, :drain_waiter) + else + state + end + + {:noreply, state} end def handle_cast( diff --git a/packages/sync-service/lib/electric/shape_cache.ex b/packages/sync-service/lib/electric/shape_cache.ex index e7796fb6df..9bedb75118 100644 --- a/packages/sync-service/lib/electric/shape_cache.ex +++ b/packages/sync-service/lib/electric/shape_cache.ex @@ -286,8 +286,14 @@ defmodule Electric.ShapeCache do Electric.Replication.PublicationManager.wait_for_restore(state.stack_id) - # Let ShapeLogCollector that it can start processing after finishing this function so that - # we're subscribed to the producer before it starts forwarding its demand. + # Subquery shapes' consumers must be fully initialized before + # ShapeLogCollector starts dispatching events. If events flow first, + # the materializer can advance past the outer shape's on-disk storage; + # the outer consumer's later init would then seed `state.views` from + # the advanced materializer view and a subsequent move-in event for + # a value already in that seeded view would be dropped as redundant. + eagerly_start_subquery_shape_consumers(state) + ShapeLogCollector.mark_as_ready(state.stack_id) duration = System.monotonic_time() - start_time @@ -305,6 +311,39 @@ defmodule Electric.ShapeCache do {:noreply, state} end + # Shapes whose where clause contains a subquery (`shape_dependencies != []`) + # rely on their materializer subscription to be notified of dependency-side + # changes. The router only delivers events for a shape when its own + # `root_table` changes, so a subquery dependent stays dormant after a + # restart until something writes to its own table — movements driven by + # the dependency (e.g. parent rows becoming active) never reach its + # on-disk view. Restoring it here re-establishes the materializer + # subscription so dependency updates flow in. + # + # `await_snapshot_start/2` is queued *after* the consumer's + # `:initialize_shape` info message, so by the time it returns + # `EventHandlerBuilder.build` has run and `state.views` is seeded. + defp eagerly_start_subquery_shape_consumers(state) do + opts = %{ + stack_id: state.stack_id, + action: :restore, + otel_ctx: nil, + feature_flags: state.feature_flags + } + + for {handle, %Shape{shape_dependencies: [_ | _]} = shape} <- + ShapeStatus.list_shapes(state.stack_id), + is_nil(Electric.Shapes.ConsumerRegistry.whereis(state.stack_id, handle)) do + case restore_shape_and_dependencies(handle, shape, opts) do + {:ok, _pid} -> + _ = Electric.Shapes.Consumer.await_snapshot_start(state.stack_id, handle) + + _ -> + :ok + end + end + end + @impl GenServer def handle_call({:create_or_wait_shape_handle, shape, otel_ctx}, _from, state) do if not is_nil(otel_ctx), do: OpenTelemetry.set_current_context(otel_ctx) diff --git a/packages/sync-service/lib/electric/shape_cache/shape_status.ex b/packages/sync-service/lib/electric/shape_cache/shape_status.ex index 80cd0246d2..ff879e05b3 100644 --- a/packages/sync-service/lib/electric/shape_cache/shape_status.ex +++ b/packages/sync-service/lib/electric/shape_cache/shape_status.ex @@ -291,6 +291,14 @@ defmodule Electric.ShapeCache.ShapeStatus do [] -> :error end + rescue + # The shape_meta_table is created by ShapeStatusOwner. During a stack + # restart there's a window when the old owner's table has been freed + # and the new owner hasn't recreated it yet. An inflight HTTP request + # held by Bandit can wake up in that window. Treat a missing table + # the same as a missing entry — the caller can fall back to + # fetch_handle_by_shape (which goes through the live SQLite store). + ArgumentError -> :error end @spec mark_snapshot_started(stack_id(), shape_handle()) :: :ok | :error diff --git a/packages/sync-service/lib/electric/shape_cache/shape_status_owner.ex b/packages/sync-service/lib/electric/shape_cache/shape_status_owner.ex index 13f4968e7b..7b466c1457 100644 --- a/packages/sync-service/lib/electric/shape_cache/shape_status_owner.ex +++ b/packages/sync-service/lib/electric/shape_cache/shape_status_owner.ex @@ -14,10 +14,16 @@ defmodule Electric.ShapeCache.ShapeStatusOwner do use GenServer alias Electric.ShapeCache.ShapeStatus + alias Electric.ShapeCache.ShapeStatus.ShapeDb + alias Electric.ShapeCache.Storage + alias Electric.StackSupervisor.ShutdownCoordinator require Logger - @schema NimbleOptions.new!(stack_id: [type: :string, required: true]) + @schema NimbleOptions.new!( + stack_id: [type: :string, required: true], + persistent_kv: [type: :any, required: false] + ) def name(stack_id) do Electric.ProcessRegistry.name(stack_id, __MODULE__) @@ -50,11 +56,14 @@ defmodule Electric.ShapeCache.ShapeStatusOwner do :ok = Electric.LsnTracker.initialize(stack_id) - {:ok, %{stack_id: stack_id}, {:continue, :initialize}} + {:ok, %{stack_id: stack_id, persistent_kv: Map.get(config, :persistent_kv)}, + {:continue, :initialize}} end @impl true def handle_continue(:initialize, state) do + maybe_reset_after_dirty_shutdown(state) + # Initialize shape metadata from SQLite so we can serve shapes in # read-only mode before the advisory lock is acquired :ok = ShapeStatus.initialize(state.stack_id) @@ -67,6 +76,45 @@ defmodule Electric.ShapeCache.ShapeStatusOwner do {:noreply, state, :hibernate} end + # If the previous stack shutdown wasn't clean, different shapes' on-disk + # `last_persisted_txn_offset`s may be at different LSNs (writers flush + # independently and the supervisor doesn't coordinate a final flush + # before tear-down). Recovering from such a state would mean that an + # outer subquery shape's `state.views` (seeded from the materializer, + # which is rebuilt from the inner shape's history) and its on-disk + # storage land at different LSNs, producing duplicate inserts and + # orphan deletes when live events resume. + # + # Sound recovery from a partial shutdown is hard; the conservative + # answer is to throw away every shape's persisted state and force a + # re-snapshot. Clients re-request their shapes from scratch on the + # next poll. + defp maybe_reset_after_dirty_shutdown(%{persistent_kv: nil}), do: :ok + + defp maybe_reset_after_dirty_shutdown(%{persistent_kv: persistent_kv} = state) do + if ShutdownCoordinator.consume_clean_shutdown_marker(persistent_kv) do + :ok + else + Logger.warning( + "Stack #{state.stack_id} did not shut down cleanly last time; " <> + "dropping all shape data to avoid inconsistent recovery" + ) + + # Wipe before ShapeStatus.initialize so that initialize sees an + # empty SQLite store and creates a fresh, empty ETS cache. We + # can't call ShapeStatus.reset/1 here because the ETS tables it + # tries to clear haven't been created yet (initialize is what + # creates them). + :ok = ShapeDb.reset(state.stack_id) + + state.stack_id + |> Storage.for_stack() + |> Storage.cleanup_all!() + + :ok + end + end + @impl true def handle_call(:refresh, _from, state) do :ok = ShapeStatus.refresh(state.stack_id) diff --git a/packages/sync-service/lib/electric/shapes/api.ex b/packages/sync-service/lib/electric/shapes/api.ex index 09717ebd22..ca26772c8c 100644 --- a/packages/sync-service/lib/electric/shapes/api.ex +++ b/packages/sync-service/lib/electric/shapes/api.ex @@ -970,6 +970,13 @@ defmodule Electric.Shapes.Api do _ -> :no_change end + rescue + # During a stack restart, the per-stack ETS tables this path reads + # (shape_meta_table, write_buffer_shapes, …) may be transiently absent + # for a held long-poll that wakes up between the old owner dying and + # the new owner recreating the table. Treat as no_change rather than + # bubbling an ArgumentError up to the request handler. + ArgumentError -> :no_change end defp stream_sse_events(%Request{} = request) do diff --git a/packages/sync-service/lib/electric/shapes/consumer.ex b/packages/sync-service/lib/electric/shapes/consumer.ex index 6ee86ca229..0187dc796b 100644 --- a/packages/sync-service/lib/electric/shapes/consumer.ex +++ b/packages/sync-service/lib/electric/shapes/consumer.ex @@ -1063,6 +1063,18 @@ defmodule Electric.Shapes.Consumer do opts ) do Materializer.new_changes(Map.take(state, [:stack_id, :shape_handle]), changes_or_bounds, opts) + catch + # The consumer monitors the materializer; if the materializer died the + # :DOWN message is already in our mailbox and handle_materializer_down/2 + # will run after the current handle_event/handle_call completes. + # Treat a `:noproc` (or transient `:normal`/`:shutdown` exit) here as + # the same condition: don't crash the consumer (which would route into + # the abnormal-shutdown path of handle_writer_termination and remove + # the shape from disk). + :exit, {:noproc, _} -> :ok + :exit, :noproc -> :ok + :exit, {:normal, _} -> :ok + :exit, {:shutdown, _} -> :ok end defp notify_materializer_of_new_changes(_state, _changes_or_bounds, _opts), do: :ok diff --git a/packages/sync-service/lib/electric/shapes/consumer/materializer.ex b/packages/sync-service/lib/electric/shapes/consumer/materializer.ex index 08c6c97bf5..2ab3512826 100644 --- a/packages/sync-service/lib/electric/shapes/consumer/materializer.ex +++ b/packages/sync-service/lib/electric/shapes/consumer/materializer.ex @@ -171,33 +171,72 @@ defmodule Electric.Shapes.Consumer.Materializer do end def handle_continue({:read_stream, storage}, state) do - {:ok, offset, stream} = - get_stream_up_to_offset(state.offset, state.subscribed_offset, storage) - - {state, _} = - stream - |> decode_json_stream() - |> apply_changes(state) - + state = read_history_up_to_subscribed(state, storage) write_link_values(state) - - {:noreply, %{state | offset: offset}} + {:noreply, state} end @doc """ - Get a stream of log entries from storage, bounded by the subscribed offset. - - The subscribed_offset is the Consumer's latest_offset at the time of subscription. - We only read up to this offset to avoid duplicates - any changes after this offset - will be delivered via new_changes messages from the Consumer. + Replay all of the source shape's persisted history (snapshot + log) up to + `state.subscribed_offset` so the materializer's value_counts reflect the + on-disk state on startup. + + `Storage.get_log_stream/3` returns at most one chunk per call **for + snapshot chunks** — but for the main log it returns the entire requested + range `[min_offset, subscribed_offset]` in one call. So we iterate + through snapshot chunks using `Storage.get_chunk_end_log_offset/2`, + and as soon as the iteration would step into the main log we stop: + the previous call already streamed everything up to the subscribed + offset. Iterating into the main log would re-read entries already + applied, producing duplicate inserts that crash the materializer. """ - def get_stream_up_to_offset(min_offset, subscribed_offset, storage) do - # If subscribed_offset is nil or at/before min_offset, nothing to read - if is_nil(subscribed_offset) or is_log_offset_lte(subscribed_offset, min_offset) do - {:ok, min_offset, []} - else - stream = Storage.get_log_stream(min_offset, subscribed_offset, storage) - {:ok, subscribed_offset, stream} + def read_history_up_to_subscribed(state, storage) do + cond do + is_nil(state.subscribed_offset) -> + state + + is_log_offset_lte(state.subscribed_offset, state.offset) -> + state + + true -> + stream = Storage.get_log_stream(state.offset, state.subscribed_offset, storage) + {state, _} = stream |> decode_json_stream() |> apply_changes(state) + + # If the read just covered the main log (because either the + # current offset is already past the snapshot or the next chunk + # boundary jumps into real-offset territory), `stream_main_log` + # returned the whole range up to `subscribed_offset` in a single + # call and we're done. + if is_real_offset(state.offset) or is_last_virtual_offset(state.offset) do + %{state | offset: state.subscribed_offset} + else + next_offset = Storage.get_chunk_end_log_offset(state.offset, storage) + + cond do + is_nil(next_offset) -> + # No further chunks past this offset — we've reached the end. + %{state | offset: state.subscribed_offset} + + is_log_offset_lte(next_offset, state.offset) -> + # Defensive: chunk_end did not advance. Stop to avoid an + # infinite loop. This shouldn't happen in normal operation. + %{state | offset: state.subscribed_offset} + + is_log_offset_lte(state.subscribed_offset, next_offset) -> + %{state | offset: state.subscribed_offset} + + is_real_offset(next_offset) -> + # The next chunk is in the main log, which means the call + # we just made (with `state.offset` past the last snapshot + # chunk) already streamed the entire main log up to + # `subscribed_offset`. Stop — iterating further would + # re-read entries we've already applied. + %{state | offset: state.subscribed_offset} + + true -> + read_history_up_to_subscribed(%{state | offset: next_offset}, storage) + end + end end end diff --git a/packages/sync-service/lib/electric/stack_supervisor.ex b/packages/sync-service/lib/electric/stack_supervisor.ex index 3d8a35b078..b0df847ab1 100644 --- a/packages/sync-service/lib/electric/stack_supervisor.ex +++ b/packages/sync-service/lib/electric/stack_supervisor.ex @@ -427,7 +427,12 @@ defmodule Electric.StackSupervisor do {Electric.MonitoredCoreSupervisor, stack_id: stack_id, connection_manager_opts: connection_manager_opts, - shape_db_opts: config.shape_db_opts} + shape_db_opts: config.shape_db_opts, + persistent_kv: config.persistent_kv}, + # ShutdownCoordinator must be the LAST child so it terminates + # FIRST on graceful stack shutdown — see the module docstring. + {Electric.StackSupervisor.ShutdownCoordinator, + stack_id: stack_id, persistent_kv: config.persistent_kv} ] |> Enum.reject(&is_nil/1) diff --git a/packages/sync-service/lib/electric/stack_supervisor/shutdown_coordinator.ex b/packages/sync-service/lib/electric/stack_supervisor/shutdown_coordinator.ex new file mode 100644 index 0000000000..33811a72f5 --- /dev/null +++ b/packages/sync-service/lib/electric/stack_supervisor/shutdown_coordinator.ex @@ -0,0 +1,135 @@ +defmodule Electric.StackSupervisor.ShutdownCoordinator do + @moduledoc """ + Writes a "clean shutdown" marker to `persistent_kv` when the stack + shuts down gracefully. On the next stack startup, `ShapeStatusOwner` + reads the marker: if it's missing the stack didn't shut down cleanly + and all shape data is dropped, because there's no way to know which + shapes' on-disk state is consistent with which other shapes' + (different shapes' writers flush independently and can land at + different `last_persisted_txn_offset`s after a crash). + + This GenServer is added as the *last* child of `StackSupervisor` so + that on graceful shutdown it terminates *first* (Supervisor stops + children in reverse start order). Its `terminate/2` runs while every + other stack process is still alive, then writes the marker. + + If the process is brutal-killed (e.g. `kill -9`, OOM, supervisor + shutdown timeout exceeded), `terminate/2` does not run, the marker + is not written, and the next startup correctly treats the previous + shutdown as dirty. + """ + + use GenServer + + alias Electric.PersistentKV + alias Electric.Postgres.ReplicationClient + alias Electric.Replication.ShapeLogCollector + + require Logger + + @marker_key "stack_supervisor:clean_shutdown_marker" + @drain_timeout 30_000 + + def child_spec(opts) do + %{ + id: __MODULE__, + start: {__MODULE__, :start_link, [opts]}, + shutdown: 5_000 + } + end + + def start_link(opts) do + GenServer.start_link(__MODULE__, opts, name: name(opts[:stack_id])) + end + + def name(stack_id) when is_binary(stack_id) do + Electric.ProcessRegistry.name(stack_id, __MODULE__) + end + + @doc """ + Returns true iff the previous stack shutdown wrote the clean-shutdown + marker. Has the side effect of clearing the marker — every startup + must observe it as missing unless the *previous* shutdown wrote it. + """ + @spec consume_clean_shutdown_marker(Electric.PersistentKV.t()) :: boolean() + def consume_clean_shutdown_marker(persistent_kv) do + case PersistentKV.get(persistent_kv, @marker_key) do + {:ok, _value} -> + :ok = PersistentKV.set(persistent_kv, @marker_key, nil) + true + + {:error, :not_found} -> + false + end + end + + @impl true + def init(opts) do + Process.flag(:trap_exit, true) + + state = %{ + stack_id: Keyword.fetch!(opts, :stack_id), + persistent_kv: Keyword.fetch!(opts, :persistent_kv) + } + + Logger.metadata(stack_id: state.stack_id) + Process.set_label({:shutdown_coordinator, state.stack_id}) + + {:ok, state} + end + + @impl true + def terminate(reason, state) do + if clean_reason?(reason) do + drain_and_mark_clean(state) + else + Logger.warning( + "ShutdownCoordinator terminating with non-clean reason #{inspect(reason)}; " <> + "clean-shutdown marker NOT written, next startup will reset shape data" + ) + end + + :ok + end + + defp clean_reason?(:shutdown), do: true + defp clean_reason?({:shutdown, _}), do: true + defp clean_reason?(:normal), do: true + defp clean_reason?(_), do: false + + # Run the drain protocol while every other stack process is still alive + # (we're the last child, so we terminate first): + # 1. Stop the replication client so no new events arrive at the SLC. + # 2. Tell the SLC to drain — it stops accepting events and replies once + # every consumer has flushed up to the last seen offset. + # 3. Only then write the clean-shutdown marker. If any step fails, leave + # the marker absent so the next startup treats this as dirty. + defp drain_and_mark_clean(state) do + with :ok <- stop_replication_client(state.stack_id), + :ok <- ShapeLogCollector.drain(state.stack_id, @drain_timeout) do + :ok = PersistentKV.set(state.persistent_kv, @marker_key, true) + Logger.notice("Stack #{state.stack_id} cleanly shut down") + else + error -> + Logger.warning( + "Stack #{state.stack_id} failed to drain cleanly (#{inspect(error)}); " <> + "next startup will reset shape data" + ) + end + end + + defp stop_replication_client(stack_id) do + case GenServer.whereis(ReplicationClient.name(stack_id)) do + nil -> + :ok + + pid -> + try do + ReplicationClient.stop(pid, :shutdown) + :ok + catch + :exit, _ -> :ok + end + end + end +end diff --git a/packages/sync-service/test/electric/shape_cache_test.exs b/packages/sync-service/test/electric/shape_cache_test.exs index 1407853478..4ac6e3fc69 100644 --- a/packages/sync-service/test/electric/shape_cache_test.exs +++ b/packages/sync-service/test/electric/shape_cache_test.exs @@ -1407,6 +1407,93 @@ defmodule Electric.ShapeCacheTest do end end + describe "wait_for_restore eager subquery consumer start" do + setup [ + :with_noop_publication_manager, + :with_log_chunking, + :with_registry, + :with_shape_log_collector + ] + + test "shapes with subquery dependencies have their consumer eagerly started", ctx do + %{stack_id: stack_id} = ctx + test_pid = self() + + # Pre-add a subquery shape to ShapeStatus, simulating a shape that + # was created in a prior incarnation of the stack and is now being + # restored. The shape's consumer is NOT yet running. + {:ok, shape_handle} = ShapeStatus.add_shape(stack_id, @shape_with_subquery) + + # We don't have a fully wired-up consumer chain in this test, so + # short-circuit `start_shape_consumer` and `start_materializer` while + # recording the calls. This proves wait_for_restore reaches the start + # path for the subquery shape; full consumer lifecycle is covered by + # the integration tests in `oracle_restore_test.exs`. + Repatch.patch( + Electric.Shapes.DynamicConsumerSupervisor, + :start_materializer, + [mode: :shared], + fn _stack_id, _config -> {:ok, self()} end + ) + + Repatch.patch( + Electric.Shapes.DynamicConsumerSupervisor, + :start_shape_consumer, + [mode: :shared], + fn _stack_id, %{shape_handle: handle} -> + send(test_pid, {:start_shape_consumer_called, handle}) + # Returning :error short-circuits restore_shape_and_dependencies' + # follow-up calls (initialize_shape, update_last_read_time) which + # would fail without a real consumer pid. + {:error, :test_short_circuit} + end + ) + + # clean_shape is called on start_shape_consumer error; stub it so the + # follow-up cleanup doesn't interfere with the test assertion. + Repatch.patch( + Electric.ShapeCache.ShapeCleaner, + :remove_shape, + [mode: :shared], + fn _stack_id, _handle -> :ok end + ) + + activate_mocks_for_descendant_procs(Electric.ShapeCache) + + with_shape_cache(ctx) + + # The eager-start path runs in handle_continue(:wait_for_restore); the + # patched start_shape_consumer captures the call. + assert_receive {:start_shape_consumer_called, ^shape_handle}, 5_000 + end + + test "non-subquery shapes are NOT eagerly started", ctx do + %{stack_id: stack_id} = ctx + test_pid = self() + + # Add a shape with no dependencies — eager-start should skip it. + {:ok, _shape_handle} = ShapeStatus.add_shape(stack_id, @shape) + + Repatch.patch( + Electric.Shapes.DynamicConsumerSupervisor, + :start_shape_consumer, + [mode: :shared], + fn _stack_id, %{shape_handle: handle} -> + send(test_pid, {:start_shape_consumer_called, handle}) + {:error, :test_short_circuit} + end + ) + + activate_mocks_for_descendant_procs(Electric.ShapeCache) + + with_shape_cache(ctx) + + # Give wait_for_restore time to complete; eager-start should NOT + # have called start_shape_consumer for the simple shape. + refute_receive {:start_shape_consumer_called, _}, 200 + end + end + describe "start_consumer_for_handle/2" do setup [ :with_noop_publication_manager, @@ -1445,19 +1532,21 @@ defmodule Electric.ShapeCacheTest do GenServer.whereis(Electric.Shapes.Consumer.Materializer.name(stack_id, dep_handle)) ) - # Register this test as the connection manager to get "consumers ready" notification restart_shape_cache(ctx) assert [{^dep_handle, _}, {^shape_handle, _}] = ShapeCache.list_shapes(stack_id) - refute Electric.Shapes.ConsumerRegistry.whereis(stack_id, shape_handle) - refute Electric.Shapes.ConsumerRegistry.whereis(stack_id, dep_handle) - - refute GenServer.whereis(Electric.Shapes.Consumer.Materializer.name(stack_id, dep_handle)) - - assert {:ok, _pid1} = ShapeCache.start_consumer_for_handle(shape_handle, stack_id) + # After restart, ShapeCache eagerly starts subquery shape consumers + # in `handle_continue(:wait_for_restore)` so the outer consumer and + # its dependency materializer come back up automatically — no + # explicit `start_consumer_for_handle/2` call is required. The + # continue runs asynchronously after `start_supervised!` returns, + # so we wait for the registry to populate. + assert wait_until(1000, fn -> + not is_nil(Electric.Shapes.ConsumerRegistry.whereis(stack_id, shape_handle)) and + not is_nil(Electric.Shapes.ConsumerRegistry.whereis(stack_id, dep_handle)) + end) - # Materializer should be started assert Process.alive?( GenServer.whereis(Electric.Shapes.Consumer.Materializer.name(stack_id, dep_handle)) ) diff --git a/packages/sync-service/test/electric/shapes/consumer/materializer_test.exs b/packages/sync-service/test/electric/shapes/consumer/materializer_test.exs index 06f9cbbd48..48f0b49839 100644 --- a/packages/sync-service/test/electric/shapes/consumer/materializer_test.exs +++ b/packages/sync-service/test/electric/shapes/consumer/materializer_test.exs @@ -1376,6 +1376,70 @@ defmodule Electric.Shapes.Consumer.MaterializerTest do end end + describe "startup history replay" do + # Storage.get_log_stream/3 returns at most one chunk per call (one + # snapshot chunk, or one main-log chunk). On startup the Materializer + # must iterate chunks until it reaches `subscribed_offset` so it + # correctly replays the source shape's full persisted history. If it + # reads only the first chunk, post-snapshot updates persisted to the + # main log are silently dropped, leaving `value_counts` reflecting only + # the snapshot. + test "replays main-log entries persisted before subscription", ctx do + shape_handle = "history-test-#{System.unique_integer()}" + + storage = Storage.for_shape(shape_handle, ctx.storage) + Storage.start_link(storage) + writer = Storage.init_writer!(storage, @shape) + Storage.mark_snapshot_as_started(storage) + + # Snapshot with one row at value=10 + Storage.make_new_snapshot!( + make_snapshot_data([%Changes.NewRecord{record: %{"id" => "1", "value" => "10"}}]), + storage + ) + + # Main-log entry that updates the row to value=99 — written to disk + # before the Materializer subscribes. + log_offset = LogOffset.new(100, 0) + + writer = + Storage.append_to_log!( + [ + {log_offset, ~s|"public"."test_table"/"1"|, :update, + ~s|{"key":"\\"public\\".\\"test_table\\"/\\"1\\"","value":{"id":"1","value":"99"},"headers":{"operation":"update"}}|} + ], + writer + ) + + Storage.hibernate(writer) + + ConsumerRegistry.register_consumer(self(), shape_handle, ctx.stack_id) + + {:ok, _pid} = + Materializer.start_link(%{ + stack_id: ctx.stack_id, + shape_handle: shape_handle, + storage: ctx.storage, + columns: ["value"], + materialized_type: {:array, :int8} + }) + + respond_to_call(:await_snapshot_start, :started) + # Subscribe at an offset past the persisted UPDATE — the Materializer + # must walk the snapshot AND the main log to reach this point. + respond_to_call(:subscribe_materializer, {:ok, log_offset}) + + mat_ctx = %{stack_id: ctx.stack_id, shape_handle: shape_handle} + assert Materializer.wait_until_ready(mat_ctx) == :ok + + # If only snapshot chunk 0 was read, value_counts would contain 10 + # (the snapshot value) and the persisted UPDATE would be lost. The + # iteration fix guarantees the UPDATE is replayed. + assert Materializer.get_link_values(mat_ctx) == MapSet.new([99]) + end + + end + describe "startup race condition handling" do # Tests for the race condition where Consumer dies between await_snapshot_start # and subscribe_materializer. See concurrency_analysis/MATERIALIZER_RACE_ANALYSIS.md diff --git a/packages/sync-service/test/integration/oracle_property_test.exs b/packages/sync-service/test/integration/oracle_property_test.exs index 2787fa9108..643fbd84ef 100644 --- a/packages/sync-service/test/integration/oracle_property_test.exs +++ b/packages/sync-service/test/integration/oracle_property_test.exs @@ -12,6 +12,14 @@ defmodule Electric.Integration.OraclePropertyTest do - MUTATIONS_PER_TXN: Number of mutations per transaction (default: 5) - RUN_COUNT: Number of property test iterations (default: 1) - LONG_POLL_TIMEOUT: Server long-poll timeout in ms (default: 100) + - RESTART_SERVER_EVERY: Stop and restart the sync stack every N batches to + test server-side restore-from-file (default: 0, disabled). After each + restart, fresh clients reconnect and check_initial_state asserts the + restored state matches the oracle. + - RESTART_CLIENT_EVERY: Throw away clients (poll cursors, materialized + rows) and reconnect every M batches to test that fresh polls correctly + assemble snapshot + log (default: 0, disabled). Independent of + RESTART_SERVER_EVERY. """ use ExUnit.Case, async: false @@ -35,6 +43,7 @@ defmodule Electric.Integration.OraclePropertyTest do @default_mutations_per_txn 5 setup [:with_unique_db] + setup :use_persistent_slot setup :with_complete_stack # Use a short long_poll_timeout to speed up tests - shapes with no changes @@ -55,12 +64,23 @@ defmodule Electric.Integration.OraclePropertyTest do ctx end + # The replication slot must survive the StackSupervisor restart used by + # RESTART_SERVER_EVERY, otherwise Electric correctly treats the new slot + # as a slot-loss event and purges all on-disk shape data — defeating the + # restore-from-file scenario. Always run with a persistent slot; the slot + # is dropped automatically with the per-test database in `after_suite`. + defp use_persistent_slot(_ctx) do + %{replication_opts_overrides: [slot_temporary?: false]} + end + test "shapes with generated where clauses and mutations", ctx do run_count = env_int("RUN_COUNT") || 1 shape_count = env_int("SHAPE_COUNT") || @default_shape_count batch_count = env_int("BATCH_COUNT") || @default_batch_count txns_per_batch = env_int("TXNS_PER_BATCH") || @default_txns_per_batch mutations_per_txn = env_int("MUTATIONS_PER_TXN") || @default_mutations_per_txn + restart_server_every = env_int("RESTART_SERVER_EVERY") || 0 + restart_client_every = env_int("RESTART_CLIENT_EVERY") || 0 total_mutations = batch_count * txns_per_batch * mutations_per_txn @@ -69,7 +89,11 @@ defmodule Electric.Integration.OraclePropertyTest do max_runs: run_count do transactions = Enum.chunk_every(mutations, mutations_per_txn) batches = Enum.chunk_every(transactions, txns_per_batch) - test_against_oracle(ctx, shapes, batches) + + test_against_oracle(ctx, shapes, batches, + restart_server_every: restart_server_every, + restart_client_every: restart_client_every + ) end end end diff --git a/packages/sync-service/test/integration/oracle_restore_test.exs b/packages/sync-service/test/integration/oracle_restore_test.exs new file mode 100644 index 0000000000..cc6cc5e45c --- /dev/null +++ b/packages/sync-service/test/integration/oracle_restore_test.exs @@ -0,0 +1,181 @@ +defmodule Electric.Integration.OracleRestoreTest do + @moduledoc """ + Targeted regression tests for restore-from-file. Each test exercises a + scenario from `bugs.md` with a deterministic, minimal mutation sequence, + reusing `Support.OracleHarness.test_against_oracle/4`. + + These tests are expected to fail until the underlying Electric bugs are + fixed. + """ + + use ExUnit.Case, async: false + + import Support.ComponentSetup + import Support.DbSetup + import Support.IntegrationSetup + alias Support.OracleHarness + alias Support.OracleHarness.StandardSchema + + @moduletag :oracle + @moduletag timeout: :infinity + @moduletag :tmp_dir + + setup [:with_unique_db] + setup :use_persistent_slot + setup :with_complete_stack + + setup ctx do + ctx = + with_electric_client(ctx, + router_opts: [long_poll_timeout: 100], + num_clients: 1 + ) + + StandardSchema.setup_standard_schema(ctx) + ctx + end + + # See `oracle_property_test.exs`: the StackSupervisor restart needs the + # replication slot to persist so Electric reconnects rather than treating + # a new slot as a slot-loss event and purging on-disk shape data. + defp use_persistent_slot(_ctx) do + %{replication_opts_overrides: [slot_temporary?: false]} + end + + @tag :oracle_restore_bug_4 + @tag chunk_size: 200 + test "bug 4: subquery shape returns 409 after restart with many persisted log entries", + ctx do + # Same shape as Bug 1 but with enough mutations between snapshot and + # restart that the persisted main log spans more than one read range. + # The materializer's startup replay re-reads main-log entries on + # subsequent iterations because `stream_main_log` returns the whole + # range in a single call but the iteration loop keeps advancing. + # Re-replaying the same INSERTs raises "Key already exists" inside the + # materializer, which crashes the dependent shape's consumer and + # causes the server to return 409 must-refetch on the next poll. + shapes = [ + %{ + name: "active_l3", + table: "level_4", + where: "level_3_id IN (SELECT id FROM level_3 WHERE active = true)", + columns: ["id", "level_3_id", "value"], + pk: ["id"], + optimized: true + } + ] + + # Build a batch with many UPDATE entries on the source table (level_3) + # so that the dependency materializer's source shape persists a long + # log on disk before the restart. The bug surfaces when the + # materializer's startup replay re-reads main-log entries. + many_l3_mutations = + for i <- 1..30 do + active = if rem(i, 2) == 0, do: "true", else: "false" + id = "l3-#{rem(i, 5) + 1}" + + %{ + name: "toggle_#{id}_#{i}", + sql: "UPDATE level_3 SET active = #{active} WHERE id = '#{id}'" + } + end + + batches = [ + Enum.map(many_l3_mutations, &[&1]), + [ + [%{name: "deactivate_l3-1", sql: "UPDATE level_3 SET active = false WHERE id = 'l3-1'"}] + ] + ] + + OracleHarness.test_against_oracle(ctx, shapes, batches, restart_server_every: 1) + end + + @tag :oracle_restore_bug_5 + @tag chunk_size: 200 + test "bug 5: multiple subquery shapes diverge after restart with long persisted log", + ctx do + # Multiple shapes whose source-shape main logs span more than one chunk + # (forced via `chunk_size: 200`). After a server restart the + # materialized view of at least one shape no longer matches the oracle + # — rows that should be in the view are missing. The single-shape + # variants of this scenario (Bug 1 and Bug 4 regression tests) pass + # cleanly, so this looks like an interaction between concurrent + # materializer recoveries — possibly a shared per-stack ETS structure + # (link-values cache, SubqueryIndex) being read by one shape's + # consumer before another shape's materializer has finished + # repopulating it on startup. + shapes = [ + %{ + name: "shape_active_true", + table: "level_4", + where: "level_3_id IN (SELECT id FROM level_3 WHERE active = true)", + columns: ["id", "level_3_id", "value"], + pk: ["id"], + optimized: true + }, + %{ + name: "shape_active_false", + table: "level_4", + where: "level_3_id IN (SELECT id FROM level_3 WHERE active = false)", + columns: ["id", "level_3_id", "value"], + pk: ["id"], + optimized: true + } + ] + + many_l3_mutations = + for i <- 1..200 do + active = if rem(i, 2) == 0, do: "true", else: "false" + id = "l3-#{rem(i, 5) + 1}" + + %{ + name: "toggle_#{id}_#{i}", + sql: "UPDATE level_3 SET active = #{active} WHERE id = '#{id}'" + } + end + + batches = [ + Enum.map(many_l3_mutations, &[&1]), + [ + [%{name: "deactivate_l3-2", sql: "UPDATE level_3 SET active = false WHERE id = 'l3-2'"}] + ] + ] + + OracleHarness.test_against_oracle(ctx, shapes, batches, restart_server_every: 1) + end + + @tag :oracle_restore_bug_1 + test "bug 1: subquery shape diverges from oracle after server restart", ctx do + # Shape on level_4 with a subquery predicate over level_3.active. After + # the server is restarted, the subquery materializer state is not + # restored from disk, so toggling level_3.active on either side of the + # restart produces a divergence between the client view and the oracle + # (or a 409 must-refetch on this `optimized: true` shape). + shapes = [ + %{ + name: "active_level_3_children", + table: "level_4", + where: "level_3_id IN (SELECT id FROM level_3 WHERE active = true)", + columns: ["id", "level_3_id", "value"], + pk: ["id"], + optimized: true + } + ] + + # Two batches with one mutation each. Restart fires after batch_1. + # The mutations move rows in/out of the shape because they flip the + # parent level_3 row's `active` flag — so the subquery's result set + # changes, and the materializer is the component responsible for + # routing the corresponding level_4 rows in or out. + batches = [ + [ + [%{name: "deactivate_l3-1", sql: "UPDATE level_3 SET active = false WHERE id = 'l3-1'"}] + ], + [ + [%{name: "reactivate_l3-1", sql: "UPDATE level_3 SET active = true WHERE id = 'l3-1'"}] + ] + ] + + OracleHarness.test_against_oracle(ctx, shapes, batches, restart_server_every: 1) + end +end diff --git a/packages/sync-service/test/support/component_setup.ex b/packages/sync-service/test/support/component_setup.ex index 284415d51c..5ba6348cc9 100644 --- a/packages/sync-service/test/support/component_setup.ex +++ b/packages/sync-service/test/support/component_setup.ex @@ -432,11 +432,68 @@ defmodule Support.ComponentSetup do stack_events_registry = Electric.stack_events_registry() - ref = Electric.StackSupervisor.subscribe_to_stack_events(stack_id) - publication_name = Map.get(ctx, :publication_name, "electric_test_pub_#{:erlang.phash2(stack_id)}") + stack_supervisor = + start_stack_supervisor!(ctx, stack_id, kv, storage, stack_events_registry, publication_name) + + %{ + stack_id: stack_id, + registry: Electric.StackSupervisor.registry_name(stack_id), + stack_events_registry: stack_events_registry, + shape_cache: {ShapeCache, [stack_id: stack_id]}, + persistent_kv: kv, + stack_supervisor: stack_supervisor, + storage: storage, + inspector: + {EtsInspector, stack_id: stack_id, server: EtsInspector.name(stack_id: stack_id)}, + feature_flags: Electric.Config.get_env(:feature_flags), + publication_name: publication_name + } + end + + @doc """ + Stops the running `Electric.StackSupervisor` started by `with_complete_stack/1` + and starts a fresh one with the same `stack_id`, `persistent_kv`, `storage`, + and `publication_name` so the new stack reads back what the old stack + persisted to disk. + + Returns a map with the updated `:stack_supervisor` pid. Other ctx keys + (stack_id, persistent_kv, storage, registry, etc.) are unchanged. + """ + def restart_complete_stack(ctx) do + :ok = stop_supervised(Electric.StackSupervisor) + + stack_supervisor = + start_stack_supervisor!( + ctx, + ctx.stack_id, + ctx.persistent_kv, + ctx.storage, + ctx.stack_events_registry, + ctx.publication_name + ) + + # The :stack_status :ready event fires before the connection pools and + # shape metadata are fully online. Polling clients hitting the server in + # that window can see spurious 409s. Wait for the StatusMonitor's + # :active level which requires all readiness conditions to be met. + :ok = Electric.StatusMonitor.wait_until_active(ctx.stack_id, timeout: 5000) + + %{stack_supervisor: stack_supervisor} + end + + defp start_stack_supervisor!( + ctx, + stack_id, + kv, + storage, + stack_events_registry, + publication_name + ) do + ref = Electric.StackSupervisor.subscribe_to_stack_events(stack_id) + connection_opts = Keyword.merge(ctx.pooled_db_config, List.wrap(ctx[:connection_opt_overrides])) @@ -492,19 +549,7 @@ defmodule Support.ComponentSetup do # potential CI slowness, including PG assert_receive {:stack_status, ^ref, :ready}, 2000 - %{ - stack_id: stack_id, - registry: Electric.StackSupervisor.registry_name(stack_id), - stack_events_registry: stack_events_registry, - shape_cache: {ShapeCache, [stack_id: stack_id]}, - persistent_kv: kv, - stack_supervisor: stack_supervisor, - storage: storage, - inspector: - {EtsInspector, stack_id: stack_id, server: EtsInspector.name(stack_id: stack_id)}, - feature_flags: Electric.Config.get_env(:feature_flags), - publication_name: publication_name - } + stack_supervisor end def secure_mode(_ctx) do diff --git a/packages/sync-service/test/support/oracle_harness.ex b/packages/sync-service/test/support/oracle_harness.ex index fb382577c0..c9b5d92cf7 100644 --- a/packages/sync-service/test/support/oracle_harness.ex +++ b/packages/sync-service/test/support/oracle_harness.ex @@ -56,9 +56,7 @@ defmodule Support.OracleHarness do # ---------------------------------------------------------------------------- def default_opts_from_env do - %{ - oracle_pool_size: env_int("ORACLE_POOL_SIZE") || @default_oracle_pool_size - } + [oracle_pool_size: env_int("ORACLE_POOL_SIZE") || @default_oracle_pool_size] end @doc """ @@ -73,11 +71,17 @@ defmodule Support.OracleHarness do - :oracle_pool_size - number of parallel oracle connections (default: 50, env: ORACLE_POOL_SIZE) - :timeout_ms - timeout for waiting on shapes (default: 10_000) + - :restart_server_every - restart the StackSupervisor every N batches to + exercise restore-from-disk (default: 0, disabled) + - :restart_client_every - throw away and recreate the shape clients every + M batches to exercise fresh-poll consistency (default: 0, disabled) """ - @spec test_against_oracle(map(), [shape()], [batch()], map()) :: :ok - def test_against_oracle(ctx, shapes, batches, opts \\ %{}) do - opts = Map.merge(default_opts_from_env(), opts) + @spec test_against_oracle(map(), [shape()], [batch()], keyword()) :: :ok + def test_against_oracle(ctx, shapes, batches, opts \\ []) do + opts = Keyword.merge(default_opts_from_env(), opts) timeout_ms = opts[:timeout_ms] || env_int("CHECK_TIMEOUT") || @default_timeout_ms + restart_server_every = opts[:restart_server_every] || 0 + restart_client_every = opts[:restart_client_every] || 0 log_test_config(shapes, batches) @@ -90,21 +94,46 @@ defmodule Support.OracleHarness do # Check initial state (all in parallel) log("Checking initial snapshot for #{length(pids)} shapes") - pids - |> Task.async_stream(&ShapeChecker.check_initial_state/1, timeout: :infinity) - |> Stream.run() + check_initial_states(pids) # Run each batch log("Running #{length(batches)} batches") - batches - |> Enum.with_index(1) - |> Enum.each(fn {transactions, batch_idx} -> - run_batch(ctx, pids, transactions, batch_idx) - end) + total_batches = length(batches) + + {final_pids, _ctx} = + batches + |> Enum.with_index(1) + |> Enum.reduce({pids, ctx}, fn {transactions, batch_idx}, {pids, ctx} -> + run_batch(ctx, pids, transactions, batch_idx) + + restart_server? = + restart_server_every > 0 and rem(batch_idx, restart_server_every) == 0 and + batch_idx < total_batches + + restart_client? = + restart_client_every > 0 and rem(batch_idx, restart_client_every) == 0 and + batch_idx < total_batches + + cond do + restart_server? -> + # restart_server tears down the old checkers (they're polling the + # server about to go down) and recreates them after the stack is + # back up. If a client restart is also due this batch, it's + # subsumed by the recreate that follows the server restart. + restart_server(ctx, pids, shapes, oracle_pool, timeout_ms, batch_idx) + + restart_client? -> + new_pids = recreate_checkers(ctx, pids, shapes, oracle_pool, timeout_ms, batch_idx) + {new_pids, ctx} + + true -> + {pids, ctx} + end + end) # Cleanup - Enum.each(pids, &GenServer.stop/1) + Enum.each(final_pids, &GenServer.stop/1) GenServer.stop(oracle_pool) :ok end @@ -134,6 +163,40 @@ defmodule Support.OracleHarness do end) end + defp check_initial_states(pids) do + pids + |> Task.async_stream(&ShapeChecker.check_initial_state/1, timeout: :infinity) + |> Stream.run() + end + + # Restarts the Electric stack (server-side restore-from-file test) and + # reconnects the clients. Old checkers are stopped because their polls are + # against the server that is about to go down. + defp restart_server(ctx, pids, shapes, oracle_pool, timeout_ms, batch_idx) do + log("Restarting server after batch_#{batch_idx}") + + Enum.each(pids, &GenServer.stop/1) + + new_ctx = Map.merge(ctx, Support.ComponentSetup.restart_complete_stack(ctx)) + + new_pids = recreate_checkers(new_ctx, [], shapes, oracle_pool, timeout_ms, batch_idx) + + {new_pids, new_ctx} + end + + # Throws away the existing checkers and creates fresh ones (client-side + # resync test). The new checkers do an initial snapshot poll and assert + # consistency against the oracle. + defp recreate_checkers(ctx, old_pids, shapes, oracle_pool, timeout_ms, batch_idx) do + log("Recreating clients after batch_#{batch_idx}") + + Enum.each(old_pids, &GenServer.stop/1) + + new_pids = start_checkers(ctx, shapes, oracle_pool, timeout_ms) + check_initial_states(new_pids) + new_pids + end + defp run_batch(ctx, pids, transactions, batch_idx) do batch_name = "batch_#{batch_idx}" total_mutations = transactions |> Enum.map(&length/1) |> Enum.sum()