fix(sync-service): cold nested subquery shape times out subscribing to a dependency materializer (#4715) (#4729)

## Summary
A cold nested subquery shape could return an initial HTTP 500 while its
dependency snapshots were still in progress. A retry succeeded once the
dependency snapshots finished.

Fixes #4715

## Problem
When an outer shape's consumer initializes,
`Consumer.all_materializers_alive?/1` subscribes to each dependency
materializer via `Materializer.subscribe/1`. That was the **only**
materializer call using the default 5s `GenServer.call` timeout —
`wait_until_ready/1` and `new_changes/3` both use `:infinity`.

A materializer can't answer any call until
`handle_continue(:start_materializer)` returns, and that blocks on
`Consumer.await_snapshot_start(..., :infinity)` until its own snapshot
starts. For a cold nested topology, the intermediate materializer itself
waits on a deeper dependency, so its snapshot can start later than 5s
after the outer consumer subscribes. The subscribe then times out,
Electric removes the outer shape, and the client's first request 500s:

```
Removing shape "..." due to abnormal shutdown: {:timeout, {GenServer, :call, [#PID, :subscribe, 5000]}}
  ... consumer.ex: Electric.Shapes.Consumer.all_materializers_alive?/1
  ... consumer.ex: Electric.Shapes.Consumer.finish_initialization/3
```

This matches the logs reported in #4715.

## Solution
`Materializer.subscribe/*` now waits with `:infinity`, consistent with
the other materializer calls. This matches the intended semantics from
the issue — *"a single cold nested shape waits for dependency
materializers and completes without an externally visible 500."*

Why this is safe:
- **Liveness** is already handled by the caller:
`all_materializers_alive?/1` does `Process.monitor(pid, ...)` *before*
subscribing, so a dead materializer surfaces as a call exit rather than
being masked by a short timeout.
- **No deadlock**: the shape-dependency graph is a DAG (outer → inner),
so no dependency can transitively wait on the outer consumer.
- **Replication is not stalled** by the longer block: the outer shape is
registered with the `ShapeLogCollector`'s `EventRouter` (via
`SetupEffects` → `ShapeLogCollector.add_shape`) only *after*
`all_materializers_alive?/1` returns, inside `initialize_event_handler`.
While the consumer is blocked in the subscribe it is not yet a routing
target, so the collector never broadcasts a transaction to it. The
overall wait is bounded by the request-level
`await_snapshot_start(outer, 45000)` budget.

## Reproduction / Test Plan
Added a `:slow` router test (`test/electric/plug/router_test.exs`) that
deterministically reproduces the bug by stalling **only** the dependency
shape's snapshot past the 5s subscribe window (via an injected
`:create_snapshot_fn`), then asserting the cold nested shape still
serves 200.

- [x] Test fails on `main` with the `{:timeout, {GenServer, :call, [_,
:subscribe, 5000]}}` signature
- [x] Test passes with the fix
- [x] `materializer_test.exs`, `consumer_test.exs`, and the `/v1/shapes
- subqueries` router tests pass

Run the reproduction:
```sh
mix test test/electric/plug/router_test.exs --include slow --only line:3505
```

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

---------

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

Author: robacourt

.changeset/cold-nested-subquery-subscribe-timeout.md

@@ -0,0 +1,5 @@
+---
+"@core/sync-service": patch
+---
+
+Fix a cold nested subquery shape returning an initial HTTP 500 while its dependency snapshots were still in progress. When an outer shape's consumer initialized, it subscribed to each dependency materializer using `GenServer.call` with the default 5s timeout. A dependency materializer stays blocked in start-up until its own snapshot starts, so if that snapshot took longer than 5s the subscribe timed out, the outer shape was removed, and the client's first request 500'd (a retry succeeded once the snapshot finished). The subscribe now waits with `:infinity`, consistent with the other materializer calls, so a cold nested shape waits for its dependency materializers and completes without an externally visible 500. Liveness is unaffected: the caller already monitors the materializer, so a dead dependency surfaces as a call exit rather than being masked by a short timeout.

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

@@ -103,9 +103,15 @@ defmodule Electric.Shapes.Consumer.Materializer do
     end)
   end
 
-  def subscribe(pid) when is_pid(pid), do: GenServer.call(pid, :subscribe)
-
-  def subscribe(opts) when is_map(opts), do: GenServer.call(name(opts), :subscribe)
+  # A materializer cannot answer any call until `handle_continue(:start_materializer)`
+  # returns, and that blocks on `await_snapshot_start(:infinity)` until its own snapshot
+  # starts. A cold dependency snapshot can take longer than the default 5s `GenServer.call`
+  # timeout, so we wait with `:infinity` (consistent with `wait_until_ready/1` and
+  # `new_changes/3`). Liveness is handled by the caller monitoring the materializer, so a
+  # dead materializer surfaces as a call exit rather than being masked by a short timeout.
+  def subscribe(pid) when is_pid(pid), do: GenServer.call(pid, :subscribe, :infinity)
+
+  def subscribe(opts) when is_map(opts), do: GenServer.call(name(opts), :subscribe, :infinity)
 
   def subscribe(stack_id, shape_handle),
     do: subscribe(%{stack_id: stack_id, shape_handle: shape_handle})

packages/sync-service/test/electric/plug/router_test.exs

@@ -3478,6 +3478,71 @@ defmodule Electric.Plug.RouterTest do
                _ -> false
              end)
     end
+
+    # Reproduces electric-sql/electric#4715.
+    #
+    # The outer consumer subscribes to each of its dependency materializers
+    # with a hardcoded 5s `GenServer.call` timeout
+    # (`Consumer.all_materializers_alive?/1` -> `Materializer.subscribe/1`).
+    # A dependency materializer stays blocked in
+    # `handle_continue(:start_materializer)` on
+    # `Consumer.await_snapshot_start(..., :infinity)` until its own snapshot
+    # starts. If the dependency snapshot takes longer than 5s, the subscribe
+    # times out, Electric removes the outer shape, and the client's first
+    # request returns 500. A later retry succeeds once the dependency snapshot
+    # has finished.
+    #
+    # We reproduce that window deterministically by stalling only the
+    # dependency (`parent`) shape's snapshot past the 5s subscribe timeout,
+    # while leaving every other shape's snapshot untouched.
+    @tag :slow
+    @tag with_sql: [
+           "CREATE TABLE parent (id INT PRIMARY KEY, value INT NOT NULL)",
+           "CREATE TABLE child (id INT PRIMARY KEY, parent_id INT NOT NULL REFERENCES parent(id), value INT NOT NULL)",
+           "INSERT INTO parent (id, value) VALUES (1, 1), (2, 2)",
+           "INSERT INTO child (id, parent_id, value) VALUES (1, 1, 10), (2, 2, 20)"
+         ]
+    test "cold dependency shape whose materializer is slow to snapshot does not surface a 500",
+         %{opts: opts, stack_id: stack_id} do
+      test_pid = self()
+
+      # Stall the dependency (`parent`) shape's snapshot past the 5s subscribe
+      # timeout, keeping its materializer blocked in start-up. Every other
+      # shape (including the outer `child` shape) snapshots normally.
+      Electric.StackConfig.put(
+        stack_id,
+        :create_snapshot_fn,
+        fn task_parent, consumer, shape_handle, shape, snapshot_ctx ->
+          if shape.root_table == {"public", "parent"} do
+            send(test_pid, :stalling_dependency_snapshot)
+            Process.sleep(6_000)
+          end
+
+          Electric.Shapes.Consumer.Snapshotter.stream_snapshot_from_db(
+            task_parent,
+            consumer,
+            shape_handle,
+            shape,
+            snapshot_ctx
+          )
+        end
+      )
+
+      where = "parent_id IN (SELECT id FROM parent WHERE value = 1)"
+
+      conn =
+        conn("GET", "/v1/shape", %{table: "child", offset: "-1", where: where})
+        |> Router.call(opts)
+
+      # Confirm we actually exercised the stalled-dependency path.
+      assert_received :stalling_dependency_snapshot
+
+      # A cold nested shape should wait for its dependency materializers and
+      # complete successfully rather than exposing the internal subscribe
+      # timeout as a 500 to the client.
+      assert %{status: 200} = conn
+      assert [%{"value" => %{"id" => "1"}}, _] = Jason.decode!(conn.resp_body)
+    end
   end
 
   describe "/v1/shapes - subset snapshots" do