From 81ebf24e3aec43e516098195a014f8f4b25381cb Mon Sep 17 00:00:00 2001 From: itzzdev09 Date: Wed, 9 Sep 2026 18:25:11 +0530 Subject: [PATCH] fix(batch): clear idempotency keys pointing at dead runs in batchTrigger MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The single-trigger path clears an idempotency key when the run it points at reached a clearable terminal state — `IdempotencyKeyConcern.handleExistingRun` guards on `shouldIdempotencyKeyBeCleared(existingRun.status)`. The batch path only tested time expiry, so `batchTrigger` with a key pointing at a dead run returned that FAILED run as `isCached: true`, and kept returning it on every retry. The batch path could not make that check: `findRunsByIdempotencyKeys` selected five columns and `status` was not one of them, so `cachedRun` had no status to test. The fix spans three files: - `run-store/types.ts`: `IdempotencyKeyRunMatch` gains `status`. - `run-store/PostgresRunStore.ts`: the lookup selects `"status"`. `delegatingRunStore` and `runOpsStore` forward unchanged. - `batchTriggerV3.server.ts`: the cached-run guard becomes `keyTimeExpired || shouldIdempotencyKeyBeCleared(cachedRun.status)`, in the same branch as time expiry so the run lands in `expiredRunIds` and the stale key is cleared — otherwise it would survive to the next batch. Policy stays owned by the webapp (`shouldIdempotencyKeyBeCleared` lives in `v3/taskStatus.ts`); the store just returns one more column. Fixes #4819. Co-Authored-By: Claude Opus 5 --- .../batch-idempotency-dead-runs.md | 6 +++ .../app/v3/services/batchTriggerV3.server.ts | 23 +++++++++-- ...RunStore.findRunsByIdempotencyKeys.test.ts | 40 ++++++++++++++++++- .../run-store/src/PostgresRunStore.ts | 2 +- .../src/runOpsStore.shardMap.test.ts | 1 + internal-packages/run-store/src/types.ts | 4 ++ 6 files changed, 70 insertions(+), 6 deletions(-) create mode 100644 .server-changes/batch-idempotency-dead-runs.md diff --git a/.server-changes/batch-idempotency-dead-runs.md b/.server-changes/batch-idempotency-dead-runs.md new file mode 100644 index 00000000000..26986bdde63 --- /dev/null +++ b/.server-changes/batch-idempotency-dead-runs.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: fix +--- + +Fix `batchTrigger` returning a stale failed run when its idempotency key points at a run that crashed, timed out, or otherwise failed. The key is now cleared and a fresh run triggered, matching single-`trigger` behaviour diff --git a/apps/webapp/app/v3/services/batchTriggerV3.server.ts b/apps/webapp/app/v3/services/batchTriggerV3.server.ts index b86e5a40a64..2b9c1b46746 100644 --- a/apps/webapp/app/v3/services/batchTriggerV3.server.ts +++ b/apps/webapp/app/v3/services/batchTriggerV3.server.ts @@ -27,7 +27,11 @@ import { mintBatchFriendlyId } from "~/v3/runOpsMigration/mintBatchFriendlyId.se import { batchTriggerWorker } from "../batchTriggerWorker.server"; import { guardQueueSizeLimitsForEnv } from "../queueSizeLimits.server"; import { downloadPacketFromObjectStore, uploadPacketToObjectStore } from "../objectStore.server"; -import { isFinalAttemptStatus, isFinalRunStatus } from "../taskStatus"; +import { + isFinalAttemptStatus, + isFinalRunStatus, + shouldIdempotencyKeyBeCleared, +} from "../taskStatus"; import { startActiveSpan } from "../tracer.server"; import { BaseService, ServiceValidationError } from "./baseService.server"; import { OutOfEntitlementError, TriggerTaskService } from "./triggerTask.server"; @@ -439,8 +443,10 @@ export class BatchTriggerV3Service extends BaseService { ) ).flat(); - // Build the run IDs in order: reuse an unexpired cached id, else mint a new id (and record any - // expired cached id so its idempotency key can be cleared below). + // Build the run IDs in order: reuse a still-valid cached id, else mint a new id (and record the + // superseded cached id so its idempotency key can be cleared below). "Still valid" means both + // that the key has not timed out and that the run it points at has not reached a clearable + // terminal state. const expiredRunIds = new Set(); const runs = await Promise.all( @@ -450,7 +456,16 @@ export class BatchTriggerV3Service extends BaseService { ); if (cachedRun) { - if (cachedRun.idempotencyKeyExpiresAt && cachedRun.idempotencyKeyExpiresAt < new Date()) { + // Reuse the cached id only if the key is still live AND the run it points at is still + // usable. A run that reached a clearable terminal state (failed statuses + EXPIRED) is + // treated exactly like an expired key: clear it and mint a fresh run, matching the + // single-trigger path in `IdempotencyKeyConcern.handleExistingRun`. Sharing the branch + // matters - it is what adds the run to `expiredRunIds`, so the stale key cannot survive + // to the next batch. + const keyTimeExpired = + !!cachedRun.idempotencyKeyExpiresAt && cachedRun.idempotencyKeyExpiresAt < new Date(); + + if (keyTimeExpired || shouldIdempotencyKeyBeCleared(cachedRun.status)) { expiredRunIds.add(cachedRun.friendlyId); return { diff --git a/internal-packages/run-store/src/PostgresRunStore.findRunsByIdempotencyKeys.test.ts b/internal-packages/run-store/src/PostgresRunStore.findRunsByIdempotencyKeys.test.ts index bc108cd1122..f04d463f9e5 100644 --- a/internal-packages/run-store/src/PostgresRunStore.findRunsByIdempotencyKeys.test.ts +++ b/internal-packages/run-store/src/PostgresRunStore.findRunsByIdempotencyKeys.test.ts @@ -1,5 +1,5 @@ import { postgresTest } from "@internal/testcontainers"; -import type { PrismaClient } from "@trigger.dev/database"; +import type { PrismaClient, TaskRunStatus } from "@trigger.dev/database"; import { describe, expect } from "vitest"; import { PostgresRunStore } from "./PostgresRunStore.js"; @@ -38,6 +38,7 @@ async function createRun( taskIdentifier: string; idempotencyKey: string; idempotencyKeyExpiresAt?: Date; + status?: TaskRunStatus; } ) { await prisma.taskRun.create({ @@ -46,6 +47,7 @@ async function createRun( taskIdentifier: params.taskIdentifier, idempotencyKey: params.idempotencyKey, idempotencyKeyExpiresAt: params.idempotencyKeyExpiresAt ?? null, + ...(params.status ? { status: params.status } : {}), payload: "{}", payloadType: "application/json", runtimeEnvironmentId: params.runtimeEnvironmentId, @@ -105,6 +107,42 @@ describe("PostgresRunStore.findRunsByIdempotencyKeys", () => { expect(byKey.get("idem-2")?.idempotencyKeyExpiresAt).toBeNull(); }); + // Regression for #4819: the batch trigger path decides whether a cached match is still reusable + // by feeding this row's status to `shouldIdempotencyKeyBeCleared`. Before the fix the query did + // not select `status` at all, so a key pointing at a dead run was returned as cached forever. + postgresTest("returns the run status so callers can reject dead runs", async ({ prisma }) => { + const { project, environment } = await seedEnvironment(prisma); + const store = new PostgresRunStore({ prisma, readOnlyPrisma: prisma }); + + await createRun(prisma, { + runtimeEnvironmentId: environment.id, + projectId: project.id, + friendlyId: "run_dead", + taskIdentifier: "task-a", + idempotencyKey: "idem-dead", + status: "CRASHED", + }); + await createRun(prisma, { + runtimeEnvironmentId: environment.id, + projectId: project.id, + friendlyId: "run_live", + taskIdentifier: "task-a", + idempotencyKey: "idem-live", + status: "EXECUTING", + }); + + const rows = await store.findRunsByIdempotencyKeys({ + runtimeEnvironmentId: environment.id, + taskIdentifier: "task-a", + idempotencyKeys: ["idem-dead", "idem-live"], + }); + + const byKey = new Map(rows.map((r) => [r.idempotencyKey, r])); + expect(rows).toHaveLength(2); + expect(byKey.get("idem-dead")?.status).toBe("CRASHED"); + expect(byKey.get("idem-live")?.status).toBe("EXECUTING"); + }); + postgresTest("short-circuits on an empty key list without querying", async ({ prisma }) => { const { environment } = await seedEnvironment(prisma); const store = new PostgresRunStore({ prisma, readOnlyPrisma: prisma }); diff --git a/internal-packages/run-store/src/PostgresRunStore.ts b/internal-packages/run-store/src/PostgresRunStore.ts index 22dc2f90c44..d3f6d9eafbc 100644 --- a/internal-packages/run-store/src/PostgresRunStore.ts +++ b/internal-packages/run-store/src/PostgresRunStore.ts @@ -1861,7 +1861,7 @@ export class PostgresRunStore implements RunStore { const branches = args.idempotencyKeys.map((key) => { const base = params.length; params.push(args.runtimeEnvironmentId, args.taskIdentifier, key); - return `SELECT "id", "createdAt", "friendlyId", "idempotencyKey", "idempotencyKeyExpiresAt" FROM "TaskRun" WHERE "runtimeEnvironmentId" = $${base + 1} AND "taskIdentifier" = $${base + 2} AND "idempotencyKey" = $${base + 3}`; + return `SELECT "id", "createdAt", "friendlyId", "idempotencyKey", "idempotencyKeyExpiresAt", "status" FROM "TaskRun" WHERE "runtimeEnvironmentId" = $${base + 1} AND "taskIdentifier" = $${base + 2} AND "idempotencyKey" = $${base + 3}`; }); return prisma.$queryRawUnsafe( branches.join(" UNION ALL "), diff --git a/internal-packages/run-store/src/runOpsStore.shardMap.test.ts b/internal-packages/run-store/src/runOpsStore.shardMap.test.ts index aff0d50764e..138becc6b11 100644 --- a/internal-packages/run-store/src/runOpsStore.shardMap.test.ts +++ b/internal-packages/run-store/src/runOpsStore.shardMap.test.ts @@ -622,6 +622,7 @@ describe("RoutingRunStore findRunsByIdempotencyKeys tiebreak", () => { friendlyId: `run_${id}`, idempotencyKey: "k", idempotencyKeyExpiresAt: null, + status: "PENDING", }); it("keeps NEW-wins across the gen-1 pair even when legacy is older", async () => { diff --git a/internal-packages/run-store/src/types.ts b/internal-packages/run-store/src/types.ts index 41fecf90bb5..23b6e3fc71f 100644 --- a/internal-packages/run-store/src/types.ts +++ b/internal-packages/run-store/src/types.ts @@ -28,6 +28,10 @@ export type IdempotencyKeyRunMatch = { friendlyId: string; idempotencyKey: string | null; idempotencyKeyExpiresAt: Date | null; + /** Callers decide whether a cached match is still reusable: a run that reached a clearable + * terminal state must not be handed back as cached. Policy lives in the webapp + * (`shouldIdempotencyKeyBeCleared`); the store just returns the column. */ + status: TaskRunStatus; }; export type CreateRunSnapshotInput = {