test(run-engine): pin the single nested write and the resumed QUEUED

The store-routing counter separated nested creates from standalone ones, so it
can assert the trigger path makes no standalone snapshot write at all rather
than just more writes than before. A new case drives a run through suspend and
resume to prove a re-enqueue still writes its own QUEUED snapshot.

Author: ericallam

internal-packages/run-engine/src/engine/systems/enqueueSystem.test.ts

@@ -50,28 +50,31 @@ function createEngineOptions(redisOptions: any, prisma: any, store?: PostgresRun
  * the routing is observed over real containers without ever mocking prisma or the store.
  */
 class CountingPostgresRunStore extends PostgresRunStore {
-  public snapshotCreates = 0;
+  /** Snapshots written nested in a run create: the trigger path's QUEUED write. */
+  public nestedSnapshotCreates = 0;
+  /** Snapshots written standalone through `createExecutionSnapshot`: every re-enqueue. */
+  public standaloneSnapshotCreates = 0;
 
   override async createExecutionSnapshot(
     input: any,
     tx?: any
   ): ReturnType<PostgresRunStore["createExecutionSnapshot"]> {
-    this.snapshotCreates++;
+    this.standaloneSnapshotCreates++;
     return super.createExecutionSnapshot(input, tx);
   }
 
   override async createRun(
     params: Parameters<PostgresRunStore["createRun"]>[0],
     tx?: any
   ): ReturnType<PostgresRunStore["createRun"]> {
-    this.snapshotCreates++;
+    this.nestedSnapshotCreates++;
     return super.createRun(params, tx);
   }
 }
 
 describe("RunEngine enqueueRun store routing", () => {
   containerTest(
-    "the QUEUED snapshot routes through the store",
+    "the QUEUED snapshot routes through the store as a single nested write",
     async ({ prisma, redisOptions }) => {
       const countingStore = new CountingPostgresRunStore({ prisma, readOnlyPrisma: prisma });
       const engine = new RunEngine(createEngineOptions(redisOptions, prisma, countingStore));
@@ -81,7 +84,8 @@ describe("RunEngine enqueueRun store routing", () => {
         const taskIdentifier = "test-task";
         await setupBackgroundWorker(engine, environment, taskIdentifier);
 
-        const before = countingStore.snapshotCreates;
+        const nestedBefore = countingStore.nestedSnapshotCreates;
+        const standaloneBefore = countingStore.standaloneSnapshotCreates;
 
         const run = await engine.trigger(
           {
@@ -103,7 +107,8 @@ describe("RunEngine enqueueRun store routing", () => {
           prisma
         );
 
-        expect(countingStore.snapshotCreates).toBeGreaterThan(before);
+        expect(countingStore.nestedSnapshotCreates).toBe(nestedBefore + 1);
+        expect(countingStore.standaloneSnapshotCreates).toBe(standaloneBefore);
 
         const latest = await getLatestExecutionSnapshot(prisma, run.id);
         assertNonNullable(latest);

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

@@ -253,4 +253,118 @@ describe("RunEngine trigger() execution snapshots", () => {
       }
     }
   );
+
+  containerTest(
+    "a resumed run still writes its own QUEUED snapshot",
+    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";
+
+        await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier);
+
+        const run = await engine.trigger(
+          {
+            number: 1,
+            friendlyId: "run_1237",
+            environment: authenticatedEnvironment,
+            taskIdentifier,
+            payload: "{}",
+            payloadType: "application/json",
+            context: {},
+            traceContext: {},
+            traceId: "t_collapse_4",
+            spanId: "s_collapse_4",
+            workerQueue: "main",
+            queue: "task/test-task",
+            isTest: false,
+            tags: [],
+          },
+          prisma
+        );
+
+        await setTimeout(500);
+        const dequeued = await engine.dequeueFromWorkerQueue({
+          consumerId: "test_collapse_4",
+          workerQueue: "main",
+        });
+        assertNonNullable(dequeued[0]);
+
+        await engine.startRunAttempt({
+          runId: dequeued[0].run.id,
+          snapshotId: dequeued[0].snapshot.id,
+        });
+
+        const waitpointResult = await engine.createManualWaitpoint({
+          environmentId: authenticatedEnvironment.id,
+          projectId: authenticatedEnvironment.projectId,
+        });
+
+        const blockedResult = await engine.blockRunWithWaitpoint({
+          runId: run.id,
+          waitpoints: waitpointResult.waitpoint.id,
+          projectId: authenticatedEnvironment.projectId,
+          organizationId: authenticatedEnvironment.organizationId,
+        });
+
+        const checkpointResult = await engine.createCheckpoint({
+          runId: run.id,
+          snapshotId: blockedResult.id,
+          checkpoint: {
+            type: "DOCKER",
+            reason: "TEST_CHECKPOINT",
+            location: "test-location",
+            imageRef: "test-image-ref",
+          },
+        });
+        expect(checkpointResult.ok).toBe(true);
+
+        await engine.completeWaitpoint({ id: waitpointResult.waitpoint.id });
+        await setTimeout(500);
+
+        expect(await snapshotStatuses(prisma, run.id)).toEqual([
+          "QUEUED",
+          "PENDING_EXECUTING",
+          "EXECUTING",
+          "EXECUTING_WITH_WAITPOINTS",
+          "SUSPENDED",
+          "QUEUED",
+        ]);
+      } finally {
+        await engine.quit();
+      }
+    }
+  );
 });