Remove QUEUED_EXECUTING because we no longer "eagerly" release before checkpointing

Author: ericallam

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

@@ -595,75 +595,45 @@ export class WaitpointSystem {
           return "skipped";
         }
         case "EXECUTING_WITH_WAITPOINTS": {
-          const result = await this.$.runQueue.reacquireConcurrency(
-            run.runtimeEnvironment.organization.id,
-            runId
-          );
-
-          if (result) {
-            const newSnapshot = await this.executionSnapshotSystem.createExecutionSnapshot(
-              this.$.prisma,
-              {
-                run: {
-                  id: runId,
-                  status: snapshot.runStatus,
-                  attemptNumber: snapshot.attemptNumber,
-                },
-                snapshot: {
-                  executionStatus: "EXECUTING",
-                  description: "Run was continued, whilst still executing.",
-                },
-                previousSnapshotId: snapshot.id,
-                environmentId: snapshot.environmentId,
-                environmentType: snapshot.environmentType,
-                projectId: snapshot.projectId,
-                organizationId: snapshot.organizationId,
-                batchId: snapshot.batchId ?? undefined,
-                completedWaitpoints: blockingWaitpoints.map((b) => ({
-                  id: b.waitpoint.id,
-                  index: b.batchIndex ?? undefined,
-                })),
-              }
-            );
-
-            this.$.logger.debug(
-              `continueRunIfUnblocked: run was still executing, sending notification`,
-              {
-                runId,
-                snapshot,
-                newSnapshot,
-              }
-            );
-
-            await sendNotificationToWorker({
-              runId,
-              snapshot: newSnapshot,
-              eventBus: this.$.eventBus,
-            });
-          } else {
-            // Because we cannot reacquire the concurrency, we need to enqueue the run again
-            // and because the run is still executing, we need to set the status to QUEUED_EXECUTING
-            const newSnapshot = await this.enqueueSystem.enqueueRun({
-              run,
-              env: run.runtimeEnvironment,
+          const newSnapshot = await this.executionSnapshotSystem.createExecutionSnapshot(
+            this.$.prisma,
+            {
+              run: {
+                id: runId,
+                status: snapshot.runStatus,
+                attemptNumber: snapshot.attemptNumber,
+              },
               snapshot: {
-                status: "QUEUED_EXECUTING",
-                description: "Run can continue, but is waiting for concurrency",
+                executionStatus: "EXECUTING",
+                description: "Run was continued, whilst still executing.",
               },
               previousSnapshotId: snapshot.id,
+              environmentId: snapshot.environmentId,
+              environmentType: snapshot.environmentType,
+              projectId: snapshot.projectId,
+              organizationId: snapshot.organizationId,
               batchId: snapshot.batchId ?? undefined,
               completedWaitpoints: blockingWaitpoints.map((b) => ({
                 id: b.waitpoint.id,
                 index: b.batchIndex ?? undefined,
               })),
-            });
+            }
+          );
 
-            this.$.logger.debug(`continueRunIfUnblocked: run goes to QUEUED_EXECUTING`, {
+          this.$.logger.debug(
+            `continueRunIfUnblocked: run was still executing, sending notification`,
+            {
               runId,
               snapshot,
               newSnapshot,
-            });
-          }
+            }
+          );
+
+          await sendNotificationToWorker({
+            runId,
+            snapshot: newSnapshot,
+            eventBus: this.$.eventBus,
+          });
 
           break;
         }

internal-packages/run-engine/src/engine/tests/checkpoints.test.ts

@@ -775,237 +775,6 @@ describe("RunEngine checkpoints", () => {
     }
   );
 
-  containerTest(
-    "when a checkpoint is created while the run is in QUEUED_EXECUTING state, the run is QUEUED",
-    async ({ prisma, redisOptions }) => {
-      const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
-
-      const engine = new RunEngine({
-        prisma,
-        worker: {
-          redis: redisOptions,
-          workers: 1,
-          tasksPerWorker: 10,
-          pollIntervalMs: 100,
-        },
-        queue: {
-          redis: redisOptions,
-          masterQueueConsumersDisabled: true,
-          processWorkerQueueDebounceMs: 50,
-        },
-        runLock: {
-          redis: redisOptions,
-        },
-        machines: {
-          defaultMachine: "small-1x",
-          machines: {
-            "small-1x": {
-              name: "small-1x" as const,
-              cpu: 0.5,
-              memory: 0.5,
-              centsPerMs: 0.0001,
-            },
-          },
-          baseCostInCents: 0.0001,
-        },
-        tracer: trace.getTracer("test", "0.0.0"),
-      });
-
-      try {
-        const taskIdentifier = "test-task";
-
-        // Create background worker
-        await setupBackgroundWorker(
-          engine,
-          authenticatedEnvironment,
-          taskIdentifier,
-          undefined,
-          undefined,
-          {
-            concurrencyLimit: 1,
-          }
-        );
-
-        // Create first run with queue concurrency limit of 1
-        const firstRun = await engine.trigger(
-          {
-            number: 1,
-            friendlyId: "run_first",
-            environment: authenticatedEnvironment,
-            taskIdentifier,
-            payload: "{}",
-            payloadType: "application/json",
-            context: {},
-            traceContext: {},
-            traceId: "t12345-first",
-            spanId: "s12345-first",
-            workerQueue: "main",
-            queue: "task/test-task",
-            isTest: false,
-            tags: [],
-          },
-          prisma
-        );
-
-        await setTimeout(500);
-
-        // Dequeue and start the first run
-        const dequeuedFirst = await engine.dequeueFromWorkerQueue({
-          consumerId: "test_12345",
-          workerQueue: "main",
-        });
-        expect(dequeuedFirst.length).toBe(1);
-        assertNonNullable(dequeuedFirst[0]);
-
-        const firstAttempt = await engine.startRunAttempt({
-          runId: dequeuedFirst[0].run.id,
-          snapshotId: dequeuedFirst[0].snapshot.id,
-        });
-        expect(firstAttempt.snapshot.executionStatus).toBe("EXECUTING");
-
-        // Create a manual waitpoint for the first run
-        const waitpoint = await engine.createManualWaitpoint({
-          environmentId: authenticatedEnvironment.id,
-          projectId: authenticatedEnvironment.projectId,
-        });
-        expect(waitpoint.waitpoint.status).toBe("PENDING");
-
-        // Block the first run
-        const blockedResult = await engine.blockRunWithWaitpoint({
-          runId: firstRun.id,
-          waitpoints: waitpoint.waitpoint.id,
-          projectId: authenticatedEnvironment.projectId,
-          organizationId: authenticatedEnvironment.organizationId,
-        });
-
-        // Verify first run is blocked
-        const firstRunData = await engine.getRunExecutionData({ runId: firstRun.id });
-        expect(firstRunData?.snapshot.executionStatus).toBe("EXECUTING_WITH_WAITPOINTS");
-
-        // Create and start second run on the same queue
-        const secondRun = await engine.trigger(
-          {
-            number: 2,
-            friendlyId: "run_second",
-            environment: authenticatedEnvironment,
-            taskIdentifier,
-            payload: "{}",
-            payloadType: "application/json",
-            context: {},
-            traceContext: {},
-            traceId: "t12345-second",
-            spanId: "s12345-second",
-            workerQueue: "main",
-            queue: "task/test-task",
-            isTest: false,
-            tags: [],
-          },
-          prisma
-        );
-
-        await setTimeout(500);
-
-        // Dequeue and start the second run
-        const dequeuedSecond = await engine.dequeueFromWorkerQueue({
-          consumerId: "test_12345",
-          workerQueue: "main",
-        });
-        expect(dequeuedSecond.length).toBe(1);
-        assertNonNullable(dequeuedSecond[0]);
-
-        const secondAttempt = await engine.startRunAttempt({
-          runId: dequeuedSecond[0].run.id,
-          snapshotId: dequeuedSecond[0].snapshot.id,
-        });
-        expect(secondAttempt.snapshot.executionStatus).toBe("EXECUTING");
-
-        // Now complete the waitpoint for the first run
-        await engine.completeWaitpoint({
-          id: waitpoint.waitpoint.id,
-        });
-
-        // Wait for the continueRunIfUnblocked to process
-        await setTimeout(500);
-
-        // Verify the first run is now in QUEUED_EXECUTING state
-        const executionDataAfter = await engine.getRunExecutionData({ runId: firstRun.id });
-        expect(executionDataAfter?.snapshot.executionStatus).toBe("QUEUED_EXECUTING");
-        expect(executionDataAfter?.snapshot.description).toBe(
-          "Run can continue, but is waiting for concurrency"
-        );
-
-        // Verify the waitpoint is no longer blocking the first run
-        const runWaitpoint = await prisma.taskRunWaitpoint.findFirst({
-          where: {
-            taskRunId: firstRun.id,
-          },
-          include: {
-            waitpoint: true,
-          },
-        });
-        expect(runWaitpoint).toBeNull();
-
-        // Verify the waitpoint itself is completed
-        const completedWaitpoint = await prisma.waitpoint.findUnique({
-          where: {
-            id: waitpoint.waitpoint.id,
-          },
-        });
-        assertNonNullable(completedWaitpoint);
-        expect(completedWaitpoint.status).toBe("COMPLETED");
-
-        // Create checkpoint after waitpoint completion
-        const checkpointResult = await engine.createCheckpoint({
-          runId: firstRun.id,
-          snapshotId: firstRunData?.snapshot.id!,
-          checkpoint: {
-            type: "DOCKER",
-            reason: "TEST_CHECKPOINT",
-            location: "test-location",
-            imageRef: "test-image-ref",
-          },
-        });
-
-        expect(checkpointResult.ok).toBe(true);
-        const checkpoint = checkpointResult.ok ? checkpointResult.snapshot : null;
-        assertNonNullable(checkpoint);
-        expect(checkpoint.executionStatus).toBe("QUEUED");
-
-        // Complete the second run so the first run can be dequeued
-        const result = await engine.completeRunAttempt({
-          runId: dequeuedSecond[0].run.id,
-          snapshotId: secondAttempt.snapshot.id,
-          completion: {
-            ok: true,
-            id: dequeuedSecond[0].run.id,
-            output: `{"foo":"bar"}`,
-            outputType: "application/json",
-          },
-        });
-
-        await setTimeout(1000);
-
-        // Verify the first run is back in the queue
-        const queuedRun = await engine.dequeueFromWorkerQueue({
-          consumerId: "test_12345",
-          workerQueue: "main",
-        });
-        expect(queuedRun.length).toBe(1);
-        assertNonNullable(queuedRun[0]);
-
-        // Now we can continue the run
-        const continueResult = await engine.continueRunExecution({
-          runId: firstRun.id,
-          snapshotId: queuedRun[0].snapshot.id,
-        });
-
-        expect(continueResult.snapshot.executionStatus).toBe("EXECUTING");
-      } finally {
-        await engine.quit();
-      }
-    }
-  );
-
   containerTest("batchTriggerAndWait resume after checkpoint", async ({ prisma, redisOptions }) => {
     //create environment
     const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");

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

@@ -952,33 +952,6 @@ describe("RunQueue", () => {
         0
       );
       expect(await queue.currentConcurrencyOfEnvironment(authenticatedEnvProd)).toBe(0);
-
-      //reacquire the concurrency
-      await queue.reacquireConcurrency(authenticatedEnvProd.organization.id, message.messageId);
-
-      //check concurrencies are back to what they were before
-      expect(await queue.currentConcurrencyOfQueue(authenticatedEnvProd, messageProd.queue)).toBe(
-        1
-      );
-      expect(await queue.currentConcurrencyOfEnvironment(authenticatedEnvProd)).toBe(1);
-
-      //release the concurrency (with the queue this time)
-      await queue.releaseAllConcurrency(authenticatedEnvProd.organization.id, message.messageId);
-
-      //concurrencies
-      expect(await queue.currentConcurrencyOfQueue(authenticatedEnvProd, messageProd.queue)).toBe(
-        0
-      );
-      expect(await queue.currentConcurrencyOfEnvironment(authenticatedEnvProd)).toBe(0);
-
-      //reacquire the concurrency
-      await queue.reacquireConcurrency(authenticatedEnvProd.organization.id, message.messageId);
-
-      //check concurrencies are back to what they were before
-      expect(await queue.currentConcurrencyOfQueue(authenticatedEnvProd, messageProd.queue)).toBe(
-        1
-      );
-      expect(await queue.currentConcurrencyOfEnvironment(authenticatedEnvProd)).toBe(1);
     } finally {
       try {
         await queue.quit();