Restart-aware oracle property tests

packages/sync-service/test/integration/oracle_property_test.exs

@@ -12,6 +12,14 @@ defmodule Electric.Integration.OraclePropertyTest do
     - MUTATIONS_PER_TXN: Number of mutations per transaction (default: 5)
     - RUN_COUNT: Number of property test iterations (default: 1)
     - LONG_POLL_TIMEOUT: Server long-poll timeout in ms (default: 100)
+    - RESTART_SERVER_EVERY: Stop and restart the sync stack every N batches to
+      test server-side restore-from-file (default: 0, disabled). After each
+      restart, fresh clients reconnect and check_initial_state asserts the
+      restored state matches the oracle.
+    - RESTART_CLIENT_EVERY: Throw away clients (poll cursors, materialized
+      rows) and reconnect every M batches to test that fresh polls correctly
+      assemble snapshot + log (default: 0, disabled). Independent of
+      RESTART_SERVER_EVERY.
   """
 
   use ExUnit.Case, async: false
@@ -35,6 +43,7 @@ defmodule Electric.Integration.OraclePropertyTest do
   @default_mutations_per_txn 5
 
   setup [:with_unique_db]
+  setup :use_persistent_slot
   setup :with_complete_stack
 
   # Use a short long_poll_timeout to speed up tests - shapes with no changes
@@ -55,12 +64,23 @@ defmodule Electric.Integration.OraclePropertyTest do
     ctx
   end
 
+  # The replication slot must survive the StackSupervisor restart used by
+  # RESTART_SERVER_EVERY, otherwise Electric correctly treats the new slot
+  # as a slot-loss event and purges all on-disk shape data — defeating the
+  # restore-from-file scenario. Always run with a persistent slot; the slot
+  # is dropped automatically with the per-test database in `after_suite`.
+  defp use_persistent_slot(_ctx) do
+    %{replication_opts_overrides: [slot_temporary?: false]}
+  end
+
   test "shapes with generated where clauses and mutations", ctx do
     run_count = env_int("RUN_COUNT") || 1
     shape_count = env_int("SHAPE_COUNT") || @default_shape_count
     batch_count = env_int("BATCH_COUNT") || @default_batch_count
     txns_per_batch = env_int("TXNS_PER_BATCH") || @default_txns_per_batch
     mutations_per_txn = env_int("MUTATIONS_PER_TXN") || @default_mutations_per_txn
+    restart_server_every = env_int("RESTART_SERVER_EVERY") || 0
+    restart_client_every = env_int("RESTART_CLIENT_EVERY") || 0
 
     total_mutations = batch_count * txns_per_batch * mutations_per_txn
 
@@ -69,7 +89,11 @@ defmodule Electric.Integration.OraclePropertyTest do
               max_runs: run_count do
       transactions = Enum.chunk_every(mutations, mutations_per_txn)
       batches = Enum.chunk_every(transactions, txns_per_batch)
-      test_against_oracle(ctx, shapes, batches)
+
+      test_against_oracle(ctx, shapes, batches,
+        restart_server_every: restart_server_every,
+        restart_client_every: restart_client_every
+      )
     end
   end
 end

packages/sync-service/test/support/component_setup.ex

@@ -432,11 +432,68 @@ defmodule Support.ComponentSetup do
 
     stack_events_registry = Electric.stack_events_registry()
 
-    ref = Electric.StackSupervisor.subscribe_to_stack_events(stack_id)
-
     publication_name =
       Map.get(ctx, :publication_name, "electric_test_pub_#{:erlang.phash2(stack_id)}")
 
+    stack_supervisor =
+      start_stack_supervisor!(ctx, stack_id, kv, storage, stack_events_registry, publication_name)
+
+    %{
+      stack_id: stack_id,
+      registry: Electric.StackSupervisor.registry_name(stack_id),
+      stack_events_registry: stack_events_registry,
+      shape_cache: {ShapeCache, [stack_id: stack_id]},
+      persistent_kv: kv,
+      stack_supervisor: stack_supervisor,
+      storage: storage,
+      inspector:
+        {EtsInspector, stack_id: stack_id, server: EtsInspector.name(stack_id: stack_id)},
+      feature_flags: Electric.Config.get_env(:feature_flags),
+      publication_name: publication_name
+    }
+  end
+
+  @doc """
+  Stops the running `Electric.StackSupervisor` started by `with_complete_stack/1`
+  and starts a fresh one with the same `stack_id`, `persistent_kv`, `storage`,
+  and `publication_name` so the new stack reads back what the old stack
+  persisted to disk.
+
+  Returns a map with the updated `:stack_supervisor` pid. Other ctx keys
+  (stack_id, persistent_kv, storage, registry, etc.) are unchanged.
+  """
+  def restart_complete_stack(ctx) do
+    :ok = stop_supervised(Electric.StackSupervisor)
+
+    stack_supervisor =
+      start_stack_supervisor!(
+        ctx,
+        ctx.stack_id,
+        ctx.persistent_kv,
+        ctx.storage,
+        ctx.stack_events_registry,
+        ctx.publication_name
+      )
+
+    # The :stack_status :ready event fires before the connection pools and
+    # shape metadata are fully online. Polling clients hitting the server in
+    # that window can see spurious 409s. Wait for the StatusMonitor's
+    # :active level which requires all readiness conditions to be met.
+    :ok = Electric.StatusMonitor.wait_until_active(ctx.stack_id, timeout: 5000)
+
+    %{stack_supervisor: stack_supervisor}
+  end
+
+  defp start_stack_supervisor!(
+         ctx,
+         stack_id,
+         kv,
+         storage,
+         stack_events_registry,
+         publication_name
+       ) do
+    ref = Electric.StackSupervisor.subscribe_to_stack_events(stack_id)
+
     connection_opts =
       Keyword.merge(ctx.pooled_db_config, List.wrap(ctx[:connection_opt_overrides]))
 
@@ -492,19 +549,7 @@ defmodule Support.ComponentSetup do
     # potential CI slowness, including PG
     assert_receive {:stack_status, ^ref, :ready}, 2000
 
-    %{
-      stack_id: stack_id,
-      registry: Electric.StackSupervisor.registry_name(stack_id),
-      stack_events_registry: stack_events_registry,
-      shape_cache: {ShapeCache, [stack_id: stack_id]},
-      persistent_kv: kv,
-      stack_supervisor: stack_supervisor,
-      storage: storage,
-      inspector:
-        {EtsInspector, stack_id: stack_id, server: EtsInspector.name(stack_id: stack_id)},
-      feature_flags: Electric.Config.get_env(:feature_flags),
-      publication_name: publication_name
-    }
+    stack_supervisor
   end
 
   def secure_mode(_ctx) do

packages/sync-service/test/support/oracle_harness.ex

@@ -56,9 +56,7 @@ defmodule Support.OracleHarness do
   # ----------------------------------------------------------------------------
 
   def default_opts_from_env do
-    %{
-      oracle_pool_size: env_int("ORACLE_POOL_SIZE") || @default_oracle_pool_size
-    }
+    [oracle_pool_size: env_int("ORACLE_POOL_SIZE") || @default_oracle_pool_size]
   end
 
   @doc """
@@ -73,11 +71,17 @@ defmodule Support.OracleHarness do
 
     - :oracle_pool_size - number of parallel oracle connections (default: 50, env: ORACLE_POOL_SIZE)
     - :timeout_ms - timeout for waiting on shapes (default: 10_000)
+    - :restart_server_every - restart the StackSupervisor every N batches to
+      exercise restore-from-disk (default: 0, disabled)
+    - :restart_client_every - throw away and recreate the shape clients every
+      M batches to exercise fresh-poll consistency (default: 0, disabled)
   """
-  @spec test_against_oracle(map(), [shape()], [batch()], map()) :: :ok
-  def test_against_oracle(ctx, shapes, batches, opts \\ %{}) do
-    opts = Map.merge(default_opts_from_env(), opts)
+  @spec test_against_oracle(map(), [shape()], [batch()], keyword()) :: :ok
+  def test_against_oracle(ctx, shapes, batches, opts \\ []) do
+    opts = Keyword.merge(default_opts_from_env(), opts)
     timeout_ms = opts[:timeout_ms] || env_int("CHECK_TIMEOUT") || @default_timeout_ms
+    restart_server_every = opts[:restart_server_every] || 0
+    restart_client_every = opts[:restart_client_every] || 0
 
     log_test_config(shapes, batches)
 
@@ -90,21 +94,46 @@ defmodule Support.OracleHarness do
     # Check initial state (all in parallel)
     log("Checking initial snapshot for #{length(pids)} shapes")
 
-    pids
-    |> Task.async_stream(&ShapeChecker.check_initial_state/1, timeout: :infinity)
-    |> Stream.run()
+    check_initial_states(pids)
 
     # Run each batch
     log("Running #{length(batches)} batches")
 
-    batches
-    |> Enum.with_index(1)
-    |> Enum.each(fn {transactions, batch_idx} ->
-      run_batch(ctx, pids, transactions, batch_idx)
-    end)
+    total_batches = length(batches)
+
+    {final_pids, _ctx} =
+      batches
+      |> Enum.with_index(1)
+      |> Enum.reduce({pids, ctx}, fn {transactions, batch_idx}, {pids, ctx} ->
+        run_batch(ctx, pids, transactions, batch_idx)
+
+        restart_server? =
+          restart_server_every > 0 and rem(batch_idx, restart_server_every) == 0 and
+            batch_idx < total_batches
+
+        restart_client? =
+          restart_client_every > 0 and rem(batch_idx, restart_client_every) == 0 and
+            batch_idx < total_batches
+
+        cond do
+          restart_server? ->
+            # restart_server tears down the old checkers (they're polling the
+            # server about to go down) and recreates them after the stack is
+            # back up. If a client restart is also due this batch, it's
+            # subsumed by the recreate that follows the server restart.
+            restart_server(ctx, pids, shapes, oracle_pool, timeout_ms, batch_idx)
+
+          restart_client? ->
+            new_pids = recreate_checkers(ctx, pids, shapes, oracle_pool, timeout_ms, batch_idx)
+            {new_pids, ctx}
+
+          true ->
+            {pids, ctx}
+        end
+      end)
 
     # Cleanup
-    Enum.each(pids, &GenServer.stop/1)
+    Enum.each(final_pids, &GenServer.stop/1)
     GenServer.stop(oracle_pool)
     :ok
   end
@@ -134,6 +163,40 @@ defmodule Support.OracleHarness do
     end)
   end
 
+  defp check_initial_states(pids) do
+    pids
+    |> Task.async_stream(&ShapeChecker.check_initial_state/1, timeout: :infinity)
+    |> Stream.run()
+  end
+
+  # Restarts the Electric stack (server-side restore-from-file test) and
+  # reconnects the clients. Old checkers are stopped because their polls are
+  # against the server that is about to go down.
+  defp restart_server(ctx, pids, shapes, oracle_pool, timeout_ms, batch_idx) do
+    log("Restarting server after batch_#{batch_idx}")
+
+    Enum.each(pids, &GenServer.stop/1)
+
+    new_ctx = Map.merge(ctx, Support.ComponentSetup.restart_complete_stack(ctx))
+
+    new_pids = recreate_checkers(new_ctx, [], shapes, oracle_pool, timeout_ms, batch_idx)
+
+    {new_pids, new_ctx}
+  end
+
+  # Throws away the existing checkers and creates fresh ones (client-side
+  # resync test). The new checkers do an initial snapshot poll and assert
+  # consistency against the oracle.
+  defp recreate_checkers(ctx, old_pids, shapes, oracle_pool, timeout_ms, batch_idx) do
+    log("Recreating clients after batch_#{batch_idx}")
+
+    Enum.each(old_pids, &GenServer.stop/1)
+
+    new_pids = start_checkers(ctx, shapes, oracle_pool, timeout_ms)
+    check_initial_states(new_pids)
+    new_pids
+  end
+
   defp run_batch(ctx, pids, transactions, batch_idx) do
     batch_name = "batch_#{batch_idx}"
     total_mutations = transactions |> Enum.map(&length/1) |> Enum.sum()