diff --git a/CHANGELOG.md b/CHANGELOG.md index 572ff4a0..46786d6b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,9 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ADDED +- Added an optional timeout to filtered orchestration purges. Callers can now +limit a purge operation's duration and inspect `PurgeInstancesResult.is_complete` +for completed, partial, or backend-unknown completion status. - Added the optional `new_version` argument to `OrchestrationContext.continue_as_new()` so continued orchestrations can switch to a new version. 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 a5974749..f8e05bab 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 @@ -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,21 +853,23 @@ 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, 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) + return new_purge_instances_result(resp) def signal_entity(self, entity_instance_id: EntityInstanceId, @@ -1374,21 +1389,23 @@ 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, 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) + return new_purge_instances_result(resp) async def signal_entity(self, entity_instance_id: EntityInstanceId, 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..3389fb9d --- /dev/null +++ b/tests/durabletask/test_purge_timeout.py @@ -0,0 +1,102 @@ +# 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") + + +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()) + + 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())