diff --git a/packages/sync-service/test/integration/oracle_property_test.exs b/packages/sync-service/test/integration/oracle_property_test.exs index 2787fa9108..643fbd84ef 100644 --- a/packages/sync-service/test/integration/oracle_property_test.exs +++ b/packages/sync-service/test/integration/oracle_property_test.exs @@ -12,6 +12,14 @@ defmodule Electric.Integration.OraclePropertyTest do - MUTATIONS_PER_TXN: Number of mutations per transaction (default: 5) - RUN_COUNT: Number of property test iterations (default: 1) - LONG_POLL_TIMEOUT: Server long-poll timeout in ms (default: 100) + - RESTART_SERVER_EVERY: Stop and restart the sync stack every N batches to + test server-side restore-from-file (default: 0, disabled). After each + restart, fresh clients reconnect and check_initial_state asserts the + restored state matches the oracle. + - RESTART_CLIENT_EVERY: Throw away clients (poll cursors, materialized + rows) and reconnect every M batches to test that fresh polls correctly + assemble snapshot + log (default: 0, disabled). Independent of + RESTART_SERVER_EVERY. """ use ExUnit.Case, async: false @@ -35,6 +43,7 @@ defmodule Electric.Integration.OraclePropertyTest do @default_mutations_per_txn 5 setup [:with_unique_db] + setup :use_persistent_slot setup :with_complete_stack # Use a short long_poll_timeout to speed up tests - shapes with no changes @@ -55,12 +64,23 @@ defmodule Electric.Integration.OraclePropertyTest do ctx end + # The replication slot must survive the StackSupervisor restart used by + # RESTART_SERVER_EVERY, otherwise Electric correctly treats the new slot + # as a slot-loss event and purges all on-disk shape data — defeating the + # restore-from-file scenario. Always run with a persistent slot; the slot + # is dropped automatically with the per-test database in `after_suite`. + defp use_persistent_slot(_ctx) do + %{replication_opts_overrides: [slot_temporary?: false]} + end + test "shapes with generated where clauses and mutations", ctx do run_count = env_int("RUN_COUNT") || 1 shape_count = env_int("SHAPE_COUNT") || @default_shape_count batch_count = env_int("BATCH_COUNT") || @default_batch_count txns_per_batch = env_int("TXNS_PER_BATCH") || @default_txns_per_batch mutations_per_txn = env_int("MUTATIONS_PER_TXN") || @default_mutations_per_txn + restart_server_every = env_int("RESTART_SERVER_EVERY") || 0 + restart_client_every = env_int("RESTART_CLIENT_EVERY") || 0 total_mutations = batch_count * txns_per_batch * mutations_per_txn @@ -69,7 +89,11 @@ defmodule Electric.Integration.OraclePropertyTest do max_runs: run_count do transactions = Enum.chunk_every(mutations, mutations_per_txn) batches = Enum.chunk_every(transactions, txns_per_batch) - test_against_oracle(ctx, shapes, batches) + + test_against_oracle(ctx, shapes, batches, + restart_server_every: restart_server_every, + restart_client_every: restart_client_every + ) end end end diff --git a/packages/sync-service/test/support/component_setup.ex b/packages/sync-service/test/support/component_setup.ex index 284415d51c..5ba6348cc9 100644 --- a/packages/sync-service/test/support/component_setup.ex +++ b/packages/sync-service/test/support/component_setup.ex @@ -432,11 +432,68 @@ defmodule Support.ComponentSetup do stack_events_registry = Electric.stack_events_registry() - ref = Electric.StackSupervisor.subscribe_to_stack_events(stack_id) - publication_name = Map.get(ctx, :publication_name, "electric_test_pub_#{:erlang.phash2(stack_id)}") + stack_supervisor = + start_stack_supervisor!(ctx, stack_id, kv, storage, stack_events_registry, publication_name) + + %{ + stack_id: stack_id, + registry: Electric.StackSupervisor.registry_name(stack_id), + stack_events_registry: stack_events_registry, + shape_cache: {ShapeCache, [stack_id: stack_id]}, + persistent_kv: kv, + stack_supervisor: stack_supervisor, + storage: storage, + inspector: + {EtsInspector, stack_id: stack_id, server: EtsInspector.name(stack_id: stack_id)}, + feature_flags: Electric.Config.get_env(:feature_flags), + publication_name: publication_name + } + end + + @doc """ + Stops the running `Electric.StackSupervisor` started by `with_complete_stack/1` + and starts a fresh one with the same `stack_id`, `persistent_kv`, `storage`, + and `publication_name` so the new stack reads back what the old stack + persisted to disk. + + Returns a map with the updated `:stack_supervisor` pid. Other ctx keys + (stack_id, persistent_kv, storage, registry, etc.) are unchanged. + """ + def restart_complete_stack(ctx) do + :ok = stop_supervised(Electric.StackSupervisor) + + stack_supervisor = + start_stack_supervisor!( + ctx, + ctx.stack_id, + ctx.persistent_kv, + ctx.storage, + ctx.stack_events_registry, + ctx.publication_name + ) + + # The :stack_status :ready event fires before the connection pools and + # shape metadata are fully online. Polling clients hitting the server in + # that window can see spurious 409s. Wait for the StatusMonitor's + # :active level which requires all readiness conditions to be met. + :ok = Electric.StatusMonitor.wait_until_active(ctx.stack_id, timeout: 5000) + + %{stack_supervisor: stack_supervisor} + end + + defp start_stack_supervisor!( + ctx, + stack_id, + kv, + storage, + stack_events_registry, + publication_name + ) do + ref = Electric.StackSupervisor.subscribe_to_stack_events(stack_id) + connection_opts = Keyword.merge(ctx.pooled_db_config, List.wrap(ctx[:connection_opt_overrides])) @@ -492,19 +549,7 @@ defmodule Support.ComponentSetup do # potential CI slowness, including PG assert_receive {:stack_status, ^ref, :ready}, 2000 - %{ - stack_id: stack_id, - registry: Electric.StackSupervisor.registry_name(stack_id), - stack_events_registry: stack_events_registry, - shape_cache: {ShapeCache, [stack_id: stack_id]}, - persistent_kv: kv, - stack_supervisor: stack_supervisor, - storage: storage, - inspector: - {EtsInspector, stack_id: stack_id, server: EtsInspector.name(stack_id: stack_id)}, - feature_flags: Electric.Config.get_env(:feature_flags), - publication_name: publication_name - } + stack_supervisor end def secure_mode(_ctx) do diff --git a/packages/sync-service/test/support/oracle_harness.ex b/packages/sync-service/test/support/oracle_harness.ex index fb382577c0..c9b5d92cf7 100644 --- a/packages/sync-service/test/support/oracle_harness.ex +++ b/packages/sync-service/test/support/oracle_harness.ex @@ -56,9 +56,7 @@ defmodule Support.OracleHarness do # ---------------------------------------------------------------------------- def default_opts_from_env do - %{ - oracle_pool_size: env_int("ORACLE_POOL_SIZE") || @default_oracle_pool_size - } + [oracle_pool_size: env_int("ORACLE_POOL_SIZE") || @default_oracle_pool_size] end @doc """ @@ -73,11 +71,17 @@ defmodule Support.OracleHarness do - :oracle_pool_size - number of parallel oracle connections (default: 50, env: ORACLE_POOL_SIZE) - :timeout_ms - timeout for waiting on shapes (default: 10_000) + - :restart_server_every - restart the StackSupervisor every N batches to + exercise restore-from-disk (default: 0, disabled) + - :restart_client_every - throw away and recreate the shape clients every + M batches to exercise fresh-poll consistency (default: 0, disabled) """ - @spec test_against_oracle(map(), [shape()], [batch()], map()) :: :ok - def test_against_oracle(ctx, shapes, batches, opts \\ %{}) do - opts = Map.merge(default_opts_from_env(), opts) + @spec test_against_oracle(map(), [shape()], [batch()], keyword()) :: :ok + def test_against_oracle(ctx, shapes, batches, opts \\ []) do + opts = Keyword.merge(default_opts_from_env(), opts) timeout_ms = opts[:timeout_ms] || env_int("CHECK_TIMEOUT") || @default_timeout_ms + restart_server_every = opts[:restart_server_every] || 0 + restart_client_every = opts[:restart_client_every] || 0 log_test_config(shapes, batches) @@ -90,21 +94,46 @@ defmodule Support.OracleHarness do # Check initial state (all in parallel) log("Checking initial snapshot for #{length(pids)} shapes") - pids - |> Task.async_stream(&ShapeChecker.check_initial_state/1, timeout: :infinity) - |> Stream.run() + check_initial_states(pids) # Run each batch log("Running #{length(batches)} batches") - batches - |> Enum.with_index(1) - |> Enum.each(fn {transactions, batch_idx} -> - run_batch(ctx, pids, transactions, batch_idx) - end) + total_batches = length(batches) + + {final_pids, _ctx} = + batches + |> Enum.with_index(1) + |> Enum.reduce({pids, ctx}, fn {transactions, batch_idx}, {pids, ctx} -> + run_batch(ctx, pids, transactions, batch_idx) + + restart_server? = + restart_server_every > 0 and rem(batch_idx, restart_server_every) == 0 and + batch_idx < total_batches + + restart_client? = + restart_client_every > 0 and rem(batch_idx, restart_client_every) == 0 and + batch_idx < total_batches + + cond do + restart_server? -> + # restart_server tears down the old checkers (they're polling the + # server about to go down) and recreates them after the stack is + # back up. If a client restart is also due this batch, it's + # subsumed by the recreate that follows the server restart. + restart_server(ctx, pids, shapes, oracle_pool, timeout_ms, batch_idx) + + restart_client? -> + new_pids = recreate_checkers(ctx, pids, shapes, oracle_pool, timeout_ms, batch_idx) + {new_pids, ctx} + + true -> + {pids, ctx} + end + end) # Cleanup - Enum.each(pids, &GenServer.stop/1) + Enum.each(final_pids, &GenServer.stop/1) GenServer.stop(oracle_pool) :ok end @@ -134,6 +163,40 @@ defmodule Support.OracleHarness do end) end + defp check_initial_states(pids) do + pids + |> Task.async_stream(&ShapeChecker.check_initial_state/1, timeout: :infinity) + |> Stream.run() + end + + # Restarts the Electric stack (server-side restore-from-file test) and + # reconnects the clients. Old checkers are stopped because their polls are + # against the server that is about to go down. + defp restart_server(ctx, pids, shapes, oracle_pool, timeout_ms, batch_idx) do + log("Restarting server after batch_#{batch_idx}") + + Enum.each(pids, &GenServer.stop/1) + + new_ctx = Map.merge(ctx, Support.ComponentSetup.restart_complete_stack(ctx)) + + new_pids = recreate_checkers(new_ctx, [], shapes, oracle_pool, timeout_ms, batch_idx) + + {new_pids, new_ctx} + end + + # Throws away the existing checkers and creates fresh ones (client-side + # resync test). The new checkers do an initial snapshot poll and assert + # consistency against the oracle. + defp recreate_checkers(ctx, old_pids, shapes, oracle_pool, timeout_ms, batch_idx) do + log("Recreating clients after batch_#{batch_idx}") + + Enum.each(old_pids, &GenServer.stop/1) + + new_pids = start_checkers(ctx, shapes, oracle_pool, timeout_ms) + check_initial_states(new_pids) + new_pids + end + defp run_batch(ctx, pids, transactions, batch_idx) do batch_name = "batch_#{batch_idx}" total_mutations = transactions |> Enum.map(&length/1) |> Enum.sum()