Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 21 additions & 2 deletions apps/webapp/app/v3/services/batchTriggerV3.server.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand All @@ -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";
Expand Down Expand Up @@ -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,
};
}
Comment on lines +468 to +481

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 Missing server release note for user-facing fix

This user-facing behavior change to batch triggers ships only server code under apps/webapp and internal-packages, with no .server-changes/ note added. The repository guidelines require a .server-changes/ entry for user-facing server-only changes, so this fix will be absent from release notes.

Prompt for agents
CONTRIBUTING.md and AGENTS.md require a .server-changes/ file for user-facing server-only changes (changes under apps/webapp with no package changes). This PR changes observable batchTrigger behavior (failed runs are now re-triggered instead of returned as a dead cached run) and touches only apps/webapp and internal-packages. Add a markdown file under .server-changes/ (e.g. fix-batch-trigger-stale-failed-runs.md) with frontmatter `area: webapp` and `type: fix`, and a one-line user-facing description. See .server-changes/README.md for the exact format.
Devin Review

Was this helpful? React with 👍 or 👎 to provide feedback.


return {
id: cachedRun.friendlyId,
isCached: true,
Expand Down
81 changes: 81 additions & 0 deletions apps/webapp/test/batchTriggerIdempotencyStatusCheck.test.ts
Original file line number Diff line number Diff line change
@@ -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);
});
});
Original file line number Diff line number Diff line change
@@ -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";

Expand Down Expand Up @@ -38,6 +38,7 @@ async function createRun(
taskIdentifier: string;
idempotencyKey: string;
idempotencyKeyExpiresAt?: Date;
status?: TaskRunStatus;
}
) {
await prisma.taskRun.create({
Expand All @@ -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,
Expand Down Expand Up @@ -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 }) => {
Expand All @@ -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");
});
});
2 changes: 1 addition & 1 deletion internal-packages/run-store/src/PostgresRunStore.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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 "),
Expand Down
1 change: 1 addition & 0 deletions internal-packages/run-store/src/types.ts
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@ export type IdempotencyKeyRunMatch = {
friendlyId: string;
idempotencyKey: string | null;
idempotencyKeyExpiresAt: Date | null;
status: string;
};

export type CreateRunSnapshotInput = {
Expand Down