Store the graphile job ID on the executions to be able to fast-forward delayed tasks

apps/webapp/app/services/runs/performRunExecution.ts

@@ -158,13 +158,22 @@ export class PerformRunExecutionService {
           },
         });
 
-        await workerQueue.enqueue(
+        const job = await workerQueue.enqueue(
           "performRunExecution",
           {
             id: runExecution.id,
           },
           { tx }
         );
+
+        await tx.jobRunExecution.update({
+          where: {
+            id: runExecution.id,
+          },
+          data: {
+            graphileJobId: job.id,
+          },
+        });
       });
     }
   }
@@ -347,13 +356,22 @@ export class PerformRunExecutionService {
           },
         });
 
-        await workerQueue.enqueue(
+        const graphileJob = await workerQueue.enqueue(
           "performRunExecution",
           {
             id: newJobExecution.id,
           },
           { tx, runAt: resumeTask.delayUntil ?? undefined }
         );
+
+        await tx.jobRunExecution.update({
+          where: {
+            id: newJobExecution.id,
+          },
+          data: {
+            graphileJobId: graphileJob.id,
+          },
+        });
       });
     }
   }
@@ -389,11 +407,20 @@ export class PerformRunExecutionService {
 
       const runAt = new Date(Date.now() + retryDelayInMs);
 
-      await workerQueue.enqueue(
+      const job = await workerQueue.enqueue(
         "performRunExecution",
         { id: execution.id },
         { runAt, tx }
       );
+
+      await tx.jobRunExecution.update({
+        where: {
+          id: execution.id,
+        },
+        data: {
+          graphileJobId: job.id,
+        },
+      });
     });
   }
 
@@ -483,7 +510,7 @@ export class PerformRunExecutionService {
             },
           });
 
-          await workerQueue.enqueue(
+          const job = await workerQueue.enqueue(
             "performRunExecution",
             {
               id: runExecution.id,
@@ -491,6 +518,15 @@ export class PerformRunExecutionService {
             { tx }
           );
 
+          await tx.jobRunExecution.update({
+            where: {
+              id: runExecution.id,
+            },
+            data: {
+              graphileJobId: job.id,
+            },
+          });
+
           break;
         }
       }

apps/webapp/app/services/runs/startRun.server.ts

@@ -141,7 +141,7 @@ export class StartRunService {
 
     const execution = await updateRunAndCreateExecution();
 
-    await workerQueue.enqueue(
+    const job = await workerQueue.enqueue(
       "performRunExecution",
       {
         id: execution.id,
@@ -149,6 +149,13 @@ export class StartRunService {
       { tx }
     );
 
+    await tx.jobRunExecution.update({
+      where: { id: execution.id },
+      data: {
+        graphileJobId: job.id,
+      },
+    });
+
     await workerQueue.enqueue(
       "startQueuedRuns",
       {

apps/webapp/prisma/migrations/20230609150207_add_graphile_job_id_to_executions/migration.sql

@@ -0,0 +1,2 @@
+-- AlterTable
+ALTER TABLE "JobRunExecution" ADD COLUMN     "graphileJobId" INTEGER;

apps/webapp/prisma/migrations/20230609150822_add_graphile_job_id_string/migration.sql

@@ -0,0 +1,2 @@
+-- AlterTable
+ALTER TABLE "JobRunExecution" ALTER COLUMN "graphileJobId" SET DATA TYPE TEXT;

apps/webapp/prisma/schema.prisma

@@ -605,6 +605,8 @@ model JobRunExecution {
 
   resumeTask   Task?   @relation(fields: [resumeTaskId], references: [id], onDelete: Cascade, onUpdate: Cascade)
   resumeTaskId String?
+
+  graphileJobId String?
 }
 
 enum JobRunExecutionReason {