Implement checkpoint tests, handle dequeuing QUEUED_EXECUTING runs

Author: ericallam

internal-packages/run-engine/src/engine/index.ts

@@ -1009,6 +1009,7 @@ export class RunEngine {
     batch,
     workerId,
     runnerId,
+    tx,
   }: {
     runId: string;
     waitpoints: string | string[];
@@ -1033,6 +1034,7 @@ export class RunEngine {
       batch,
       workerId,
       runnerId,
+      tx,
     });
   }
 

internal-packages/run-engine/src/engine/statuses.ts

@@ -26,6 +26,7 @@ export function isCheckpointable(status: TaskRunExecutionStatus): boolean {
     //executing
     "EXECUTING",
     "EXECUTING_WITH_WAITPOINTS",
+    "QUEUED_EXECUTING",
   ];
   return checkpointableStatuses.includes(status);
 }

internal-packages/run-engine/src/engine/systems/dequeueSystem.ts

@@ -10,6 +10,7 @@ import { RunEngineOptions } from "../types.js";
 import { ExecutionSnapshotSystem, getLatestExecutionSnapshot } from "./executionSnapshotSystem.js";
 import { RunAttemptSystem } from "./runAttemptSystem.js";
 import { SystemResources } from "./systems.js";
+import { sendNotificationToWorker } from "../eventBus.js";
 
 export type DequeueSystemOptions = {
   resources: SystemResources;
@@ -128,6 +129,38 @@ export class DequeueSystem {
                 return null;
               }
 
+              if (snapshot.executionStatus === "QUEUED_EXECUTING") {
+                const newSnapshot = await this.executionSnapshotSystem.createExecutionSnapshot(
+                  prisma,
+                  {
+                    run: {
+                      id: runId,
+                      status: snapshot.runStatus,
+                      attemptNumber: snapshot.attemptNumber,
+                    },
+                    snapshot: {
+                      executionStatus: "EXECUTING",
+                      description: "Run was continued, whilst still executing.",
+                    },
+                    environmentId: snapshot.environmentId,
+                    environmentType: snapshot.environmentType,
+                    batchId: snapshot.batchId ?? undefined,
+                    completedWaitpoints: snapshot.completedWaitpoints.map((waitpoint) => ({
+                      id: waitpoint.id,
+                      index: waitpoint.index,
+                    })),
+                  }
+                );
+
+                await sendNotificationToWorker({
+                  runId,
+                  snapshot: newSnapshot,
+                  eventBus: this.$.eventBus,
+                });
+
+                return null;
+              }
+
               const result = await getRunWithBackgroundWorkerTasks(
                 prisma,
                 runId,