From bcdaa6e5bb47ff16edbb534c2ff8e17ded559642 Mon Sep 17 00:00:00 2001 From: rob Date: Tue, 28 Jul 2026 13:34:24 +0100 Subject: [PATCH 1/7] fix(sync-service): carry dropped txn boundaries in consumer flush alignment MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A txn_offset_mapping entry {written, boundary} promises "once the writer has flushed `written`, this shape has processed everything up to `boundary`". Subquery consumers append move-in/move-out control messages to the shape log with no boundary bookkeeping of their own, so a storage flush can land past a mapped written offset without matching it exactly. align_offset_to_txn_boundary/2 dropped such entries silently and reported the raw flushed offset, which can sit *below* the dropped boundary — e.g. splice writes end at (tx,16) with boundary (tx,18), a drained move-out control message lands at (tx,17), storage flushes (tx,17). The ShapeLogCollector's flush entry for that transaction then waits for (tx,18) forever while the consumer has nothing left to flush, pinning the stack-wide flush boundary and, since the stall watchdog landed, getting the healthy shape removed after 60s of quiet WAL — observed as spurious 409s for optimized subquery shapes in the first batch after a rolling deploy (shared dependency shapes cascade their removal into every dependent). Report max(flushed offset, max dropped boundary) instead — the same claim the exact-match path already made, generalized to non-exact flushes. Co-Authored-By: Claude Fable 5 --- .../lib/electric/shapes/consumer/state.ex | 34 ++++++++++++----- .../electric/shapes/consumer/state_test.exs | 37 +++++++++++++++++++ 2 files changed, 62 insertions(+), 9 deletions(-) diff --git a/packages/sync-service/lib/electric/shapes/consumer/state.ex b/packages/sync-service/lib/electric/shapes/consumer/state.ex index e23edee4dc..d6693df4be 100644 --- a/packages/sync-service/lib/electric/shapes/consumer/state.ex +++ b/packages/sync-service/lib/electric/shapes/consumer/state.ex @@ -181,21 +181,37 @@ defmodule Electric.Shapes.Consumer.State do defp validate_storage_capabilities(state, _storage), do: state @doc """ - For the given offset, find the appropriate transaction boundary and - remove all transactions that are less than or equal to the boundary. + For the given flushed offset, drop all mapping entries whose written offset + is covered by the flush and return the offset to report to the + ShapeLogCollector: the max of the flushed offset and the dropped entries' + transaction boundaries. + + A mapping entry `{written, boundary}` promises "once the writer has flushed + `written`, this shape has processed everything up to `boundary`". A flush can + land past `written` without matching it exactly — trailing writes that carry + no boundary of their own (subquery move-in/move-out control messages) advance + the writer beyond the mapped offset. The dropped boundaries must still be + honoured in that case: reporting only the raw flushed offset would + permanently under-report when a boundary lies above it, leaving the + ShapeLogCollector's flush entry for that transaction uncompletable. """ @spec align_offset_to_txn_boundary(t(), LogOffset.t()) :: {t(), LogOffset.t()} def align_offset_to_txn_boundary( %__MODULE__{txn_offset_mapping: txn_offset_mapping} = state, offset ) do - case Enum.drop_while(txn_offset_mapping, &(LogOffset.compare(elem(&1, 0), offset) == :lt)) do - [{^offset, boundary} | rest] -> - {%{state | txn_offset_mapping: rest}, boundary} - - rest -> - {%{state | txn_offset_mapping: rest}, offset} - end + {dropped, rest} = + Enum.split_while(txn_offset_mapping, &(LogOffset.compare(elem(&1, 0), offset) != :gt)) + + boundary = + case dropped do + # Entries are appended in write order, so the last dropped entry + # carries the highest boundary. + [] -> offset + _ -> LogOffset.max(offset, dropped |> List.last() |> elem(1)) + end + + {%{state | txn_offset_mapping: rest}, boundary} end @spec add_to_buffer(t(), TransactionFragment.t()) :: t() diff --git a/packages/sync-service/test/electric/shapes/consumer/state_test.exs b/packages/sync-service/test/electric/shapes/consumer/state_test.exs index c00bfdc199..b7dd9f3e83 100644 --- a/packages/sync-service/test/electric/shapes/consumer/state_test.exs +++ b/packages/sync-service/test/electric/shapes/consumer/state_test.exs @@ -64,6 +64,43 @@ defmodule Electric.Shapes.Consumer.StateTest do assert state.txn_offset_mapping == [] end + test "carries a dropped boundary forward when the flush lands between written offset and boundary", + %{state: state} do + # The writer wrote up to (100, 5) for a txn whose boundary is (100, 10), + # then a trailing unmapped write (e.g. a subquery move-out control + # message) landed at (100, 7) and storage flushed through it. Everything + # the shape owes through (100, 10) is durable — nothing was written in + # ((100, 5), (100, 10)] except the flushed control message — so the + # notification must be the boundary, not the raw flushed offset. + # Returning (100, 7) here permanently under-reports: the mapping entry is + # gone, no further writes are pending, and the ShapeLogCollector's flush + # entry for the txn at (100, 10) can never complete. + state = + %{state | txn_offset_mapping: [{LogOffset.new(100, 5), LogOffset.new(100, 10)}]} + + {state, result} = State.align_offset_to_txn_boundary(state, LogOffset.new(100, 7)) + + assert result == LogOffset.new(100, 10) + assert state.txn_offset_mapping == [] + end + + test "uses the max dropped boundary when multiple entries are dropped", %{state: state} do + state = + %{ + state + | txn_offset_mapping: [ + {LogOffset.new(100, 5), LogOffset.new(100, 10)}, + {LogOffset.new(200, 3), LogOffset.new(200, 8)}, + {LogOffset.new(300, 1), LogOffset.new(300, 9)} + ] + } + + {state, result} = State.align_offset_to_txn_boundary(state, LogOffset.new(200, 5)) + + assert result == LogOffset.new(200, 8) + assert state.txn_offset_mapping == [{LogOffset.new(300, 1), LogOffset.new(300, 9)}] + end + test "handles empty mapping", %{state: state} do state = %{state | txn_offset_mapping: []} offset = LogOffset.new(100, 5) From 1edaecdd8aeaffcda600c48130c827d8617bbffc Mon Sep 17 00:00:00 2001 From: rob Date: Tue, 28 Jul 2026 13:34:36 +0100 Subject: [PATCH 2/7] fix(sync-service): complete FlushTracker entries on flush at or past last_sent MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit FlushTracker cleared a shape's pending-flush entry only when the consumer's notification exactly equalled the recorded last_sent offset. Subquery consumers write move-in/move-out control messages at op offsets past the transaction's last replicated operation, so their flush notifications can overshoot last_sent — e.g. notified (tx,19) against last_sent (tx,18). The entry then survived forever with last_flushed > last_sent: the consumer owed nothing, the entry could never complete, the global flush boundary (and the slot's confirmed_flush_lsn) dragged behind, and the stall watchdog eventually removed the healthy shape. A shape's log offsets are monotone, so a flush at X covers every write at or below X: a notification at or past last_sent means the shape has durably flushed everything it was sent and is caught up. The changeset covers this and the companion consumer-side alignment fix. Co-Authored-By: Claude Fable 5 --- .../flush-tracker-overshoot-completion.md | 5 ++++ .../shape_log_collector/flush_tracker.ex | 13 ++++++++++- .../flush_tracker_test.exs | 23 +++++++++++++++++++ 3 files changed, 40 insertions(+), 1 deletion(-) create mode 100644 .changeset/flush-tracker-overshoot-completion.md diff --git a/.changeset/flush-tracker-overshoot-completion.md b/.changeset/flush-tracker-overshoot-completion.md new file mode 100644 index 0000000000..9960254832 --- /dev/null +++ b/.changeset/flush-tracker-overshoot-completion.md @@ -0,0 +1,5 @@ +--- +'@core/sync-service': patch +--- + +Fix flush-boundary accounting for shapes that append log entries past the transaction boundary (subquery move-in/move-out control messages). The consumer's flush-offset alignment now carries a dropped transaction boundary forward when the storage flush lands past the mapped written offset, and FlushTracker treats a notification at or past its recorded last-sent offset as completion instead of requiring exact equality. Previously such shapes' flush entries could pin forever in either direction, dragging the stack-wide flush boundary (WAL retention) and — since the stall watchdog was added — getting healthy shapes removed after 60s of quiet WAL, seen as spurious 409s for optimized subquery shapes after rolling deploys. diff --git a/packages/sync-service/lib/electric/replication/shape_log_collector/flush_tracker.ex b/packages/sync-service/lib/electric/replication/shape_log_collector/flush_tracker.ex index 49a858b0b6..9d189fd294 100644 --- a/packages/sync-service/lib/electric/replication/shape_log_collector/flush_tracker.ex +++ b/packages/sync-service/lib/electric/replication/shape_log_collector/flush_tracker.ex @@ -2,6 +2,8 @@ defmodule Electric.Replication.ShapeLogCollector.FlushTracker do alias Electric.Replication.LogOffset alias Electric.Replication.Changes.{Commit, TransactionFragment} + import Electric.Replication.LogOffset, only: [is_log_offset_lt: 2] + @type shape_id() :: term() defstruct [ @@ -190,7 +192,16 @@ defmodule Electric.Replication.ShapeLogCollector.FlushTracker do when is_map_key(last_flushed, shape_id) do {last_flushed, min_incomplete_flush_tree} = case Map.fetch!(last_flushed, shape_id) do - {^last_flushed_offset, prev_flushed_offset, _last_progress_at} -> + # A flush at or past last_sent means the writer has durably persisted + # everything this shape was sent, so it is caught up. Strict equality + # is not enough here: shape writers can append entries past the + # transaction's last replicated operation (e.g. subquery move-in/ + # move-out control messages at incremented op offsets), making their + # flush notification overshoot last_sent. Keeping such an entry would + # pin it forever — the consumer has nothing left to flush — dragging + # the global flush boundary and eventually tripping the stall check. + {last_sent, prev_flushed_offset, _last_progress_at} + when not is_log_offset_lt(last_flushed_offset, last_sent) -> {Map.delete(last_flushed, shape_id), min_incomplete_flush_tree |> delete_from_tree(prev_flushed_offset, shape_id)} diff --git a/packages/sync-service/test/electric/replication/shape_log_collector/flush_tracker_test.exs b/packages/sync-service/test/electric/replication/shape_log_collector/flush_tracker_test.exs index ae61a4480c..c5ffd98c5a 100644 --- a/packages/sync-service/test/electric/replication/shape_log_collector/flush_tracker_test.exs +++ b/packages/sync-service/test/electric/replication/shape_log_collector/flush_tracker_test.exs @@ -192,6 +192,29 @@ defmodule Electric.Replication.ShapeLogCollector.FlushTrackerTest do assert FlushTracker.empty?(tracker) end + test "a notification past last_sent completes the shape", %{tracker: tracker} do + tracker = handle_txn(tracker, batch(lsn: 5, last_offset: 10), ["shape1"]) + + # Subquery consumers append move-in/move-out control messages to the + # shape log after the transaction's last replicated operation, so their + # writer's flush notification can land past the offset the tracker + # recorded as last_sent. Having flushed past everything it was sent, + # the shape owes nothing and must be considered caught up. + tracker = FlushTracker.handle_flush_notification(tracker, "shape1", LogOffset.new(5, 12), 0) + + assert_receive {:flush_confirmed, 5} + assert FlushTracker.empty?(tracker) + end + + test "a notification past last_sent is not reported as stalled", %{tracker: tracker} do + tracker = handle_txn(tracker, batch(lsn: 5, last_offset: 10), ["shape1"], 100) + + tracker = + FlushTracker.handle_flush_notification(tracker, "shape1", LogOffset.new(5, 12), 150) + + assert FlushTracker.stalled_shapes(tracker, 100_000, 100) == [] + end + test "should notify flushes under continuous updates", %{tracker: tracker} do tracker |> handle_txn(batch(lsn: 10, last_offset: 10), ["shape1"]) From d2e1cd4b5d383b54d6f0e73cb3ffad72ef3971ef Mon Sep 17 00:00:00 2001 From: rob Date: Tue, 28 Jul 2026 13:34:58 +0100 Subject: [PATCH 3/7] test(sync-service): scope the shape metadata db per stack in the harness MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The shape-db sqlite lived at storage_dir/meta/shape-db with storage_dir = ctx.tmp_dir, unscoped by stack_id — so the two stacks of a rolling-deploy test (old and replacement, sharing one tmp_dir) shared one metadata database while PureFileStorage keeps their shape data in separate per-stack namespaces. The replacement stack then "restored" hundreds of shape records it had no data for: a storage layout no production deploy has (per-instance disks share neither; a shared bind-mount shares both). Non-optimized restored shapes churned through 409/refetch cycles, stretching post-restart checks by minutes, and a race let a dataless restored shape serve an empty snapshot as up-to-date. Scope the shape-db dir by stack_id, placed *beside* the PureFileStorage data namespace (tmp_dir/) rather than inside it — reset_storage/cleanup_all! trashes that whole directory and must never take the metadata db with it. Graceful and brutal restarts reuse the stack_id and therefore the same path, so restore-from-disk scenarios are unchanged. Co-Authored-By: Claude Fable 5 --- .../test/support/component_setup.ex | 101 ++++++++++-------- 1 file changed, 58 insertions(+), 43 deletions(-) diff --git a/packages/sync-service/test/support/component_setup.ex b/packages/sync-service/test/support/component_setup.ex index bb35e218fa..7c7ed65d41 100644 --- a/packages/sync-service/test/support/component_setup.ex +++ b/packages/sync-service/test/support/component_setup.ex @@ -711,49 +711,64 @@ defmodule Support.ComponentSetup do stack_supervisor = start_supervised!( - {Electric.StackSupervisor, - stack_id: stack_id, - stack_events_registry: stack_events_registry, - chunk_bytes_threshold: - Map.get( - ctx, - :chunk_size, - Electric.ShapeCache.LogChunker.default_chunk_size_threshold() - ), - persistent_kv: kv, - storage: storage, - storage_dir: ctx.tmp_dir, - connection_opts: connection_opts, - replication_opts: - Keyword.merge( - [ - connection_opts: replication_connection_opts, - slot_name: "electric_test_slot_#{:erlang.phash2(stack_id)}", - publication_name: publication_name, - try_creating_publication?: true, - slot_temporary?: true - ], - List.wrap(ctx[:replication_opts_overrides]) - ), - pool_opts: [ - backoff_type: :stop, - max_restarts: 0, - # 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, - shape_cleaner_opts: shape_cleaner_opts(ctx) - ], - manual_table_publishing?: Map.get(ctx, :manual_table_publishing?, false), - telemetry_opts: [instance_id: "test_instance", version: Electric.version()], - feature_flags: Electric.Config.get_env(:feature_flags), - shape_db_opts: [ - storage_dir: ctx.tmp_dir - ]}, + { + Electric.StackSupervisor, + # Scope the shape metadata db by stack_id so two stacks sharing a + # tmp_dir (a rolling deploy's old and new stack) don't share one + # shape-db: with a shared db the new stack "restores" the old stack's + # shape metadata while having none of its data, a storage layout no + # production deploy has (per-instance disks have neither; a shared + # bind-mount has both). The scoped dir must live *beside* the + # PureFileStorage data namespace (`tmp_dir/`), not inside + # it: `reset_storage`/`cleanup_all!` trashes that whole directory, + # which must never take the metadata db with it. The stack_id (a + # full test name) is hashed rather than embedded because sqlite's + # unix VFS caps database pathnames at 512 bytes, which long test + # names exceed. Same stack_id (graceful and brutal restarts) hashes + # to the same path, so restore-from-disk scenarios are unaffected. + stack_id: stack_id, + stack_events_registry: stack_events_registry, + chunk_bytes_threshold: + Map.get( + ctx, + :chunk_size, + Electric.ShapeCache.LogChunker.default_chunk_size_threshold() + ), + persistent_kv: kv, + storage: storage, + storage_dir: ctx.tmp_dir, + connection_opts: connection_opts, + replication_opts: + Keyword.merge( + [ + connection_opts: replication_connection_opts, + slot_name: "electric_test_slot_#{:erlang.phash2(stack_id)}", + publication_name: publication_name, + try_creating_publication?: true, + slot_temporary?: true + ], + List.wrap(ctx[:replication_opts_overrides]) + ), + pool_opts: [ + backoff_type: :stop, + max_restarts: 0, + # 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, + shape_cleaner_opts: shape_cleaner_opts(ctx) + ], + manual_table_publishing?: Map.get(ctx, :manual_table_publishing?, false), + telemetry_opts: [instance_id: "test_instance", version: Electric.version()], + feature_flags: Electric.Config.get_env(:feature_flags), + shape_db_opts: [ + storage_dir: Path.join(ctx.tmp_dir, "meta-#{:erlang.phash2(stack_id)}") + ] + }, id: id, restart: :temporary, significant: false From 2c4753ae51999b83baf979bb9fb0c1c06bd58312 Mon Sep 17 00:00:00 2001 From: rob Date: Tue, 28 Jul 2026 13:35:10 +0100 Subject: [PATCH 4/7] test(sync-service): stop compounding rolling-restart stack ids in the harness MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Each rolling restart derived the replacement stack's id from the previous generation's id, compounding the suffix ("…_roll1_roll2_…"). Two consequences: registry-name atoms eventually exceed Erlang's 255-char atom limit, crashing the replacement stack's boot around the 17th restart of a long run (StreamData then swallows the exception and re-runs the test body against a stale ctx, surfacing only as connection-refused noise); and — worse for fidelity — the slot name was hashed from the evolving id, so from the second restart onward each replacement silently created a fresh replication slot instead of contending for and taking over the original one. Only the first restart ever exercised a true rolling-deploy slot handover. Derive every generation's id and the shared slot name from the base (pre-roll) stack_id, carried through the ctx as base_stack_id: ids stay bounded and all 66 restarts of a 200-batch oracle run now perform a genuine advisory-lock handover on one slot. Co-Authored-By: Claude Fable 5 --- .../test/support/component_setup.ex | 22 ++++++++++++++----- 1 file changed, 16 insertions(+), 6 deletions(-) diff --git a/packages/sync-service/test/support/component_setup.ex b/packages/sync-service/test/support/component_setup.ex index 7c7ed65d41..1944532252 100644 --- a/packages/sync-service/test/support/component_setup.ex +++ b/packages/sync-service/test/support/component_setup.ex @@ -565,13 +565,23 @@ defmodule Support.ComponentSetup do 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)}" + # All generations derive their identity from the base (pre-roll) stack_id. + # Deriving from the *previous* generation's id instead compounds the + # suffix ("…_roll1_roll2_…"): registry-name atoms eventually exceed the + # 255-char atom limit (crashing the stack boot around gen 17 with this + # test's names), and — worse for fidelity — a slot name hashed from the + # evolving id changes every generation, so restarts 2+ would silently + # create fresh replication slots instead of contending for and taking over + # the original one. + base_stack_id = Map.get(ctx, :base_stack_id, old_stack_id) + + # The original stack derived its slot name from its stack_id in + # start_stack_supervisor!/7; force every replacement stack onto that same + # slot so old and new always contend on one advisory lock. + old_slot_name = "electric_test_slot_#{:erlang.phash2(base_stack_id)}" gen = Map.get(ctx, :rolling_gen, 0) + 1 - new_stack_id = "#{old_stack_id}_roll#{gen}" + new_stack_id = "#{base_stack_id}_roll#{gen}" new_sup_id = {Electric.StackSupervisor, new_stack_id} new_server_id = {Bandit, new_stack_id} @@ -651,7 +661,7 @@ defmodule Support.ComponentSetup do new_ctx |> Map.merge(server) - |> Map.merge(%{server_id: new_server_id, rolling_gen: gen}) + |> Map.merge(%{server_id: new_server_id, rolling_gen: gen, base_stack_id: base_stack_id}) end # Polls until the stack's registered processes are gone. Uses From 91bb9aec6972e91f00812e9710e53e4d71083bb7 Mon Sep 17 00:00:00 2001 From: rob Date: Tue, 28 Jul 2026 13:35:23 +0100 Subject: [PATCH 5/7] test(sync-service): use sized http1 Finch pools for oracle test clients MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The shared Finch pool for many-client oracle tests used the default pool protocol, which resolves to HTTP/2: every checker multiplexes onto a handful of connections and excess concurrent streams are shed with :pool_not_available, randomly starving checkers under 800 concurrent long-polls — sometimes for a full check window, failing the run as "server unreachable". An http1 pool sized to the client count queues instead of shedding. The `default` pool key applies to every origin, so replacement servers on new ports (rolling deploys) inherit the same configuration. Co-Authored-By: Claude Fable 5 --- .../sync-service/test/support/integration_setup.ex | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/packages/sync-service/test/support/integration_setup.ex b/packages/sync-service/test/support/integration_setup.ex index bb061a5f2e..f1c7099a7f 100644 --- a/packages/sync-service/test/support/integration_setup.ex +++ b/packages/sync-service/test/support/integration_setup.ex @@ -31,9 +31,19 @@ defmodule Support.IntegrationSetup do if num_clients > 1 do finch_name = :"Electric.Client.Finch.Test.#{System.unique_integer([:positive])}" + # Explicit http1: an http2 pool multiplexes every checker onto a + # handful of connections and sheds excess concurrent streams with + # `:pool_not_available`, which starves random checkers under + # many-client load (e.g. 800 concurrent long-polls in the oracle + # property tests). An http1 pool sized to the client count queues + # instead of shedding. The `default` key applies to every origin, so + # replacement servers on new ports (rolling deploys) get the same + # pool configuration. {:ok, _} = ExUnit.Callbacks.start_supervised( - {Finch, name: finch_name, pools: %{default: [size: num_clients]}} + {Finch, + name: finch_name, + pools: %{default: [size: num_clients, count: 1, protocols: [:http1]]}} ) finch_name From 36a2e20e94b290859f628b2d6c0316799d514d44 Mon Sep 17 00:00:00 2001 From: rob Date: Tue, 28 Jul 2026 13:35:23 +0100 Subject: [PATCH 6/7] test(sync-service): add ORACLE_LOG_FILE to mirror server logs to a file ExUnit's capture_log swallows all server-side logs in the long-running oracle tests, including on failure, which made server-side diagnosis of oracle failures effectively impossible. When ORACLE_LOG_FILE is set, attach a plain :logger_std_h handler writing every Logger event at or above ORACLE_LOG_LEVEL (default info) to the given file, bypassing the capture. No-op when the variable is unset. Co-Authored-By: Claude Fable 5 --- packages/sync-service/test/test_helper.exs | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/packages/sync-service/test/test_helper.exs b/packages/sync-service/test/test_helper.exs index f629e23204..7f43111edd 100644 --- a/packages/sync-service/test/test_helper.exs +++ b/packages/sync-service/test/test_helper.exs @@ -12,6 +12,25 @@ ExUnit.start( capture_log: true ) +# Debug aid: mirror all Logger events at/above ORACLE_LOG_LEVEL (default info) +# to a file, bypassing ExUnit's capture_log which otherwise swallows server +# logs in the long-running oracle tests. Enable with ORACLE_LOG_FILE=/path. +if log_file = System.get_env("ORACLE_LOG_FILE") do + level = String.to_existing_atom(System.get_env("ORACLE_LOG_LEVEL", "info")) + + :ok = + :logger.add_handler(:oracle_file_log, :logger_std_h, %{ + config: %{file: to_charlist(log_file), file_check: 100}, + level: level, + formatter: + Logger.Formatter.new( + format: "$time $metadata[$level] $message\n", + metadata: [:pid, :stack_id, :shape_handle, :failed_handle], + colors: [enabled: false] + ) + }) +end + # Start electric_client application directly, bypassing OTP's dependency resolution. # This avoids a circular dependency: electric_client has :electric as an optional dep, # which gets added to its applications list when compiled in sync-service context, From 7e8c1879f290a97fa46fb5e0c3eb9059fcc551ed Mon Sep 17 00:00:00 2001 From: rob Date: Tue, 28 Jul 2026 13:35:23 +0100 Subject: [PATCH 7/7] ci(sync-service): run the oracle restart property test with rolling deploys Adds "rolling" to the restart_type matrix alongside graceful and brutal: a replacement stack with its own storage contends on the running stack's replication slot via the advisory lock and takes over when the old stack stops, exercising slot handover, WAL redelivery into freshly created shapes, and the flush-boundary accounting fixed in the preceding commits. Co-Authored-By: Claude Fable 5 --- .github/workflows/sync_service_tests.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/sync_service_tests.yml b/.github/workflows/sync_service_tests.yml index 40402c88b2..047af693b9 100644 --- a/.github/workflows/sync_service_tests.yml +++ b/.github/workflows/sync_service_tests.yml @@ -178,7 +178,7 @@ jobs: strategy: fail-fast: false matrix: - restart_type: [graceful, brutal] + restart_type: [graceful, brutal, rolling] env: MIX_ENV: test MIX_TARGET: application