Skip to content

Commit 81ebf24

Browse files
itzzdev09claude
andcommitted
fix(batch): clear idempotency keys pointing at dead runs in batchTrigger
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 <noreply@anthropic.com>
1 parent 1f8f230 commit 81ebf24

6 files changed

Lines changed: 70 additions & 6 deletions

File tree

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
---
2+
area: webapp
3+
type: fix
4+
---
5+
6+
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

‎apps/webapp/app/v3/services/batchTriggerV3.server.ts‎

Lines changed: 19 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,11 @@ import { mintBatchFriendlyId } from "~/v3/runOpsMigration/mintBatchFriendlyId.se
2727
import { batchTriggerWorker } from "../batchTriggerWorker.server";
2828
import { guardQueueSizeLimitsForEnv } from "../queueSizeLimits.server";
2929
import { downloadPacketFromObjectStore, uploadPacketToObjectStore } from "../objectStore.server";
30-
import { isFinalAttemptStatus, isFinalRunStatus } from "../taskStatus";
30+
import {
31+
isFinalAttemptStatus,
32+
isFinalRunStatus,
33+
shouldIdempotencyKeyBeCleared,
34+
} from "../taskStatus";
3135
import { startActiveSpan } from "../tracer.server";
3236
import { BaseService, ServiceValidationError } from "./baseService.server";
3337
import { OutOfEntitlementError, TriggerTaskService } from "./triggerTask.server";
@@ -439,8 +443,10 @@ export class BatchTriggerV3Service extends BaseService {
439443
)
440444
).flat();
441445

442-
// Build the run IDs in order: reuse an unexpired cached id, else mint a new id (and record any
443-
// expired cached id so its idempotency key can be cleared below).
446+
// Build the run IDs in order: reuse a still-valid cached id, else mint a new id (and record the
447+
// superseded cached id so its idempotency key can be cleared below). "Still valid" means both
448+
// that the key has not timed out and that the run it points at has not reached a clearable
449+
// terminal state.
444450
const expiredRunIds = new Set<string>();
445451

446452
const runs = await Promise.all(
@@ -450,7 +456,16 @@ export class BatchTriggerV3Service extends BaseService {
450456
);
451457

452458
if (cachedRun) {
453-
if (cachedRun.idempotencyKeyExpiresAt && cachedRun.idempotencyKeyExpiresAt < new Date()) {
459+
// Reuse the cached id only if the key is still live AND the run it points at is still
460+
// usable. A run that reached a clearable terminal state (failed statuses + EXPIRED) is
461+
// treated exactly like an expired key: clear it and mint a fresh run, matching the
462+
// single-trigger path in `IdempotencyKeyConcern.handleExistingRun`. Sharing the branch
463+
// matters - it is what adds the run to `expiredRunIds`, so the stale key cannot survive
464+
// to the next batch.
465+
const keyTimeExpired =
466+
!!cachedRun.idempotencyKeyExpiresAt && cachedRun.idempotencyKeyExpiresAt < new Date();
467+
468+
if (keyTimeExpired || shouldIdempotencyKeyBeCleared(cachedRun.status)) {
454469
expiredRunIds.add(cachedRun.friendlyId);
455470

456471
return {

‎internal-packages/run-store/src/PostgresRunStore.findRunsByIdempotencyKeys.test.ts‎

Lines changed: 39 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
import { postgresTest } from "@internal/testcontainers";
2-
import type { PrismaClient } from "@trigger.dev/database";
2+
import type { PrismaClient, TaskRunStatus } from "@trigger.dev/database";
33
import { describe, expect } from "vitest";
44
import { PostgresRunStore } from "./PostgresRunStore.js";
55

@@ -38,6 +38,7 @@ async function createRun(
3838
taskIdentifier: string;
3939
idempotencyKey: string;
4040
idempotencyKeyExpiresAt?: Date;
41+
status?: TaskRunStatus;
4142
}
4243
) {
4344
await prisma.taskRun.create({
@@ -46,6 +47,7 @@ async function createRun(
4647
taskIdentifier: params.taskIdentifier,
4748
idempotencyKey: params.idempotencyKey,
4849
idempotencyKeyExpiresAt: params.idempotencyKeyExpiresAt ?? null,
50+
...(params.status ? { status: params.status } : {}),
4951
payload: "{}",
5052
payloadType: "application/json",
5153
runtimeEnvironmentId: params.runtimeEnvironmentId,
@@ -105,6 +107,42 @@ describe("PostgresRunStore.findRunsByIdempotencyKeys", () => {
105107
expect(byKey.get("idem-2")?.idempotencyKeyExpiresAt).toBeNull();
106108
});
107109

110+
// Regression for #4819: the batch trigger path decides whether a cached match is still reusable
111+
// by feeding this row's status to `shouldIdempotencyKeyBeCleared`. Before the fix the query did
112+
// not select `status` at all, so a key pointing at a dead run was returned as cached forever.
113+
postgresTest("returns the run status so callers can reject dead runs", async ({ prisma }) => {
114+
const { project, environment } = await seedEnvironment(prisma);
115+
const store = new PostgresRunStore({ prisma, readOnlyPrisma: prisma });
116+
117+
await createRun(prisma, {
118+
runtimeEnvironmentId: environment.id,
119+
projectId: project.id,
120+
friendlyId: "run_dead",
121+
taskIdentifier: "task-a",
122+
idempotencyKey: "idem-dead",
123+
status: "CRASHED",
124+
});
125+
await createRun(prisma, {
126+
runtimeEnvironmentId: environment.id,
127+
projectId: project.id,
128+
friendlyId: "run_live",
129+
taskIdentifier: "task-a",
130+
idempotencyKey: "idem-live",
131+
status: "EXECUTING",
132+
});
133+
134+
const rows = await store.findRunsByIdempotencyKeys({
135+
runtimeEnvironmentId: environment.id,
136+
taskIdentifier: "task-a",
137+
idempotencyKeys: ["idem-dead", "idem-live"],
138+
});
139+
140+
const byKey = new Map(rows.map((r) => [r.idempotencyKey, r]));
141+
expect(rows).toHaveLength(2);
142+
expect(byKey.get("idem-dead")?.status).toBe("CRASHED");
143+
expect(byKey.get("idem-live")?.status).toBe("EXECUTING");
144+
});
145+
108146
postgresTest("short-circuits on an empty key list without querying", async ({ prisma }) => {
109147
const { environment } = await seedEnvironment(prisma);
110148
const store = new PostgresRunStore({ prisma, readOnlyPrisma: prisma });

‎internal-packages/run-store/src/PostgresRunStore.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1861,7 +1861,7 @@ export class PostgresRunStore implements RunStore {
18611861
const branches = args.idempotencyKeys.map((key) => {
18621862
const base = params.length;
18631863
params.push(args.runtimeEnvironmentId, args.taskIdentifier, key);
1864-
return `SELECT "id", "createdAt", "friendlyId", "idempotencyKey", "idempotencyKeyExpiresAt" FROM "TaskRun" WHERE "runtimeEnvironmentId" = $${base + 1} AND "taskIdentifier" = $${base + 2} AND "idempotencyKey" = $${base + 3}`;
1864+
return `SELECT "id", "createdAt", "friendlyId", "idempotencyKey", "idempotencyKeyExpiresAt", "status" FROM "TaskRun" WHERE "runtimeEnvironmentId" = $${base + 1} AND "taskIdentifier" = $${base + 2} AND "idempotencyKey" = $${base + 3}`;
18651865
});
18661866
return prisma.$queryRawUnsafe<IdempotencyKeyRunMatch[]>(
18671867
branches.join(" UNION ALL "),

‎internal-packages/run-store/src/runOpsStore.shardMap.test.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -622,6 +622,7 @@ describe("RoutingRunStore findRunsByIdempotencyKeys tiebreak", () => {
622622
friendlyId: `run_${id}`,
623623
idempotencyKey: "k",
624624
idempotencyKeyExpiresAt: null,
625+
status: "PENDING",
625626
});
626627

627628
it("keeps NEW-wins across the gen-1 pair even when legacy is older", async () => {

‎internal-packages/run-store/src/types.ts‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,10 @@ export type IdempotencyKeyRunMatch = {
2828
friendlyId: string;
2929
idempotencyKey: string | null;
3030
idempotencyKeyExpiresAt: Date | null;
31+
/** Callers decide whether a cached match is still reusable: a run that reached a clearable
32+
* terminal state must not be handed back as cached. Policy lives in the webapp
33+
* (`shouldIdempotencyKeyBeCleared`); the store just returns the column. */
34+
status: TaskRunStatus;
3135
};
3236

3337
export type CreateRunSnapshotInput = {

0 commit comments

Comments
 (0)