Address automated review feedback

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

Author: Bernd Verst

docs/superpowers/plans/2026-04-23-grpc-resiliency.md

@@ -1,6 +1,10 @@
 # gRPC Resiliency Implementation Plan
 
-> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+> [!NOTE]
+> For agentic workers: REQUIRED SUB-SKILL:
+> Use superpowers:subagent-driven-development (recommended) or
+> superpowers:executing-plans to implement this plan task-by-task.
+> Steps use checkbox (`- [ ]`) syntax for tracking.
 
 **Goal:** Implement automatic healing of stale gRPC worker streams and client channels in `durabletask-python`, aligned with the behavior added in `durabletask-dotnet` PR 708.
 

durabletask/client.py

@@ -224,7 +224,7 @@ def _invoke_unary(
             method_name: str,
             request: Any,
             *,
-            timeout: Optional[int] = None):
+            timeout: Optional[float] = None):
         method = getattr(self._stub, method_name)
         try:
             if timeout is None:
@@ -406,7 +406,7 @@ def get_all_orchestration_states(self,
 
     def wait_for_orchestration_start(self, instance_id: str, *,
                                      fetch_payloads: bool = False,
-                                     timeout: int = 60) -> Optional[OrchestrationState]:
+                                     timeout: float = 60) -> Optional[OrchestrationState]:
         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.")
@@ -427,7 +427,7 @@ def wait_for_orchestration_start(self, instance_id: str, *,
 
     def wait_for_orchestration_completion(self, instance_id: str, *,
                                           fetch_payloads: bool = True,
-                                          timeout: int = 60) -> Optional[OrchestrationState]:
+                                          timeout: float = 60) -> Optional[OrchestrationState]:
         req = pb.GetInstanceRequest(instanceId=instance_id, getInputsAndOutputs=fetch_payloads)
         try:
             self._logger.info(f"Waiting {timeout}s for instance '{instance_id}' to complete.")
@@ -685,7 +685,7 @@ async def _invoke_unary(
             method_name: str,
             request: Any,
             *,
-            timeout: Optional[int] = None):
+            timeout: Optional[float] = None):
         method = getattr(self._stub, method_name)
         try:
             if timeout is None:
@@ -733,10 +733,9 @@ async def _close_retired_channel(self, channel: grpc.aio.Channel) -> None:
             await asyncio.sleep(30.0)
             await channel.close()
         finally:
-            try:
-                self._retired_channels.remove(channel)
-            except ValueError:
-                pass
+            async with self._recreate_lock:
+                if channel in self._retired_channels:
+                    self._retired_channels.remove(channel)
 
     async def schedule_new_orchestration(self, orchestrator: Union[task.Orchestrator[TInput, TOutput], str], *,
                                          input: Optional[TInput] = None,
@@ -843,7 +842,7 @@ async def get_all_orchestration_states(self,
 
     async def wait_for_orchestration_start(self, instance_id: str, *,
                                            fetch_payloads: bool = False,
-                                           timeout: int = 60) -> Optional[OrchestrationState]:
+                                           timeout: float = 60) -> Optional[OrchestrationState]:
         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.")
@@ -863,7 +862,7 @@ async def wait_for_orchestration_start(self, instance_id: str, *,
 
     async def wait_for_orchestration_completion(self, instance_id: str, *,
                                                 fetch_payloads: bool = True,
-                                                timeout: int = 60) -> Optional[OrchestrationState]:
+                                                timeout: float = 60) -> Optional[OrchestrationState]:
         req = pb.GetInstanceRequest(instanceId=instance_id, getInputsAndOutputs=fetch_payloads)
         try:
             self._logger.info(f"Waiting {timeout}s for instance '{instance_id}' to complete.")

durabletask/internal/grpc_resiliency.py

@@ -13,7 +13,8 @@ def get_full_jitter_delay_seconds(
         attempt: int,
         *,
         base_seconds: float,
-        cap_seconds: float) -> float:
+        cap_seconds: float,
+) -> float:
     capped_attempt = min(attempt, 30)
     upper_bound = min(cap_seconds, base_seconds * (2 ** capped_attempt))
     return random.random() * upper_bound

durabletask/worker.py

@@ -194,7 +194,7 @@ def _close_channel(channel: Any) -> None:
         try:
             channel.close()
         except Exception:
-            pass
+            logging.debug("Ignoring channel close failure during worker cleanup.", exc_info=True)
 
 
 class VersioningOptions:
@@ -744,9 +744,10 @@ def create_fresh_connection():
 
         def wrap_execution(handler, release):
             def wrapped(*args, **kwargs):
-                result = handler(*args, **kwargs)
-                release()
-                return result
+                try:
+                    return handler(*args, **kwargs)
+                finally:
+                    release()
 
             return wrapped
 

tests/durabletask/test_client.py

@@ -801,8 +801,8 @@ async def test_async_client_close_prevents_channel_recreation_race():
             client._recreate_lock.release()
 
         with pytest.raises(grpc.aio.AioRpcError):
-            await rpc_task
-        await close_task
+            _ = await rpc_task
+        _ = await close_task
 
     assert mock_get_channel.call_count == 1
     first_channel.close.assert_awaited_once()

tests/durabletask/test_worker_resiliency.py

@@ -449,6 +449,56 @@ def create_stub(channel):
     created_channels[0].close.assert_called_once()
 
 
+@pytest.mark.asyncio
+async def test_worker_releases_inflight_channel_when_activity_handler_raises(monkeypatch):
+    worker = TaskHubGrpcWorker()
+    worker_manager = DummyWorkerManager()
+    worker._async_worker_manager = worker_manager
+    monkeypatch.setattr(worker._shutdown, "wait", lambda timeout: False)
+
+    def fail_activity(req, stub, completion_token):
+        raise RuntimeError("boom")
+
+    worker._execute_activity = fail_activity
+
+    created_channels = []
+
+    def get_grpc_channel(*args, **kwargs):
+        channel = MagicMock(name=f"channel-{len(created_channels) + 1}")
+        created_channels.append(channel)
+        return channel
+
+    first_stub = MagicMock()
+    first_stub.GetWorkItems.return_value = FakeResponseStream(items=[_make_activity_work_item()])
+
+    second_stub = MagicMock()
+    second_stub.GetWorkItems.side_effect = FakeRpcError(
+        grpc.StatusCode.CANCELLED,
+        "stop",
+    )
+
+    stubs = [first_stub, second_stub]
+
+    def create_stub(channel):
+        return stubs.pop(0)
+
+    monkeypatch.setattr("durabletask.worker.shared.get_grpc_channel", get_grpc_channel)
+    monkeypatch.setattr("durabletask.worker.stubs.TaskHubSidecarServiceStub", create_stub)
+
+    await worker._async_run_loop()
+
+    assert len(worker_manager.submissions) == 1
+    created_channels[0].close.assert_not_called()
+
+    _, submission = worker_manager.submissions[0]
+    func, _, req, stub, completion_token = submission
+    with pytest.raises(RuntimeError, match="boom"):
+        func(req, stub, completion_token)
+
+    created_channels[0].close.assert_called_once()
+    created_channels[1].close.assert_called_once()
+
+
 @pytest.mark.asyncio
 async def test_worker_shutdown_drains_real_manager_work_before_closing_retired_sdk_channel(monkeypatch):
     worker = TaskHubGrpcWorker(