feat(engine): Improve execution stalls troubleshooting, align dev and prod behavior, adding heartbeats.yield utility

Author: ericallam

apps/webapp/app/env.server.ts

@@ -519,8 +519,8 @@ const EnvironmentSchema = z
     RUN_ENGINE_WORKER_IMMEDIATE_POLL_INTERVAL: z.coerce.number().int().default(100),
     RUN_ENGINE_TIMEOUT_PENDING_EXECUTING: z.coerce.number().int().default(60_000),
     RUN_ENGINE_TIMEOUT_PENDING_CANCEL: z.coerce.number().int().default(60_000),
-    RUN_ENGINE_TIMEOUT_EXECUTING: z.coerce.number().int().default(60_000),
-    RUN_ENGINE_TIMEOUT_EXECUTING_WITH_WAITPOINTS: z.coerce.number().int().default(60_000),
+    RUN_ENGINE_TIMEOUT_EXECUTING: z.coerce.number().int().default(600_000), // 10 minutes
+    RUN_ENGINE_TIMEOUT_EXECUTING_WITH_WAITPOINTS: z.coerce.number().int().default(600_000), // 10 minutes
     RUN_ENGINE_TIMEOUT_SUSPENDED: z.coerce
       .number()
       .int()

docs/troubleshooting.mdx

@@ -181,6 +181,38 @@ View the [rate limits](/limits) page for more information.
 
 This can happen in different situations, for example when using plain strings as idempotency keys. Support for `Crypto` without a special flag was added in Node `v19.0.0`. You will have to upgrade Node - we recommend even-numbered major releases, e.g. `v20` or `v22`. Alternatively, you can switch from plain strings to the `idempotencyKeys.create` SDK function. [Read the guide](/idempotency).
 
+### Task run stalled executing
+
+If you see a `TASK_RUN_STALLED_EXECUTING` error it means that we didn't receive a heartbeat from your task before the stall timeout. We automatically heartbeat runs every 30 seconds, and the heartbeat timeout is 10 minutes.
+
+<Note>
+
+If this was a dev run, then most likely the `trigger.dev dev` CLI was stopped, and it wasn't an issue with your code.
+
+</Note>
+
+These errors can happen when code inside your task is blocking the event loop for too long. The most likely cause would be an accidental infinite loop. It could also be a CPU-heavy operation that's blocking the event loop, like nested loops with very large arrays. We recommend reading the [Don't Block the Event Loop](https://nodejs.org/en/learn/asynchronous-work/dont-block-the-event-loop) guide from Node.js for common patterns that can cause this.
+
+If you are doing a continuous CPU-heavy task, then we recommend you try using our `heartbeats.yield` function to automatically yield to the event loop periodically:
+
+```ts
+import { heartbeats } from "@trigger.dev/sdk";
+
+// code inside your task
+for (const row of bigDataset) {
+  await heartbeats.yield(); // safe to call every iteration, we will only actually yield when we need to
+  process(row); // this is a synchronous operation
+}
+```
+
+<Note>
+
+You could also offload the CPU-heavy work to a Node.js worker thread, but this is more complex to setup currently. We are planning on adding support for this in the future.
+
+</Note>
+
+If the above doesn't work, then we recommend you try increasing the machine size of your task. See our [machines guide](/machines) for more information.
+
 ## Framework specific issues
 
 ### NestJS swallows all errors/exceptions

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

@@ -1,8 +1,11 @@
 import { assertExhaustive } from "@trigger.dev/core";
 import { TaskRunError } from "@trigger.dev/core/v3";
-import { TaskRunStatus } from "@trigger.dev/database";
+import { RuntimeEnvironmentType, TaskRunStatus } from "@trigger.dev/database";
 
-export function runStatusFromError(error: TaskRunError): TaskRunStatus {
+export function runStatusFromError(
+  error: TaskRunError,
+  environmentType: RuntimeEnvironmentType
+): TaskRunStatus {
   if (error.type !== "INTERNAL_ERROR") {
     return "COMPLETED_WITH_ERRORS";
   }
@@ -21,6 +24,15 @@ export function runStatusFromError(error: TaskRunError): TaskRunStatus {
       return "CANCELED";
     case "MAX_DURATION_EXCEEDED":
       return "TIMED_OUT";
+    case "TASK_RUN_STALLED_EXECUTING":
+    case "TASK_RUN_STALLED_EXECUTING_WITH_WAITPOINTS": {
+      if (environmentType === "DEVELOPMENT") {
+        return "CANCELED";
+      }
+
+      return "COMPLETED_WITH_ERRORS";
+    }
+
     case "TASK_PROCESS_OOM_KILLED":
     case "TASK_PROCESS_MAYBE_OOM_KILLED":
     case "TASK_PROCESS_SIGSEGV":
@@ -40,8 +52,6 @@ export function runStatusFromError(error: TaskRunError): TaskRunStatus {
     case "TASK_DEQUEUED_INVALID_STATE":
     case "TASK_DEQUEUED_QUEUE_NOT_FOUND":
     case "TASK_RUN_DEQUEUED_MAX_RETRIES":
-    case "TASK_RUN_STALLED_EXECUTING":
-    case "TASK_RUN_STALLED_EXECUTING_WITH_WAITPOINTS":
     case "TASK_HAS_N0_EXECUTION_SNAPSHOT":
     case "GRACEFUL_EXIT_TIMEOUT":
     case "POD_EVICTED":

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

@@ -8,6 +8,7 @@ import {
   CreateCheckpointResult,
   DequeuedMessage,
   ExecutionResult,
+  formatDurationMilliseconds,
   RunExecutionData,
   StartRunAttemptResult,
   TaskRunContext,
@@ -212,6 +213,8 @@ export class RunEngine {
     });
 
     if (!options.worker.disabled) {
+      console.log("✅ Starting run engine worker");
+
       this.worker.start();
     }
 
@@ -1190,23 +1193,6 @@ export class RunEngine {
         snapshot: latestSnapshot,
       });
 
-      // For dev, we just cancel runs that are stuck
-      if (latestSnapshot.environmentType === "DEVELOPMENT") {
-        this.logger.log("RunEngine.#handleStalledSnapshot() cancelling DEV run", {
-          runId,
-          snapshot: latestSnapshot,
-        });
-
-        await this.cancelRun({
-          runId: latestSnapshot.runId,
-          finalizeRun: true,
-          reason:
-            "Run was disconnected, check you're running the CLI dev command and your network connection is healthy.",
-          tx,
-        });
-        return;
-      }
-
       switch (latestSnapshot.executionStatus) {
         case "RUN_CREATED": {
           throw new NotImplementedError("There shouldn't be a heartbeat for RUN_CREATED");
@@ -1265,7 +1251,19 @@ export class RunEngine {
         case "EXECUTING_WITH_WAITPOINTS": {
           const retryDelay = 250;
 
-          //todo call attemptFailed and force requeuing
+          const timeoutDuration =
+            latestSnapshot.executionStatus === "EXECUTING"
+              ? formatDurationMilliseconds(this.heartbeatTimeouts.EXECUTING)
+              : formatDurationMilliseconds(this.heartbeatTimeouts.EXECUTING_WITH_WAITPOINTS);
+
+          // Dev runs don't retry, because the vast majority of the time these snapshots stall because
+          // they have quit the CLI
+          const shouldRetry = latestSnapshot.environmentType !== "DEVELOPMENT";
+          const errorMessage =
+            latestSnapshot.environmentType === "DEVELOPMENT"
+              ? `Run timed out after ${timeoutDuration} due to missing heartbeats (sent every 30s). Check if your \`trigger.dev dev\` CLI is still running, or if CPU-heavy work is blocking the main thread.`
+              : `Run timed out after ${timeoutDuration} due to missing heartbeats (sent every 30s). This typically happens when CPU-heavy work blocks the main thread.`;
+
           await this.runAttemptSystem.attemptFailed({
             runId,
             snapshotId: latestSnapshot.id,
@@ -1278,13 +1276,15 @@ export class RunEngine {
                   latestSnapshot.executionStatus === "EXECUTING"
                     ? "TASK_RUN_STALLED_EXECUTING"
                     : "TASK_RUN_STALLED_EXECUTING_WITH_WAITPOINTS",
-                message: `Run stalled while executing. This can happen when the run becomes unresponsive, for example because the CPU is overloaded.`,
-              },
-              retry: {
-                //250ms in the future
-                timestamp: Date.now() + retryDelay,
-                delay: retryDelay,
+                message: errorMessage,
               },
+              retry: shouldRetry
+                ? {
+                    //250ms in the future
+                    timestamp: Date.now() + retryDelay,
+                    delay: retryDelay,
+                  }
+                : undefined,
             },
             forceRequeue: true,
             tx: prisma,

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

@@ -51,6 +51,7 @@ import { RunEngineOptions } from "../types.js";
 import { BatchSystem } from "./batchSystem.js";
 import { DelayedRunSystem } from "./delayedRunSystem.js";
 import {
+  EnhancedExecutionSnapshot,
   executionResultFromSnapshot,
   ExecutionSnapshotSystem,
   getLatestExecutionSnapshot,
@@ -690,6 +691,10 @@ export class RunAttemptSystem {
             throw new ServiceValidationError("Snapshot ID doesn't match the latest snapshot", 400);
           }
 
+          if (latestSnapshot.executionStatus === "FINISHED") {
+            throw new ServiceValidationError("Run is already finished", 400);
+          }
+
           span.setAttribute("completionStatus", completion.ok);
 
           const completedAt = new Date();
@@ -843,6 +848,10 @@ export class RunAttemptSystem {
             throw new ServiceValidationError("Snapshot ID doesn't match the latest snapshot", 400);
           }
 
+          if (latestSnapshot.executionStatus === "FINISHED") {
+            throw new ServiceValidationError("Run is already finished", 400);
+          }
+
           span.setAttribute("completionStatus", completion.ok);
 
           //remove waitpoints blocking the run
@@ -923,7 +932,7 @@ export class RunAttemptSystem {
             case "fail_run": {
               return await this.#permanentlyFailRun({
                 runId,
-                snapshotId,
+                latestSnapshot,
                 failedAt,
                 error: retryResult.sanitizedError,
                 workerId,
@@ -1443,14 +1452,14 @@ export class RunAttemptSystem {
 
   async #permanentlyFailRun({
     runId,
-    snapshotId,
+    latestSnapshot,
     failedAt,
     error,
     workerId,
     runnerId,
   }: {
     runId: string;
-    snapshotId?: string;
+    latestSnapshot: EnhancedExecutionSnapshot;
     failedAt: Date;
     error: TaskRunError;
     workerId?: string;
@@ -1459,7 +1468,7 @@ export class RunAttemptSystem {
     const prisma = this.$.prisma;
 
     return startSpan(this.$.tracer, "permanentlyFailRun", async (span) => {
-      const status = runStatusFromError(error);
+      const status = runStatusFromError(error, latestSnapshot.environmentType);
 
       //run permanently failed
       const run = await prisma.taskRun.update({
@@ -1512,7 +1521,7 @@ export class RunAttemptSystem {
           executionStatus: "FINISHED",
           description: "Run failed",
         },
-        previousSnapshotId: snapshotId,
+        previousSnapshotId: latestSnapshot.id,
         environmentId: run.runtimeEnvironment.id,
         environmentType: run.runtimeEnvironment.type,
         projectId: run.runtimeEnvironment.project.id,