Serialize metadata to prevent invalid data from breaking run completions

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

@@ -2,6 +2,7 @@ import { startSpan } from "@internal/tracing";
 import {
   CompleteRunAttemptResult,
   ExecutionResult,
+  FlushedRunMetadata,
   GitMeta,
   StartRunAttemptResult,
   TaskRunError,
@@ -35,6 +36,7 @@ import {
 import { ReleaseConcurrencySystem } from "./releaseConcurrencySystem.js";
 import { SystemResources } from "./systems.js";
 import { WaitpointSystem } from "./waitpointSystem.js";
+import { tryCatch } from "@trigger.dev/core/utils";
 
 export type RunAttemptSystemOptions = {
   resources: SystemResources;
@@ -386,15 +388,7 @@ export class RunAttemptSystem {
     workerId?: string;
     runnerId?: string;
   }): Promise<CompleteRunAttemptResult> {
-    if (completion.metadata) {
-      this.$.eventBus.emit("runMetadataUpdated", {
-        time: new Date(),
-        run: {
-          id: runId,
-          metadata: completion.metadata,
-        },
-      });
-    }
+    await this.#notifyMetadataUpdated(runId, completion);
 
     switch (completion.ok) {
       case true: {
@@ -1314,4 +1308,56 @@ export class RunAttemptSystem {
 
     return taskRun?.runtimeEnvironment;
   }
+
+  async #notifyMetadataUpdated(runId: string, completion: TaskRunExecutionResult) {
+    if (completion.metadata) {
+      this.$.eventBus.emit("runMetadataUpdated", {
+        time: new Date(),
+        run: {
+          id: runId,
+          metadata: completion.metadata,
+        },
+      });
+
+      return;
+    }
+
+    if (completion.flushedMetadata) {
+      const [packetError, packet] = await tryCatch(parsePacket(completion.flushedMetadata));
+
+      if (!packet) {
+        return;
+      }
+
+      if (packetError) {
+        this.$.logger.error("RunEngine.completeRunAttempt(): failed to parse flushed metadata", {
+          runId,
+          flushedMetadata: completion.flushedMetadata,
+          error: packetError,
+        });
+
+        return;
+      }
+
+      const metadata = FlushedRunMetadata.safeParse(packet);
+
+      if (!metadata.success) {
+        this.$.logger.error("RunEngine.completeRunAttempt(): failed to parse flushed metadata", {
+          runId,
+          flushedMetadata: completion.flushedMetadata,
+          error: metadata.error,
+        });
+
+        return;
+      }
+
+      this.$.eventBus.emit("runMetadataUpdated", {
+        time: new Date(),
+        run: {
+          id: runId,
+          metadata: metadata.data,
+        },
+      });
+    }
+  }
 }

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

@@ -351,7 +351,7 @@ const zodIpc = new ZodIpcConnection({
             usage: {
               durationMs: 0,
             },
-            metadata: runMetadataManager.stopAndReturnLastFlush(),
+            flushedMetadata: await runMetadataManager.stopAndReturnLastFlush(),
           },
         });
 
@@ -382,7 +382,7 @@ const zodIpc = new ZodIpcConnection({
               usage: {
                 durationMs: 0,
               },
-              metadata: runMetadataManager.stopAndReturnLastFlush(),
+              flushedMetadata: await runMetadataManager.stopAndReturnLastFlush(),
             },
           });
 
@@ -447,7 +447,7 @@ const zodIpc = new ZodIpcConnection({
                 usage: {
                   durationMs: 0,
                 },
-                metadata: runMetadataManager.stopAndReturnLastFlush(),
+                flushedMetadata: await runMetadataManager.stopAndReturnLastFlush(),
               },
             });
 
@@ -473,7 +473,7 @@ const zodIpc = new ZodIpcConnection({
               usage: {
                 durationMs: 0,
               },
-              metadata: runMetadataManager.stopAndReturnLastFlush(),
+              flushedMetadata: await runMetadataManager.stopAndReturnLastFlush(),
             },
           });
 
@@ -518,7 +518,7 @@ const zodIpc = new ZodIpcConnection({
                 usage: {
                   durationMs: usageSample.cpuTime,
                 },
-                metadata: runMetadataManager.stopAndReturnLastFlush(),
+                flushedMetadata: await runMetadataManager.stopAndReturnLastFlush(),
               },
             });
           }
@@ -544,7 +544,7 @@ const zodIpc = new ZodIpcConnection({
             usage: {
               durationMs: 0,
             },
-            metadata: runMetadataManager.stopAndReturnLastFlush(),
+            flushedMetadata: await runMetadataManager.stopAndReturnLastFlush(),
           },
         });
       }

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

@@ -350,7 +350,7 @@ const zodIpc = new ZodIpcConnection({
             usage: {
               durationMs: 0,
             },
-            metadata: runMetadataManager.stopAndReturnLastFlush(),
+            flushedMetadata: await runMetadataManager.stopAndReturnLastFlush(),
           },
         });
 
@@ -381,7 +381,7 @@ const zodIpc = new ZodIpcConnection({
               usage: {
                 durationMs: 0,
               },
-              metadata: runMetadataManager.stopAndReturnLastFlush(),
+              flushedMetadata: await runMetadataManager.stopAndReturnLastFlush(),
             },
           });
 
@@ -444,7 +444,7 @@ const zodIpc = new ZodIpcConnection({
                 usage: {
                   durationMs: 0,
                 },
-                metadata: runMetadataManager.stopAndReturnLastFlush(),
+                flushedMetadata: await runMetadataManager.stopAndReturnLastFlush(),
               },
             });
 
@@ -472,7 +472,7 @@ const zodIpc = new ZodIpcConnection({
               usage: {
                 durationMs: 0,
               },
-              metadata: runMetadataManager.stopAndReturnLastFlush(),
+              flushedMetadata: await runMetadataManager.stopAndReturnLastFlush(),
             },
           });
 
@@ -517,7 +517,7 @@ const zodIpc = new ZodIpcConnection({
                 usage: {
                   durationMs: usageSample.cpuTime,
                 },
-                metadata: runMetadataManager.stopAndReturnLastFlush(),
+                flushedMetadata: await runMetadataManager.stopAndReturnLastFlush(),
               },
             });
           }
@@ -544,7 +544,7 @@ const zodIpc = new ZodIpcConnection({
             usage: {
               durationMs: 0,
             },
-            metadata: runMetadataManager.stopAndReturnLastFlush(),
+            flushedMetadata: await runMetadataManager.stopAndReturnLastFlush(),
           },
         });
       }

packages/core/src/v3/runMetadata/manager.ts

@@ -7,6 +7,7 @@ import { MetadataStream } from "./metadataStream.js";
 import { applyMetadataOperations, collapseOperations } from "./operations.js";
 import { RunMetadataManager, RunMetadataUpdater } from "./types.js";
 import { AsyncIterableStream } from "../streams/asyncIterableStream.js";
+import { IOPacket, stringifyIO } from "../utils/ioSerialization.js";
 
 const MAXIMUM_ACTIVE_STREAMS = 5;
 const MAXIMUM_TOTAL_STREAMS = 10;
@@ -422,23 +423,27 @@ export class StandardMetadataManager implements RunMetadataManager {
     }
   }
 
-  stopAndReturnLastFlush(): FlushedRunMetadata | undefined {
+  async stopAndReturnLastFlush(): Promise<IOPacket> {
     this.stopPeriodicFlush();
     this.isFlushing = true;
 
     if (!this.#needsFlush()) {
-      return;
+      return { dataType: "application/json" };
     }
 
     const operations = Array.from(this.queuedOperations);
     const parentOperations = Array.from(this.queuedParentOperations);
     const rootOperations = Array.from(this.queuedRootOperations);
 
-    return {
+    const data = {
       operations: collapseOperations(operations),
       parentOperations: collapseOperations(parentOperations),
       rootOperations: collapseOperations(rootOperations),
     };
+
+    const packet = await stringifyIO(data);
+
+    return packet;
   }
 
   #needsFlush(): boolean {

packages/core/src/v3/schemas/common.ts

@@ -376,7 +376,15 @@ export const TaskRunFailedExecutionResult = z.object({
   usage: TaskRunExecutionUsage.optional(),
   // Optional for now for backwards compatibility
   taskIdentifier: z.string().optional(),
+  // This is deprecated, use flushedMetadata instead
   metadata: FlushedRunMetadata.optional(),
+  // This is the new way to flush metadata
+  flushedMetadata: z
+    .object({
+      data: z.string().optional(),
+      dataType: z.string(),
+    })
+    .optional(),
 });
 
 export type TaskRunFailedExecutionResult = z.infer<typeof TaskRunFailedExecutionResult>;
@@ -389,7 +397,15 @@ export const TaskRunSuccessfulExecutionResult = z.object({
   usage: TaskRunExecutionUsage.optional(),
   // Optional for now for backwards compatibility
   taskIdentifier: z.string().optional(),
+  // This is deprecated, use flushedMetadata instead
   metadata: FlushedRunMetadata.optional(),
+  // This is the new way to flush metadata
+  flushedMetadata: z
+    .object({
+      data: z.string().optional(),
+      dataType: z.string(),
+    })
+    .optional(),
 });
 
 export type TaskRunSuccessfulExecutionResult = z.infer<typeof TaskRunSuccessfulExecutionResult>;

references/hello-world/src/trigger/metadata.ts

@@ -0,0 +1,16 @@
+import { metadata, task } from "@trigger.dev/sdk";
+
+export const metadataTestTask = task({
+  id: "metadata-tester",
+  retry: {
+    maxAttempts: 3,
+    minTimeoutInMs: 500,
+    maxTimeoutInMs: 1000,
+    factor: 1.5,
+  },
+  run: async (payload: any, { ctx }) => {
+    metadata.set("test-key", "test-value");
+    metadata.append("test-keys", "test-value");
+    metadata.increment("test-counter", 1);
+  },
+});