Prevent shape removal and request crashes during server restart

.changeset/restore-shutdown-shape-removal.md

@@ -0,0 +1,6 @@
+---
+'@core/sync-service': patch
+---
+
+Stop subquery shapes from being spuriously removed (and held requests from
+crashing) during a server restart.

packages/sync-service/lib/electric/shape_cache/shape_status.ex

@@ -291,6 +291,14 @@ defmodule Electric.ShapeCache.ShapeStatus do
       [] ->
         :error
     end
+  rescue
+    # The shape_meta_table is created by ShapeStatusOwner. During a stack
+    # restart there's a window when the old owner's table has been freed
+    # and the new owner hasn't recreated it yet. An inflight HTTP request
+    # held by Bandit can wake up in that window. Treat a missing table
+    # the same as a missing entry — the caller can fall back to
+    # fetch_handle_by_shape (which goes through the live SQLite store).
+    ArgumentError -> :error
   end
 
   @spec mark_snapshot_started(stack_id(), shape_handle()) :: :ok | :error

packages/sync-service/lib/electric/shapes/api.ex

@@ -970,6 +970,13 @@ defmodule Electric.Shapes.Api do
       _ ->
         :no_change
     end
+  rescue
+    # During a stack restart, the per-stack ETS tables this path reads
+    # (shape_meta_table, write_buffer_shapes, …) may be transiently absent
+    # for a held long-poll that wakes up between the old owner dying and
+    # the new owner recreating the table. Treat as no_change rather than
+    # bubbling an ArgumentError up to the request handler.
+    ArgumentError -> :no_change
   end
 
   defp stream_sse_events(%Request{} = request) do

packages/sync-service/lib/electric/shapes/consumer.ex

@@ -1037,6 +1037,18 @@ defmodule Electric.Shapes.Consumer do
          opts
        ) do
     Materializer.new_changes(Map.take(state, [:stack_id, :shape_handle]), changes_or_bounds, opts)
+  catch
+    # The consumer monitors the materializer; if the materializer died the
+    # :DOWN message is already in our mailbox and handle_materializer_down/2
+    # will run after the current handle_event/handle_call completes.
+    # Treat a `:noproc` (or transient `:normal`/`:shutdown` exit) here as
+    # the same condition: don't crash the consumer (which would route into
+    # the abnormal-shutdown path of handle_writer_termination and remove
+    # the shape from disk).
+    :exit, {:noproc, _} -> :ok
+    :exit, :noproc -> :ok
+    :exit, {:normal, _} -> :ok
+    :exit, {:shutdown, _} -> :ok
   end
 
   defp notify_materializer_of_new_changes(_state, _changes_or_bounds, _opts), do: :ok

packages/sync-service/test/electric/shape_cache/shape_status_test.exs

@@ -141,6 +141,18 @@ defmodule Electric.ShapeCache.ShapeStatusTest do
     assert :error = ShapeStatus.validate_shape_handle(state, shape_handle1, shape2)
   end
 
+  test "validate_shape_handle/3 returns :error when the meta table is absent" do
+    # During a stack restart, ShapeStatusOwner frees the shape_meta_table and
+    # the new owner recreates it. An inflight request held by Bandit can wake
+    # up in that window and call validate_shape_handle while the table is gone.
+    # It must return :error (so the caller can fall back to the live SQLite
+    # store) rather than raising an ArgumentError that crashes the request.
+    absent_stack_id = "absent-stack-#{System.unique_integer([:positive])}"
+
+    assert :error =
+             ShapeStatus.validate_shape_handle(absent_stack_id, "any-handle", shape!("one"))
+  end
+
   describe "list_shapes/2" do
     test "returns shapes with dependencies in a topological order", ctx do
       {:ok, state, []} = new_state(ctx)

packages/sync-service/test/electric/shapes/api_test.exs

@@ -924,6 +924,55 @@ defmodule Electric.Shapes.ApiTest do
              }
     end
 
+    @tag long_poll_timeout: 200
+    test "disk-update check survives ETS tables vanishing during a restart", ctx do
+      # During a stack restart the per-stack ETS tables read by
+      # check_for_disk_updates can be transiently absent for a held long-poll
+      # that wakes up in the restart window. resolve_shape_handle then raises
+      # ArgumentError. The rescue must turn that into :no_change so the request
+      # completes gracefully (here: the normal out-of-bounds 400) instead of
+      # crashing the request handler with the ArgumentError.
+      expect_shape_cache(
+        resolve_shape_handle: fn @test_shape_handle, @test_shape, _stack_id, _opts ->
+          {@test_shape_handle, @test_offset}
+        end,
+        resolve_shape_handle: fn @test_shape_handle, @test_shape, _stack_id, _opts ->
+          {@test_shape_handle, @test_offset}
+        end,
+        # check_for_disk_updates at the out-of-bounds timeout — the ETS table
+        # has been freed by the restarting stack, so the lookup raises.
+        resolve_shape_handle: fn @test_shape_handle, @test_shape, _stack_id, _opts ->
+          raise ArgumentError, "shape_meta_table gone (restart window)"
+        end
+      )
+
+      out_of_bounds_offset = LogOffset.increment(@test_offset)
+
+      assert {:ok, request} =
+               Api.validate(
+                 ctx.api,
+                 %{
+                   table: "public.users",
+                   handle: "#{@test_shape_handle}",
+                   offset: "#{out_of_bounds_offset}",
+                   live: true
+                 }
+               )
+
+      assert response = Api.serve_shape_response(request)
+
+      # With the rescue the raised ArgumentError becomes :no_change, so the
+      # out-of-bounds branch returns its normal 400 instead of crashing.
+      assert response.status == 400
+
+      assert response_body(response) == %{
+               message: "Invalid request",
+               errors: %{
+                 offset: ["out of bounds for this shape"]
+               }
+             }
+    end
+
     for live <- [true, false] do
       @live live
       test "live=#{@live} request recovers from out of bounds offset via subscription", ctx do

packages/sync-service/test/electric/shapes/consumer_test.exs

@@ -2617,6 +2617,64 @@ defmodule Electric.Shapes.ConsumerTest do
       # After cleanup, the shape's rows should be removed from the index
       refute SubqueryIndex.has_positions?(index, shape_handle)
     end
+
+    test "dependency consumer survives a :noproc from its materializer without removing the shape",
+         ctx do
+      # Bug 6 cascade route: during a stack restart's shutdown, the dependency
+      # consumer's inline call into its materializer can race the
+      # materializer's death and exit with :noproc. Without the catch in
+      # notify_materializer_of_new_changes/3, that crashes the consumer with a
+      # non-shutdown reason, which routes through handle_writer_termination and
+      # removes the shape from disk — mid stack-shutdown that leaves the shape
+      # half-removed and 409s on the next poll after restart. The catch must
+      # absorb the exit so the pending :DOWN can drive a clean stop instead.
+
+      # Make the dependency consumer's notification call into the materializer
+      # exit exactly as a GenServer.call to an already-dead process would.
+      Repatch.patch(Consumer.Materializer, :new_changes, [mode: :shared], fn _, _, _ ->
+        exit({:noproc, {GenServer, :call, [:materializer, :new_changes, 5000]}})
+      end)
+
+      Support.TestUtils.activate_mocks_for_descendant_procs(Consumer)
+
+      # If the bug were present the consumer would crash and remove the shape;
+      # assert remove_shape is never called.
+      patch_shape_status(
+        remove_shape: fn _, handle ->
+          raise "Unexpected remove_shape for #{handle}"
+        end
+      )
+
+      {shape_handle, _} =
+        ShapeCache.get_or_create_shape_handle(@shape_with_subquery, ctx.stack_id)
+
+      :started = ShapeCache.await_snapshot_start(shape_handle, ctx.stack_id)
+
+      {:ok, shape} = Electric.Shapes.fetch_shape_by_handle(ctx.stack_id, shape_handle)
+      [dep_handle] = shape.shape_dependencies_handles
+
+      dep_consumer = Consumer.whereis(ctx.stack_id, dep_handle)
+      assert is_pid(dep_consumer)
+      ref = Process.monitor(dep_consumer)
+
+      # A change to the dependency table makes the dependency consumer notify
+      # its materializer — hitting the patched, exiting call.
+      ShapeLogCollector.handle_event(
+        complete_txn_fragment(100, Lsn.from_integer(50), [
+          %Changes.NewRecord{
+            relation: {"public", "other_table"},
+            record: %{"id" => "1"},
+            log_offset: LogOffset.new(Lsn.from_integer(50), 0)
+          }
+        ]),
+        ctx.stack_id
+      )
+
+      # With the catch, the dependency consumer absorbs the :noproc and stays
+      # alive; the shape is not removed.
+      refute_receive {:DOWN, ^ref, :process, _, _}, 500
+      assert Consumer.whereis(ctx.stack_id, dep_handle) == dep_consumer
+    end
   end
 
   defp refute_storage_calls_for_txn_fragment(shape_handle) do