add fast idempotency key extraction for the run replication service

Author: ericallam

apps/webapp/app/presenters/v3/SpanPresenter.server.ts

@@ -9,7 +9,7 @@ import {
 } from "@trigger.dev/core/v3";
 import { AttemptId, getMaxDuration, parseTraceparent } from "@trigger.dev/core/v3/isomorphic";
 import {
-  getIdempotencyKeyScope,
+  extractIdempotencyKeyScope,
   getUserProvidedIdempotencyKey,
 } from "@trigger.dev/core/v3/serverOnly";
 import { RUNNING_STATUSES } from "~/components/runs/v3/TaskRunStatus";
@@ -235,7 +235,7 @@ export class SpanPresenter extends BasePresenter {
       environmentId: run.runtimeEnvironment.id,
       idempotencyKey: getUserProvidedIdempotencyKey(run),
       idempotencyKeyExpiresAt: run.idempotencyKeyExpiresAt,
-      idempotencyKeyScope: getIdempotencyKeyScope(run),
+      idempotencyKeyScope: extractIdempotencyKeyScope(run),
       idempotencyKeyStatus: this.getIdempotencyKeyStatus(run),
       debounce: run.debounce as { key: string; delay: string; createdAt: Date } | null,
       schedule: await this.resolveSchedule(run.scheduleId ?? undefined),
@@ -288,7 +288,7 @@ export class SpanPresenter extends BasePresenter {
     idempotencyKeyOptions: unknown;
   }): "active" | "inactive" | "expired" | undefined {
     // No idempotency configured if no scope exists
-    const scope = getIdempotencyKeyScope(run);
+    const scope = extractIdempotencyKeyScope(run);
     if (!scope) {
       return undefined;
     }

apps/webapp/app/services/runsReplicationService.server.ts

@@ -21,7 +21,7 @@ import {
 import { Logger, type LogLevel } from "@trigger.dev/core/logger";
 import { tryCatch } from "@trigger.dev/core/utils";
 import { parsePacketAsJson } from "@trigger.dev/core/v3/utils/ioSerialization";
-import { extractIdempotencyKeyUser, getIdempotencyKeyScope } from "@trigger.dev/core/v3/serverOnly";
+import { unsafeExtractIdempotencyKeyScope, unsafeExtractIdempotencyKeyUser } from "@trigger.dev/core/v3/serverOnly";
 import { type TaskRun } from "@trigger.dev/database";
 import { nanoid } from "nanoid";
 import EventEmitter from "node:events";
@@ -892,8 +892,8 @@ export class RunsReplicationService {
       run.spanId, // span_id
       run.traceId, // trace_id
       run.idempotencyKey ?? "", // idempotency_key
-      extractIdempotencyKeyUser(run) ?? "", // idempotency_key_user
-      getIdempotencyKeyScope(run) ?? "", // idempotency_key_scope
+      unsafeExtractIdempotencyKeyUser(run) ?? "", // idempotency_key_user
+      unsafeExtractIdempotencyKeyScope(run) ?? "", // idempotency_key_scope
       run.ttl ?? "", // expiration_ttl
       run.isTest ?? false, // is_test
       _version.toString(), // _version

apps/webapp/test/runsReplicationService.part2.test.ts

@@ -1292,7 +1292,11 @@ describe("RunsReplicationService (part 2/2)", () => {
 
           // Execution settings
           machinePreset: "large-1x",
-          idempotencyKey: "exhaustive-idempotency-key",
+          idempotencyKey: "exhaustive-idempotency-key-hashed",
+          idempotencyKeyOptions: {
+            key: "exhaustive-idempotency-key",
+            scope: "run",
+          },
           ttl: "1h",
           isTest: true,
           concurrencyKey: "exhaustive-concurrency-key",
@@ -1388,7 +1392,9 @@ describe("RunsReplicationService (part 2/2)", () => {
 
       // Execution settings
       expect(clickhouseRun.machine_preset).toBe("large-1x");
-      expect(clickhouseRun.idempotency_key).toBe("exhaustive-idempotency-key");
+      expect(clickhouseRun.idempotency_key).toBe("exhaustive-idempotency-key-hashed");
+      expect(clickhouseRun.idempotency_key_user).toBe("exhaustive-idempotency-key");
+      expect(clickhouseRun.idempotency_key_scope).toBe("run");
       expect(clickhouseRun.expiration_ttl).toBe("1h");
       expect(clickhouseRun.is_test).toBe(1); // ClickHouse returns booleans as integers
       expect(clickhouseRun.concurrency_key).toBe("exhaustive-concurrency-key");

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

@@ -31,7 +31,7 @@ import {
   TaskRunSuccessfulExecutionResult,
 } from "@trigger.dev/core/v3/schemas";
 import {
-  getIdempotencyKeyScope,
+  extractIdempotencyKeyScope,
   getUserProvidedIdempotencyKey,
 } from "@trigger.dev/core/v3/serverOnly";
 import { parsePacket } from "@trigger.dev/core/v3/utils/ioSerialization";
@@ -267,7 +267,7 @@ export class RunAttemptSystem {
         createdAt: run.createdAt,
         startedAt: run.startedAt ?? run.createdAt,
         idempotencyKey: getUserProvidedIdempotencyKey(run) ?? undefined,
-        idempotencyKeyScope: getIdempotencyKeyScope(run),
+        idempotencyKeyScope: extractIdempotencyKeyScope(run),
         maxAttempts: run.maxAttempts ?? undefined,
         version: run.taskVersion ?? "unknown",
         maxDuration: run.maxDurationInSeconds ?? undefined,
@@ -578,7 +578,7 @@ export class RunAttemptSystem {
               tags: updatedRun.runTags,
               isTest: updatedRun.isTest,
               idempotencyKey: getUserProvidedIdempotencyKey(updatedRun) ?? undefined,
-              idempotencyKeyScope: getIdempotencyKeyScope(updatedRun),
+              idempotencyKeyScope: extractIdempotencyKeyScope(updatedRun),
               startedAt: updatedRun.startedAt ?? updatedRun.createdAt,
               maxAttempts: updatedRun.maxAttempts ?? undefined,
               version: updatedRun.taskVersion ?? "unknown",

packages/core/src/v3/serverOnly/idempotencyKeys.ts

@@ -24,7 +24,7 @@ export function getUserProvidedIdempotencyKey(run: {
  * @param run - Object containing idempotencyKeyOptions (JSON from DB)
  * @returns The scope if valid options exist, otherwise undefined
  */
-export function getIdempotencyKeyScope(run: {
+export function extractIdempotencyKeyScope(run: {
   idempotencyKeyOptions: unknown;
 }): "run" | "attempt" | "global" | undefined {
   const parsed = IdempotencyKeyOptionsSchema.safeParse(run.idempotencyKeyOptions);
@@ -34,6 +34,13 @@ export function getIdempotencyKeyScope(run: {
   return undefined;
 }
 
+export function unsafeExtractIdempotencyKeyScope(run: {
+  idempotencyKeyOptions: unknown;
+}): "run" | "attempt" | "global" | undefined {
+  const unsafe = run.idempotencyKeyOptions as { scope?: "run" | "attempt" | "global" } | undefined | null;
+  return unsafe?.scope ?? undefined;
+}
+
 /**
  * Extracts just the user-provided key from idempotencyKeyOptions, without falling back to the hash.
  * Useful for ClickHouse replication where we want to store only the explicit user key.
@@ -50,3 +57,13 @@ export function extractIdempotencyKeyUser(run: {
   }
   return undefined;
 }
+
+export function unsafeExtractIdempotencyKeyUser(run: {
+  idempotencyKeyOptions: unknown;
+}): string | undefined {
+  const unsafe = run.idempotencyKeyOptions as { key?: string } | undefined | null;
+
+  return unsafe?.key ?? undefined;
+}
+
+