Implement request idempotency on trigger

Author: ericallam

apps/webapp/app/routes/api.v1.tasks.$taskId.trigger.ts

@@ -6,6 +6,7 @@ import {
 } from "@trigger.dev/core/v3";
 import { TaskRun } from "@trigger.dev/database";
 import { z } from "zod";
+import { prisma } from "~/db.server";
 import { env } from "~/env.server";
 import { EngineServiceValidationError } from "~/runEngine/concerns/errors";
 import {
@@ -14,6 +15,7 @@ import {
   getOneTimeUseToken,
 } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
+import { requestIdempotency } from "~/services/requestIdempotencyInstance.server";
 import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
 import { resolveIdempotencyKeyTTL } from "~/utils/idempotencyKeys.server";
 import { ServiceValidationError } from "~/v3/services/baseService.server";
@@ -31,6 +33,7 @@ export const HeadersSchema = z.object({
   "x-trigger-worker": z.string().nullish(),
   "x-trigger-client": z.string().nullish(),
   "x-trigger-engine-version": RunEngineVersionSchema.nullish(),
+  "x-trigger-request-id": z.string().nullish(),
   traceparent: z.string().optional(),
   tracestate: z.string().optional(),
 });
@@ -60,8 +63,53 @@ const { action, loader } = createActionApiRoute(
       "x-trigger-worker": isFromWorker,
       "x-trigger-client": triggerClient,
       "x-trigger-engine-version": engineVersion,
+      "x-trigger-request-id": requestId,
     } = headers;
 
+    if (requestId) {
+      logger.debug("request-idempotency: checking for cached trigger request", {
+        requestId,
+      });
+
+      const cachedRequest = await requestIdempotency.checkRequest("trigger", requestId);
+
+      if (cachedRequest) {
+        logger.info("request-idempotency: found cached trigger request", {
+          requestId,
+          cachedRequest,
+        });
+
+        const cachedRun = await prisma.taskRun.findFirst({
+          where: {
+            id: cachedRequest.id,
+          },
+          select: {
+            friendlyId: true,
+          },
+        });
+
+        if (cachedRun) {
+          logger.info("request-idempotency: found cached trigger run", {
+            requestId,
+            cachedRun,
+          });
+
+          const $responseHeaders = await responseHeaders(cachedRun, authentication, triggerClient);
+
+          return json(
+            {
+              id: cachedRun.friendlyId,
+              isCached: false,
+            },
+            {
+              headers: $responseHeaders,
+              status: 200,
+            }
+          );
+        }
+      }
+    }
+
     const service = new TriggerTaskService();
 
     try {
@@ -104,6 +152,12 @@ const { action, loader } = createActionApiRoute(
         return json({ error: "Task not found" }, { status: 404 });
       }
 
+      if (requestId) {
+        await requestIdempotency.saveRequest("trigger", requestId, {
+          id: result.run.id,
+        });
+      }
+
       const $responseHeaders = await responseHeaders(result.run, authentication, triggerClient);
 
       return json(
@@ -113,6 +167,7 @@ const { action, loader } = createActionApiRoute(
         },
         {
           headers: $responseHeaders,
+          status: 200,
         }
       );
     } catch (error) {
@@ -132,7 +187,7 @@ const { action, loader } = createActionApiRoute(
 );
 
 async function responseHeaders(
-  run: TaskRun,
+  run: Pick<TaskRun, "friendlyId">,
   authentication: ApiAuthenticationResultSuccess,
   triggerClient?: string | null
 ): Promise<Record<string, string>> {

apps/webapp/app/routes/api.v2.tasks.batch.ts

@@ -21,7 +21,6 @@ const { action, loader } = createActionApiRoute(
   {
     headers: HeadersSchema.extend({
       "batch-processing-strategy": BatchProcessingStrategy.nullish(),
-      "x-trigger-request-id": z.string().nullish(),
     }),
     body: BatchTriggerTaskV3RequestBody,
     allowJWT: true,

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

@@ -3,6 +3,7 @@ import { createCache, DefaultStatefulContext, Namespace, Cache as UnkeyCache } f
 import { MemoryStore } from "@unkey/cache/stores";
 import { RedisCacheStore } from "./unkey/redisCacheStore.server";
 import { RedisWithClusterOptions } from "~/redis.server";
+import { validate as uuidValidate, version as uuidVersion } from "uuid";
 
 export type RequestIdempotencyServiceOptions<TTypes extends string> = {
   types: TTypes[];
@@ -53,6 +54,14 @@ export class RequestIdempotencyService<TTypes extends string> {
   }
 
   async checkRequest(type: TTypes, requestId: string) {
+    if (!this.#validateRequestId(requestId)) {
+      this.logger.warn("RequestIdempotency: invalid requestId", {
+        requestId,
+      });
+
+      return undefined;
+    }
+
     const key = `${type}:${requestId}`;
     const result = await this.cache.requests.get(key);
 
@@ -67,6 +76,13 @@ export class RequestIdempotencyService<TTypes extends string> {
   }
 
   async saveRequest(type: TTypes, requestId: string, value: RequestIdempotencyCacheEntry) {
+    if (!this.#validateRequestId(requestId)) {
+      this.logger.warn("RequestIdempotency: invalid requestId", {
+        requestId,
+      });
+      return undefined;
+    }
+
     const key = `${type}:${requestId}`;
     const result = await this.cache.requests.set(key, value);
 
@@ -86,4 +102,13 @@ export class RequestIdempotencyService<TTypes extends string> {
 
     return result;
   }
+
+  // The requestId should be a valid UUID
+  #validateRequestId(requestId: string): boolean {
+    return isValidV4UUID(requestId);
+  }
+}
+
+function isValidV4UUID(uuid: string): boolean {
+  return uuidValidate(uuid) && uuidVersion(uuid) === 4;
 }

apps/webapp/app/v3/services/triggerTask.server.ts

@@ -15,6 +15,7 @@ import { TriggerTaskServiceV1 } from "./triggerTaskV1.server";
 import { DefaultTraceEventsConcern } from "~/runEngine/concerns/traceEvents.server";
 import { DefaultRunChainStateManager } from "~/runEngine/concerns/runChainStates.server";
 import { env } from "~/env.server";
+import { Evt } from "evt";
 
 export type TriggerTaskServiceOptions = {
   idempotencyKey?: string;
@@ -113,6 +114,7 @@ export class TriggerTaskService extends WithRunEngine {
       ),
       tracer: tracer,
     });
+
     return await service.call({
       taskId,
       environment,

apps/webapp/package.json

@@ -198,6 +198,7 @@
     "tiny-invariant": "^1.2.0",
     "ulid": "^2.3.0",
     "ulidx": "^2.2.1",
+    "uuid": "^9.0.0",
     "ws": "^8.11.0",
     "zod": "3.23.8",
     "zod-error": "1.5.0",
@@ -239,6 +240,7 @@
     "@types/slug": "^5.0.3",
     "@types/supertest": "^6.0.2",
     "@types/tar": "^6.1.4",
+    "@types/uuid": "^9.0.0",
     "@types/ws": "^8.5.3",
     "@typescript-eslint/eslint-plugin": "^5.59.6",
     "@typescript-eslint/parser": "^5.59.6",