extract out the getUserProvidedIdempotencyKey/Scope stuff to core and also added a test for the resetting changes

Author: ericallam

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

@@ -9,6 +9,7 @@ import {
   logger,
 } from "@trigger.dev/core/v3";
 import { parsePacketAsJson } from "@trigger.dev/core/v3/utils/ioSerialization";
+import { getUserProvidedIdempotencyKey } from "@trigger.dev/core/v3/serverOnly";
 import { Prisma, TaskRunAttemptStatus, TaskRunStatus } from "@trigger.dev/database";
 import assertNever from "assert-never";
 import { API_VERSIONS, CURRENT_API_VERSION, RunStatusUnspecifiedApiVersion } from "~/api/versions";
@@ -18,26 +19,6 @@ import { generatePresignedUrl } from "~/v3/r2.server";
 import { tracer } from "~/v3/tracer.server";
 import { startSpanWithEnv } from "~/v3/tracing.server";
 
-/**
- * Returns the user-provided idempotency key if available (from idempotencyKeyOptions),
- * otherwise falls back to the stored idempotency key (which is the hash).
- */
-function getUserProvidedIdempotencyKey(run: {
-  idempotencyKey: string | null;
-  idempotencyKeyOptions: unknown;
-}): string | undefined {
-  // If we have the user-provided key options, return the original key
-  const options = run.idempotencyKeyOptions as {
-    key?: string;
-    scope?: string;
-  } | null;
-  if (options?.key) {
-    return options.key;
-  }
-  // Fallback to the hash (for runs created before this feature)
-  return run.idempotencyKey ?? undefined;
-}
-
 // Build 'select' object
 const commonRunSelect = {
   id: true,

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

@@ -8,6 +8,7 @@ import {
   type V3TaskRunContext,
 } from "@trigger.dev/core/v3";
 import { AttemptId, getMaxDuration, parseTraceparent } from "@trigger.dev/core/v3/isomorphic";
+import { getUserProvidedIdempotencyKey } from "@trigger.dev/core/v3/serverOnly";
 import { RUNNING_STATUSES } from "~/components/runs/v3/TaskRunStatus";
 import { logger } from "~/services/logger.server";
 import { rehydrateAttribute } from "~/v3/eventRepository/eventRepository.server";
@@ -229,7 +230,7 @@ export class SpanPresenter extends BasePresenter {
       isTest: run.isTest,
       replayedFromTaskRunFriendlyId: run.replayedFromTaskRunFriendlyId,
       environmentId: run.runtimeEnvironment.id,
-      idempotencyKey: this.getUserProvidedIdempotencyKey(run),
+      idempotencyKey: getUserProvidedIdempotencyKey(run),
       idempotencyKeyExpiresAt: run.idempotencyKeyExpiresAt,
       debounce: run.debounce as { key: string; delay: string; createdAt: Date } | null,
       schedule: await this.resolveSchedule(run.scheduleId ?? undefined),
@@ -645,7 +646,7 @@ export class SpanPresenter extends BasePresenter {
         createdAt: run.createdAt,
         tags: run.runTags,
         isTest: run.isTest,
-        idempotencyKey: this.getUserProvidedIdempotencyKey(run) ?? undefined,
+        idempotencyKey: getUserProvidedIdempotencyKey(run) ?? undefined,
         startedAt: run.startedAt ?? run.createdAt,
         durationMs: run.usageDurationMs,
         costInCents: run.costInCents,
@@ -706,22 +707,4 @@ export class SpanPresenter extends BasePresenter {
     return parsedTraceparent?.traceId;
   }
 
-  /**
-   * Returns the user-provided idempotency key if available (from idempotencyKeyOptions),
-   * otherwise falls back to the stored idempotency key (which is the hash).
-   */
-  getUserProvidedIdempotencyKey(
-    run: Pick<FindRunResult, "idempotencyKey" | "idempotencyKeyOptions">
-  ): string | null {
-    // If we have the user-provided key options, return the original key
-    const options = run.idempotencyKeyOptions as {
-      key?: string;
-      scope?: string;
-    } | null;
-    if (options?.key) {
-      return options.key;
-    }
-    // Fallback to the hash (for runs created before this feature)
-    return run.idempotencyKey;
-  }
 }

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

@@ -21,6 +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 { type TaskRun } from "@trigger.dev/database";
 import { nanoid } from "nanoid";
 import EventEmitter from "node:events";
@@ -891,8 +892,8 @@ export class RunsReplicationService {
       run.spanId, // span_id
       run.traceId, // trace_id
       run.idempotencyKey ?? "", // idempotency_key
-      this.#extractIdempotencyKeyUser(run), // idempotency_key_user
-      this.#extractIdempotencyKeyScope(run), // idempotency_key_scope
+      extractIdempotencyKeyUser(run) ?? "", // idempotency_key_user
+      getIdempotencyKeyScope(run) ?? "", // idempotency_key_scope
       run.ttl ?? "", // expiration_ttl
       run.isTest ?? false, // is_test
       _version.toString(), // _version
@@ -954,15 +955,6 @@ export class RunsReplicationService {
     return { data: parsedData };
   }
 
-  #extractIdempotencyKeyUser(run: TaskRun): string {
-    const options = run.idempotencyKeyOptions as { key?: string; scope?: string } | null;
-    return options?.key ?? "";
-  }
-
-  #extractIdempotencyKeyScope(run: TaskRun): string {
-    const options = run.idempotencyKeyOptions as { key?: string; scope?: string } | null;
-    return options?.scope ?? "";
-  }
 }
 
 export type ConcurrentFlushSchedulerConfig<T> = {

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

@@ -30,6 +30,10 @@ import {
   TaskRunInternalError,
   TaskRunSuccessfulExecutionResult,
 } from "@trigger.dev/core/v3/schemas";
+import {
+  getIdempotencyKeyScope,
+  getUserProvidedIdempotencyKey,
+} from "@trigger.dev/core/v3/serverOnly";
 import { parsePacket } from "@trigger.dev/core/v3/utils/ioSerialization";
 import {
   $transaction,
@@ -262,8 +266,8 @@ export class RunAttemptSystem {
         isTest: run.isTest,
         createdAt: run.createdAt,
         startedAt: run.startedAt ?? run.createdAt,
-        idempotencyKey: this.#getUserProvidedIdempotencyKey(run) ?? undefined,
-        idempotencyKeyScope: this.#getIdempotencyKeyScope(run),
+        idempotencyKey: getUserProvidedIdempotencyKey(run) ?? undefined,
+        idempotencyKeyScope: getIdempotencyKeyScope(run),
         maxAttempts: run.maxAttempts ?? undefined,
         version: run.taskVersion ?? "unknown",
         maxDuration: run.maxDurationInSeconds ?? undefined,
@@ -573,8 +577,8 @@ export class RunAttemptSystem {
               createdAt: updatedRun.createdAt,
               tags: updatedRun.runTags,
               isTest: updatedRun.isTest,
-              idempotencyKey: this.#getUserProvidedIdempotencyKey(updatedRun) ?? undefined,
-              idempotencyKeyScope: this.#getIdempotencyKeyScope(updatedRun),
+              idempotencyKey: getUserProvidedIdempotencyKey(updatedRun) ?? undefined,
+              idempotencyKeyScope: getIdempotencyKeyScope(updatedRun),
               startedAt: updatedRun.startedAt ?? updatedRun.createdAt,
               maxAttempts: updatedRun.maxAttempts ?? undefined,
               version: updatedRun.taskVersion ?? "unknown",
@@ -1919,24 +1923,6 @@ export class RunAttemptSystem {
     };
   }
 
-  #getUserProvidedIdempotencyKey(
-    run: { idempotencyKey: string | null; idempotencyKeyOptions: unknown }
-  ): string | null {
-    const options = run.idempotencyKeyOptions as { key?: string; scope?: string } | null;
-    // Return user-provided key if available, otherwise fall back to the hash
-    return options?.key ?? run.idempotencyKey;
-  }
-
-  #getIdempotencyKeyScope(
-    run: { idempotencyKeyOptions: unknown }
-  ): "run" | "attempt" | "global" | undefined {
-    const options = run.idempotencyKeyOptions as { key?: string; scope?: string } | null;
-    const scope = options?.scope;
-    if (scope === "run" || scope === "attempt" || scope === "global") {
-      return scope;
-    }
-    return undefined;
-  }
 }
 
 export function safeParseGitMeta(git: unknown): GitMeta | undefined {

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

@@ -0,0 +1,52 @@
+import { IdempotencyKeyOptionsSchema } from "../schemas/api.js";
+
+/**
+ * Safely parses idempotencyKeyOptions from a database record and extracts the user-provided key.
+ * Returns the user-provided key if valid options exist, otherwise falls back to the hash.
+ *
+ * @param run - Object containing idempotencyKey (the hash) and idempotencyKeyOptions (JSON from DB)
+ * @returns The user-provided key, the hash as fallback, or null if neither exists
+ */
+export function getUserProvidedIdempotencyKey(run: {
+  idempotencyKey: string | null;
+  idempotencyKeyOptions: unknown;
+}): string | null {
+  const parsed = IdempotencyKeyOptionsSchema.safeParse(run.idempotencyKeyOptions);
+  if (parsed.success) {
+    return parsed.data.key;
+  }
+  return run.idempotencyKey;
+}
+
+/**
+ * Safely parses idempotencyKeyOptions and extracts the scope.
+ *
+ * @param run - Object containing idempotencyKeyOptions (JSON from DB)
+ * @returns The scope if valid options exist, otherwise undefined
+ */
+export function getIdempotencyKeyScope(run: {
+  idempotencyKeyOptions: unknown;
+}): "run" | "attempt" | "global" | undefined {
+  const parsed = IdempotencyKeyOptionsSchema.safeParse(run.idempotencyKeyOptions);
+  if (parsed.success) {
+    return parsed.data.scope;
+  }
+  return 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.
+ *
+ * @param run - Object containing idempotencyKeyOptions (JSON from DB)
+ * @returns The user-provided key if valid options exist, otherwise undefined
+ */
+export function extractIdempotencyKeyUser(run: {
+  idempotencyKeyOptions: unknown;
+}): string | undefined {
+  const parsed = IdempotencyKeyOptionsSchema.safeParse(run.idempotencyKeyOptions);
+  if (parsed.success) {
+    return parsed.data.key;
+  }
+  return undefined;
+}

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

@@ -8,3 +8,4 @@ export * from "./jumpHash.js";
 export * from "../apiClient/version.js";
 export * from "./placementTags.js";
 export * from "./resourceMonitor.js";
+export * from "./idempotencyKeys.js";

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

@@ -1,4 +1,4 @@
-import { batch, idempotencyKeys, logger, task, timeout, usage, wait } from "@trigger.dev/sdk/v3";
+import { batch, idempotencyKeys, logger, runs, task, timeout, usage, wait } from "@trigger.dev/sdk/v3";
 import { setTimeout } from "timers/promises";
 import { childTask } from "./example.js";
 
@@ -388,3 +388,155 @@ export const idempotencyKeyOptionsTest = task({
     };
   },
 });
+
+// Test task for verifying idempotencyKeys.reset works with the new API (TRI-4352)
+export const idempotencyKeyResetTest = task({
+  id: "idempotency-key-reset-test",
+  maxDuration: 120,
+  run: async (payload: any, { ctx }) => {
+    logger.log("Testing idempotencyKeys.reset feature (TRI-4352)");
+
+    const testResults: Array<{
+      test: string;
+      success: boolean;
+      details: Record<string, unknown>;
+    }> = [];
+
+    // Test 1: Reset using IdempotencyKey object (options extracted automatically)
+    {
+      const key = await idempotencyKeys.create("reset-test-key-1", { scope: "global" });
+      logger.log("Test 1: Created global-scoped key", { key: key.toString() });
+
+      // First trigger - should create a new run
+      const result1 = await idempotencyKeyOptionsChild.triggerAndWait(
+        { message: "First trigger" },
+        { idempotencyKey: key, idempotencyKeyTTL: "300s" }
+      );
+      const firstRunId = result1.ok ? result1.id : null;
+      logger.log("Test 1: First trigger", { runId: firstRunId });
+
+      // Second trigger - should be deduplicated (same run ID)
+      const result2 = await idempotencyKeyOptionsChild.triggerAndWait(
+        { message: "Second trigger (should dedupe)" },
+        { idempotencyKey: key, idempotencyKeyTTL: "300s" }
+      );
+      const secondRunId = result2.ok ? result2.id : null;
+      logger.log("Test 1: Second trigger (dedupe check)", { runId: secondRunId });
+
+      const wasDeduplicated = firstRunId === secondRunId;
+
+      // Reset the idempotency key using the IdempotencyKey object
+      logger.log("Test 1: Resetting idempotency key using IdempotencyKey object");
+      await idempotencyKeys.reset("idempotency-key-options-child", key);
+
+      // Third trigger - should create a NEW run after reset
+      const result3 = await idempotencyKeyOptionsChild.triggerAndWait(
+        { message: "Third trigger (after reset)" },
+        { idempotencyKey: key, idempotencyKeyTTL: "300s" }
+      );
+      const thirdRunId = result3.ok ? result3.id : null;
+      logger.log("Test 1: Third trigger (after reset)", { runId: thirdRunId });
+
+      const wasResetSuccessful = thirdRunId !== firstRunId && thirdRunId !== null;
+
+      testResults.push({
+        test: "Reset with IdempotencyKey object (global scope)",
+        success: wasDeduplicated && wasResetSuccessful,
+        details: {
+          firstRunId,
+          secondRunId,
+          thirdRunId,
+          wasDeduplicated,
+          wasResetSuccessful,
+        },
+      });
+    }
+
+    // Test 2: Reset using raw string with scope option
+    {
+      const keyString = "reset-test-key-2";
+      const key = await idempotencyKeys.create(keyString, { scope: "global" });
+      logger.log("Test 2: Created global-scoped key from string", { key: key.toString() });
+
+      // First trigger
+      const result1 = await idempotencyKeyOptionsChild.triggerAndWait(
+        { message: "First trigger (raw string test)" },
+        { idempotencyKey: key, idempotencyKeyTTL: "300s" }
+      );
+      const firstRunId = result1.ok ? result1.id : null;
+      logger.log("Test 2: First trigger", { runId: firstRunId });
+
+      // Reset using raw string + scope option
+      logger.log("Test 2: Resetting idempotency key using raw string + scope");
+      await idempotencyKeys.reset("idempotency-key-options-child", keyString, { scope: "global" });
+
+      // Second trigger - should create a NEW run after reset
+      const result2 = await idempotencyKeyOptionsChild.triggerAndWait(
+        { message: "Second trigger (after reset with raw string)" },
+        { idempotencyKey: key, idempotencyKeyTTL: "300s" }
+      );
+      const secondRunId = result2.ok ? result2.id : null;
+      logger.log("Test 2: Second trigger (after reset)", { runId: secondRunId });
+
+      const wasResetSuccessful = secondRunId !== firstRunId && secondRunId !== null;
+
+      testResults.push({
+        test: "Reset with raw string + scope option (global scope)",
+        success: wasResetSuccessful,
+        details: {
+          firstRunId,
+          secondRunId,
+          wasResetSuccessful,
+        },
+      });
+    }
+
+    // Test 3: Reset with run scope (uses current run context)
+    {
+      const key = await idempotencyKeys.create("reset-test-key-3", { scope: "run" });
+      logger.log("Test 3: Created run-scoped key", { key: key.toString() });
+
+      // First trigger
+      const result1 = await idempotencyKeyOptionsChild.triggerAndWait(
+        { message: "First trigger (run scope)" },
+        { idempotencyKey: key, idempotencyKeyTTL: "300s" }
+      );
+      const firstRunId = result1.ok ? result1.id : null;
+      logger.log("Test 3: First trigger", { runId: firstRunId });
+
+      // Reset using IdempotencyKey (run scope - should use current run context)
+      logger.log("Test 3: Resetting idempotency key with run scope");
+      await idempotencyKeys.reset("idempotency-key-options-child", key);
+
+      // Second trigger - should create a NEW run after reset
+      const result2 = await idempotencyKeyOptionsChild.triggerAndWait(
+        { message: "Second trigger (after reset, run scope)" },
+        { idempotencyKey: key, idempotencyKeyTTL: "300s" }
+      );
+      const secondRunId = result2.ok ? result2.id : null;
+      logger.log("Test 3: Second trigger (after reset)", { runId: secondRunId });
+
+      const wasResetSuccessful = secondRunId !== firstRunId && secondRunId !== null;
+
+      testResults.push({
+        test: "Reset with IdempotencyKey object (run scope)",
+        success: wasResetSuccessful,
+        details: {
+          firstRunId,
+          secondRunId,
+          wasResetSuccessful,
+          parentRunId: ctx.run.id,
+        },
+      });
+    }
+
+    // Summary
+    const allPassed = testResults.every((r) => r.success);
+    logger.log("Test summary", { allPassed, testResults });
+
+    return {
+      allPassed,
+      testResults,
+    };
+  },
+});