From 0e25699a35eb8686e020f3b3aadefc868ed25ea7 Mon Sep 17 00:00:00 2001 From: rob Date: Mon, 20 Jul 2026 15:44:53 +0100 Subject: [PATCH 1/2] fix(sync-service): drop subquery shapes on restart instead of restoring them --- .changeset/drop-subquery-shapes-on-restart.md | 5 + .../replication/shape_log_collector.ex | 8 + .../sync-service/lib/electric/shape_cache.ex | 66 +----- .../lib/electric/shape_cache/shape_status.ex | 58 +++++ .../replication/shape_log_collector_test.exs | 69 +----- .../test/electric/shape_cache_test.exs | 219 ++---------------- 6 files changed, 105 insertions(+), 320 deletions(-) create mode 100644 .changeset/drop-subquery-shapes-on-restart.md diff --git a/.changeset/drop-subquery-shapes-on-restart.md b/.changeset/drop-subquery-shapes-on-restart.md new file mode 100644 index 0000000000..966d3ee7d9 --- /dev/null +++ b/.changeset/drop-subquery-shapes-on-restart.md @@ -0,0 +1,5 @@ +--- +'@core/sync-service': patch +--- + +Drop shapes that involve subqueries on server restart to prevent consistency issues. 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 ff1cf9bfa3..33289197e0 100644 --- a/packages/sync-service/lib/electric/replication/shape_log_collector.ex +++ b/packages/sync-service/lib/electric/replication/shape_log_collector.ex @@ -296,6 +296,14 @@ defmodule Electric.Replication.ShapeLogCollector do fn -> start = System.monotonic_time() + # Restoring subquery shapes consistently across a restart is not yet + # implemented, so for now they are dropped rather than restored and + # clients re-request them. This is the first restore path in the shape + # subsystem to read from ShapeStatus, so pruning them here — before we + # build routing and before the shape consumers start — is the single + # point that keeps every restore path from reinstating a subquery shape. + :ok = Electric.ShapeCache.ShapeStatus.prune_subquery_shapes(state.stack_id) + {partitions, event_router, layers, count} = state.stack_id |> Electric.ShapeCache.ShapeStatus.list_shapes() diff --git a/packages/sync-service/lib/electric/shape_cache.ex b/packages/sync-service/lib/electric/shape_cache.ex index f406413a6a..fca967205e 100644 --- a/packages/sync-service/lib/electric/shape_cache.ex +++ b/packages/sync-service/lib/electric/shape_cache.ex @@ -286,13 +286,10 @@ defmodule Electric.ShapeCache do Electric.Replication.PublicationManager.wait_for_restore(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) + # Shapes involved in a subquery are dropped rather than restored on restart + # (see `ShapeStatus.prune_subquery_shapes/1`, called from ShapeLogCollector's + # restore before any routing or consumer state is rebuilt), so there is + # nothing to eagerly start here. # 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. @@ -313,61 +310,6 @@ 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 — 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, - 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 - case restore_shape_and_dependencies(handle, shape, opts) do - {:ok, _pid} -> - # await_snapshot_start/2 is a GenServer.call into the just-started - # consumer. If that consumer dies before/during the call it exits; - # left unguarded that would propagate out of handle_continue and - # crash ShapeCache before mark_as_ready — turning a single shape - # that reliably fails its snapshot into a stack-wide restart loop. - # A call timeout (the consumer is alive but wedged) exits the same - # way. In either case we can't confirm the shape's consumer came up - # subscribed-and-correct, and the eager start exists precisely to - # guarantee that consistency. Leaving the shape alive-but-unconfirmed - # would silently reintroduce the divergence this restore path fixes, - # so we purge it (mirroring restore_shape_and_dependencies' own - # clean_shape-on-failure) and let the client refetch from scratch. - try do - _ = Electric.Shapes.Consumer.await_snapshot_start(state.stack_id, handle) - catch - :exit, reason -> - Logger.warning( - "Eager subquery consumer await failed for #{handle}: #{inspect(reason)}; " <> - "purging shape to force a clean refetch" - ) - - clean_shape(handle, state.stack_id) - end - - _ -> - :ok - end - 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/shape_cache/shape_status.ex b/packages/sync-service/lib/electric/shape_cache/shape_status.ex index 8c05cbe127..0c11b57415 100644 --- a/packages/sync-service/lib/electric/shape_cache/shape_status.ex +++ b/packages/sync-service/lib/electric/shape_cache/shape_status.ex @@ -147,6 +147,64 @@ defmodule Electric.ShapeCache.ShapeStatus do end) end + @doc """ + Given a list of `{handle, shape}` pairs (e.g. from `list_shapes/1`), return the + set of handles for every shape involved in a subquery: each shape with a + non-empty `shape_dependencies` plus all of its dependency handles. + + This is the transitive closure of the subquery hierarchy — a nested dependency + that itself has a subquery matches the same filter and contributes its own + dependencies — so it covers outer shapes, intermediate dependencies and leaf + materializers. Used by `prune_subquery_shapes/1`. + """ + @spec subquery_shape_handles([{shape_handle(), Shape.t()}]) :: MapSet.t(shape_handle()) + def subquery_shape_handles(handles_and_shapes) do + for {handle, %Shape{shape_dependencies: [_ | _]} = shape} <- handles_and_shapes, + h <- [handle | shape.shape_dependencies_handles], + into: MapSet.new(), + do: h + end + + @doc """ + Remove every shape involved in a subquery (the outer shape plus its dependency + materializers) from shape metadata and on-disk storage. + + Correctly restoring a subquery shape's on-disk view together with its + dependency materializer across a restart is not yet implemented, so for now we + drop every subquery shape on restart and let clients re-request them from + scratch. This is called once at the start of the shape subsystem's restore — + from `ShapeLogCollector`'s `restore_shapes`, before it (or the shape consumers) + rebuild any state from `list_shapes/1` — so no restore path ever reinstates a + subquery shape. At that point no consumer or routing entry exists for these + shapes yet, so a direct metadata + storage delete is sufficient; there is no + consumer to stop or routing entry to clear. + """ + @spec prune_subquery_shapes(stack_id()) :: :ok + def prune_subquery_shapes(stack_id) when is_stack_id(stack_id) do + handles = + stack_id + |> list_shapes() + |> subquery_shape_handles() + + unless Enum.empty?(handles) do + Logger.notice( + "Dropping #{MapSet.size(handles)} shape(s) involved in subqueries on restart; " <> + "clients will re-request them from scratch" + ) + + stack_storage = Electric.ShapeCache.Storage.for_stack(stack_id) + + for handle <- handles do + case remove_shape(stack_id, handle) do + :ok -> Electric.ShapeCache.Storage.cleanup!(stack_storage, handle) + {:error, _reason} -> :ok + end + end + end + + :ok + end + @spec topological_sort([{shape_handle(), Shape.t()}]) :: [{shape_handle(), Shape.t()}] defp topological_sort(handles_and_shapes, acc \\ [], visited \\ MapSet.new()) defp topological_sort([], acc, _visited), do: Enum.reverse(acc) |> List.flatten() diff --git a/packages/sync-service/test/electric/replication/shape_log_collector_test.exs b/packages/sync-service/test/electric/replication/shape_log_collector_test.exs index ed449a8c48..81804316be 100644 --- a/packages/sync-service/test/electric/replication/shape_log_collector_test.exs +++ b/packages/sync-service/test/electric/replication/shape_log_collector_test.exs @@ -46,16 +46,6 @@ defmodule Electric.Replication.ShapeLogCollectorTest do @shape Shape.new!("test_table", inspector: @inspector) @shape_handle "the-shape-handle" - @subquery_inspector Support.StubInspector.new( - tables: [{1234, {"public", "test_table"}}, {5678, {"public", "parent"}}], - columns: [%{name: "id", type: "int8", type_id: {20, 1}, pk_position: 0}] - ) - @subquery_shape Shape.new!("test_table", - inspector: @subquery_inspector, - where: "id IN (SELECT id FROM public.parent)" - ) - @subquery_shape_handle "subquery-shape-handle" - def setup_log_collector(ctx) do %{stack_id: stack_id} = ctx # Start a test Registry @@ -235,60 +225,11 @@ defmodule Electric.Replication.ShapeLogCollectorTest do assert xids == [xmin] end - @tag restore_shapes: [{@subquery_shape_handle, @subquery_shape}], - inspector: @subquery_inspector - test "restored subquery shape routes via fallback before consumer seeds index", ctx do - alias Electric.Shapes.Filter.Indexes.SubqueryIndex - - # After restore, the subquery shape should be in fallback because - # no consumer has seeded the SubqueryIndex yet. - index = SubqueryIndex.for_stack(ctx.stack_id) - assert index != nil - assert SubqueryIndex.fallback?(index, @subquery_shape_handle) - - parent = self() - - consumer = - start_link_supervised!( - {Support.TransactionConsumer, - [ - id: 1, - stack_id: ctx.stack_id, - parent: parent, - shape: @subquery_shape, - shape_handle: @subquery_shape_handle, - stack_id: ctx.stack_id, - action: :restore - ]} - ) - - :ok = - Electric.Shapes.ConsumerRegistry.register_consumer( - consumer, - @subquery_shape_handle, - ctx.stack_id - ) - - xmin = 100 - lsn = Lsn.from_string("0/10") - last_log_offset = LogOffset.new(lsn, 0) - - # Any root-table change should route to the shape via fallback, - # even if the record wouldn't match the subquery membership. - txn = - complete_txn_fragment(xmin, lsn, [ - %Changes.NewRecord{ - relation: {"public", "test_table"}, - record: %{"id" => "999"}, - log_offset: last_log_offset - } - ]) - - assert :ok = ShapeLogCollector.handle_event(txn, ctx.stack_id) - - xids = Support.TransactionConsumer.assert_consume([{1, consumer}], [txn]) - assert xids == [xmin] - end + # Subquery shapes are pruned (not restored) at the start of the collector's + # restore via `ShapeStatus.prune_subquery_shapes/1`. Its effect — the subquery + # hierarchy absent from both `ShapeCache.list_shapes/1` and `active_shapes/1` + # after a real restart, while plain shapes are retained — is covered by the + # "after restart" tests in `shape_cache_test.exs`. @tag restore_shapes: [{@shape_handle, @shape}, {@shape_handle <> "-2", @shape}], inspector: @inspector diff --git a/packages/sync-service/test/electric/shape_cache_test.exs b/packages/sync-service/test/electric/shape_cache_test.exs index 856cf86e42..83ec782a39 100644 --- a/packages/sync-service/test/electric/shape_cache_test.exs +++ b/packages/sync-service/test/electric/shape_cache_test.exs @@ -1301,13 +1301,18 @@ defmodule Electric.ShapeCacheTest do ShapeCache.get_or_create_shape_handle(@shape, ctx.stack_id) end - test "restores shapes with subqueries and their materializers", ctx do + test "drops shapes with subqueries and their materializers on restart", ctx do {shape_handle, _} = ShapeCache.get_or_create_shape_handle(@shape_with_subquery, ctx.stack_id) + # A plain (non-subquery) shape alongside it, to prove the prune is selective. + {plain_handle, _} = ShapeCache.get_or_create_shape_handle(@shape, ctx.stack_id) + :started = ShapeCache.await_snapshot_start(shape_handle, ctx.stack_id) + :started = ShapeCache.await_snapshot_start(plain_handle, ctx.stack_id) - assert [{dep_handle, _}, {^shape_handle, _}] = ShapeCache.list_shapes(ctx.stack_id) + assert [{^plain_handle, _}, {dep_handle, _}, {^shape_handle, _}] = + ShapeCache.list_shapes(ctx.stack_id) # Materializer should be started assert Process.alive?( @@ -1316,42 +1321,29 @@ defmodule Electric.ShapeCacheTest do ) ) - # Register this test as the connection manager to get "consumers ready" notification restart_shape_cache(ctx) - assert [{^dep_handle, _}, {^shape_handle, _}] = ShapeCache.list_shapes(ctx.stack_id) - end - - test "restarted subquery shape reseeds the subquery index after restart", ctx do - alias Electric.Shapes.Filter.Indexes.SubqueryIndex - - {shape_handle, _} = - ShapeCache.get_or_create_shape_handle(@shape_with_subquery, ctx.stack_id) - - :started = ShapeCache.await_snapshot_start(shape_handle, ctx.stack_id) - - # Before restart: shape should have positions in the SubqueryIndex - index_before = SubqueryIndex.for_stack(ctx.stack_id) - assert index_before != nil - assert SubqueryIndex.has_positions?(index_before, shape_handle) - - restart_shape_cache(ctx) - - # After restart: the SubqueryIndex is recreated by the ShapeLogCollector. - # The consumer re-initializes and reseeds the index. - # Wait for the consumer to finish restoring. - :started = ShapeCache.await_snapshot_start(shape_handle, ctx.stack_id) - - index_after = SubqueryIndex.for_stack(ctx.stack_id) - assert index_after != nil - + # Restoring subquery shapes consistently across a restart is not yet + # implemented, so for now we prune every shape involved in a subquery (the + # outer shape and its dependency materializers) from ShapeStatus + storage, + # at the start of the ShapeLogCollector's restore, and let clients + # re-request them from scratch. The plain shape is untouched. assert wait_until( - fn -> SubqueryIndex.has_positions?(index_after, shape_handle) end, - 200 + fn -> match?([{^plain_handle, _}], ShapeCache.list_shapes(ctx.stack_id)) end, + 500 ) + + # The ShapeLogCollector rebuilds its routing indexes independently from + # ShapeStatus. Because it prunes the subquery hierarchy before building + # routing, those handles must be absent there too — not just from + # ShapeCache.list_shapes/1 — otherwise events could still be routed to them. + active = Electric.Replication.ShapeLogCollector.active_shapes(ctx.stack_id) + refute shape_handle in active + refute dep_handle in active + assert plain_handle in active end - test "restores shapes with subqueries and their materializers when backup missing", ctx do + test "drops shapes with subqueries on restart even when backup missing", ctx do {shape_handle, _} = ShapeCache.get_or_create_shape_handle(@shape_with_subquery, ctx.stack_id) @@ -1368,7 +1360,7 @@ defmodule Electric.ShapeCacheTest do restart_shape_cache(ctx) - assert [{^dep_handle, _}, {^shape_handle, _}] = ShapeCache.list_shapes(ctx.stack_id) + assert wait_until(fn -> ShapeCache.list_shapes(ctx.stack_id) == [] end, 500) end defp restart_shape_cache(ctx, opts \\ []) do @@ -1399,144 +1391,6 @@ 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 - - test "purges the shape when the eager consumer await fails", ctx do - %{stack_id: stack_id} = ctx - test_pid = self() - - {:ok, shape_handle} = ShapeStatus.add_shape(stack_id, @shape_with_subquery) - - # Make the consumer start succeed so restore_shape_and_dependencies - # returns {:ok, pid} and the eager-start reaches await_snapshot_start. - 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, _config -> {:ok, self()} end - ) - - # ...then make the await fail (consumer died, or timed out while - # wedged) so the catch fires. We can't confirm the shape came up - # subscribed-and-correct, so it must be purged rather than left alive. - Repatch.patch( - Electric.Shapes.Consumer, - :await_snapshot_start, - [mode: :shared], - fn _stack_id, _handle -> exit(:test_consumer_died) end - ) - - # clean_shape goes through ShapeCleaner.remove_shape; capture the call - # to prove the failed shape is purged so a client refetches from scratch. - Repatch.patch( - Electric.ShapeCache.ShapeCleaner, - :remove_shape, - [mode: :shared], - fn _stack_id, handle -> - send(test_pid, {:remove_shape_called, handle}) - :ok - end - ) - - activate_mocks_for_descendant_procs(Electric.ShapeCache) - - with_shape_cache(ctx) - - assert_receive {:remove_shape_called, ^shape_handle}, 5_000 - end - end - describe "start_consumer_for_handle/2" do setup [ :with_noop_publication_manager, @@ -1574,29 +1428,6 @@ defmodule Electric.ShapeCacheTest do assert Process.alive?( GenServer.whereis(Electric.Shapes.Consumer.Materializer.name(stack_id, dep_handle)) ) - - restart_shape_cache(ctx) - - assert [{^dep_handle, _}, {^shape_handle, _}] = ShapeCache.list_shapes(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( - 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, - 1000 - ) - - # Materializer should be started - assert Process.alive?( - GenServer.whereis(Electric.Shapes.Consumer.Materializer.name(stack_id, dep_handle)) - ) end end From a2b11a53982708fe5563cd3022e2b90f8611bd80 Mon Sep 17 00:00:00 2001 From: rob Date: Mon, 27 Jul 2026 16:06:04 +0100 Subject: [PATCH 2/2] ci(sync-service): add oracle property test job with server restarts --- .github/workflows/sync_service_tests.yml | 49 +++++++++++++++++++ .../test/support/component_setup.ex | 13 ++++- 2 files changed, 61 insertions(+), 1 deletion(-) diff --git a/.github/workflows/sync_service_tests.yml b/.github/workflows/sync_service_tests.yml index 20ce9e5d1a..40402c88b2 100644 --- a/.github/workflows/sync_service_tests.yml +++ b/.github/workflows/sync_service_tests.yml @@ -169,6 +169,55 @@ jobs: - *upload_test_results_to_codecov + oracle_property_test_with_restarts: + name: 'Oracle property test with restarts (${{ matrix.restart_type }})' + runs-on: blacksmith-4vcpu-ubuntu-2404 + defaults: + run: + working-directory: packages/sync-service + strategy: + fail-fast: false + matrix: + restart_type: [graceful, brutal] + env: + MIX_ENV: test + MIX_TARGET: application + POSTGRES_VERSION: '170000' + CODECOV_FLAGS: elixir,oracle-tests,sync-service,postgres-170000 + CODECOV_TEST_RESULTS_FILES: ./junit/regular-test-junit-report.xml + CHECK_TIMEOUT: 60000 + SHAPE_COUNT: 200 + MUTATIONS_PER_TXN: 10 + TXNS_PER_BATCH: 10 + BATCH_COUNT: 50 + SKIP_REPATCH_PREWARM: 'true' + RESTART_SERVER_EVERY: 3 + RESTART_TYPE: ${{ matrix.restart_type }} + TEST_POOL_SIZE: 20 + services: + postgres: + image: 'ghcr.io/${{ github.repository }}/postgres:17-alpine-logical' + env: + POSTGRES_PASSWORD: password + options: *postgres_health_check + ports: + - 54321:5432 + + pgbouncer: *pgbouncer_service + steps: + - *checkout_source + - *seed_database + - *setup_beam + - *cache_dependencies + - *cache_compiled_code + - *install_dependencies + - *compile_package + + - name: Run oracle property test with ${{ matrix.restart_type }} server restarts + run: mix test --only oracle test/integration/oracle_property_test.exs + + - *upload_test_results_to_codecov + performance_test: name: 'Performance test, pg17' runs-on: blacksmith-4vcpu-ubuntu-2404 diff --git a/packages/sync-service/test/support/component_setup.ex b/packages/sync-service/test/support/component_setup.ex index 7acec725cd..bb35e218fa 100644 --- a/packages/sync-service/test/support/component_setup.ex +++ b/packages/sync-service/test/support/component_setup.ex @@ -419,6 +419,13 @@ defmodule Support.ComponentSetup do [on_cleanup: on_cleanup] end + defp env_pool_size do + case System.get_env("TEST_POOL_SIZE") do + nil -> 2 + value -> String.to_integer(value) + end + end + def with_complete_stack(ctx) do stack_id = full_test_name(ctx) @@ -731,7 +738,11 @@ defmodule Support.ComponentSetup do pool_opts: [ backoff_type: :stop, max_restarts: 0, - pool_size: 2 + # Default of 2 leaves a snapshot pool of 1 connection + # (Connection.Manager.pool_sizes/1), which under many-shape loads + # (e.g. the oracle property test) starves move-in queries into + # queue timeouts and 409 load-shedding. Override for such tests. + pool_size: Map.get(ctx, :db_pool_size, env_pool_size()) ], tweaks: [ registry_partitions: 1,