Fix controller waitpoint resolution, suspendable state, and snapshot race conditions (#2006)

* remove dead code

* rename managed to shared runtime manager

* rename to resolve waitpoint for clarity

* add resolver id helper

* store and correctly resolve waipoints that come in early

* fix ipc message type change

* branded type for resolver ids

* add fixme comments

* remove more unused ipc schemas

* fix entitlement validation when client doesn't exist

* restore hello world reference workspace imports

* runtime manager debug logs

* prefix engine run logs

* managed run logger accepts nested props

* runtime suspendable state and improved logs

* require suspendable state for checkpoints, fix snapshot processing queue

* add terminal link as cli module so we can more easily patch it

* apply cursor patch

* add license info

* remove terminal-link package and add deprecation notice

* remove old patch

* remove terminal-link from sdk

* rename snapshot module

* add cli test tsconfig

* add run logger base type

* add snapshot manager tests

* fix cli builds

* improve QUEUED_EXECUTING test

* changeset

* make testcontainers wait until container has stopped

* require unit tests for publishing again

* avoid mutation during iteration when resolving pending waitpoints

* improve debug logs and make them less noisy

* always update poller snapshot id for accurate logs

* detach task run process handlers

* check for env overrides in a few more places and add verbose logs

* log when poller is still executing when we stop it

* add supervisor to publish workflow

* always print full deploy logs in CI

* Revert "avoid mutation during iteration when resolving pending waitpoints"

This reverts commit 87b0ce1.

* disable pre

* print prerelease script errors

* Revert "disable pre"

This reverts commit 9403409.

* misc fixes

* better debug logs

* add snapshots since methods and route

* prep for snapshots since

* improve deprecated execution detection

* update supervisor and schema

* properly log http server errors

* detect restore after failed snapshot fetch

* run and snapshot id can be overridden

* fix restore detection

* fix deprecation checks, move into snapshot manager

* less logs

* rename snapshot manager stop

* restore detection was moved into snapshot manager

* fix notifier logs

* make runtime manager status a debug log

* no need to attach runtime status twice

* findUnique -> findFirst

* sort snapshots by created at everywhere

Author: nicktrn

.changeset/plenty-dolphins-act.md

@@ -0,0 +1,8 @@
+---
+"trigger.dev": patch
+"@trigger.dev/core": patch
+---
+
+- Correctly resolve waitpoints that come in early
+- Ensure correct state before requesting suspension
+- Fix race conditions in snapshot processing

.changeset/sweet-dolphins-invent.md

@@ -0,0 +1,5 @@
+---
+"trigger.dev": patch
+---
+
+Always print full deploy logs in CI

apps/supervisor/src/workloadServer/index.ts

@@ -17,6 +17,7 @@ import {
   WorkloadRunAttemptStartRequestBody,
   type WorkloadRunAttemptStartResponseBody,
   type WorkloadRunLatestSnapshotResponseBody,
+  WorkloadRunSnapshotsSinceResponseBody,
   type WorkloadServerToClientEvents,
   type WorkloadSuspendRunResponseBody,
 } from "@trigger.dev/core/v3/workers";
@@ -341,6 +342,31 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
           } satisfies WorkloadRunLatestSnapshotResponseBody);
         },
       })
+      .route(
+        "/api/v1/workload-actions/runs/:runFriendlyId/snapshots/since/:snapshotFriendlyId",
+        "GET",
+        {
+          paramsSchema: WorkloadActionParams,
+          handler: async ({ req, reply, params }) => {
+            const sinceSnapshotResponse = await this.workerClient.getSnapshotsSince(
+              params.runFriendlyId,
+              params.snapshotFriendlyId,
+              this.runnerIdFromRequest(req)
+            );
+
+            if (!sinceSnapshotResponse.success) {
+              console.error("Failed to get snapshots since", {
+                runId: params.runFriendlyId,
+                error: sinceSnapshotResponse.error,
+              });
+              reply.empty(500);
+              return;
+            }
+
+            reply.json(sinceSnapshotResponse.data satisfies WorkloadRunSnapshotsSinceResponseBody);
+          },
+        }
+      )
       .route("/api/v1/workload-actions/runs/:runFriendlyId/logs/debug", "POST", {
         paramsSchema: WorkloadActionParams.pick({ runFriendlyId: true }),
         bodySchema: WorkloadDebugLogRequestBody,

apps/webapp/app/routes/engine.v1.worker-actions.runs.$runFriendlyId.snapshots.since.$snapshotId.ts

@@ -0,0 +1,30 @@
+import { json, TypedResponse } from "@remix-run/server-runtime";
+import { WorkerApiRunSnapshotsSinceResponseBody } from "@trigger.dev/core/v3/workers";
+import { z } from "zod";
+import { createLoaderWorkerApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+
+export const loader = createLoaderWorkerApiRoute(
+  {
+    params: z.object({
+      runFriendlyId: z.string(),
+      snapshotId: z.string(),
+    }),
+  },
+  async ({
+    authenticatedWorker,
+    params,
+  }): Promise<TypedResponse<WorkerApiRunSnapshotsSinceResponseBody>> => {
+    const { runFriendlyId, snapshotId } = params;
+
+    const snapshots = await authenticatedWorker.getSnapshotsSince({
+      runFriendlyId,
+      snapshotId,
+    });
+
+    if (!snapshots) {
+      throw new Error("Failed to retrieve snapshots since given snapshot");
+    }
+
+    return json({ snapshots });
+  }
+);

apps/webapp/app/v3/runEngineHandlers.server.ts

@@ -401,7 +401,7 @@ export function registerRunEngineEventBusHandlers() {
   engine.eventBus.on("executionSnapshotCreated", async ({ time, run, snapshot }) => {
     const eventResult = await recordRunDebugLog(
       run.id,
-      `${snapshot.executionStatus} - ${snapshot.description}`,
+      `[engine] ${snapshot.executionStatus} - ${snapshot.description}`,
       {
         attributes: {
           properties: {
@@ -450,6 +450,7 @@ export function registerRunEngineEventBusHandlers() {
       // Record notification event
       const eventResult = await recordRunDebugLog(
         run.id,
+        // don't prefix this with [engine] - "run:notify" is the correct prefix
         `run:notify platform -> supervisor: ${snapshot.executionStatus}`,
         {
           attributes: {
@@ -479,6 +480,7 @@ export function registerRunEngineEventBusHandlers() {
       // Record notification event
       const eventResult = await recordRunDebugLog(
         run.id,
+        // don't prefix this with [engine] - "run:notify" is the correct prefix
         `run:notify ERROR platform -> supervisor: ${snapshot.executionStatus}`,
         {
           attributes: {
@@ -505,7 +507,7 @@ export function registerRunEngineEventBusHandlers() {
   engine.eventBus.on("incomingCheckpointDiscarded", async ({ time, run, snapshot, checkpoint }) => {
     const eventResult = await recordRunDebugLog(
       run.id,
-      `Checkpoint discarded: ${checkpoint.discardReason}`,
+      `[engine] Checkpoint discarded: ${checkpoint.discardReason}`,
       {
         attributes: {
           properties: {

apps/webapp/app/v3/services/worker/workerGroupTokenService.server.ts

@@ -759,6 +759,19 @@ export class AuthenticatedWorkerInstance extends WithRunEngine {
     });
   }
 
+  async getSnapshotsSince({
+    runFriendlyId,
+    snapshotId,
+  }: {
+    runFriendlyId: string;
+    snapshotId: string;
+  }) {
+    return await this._engine.getSnapshotsSince({
+      runId: fromFriendlyId(runFriendlyId),
+      snapshotId: fromFriendlyId(snapshotId),
+    });
+  }
+
   toJSON(): WorkerGroupTokenAuthenticationResponse {
     if (this.type === WorkerInstanceGroupType.MANAGED) {
       return {

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

@@ -43,6 +43,8 @@ import { EnqueueSystem } from "./systems/enqueueSystem.js";
 import {
   ExecutionSnapshotSystem,
   getLatestExecutionSnapshot,
+  getExecutionSnapshotsSince,
+  executionDataFromSnapshot,
 } from "./systems/executionSnapshotSystem.js";
 import { PendingVersionSystem } from "./systems/pendingVersionSystem.js";
 import { ReleaseConcurrencySystem } from "./systems/releaseConcurrencySystem.js";
@@ -1100,43 +1102,31 @@ export class RunEngine {
     const prisma = tx ?? this.prisma;
     try {
       const snapshot = await getLatestExecutionSnapshot(prisma, runId);
+      return executionDataFromSnapshot(snapshot);
+    } catch (e) {
+      this.logger.error("Failed to getRunExecutionData", {
+        message: e instanceof Error ? e.message : e,
+      });
+      return null;
+    }
+  }
 
-      const executionData: RunExecutionData = {
-        version: "1" as const,
-        snapshot: {
-          id: snapshot.id,
-          friendlyId: snapshot.friendlyId,
-          executionStatus: snapshot.executionStatus,
-          description: snapshot.description,
-        },
-        run: {
-          id: snapshot.runId,
-          friendlyId: snapshot.runFriendlyId,
-          status: snapshot.runStatus,
-          attemptNumber: snapshot.attemptNumber ?? undefined,
-        },
-        batch: snapshot.batchId
-          ? {
-              id: snapshot.batchId,
-              friendlyId: BatchId.toFriendlyId(snapshot.batchId),
-            }
-          : undefined,
-        checkpoint: snapshot.checkpoint
-          ? {
-              id: snapshot.checkpoint.id,
-              friendlyId: snapshot.checkpoint.friendlyId,
-              type: snapshot.checkpoint.type,
-              location: snapshot.checkpoint.location,
-              imageRef: snapshot.checkpoint.imageRef,
-              reason: snapshot.checkpoint.reason ?? undefined,
-            }
-          : undefined,
-        completedWaitpoints: snapshot.completedWaitpoints,
-      };
+  async getSnapshotsSince({
+    runId,
+    snapshotId,
+    tx,
+  }: {
+    runId: string;
+    snapshotId: string;
+    tx?: PrismaClientOrTransaction;
+  }): Promise<RunExecutionData[] | null> {
+    const prisma = tx ?? this.prisma;
 
-      return executionData;
+    try {
+      const snapshots = await getExecutionSnapshotsSince(prisma, runId, snapshotId);
+      return snapshots.map(executionDataFromSnapshot);
     } catch (e) {
-      this.logger.error("Failed to getRunExecutionData", {
+      this.logger.error("Failed to getSnapshotsSince", {
         message: e instanceof Error ? e.message : e,
       });
       return null;
@@ -1158,9 +1148,6 @@ export class RunEngine {
     }
   }
 
-  //#endregion
-
-  //#region Heartbeat
   async #handleStalledSnapshot({
     runId,
     snapshotId,

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

@@ -409,6 +409,7 @@ export class DequeueSystem {
                     friendlyId: newSnapshot.friendlyId,
                     executionStatus: newSnapshot.executionStatus,
                     description: newSnapshot.description,
+                    createdAt: newSnapshot.createdAt,
                   },
                   image: result.deployment?.imageReference ?? undefined,
                   checkpoint: newSnapshot.checkpoint ?? undefined,

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

@@ -1,4 +1,4 @@
-import { CompletedWaitpoint, ExecutionResult } from "@trigger.dev/core/v3";
+import { CompletedWaitpoint, ExecutionResult, RunExecutionData } from "@trigger.dev/core/v3";
 import { BatchId, RunId, SnapshotId } from "@trigger.dev/core/v3/isomorphic";
 import {
   Prisma,
@@ -17,31 +17,23 @@ export type ExecutionSnapshotSystemOptions = {
   heartbeatTimeouts: HeartbeatTimeouts;
 };
 
-export interface LatestExecutionSnapshot extends TaskRunExecutionSnapshot {
+export interface EnhancedExecutionSnapshot extends TaskRunExecutionSnapshot {
   friendlyId: string;
   runFriendlyId: string;
   checkpoint: TaskRunCheckpoint | null;
   completedWaitpoints: CompletedWaitpoint[];
 }
 
-/* Gets the most recent valid snapshot for a run */
-export async function getLatestExecutionSnapshot(
-  prisma: PrismaClientOrTransaction,
-  runId: string
-): Promise<LatestExecutionSnapshot> {
-  const snapshot = await prisma.taskRunExecutionSnapshot.findFirst({
-    where: { runId, isValid: true },
-    include: {
-      completedWaitpoints: true,
-      checkpoint: true,
-    },
-    orderBy: { createdAt: "desc" },
-  });
-
-  if (!snapshot) {
-    throw new Error(`No execution snapshot found for TaskRun ${runId}`);
-  }
+type ExecutionSnapshotWithCheckAndWaitpoints = Prisma.TaskRunExecutionSnapshotGetPayload<{
+  include: {
+    checkpoint: true;
+    completedWaitpoints: true;
+  };
+}>;
 
+function enhanceExecutionSnapshot(
+  snapshot: ExecutionSnapshotWithCheckAndWaitpoints
+): EnhancedExecutionSnapshot {
   return {
     ...snapshot,
     friendlyId: SnapshotId.toFriendlyId(snapshot.id),
@@ -99,6 +91,27 @@ export async function getLatestExecutionSnapshot(
   };
 }
 
+/* Gets the most recent valid snapshot for a run */
+export async function getLatestExecutionSnapshot(
+  prisma: PrismaClientOrTransaction,
+  runId: string
+): Promise<EnhancedExecutionSnapshot> {
+  const snapshot = await prisma.taskRunExecutionSnapshot.findFirst({
+    where: { runId, isValid: true },
+    include: {
+      completedWaitpoints: true,
+      checkpoint: true,
+    },
+    orderBy: { createdAt: "desc" },
+  });
+
+  if (!snapshot) {
+    throw new Error(`No execution snapshot found for TaskRun ${runId}`);
+  }
+
+  return enhanceExecutionSnapshot(snapshot);
+}
+
 export async function getExecutionSnapshotCompletedWaitpoints(
   prisma: PrismaClientOrTransaction,
   snapshotId: string
@@ -131,6 +144,7 @@ export function executionResultFromSnapshot(snapshot: TaskRunExecutionSnapshot):
       friendlyId: SnapshotId.toFriendlyId(snapshot.id),
       executionStatus: snapshot.executionStatus,
       description: snapshot.description,
+      createdAt: snapshot.createdAt,
     },
     run: {
       id: snapshot.runId,
@@ -141,6 +155,73 @@ export function executionResultFromSnapshot(snapshot: TaskRunExecutionSnapshot):
   };
 }
 
+export function executionDataFromSnapshot(snapshot: EnhancedExecutionSnapshot): RunExecutionData {
+  return {
+    version: "1" as const,
+    snapshot: {
+      id: snapshot.id,
+      friendlyId: snapshot.friendlyId,
+      executionStatus: snapshot.executionStatus,
+      description: snapshot.description,
+      createdAt: snapshot.createdAt,
+    },
+    run: {
+      id: snapshot.runId,
+      friendlyId: snapshot.runFriendlyId,
+      status: snapshot.runStatus,
+      attemptNumber: snapshot.attemptNumber ?? undefined,
+    },
+    batch: snapshot.batchId
+      ? {
+          id: snapshot.batchId,
+          friendlyId: BatchId.toFriendlyId(snapshot.batchId),
+        }
+      : undefined,
+    checkpoint: snapshot.checkpoint
+      ? {
+          id: snapshot.checkpoint.id,
+          friendlyId: snapshot.checkpoint.friendlyId,
+          type: snapshot.checkpoint.type,
+          location: snapshot.checkpoint.location,
+          imageRef: snapshot.checkpoint.imageRef,
+          reason: snapshot.checkpoint.reason ?? undefined,
+        }
+      : undefined,
+    completedWaitpoints: snapshot.completedWaitpoints,
+  };
+}
+
+export async function getExecutionSnapshotsSince(
+  prisma: PrismaClientOrTransaction,
+  runId: string,
+  sinceSnapshotId: string
+): Promise<EnhancedExecutionSnapshot[]> {
+  // Find the createdAt of the sinceSnapshotId
+  const sinceSnapshot = await prisma.taskRunExecutionSnapshot.findFirst({
+    where: { id: sinceSnapshotId },
+    select: { createdAt: true },
+  });
+
+  if (!sinceSnapshot) {
+    throw new Error(`No execution snapshot found for id ${sinceSnapshotId}`);
+  }
+
+  const snapshots = await prisma.taskRunExecutionSnapshot.findMany({
+    where: {
+      runId,
+      isValid: true,
+      createdAt: { gt: sinceSnapshot.createdAt },
+    },
+    include: {
+      completedWaitpoints: true,
+      checkpoint: true,
+    },
+    orderBy: { createdAt: "asc" },
+  });
+
+  return snapshots.map(enhanceExecutionSnapshot);
+}
+
 export class ExecutionSnapshotSystem {
   private readonly $: SystemResources;
   private readonly heartbeatTimeouts: HeartbeatTimeouts;

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

@@ -892,6 +892,7 @@ export class RunAttemptSystem {
           friendlyId: newSnapshot.friendlyId,
           executionStatus: newSnapshot.executionStatus,
           description: newSnapshot.description,
+          createdAt: newSnapshot.createdAt,
         },
         run: {
           id: newSnapshot.runId,