Add afd rewind

Author: andystaples

CHANGELOG.md

@@ -18,6 +18,12 @@ ADDED
 
 FIXED
 
+- Fixed orchestrations failing with `OrchestrationStateError: Don't know how to
+  handle event of type 'genericEvent'` after being rewound over the Azure
+  Functions Durable extension. The `genericEvent` history event (an
+  informational marker with no execution semantics) is now ignored during
+  replay, matching the .NET worker, so `rewind_orchestration` completes the
+  replay instead of re-failing.
 - Fixed durabletask scheduled tasks (`durabletask.scheduled`) failing under
   data converters that reconstruct nested custom-object envelopes bottom-up
   (such as the Azure Functions Durable `df` codec). `ScheduleState.from_json`

azure-functions-durable/CHANGELOG.md

@@ -9,6 +9,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- `DurableFunctionsClient.rewind_orchestration(...)` (inherited from durabletask)
+  rewinds a failed orchestration to its last known good state. The deprecated v1
+  `rewind(...)` method now delegates to it instead of raising
+  `NotImplementedError`.
+
 - `DFApp.configure_scheduled_tasks()` opts an app in to durabletask scheduled
   tasks by registering the schedule entity and operation orchestrator. Once
   enabled, schedules are managed from a client via

azure-functions-durable/azure/durable_functions/client.py

@@ -393,18 +393,23 @@ async def wait_for_completion_or_create_check_status_response(
             return self._create_http_response(500, state.serialized_output)
         return self.create_check_status_response(request, instance_id)
 
-    @deprecated(
-        "rewind is not yet supported in durabletask; this shim raises "
-        "NotImplementedError.")
+    @deprecated("rewind is deprecated; use rewind_orchestration instead.")
     async def rewind(
             self,
             instance_id: str,
             reason: str,
             task_hub_name: Optional[str] = None,
             connection_name: Optional[str] = None) -> None:
-        """Not implemented: durabletask has no rewind equivalent yet."""
-        raise NotImplementedError(
-            "rewind is not yet supported by durabletask.")
+        """Deprecated alias for :meth:`rewind_orchestration`.
+
+        Rewinds a failed orchestration instance to its last known good state,
+        removing failed task and sub-orchestration results from the history and
+        replaying from the last successful checkpoint.
+
+        The ``task_hub_name`` and ``connection_name`` arguments have no
+        equivalent in durabletask and are ignored.
+        """
+        await self.rewind_orchestration(instance_id, reason=reason)
 
     @staticmethod
     def _create_http_response(status_code: int, body: Union[str, Any]) -> func.HttpResponse:

durabletask/worker.py

@@ -2917,6 +2917,12 @@ def _cancel_timer() -> None:
             elif event.HasField("executionRewound"):
                 # Informational event added when an orchestration is rewound. No action needed.
                 pass
+            elif event.HasField("genericEvent"):
+                # Informational history event with no execution semantics (for
+                # example, the marker the Durable Functions extension appends
+                # when rewinding an orchestration). Ignored during replay,
+                # matching the .NET worker.
+                pass
             elif event.HasField("eventSent"):
                 # Check if this eventSent corresponds to an entity operation call after being translated to the old
                 # entity protocol by the Durable WebJobs extension. If so, treat this message similarly to

tests/azure-functions-durable/e2e/apps/v1_style/activities.py

@@ -48,3 +48,18 @@ def flaky(payload: dict) -> dict:
     if attempts < threshold:
         raise ValueError(f"flaky failure {attempts}/{threshold}")
     return {"attempts": attempts}
+
+
+@bp.activity_trigger(input_name="token")
+def fail_once(token: str) -> str:
+    """Fail on the first invocation for ``token``, then succeed.
+
+    Used to exercise rewind: the first attempt fails the orchestration, and a
+    rewind replays the failed activity -- which now succeeds because the
+    in-process attempt counter has advanced.
+    """
+    _ATTEMPTS[token] = _ATTEMPTS.get(token, 0) + 1
+    attempts = _ATTEMPTS[token]
+    if attempts == 1:
+        raise ValueError("failing on first attempt (rewind to retry)")
+    return f"succeeded on attempt {attempts}"

tests/azure-functions-durable/e2e/apps/v1_style/client_routes.py

@@ -183,10 +183,7 @@ async def wait_or_check(
 @bp.durable_client_input(client_name="client")
 async def rewind_orchestration(
         req: func.HttpRequest, client: df.DurableFunctionsClient) -> func.HttpResponse:
-    try:
-        await client.rewind(req.route_params["id"], reason="e2e-rewind")
-    except NotImplementedError as exc:
-        return func.HttpResponse(str(exc), status_code=501)
+    await client.rewind(req.route_params["id"], reason="e2e-rewind")
     return func.HttpResponse(status_code=202)
 
 

tests/azure-functions-durable/e2e/apps/v1_style/orchestrators.py

@@ -206,6 +206,14 @@ def sub_orch_fails(context: df.DurableOrchestrationContext):
     return result
 
 
+@bp.orchestration_trigger(context_name="context")
+def rewind_target(context: df.DurableOrchestrationContext):
+    # Calls an activity that fails on its first attempt. The orchestration
+    # fails; a client rewind replays the failed activity, which then succeeds.
+    result = yield context.call_activity("fail_once", context.instance_id)
+    return result
+
+
 @bp.orchestration_trigger(context_name="context")
 def access_histories(context: df.DurableOrchestrationContext):
     # histories is intentionally unsupported and raises NotImplementedError,

tests/azure-functions-durable/e2e/test_v1_client_e2e.py

@@ -104,10 +104,16 @@ def test_wait_for_completion_or_check_status(v1_app):
     assert result.json() == ["Hello Tokyo!", "Hello Seattle!", "Hello London!"]
 
 
-def test_rewind_not_implemented(v1_app):
-    instance_id = v1_app.start_orchestration("activity_chain")
-    v1_app.wait_for_completion(instance_id)
+def test_rewind(v1_app):
+    # The orchestration's activity fails on its first attempt, so it lands in a
+    # Failed state; rewinding replays the failed activity, which now succeeds.
+    instance_id = v1_app.start_orchestration("rewind_target")
+    failed = v1_app.wait_for_completion(instance_id)
+    assert failed["runtimeStatus"] == "Failed"
 
     result = http_request("POST", f"{v1_app.base_url}/api/rewind/{instance_id}")
-    # rewind is a deprecated stub that raises NotImplementedError.
-    assert result.status == 501
+    assert result.status == 202
+
+    status = v1_app.wait_for_status(instance_id, "Completed")
+    assert status["runtimeStatus"] == "Completed"
+    assert status["output"] == "succeeded on attempt 2"

tests/azure-functions-durable/test_client_compat.py

@@ -332,15 +332,17 @@ async def test_wait_for_completion_returns_check_status_on_timeout():
 
 
 # ---------------------------------------------------------------------------
-# rewind (not implemented)
+# rewind
 # ---------------------------------------------------------------------------
 
-async def test_rewind_raises_not_implemented():
+async def test_rewind_delegates_to_rewind_orchestration():
     client = _make_client()
     try:
-        with pytest.warns(DeprecationWarning):
-            with pytest.raises(NotImplementedError):
+        with patch.object(client, "rewind_orchestration",
+                          new=AsyncMock()) as mock:
+            with pytest.warns(DeprecationWarning):
                 await client.rewind("abc", "reason")
+        mock.assert_awaited_once_with("abc", reason="reason")
     finally:
         await client.close()