Feature/1794 workflow list history rerun

durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskClient.java

@@ -13,9 +13,13 @@
 
 package io.dapr.durabletask;
 
+import io.dapr.durabletask.implementation.protobuf.HistoryEvents.HistoryEvent;
+import io.dapr.durabletask.implementation.protobuf.OrchestratorService;
+
 import javax.annotation.Nullable;
 
 import java.time.Duration;
+import java.util.List;
 import java.util.concurrent.TimeoutException;
 
 /**
@@ -318,4 +322,35 @@ public void resumeInstance(String instanceId) {
    * @param reason     the reason for resuming the orchestration instance
    */
   public abstract void resumeInstance(String instanceId, @Nullable String reason);
+
+  /**
+   * Lists workflow instance IDs with optional pagination.
+   *
+   * @param continuationToken the continuation token from a previous call, or null for the first page
+   * @param pageSize          the maximum number of instance IDs to return, or null for no limit
+   * @return the raw list-instance-IDs response from the sidecar
+   */
+  public abstract OrchestratorService.ListInstanceIDsResponse listInstanceIds(
+      @Nullable String continuationToken, @Nullable Integer pageSize);
+
+  /**
+   * Gets the full execution history of a workflow instance.
+   *
+   * @param instanceId the ID of the workflow instance to get history for
+   * @return the list of history events for the workflow instance
+   */
+  public abstract List<HistoryEvent> getInstanceHistory(String instanceId);
+
+  /**
+   * Reruns a workflow from a specific history event, creating a new workflow instance.
+   *
+   * @param sourceInstanceId the ID of the source workflow instance to rerun from
+   * @param eventId          the history event ID to rerun from
+   * @param newInstanceId    the instance ID to use for the new instance, or null for a random ID
+   * @param input            the input applied at the next activity event, used only when overwriteInput is true
+   * @param overwriteInput   true to overwrite the input at the rerun point with input
+   * @return the instance ID of the new workflow instance
+   */
+  public abstract String rerunWorkflowFromEvent(String sourceInstanceId, int eventId,
+      @Nullable String newInstanceId, @Nullable Object input, boolean overwriteInput);
 }

durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcClient.java

@@ -15,6 +15,7 @@
 
 import com.google.protobuf.StringValue;
 import com.google.protobuf.Timestamp;
+import io.dapr.durabletask.implementation.protobuf.HistoryEvents.HistoryEvent;
 import io.dapr.durabletask.implementation.protobuf.Orchestration;
 import io.dapr.durabletask.implementation.protobuf.OrchestratorService;
 import io.dapr.durabletask.implementation.protobuf.TaskHubSidecarServiceGrpc;
@@ -45,6 +46,7 @@
 import java.time.Duration;
 import java.time.Instant;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.UUID;
@@ -448,6 +450,57 @@ public String restartInstance(String instanceId, boolean restartWithNewInstanceI
     }
   }
 
+  @Override
+  public OrchestratorService.ListInstanceIDsResponse listInstanceIds(
+      @Nullable String continuationToken, @Nullable Integer pageSize) {
+    OrchestratorService.ListInstanceIDsRequest.Builder builder =
+        OrchestratorService.ListInstanceIDsRequest.newBuilder();
+    if (continuationToken != null) {
+      builder.setContinuationToken(continuationToken);
+    }
+    if (pageSize != null) {
+      if (pageSize <= 0) {
+        throw new IllegalArgumentException("pageSize must be greater than zero.");
+      }
+      builder.setPageSize(pageSize);
+    }
+    return this.sidecarClient.listInstanceIDs(builder.build());
+  }
+
+  @Override
+  public List<HistoryEvent> getInstanceHistory(String instanceId) {
+    Helpers.throwIfArgumentNull(instanceId, "instanceId");
+    OrchestratorService.GetInstanceHistoryRequest request =
+        OrchestratorService.GetInstanceHistoryRequest.newBuilder()
+            .setInstanceId(instanceId)
+            .build();
+    OrchestratorService.GetInstanceHistoryResponse response = this.sidecarClient.getInstanceHistory(request);
+    return response.getEventsList();
+  }
+
+  @Override
+  public String rerunWorkflowFromEvent(String sourceInstanceId, int eventId,
+      @Nullable String newInstanceId, @Nullable Object input, boolean overwriteInput) {
+    Helpers.throwIfArgumentNull(sourceInstanceId, "sourceInstanceId");
+    OrchestratorService.RerunWorkflowFromEventRequest.Builder builder =
+        OrchestratorService.RerunWorkflowFromEventRequest.newBuilder()
+            .setSourceInstanceID(sourceInstanceId)
+            .setEventID(eventId)
+            .setOverwriteInput(overwriteInput);
+    if (newInstanceId != null) {
+      builder.setNewInstanceID(newInstanceId);
+    }
+    if (overwriteInput) {
+      String serializedInput = this.dataConverter.serialize(input);
+      if (serializedInput != null) {
+        builder.setInput(StringValue.of(serializedInput));
+      }
+    }
+    OrchestratorService.RerunWorkflowFromEventResponse response =
+        this.sidecarClient.rerunWorkflowFromEvent(builder.build());
+    return response.getNewInstanceID();
+  }
+
   private PurgeResult toPurgeResult(OrchestratorService.PurgeInstancesResponse response) {
     return new PurgeResult(response.getDeletedInstanceCount());
   }

durabletask-client/src/test/java/io/dapr/durabletask/DurableTaskClientIT.java

@@ -40,6 +40,9 @@
 import java.util.stream.IntStream;
 import java.util.stream.Stream;
 
+import io.dapr.durabletask.implementation.protobuf.HistoryEvents;
+import io.dapr.durabletask.implementation.protobuf.OrchestratorService;
+
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotEquals;
@@ -1746,6 +1749,88 @@ public void taskExecutionIdTest() {
 
   }
 
+  @Test
+  void getInstanceHistoryReturnsEvents() throws TimeoutException {
+    final String orchestratorName = "historyTest";
+    DurableTaskGrpcWorker worker = this.createWorkerBuilder()
+            .addOrchestrator(orchestratorName, ctx -> ctx.complete(ctx.getInput(String.class)))
+            .buildAndStart();
+
+    DurableTaskClient client = new DurableTaskGrpcClientBuilder().build();
+    try (worker; client) {
+      String instanceId = client.scheduleNewOrchestrationInstance(orchestratorName, "hello");
+      client.waitForInstanceCompletion(instanceId, defaultTimeout, true);
+
+      List<HistoryEvents.HistoryEvent> history = client.getInstanceHistory(instanceId);
+
+      assertFalse(history.isEmpty());
+    }
+  }
+
+  @Test
+  void listInstanceIdsReturnsScheduledInstance() throws TimeoutException {
+    final String orchestratorName = "listTest";
+    DurableTaskGrpcWorker worker = this.createWorkerBuilder()
+            .addOrchestrator(orchestratorName, ctx -> ctx.complete(ctx.getInput(String.class)))
+            .buildAndStart();
+
+    DurableTaskClient client = new DurableTaskGrpcClientBuilder().build();
+    try (worker; client) {
+      String instanceId = client.scheduleNewOrchestrationInstance(orchestratorName, "hello");
+      client.waitForInstanceCompletion(instanceId, defaultTimeout, true);
+
+      List<String> ids = new ArrayList<>();
+      String token = null;
+      do {
+        OrchestratorService.ListInstanceIDsResponse page = client.listInstanceIds(token, 100);
+        ids.addAll(page.getInstanceIdsList());
+        token = page.hasContinuationToken() ? page.getContinuationToken() : null;
+      } while (token != null);
+
+      assertTrue(ids.contains(instanceId));
+    }
+  }
+
+  @Test
+  void rerunWorkflowFromEventCreatesNewInstance() throws TimeoutException {
+    final String orchestratorName = "rerunTest";
+    DurableTaskGrpcWorker worker = this.createWorkerBuilder()
+            .addOrchestrator(orchestratorName, ctx -> ctx.complete(ctx.getInput(String.class)))
+            .buildAndStart();
+
+    DurableTaskClient client = new DurableTaskGrpcClientBuilder().build();
+    try (worker; client) {
+      String instanceId = client.scheduleNewOrchestrationInstance(orchestratorName, "hello");
+      client.waitForInstanceCompletion(instanceId, defaultTimeout, true);
+
+      String newInstanceId = client.rerunWorkflowFromEvent(instanceId, 0, null, null, false);
+
+      assertNotEquals(instanceId, newInstanceId);
+      OrchestrationMetadata instance = client.waitForInstanceCompletion(newInstanceId, defaultTimeout, true);
+      assertEquals(OrchestrationRuntimeStatus.COMPLETED, instance.getRuntimeStatus());
+    }
+  }
+
+  @Test
+  void rerunWorkflowFromEventWithOverwriteNullInputDoesNotThrow() throws TimeoutException {
+    final String orchestratorName = "rerunNullInputTest";
+    DurableTaskGrpcWorker worker = this.createWorkerBuilder()
+            .addOrchestrator(orchestratorName, ctx -> ctx.complete(ctx.getInput(String.class)))
+            .buildAndStart();
+
+    DurableTaskClient client = new DurableTaskGrpcClientBuilder().build();
+    try (worker; client) {
+      String instanceId = client.scheduleNewOrchestrationInstance(orchestratorName, "hello");
+      client.waitForInstanceCompletion(instanceId, defaultTimeout, true);
+
+      String newInstanceId = client.rerunWorkflowFromEvent(instanceId, 0, null, null, true);
+
+      assertNotEquals(instanceId, newInstanceId);
+      OrchestrationMetadata instance = client.waitForInstanceCompletion(newInstanceId, defaultTimeout, true);
+      assertEquals(OrchestrationRuntimeStatus.COMPLETED, instance.getRuntimeStatus());
+    }
+  }
+
 }
 
 

examples/src/main/java/io/dapr/examples/workflows/README.md

@@ -984,4 +984,53 @@ The client log:
 ```text
 Started a new external-event model workflow with instance ID: 23410d96-1afe-4698-9fcd-c01c1e0db255
 workflow instance with ID: 23410d96-1afe-4698-9fcd-c01c1e0db255 completed.
-```
+```
+
+### Workflow Management (List, History, Rerun) Pattern
+
+The `DaprWorkflowClient` can list workflow instance IDs, read a workflow instance's full
+execution history, and rerun a workflow from a specific history event. This example shows
+all three operations.
+
+<!-- STEP
+name: Run Workflow Management Worker
+match_order: none
+expected_stdout_lines:
+  - "Start workflow runtime"
+background: true
+sleep: 20
+timeout_seconds: 45
+-->
+
+Run the worker:
+
+```sh
+dapr run --app-id demoworkflowworker --resources-path ./components/workflows --dapr-grpc-port 50005 -- java -jar target/dapr-java-sdk-examples-exec.jar io.dapr.examples.workflows.management.DemoWorkflowManagementWorker 50005
+```
+
+<!-- END_STEP -->
+
+<!-- STEP
+name: Run Workflow Management Client
+match_order: none
+expected_stdout_lines:
+  - "Started a new workflow with instance ID"
+  - "Workflow completed with result"
+  - "Reran workflow from event"
+  - "Listed"
+timeout_seconds: 60
+-->
+
+In a separate terminal, run the client. It connects to the worker's sidecar on
+gRPC port 50005, so the workflow it schedules runs on the worker:
+
+```sh
+java -jar target/dapr-java-sdk-examples-exec.jar io.dapr.examples.workflows.management.DemoWorkflowManagementClient 50005
+dapr stop --app-id demoworkflowworker
+```
+
+<!-- END_STEP -->
+
+The client output shows the started instance ID, the completed result, the list of history
+events (each with its event ID, type, and timestamp), the new instance ID from the rerun,
+and the count of listed instance IDs.

examples/src/main/java/io/dapr/examples/workflows/management/DemoWorkflowManagementActivity.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2023 The Dapr Authors
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package io.dapr.examples.workflows.management;
+
+import io.dapr.workflows.WorkflowActivity;
+import io.dapr.workflows.WorkflowActivityContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class DemoWorkflowManagementActivity implements WorkflowActivity {
+  @Override
+  public Object run(WorkflowActivityContext ctx) {
+    Logger logger = LoggerFactory.getLogger(DemoWorkflowManagementActivity.class);
+    logger.info("Starting Activity: " + ctx.getName());
+    String message = ctx.getInput(String.class);
+    String newMessage = message.toUpperCase();
+    logger.info("Message Received from input: " + message);
+    return newMessage;
+  }
+}

examples/src/main/java/io/dapr/examples/workflows/management/DemoWorkflowManagementClient.java

@@ -0,0 +1,63 @@
+/*
+ * Copyright 2023 The Dapr Authors
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package io.dapr.examples.workflows.management;
+
+import io.dapr.examples.workflows.utils.PropertyUtils;
+import io.dapr.workflows.client.DaprWorkflowClient;
+import io.dapr.workflows.client.RerunWorkflowFromEventOptions;
+import io.dapr.workflows.client.WorkflowHistoryEvent;
+import io.dapr.workflows.client.WorkflowInstancePage;
+import io.dapr.workflows.client.WorkflowState;
+
+import java.util.List;
+import java.util.concurrent.TimeoutException;
+
+public class DemoWorkflowManagementClient {
+  /**
+   * The main method to start the client.
+   *
+   * @param args Input arguments (unused).
+   */
+  public static void main(String[] args) {
+    try (DaprWorkflowClient client = new DaprWorkflowClient(PropertyUtils.getProperties(args))) {
+      String instanceId = client.scheduleNewWorkflow(DemoWorkflowManagementWorkflow.class);
+      System.out.printf("Started a new workflow with instance ID: %s%n", instanceId);
+
+      WorkflowState state = client.waitForWorkflowCompletion(instanceId, null, true);
+      System.out.printf("Workflow completed with result: %s%n", state.readOutputAs(String.class));
+
+      // Read the full execution history.
+      List<WorkflowHistoryEvent> history = client.getInstanceHistory(instanceId);
+      System.out.printf("History for %s has %d events:%n", instanceId, history.size());
+      for (WorkflowHistoryEvent event : history) {
+        System.out.printf("  eventId=%d type=%s at=%s%n",
+            event.getEventId(), event.getEventType(), event.getTimestamp());
+      }
+
+      // Rerun the workflow from the first history event.
+      int firstEventId = history.get(0).getEventId();
+      String rerunId = client.rerunWorkflowFromEvent(instanceId, firstEventId,
+          new RerunWorkflowFromEventOptions().setInput("Osaka").setOverwriteInput(true));
+      System.out.printf("Reran workflow from event %d as new instance: %s%n", firstEventId, rerunId);
+      client.waitForWorkflowCompletion(rerunId, null, true);
+
+      // List workflow instance IDs (first page).
+      WorkflowInstancePage page = client.listInstanceIds(null, 100);
+      System.out.printf("Listed %d instance ID(s); continuationToken=%s%n",
+          page.getInstanceIds().size(), page.getContinuationToken());
+    } catch (TimeoutException | InterruptedException e) {
+      throw new RuntimeException(e);
+    }
+  }
+}