Merge remote-tracking branch 'upstream-ms/main' into update-upstream

Signed-off-by: Albert Callarisa <[email protected]>

Author: acroca

CHANGELOG.md

@@ -5,12 +5,23 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
-## v0.2.0 (Unreleased)
+## v0.3.0
+
+### New
+
+- Added `ConcurrencyOptions` class for fine-grained concurrency control with separate limits for activities and orchestrations. The thread pool worker count can also be configured.
+
+### Fixed
+
+- Fixed an issue where a worker could not recover after its connection was interrupted or severed
+
+## v0.2.1
 
 ### New
 
 - Added `set_custom_status` orchestrator API ([#31](https://github.com/microsoft/durabletask-python/pull/31)) - contributed by [@famarting](https://github.com/famarting)
 - Added `purge_orchestration` client API ([#34](https://github.com/microsoft/durabletask-python/pull/34)) - contributed by [@famarting](https://github.com/famarting)
+- Added new `durabletask-azuremanaged` package for use with the [Durable Task Scheduler](https://learn.microsoft.com/azure/azure-functions/durable/durable-task-scheduler/durable-task-scheduler) - by [@RyanLettieri](https://github.com/RyanLettieri)
 
 ### Changes
 

dev-requirements.txt

@@ -1 +1 @@
-grpcio-tools==1.62.3 # 1.62.X is the latest version before protobuf 1.26.X is used which has breaking changes for Python
+grpcio-tools

durabletask/client.py

@@ -6,7 +6,7 @@
 from dataclasses import dataclass
 from datetime import datetime
 from enum import Enum
-from typing import Any, Optional, TypeVar, Union
+from typing import Any, Optional, Sequence, TypeVar, Union
 
 import grpc
 from google.protobuf import wrappers_pb2
@@ -16,6 +16,7 @@
 import durabletask.internal.orchestrator_service_pb2_grpc as stubs
 import durabletask.internal.shared as shared
 from durabletask import task
+from durabletask.internal.grpc_interceptor import DefaultClientInterceptorImpl
 
 TInput = TypeVar('TInput')
 TOutput = TypeVar('TOutput')
@@ -96,8 +97,25 @@ def __init__(self, *,
                  metadata: Optional[list[tuple[str, str]]] = None,
                  log_handler: Optional[logging.Handler] = None,
                  log_formatter: Optional[logging.Formatter] = None,
-                 secure_channel: bool = False):
-        channel = shared.get_grpc_channel(host_address, metadata, secure_channel=secure_channel)
+                 secure_channel: bool = False,
+                 interceptors: Optional[Sequence[shared.ClientInterceptor]] = None):
+
+        # If the caller provided metadata, we need to create a new interceptor for it and
+        # add it to the list of interceptors.
+        if interceptors is not None:
+            interceptors = list(interceptors)
+            if metadata is not None:
+                interceptors.append(DefaultClientInterceptorImpl(metadata))
+        elif metadata is not None:
+            interceptors = [DefaultClientInterceptorImpl(metadata)]
+        else:
+            interceptors = None
+
+        channel = shared.get_grpc_channel(
+            host_address=host_address,
+            secure_channel=secure_channel,
+            interceptors=interceptors
+        )
         self._stub = stubs.TaskHubSidecarServiceStub(channel)
         self._logger = shared.get_logger("client", log_handler, log_formatter)
 
@@ -116,7 +134,7 @@ def schedule_new_orchestration(self, orchestrator: Union[task.Orchestrator[TInpu
             scheduledStartTimestamp=helpers.new_timestamp(start_at) if start_at else None,
             version=wrappers_pb2.StringValue(value=""),
             orchestrationIdReusePolicy=reuse_id_policy,
-            )
+        )
 
         self._logger.info(f"Starting new '{name}' instance with ID = '{req.instanceId}'.")
         res: pb.CreateInstanceResponse = self._stub.StartInstance(req)

durabletask/internal/grpc_interceptor.py

@@ -19,28 +19,28 @@ class _ClientCallDetails(
 
 
 class DefaultClientInterceptorImpl (
-    grpc.UnaryUnaryClientInterceptor, grpc.UnaryStreamClientInterceptor,
-    grpc.StreamUnaryClientInterceptor, grpc.StreamStreamClientInterceptor):
+        grpc.UnaryUnaryClientInterceptor, grpc.UnaryStreamClientInterceptor,
+        grpc.StreamUnaryClientInterceptor, grpc.StreamStreamClientInterceptor):
     """The class implements a UnaryUnaryClientInterceptor, UnaryStreamClientInterceptor,
-    StreamUnaryClientInterceptor and StreamStreamClientInterceptor from grpc to add an 
+    StreamUnaryClientInterceptor and StreamStreamClientInterceptor from grpc to add an
     interceptor to add additional headers to all calls as needed."""
 
     def __init__(self, metadata: list[tuple[str, str]]):
         super().__init__()
         self._metadata = metadata
 
     def _intercept_call(
-                self, client_call_details: _ClientCallDetails) -> grpc.ClientCallDetails:
+            self, client_call_details: _ClientCallDetails) -> grpc.ClientCallDetails:
         """Internal intercept_call implementation which adds metadata to grpc metadata in the RPC
             call details."""
         if self._metadata is None:
             return client_call_details
-        
+
         if client_call_details.metadata is not None:
             metadata = list(client_call_details.metadata)
         else:
             metadata = []
-        
+
         metadata.extend(self._metadata)
         client_call_details = _ClientCallDetails(
             client_call_details.method, client_call_details.timeout, metadata,

durabletask/internal/shared.py

@@ -5,11 +5,16 @@
 import json
 import logging
 from types import SimpleNamespace
-from typing import Any, Optional
+from typing import Any, Optional, Sequence, Union
 
 import grpc
 
-from durabletask.internal.grpc_interceptor import DefaultClientInterceptorImpl
+ClientInterceptor = Union[
+    grpc.UnaryUnaryClientInterceptor,
+    grpc.UnaryStreamClientInterceptor,
+    grpc.StreamUnaryClientInterceptor,
+    grpc.StreamStreamClientInterceptor
+]
 
 # Field name used to indicate that an object was automatically serialized
 # and should be deserialized as a SimpleNamespace
@@ -25,8 +30,8 @@ def get_default_host_address() -> str:
 
 def get_grpc_channel(
         host_address: Optional[str],
-        metadata: Optional[list[tuple[str, str]]],
-        secure_channel: bool = False) -> grpc.Channel:
+        secure_channel: bool = False,
+        interceptors: Optional[Sequence[ClientInterceptor]] = None) -> grpc.Channel:
     if host_address is None:
         host_address = get_default_host_address()
 
@@ -44,16 +49,18 @@ def get_grpc_channel(
             host_address = host_address[len(protocol):]
             break
 
+    # Create the base channel
     if secure_channel:
         channel = grpc.secure_channel(host_address, grpc.ssl_channel_credentials())
     else:
         channel = grpc.insecure_channel(host_address)
 
-    if metadata is not None and len(metadata) > 0:
-        interceptors = [DefaultClientInterceptorImpl(metadata)]
+    # Apply interceptors ONLY if they exist
+    if interceptors:
         channel = grpc.intercept_channel(channel, *interceptors)
     return channel
 
+
 def get_logger(
         name_suffix: str,
         log_handler: Optional[logging.Handler] = None,
@@ -98,7 +105,7 @@ def default(self, obj):
         if dataclasses.is_dataclass(obj):
             # Dataclasses are not serializable by default, so we convert them to a dict and mark them for
             # automatic deserialization by the receiver
-            d = dataclasses.asdict(obj)  # type: ignore 
+            d = dataclasses.asdict(obj)  # type: ignore
             d[AUTO_SERIALIZED] = True
             return d
         elif isinstance(obj, SimpleNamespace):

durabletask/task.py

@@ -333,7 +333,7 @@ class RetryableTask(CompletableTask[T]):
     """A task that can be retried according to a retry policy."""
 
     def __init__(self, retry_policy: RetryPolicy, action: pb.OrchestratorAction,
-                 start_time:datetime, is_sub_orch: bool) -> None:
+                 start_time: datetime, is_sub_orch: bool) -> None:
         super().__init__()
         self._action = action
         self._retry_policy = retry_policy
@@ -343,15 +343,15 @@ def __init__(self, retry_policy: RetryPolicy, action: pb.OrchestratorAction,
 
     def increment_attempt_count(self) -> None:
         self._attempt_count += 1
-    
+
     def compute_next_delay(self) -> Optional[timedelta]:
         if self._attempt_count >= self._retry_policy.max_number_of_attempts:
             return None
 
         retry_expiration: datetime = datetime.max
         if self._retry_policy.retry_timeout is not None and self._retry_policy.retry_timeout != datetime.max:
             retry_expiration = self._start_time + self._retry_policy.retry_timeout
-        
+
         if self._retry_policy.backoff_coefficient is None:
             backoff_coefficient = 1.0
         else: