Adding rewind client API

.github/workflows/build-validation.yml

@@ -156,8 +156,8 @@ jobs:
       - name: Archive test report
         uses: actions/upload-artifact@v4
         with:
-          name: Integration test report
-          path: client/build/reports/tests/endToEndTest
+          name: E2E test report
+          path: endtoendtests/build/reports/tests/endToEndTest
 
   functions-sample-tests:
 
@@ -195,5 +195,5 @@ jobs:
       - name: Archive test report
         uses: actions/upload-artifact@v4
         with:
-          name: Integration test report
-          path: client/build/reports/tests/endToEndTest
+          name: Functions Sample test report
+          path: samples-azure-functions/build/reports/tests/sampleTest

CHANGELOG.md

@@ -1,5 +1,7 @@
 ## Unreleased
 
+* Adding rewind client API ([#253](https://github.com/microsoft/durabletask-java/pull/253))
+
 ## v1.7.0
 * Add descriptive error when orchestration type is not registered ([#261](https://github.com/microsoft/durabletask-java/pull/261))
 * Update all dependencies to latest versions ([#260](https://github.com/microsoft/durabletask-java/pull/260))

azurefunctions/src/main/java/com/microsoft/durabletask/azurefunctions/HttpManagementPayload.java

@@ -13,6 +13,7 @@ public class HttpManagementPayload {
     private final String id;
     private final String purgeHistoryDeleteUri;
     private final String restartPostUri;
+    private final String rewindPostUri;
     private final String sendEventPostUri;
     private final String statusQueryGetUri;
     private final String terminatePostUri;
@@ -33,6 +34,7 @@ public HttpManagementPayload(
         this.id = instanceId;
         this.purgeHistoryDeleteUri = instanceStatusURL + "?" + requiredQueryStringParameters;
         this.restartPostUri = instanceStatusURL + "/restart?" + requiredQueryStringParameters;
+        this.rewindPostUri = instanceStatusURL + "/rewind?reason={text}&" + requiredQueryStringParameters;
         this.sendEventPostUri = instanceStatusURL + "/raiseEvent/{eventName}?" + requiredQueryStringParameters;
         this.statusQueryGetUri = instanceStatusURL + "?" + requiredQueryStringParameters;
         this.terminatePostUri = instanceStatusURL + "/terminate?reason={text}&" + requiredQueryStringParameters;
@@ -94,4 +96,13 @@ public String getRestartPostUri() {
         return restartPostUri;
     }
 
+    /**
+     * Gets the HTTP POST instance rewind endpoint.
+     *
+     * @return The HTTP URL for posting instance rewind commands.
+     */
+    public String getRewindPostUri() {
+        return rewindPostUri;
+    }
+
 }

client/src/main/java/com/microsoft/durabletask/DurableTaskClient.java

@@ -292,6 +292,32 @@ public abstract OrchestrationMetadata waitForInstanceCompletion(
      */
     public abstract String restartInstance(String instanceId, boolean restartWithNewInstanceId);
 
+    /**
+     * Rewinds a failed orchestration instance to the last known good state and replays from there.
+     * <p>
+     * This method can only be used on orchestration instances that are in a <code>Failed</code> state.
+     * When rewound, the orchestration instance will restart from the point of failure as if the failure
+     * never occurred. It rewinds the orchestration by replaying any
+     * Failed Activities and Failed suborchestrations that themselves have Failed Activities
+     *
+     * @param instanceId the ID of the orchestration instance to rewind
+     */
+    public void rewindInstance(String instanceId) {
+        this.rewindInstance(instanceId, null);
+    }
+
+    /**
+     * Rewinds a failed orchestration instance to the last known good state and replays from there.
+     * <p>
+     * This method can only be used on orchestration instances that are in a <code>Failed</code> state.
+     * When rewound, the orchestration instance will restart from the point of failure as if the failure
+     * never occurred.
+     *
+     * @param instanceId the ID of the orchestration instance to rewind
+     * @param reason the reason for rewinding the orchestration instance
+     */
+    public abstract void rewindInstance(String instanceId, @Nullable String reason);
+
     /**
      * Suspends a running orchestration instance.
      * @param instanceId the ID of the orchestration instance to suspend

client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcClient.java

@@ -343,6 +343,17 @@ public PurgeResult purgeInstances(PurgeInstanceCriteria purgeInstanceCriteria) t
         }
     }
 
+    @Override
+    public void rewindInstance(String instanceId, @Nullable String reason) {
+        Helpers.throwIfArgumentNull(instanceId, "instanceId");
+        RewindInstanceRequest.Builder rewindRequestBuilder = RewindInstanceRequest.newBuilder();
+        rewindRequestBuilder.setInstanceId(instanceId);
+        if (reason != null) {
+            rewindRequestBuilder.setReason(StringValue.of(reason));
+        }
+        this.sidecarClient.rewindInstance(rewindRequestBuilder.build());
+    }
+
     @Override
     public void suspendInstance(String instanceId, @Nullable String reason) {
         SuspendRequest.Builder suspendRequestBuilder = SuspendRequest.newBuilder();

client/src/test/java/com/microsoft/durabletask/IntegrationTests.java

@@ -413,6 +413,40 @@ void subOrchestration() throws TimeoutException {
         }
     }
 
+    @Test
+    void subOrchestrationWithFailedActivity() throws TimeoutException {
+        final String parentOrchestratorName = "ParentOrchestration";
+        final String childOrchestratorName = "ChildOrchestration";
+        final String activityName = "FailingActivity";
+        final String failureMessage = "Simulated activity failure in sub-orchestration";
+
+        DurableTaskGrpcWorker worker = this.createWorkerBuilder()
+                .addOrchestrator(parentOrchestratorName, ctx -> {
+                    String result = ctx.callSubOrchestrator(childOrchestratorName, null, String.class).await();
+                    ctx.complete(result);
+                })
+                .addOrchestrator(childOrchestratorName, ctx -> {
+                    String result = ctx.callActivity(activityName, null, String.class).await();
+                    ctx.complete(result);
+                })
+                .addActivity(activityName, ctx -> {
+                    throw new RuntimeException(failureMessage);
+                })
+                .buildAndStart();
+
+        DurableTaskClient client = this.createClientBuilder().build();
+        try (worker; client) {
+            String instanceId = client.scheduleNewOrchestrationInstance(parentOrchestratorName);
+            OrchestrationMetadata instance = client.waitForInstanceCompletion(instanceId, defaultTimeout, true);
+            assertNotNull(instance);
+            assertEquals(OrchestrationRuntimeStatus.FAILED, instance.getRuntimeStatus());
+
+            FailureDetails details = instance.getFailureDetails();
+            assertNotNull(details);
+            assertTrue(details.getErrorMessage().contains(failureMessage));
+        }
+    }
+
     @Test
     void continueAsNew() throws TimeoutException {
         final String orchestratorName = "continueAsNew";
@@ -605,6 +639,107 @@ void terminateSuspendOrchestration() throws TimeoutException, InterruptedExcepti
         }
     }
 
+    private void rewindFailedOrchestrationHelper(String rewindReason) throws TimeoutException {
+        final String orchestratorName = "RewindOrchestration" + (rewindReason == null ? "NoReason" : "");
+        final String activityName = "FailOnceActivity" + (rewindReason == null ? "NoReason" : "");
+        final String expectedOutput = "Success after rewind" + (rewindReason == null ? " without reason" : "");
+        final AtomicBoolean shouldFail = new AtomicBoolean(true);
+
+        DurableTaskGrpcWorker worker = this.createWorkerBuilder()
+                .addOrchestrator(orchestratorName, ctx -> {
+                    String result = ctx.callActivity(activityName, null, String.class).await();
+                    ctx.complete(result);
+                })
+                .addActivity(activityName, ctx -> {
+                    if (shouldFail.compareAndSet(true, false)) {
+                        throw new RuntimeException("Simulated transient failure");
+                    }
+                    return expectedOutput;
+                })
+                .buildAndStart();
+
+        DurableTaskClient client = this.createClientBuilder().build();
+        try (worker; client) {
+            String instanceId = client.scheduleNewOrchestrationInstance(orchestratorName);
+
+            // Wait for the orchestration to fail
+            OrchestrationMetadata instance = client.waitForInstanceCompletion(instanceId, defaultTimeout, false);
+            assertNotNull(instance);
+            assertEquals(OrchestrationRuntimeStatus.FAILED, instance.getRuntimeStatus());
+
+            // Rewind the failed orchestration
+            if (rewindReason != null) {
+                client.rewindInstance(instanceId, rewindReason);
+            } else {
+                client.rewindInstance(instanceId);
+            }
+
+            // Wait for the orchestration to complete after rewind
+            instance = client.waitForInstanceCompletion(instanceId, defaultTimeout, true);
+            assertNotNull(instance);
+            assertEquals(OrchestrationRuntimeStatus.COMPLETED, instance.getRuntimeStatus());
+            assertEquals(expectedOutput, instance.readOutputAs(String.class));
+        }
+    }
+
+    @Test
+    void rewindFailedOrchestration() throws TimeoutException {
+        rewindFailedOrchestrationHelper("Rewinding after transient failure");
+    }
+
+    @Test
+    void rewindFailedOrchestrationWithoutReason() throws TimeoutException {
+        rewindFailedOrchestrationHelper(null);
+    }
+
+    @Test
+    void rewindCompletedOrchestrationThrowsException() throws TimeoutException {
+        final String orchestratorName = "RewindCompletedOrchestration";
+
+        DurableTaskGrpcWorker worker = this.createWorkerBuilder()
+                .addOrchestrator(orchestratorName, ctx -> {
+                    ctx.complete("Completed successfully");
+                })
+                .buildAndStart();
+
+        DurableTaskClient client = this.createClientBuilder().build();
+        try (worker; client) {
+            String instanceId = client.scheduleNewOrchestrationInstance(orchestratorName);
+
+            // Wait for the orchestration to complete
+            OrchestrationMetadata instance = client.waitForInstanceCompletion(instanceId, defaultTimeout, true);
+            assertNotNull(instance);
+            assertEquals(OrchestrationRuntimeStatus.COMPLETED, instance.getRuntimeStatus());
+
+            // Attempt to rewind a completed orchestration - should throw or be a no-op
+            // Based on API behavior, rewind is only valid for FAILED orchestrations
+            assertThrows(
+                Exception.class,
+                () -> client.rewindInstance(instanceId, "Attempting to rewind completed orchestration"),
+                "Rewinding a completed orchestration should throw an exception"
+            );
+        }
+    }
+
+    @Test
+    void rewindNonExistentOrchestrationThrowsException() {
+        final String orchestratorName = "RewindNonExistent";
+        final String nonExistentId = "non-existent-instance-id";
+
+        DurableTaskGrpcWorker worker = this.createWorkerBuilder()
+                .addOrchestrator(orchestratorName, ctx -> ctx.complete("done"))
+                .buildAndStart();
+
+        DurableTaskClient client = this.createClientBuilder().build();
+        try (worker; client) {
+            assertThrows(
+                Exception.class,
+                () -> client.rewindInstance(nonExistentId, "Attempting to rewind non-existent orchestration"),
+                "Rewinding a non-existent orchestration should throw an exception"
+            );
+        }
+    }
+
     @Test
     void activityFanOut() throws IOException, TimeoutException {
         final String orchestratorName = "ActivityFanOut";

endtoendtests/src/main/java/com/functions/RewindTest.java

@@ -0,0 +1,89 @@
+package com.functions;
+
+import com.microsoft.azure.functions.ExecutionContext;
+import com.microsoft.azure.functions.HttpMethod;
+import com.microsoft.azure.functions.HttpRequestMessage;
+import com.microsoft.azure.functions.HttpResponseMessage;
+import com.microsoft.azure.functions.annotation.AuthorizationLevel;
+import com.microsoft.azure.functions.annotation.FunctionName;
+import com.microsoft.azure.functions.annotation.HttpTrigger;
+import com.microsoft.durabletask.DurableTaskClient;
+import com.microsoft.durabletask.TaskOrchestrationContext;
+import com.microsoft.durabletask.azurefunctions.DurableActivityTrigger;
+import com.microsoft.durabletask.azurefunctions.DurableClientContext;
+import com.microsoft.durabletask.azurefunctions.DurableClientInput;
+import com.microsoft.durabletask.azurefunctions.DurableOrchestrationTrigger;
+
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * Sample functions to test the rewind functionality.
+ * Rewind allows a failed orchestration to be replayed from its last known good state.
+ */
+public class RewindTest {
+
+    // Flag to control whether the activity should fail (first call fails, subsequent calls succeed)
+    private static final AtomicBoolean shouldFail = new AtomicBoolean(true);
+
+    /**
+     * HTTP trigger to start the rewindable orchestration.
+     */
+    @FunctionName("StartRewindableOrchestration")
+    public HttpResponseMessage startRewindableOrchestration(
+            @HttpTrigger(name = "req", methods = {HttpMethod.GET, HttpMethod.POST}, authLevel = AuthorizationLevel.ANONYMOUS) HttpRequestMessage<Optional<String>> request,
+            @DurableClientInput(name = "durableContext") DurableClientContext durableContext,
+            final ExecutionContext context) {
+        context.getLogger().info("Starting rewindable orchestration.");
+
+        // Reset the failure flag so the first activity call will fail
+        shouldFail.set(true);
+
+        DurableTaskClient client = durableContext.getClient();
+        String instanceId = client.scheduleNewOrchestrationInstance("RewindableOrchestration");
+        context.getLogger().info("Created new Java orchestration with instance ID = " + instanceId);
+        return durableContext.createCheckStatusResponse(request, instanceId);
+    }
+
+    /**
+     * Orchestration that calls an activity which will fail on the first attempt.
+     * After rewinding, the orchestration will replay and the activity will succeed.
+     */
+    @FunctionName("RewindableOrchestration")
+    public String rewindableOrchestration(
+            @DurableOrchestrationTrigger(name = "ctx") TaskOrchestrationContext ctx) {
+        // Call the activity that may fail
+        String result = ctx.callActivity("FailOnceActivity", "RewindTest", String.class).await();
+        return result;
+    }
+
+    /**
+     * Activity that fails on the first call but succeeds on subsequent calls.
+     * This simulates a transient failure that can be recovered by rewinding.
+     */
+    @FunctionName("FailOnceActivity")
+    public String failOnceActivity(
+            @DurableActivityTrigger(name = "input") String input,
+            final ExecutionContext context) {
+        if (shouldFail.compareAndSet(true, false)) {
+            context.getLogger().warning("FailOnceActivity: Simulating failure for input: " + input);
+            throw new RuntimeException("Simulated transient failure - rewind to retry");
+        }
+        context.getLogger().info("FailOnceActivity: Success for input: " + input);
+        return input + "-rewound-success";
+    }
+
+    /**
+     * HTTP trigger to reset the failure flag (useful for testing).
+     */
+    @FunctionName("ResetRewindFailureFlag")
+    public HttpResponseMessage resetRewindFailureFlag(
+            @HttpTrigger(name = "req", methods = {HttpMethod.POST}, authLevel = AuthorizationLevel.ANONYMOUS) HttpRequestMessage<Optional<String>> request,
+            final ExecutionContext context) {
+        shouldFail.set(true);
+        context.getLogger().info("Reset failure flag to true.");
+        return request.createResponseBuilder(com.microsoft.azure.functions.HttpStatus.OK)
+                .body("Failure flag reset to true")
+                .build();
+    }
+}