diff --git a/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts b/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts index a0d5b733494..b1143366aa0 100644 --- a/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts +++ b/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts @@ -1,5 +1,5 @@ import { timeoutError, tryCatch } from "@trigger.dev/core/v3"; -import { WaitpointId } from "@trigger.dev/core/v3/isomorphic"; +import { UnclassifiableRunId, WaitpointId } from "@trigger.dev/core/v3/isomorphic"; import type { PrismaClientOrTransaction, TaskRun, @@ -91,11 +91,24 @@ export class WaitpointSystem { try { store = await this.$.runStore.forWaitpointCompletion(id, { routeKind: "MANUAL" }); } catch (error) { - this.$.logger.error("completeWaitpoint: unclassifiable waitpointId", { + // Only a genuine id-classification failure should become UnclassifiableWaitpointId. + // forWaitpointCompletion also probes the DB to resolve the owning store, so a transient + // database/infra error (e.g. can't reach the database) can surface here too. Those MUST + // bubble up unchanged so they keep their original type, retryability, and error grouping + // instead of being mislabelled as an unclassifiable id. + if (error instanceof UnclassifiableRunId) { + this.$.logger.error("completeWaitpoint: unclassifiable waitpointId", { + waitpointId: id, + error, + }); + throw new UnclassifiableWaitpointId(id, { cause: error }); + } + + this.$.logger.error("completeWaitpoint: error resolving waitpoint store", { waitpointId: id, error, }); - throw new UnclassifiableWaitpointId(id, { cause: error }); + throw error; } // 1. Complete the Waitpoint (if not completed) diff --git a/internal-packages/run-engine/src/engine/tests/completeWaitpointErrorClassification.test.ts b/internal-packages/run-engine/src/engine/tests/completeWaitpointErrorClassification.test.ts new file mode 100644 index 00000000000..bc6ebd5f8db --- /dev/null +++ b/internal-packages/run-engine/src/engine/tests/completeWaitpointErrorClassification.test.ts @@ -0,0 +1,68 @@ +// completeWaitpoint's store-selection guard must only turn a genuine id-classification +// failure into UnclassifiableWaitpointId. forWaitpointCompletion also probes the DB to +// resolve the owning store, so a transient database/infra error can surface from the same +// call — and those must bubble up UNCHANGED (keeping their original type, retryability, and +// error grouping) rather than being mislabelled as an unclassifiable id. +// +// This is a hermetic unit test: the error is thrown on the very first line of +// completeWaitpoint (runStore.forWaitpointCompletion), before any snapshot/enqueue work, +// so we can drive it with a minimal SystemResources and a fake runStore — no DB, no Redis. +import { UnclassifiableRunId } from "@trigger.dev/core/v3/isomorphic"; +import { expect } from "vitest"; +import { UnclassifiableWaitpointId } from "../errors.js"; +import type { SystemResources } from "../systems/systems.js"; +import { WaitpointSystem } from "../systems/waitpointSystem.js"; + +function createWaitpointSystem(forWaitpointCompletion: () => Promise) { + const runStore = { forWaitpointCompletion }; + + const resources = { + runStore, + logger: { + error: vi.fn(), + warn: vi.fn(), + info: vi.fn(), + debug: vi.fn(), + }, + } as unknown as SystemResources; + + return new WaitpointSystem({ + resources, + // Never reached on the store-resolution error path. + executionSnapshotSystem: {} as any, + enqueueSystem: {} as any, + }); +} + +describe("completeWaitpoint store-resolution error classification", () => { + it("rethrows a transient database error unchanged (never wraps it as UnclassifiableWaitpointId)", async () => { + const dbError = new Error("Can't reach database server at db:5432"); + const waitpointSystem = createWaitpointSystem(() => Promise.reject(dbError)); + + // The original error bubbles up as-is... + await expect(waitpointSystem.completeWaitpoint({ id: "waitpoint_transient" })).rejects.toBe( + dbError + ); + // ...and is NOT relabelled as a classification failure. + await expect( + waitpointSystem.completeWaitpoint({ id: "waitpoint_transient" }) + ).rejects.not.toBeInstanceOf(UnclassifiableWaitpointId); + }); + + it("wraps a genuine UnclassifiableRunId as UnclassifiableWaitpointId with the original as cause", async () => { + const waitpointId = "waitpoint_unclassifiable"; + const classificationError = new UnclassifiableRunId(waitpointId); + const waitpointSystem = createWaitpointSystem(() => Promise.reject(classificationError)); + + await expect(waitpointSystem.completeWaitpoint({ id: waitpointId })).rejects.toBeInstanceOf( + UnclassifiableWaitpointId + ); + + const caught = (await waitpointSystem + .completeWaitpoint({ id: waitpointId }) + .catch((error: unknown) => error)) as UnclassifiableWaitpointId; + expect(caught).toBeInstanceOf(UnclassifiableWaitpointId); + expect(caught.waitpointId).toBe(waitpointId); + expect(caught.cause).toBe(classificationError); + }); +});