add non-retryable error, add grpc options

filintod · filintod · commit f09216ec3250 · 2025-09-12T08:52:00.000-05:00
Signed-off-by: Filinto Duran &lt;1373693+filintod@users.noreply.github.com&gt;
diff --git a/durabletask/client.py b/durabletask/client.py
@@ -98,7 +98,8 @@ def __init__(self, *,
                  log_handler: Optional[logging.Handler] = None,
                  log_formatter: Optional[logging.Formatter] = None,
                  secure_channel: bool = False,
-                 interceptors: Optional[Sequence[shared.ClientInterceptor]] = None):
+                 interceptors: Optional[Sequence[shared.ClientInterceptor]] = None,
+                 options: Optional[Sequence[tuple[str, Any]]] = None):
 
         # If the caller provided metadata, we need to create a new interceptor for it and
         # add it to the list of interceptors.
@@ -114,7 +115,8 @@ def __init__(self, *,
         channel = shared.get_grpc_channel(
             host_address=host_address,
             secure_channel=secure_channel,
-            interceptors=interceptors
+            interceptors=interceptors,
+            options=options,
         )
         self._stub = stubs.TaskHubSidecarServiceStub(channel)
         self._logger = shared.get_logger("client", log_handler, log_formatter)
diff --git a/durabletask/internal/shared.py b/durabletask/internal/shared.py
@@ -6,7 +6,7 @@
 import logging
 import os
 from types import SimpleNamespace
-from typing import Any, Optional, Sequence, Union
+from typing import Any, Iterable, Optional, Sequence, Union
 
 import grpc
 
@@ -53,7 +53,8 @@ def get_default_host_address() -> str:
 def get_grpc_channel(
         host_address: Optional[str],
         secure_channel: bool = False,
-        interceptors: Optional[Sequence[ClientInterceptor]] = None) -> grpc.Channel:
+        interceptors: Optional[Sequence[ClientInterceptor]] = None,
+        options: Optional[Sequence[tuple[str, Any]]] = None) -> grpc.Channel:
     if host_address is None:
         host_address = get_default_host_address()
 
@@ -71,18 +72,139 @@ def get_grpc_channel(
             host_address = host_address[len(protocol):]
             break
 
-    # Create the base channel
+    # Build channel options (merge provided options with env-driven keepalive/retry)
+    channel_options = build_grpc_channel_options(options)
+
+    # Create the base channel (preserve original call signature when no options)
     if secure_channel:
-        channel = grpc.secure_channel(host_address, grpc.ssl_channel_credentials())
+        if channel_options is not None:
+            channel = grpc.secure_channel(
+                host_address, grpc.ssl_channel_credentials(), options=channel_options
+            )
+        else:
+            channel = grpc.secure_channel(host_address, grpc.ssl_channel_credentials())
     else:
-        channel = grpc.insecure_channel(host_address)
+        if channel_options is not None:
+            channel = grpc.insecure_channel(host_address, options=channel_options)
+        else:
+            channel = grpc.insecure_channel(host_address)
 
     # Apply interceptors ONLY if they exist
     if interceptors:
         channel = grpc.intercept_channel(channel, *interceptors)
     return channel
 
 
+def _get_env_bool(name: str, default: bool) -> bool:
+    val = os.environ.get(name)
+    if val is None:
+        return default
+    return val.strip().lower() in {"1", "true", "t", "yes", "y"}
+
+
+def _get_env_int(name: str, default: int) -> int:
+    val = os.environ.get(name)
+    if val is None:
+        return default
+    try:
+        return int(val)
+    except Exception:
+        return default
+
+
+def _get_env_float(name: str, default: float) -> float:
+    val = os.environ.get(name)
+    if val is None:
+        return default
+    try:
+        return float(val)
+    except Exception:
+        return default
+
+
+def _get_env_csv(name: str, default_csv: str) -> list[str]:
+    val = os.environ.get(name, default_csv)
+    return [s.strip().upper() for s in val.split(",") if s.strip()]
+
+
+def get_grpc_keepalive_options() -> list[tuple[str, Any]]:
+    """Build gRPC keepalive channel options from environment variables.
+
+    Environment variables (defaults in parentheses):
+    - DAPR_GRPC_KEEPALIVE_ENABLED (false)
+    - DAPR_GRPC_KEEPALIVE_TIME_MS (120000)
+    - DAPR_GRPC_KEEPALIVE_TIMEOUT_MS (20000)
+    - DAPR_GRPC_KEEPALIVE_PERMIT_WITHOUT_CALLS (false)
+    """
+    enabled = _get_env_bool("DAPR_GRPC_KEEPALIVE_ENABLED", False)
+    if not enabled:
+        return []
+    time_ms = _get_env_int("DAPR_GRPC_KEEPALIVE_TIME_MS", 120000)
+    timeout_ms = _get_env_int("DAPR_GRPC_KEEPALIVE_TIMEOUT_MS", 20000)
+    permit_without_calls = 1 if _get_env_bool("DAPR_GRPC_KEEPALIVE_PERMIT_WITHOUT_CALLS", False) else 0
+    return [
+        ("grpc.keepalive_time_ms", time_ms),
+        ("grpc.keepalive_timeout_ms", timeout_ms),
+        ("grpc.keepalive_permit_without_calls", permit_without_calls),
+    ]
+
+
+def get_grpc_retry_service_config_option() -> Optional[tuple[str, str]]:
+    """Return ("grpc.service_config", json_str) if retry is enabled via env; else None.
+
+    Environment variables (defaults in parentheses):
+    - DAPR_GRPC_RETRY_ENABLED (false)
+    - DAPR_GRPC_RETRY_MAX_ATTEMPTS (4)
+    - DAPR_GRPC_RETRY_INITIAL_BACKOFF_MS (100)
+    - DAPR_GRPC_RETRY_MAX_BACKOFF_MS (1000)
+    - DAPR_GRPC_RETRY_BACKOFF_MULTIPLIER (2.0)
+    - DAPR_GRPC_RETRY_CODES (UNAVAILABLE,DEADLINE_EXCEEDED)
+    """
+    enabled = _get_env_bool("DAPR_GRPC_RETRY_ENABLED", False)
+    if not enabled:
+        return None
+
+    max_attempts = _get_env_int("DAPR_GRPC_RETRY_MAX_ATTEMPTS", 4)
+    initial_backoff_ms = _get_env_int("DAPR_GRPC_RETRY_INITIAL_BACKOFF_MS", 100)
+    max_backoff_ms = _get_env_int("DAPR_GRPC_RETRY_MAX_BACKOFF_MS", 1000)
+    backoff_multiplier = _get_env_float("DAPR_GRPC_RETRY_BACKOFF_MULTIPLIER", 2.0)
+    codes = _get_env_csv("DAPR_GRPC_RETRY_CODES", "UNAVAILABLE,DEADLINE_EXCEEDED")
+
+    service_config = {
+        "methodConfig": [
+            {
+                "name": [{"service": ""}],
+                "retryPolicy": {
+                    "maxAttempts": max_attempts,
+                    "initialBackoff": f"{initial_backoff_ms/1000.0}s",
+                    "maxBackoff": f"{max_backoff_ms/1000.0}s",
+                    "backoffMultiplier": backoff_multiplier,
+                    "retryableStatusCodes": codes,
+                },
+            }
+        ]
+    }
+    return ("grpc.service_config", json.dumps(service_config))
+
+
+def build_grpc_channel_options(base_options: Optional[Iterable[tuple[str, Any]]] = None) -> Optional[list[tuple[str, Any]]]:
+    """Combine base options + env-driven keepalive and retry service config.
+
+    The returned list is safe to pass as the `options` argument to grpc.secure_channel/insecure_channel.
+    """
+    combined: list[tuple[str, Any]] = []
+    if base_options:
+        combined.extend(list(base_options))
+
+    keepalive = get_grpc_keepalive_options()
+    if keepalive:
+        combined.extend(keepalive)
+    retry_opt = get_grpc_retry_service_config_option()
+    if retry_opt is not None:
+        combined.append(retry_opt)
+    return combined if combined else None
+
+
 def get_logger(
         name_suffix: str,
         log_handler: Optional[logging.Handler] = None,
diff --git a/durabletask/task.py b/durabletask/task.py
@@ -219,6 +219,15 @@ class OrchestrationStateError(Exception):
     pass
 
 
+class NonRetryableError(Exception):
+    """Exception indicating the operation should not be retried.
+
+    If an activity or sub-orchestration raises this exception, retry logic will be
+    bypassed and the failure will be returned immediately to the orchestrator.
+    """
+    pass
+
+
 class Task(ABC, Generic[T]):
     """Abstract base class for asynchronous tasks in a durable orchestration."""
     _result: T
@@ -447,7 +456,8 @@ def __init__(self, *,
                  max_number_of_attempts: int,
                  backoff_coefficient: Optional[float] = 1.0,
                  max_retry_interval: Optional[timedelta] = None,
-                 retry_timeout: Optional[timedelta] = None):
+                 retry_timeout: Optional[timedelta] = None,
+                 non_retryable_error_types: Optional[list[Union[str, type]]] = None):
         """Creates a new RetryPolicy instance.
 
         Parameters
@@ -462,6 +472,11 @@ def __init__(self, *,
             The maximum retry interval to use for any retry attempt.
         retry_timeout : Optional[timedelta]
             The maximum amount of time to spend retrying the operation.
+        non_retryable_error_types : Optional[list[Union[str, type]]]
+            A list of exception type names or classes that should not be retried.
+            If a failure's error type matches any of these, the task fails immediately.
+            The built-in NonRetryableError is always treated as non-retryable regardless
+            of this setting.
         """
         # validate inputs
         if first_retry_interval < timedelta(seconds=0):
@@ -480,6 +495,17 @@ def __init__(self, *,
         self._backoff_coefficient = backoff_coefficient
         self._max_retry_interval = max_retry_interval
         self._retry_timeout = retry_timeout
+        # Normalize non-retryable error type names to a set of strings
+        names: Optional[set[str]] = None
+        if non_retryable_error_types:
+            names = set()
+            for t in non_retryable_error_types:
+                if isinstance(t, str):
+                    if t:
+                        names.add(t)
+                elif isinstance(t, type):
+                    names.add(t.__name__)
+        self._non_retryable_error_types = names
 
     @property
     def first_retry_interval(self) -> timedelta:
@@ -506,6 +532,15 @@ def retry_timeout(self) -> Optional[timedelta]:
         """The maximum amount of time to spend retrying the operation."""
         return self._retry_timeout
 
+    @property
+    def non_retryable_error_types(self) -> Optional[set[str]]:
+        """Set of error type names that should not be retried.
+
+        Comparison is performed against the errorType string provided by the
+        backend (typically the exception class name).
+        """
+        return self._non_retryable_error_types
+
 
 def get_name(fn: Callable) -> str:
     """Returns the name of the provided function"""
diff --git a/durabletask/worker.py b/durabletask/worker.py
@@ -20,8 +20,7 @@
 import durabletask.internal.orchestrator_service_pb2_grpc as stubs
 import durabletask.internal.shared as shared
 from durabletask import task
-from durabletask.asyncio_compat import (AsyncWorkflowContext,
-                                        CoroutineOrchestratorRunner)
+from durabletask.asyncio_compat import AsyncWorkflowContext, CoroutineOrchestratorRunner
 from durabletask.internal.grpc_interceptor import DefaultClientInterceptorImpl
 
 TInput = TypeVar("TInput")
@@ -763,7 +762,17 @@ def resume(self):
             else:
                 # Resume the generator with the previous result.
                 # This will either return a Task or raise StopIteration if it's done.
-                next_task = self._generator.send(self._previous_task.get_result())
+                try:
+                    _val = self._previous_task.get_result()
+                    import os as _os
+                    if _os.getenv('DAPR_WF_DEBUG') or _os.getenv('DT_DEBUG'):
+                        print(f"[DT] resume send instance={self._instance_id} type={type(_val)} is_none={_val is None}")
+                except Exception as _e:
+                    import os as _os
+                    if _os.getenv('DAPR_WF_DEBUG') or _os.getenv('DT_DEBUG'):
+                        print(f"[DT] resume send error instance={self._instance_id} err={_e}")
+                    raise
+                next_task = self._generator.send(_val)
 
             if not isinstance(next_task, task.Task):
                 raise TypeError("The orchestrator generator yielded a non-Task object")
@@ -1047,6 +1056,7 @@ def execute(
             )
             ctx._is_replaying = True
             for old_event in old_events:
+                self._logger.debug(f"OLD-EVENT: {instance_id}: {old_event}")
                 self.process_event(ctx, old_event)
 
             # Get new actions by executing newly received events into the orchestrator function
@@ -1119,6 +1129,13 @@ def process_event(
                 result = fn(
                     ctx, input
                 )  # this does not execute the generator, only creates it
+                try:
+                    from types import GeneratorType as _GenT
+                    import os as _os
+                    if _os.getenv('DAPR_WF_DEBUG') or _os.getenv('DT_DEBUG'):
+                        print(f"[DT] executionStarted orchestrator returned type={type(result)} is_gen={isinstance(result, _GenT)} id={id(result) if isinstance(result, _GenT) else 'n/a'}")
+                except Exception:
+                    pass
                 if isinstance(result, GeneratorType):
                     # Start the orchestrator's generator function
                     ctx.run(result)
@@ -1206,6 +1223,13 @@ def process_event(
                 result = None
                 if not ph.is_empty(event.taskCompleted.result):
                     result = shared.from_json(event.taskCompleted.result.value)
+                try:
+                    import os as _os
+                    if _os.getenv('DAPR_WF_DEBUG') or _os.getenv('DT_DEBUG'):
+                        print(f"[DT] taskCompleted decode instance={ctx.instance_id} task_id={task_id} type={type(result)} is_none={result is None}")
+                        print(f"[DT] pending_task_present={activity_task is not None}")
+                except Exception:
+                    pass
                 activity_task.complete(result)
                 ctx.resume()
             elif event.HasField("taskFailed"):
@@ -1221,16 +1245,32 @@ def process_event(
 
                 if isinstance(activity_task, task.RetryableTask):
                     if activity_task._retry_policy is not None:
-                        next_delay = activity_task.compute_next_delay()
-                        if next_delay is None:
+                        # Check for non-retryable errors by type name
+                        error_type = event.taskFailed.failureDetails.errorType
+                        policy = activity_task._retry_policy
+                        is_non_retryable = False
+                        if error_type == getattr(task.NonRetryableError, "__name__", "NonRetryableError"):
+                            is_non_retryable = True
+                        elif policy.non_retryable_error_types is not None and error_type in policy.non_retryable_error_types:
+                            is_non_retryable = True
+
+                        if is_non_retryable:
                             activity_task.fail(
                                 f"{ctx.instance_id}: Activity task #{task_id} failed: {event.taskFailed.failureDetails.errorMessage}",
                                 event.taskFailed.failureDetails,
                             )
                             ctx.resume()
                         else:
-                            activity_task.increment_attempt_count()
-                            ctx.create_timer_internal(next_delay, activity_task)
+                            next_delay = activity_task.compute_next_delay()
+                            if next_delay is None:
+                                activity_task.fail(
+                                    f"{ctx.instance_id}: Activity task #{task_id} failed: {event.taskFailed.failureDetails.errorMessage}",
+                                    event.taskFailed.failureDetails,
+                                )
+                                ctx.resume()
+                            else:
+                                activity_task.increment_attempt_count()
+                                ctx.create_timer_internal(next_delay, activity_task)
                 elif isinstance(activity_task, task.CompletableTask):
                     activity_task.fail(
                         f"{ctx.instance_id}: Activity task #{task_id} failed: {event.taskFailed.failureDetails.errorMessage}",
@@ -1292,16 +1332,32 @@ def process_event(
                     return
                 if isinstance(sub_orch_task, task.RetryableTask):
                     if sub_orch_task._retry_policy is not None:
-                        next_delay = sub_orch_task.compute_next_delay()
-                        if next_delay is None:
+                        # Check for non-retryable errors by type name
+                        error_type = failedEvent.failureDetails.errorType
+                        policy = sub_orch_task._retry_policy
+                        is_non_retryable = False
+                        if error_type == getattr(task.NonRetryableError, "__name__", "NonRetryableError"):
+                            is_non_retryable = True
+                        elif policy.non_retryable_error_types is not None and error_type in policy.non_retryable_error_types:
+                            is_non_retryable = True
+
+                        if is_non_retryable:
                             sub_orch_task.fail(
                                 f"Sub-orchestration task #{task_id} failed: {failedEvent.failureDetails.errorMessage}",
                                 failedEvent.failureDetails,
                             )
                             ctx.resume()
                         else:
-                            sub_orch_task.increment_attempt_count()
-                            ctx.create_timer_internal(next_delay, sub_orch_task)
+                            next_delay = sub_orch_task.compute_next_delay()
+                            if next_delay is None:
+                                sub_orch_task.fail(
+                                    f"Sub-orchestration task #{task_id} failed: {failedEvent.failureDetails.errorMessage}",
+                                    failedEvent.failureDetails,
+                                )
+                                ctx.resume()
+                            else:
+                                sub_orch_task.increment_attempt_count()
+                                ctx.create_timer_internal(next_delay, sub_orch_task)
                 elif isinstance(sub_orch_task, task.CompletableTask):
                     sub_orch_task.fail(
                         f"Sub-orchestration task #{task_id} failed: {failedEvent.failureDetails.errorMessage}",
diff --git a/tests/durabletask/test_orchestration_executor.py b/tests/durabletask/test_orchestration_executor.py
