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/.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 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/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/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"]) 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) diff --git a/packages/sync-service/test/support/component_setup.ex b/packages/sync-service/test/support/component_setup.ex index bb35e218fa..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 @@ -711,49 +721,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 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 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,