v4: fix batchTriggerAndWait completion issues by processing batch chunks sequentially

Author: ericallam

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

@@ -58,7 +58,10 @@ export class RunEngineBatchTriggerService extends WithRunEngine {
   ) {
     super({ prisma });
 
-    this._batchProcessingStrategy = batchProcessingStrategy ?? "parallel";
+    // Eric note: We need to force sequential processing because when doing parallel, we end up with high-contention on the parent run lock
+    // becuase we are triggering a lot of runs at once, and each one is trying to lock the parent run.
+    // by forcing sequential, we are only ever locking the parent run for a single run at a time.
+    this._batchProcessingStrategy = "sequential";
   }
 
   public async call(
@@ -316,6 +319,14 @@ export class RunEngineBatchTriggerService extends WithRunEngine {
     }
   }
 
+  async #enqueueBatchTaskRun(options: BatchProcessingOptions, tx?: PrismaClientOrTransaction) {
+    await workerQueue.enqueue("runengine.processBatchTaskRun", options, {
+      tx,
+      jobKey: `RunEngineBatchTriggerService.process:${options.batchId}:${options.processingId}`,
+    });
+  }
+
+  // This is the function that the worker will call
   async processBatchTaskRun(options: BatchProcessingOptions) {
     logger.debug("[RunEngineBatchTrigger][processBatchTaskRun] Processing batch", {
       options,
@@ -648,13 +659,6 @@ export class RunEngineBatchTriggerService extends WithRunEngine {
       : undefined;
   }
 
-  async #enqueueBatchTaskRun(options: BatchProcessingOptions, tx?: PrismaClientOrTransaction) {
-    await workerQueue.enqueue("runengine.processBatchTaskRun", options, {
-      tx,
-      jobKey: `RunEngineBatchTriggerService.process:${options.batchId}:${options.processingId}`,
-    });
-  }
-
   async #handlePayloadPacket(
     payload: any,
     pathPrefix: string,

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

@@ -0,0 +1,39 @@
+import { task } from "@trigger.dev/sdk/v3";
+import { setTimeout } from "timers/promises";
+
+export const batchTriggerAndWait = task({
+  id: "batch-trigger-and-wait",
+  maxDuration: 60,
+  run: async (payload: { count: number }, { ctx }) => {
+    const payloads = Array.from({ length: payload.count }, (_, i) => ({
+      payload: { waitSeconds: 1, output: `test${i}` },
+    }));
+
+    // First batch triggerAndWait with idempotency keys
+    const firstResults = await fixedLengthTask.batchTriggerAndWait(payloads);
+  },
+});
+
+type Payload = {
+  waitSeconds: number;
+  error?: string;
+  output?: any;
+};
+
+export const fixedLengthTask = task({
+  id: "fixed-length-lask",
+  retry: {
+    maxAttempts: 2,
+    maxTimeoutInMs: 100,
+  },
+  machine: "micro",
+  run: async ({ waitSeconds = 1, error, output }: Payload) => {
+    await setTimeout(waitSeconds * 1000);
+
+    if (error) {
+      throw new Error(error);
+    }
+
+    return output;
+  },
+});