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

Author: ericallam

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

@@ -9,7 +9,7 @@ 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 } from "effect";
+import { Effect, Schedule, Duration, Fiber } from "effect";
 import { type RuntimeFiber } from "effect/Fiber";
 import { setTimeout } from "timers/promises";
 import { Logger, LogLevel } from "@trigger.dev/core/logger";
@@ -102,6 +102,12 @@ export class UpdateMetadataService {
     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[]>
   ) => {
@@ -226,6 +232,11 @@ export class UpdateMetadataService {
         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({
@@ -378,6 +389,11 @@ export class UpdateMetadataService {
         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 })
@@ -401,6 +417,11 @@ export class UpdateMetadataService {
         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: {

apps/webapp/app/utils/packets.ts

@@ -13,8 +13,6 @@ export function handleMetadataPacket(
   metadataType: string,
   maximumSize: number
 ): IOPacket | undefined {
-  console.log("handleMetadataPacket.maximumSize", maximumSize);
-
   let metadataPacket: IOPacket | undefined = undefined;
 
   if (typeof metadata === "string") {
@@ -31,8 +29,6 @@ export function handleMetadataPacket(
 
   const byteLength = Buffer.byteLength(metadataPacket.data, "utf8");
 
-  console.log("handleMetadataPacket.byteLength", byteLength);
-
   if (byteLength > maximumSize) {
     throw new MetadataTooLargeError(`Metadata exceeds maximum size of ${maximumSize} bytes`);
   }