Defer async client channel creation

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

Copilot-Session: 7bc5f597-1596-460d-bfe6-738676344ef0

Author: andystaples

CHANGELOG.md

@@ -33,6 +33,9 @@ import paths, `__all__`, `dir()`, and star-imports behave exactly as before.
 
 FIXED
 
+- Fixed `AsyncTaskHubGrpcClient` failing during construction when no current
+event loop was set. SDK-owned async gRPC channels are now created on first use,
+binding them to the event loop that performs the RPC.
 - Fixed the worker allocating one `asyncio` task per queued work item before applying the concurrency limit, which made memory use and event-loop scheduling overhead grow with the queue backlog during bursts. In-flight work item tasks are now bounded by the configured `ConcurrencyOptions` limits.
 - Fixed input/output type discovery raising `TypeError: unhashable type` for handlers that are unhashable callables. A callable object registered through `add_named_activity()`, `add_named_orchestrator()`, or `add_named_entity()` is unhashable whenever its class defines `__eq__` without `__hash__` — most commonly a `@dataclass` with a `__call__` method, since dataclasses default to `eq=True`. Annotations on such handlers are now discovered normally instead of failing.
 

azure-functions-durable/CHANGELOG.md

@@ -22,6 +22,11 @@ CHANGED
 - Reused host-driven orchestration and entity workers across invocations,
 avoiding repeated allocation of unused worker resources.
 
+FIXED
+
+- Fixed asynchronous durable-client construction failing after an application
+event loop had been closed or cleared.
+
 ## 2.0.0b1
 
 First preview (beta) release of `azure-functions-durable` 2.x — a ground-up

durabletask/client.py

@@ -988,20 +988,17 @@ def __init__(self, *,
             else None
         )
         self._interceptors = self._compose_interceptors(user_interceptors)
-        if channel is None:
-            channel = shared.get_async_grpc_channel(
-                host_address=self._host_address,
-                secure_channel=secure_channel,
-                interceptors=self._interceptors,
-                channel_options=channel_options,
-            )
         # Caller-owned channels cannot be retroactively wrapped with our
         # interceptor (``grpc.aio`` exposes no public equivalent of
         # ``grpc.intercept_channel``). We document this in :meth:`__init__` and
         # leave the failure-tracking opt-out implicit: callers wanting full
         # resiliency should let us create the channel.
         self._channel = channel
-        self._stub = cast(_AsyncTaskHubSidecarServiceStub, stubs.TaskHubSidecarServiceStub(channel))
+        self._stub = (
+            cast(_AsyncTaskHubSidecarServiceStub, stubs.TaskHubSidecarServiceStub(channel))
+            if channel is not None
+            else None
+        )
         self._logger = shared.get_logger("async_client", log_handler, log_formatter)
         self.default_version = default_version
         self._payload_store = payload_store
@@ -1016,6 +1013,25 @@ def _compose_interceptors(
             composed.extend(user_interceptors)
         return composed
 
+    def _get_stub(self) -> _AsyncTaskHubSidecarServiceStub:
+        """Create the SDK-owned channel on the event loop that first uses it."""
+        if self._closing:
+            raise RuntimeError("The client is closed")
+        if self._stub is None:
+            channel = shared.get_async_grpc_channel(
+                host_address=self._host_address,
+                secure_channel=self._secure_channel,
+                interceptors=self._interceptors,
+                channel_options=self._channel_options,
+            )
+            stub = cast(
+                _AsyncTaskHubSidecarServiceStub,
+                stubs.TaskHubSidecarServiceStub(channel),
+            )
+            self._channel = channel
+            self._stub = stub
+        return self._stub
+
     async def close(self) -> None:
         """Closes the underlying gRPC channel.
 
@@ -1047,7 +1063,9 @@ async def close(self) -> None:
                 await asyncio.gather(*close_tasks, return_exceptions=True)
             for retired_channel in retired_channels:
                 await retired_channel.close()
-            await self._channel.close()
+            channel = self._channel
+            if channel is not None:
+                await channel.close()
 
     async def __aenter__(self) -> "AsyncTaskHubGrpcClient":
         return self
@@ -1093,6 +1111,8 @@ async def _maybe_recreate_channel(self) -> None:
             if now - self._last_recreate_time < self._resiliency_options.min_recreate_interval_seconds:
                 return
             old_channel = self._channel
+            if old_channel is None:
+                return
             self._channel = shared.get_async_grpc_channel(
                 host_address=self._host_address,
                 secure_channel=self._secure_channel,
@@ -1147,13 +1167,13 @@ async def schedule_new_orchestration(self, orchestrator: task.Orchestrator[TInpu
                 await payload_helpers.externalize_payloads_async(
                     req, self._payload_store, instance_id=req.instanceId,
                 )
-            res: pb.CreateInstanceResponse = await self._stub.StartInstance(req)
+            res: pb.CreateInstanceResponse = await self._get_stub().StartInstance(req)
             return res.instanceId
 
     async def get_orchestration_state(self, instance_id: str, *,
                                       fetch_payloads: bool = True) -> OrchestrationState | None:
         req = pb.GetInstanceRequest(instanceId=instance_id, getInputsAndOutputs=fetch_payloads)
-        res: pb.GetInstanceResponse = await self._stub.GetInstance(req)
+        res: pb.GetInstanceResponse = await self._get_stub().GetInstance(req)
         if self._payload_store is not None and res.exists:
             await payload_helpers.deexternalize_payloads_async(res, self._payload_store)
         return new_orchestration_state(req.instanceId, res, self._data_converter)
@@ -1168,7 +1188,7 @@ async def get_orchestration_history(self,
             forWorkItemProcessing=for_work_item_processing,
         )
         self._logger.info(f"Retrieving history for instance '{instance_id}'.")
-        stream = self._stub.StreamInstanceHistory(req)
+        stream = self._get_stub().StreamInstanceHistory(req)
         return await history_helpers.collect_history_events_async(stream, self._payload_store)
 
     async def list_instance_ids(self,
@@ -1192,7 +1212,7 @@ async def list_instance_ids(self,
             f"page_size={page_size}, "
             f"continuation_token={continuation_token}"
         )
-        resp: pb.ListInstanceIdsResponse = await self._stub.ListInstanceIds(req)
+        resp: pb.ListInstanceIdsResponse = await self._get_stub().ListInstanceIds(req)
         next_token = resp.lastInstanceKey.value if resp.HasField("lastInstanceKey") else None
         return Page(items=list(resp.instanceIds), continuation_token=next_token)
 
@@ -1209,7 +1229,7 @@ async def get_all_orchestration_states(self,
 
         while True:
             req = build_query_instances_req(orchestration_query, _continuation_token)
-            resp: pb.QueryInstancesResponse = await self._stub.QueryInstances(req)
+            resp: pb.QueryInstancesResponse = await self._get_stub().QueryInstances(req)
             if self._payload_store is not None:
                 await payload_helpers.deexternalize_payloads_async(resp, self._payload_store)
             states += [parse_orchestration_state(res, self._data_converter) for res in resp.orchestrationState]
@@ -1226,7 +1246,7 @@ async def wait_for_orchestration_start(self, instance_id: str, *,
         req = pb.GetInstanceRequest(instanceId=instance_id, getInputsAndOutputs=fetch_payloads)
         try:
             self._logger.info(f"Waiting up to {timeout}s for instance '{instance_id}' to start.")
-            res: pb.GetInstanceResponse = await self._stub.WaitForInstanceStart(
+            res: pb.GetInstanceResponse = await self._get_stub().WaitForInstanceStart(
                 req,
                 timeout=timeout,
             )
@@ -1245,7 +1265,7 @@ async def wait_for_orchestration_completion(self, instance_id: str, *,
         req = pb.GetInstanceRequest(instanceId=instance_id, getInputsAndOutputs=fetch_payloads)
         try:
             self._logger.info(f"Waiting {timeout}s for instance '{instance_id}' to complete.")
-            res: pb.GetInstanceResponse = await self._stub.WaitForInstanceCompletion(
+            res: pb.GetInstanceResponse = await self._get_stub().WaitForInstanceCompletion(
                 req,
                 timeout=timeout,
             )
@@ -1269,7 +1289,7 @@ async def raise_orchestration_event(self, instance_id: str, event_name: str, *,
                 await payload_helpers.externalize_payloads_async(
                     req, self._payload_store, instance_id=instance_id,
                 )
-            await self._stub.RaiseEvent(req)
+            await self._get_stub().RaiseEvent(req)
 
     async def terminate_orchestration(self, instance_id: str, *,
                                       output: Any | None = None,
@@ -1281,17 +1301,17 @@ async def terminate_orchestration(self, instance_id: str, *,
             await payload_helpers.externalize_payloads_async(
                 req, self._payload_store, instance_id=instance_id,
             )
-        await self._stub.TerminateInstance(req)
+        await self._get_stub().TerminateInstance(req)
 
     async def suspend_orchestration(self, instance_id: str) -> None:
         req = pb.SuspendRequest(instanceId=instance_id)
         self._logger.info(f"Suspending instance '{instance_id}'.")
-        await self._stub.SuspendInstance(req)
+        await self._get_stub().SuspendInstance(req)
 
     async def resume_orchestration(self, instance_id: str) -> None:
         req = pb.ResumeRequest(instanceId=instance_id)
         self._logger.info(f"Resuming instance '{instance_id}'.")
-        await self._stub.ResumeInstance(req)
+        await self._get_stub().ResumeInstance(req)
 
     async def rewind_orchestration(self, instance_id: str, *,
                                    reason: str | None = None) -> None:
@@ -1311,7 +1331,7 @@ async def rewind_orchestration(self, instance_id: str, *,
             reason=helpers.get_string_value(reason))
 
         self._logger.info(f"Rewinding instance '{instance_id}'.")
-        await self._stub.RewindInstance(req)
+        await self._get_stub().RewindInstance(req)
 
     async def restart_orchestration(self, instance_id: str, *,
                                     restart_with_new_instance_id: bool = False) -> str:
@@ -1330,13 +1350,13 @@ async def restart_orchestration(self, instance_id: str, *,
             restartWithNewInstanceId=restart_with_new_instance_id)
 
         self._logger.info(f"Restarting instance '{instance_id}'.")
-        res: pb.RestartInstanceResponse = await self._stub.RestartInstance(req)
+        res: pb.RestartInstanceResponse = await self._get_stub().RestartInstance(req)
         return res.instanceId
 
     async def purge_orchestration(self, instance_id: str, recursive: bool = True) -> PurgeInstancesResult:
         req = pb.PurgeInstancesRequest(instanceId=instance_id, recursive=recursive)
         self._logger.info(f"Purging instance '{instance_id}'.")
-        resp: pb.PurgeInstancesResponse = await self._stub.PurgeInstances(req)
+        resp: pb.PurgeInstancesResponse = await self._get_stub().PurgeInstances(req)
         return PurgeInstancesResult(resp.deletedInstanceCount, resp.isComplete.value)
 
     async def purge_orchestrations_by(self,
@@ -1350,7 +1370,7 @@ async def purge_orchestrations_by(self,
                           f"runtime_status={[str(status) for status in runtime_status] if runtime_status else None}, "
                           f"recursive={recursive}")
         req = build_purge_by_filter_req(created_time_from, created_time_to, runtime_status, recursive)
-        resp: pb.PurgeInstancesResponse = await self._stub.PurgeInstances(req)
+        resp: pb.PurgeInstancesResponse = await self._get_stub().PurgeInstances(req)
         return PurgeInstancesResult(resp.deletedInstanceCount, resp.isComplete.value)
 
     async def signal_entity(self,
@@ -1365,15 +1385,15 @@ async def signal_entity(self,
             await payload_helpers.externalize_payloads_async(
                 req, self._payload_store, instance_id=str(entity_instance_id),
             )
-        await self._stub.SignalEntity(req)
+        await self._get_stub().SignalEntity(req)
 
     async def get_entity(self,
                          entity_instance_id: EntityInstanceId,
                          include_state: bool = True
                          ) -> EntityMetadata | None:
         req = pb.GetEntityRequest(instanceId=str(entity_instance_id), includeState=include_state)
         self._logger.info(f"Getting entity '{entity_instance_id}'.")
-        res: pb.GetEntityResponse = await self._stub.GetEntity(req)
+        res: pb.GetEntityResponse = await self._get_stub().GetEntity(req)
         if not res.exists:
             return None
         if self._payload_store is not None:
@@ -1392,7 +1412,7 @@ async def get_all_entities(self,
 
         while True:
             query_request = build_query_entities_req(entity_query, _continuation_token)
-            resp: pb.QueryEntitiesResponse = await self._stub.QueryEntities(query_request)
+            resp: pb.QueryEntitiesResponse = await self._get_stub().QueryEntities(query_request)
             if self._payload_store is not None:
                 await payload_helpers.deexternalize_payloads_async(resp, self._payload_store)
             entities += [EntityMetadata.from_entity_metadata(entity, query_request.query.includeState, self._data_converter) for entity in resp.entities]
@@ -1418,7 +1438,7 @@ async def clean_entity_storage(self,
                 releaseOrphanedLocks=release_orphaned_locks,
                 continuationToken=_continuation_token
             )
-            resp: pb.CleanEntityStorageResponse = await self._stub.CleanEntityStorage(req)
+            resp: pb.CleanEntityStorageResponse = await self._get_stub().CleanEntityStorage(req)
             empty_entities_removed += resp.emptyEntitiesRemoved
             orphaned_locks_released += resp.orphanedLocksReleased
 

tests/durabletask/test_client.py

@@ -223,6 +223,8 @@ def install_resilient_test_stubs(client):
     wrapper_cls = _ResilientAsyncTestStub if is_async else _ResilientSyncTestStub
 
     def wrap_if_needed():
+        if is_async:
+            client._get_stub()
         if not isinstance(client._stub, wrapper_cls):
             client._stub = wrapper_cls(client._stub, client._resiliency_interceptor)
 
@@ -490,20 +492,24 @@ def test_async_grpc_channel_protocol_stripping():
 # ==== Async client construction tests ====
 
 
-def test_async_client_creates_with_defaults():
+@pytest.mark.asyncio
+async def test_async_client_creates_with_defaults():
     with patch('grpc.aio.insecure_channel') as mock_channel:
         mock_channel.return_value = MagicMock()
         client = AsyncTaskHubGrpcClient()
+        client._get_stub()
         mock_channel.assert_called_once_with(
             get_default_host_address(),
             interceptors=[client._resiliency_interceptor])
 
 
-def test_async_client_creates_with_metadata():
+@pytest.mark.asyncio
+async def test_async_client_creates_with_metadata():
     with patch('grpc.aio.insecure_channel') as mock_channel:
         mock_channel.return_value = MagicMock()
         client = AsyncTaskHubGrpcClient(
             host_address=HOST_ADDRESS, metadata=METADATA)
+        client._get_stub()
         mock_channel.assert_called_once()
         args, kwargs = mock_channel.call_args
         assert args[0] == HOST_ADDRESS
@@ -515,6 +521,33 @@ def test_async_client_creates_with_metadata():
         assert isinstance(interceptors[1], DefaultAsyncClientInterceptorImpl)
 
 
+def test_async_client_defers_channel_until_async_use():
+    asyncio.run(asyncio.sleep(0))
+    channel = MagicMock()
+    channel.close = AsyncMock()
+    stub = MagicMock()
+    stub.GetInstance = AsyncMock(return_value=MagicMock(exists=False))
+
+    with patch(
+            'durabletask.client.shared.get_async_grpc_channel',
+            return_value=channel,
+    ) as mock_get_channel, patch(
+            'durabletask.client.stubs.TaskHubSidecarServiceStub',
+            return_value=stub,
+    ):
+        client = AsyncTaskHubGrpcClient()
+        mock_get_channel.assert_not_called()
+
+        async def use_client() -> None:
+            assert await client.get_orchestration_state("abc") is None
+            await client.close()
+
+        asyncio.run(use_client())
+
+    mock_get_channel.assert_called_once()
+    channel.close.assert_awaited_once()
+
+
 def test_client_uses_provided_channel_directly():
     provided_channel = MagicMock()
     with patch('durabletask.internal.shared.get_grpc_channel') as mock_get_channel: