Removed batch idempotency because it gets put on the runs instead

matt-aitken · matt-aitken · commit efc83c989419 · 2025-01-13T15:00:47.000Z
diff --git a/apps/webapp/app/routes/api.v2.tasks.batch.ts b/apps/webapp/app/routes/api.v2.tasks.batch.ts
@@ -1,7 +1,5 @@
 import { json } from "@remix-run/server-runtime";
 import {
-  BatchTriggerTaskV2RequestBody,
-  BatchTriggerTaskV2Response,
   BatchTriggerTaskV3RequestBody,
   BatchTriggerTaskV3Response,
   generateJWT,
@@ -10,7 +8,6 @@ import { env } from "~/env.server";
 import { AuthenticatedEnvironment, getOneTimeUseToken } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
 import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
-import { resolveIdempotencyKeyTTL } from "~/utils/idempotencyKeys.server";
 import { ServiceValidationError } from "~/v3/services/baseService.server";
 import {
   BatchProcessingStrategy,
@@ -52,8 +49,6 @@ const { action, loader } = createActionApiRoute(
     }
 
     const {
-      "idempotency-key": idempotencyKey,
-      "idempotency-key-ttl": idempotencyKeyTTL,
       "trigger-version": triggerVersion,
       "x-trigger-span-parent-as-link": spanParentAsLink,
       "x-trigger-worker": isFromWorker,
@@ -67,8 +62,6 @@ const { action, loader } = createActionApiRoute(
     const oneTimeUseToken = await getOneTimeUseToken(authentication);
 
     logger.debug("Batch trigger request", {
-      idempotencyKey,
-      idempotencyKeyTTL,
       triggerVersion,
       spanParentAsLink,
       isFromWorker,
@@ -83,17 +76,10 @@ const { action, loader } = createActionApiRoute(
         ? { traceparent, tracestate }
         : undefined;
 
-    // By default, the idempotency key expires in 30 days
-    const idempotencyKeyExpiresAt =
-      resolveIdempotencyKeyTTL(idempotencyKeyTTL) ??
-      new Date(Date.now() + 24 * 60 * 60 * 1000 * 30);
-
     const service = new BatchTriggerV3Service(batchProcessingStrategy ?? undefined);
 
     try {
       const batch = await service.call(authentication.environment, body, {
-        idempotencyKey: idempotencyKey ?? undefined,
-        idempotencyKeyExpiresAt,
         triggerVersion: triggerVersion ?? undefined,
         traceContext,
         spanParentAsLink: spanParentAsLink === 1,
diff --git a/apps/webapp/app/v3/services/batchTriggerV3.server.ts b/apps/webapp/app/v3/services/batchTriggerV3.server.ts
@@ -18,7 +18,6 @@ import { downloadPacketFromObjectStore, uploadPacketToObjectStore } from "../r2.
 import { startActiveSpan } from "../tracer.server";
 import { ServiceValidationError, WithRunEngine } from "./baseService.server";
 import { OutOfEntitlementError, TriggerTaskService } from "./triggerTask.server";
-import { guardQueueSizeLimitsForEnv } from "./triggerTaskV2.server";
 
 const PROCESSING_BATCH_SIZE = 50;
 const ASYNC_BATCH_PROCESS_SIZE_THRESHOLD = 20;
@@ -40,8 +39,6 @@ export const BatchProcessingOptions = z.object({
 export type BatchProcessingOptions = z.infer<typeof BatchProcessingOptions>;
 
 export type BatchTriggerTaskServiceOptions = {
-  idempotencyKey?: string;
-  idempotencyKeyExpiresAt?: Date;
   triggerVersion?: string;
   traceContext?: Record<string, string | undefined>;
   spanParentAsLink?: boolean;
@@ -73,61 +70,6 @@ export class BatchTriggerV3Service extends WithRunEngine {
         "call()",
         environment,
         async (span) => {
-          const existingBatch = options.idempotencyKey
-            ? await this._prisma.batchTaskRun.findUnique({
-                where: {
-                  runtimeEnvironmentId_idempotencyKey: {
-                    runtimeEnvironmentId: environment.id,
-                    idempotencyKey: options.idempotencyKey,
-                  },
-                },
-              })
-            : undefined;
-
-          if (existingBatch) {
-            if (
-              existingBatch.idempotencyKeyExpiresAt &&
-              existingBatch.idempotencyKeyExpiresAt < new Date()
-            ) {
-              logger.debug("[BatchTriggerV3][call] Idempotency key has expired", {
-                idempotencyKey: options.idempotencyKey,
-                batch: {
-                  id: existingBatch.id,
-                  friendlyId: existingBatch.friendlyId,
-                  runCount: existingBatch.runCount,
-                  idempotencyKeyExpiresAt: existingBatch.idempotencyKeyExpiresAt,
-                  idempotencyKey: existingBatch.idempotencyKey,
-                },
-              });
-
-              // Update the existing batch to remove the idempotency key
-              await this._prisma.batchTaskRun.update({
-                where: { id: existingBatch.id },
-                data: { idempotencyKey: null },
-              });
-
-              // Don't return, just continue with the batch trigger
-            } else {
-              span.setAttribute("batchId", existingBatch.friendlyId);
-
-              //block the parent with all of the children
-              if (body.resumeParentOnCompletion && body.parentRunId) {
-                await this.#blockParentRun({
-                  parentRunId: body.parentRunId,
-                  childFriendlyIds: existingBatch.runIds,
-                  environment,
-                });
-              }
-
-              return {
-                id: existingBatch.friendlyId,
-                idempotencyKey: existingBatch.idempotencyKey ?? undefined,
-                isCached: true,
-                runCount: existingBatch.runCount,
-              };
-            }
-          }
-
           const { id, friendlyId } = BatchId.generate();
 
           span.setAttribute("batchId", friendlyId);
@@ -212,8 +154,6 @@ export class BatchTriggerV3Service extends WithRunEngine {
           id: BatchId.fromFriendlyId(batchId),
           friendlyId: batchId,
           runtimeEnvironmentId: environment.id,
-          idempotencyKey: options.idempotencyKey,
-          idempotencyKeyExpiresAt: options.idempotencyKeyExpiresAt,
           runCount: body.items.length,
           runIds: [],
           payload: payloadPacket.data,
@@ -305,8 +245,6 @@ export class BatchTriggerV3Service extends WithRunEngine {
             id: BatchId.fromFriendlyId(batchId),
             friendlyId: batchId,
             runtimeEnvironmentId: environment.id,
-            idempotencyKey: options.idempotencyKey,
-            idempotencyKeyExpiresAt: options.idempotencyKeyExpiresAt,
             runCount: body.items.length,
             runIds: [],
             payload: payloadPacket.data,
@@ -728,40 +666,4 @@ export class BatchTriggerV3Service extends WithRunEngine {
       };
     });
   }
-
-  //todo what if the idempotent batch hasn't finished creating all the runs yet?!
-  async #blockParentRun({
-    parentRunId,
-    childFriendlyIds,
-    environment,
-  }: {
-    parentRunId: string;
-    childFriendlyIds: string[];
-    environment: AuthenticatedEnvironment;
-  }) {
-    const runsWithAssociatedWaitpoints = await this._prisma.taskRun.findMany({
-      where: {
-        id: {
-          in: childFriendlyIds.map((r) => RunId.fromFriendlyId(r)),
-        },
-      },
-      select: {
-        associatedWaitpoint: {
-          select: {
-            id: true,
-          },
-        },
-      },
-    });
-
-    await this._engine.blockRunWithWaitpoint({
-      runId: RunId.fromFriendlyId(parentRunId),
-      waitpoints: runsWithAssociatedWaitpoints.flatMap((r) => {
-        if (!r.associatedWaitpoint) return [];
-        return [r.associatedWaitpoint.id];
-      }),
-      environmentId: environment.id,
-      projectId: environment.projectId,
-    });
-  }
 }
diff --git a/references/hello-world/src/trigger/idempotency.ts b/references/hello-world/src/trigger/idempotency.ts
@@ -118,23 +118,51 @@ export const idempotencyBatch = task({
     const successfulKey = await idempotencyKeys.create("a", { scope: "global" });
     const failureKey = await idempotencyKeys.create("b", { scope: "global" });
     const anotherKey = await idempotencyKeys.create("c", { scope: "global" });
+    const batchKey = await idempotencyKeys.create("batch", { scope: "global" });
+
+    const batch1 = await childTask.batchTriggerAndWait(
+      [
+        {
+          payload: { message: "Hello, world!" },
+          options: { idempotencyKey: successfulKey, idempotencyKeyTTL: "120s" },
+        },
+        {
+          payload: { message: "Hello, world 2!" },
+          options: { idempotencyKey: failureKey, idempotencyKeyTTL: "120s" },
+        },
+        {
+          payload: { message: "Hello, world 3", duration: 500, failureChance: 0 },
+        },
+      ],
+      {
+        idempotencyKey: batchKey,
+        idempotencyKeyTTL: "120s",
+      }
+    );
+    logger.log("Batch 1", { batch1 });
 
-    const batch1 = await childTask.batchTriggerAndWait([
-      {
-        payload: { message: "Hello, world!" },
-        options: { idempotencyKey: successfulKey, idempotencyKeyTTL: "120s" },
-      },
-      {
-        payload: { message: "Hello, world 2!" },
-        options: { idempotencyKey: failureKey, idempotencyKeyTTL: "120s" },
-      },
-      {
-        payload: { message: "Hello, world 3", duration: 500, failureChance: 0 },
-      },
-    ]);
+    const batch2 = await childTask.batchTriggerAndWait(
+      [
+        {
+          payload: { message: "Hello, world!" },
+          options: { idempotencyKey: successfulKey, idempotencyKeyTTL: "120s" },
+        },
+        {
+          payload: { message: "Hello, world 2!" },
+          options: { idempotencyKey: failureKey, idempotencyKeyTTL: "120s" },
+        },
+        {
+          payload: { message: "Hello, world 3", duration: 500, failureChance: 0 },
+        },
+      ],
+      {
+        idempotencyKey: batchKey,
+        idempotencyKeyTTL: "120s",
+      }
+    );
     logger.log("Batch 1", { batch1 });
 
-    const batch2 = await childTask.batchTriggerAndWait([
+    await childTask.batchTrigger([
       {
         payload: { message: "Hello, world!" },
         options: { idempotencyKey: successfulKey, idempotencyKeyTTL: "120s" },
@@ -143,21 +171,14 @@ export const idempotencyBatch = task({
         payload: { message: "Hello, world 2!" },
         options: { idempotencyKey: failureKey, idempotencyKeyTTL: "120s" },
       },
-      {
-        payload: { message: "Hello, world 3", duration: 500, failureChance: 0 },
-        options: { idempotencyKey: anotherKey, idempotencyKeyTTL: "120s" },
-      },
     ]);
-    logger.log("Batch 1", { batch1 });
 
     await childTask.batchTrigger([
       {
         payload: { message: "Hello, world!" },
-        options: { idempotencyKey: successfulKey, idempotencyKeyTTL: "120s" },
       },
       {
         payload: { message: "Hello, world 2!" },
-        options: { idempotencyKey: failureKey, idempotencyKeyTTL: "120s" },
       },
     ]);
 
