Address PR review feedback; enable DTS rewind e2e tests

- Add async rewind_orchestration parity on AsyncTaskHubGrpcClient

- in-memory backend: stamp ExecutionRewound with current time; avoid duplicate executionRewound on post-rewind replay

- worker: fix rewind comments, guard parentInstance, drop dead sub-orchestration branch

- tests: free-port fixture, clarify docstrings, un-skip DTS rewind e2e (emulator now supports RewindInstance)

- azuremanaged CHANGELOG entry

Author: andystaples

durabletask-azuremanaged/CHANGELOG.md

@@ -7,6 +7,8 @@ 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.

durabletask/client.py

@@ -384,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:
         ...
 
@@ -786,7 +789,7 @@ def resume_orchestration(self, instance_id: str) -> None:
         self._stub.ResumeInstance(req)
 
     def rewind_orchestration(self, instance_id: str, *,
-                             reason: str | None = None):
+                             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
@@ -1290,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/testing/in_memory_backend.py

@@ -613,8 +613,12 @@ def GetWorkItems(self, request: pb.GetWorkItemsRequest, context: grpc.ServicerCo
                         instance.pending_events.clear()
 
                         # Add OrchestratorStarted for re-dispatches so that
-                        # ctx.current_utc_datetime advances correctly
-                        if instance.history:
+                        # ctx.current_utc_datetime advances correctly, unless
+                        # the pending batch already begins with one (e.g. a
+                        # post-rewind replay supplies its own).
+                        if instance.history and not (
+                                instance.dispatched_events
+                                and instance.dispatched_events[0].HasField("orchestratorStarted")):
                             now = datetime.now(timezone.utc)
                             orch_started = helpers.new_orchestrator_started_event(now)
                             instance.dispatched_events.insert(0, orch_started)
@@ -1901,10 +1905,15 @@ def _prepare_rewind(self, instance: OrchestrationInstance,
         # Clear any stale dispatched events.
         instance.dispatched_events.clear()
 
-        # Add the ExecutionRewound event as a new pending event.
+        # Add the ExecutionRewound event as a new pending event. Stamp it
+        # with the current time (rather than the default epoch) so the
+        # audit event carries a meaningful timestamp, consistent with
+        # other backend-generated events.
+        rewind_timestamp = timestamp_pb2.Timestamp()
+        rewind_timestamp.FromDatetime(datetime.now(timezone.utc))
         rewind_event = pb.HistoryEvent(
             eventId=-1,
-            timestamp=timestamp_pb2.Timestamp(),
+            timestamp=rewind_timestamp,
             executionRewound=pb.ExecutionRewoundEvent(
                 reason=wrappers_pb2.StringValue(value=reason) if reason else None,
             ),
@@ -1977,22 +1986,18 @@ def _process_rewind_orchestration_action(
                     instance.instance_id, sub_instance_id, task_id)
 
         # Re-enqueue so the orchestration replays with the clean history.
-        # The executionRewound event is added to pending_events so the
-        # worker can see it in new_events; the worker uses the presence
-        # of executionRewound in old_events (history) to distinguish
-        # this normal post-rewind replay from the initial rewind
-        # short-circuit.  Note: we do NOT add orchestratorStarted here
-        # because the work-item dispatch loop already inserts one when
-        # the instance has non-empty history.
-        rewind_event = None
-        for event in new_history:
-            if event.HasField("executionRewound"):
-                rewind_event = event
-                break
+        # The executionRewound event is already present in the clean
+        # history (it was kept by the SDK's rewind rewrite), so it must
+        # NOT be re-sent as a pending event — doing so would duplicate it
+        # in the instance history once the dispatched events are
+        # committed. A lone orchestratorStarted is enough to make the
+        # instance dispatchable for the post-rewind replay; the worker
+        # replays normally because executionCompleted is no longer in the
+        # history.
         instance.pending_events.clear()
         instance.dispatched_events.clear()
-        if rewind_event is not None:
-            instance.pending_events.append(rewind_event)
+        instance.pending_events.append(
+            helpers.new_orchestrator_started_event(datetime.now(timezone.utc)))
         instance.completion_token = self._next_completion_token
         self._next_completion_token += 1
         self._orchestration_in_flight.discard(instance.instance_id)

durabletask/worker.py

@@ -2142,8 +2142,8 @@ def execute(
 
         # Check for rewind BEFORE replay.  A rewind is indicated by an
         # executionRewound event in new_events.  We look for an
-        # executionCompleted event anywhere in the history (old or new
-        # events) to decide whether to rewind or replay:
+        # executionCompleted event in the committed history (old_events)
+        # to decide whether to rewind or replay:
         # 1. executionCompleted IS present → the orchestration reached a
         #    terminal state (e.g. failed).  This is a *new* rewind that
         #    the worker must short-circuit by building clean history.
@@ -2252,6 +2252,14 @@ def _build_rewind_result(
         event is kept so the backend can identify which sub-orchestration
         instances to recursively rewind.
 
+        Known limitation (shared with the Core/.NET rewind): timer events
+        (``timerCreated`` / ``timerFired``) emitted between retry attempts
+        of an activity scheduled with a ``RetryPolicy`` are not removed.
+        On the next replay the regenerated ``scheduleTask`` action may not
+        line up with those retained timer events, which can surface as a
+        non-determinism mismatch. Rewinding an activity that used a retry
+        policy is therefore not currently supported.
+
         WARNING!!!:
         If any changes are made to how this method modifies the orchestration's history, then corresponding changes *must*
         be made in the backend implementations that rely on this method for executing a rewind.
@@ -2272,14 +2280,12 @@ def _build_rewind_result(
         new_execution_id = uuid.uuid4().hex
 
         # First pass: collect the task-scheduled IDs that correspond to
-        # failed activities / sub-orchestrations so we can remove the
-        # matching taskScheduled events in the second pass.
+        # failed activities so we can remove the matching taskScheduled
+        # events in the second pass.
         failed_task_ids: set[int] = set()
         for event in all_events:
             if event.HasField("taskFailed"):
                 failed_task_ids.add(event.taskFailed.taskScheduledId)
-            elif event.HasField("subOrchestrationInstanceFailed"):
-                failed_task_ids.add(event.subOrchestrationInstanceFailed.taskScheduledId)
 
         # Second pass: build the clean history.
         clean_history: list[pb.HistoryEvent] = []
@@ -2301,10 +2307,15 @@ def _build_rewind_result(
                 event_copy.CopyFrom(event)
                 event_copy.executionStarted.orchestrationInstance.executionId.CopyFrom(
                     ph.get_string_value_or_empty(new_execution_id))
-                if rewind_event.HasField("parentExecutionId"):
-                    if rewind_event.parentExecutionId.value:
-                        event_copy.executionStarted.parentInstance.orchestrationInstance.executionId.CopyFrom(
-                            rewind_event.parentExecutionId)
+                # Only update the parent's execution ID when this is
+                # actually a sub-orchestration (its executionStarted has a
+                # parentInstance). Writing through parentInstance for a
+                # top-level orchestration would materialize an empty one.
+                if (rewind_event.HasField("parentExecutionId")
+                        and rewind_event.parentExecutionId.value
+                        and event_copy.executionStarted.HasField("parentInstance")):
+                    event_copy.executionStarted.parentInstance.orchestrationInstance.executionId.CopyFrom(
+                        rewind_event.parentExecutionId)
                 clean_history.append(event_copy)
                 continue
 

tests/durabletask-azuremanaged/test_dts_rewind_e2e.py

@@ -14,7 +14,6 @@
 #       docker run -i -p 8080:8080 -p 8082:8082 -d mcr.microsoft.com/dts/dts-emulator:latest
 pytestmark = [
     pytest.mark.dts,
-    pytest.mark.skip(reason="Rewind support is not yet available in the public DTS emulator"),
 ]
 
 # Read the environment variables

tests/durabletask/test_build_rewind_result.py

@@ -13,8 +13,9 @@
   ``executionStarted`` copy is updated accordingly.
 * ``taskFailed`` events and their corresponding ``taskScheduled``
   events are removed.
-* ``subOrchestrationInstanceFailed`` events and their corresponding
-  ``taskScheduled`` events are removed.
+* ``subOrchestrationInstanceFailed`` events are removed. Sub-orchestrations
+  are represented by ``subOrchestrationInstanceCreated`` (not
+  ``taskScheduled``), so no ``taskScheduled`` event is dropped for them.
 * ``subOrchestrationInstanceCreated`` events for failed sub-orchestrations
   are kept so the backend can recursively rewind them.
 * ``executionCompleted`` events are removed.
@@ -344,8 +345,8 @@ def test_rewind_preserves_successful_activity():
 
 
 def test_rewind_removes_failed_sub_orch_events():
-    """subOrchestrationInstanceFailed and its corresponding taskScheduled
-    should be removed, but subOrchestrationInstanceCreated is kept."""
+    """subOrchestrationInstanceFailed should be removed, but
+    subOrchestrationInstanceCreated is kept."""
     executor = _make_executor()
 
     old_events = [
@@ -373,14 +374,10 @@ def test_rewind_removes_failed_sub_orch_events():
     assert not any(
         e.HasField("subOrchestrationInstanceFailed") for e in clean
     )
-    # The corresponding taskScheduled (eventId == 1) used by
-    # subOrchestrationInstanceFailed should also be removed, since
-    # _build_rewind_result collects the taskScheduledId from
-    # subOrchestrationInstanceFailed too.
-    # Note: subOrchestrationInstanceCreated uses a separate event with
-    # its own eventId and is NOT removed.
+    # Sub-orchestrations are represented by subOrchestrationInstanceCreated,
+    # not taskScheduled, so there is no taskScheduled event to remove here.
     assert not any(
-        e.HasField("taskScheduled") and e.eventId == 1 for e in clean
+        e.HasField("taskScheduled") for e in clean
     )
     # The subOrchestrationInstanceCreated event should be preserved
     # so the backend can identify which sub-orchestration to rewind.

tests/durabletask/test_rewind_e2e.py

@@ -8,13 +8,16 @@
 from durabletask import client, task, worker
 from durabletask.testing import create_test_backend
 
-HOST = "localhost:50055"
+from tests.durabletask._port_utils import find_free_port
+
+PORT = find_free_port()
+HOST = f"localhost:{PORT}"
 
 
 @pytest.fixture(autouse=True)
 def backend():
     """Create an in-memory backend for testing."""
-    b = create_test_backend(port=50055)
+    b = create_test_backend(port=PORT)
     yield b
     b.stop()
     b.reset()