Initial implementation

Author: andystaples

durabletask/client.py

@@ -299,6 +299,26 @@ def resume_orchestration(self, instance_id: str):
         self._logger.info(f"Resuming instance '{instance_id}'.")
         self._stub.ResumeInstance(req)
 
+    def rewind_orchestration(self, instance_id: str, *,
+                             reason: Optional[str] = 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.

durabletask/internal/PROTO_SOURCE_COMMIT_HASH

@@ -1,3 +1,4 @@
 443b333f4f65a438dc9eb4f090560d232afec4b7
 fd9369c6a03d6af4e95285e432b7c4e943c06970
-026329c53fe6363985655857b9ca848ec7238bd2
+026329c53fe6363985655857b9ca848ec7238bd2
+57930bf659bb4c90cfeae44eaf465e000a67ecf1 // DO NOT MERGE - FEATURE BRANCH

durabletask/internal/orchestrator_service_pb2.py

@@ -593,8 +593,14 @@ class SendEntityMessageAction(_message.Message):
     entityUnlockSent: EntityUnlockSentEvent
     def __init__(self, entityOperationSignaled: _Optional[_Union[EntityOperationSignaledEvent, _Mapping]] = ..., entityOperationCalled: _Optional[_Union[EntityOperationCalledEvent, _Mapping]] = ..., entityLockRequested: _Optional[_Union[EntityLockRequestedEvent, _Mapping]] = ..., entityUnlockSent: _Optional[_Union[EntityUnlockSentEvent, _Mapping]] = ...) -> None: ...
 
+class RewindOrchestrationAction(_message.Message):
+    __slots__ = ("newHistory",)
+    NEWHISTORY_FIELD_NUMBER: _ClassVar[int]
+    newHistory: _containers.RepeatedCompositeFieldContainer[HistoryEvent]
+    def __init__(self, newHistory: _Optional[_Iterable[_Union[HistoryEvent, _Mapping]]] = ...) -> None: ...
+
 class OrchestratorAction(_message.Message):
-    __slots__ = ("id", "scheduleTask", "createSubOrchestration", "createTimer", "sendEvent", "completeOrchestration", "terminateOrchestration", "sendEntityMessage")
+    __slots__ = ("id", "scheduleTask", "createSubOrchestration", "createTimer", "sendEvent", "completeOrchestration", "terminateOrchestration", "sendEntityMessage", "rewindOrchestration")
     ID_FIELD_NUMBER: _ClassVar[int]
     SCHEDULETASK_FIELD_NUMBER: _ClassVar[int]
     CREATESUBORCHESTRATION_FIELD_NUMBER: _ClassVar[int]
@@ -603,6 +609,7 @@ class OrchestratorAction(_message.Message):
     COMPLETEORCHESTRATION_FIELD_NUMBER: _ClassVar[int]
     TERMINATEORCHESTRATION_FIELD_NUMBER: _ClassVar[int]
     SENDENTITYMESSAGE_FIELD_NUMBER: _ClassVar[int]
+    REWINDORCHESTRATION_FIELD_NUMBER: _ClassVar[int]
     id: int
     scheduleTask: ScheduleTaskAction
     createSubOrchestration: CreateSubOrchestrationAction
@@ -611,7 +618,8 @@ class OrchestratorAction(_message.Message):
     completeOrchestration: CompleteOrchestrationAction
     terminateOrchestration: TerminateOrchestrationAction
     sendEntityMessage: SendEntityMessageAction
-    def __init__(self, id: _Optional[int] = ..., scheduleTask: _Optional[_Union[ScheduleTaskAction, _Mapping]] = ..., createSubOrchestration: _Optional[_Union[CreateSubOrchestrationAction, _Mapping]] = ..., createTimer: _Optional[_Union[CreateTimerAction, _Mapping]] = ..., sendEvent: _Optional[_Union[SendEventAction, _Mapping]] = ..., completeOrchestration: _Optional[_Union[CompleteOrchestrationAction, _Mapping]] = ..., terminateOrchestration: _Optional[_Union[TerminateOrchestrationAction, _Mapping]] = ..., sendEntityMessage: _Optional[_Union[SendEntityMessageAction, _Mapping]] = ...) -> None: ...
+    rewindOrchestration: RewindOrchestrationAction
+    def __init__(self, id: _Optional[int] = ..., scheduleTask: _Optional[_Union[ScheduleTaskAction, _Mapping]] = ..., createSubOrchestration: _Optional[_Union[CreateSubOrchestrationAction, _Mapping]] = ..., createTimer: _Optional[_Union[CreateTimerAction, _Mapping]] = ..., sendEvent: _Optional[_Union[SendEventAction, _Mapping]] = ..., completeOrchestration: _Optional[_Union[CompleteOrchestrationAction, _Mapping]] = ..., terminateOrchestration: _Optional[_Union[TerminateOrchestrationAction, _Mapping]] = ..., sendEntityMessage: _Optional[_Union[SendEntityMessageAction, _Mapping]] = ..., rewindOrchestration: _Optional[_Union[RewindOrchestrationAction, _Mapping]] = ...) -> None: ...
 
 class OrchestrationTraceContext(_message.Message):
     __slots__ = ("spanID", "spanStartTime")
@@ -1250,16 +1258,50 @@ class SkipGracefulOrchestrationTerminationsResponse(_message.Message):
     def __init__(self, unterminatedInstanceIds: _Optional[_Iterable[str]] = ...) -> None: ...
 
 class GetWorkItemsRequest(_message.Message):
-    __slots__ = ("maxConcurrentOrchestrationWorkItems", "maxConcurrentActivityWorkItems", "maxConcurrentEntityWorkItems", "capabilities")
+    __slots__ = ("maxConcurrentOrchestrationWorkItems", "maxConcurrentActivityWorkItems", "maxConcurrentEntityWorkItems", "capabilities", "workItemFilters")
     MAXCONCURRENTORCHESTRATIONWORKITEMS_FIELD_NUMBER: _ClassVar[int]
     MAXCONCURRENTACTIVITYWORKITEMS_FIELD_NUMBER: _ClassVar[int]
     MAXCONCURRENTENTITYWORKITEMS_FIELD_NUMBER: _ClassVar[int]
     CAPABILITIES_FIELD_NUMBER: _ClassVar[int]
+    WORKITEMFILTERS_FIELD_NUMBER: _ClassVar[int]
     maxConcurrentOrchestrationWorkItems: int
     maxConcurrentActivityWorkItems: int
     maxConcurrentEntityWorkItems: int
     capabilities: _containers.RepeatedScalarFieldContainer[WorkerCapability]
-    def __init__(self, maxConcurrentOrchestrationWorkItems: _Optional[int] = ..., maxConcurrentActivityWorkItems: _Optional[int] = ..., maxConcurrentEntityWorkItems: _Optional[int] = ..., capabilities: _Optional[_Iterable[_Union[WorkerCapability, str]]] = ...) -> None: ...
+    workItemFilters: WorkItemFilters
+    def __init__(self, maxConcurrentOrchestrationWorkItems: _Optional[int] = ..., maxConcurrentActivityWorkItems: _Optional[int] = ..., maxConcurrentEntityWorkItems: _Optional[int] = ..., capabilities: _Optional[_Iterable[_Union[WorkerCapability, str]]] = ..., workItemFilters: _Optional[_Union[WorkItemFilters, _Mapping]] = ...) -> None: ...
+
+class WorkItemFilters(_message.Message):
+    __slots__ = ("orchestrations", "activities", "entities")
+    ORCHESTRATIONS_FIELD_NUMBER: _ClassVar[int]
+    ACTIVITIES_FIELD_NUMBER: _ClassVar[int]
+    ENTITIES_FIELD_NUMBER: _ClassVar[int]
+    orchestrations: _containers.RepeatedCompositeFieldContainer[OrchestrationFilter]
+    activities: _containers.RepeatedCompositeFieldContainer[ActivityFilter]
+    entities: _containers.RepeatedCompositeFieldContainer[EntityFilter]
+    def __init__(self, orchestrations: _Optional[_Iterable[_Union[OrchestrationFilter, _Mapping]]] = ..., activities: _Optional[_Iterable[_Union[ActivityFilter, _Mapping]]] = ..., entities: _Optional[_Iterable[_Union[EntityFilter, _Mapping]]] = ...) -> None: ...
+
+class OrchestrationFilter(_message.Message):
+    __slots__ = ("name", "versions")
+    NAME_FIELD_NUMBER: _ClassVar[int]
+    VERSIONS_FIELD_NUMBER: _ClassVar[int]
+    name: str
+    versions: _containers.RepeatedScalarFieldContainer[str]
+    def __init__(self, name: _Optional[str] = ..., versions: _Optional[_Iterable[str]] = ...) -> None: ...
+
+class ActivityFilter(_message.Message):
+    __slots__ = ("name", "versions")
+    NAME_FIELD_NUMBER: _ClassVar[int]
+    VERSIONS_FIELD_NUMBER: _ClassVar[int]
+    name: str
+    versions: _containers.RepeatedScalarFieldContainer[str]
+    def __init__(self, name: _Optional[str] = ..., versions: _Optional[_Iterable[str]] = ...) -> None: ...
+
+class EntityFilter(_message.Message):
+    __slots__ = ("name",)
+    NAME_FIELD_NUMBER: _ClassVar[int]
+    name: str
+    def __init__(self, name: _Optional[str] = ...) -> None: ...
 
 class WorkItem(_message.Message):
     __slots__ = ("orchestratorRequest", "activityRequest", "entityRequest", "healthPing", "entityRequestV2", "completionToken")

durabletask/internal/orchestrator_service_pb2.pyi

@@ -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
 

durabletask/testing/README.md

@@ -981,8 +981,33 @@ def DeleteTaskHub(self, request: pb.DeleteTaskHubRequest, context):
         return pb.DeleteTaskHubResponse()
 
     def RewindInstance(self, request: pb.RewindInstanceRequest, context):
-        """Rewinds an orchestration instance (not implemented)."""
-        context.abort(grpc.StatusCode.UNIMPLEMENTED, "RewindInstance not implemented")
+        """Rewinds a failed orchestration instance.
+
+        The backend validates the instance is in a failed state, appends
+        an ``ExecutionRewoundEvent`` to the pending events, resets the
+        instance status to RUNNING, and re-enqueues the orchestration
+        so the worker can replay it and produce a
+        ``RewindOrchestrationAction`` with the corrected history.
+        """
+        with self._lock:
+            instance = self._instances.get(request.instanceId)
+            if not instance:
+                context.abort(
+                    grpc.StatusCode.NOT_FOUND,
+                    f"Orchestration instance '{request.instanceId}' not found")
+                return pb.RewindInstanceResponse()
+
+            if instance.status != pb.ORCHESTRATION_STATUS_FAILED:
+                context.abort(
+                    grpc.StatusCode.FAILED_PRECONDITION,
+                    f"Orchestration instance '{request.instanceId}' is not in a failed state")
+                return pb.RewindInstanceResponse()
+
+            reason = request.reason.value if request.HasField("reason") else None
+            self._prepare_rewind(instance, reason)
+
+        self._logger.info(f"Rewound instance '{request.instanceId}'")
+        return pb.RewindInstanceResponse()
 
     def AbandonTaskActivityWorkItem(self, request: pb.AbandonActivityTaskRequest, context):
         """Abandons an activity work item."""
@@ -1196,6 +1221,8 @@ def _process_action(self, instance: OrchestrationInstance, action: pb.Orchestrat
             self._process_send_event_action(action.sendEvent)
         elif action.HasField("sendEntityMessage"):
             self._process_send_entity_message_action(instance, action)
+        elif action.HasField("rewindOrchestration"):
+            self._process_rewind_orchestration_action(instance, action.rewindOrchestration)
 
     def _process_complete_orchestration_action(self, instance: OrchestrationInstance,
                                                complete_action: pb.CompleteOrchestrationAction):
@@ -1205,6 +1232,14 @@ def _process_complete_orchestration_action(self, instance: OrchestrationInstance
         instance.output = complete_action.result.value if complete_action.result else None
         instance.failure_details = complete_action.failureDetails if complete_action.failureDetails else None
 
+        # Append orchestratorCompleted to history when the orchestration
+        # reaches a terminal state.  This positional marker allows the
+        # SDK to distinguish a post-rewind replay from a new rewind
+        # request by comparing the position of the last
+        # orchestratorCompleted against the last executionRewound.
+        if status != pb.ORCHESTRATION_STATUS_CONTINUED_AS_NEW:
+            instance.history.append(helpers.new_orchestrator_completed_event())
+
         if status == pb.ORCHESTRATION_STATUS_CONTINUED_AS_NEW:
             # Handle continue-as-new
             new_input = complete_action.result.value if complete_action.result else None
@@ -1558,6 +1593,119 @@ def _signal_entity_internal(self, entity_id: str, operation: str,
         )
         self._queue_entity_operation(entity_id, event)
 
+    def _prepare_rewind(self, instance: OrchestrationInstance,
+                        reason: Optional[str] = None):
+        """Prepares an orchestration instance for rewind.
+
+        Appends an ``ExecutionRewoundEvent`` to the pending events, resets
+        the instance status to RUNNING, and re-enqueues it so the worker
+        can replay it. The actual history rewriting is done by the SDK
+        worker when it processes the rewind event.
+
+        Args:
+            instance: The orchestration instance to rewind.
+            reason: Optional reason string for the rewind.
+
+        Note:
+            Must be called while holding ``self._lock``.
+        """
+        # Reset instance state so it can be re-processed.
+        instance.status = pb.ORCHESTRATION_STATUS_RUNNING
+        instance.output = None
+        instance.failure_details = None
+        instance.last_updated_at = datetime.now(timezone.utc)
+
+        # Clear any stale dispatched events.
+        instance.dispatched_events.clear()
+
+        # Add the ExecutionRewound event as a new pending event.
+        rewind_event = pb.HistoryEvent(
+            eventId=-1,
+            timestamp=timestamp_pb2.Timestamp(),
+            executionRewound=pb.ExecutionRewoundEvent(
+                reason=wrappers_pb2.StringValue(value=reason) if reason else None,
+            ),
+        )
+        instance.pending_events.append(rewind_event)
+
+        # Refresh the completion token and enqueue.
+        instance.completion_token = self._next_completion_token
+        self._next_completion_token += 1
+        self._orchestration_in_flight.discard(instance.instance_id)
+        self._enqueue_orchestration(instance.instance_id)
+
+    def _process_rewind_orchestration_action(
+            self, instance: OrchestrationInstance,
+            rewind_action: pb.RewindOrchestrationAction):
+        """Processes a RewindOrchestrationAction returned by the SDK.
+
+        The action contains a ``newHistory`` field with the rewritten
+        history computed by the SDK (failed tasks and sub-orchestration
+        failures removed).  The backend replaces the instance's history
+        with this new history, recursively rewinds any failed
+        sub-orchestrations, and re-enqueues the orchestration.
+        """
+        new_history = list(rewind_action.newHistory)
+
+        # Replace history with the rewritten version.
+        instance.history = new_history
+        instance.status = pb.ORCHESTRATION_STATUS_RUNNING
+        instance.output = None
+        instance.failure_details = None
+        instance.last_updated_at = datetime.now(timezone.utc)
+
+        # Identify sub-orchestrations that were created but did not
+        # complete successfully — they need to be recursively rewound.
+        completed_sub_orch_task_ids: set[int] = set()
+        created_sub_orchs: dict[int, str] = {}
+        for event in new_history:
+            if event.HasField("subOrchestrationInstanceCreated"):
+                created_sub_orchs[event.eventId] = (
+                    event.subOrchestrationInstanceCreated.instanceId)
+            elif event.HasField("subOrchestrationInstanceCompleted"):
+                completed_sub_orch_task_ids.add(
+                    event.subOrchestrationInstanceCompleted.taskScheduledId)
+
+        # Extract the rewind reason from the last ExecutionRewound event.
+        reason: Optional[str] = None
+        for event in reversed(new_history):
+            if event.HasField("executionRewound"):
+                if event.executionRewound.HasField("reason"):
+                    reason = event.executionRewound.reason.value
+                break
+
+        # Recursively rewind failed sub-orchestrations.
+        for task_id, sub_instance_id in created_sub_orchs.items():
+            if task_id not in completed_sub_orch_task_ids:
+                sub_instance = self._instances.get(sub_instance_id)
+                if (sub_instance
+                        and sub_instance.status == pb.ORCHESTRATION_STATUS_FAILED):
+                    self._prepare_rewind(sub_instance, reason)
+                self._watch_sub_orchestration(
+                    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
+        instance.pending_events.clear()
+        instance.dispatched_events.clear()
+        if rewind_event is not None:
+            instance.pending_events.append(rewind_event)
+        instance.completion_token = self._next_completion_token
+        self._next_completion_token += 1
+        self._orchestration_in_flight.discard(instance.instance_id)
+        self._enqueue_orchestration(instance.instance_id)
+
     def _enqueue_entity(self, entity_id: str):
         """Enqueues an entity for processing."""
         if entity_id not in self._entity_queue_set: