feedback

Signed-off-by: Filinto Duran <[email protected]>

Author: filintod

durabletask/worker.py

@@ -253,8 +253,6 @@ def __init__(
             self._interceptors = None
 
         self._async_worker_manager = _AsyncWorkerManager(self._concurrency_options)
-        # Readiness flag set once the worker has an active stream to the sidecar
-        self._ready = Event()
 
     @property
     def concurrency_options(self) -> ConcurrencyOptions:
@@ -357,8 +355,6 @@ def invalidate_connection():
                     pass
             current_channel = None
             current_stub = None
-            # No longer ready if connection is gone
-            self._ready.clear()
 
         def should_invalidate_connection(rpc_error):
             error_code = rpc_error.code()  # type: ignore
@@ -398,8 +394,6 @@ def should_invalidate_connection(rpc_error):
                 self._logger.info(
                     f"Successfully connected to {self._host_address}. Waiting for work items..."
                 )
-                # Signal readiness once stream is established
-                self._ready.set()
 
                 # Use a thread to read from the blocking gRPC stream and forward to asyncio
                 import queue
@@ -508,14 +502,34 @@ def stop(self):
         self._async_worker_manager.shutdown()
         self._logger.info("Worker shutdown completed")
         self._is_running = False
-        self._ready.clear()
 
-    def wait_for_ready(self, timeout: Optional[float] = None) -> bool:
-        """Block until the worker has an active connection to the sidecar.
-
-        Returns True if the worker became ready within the timeout; otherwise False.
-        """
-        return self._ready.wait(timeout)
+    def _handle_grpc_execution_error(self, rpc_error: grpc.RpcError, request_type: str):
+        """Handle a gRPC execution error during shutdown or benign condition."""
+        # During shutdown or if the instance was terminated, the channel may be close
+        # or the instance may no longer be recognized by the sidecar. Treat these as benign
+        # to reduce noisy logging when shutting down.
+        details = str(rpc_error).lower()
+        benign_errors = {
+            grpc.StatusCode.CANCELLED,
+            grpc.StatusCode.UNAVAILABLE,
+            grpc.StatusCode.UNKNOWN,
+        }
+        if (
+            self._shutdown.is_set()
+            and rpc_error.code() in benign_errors
+            or (
+                "unknown instance id/task id combo" in details
+                or "channel closed" in details
+                or "locally cancelled by application" in details
+            )
+        ):
+            self._logger.debug(
+                f"Ignoring gRPC {request_type} execution error during shutdown/benign condition: {rpc_error}"
+            )
+        else:
+            self._logger.exception(
+                f"Failed to execute gRPC {request_type} execution error: {rpc_error}"
+            )
 
     def _execute_orchestrator(
         self,
@@ -551,24 +565,7 @@ def _execute_orchestrator(
         try:
             stub.CompleteOrchestratorTask(res)
         except grpc.RpcError as rpc_error:  # type: ignore
-            # During shutdown or if the instance was terminated, the channel may be closed
-            # or the instance may no longer be recognized by the sidecar. Treat these as benign
-            # to reduce noisy logging when shutting down.
-            code = rpc_error.code()  # type: ignore
-            details = str(rpc_error)
-            benign = (
-                code in {grpc.StatusCode.CANCELLED, grpc.StatusCode.UNAVAILABLE}
-                or "unknown instance ID/task ID combo" in details
-                or "Channel closed" in details
-            )
-            if self._shutdown.is_set() or benign:
-                self._logger.debug(
-                    f"Ignoring orchestrator completion delivery error during shutdown/benign condition: {rpc_error}"
-                )
-            else:
-                self._logger.exception(
-                    f"Failed to deliver orchestrator response for '{req.instanceId}' to sidecar: {rpc_error}"
-                )
+            self._handle_grpc_execution_error(rpc_error, "orchestrator")
         except Exception as ex:
             self._logger.exception(
                 f"Failed to deliver orchestrator response for '{req.instanceId}' to sidecar: {ex}"
@@ -601,26 +598,7 @@ def _execute_activity(
         try:
             stub.CompleteActivityTask(res)
         except grpc.RpcError as rpc_error:  # type: ignore
-            # Treat common shutdown/termination races as benign to avoid noisy logs
-            code = rpc_error.code()  # type: ignore
-            details = str(rpc_error)
-            benign = code in {
-                grpc.StatusCode.CANCELLED,
-                grpc.StatusCode.UNAVAILABLE,
-                grpc.StatusCode.UNKNOWN,
-            } and (
-                "unknown instance ID/task ID combo" in details
-                or "Channel closed" in details
-                or "Locally cancelled by application" in details
-            )
-            if self._shutdown.is_set() or benign:
-                self._logger.debug(
-                    f"Ignoring activity completion delivery error during shutdown/benign condition: {rpc_error}"
-                )
-            else:
-                self._logger.exception(
-                    f"Failed to deliver activity response for '{req.name}#{req.taskId}' of orchestration ID '{instance_id}' to sidecar: {rpc_error}"
-                )
+            self._handle_grpc_execution_error(rpc_error, "activity")
         except Exception as ex:
             self._logger.exception(
                 f"Failed to deliver activity response for '{req.name}#{req.taskId}' of orchestration ID '{instance_id}' to sidecar: {ex}"

tests/durabletask/test_client.py

@@ -1,5 +1,6 @@
 from unittest.mock import MagicMock, patch
 
+from durabletask import client
 from durabletask.internal.grpc_interceptor import DefaultClientInterceptorImpl
 from durabletask.internal.shared import get_default_host_address, get_grpc_channel
 
@@ -149,8 +150,18 @@ def test_taskhub_client_close_handles_exceptions():
         mock_channel.close.side_effect = Exception("close failed")
         mock_get_channel.return_value = mock_channel
 
-        from durabletask import client
-
         task_hub_client = client.TaskHubGrpcClient()
         # Should not raise exception
         task_hub_client.close()
+
+
+def test_taskhub_client_close_closes_channel_handles_exceptions():
+    """Test that close() closes the channel and handles exceptions gracefully."""
+    with patch("durabletask.internal.shared.get_grpc_channel") as mock_get_channel:
+        mock_channel = MagicMock()
+        mock_channel.close.side_effect = Exception("close failed")
+        mock_get_channel.return_value = mock_channel
+
+        task_hub_client = client.TaskHubGrpcClient()
+        task_hub_client.close()
+        mock_channel.close.assert_called_once()

tests/durabletask/test_orchestration_e2e.py

@@ -58,7 +58,6 @@ def empty_orchestrator(ctx: task.OrchestrationContext, _):
     with worker.TaskHubGrpcWorker(channel_options=channel_options) as w:
         w.add_orchestrator(empty_orchestrator)
         w.start()
-        w.wait_for_ready(timeout=10)
 
         # set a custom max send length option
         c = client.TaskHubGrpcClient(channel_options=channel_options)
@@ -98,7 +97,6 @@ def sequence(ctx: task.OrchestrationContext, start_val: int):
         w.add_orchestrator(sequence)
         w.add_activity(plus_one)
         w.start()
-        w.wait_for_ready(timeout=10)
 
         with client.TaskHubGrpcClient() as task_hub_client:
             id = task_hub_client.schedule_new_orchestration(sequence, input=1)
@@ -143,7 +141,6 @@ def orchestrator(ctx: task.OrchestrationContext, input: int):
         w.add_activity(throw)
         w.add_activity(increment_counter)
         w.start()
-        w.wait_for_ready(timeout=10)
 
         with client.TaskHubGrpcClient() as task_hub_client:
             id = task_hub_client.schedule_new_orchestration(orchestrator, input=1)
@@ -186,7 +183,6 @@ def parent_orchestrator(ctx: task.OrchestrationContext, count: int):
         w.add_orchestrator(orchestrator_child)
         w.add_orchestrator(parent_orchestrator)
         w.start()
-        w.wait_for_ready(timeout=10)
 
         with client.TaskHubGrpcClient() as task_hub_client:
             id = task_hub_client.schedule_new_orchestration(parent_orchestrator, input=10)
@@ -209,7 +205,6 @@ def orchestrator(ctx: task.OrchestrationContext, _):
     with worker.TaskHubGrpcWorker(stop_timeout=2.0) as w:
         w.add_orchestrator(orchestrator)
         w.start()
-        w.wait_for_ready(timeout=10)
 
         # Start the orchestration and immediately raise events to it.
         task_hub_client = client.TaskHubGrpcClient()
@@ -239,7 +234,6 @@ def orchestrator(ctx: task.OrchestrationContext, _):
     with worker.TaskHubGrpcWorker(stop_timeout=2.0) as w:
         w.add_orchestrator(orchestrator)
         w.start()
-        w.wait_for_ready(timeout=10)
 
         # Start the orchestration and immediately raise events to it.
         with client.TaskHubGrpcClient() as task_hub_client:
@@ -265,7 +259,7 @@ def orchestrator(ctx: task.OrchestrationContext, _):
     with worker.TaskHubGrpcWorker(stop_timeout=2.0) as w:
         w.add_orchestrator(orchestrator)
         w.start()
-        w.wait_for_ready(timeout=10)
+
         with client.TaskHubGrpcClient() as task_hub_client:
             id = task_hub_client.schedule_new_orchestration(orchestrator)
             state = task_hub_client.wait_for_orchestration_start(id, timeout=30)
@@ -304,7 +298,7 @@ def orchestrator(ctx: task.OrchestrationContext, _):
     with worker.TaskHubGrpcWorker(stop_timeout=2.0) as w:
         w.add_orchestrator(orchestrator)
         w.start()
-        w.wait_for_ready(timeout=10)
+
         with client.TaskHubGrpcClient() as task_hub_client:
             id = task_hub_client.schedule_new_orchestration(orchestrator)
             state = task_hub_client.wait_for_orchestration_start(id, timeout=30)
@@ -348,7 +342,7 @@ def parent_orchestrator(ctx: task.OrchestrationContext, count: int):
             w.add_orchestrator(orchestrator_child)
             w.add_orchestrator(parent_orchestrator)
             w.start()
-            w.wait_for_ready(timeout=10)
+
             with client.TaskHubGrpcClient() as task_hub_client:
                 instance_id = task_hub_client.schedule_new_orchestration(
                     parent_orchestrator, input=5
@@ -397,7 +391,6 @@ def orchestrator(ctx: task.OrchestrationContext, input: int):
     with worker.TaskHubGrpcWorker(stop_timeout=2.0) as w:
         w.add_orchestrator(orchestrator)
         w.start()
-        w.wait_for_ready(timeout=10)
 
         task_hub_client = client.TaskHubGrpcClient()
         id = task_hub_client.schedule_new_orchestration(orchestrator, input=0)
@@ -498,7 +491,6 @@ def throw_activity_with_retry(ctx: task.ActivityContext, _):
         w.add_orchestrator(child_orchestrator_with_retry)
         w.add_activity(throw_activity_with_retry)
         w.start()
-        w.wait_for_ready(timeout=10)
 
         task_hub_client = client.TaskHubGrpcClient()
         id = task_hub_client.schedule_new_orchestration(parent_orchestrator_with_retry)
@@ -529,7 +521,6 @@ def orchestrator_with_non_retryable(ctx: task.OrchestrationContext, _):
         w.add_orchestrator(orchestrator_with_non_retryable)
         w.add_activity(throw_non_retryable)
         w.start()
-        w.wait_for_ready(timeout=10)
 
         task_hub_client = client.TaskHubGrpcClient()
         id = task_hub_client.schedule_new_orchestration(orchestrator_with_non_retryable)
@@ -568,7 +559,6 @@ def throw_activity(ctx: task.ActivityContext, _):
         w.add_orchestrator(mock_orchestrator)
         w.add_activity(throw_activity)
         w.start()
-        w.wait_for_ready(timeout=10)
 
         task_hub_client = client.TaskHubGrpcClient()
         id = task_hub_client.schedule_new_orchestration(mock_orchestrator)