build uploadthing/fal demo and change how run metadata is synced to the server

Author: ericallam

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

@@ -14,6 +14,7 @@ import {
   WorkerManifest,
   ExecutorToWorkerMessageCatalog,
   timeout,
+  runMetadata,
 } from "@trigger.dev/core/v3";
 import { TriggerTracer } from "@trigger.dev/core/v3/tracer";
 import { ProdRuntimeManager } from "@trigger.dev/core/v3/prod";
@@ -22,6 +23,7 @@ import {
   DevUsageManager,
   DurableClock,
   getEnvVar,
+  getNumberEnvVar,
   logLevels,
   OtelTaskLogger,
   ProdUsageManager,
@@ -303,6 +305,10 @@ const zodIpc = new ZodIpcConnection({
           _execution = execution;
           _isRunning = true;
 
+          runMetadata.startPeriodicFlush(
+            getNumberEnvVar("TRIGGER_RUN_METADATA_FLUSH_INTERVAL", 1000)
+          );
+
           const measurement = usage.start();
 
           // This lives outside of the executor because this will eventually be moved to the controller level
@@ -397,7 +403,11 @@ const zodIpc = new ZodIpcConnection({
 async function flushAll(timeoutInMs: number = 10_000) {
   const now = performance.now();
 
-  await Promise.all([flushUsage(timeoutInMs), flushTracingSDK(timeoutInMs)]);
+  await Promise.all([
+    flushUsage(timeoutInMs),
+    flushTracingSDK(timeoutInMs),
+    flushMetadata(timeoutInMs),
+  ]);
 
   const duration = performance.now() - now;
 
@@ -424,6 +434,16 @@ async function flushTracingSDK(timeoutInMs: number = 10_000) {
   console.log(`Flushed tracingSDK in ${duration}ms`);
 }
 
+async function flushMetadata(timeoutInMs: number = 10_000) {
+  const now = performance.now();
+
+  await Promise.race([runMetadata.flush(), setTimeout(timeoutInMs)]);
+
+  const duration = performance.now() - now;
+
+  console.log(`Flushed runMetadata in ${duration}ms`);
+}
+
 const prodRuntimeManager = new ProdRuntimeManager(zodIpc, {
   waitThresholdInMs: parseInt(env.TRIGGER_RUNTIME_WAIT_THRESHOLD_IN_MS ?? "30000", 10),
 });

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

@@ -14,6 +14,7 @@ import {
   WorkerManifest,
   ExecutorToWorkerMessageCatalog,
   timeout,
+  runMetadata,
 } from "@trigger.dev/core/v3";
 import { TriggerTracer } from "@trigger.dev/core/v3/tracer";
 import { DevRuntimeManager } from "@trigger.dev/core/v3/dev";
@@ -30,6 +31,7 @@ import {
   TracingDiagnosticLogLevel,
   TracingSDK,
   usage,
+  getNumberEnvVar,
 } from "@trigger.dev/core/v3/workers";
 import { ZodIpcConnection } from "@trigger.dev/core/v3/zodIpc";
 import { readFile } from "node:fs/promises";
@@ -273,6 +275,9 @@ const zodIpc = new ZodIpcConnection({
         _execution = execution;
         _isRunning = true;
 
+        runMetadata.startPeriodicFlush(
+          getNumberEnvVar("TRIGGER_RUN_METADATA_FLUSH_INTERVAL", 1000)
+        );
         const measurement = usage.start();
 
         // This lives outside of the executor because this will eventually be moved to the controller level
@@ -345,7 +350,7 @@ const zodIpc = new ZodIpcConnection({
       }
     },
     FLUSH: async ({ timeoutInMs }, sender) => {
-      await _tracingSDK?.flush();
+      await Promise.allSettled([_tracingSDK?.flush(), runMetadata.flush()]);
     },
   },
 });

packages/core/package.json

@@ -181,6 +181,7 @@
   "dependencies": {
     "@electric-sql/client": "0.6.1",
     "@google-cloud/precise-date": "^4.0.0",
+    "@jsonhero/path": "^1.0.21",
     "@opentelemetry/api": "1.9.0",
     "@opentelemetry/api-logs": "0.52.1",
     "@opentelemetry/exporter-logs-otlp-http": "0.52.1",
@@ -192,10 +193,11 @@
     "@opentelemetry/sdk-trace-base": "1.25.1",
     "@opentelemetry/sdk-trace-node": "1.25.1",
     "@opentelemetry/semantic-conventions": "1.25.1",
+    "dequal": "^2.0.3",
     "execa": "^8.0.1",
     "humanize-duration": "^3.27.3",
-    "nanoid": "^3.3.4",
     "jose": "^5.4.0",
+    "nanoid": "^3.3.4",
     "socket.io-client": "4.7.5",
     "superjson": "^2.2.1",
     "zod": "3.22.3",

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

@@ -1,13 +1,17 @@
+import { dequal } from "dequal/lite";
 import { DeserializedJson } from "../../schemas/json.js";
 import { apiClientManager } from "../apiClientManager-api.js";
 import { taskContext } from "../task-context-api.js";
 import { getGlobal, registerGlobal } from "../utils/globals.js";
 import { ApiRequestOptions } from "../zodfetch.js";
+import { JSONHeroPath } from "@jsonhero/path";
 
 const API_NAME = "run-metadata";
 
 export class RunMetadataAPI {
   private static _instance?: RunMetadataAPI;
+  private flushTimeoutId: NodeJS.Timeout | null = null;
+  private hasChanges: boolean = false;
 
   private constructor() {}
 
@@ -39,52 +43,53 @@ export class RunMetadataAPI {
     return this.store?.[key];
   }
 
-  public async setKey(
-    key: string,
-    value: DeserializedJson,
-    requestOptions?: ApiRequestOptions
-  ): Promise<void> {
+  public setKey(key: string, value: DeserializedJson) {
     const runId = taskContext.ctx?.run.id;
 
     if (!runId) {
       return;
     }
 
-    const apiClient = apiClientManager.clientOrThrow();
+    let nextStore: Record<string, DeserializedJson> | undefined = this.store
+      ? structuredClone(this.store)
+      : undefined;
+
+    if (key.startsWith("$.")) {
+      const path = new JSONHeroPath(key);
+      path.set(nextStore, value);
+    } else {
+      nextStore = {
+        ...(nextStore ?? {}),
+        [key]: value,
+      };
+    }
 
-    const nextStore = {
-      ...(this.store ?? {}),
-      [key]: value,
-    };
+    if (!nextStore) {
+      return;
+    }
 
-    const response = await apiClient.updateRunMetadata(
-      runId,
-      { metadata: nextStore },
-      requestOptions
-    );
+    if (!dequal(this.store, nextStore)) {
+      this.hasChanges = true;
+    }
 
-    this.store = response.metadata;
+    this.store = nextStore;
   }
 
-  public async deleteKey(key: string, requestOptions?: ApiRequestOptions): Promise<void> {
+  public deleteKey(key: string, requestOptions?: ApiRequestOptions) {
     const runId = taskContext.ctx?.run.id;
 
     if (!runId) {
       return;
     }
 
-    const apiClient = apiClientManager.clientOrThrow();
-
     const nextStore = { ...(this.store ?? {}) };
     delete nextStore[key];
 
-    const response = await apiClient.updateRunMetadata(
-      runId,
-      { metadata: nextStore },
-      requestOptions
-    );
+    if (!dequal(this.store, nextStore)) {
+      this.hasChanges = true;
+    }
 
-    this.store = response.metadata;
+    this.store = nextStore;
   }
 
   public async update(
@@ -103,4 +108,56 @@ export class RunMetadataAPI {
 
     this.store = response.metadata;
   }
+
+  public async flush(requestOptions?: ApiRequestOptions): Promise<void> {
+    const runId = taskContext.ctx?.run.id;
+
+    if (!runId) {
+      return;
+    }
+
+    if (!this.store) {
+      return;
+    }
+
+    if (!this.hasChanges) {
+      return;
+    }
+
+    const apiClient = apiClientManager.clientOrThrow();
+
+    try {
+      this.hasChanges = false;
+      await apiClient.updateRunMetadata(runId, { metadata: this.store }, requestOptions);
+    } catch (error) {
+      this.hasChanges = true;
+      throw error;
+    }
+  }
+
+  public startPeriodicFlush(intervalMs: number = 1000) {
+    const periodicFlush = async (intervalMs: number) => {
+      try {
+        await this.flush();
+      } catch (error) {
+        console.error("Failed to flush metadata", error);
+        throw error;
+      } finally {
+        scheduleNext();
+      }
+    };
+
+    const scheduleNext = () => {
+      this.flushTimeoutId = setTimeout(() => periodicFlush(intervalMs), intervalMs);
+    };
+
+    scheduleNext();
+  }
+
+  stopPeriodicFlush(): void {
+    if (this.flushTimeoutId) {
+      clearTimeout(this.flushTimeoutId);
+      this.flushTimeoutId = null;
+    }
+  }
 }

packages/core/src/v3/utils/getEnv.ts

@@ -6,3 +6,19 @@ export function getEnvVar(name: string): string | undefined {
 
   return;
 }
+
+export function getNumberEnvVar(name: string, defaultValue?: number): number | undefined {
+  const value = getEnvVar(name);
+
+  if (value === undefined) {
+    return defaultValue;
+  }
+
+  const parsed = Number(value);
+
+  if (isNaN(parsed)) {
+    return defaultValue;
+  }
+
+  return parsed;
+}

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

@@ -1,7 +1,7 @@
 export { TaskExecutor, type TaskExecutorOptions } from "./taskExecutor.js";
 export type { RuntimeManager } from "../runtime/manager.js";
 export { PreciseWallClock as DurableClock } from "../clock/preciseWallClock.js";
-export { getEnvVar } from "../utils/getEnv.js";
+export { getEnvVar, getNumberEnvVar } from "../utils/getEnv.js";
 export { OtelTaskLogger, logLevels } from "../logger/taskLogger.js";
 export { ConsoleInterceptor } from "../consoleInterceptor.js";
 export { TracingSDK, type TracingDiagnosticLogLevel, recordSpanException } from "../otel/index.js";