Create gRPC channel in the caller's event loop

Signed-off-by: Sergio Herrera <627709+seherv@users.noreply.github.com>

Author: seherv

ext/dapr-ext-workflow/dapr/ext/workflow/_durabletask/aio/client.py

@@ -70,18 +70,32 @@ def __init__(
         else:
             interceptors = None
 
-        channel = get_grpc_aio_channel(
-            host_address=host_address,
-            secure_channel=secure_channel,
-            interceptors=interceptors,
-            options=channel_options,
-        )
-        self._channel = channel
-        self._stub = stubs.TaskHubSidecarServiceStub(channel)
+        self._host_address = host_address
+        self._secure_channel = secure_channel
+        self._interceptors = interceptors
+        self._channel_options = channel_options
+        self._channel: grpc.aio.Channel | None = None
+        self._stub: stubs.TaskHubSidecarServiceStub | None = None
         self._logger = shared.get_logger('client', log_handler, log_formatter)
 
+    def _get_stub(self) -> stubs.TaskHubSidecarServiceStub:
+        """Lazily create the channel and stub on first use.
+
+        Async grpc binds a channel to the loop active at creation, deferring it avoids binding to the wrong loop.
+        """
+        if self._stub is None:
+            self._channel = get_grpc_aio_channel(
+                host_address=self._host_address,
+                secure_channel=self._secure_channel,
+                interceptors=self._interceptors,
+                options=self._channel_options,
+            )
+            self._stub = stubs.TaskHubSidecarServiceStub(self._channel)
+        return self._stub
+
     async def aclose(self):
-        await self._channel.close()
+        if self._channel is not None:
+            await self._channel.close()
 
     async def __aenter__(self):
         return self
@@ -112,14 +126,14 @@ async def schedule_new_orchestration(
         )
 
         self._logger.info(f"Starting new '{name}' instance with 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
     ) -> Optional[WorkflowState]:
         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)
         return new_orchestration_state(req.instanceId, res)
 
     async def wait_for_orchestration_start(
@@ -131,7 +145,7 @@ async def wait_for_orchestration_start(
         )
 
         async def _call(grpc_timeout):
-            res: pb.GetInstanceResponse = await self._stub.WaitForInstanceStart(
+            res: pb.GetInstanceResponse = await self._get_stub().WaitForInstanceStart(
                 req, timeout=grpc_timeout
             )
             return new_orchestration_state(req.instanceId, res)
@@ -150,7 +164,7 @@ async def wait_for_orchestration_completion(
         )
 
         async def _call(grpc_timeout):
-            res: pb.GetInstanceResponse = await self._stub.WaitForInstanceCompletion(
+            res: pb.GetInstanceResponse = await self._get_stub().WaitForInstanceCompletion(
                 req, timeout=grpc_timeout
             )
             state = new_orchestration_state(req.instanceId, res)
@@ -261,7 +275,7 @@ async def raise_orchestration_event(
         )
 
         self._logger.info(f"Raising event '{event_name}' for instance '{instance_id}'.")
-        await self._stub.RaiseEvent(req)
+        await self._get_stub().RaiseEvent(req)
 
     async def terminate_orchestration(
         self, instance_id: str, *, output: Optional[Any] = None, recursive: bool = True
@@ -273,19 +287,19 @@ async def terminate_orchestration(
         )
 
         self._logger.info(f"Terminating instance '{instance_id}'.")
-        await self._stub.TerminateInstance(req)
+        await self._get_stub().TerminateInstance(req)
 
     async def suspend_orchestration(self, instance_id: str):
         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):
         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 purge_orchestration(self, instance_id: str, recursive: bool = True):
         req = pb.PurgeInstancesRequest(instanceId=instance_id, recursive=recursive)
         self._logger.info(f"Purging instance '{instance_id}'.")
-        await self._stub.PurgeInstances(req)
+        await self._get_stub().PurgeInstances(req)

ext/dapr-ext-workflow/tests/durabletask/test_client_async.py

@@ -165,7 +165,11 @@ def test_async_client_construct_with_metadata():
     with patch(
         'dapr.ext.workflow._durabletask.aio.internal.shared.grpc_aio.insecure_channel'
     ) as mock_channel:
-        AsyncTaskHubGrpcClient(host_address=HOST_ADDRESS, metadata=METADATA)
+        client = AsyncTaskHubGrpcClient(host_address=HOST_ADDRESS, metadata=METADATA)
+        assert mock_channel.call_count == 0  # channel is built lazily, not at construction
+
+        client._get_stub()
+
         # Ensure channel created with an interceptor that has the expected metadata
         args, kwargs = mock_channel.call_args
         assert args[0] == HOST_ADDRESS
@@ -175,6 +179,18 @@ def test_async_client_construct_with_metadata():
         assert interceptors[0]._metadata == METADATA
 
 
+def test_async_client_channel_is_lazy():
+    with patch(
+        'dapr.ext.workflow._durabletask.aio.internal.shared.grpc_aio.insecure_channel'
+    ) as mock_channel:
+        client = AsyncTaskHubGrpcClient(host_address=HOST_ADDRESS)
+        assert mock_channel.call_count == 0  # not built at construction
+
+        client._get_stub()
+        client._get_stub()
+        assert mock_channel.call_count == 1  # built once on first use, then cached
+
+
 def test_aio_channel_passes_base_options_and_max_lengths():
     base_options = [
         ('grpc.max_send_message_length', 4321),