test(oracle): graceful/brutal/rolling restart strategies for postgres oracle tests

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

@@ -16,6 +16,18 @@ defmodule Electric.Integration.OraclePropertyTest do
       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_TYPE: How the RESTART_SERVER_EVERY restart is performed:
+      "graceful" (default, clean shutdown + restore from disk), "brutal"
+      (kill -9 style crash + recover), or "rolling" (rolling deploy: a new
+      stack takes over the replication slot before the old one is stopped).
+    - RETRY_TRANSIENT_ERRORS: When "true"/"1", a transient poll error (5xx or a
+      connection error) is retried within the check timeout instead of failing
+      the test. These are availability blips that production hides from clients
+      during a restart/deploy, so they are not consistency signals. Default off
+      (any poll error fails, preserving current behaviour). A server that never
+      recovers within the timeout still fails (a window with no successful poll
+      is treated as an error, not a pass). 409/must-refetch, 4xx errors and data
+      mismatches always fail regardless of this flag.
     - 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

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

@@ -463,7 +463,8 @@ defmodule Support.ComponentSetup do
   (stack_id, persistent_kv, storage, registry, etc.) are unchanged.
   """
   def restart_complete_stack(ctx) do
-    :ok = stop_supervised(Electric.StackSupervisor)
+    sup_id = Map.get(ctx, :stack_supervisor_id, Electric.StackSupervisor)
+    :ok = stop_supervised(sup_id)
 
     stack_supervisor =
       start_stack_supervisor!(
@@ -472,7 +473,8 @@ defmodule Support.ComponentSetup do
         ctx.persistent_kv,
         ctx.storage,
         ctx.stack_events_registry,
-        ctx.publication_name
+        ctx.publication_name,
+        id: sup_id
       )
 
     # The :stack_status :ready event fires before the connection pools and
@@ -481,17 +483,217 @@ defmodule Support.ComponentSetup do
     # :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}
+    %{stack_supervisor: stack_supervisor, stack_supervisor_id: sup_id}
   end
 
-  defp start_stack_supervisor!(
-         ctx,
-         stack_id,
-         kv,
-         storage,
-         stack_events_registry,
-         publication_name
-       ) do
+  @doc """
+  Like `restart_complete_stack/1`, but simulates a crash: the running
+  `Electric.StackSupervisor` is killed outright (`Process.exit(pid, :kill)`,
+  no terminate callbacks) rather than shut down gracefully. A fresh stack is
+  then started on the same `stack_id`/storage so recovery from on-disk state
+  (possibly torn writes, an orphaned replication slot, a stale advisory lock)
+  is exercised.
+
+  Returns a map with the updated `:stack_supervisor` pid.
+  """
+  def brutally_restart_complete_stack(ctx) do
+    sup_id = Map.get(ctx, :stack_supervisor_id, Electric.StackSupervisor)
+    pid = ctx.stack_supervisor
+    ref = Process.monitor(pid)
+    Process.exit(pid, :kill)
+
+    receive do
+      {:DOWN, ^ref, :process, ^pid, _reason} -> :ok
+    after
+      5_000 -> flunk("stack supervisor #{inspect(pid)} did not die after :kill")
+    end
+
+    # The child was `restart: :temporary`, so ExUnit does not restart it; its
+    # spec lingers with an :undefined pid. Remove it so the id is free to reuse.
+    case stop_supervised(sup_id) do
+      :ok -> :ok
+      {:error, :not_found} -> :ok
+    end
+
+    # A brutal kill propagates asynchronously; wait for the stack's registered
+    # processes (and their ETS tables) to actually be gone before starting a
+    # new stack with the same stack_id, else the new stack races name/table
+    # reuse and crashes on boot.
+    wait_for_stack_processes_down(ctx.stack_id)
+
+    stack_supervisor =
+      start_stack_supervisor!(
+        ctx,
+        ctx.stack_id,
+        ctx.persistent_kv,
+        ctx.storage,
+        ctx.stack_events_registry,
+        ctx.publication_name,
+        id: sup_id
+      )
+
+    # Generous timeout: worst case the stale advisory lock is only cleared on
+    # the periodic lock-breaker cycle (~10s), not immediately on kill.
+    :ok = Electric.StatusMonitor.wait_until_active(ctx.stack_id, timeout: 15_000)
+
+    %{stack_supervisor: stack_supervisor, stack_supervisor_id: sup_id}
+  end
+
+  @doc """
+  Performs a rolling deploy: a brand new stack (its own `stack_id`, storage and
+  HTTP server) is brought up that contends on the *same* replication slot as
+  the running stack via the Postgres advisory lock. The new stack boots to
+  `read_only` (blocked acquiring the lock the old stack holds), then the old
+  stack is gracefully stopped — releasing the lock and leaving the persistent
+  slot and publication in place — and the new stack takes over as the single
+  active writer.
+
+  Returns a ctx-merge map swapping in the new stack: `stack_id`, `storage`,
+  `persistent_kv`, `registry`, `inspector`, `shape_cache`, `stack_supervisor`,
+  `stack_supervisor_id`, plus a new `client`/`base_url`/`server_pid`/`port` and
+  `server_id`/`rolling_gen` bookkeeping.
+  """
+  def rolling_restart_complete_stack(ctx) do
+    old_stack_id = ctx.stack_id
+    old_sup_id = Map.get(ctx, :stack_supervisor_id, Electric.StackSupervisor)
+    old_server_id = Map.get(ctx, :server_id, Bandit)
+
+    # The old stack derived its slot name from its stack_id in
+    # start_stack_supervisor!/7; force the new stack onto the same slot so they
+    # contend on one advisory lock.
+    old_slot_name = "electric_test_slot_#{:erlang.phash2(old_stack_id)}"
+
+    gen = Map.get(ctx, :rolling_gen, 0) + 1
+    new_stack_id = "#{old_stack_id}_roll#{gen}"
+    new_sup_id = {Electric.StackSupervisor, new_stack_id}
+    new_server_id = {Bandit, new_stack_id}
+
+    new_kv = %Electric.PersistentKV.Memory{
+      parent: self(),
+      pid:
+        start_supervised!(Electric.PersistentKV.Memory,
+          id: {Electric.PersistentKV.Memory, new_stack_id},
+          restart: :temporary
+        )
+    }
+
+    new_storage = {PureFileStorage, stack_id: new_stack_id, storage_dir: ctx.tmp_dir}
+
+    # (1) Start the new stack forced onto the old slot + same publication. It
+    #     will block acquiring the advisory lock (held by the old stack), so it
+    #     never emits :ready — don't wait for it.
+    start_ctx =
+      Map.put(
+        ctx,
+        :replication_opts_overrides,
+        Keyword.merge(
+          List.wrap(ctx[:replication_opts_overrides]),
+          slot_name: old_slot_name,
+          slot_temporary?: false
+        )
+      )
+
+    new_stack_supervisor =
+      start_stack_supervisor!(
+        start_ctx,
+        new_stack_id,
+        new_kv,
+        new_storage,
+        ctx.stack_events_registry,
+        ctx.publication_name,
+        id: new_sup_id,
+        await_ready?: false
+      )
+
+    # (2) read_only is reachable without the lock (shape metadata is loaded
+    #     before lock acquisition), confirming the new tree booted.
+    {:ok, :read_only} =
+      Electric.StatusMonitor.wait_until(new_stack_id, :read_only, timeout: 5_000)
+
+    new_ctx =
+      Map.merge(ctx, %{
+        stack_id: new_stack_id,
+        storage: new_storage,
+        persistent_kv: new_kv,
+        registry: Electric.StackSupervisor.registry_name(new_stack_id),
+        inspector:
+          {EtsInspector,
+           stack_id: new_stack_id, server: EtsInspector.name(stack_id: new_stack_id)},
+        shape_cache: {ShapeCache, [stack_id: new_stack_id]},
+        stack_supervisor: new_stack_supervisor,
+        stack_supervisor_id: new_sup_id
+      })
+
+    # (3) New Bandit + client pointed at the new stack (no readiness wait — the
+    #     new stack is still lock-blocked). Reuse the shared Finch pool.
+    server =
+      Support.IntegrationSetup.start_bandit_client(new_ctx,
+        id: new_server_id,
+        router_opts: Keyword.get(Map.get(ctx, :electric_client_opts, []), :router_opts, []),
+        client_opts: Support.IntegrationSetup.finch_client_opts(Map.get(ctx, :finch_name))
+      )
+
+    # (4) Gracefully stop the old stack: releases the advisory lock; the
+    #     persistent slot and publication survive (no drop on a normal stop).
+    :ok = stop_supervised(old_sup_id)
+    _ = stop_supervised(old_server_id)
+
+    # (5) The new stack now acquires the lock, finds the existing slot +
+    #     publication (duplicate → no shape purge) and becomes active.
+    :ok = Electric.StatusMonitor.wait_until_active(new_stack_id, timeout: 10_000)
+
+    new_ctx
+    |> Map.merge(server)
+    |> Map.merge(%{server_id: new_server_id, rolling_gen: gen})
+  end
+
+  # Polls until the stack's registered processes are gone. Uses
+  # ProcessRegistry.alive?/2 which tolerates the per-stack registry itself
+  # being torn down (returns false rather than raising).
+  defp wait_for_stack_processes_down(stack_id, deadline \\ nil) do
+    deadline = deadline || System.monotonic_time(:millisecond) + 5_000
+
+    down? =
+      not Electric.ProcessRegistry.alive?(stack_id, Electric.StatusMonitor) and
+        not Electric.ProcessRegistry.alive?(stack_id, Electric.Connection.Manager) and
+        GenServer.whereis(Electric.ProcessRegistry.registry_name(stack_id)) == nil
+
+    cond do
+      down? ->
+        :ok
+
+      System.monotonic_time(:millisecond) > deadline ->
+        flunk("stack #{stack_id} did not fully terminate after brutal kill")
+
+      true ->
+        Process.sleep(10)
+        wait_for_stack_processes_down(stack_id, deadline)
+    end
+  end
+
+  @doc """
+  Starts an `Electric.StackSupervisor` under the test supervision tree.
+
+  Options:
+    - `:id` - ExUnit child id (default `Electric.StackSupervisor`). Must be
+      distinct when running two stacks concurrently (e.g. a rolling deploy).
+    - `:await_ready?` - block until the `{:stack_status, :ready}` event fires
+      (default `true`). Pass `false` for a stack that is expected to block
+      acquiring the replication advisory lock and therefore never reach `:ready`
+      until another stack releases it.
+  """
+  def start_stack_supervisor!(
+        ctx,
+        stack_id,
+        kv,
+        storage,
+        stack_events_registry,
+        publication_name,
+        opts \\ []
+      ) do
+    id = Keyword.get(opts, :id, Electric.StackSupervisor)
+    await_ready? = Keyword.get(opts, :await_ready?, true)
+
     ref = Electric.StackSupervisor.subscribe_to_stack_events(stack_id)
 
     connection_opts =
@@ -541,13 +743,14 @@ defmodule Support.ComponentSetup do
          shape_db_opts: [
            storage_dir: ctx.tmp_dir
          ]},
+        id: id,
         restart: :temporary,
         significant: false
       )
 
     # allow a reasonable time for full stack setup to account for
     # potential CI slowness, including PG
-    assert_receive {:stack_status, ^ref, :ready}, 2000
+    if await_ready?, do: assert_receive({:stack_status, ^ref, :ready}, 2000)
 
     stack_supervisor
   end

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

@@ -13,46 +13,86 @@ defmodule Support.IntegrationSetup do
   - `base_url` - The base URL of the server
   - `server_pid` - The Bandit server process
   - `port` - The port the server is listening on
+  - `finch_name` - the shared Finch pool name (or nil), so later HTTP servers
+    (e.g. a rolling deploy's replacement stack) can reuse it
+  - `electric_client_opts` - the opts this was called with, so a replacement
+    server can be built with the same router/client configuration
   """
   def with_electric_client(ctx, opts \\ []) do
     :ok = Electric.StatusMonitor.wait_until_active(ctx.stack_id, timeout: 2000)
 
-    router_opts = build_router_opts(ctx, Keyword.get(opts, :router_opts, []))
     num_clients = Keyword.get(opts, :num_clients, 1)
 
+    # Start a shared Finch pool once (when pooling is needed) so that a second
+    # HTTP server started later in the test — e.g. the replacement stack in a
+    # rolling deploy — can reuse it rather than starting a second Finch under
+    # the same child id.
+    finch_name =
+      if num_clients > 1 do
+        finch_name = :"Electric.Client.Finch.Test.#{System.unique_integer([:positive])}"
+
+        {:ok, _} =
+          ExUnit.Callbacks.start_supervised(
+            {Finch, name: finch_name, pools: %{default: [size: num_clients]}}
+          )
+
+        finch_name
+      end
+
+    server =
+      start_bandit_client(ctx,
+        id: Bandit,
+        router_opts: Keyword.get(opts, :router_opts, []),
+        client_opts: finch_client_opts(finch_name)
+      )
+
+    ctx
+    |> Map.merge(server)
+    |> Map.merge(%{finch_name: finch_name, electric_client_opts: opts})
+  end
+
+  @doc """
+  Starts a Bandit HTTP server bound to `ctx`'s stack and an `Electric.Client`
+  pointed at it. Unlike `with_electric_client/2` this does NOT wait for the
+  stack to be active (the caller manages readiness) and takes an explicit child
+  `:id` so several servers can coexist in one test.
+
+  Options:
+    - `:id` - Bandit child id (default `Bandit`)
+    - `:router_opts` - extra router opts merged into `build_router_opts/2`
+    - `:client_opts` - extra opts passed to `Electric.Client.new/1`
+
+  Returns `%{client:, base_url:, server_pid:, port:}`.
+  """
+  def start_bandit_client(ctx, opts \\ []) do
+    id = Keyword.get(opts, :id, Bandit)
+    router_opts = build_router_opts(ctx, Keyword.get(opts, :router_opts, []))
+    client_opts = Keyword.get(opts, :client_opts, [])
+
     {:ok, server_pid} =
       ExUnit.Callbacks.start_supervised(
         {Bandit,
          plug: {Electric.Plug.Router, router_opts},
          port: 0,
          ip: :loopback,
-         thousand_island_options: [num_acceptors: 1]}
+         thousand_island_options: [num_acceptors: 1]},
+        id: id
       )
 
     {:ok, {_ip, port}} = ThousandIsland.listener_info(server_pid)
     base_url = "http://localhost:#{port}"
 
-    client_opts =
-      if num_clients > 1 do
-        finch_name = :"Electric.Client.Finch.Test.#{System.unique_integer([:positive])}"
-
-        {:ok, _} =
-          ExUnit.Callbacks.start_supervised(
-            {Finch, name: finch_name, pools: %{default: [size: num_clients]}}
-          )
-
-        [fetch: {Electric.Client.Fetch.HTTP, [request: [finch: finch_name]]}]
-      else
-        []
-      end
-
     {:ok, client} = Electric.Client.new([base_url: base_url] ++ client_opts)
 
-    Map.merge(ctx, %{
-      client: client,
-      base_url: base_url,
-      server_pid: server_pid,
-      port: port
-    })
+    %{client: client, base_url: base_url, server_pid: server_pid, port: port}
   end
+
+  @doc """
+  Builds the `Electric.Client` opts needed to route requests through a shared
+  Finch pool, or `[]` when there is no pool.
+  """
+  def finch_client_opts(nil), do: []
+
+  def finch_client_opts(finch_name),
+    do: [fetch: {Electric.Client.Fetch.HTTP, [request: [finch: finch_name]]}]
 end