fix(batch): clear idempotency keys pointing at dead runs in batchTrigger

.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

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<string>();
 
     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 {

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 });

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<IdempotencyKeyRunMatch[]>(
       branches.join(" UNION ALL "),

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 () => {

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 = {