Resilient Temporal client: retry dead channels, concurrent RPC I/O

[acrogenesis]

Summary

Addresses the pipex zn-prod Pipex.Schedules.Reconciler: upsert failed … :down: :normal noise (and the related EventSubscriber get_namespace / get_data_converter timeout storms).

Before

• All Temporal RPCs ran inside a single GenServer handle_call, so one long or stuck call blocked every other caller (including cheap metadata reads).
• When gun died mid-RPC (:down: :normal / :noproc), the client kept the dead channel; only a later {:gun_down, …} cleared it. Queued RPCs reused the corpse until the GenServer was restarted.

After

• Caller-side I/O — GenServer only checks out / invalidates the shared gun channel. Network I/O runs in the caller process so concurrent RPCs multiplex over HTTP/2 and namespace/1 / data_converter/1 stay responsive.
• Transparent retry — on transport-death errors, drop the channel, reconnect, and retry the same rpc/4 once. A single blip is usually invisible when Temporal is reachable again.
• Regression tests for :down: :normal, :noproc, reconnect attempt counting, and non-blocking metadata during a slow RPC.
• Version 0.2.2.

Test plan

• mix test (167 passed)
• Dying-channel tests (normal + noproc) + connect attempt counter
• Concurrent namespace/1 while RPC awaits gun
• Publish 0.2.2; pipex ~> 0.2.1 accepts it

[palantir-valiot]

Overall Assessment

This PR correctly fixes sticky dead-channel reuse: when gun dies mid-RPC, the in-flight call surfaces a transport error before {:gun_down, …} runs, so clearing channel on those errors is the right fix and lets the next ensure_channel/1 reconnect. Implementation is small, comments match the production failure mode, and the new tests exercise the real gun cast/await path for both :normal and :noproc. No blocking bugs found.

Findings

No actionable findings.

Notes

• Regression coverage is solid: fixtures install a real GRPC.Channel with a dying/dead conn_pid, assert the exact :down: RPCError shape, assert channel == nil, then assert the next call takes the reconnect path ("Temporal connection error:") rather than another corpse :down:.
• Detection is intentionally message-based (:down: / connection_error) so Temporal business failures (NOT_FOUND, ALREADY_EXISTS, …) keep a live channel. That couples recovery to gun/grpc error formatting; the new tests lock the current grpc 0.11.x gun strings, which is the right safety net if the adapter wording changes.
• Considered flagging that handle_info({:gun_down, _pid, …}) still ignores pid (a late down from an old gun could clear a freshly reconnected channel). That handler is pre-existing and unchanged; in the common mid-RPC path the down is already queued and drained before the next connect, so it was left out of Findings.
• String.contains?(message, "connection_error") is broader than the two starts_with? checks; low false-positive risk for Temporal RPCError messages, and useful if formatting varies slightly.

---

🤖 PR Reviewer · Grok 4.5 (xAI)

[palantir-valiot]

Overall Assessment

This PR moves Temporal RPC I/O into the caller, keeps the GenServer to channel checkout/invalidate, and retries once after transport-level channel death. That design matches the stated pipex failure modes and is covered by focused regression tests. One reconnect race on the existing gun_down handler becomes much more likely with the new invalidate+reconnect path and can drop a healthy channel (and leak the orphaned gun connection).

Findings

#	Severity	File	Issue
1	Medium	lib/temporal_ex/client.ex	gun_down clears state.channel without matching conn_pid, so a late down from a dead channel can wipe a successful reconnect
2	Low	lib/temporal_ex/client.ex	invalidate only nils the stored channel and never disconnects; half-open guns can leak under repeated blips

Notes

• Caller-side I/O + single transparent retry is a solid fit for the :down: :normal / :noproc sticky-channel problem; same_channel?/2 correctly avoids clobbering a newer channel when two callers invalidate.
• Tests cover dying/dead channels, connect-attempt counting on retry, and non-blocking namespace/1 during a slow RPC. Full “retry returns {:ok, _} against a live Temporal” is intentionally left unasserted (commented in the test).
• Considered flagging that start option :call_timeout is still stored on state but unused when defaulting RPC timeout — that was already true before this PR (rpc/4 always defaulted to @default_call_timeout), so it is not introduced here.
• Retrying once means non-idempotent RPCs can run twice after a mid-response transport death; that is consistent with the moduledoc and typical Temporal client at-least-once behavior (workflow starts should still use request_id / server idempotency).

---

🤖 PR Reviewer · Grok 4.5 (xAI)

[Copilot]

Pull request overview

Note

Copilot couldn't run its full agentic review because it didn't start before the timeout. Make sure your repository has a runner available, or add a copilot-code-review.yml file specifying one with the runs-on attribute. See the docs for more details.

Updates TemporalEx.Client to avoid reusing a dead gRPC (gun) channel after mid-RPC transport death by invalidating the channel and retrying the same RPC once, while also shifting RPC network I/O to run in the caller process.

Changes:

• Run RPC I/O in the caller process; GenServer only manages shared channel checkout/invalidation.
• Invalidate channel and retry once when gun dies mid-RPC (:down:* / connection errors).
• Add regression tests, bump version to 0.2.2, and document the changes in CHANGELOG.md.

Reviewed changes

Copilot reviewed 4 out of 4 changed files in this pull request and generated 4 comments.

File	Description
lib/temporal_ex/client.ex	Implements caller-side RPC invoke, channel checkout/invalidate, and one-time retry on transport death; adds docs and connect test seam.
test/temporal_ex/client_test.exs	Adds regression tests and fixtures that simulate gun mid-RPC death and verify reconnect/retry behavior and responsiveness.
mix.exs	Bumps library version to 0.2.2.
CHANGELOG.md	Documents retry/channel invalidation behavior and caller-side I/O change for 0.2.2.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[palantir-valiot]

Overall Assessment

This PR moves Temporal RPC I/O into the caller, keeps the GenServer as a channel checkout/invalidate owner, and retries once on gun transport death (:down:* / connection errors). That design matches the stated pipex failures (stuck dead channels, namespace/1 blocked behind long RPCs) and is covered by focused regression tests. The implementation is generally sound; the main risk is changed :timeout semantics (no single overall deadline, and retries can roughly double wall-clock wait).

Findings

#	Severity	File	Issue
1	Medium	lib/temporal_ex/client.ex	:timeout no longer bounds the whole rpc/4 call; with checkout + invoke (+ one retry) wall-clock can exceed the caller's budget and no longer surfaces as a GenServer.call exit timeout
2	Low	lib/temporal_ex/client.ex	transport_dead_message?/1 treats any message containing "connection_error" as fatal channel death, which is broader than the :down: / :connection_error: prefix cases and can invalidate a still-shared channel

Notes

• Regression coverage is strong for the production modes called out in the description: :down :normal, :noproc, connect-attempt counting, late gun_down not wiping a newer channel, and non-blocking namespace/1 during a slow RPC.
• same_channel?/2 + pid-guarded gun_down handling is the right shape for concurrent checkout/invalidate without dropping a healthy reconnect.
• Passing :timeout into the stub via put_timeout/2 is an improvement over the old path, where only GenServer.call/3 was timed and invoke opts were metadata-only.
• Considered flagging that start_link/1's :call_timeout is still stored and never read (rpc/4 always defaults to @default_call_timeout). That predates this PR, so left out of Findings.
• Residual limitation (not a regression): GRPC.Stub.connect/2 still runs inside handle_call({:checkout, ...}), so a slow connect can still stall metadata reads until connect returns; only the caller-side wait is capped by @checkout_timeout.
• Transparent once-retry is appropriate for transport blips; callers of non-idempotent RPCs should be aware of at-least-once delivery when the server applied the first attempt but the response was lost (Temporal workflow-id semantics usually prevent double-start, but the retry may surface ALREADY_EXISTS instead of the original success).

---

🤖 PR Reviewer · Grok 4.5 (xAI)

[Copilot]

Pull request overview

Copilot reviewed 4 out of 4 changed files in this pull request and generated no new comments.

Comments suppressed due to low confidence (4)

lib/temporal_ex/client.ex:134

• The comment says to use a 1ms floor when the caller timeout is 0, but the implementation sends all non-positive/non-integer values to @checkout_timeout (10s). This means rpc(..., timeout: 0) can block up to 10s on checkout/invalidate, which contradicts the intent and can surprise callers. Consider treating 0 (and possibly negative values) as 1 for the GenServer call timeout, while keeping :infinity capped at @checkout_timeout.

  # GenServer.call rejects 0; use 1ms as the floor for "already timed out".
  defp genserver_timeout(:infinity), do: @checkout_timeout

  defp genserver_timeout(timeout) when is_integer(timeout) and timeout > 0,
    do: min(timeout, @checkout_timeout)

  defp genserver_timeout(_), do: @checkout_timeout

lib/temporal_ex/client.ex:117

• Using gs_timeout (derived from the caller RPC timeout) for {:invalidate, channel} can cause invalidate to time out under contention (e.g., many concurrent checkouts), leaving the dead channel in state. In that case, the retry may immediately re-checkout the same corpse channel and fail again, undermining the 'transparent retry' guarantee. Consider using a bounded but independent timeout for invalidate (e.g., @checkout_timeout), or switching invalidate to an async GenServer.cast (and ensure checkout won’t reuse the channel once a transport-death is detected).

          true ->
            # Drop this channel (no-op if another caller already replaced it),
            # then retry the whole checkout+invoke once.
            _ = GenServer.call(client, {:invalidate, channel}, gs_timeout)

            case retried? do
              true -> result
              false -> do_rpc(client, rpc_name, request, opts, timeout, true)
            end

lib/temporal_ex/client.ex:291

• The fallback String.contains?(message, \"connection_error\") is broad enough to match unrelated RPC failures whose message happens to include that substring, which could trigger an unintended retry of non-transport errors. This is risky because retries can duplicate non-idempotent RPCs. Prefer matching only the known gun-adapter message formats (e.g., strict prefixes), and/or tightening detection using additional fields on %GRPC.RPCError{} (such as status) if available in your GRPC library version.

  defp transport_dead_message?(message) do
    String.starts_with?(message, ":down:") or
      String.starts_with?(message, ":connection_error:") or
      String.contains?(message, "connection_error")
  end

test/temporal_ex/client_test.exs:211

• This test can be flaky on slower/loaded CI runners: the fixed Process.sleep(30) may be insufficient to reliably reach 'await', and the hard < 100_000µs threshold may fail due to scheduler latency/GC pauses even when the GenServer is non-blocking. Consider synchronizing on a deterministic signal that the RPC has entered the await path (e.g., an instrumented connect_fun / message from the dying conn process), and/or relaxing the timing threshold to a less fragile bound while still proving it doesn’t wait ~400ms.

      # Let the RPC check out the channel and enter gun.await.
      Process.sleep(30)

      {elapsed_us, namespace} = :timer.tc(fn -> Client.namespace(pid) end)
      assert namespace == "zn-prod"
      # Must not wait on the 400ms hold — GenServer is free during await.
      assert elapsed_us < 100_000

[Copilot]

Pull request overview

Copilot reviewed 4 out of 4 changed files in this pull request and generated no new comments.

Comments suppressed due to low confidence (3)

lib/temporal_ex/client.ex:134

• The comment states GenServer.call should use a 1ms floor for already-timed-out calls, but the implementation returns @checkout_timeout for timeout <= 0 (via the catch-all clause). This can cause rpc/4 with timeout: 0 (or negative) to block up to 10s during checkout/invalidate, contradicting the intended semantics. Handle is_integer(timeout) and timeout <= 0 explicitly (e.g., return 1) so the GenServer wait aligns with the caller timeout behavior.

  # GenServer.call rejects 0; use 1ms as the floor for "already timed out".
  defp genserver_timeout(:infinity), do: @checkout_timeout

  defp genserver_timeout(timeout) when is_integer(timeout) and timeout > 0,
    do: min(timeout, @checkout_timeout)

  defp genserver_timeout(_), do: @checkout_timeout

lib/temporal_ex/client.ex:291

• Transport-death classification is currently partly based on String.contains?(message, \"connection_error\"), which is broad and can match unrelated error messages containing that substring, causing an unnecessary retry and channel invalidation. Prefer more specific checks (e.g., only known prefixes like \":down:\" / \":connection_error:\", or structured fields if available) to reduce false positives and keep retry behavior predictable.

  defp transport_dead_message?(message) do
    String.starts_with?(message, ":down:") or
      String.starts_with?(message, ":connection_error:") or
      String.contains?(message, "connection_error")
  end

test/temporal_ex/client_test.exs:211

• This test relies on a fixed Process.sleep(30) and a relatively tight elapsed-time threshold, which can be flaky under CI load or scheduler contention (the RPC task may not have entered the awaited state yet, or the namespace call may exceed 100ms sporadically). Consider synchronizing explicitly (e.g., have the dying conn process or the RPC task signal when it has received the cast / entered the hold) and/or relax the timing assertion to a more robust bound to avoid intermittent failures.

      # Let the RPC check out the channel and enter gun.await.
      Process.sleep(30)

      {elapsed_us, namespace} = :timer.tc(fn -> Client.namespace(pid) end)
      assert namespace == "zn-prod"
      # Must not wait on the 400ms hold — GenServer is free during await.
      assert elapsed_us < 100_000