fixing metadata WIP

Author: ericallam

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/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,13 @@ 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 {
+  console.log("handleMetadataPacket.maximumSize", maximumSize);
+
   let metadataPacket: IOPacket | undefined = undefined;
 
   if (typeof metadata === "string") {
@@ -26,10 +31,10 @@ 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`
-    );
+  console.log("handleMetadataPacket.byteLength", byteLength);
+
+  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/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/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