diff --git a/CHANGELOG.md b/CHANGELOG.md index 40aa2e88..f2511101 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,13 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ## Unreleased +ADDED + +- Added public `OrchestrationIdReusePolicy` with typed deduplication statuses for +synchronous and asynchronous orchestration scheduling, including legacy +`OrchestrationStatus.CANCELED`. Omission preserves backend defaults; existing +protobuf policies remain supported. See [ID reuse](docs/features.md#orchestration-id-reuse). + FIXED - Activity and sub-orchestration `retry_timeout` decisions now use replay-safe diff --git a/README.md b/README.md index fb947a98..93fd0f79 100644 --- a/README.md +++ b/README.md @@ -32,6 +32,14 @@ The Azure Functions Durable provider requires Python 3.13 or later. ## Optional Features +### Orchestration ID Reuse + +Use `OrchestrationIdReusePolicy` from `durabletask.client` to select which +existing-instance statuses reject duplicate IDs. Omission preserves backend +defaults; an explicit empty deduplication list permits replacement in every +stable status, subject to backend support. See the +[policy documentation](./docs/features.md#orchestration-id-reuse). + ### Task Hub Query Filters Set `OrchestrationQuery.task_hub_names` to filter orchestration queries by one or diff --git a/azure-functions-durable/CHANGELOG.md b/azure-functions-durable/CHANGELOG.md index 4dd19e87..f931a4ac 100644 --- a/azure-functions-durable/CHANGELOG.md +++ b/azure-functions-durable/CHANGELOG.md @@ -9,6 +9,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ADDED +- With the corresponding core SDK release, Functions-bound synchronous and +asynchronous clients accept public `OrchestrationIdReusePolicy` values through +`schedule_new_orchestration()`, preserving backend defaults when omitted. +See [ID reuse](../docs/features.md#orchestration-id-reuse). Core release and +dependency minimum updates are coordinated in the dedicated release PR. - A best-effort background check warns when a Function App's requirements allow a future major version, which may introduce replay-breaking changes. diff --git a/docs/features.md b/docs/features.md index 8573aa0d..3a401d3b 100644 --- a/docs/features.md +++ b/docs/features.md @@ -11,6 +11,64 @@ fault-tolerant and durable, meaning that they can automatically recover from fai their local execution state. Orchestrator functions must be deterministic, meaning that they must always produce the same output given the same input. +### Orchestration ID reuse + +Use the public `OrchestrationIdReusePolicy` to control what happens when scheduling +an orchestration whose instance ID already exists: + +```python +from durabletask.client import OrchestrationIdReusePolicy, OrchestrationStatus + +instance_id = client.schedule_new_orchestration( + "process_order", + instance_id="order-123", + reuse_id_policy=OrchestrationIdReusePolicy( + dedupe_statuses=[ + OrchestrationStatus.RUNNING, + OrchestrationStatus.PENDING, + OrchestrationStatus.SUSPENDED, + OrchestrationStatus.TERMINATED, + OrchestrationStatus.CANCELED, + ], + ), +) +``` + +In this example, only completed or failed instances are eligible for replacement. +`dedupe_statuses` contains the statuses that **must not** be replaced, matching the +.NET and Go SDKs' deduplication semantics. + +- Omitting `reuse_id_policy`, or using `OrchestrationIdReusePolicy()` with + `dedupe_statuses=None`, preserves the backend's default duplicate-ID behavior. +- An explicit `dedupe_statuses=[]` permits replacement in every stable status, + including running instances. Include all stable statuses to reject every + duplicate. +- Stable statuses are `RUNNING`, `PENDING`, `SUSPENDED`, `COMPLETED`, `FAILED`, + `TERMINATED`, and the legacy `CANCELED` status. `CONTINUED_AS_NEW` is an execution + transition, not a configurable instance status. Invalid status values raise + `ValueError` before the scheduling request is sent. + +The policy works with synchronous and asynchronous core, Azure Managed, and +Functions-bound clients. Use `await client.schedule_new_orchestration(...)` with +an asynchronous client. The deprecated Functions `start_new()` alias does not +accept policy options. + +Existing callers may continue passing a generated protobuf policy through +`reuse_id_policy`. Its `replaceableStatus` list has the **opposite polarity**: +it lists statuses that may be replaced. An explicitly empty protobuf policy still +permits no replacements; it is not equivalent to an empty public deduplication list. + +> [!WARNING] +> Replacing an active instance can terminate its execution. Support, replacement +> guarantees, and duplicate rejection are enforced by the backend, not by the SDK. +> The SDK sends one scheduling request; it does not perform a separate +> terminate-then-start sequence. The shared protocol does not provide an atomic +> no-op/`IGNORE` action. + +Provider releases exposing this public policy require the corresponding core SDK +release first. Package versions and dependency minimums are coordinated in the +dedicated release PR. + ### Orchestration versioning Orchestrations may be assigned a version when they are first created. If an orchestration is given a diff --git a/durabletask-azuremanaged/CHANGELOG.md b/durabletask-azuremanaged/CHANGELOG.md index 1b327beb..15dbdea2 100644 --- a/durabletask-azuremanaged/CHANGELOG.md +++ b/durabletask-azuremanaged/CHANGELOG.md @@ -7,6 +7,13 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ## Unreleased +ADDED + +- With the corresponding core SDK release, synchronous and asynchronous clients +accept public `OrchestrationIdReusePolicy` values to control duplicate instance +IDs without changing backend defaults. See [ID reuse](../docs/features.md#orchestration-id-reuse). +Core release and dependency minimum updates are coordinated in the dedicated release PR. + FIXED - With the corresponding core SDK update, activity and sub-orchestration diff --git a/durabletask/client.py b/durabletask/client.py index 64d9525b..88730f8d 100644 --- a/durabletask/client.py +++ b/durabletask/client.py @@ -63,6 +63,7 @@ class OrchestrationStatus(Enum): RUNNING = pb.ORCHESTRATION_STATUS_RUNNING COMPLETED = pb.ORCHESTRATION_STATUS_COMPLETED FAILED = pb.ORCHESTRATION_STATUS_FAILED + CANCELED = pb.ORCHESTRATION_STATUS_CANCELED TERMINATED = pb.ORCHESTRATION_STATUS_TERMINATED CONTINUED_AS_NEW = pb.ORCHESTRATION_STATUS_CONTINUED_AS_NEW PENDING = pb.ORCHESTRATION_STATUS_PENDING @@ -72,6 +73,23 @@ def __str__(self) -> str: return cast(str, helpers.get_orchestration_status_str(self.value)) +@dataclass +class OrchestrationIdReusePolicy: + """Control duplicate instance IDs when scheduling an orchestration. + + ``dedupe_statuses`` lists existing-instance statuses that must not be + replaced. Other stable statuses permit replacement if supported by the + backend. ``None`` preserves backend defaults; an empty sequence permits + replacement in every stable status. ``CONTINUED_AS_NEW`` is a transition, + not a configurable instance status. + + This policy does not provide an atomic no-op/ignore action. Replacement + behavior and support are determined by the backend. + """ + + dedupe_statuses: Sequence[OrchestrationStatus] | None = None + + @dataclass class OrchestrationState: instance_id: str @@ -631,13 +649,18 @@ def schedule_new_orchestration(self, orchestrator: task.Orchestrator[TInput, TOu input: TInput | None = None, instance_id: str | None = None, start_at: datetime | None = None, - reuse_id_policy: pb.OrchestrationIdReusePolicy | None = None, + reuse_id_policy: OrchestrationIdReusePolicy | pb.OrchestrationIdReusePolicy | None = None, tags: dict[str, str] | None = None, version: str | None = None) -> str: """Schedule a new orchestration instance for execution. If instance_id is not specified, a random UUID hex string is generated upfront and shared across both the OpenTelemetry producer span and the CreateInstanceRequest. + + ``reuse_id_policy`` controls which existing-instance statuses reject + duplicate IDs. Omission preserves backend defaults. Existing protobuf + policies remain supported and retain their replaceable-status semantics. + Invalid public policy statuses raise ``ValueError`` before sending the request. """ name = orchestrator if isinstance(orchestrator, str) else task.get_name(orchestrator) resolved_instance_id = instance_id if instance_id else uuid.uuid4().hex @@ -1183,13 +1206,18 @@ async def schedule_new_orchestration(self, orchestrator: task.Orchestrator[TInpu input: TInput | None = None, instance_id: str | None = None, start_at: datetime | None = None, - reuse_id_policy: pb.OrchestrationIdReusePolicy | None = None, + reuse_id_policy: OrchestrationIdReusePolicy | pb.OrchestrationIdReusePolicy | None = None, tags: dict[str, str] | None = None, version: str | None = None) -> str: """Schedule a new orchestration instance for execution asynchronously. If instance_id is not specified, a random UUID hex string is generated upfront and shared across both the OpenTelemetry producer span and the CreateInstanceRequest. + + ``reuse_id_policy`` controls which existing-instance statuses reject + duplicate IDs. Omission preserves backend defaults. Existing protobuf + policies remain supported and retain their replaceable-status semantics. + Invalid public policy statuses raise ``ValueError`` before sending the request. """ name = orchestrator if isinstance(orchestrator, str) else task.get_name(orchestrator) resolved_instance_id = instance_id if instance_id else uuid.uuid4().hex diff --git a/durabletask/internal/client_helpers.py b/durabletask/internal/client_helpers.py index fc4fd7c9..995afd76 100644 --- a/durabletask/internal/client_helpers.py +++ b/durabletask/internal/client_helpers.py @@ -23,6 +23,7 @@ if TYPE_CHECKING: from durabletask.client import ( EntityQuery, + OrchestrationIdReusePolicy, OrchestrationQuery, OrchestrationState, OrchestrationStatus, @@ -77,11 +78,30 @@ def build_schedule_new_orchestration_req( input: TInput | None, instance_id: str | None, start_at: datetime | None, - reuse_id_policy: pb.OrchestrationIdReusePolicy | None, + reuse_id_policy: OrchestrationIdReusePolicy | pb.OrchestrationIdReusePolicy | None, tags: dict[str, str] | None, version: str | None, data_converter: DataConverter) -> pb.CreateInstanceRequest: """Build a CreateInstanceRequest for scheduling a new orchestration.""" + if reuse_id_policy is not None and not isinstance(reuse_id_policy, pb.OrchestrationIdReusePolicy): + from durabletask.client import OrchestrationStatus + + dedupe_statuses = reuse_id_policy.dedupe_statuses + if dedupe_statuses is None: + reuse_id_policy = None + else: + stable_statuses = tuple( + status for status in OrchestrationStatus + if status is not OrchestrationStatus.CONTINUED_AS_NEW) + for status in dedupe_statuses: + if status not in stable_statuses: + raise ValueError(f"Invalid orchestration dedupe status: {status!r}") + reuse_id_policy = pb.OrchestrationIdReusePolicy( + replaceableStatus=[ + status.value for status in stable_statuses + if status not in dedupe_statuses + ]) + name = orchestrator if isinstance(orchestrator, str) else task.get_name(orchestrator) return pb.CreateInstanceRequest( name=name, diff --git a/tests/azure-functions-durable/test_client_transport.py b/tests/azure-functions-durable/test_client_transport.py index 2a1ac7c7..576156e6 100644 --- a/tests/azure-functions-durable/test_client_transport.py +++ b/tests/azure-functions-durable/test_client_transport.py @@ -11,7 +11,13 @@ from google.protobuf.wrappers_pb2 import StringValue from azure.durable_functions import DurableFunctionsClient, SyncDurableFunctionsClient -from durabletask.client import AsyncTaskHubGrpcClient, OrchestrationQuery, OrchestrationStatus, TaskHubGrpcClient +from durabletask.client import ( + AsyncTaskHubGrpcClient, + OrchestrationIdReusePolicy, + OrchestrationQuery, + OrchestrationStatus, + TaskHubGrpcClient, +) from durabletask.internal import orchestrator_service_pb2 as pb from durabletask.internal import orchestrator_service_pb2_grpc as rpc @@ -26,6 +32,35 @@ def _config(address: str) -> str: }) +@pytest.mark.asyncio +@pytest.mark.parametrize("use_async", [False, True]) +async def test_functions_client_accepts_public_reuse_policy(use_async): + stub = Mock() + response = pb.CreateInstanceResponse(instanceId="instance") + policy = OrchestrationIdReusePolicy(dedupe_statuses=[OrchestrationStatus.RUNNING]) + if use_async: + stub.StartInstance = AsyncMock(return_value=response) + async with DurableFunctionsClient(_config("localhost:8080")) as client: + with patch.object(client, "_get_stub", return_value=stub): + result = await client.schedule_new_orchestration( + "orchestrator", instance_id="instance", reuse_id_policy=policy) + else: + stub.StartInstance.return_value = response + with SyncDurableFunctionsClient(_config("localhost:8080")) as client: + with patch.object(client, "_stub", stub): + result = client.schedule_new_orchestration( + "orchestrator", instance_id="instance", reuse_id_policy=policy) + + assert result == "instance" + request = stub.StartInstance.call_args.args[0] + assert request.HasField("orchestrationIdReusePolicy") + assert list(request.orchestrationIdReusePolicy.replaceableStatus) == [ + pb.ORCHESTRATION_STATUS_COMPLETED, pb.ORCHESTRATION_STATUS_FAILED, + pb.ORCHESTRATION_STATUS_CANCELED, pb.ORCHESTRATION_STATUS_TERMINATED, + pb.ORCHESTRATION_STATUS_PENDING, pb.ORCHESTRATION_STATUS_SUSPENDED, + ] + + @pytest.mark.parametrize("use_async", [False, True]) @pytest.mark.parametrize("address, expected", [ ("http://127.0.0.1:8080/durabletask/", "http://127.0.0.1:8080"), diff --git a/tests/durabletask-azuremanaged/test_azuremanaged_grpc_resiliency.py b/tests/durabletask-azuremanaged/test_azuremanaged_grpc_resiliency.py index 1c9b1a86..2637a381 100644 --- a/tests/durabletask-azuremanaged/test_azuremanaged_grpc_resiliency.py +++ b/tests/durabletask-azuremanaged/test_azuremanaged_grpc_resiliency.py @@ -1,13 +1,14 @@ # Copyright (c) Microsoft Corporation. # Licensed under the MIT License. -from unittest.mock import Mock, patch +from unittest.mock import AsyncMock, Mock, patch from collections.abc import Mapping from typing import Any import pytest from durabletask import task +from durabletask.client import OrchestrationIdReusePolicy, OrchestrationStatus from durabletask.internal import helpers from durabletask.internal import orchestrator_service_pb2 as pb from durabletask.azuremanaged.client import ( @@ -21,6 +22,35 @@ ) +@pytest.mark.asyncio +@pytest.mark.parametrize("use_async", [False, True]) +async def test_dts_client_accepts_public_reuse_policy(use_async): + stub = Mock() + response = pb.CreateInstanceResponse(instanceId="instance") + stub.StartInstance = AsyncMock(return_value=response) if use_async else Mock(return_value=response) + policy = OrchestrationIdReusePolicy(dedupe_statuses=[OrchestrationStatus.RUNNING]) + kwargs = dict(host_address="localhost:4001", taskhub="hub", + token_credential=None, channel=Mock()) + + with patch("durabletask.client.stubs.TaskHubSidecarServiceStub", return_value=stub): + if use_async: + async with AsyncDurableTaskSchedulerClient(**kwargs) as client: + assert await client.schedule_new_orchestration( + "orchestrator", instance_id="instance", reuse_id_policy=policy) == "instance" + else: + with DurableTaskSchedulerClient(**kwargs) as client: + assert client.schedule_new_orchestration( + "orchestrator", instance_id="instance", reuse_id_policy=policy) == "instance" + + request = stub.StartInstance.call_args.args[0] + assert request.HasField("orchestrationIdReusePolicy") + assert list(request.orchestrationIdReusePolicy.replaceableStatus) == [ + pb.ORCHESTRATION_STATUS_COMPLETED, pb.ORCHESTRATION_STATUS_FAILED, + pb.ORCHESTRATION_STATUS_CANCELED, pb.ORCHESTRATION_STATUS_TERMINATED, + pb.ORCHESTRATION_STATUS_PENDING, pb.ORCHESTRATION_STATUS_SUSPENDED, + ] + + def test_dts_client_passes_resiliency_options_to_base_client(): resiliency = GrpcClientResiliencyOptions() with patch("durabletask.azuremanaged.client.TaskHubGrpcClient.__init__", return_value=None) as mock_init: diff --git a/tests/durabletask/test_client.py b/tests/durabletask/test_client.py index ae53f163..28908453 100644 --- a/tests/durabletask/test_client.py +++ b/tests/durabletask/test_client.py @@ -14,7 +14,12 @@ import durabletask.history as history import durabletask.internal.orchestrator_service_pb2 as pb -from durabletask.client import AsyncTaskHubGrpcClient, OrchestrationStatus, TaskHubGrpcClient +from durabletask.client import ( + AsyncTaskHubGrpcClient, + OrchestrationIdReusePolicy, + OrchestrationStatus, + TaskHubGrpcClient, +) from durabletask.grpc_options import ( GrpcChannelOptions, GrpcClientResiliencyOptions, @@ -39,6 +44,81 @@ INTERCEPTORS = [DefaultClientInterceptorImpl(METADATA)] +@pytest.mark.asyncio +@pytest.mark.parametrize("use_async", [False, True]) +@pytest.mark.parametrize("policy, expected_statuses", [ + (None, None), + (OrchestrationIdReusePolicy(), None), + (OrchestrationIdReusePolicy(dedupe_statuses=[]), [ + pb.ORCHESTRATION_STATUS_RUNNING, pb.ORCHESTRATION_STATUS_COMPLETED, + pb.ORCHESTRATION_STATUS_FAILED, pb.ORCHESTRATION_STATUS_CANCELED, + pb.ORCHESTRATION_STATUS_TERMINATED, pb.ORCHESTRATION_STATUS_PENDING, + pb.ORCHESTRATION_STATUS_SUSPENDED, + ]), + (OrchestrationIdReusePolicy(dedupe_statuses=( + OrchestrationStatus.RUNNING, OrchestrationStatus.PENDING, + OrchestrationStatus.SUSPENDED, OrchestrationStatus.TERMINATED, + OrchestrationStatus.RUNNING, + )), [ + pb.ORCHESTRATION_STATUS_COMPLETED, pb.ORCHESTRATION_STATUS_FAILED, + pb.ORCHESTRATION_STATUS_CANCELED, + ]), + (OrchestrationIdReusePolicy(dedupe_statuses=[ + OrchestrationStatus.RUNNING, OrchestrationStatus.COMPLETED, + OrchestrationStatus.FAILED, OrchestrationStatus.CANCELED, + OrchestrationStatus.TERMINATED, OrchestrationStatus.PENDING, + OrchestrationStatus.SUSPENDED, + ]), []), + (pb.OrchestrationIdReusePolicy(), []), + (pb.OrchestrationIdReusePolicy(replaceableStatus=[pb.ORCHESTRATION_STATUS_FAILED]), + [pb.ORCHESTRATION_STATUS_FAILED]), +]) +async def test_schedule_new_orchestration_reuse_policy(use_async, policy, expected_statuses): + stub = MagicMock() + response = pb.CreateInstanceResponse(instanceId="instance") + stub.StartInstance = AsyncMock(return_value=response) if use_async else MagicMock(return_value=response) + + with patch('durabletask.client.stubs.TaskHubSidecarServiceStub', return_value=stub): + if use_async: + async with AsyncTaskHubGrpcClient(channel=MagicMock()) as client: + result = await client.schedule_new_orchestration( + "orchestrator", instance_id="instance", reuse_id_policy=policy) + else: + with TaskHubGrpcClient(channel=MagicMock()) as client: + result = client.schedule_new_orchestration( + "orchestrator", instance_id="instance", reuse_id_policy=policy) + + assert result == "instance" + stub.StartInstance.assert_called_once() + request = stub.StartInstance.call_args.args[0] + assert request.HasField("orchestrationIdReusePolicy") == (expected_statuses is not None) + if expected_statuses is not None: + assert list(request.orchestrationIdReusePolicy.replaceableStatus) == expected_statuses + stub.GetInstance.assert_not_called() + stub.TerminateInstance.assert_not_called() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("use_async", [False, True]) +@pytest.mark.parametrize("status", [ + OrchestrationStatus.CONTINUED_AS_NEW, "Running", pb.ORCHESTRATION_STATUS_RUNNING, +]) +async def test_schedule_new_orchestration_rejects_invalid_dedupe_status(use_async, status): + stub = MagicMock() + policy = OrchestrationIdReusePolicy(dedupe_statuses=[status]) + + with patch('durabletask.client.stubs.TaskHubSidecarServiceStub', return_value=stub): + with pytest.raises(ValueError, match="Invalid orchestration dedupe status"): + if use_async: + async with AsyncTaskHubGrpcClient(channel=MagicMock()) as client: + await client.schedule_new_orchestration("orchestrator", reuse_id_policy=policy) + else: + with TaskHubGrpcClient(channel=MagicMock()) as client: + client.schedule_new_orchestration("orchestrator", reuse_id_policy=policy) + + stub.StartInstance.assert_not_called() + + class FakeRpcError(grpc.RpcError): def __init__(self, status_code: grpc.StatusCode): super().__init__()