diff --git a/packages/sync-service/test/integration/oracle_property_test.exs b/packages/sync-service/test/integration/oracle_property_test.exs
index 643fbd84ef..9729f80bc9 100644
--- a/packages/sync-service/test/integration/oracle_property_test.exs
+++ b/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
diff --git a/packages/sync-service/test/support/component_setup.ex b/packages/sync-service/test/support/component_setup.ex
index 5ba6348cc9..38ab0ce917 100644
--- a/packages/sync-service/test/support/component_setup.ex
+++ b/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
diff --git a/packages/sync-service/test/support/integration_setup.ex b/packages/sync-service/test/support/integration_setup.ex
index 2873c80bbb..bb061a5f2e 100644
--- a/packages/sync-service/test/support/integration_setup.ex
+++ b/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
diff --git a/packages/sync-service/test/support/oracle_harness.ex b/packages/sync-service/test/support/oracle_harness.ex
index c9b5d92cf7..37103e054a 100644
--- a/packages/sync-service/test/support/oracle_harness.ex
+++ b/packages/sync-service/test/support/oracle_harness.ex
@@ -56,7 +56,10 @@ 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,
+      restart_type: env_string("RESTART_TYPE") || "graceful"
+    ]
   end
 
   @doc """
@@ -73,6 +76,9 @@ defmodule Support.OracleHarness do
     - :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_type - how the server restart is performed: "graceful" (default),
+      "brutal" (crash + recover), or "rolling" (rolling deploy). See
+      `restart_stack/2`. Env: RESTART_TYPE.
     - :restart_client_every - throw away and recreate the shape clients every
       M batches to exercise fresh-poll consistency (default: 0, disabled)
   """
@@ -82,6 +88,7 @@ defmodule Support.OracleHarness do
     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
+    restart_type = opts[:restart_type] || "graceful"
 
     log_test_config(shapes, batches)
 
@@ -121,7 +128,15 @@ defmodule Support.OracleHarness do
             # 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_server(
+              ctx,
+              pids,
+              shapes,
+              oracle_pool,
+              timeout_ms,
+              batch_idx,
+              restart_type
+            )
 
           restart_client? ->
             new_pids = recreate_checkers(ctx, pids, shapes, oracle_pool, timeout_ms, batch_idx)
@@ -172,18 +187,39 @@ defmodule Support.OracleHarness do
   # 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}")
+  defp restart_server(ctx, pids, shapes, oracle_pool, timeout_ms, batch_idx, restart_type) do
+    log("Restarting server (#{restart_type}) after batch_#{batch_idx}")
 
     Enum.each(pids, &GenServer.stop/1)
 
-    new_ctx = Map.merge(ctx, Support.ComponentSetup.restart_complete_stack(ctx))
+    new_ctx = Map.merge(ctx, restart_stack(restart_type, ctx))
 
     new_pids = recreate_checkers(new_ctx, [], shapes, oracle_pool, timeout_ms, batch_idx)
 
     {new_pids, new_ctx}
   end
 
+  # Restarts the Electric stack according to RESTART_TYPE, returning a ctx-merge
+  # map. Each variant is implemented in `Support.ComponentSetup`:
+  #   - "graceful" - clean stop + restore from disk (default)
+  #   - "brutal"   - kill -9 style crash, then recover the same stack_id
+  #   - "rolling"  - rolling deploy: a new stack takes over the replication slot
+  #                  before the old one is stopped
+  defp restart_stack("graceful", ctx), do: Support.ComponentSetup.restart_complete_stack(ctx)
+
+  defp restart_stack("brutal", ctx),
+    do: Support.ComponentSetup.brutally_restart_complete_stack(ctx)
+
+  defp restart_stack("rolling", ctx),
+    do: Support.ComponentSetup.rolling_restart_complete_stack(ctx)
+
+  defp restart_stack(other, _ctx),
+    do:
+      raise(
+        ArgumentError,
+        "unknown RESTART_TYPE=#{inspect(other)} (expected \"graceful\", \"brutal\" or \"rolling\")"
+      )
+
   # 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.
@@ -305,6 +341,18 @@ defmodule Support.OracleHarness do
     end
   end
 
+  @doc """
+  Reads an environment variable as a string.
+  Returns nil if unset or empty.
+  """
+  def env_string(name) do
+    case System.get_env(name) do
+      nil -> nil
+      "" -> nil
+      value -> value
+    end
+  end
+
   defp log(message) do
     IO.puts("[oracle] #{message}")
   end
diff --git a/packages/sync-service/test/support/oracle_harness/shape_checker.ex b/packages/sync-service/test/support/oracle_harness/shape_checker.ex
index 82161fc2cb..8f4f856349 100644
--- a/packages/sync-service/test/support/oracle_harness/shape_checker.ex
+++ b/packages/sync-service/test/support/oracle_harness/shape_checker.ex
@@ -43,9 +43,16 @@ defmodule Support.OracleHarness.ShapeChecker do
     poll_state: nil,
     rows: %{},
     # Cached oracle state from previous check (used as "before" for next check)
-    oracle_before: nil
+    oracle_before: nil,
+    # When true, transient poll errors (5xx / connection errors) are retried
+    # within timeout_ms instead of failing immediately. Opt-in via
+    # RETRY_TRANSIENT_ERRORS; see do_await/3.
+    retry_transient_errors?: false
   ]
 
+  # Backoff between retries of a transient poll error.
+  @transient_retry_backoff_ms 50
+
   # ---------------------------------------------------------------------------
   # Public API
   # ---------------------------------------------------------------------------
@@ -101,7 +108,8 @@ defmodule Support.OracleHarness.ShapeChecker do
       shape_def: shape_def,
       oracle_pool: oracle_pool,
       timeout_ms: timeout_ms,
-      poll_state: ShapeState.new()
+      poll_state: ShapeState.new(),
+      retry_transient_errors?: retry_transient_errors?()
     }
 
     {:ok, state}
@@ -148,14 +156,27 @@ defmodule Support.OracleHarness.ShapeChecker do
   # ---------------------------------------------------------------------------
 
   defp await_up_to_date(state) do
-    do_await(state, System.monotonic_time(:millisecond))
+    # polled_ok? tracks whether at least one poll completed (vs only transient
+    # errors) in this window, so a persistently-down server can't pass silently.
+    do_await(state, System.monotonic_time(:millisecond), false)
   end
 
-  defp do_await(state, start_ms) do
+  defp do_await(state, start_ms, polled_ok?) do
     elapsed = System.monotonic_time(:millisecond) - start_ms
 
     if elapsed >= state.timeout_ms do
-      # Timed out — return state and let assert_consistent! report the mismatch
+      # When retrying transient errors, a window that timed out without a single
+      # successful poll means the server was unreachable the whole time. Returning
+      # stale state would let a genuinely-down server pass if the oracle happens to
+      # be unchanged for this check, so fail explicitly instead. Otherwise return
+      # state and let assert_consistent! report any data mismatch.
+      if state.retry_transient_errors? and not polled_ok? do
+        flunk(
+          "shape=#{state.name}: no successful poll within #{state.timeout_ms}ms " <>
+            "(server unreachable for the whole window)"
+        )
+      end
+
       state
     else
       case Client.poll(state.client, state.shape_def, state.poll_state, replica: :full) do
@@ -166,7 +187,7 @@ defmodule Support.OracleHarness.ShapeChecker do
           if new_state.up_to_date? do
             handle_up_to_date(state, start_ms)
           else
-            do_await(state, start_ms)
+            do_await(state, start_ms, true)
           end
 
         {:must_refetch, messages, new_state} ->
@@ -178,14 +199,42 @@ defmodule Support.OracleHarness.ShapeChecker do
 
           state = %{state | poll_state: new_state, rows: %{}}
           state = apply_messages(state, messages)
-          do_await(state, start_ms)
+          do_await(state, start_ms, true)
 
         {:error, error} ->
-          flunk("Poll error for shape=#{state.name} where=#{state.where}: #{inspect(error)}")
+          # A 5xx or a connection error is an availability signal, never a
+          # data-consistency one. Production hides restart/deploy windows from
+          # clients (HTTP drains before teardown; the LB switches after the new
+          # instance is ready), so a transient error here does not reflect a
+          # production bug. When opted in, retry within timeout_ms; a genuine
+          # persistent error still fails via the no-successful-poll guard above
+          # (or, once some poll succeeded, as a consistency mismatch). 409s are
+          # handled separately (:must_refetch) and 4xx errors fail immediately.
+          if state.retry_transient_errors? and transient_error?(error) do
+            Process.sleep(@transient_retry_backoff_ms)
+            do_await(state, start_ms, polled_ok?)
+          else
+            flunk("Poll error for shape=#{state.name} where=#{state.where}: #{inspect(error)}")
+          end
       end
     end
   end
 
+  # A 5xx response or a transport-level failure (no HTTP status) is transient.
+  # Any other HTTP status (e.g. 4xx) is treated as a real error.
+  defp transient_error?(%Client.Error{resp: %{status: status}}) when is_integer(status),
+    do: status in 500..599
+
+  defp transient_error?(%Client.Error{}), do: true
+  defp transient_error?(_), do: false
+
+  defp retry_transient_errors? do
+    case System.get_env("RETRY_TRANSIENT_ERRORS") do
+      value when value in ["1", "true", "TRUE", "yes", "YES"] -> true
+      _ -> false
+    end
+  end
+
   defp handle_up_to_date(state, start_ms) do
     oracle_rows = query_oracle(state)
     materialized = materialized_rows(state)
@@ -194,7 +243,7 @@ defmodule Support.OracleHarness.ShapeChecker do
       state
     else
       # Electric reported up_to_date but data doesn't match yet - keep polling
-      do_await(state, start_ms)
+      do_await(state, start_ms, true)
     end
   end