fallback to graphile worker when a tx is required

Author: ericallam

apps/webapp/app/services/worker.server.ts

@@ -33,7 +33,7 @@ import { logger } from "./logger.server";
 const workerCatalog = {
   // @deprecated, moved to commonWorker.server.ts
   scheduleEmail: DeliverEmailSchema,
-  // @deprecated, moved to commonWorker.server.ts
+  // @deprecated, but still used when resuming batch runs in a transaction
   "v3.resumeBatchRun": z.object({
     batchRunId: z.string(),
   }),
@@ -164,7 +164,7 @@ function getWorkerQueue() {
           await sendEmail(payload);
         },
       },
-      // @deprecated, moved to commonWorker.server.ts
+      // @deprecated, moved to commonWorker.server.ts but still used when resuming batch runs in a transaction
       "v3.resumeBatchRun": {
         priority: 0,
         maxAttempts: 5,

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

@@ -942,6 +942,15 @@ export async function completeBatchTaskRunItemV3(
 ) {
   const isRetry = retryAttempt !== undefined;
 
+  logger.debug("completeBatchTaskRunItemV3", {
+    itemId,
+    batchTaskRunId,
+    scheduleResumeOnComplete,
+    taskRunAttemptId,
+    retryAttempt,
+    isRetry,
+  });
+
   if (isRetry) {
     logger.debug("completeBatchTaskRunItemV3 retrying", {
       itemId,

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

@@ -392,11 +392,7 @@ export class CreateCheckpointService extends BaseService {
             checkpointEventId: checkpointEvent.id,
           });
 
-          await ResumeBatchRunService.enqueue(
-            batchRun.id,
-            batchRun.batchVersion === "v3",
-            this._prisma
-          );
+          await ResumeBatchRunService.enqueue(batchRun.id, batchRun.batchVersion === "v3");
 
           return {
             success: true,

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

@@ -244,7 +244,7 @@ export class FinalizeTaskRunService extends BaseService {
 
         // This won't resume because this batch does not have a dependent task attempt ID
         // or is in development, but this service will mark the batch as completed
-        await ResumeBatchRunService.enqueue(item.batchTaskRunId, false, this._prisma);
+        await ResumeBatchRunService.enqueue(item.batchTaskRunId, false);
       }
     }
   }

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

@@ -4,6 +4,7 @@ import { marqs } from "~/v3/marqs/index.server";
 import { BaseService } from "./baseService.server";
 import { logger } from "~/services/logger.server";
 import { BatchTaskRun } from "@trigger.dev/database";
+import { workerQueue } from "~/services/worker.server";
 
 const finishedBatchRunStatuses = ["COMPLETED", "FAILED", "CANCELED"];
 
@@ -331,17 +332,43 @@ export class ResumeBatchRunService extends BaseService {
   static async enqueue(
     batchRunId: string,
     skipJobKey: boolean,
-    tx: PrismaClientOrTransaction,
+    tx?: PrismaClientOrTransaction,
     runAt?: Date
   ) {
-    return await commonWorker.enqueue({
-      id: skipJobKey ? undefined : `resumeBatchRun-${batchRunId}`,
-      job: "v3.resumeBatchRun",
-      payload: {
+    if (tx) {
+      logger.debug("ResumeBatchRunService: Enqueuing resume batch run using workerQueue", {
         batchRunId,
-      },
-      availableAt: runAt,
-    });
+        skipJobKey,
+        runAt,
+      });
+
+      return await workerQueue.enqueue(
+        "v3.resumeBatchRun",
+        {
+          batchRunId,
+        },
+        {
+          jobKey: skipJobKey ? undefined : `resumeBatchRun-${batchRunId}`,
+          runAt,
+          tx,
+        }
+      );
+    } else {
+      logger.debug("ResumeBatchRunService: Enqueuing resume batch run using commonWorker", {
+        batchRunId,
+        skipJobKey,
+        runAt,
+      });
+
+      return await commonWorker.enqueue({
+        id: skipJobKey ? undefined : `resumeBatchRun-${batchRunId}`,
+        job: "v3.resumeBatchRun",
+        payload: {
+          batchRunId,
+        },
+        availableAt: runAt,
+      });
+    }
   }
 }