Make sure to complete batches, even if they’re not andWait ones

Author: matt-aitken

apps/webapp/app/v3/services/batchTriggerV3.server.ts

@@ -567,17 +567,21 @@ export class BatchTriggerV3Service extends WithRunEngine {
       return { status: "INCOMPLETE", workingIndex };
     }
 
-    if (
-      parentRunId &&
-      resumeParentOnCompletion &&
-      updatedBatch.runIds.length === updatedBatch.runCount
-    ) {
-      await this._engine.unblockRunForCreatedBatch({
-        runId: RunId.fromFriendlyId(parentRunId),
-        batchId: batch.id,
-        environmentId: environment.id,
-        projectId: environment.projectId,
-      });
+    //triggered all the runs
+    if (updatedBatch.runIds.length === updatedBatch.runCount) {
+      //unblock the parent run from the batch
+      //this prevents the parent continuing before all the runs are created
+      if (parentRunId && resumeParentOnCompletion) {
+        await this._engine.unblockRunForCreatedBatch({
+          runId: RunId.fromFriendlyId(parentRunId),
+          batchId: batch.id,
+          environmentId: environment.id,
+          projectId: environment.projectId,
+        });
+      }
+
+      //if all the runs were idempotent, it's possible the batch is already completed
+      await this._engine.tryCompleteBatch({ batchId: batch.id });
     }
 
     return { status: "COMPLETE" };

internal-packages/run-engine/src/engine/index.ts

@@ -1662,7 +1662,9 @@ export class RunEngine {
       id: waitpoint.id,
       output: { value: "Batch waitpoint completed", isError: false },
     });
+  }
 
+  async tryCompleteBatch({ batchId }: { batchId: string }): Promise<void> {
     await this.worker.enqueue({
       //this will debounce the call
       id: `tryCompleteBatch:${batchId}`,
@@ -3187,14 +3189,7 @@ export class RunEngine {
    */
   async #finalizeRun({ id, batchId }: { id: string; batchId: string | null }) {
     if (batchId) {
-      await this.worker.enqueue({
-        //this will debounce the call
-        id: `tryCompleteBatch:${batchId}`,
-        job: "tryCompleteBatch",
-        payload: { batchId: batchId },
-        //2s in the future
-        availableAt: new Date(Date.now() + 2_000),
-      });
+      await this.tryCompleteBatch({ batchId });
     }
   }
 

references/hello-world/src/trigger/idempotency.ts

@@ -245,3 +245,40 @@ export const idempotencyBatch = task({
     logger.log("Results 4", { results4 });
   },
 });
+
+export const idempotencyTriggerByTaskAndWait = task({
+  id: "idempotency-trigger-by-task-and-wait",
+  maxDuration: 60,
+  run: async () => {
+    const successfulKey = await idempotencyKeys.create("a", { scope: "global" });
+    const failureKey = await idempotencyKeys.create("b", { scope: "global" });
+
+    const results1 = await batch.triggerByTaskAndWait([
+      {
+        task: childTask,
+        payload: { message: "Hello, world !" },
+        options: { idempotencyKey: successfulKey, idempotencyKeyTTL: "60s" },
+      },
+      {
+        task: childTask,
+        payload: { message: "Hello, world 2!" },
+        options: { idempotencyKey: failureKey, idempotencyKeyTTL: "60s" },
+      },
+    ]);
+    logger.log("Results 1", { results1 });
+
+    const results2 = await batch.triggerByTaskAndWait([
+      {
+        task: childTask,
+        payload: { message: "Hello, world !" },
+        options: { idempotencyKey: successfulKey, idempotencyKeyTTL: "60s" },
+      },
+      {
+        task: childTask,
+        payload: { message: "Hello, world 2!" },
+        options: { idempotencyKey: failureKey, idempotencyKeyTTL: "60s" },
+      },
+    ]);
+    logger.log("Results 2", { results2 });
+  },
+});