Address review: catch-all for stale replies, mode tests, :hang doc note

- Add catch-all handle_info for stale {ref, _} replies (defensive, matches
  StatusMonitor convention)
- Add tests for set_event_processing_mode :suspend, :crash, and :hang
- Add doc note that :hang is only recoverable by killing the process

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: msfstef

packages/sync-service/lib/electric/postgres/replication_client.ex

@@ -409,6 +409,11 @@ defmodule Electric.Postgres.ReplicationClient do
     end
   end
 
+  # Stale $gen_call reply from a previous event processing cycle — ignore.
+  def handle_info({ref, _result}, state) when is_reference(ref) do
+    {:noreply, state}
+  end
+
   # This callback is invoked when the connection process receives a shutdown signal.
   def handle_info({:EXIT, _pid, :shutdown}, _state) do
     Logger.debug("Replication client #{inspect(self())} received shutdown signal, stopping")

packages/sync-service/lib/electric/replication/shape_log_collector.ex

@@ -130,7 +130,7 @@ defmodule Electric.Replication.ShapeLogCollector do
 
   Modes:
   - `:suspend` — return `{:error, :not_ready}` without processing
-  - `:hang` — accept the event but never reply (simulates slow handler)
+  - `:hang` — accept the event but never reply (simulates slow handler; only recoverable by killing the process)
   - `:crash` — raise an error on the next event
   - `:normal` — process events normally (clears any test mode)
   """

packages/sync-service/test/electric/replication/shape_log_collector_test.exs

@@ -101,6 +101,37 @@ defmodule Electric.Replication.ShapeLogCollectorTest do
     end
   end
 
+  describe "set_event_processing_mode" do
+    setup :setup_log_collector
+
+    test ":suspend causes handle_event to return {:error, :not_ready}", ctx do
+      ShapeLogCollector.set_event_processing_mode(ctx.stack_id, :suspend)
+      txn = transaction(100, Lsn.from_string("0/10"), [])
+      assert {:error, :not_ready} = ShapeLogCollector.handle_event(txn, ctx.stack_id)
+
+      ShapeLogCollector.set_event_processing_mode(ctx.stack_id, :normal)
+      assert :ok = ShapeLogCollector.handle_event(txn, ctx.stack_id)
+    end
+
+    test ":crash causes handle_event to exit the caller", ctx do
+      ShapeLogCollector.set_event_processing_mode(ctx.stack_id, :crash)
+      txn = transaction(100, Lsn.from_string("0/10"), [])
+
+      assert catch_exit(ShapeLogCollector.handle_event(txn, ctx.stack_id))
+    end
+
+    test ":hang causes handle_event to block indefinitely", ctx do
+      ShapeLogCollector.set_event_processing_mode(ctx.stack_id, :hang)
+      txn = transaction(100, Lsn.from_string("0/10"), [])
+
+      task = Task.async(fn -> ShapeLogCollector.handle_event(txn, ctx.stack_id) end)
+      refute_receive _, 100
+      # The task is stuck — verify it hasn't completed
+      assert nil == Task.yield(task, 0)
+      Task.shutdown(task, :brutal_kill)
+    end
+  end
+
   describe "shape restoration" do
     setup :setup_log_collector