v4: batches can only complete when all their runs have been created (#2101)

* v4: batches can only complete when all their runs have been created

* Fix flaky runs replication test

apps/webapp/app/runEngine/services/batchTrigger.server.ts

@@ -576,6 +576,13 @@ export class RunEngineBatchTriggerService extends WithRunEngine {
 
     //triggered all the runs
     if (updatedBatch.processingJobsCount >= updatedBatch.runCount) {
+      logger.debug("[RunEngineBatchTrigger][processBatchTaskRun] All runs created", {
+        batchId: batch.friendlyId,
+        processingJobsCount: updatedBatch.processingJobsCount,
+        runCount: updatedBatch.runCount,
+        workingIndex,
+      });
+
       //if all the runs were idempotent, it's possible the batch is already completed
       await this._engine.tryCompleteBatch({ batchId: batch.id });
     }

apps/webapp/test/runsReplicationService.part1.test.ts

@@ -128,7 +128,7 @@ describe("RunsReplicationService (part 1/2)", () => {
 
       expect(transactionSpan).not.toBeNull();
       expect(transactionSpan?.attributes["transaction.parse_duration_ms"]).toBeGreaterThan(0);
-      expect(transactionSpan?.attributes["transaction.parse_duration_ms"]).toBeLessThan(1);
+      expect(transactionSpan?.attributes["transaction.parse_duration_ms"]).toBeLessThan(5);
 
       await runsReplicationService.stop();
     }

internal-packages/run-engine/src/engine/systems/batchSystem.ts

@@ -42,6 +42,8 @@ export class BatchSystem {
         select: {
           status: true,
           runtimeEnvironmentId: true,
+          processingJobsCount: true,
+          runCount: true,
         },
         where: {
           id: batchId,
@@ -58,6 +60,15 @@ export class BatchSystem {
         return;
       }
 
+      if (batch.processingJobsCount < batch.runCount) {
+        this.$.logger.debug("#tryCompleteBatch: Not all runs are created yet", {
+          batchId,
+          processingJobsCount: batch.processingJobsCount,
+          runCount: batch.runCount,
+        });
+        return;
+      }
+
       const runs = await this.$.prisma.taskRun.findMany({
         select: {
           id: true,