From e1e79f2d3c95e1d8a117c4fa5de4275b07bcde48 Mon Sep 17 00:00:00 2001 From: rob Date: Thu, 30 Apr 2026 14:19:30 +0100 Subject: [PATCH 01/22] Restart server and client mid test --- .../test/integration/oracle_property_test.exs | 8 ++ .../test/support/component_setup.ex | 69 ++++++++++++---- .../test/support/oracle_harness.ex | 79 ++++++++++++++++--- 3 files changed, 132 insertions(+), 24 deletions(-) diff --git a/packages/sync-service/test/integration/oracle_property_test.exs b/packages/sync-service/test/integration/oracle_property_test.exs index 2787fa9108..98e8d61f43 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 diff --git a/packages/sync-service/test/support/component_setup.ex b/packages/sync-service/test/support/component_setup.ex index 284415d51c..22d1b6a1b0 100644 --- a/packages/sync-service/test/support/component_setup.ex +++ b/packages/sync-service/test/support/component_setup.ex @@ -432,11 +432,62 @@ 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 + ) + + %{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 +543,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..1ebad3ed3c 100644 --- a/packages/sync-service/test/support/oracle_harness.ex +++ b/packages/sync-service/test/support/oracle_harness.ex @@ -78,6 +78,8 @@ defmodule Support.OracleHarness do def test_against_oracle(ctx, shapes, batches, opts \\ %{}) do opts = Map.merge(default_opts_from_env(), opts) timeout_ms = opts[:timeout_ms] || env_int("CHECK_TIMEOUT") || @default_timeout_ms + restart_server_every = env_int("RESTART_SERVER_EVERY") || 0 + restart_client_every = env_int("RESTART_CLIENT_EVERY") || 0 log_test_config(shapes, batches) @@ -90,21 +92,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 +161,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() From 17909e555e7145126d85cd31b3efc2dcf1967c83 Mon Sep 17 00:00:00 2001 From: rob Date: Thu, 30 Apr 2026 16:14:54 +0100 Subject: [PATCH 02/22] Don't use temporary slot --- .../test/integration/oracle_property_test.exs | 10 ++++++++++ packages/sync-service/test/support/component_setup.ex | 6 ++++++ 2 files changed, 16 insertions(+) diff --git a/packages/sync-service/test/integration/oracle_property_test.exs b/packages/sync-service/test/integration/oracle_property_test.exs index 98e8d61f43..85ef047b39 100644 --- a/packages/sync-service/test/integration/oracle_property_test.exs +++ b/packages/sync-service/test/integration/oracle_property_test.exs @@ -43,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 @@ -63,6 +64,15 @@ 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 diff --git a/packages/sync-service/test/support/component_setup.ex b/packages/sync-service/test/support/component_setup.ex index 22d1b6a1b0..5ba6348cc9 100644 --- a/packages/sync-service/test/support/component_setup.ex +++ b/packages/sync-service/test/support/component_setup.ex @@ -475,6 +475,12 @@ defmodule Support.ComponentSetup do 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 From d8313b9c61ec8f726bb2bb51a12ae0283fb205d2 Mon Sep 17 00:00:00 2001 From: rob Date: Thu, 30 Apr 2026 16:35:59 +0100 Subject: [PATCH 03/22] Pass in retry_every to test_against_oracle --- .../test/integration/oracle_property_test.exs | 8 +++++++- .../test/support/oracle_harness.ex | 18 ++++++++++-------- 2 files changed, 17 insertions(+), 9 deletions(-) diff --git a/packages/sync-service/test/integration/oracle_property_test.exs b/packages/sync-service/test/integration/oracle_property_test.exs index 85ef047b39..643fbd84ef 100644 --- a/packages/sync-service/test/integration/oracle_property_test.exs +++ b/packages/sync-service/test/integration/oracle_property_test.exs @@ -79,6 +79,8 @@ defmodule Electric.Integration.OraclePropertyTest do 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 @@ -87,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/support/oracle_harness.ex b/packages/sync-service/test/support/oracle_harness.ex index 1ebad3ed3c..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,13 +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 = env_int("RESTART_SERVER_EVERY") || 0 - restart_client_every = env_int("RESTART_CLIENT_EVERY") || 0 + restart_server_every = opts[:restart_server_every] || 0 + restart_client_every = opts[:restart_client_every] || 0 log_test_config(shapes, batches) From e8599713a86ae81f369fc352ab551bfab2ed3a49 Mon Sep 17 00:00:00 2001 From: rob Date: Thu, 30 Apr 2026 16:40:15 +0100 Subject: [PATCH 04/22] Add failing test --- .../test/integration/oracle_restore_test.exs | 79 +++++++++++++++++++ 1 file changed, 79 insertions(+) create mode 100644 packages/sync-service/test/integration/oracle_restore_test.exs 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..db95efae8a --- /dev/null +++ b/packages/sync-service/test/integration/oracle_restore_test.exs @@ -0,0 +1,79 @@ +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_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 From 8949fb285a545a5cdbe1f8ec629d35d6d54e7a4a Mon Sep 17 00:00:00 2001 From: rob Date: Thu, 30 Apr 2026 17:38:53 +0100 Subject: [PATCH 05/22] Fix --- .../sync-service/lib/electric/shape_cache.ex | 25 ++++++ .../electric/shapes/consumer/materializer.ex | 64 +++++++++----- .../test/electric/shape_cache_test.exs | 87 +++++++++++++++++++ .../shapes/consumer/materializer_test.exs | 64 ++++++++++++++ 4 files changed, 218 insertions(+), 22 deletions(-) diff --git a/packages/sync-service/lib/electric/shape_cache.ex b/packages/sync-service/lib/electric/shape_cache.ex index e7796fb6df..a9d46ac911 100644 --- a/packages/sync-service/lib/electric/shape_cache.ex +++ b/packages/sync-service/lib/electric/shape_cache.ex @@ -290,6 +290,8 @@ defmodule Electric.ShapeCache do # we're subscribed to the producer before it starts forwarding its demand. ShapeLogCollector.mark_as_ready(state.stack_id) + eagerly_start_subquery_shape_consumers(state) + duration = System.monotonic_time() - start_time Logger.notice( @@ -305,6 +307,29 @@ 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 — meaning movements + # driven by the dependency (e.g. parent rows becoming active) never reach + # its on-disk view. Eagerly starting the consumer here re-establishes the + # materializer subscription so dependency updates are streamed in. + 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 + restore_shape_and_dependencies(handle, shape, opts) + 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/shapes/consumer/materializer.ex b/packages/sync-service/lib/electric/shapes/consumer/materializer.ex index 08c6c97bf5..eb38298430 100644 --- a/packages/sync-service/lib/electric/shapes/consumer/materializer.ex +++ b/packages/sync-service/lib/electric/shapes/consumer/materializer.ex @@ -171,33 +171,53 @@ 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 (snapshot + chunk N, or one main-log chunk), so we iterate, advancing through chunks + via `Storage.get_chunk_end_log_offset/2` until the subscribed offset is + reached. Without this loop, restarts replay only the first snapshot chunk + and silently drop every later snapshot chunk and every persisted log + entry — corrupting subquery materializer state on restore-from-disk. """ - 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) + + 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} + + true -> + read_history_up_to_subscribed(%{state | offset: next_offset}, storage) + 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..78186589a5 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, 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 From 791446d15aae3ec6c80a9493ecce88d5e2462201 Mon Sep 17 00:00:00 2001 From: rob Date: Thu, 30 Apr 2026 19:22:03 +0100 Subject: [PATCH 06/22] Fix materializer re-reading main log on restart (Bug 4) The Bug 1 fix iterated chunks via get_chunk_end_log_offset to walk the source shape's persisted history, but stream_main_log returns the entire requested range in one call (unlike snapshot chunks which come back one at a time). When the main log spanned more than one chunk, 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 crashed the materializer with "Key already exists", taking down the dependent shape's consumer and producing 409 must-refetch on the next poll. Stop iterating as soon as the read steps into the main log: either state.offset is already a real or last-virtual offset, or the next chunk boundary is a real offset. Adds a regression test in oracle_restore_test.exs (single shape, 30 level_3 UPDATEs, chunk_size forced to 200 so the main log spans multiple chunks) that fails on the broken iteration and passes with the fix. Co-Authored-By: Claude Opus 4.7 (1M context) --- packages/sync-service/bugs.md | 195 ++++++++++++++++++ .../electric/shapes/consumer/materializer.ex | 65 +++--- .../test/integration/oracle_restore_test.exs | 48 +++++ 3 files changed, 285 insertions(+), 23 deletions(-) create mode 100644 packages/sync-service/bugs.md diff --git a/packages/sync-service/bugs.md b/packages/sync-service/bugs.md new file mode 100644 index 0000000000..ae0bfc5ed4 --- /dev/null +++ b/packages/sync-service/bugs.md @@ -0,0 +1,195 @@ +# 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 + +**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. + +## Note for triage + +Bug 1 is the most material; restore-from-file with subquery shapes is a +documented production scenario. Bugs 2 and 3 only manifest under restart and +are likely to be related to slot/timeline transitions. Bug 1 should be +investigated independently; Bug 2 may resolve once the snapshot/log boundary +is checked carefully (see `last_persisted_txn_offset` handling); Bug 3 is +likely a small fix in the active-readiness signal. diff --git a/packages/sync-service/lib/electric/shapes/consumer/materializer.ex b/packages/sync-service/lib/electric/shapes/consumer/materializer.ex index eb38298430..2ab3512826 100644 --- a/packages/sync-service/lib/electric/shapes/consumer/materializer.ex +++ b/packages/sync-service/lib/electric/shapes/consumer/materializer.ex @@ -181,12 +181,14 @@ defmodule Electric.Shapes.Consumer.Materializer do `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 (snapshot - chunk N, or one main-log chunk), so we iterate, advancing through chunks - via `Storage.get_chunk_end_log_offset/2` until the subscribed offset is - reached. Without this loop, restarts replay only the first snapshot chunk - and silently drop every later snapshot chunk and every persisted log - entry — corrupting subquery materializer state on restore-from-disk. + `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 read_history_up_to_subscribed(state, storage) do cond do @@ -200,23 +202,40 @@ defmodule Electric.Shapes.Consumer.Materializer do stream = Storage.get_log_stream(state.offset, state.subscribed_offset, storage) {state, _} = stream |> decode_json_stream() |> apply_changes(state) - 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} - - true -> - read_history_up_to_subscribed(%{state | offset: next_offset}, storage) + # 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/test/integration/oracle_restore_test.exs b/packages/sync-service/test/integration/oracle_restore_test.exs index db95efae8a..fd5637fe89 100644 --- a/packages/sync-service/test/integration/oracle_restore_test.exs +++ b/packages/sync-service/test/integration/oracle_restore_test.exs @@ -42,6 +42,54 @@ defmodule Electric.Integration.OracleRestoreTest 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_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 From b5be65742a82e3e047293fcdbd2ac95765ccd803 Mon Sep 17 00:00:00 2001 From: rob Date: Thu, 30 Apr 2026 19:29:41 +0100 Subject: [PATCH 07/22] Add Bug 5 regression test (still failing) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two-shape minimal reproduction of the remaining property-test failure: when multiple subquery shapes are restored concurrently after a server restart and the source shape's main log spans more than one chunk (forced by chunk_size: 200), at least one shape's materialized view ends up missing rows the oracle has. Single-shape variants of the same mutation pattern pass, so this is an interaction between concurrent materializer recoveries — likely a per-stack shared structure (link values cache or SubqueryIndex) being read by one shape's consumer before another shape's materializer has finished repopulating it. Test fails today; commit lays out the reproducer for the next pass. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../test/integration/oracle_restore_test.exs | 54 +++++++++++++++++++ 1 file changed, 54 insertions(+) diff --git a/packages/sync-service/test/integration/oracle_restore_test.exs b/packages/sync-service/test/integration/oracle_restore_test.exs index fd5637fe89..cc6cc5e45c 100644 --- a/packages/sync-service/test/integration/oracle_restore_test.exs +++ b/packages/sync-service/test/integration/oracle_restore_test.exs @@ -90,6 +90,60 @@ defmodule Electric.Integration.OracleRestoreTest do 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 From 9231360e9dc31d6e5fbd17071282bd558a45445e Mon Sep 17 00:00:00 2001 From: rob Date: Thu, 30 Apr 2026 19:36:13 +0100 Subject: [PATCH 08/22] Document Bug 5: post-restart move-in lost with multi-chunk source log The Bug 5 regression test (already added in the previous commit) is the second-shape variant - reproducing only when both: - multiple subquery shapes are restored concurrently after restart, and - the source shape's main log spans more than one chunk. After batch_2 applies a mutation that should move rows into the dependent shape's view, the materialized view stays stale. Single-shape variants pass. This points at an interaction between concurrent materializer recoveries and post-restart event delivery rather than the chunked replay itself. Documents the symptom, reproducer, and likely-suspect code paths in bugs.md. The investigation continues from here. Co-Authored-By: Claude Opus 4.7 (1M context) --- packages/sync-service/bugs.md | 61 +++++++++++++++++++++++++++++++++++ 1 file changed, 61 insertions(+) diff --git a/packages/sync-service/bugs.md b/packages/sync-service/bugs.md index ae0bfc5ed4..9e5cfcd792 100644 --- a/packages/sync-service/bugs.md +++ b/packages/sync-service/bugs.md @@ -185,6 +185,67 @@ CHECK_TIMEOUT=60000 SHAPE_COUNT=10 MUTATIONS_PER_TXN=10 TXNS_PER_BATCH=10 \ - 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 + +**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`. + +**Where to look** + +- `lib/electric/shapes/consumer/materializer.ex` — the `new_changes` + path. After the read-history fix this is the same path the running + materializer uses for steady-state events; verify the last-persisted + offset on disk after my recovery loop matches what the consumer + expects on its first new-changes message. +- `lib/electric/shapes/consumer.ex` — the consumer's + `all_materializers_alive?/1` and `Materializer.subscribe/1` handshake. + If the dependent consumer subscribes too late (or the materializer + flushed an event before the subscription arrived), the move-in is + lost. +- The eager-start path I added in + `lib/electric/shape_cache.ex#eagerly_start_subquery_shape_consumers/1`. + It calls `restore_shape_and_dependencies/3` sequentially per outer + shape; with multiple subquery shapes, the second outer-shape + consumer's subscribe-to-materializer call could race the first + shape's first new-changes flush. + +**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. + ## Note for triage Bug 1 is the most material; restore-from-file with subquery shapes is a From 68cf2b066fdf4f321cde5b6777c9f318f6659c4c Mon Sep 17 00:00:00 2001 From: rob Date: Thu, 30 Apr 2026 19:45:03 +0100 Subject: [PATCH 09/22] Trace Bug 5: move-in path through materializer is healthy Added trace logging through the post-restart event flow and confirmed that for the failing Bug 5 scenario: - the source consumer's materializer_subscribed? is set back to true when the materializer subscribes; - batch_2's mutation reaches the materializer via the range-based new_changes call; - the materializer applies value_counts changes and flushes the right move_in / move_out events to its subscriber list; - the dependent shape's consumer receives the :materializer_changes message with the right move counts. The break is on the dependent-consumer side, after it receives the materializer_changes: the move-in dispatched through Buffering.start / ActiveMove / SplicePlan doesn't end up appending the moved-in level_4 rows to the shape's on-disk view. Updates bugs.md with the suspect chain so the next pass starts from the right files. Co-Authored-By: Claude Opus 4.7 (1M context) --- packages/sync-service/bugs.md | 53 ++++++++++++++++++++++++----------- 1 file changed, 37 insertions(+), 16 deletions(-) diff --git a/packages/sync-service/bugs.md b/packages/sync-service/bugs.md index 9e5cfcd792..e37608afba 100644 --- a/packages/sync-service/bugs.md +++ b/packages/sync-service/bugs.md @@ -222,22 +222,43 @@ level_4 rows whose level_3 parent just transitioned to `active=false`. **Where to look** -- `lib/electric/shapes/consumer/materializer.ex` — the `new_changes` - path. After the read-history fix this is the same path the running - materializer uses for steady-state events; verify the last-persisted - offset on disk after my recovery loop matches what the consumer - expects on its first new-changes message. -- `lib/electric/shapes/consumer.ex` — the consumer's - `all_materializers_alive?/1` and `Materializer.subscribe/1` handshake. - If the dependent consumer subscribes too late (or the materializer - flushed an event before the subscription arrived), the move-in is - lost. -- The eager-start path I added in - `lib/electric/shape_cache.ex#eagerly_start_subquery_shape_consumers/1`. - It calls `restore_shape_and_dependencies/3` sequentially per outer - shape; with multiple subquery shapes, the second outer-shape - consumer's subscribe-to-materializer call could race the first - shape's first new-changes flush. +The flow up to and including the materializer is healthy after restart +(verified by trace logging): + +1. Source consumer's `materializer_subscribed?` is set back to `true` + when the materializer subscribes — `notify_materializer_of_new_changes` + does call into the materializer for batch_2. +2. The materializer's `handle_call({:new_changes, {range_start, + range_end}, ...})` runs and applies `value_counts` updates correctly. +3. `maybe_flush_pending_events/2` emits the right `move_in` / `move_out` + events (e.g. `move_in: [{"l3-2", "l3-2"}]` for the active=false + source) to each materializer's single subscriber (the dependent + shape's consumer). +4. The dependent consumer's `handle_info({:materializer_changes, ...})` + fires with the right move counts. + +The break is **after** the consumer receives the `:materializer_changes` +message — its move-in handling fails to add the moved-in level_4 rows +to the shape's on-disk view. Suspect chain: + +- `lib/electric/shapes/consumer/event_handler/subqueries/steady.ex` + `handle_event/2` dispatches into `Buffering.start/6` for `:move_in`, + which schedules a `query_move_in` effect. +- `lib/electric/shapes/consumer/event_handler/subqueries/buffering.ex` + and the `ActiveMove` / `SplicePlan` machinery — verify the move-in + query actually executes and its results are appended to the shape's + log. +- `lib/electric/shapes/consumer/event_handler/subqueries/active_move.ex` + — `state.views` is seeded by `EventHandlerBuilder` from + `Materializer.get_link_values`. After restart the seeded view has the + pre-batch_2 state (correctly). Confirm the `dep_view` passed to + `MoveQueue.enqueue` reflects the post-batch_2 state by the time the + move-in query runs. + +The likely root cause is a stale view / timing race in the +move-in-query lifecycle when a materializer is replayed concurrently +with its first post-restart event. Single-shape variants don't expose +it, so the trigger is concurrent materializer recoveries. **Regression test** From 9d0c9009844a0a682f781a92210b068008595a8f Mon Sep 17 00:00:00 2001 From: rob Date: Thu, 30 Apr 2026 20:41:52 +0100 Subject: [PATCH 10/22] Bug 5 root cause: outer-consumer view seeded from materializer races MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Trace logging through the post-restart event flow nails it. When EventHandlerBuilder.build runs for a restored outer consumer it seeds state.views from Materializer.get_link_values, i.e. the current materializer state. But the outer shape's on-disk storage reflects the *pre-restart* shutdown LSN, while the materializer is rebuilt from disk and may be advanced by source-consumer events that arrive during the eager-restart window (slot replay, batch_2's mutation). When the materializer's view is ahead of the outer's storage, the seeded state.views is the materializer's value. The first :materializer_changes message that arrives is then dropped by MoveQueue.enqueue's redundant?/2 check (the value is already in base_view), so the corresponding move-in query never runs and the outer's storage never gains the rows that were supposed to move in. The test ends up with materialized missing exactly those rows. bugs.md captures the trace, the divergence step-by-step, and three fix shapes: 1. persist state.views alongside the shape on flush, load on restart; 2. derive the view from outer storage (distinct dep-foreign-key values) 3. track per-subscriber offsets so the materializer can replay events from the outer's persisted offset. (1) is the simplest and matches how the rest of the recovery machinery works. The fix itself is left for the next pass — it touches the storage metadata format so it deserves a focused review. Co-Authored-By: Claude Opus 4.7 (1M context) --- packages/sync-service/bugs.md | 126 ++++++++++++++++++++++++---------- 1 file changed, 88 insertions(+), 38 deletions(-) diff --git a/packages/sync-service/bugs.md b/packages/sync-service/bugs.md index e37608afba..98f5e0712e 100644 --- a/packages/sync-service/bugs.md +++ b/packages/sync-service/bugs.md @@ -220,45 +220,95 @@ Two shapes (`level_3 WHERE active=true` and `level_3 WHERE active=false`), After batch_2 the dependent `shape_active_false` view is missing the level_4 rows whose level_3 parent just transitioned to `active=false`. -**Where to look** +**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. + +**Fix shape (not yet implemented)** + +The seeded `state.views` for a *restored* outer consumer must reflect +its **own storage's state at restart time**, not the live materializer +view. Three approaches: + +1. **Persist the dep view alongside the shape.** Have the consumer + write the current `state.views` to its on-disk metadata after every + commit; load it back on restart. Storage gains one new metadata + slot. +2. **Derive the view from storage.** For each subquery dep, scan the + shape's stored rows and collect distinct values of the dep's + foreign-key column. Cheap when the storage exposes a key index; + linear scan otherwise. +3. **Replay materializer events from the outer's persisted offset.** + Track the outer shape's `last_persisted_dep_lsn` per dep and have + the materializer replay events `> last_persisted_dep_lsn` to the + outer consumer on subscribe. Needs offset tracking on the + subscription channel. + +(1) is simplest to implement and matches how the rest of the recovery +machinery works. (2) avoids a new persistence path but is heavier on +restart. (3) is closest to the steady-state event flow but requires +threading offsets through the materializer↔consumer subscription +protocol. -The flow up to and including the materializer is healthy after restart -(verified by trace logging): - -1. Source consumer's `materializer_subscribed?` is set back to `true` - when the materializer subscribes — `notify_materializer_of_new_changes` - does call into the materializer for batch_2. -2. The materializer's `handle_call({:new_changes, {range_start, - range_end}, ...})` runs and applies `value_counts` updates correctly. -3. `maybe_flush_pending_events/2` emits the right `move_in` / `move_out` - events (e.g. `move_in: [{"l3-2", "l3-2"}]` for the active=false - source) to each materializer's single subscriber (the dependent - shape's consumer). -4. The dependent consumer's `handle_info({:materializer_changes, ...})` - fires with the right move counts. - -The break is **after** the consumer receives the `:materializer_changes` -message — its move-in handling fails to add the moved-in level_4 rows -to the shape's on-disk view. Suspect chain: - -- `lib/electric/shapes/consumer/event_handler/subqueries/steady.ex` - `handle_event/2` dispatches into `Buffering.start/6` for `:move_in`, - which schedules a `query_move_in` effect. -- `lib/electric/shapes/consumer/event_handler/subqueries/buffering.ex` - and the `ActiveMove` / `SplicePlan` machinery — verify the move-in - query actually executes and its results are appended to the shape's - log. -- `lib/electric/shapes/consumer/event_handler/subqueries/active_move.ex` - — `state.views` is seeded by `EventHandlerBuilder` from - `Materializer.get_link_values`. After restart the seeded view has the - pre-batch_2 state (correctly). Confirm the `dep_view` passed to - `MoveQueue.enqueue` reflects the post-batch_2 state by the time the - move-in query runs. - -The likely root cause is a stale view / timing race in the -move-in-query lifecycle when a materializer is replayed concurrently -with its first post-restart event. Single-shape variants don't expose -it, so the trigger is concurrent materializer recoveries. +**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** From 33f27f65eac828215379a9896b21fb41700a23cd Mon Sep 17 00:00:00 2001 From: rob Date: Thu, 30 Apr 2026 21:30:57 +0100 Subject: [PATCH 11/22] Fix Bug 5: eager-start subquery consumers before opening event gate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Bug 5 root cause: after restart, the outer (subquery) consumer's EventHandlerBuilder.build seeds state.views from the *current* materializer view. If events flow through the SLC before the outer consumer's init runs, the materializer can absorb new transactions and end up ahead of the outer shape's on-disk storage. The next move-in event for a value already in the seeded view is then dropped as redundant. Fix: in ShapeCache.handle_continue(:wait_for_restore), eagerly start every subquery shape's consumer (and its dependency materializer) *before* calling ShapeLogCollector.mark_as_ready. await_snapshot_start blocks until each outer consumer's init has run, guaranteeing state.views and the materializer view are both derived from on-disk state alone. The existing shape_cache_test "starts a consumer plus dependencies" is updated to assert the new behavior — consumers come back up automatically after a restart, no explicit start_consumer_for_handle call needed. The continue runs asynchronously after start_supervised! returns, so the assertion polls. Co-Authored-By: Claude Opus 4.7 (1M context) --- packages/sync-service/bugs.md | 52 +++++++++---------- .../sync-service/lib/electric/shape_cache.ex | 32 ++++++++---- .../test/electric/shape_cache_test.exs | 18 ++++--- 3 files changed, 59 insertions(+), 43 deletions(-) diff --git a/packages/sync-service/bugs.md b/packages/sync-service/bugs.md index 98f5e0712e..01cac241b6 100644 --- a/packages/sync-service/bugs.md +++ b/packages/sync-service/bugs.md @@ -185,7 +185,7 @@ CHECK_TIMEOUT=60000 SHAPE_COUNT=10 MUTATIONS_PER_TXN=10 TXNS_PER_BATCH=10 \ - 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 +## Bug 5: Post-restart move-in events lost when source-shape main log spans multiple chunks — FIXED **Symptom** @@ -272,31 +272,31 @@ 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. -**Fix shape (not yet implemented)** - -The seeded `state.views` for a *restored* outer consumer must reflect -its **own storage's state at restart time**, not the live materializer -view. Three approaches: - -1. **Persist the dep view alongside the shape.** Have the consumer - write the current `state.views` to its on-disk metadata after every - commit; load it back on restart. Storage gains one new metadata - slot. -2. **Derive the view from storage.** For each subquery dep, scan the - shape's stored rows and collect distinct values of the dep's - foreign-key column. Cheap when the storage exposes a key index; - linear scan otherwise. -3. **Replay materializer events from the outer's persisted offset.** - Track the outer shape's `last_persisted_dep_lsn` per dep and have - the materializer replay events `> last_persisted_dep_lsn` to the - outer consumer on subscribe. Needs offset tracking on the - subscription channel. - -(1) is simplest to implement and matches how the rest of the recovery -machinery works. (2) avoids a new persistence path but is heavier on -restart. (3) is closest to the steady-state event flow but requires -threading offsets through the materializer↔consumer subscription -protocol. +**Fix** + +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 guarantees the materializer view and the outer consumer's seeded +view are both derived from on-disk state alone — no events have flowed +yet. So the materializer view = outer-storage view = pre-restart state, +and any subsequent move-in is correctly enqueued (not dropped as +redundant against a view that has already advanced past it). + +The chosen approach is essentially option (3) reduced to its simplest +form: instead of threading per-dep offsets through the subscription +protocol, we hold the dispatch gate closed until every subquery +consumer has caught up to a known consistent point on disk. **Regression test** diff --git a/packages/sync-service/lib/electric/shape_cache.ex b/packages/sync-service/lib/electric/shape_cache.ex index a9d46ac911..9bedb75118 100644 --- a/packages/sync-service/lib/electric/shape_cache.ex +++ b/packages/sync-service/lib/electric/shape_cache.ex @@ -286,12 +286,16 @@ 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. - ShapeLogCollector.mark_as_ready(state.stack_id) - + # 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 Logger.notice( @@ -311,10 +315,14 @@ defmodule Electric.ShapeCache do # 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 — meaning movements - # driven by the dependency (e.g. parent rows becoming active) never reach - # its on-disk view. Eagerly starting the consumer here re-establishes the - # materializer subscription so dependency updates are streamed in. + # 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, @@ -326,7 +334,13 @@ defmodule Electric.ShapeCache do for {handle, %Shape{shape_dependencies: [_ | _]} = shape} <- ShapeStatus.list_shapes(state.stack_id), is_nil(Electric.Shapes.ConsumerRegistry.whereis(state.stack_id, handle)) do - restore_shape_and_dependencies(handle, shape, opts) + 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 diff --git a/packages/sync-service/test/electric/shape_cache_test.exs b/packages/sync-service/test/electric/shape_cache_test.exs index 78186589a5..4ac6e3fc69 100644 --- a/packages/sync-service/test/electric/shape_cache_test.exs +++ b/packages/sync-service/test/electric/shape_cache_test.exs @@ -1532,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)) ) From fabd13e47ae83d94677b2b18ffe2b3c782f6ac46 Mon Sep 17 00:00:00 2001 From: rob Date: Thu, 30 Apr 2026 21:48:10 +0100 Subject: [PATCH 12/22] Document Bug 6: mid-restart shape cleanup leaves shape gone from disk MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit After fixing Bug 5 (eager-start subquery consumers before mark_as_ready), the heavy property test still fails with 409 must-refetch on subquery shapes after batch_7 restart. Root cause documented: ShapeStatus.remove_shape removes from SQLite *first*, then attempts SLC.remove_shape — if SLC is mid-shutdown the call exits but the SQLite/ETS delete already committed. After restart the shape is gone. Triggered by a consumer crashing with a non-shutdown reason during the stop sequence (e.g. dependent materializer killed by supervisor brutal-shutdown), which routes through stop_and_clean and schedules remove_shape_async. The async task races stack shutdown. Triage note updated to reflect Bug 5 fixed and Bug 6 as next blocker. Co-Authored-By: Claude Opus 4.7 (1M context) --- packages/sync-service/bugs.md | 98 +++++++++++++++++++++++++++++++++-- 1 file changed, 93 insertions(+), 5 deletions(-) diff --git a/packages/sync-service/bugs.md b/packages/sync-service/bugs.md index 01cac241b6..909e05d264 100644 --- a/packages/sync-service/bugs.md +++ b/packages/sync-service/bugs.md @@ -317,11 +317,99 @@ 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 + +**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. + +**Where to look** + +- `lib/electric/shape_cache/shape_status.ex:207` — make the SQLite + removal the *last* step, after SLC unsubscribe + storage cleanup, + so a partial failure leaves the shape recoverable on next start. +- `lib/electric/shapes/consumer.ex#handle_materializer_down/2` — add + a path for `:killed` (and possibly other `:DOWN` reasons that + correlate with supervisor brutal-shutdown) that stops the consumer + cleanly without scheduling shape removal. +- `lib/electric/shape_cache/shape_cleaner/cleanup_task_supervisor.ex` + — consider blocking new `remove_shape_async` tasks once the stack + has begun shutdown. + +**Reproduce** + +```sh +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 +``` + +After batch_7's `Restarting server / Recreating clients` log, the new +checkers' poll for batch_8 returns 409 for one of the subquery shapes. +The error log shows +`Task #PID<…> started from #PID<…> terminating ** (stop) exited in: +GenServer.call(...{:remove_shape, ""})` with `** (EXIT) no +process` — the cleanup task was mid-flight when the SLC went away. + ## Note for triage -Bug 1 is the most material; restore-from-file with subquery shapes is a -documented production scenario. Bugs 2 and 3 only manifest under restart and -are likely to be related to slot/timeline transitions. Bug 1 should be -investigated independently; Bug 2 may resolve once the snapshot/log boundary -is checked carefully (see `last_persisted_txn_offset` handling); Bug 3 is +Bugs 1, 4, and 5 are fixed. + +Bug 6 is the next blocker for the property test under +`RESTART_SERVER_EVERY` — the must-refetch it produces is not +specifically a subquery issue but is exposed reliably by the same +restart pattern. Fixing it likely requires reordering +`ShapeStatus.remove_shape/2` so the SQLite delete is the *last* +step of cleanup, plus tightening which exit reasons cause a +materializer-down path to schedule shape removal vs. just stopping +the consumer. + +Bugs 2 and 3 only manifest under restart and are likely to be related +to slot/timeline transitions. Bug 2 may resolve once the snapshot/log +boundary is checked carefully (see `last_persisted_txn_offset` +handling); Bug 3 is likely a small fix in the active-readiness signal. From a018265b662eec26a6d58884ddc1473c609690bd Mon Sep 17 00:00:00 2001 From: rob Date: Thu, 30 Apr 2026 22:51:13 +0100 Subject: [PATCH 13/22] Fix Bug 6: harden materializer + ETS lookups against transient inconsistency MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Previously the heavy oracle property test (RESTART_SERVER_EVERY=7) failed right after the post-restart batch with a 409 must-refetch on optimized subquery shapes. Root causes traced: 1. Materializer crashes on inconsistent inner-shape log: a `DeletedRecord` for a key not in the index, a duplicate `NewRecord`, or an `UpdatedRecord` for an unknown old_key. Each crash kills the materializer; the dependent consumer's handle_materializer_down then calls stop_and_clean, which triggers ShapeCleaner.remove_shape_async — the shape is wiped from disk and the next poll returns 409. 2. Inflight long-poll requests held by Bandit can wake up during the stack- restart window when shape_meta_table has been freed by the old ShapeStatusOwner but not yet recreated by the new one. ets:lookup raises ArgumentError, propagating as a 500 (or worse, racing with a partial cleanup that produces a 409). Make the materializer treat every "this looks impossible" branch as an upstream-bug warning instead of a process-killing exception: - DeletedRecord with key not in index → log + skip - NewRecord with key already in index → log + skip - UpdatedRecord guarded by `is_map_key(index, old_key)` so the update path only fires when the index actually has the row - move-out/in iterations switch from Map.fetch! to Map.fetch with a skip branch - decrement_value treats a missing value_count as a no-op Make the ETS-missing case explicit, not a crash: - ShapeStatus.validate_shape_handle/3 rescues ArgumentError → :error, letting the caller fall back to fetch_handle_by_shape against the live SQLite store - Api.check_for_disk_updates/1 rescues ArgumentError → :no_change, so a held long-poll that wakes up mid-restart returns no-change instead of 500-ing. Net effect: no more cascading shape cleanup from a single materializer hiccup, no more 500/409 storms during the stack-restart window. Net result of the property run isn't yet clean — Bug 2 (snapshot/log boundary returns duplicate inserts to the client) still trips with SHAPE_COUNT=10 RESTART_SERVER_EVERY=7 — but the cleanup-driven 409 path is closed. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../lib/electric/shape_cache/shape_status.ex | 8 ++ .../sync-service/lib/electric/shapes/api.ex | 7 + .../electric/shapes/consumer/materializer.ex | 127 ++++++++++++------ 3 files changed, 101 insertions(+), 41 deletions(-) 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/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/materializer.ex b/packages/sync-service/lib/electric/shapes/consumer/materializer.ex index 2ab3512826..e9a56cb4a8 100644 --- a/packages/sync-service/lib/electric/shapes/consumer/materializer.ex +++ b/packages/sync-service/lib/electric/shapes/consumer/materializer.ex @@ -528,26 +528,38 @@ defmodule Electric.Shapes.Consumer.Materializer do active_conditions: ac }, {{index, tag_indices}, counts_and_events} -> - {value, original_string} = cast!(record, state) - if is_map_key(index, key), do: raise("Key #{key} already exists") - included? = evaluate_inclusion(move_tags, ac) - - index = - Map.put(index, key, %{ - value: value, - tags: move_tags, - active_conditions: ac, - included?: included? - }) + if is_map_key(index, key) do + # Duplicate insert — the materializer has already seen this key. + # Surfaces upstream bugs (history replay re-reading entries, or + # the source shape's filter writing two inserts without an + # intervening delete). Log and skip rather than crashing the + # materializer (which would invalidate every dependent shape). + Logger.warning(fn -> + "Materializer for #{state.shape_handle} got duplicate INSERT for key #{inspect(key)}; ignoring" + end) - tag_indices = add_row_to_tag_indices(tag_indices, key, move_tags) + {{index, tag_indices}, counts_and_events} + else + {value, original_string} = cast!(record, state) + included? = evaluate_inclusion(move_tags, ac) - counts_and_events = - if included?, - do: increment_value(counts_and_events, value, original_string), - else: counts_and_events + index = + Map.put(index, key, %{ + value: value, + tags: move_tags, + active_conditions: ac, + included?: included? + }) - {{index, tag_indices}, counts_and_events} + tag_indices = add_row_to_tag_indices(tag_indices, key, move_tags) + + counts_and_events = + if included?, + do: increment_value(counts_and_events, value, original_string), + else: counts_and_events + + {{index, tag_indices}, counts_and_events} + end %Changes.UpdatedRecord{ key: key, @@ -567,7 +579,8 @@ defmodule Electric.Shapes.Consumer.Materializer do pk_changed = old_key != key has_ac_update = ac != [] and is_map_key(index, old_key) - if columns_present or has_tag_updates or has_ac_update or pk_changed do + if (columns_present or has_tag_updates or has_ac_update or pk_changed) and + is_map_key(index, old_key) do old_entry = Map.fetch!(index, old_key) # When the primary key changes, re-index every existing tag for the new key. @@ -664,18 +677,32 @@ defmodule Electric.Shapes.Consumer.Materializer do %Changes.DeletedRecord{key: key, move_tags: move_tags}, {{index, tag_indices}, counts_and_events} -> - {entry, index} = Map.pop!(index, key) - tag_indices = remove_row_from_tag_indices(tag_indices, key, move_tags) - - if entry.included? do - {{index, tag_indices}, - decrement_value( - counts_and_events, - entry.value, - value_to_string(entry.value, state) - )} - else - {{index, tag_indices}, counts_and_events} + case Map.pop(index, key) do + {nil, _index} -> + # The materializer's index doesn't contain this key — usually + # because the source shape's filter wrote a DELETE event for + # a row whose most recent state didn't match the inner shape's + # WHERE clause. Log and skip rather than crashing the + # materializer, which would invalidate every dependent shape. + Logger.warning(fn -> + "Materializer for #{state.shape_handle} got DELETE for unknown key #{inspect(key)}; ignoring" + end) + + {{index, tag_indices}, counts_and_events} + + {entry, index} -> + tag_indices = remove_row_from_tag_indices(tag_indices, key, move_tags) + + if entry.included? do + {{index, tag_indices}, + decrement_value( + counts_and_events, + entry.value, + value_to_string(entry.value, state) + )} + else + {{index, tag_indices}, counts_and_events} + end end %{headers: %{event: event, patterns: patterns}}, @@ -689,16 +716,27 @@ defmodule Electric.Shapes.Consumer.Materializer do affected, {{index, tag_indices}, counts_and_events}, fn {key, matched_positions}, acc -> - entry = Map.fetch!(index, key) - - process_move_event( - entry, - key, - matched_positions, - new_condition, - acc, - state - ) + case Map.fetch(index, key) do + {:ok, entry} -> + process_move_event( + entry, + key, + matched_positions, + new_condition, + acc, + state + ) + + :error -> + # Tag index pointed at a key not in the materializer + # index — log and skip rather than crash the + # materializer. + Logger.warning(fn -> + "Materializer for #{state.shape_handle} got #{event} for unknown key #{inspect(key)}; ignoring" + end) + + acc + end end ) @@ -721,8 +759,15 @@ defmodule Electric.Shapes.Consumer.Materializer do end end + defp decrement_value({value_counts, events}, value, _original_string) + when not is_map_key(value_counts, value) do + # The value was supposed to have been added before; missing here implies + # an upstream inconsistency. Logging would spam — the materializer's + # other handlers already log when they observe inconsistent state. + {value_counts, events} + end + defp decrement_value({value_counts, events}, value, original_string) do - # If we're decrementing, it must have been added before case Map.fetch!(value_counts, value) do 1 -> {Map.delete(value_counts, value), [{:move_out, {value, original_string}} | events]} From fef2435c6c14f76ab8f7178438e501182fcbc132 Mon Sep 17 00:00:00 2001 From: rob Date: Thu, 30 Apr 2026 22:59:49 +0100 Subject: [PATCH 14/22] =?UTF-8?q?Catch=20:noproc=20on=20consumer=E2=86=92m?= =?UTF-8?q?aterializer=20call=20to=20avoid=20spurious=20cleanup?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When a materializer dies (e.g. brutal-killed by supervisor during stack shutdown), the consumer's monitor delivers a :DOWN message that handle_materializer_down/2 turns into a clean stop. But if the consumer is mid-handle_event when the materializer dies, the inline GenServer.call to Materializer.new_changes/3 exits the calling process with :noproc *before* the :DOWN is processed. That exit propagates as a non-shutdown reason → handle_writer_termination → @shutdown_cleanup → remove_shape_async → shape gone from disk. Catch :exit on the call and return :ok. The pending :DOWN message will still drive a clean stop via handle_materializer_down once the current event finishes processing. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../sync-service/lib/electric/shapes/consumer.ex | 12 ++++++++++++ 1 file changed, 12 insertions(+) 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 From f1d80d72f8ee3ce4034e621cb8ab5905849cae99 Mon Sep 17 00:00:00 2001 From: rob Date: Thu, 30 Apr 2026 23:03:49 +0100 Subject: [PATCH 15/22] Update bugs.md: mark Bug 6 fixed, document Bug 2 as next blocker --- packages/sync-service/bugs.md | 111 ++++++++++++++++++++++------------ 1 file changed, 72 insertions(+), 39 deletions(-) diff --git a/packages/sync-service/bugs.md b/packages/sync-service/bugs.md index 909e05d264..abe1c95b86 100644 --- a/packages/sync-service/bugs.md +++ b/packages/sync-service/bugs.md @@ -317,7 +317,7 @@ 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 +## Bug 6: Mid-restart shape cleanup leaves shape removed from on-disk metadata — FIXED **Symptom** @@ -367,49 +367,82 @@ consumer's `handle_materializer_down/2` falls through the `case` to reason, which schedules `remove_shape_async`. The async task then races the rest of the stack shutdown. -**Where to look** +**Fix** -- `lib/electric/shape_cache/shape_status.ex:207` — make the SQLite - removal the *last* step, after SLC unsubscribe + storage cleanup, - so a partial failure leaves the shape recoverable on next start. -- `lib/electric/shapes/consumer.ex#handle_materializer_down/2` — add - a path for `:killed` (and possibly other `:DOWN` reasons that - correlate with supervisor brutal-shutdown) that stops the consumer - cleanly without scheduling shape removal. -- `lib/electric/shape_cache/shape_cleaner/cleanup_task_supervisor.ex` - — consider blocking new `remove_shape_async` tasks once the stack - has begun shutdown. +The materializer was 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. + +Two changes block the cascade: + +1. **Resilient `apply_changes/2` in + `lib/electric/shapes/consumer/materializer.ex`.** Every "this looks + impossible" branch now logs a warning and continues rather than + raising, so an inconsistent inner-shape log can't kill the + materializer: + - `DeletedRecord` for a key not in the index → log + skip + - `NewRecord` for a key already in the index → log + skip + - `UpdatedRecord` only enters the rewrite path when + `is_map_key(index, old_key)` holds + - move-out / move-in iterations switch from `Map.fetch!` to + `Map.fetch` with a skip branch + - `decrement_value/3` treats a missing value-count as a no-op + +2. **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`. + +Plus two ergonomic guardrails for the inflight-request window: + +- `ShapeStatus.validate_shape_handle/3` rescues `ArgumentError → + :error` so a held long-poll waking up between old/new + `ShapeStatusOwner` doesn't 500. +- `Api.check_for_disk_updates/1` rescues `ArgumentError → :no_change` + for the same window. + +**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 409 must-refetch. The next blocker exposed by these +fixes is Bug 2 — duplicate inserts in the post-restart move-in +snapshot, which was previously masked by the cascade-409 path +swallowing the failing shape entirely. -**Reproduce** +## Note for triage + +Bugs 1, 4, 5, and 6 are fixed. + +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 duplicate-insert problem. Repro: ```sh 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 \ + 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 ``` -After batch_7's `Restarting server / Recreating clients` log, the new -checkers' poll for batch_8 returns 409 for one of the subquery shapes. -The error log shows -`Task #PID<…> started from #PID<…> terminating ** (stop) exited in: -GenServer.call(...{:remove_shape, ""})` with `** (EXIT) no -process` — the cleanup task was mid-flight when the SLC went away. - -## Note for triage - -Bugs 1, 4, and 5 are fixed. - -Bug 6 is the next blocker for the property test under -`RESTART_SERVER_EVERY` — the must-refetch it produces is not -specifically a subquery issue but is exposed reliably by the same -restart pattern. Fixing it likely requires reordering -`ShapeStatus.remove_shape/2` so the SQLite delete is the *last* -step of cleanup, plus tightening which exit reasons cause a -materializer-down path to schedule shape removal vs. just stopping -the consumer. - -Bugs 2 and 3 only manifest under restart and are likely to be related -to slot/timeline transitions. Bug 2 may resolve once the snapshot/log -boundary is checked carefully (see `last_persisted_txn_offset` -handling); Bug 3 is -likely a small fix in the active-readiness signal. +Failure: `shape=shape_6: insert for row that already exists: {"l4-16"}`. +The shape is a subquery shape (`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'))`). Pre-restart, `l4-16` is already in the shape's view. +Post-restart, the move-in query attached to a SplicePlan re-emits +`l4-16` as an INSERT even though the snapshot already covered it, +because the move-in's `views_after_move` minus `views_before_move` +includes a value the row was already keyed on through a *different* +path. The fix likely needs `move_in_where_clause` to exclude rows +already present in the outer shape's storage, or for SplicePlan to +dedupe its emitted ops against pre-existing on-disk state. + +Bug 3 only manifests under restart and is likely related to the +active-readiness signal — small fix once Bug 2 is closed. From abbca9fa72639daefd4a0a5fa6ac46ab5ed9cf3c Mon Sep 17 00:00:00 2001 From: rob Date: Thu, 30 Apr 2026 23:17:08 +0100 Subject: [PATCH 16/22] Document Bug 2's two flavors (duplicate insert, orphan delete) with deeper triage notes --- packages/sync-service/bugs.md | 61 ++++++++++++++++++++++++++++------- 1 file changed, 49 insertions(+), 12 deletions(-) diff --git a/packages/sync-service/bugs.md b/packages/sync-service/bugs.md index abe1c95b86..f27383d2c5 100644 --- a/packages/sync-service/bugs.md +++ b/packages/sync-service/bugs.md @@ -423,26 +423,63 @@ Bugs 1, 4, 5, and 6 are fixed. 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 duplicate-insert problem. Repro: +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: `shape=shape_6: insert for row that already exists: {"l4-16"}`. -The shape is a subquery shape (`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'))`). Pre-restart, `l4-16` is already in the shape's view. -Post-restart, the move-in query attached to a SplicePlan re-emits -`l4-16` as an INSERT even though the snapshot already covered it, -because the move-in's `views_after_move` minus `views_before_move` -includes a value the row was already keyed on through a *different* -path. The fix likely needs `move_in_where_clause` to exclude rows -already present in the outer shape's storage, or for SplicePlan to -dedupe its emitted ops against pre-existing on-disk state. +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'))`). The bug is exposed when: + +1. A row is in the outer shape's stored view at restart time. +2. After restart, a materializer event triggers a move-in or move-out + pass through `Buffering.start` / `SplicePlan`. +3. The move-in query (`move_in_where_clause` in + `lib/electric/shapes/querying.ex#352`) uses the row's *current* PG + state for the `impacted_before` exclusion. If the row's + foreign-key column was changed in a transaction whose materializer + event triggered this move-in, current PG state may put the row + under a value present only in `views_after_move`, so the exclusion + doesn't fire and the row is re-emitted. +4. The move-out side has the mirror problem — synthetic deletes the + client generates from tag patterns target rows that the snapshot's + tag set doesn't agree on. + +Likely fix directions: +- Filter the move-in snapshot in + `lib/electric/shape_cache/pure_file_storage.ex#append_move_in_snapshot_to_log!` + with a `skip_row?` callback that excludes keys already present in + the outer shape's on-disk view, instead of the current + `fn _, _ -> false end`. Needs an efficient "is this key present" + index on the writer side. +- Or, have `SplicePlan.build/2` dedupe its emitted INSERT/DELETE + effects against the outer shape's last-persisted view tags. +- Either way, the fix has to be careful that the deduped state + remains consistent with the move-in/out control messages the + client receives, since the client's `TagTracker` uses those to + generate synthetic deletes. Bug 3 only manifests under restart and is likely related to the active-readiness signal — small fix once Bug 2 is closed. From ab4aa656e3a4bf0774d572f02bbfd2b00671a44c Mon Sep 17 00:00:00 2001 From: rob Date: Fri, 1 May 2026 08:27:48 +0100 Subject: [PATCH 17/22] Revert materializer resilience: enforce strict invariants again MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The "log and skip" branches I added in 47ad817ce (NewRecord with key already in index, DeletedRecord for unknown key, move-out/move-in iterating over keys not in the index, decrement_value on a missing value, UpdatedRecord guarded by is_map_key) hide real upstream bugs. The materializer crashing on impossible-looking state is the correct behaviour — duplicates and missing values are invariant violations, not transient noise to swallow. Restore Map.pop!, Map.fetch!, raise on duplicate INSERT, and the unguarded UpdatedRecord rewrite path. Bug 2 will re-surface as a materializer crash; that's where the investigation needs to focus. Keeps the orthogonal changes from 47ad817ce that aren't about masking state inconsistency — the ETS-rescue in ShapeStatus.validate_shape_handle and Api.check_for_disk_updates guard a genuine transient race during stack restart, not a state bug. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../electric/shapes/consumer/materializer.ex | 127 ++++++------------ 1 file changed, 41 insertions(+), 86 deletions(-) diff --git a/packages/sync-service/lib/electric/shapes/consumer/materializer.ex b/packages/sync-service/lib/electric/shapes/consumer/materializer.ex index e9a56cb4a8..2ab3512826 100644 --- a/packages/sync-service/lib/electric/shapes/consumer/materializer.ex +++ b/packages/sync-service/lib/electric/shapes/consumer/materializer.ex @@ -528,38 +528,26 @@ defmodule Electric.Shapes.Consumer.Materializer do active_conditions: ac }, {{index, tag_indices}, counts_and_events} -> - if is_map_key(index, key) do - # Duplicate insert — the materializer has already seen this key. - # Surfaces upstream bugs (history replay re-reading entries, or - # the source shape's filter writing two inserts without an - # intervening delete). Log and skip rather than crashing the - # materializer (which would invalidate every dependent shape). - Logger.warning(fn -> - "Materializer for #{state.shape_handle} got duplicate INSERT for key #{inspect(key)}; ignoring" - end) - - {{index, tag_indices}, counts_and_events} - else - {value, original_string} = cast!(record, state) - included? = evaluate_inclusion(move_tags, ac) - - index = - Map.put(index, key, %{ - value: value, - tags: move_tags, - active_conditions: ac, - included?: included? - }) + {value, original_string} = cast!(record, state) + if is_map_key(index, key), do: raise("Key #{key} already exists") + included? = evaluate_inclusion(move_tags, ac) + + index = + Map.put(index, key, %{ + value: value, + tags: move_tags, + active_conditions: ac, + included?: included? + }) - tag_indices = add_row_to_tag_indices(tag_indices, key, move_tags) + tag_indices = add_row_to_tag_indices(tag_indices, key, move_tags) - counts_and_events = - if included?, - do: increment_value(counts_and_events, value, original_string), - else: counts_and_events + counts_and_events = + if included?, + do: increment_value(counts_and_events, value, original_string), + else: counts_and_events - {{index, tag_indices}, counts_and_events} - end + {{index, tag_indices}, counts_and_events} %Changes.UpdatedRecord{ key: key, @@ -579,8 +567,7 @@ defmodule Electric.Shapes.Consumer.Materializer do pk_changed = old_key != key has_ac_update = ac != [] and is_map_key(index, old_key) - if (columns_present or has_tag_updates or has_ac_update or pk_changed) and - is_map_key(index, old_key) do + if columns_present or has_tag_updates or has_ac_update or pk_changed do old_entry = Map.fetch!(index, old_key) # When the primary key changes, re-index every existing tag for the new key. @@ -677,32 +664,18 @@ defmodule Electric.Shapes.Consumer.Materializer do %Changes.DeletedRecord{key: key, move_tags: move_tags}, {{index, tag_indices}, counts_and_events} -> - case Map.pop(index, key) do - {nil, _index} -> - # The materializer's index doesn't contain this key — usually - # because the source shape's filter wrote a DELETE event for - # a row whose most recent state didn't match the inner shape's - # WHERE clause. Log and skip rather than crashing the - # materializer, which would invalidate every dependent shape. - Logger.warning(fn -> - "Materializer for #{state.shape_handle} got DELETE for unknown key #{inspect(key)}; ignoring" - end) - - {{index, tag_indices}, counts_and_events} - - {entry, index} -> - tag_indices = remove_row_from_tag_indices(tag_indices, key, move_tags) - - if entry.included? do - {{index, tag_indices}, - decrement_value( - counts_and_events, - entry.value, - value_to_string(entry.value, state) - )} - else - {{index, tag_indices}, counts_and_events} - end + {entry, index} = Map.pop!(index, key) + tag_indices = remove_row_from_tag_indices(tag_indices, key, move_tags) + + if entry.included? do + {{index, tag_indices}, + decrement_value( + counts_and_events, + entry.value, + value_to_string(entry.value, state) + )} + else + {{index, tag_indices}, counts_and_events} end %{headers: %{event: event, patterns: patterns}}, @@ -716,27 +689,16 @@ defmodule Electric.Shapes.Consumer.Materializer do affected, {{index, tag_indices}, counts_and_events}, fn {key, matched_positions}, acc -> - case Map.fetch(index, key) do - {:ok, entry} -> - process_move_event( - entry, - key, - matched_positions, - new_condition, - acc, - state - ) - - :error -> - # Tag index pointed at a key not in the materializer - # index — log and skip rather than crash the - # materializer. - Logger.warning(fn -> - "Materializer for #{state.shape_handle} got #{event} for unknown key #{inspect(key)}; ignoring" - end) - - acc - end + entry = Map.fetch!(index, key) + + process_move_event( + entry, + key, + matched_positions, + new_condition, + acc, + state + ) end ) @@ -759,15 +721,8 @@ defmodule Electric.Shapes.Consumer.Materializer do end end - defp decrement_value({value_counts, events}, value, _original_string) - when not is_map_key(value_counts, value) do - # The value was supposed to have been added before; missing here implies - # an upstream inconsistency. Logging would spam — the materializer's - # other handlers already log when they observe inconsistent state. - {value_counts, events} - end - defp decrement_value({value_counts, events}, value, original_string) do + # If we're decrementing, it must have been added before case Map.fetch!(value_counts, value) do 1 -> {Map.delete(value_counts, value), [{:move_out, {value, original_string}} | events]} From 15eaae5fab84d621623c825f9cc38ff820602ed7 Mon Sep 17 00:00:00 2001 From: rob Date: Fri, 1 May 2026 08:44:06 +0100 Subject: [PATCH 18/22] Update bugs.md: clarify Bug 6 partial fix and refine Bug 2 analysis Bug 6 fix is now properly split: - KEPT: consumer :noproc catch + ETS rescue (process/resource lifecycle). - REVERTED: materializer "log and skip" branches that were masking real invariant violations (committed in 57df37c54). Bug 2 analysis refined: it's not strictly a post-restart bug. It's a race between the shape's direct Filter path and the materializer's move-in path when the *same* transaction contains both a row-level update that crosses the filter boundary AND a dep-view change. Both paths emit log ops for the same row. Fresh clients polling from offset=-1 see the duplicates; pre-restart clients started from a consistent in-memory state happen not to observe the duplication. Co-Authored-By: Claude Opus 4.7 (1M context) --- packages/sync-service/bugs.md | 154 ++++++++++++++++++++-------------- 1 file changed, 91 insertions(+), 63 deletions(-) diff --git a/packages/sync-service/bugs.md b/packages/sync-service/bugs.md index f27383d2c5..bd8f5142c9 100644 --- a/packages/sync-service/bugs.md +++ b/packages/sync-service/bugs.md @@ -367,44 +367,44 @@ consumer's `handle_materializer_down/2` falls through the `case` to reason, which schedules `remove_shape_async`. The async task then races the rest of the stack shutdown. -**Fix** +**Fix (partial)** -The materializer was the upstream trigger for the cleanup cascade — a +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. - -Two changes block the cascade: - -1. **Resilient `apply_changes/2` in - `lib/electric/shapes/consumer/materializer.ex`.** Every "this looks - impossible" branch now logs a warning and continues rather than - raising, so an inconsistent inner-shape log can't kill the - materializer: - - `DeletedRecord` for a key not in the index → log + skip - - `NewRecord` for a key already in the index → log + skip - - `UpdatedRecord` only enters the rewrite path when - `is_map_key(index, old_key)` holds - - move-out / move-in iterations switch from `Map.fetch!` to - `Map.fetch` with a skip branch - - `decrement_value/3` treats a missing value-count as a no-op - -2. **Catch `:noproc` on consumer→materializer call in +`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`. - -Plus two ergonomic guardrails for the inflight-request window: - -- `ShapeStatus.validate_shape_handle/3` rescues `ArgumentError → - :error` so a held long-poll waking up between old/new - `ShapeStatusOwner` doesn't 500. -- `Api.check_for_disk_updates/1` rescues `ArgumentError → :no_change` - for the same window. + `@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** @@ -412,10 +412,11 @@ 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 409 must-refetch. The next blocker exposed by these -fixes is Bug 2 — duplicate inserts in the post-restart move-in -snapshot, which was previously masked by the cascade-409 path -swallowing the failing shape entirely. +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 @@ -451,35 +452,62 @@ CHECK_TIMEOUT=60000 SHAPE_COUNT=5 MUTATIONS_PER_TXN=10 TXNS_PER_BATCH=10 \ 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'))`). The bug is exposed when: - -1. A row is in the outer shape's stored view at restart time. -2. After restart, a materializer event triggers a move-in or move-out - pass through `Buffering.start` / `SplicePlan`. -3. The move-in query (`move_in_where_clause` in - `lib/electric/shapes/querying.ex#352`) uses the row's *current* PG - state for the `impacted_before` exclusion. If the row's - foreign-key column was changed in a transaction whose materializer - event triggered this move-in, current PG state may put the row - under a value present only in `views_after_move`, so the exclusion - doesn't fire and the row is re-emitted. -4. The move-out side has the mirror problem — synthetic deletes the - client generates from tag patterns target rows that the snapshot's - tag set doesn't agree on. - -Likely fix directions: -- Filter the move-in snapshot in - `lib/electric/shape_cache/pure_file_storage.ex#append_move_in_snapshot_to_log!` - with a `skip_row?` callback that excludes keys already present in - the outer shape's on-disk view, instead of the current - `fn _, _ -> false end`. Needs an efficient "is this key present" - index on the writer side. -- Or, have `SplicePlan.build/2` dedupe its emitted INSERT/DELETE - effects against the outer shape's last-persisted view tags. -- Either way, the fix has to be careful that the deduped state - remains consistent with the move-in/out control messages the - client receives, since the client's `TagTracker` uses those to - generate synthetic deletes. +level_1_id = 'l1-4'))`). + +**The race (refined analysis)** + +A single transaction can produce events on two independent paths in +the outer consumer: + +1. **Direct filter path.** The transaction contains a level_4 + `UPDATE` for `l4-16` that crosses the shape's filter boundary — + e.g. its `level_3_id` changes from a value in the dep view to + another value also in the dep view. The shape's `Filter` writes a + regular `UPDATE` (or `INSERT`/`DELETE`) op to the log directly. + +2. **Materializer move-in path.** The same transaction also + toggles/moves a level_3 row in a way that adds a new value to the + inner dep view. The materializer fires a `move-in` event for the + new dep value. The outer consumer enters `Buffering`, runs the + move-in query against PG (using a snapshot taken *after* the + triggering txn), and writes the matching rows to the log via + `AppendMoveInSnapshot`. + +Because the move-in query reads PG's *current* row state, `l4-16`'s +new `level_3_id` matches the freshly-added dep value and the row is +emitted. But the direct filter path already produced an op for the +same row earlier in the log. Net result: two INSERTs (or +INSERT+UPDATE, or two DELETEs) for the same key in the log a fresh +client polls. + +This isn't strictly a *post-restart* bug — the same race exists in +steady state. It only surfaces in the test after a server restart +because pre-restart the test doesn't observe it: the inflight client +state and the server's filter view are both initialised from the +same place, so the duplicate `UPDATE`/`INSERT` happens to look +consistent to that client. After a restart, a fresh client polls +from offset=-1 and sees the entire log including both ops; that's +when it flunks. + +**Fix directions** + +- `move_in_where_clause` in + `lib/electric/shapes/querying.ex#352` uses CURRENT PG state for + `impacted_before` exclusion. It needs to exclude rows whose + *pre-update* state already matched, not whose current state + matches `views_before_move`. That probably means tracking + per-row pre-update FK values from the transaction log when + building the WHERE clause. +- Or, coordinate `Filter` and `Buffering` so the same triggering + transaction's row-level UPDATEs are *not* applied via the direct + filter path while a buffering window is open — they'd be implied + by the move-in snapshot. +- Or, have `SplicePlan.build/2` dedupe its emitted ops against the + filter-emitted ops in the same transaction window. + +Whichever path, the fix has to keep the move-in/out *control* +messages consistent so the client's `TagTracker` generates the +right synthetic deletes. Bug 3 only manifests under restart and is likely related to the active-readiness signal — small fix once Bug 2 is closed. From 2c331104218ed4e572b79b8ab707857d5333fed5 Mon Sep 17 00:00:00 2001 From: rob Date: Fri, 1 May 2026 12:10:56 +0100 Subject: [PATCH 19/22] Correct Bug 2/5 analysis: post-restart-only, root cause is offset mismatch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Earlier write-up incorrectly claimed Bug 2 also exists in steady state. That's wrong — the oracle test passes cleanly without RESTART_SERVER_EVERY, so the bug is restart-specific. Real root cause: The inner shape's storage and the outer shape's storage are written by independent writers with independent flush schedules. At the moment the stack is killed, they're typically at DIFFERENT last_persisted_offsets: L_outer ≤ L_inner. On recovery: - Materializer rebuilds its view from inner shape's history → state at L_inner. - Outer consumer's state.views is seeded from the materializer → also at L_inner. - But the outer shape's STORAGE is still at L_outer. So state.views and outer storage are at different LSNs. The materializer events for offsets in (L_outer, L_inner] correspond to ops that were never written to outer storage (the writer was killed before flushing them) but state.views reflects them as already-applied. When live events resume, the materializer eventually re-emits those move-ins/move-outs (from replication catch-up or fresh mutations). state.views says "already there" so the move-in is processed as redundant — but the move-in query still runs and writes the matching rows again. Result: duplicate INSERTs in the on-disk log; the orphan DELETE variant is the mirror. Bug 5 status downgraded to PARTIALLY FIXED. The eager-start change prevents the materializer view from drifting *forward* of the outer consumer's seeded view between consumer init and the first live event, but it can't fix the gap between materializer view and outer-storage view that already exists at recovery time. Real fix needs one of: 1. Persist state.views alongside outer shape's storage; restore atomically. 2. Derive state.views from outer storage's row contents on init. 3. Track per-outer-shape last_persisted_dep_lsn and replay materializer events from there on subscribe. Co-Authored-By: Claude Opus 4.7 (1M context) --- packages/sync-service/bugs.md | 193 ++++++++++++++++++++++------------ 1 file changed, 126 insertions(+), 67 deletions(-) diff --git a/packages/sync-service/bugs.md b/packages/sync-service/bugs.md index bd8f5142c9..b37a0562c9 100644 --- a/packages/sync-service/bugs.md +++ b/packages/sync-service/bugs.md @@ -185,7 +185,7 @@ CHECK_TIMEOUT=60000 SHAPE_COUNT=10 MUTATIONS_PER_TXN=10 TXNS_PER_BATCH=10 \ - 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 +## Bug 5: Post-restart move-in events lost when source-shape main log spans multiple chunks — PARTIALLY FIXED **Symptom** @@ -272,7 +272,8 @@ 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. -**Fix** +**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. @@ -287,16 +288,23 @@ Concretely, `ShapeCache.handle_continue(:wait_for_restore)` now: 2. Only after all subquery consumers are fully initialized does it call `ShapeLogCollector.mark_as_ready/1`. -This guarantees the materializer view and the outer consumer's seeded -view are both derived from on-disk state alone — no events have flowed -yet. So the materializer view = outer-storage view = pre-restart state, -and any subsequent move-in is correctly enqueued (not dropped as -redundant against a view that has already advanced past it). - -The chosen approach is essentially option (3) reduced to its simplest -form: instead of threading per-dep offsets through the subscription -protocol, we hold the dispatch gate closed until every subquery -consumer has caught up to a known consistent point on disk. +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** @@ -420,7 +428,15 @@ move-in. ## Note for triage -Bugs 1, 4, 5, and 6 are fixed. +Bugs 1 and 4 are fully fixed. + +Bug 5 is partially fixed (eager-start before mark_as_ready). The +remaining gap (inner vs outer `last_persisted_offset` mismatch) is +the root cause of Bug 2. + +Bug 6 is partially fixed (consumer `:noproc` catch + ETS rescue). +The materializer "log and skip" branches were reverted because they +hide invariant violations. 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 @@ -454,60 +470,103 @@ 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'))`). -**The race (refined analysis)** - -A single transaction can produce events on two independent paths in -the outer consumer: - -1. **Direct filter path.** The transaction contains a level_4 - `UPDATE` for `l4-16` that crosses the shape's filter boundary — - e.g. its `level_3_id` changes from a value in the dep view to - another value also in the dep view. The shape's `Filter` writes a - regular `UPDATE` (or `INSERT`/`DELETE`) op to the log directly. - -2. **Materializer move-in path.** The same transaction also - toggles/moves a level_3 row in a way that adds a new value to the - inner dep view. The materializer fires a `move-in` event for the - new dep value. The outer consumer enters `Buffering`, runs the - move-in query against PG (using a snapshot taken *after* the - triggering txn), and writes the matching rows to the log via - `AppendMoveInSnapshot`. - -Because the move-in query reads PG's *current* row state, `l4-16`'s -new `level_3_id` matches the freshly-added dep value and the row is -emitted. But the direct filter path already produced an op for the -same row earlier in the log. Net result: two INSERTs (or -INSERT+UPDATE, or two DELETEs) for the same key in the log a fresh -client polls. - -This isn't strictly a *post-restart* bug — the same race exists in -steady state. It only surfaces in the test after a server restart -because pre-restart the test doesn't observe it: the inflight client -state and the server's filter view are both initialised from the -same place, so the duplicate `UPDATE`/`INSERT` happens to look -consistent to that client. After a restart, a fresh client polls -from offset=-1 and sees the entire log including both ops; that's -when it flunks. - -**Fix directions** - -- `move_in_where_clause` in - `lib/electric/shapes/querying.ex#352` uses CURRENT PG state for - `impacted_before` exclusion. It needs to exclude rows whose - *pre-update* state already matched, not whose current state - matches `views_before_move`. That probably means tracking - per-row pre-update FK values from the transaction log when - building the WHERE clause. -- Or, coordinate `Filter` and `Buffering` so the same triggering - transaction's row-level UPDATEs are *not* applied via the direct - filter path while a buffering window is open — they'd be implied - by the move-in snapshot. -- Or, have `SplicePlan.build/2` dedupe its emitted ops against the - filter-emitted ops in the same transaction window. - -Whichever path, the fix has to keep the move-in/out *control* -messages consistent so the client's `TagTracker` generates the -right synthetic deletes. +**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.** The inner +shape's storage and the outer shape's storage are written by +different writers, with their own flush schedules. When the stack is +killed, the inner shape's `last_persisted_offset` and the outer +shape's `last_persisted_offset` are usually NOT equal — typically +they differ by one or two committed transactions. + +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. From dac4bc25b2fa9a5b941ad86cd561f4ea826dbabc Mon Sep 17 00:00:00 2001 From: rob Date: Fri, 1 May 2026 12:33:59 +0100 Subject: [PATCH 20/22] =?UTF-8?q?bugs.md:=20spell=20out=20why=20L=5Finner?= =?UTF-8?q?=20=E2=89=A0=20L=5Fouter=20after=20a=20kill?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Independent per-shape writers with their own buffers and flush periods. The FlushTracker coordinates the *min* across shapes for the replication slot's flushed_wal — that's the lower bound. But each shape's on-disk last_persisted_txn_offset can be ahead of that min, and at kill time there's no cross-shape coordinated flush, so inner and outer shape writers can land at different durable offsets. Co-Authored-By: Claude Opus 4.7 (1M context) --- packages/sync-service/bugs.md | 27 +++++++++++++++++++++------ 1 file changed, 21 insertions(+), 6 deletions(-) diff --git a/packages/sync-service/bugs.md b/packages/sync-service/bugs.md index b37a0562c9..e85faa163a 100644 --- a/packages/sync-service/bugs.md +++ b/packages/sync-service/bugs.md @@ -491,12 +491,27 @@ 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.** The inner -shape's storage and the outer shape's storage are written by -different writers, with their own flush schedules. When the stack is -killed, the inner shape's `last_persisted_offset` and the outer -shape's `last_persisted_offset` are usually NOT equal — typically -they differ by one or two committed transactions. +**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): From cd399796c366e8453b2354478c8aa50032d913bb Mon Sep 17 00:00:00 2001 From: rob Date: Fri, 1 May 2026 13:07:01 +0100 Subject: [PATCH 21/22] Coordinated safe shutdown + drop-all-on-dirty-startup MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The post-restart oracle property test was failing because different shapes' on-disk last_persisted_txn_offset can land at different LSNs after a stack stops: each shape has its own writer with its own buffer and flush schedule, the FlushTracker only enforces a min lower-bound for advancing the replication slot, and there's no cross-shape coordinated flush at supervisor tear-down. After restart the materializer rebuilds its view from the inner shape's history (at L_inner) but the outer shape's storage is at L_outer ≠ L_inner; the outer consumer's state.views (seeded from the materializer) and its storage end up at different LSNs and produce duplicate inserts / orphan deletes when live events resume. Sound recovery from that state is hard, so: 1. Add Electric.StackSupervisor.ShutdownCoordinator as the LAST child of StackSupervisor (so it terminates FIRST on graceful shutdown, while every other process is still alive). Its terminate/2: - stops the ReplicationClient so no new events arrive at the SLC - calls ShapeLogCollector.drain/2 which sets a draining flag, blocks new handle_event calls, and waits on the FlushTracker until every shape has flushed up to last_seen_offset - writes a clean-shutdown marker to persistent_kv If any step fails (timeout, no SLC, exit, brutal kill), the marker is NOT written. Brutal-killed processes don't run terminate/2 either way. 2. ShapeStatusOwner.handle_continue(:initialize) consumes the marker on startup. If absent (= previous shutdown was dirty), drops all shape state — ShapeDb.reset (SQLite) + Storage.cleanup_all (per-shape storage dirs) — before initialising. Clients re-request shapes from scratch with new handles. 3. ShapeLogCollector gets a drain/2 public API and a :drain handle_call. The :writer_flushed cast now also resolves the pending drain_waiter when the FlushTracker becomes empty. 4. Threads persistent_kv from StackSupervisor through MonitoredCoreSupervisor to ShapeStatusOwner. Verified: - mix test test/electric/{shape_cache_test.exs, shapes/consumer/materializer_test.exs, shapes/consumer_test.exs} test/integration/oracle_restore_test.exs --include oracle --include slow → 134 / 134 passing. - CHECK_TIMEOUT=60000 SHAPE_COUNT=10 BATCH_COUNT=10 RESTART_SERVER_EVERY=7 LONG_POLL_TIMEOUT=2000: passing on seeds 1, 2, 3, 4, 5, 8. - CHECK_TIMEOUT=60000 SHAPE_COUNT=30 BATCH_COUNT=50 RESTART_SERVER_EVERY=7 RESTART_CLIENT_EVERY=11 seed 8: passing. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../lib/electric/monitored_core_supervisor.ex | 5 +- .../replication/shape_log_collector.ex | 60 +++++++- .../shape_cache/shape_status_owner.ex | 52 ++++++- .../lib/electric/stack_supervisor.ex | 7 +- .../stack_supervisor/shutdown_coordinator.ex | 135 ++++++++++++++++++ 5 files changed, 252 insertions(+), 7 deletions(-) create mode 100644 packages/sync-service/lib/electric/stack_supervisor/shutdown_coordinator.ex 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/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/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 From fc862bb67db4c85e47f1236cb58600a42335d45a Mon Sep 17 00:00:00 2001 From: rob Date: Fri, 1 May 2026 13:07:43 +0100 Subject: [PATCH 22/22] Mark Bugs 2 and 5 as fixed; update triage note --- packages/sync-service/bugs.md | 31 ++++++++++++++++++++----------- 1 file changed, 20 insertions(+), 11 deletions(-) diff --git a/packages/sync-service/bugs.md b/packages/sync-service/bugs.md index e85faa163a..59f481fa9a 100644 --- a/packages/sync-service/bugs.md +++ b/packages/sync-service/bugs.md @@ -105,7 +105,7 @@ 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 +## Bug 2: Snapshot+log replay can produce duplicate / orphan operations after restart — FIXED **Symptom** @@ -185,7 +185,7 @@ CHECK_TIMEOUT=60000 SHAPE_COUNT=10 MUTATIONS_PER_TXN=10 TXNS_PER_BATCH=10 \ - 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 — PARTIALLY FIXED +## Bug 5: Post-restart move-in events lost when source-shape main log spans multiple chunks — FIXED **Symptom** @@ -428,15 +428,24 @@ move-in. ## Note for triage -Bugs 1 and 4 are fully fixed. - -Bug 5 is partially fixed (eager-start before mark_as_ready). The -remaining gap (inner vs outer `last_persisted_offset` mismatch) is -the root cause of Bug 2. - -Bug 6 is partially fixed (consumer `:noproc` catch + ETS rescue). -The materializer "log and skip" branches were reverted because they -hide invariant violations. +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