Bulk WF Operations (#325)

cancel/resume/delete workflows in bulk, either via DBOS/DBOSClient API
or Conductor.

fixes #311

Author: devhawk

transact/src/main/java/dev/dbos/transact/DBOS.java

@@ -60,6 +60,12 @@ public class DBOS implements AutoCloseable {
 
   private AlertHandler alertHandler;
 
+  /**
+   * Construct a new DBOS instance with the provided configuration.
+   *
+   * @param config the DBOS configuration; must not be null
+   * @throws NullPointerException if config or required config fields are null
+   */
   public DBOS(@NonNull DBOSConfig config) {
     Objects.requireNonNull(config, "DBOSConfig must not be null");
     Objects.requireNonNull(config.appName(), "DBOSConfig.appName must not be null");
@@ -72,6 +78,10 @@ public DBOS(@NonNull DBOSConfig config) {
     this.config = config;
   }
 
+  /**
+   * Close this DBOS instance and shut down all associated resources. This method delegates to
+   * {@link #shutdown()}.
+   */
   @Override
   public void close() throws Exception {
     shutdown();
@@ -98,6 +108,11 @@ public void close() throws Exception {
     }
   }
 
+  /**
+   * Get the current DBOS version.
+   *
+   * @return the DBOS version string
+   */
   public static String version() {
     return DBOS_VERSION;
   }
@@ -592,9 +607,26 @@ public <E extends Exception> void runStep(
    * @param workflowId id of the workflow
    * @return A handle to the workflow
    */
+  @SuppressWarnings("unchecked")
   public <T, E extends Exception> @NonNull WorkflowHandle<T, E> resumeWorkflow(
       @NonNull String workflowId) {
-    return ensureLaunched("resumeWorkflow").resumeWorkflow(workflowId);
+    var handles = resumeWorkflows(List.of(workflowId));
+    assert (handles.size() == 1);
+    return (WorkflowHandle<T, E>) handles.get(0);
+  }
+
+  /**
+   * Resume multiple workflows starting from the step after the last complete step for each
+   * workflow. This method allows bulk resumption of workflows that were previously interrupted or
+   * failed.
+   *
+   * @param workflowIds a list of workflow IDs to resume; must not be null
+   * @return A list of handles to the resumed workflows
+   * @throws IllegalStateException if called before DBOS is launched
+   */
+  public @NonNull List<WorkflowHandle<Object, Exception>> resumeWorkflows(
+      @NonNull List<String> workflowIds) {
+    return ensureLaunched("resumeWorkflow").resumeWorkflows(workflowIds);
   }
 
   /***
@@ -603,9 +635,69 @@ public <E extends Exception> void runStep(
    * current one) will not execute
    *
    * @param workflowId ID of the workflow to cancel
+   * @throws IllegalStateException if called before DBOS is launched
    */
   public void cancelWorkflow(@NonNull String workflowId) {
-    ensureLaunched("cancelWorkflow").cancelWorkflow(workflowId);
+    cancelWorkflows(List.of(workflowId));
+  }
+
+  /**
+   * Cancels multiple workflows. After this function is called, the next step (not the current one)
+   * of each specified workflow will not execute.
+   *
+   * @param workflowIds a list of workflow IDs to cancel; must not be null
+   * @throws IllegalStateException if called before DBOS is launched
+   */
+  public void cancelWorkflows(@NonNull List<String> workflowIds) {
+    ensureLaunched("cancelWorkflow").cancelWorkflows(workflowIds);
+  }
+
+  /**
+   * Delete a workflow from the system. This permanently removes the workflow and its associated
+   * data from the database. Child workflows are preserved by default.
+   *
+   * @param workflowId ID of the workflow to delete; must not be null
+   * @throws IllegalStateException if called before DBOS is launched
+   */
+  public void deleteWorkflow(@NonNull String workflowId) {
+    deleteWorkflows(List.of(workflowId), false);
+  }
+
+  /**
+   * Delete a workflow from the system. This permanently removes the workflow and its associated
+   * data from the database.
+   *
+   * @param workflowId ID of the workflow to delete; must not be null
+   * @param deleteChildren if true, also delete any child workflows; if false, preserve child
+   *     workflows
+   * @throws IllegalStateException if called before DBOS is launched
+   */
+  public void deleteWorkflow(@NonNull String workflowId, boolean deleteChildren) {
+    deleteWorkflows(List.of(workflowId), deleteChildren);
+  }
+
+  /**
+   * Delete multiple workflows from the system. This permanently removes the workflows and their
+   * associated data from the database. Child workflows are preserved by default.
+   *
+   * @param workflowIds a list of workflow IDs to delete; must not be null
+   * @throws IllegalStateException if called before DBOS is launched
+   */
+  public void deleteWorkflows(@NonNull List<String> workflowIds) {
+    deleteWorkflows(workflowIds, false);
+  }
+
+  /**
+   * Delete multiple workflows from the system. This permanently removes the workflows and their
+   * associated data from the database.
+   *
+   * @param workflowIds a list of workflow IDs to delete; must not be null
+   * @param deleteChildren if true, also delete any child workflows; if false, preserve child
+   *     workflows
+   * @throws IllegalStateException if called before DBOS is launched
+   */
+  public void deleteWorkflows(@NonNull List<String> workflowIds, boolean deleteChildren) {
+    ensureLaunched("deleteWorkflows").deleteWorkflows(workflowIds, deleteChildren);
   }
 
   /**
@@ -639,28 +731,6 @@ public void cancelWorkflow(@NonNull String workflowId) {
     return forkWorkflow(workflowId, startStep, new ForkOptions());
   }
 
-  /**
-   * Deletes a workflow from the system. Does not delete child workflows.
-   *
-   * @param workflowId the unique identifier of the workflow to delete. Must not be null.
-   * @throws IllegalArgumentException if workflowId is null
-   */
-  public void deleteWorkflow(@NonNull String workflowId) {
-    deleteWorkflow(workflowId, false);
-  }
-
-  /**
-   * Deletes a workflow and optionally its child workflows from the system.
-   *
-   * @param workflowId the unique identifier of the workflow to delete. Must not be null.
-   * @param deleteChildren if true, also deletes all child workflows associated with the specified
-   *     workflow; if false, only deletes the specified workflow
-   * @throws IllegalArgumentException if workflowId is null
-   */
-  public void deleteWorkflow(@NonNull String workflowId, boolean deleteChildren) {
-    ensureLaunched("deleteWorkflow").deleteWorkflow(workflowId, deleteChildren);
-  }
-
   /**
    * Retrieve a handle to a workflow, given its ID. Note that a handle is always returned, whether
    * the workflow exists or not; getStatus() can be used to tell the difference

transact/src/main/java/dev/dbos/transact/DBOSClient.java

@@ -660,7 +660,7 @@ public void send(
    */
   public <T, E extends Exception> @NonNull WorkflowHandle<T, E> retrieveWorkflow(
       @NonNull String workflowId) {
-    return new WorkflowHandleClient<T, E>(workflowId);
+    return new WorkflowHandleClient<>(workflowId);
   }
 
   /**
@@ -669,7 +669,17 @@ public void send(
    * @param workflowId ID of the workflow to cancel
    */
   public void cancelWorkflow(@NonNull String workflowId) {
-    systemDatabase.cancelWorkflow(workflowId);
+    systemDatabase.cancelWorkflows(List.of(workflowId));
+  }
+
+  /**
+   * Cancel multiple workflows. After this function is called, the next step (not the current one)
+   * of each specified workflow will not execute.
+   *
+   * @param workflowIds a list of workflow IDs to cancel; must not be null
+   */
+  public void cancelWorkflows(@NonNull List<String> workflowIds) {
+    systemDatabase.cancelWorkflows(workflowIds);
   }
 
   /**
@@ -682,10 +692,68 @@ public void cancelWorkflow(@NonNull String workflowId) {
    */
   public <T, E extends Exception> @NonNull WorkflowHandle<T, E> resumeWorkflow(
       @NonNull String workflowId) {
-    systemDatabase.resumeWorkflow(workflowId);
+    systemDatabase.resumeWorkflows(List.of(workflowId));
     return retrieveWorkflow(workflowId);
   }
 
+  /**
+   * Resume multiple workflows starting from the step after the last complete step for each
+   * workflow. This method allows bulk resumption of workflows that were previously interrupted or
+   * failed.
+   *
+   * @param workflowIds a list of workflow IDs to resume; must not be null
+   * @return A list of handles to the resumed workflows
+   */
+  public @NonNull List<WorkflowHandle<Object, Exception>> resumeWorkflows(
+      @NonNull List<String> workflowIds) {
+    systemDatabase.resumeWorkflows(workflowIds);
+    return workflowIds.stream().map(this::retrieveWorkflow).toList();
+  }
+
+  /**
+   * Delete a workflow from the system. This permanently removes the workflow and its associated
+   * data from the database. Child workflows are preserved by default.
+   *
+   * @param workflowId ID of the workflow to delete; must not be null
+   */
+  public void deleteWorkflow(@NonNull String workflowId) {
+    deleteWorkflows(List.of(workflowId), false);
+  }
+
+  /**
+   * Delete a workflow from the system. This permanently removes the workflow and its associated
+   * data from the database.
+   *
+   * @param workflowId ID of the workflow to delete; must not be null
+   * @param deleteChildren if true, also delete any child workflows; if false, preserve child
+   *     workflows
+   */
+  public void deleteWorkflow(@NonNull String workflowId, boolean deleteChildren) {
+    deleteWorkflows(List.of(workflowId), deleteChildren);
+  }
+
+  /**
+   * Delete multiple workflows from the system. This permanently removes the workflows and their
+   * associated data from the database. Child workflows are preserved by default.
+   *
+   * @param workflowIds a list of workflow IDs to delete; must not be null
+   */
+  public void deleteWorkflows(@NonNull List<String> workflowIds) {
+    deleteWorkflows(workflowIds, false);
+  }
+
+  /**
+   * Delete multiple workflows from the system. This permanently removes the workflows and their
+   * associated data from the database.
+   *
+   * @param workflowIds a list of workflow IDs to delete; must not be null
+   * @param deleteChildren if true, also delete any child workflows; if false, preserve child
+   *     workflows
+   */
+  public void deleteWorkflows(@NonNull List<String> workflowIds, boolean deleteChildren) {
+    systemDatabase.deleteWorkflows(workflowIds, deleteChildren);
+  }
+
   /**
    * Fork a workflow, providing a handle to the new workflow
    *

transact/src/main/java/dev/dbos/transact/admin/AdminServer.java

@@ -230,7 +230,7 @@ private void cancel(HttpExchange exchange, String wfid) throws IOException {
 
     logger.info("cancel workflow {}", wfid);
 
-    dbosExecutor.cancelWorkflow(wfid);
+    dbosExecutor.cancelWorkflows(List.of(wfid));
     exchange.sendResponseHeaders(204, 0);
   }
 
@@ -239,7 +239,7 @@ private void resume(HttpExchange exchange, String wfid) throws IOException {
 
     logger.info("resume workflow {}", wfid);
 
-    dbosExecutor.resumeWorkflow(wfid);
+    dbosExecutor.resumeWorkflows(List.of(wfid));
     exchange.sendResponseHeaders(204, 0);
   }
 

transact/src/main/java/dev/dbos/transact/conductor/Conductor.java

@@ -183,7 +183,7 @@ public void onOpen(WebSocket ws) {
 
     @Override
     public CompletableFuture<?> onText(WebSocket ws, CharSequence data, boolean last) {
-      logger.debug("onText data size {} last {}", data.length(), last);
+      logger.trace("onText data size {} last {}", data.length(), last);
 
       // Streaming import path: queue each frame for the worker thread (non-blocking)
       if (importFrameQueue != null) {
@@ -754,12 +754,15 @@ static CompletableFuture<BaseResponse> handleCancel(Conductor conductor, BaseMes
     return CompletableFuture.supplyAsync(
         () -> {
           CancelRequest request = (CancelRequest) message;
+          List<String> ids =
+              (request.workflow_ids != null && !request.workflow_ids.isEmpty())
+                  ? request.workflow_ids
+                  : List.of(request.workflow_id);
           try {
-            conductor.dbosExecutor.cancelWorkflow(request.workflow_id);
+            conductor.dbosExecutor.cancelWorkflows(ids);
             return new SuccessResponse(request, true);
           } catch (Exception e) {
-            logger.error(
-                "Exception encountered when cancelling workflow {}", request.workflow_id, e);
+            logger.error("Exception encountered when cancelling workflow(s) {}", ids, e);
             return new SuccessResponse(request, e);
           }
         });
@@ -769,11 +772,15 @@ static CompletableFuture<BaseResponse> handleDelete(Conductor conductor, BaseMes
     return CompletableFuture.supplyAsync(
         () -> {
           DeleteRequest request = (DeleteRequest) message;
+          List<String> ids =
+              (request.workflow_ids != null && !request.workflow_ids.isEmpty())
+                  ? request.workflow_ids
+                  : List.of(request.workflow_id);
           try {
-            conductor.dbosExecutor.deleteWorkflow(request.workflow_id, request.delete_children);
+            conductor.systemDatabase.deleteWorkflows(ids, request.delete_children);
             return new SuccessResponse(request, true);
           } catch (Exception e) {
-            logger.error("Exception encountered when deleting workflow {}", request.workflow_id, e);
+            logger.error("Exception encountered when deleting workflow(s) {}", ids, e);
             return new SuccessResponse(request, e);
           }
         });
@@ -783,11 +790,15 @@ static CompletableFuture<BaseResponse> handleResume(Conductor conductor, BaseMes
     return CompletableFuture.supplyAsync(
         () -> {
           ResumeRequest request = (ResumeRequest) message;
+          List<String> ids =
+              (request.workflow_ids != null && !request.workflow_ids.isEmpty())
+                  ? request.workflow_ids
+                  : List.of(request.workflow_id);
           try {
-            conductor.dbosExecutor.resumeWorkflow(request.workflow_id);
+            conductor.dbosExecutor.resumeWorkflows(ids);
             return new SuccessResponse(request, true);
           } catch (Exception e) {
-            logger.error("Exception encountered when resuming workflow {}", request.workflow_id, e);
+            logger.error("Exception encountered when resuming workflow(s) {}", ids, e);
             return new SuccessResponse(request, e);
           }
         });

transact/src/main/java/dev/dbos/transact/conductor/protocol/CancelRequest.java

@@ -1,7 +1,10 @@
 package dev.dbos.transact.conductor.protocol;
 
+import java.util.List;
+
 public class CancelRequest extends BaseMessage {
   public String workflow_id;
+  public List<String> workflow_ids;
 
   public CancelRequest() {}
 

transact/src/main/java/dev/dbos/transact/conductor/protocol/DeleteRequest.java

@@ -1,7 +1,10 @@
 package dev.dbos.transact.conductor.protocol;
 
+import java.util.List;
+
 public class DeleteRequest extends BaseMessage {
   public String workflow_id;
+  public List<String> workflow_ids;
   public boolean delete_children;
 
   public DeleteRequest() {}

transact/src/main/java/dev/dbos/transact/conductor/protocol/ResumeRequest.java

@@ -1,7 +1,10 @@
 package dev.dbos.transact.conductor.protocol;
 
+import java.util.List;
+
 public class ResumeRequest extends BaseMessage {
   public String workflow_id;
+  public List<String> workflow_ids;
 
   public ResumeRequest() {}