fix: address PR review comments

- Track channel ownership in sync/async clients; close() is a no-op
  when the channel was provided externally via the channel= parameter.
- Raise ValueError in GrpcRetryPolicyOptions.__post_init__ when either
  backoff duration rounds to zero at 9-decimal-place precision, rather
  than silently emitting a zero-duration string that gRPC rejects.

Author: andystaples

durabletask/client.py

@@ -160,6 +160,7 @@ def __init__(self, *,
                  default_version: Optional[str] = None,
                  payload_store: Optional[PayloadStore] = None):
 
+        self._owns_channel = channel is None
         if channel is None:
             interceptors = prepare_sync_interceptors(metadata, interceptors)
             channel = shared.get_grpc_channel(
@@ -175,8 +176,15 @@ def __init__(self, *,
         self._payload_store = payload_store
 
     def close(self) -> None:
-        """Closes the underlying gRPC channel."""
-        self._channel.close()
+        """Closes the underlying gRPC channel.
+
+        Only closes channels created internally. If a pre-configured channel
+        was passed via the ``channel`` constructor parameter, this method is
+        a no-op — the caller retains ownership and is responsible for closing
+        it.
+        """
+        if self._owns_channel:
+            self._channel.close()
 
     def schedule_new_orchestration(self, orchestrator: Union[task.Orchestrator[TInput, TOutput], str], *,
                                    input: Optional[TInput] = None,
@@ -444,6 +452,7 @@ def __init__(self, *,
                  default_version: Optional[str] = None,
                  payload_store: Optional[PayloadStore] = None):
 
+        self._owns_channel = channel is None
         if channel is None:
             interceptors = prepare_async_interceptors(metadata, interceptors)
             channel = shared.get_async_grpc_channel(
@@ -459,8 +468,15 @@ def __init__(self, *,
         self._payload_store = payload_store
 
     async def close(self) -> None:
-        """Closes the underlying gRPC channel."""
-        await self._channel.close()
+        """Closes the underlying gRPC channel.
+
+        Only closes channels created internally. If a pre-configured channel
+        was passed via the ``channel`` constructor parameter, this method is
+        a no-op — the caller retains ownership and is responsible for closing
+        it.
+        """
+        if self._owns_channel:
+            await self._channel.close()
 
     async def __aenter__(self):
         return self

durabletask/grpc_options.py

@@ -31,10 +31,21 @@ def __post_init__(self) -> None:
             raise ValueError("max_backoff_seconds must be >= initial_backoff_seconds")
         if len(self.retryable_status_codes) == 0:
             raise ValueError("retryable_status_codes cannot be empty")
+        # Validate that backoff values are representable as non-zero gRPC duration strings.
+        self._format_duration(self.initial_backoff_seconds)
+        self._format_duration(self.max_backoff_seconds)
 
     @staticmethod
     def _format_duration(seconds: float) -> str:
-        return f"{seconds:.3f}s"
+        formatted = f"{seconds:.9f}".rstrip('0')
+        if formatted.endswith('.'):
+            formatted += '0'
+        if float(formatted) == 0:
+            raise ValueError(
+                f"Duration {seconds!r} rounds to zero; use a value large enough to "
+                "produce a non-zero gRPC duration string."
+            )
+        return f"{formatted}s"
 
     def to_service_config(self) -> dict[str, Any]:
         return {

tests/durabletask/test_client.py

@@ -1,4 +1,5 @@
 import json
+import pytest
 from unittest.mock import ANY, MagicMock, patch
 
 from durabletask.client import AsyncTaskHubGrpcClient, TaskHubGrpcClient
@@ -62,6 +63,15 @@ def test_get_grpc_channel_with_retry_policy_service_config():
         assert retry_policy['retryableStatusCodes'] == ['UNAVAILABLE']
 
 
+def test_retry_policy_format_duration_raises_on_zero():
+    with pytest.raises(ValueError, match="rounds to zero"):
+        GrpcRetryPolicyOptions(
+            max_attempts=2,
+            initial_backoff_seconds=1e-15,
+            max_backoff_seconds=1e-15,
+        )
+
+
 def test_get_grpc_channel_default_host_address():
     with patch('grpc.insecure_channel') as mock_channel:
         get_grpc_channel(None, False, interceptors=INTERCEPTORS)