Prevent v2 batches from being resumed before all items have been created (#1614)

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

@@ -51,8 +51,25 @@ export class ResumeBatchRunService extends BaseService {
       return "ERROR";
     }
 
+    if (batchRun.batchVersion === "v2") {
+      // Make sure batchRun.items.length is equal to or greater than batchRun.runCount
+      if (batchRun.items.length < batchRun.runCount) {
+        logger.debug("ResumeBatchRunService: All items aren't yet completed [v2]", {
+          batchRunId: batchRun.id,
+          batchRun: {
+            id: batchRun.id,
+            status: batchRun.status,
+            itemsLength: batchRun.items.length,
+            runCount: batchRun.runCount,
+          },
+        });
+
+        return "PENDING";
+      }
+    }
+
     if (batchRun.items.some((item) => !finishedBatchRunStatuses.includes(item.status))) {
-      logger.debug("ResumeBatchRunService: All items aren't yet completed", {
+      logger.debug("ResumeBatchRunService: All items aren't yet completed [v1]", {
         batchRunId: batchRun.id,
         batchRun: {
           id: batchRun.id,

references/v3-catalog/src/trigger/batch.ts

@@ -376,10 +376,18 @@ export const batchV2TestTask = task({
 
       for await (const liveRun0 of runs.subscribeToRun(response1.runs[0].id)) {
         logger.debug("subscribed to run0", { liveRun0 });
+
+        if (liveRun0.finishedAt) {
+          break;
+        }
       }
 
       for await (const liveRun1 of runs.subscribeToRun(response1.runs[1].id)) {
         logger.debug("subscribed to run1", { liveRun1 });
+
+        if (liveRun1.finishedAt) {
+          break;
+        }
       }
     });