fix(sync-service): don't suspend a consumer mid-transaction (#4501) (#4503)

## Summary

Fixes #4501 — `KeyError: key :consider_flushed? not found in: nil` in
`Electric.Shapes.Consumer.process_txn_fragment/2`.

A shape consumer could **suspend (terminate to reclaim memory) on its
idle timeout while still holding a `pending_txn`** for an in-flight
multi-fragment transaction. The producer's `EventRouter` tracks "this
shape already saw the begin for the current xid" keyed by
`shape_handle`, independently of consumer liveness — so when a later
fragment of that transaction arrived, `ConsumerRegistry` started a fresh
consumer and delivered a `has_begin?: false` fragment to it. The fresh
consumer has `pending_txn: nil`, so `process_txn_fragment/2`
dereferenced `nil` at `txn.consider_flushed?`.

## Fix

`consumer_can_suspend?/1` now also requires `is_nil(state.pending_txn)`.
A consumer that is mid-transaction hibernates instead of suspending, and
only suspends once the transaction completes and it goes idle again.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Oleksii Sholik <oleksii@sholik.dev>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: 3 people

.changeset/fix-suspend-mid-txn.md

@@ -0,0 +1,5 @@
+---
+"@core/sync-service": patch
+---
+
+Fix a `KeyError: key :consider_flushed? not found in: nil` crash in `Electric.Shapes.Consumer`. `consumer_can_suspend?/1` now refuses to suspend while a transaction is pending, so the consumer hibernates instead and suspends only once the transaction completes.

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

@@ -399,6 +399,7 @@ defmodule Electric.Shapes.Consumer do
     # 3. we are not part of a subquery dependency tree, that is either
     #   a. we have no dependent shapes
     #   b. we don't have a materializer subscribed
+    # 4. we're not in the middle of processing a multi-fragment transaction
 
     if consumer_suspend_enabled?(state) and consumer_can_suspend?(state) do
       Logger.debug(fn -> ["Suspending consumer ", to_string(state.shape_handle)] end)
@@ -416,7 +417,7 @@ defmodule Electric.Shapes.Consumer do
 
   defp consumer_can_suspend?(state) do
     is_snapshot_started(state) and not Shape.has_dependencies(state.shape) and
-      not state.materializer_subscribed?
+      not state.materializer_subscribed? and is_nil(state.pending_txn)
   end
 
   @impl GenServer

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

@@ -1461,6 +1461,76 @@ defmodule Electric.Shapes.ConsumerTest do
       assert is_pid(Consumer.whereis(ctx.stack_id, shape_handle))
     end
 
+    @tag hibernate_after: 10, with_pure_file_storage_opts: [flush_period: 1]
+    @tag suspend: true
+    test "should hibernate not suspend while a multi-fragment transaction is pending", ctx do
+      register_as_replication_client(ctx.stack_id)
+
+      {shape_handle, _} = ShapeCache.get_or_create_shape_handle(@shape1, ctx.stack_id)
+      lsn1 = Lsn.from_integer(300)
+
+      :started = ShapeCache.await_snapshot_start(shape_handle, ctx.stack_id)
+
+      consumer_pid = Consumer.whereis(ctx.stack_id, shape_handle)
+      assert is_pid(consumer_pid)
+      ref = Process.monitor(consumer_pid)
+
+      # The begin fragment of a multi-fragment transaction leaves the consumer
+      # holding a pending_txn until the matching commit fragment arrives.
+      begin_fragment =
+        txn_fragment(
+          2,
+          lsn1,
+          [
+            %Changes.NewRecord{
+              relation: {"public", "test_table"},
+              record: %{"id" => "21"},
+              log_offset: LogOffset.new(lsn1, 0)
+            }
+          ],
+          has_begin?: true,
+          has_commit?: false
+        )
+
+      assert :ok = ShapeLogCollector.handle_event(begin_fragment, ctx.stack_id)
+
+      # The idle timer (hibernate_after: 10ms) fires, but with a transaction still
+      # pending the consumer must hibernate rather than suspend, so it survives to
+      # receive the rest of the transaction. Suspending here would drop pending_txn
+      # and crash on the next fragment (issue #4501).
+      refute_receive {:DOWN, ^ref, :process, ^consumer_pid, {:shutdown, :suspend}}, 400
+
+      assert {:current_function, {:gen_server, :loop_hibernate, 4}} =
+               Process.info(consumer_pid, :current_function)
+
+      assert is_pid(Consumer.whereis(ctx.stack_id, shape_handle))
+
+      # Completing the transaction clears pending_txn, so the consumer is free to
+      # suspend on the next idle timeout.
+      commit_fragment =
+        txn_fragment(
+          2,
+          lsn1,
+          [
+            %Changes.NewRecord{
+              relation: {"public", "test_table"},
+              record: %{"id" => "22"},
+              log_offset: LogOffset.new(lsn1, 1)
+            }
+          ],
+          has_begin?: false,
+          has_commit?: true
+        )
+
+      assert :ok = ShapeLogCollector.handle_event(commit_fragment, ctx.stack_id)
+
+      assert_receive {:flush_boundary_updated, 300}, 1_000
+
+      assert_receive {:DOWN, ^ref, :process, ^consumer_pid, {:shutdown, :suspend}}
+
+      refute Consumer.whereis(ctx.stack_id, shape_handle)
+    end
+
     @tag with_pure_file_storage_opts: [flush_period: 1]
     @tag suspend: false
     test "ConsumerRegistry.enable_suspend should suspend hibernated consumers", ctx do