Prevent large outputs from overwriting each other

.changeset/spotty-pants-wink.md

@@ -0,0 +1,6 @@
+---
+"trigger.dev": patch
+"@trigger.dev/core": patch
+---
+
+Prevent large outputs from overwriting each other

packages/cli-v3/src/entryPoints/dev-run-controller.ts

@@ -623,7 +623,7 @@ export class DevRunController {
     }).initialize();
 
     logger.debug("executing task run process", {
-      attemptId: execution.attempt.id,
+      attemptNumber: execution.attempt.number,
       runId: execution.run.id,
     });
 

packages/cli-v3/src/entryPoints/dev-run-worker.ts

@@ -7,6 +7,7 @@ import {
   AnyOnStartHookFunction,
   AnyOnSuccessHookFunction,
   apiClientManager,
+  attemptKey,
   clock,
   ExecutorToWorkerMessageCatalog,
   type HandleErrorFunction,
@@ -546,7 +547,7 @@ const heartbeatInterval = parseInt(heartbeatIntervalMs ?? "30000", 10);
 for await (const _ of setInterval(heartbeatInterval)) {
   if (_isRunning && _execution) {
     try {
-      await zodIpc.send("TASK_HEARTBEAT", { id: _execution.attempt.id });
+      await zodIpc.send("TASK_HEARTBEAT", { id: attemptKey(_execution) });
     } catch (err) {
       logError("Failed to send HEARTBEAT message", err);
     }

packages/cli-v3/src/executions/taskRunProcess.ts

@@ -1,4 +1,5 @@
 import {
+  attemptKey,
   CompletedWaitpoint,
   ExecutorToWorkerMessageCatalog,
   MachinePresetResources,
@@ -164,15 +165,17 @@ export class TaskRunProcess {
         TASK_RUN_COMPLETED: async (message) => {
           const { result, execution } = message;
 
-          const promiseStatus = this._attemptStatuses.get(execution.attempt.id);
+          const key = attemptKey(execution);
+
+          const promiseStatus = this._attemptStatuses.get(key);
 
           if (promiseStatus !== "PENDING") {
             return;
           }
 
-          this._attemptStatuses.set(execution.attempt.id, "RESOLVED");
+          this._attemptStatuses.set(key, "RESOLVED");
 
-          const attemptPromise = this._attemptPromises.get(execution.attempt.id);
+          const attemptPromise = this._attemptPromises.get(key);
 
           if (!attemptPromise) {
             return;
@@ -229,10 +232,12 @@ export class TaskRunProcess {
       rejecter = reject;
     });
 
-    this._attemptStatuses.set(params.payload.execution.attempt.id, "PENDING");
+    const key = attemptKey(params.payload.execution);
+
+    this._attemptStatuses.set(key, "PENDING");
 
     // @ts-expect-error - We know that the resolver and rejecter are defined
-    this._attemptPromises.set(params.payload.execution.attempt.id, { resolver, rejecter });
+    this._attemptPromises.set(key, { resolver, rejecter });
 
     const { execution, traceContext, metrics } = params.payload;
 

packages/core/src/v3/idempotencyKeys.ts

@@ -105,7 +105,7 @@ function injectScope(scope: "run" | "attempt" | "global"): string[] {
     }
     case "attempt": {
       if (taskContext?.ctx) {
-        return [taskContext.ctx.attempt.id];
+        return [taskContext.ctx.run.id, taskContext.ctx.attempt.number.toString()];
       }
       break;
     }
@@ -125,3 +125,17 @@ async function generateIdempotencyKey(keyMaterial: string[]) {
     .map((byte) => byte.toString(16).padStart(2, "0"))
     .join("");
 }
+
+type AttemptKeyMaterial = {
+  run: {
+    id: string;
+  };
+  attempt: {
+    number: number;
+  };
+};
+
+/** Creates a unique key for each attempt. */
+export function attemptKey(ctx: AttemptKeyMaterial): string {
+  return `${ctx.run.id}-${ctx.attempt.number}`;
+}

packages/core/src/v3/workers/taskExecutor.ts

@@ -12,6 +12,7 @@ import {
 } from "../errors.js";
 import {
   accessoryAttributes,
+  attemptKey,
   flattenAttributes,
   lifecycleHooks,
   runMetadata,
@@ -235,7 +236,7 @@ export class TaskExecutor {
             const [exportError, finalOutput] = await tryCatch(
               conditionallyExportPacket(
                 stringifiedOutput,
-                `${execution.attempt.id}/output`,
+                `${attemptKey(ctx)}/output`,
                 this._tracer
               )
             );