Fire-and-forget channel recreate; hoist client state; tighten async interceptor exception handling

Addresses three follow-up review comments on the resiliency interceptor refactor:

[1/3] Channel recreate now runs fire-and-forget (daemon thread for sync,
asyncio.create_task for async). The original RPC error propagates to the
caller without being delayed by DNS, TLS handshake, or contention on
_recreate_lock. A client-side single-flight guard avoids spawning duplicate
work when many failures land in a burst; the existing cooldown still
prevents thrash. close() waits for any in-flight recreate to finish so the
teardown path stays deterministic. A _recreate_done_event (test seam) lets
tests synchronise on completion without polling.

[2/3] Hoisted _closing, _recreate_lock, _last_recreate_time,
_retired_channels / _retired_channel_close_tasks above ClientResiliencyInterceptor
construction in both __init__ methods so the bound recreate callback is safe
to invoke at any time during construction.

[3/3] AsyncClientResiliencyInterceptor now uses 'except Exception' (so
asyncio.CancelledError, KeyboardInterrupt and SystemExit propagate
unchanged) and mirrors the sync interceptor's policy by resetting the
failure counter on non-AioRpcError exceptions. _record_outcome is now
synchronous on both interceptors because the on_recreate callback no
longer awaits the recreate.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: Bernd Verst

durabletask/client.py

@@ -198,12 +198,28 @@ def __init__(self, *,
             if resiliency_options is not None
             else GrpcClientResiliencyOptions()
         )
+        # Resiliency state must be initialised BEFORE the interceptor is
+        # constructed because the interceptor receives a bound reference to
+        # ``self._schedule_recreate``; any failure handled during construction
+        # of the underlying channel could otherwise observe a half-built
+        # client.
+        self._closing = False
+        self._last_recreate_time = 0.0
+        self._recreate_lock = threading.Lock()
+        self._retired_channels: dict[grpc.Channel, threading.Timer] = {}
+        self._recreate_thread_lock = threading.Lock()
+        self._recreate_thread: Optional[threading.Thread] = None
+        # Test seam: set after each fire-and-forget recreate attempt finishes
+        # (whether it actually recreated the channel or short-circuited on
+        # close / cooldown). Lets tests synchronise without polling and lets
+        # ``close()`` wait deterministically for an in-flight recreate.
+        self._recreate_done_event = threading.Event()
         self._client_failure_tracker = FailureTracker(
             self._resiliency_options.channel_recreate_failure_threshold
         )
         self._resiliency_interceptor = ClientResiliencyInterceptor(
             self._client_failure_tracker,
-            self._maybe_recreate_channel,
+            self._schedule_recreate,
         )
         resolved_interceptors = (
             prepare_sync_interceptors(metadata, interceptors) if channel is None else interceptors
@@ -230,10 +246,6 @@ def __init__(self, *,
         # can prepend the interceptor themselves via grpc.intercept_channel.
         self._channel = channel
         self._stub = stubs.TaskHubSidecarServiceStub(channel)
-        self._closing = False
-        self._last_recreate_time = 0.0
-        self._recreate_lock = threading.Lock()
-        self._retired_channels: dict[grpc.Channel, threading.Timer] = {}
         self._logger = shared.get_logger("client", log_handler, log_formatter)
         self.default_version = default_version
         self._payload_store = payload_store
@@ -252,6 +264,48 @@ def _compose_interceptors(
             composed.extend(user_interceptors)
         return composed
 
+    def _schedule_recreate(self) -> None:
+        """Spawn a daemon thread that recreates the channel fire-and-forget.
+
+        Called from the resiliency interceptor on the caller's thread when a
+        unary RPC fails with a transport error. The interceptor returns to its
+        caller as soon as this method returns, so the failing RPC's original
+        error propagates without being delayed by DNS, TLS handshake, or
+        contention on ``_recreate_lock``.
+
+        Single-flight under ``_recreate_thread_lock``: if a recreate thread is
+        still alive, the new trigger is dropped. The in-flight recreate will
+        pick up the latest channel state on completion; the cooldown inside
+        ``_maybe_recreate_channel`` further prevents thrash. ``thread.start()``
+        is called under the lock so a follow-up caller's ``is_alive()`` check
+        observes the running state rather than racing the start.
+        """
+        try:
+            if self._closing:
+                return
+            with self._recreate_thread_lock:
+                existing = self._recreate_thread
+                if existing is not None and existing.is_alive():
+                    return
+                self._recreate_done_event.clear()
+                thread = threading.Thread(
+                    target=self._run_recreate,
+                    name="durabletask-client-recreate",
+                    daemon=True,
+                )
+                self._recreate_thread = thread
+                thread.start()
+        except Exception:
+            self._logger.exception("Failed to schedule channel recreate")
+
+    def _run_recreate(self) -> None:
+        try:
+            self._maybe_recreate_channel()
+        except Exception:
+            self._logger.exception("Channel recreate failed")
+        finally:
+            self._recreate_done_event.set()
+
     def _maybe_recreate_channel(self) -> None:
         if not self._owns_channel or self._closing:
             return
@@ -296,8 +350,14 @@ def close(self) -> None:
         it.
         """
         if self._owns_channel:
+            # Signal early so any in-flight recreate thread bails out of
+            # ``_maybe_recreate_channel`` before we tear the channel down.
+            self._closing = True
+            with self._recreate_thread_lock:
+                recreate_thread = self._recreate_thread
+            if recreate_thread is not None and recreate_thread.is_alive():
+                recreate_thread.join(timeout=5.0)
             with self._recreate_lock:
-                self._closing = True
                 retired_channels = list(self._retired_channels.items())
                 self._retired_channels.clear()
                 current_channel = self._channel
@@ -628,12 +688,28 @@ def __init__(self, *,
             if resiliency_options is not None
             else GrpcClientResiliencyOptions()
         )
+        # Resiliency state must be initialised BEFORE the interceptor is
+        # constructed because the interceptor receives a bound reference to
+        # ``self._schedule_recreate``; any failure handled during construction
+        # of the underlying channel could otherwise observe a half-built
+        # client.
+        self._closing = False
+        self._recreate_lock = asyncio.Lock()
+        self._last_recreate_time = 0.0
+        self._retired_channels: list[grpc.aio.Channel] = []
+        self._retired_channel_close_tasks: set[asyncio.Task[None]] = set()
+        self._recreate_task: Optional[asyncio.Task[None]] = None
+        # Test seam: set after each fire-and-forget recreate attempt finishes
+        # (whether it actually recreated the channel or short-circuited on
+        # close / cooldown). Lets tests synchronise without polling and lets
+        # ``close()`` await an in-flight recreate deterministically.
+        self._recreate_done_event = asyncio.Event()
         self._client_failure_tracker = FailureTracker(
             self._resiliency_options.channel_recreate_failure_threshold
         )
         self._resiliency_interceptor = AsyncClientResiliencyInterceptor(
             self._client_failure_tracker,
-            self._maybe_recreate_channel,
+            self._schedule_recreate,
         )
         resolved_interceptors = (
             prepare_async_interceptors(metadata, interceptors) if channel is None else interceptors
@@ -660,11 +736,6 @@ def __init__(self, *,
         # resiliency should let us create the channel.
         self._channel = channel
         self._stub = stubs.TaskHubSidecarServiceStub(channel)
-        self._closing = False
-        self._recreate_lock = asyncio.Lock()
-        self._last_recreate_time = 0.0
-        self._retired_channels: list[grpc.aio.Channel] = []
-        self._retired_channel_close_tasks: set[asyncio.Task[None]] = set()
         self._logger = shared.get_logger("async_client", log_handler, log_formatter)
         self.default_version = default_version
         self._payload_store = payload_store
@@ -688,7 +759,17 @@ async def close(self) -> None:
         it.
         """
         if self._owns_channel:
+            # Signal early so any in-flight recreate task bails out of
+            # ``_maybe_recreate_channel`` before we tear the channel down.
             self._closing = True
+            recreate_task = self._recreate_task
+            if recreate_task is not None and not recreate_task.done():
+                try:
+                    await recreate_task
+                except Exception:
+                    # Already logged by ``_run_recreate``; suppressing here
+                    # ensures close() always tears down cleanly.
+                    pass
             async with self._recreate_lock:
                 retired_channels = list(self._retired_channels)
                 self._retired_channels.clear()
@@ -708,6 +789,34 @@ async def __aenter__(self):
     async def __aexit__(self, exc_type, exc_val, exc_tb):
         await self.close()
 
+    def _schedule_recreate(self) -> None:
+        """Schedule a fire-and-forget channel recreate on the event loop.
+
+        Called from the resiliency interceptor when a unary RPC fails with a
+        transport error. Single-flight: if ``_recreate_task`` is still
+        pending, the trigger is dropped — the in-flight recreate will pick up
+        the latest channel state on completion. asyncio is single-threaded
+        so ``done()`` is race-free; no extra lock is required.
+        """
+        try:
+            if self._closing:
+                return
+            existing = self._recreate_task
+            if existing is not None and not existing.done():
+                return
+            self._recreate_done_event.clear()
+            self._recreate_task = asyncio.create_task(self._run_recreate())
+        except Exception:
+            self._logger.exception("Failed to schedule channel recreate")
+
+    async def _run_recreate(self) -> None:
+        try:
+            await self._maybe_recreate_channel()
+        except Exception:
+            self._logger.exception("Channel recreate failed")
+        finally:
+            self._recreate_done_event.set()
+
     async def _maybe_recreate_channel(self) -> None:
         if not self._owns_channel or self._closing:
             return

durabletask/internal/grpc_resiliency.py

@@ -1,11 +1,10 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
 
-import inspect
 import random
 import threading
 from dataclasses import dataclass, field
-from typing import Awaitable, Callable, Optional, Union
+from typing import Callable, Optional
 
 import grpc
 import grpc.aio
@@ -92,6 +91,13 @@ class ClientResiliencyInterceptor(grpc.UnaryUnaryClientInterceptor):
     need to wrap every stub call: any unary RPC sent through the intercepted
     channel automatically participates in failure tracking and channel
     recreation, including future RPCs that are added to the service stub.
+
+    The ``on_recreate`` callback is invoked **fire-and-forget** by the owning
+    client (it schedules the actual recreate on a daemon thread / asyncio
+    task), so this interceptor never blocks the calling thread or event loop
+    on DNS, TLS handshake, or any other channel construction work. The
+    triggering call's original error propagates to the caller without added
+    latency; subsequent calls benefit from the recreated channel.
     """
 
     def __init__(
@@ -121,37 +127,48 @@ def _record_outcome(self, method: str, error: Optional[BaseException]) -> None:
 
 
 class AsyncClientResiliencyInterceptor(grpc.aio.UnaryUnaryClientInterceptor):
-    """Async counterpart of :class:`ClientResiliencyInterceptor`."""
+    """Async counterpart of :class:`ClientResiliencyInterceptor`.
+
+    The ``on_recreate`` callback is a *synchronous* function (it schedules an
+    ``asyncio.Task`` for the actual recreate); this keeps the original RPC
+    error free of any extra latency from DNS / TLS handshake / lock waits and
+    guarantees the caller sees its original ``AioRpcError`` rather than an
+    exception that happened during recreate scheduling.
+
+    Non-``AioRpcError`` exceptions reset the failure counter (matching the
+    sync interceptor's policy, where ``.exception()`` returning a non-RpcError
+    falls through to ``record_success``). ``CancelledError`` and other
+    non-``Exception`` ``BaseException`` subclasses propagate without bookkeeping,
+    which is the correct asyncio convention.
+    """
 
     def __init__(
             self,
             failure_tracker: FailureTracker,
-            on_recreate: Callable[[], Union[None, Awaitable[object]]],
+            on_recreate: Callable[[], None],
     ):
         self._failure_tracker = failure_tracker
         self._on_recreate = on_recreate
 
     async def intercept_unary_unary(self, continuation, client_call_details, request):
         try:
             response = await continuation(client_call_details, request)
-        except grpc.aio.AioRpcError as rpc_error:
-            await self._record_outcome(client_call_details.method, rpc_error)
+        except Exception as exc:
+            if isinstance(exc, grpc.aio.AioRpcError):
+                self._record_outcome(client_call_details.method, exc)
+            else:
+                self._failure_tracker.record_success()
             raise
-        await self._record_outcome(client_call_details.method, None)
+        self._record_outcome(client_call_details.method, None)
         return response
 
-    async def _record_outcome(self, method: str, error: Optional[BaseException]) -> None:
+    def _record_outcome(self, method: str, error: Optional[BaseException]) -> None:
         if error is None:
             self._failure_tracker.record_success()
             return
         status_code = getattr(error, "code", lambda: None)()
         if status_code is not None and is_client_transport_failure(method, status_code):
             if self._failure_tracker.record_failure():
-                result = self._on_recreate()
-                if inspect.isawaitable(result):
-                    # ``_ =`` signals that we await purely for the side effect
-                    # of running the recreate callback to completion; the
-                    # awaitable's return value is intentionally discarded.
-                    _ = await result
+                self._on_recreate()
         else:
             self._failure_tracker.record_success()

tests/durabletask/test_client.py

@@ -209,6 +209,11 @@ def install_resilient_test_stubs(client):
     interception pipeline. This helper wraps the current stub in a thin
     interceptor-aware shim, and re-wraps after each ``_maybe_recreate_channel``
     call so newly created stubs continue to participate in failure tracking.
+
+    The interceptor's ``_on_recreate`` callback is intentionally left alone:
+    it is the client's ``_schedule_recreate`` (fire-and-forget), and tests
+    that need to observe the recreate's completion can wait on
+    ``client._recreate_done_event``.
     """
     is_async = inspect.iscoroutinefunction(client._maybe_recreate_channel)
     wrapper_cls = _ResilientAsyncTestStub if is_async else _ResilientSyncTestStub
@@ -231,7 +236,6 @@ def wrapped_recreate():
             wrap_if_needed()
 
     client._maybe_recreate_channel = wrapped_recreate
-    client._resiliency_interceptor._on_recreate = wrapped_recreate
 
 
 class FakePayloadStore(PayloadStore):
@@ -557,6 +561,9 @@ def test_sync_client_recreates_sdk_owned_channel_with_original_transport_inputs(
         install_resilient_test_stubs(client)
         with pytest.raises(FakeRpcError):
             client.get_orchestration_state("abc")
+        # Fire-and-forget recreate runs on a daemon thread; wait for it to
+        # complete before issuing the call that should see the new channel.
+        assert client._recreate_done_event.wait(timeout=5.0)
         client.get_orchestration_state("abc")
 
     expected_channel_call = call(
@@ -641,8 +648,13 @@ def test_sync_client_close_closes_all_retired_sdk_channels_immediately():
         install_resilient_test_stubs(client)
         with pytest.raises(FakeRpcError):
             client.get_orchestration_state("abc")
+        # Wait for the first fire-and-forget recreate to complete so the
+        # single-flight guard in _schedule_recreate does not drop the second
+        # trigger.
+        assert client._recreate_done_event.wait(timeout=5.0)
         with pytest.raises(FakeRpcError):
             client.get_orchestration_state("abc")
+        assert client._recreate_done_event.wait(timeout=5.0)
 
         client.close()
 
@@ -746,16 +758,23 @@ def test_sync_client_recreate_cooldown_prevents_immediate_repeated_recreation():
         install_resilient_test_stubs(client)
         with pytest.raises(FakeRpcError):
             client.get_orchestration_state("abc")
+        # Wait for the fire-and-forget recreate to complete before asserting
+        # the channel was swapped.
+        assert client._recreate_done_event.wait(timeout=5.0)
         assert client._channel is second_channel
         assert mock_get_channel.call_count == 2
 
         with pytest.raises(FakeRpcError):
             client.get_orchestration_state("abc")
+        # Cooldown should fire-and-forget but exit without recreating; wait
+        # for the no-op recreate to complete so the assertion is deterministic.
+        assert client._recreate_done_event.wait(timeout=5.0)
         assert client._channel is second_channel
         assert mock_get_channel.call_count == 2
 
         with pytest.raises(FakeRpcError):
             client.get_orchestration_state("abc")
+        assert client._recreate_done_event.wait(timeout=5.0)
         assert client._channel is third_channel
 
     expected_channel_call = call(
@@ -861,6 +880,10 @@ async def test_async_client_recreates_sdk_owned_channel_with_original_transport_
         try:
             with pytest.raises(grpc.aio.AioRpcError):
                 await client.get_orchestration_state("abc")
+            # Fire-and-forget recreate runs as an asyncio task; await its
+            # completion before issuing the call that should see the new
+            # channel.
+            await asyncio.wait_for(client._recreate_done_event.wait(), timeout=5.0)
             await client.get_orchestration_state("abc")
         finally:
             await client.close()