feat(workflow): add a per-client default deadline to DaprWorkflowClient

dapr/aio/clients/grpc/interceptors.py

@@ -14,16 +14,19 @@
 """
 
 from collections import namedtuple
-from typing import List, Tuple
+from typing import List, Optional, Tuple
 
 from grpc.aio import (  # type: ignore
     ClientCallDetails,
     StreamStreamClientInterceptor,
     UnaryUnaryClientInterceptor,
 )
 
-from dapr.clients.grpc.interceptors import _is_unbounded_wait
-from dapr.conf import settings
+from dapr.clients.grpc.interceptors import (
+    _is_unbounded_wait,
+    _resolve_timeout_seconds,
+    _validate_timeout_seconds,
+)
 
 
 class _ClientCallDetailsAsync(
@@ -41,19 +44,39 @@ class _ClientCallDetailsAsync(
 
 
 class DaprClientTimeoutInterceptorAsync(UnaryUnaryClientInterceptor):
+    """Applies a default deadline to unary calls that do not set their own.
+
+    Calls that already carry a timeout keep it, and the workflow waits in
+    ``_UNBOUNDED_WAIT_METHODS`` are never given a default deadline.
+    """
+
+    def __init__(self, timeout_seconds: Optional[float] = None):
+        """Initializes the interceptor.
+
+        Args:
+            timeout_seconds: Default deadline in seconds for calls made through this
+                interceptor. ``None`` falls back to ``settings.DAPR_API_TIMEOUT_SECONDS``,
+                read on every call; when that is also ``None`` no deadline is applied.
+
+        Raises:
+            ValueError: If ``timeout_seconds`` is not a finite number greater than 0.
+        """
+        self._timeout_seconds = _validate_timeout_seconds(timeout_seconds)
+
     def intercept_unary_unary(self, continuation, client_call_details, request):
-        # Only apply a deadline when DAPR_API_TIMEOUT_SECONDS is explicitly configured.
-        # Without an explicit setting there is no SDK-level default deadline: Dapr's own
-        # resiliency policies and component timeouts act as the authoritative bounds for
+        # Only apply a deadline when one is explicitly configured, here or through
+        # DAPR_API_TIMEOUT_SECONDS. Without one there is no SDK-level default deadline: Dapr's
+        # own resiliency policies and component timeouts act as the authoritative bounds for
         # long-running operations such as LLM calls and workflow activities.
+        timeout_seconds = _resolve_timeout_seconds(self._timeout_seconds)
         if (
-            settings.DAPR_API_TIMEOUT_SECONDS is not None
+            timeout_seconds is not None
             and client_call_details.timeout is None
             and not _is_unbounded_wait(client_call_details.method)
         ):
             new_client_call_details = _ClientCallDetailsAsync(
                 client_call_details.method,
-                float(settings.DAPR_API_TIMEOUT_SECONDS),
+                timeout_seconds,
                 client_call_details.metadata,
                 client_call_details.credentials,
                 client_call_details.wait_for_ready,

dapr/clients/grpc/interceptors.py

@@ -1,5 +1,6 @@
+import math
 from collections import namedtuple
-from typing import List, Tuple, Union
+from typing import List, Optional, Tuple, Union
 
 from grpc import (  # type: ignore
     ClientCallDetails,
@@ -40,20 +41,58 @@ def _is_unbounded_wait(method: Union[str, bytes]) -> bool:
     return method in _UNBOUNDED_WAIT_METHODS
 
 
+def _validate_timeout_seconds(timeout_seconds: Optional[float]) -> Optional[float]:
+    if timeout_seconds is None:
+        return None
+    if not math.isfinite(timeout_seconds) or timeout_seconds <= 0:
+        raise ValueError(
+            f'timeout_seconds must be a finite number greater than 0, got {timeout_seconds}'
+        )
+    return float(timeout_seconds)
+
+
+def _resolve_timeout_seconds(timeout_seconds: Optional[float]) -> Optional[float]:
+    if timeout_seconds is not None:
+        return timeout_seconds
+    if settings.DAPR_API_TIMEOUT_SECONDS is None:
+        return None
+    return float(settings.DAPR_API_TIMEOUT_SECONDS)
+
+
 class DaprClientTimeoutInterceptor(UnaryUnaryClientInterceptor):
+    """Applies a default deadline to unary calls that do not set their own.
+
+    Calls that already carry a timeout keep it, and the workflow waits in
+    ``_UNBOUNDED_WAIT_METHODS`` are never given a default deadline.
+    """
+
+    def __init__(self, timeout_seconds: Optional[float] = None):
+        """Initializes the interceptor.
+
+        Args:
+            timeout_seconds: Default deadline in seconds for calls made through this
+                interceptor. ``None`` falls back to ``settings.DAPR_API_TIMEOUT_SECONDS``,
+                read on every call; when that is also ``None`` no deadline is applied.
+
+        Raises:
+            ValueError: If ``timeout_seconds`` is not a finite number greater than 0.
+        """
+        self._timeout_seconds = _validate_timeout_seconds(timeout_seconds)
+
     def intercept_unary_unary(self, continuation, client_call_details, request):
-        # Only apply a deadline when DAPR_API_TIMEOUT_SECONDS is explicitly configured.
-        # Without an explicit setting there is no SDK-level default deadline: Dapr's own
-        # resiliency policies and component timeouts act as the authoritative bounds for
+        # Only apply a deadline when one is explicitly configured, here or through
+        # DAPR_API_TIMEOUT_SECONDS. Without one there is no SDK-level default deadline: Dapr's
+        # own resiliency policies and component timeouts act as the authoritative bounds for
         # long-running operations such as LLM calls and workflow activities.
+        timeout_seconds = _resolve_timeout_seconds(self._timeout_seconds)
         if (
-            settings.DAPR_API_TIMEOUT_SECONDS is not None
+            timeout_seconds is not None
             and client_call_details.timeout is None
             and not _is_unbounded_wait(client_call_details.method)
         ):
             new_client_call_details = _ClientCallDetails(
                 client_call_details.method,
-                float(settings.DAPR_API_TIMEOUT_SECONDS),
+                timeout_seconds,
                 client_call_details.metadata,
                 client_call_details.credentials,
                 client_call_details.wait_for_ready,

dapr/ext/workflow/AGENTS.md

@@ -140,6 +140,8 @@ Workflow (orchestrator) functions must remain generators (`def` with `yield`). T
 
 Client for workflow lifecycle management:
 
+Constructor: `DaprWorkflowClient(host, port, logger_options, max_grpc_message_length, api_timeout_seconds)`. `api_timeout_seconds` is a default deadline for this client's unary calls; it overrides `DAPR_API_TIMEOUT_SECONDS` for this client only, through `DaprClientTimeoutInterceptor(timeout_seconds)`. `None` falls back to that setting, then to no deadline. Calls that set their own timeout keep it, and `wait_for_workflow_start` / `wait_for_workflow_completion` are never given it (`_UNBOUNDED_WAIT_METHODS`); their `timeout_in_seconds` bounds them.
+
 - `schedule_new_workflow(workflow, *, input, instance_id, start_at, reuse_id_policy)` → returns `instance_id`. `reuse_id_policy` is deprecated (no effect, emits `DeprecationWarning`): an instance ID can always be reused once the existing instance is in a terminal state.
 - `get_workflow_state(instance_id, *, fetch_payloads=True)` → `Optional[WorkflowState]`
 - `wait_for_workflow_start(instance_id, *, fetch_payloads, timeout_in_seconds)`

dapr/ext/workflow/aio/dapr_workflow_client.py

@@ -59,6 +59,7 @@ def __init__(
         port: Optional[str] = None,
         logger_options: Optional[LoggerOptions] = None,
         max_grpc_message_length: Optional[int] = None,
+        api_timeout_seconds: Optional[float] = None,
     ):
         """Initializes the async workflow client.
 
@@ -73,6 +74,17 @@ def __init__(
                 ``DAPR_GRPC_MAX_INBOUND_MESSAGE_SIZE_BYTES`` env var (if non-zero),
                 then the gRPC default (4 MiB). ``0`` in either source means
                 "no opinion" and falls through to the next source.
+            api_timeout_seconds: Default deadline in seconds for this client's calls to the
+                sidecar, such as ``schedule_new_workflow`` and ``get_workflow_state``. It
+                overrides ``DAPR_API_TIMEOUT_SECONDS`` for this client only, so a hung call
+                fails with ``DEADLINE_EXCEEDED`` without bounding other Dapr calls in the
+                process. ``None`` (the default) falls back to ``DAPR_API_TIMEOUT_SECONDS``,
+                and to no deadline when that is unset. It does not apply to
+                ``wait_for_workflow_start`` and ``wait_for_workflow_completion``, which are
+                bounded only by their own ``timeout_in_seconds``.
+
+        Raises:
+            ValueError: If ``api_timeout_seconds`` is not a finite number greater than 0.
         """
         address = getAddress(host, port)
 
@@ -88,13 +100,14 @@ def __init__(
             metadata = ((DAPR_API_TOKEN_HEADER, settings.DAPR_API_TOKEN),)
         options = self._logger.get_options()
         channel_options = get_grpc_channel_options(max_grpc_message_length)
+        timeout_interceptor = DaprClientTimeoutInterceptorAsync(api_timeout_seconds)
         self.__obj = aioclient.AsyncTaskHubGrpcClient(
             host_address=uri.endpoint,
             metadata=metadata,
             secure_channel=uri.tls,
             log_handler=options.log_handler,
             log_formatter=options.log_formatter,
-            interceptors=[DaprClientTimeoutInterceptorAsync()],
+            interceptors=[timeout_interceptor],
             channel_options=channel_options,
         )
 

dapr/ext/workflow/dapr_workflow_client.py

@@ -60,6 +60,7 @@ def __init__(
         port: Optional[str] = None,
         logger_options: Optional[LoggerOptions] = None,
         max_grpc_message_length: Optional[int] = None,
+        api_timeout_seconds: Optional[float] = None,
     ):
         """Initializes the sync workflow client.
 
@@ -74,6 +75,17 @@ def __init__(
                 ``DAPR_GRPC_MAX_INBOUND_MESSAGE_SIZE_BYTES`` env var (if non-zero),
                 then the gRPC default (4 MiB). ``0`` in either source means
                 "no opinion" and falls through to the next source.
+            api_timeout_seconds: Default deadline in seconds for this client's calls to the
+                sidecar, such as ``schedule_new_workflow`` and ``get_workflow_state``. It
+                overrides ``DAPR_API_TIMEOUT_SECONDS`` for this client only, so a hung call
+                fails with ``DEADLINE_EXCEEDED`` without bounding other Dapr calls in the
+                process. ``None`` (the default) falls back to ``DAPR_API_TIMEOUT_SECONDS``,
+                and to no deadline when that is unset. It does not apply to
+                ``wait_for_workflow_start`` and ``wait_for_workflow_completion``, which are
+                bounded only by their own ``timeout_in_seconds``.
+
+        Raises:
+            ValueError: If ``api_timeout_seconds`` is not a finite number greater than 0.
         """
         address = getAddress(host, port)
 
@@ -89,13 +101,14 @@ def __init__(
             metadata = ((DAPR_API_TOKEN_HEADER, settings.DAPR_API_TOKEN),)
         options = self._logger.get_options()
         channel_options = get_grpc_channel_options(max_grpc_message_length)
+        timeout_interceptor = DaprClientTimeoutInterceptor(api_timeout_seconds)
         self.__obj = client.TaskHubGrpcClient(
             host_address=uri.endpoint,
             metadata=metadata,
             secure_channel=uri.tls,
             log_handler=options.log_handler,
             log_formatter=options.log_formatter,
-            interceptors=[DaprClientTimeoutInterceptor()],
+            interceptors=[timeout_interceptor],
             channel_options=channel_options,
         )
 

tests/clients/test_timeout_interceptor.py

@@ -14,7 +14,8 @@
 """
 
 import unittest
-from unittest.mock import Mock, patch
+from typing import Optional, Union
+from unittest.mock import ANY, Mock, patch
 
 from dapr.clients.grpc.interceptors import DaprClientTimeoutInterceptor
 from dapr.conf import settings
@@ -159,3 +160,79 @@ def test_workflow_waits_keep_an_explicit_timeout(self):
         )
         continuation.assert_called_once_with(client_call_details, request)
         self.assertEqual(5, continuation.call_args[0][0].timeout)
+
+
+def _call_details(method: Union[str, bytes], timeout: Optional[float]) -> Mock:
+    client_call_details = Mock()
+    client_call_details.method = method
+    client_call_details.timeout = timeout
+    client_call_details.metadata = 'metadata'
+    client_call_details.credentials = 'credentials'
+    client_call_details.wait_for_ready = 'wait_for_ready'
+    client_call_details.compression = 'compression'
+    return client_call_details
+
+
+class DaprClientTimeoutInterceptorExplicitTimeoutTests(unittest.TestCase):
+    @patch.object(settings, 'DAPR_API_TIMEOUT_SECONDS', None)
+    def test_explicit_timeout_applies_without_global_timeout(self):
+        continuation = Mock()
+        client_call_details = _call_details('method', None)
+
+        DaprClientTimeoutInterceptor(3).intercept_unary_unary(
+            continuation, client_call_details, Mock()
+        )
+        self.assertEqual(3.0, continuation.call_args[0][0].timeout)
+
+    @patch.object(settings, 'DAPR_API_TIMEOUT_SECONDS', 7)
+    def test_explicit_timeout_overrides_global_timeout(self):
+        continuation = Mock()
+        client_call_details = _call_details('method', None)
+
+        DaprClientTimeoutInterceptor(3).intercept_unary_unary(
+            continuation, client_call_details, Mock()
+        )
+        self.assertEqual(3.0, continuation.call_args[0][0].timeout)
+
+    @patch.object(settings, 'DAPR_API_TIMEOUT_SECONDS', 7)
+    def test_explicit_timeout_keeps_per_call_timeout(self):
+        continuation = Mock()
+        client_call_details = _call_details('method', 10)
+
+        DaprClientTimeoutInterceptor(3).intercept_unary_unary(
+            continuation, client_call_details, Mock()
+        )
+        continuation.assert_called_once_with(client_call_details, ANY)
+        self.assertEqual(10, continuation.call_args[0][0].timeout)
+
+    @patch.object(settings, 'DAPR_API_TIMEOUT_SECONDS', None)
+    def test_explicit_timeout_skips_workflow_waits(self):
+        for method in (
+            '/TaskHubSidecarService/WaitForInstanceStart',
+            b'/TaskHubSidecarService/WaitForInstanceCompletion',
+        ):
+            with self.subTest(method=method):
+                continuation = Mock()
+                client_call_details = _call_details(method, None)
+
+                DaprClientTimeoutInterceptor(3).intercept_unary_unary(
+                    continuation, client_call_details, Mock()
+                )
+                self.assertIsNone(continuation.call_args[0][0].timeout)
+
+    def test_default_interceptor_reads_global_timeout_on_every_call(self):
+        interceptor = DaprClientTimeoutInterceptor()
+        for global_timeout, expected_timeout in ((None, None), (7, 7.0)):
+            with self.subTest(global_timeout=global_timeout):
+                continuation = Mock()
+                client_call_details = _call_details('method', None)
+
+                with patch.object(settings, 'DAPR_API_TIMEOUT_SECONDS', global_timeout):
+                    interceptor.intercept_unary_unary(continuation, client_call_details, Mock())
+                self.assertEqual(expected_timeout, continuation.call_args[0][0].timeout)
+
+    def test_non_positive_timeout_is_rejected(self):
+        for timeout in (0, -1, float('nan'), float('inf')):
+            with self.subTest(timeout=timeout):
+                with self.assertRaises(ValueError):
+                    DaprClientTimeoutInterceptor(timeout)