fix: prevent large root/parent metadata updates from endlessly retrying (#2290)

* fixing metadata WIP

* WIP

* fix: prevent large root/parent metadata updates from endlessly retrying

* Fixed other calls to handleMetadataPacket

apps/webapp/app/routes/api.v1.runs.$runId.metadata.ts

@@ -2,9 +2,9 @@ import { json } from "@remix-run/server-runtime";
 import { tryCatch } from "@trigger.dev/core/utils";
 import { UpdateMetadataRequestBody } from "@trigger.dev/core/v3";
 import { z } from "zod";
-import { updateMetadataService } from "~/services/metadata/updateMetadata.server";
+import { updateMetadataService } from "~/services/metadata/updateMetadataInstance.server";
 import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
-import { ServiceValidationError } from "~/v3/services/baseService.server";
+import { ServiceValidationError } from "~/v3/services/common.server";
 
 const ParamsSchema = z.object({
   runId: z.string(),

apps/webapp/app/runEngine/services/triggerTask.server.ts

@@ -46,6 +46,7 @@ export class RunEngineTriggerTaskService {
   private readonly engine: RunEngine;
   private readonly tracer: Tracer;
   private readonly traceEventConcern: TraceEventConcern;
+  private readonly metadataMaximumSize: number;
 
   constructor(opts: {
     prisma: PrismaClientOrTransaction;
@@ -57,6 +58,7 @@ export class RunEngineTriggerTaskService {
     runNumberIncrementer: RunNumberIncrementer;
     traceEventConcern: TraceEventConcern;
     tracer: Tracer;
+    metadataMaximumSize: number;
   }) {
     this.prisma = opts.prisma;
     this.engine = opts.engine;
@@ -67,6 +69,7 @@ export class RunEngineTriggerTaskService {
     this.runNumberIncrementer = opts.runNumberIncrementer;
     this.tracer = opts.tracer;
     this.traceEventConcern = opts.traceEventConcern;
+    this.metadataMaximumSize = opts.metadataMaximumSize;
   }
 
   public async call({
@@ -188,7 +191,8 @@ export class RunEngineTriggerTaskService {
       const metadataPacket = body.options?.metadata
         ? handleMetadataPacket(
             body.options?.metadata,
-            body.options?.metadataType ?? "application/json"
+            body.options?.metadataType ?? "application/json",
+            this.metadataMaximumSize
           )
         : undefined;
 

apps/webapp/app/services/metadata/updateMetadata.server.ts

@@ -5,17 +5,14 @@ import {
   RunMetadataChangeOperation,
   UpdateMetadataRequestBody,
 } from "@trigger.dev/core/v3";
-import { prisma, PrismaClientOrTransaction } from "~/db.server";
-import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
-import { handleMetadataPacket } from "~/utils/packets";
-import { BaseService, ServiceValidationError } from "~/v3/services/baseService.server";
-
-import { Effect, Schedule, Duration } from "effect";
+import type { PrismaClientOrTransaction } from "~/db.server";
+import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
+import { handleMetadataPacket, MetadataTooLargeError } from "~/utils/packets";
+import { ServiceValidationError } from "~/v3/services/common.server";
+import { Effect, Schedule, Duration, Fiber } from "effect";
 import { type RuntimeFiber } from "effect/Fiber";
-import { logger } from "../logger.server";
-import { singleton } from "~/utils/singleton";
-import { env } from "~/env.server";
 import { setTimeout } from "timers/promises";
+import { Logger, LogLevel } from "@trigger.dev/core/logger";
 
 const RUN_UPDATABLE_WINDOW_MS = 60 * 60 * 1000; // 1 hour
 
@@ -25,17 +22,36 @@ type BufferedRunMetadataChangeOperation = {
   operation: RunMetadataChangeOperation;
 };
 
-export class UpdateMetadataService extends BaseService {
+export type UpdateMetadataServiceOptions = {
+  prisma: PrismaClientOrTransaction;
+  flushIntervalMs?: number;
+  flushEnabled?: boolean;
+  flushLoggingEnabled?: boolean;
+  maximumSize?: number;
+  logger?: Logger;
+  logLevel?: LogLevel;
+  // Testing hooks
+  onBeforeUpdate?: (runId: string) => Promise<void>;
+  onAfterRead?: (runId: string, metadataVersion: number) => Promise<void>;
+};
+
+export class UpdateMetadataService {
   private _bufferedOperations: Map<string, BufferedRunMetadataChangeOperation[]> = new Map();
   private _flushFiber: RuntimeFiber<void> | null = null;
+  private readonly _prisma: PrismaClientOrTransaction;
+  private readonly flushIntervalMs: number;
+  private readonly flushEnabled: boolean;
+  private readonly flushLoggingEnabled: boolean;
+  private readonly maximumSize: number;
+  private readonly logger: Logger;
 
-  constructor(
-    protected readonly _prisma: PrismaClientOrTransaction = prisma,
-    private readonly flushIntervalMs: number = 5000,
-    private readonly flushEnabled: boolean = true,
-    private readonly flushLoggingEnabled: boolean = true
-  ) {
-    super();
+  constructor(private readonly options: UpdateMetadataServiceOptions) {
+    this._prisma = options.prisma;
+    this.flushIntervalMs = options.flushIntervalMs ?? 5000;
+    this.flushEnabled = options.flushEnabled ?? true;
+    this.flushLoggingEnabled = options.flushLoggingEnabled ?? true;
+    this.maximumSize = options.maximumSize ?? 1024 * 1024 * 1; // 1MB
+    this.logger = options.logger ?? new Logger("UpdateMetadataService", options.logLevel ?? "info");
 
     this._startFlushing();
   }
@@ -43,12 +59,12 @@ export class UpdateMetadataService extends BaseService {
   // Start a loop that periodically flushes buffered operations
   private _startFlushing() {
     if (!this.flushEnabled) {
-      logger.info("[UpdateMetadataService] 🚽 Flushing disabled");
+      this.logger.info("[UpdateMetadataService] 🚽 Flushing disabled");
 
       return;
     }
 
-    logger.info("[UpdateMetadataService] 🚽 Flushing started");
+    this.logger.info("[UpdateMetadataService] 🚽 Flushing started");
 
     // Create a program that sleeps, then processes buffered ops
     const program = Effect.gen(this, function* (_) {
@@ -62,7 +78,7 @@ export class UpdateMetadataService extends BaseService {
 
         yield* Effect.sync(() => {
           if (this.flushLoggingEnabled) {
-            logger.debug(`[UpdateMetadataService] Flushing operations`, {
+            this.logger.debug(`[UpdateMetadataService] Flushing operations`, {
               operations: Object.fromEntries(currentOperations),
             });
           }
@@ -77,7 +93,7 @@ export class UpdateMetadataService extends BaseService {
       // Handle any unexpected errors, ensuring program does not fail
       Effect.catchAll((error) =>
         Effect.sync(() => {
-          logger.error("Error in flushing program:", { error });
+          this.logger.error("Error in flushing program:", { error });
         })
       )
     );
@@ -86,6 +102,12 @@ export class UpdateMetadataService extends BaseService {
     this._flushFiber = Effect.runFork(program as Effect.Effect<void, never, never>);
   }
 
+  stopFlushing() {
+    if (this._flushFiber) {
+      Effect.runFork(Fiber.interrupt(this._flushFiber));
+    }
+  }
+
   private _processBufferedOperations = (
     operations: Map<string, BufferedRunMetadataChangeOperation[]>
   ) => {
@@ -101,7 +123,7 @@ export class UpdateMetadataService extends BaseService {
 
         yield* Effect.sync(() => {
           if (this.flushLoggingEnabled) {
-            logger.debug(`[UpdateMetadataService] Processing operations for run`, {
+            this.logger.debug(`[UpdateMetadataService] Processing operations for run`, {
               runId,
               operationsCount: processedOps.length,
             });
@@ -111,6 +133,25 @@ export class UpdateMetadataService extends BaseService {
         // Update run with retry
         yield* _(
           this._updateRunWithOperations(runId, processedOps).pipe(
+            // Catch MetadataTooLargeError before retry logic
+            Effect.catchIf(
+              (error) => error instanceof MetadataTooLargeError,
+              (error) =>
+                Effect.sync(() => {
+                  // Log the error but don't return operations to buffer
+                  console.error(
+                    `[UpdateMetadataService] Dropping operations for run ${runId} due to metadata size limit:`,
+                    error.message
+                  );
+                  if (this.flushLoggingEnabled) {
+                    this.logger.warn(`[UpdateMetadataService] Metadata too large for run`, {
+                      runId,
+                      operationsCount: processedOps.length,
+                      error: error.message,
+                    });
+                  }
+                })
+            ),
             Effect.retry(Schedule.exponential(Duration.millis(100), 1.4)),
             Effect.catchAll((error) =>
               Effect.sync(() => {
@@ -145,6 +186,11 @@ export class UpdateMetadataService extends BaseService {
         return yield* _(Effect.fail(new Error(`Run ${runId} not found`)));
       }
 
+      // Testing hook after read
+      if (this.options.onAfterRead) {
+        yield* _(Effect.tryPromise(() => this.options.onAfterRead!(runId, run.metadataVersion)));
+      }
+
       const metadata = yield* _(
         Effect.tryPromise(() =>
           run.metadata
@@ -160,22 +206,37 @@ export class UpdateMetadataService extends BaseService {
       );
 
       if (applyResult.unappliedOperations.length === operations.length) {
-        logger.warn(`No operations applied for run ${runId}`);
+        this.logger.warn(`No operations applied for run ${runId}`);
         // If no operations were applied, return
         return;
       }
 
       // Stringify the metadata
       const newMetadataPacket = yield* _(
-        Effect.try(() => handleMetadataPacket(applyResult.newMetadata, run.metadataType))
+        Effect.try(() =>
+          handleMetadataPacket(applyResult.newMetadata, run.metadataType, this.maximumSize)
+        ).pipe(
+          Effect.mapError((error) => {
+            // Preserve the original error if it's MetadataTooLargeError
+            if ("cause" in error && error.cause instanceof MetadataTooLargeError) {
+              return error.cause;
+            }
+            return error;
+          })
+        )
       );
 
       if (!newMetadataPacket) {
         // Log and skip if metadata is invalid
-        logger.warn(`Invalid metadata after operations, skipping update`);
+        this.logger.warn(`Invalid metadata after operations, skipping update`);
         return;
       }
 
+      // Testing hook before update
+      if (this.options.onBeforeUpdate) {
+        yield* _(Effect.tryPromise(() => this.options.onBeforeUpdate!(runId)));
+      }
+
       const result = yield* _(
         Effect.tryPromise(() =>
           this._prisma.taskRun.updateMany({
@@ -193,7 +254,7 @@ export class UpdateMetadataService extends BaseService {
 
       if (result.count === 0) {
         yield* Effect.sync(() => {
-          logger.warn(`Optimistic lock failed for run ${runId}`, {
+          this.logger.warn(`Optimistic lock failed for run ${runId}`, {
             metadataVersion: run.metadataVersion,
           });
         });
@@ -275,12 +336,12 @@ export class UpdateMetadataService extends BaseService {
       throw new ServiceValidationError("Cannot update metadata for a completed run");
     }
 
-    if (body.parentOperations && body.parentOperations.length > 0 && taskRun.parentTaskRun) {
-      this.#ingestRunOperations(taskRun.parentTaskRun.id, body.parentOperations);
+    if (body.parentOperations && body.parentOperations.length > 0) {
+      this.#ingestRunOperations(taskRun.parentTaskRun?.id ?? taskRun.id, body.parentOperations);
     }
 
-    if (body.rootOperations && body.rootOperations.length > 0 && taskRun.rootTaskRun) {
-      this.#ingestRunOperations(taskRun.rootTaskRun.id, body.rootOperations);
+    if (body.rootOperations && body.rootOperations.length > 0) {
+      this.#ingestRunOperations(taskRun.rootTaskRun?.id ?? taskRun.id, body.rootOperations);
     }
 
     const newMetadata = await this.#updateRunMetadata({
@@ -328,6 +389,11 @@ export class UpdateMetadataService extends BaseService {
         throw new Error(`Run ${runId} not found`);
       }
 
+      // Testing hook after read
+      if (this.options.onAfterRead) {
+        await this.options.onAfterRead(runId, run.metadataVersion);
+      }
+
       // Parse the current metadata
       const currentMetadata = await (run.metadata
         ? parsePacket({ data: run.metadata, dataType: run.metadataType })
@@ -341,6 +407,21 @@ export class UpdateMetadataService extends BaseService {
         return currentMetadata;
       }
 
+      const newMetadataPacket = handleMetadataPacket(
+        applyResults.newMetadata,
+        run.metadataType,
+        this.maximumSize
+      );
+
+      if (!newMetadataPacket) {
+        throw new ServiceValidationError("Unable to update metadata");
+      }
+
+      // Testing hook before update
+      if (this.options.onBeforeUpdate) {
+        await this.options.onBeforeUpdate(runId);
+      }
+
       // Update with optimistic locking
       const result = await this._prisma.taskRun.updateMany({
         where: {
@@ -348,8 +429,8 @@ export class UpdateMetadataService extends BaseService {
           metadataVersion: run.metadataVersion,
         },
         data: {
-          metadata: JSON.stringify(applyResults.newMetadata),
-          metadataType: run.metadataType,
+          metadata: newMetadataPacket.data,
+          metadataType: newMetadataPacket.dataType,
           metadataVersion: {
             increment: 1,
           },
@@ -358,8 +439,8 @@ export class UpdateMetadataService extends BaseService {
 
       if (result.count === 0) {
         if (this.flushLoggingEnabled) {
-          logger.debug(
-            `[UpdateMetadataService][updateRunMetadataWithOperations] Optimistic lock failed for run ${runId}`,
+          this.logger.debug(
+            `[updateRunMetadataWithOperations] Optimistic lock failed for run ${runId}`,
             {
               metadataVersion: run.metadataVersion,
             }
@@ -379,13 +460,11 @@ export class UpdateMetadataService extends BaseService {
       }
 
       if (this.flushLoggingEnabled) {
-        logger.debug(
-          `[UpdateMetadataService][updateRunMetadataWithOperations] Updated metadata for run ${runId}`,
-          {
-            metadata: applyResults.newMetadata,
-            operations: operations,
-          }
-        );
+        this.logger.debug(`[updateRunMetadataWithOperations] Updated metadata for run`, {
+          metadata: applyResults.newMetadata,
+          operations: operations,
+          runId,
+        });
       }
 
       // Success! Return the new metadata
@@ -409,10 +488,14 @@ export class UpdateMetadataService extends BaseService {
     body: UpdateMetadataRequestBody,
     existingMetadata: IOPacket
   ) {
-    const metadataPacket = handleMetadataPacket(body.metadata, "application/json");
+    const metadataPacket = handleMetadataPacket(
+      body.metadata,
+      "application/json",
+      this.maximumSize
+    );
 
     if (!metadataPacket) {
-      throw new ServiceValidationError("Invalid metadata");
+      return {};
     }
 
     if (
@@ -420,13 +503,10 @@ export class UpdateMetadataService extends BaseService {
       (existingMetadata.data && metadataPacket.data !== existingMetadata.data)
     ) {
       if (this.flushLoggingEnabled) {
-        logger.debug(
-          `[UpdateMetadataService][updateRunMetadataDirectly] Updating metadata directly for run`,
-          {
-            metadata: metadataPacket.data,
-            runId,
-          }
-        );
+        this.logger.debug(`[updateRunMetadataDirectly] Updating metadata directly for run`, {
+          metadata: metadataPacket.data,
+          runId,
+        });
       }
 
       // Update the metadata without version check
@@ -458,7 +538,7 @@ export class UpdateMetadataService extends BaseService {
     });
 
     if (this.flushLoggingEnabled) {
-      logger.debug(`[UpdateMetadataService] Ingesting operations for run`, {
+      this.logger.debug(`[ingestRunOperations] Ingesting operations for run`, {
         runId,
         bufferedOperations,
       });
@@ -468,15 +548,14 @@ export class UpdateMetadataService extends BaseService {
 
     this._bufferedOperations.set(runId, [...existingBufferedOperations, ...bufferedOperations]);
   }
-}
 
-export const updateMetadataService = singleton(
-  "update-metadata-service",
-  () =>
-    new UpdateMetadataService(
-      prisma,
-      env.BATCH_METADATA_OPERATIONS_FLUSH_INTERVAL_MS,
-      env.BATCH_METADATA_OPERATIONS_FLUSH_ENABLED === "1",
-      env.BATCH_METADATA_OPERATIONS_FLUSH_LOGGING_ENABLED === "1"
-    )
-);
+  // Testing method to manually trigger flush
+  async flushOperations() {
+    const currentOperations = new Map(this._bufferedOperations);
+    this._bufferedOperations.clear();
+
+    if (currentOperations.size > 0) {
+      await Effect.runPromise(this._processBufferedOperations(currentOperations));
+    }
+  }
+}

apps/webapp/app/services/metadata/updateMetadataInstance.server.ts

@@ -0,0 +1,17 @@
+import { singleton } from "~/utils/singleton";
+import { env } from "~/env.server";
+import { UpdateMetadataService } from "./updateMetadata.server";
+import { prisma } from "~/db.server";
+
+export const updateMetadataService = singleton(
+  "update-metadata-service",
+  () =>
+    new UpdateMetadataService({
+      prisma,
+      flushIntervalMs: env.BATCH_METADATA_OPERATIONS_FLUSH_INTERVAL_MS,
+      flushEnabled: env.BATCH_METADATA_OPERATIONS_FLUSH_ENABLED === "1",
+      flushLoggingEnabled: env.BATCH_METADATA_OPERATIONS_FLUSH_LOGGING_ENABLED === "1",
+      maximumSize: env.TASK_RUN_METADATA_MAXIMUM_SIZE,
+      logLevel: env.BATCH_METADATA_OPERATIONS_FLUSH_LOGGING_ENABLED === "1" ? "debug" : "info",
+    })
+);

apps/webapp/app/utils/packets.ts

@@ -1,6 +1,5 @@
 import { IOPacket } from "@trigger.dev/core/v3/utils/ioSerialization";
-import { env } from "~/env.server";
-import { ServiceValidationError } from "~/v3/services/baseService.server";
+import { ServiceValidationError } from "~/v3/services/common.server";
 
 export class MetadataTooLargeError extends ServiceValidationError {
   constructor(message: string) {
@@ -9,7 +8,11 @@ export class MetadataTooLargeError extends ServiceValidationError {
   }
 }
 
-export function handleMetadataPacket(metadata: any, metadataType: string): IOPacket | undefined {
+export function handleMetadataPacket(
+  metadata: any,
+  metadataType: string,
+  maximumSize: number
+): IOPacket | undefined {
   let metadataPacket: IOPacket | undefined = undefined;
 
   if (typeof metadata === "string") {
@@ -26,10 +29,8 @@ export function handleMetadataPacket(metadata: any, metadataType: string): IOPac
 
   const byteLength = Buffer.byteLength(metadataPacket.data, "utf8");
 
-  if (byteLength > env.TASK_RUN_METADATA_MAXIMUM_SIZE) {
-    throw new MetadataTooLargeError(
-      `Metadata exceeds maximum size of ${env.TASK_RUN_METADATA_MAXIMUM_SIZE} bytes`
-    );
+  if (byteLength > maximumSize) {
+    throw new MetadataTooLargeError(`Metadata exceeds maximum size of ${maximumSize} bytes`);
   }
 
   return metadataPacket;

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

@@ -13,7 +13,7 @@ import { roomFromFriendlyRunId, socketIo } from "./handleSocketIo.server";
 import { engine } from "./runEngine.server";
 import { PerformTaskRunAlertsService } from "./services/alerts/performTaskRunAlerts.server";
 import { RunId } from "@trigger.dev/core/v3/isomorphic";
-import { updateMetadataService } from "~/services/metadata/updateMetadata.server";
+import { updateMetadataService } from "~/services/metadata/updateMetadataInstance.server";
 import { findEnvironmentFromRun } from "~/models/runtimeEnvironment.server";
 import { env } from "~/env.server";
 import { getTaskEventStoreTableForRun } from "./taskEventStore.server";

apps/webapp/app/v3/services/baseService.server.ts

@@ -3,6 +3,9 @@ import { $replica, PrismaClientOrTransaction, prisma } from "~/db.server";
 import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { attributesFromAuthenticatedEnv, tracer } from "../tracer.server";
 import { engine, RunEngine } from "../runEngine.server";
+import { ServiceValidationError } from "./common.server";
+
+export { ServiceValidationError };
 
 export abstract class BaseService {
   constructor(
@@ -54,10 +57,3 @@ export class WithRunEngine extends BaseService {
     this._engine = opts.engine ?? engine;
   }
 }
-
-export class ServiceValidationError extends Error {
-  constructor(message: string, public status?: number) {
-    super(message);
-    this.name = "ServiceValidationError";
-  }
-}

apps/webapp/app/v3/services/common.server.ts

@@ -0,0 +1,6 @@
+export class ServiceValidationError extends Error {
+  constructor(message: string, public status?: number) {
+    super(message);
+    this.name = "ServiceValidationError";
+  }
+}

apps/webapp/app/v3/services/finalizeTaskRun.server.ts

@@ -3,7 +3,7 @@ import { type Prisma, type TaskRun } from "@trigger.dev/database";
 import { findQueueInEnvironment } from "~/models/taskQueue.server";
 import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
-import { updateMetadataService } from "~/services/metadata/updateMetadata.server";
+import { updateMetadataService } from "~/services/metadata/updateMetadataInstance.server";
 import { marqs } from "~/v3/marqs/index.server";
 import { generateFriendlyId } from "../friendlyIdentifiers";
 import { socketIo } from "../handleSocketIo.server";

apps/webapp/app/v3/services/taskRunTemplate.server.ts

@@ -24,7 +24,11 @@ export class TaskRunTemplateService extends BaseService {
         }
 
         const metadataPacket = data.metadata
-          ? handleMetadataPacket(data.metadata, "application/json")
+          ? handleMetadataPacket(
+              data.metadata,
+              "application/json",
+              env.TASK_RUN_METADATA_MAXIMUM_SIZE
+            )
           : undefined;
 
         const taskRunTemplate = await this._prisma.taskRunTemplate.create({

apps/webapp/app/v3/services/triggerTask.server.ts

@@ -13,6 +13,7 @@ import { eventRepository } from "../eventRepository.server";
 import { tracer } from "../tracer.server";
 import { WithRunEngine } from "./baseService.server";
 import { TriggerTaskServiceV1 } from "./triggerTaskV1.server";
+import { env } from "~/env.server";
 
 export type TriggerTaskServiceOptions = {
   idempotencyKey?: string;
@@ -107,6 +108,7 @@ export class TriggerTaskService extends WithRunEngine {
       runNumberIncrementer: new DefaultRunNumberIncrementer(),
       traceEventConcern,
       tracer: tracer,
+      metadataMaximumSize: env.TASK_RUN_METADATA_MAXIMUM_SIZE,
     });
 
     return await service.call({

apps/webapp/app/v3/services/triggerTaskV1.server.ts

@@ -163,7 +163,8 @@ export class TriggerTaskServiceV1 extends BaseService {
       const metadataPacket = body.options?.metadata
         ? handleMetadataPacket(
             body.options?.metadata,
-            body.options?.metadataType ?? "application/json"
+            body.options?.metadataType ?? "application/json",
+            env.TASK_RUN_METADATA_MAXIMUM_SIZE
           )
         : undefined;
 

apps/webapp/test/engine/triggerTask.test.ts

@@ -164,6 +164,7 @@ describe("RunEngineTriggerTaskService", () => {
       validator: new MockTriggerTaskValidator(),
       traceEventConcern: new MockTraceEventConcern(),
       tracer: trace.getTracer("test", "0.0.0"),
+      metadataMaximumSize: 1024 * 1024 * 1, // 1MB
     });
 
     const result = await triggerTaskService.call({
@@ -254,6 +255,7 @@ describe("RunEngineTriggerTaskService", () => {
       validator: new MockTriggerTaskValidator(),
       traceEventConcern: new MockTraceEventConcern(),
       tracer: trace.getTracer("test", "0.0.0"),
+      metadataMaximumSize: 1024 * 1024 * 1, // 1MB
     });
 
     const result = await triggerTaskService.call({
@@ -395,6 +397,7 @@ describe("RunEngineTriggerTaskService", () => {
         validator: new MockTriggerTaskValidator(),
         traceEventConcern: new MockTraceEventConcern(),
         tracer: trace.getTracer("test", "0.0.0"),
+        metadataMaximumSize: 1024 * 1024 * 1, // 1MB
       });
 
       // Test case 1: Trigger with lockToVersion but no specific queue

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

@@ -1,4 +1,4 @@
-import { logger, metadata, task } from "@trigger.dev/sdk";
+import { logger, metadata, task, wait } from "@trigger.dev/sdk";
 import { setTimeout } from "node:timers/promises";
 
 export const metadataTestTask = task({
@@ -35,3 +35,31 @@ export const metadataTestTask = task({
     await runPromise;
   },
 });
+
+export const parentTask = task({
+  id: "metadata-parent-task",
+  run: async (payload: any, { ctx }) => {
+    // will not be set
+    metadata.root.set("test.root.set", true);
+    metadata.parent.set("test.parent.set", true);
+    metadata.set("test.set", "test");
+
+    await childTask.triggerAndWait({});
+
+    return {
+      ok: true,
+    };
+  },
+});
+
+export const childTask = task({
+  id: "metadata-child-task",
+  run: async (payload: any, { ctx }) => {
+    // will not be set
+    metadata.root.set("child.root.before", true);
+    await wait.for({ seconds: 15 });
+    // will be set
+    metadata.root.set("child.root.after", true);
+    return { ok: true };
+  },
+});