Run workers in a separate Docker container

We've outgrown the run-everything-on-one-machine setup. This change moves background jobs to a different Docker image in production. It also adds a `jobKey` to certain jobs so if we try to process the same cell multiple times it'll only actually run the job once.

app/src/env.mjs

@@ -26,6 +26,10 @@ export const env = createEnv({
     SMTP_PORT: z.string().default("placeholder"),
     SMTP_LOGIN: z.string().default("placeholder"),
     SMTP_PASSWORD: z.string().default("placeholder"),
+    WORKER_CONCURRENCY: z
+      .string()
+      .default("10")
+      .transform((val) => parseInt(val)),
   },
 
   /**
@@ -68,6 +72,7 @@ export const env = createEnv({
     SMTP_PORT: process.env.SMTP_PORT,
     SMTP_LOGIN: process.env.SMTP_LOGIN,
     SMTP_PASSWORD: process.env.SMTP_PASSWORD,
+    WORKER_CONCURRENCY: process.env.WORKER_CONCURRENCY,
   },
   /**
    * Run `build` or `dev` with `SKIP_ENV_VALIDATION` to skip env validation.

app/src/server/tasks/defineTask.ts

@@ -1,4 +1,4 @@
-import { type Helpers, type Task, makeWorkerUtils } from "graphile-worker";
+import { type Helpers, type Task, makeWorkerUtils, TaskSpec } from "graphile-worker";
 import { env } from "~/env.mjs";
 
 let workerUtilsPromise: ReturnType<typeof makeWorkerUtils> | null = null;
@@ -16,9 +16,11 @@ function defineTask<TPayload>(
   taskIdentifier: string,
   taskHandler: (payload: TPayload, helpers: Helpers) => Promise<void>,
 ) {
-  const enqueue = async (payload: TPayload, runAt?: Date) => {
+  const enqueue = async (payload: TPayload, spec?: TaskSpec) => {
     console.log("Enqueuing task", taskIdentifier, payload);
-    await (await workerUtils()).addJob(taskIdentifier, payload, { runAt });
+
+    const utils = await workerUtils();
+    return await utils.addJob(taskIdentifier, payload, spec);
   };
 
   const handler = (payload: TPayload, helpers: Helpers) => {

app/src/server/tasks/queryModel.task.ts

@@ -153,7 +153,7 @@ export const queryModel = defineTask<QueryModelJob>("queryModel", async (task) =
           stream,
           numPreviousTries: numPreviousTries + 1,
         },
-        retryTime,
+        { runAt: retryTime, jobKey: cellId },
       );
       await prisma.scenarioVariantCell.update({
         where: { id: cellId },
@@ -184,6 +184,6 @@ export const queueQueryModel = async (cellId: string, stream: boolean) => {
         jobQueuedAt: new Date(),
       },
     }),
-    queryModel.enqueue({ cellId, stream, numPreviousTries: 0 }),
+    queryModel.enqueue({ cellId, stream, numPreviousTries: 0 }, { jobKey: cellId }),
   ]);
 };

app/src/server/tasks/worker.ts

@@ -17,7 +17,7 @@ const taskList = registeredTasks.reduce((acc, task) => {
 // Run a worker to execute jobs:
 const runner = await run({
   connectionString: env.DATABASE_URL,
-  concurrency: 10,
+  concurrency: env.WORKER_CONCURRENCY,
   // Install signal handlers for graceful shutdown on SIGINT, SIGTERM, etc
   noHandleSignals: false,
   pollInterval: 1000,