Move more of graphile worker jobs into Redis Worker (workerQueue -> commonWorker)

Author: ericallam

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

@@ -14,7 +14,7 @@ import { env } from "~/env.server";
 import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
 import { getEntitlement } from "~/services/platform.v3.server";
-import { workerQueue } from "~/services/worker.server";
+import { commonWorker } from "~/v3/commonWorker.server";
 import { downloadPacketFromObjectStore, uploadPacketToObjectStore } from "../../v3/r2.server";
 import { ServiceValidationError, WithRunEngine } from "../../v3/services/baseService.server";
 import { OutOfEntitlementError, TriggerTaskService } from "../../v3/services/triggerTask.server";
@@ -244,88 +244,80 @@ export class RunEngineBatchTriggerService extends WithRunEngine {
         }
       }
     } else {
-      return await $transaction(this._prisma, async (tx) => {
-        const batch = await tx.batchTaskRun.create({
-          data: {
-            id: BatchId.fromFriendlyId(batchId),
-            friendlyId: batchId,
-            runtimeEnvironmentId: environment.id,
-            runCount: body.items.length,
-            runIds: [],
-            payload: payloadPacket.data,
-            payloadType: payloadPacket.dataType,
-            options,
-            batchVersion: "runengine:v1",
-            oneTimeUseToken: options.oneTimeUseToken,
-          },
+      const batch = await this._prisma.batchTaskRun.create({
+        data: {
+          id: BatchId.fromFriendlyId(batchId),
+          friendlyId: batchId,
+          runtimeEnvironmentId: environment.id,
+          runCount: body.items.length,
+          runIds: [],
+          payload: payloadPacket.data,
+          payloadType: payloadPacket.dataType,
+          options,
+          batchVersion: "runengine:v1",
+          oneTimeUseToken: options.oneTimeUseToken,
+        },
+      });
+
+      if (body.parentRunId && body.resumeParentOnCompletion) {
+        await this._engine.blockRunWithCreatedBatch({
+          runId: RunId.fromFriendlyId(body.parentRunId),
+          batchId: batch.id,
+          environmentId: environment.id,
+          projectId: environment.projectId,
+          organizationId: environment.organizationId,
         });
+      }
 
-        if (body.parentRunId && body.resumeParentOnCompletion) {
-          await this._engine.blockRunWithCreatedBatch({
-            runId: RunId.fromFriendlyId(body.parentRunId),
+      switch (this._batchProcessingStrategy) {
+        case "sequential": {
+          await this.#enqueueBatchTaskRun({
             batchId: batch.id,
-            environmentId: environment.id,
-            projectId: environment.projectId,
-            organizationId: environment.organizationId,
-            tx,
+            processingId: batchId,
+            range: { start: 0, count: PROCESSING_BATCH_SIZE },
+            attemptCount: 0,
+            strategy: this._batchProcessingStrategy,
+            parentRunId: body.parentRunId,
+            resumeParentOnCompletion: body.resumeParentOnCompletion,
           });
-        }
 
-        switch (this._batchProcessingStrategy) {
-          case "sequential": {
-            await this.#enqueueBatchTaskRun(
-              {
+          break;
+        }
+        case "parallel": {
+          const ranges = Array.from({
+            length: Math.ceil(body.items.length / PROCESSING_BATCH_SIZE),
+          }).map((_, index) => ({
+            start: index * PROCESSING_BATCH_SIZE,
+            count: PROCESSING_BATCH_SIZE,
+          }));
+
+          await Promise.all(
+            ranges.map((range, index) =>
+              this.#enqueueBatchTaskRun({
                 batchId: batch.id,
-                processingId: batchId,
-                range: { start: 0, count: PROCESSING_BATCH_SIZE },
+                processingId: `${index}`,
+                range,
                 attemptCount: 0,
                 strategy: this._batchProcessingStrategy,
                 parentRunId: body.parentRunId,
                 resumeParentOnCompletion: body.resumeParentOnCompletion,
-              },
-              tx
-            );
-
-            break;
-          }
-          case "parallel": {
-            const ranges = Array.from({
-              length: Math.ceil(body.items.length / PROCESSING_BATCH_SIZE),
-            }).map((_, index) => ({
-              start: index * PROCESSING_BATCH_SIZE,
-              count: PROCESSING_BATCH_SIZE,
-            }));
-
-            await Promise.all(
-              ranges.map((range, index) =>
-                this.#enqueueBatchTaskRun(
-                  {
-                    batchId: batch.id,
-                    processingId: `${index}`,
-                    range,
-                    attemptCount: 0,
-                    strategy: this._batchProcessingStrategy,
-                    parentRunId: body.parentRunId,
-                    resumeParentOnCompletion: body.resumeParentOnCompletion,
-                  },
-                  tx
-                )
-              )
-            );
+              })
+            )
+          );
 
-            break;
-          }
+          break;
         }
+      }
 
-        return batch;
-      });
+      return batch;
     }
   }
 
   async #enqueueBatchTaskRun(options: BatchProcessingOptions, tx?: PrismaClientOrTransaction) {
-    await workerQueue.enqueue("runengine.processBatchTaskRun", options, {
-      tx,
-      jobKey: `RunEngineBatchTriggerService.process:${options.batchId}:${options.processingId}`,
+    await commonWorker.enqueue({
+      id: `RunEngineBatchTriggerService.process:${options.batchId}:${options.processingId}`,
+      job: "runengine.processBatchTaskRun",
+      payload: options,
     });
   }
 

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

@@ -4,7 +4,7 @@ import type { SendEmailOptions } from "remix-auth-email-link";
 import { redirect } from "remix-typedjson";
 import { env } from "~/env.server";
 import type { AuthUser } from "./authUser";
-import { workerQueue } from "./worker.server";
+import { commonWorker } from "~/v3/commonWorker.server";
 import { logger } from "./logger.server";
 import { singleton } from "~/utils/singleton";
 import { assertEmailAllowed } from "~/utils/email";
@@ -93,8 +93,12 @@ export async function sendPlainTextEmail(options: SendPlainTextOptions) {
 }
 
 export async function scheduleEmail(data: DeliverEmail, delay?: { seconds: number }) {
-  const runAt = delay ? new Date(Date.now() + delay.seconds * 1000) : undefined;
-  await workerQueue.enqueue("scheduleEmail", data, { runAt });
+  const availableAt = delay ? new Date(Date.now() + delay.seconds * 1000) : undefined;
+  await commonWorker.enqueue({
+    job: "scheduleEmail",
+    payload: data,
+    availableAt,
+  });
 }
 
 export async function sendEmail(data: DeliverEmail) {