Introduce request idempotency to prevent duplicate triggers

ericallam · ericallam · commit 0419fe8dd867 · 2025-07-10T14:19:57.000+01:00
diff --git a/apps/webapp/app/env.server.ts b/apps/webapp/app/env.server.ts
@@ -932,6 +932,44 @@ const EnvironmentSchema = z.object({
   TRIGGER_CLI_TAG: z.string().default("latest"),
 
   HEALTHCHECK_DATABASE_DISABLED: z.string().default("0"),
+
+  REQUEST_IDEMPOTENCY_REDIS_HOST: z
+    .string()
+    .optional()
+    .transform((v) => v ?? process.env.REDIS_HOST),
+  REQUEST_IDEMPOTENCY_REDIS_READER_HOST: z
+    .string()
+    .optional()
+    .transform((v) => v ?? process.env.REDIS_READER_HOST),
+  REQUEST_IDEMPOTENCY_REDIS_READER_PORT: z.coerce
+    .number()
+    .optional()
+    .transform(
+      (v) =>
+        v ?? (process.env.REDIS_READER_PORT ? parseInt(process.env.REDIS_READER_PORT) : undefined)
+    ),
+  REQUEST_IDEMPOTENCY_REDIS_PORT: z.coerce
+    .number()
+    .optional()
+    .transform((v) => v ?? (process.env.REDIS_PORT ? parseInt(process.env.REDIS_PORT) : undefined)),
+  REQUEST_IDEMPOTENCY_REDIS_USERNAME: z
+    .string()
+    .optional()
+    .transform((v) => v ?? process.env.REDIS_USERNAME),
+  REQUEST_IDEMPOTENCY_REDIS_PASSWORD: z
+    .string()
+    .optional()
+    .transform((v) => v ?? process.env.REDIS_PASSWORD),
+  REQUEST_IDEMPOTENCY_REDIS_TLS_DISABLED: z
+    .string()
+    .default(process.env.REDIS_TLS_DISABLED ?? "false"),
+
+  REQUEST_IDEMPOTENCY_LOG_LEVEL: z.enum(["log", "error", "warn", "info", "debug"]).default("info"),
+
+  REQUEST_IDEMPOTENCY_TTL_IN_MS: z.coerce
+    .number()
+    .int()
+    .default(60_000 * 60 * 24),
 });
 
 export type Environment = z.infer<typeof EnvironmentSchema>;
diff --git a/apps/webapp/app/routes/api.v2.tasks.batch.ts b/apps/webapp/app/routes/api.v2.tasks.batch.ts
@@ -13,11 +13,15 @@ import { BatchProcessingStrategy } from "~/v3/services/batchTriggerV3.server";
 import { OutOfEntitlementError } from "~/v3/services/triggerTask.server";
 import { HeadersSchema } from "./api.v1.tasks.$taskId.trigger";
 import { RunEngineBatchTriggerService } from "~/runEngine/services/batchTrigger.server";
+import { z } from "zod";
+import { requestIdempotency } from "~/services/requestIdempotencyInstance.server";
+import { prisma } from "~/db.server";
 
 const { action, loader } = createActionApiRoute(
   {
     headers: HeadersSchema.extend({
       "batch-processing-strategy": BatchProcessingStrategy.nullish(),
+      "x-trigger-request-id": z.string().nullish(),
     }),
     body: BatchTriggerTaskV3RequestBody,
     allowJWT: true,
@@ -53,6 +57,7 @@ const { action, loader } = createActionApiRoute(
       "x-trigger-client": triggerClient,
       "x-trigger-engine-version": engineVersion,
       "batch-processing-strategy": batchProcessingStrategy,
+      "x-trigger-request-id": requestId,
       traceparent,
       tracestate,
     } = headers;
@@ -67,15 +72,72 @@ const { action, loader } = createActionApiRoute(
       traceparent,
       tracestate,
       batchProcessingStrategy,
+      requestId,
     });
 
+    if (requestId) {
+      logger.debug("request-idempotency: checking for cached batch-trigger request", {
+        requestId,
+      });
+
+      const cachedRequest = await requestIdempotency.checkRequest("batch-trigger", requestId);
+
+      if (cachedRequest) {
+        logger.info("request-idempotency: found cached batch-trigger request", {
+          requestId,
+          cachedRequest,
+        });
+
+        // Find the batch and return it as a 200 response
+        const cachedBatch = await prisma.batchTaskRun.findFirst({
+          where: {
+            id: cachedRequest.id,
+            runtimeEnvironmentId: authentication.environment.id,
+          },
+          select: {
+            friendlyId: true,
+            runCount: true,
+          },
+        });
+
+        if (cachedBatch) {
+          logger.info("request-idempotency: found cached batch-trigger request", {
+            requestId,
+            cachedRequest,
+            cachedBatch,
+          });
+
+          const responseBody = {
+            id: cachedBatch.friendlyId,
+            runCount: cachedBatch.runCount,
+          };
+
+          const $responseHeaders = await responseHeaders(
+            responseBody,
+            authentication.environment,
+            triggerClient
+          );
+
+          return json(responseBody, { status: 200, headers: $responseHeaders });
+        }
+      }
+    }
+
     const traceContext =
       traceparent && isFromWorker // If the request is from a worker, we should pass the trace context
         ? { traceparent, tracestate }
         : undefined;
 
     const service = new RunEngineBatchTriggerService(batchProcessingStrategy ?? undefined);
 
+    service.onBatchTaskRunCreated.attachOnce(async (batch) => {
+      if (requestId) {
+        await requestIdempotency.saveRequest("batch-trigger", requestId, {
+          id: batch.id,
+        });
+      }
+    });
+
     try {
       const batch = await service.call(authentication.environment, body, {
         triggerVersion: triggerVersion ?? undefined,
@@ -90,7 +152,10 @@ const { action, loader } = createActionApiRoute(
         triggerClient
       );
 
-      return json(batch, { status: 202, headers: $responseHeaders });
+      return json(batch, {
+        status: 202,
+        headers: $responseHeaders,
+      });
     } catch (error) {
       logger.error("Batch trigger error", {
         error: {
diff --git a/apps/webapp/app/runEngine/services/batchTrigger.server.ts b/apps/webapp/app/runEngine/services/batchTrigger.server.ts
@@ -8,8 +8,9 @@ import {
 } from "@trigger.dev/core/v3";
 import { BatchId, RunId } from "@trigger.dev/core/v3/isomorphic";
 import { BatchTaskRun, Prisma } from "@trigger.dev/database";
+import { Evt } from "evt";
 import { z } from "zod";
-import { $transaction, prisma, PrismaClientOrTransaction } from "~/db.server";
+import { prisma, PrismaClientOrTransaction } from "~/db.server";
 import { env } from "~/env.server";
 import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
@@ -51,6 +52,7 @@ export type BatchTriggerTaskServiceOptions = {
  */
 export class RunEngineBatchTriggerService extends WithRunEngine {
   private _batchProcessingStrategy: BatchProcessingStrategy;
+  public onBatchTaskRunCreated: Evt<BatchTaskRun> = new Evt();
 
   constructor(
     batchProcessingStrategy?: BatchProcessingStrategy,
@@ -168,6 +170,8 @@ export class RunEngineBatchTriggerService extends WithRunEngine {
         },
       });
 
+      this.onBatchTaskRunCreated.post(batch);
+
       if (body.parentRunId && body.resumeParentOnCompletion) {
         await this._engine.blockRunWithCreatedBatch({
           runId: RunId.fromFriendlyId(body.parentRunId),
@@ -259,6 +263,8 @@ export class RunEngineBatchTriggerService extends WithRunEngine {
         },
       });
 
+      this.onBatchTaskRunCreated.post(batch);
+
       if (body.parentRunId && body.resumeParentOnCompletion) {
         await this._engine.blockRunWithCreatedBatch({
           runId: RunId.fromFriendlyId(body.parentRunId),
diff --git a/apps/webapp/app/services/requestIdempotency.server.ts b/apps/webapp/app/services/requestIdempotency.server.ts
@@ -0,0 +1,89 @@
+import { Logger, LogLevel } from "@trigger.dev/core/logger";
+import { createCache, DefaultStatefulContext, Namespace, Cache as UnkeyCache } from "@unkey/cache";
+import { MemoryStore } from "@unkey/cache/stores";
+import { RedisCacheStore } from "./unkey/redisCacheStore.server";
+import { RedisWithClusterOptions } from "~/redis.server";
+
+export type RequestIdempotencyServiceOptions<TTypes extends string> = {
+  types: TTypes[];
+  redis: RedisWithClusterOptions;
+  logger?: Logger;
+  logLevel?: LogLevel;
+  ttlInMs?: number;
+};
+
+const DEFAULT_TTL_IN_MS = 60_000 * 60 * 24;
+
+type RequestIdempotencyCacheEntry = {
+  id: string;
+};
+
+export class RequestIdempotencyService<TTypes extends string> {
+  private readonly logger: Logger;
+  private readonly cache: UnkeyCache<{ requests: RequestIdempotencyCacheEntry }>;
+
+  constructor(private readonly options: RequestIdempotencyServiceOptions<TTypes>) {
+    this.logger =
+      options.logger ?? new Logger("RequestIdempotencyService", options.logLevel ?? "info");
+
+    const keyPrefix = options.redis.keyPrefix
+      ? `request-idempotency:${options.redis.keyPrefix}`
+      : "request-idempotency:";
+
+    const ctx = new DefaultStatefulContext();
+    const memory = new MemoryStore({ persistentMap: new Map() });
+    const redisCacheStore = new RedisCacheStore({
+      name: "request-idempotency",
+      connection: {
+        keyPrefix: keyPrefix,
+        ...options.redis,
+      },
+    });
+
+    // This cache holds the rate limit configuration for each org, so we don't have to fetch it every request
+    const cache = createCache({
+      requests: new Namespace<RequestIdempotencyCacheEntry>(ctx, {
+        stores: [memory, redisCacheStore],
+        fresh: options.ttlInMs ?? DEFAULT_TTL_IN_MS,
+        stale: options.ttlInMs ?? DEFAULT_TTL_IN_MS,
+      }),
+    });
+
+    this.cache = cache;
+  }
+
+  async checkRequest(type: TTypes, requestId: string) {
+    const key = `${type}:${requestId}`;
+    const result = await this.cache.requests.get(key);
+
+    this.logger.debug("RequestIdempotency: checking request", {
+      type,
+      requestId,
+      key,
+      result,
+    });
+
+    return result.val ? result.val : undefined;
+  }
+
+  async saveRequest(type: TTypes, requestId: string, value: RequestIdempotencyCacheEntry) {
+    const key = `${type}:${requestId}`;
+    const result = await this.cache.requests.set(key, value);
+
+    if (result.err) {
+      this.logger.error("RequestIdempotency: error saving request", {
+        key,
+        error: result.err,
+      });
+    } else {
+      this.logger.debug("RequestIdempotency: saved request", {
+        type,
+        requestId,
+        key,
+        value,
+      });
+    }
+
+    return result;
+  }
+}
diff --git a/apps/webapp/app/services/requestIdempotencyInstance.server.ts b/apps/webapp/app/services/requestIdempotencyInstance.server.ts
@@ -0,0 +1,22 @@
+import { singleton } from "~/utils/singleton";
+import { RequestIdempotencyService } from "./requestIdempotency.server";
+import { env } from "~/env.server";
+
+export const requestIdempotency = singleton("requestIdempotency", createRequestIdempotencyInstance);
+
+function createRequestIdempotencyInstance() {
+  return new RequestIdempotencyService({
+    redis: {
+      keyPrefix: "request-idempotency:",
+      port: env.REQUEST_IDEMPOTENCY_REDIS_PORT ?? undefined,
+      host: env.REQUEST_IDEMPOTENCY_REDIS_HOST ?? undefined,
+      username: env.REQUEST_IDEMPOTENCY_REDIS_USERNAME ?? undefined,
+      password: env.REQUEST_IDEMPOTENCY_REDIS_PASSWORD ?? undefined,
+      tlsDisabled: env.REQUEST_IDEMPOTENCY_REDIS_TLS_DISABLED === "true",
+      clusterMode: false,
+    },
+    logLevel: env.REQUEST_IDEMPOTENCY_LOG_LEVEL,
+    ttlInMs: env.REQUEST_IDEMPOTENCY_TTL_IN_MS,
+    types: ["batch-trigger", "trigger"],
+  });
+}
diff --git a/apps/webapp/app/services/unkey/redisCacheStore.server.ts b/apps/webapp/app/services/unkey/redisCacheStore.server.ts
@@ -5,6 +5,7 @@ import { createRedisClient, RedisClient, RedisWithClusterOptions } from "~/redis
 
 export type RedisCacheStoreConfig = {
   connection: RedisWithClusterOptions;
+  name?: string;
 };
 
 export class RedisCacheStore<TNamespace extends string, TValue = any>
@@ -14,7 +15,7 @@ export class RedisCacheStore<TNamespace extends string, TValue = any>
   private readonly redis: RedisClient;
 
   constructor(config: RedisCacheStoreConfig) {
-    this.redis = createRedisClient("trigger:cacheStore", config.connection);
+    this.redis = createRedisClient(config.name ?? "trigger:cacheStore", config.connection);
   }
 
   private buildCacheKey(namespace: TNamespace, key: string): string {
diff --git a/packages/core/package.json b/packages/core/package.json
@@ -195,6 +195,7 @@
     "std-env": "^3.8.1",
     "superjson": "^2.2.1",
     "tinyexec": "^0.3.2",
+    "uncrypto": "^0.1.3",
     "zod": "3.23.8",
     "zod-error": "1.5.0",
     "zod-validation-error": "^1.5.0"
diff --git a/packages/core/src/v3/apiClient/core.ts b/packages/core/src/v3/apiClient/core.ts
@@ -18,6 +18,7 @@ import {
   OffsetLimitPageResponse,
 } from "./pagination.js";
 import { EventSource, type ErrorEvent } from "eventsource";
+import { randomUUID } from "../utils/crypto.js";
 
 export const defaultRetryOptions = {
   maxAttempts: 3,
@@ -200,7 +201,10 @@ async function _doZodFetch<TResponseBodySchema extends z.ZodTypeAny>(
   let $requestInit = await requestInit;
 
   return traceZodFetch({ url, requestInit: $requestInit, options }, async (span) => {
+    const requestId = await randomUUID();
+
     $requestInit = injectPropagationHeadersIfInWorker($requestInit);
+    $requestInit = injectRequestId(requestId, $requestInit);
 
     const result = await _doZodFetchWithRetries(schema, url, $requestInit, options);
 
@@ -627,6 +631,17 @@ function injectPropagationHeadersIfInWorker(requestInit?: RequestInit): RequestI
   };
 }
 
+function injectRequestId(requestId: string, requestInit?: RequestInit): RequestInit | undefined {
+  const headers = new Headers(requestInit?.headers);
+
+  headers.set("x-trigger-request-id", requestId);
+
+  return {
+    ...requestInit,
+    headers,
+  };
+}
+
 export type ZodFetchSSEMessageValueSchema<
   TDiscriminatedUnion extends z.ZodDiscriminatedUnion<any, any>,
 > = z.ZodFirstPartySchemaTypes | TDiscriminatedUnion;
diff --git a/packages/core/src/v3/idempotencyKeys.ts b/packages/core/src/v3/idempotencyKeys.ts
@@ -1,5 +1,6 @@
 import { taskContext } from "./task-context-api.js";
 import { IdempotencyKey } from "./types/idempotencyKeys.js";
+import { digestSHA256 } from "./utils/crypto.js";
 
 export function isIdempotencyKey(
   value: string | string[] | IdempotencyKey
@@ -115,15 +116,7 @@ function injectScope(scope: "run" | "attempt" | "global"): string[] {
 }
 
 async function generateIdempotencyKey(keyMaterial: string[]) {
-  const hash = await crypto.subtle.digest(
-    "SHA-256",
-    new TextEncoder().encode(keyMaterial.join("-"))
-  );
-
-  // Return a hex string, using cross-runtime compatible methods
-  return Array.from(new Uint8Array(hash))
-    .map((byte) => byte.toString(16).padStart(2, "0"))
-    .join("");
+  return await digestSHA256(keyMaterial.join("-"));
 }
 
 type AttemptKeyMaterial = {
diff --git a/packages/core/src/v3/utils/crypto.ts b/packages/core/src/v3/utils/crypto.ts
diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml
