diff --git a/CHANGELOG.md b/CHANGELOG.md index c2015651..1a6fa8d9 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/azure-functions-durable/CHANGELOG.md b/azure-functions-durable/CHANGELOG.md index 21ab81cd..cb753941 100644 --- a/azure-functions-durable/CHANGELOG.md +++ b/azure-functions-durable/CHANGELOG.md @@ -24,6 +24,8 @@ avoiding repeated allocation of unused worker resources. FIXED +- Fixed asynchronous durable-client construction failing after an application +event loop had been closed or cleared. - Prevented Durable HTTP calls from forwarding managed identity tokens, authorization headers, cookies, proxy credentials, or function keys to cross-origin redirect and polling targets. Function keys are now removed from diff --git a/durabletask-azuremanaged/CHANGELOG.md b/durabletask-azuremanaged/CHANGELOG.md index 5ca50774..f5c6c6fe 100644 --- a/durabletask-azuremanaged/CHANGELOG.md +++ b/durabletask-azuremanaged/CHANGELOG.md @@ -21,6 +21,9 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). - Improved async access token refresh concurrency handling to avoid duplicate refresh operations under concurrent access, matching the existing sync behavior. +- Fixed `AsyncDurableTaskSchedulerClient` failing during construction when no + current event loop was set. Its async gRPC channel is now created on first + use and bound to the event loop performing the request. ## v1.8.0 diff --git a/durabletask/client.py b/durabletask/client.py index 4fbeb554..a4851b8d 100644 --- a/durabletask/client.py +++ b/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,7 +1385,7 @@ 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, @@ -1373,7 +1393,7 @@ async def get_entity(self, ) -> 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 diff --git a/tests/durabletask/test_client.py b/tests/durabletask/test_client.py index 5b5d6520..8a9cb5e3 100644 --- a/tests/durabletask/test_client.py +++ b/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: