diff --git a/apps/webapp/app/v3/services/batchTriggerV3.server.ts b/apps/webapp/app/v3/services/batchTriggerV3.server.ts index b86e5a40a64..9e903fa2aa8 100644 --- a/apps/webapp/app/v3/services/batchTriggerV3.server.ts +++ b/apps/webapp/app/v3/services/batchTriggerV3.server.ts @@ -4,7 +4,7 @@ import type { IOPacket, } from "@trigger.dev/core/v3"; import { packetRequiresOffloading, parsePacket } from "@trigger.dev/core/v3"; -import type { BatchTaskRun, TaskRunAttempt } from "@trigger.dev/database"; +import type { BatchTaskRun, TaskRunAttempt, TaskRunStatus } from "@trigger.dev/database"; import { isUniqueConstraintError, Prisma } from "@trigger.dev/database"; import type { RunStore } from "@internal/run-store"; import pMap from "p-map"; @@ -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"; @@ -461,6 +465,21 @@ export class BatchTriggerV3Service extends BaseService { }; } + // Mirror the single-trigger path (IdempotencyKeyConcern.handleExistingRun): + // if the cached run is in a terminal failure state (CRASHED, SYSTEM_FAILURE, + // TIMED_OUT, EXPIRED, COMPLETED_WITH_ERRORS, INTERRUPTED), clear the + // idempotency key and re-trigger instead of returning the dead run. + if (shouldIdempotencyKeyBeCleared(cachedRun.status as TaskRunStatus)) { + expiredRunIds.add(cachedRun.friendlyId); + + return { + id: await this.mintChildFriendlyId(environment, childAnchor, item.options?.region), + isCached: false, + idempotencyKey: item.options?.idempotencyKey ?? undefined, + taskIdentifier: item.task, + }; + } + return { id: cachedRun.friendlyId, isCached: true, diff --git a/apps/webapp/test/batchTriggerIdempotencyStatusCheck.test.ts b/apps/webapp/test/batchTriggerIdempotencyStatusCheck.test.ts new file mode 100644 index 00000000000..9862c577799 --- /dev/null +++ b/apps/webapp/test/batchTriggerIdempotencyStatusCheck.test.ts @@ -0,0 +1,81 @@ +import { describe, expect, it } from "vitest"; +import type { TaskRunStatus } from "@trigger.dev/database"; +import { shouldIdempotencyKeyBeCleared } from "~/v3/taskStatus"; + +/** + * Tests the shouldIdempotencyKeyBeCleared function which is now used by both the + * single-trigger path (IdempotencyKeyConcern.handleExistingRun) and the batch-trigger + * path (BatchTriggerV3Service.#prepareRunData). + * + * Before the fix, the batch path did not call this function at all — it only + * checked time-based expiration, silently returning dead runs as cached hits. + * + * These tests validate: + * 1. All failure statuses that should trigger re-triggering + * 2. All non-failure statuses that should preserve the cached run + * 3. Edge cases around status classification + */ +describe("shouldIdempotencyKeyBeCleared — batch trigger parity", () => { + // ─── Statuses that MUST clear the idempotency key ────────────────── + // These are the statuses where the single-trigger path (and now the + // batch-trigger path) clears the key and creates a fresh run. + + const clearableStatuses: TaskRunStatus[] = [ + "CRASHED", + "SYSTEM_FAILURE", + "TIMED_OUT", + "EXPIRED", + "COMPLETED_WITH_ERRORS", + "INTERRUPTED", + ]; + + it.each(clearableStatuses)( + "returns true for %s — dead runs must not be returned as cached hits in batch triggers", + (status) => { + expect(shouldIdempotencyKeyBeCleared(status)).toBe(true); + } + ); + + // ─── Statuses that MUST NOT clear the idempotency key ────────────── + // These runs are either still in progress, successfully completed, + // or canceled by the user. Returning them as cached is correct. + + const nonClearableStatuses: TaskRunStatus[] = [ + "PENDING", + "PENDING_VERSION", + "WAITING_FOR_DEPLOY", + "DEQUEUED", + "EXECUTING", + "WAITING_TO_RESUME", + "RETRYING_AFTER_FAILURE", + "PAUSED", + "DELAYED", + "COMPLETED_SUCCESSFULLY", + "CANCELED", + ]; + + it.each(nonClearableStatuses)( + "returns false for %s — these runs should be returned as cached hits", + (status) => { + expect(shouldIdempotencyKeyBeCleared(status)).toBe(false); + } + ); + + // ─── Specific edge cases ─────────────────────────────────────────── + + it("COMPLETED_SUCCESSFULLY should NOT be cleared — it is a valid cached result", () => { + expect(shouldIdempotencyKeyBeCleared("COMPLETED_SUCCESSFULLY")).toBe(false); + }); + + it("CANCELED should NOT be cleared — user-initiated cancellation is intentional", () => { + expect(shouldIdempotencyKeyBeCleared("CANCELED")).toBe(false); + }); + + it("EXPIRED should be cleared — expired runs should be re-triggerable via batch", () => { + expect(shouldIdempotencyKeyBeCleared("EXPIRED")).toBe(true); + }); + + it("RETRYING_AFTER_FAILURE should NOT be cleared — the run is still in progress", () => { + expect(shouldIdempotencyKeyBeCleared("RETRYING_AFTER_FAILURE")).toBe(false); + }); +}); diff --git a/internal-packages/run-store/src/PostgresRunStore.findRunsByIdempotencyKeys.test.ts b/internal-packages/run-store/src/PostgresRunStore.findRunsByIdempotencyKeys.test.ts index bc108cd1122..d384716cbde 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, + status: params.status ?? "PENDING", payload: "{}", payloadType: "application/json", runtimeEnvironmentId: params.runtimeEnvironmentId, @@ -103,6 +105,9 @@ describe("PostgresRunStore.findRunsByIdempotencyKeys", () => { expiresAt.toISOString() ); expect(byKey.get("idem-2")?.idempotencyKeyExpiresAt).toBeNull(); + // status must be returned so callers can check shouldIdempotencyKeyBeCleared + expect(byKey.get("idem-1")?.status).toBe("PENDING"); + expect(byKey.get("idem-2")?.status).toBe("PENDING"); }); postgresTest("short-circuits on an empty key list without querying", async ({ prisma }) => { @@ -117,4 +122,71 @@ describe("PostgresRunStore.findRunsByIdempotencyKeys", () => { expect(rows).toEqual([]); }); + + postgresTest( + "returns status for runs in failure states (CRASHED, SYSTEM_FAILURE, TIMED_OUT, EXPIRED, COMPLETED_WITH_ERRORS, INTERRUPTED)", + async ({ prisma }) => { + const { project, environment } = await seedEnvironment(prisma); + const store = new PostgresRunStore({ prisma, readOnlyPrisma: prisma }); + + const failureStatuses: TaskRunStatus[] = [ + "CRASHED", + "SYSTEM_FAILURE", + "TIMED_OUT", + "EXPIRED", + "COMPLETED_WITH_ERRORS", + "INTERRUPTED", + ]; + + for (let i = 0; i < failureStatuses.length; i++) { + await createRun(prisma, { + runtimeEnvironmentId: environment.id, + projectId: project.id, + friendlyId: `run_fail_${i}`, + taskIdentifier: "task-a", + idempotencyKey: `idem-fail-${i}`, + status: failureStatuses[i], + }); + } + + const rows = await store.findRunsByIdempotencyKeys({ + runtimeEnvironmentId: environment.id, + taskIdentifier: "task-a", + idempotencyKeys: failureStatuses.map((_, i) => `idem-fail-${i}`), + }); + + expect(rows).toHaveLength(failureStatuses.length); + + const byKey = new Map(rows.map((r) => [r.idempotencyKey, r])); + for (let i = 0; i < failureStatuses.length; i++) { + const row = byKey.get(`idem-fail-${i}`); + expect(row).toBeDefined(); + expect(row?.status).toBe(failureStatuses[i]); + expect(row?.friendlyId).toBe(`run_fail_${i}`); + } + } + ); + + postgresTest("returns status for a successfully completed run", 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_success", + taskIdentifier: "task-a", + idempotencyKey: "idem-success", + status: "COMPLETED_SUCCESSFULLY", + }); + + const rows = await store.findRunsByIdempotencyKeys({ + runtimeEnvironmentId: environment.id, + taskIdentifier: "task-a", + idempotencyKeys: ["idem-success"], + }); + + expect(rows).toHaveLength(1); + expect(rows[0].status).toBe("COMPLETED_SUCCESSFULLY"); + }); }); 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/types.ts b/internal-packages/run-store/src/types.ts index 41fecf90bb5..ce8dae5fc4a 100644 --- a/internal-packages/run-store/src/types.ts +++ b/internal-packages/run-store/src/types.ts @@ -28,6 +28,7 @@ export type IdempotencyKeyRunMatch = { friendlyId: string; idempotencyKey: string | null; idempotencyKeyExpiresAt: Date | null; + status: string; }; export type CreateRunSnapshotInput = {