From 85a70f93675ecf52d873f891d523b353acefb283 Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Mon, 10 Aug 2026 11:25:32 -0600 Subject: [PATCH 1/2] Add timeout to filtered orchestration purge Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 2af7eb6a-76b9-4c7e-98d6-1efc7e873fc0 --- CHANGELOG.md | 6 ++ durabletask/client.py | 18 +++-- durabletask/internal/client_helpers.py | 26 +++++--- tests/durabletask/test_purge_timeout.py | 88 +++++++++++++++++++++++++ 4 files changed, 123 insertions(+), 15 deletions(-) create mode 100644 tests/durabletask/test_purge_timeout.py diff --git a/CHANGELOG.md b/CHANGELOG.md index d881ef7e..ce8aee6b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,12 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ## Unreleased +ADDED + +- Added an optional timeout to filtered orchestration purges. Callers can now +limit a purge operation's duration and inspect `PurgeInstancesResult.is_complete` +to determine whether it finished. + ## v1.9.0 ADDED diff --git a/durabletask/client.py b/durabletask/client.py index a5974749..f6f9ac85 100644 --- a/durabletask/client.py +++ b/durabletask/client.py @@ -8,7 +8,7 @@ import uuid from collections.abc import AsyncIterable, Iterable, Sequence from dataclasses import dataclass, field -from datetime import datetime +from datetime import datetime, timedelta from enum import Enum from typing import Any, Generic, Protocol, TypeVar, cast, overload @@ -846,13 +846,15 @@ def purge_orchestrations_by(self, created_time_from: datetime | None = None, created_time_to: datetime | None = None, runtime_status: list[OrchestrationStatus] | None = None, - recursive: bool = False) -> PurgeInstancesResult: + recursive: bool = False, + timeout: timedelta | None = None) -> PurgeInstancesResult: self._logger.info("Purging orchestrations by filter: " f"created_time_from={created_time_from}, " f"created_time_to={created_time_to}, " f"runtime_status={[str(status) for status in runtime_status] if runtime_status else None}, " - f"recursive={recursive}") - req = build_purge_by_filter_req(created_time_from, created_time_to, runtime_status, recursive) + f"recursive={recursive}, " + f"timeout={timeout}") + req = build_purge_by_filter_req(created_time_from, created_time_to, runtime_status, recursive, timeout) resp: pb.PurgeInstancesResponse = self._stub.PurgeInstances(req) return PurgeInstancesResult(resp.deletedInstanceCount, resp.isComplete.value) @@ -1380,13 +1382,15 @@ async def purge_orchestrations_by(self, created_time_from: datetime | None = None, created_time_to: datetime | None = None, runtime_status: list[OrchestrationStatus] | None = None, - recursive: bool = False) -> PurgeInstancesResult: + recursive: bool = False, + timeout: timedelta | None = None) -> PurgeInstancesResult: self._logger.info("Purging orchestrations by filter: " f"created_time_from={created_time_from}, " f"created_time_to={created_time_to}, " f"runtime_status={[str(status) for status in runtime_status] if runtime_status else None}, " - f"recursive={recursive}") - req = build_purge_by_filter_req(created_time_from, created_time_to, runtime_status, recursive) + f"recursive={recursive}, " + f"timeout={timeout}") + req = build_purge_by_filter_req(created_time_from, created_time_to, runtime_status, recursive, timeout) resp: pb.PurgeInstancesResponse = await self._get_stub().PurgeInstances(req) return PurgeInstancesResult(resp.deletedInstanceCount, resp.isComplete.value) diff --git a/durabletask/internal/client_helpers.py b/durabletask/internal/client_helpers.py index fe0c828b..a4f60ffa 100644 --- a/durabletask/internal/client_helpers.py +++ b/durabletask/internal/client_helpers.py @@ -6,10 +6,10 @@ import logging import uuid from collections.abc import Sequence -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from typing import TYPE_CHECKING, Any, TypeVar -from google.protobuf import wrappers_pb2 +from google.protobuf import duration_pb2, wrappers_pb2 import durabletask.internal.helpers as helpers import durabletask.internal.orchestrator_service_pb2 as pb @@ -114,14 +114,24 @@ def build_purge_by_filter_req( created_time_from: datetime | None, created_time_to: datetime | None, runtime_status: list[OrchestrationStatus] | None, - recursive: bool) -> pb.PurgeInstancesRequest: + recursive: bool, + timeout: timedelta | None = None) -> pb.PurgeInstancesRequest: """Build a PurgeInstancesRequest for purging orchestrations by filter.""" + if timeout is not None and timeout <= timedelta(): + raise ValueError("timeout must be greater than zero.") + + purge_filter = pb.PurgeInstanceFilter( + createdTimeFrom=helpers.new_timestamp(created_time_from) if created_time_from else None, + createdTimeTo=helpers.new_timestamp(created_time_to) if created_time_to else None, + runtimeStatus=[status.value for status in runtime_status] if runtime_status else None + ) + if timeout is not None: + timeout_duration = duration_pb2.Duration() + timeout_duration.FromTimedelta(timeout) + purge_filter.timeout.CopyFrom(timeout_duration) + return pb.PurgeInstancesRequest( - purgeInstanceFilter=pb.PurgeInstanceFilter( - createdTimeFrom=helpers.new_timestamp(created_time_from) if created_time_from else None, - createdTimeTo=helpers.new_timestamp(created_time_to) if created_time_to else None, - runtimeStatus=[status.value for status in runtime_status] if runtime_status else None - ), + purgeInstanceFilter=purge_filter, recursive=recursive ) diff --git a/tests/durabletask/test_purge_timeout.py b/tests/durabletask/test_purge_timeout.py new file mode 100644 index 00000000..708309ec --- /dev/null +++ b/tests/durabletask/test_purge_timeout.py @@ -0,0 +1,88 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +from datetime import timedelta +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from google.protobuf import wrappers_pb2 + +import durabletask.internal.orchestrator_service_pb2 as pb +from durabletask.client import AsyncTaskHubGrpcClient, TaskHubGrpcClient + + +def test_sync_filtered_purge_serializes_timeout(): + stub = MagicMock() + stub.PurgeInstances.return_value = pb.PurgeInstancesResponse( + deletedInstanceCount=1, + isComplete=wrappers_pb2.BoolValue(value=False), + ) + + with ( + patch("durabletask.client.shared.get_grpc_channel", return_value=MagicMock()), + patch("durabletask.client.stubs.TaskHubSidecarServiceStub", return_value=stub), + ): + client = TaskHubGrpcClient() + result = client.purge_orchestrations_by(timeout=timedelta(seconds=1, microseconds=500000)) + + request = stub.PurgeInstances.call_args.args[0] + assert request.purgeInstanceFilter.timeout.seconds == 1 + assert request.purgeInstanceFilter.timeout.nanos == 500000000 + assert result.is_complete is False + + +def test_sync_filtered_purge_omits_timeout_when_not_supplied(): + stub = MagicMock() + stub.PurgeInstances.return_value = pb.PurgeInstancesResponse( + isComplete=wrappers_pb2.BoolValue(value=True), + ) + + with ( + patch("durabletask.client.shared.get_grpc_channel", return_value=MagicMock()), + patch("durabletask.client.stubs.TaskHubSidecarServiceStub", return_value=stub), + ): + client = TaskHubGrpcClient() + client.purge_orchestrations_by() + + request = stub.PurgeInstances.call_args.args[0] + assert not request.purgeInstanceFilter.HasField("timeout") + + +@pytest.mark.parametrize("timeout", [timedelta(), timedelta(seconds=-1)]) +def test_sync_filtered_purge_rejects_non_positive_timeout(timeout): + client = TaskHubGrpcClient(channel=MagicMock()) + + with pytest.raises(ValueError, match="timeout must be greater than zero"): + client.purge_orchestrations_by(timeout=timeout) + + +@pytest.mark.asyncio +async def test_async_filtered_purge_serializes_timeout(): + stub = MagicMock() + stub.PurgeInstances = AsyncMock(return_value=pb.PurgeInstancesResponse( + deletedInstanceCount=1, + isComplete=wrappers_pb2.BoolValue(value=False), + )) + channel = MagicMock() + channel.close = AsyncMock() + + with ( + patch("durabletask.client.shared.get_async_grpc_channel", return_value=channel), + patch("durabletask.client.stubs.TaskHubSidecarServiceStub", return_value=stub), + ): + client = AsyncTaskHubGrpcClient() + result = await client.purge_orchestrations_by(timeout=timedelta(milliseconds=250)) + await client.close() + + request = stub.PurgeInstances.call_args.args[0] + assert request.purgeInstanceFilter.timeout.seconds == 0 + assert request.purgeInstanceFilter.timeout.nanos == 250000000 + assert result.is_complete is False + + +@pytest.mark.asyncio +async def test_async_filtered_purge_rejects_non_positive_timeout(): + client = AsyncTaskHubGrpcClient(channel=MagicMock()) + + with pytest.raises(ValueError, match="timeout must be greater than zero"): + await client.purge_orchestrations_by(timeout=timedelta()) From aa8942cfea8242f1288b4f20528fabaa16cc11f5 Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Thu, 13 Aug 2026 10:43:43 -0600 Subject: [PATCH 2/2] Preserve purge completion status Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 2af7eb6a-76b9-4c7e-98d6-1efc7e873fc0 --- CHANGELOG.md | 2 +- azure-functions-durable/CHANGELOG.md | 5 +++++ durabletask-azuremanaged/CHANGELOG.md | 5 +++++ durabletask/client.py | 23 ++++++++++++++++++----- tests/durabletask/test_purge_timeout.py | 14 ++++++++++++++ 5 files changed, 43 insertions(+), 6 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ce8aee6b..7a8fc772 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,7 +11,7 @@ ADDED - Added an optional timeout to filtered orchestration purges. Callers can now limit a purge operation's duration and inspect `PurgeInstancesResult.is_complete` -to determine whether it finished. +for completed, partial, or backend-unknown completion status. ## v1.9.0 diff --git a/azure-functions-durable/CHANGELOG.md b/azure-functions-durable/CHANGELOG.md index f5e4adf3..0742c240 100644 --- a/azure-functions-durable/CHANGELOG.md +++ b/azure-functions-durable/CHANGELOG.md @@ -7,6 +7,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## Unreleased +ADDED + +- Added an optional timeout to filtered orchestration purges through +`DurableFunctionsClient` and `SyncDurableFunctionsClient`. + ## v2.0.0b2 ADDED diff --git a/durabletask-azuremanaged/CHANGELOG.md b/durabletask-azuremanaged/CHANGELOG.md index dde10137..014c307a 100644 --- a/durabletask-azuremanaged/CHANGELOG.md +++ b/durabletask-azuremanaged/CHANGELOG.md @@ -7,6 +7,11 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ## Unreleased +ADDED + +- Added an optional timeout to filtered orchestration purges through +`DurableTaskSchedulerClient` and `AsyncDurableTaskSchedulerClient`. + ## v1.9.0 CHANGED diff --git a/durabletask/client.py b/durabletask/client.py index f6f9ac85..f8e05bab 100644 --- a/durabletask/client.py +++ b/durabletask/client.py @@ -210,8 +210,15 @@ class EntityQuery: @dataclass class PurgeInstancesResult: + """The outcome of a purge operation. + + ``is_complete`` is ``None`` when the backend does not report whether the + purge completed, ``False`` when the operation stopped before completion, + and ``True`` when it completed. + """ + deleted_instance_count: int - is_complete: bool + is_complete: bool | None @dataclass @@ -272,6 +279,12 @@ def parse_orchestration_state( data_converter if data_converter is not None else DEFAULT_DATA_CONVERTER) +def new_purge_instances_result(response: pb.PurgeInstancesResponse) -> PurgeInstancesResult: + """Build a purge result while preserving the completion field's presence.""" + is_complete = response.isComplete.value if response.HasField("isComplete") else None + return PurgeInstancesResult(response.deletedInstanceCount, is_complete) + + # Grace period before a retired SDK-owned channel is force-closed. Long enough # for in-flight unary RPCs to drain on their own, short enough that recreate # storms don't pile up dozens of half-closed channels. @@ -840,7 +853,7 @@ def purge_orchestration(self, instance_id: str, recursive: bool = True) -> Purge req = pb.PurgeInstancesRequest(instanceId=instance_id, recursive=recursive) self._logger.info(f"Purging instance '{instance_id}'.") resp: pb.PurgeInstancesResponse = self._stub.PurgeInstances(req) - return PurgeInstancesResult(resp.deletedInstanceCount, resp.isComplete.value) + return new_purge_instances_result(resp) def purge_orchestrations_by(self, created_time_from: datetime | None = None, @@ -856,7 +869,7 @@ def purge_orchestrations_by(self, f"timeout={timeout}") req = build_purge_by_filter_req(created_time_from, created_time_to, runtime_status, recursive, timeout) resp: pb.PurgeInstancesResponse = self._stub.PurgeInstances(req) - return PurgeInstancesResult(resp.deletedInstanceCount, resp.isComplete.value) + return new_purge_instances_result(resp) def signal_entity(self, entity_instance_id: EntityInstanceId, @@ -1376,7 +1389,7 @@ async def purge_orchestration(self, instance_id: str, recursive: bool = True) -> req = pb.PurgeInstancesRequest(instanceId=instance_id, recursive=recursive) self._logger.info(f"Purging instance '{instance_id}'.") resp: pb.PurgeInstancesResponse = await self._get_stub().PurgeInstances(req) - return PurgeInstancesResult(resp.deletedInstanceCount, resp.isComplete.value) + return new_purge_instances_result(resp) async def purge_orchestrations_by(self, created_time_from: datetime | None = None, @@ -1392,7 +1405,7 @@ async def purge_orchestrations_by(self, f"timeout={timeout}") req = build_purge_by_filter_req(created_time_from, created_time_to, runtime_status, recursive, timeout) resp: pb.PurgeInstancesResponse = await self._get_stub().PurgeInstances(req) - return PurgeInstancesResult(resp.deletedInstanceCount, resp.isComplete.value) + return new_purge_instances_result(resp) async def signal_entity(self, entity_instance_id: EntityInstanceId, diff --git a/tests/durabletask/test_purge_timeout.py b/tests/durabletask/test_purge_timeout.py index 708309ec..3389fb9d 100644 --- a/tests/durabletask/test_purge_timeout.py +++ b/tests/durabletask/test_purge_timeout.py @@ -48,6 +48,20 @@ def test_sync_filtered_purge_omits_timeout_when_not_supplied(): assert not request.purgeInstanceFilter.HasField("timeout") +def test_sync_filtered_purge_preserves_unknown_completion_state(): + stub = MagicMock() + stub.PurgeInstances.return_value = pb.PurgeInstancesResponse() + + with ( + patch("durabletask.client.shared.get_grpc_channel", return_value=MagicMock()), + patch("durabletask.client.stubs.TaskHubSidecarServiceStub", return_value=stub), + ): + client = TaskHubGrpcClient() + result = client.purge_orchestrations_by() + + assert result.is_complete is None + + @pytest.mark.parametrize("timeout", [timedelta(), timedelta(seconds=-1)]) def test_sync_filtered_purge_rejects_non_positive_timeout(timeout): client = TaskHubGrpcClient(channel=MagicMock())