Rewind API

azure/durable_functions/models/DurableOrchestrationClient.py

@@ -546,3 +546,57 @@ def _get_raise_event_url(
             request_url += "?" + "&".join(query)
 
         return request_url
+
+    async def rewind(self,
+                     instance_id: str,
+                     reason: str,
+                     task_hub_name: Optional[str] = None,
+                     connection_name: Optional[str] = None):
+        """Return / "rewind" a failed orchestration instance to a prior "healthy" state.
+
+        Parameters
+        ----------
+        instance_id: str
+            The ID of the orchestration instance to rewind.
+        reason: str
+            The reason for rewinding the orchestration instance.
+        task_hub_name: Optional[str]
+            The TaskHub of the orchestration to rewind
+        connection_name: Optional[str]
+            Name of the application setting containing the storage
+            connection string to use.
+
+        Raises
+        ------
+        Exception:
+            In case of a failure, it reports the reason for the exception
+        """
+        request_url: str = ""
+        if self._orchestration_bindings.rpc_base_url:
+            path = f"instances/{instance_id}/rewind?reason={reason}"
+            query: List[str] = []
+            if not (task_hub_name is None):
+                query.append(f"taskHub={task_hub_name}")
+            if not (connection_name is None):
+                query.append(f"connection={connection_name}")
+            if len(query) > 0:
+                path += "&" + "&".join(query)
+
+            request_url = f"{self._orchestration_bindings.rpc_base_url}" + path
+        else:
+            raise Exception("The Python SDK only supports RPC endpoints."
+                            + "Please remove the `localRpcEnabled` setting from host.json")
+
+        response = await self._post_async_request(request_url, None)
+        status: int = response[0]
+        if status == 200 or status == 202:
+            return
+        elif status == 404:
+            ex_msg = f"No instance with ID {instance_id} found."
+            raise Exception(ex_msg)
+        elif status == 410:
+            ex_msg = "The rewind operation is only supported on failed orchestration instances."
+            raise Exception(ex_msg)
+        else:
+            ex_msg = response[1]
+            raise Exception(ex_msg)

tests/models/test_DurableOrchestrationClient.py

@@ -19,6 +19,9 @@
 MESSAGE_500 = 'instance failed with unhandled exception'
 MESSAGE_501 = "well we didn't expect that"
 
+INSTANCE_ID = "2e2568e7-a906-43bd-8364-c81733c5891e"
+REASON = "Stuff"
+
 TEST_ORCHESTRATOR = "MyDurableOrchestrator"
 EXCEPTION_ORCHESTRATOR_NOT_FOUND_EXMESSAGE = "The function <orchestrator> doesn't exist,"\
     " is disabled, or is not an orchestrator function. Additional info: "\
@@ -540,3 +543,52 @@ async def test_start_new_orchestrator_internal_exception(binding_string):
     with pytest.raises(Exception) as ex:
         await client.start_new(TEST_ORCHESTRATOR)
     ex.match(status_str)
+
+@pytest.mark.asyncio
+async def test_rewind_works_under_200_and_200_http_codes(binding_string):
+    """Tests that the rewind API works as expected under 'successful' http codes: 200, 202"""
+    client = DurableOrchestrationClient(binding_string)
+    for code in [200, 202]:
+        mock_request = MockRequest(
+            expected_url=f"{RPC_BASE_URL}instances/{INSTANCE_ID}/rewind?reason={REASON}",
+            response=[code, ""])
+        client._post_async_request = mock_request.post
+        result = await client.rewind(INSTANCE_ID, REASON)
+        assert result is None
+
+@pytest.mark.asyncio
+async def test_rewind_throws_exception_during_404_410_and_500_errors(binding_string):
+    """Tests the behaviour of rewind under 'exception' http codes: 404, 410, 500"""
+    client = DurableOrchestrationClient(binding_string)
+    codes = [404, 410, 500]
+    exception_strs = [
+        f"No instance with ID {INSTANCE_ID} found.",
+        "The rewind operation is only supported on failed orchestration instances.",
+        "Something went wrong"
+    ]
+    for http_code, expected_exception_str in zip(codes, exception_strs):
+        mock_request = MockRequest(
+            expected_url=f"{RPC_BASE_URL}instances/{INSTANCE_ID}/rewind?reason={REASON}",
+            response=[http_code, "Something went wrong"])
+        client._post_async_request = mock_request.post
+
+        with pytest.raises(Exception) as ex:
+            await client.rewind(INSTANCE_ID, REASON)
+        ex_message = str(ex.value)
+        assert ex_message == expected_exception_str
+
+@pytest.mark.asyncio
+async def test_rewind_with_no_rpc_endpoint(binding_string):
+    """Tests the behaviour of rewind without an RPC endpoint / under the legacy HTTP endpoint."""
+    client = DurableOrchestrationClient(binding_string)
+    mock_request = MockRequest(
+        expected_url=f"{RPC_BASE_URL}instances/{INSTANCE_ID}/rewind?reason={REASON}",
+        response=[-1, ""])
+    client._post_async_request = mock_request.post  
+    client._orchestration_bindings._rpc_base_url = None
+    expected_exception_str = "The Python SDK only supports RPC endpoints."\
+        + "Please remove the `localRpcEnabled` setting from host.json"
+    with pytest.raises(Exception) as ex:
+        await client.rewind(INSTANCE_ID, REASON)
+    ex_message = str(ex.value)
+    assert ex_message == expected_exception_str