Merge remote-tracking branch 'origin/main' into andystaples/add-functions-support

# Conflicts:
#	CHANGELOG.md

Author: andystaples

CHANGELOG.md

@@ -10,18 +10,12 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 ADDED
 
 - Added a `result` property to `Task` as a convenience alias for `get_result()`.
-- Added `OrchestrationContext.parent_instance_id`, which returns the instance
-  ID of the parent orchestration for a sub-orchestration, or `None` for a
-  top-level orchestration.
+- Added `TaskHubGrpcClient.rewind_orchestration()` to rewind a failed orchestration instance to its last known good state. Failed activity and sub-orchestration results are removed from the history and the orchestration replays from the last successful checkpoint, retrying only the failed work. The in-memory testing backend supports rewind as well.
 - Exported `bind_context` and `clear_context` from
   `durabletask.extensions.history_export` so hosts that register the export
   functions themselves (rather than via `ExportHistoryClient.register_worker`)
   can supply the activities' runtime dependencies.
 
-CHANGED
-
-- Changed the default large-payload externalization threshold (`LargePayloadStorageOptions.threshold_bytes`) from 900,000 bytes to 262,144 bytes (256 KiB), matching the .NET SDK default. Behavioral change (not source/binary breaking): payloads larger than 256 KiB are now externalized by default.
-
 FIXED
 
 - Fixed durabletask scheduled tasks (`durabletask.scheduled`) failing under
@@ -44,6 +38,25 @@ FIXED
 - `OrchestrationContext.create_timer` now accepts timezone-aware `datetime`
   values, normalizing them to UTC instead of raising when compared against the
   orchestration's internal clock.
+
+## v1.7.2
+
+FIXED
+
+- Fixed history export failing for orchestrations that completed with an error. Events carrying failure information (orchestration completion, activity failure, sub-orchestration failure, and entity operation failure) were not JSON-serializable, causing the `durabletask.extensions.history_export` module to raise `TypeError: Object of type FailureDetails is not JSON serializable`. `FailureDetails` now serializes to a JSON object with `message`, `error_type`, and `stack_trace` fields.
+
+## v1.7.1
+
+ADDED
+
+- Added `OrchestrationContext.parent_instance_id`, which returns the instance ID of the parent orchestration for a sub-orchestration, or `None` for a top-level orchestration.
+
+CHANGED
+
+- Changed the default large-payload externalization threshold (`LargePayloadStorageOptions.threshold_bytes`) from 900,000 bytes to 262,144 bytes (256 KiB), matching the .NET SDK default. Behavioral change (not source/binary breaking): payloads larger than 256 KiB are now externalized by default.
+
+FIXED
+
 - Fixed `OrchestrationContext.call_entity` not propagating entity operation failures when running under the legacy entity protocol (used by the Azure Functions Durable extension). A failed entity operation now raises `TaskFailedError` in the calling orchestration instead of silently completing, matching the behavior of the current entity protocol and the .NET SDK.
 - Fixed `OrchestrationContext.call_entity` returning a double-encoded result under the legacy entity protocol. The entity's return value was left as a raw serialized JSON string (for example, a returned string arrived with extra quotes and dicts/lists arrived as strings); it is now fully deserialized and coerced to the requested `return_type`, matching the current entity protocol.
 - Fixed unbounded growth of the internal entity request/lock tracking maps when using entities over the legacy entity protocol. Entries are now released as each response is handled, reducing memory use in long-running orchestrations that call or lock entities.

durabletask-azuremanaged/CHANGELOG.md

@@ -7,6 +7,16 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## Unreleased
 
+- Added `rewind_orchestration()` to `DurableTaskSchedulerClient` and `AsyncDurableTaskSchedulerClient` (inherited from the base clients) to rewind a failed orchestration instance to its last known good state.
+
+## v1.7.2
+
+- Updates base dependency to durabletask v1.7.2.
+
+## v1.7.1
+
+- Updates base dependency to durabletask v1.7.1
+
 ## v1.7.0
 
 - Updates base dependency to durabletask v1.7.0.

durabletask-azuremanaged/pyproject.toml

@@ -9,7 +9,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "durabletask.azuremanaged"
-version = "1.7.0"
+version = "1.7.2"
 description = "Durable Task Python SDK provider implementation for the Azure Durable Task Scheduler"
 keywords = [
     "durable",
@@ -26,13 +26,13 @@ requires-python = ">=3.10"
 license = {file = "LICENSE"}
 readme = "README.md"
 dependencies = [
-    "durabletask>=1.7.0",
+    "durabletask>=1.7.2",
     "azure-identity>=1.19.0"
 ]
 
 [project.optional-dependencies]
 azure-blob-payloads = [
-    "durabletask[azure-blob-payloads]>=1.7.0"
+    "durabletask[azure-blob-payloads]>=1.7.2"
 ]
 
 [project.urls]

durabletask/client.py

@@ -320,6 +320,9 @@ def SuspendInstance(self, request: pb.SuspendRequest) -> pb.SuspendResponse:
     def ResumeInstance(self, request: pb.ResumeRequest) -> pb.ResumeResponse:
         ...
 
+    def RewindInstance(self, request: pb.RewindInstanceRequest) -> pb.RewindInstanceResponse:
+        ...
+
     def RestartInstance(self, request: pb.RestartInstanceRequest) -> pb.RestartInstanceResponse:
         ...
 
@@ -381,6 +384,9 @@ async def SuspendInstance(self, request: pb.SuspendRequest) -> pb.SuspendRespons
     async def ResumeInstance(self, request: pb.ResumeRequest) -> pb.ResumeResponse:
         ...
 
+    async def RewindInstance(self, request: pb.RewindInstanceRequest) -> pb.RewindInstanceResponse:
+        ...
+
     async def RestartInstance(self, request: pb.RestartInstanceRequest) -> pb.RestartInstanceResponse:
         ...
 
@@ -782,6 +788,26 @@ def resume_orchestration(self, instance_id: str) -> None:
         self._logger.info(f"Resuming instance '{instance_id}'.")
         self._stub.ResumeInstance(req)
 
+    def rewind_orchestration(self, instance_id: str, *,
+                             reason: str | None = None) -> None:
+        """Rewinds a failed orchestration instance to its last known good state.
+
+        Rewind removes failed task and sub-orchestration results from the
+        orchestration history and replays the orchestration from the last
+        successful checkpoint.  Activities that previously succeeded are
+        not re-executed; only failed work is retried.
+
+        Args:
+            instance_id: The ID of the orchestration instance to rewind.
+            reason: An optional reason string describing why the orchestration is being rewound.
+        """
+        req = pb.RewindInstanceRequest(
+            instanceId=instance_id,
+            reason=helpers.get_string_value(reason))
+
+        self._logger.info(f"Rewinding instance '{instance_id}'.")
+        self._stub.RewindInstance(req)
+
     def restart_orchestration(self, instance_id: str, *,
                               restart_with_new_instance_id: bool = False) -> str:
         """Restarts an existing orchestration instance.
@@ -1267,6 +1293,26 @@ async def resume_orchestration(self, instance_id: str) -> None:
         self._logger.info(f"Resuming instance '{instance_id}'.")
         await self._stub.ResumeInstance(req)
 
+    async def rewind_orchestration(self, instance_id: str, *,
+                                   reason: str | None = None) -> None:
+        """Rewinds a failed orchestration instance to its last known good state.
+
+        Rewind removes failed task and sub-orchestration results from the
+        orchestration history and replays the orchestration from the last
+        successful checkpoint.  Activities that previously succeeded are
+        not re-executed; only failed work is retried.
+
+        Args:
+            instance_id: The ID of the orchestration instance to rewind.
+            reason: An optional reason string describing why the orchestration is being rewound.
+        """
+        req = pb.RewindInstanceRequest(
+            instanceId=instance_id,
+            reason=helpers.get_string_value(reason))
+
+        self._logger.info(f"Rewinding instance '{instance_id}'.")
+        await self._stub.RewindInstance(req)
+
     async def restart_orchestration(self, instance_id: str, *,
                                     restart_with_new_instance_id: bool = False) -> str:
         """Restarts an existing orchestration instance.

durabletask/internal/helpers.py

@@ -236,6 +236,21 @@ def new_terminated_event(*, encoded_output: str | None = None) -> pb.HistoryEven
     )
 
 
+def new_execution_completed_event(
+        status: pb.OrchestrationStatus,
+        encoded_result: str | None = None,
+        failure_details: pb.TaskFailureDetails | None = None) -> pb.HistoryEvent:
+    return pb.HistoryEvent(
+        eventId=-1,
+        timestamp=timestamp_pb2.Timestamp(),
+        executionCompleted=pb.ExecutionCompletedEvent(
+            orchestrationStatus=status,
+            result=get_string_value(encoded_result),
+            failureDetails=failure_details,
+        )
+    )
+
+
 def get_string_value(val: str | None) -> wrappers_pb2.StringValue | None:
     if val is None:
         return None

durabletask/task.py

@@ -8,6 +8,7 @@
 import math
 from abc import ABC, abstractmethod
 from collections.abc import Callable, Generator, Sequence
+from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
 from typing import TYPE_CHECKING, Any, Generic, TypeAlias, TypeVar, cast, overload
 
@@ -449,23 +450,11 @@ def isEnabledFor(self, level: int) -> bool:
         return self.logger.isEnabledFor(level)
 
 
+@dataclass(frozen=True)
 class FailureDetails:
-    def __init__(self, message: str, error_type: str, stack_trace: str | None):
-        self._message = message
-        self._error_type = error_type
-        self._stack_trace = stack_trace
-
-    @property
-    def message(self) -> str:
-        return self._message
-
-    @property
-    def error_type(self) -> str:
-        return self._error_type
-
-    @property
-    def stack_trace(self) -> str | None:
-        return self._stack_trace
+    message: str
+    error_type: str
+    stack_trace: str | None
 
 
 class TaskFailedError(Exception):

durabletask/testing/README.md

@@ -255,8 +255,7 @@ The in-memory backend is designed for testing and has some limitations compared
 1. **No persistence**: All state is lost when the backend is stopped
 2. **No distributed execution**: Runs in a single process
 3. **No history streaming**: StreamInstanceHistory is not implemented
-4. **No rewind**: RewindInstance is not implemented
-5. **No recursive termination**: Recursive termination is not supported
+4. **No recursive termination**: Recursive termination is not supported
 
 ### Best Practices