fix(run-engine): close guard gaps found in adversarial review

Three hardening changes to the run finalization guard:

The cancel path enqueued the guard after the CANCELED status commit,
so a crash between the two could still strand a waiting parent. The
enqueue now happens before the status write, restoring the write-ahead
invariant on every finalizing path.

A guard that finds a canceled run whose execution has not reached
FINISHED now re-arms itself instead of exiting, so the watcher
survives the PENDING_CANCEL window without resuming the parent early.

The guard now also re-delivers the queue ack, releasing the queue
message and concurrency slot when the inline path died before
acknowledging, and TTL expiry gets the same guard: both the per-run
path (whose worker retry was defeated by the not-PENDING early return)
and the batch path (which swallowed a failed waitpoint-job enqueue)
now arm it before the EXPIRED commit.

Author: matt-aitken

.server-changes/run-finalization-guard.md

@@ -3,4 +3,4 @@ area: webapp
 type: fix
 ---
 
-Runs waiting on triggerAndWait or batchTriggerAndWait can no longer get stuck forever when a database error interrupts a child run's completion. A durable guard now re-delivers the lost completion signal and resumes the waiting run automatically.
+Runs waiting on triggerAndWait or batchTriggerAndWait can no longer get stuck forever when a database error interrupts a child run finishing (completing, failing, being canceled, or expiring). A durable guard now re-delivers the lost completion signal and resumes the waiting run automatically.

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

@@ -427,6 +427,7 @@ export class RunEngine {
     this.ttlSystem = new TtlSystem({
       resources,
       waitpointSystem: this.waitpointSystem,
+      finalizationGuardDelayMs: this.options.finalizationGuardDelayMs,
     });
 
     const ttlWorkerCatalog = createTtlWorkerCatalog({

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

@@ -1454,6 +1454,8 @@ export class RunAttemptSystem {
           });
         }
 
+        await this.#scheduleFinalizationGuard(runId);
+
         const run = await this.$.runStore.cancelRun(
           runId,
           {
@@ -1547,8 +1549,6 @@ export class RunAttemptSystem {
           // finalizeRun is true — fall through to finish the run immediately
         }
 
-        await this.#scheduleFinalizationGuard(runId);
-
         //not executing, so we will actually finish the run
         const newSnapshot = await this.executionSnapshotSystem.createExecutionSnapshot(prisma, {
           run,
@@ -1808,13 +1808,16 @@ export class RunAttemptSystem {
   }
 
   /**
-   * Re-delivers a finished run's finalization side effects: the associated waitpoint's
-   * completion, the parent unblock fan-out, and the batch completion nudge. Safe to run
-   * at-least-once and to race the inline path — every leg is idempotent, and completing
-   * an already-completed waitpoint still re-runs the blocked-run fan-out (which covers a
-   * lost `continueRunIfUnblocked` enqueue). A non-final run means the finish commit
-   * itself never landed; the caller's retry re-runs the whole completion, so there is
-   * nothing to re-deliver.
+   * Re-delivers a finished run's finalization side effects: the queue ack, the
+   * associated waitpoint's completion, the parent unblock fan-out, and the batch
+   * completion nudge. Safe to run at-least-once and to race the inline path — every leg
+   * is idempotent, and completing an already-completed waitpoint still re-runs the
+   * blocked-run fan-out (which covers a lost `continueRunIfUnblocked` enqueue). A
+   * non-final run means the finish commit itself never landed; the caller's retry
+   * re-runs the whole completion, so there is nothing to re-deliver. A canceled run
+   * whose execution has not reached FINISHED re-arms the guard and waits: the
+   * cancellation finalize path owns that window, and completing early would resume the
+   * parent while the child is still winding down.
    */
   public async ensureRunFinalized({ runId }: { runId: string }): Promise<void> {
     return startSpan(this.$.tracer, "ensureRunFinalized", async (span) => {
@@ -1830,6 +1833,7 @@ export class RunAttemptSystem {
             outputType: true,
             error: true,
             batchId: true,
+            runtimeEnvironmentId: true,
             associatedWaitpoint: {
               select: {
                 id: true,
@@ -1863,18 +1867,32 @@ export class RunAttemptSystem {
 
         if (latestSnapshot.executionStatus !== "FINISHED") {
           this.$.logger.info(
-            "ensureRunFinalized: run is canceled but execution has not finished, the cancellation finalize path owns the re-delivery",
+            "ensureRunFinalized: run is canceled but execution has not finished, keeping watch until the cancellation finalize path completes it",
             {
               runId,
               executionStatus: latestSnapshot.executionStatus,
             }
           );
+          await this.#scheduleFinalizationGuard(runId);
           return;
         }
       }
 
       span.setAttribute("runStatus", run.status);
 
+      const env = await this.$.controlPlaneResolver.resolveEnv(run.runtimeEnvironmentId);
+
+      if (env) {
+        await this.$.runQueue.acknowledgeMessage(env.organizationId, runId, {
+          removeFromWorkerQueue: true,
+        });
+      } else {
+        this.$.logger.error("ensureRunFinalized: environment not found, skipping queue ack", {
+          runId,
+          runtimeEnvironmentId: run.runtimeEnvironmentId,
+        });
+      }
+
       if (run.associatedWaitpoint) {
         const wasPending = run.associatedWaitpoint.status === "PENDING";
 

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

@@ -12,15 +12,33 @@ import { boundedIn } from "@trigger.dev/database";
 export type TtlSystemOptions = {
   resources: SystemResources;
   waitpointSystem: WaitpointSystem;
+  finalizationGuardDelayMs?: number;
 };
 
 export class TtlSystem {
   private readonly $: SystemResources;
   private readonly waitpointSystem: WaitpointSystem;
+  private readonly finalizationGuardDelayMs: number;
 
   constructor(private readonly options: TtlSystemOptions) {
     this.$ = options.resources;
     this.waitpointSystem = options.waitpointSystem;
+    this.finalizationGuardDelayMs = options.finalizationGuardDelayMs ?? 60_000;
+  }
+
+  /**
+   * Write-ahead guard for TTL expiry, mirroring the run attempt system's: enqueued
+   * before the EXPIRED commit so a crash or error between that commit and the
+   * waitpoint completion cannot strand a waiting parent, and acked once the inline
+   * side effects succeed.
+   */
+  async #scheduleFinalizationGuard(runId: string): Promise<void> {
+    await this.$.worker.enqueue({
+      id: `ensureRunFinalized:${runId}`,
+      job: "ensureRunFinalized",
+      payload: { runId },
+      availableAt: new Date(Date.now() + this.finalizationGuardDelayMs),
+    });
   }
 
   async expireRun({ runId, tx }: { runId: string; tx?: PrismaClientOrTransaction }) {
@@ -62,6 +80,8 @@ export class TtlSystem {
         raw: `Run expired because the TTL (${run.ttl}) was reached`,
       };
 
+      await this.#scheduleFinalizationGuard(runId);
+
       const updatedRun = await this.$.runStore.expireRun(
         runId,
         {
@@ -121,6 +141,8 @@ export class TtlSystem {
         project: { id: snapshot.projectId },
         environment: { id: snapshot.environmentId },
       });
+
+      await this.$.worker.ack(`ensureRunFinalized:${runId}`);
     });
   }
 
@@ -220,6 +242,10 @@ export class TtlSystem {
         raw: "Run expired because the TTL was reached",
       };
 
+      await pMap(runsToExpire, (run) => this.#scheduleFinalizationGuard(run.id), {
+        concurrency: 10,
+      });
+
       await this.$.runStore.expireRunsBatch(runIdsToExpire, { error, now }, this.$.prisma);
 
       // Process each run: enqueue waitpoint completion jobs and emit events
@@ -262,6 +288,16 @@ export class TtlSystem {
               environment: { id: run.runtimeEnvironmentId },
             });
 
+            /**
+             * Waitpoint completion in this path is delegated to the finishWaitpoint job,
+             * which can still exhaust its retries, so the guard stays armed for runs with
+             * a waiting parent and verifies the completion landed. Runs with no waitpoint
+             * have nothing left to re-deliver, so release their guard now.
+             */
+            if (!run.associatedWaitpoint) {
+              await this.$.worker.ack(`ensureRunFinalized:${run.id}`);
+            }
+
             expired.push(run.id);
           } catch (e) {
             this.$.logger.error("Failed to process expired run", {

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

@@ -28,6 +28,11 @@ describe("RunEngine ensureRunFinalized guard", () => {
         redis: redisOptions,
         masterQueueConsumersDisabled: true,
         processWorkerQueueDebounceMs: 50,
+        ttlSystem: {
+          pollIntervalMs: 100,
+          batchSize: 10,
+          batchMaxWaitMs: 100,
+        },
       },
       runLock: {
         redis: redisOptions,
@@ -52,7 +57,8 @@ describe("RunEngine ensureRunFinalized guard", () => {
   async function setupParentAndChild(
     engine: RunEngine,
     prisma: any,
-    authenticatedEnvironment: any
+    authenticatedEnvironment: any,
+    options?: { childTtl?: string; dequeueChild?: boolean }
   ) {
     const parentTask = "parent-task";
     const childTask = "child-task";
@@ -109,10 +115,16 @@ describe("RunEngine ensureRunFinalized guard", () => {
         resumeParentOnCompletion: true,
         parentTaskRunId: parentRun.id,
         workerQueue: "main",
+        ttl: options?.childTtl,
+        enableFastPath: options?.childTtl ? false : undefined,
       },
       prisma
     );
 
+    if (options?.dequeueChild === false) {
+      return { parentRun, childRun, childAttempt: undefined };
+    }
+
     await setTimeout(500);
     const dequeuedChild = await engine.dequeueFromWorkerQueue({
       consumerId: "test_12345",
@@ -376,6 +388,55 @@ describe("RunEngine ensureRunFinalized guard", () => {
     }
   );
 
+  containerTest(
+    "re-derives a waitpoint completion lost during TTL expiry",
+    async ({ prisma, redisOptions }) => {
+      const engine = createEngine(prisma, redisOptions);
+
+      try {
+        const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
+        const { parentRun, childRun } = await setupParentAndChild(
+          engine,
+          prisma,
+          authenticatedEnvironment,
+          { childTtl: "1s", dequeueChild: false }
+        );
+
+        vi.spyOn(engine.waitpointSystem, "completeWaitpoint").mockRejectedValueOnce(
+          new Error("simulated DB connection loss")
+        );
+
+        await expect(engine.ttlSystem.expireRun({ runId: childRun.id })).rejects.toThrow(
+          "simulated DB connection loss"
+        );
+
+        const childAfterFailure = await prisma.taskRun.findFirstOrThrow({
+          where: { id: childRun.id },
+          include: { associatedWaitpoint: true },
+        });
+        expect(childAfterFailure.status).toBe("EXPIRED");
+        expect(childAfterFailure.associatedWaitpoint?.status).toBe("PENDING");
+
+        await setTimeout(5_000);
+
+        const childAfterGuard = await prisma.taskRun.findFirstOrThrow({
+          where: { id: childRun.id },
+          include: { associatedWaitpoint: true },
+        });
+        expect(childAfterGuard.status).toBe("EXPIRED");
+        expect(childAfterGuard.associatedWaitpoint?.status).toBe("COMPLETED");
+        expect(childAfterGuard.associatedWaitpoint?.outputIsError).toBe(true);
+
+        const parentExecutionData = await engine.getRunExecutionData({ runId: parentRun.id });
+        assertNonNullable(parentExecutionData);
+        expect(parentExecutionData.snapshot.executionStatus).toBe("EXECUTING");
+        expect(parentExecutionData.completedWaitpoints?.length).toBe(1);
+      } finally {
+        await engine.quit();
+      }
+    }
+  );
+
   containerTest(
     "guard is acked on the happy path and never executes",
     async ({ prisma, redisOptions }) => {