Fix flush-boundary accounting that got healthy subquery shapes killed after rolling deploys

.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.

.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

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)}

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()

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"])

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)

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/<stack_id>`), 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

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

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,