diff --git a/internal-packages/run-engine/src/engine/systems/completedWaitpointFreeze.test.ts b/internal-packages/run-engine/src/engine/systems/completedWaitpointFreeze.test.ts index a1af90582b6..98973540b7a 100644 --- a/internal-packages/run-engine/src/engine/systems/completedWaitpointFreeze.test.ts +++ b/internal-packages/run-engine/src/engine/systems/completedWaitpointFreeze.test.ts @@ -40,7 +40,7 @@ const _pointerKeys: Exact = true; import { enhanceExecutionSnapshotWithWaitpoints } from "./executionSnapshotSystem.js"; @@ -193,6 +193,7 @@ async function assertParity( batchId: batchId ?? undefined, pointer: { cycleSeq: 1, count: order.length }, order, + distinctIds: [...new Set(waitpoints.map((w) => w.id))], records: waitpoints.map(toRecord), }; // count-carried-forward behaviour (order.length, not the record count) is covered by @@ -406,6 +407,7 @@ describe("the completed-waitpoints freeze", () => { batchId: undefined, pointer: { cycleSeq: 1, count: 1 }, order: ["wp_hook"], + distinctIds: ["wp_hook"], records: [toRecord(w)], }); expect(resolved).toHaveLength(1); @@ -611,6 +613,7 @@ describe("the exhaustive parity grid", () => { batchId: readingBatchId ?? undefined, pointer: { cycleSeq: 1, count: order.length }, order, + distinctIds: [w.id], records: [toRecord(w)], }; const resolved = await referenceResolver( diff --git a/internal-packages/run-engine/src/engine/systems/enqueueSystem.ts b/internal-packages/run-engine/src/engine/systems/enqueueSystem.ts index 38c681c511e..850f3714239 100644 --- a/internal-packages/run-engine/src/engine/systems/enqueueSystem.ts +++ b/internal-packages/run-engine/src/engine/systems/enqueueSystem.ts @@ -4,7 +4,7 @@ import type { TaskRun, TaskRunExecutionStatus, } from "@trigger.dev/database"; -import type { RunStore } from "@internal/run-store"; +import type { CompletedWaitpointRecord, RunStore } from "@internal/run-store"; import { parseNaturalLanguageDuration } from "@trigger.dev/core/v3/isomorphic"; import type { MinimalAuthenticatedEnvironment } from "../../shared/index.js"; import { QUEUED_SNAPSHOT_DESCRIPTION, QUEUED_SNAPSHOT_STATUS } from "../consts.js"; @@ -34,6 +34,7 @@ export class EnqueueSystem { batchId, checkpointId, completedWaitpoints, + completedWaitpointRecords, workerId, runnerId, skipRunLock, @@ -57,6 +58,7 @@ export class EnqueueSystem { id: string; index?: number; }[]; + completedWaitpointRecords?: CompletedWaitpointRecord[]; workerId?: string; runnerId?: string; skipRunLock?: boolean; @@ -108,6 +110,7 @@ export class EnqueueSystem { organizationId: env.organization.id, checkpointId, completedWaitpoints, + completedWaitpointRecords, workerId, runnerId, }, diff --git a/internal-packages/run-engine/src/engine/systems/executionSnapshotSystem.ts b/internal-packages/run-engine/src/engine/systems/executionSnapshotSystem.ts index 81c41d2c2ae..fd01e70303b 100644 --- a/internal-packages/run-engine/src/engine/systems/executionSnapshotSystem.ts +++ b/internal-packages/run-engine/src/engine/systems/executionSnapshotSystem.ts @@ -10,7 +10,7 @@ import type { TaskRunStatus, Waitpoint, } from "@trigger.dev/database"; -import type { RunStore } from "@internal/run-store"; +import type { CompletedWaitpointRecord, RunStore } from "@internal/run-store"; import { ExecutionSnapshotNotFoundError, ServiceValidationError } from "../errors.js"; import type { HeartbeatTimeouts } from "../types.js"; import type { SystemResources } from "./systems.js"; @@ -200,6 +200,67 @@ async function fetchWaitpointsInChunks( return allWaitpoints; } +/** + * The columns the completion envelope is built from, and only those. + * + * `fetchWaitpointsInChunks` reads whole rows because the snapshot hydration builds a full + * executor waitpoint from them. The envelope needs fewer, so projecting drops `tags`, + * `projectId`, `environmentId`, `createdAt`, `updatedAt` and `idempotencyKeyExpiresAt` from a + * read that happens on the writer inside the run lock. It cannot drop `output`, which is the + * 100KB+ column and also the payload the envelope exists to carry. + * + * `status` is here for the arm's COMPLETED filter, not for the mapper. + */ +const WAITPOINT_ENVELOPE_SELECT = { + id: true, + friendlyId: true, + type: true, + status: true, + completedAt: true, + output: true, + outputType: true, + outputIsError: true, + completedByTaskRunId: true, + completedByBatchId: true, + completedAfter: true, + idempotencyKey: true, + userProvidedIdempotencyKey: true, + inactiveIdempotencyKey: true, +} satisfies Prisma.WaitpointSelect; + +export type WaitpointEnvelopeRow = Pick; + +/** + * The projected sibling of `fetchWaitpointsInChunks`, for the envelope read. + * + * Chunked identically, and for the same reason: a waitpoint output can be 100KB+, so a large + * fan-in read whole can exceed Node's string limits. `boundedIn` pads for plan-cache stability, + * it does not bound the set. `runId` is the routing hint the router needs to read the run's own + * store instead of fanning every chunk across both run-ops databases. + */ +export async function fetchWaitpointEnvelopeRowsInChunks( + prisma: PrismaClientOrTransaction, + waitpointIds: string[], + runStore?: RunStore, + runId?: string +): Promise { + if (waitpointIds.length === 0) return []; + + const rows: WaitpointEnvelopeRow[] = []; + for (let i = 0; i < waitpointIds.length; i += WAITPOINT_CHUNK_SIZE) { + const chunk = waitpointIds.slice(i, i + WAITPOINT_CHUNK_SIZE); + const args = { + where: { id: { in: boundedIn(chunk) } }, + select: WAITPOINT_ENVELOPE_SELECT, + }; + const found = runStore + ? await runStore.findManyWaitpoints(args, prisma, runId) + : await prisma.waitpoint.findMany(args); + rows.push(...found); + } + return rows; +} + /** * Gets the most recent valid snapshot for a run. When `environmentId` is provided the read is scoped * to that environment (tenant boundary): a run in another environment reads as not-found and rejects @@ -449,6 +510,7 @@ export class ExecutionSnapshotSystem { workerId, runnerId, completedWaitpoints, + completedWaitpointRecords, error, }: { run: { id: string; status: TaskRunStatus; attemptNumber?: number | null }; @@ -470,6 +532,7 @@ export class ExecutionSnapshotSystem { id: string; index?: number; }[]; + completedWaitpointRecords?: CompletedWaitpointRecord[]; error?: string; }, // When set (inside runStore.runInTransaction), the snapshot write goes through the owning store @@ -492,6 +555,7 @@ export class ExecutionSnapshotSystem { workerId, runnerId, completedWaitpoints, + completedWaitpointRecords, error, }, prisma diff --git a/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts b/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts index 0ab62c5cf9a..df16e403bad 100644 --- a/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts +++ b/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts @@ -1,5 +1,7 @@ import { timeoutError } from "@trigger.dev/core/v3"; +import { parseWaitpointId } from "@trigger.dev/core/v3/isomorphic"; import type { ShardKey } from "@trigger.dev/core/v3/isomorphic"; +import type { CompletedWaitpointRecord } from "@internal/run-store"; import type { PrismaClientOrTransaction, TaskRun, @@ -11,7 +13,8 @@ import { assertNever } from "assert-never"; import { sendNotificationToWorker } from "../eventBus.js"; import { isFinalRunStatus } from "../statuses.js"; import { LegacyPostgresWaitpointCoordinator } from "../waitpointCoordinator/legacyPostgresCoordinator.js"; -import type { WaitpointCoordinator } from "../waitpointCoordinator/types.js"; +import { buildCompletedWaitpointRecords } from "../waitpointCoordinator/completedWaitpointRecords.js"; +import type { RunBlockEdge, WaitpointCoordinator } from "../waitpointCoordinator/types.js"; import type { EnqueueSystem } from "./enqueueSystem.js"; import type { ExecutionSnapshotSystem } from "./executionSnapshotSystem.js"; import { getLatestExecutionSnapshot } from "./executionSnapshotSystem.js"; @@ -605,6 +608,13 @@ export class WaitpointSystem { }; } case "EXECUTING_WITH_WAITPOINTS": { + // Built inside the branch, not before the switch: the statuses above return without + // appending, and they must not pay an envelope read to do it. + const completedWaitpointRecords = await this.#completedWaitpointRecordsFor( + runId, + blockingWaitpoints + ); + const newSnapshot = await this.executionSnapshotSystem.createExecutionSnapshot( this.$.prisma, { @@ -627,6 +637,7 @@ export class WaitpointSystem { id: b.waitpoint.id, index: b.batchIndex ?? undefined, })), + ...(completedWaitpointRecords && { completedWaitpointRecords }), } ); @@ -672,6 +683,11 @@ export class WaitpointSystem { ); } + const completedWaitpointRecords = await this.#completedWaitpointRecordsFor( + runId, + blockingWaitpoints + ); + //put it back in the queue, with the original timestamp (w/ priority) //this prioritizes dequeuing waiting runs over new runs const newSnapshot = await this.enqueueSystem.enqueueRun({ @@ -686,6 +702,7 @@ export class WaitpointSystem { id: b.waitpoint.id, index: b.batchIndex ?? undefined, })), + ...(completedWaitpointRecords && { completedWaitpointRecords }), checkpointId: snapshot.checkpointId ?? undefined, }); @@ -738,6 +755,54 @@ export class WaitpointSystem { }); } + /** + * The record set for one resume, or undefined when no blocking waitpoint carries a store-format + * id. + * + * Gated on id FORMAT, not residency. The two are not the same during a migration: a + * store-format id can still be served by the Postgres arm, exactly as run-ops ids were for + * runs. Whichever arm owns it answers, so the gate only decides whether to ask at all. + * + * That gate is what keeps this inert. `parseWaitpointId` reports legacy for every id minted + * today, so no live resume reads an envelope or writes a record until a waitpoint mints in + * store format. + * + * ROLLOUT ORDER. Because the gate reads the id and not the organisation, the first store-format + * mint is what starts the cost, for every organisation that then holds one -- not the first + * snapshot-store flip. The arm that answers here is the Postgres one until a store arm is + * wired, so a mint enabled ahead of the store means a projected `Waitpoint` read on the + * WRITER, inside the run lock, once per resume. That is bounded and correct, but it is not + * free, so store-format minting should follow the snapshot store rather than lead it. + */ + async #completedWaitpointRecordsFor( + runId: string, + blockingWaitpoints: RunBlockEdge[] + ): Promise { + // `.some` before the dedup, so the gate allocates nothing in the case that is universal + // today and stays common through a partial rollout: no blocking waitpoint is store-format. + // Building the mapped array, the filtered array and the Set first meant three allocations + // per resume -- for a 1000-wide batch fan-in too -- to discover there was nothing to ask for. + if (!blockingWaitpoints.some((b) => parseWaitpointId(b.waitpoint.id).format === "b32hexW")) { + return undefined; + } + + // Only a set that really holds one pays for the dedup. + const storeFormatIds = [ + ...new Set( + blockingWaitpoints + .map((b) => b.waitpoint.id) + .filter((id) => parseWaitpointId(id).format === "b32hexW") + ), + ]; + + const sources = await this.coordinator.readCompletionEnvelopes({ + runId, + waitpointIds: storeFormatIds, + }); + + return buildCompletedWaitpointRecords(sources); + } + /** * Builds the waitpoint output payload from a completed run's stored output/error. */ diff --git a/internal-packages/run-engine/src/engine/waitpointCoordinator/completedWaitpointEquivalence.test.ts b/internal-packages/run-engine/src/engine/waitpointCoordinator/completedWaitpointEquivalence.test.ts new file mode 100644 index 00000000000..bc8a551179d --- /dev/null +++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/completedWaitpointEquivalence.test.ts @@ -0,0 +1,422 @@ +// The resolver must produce what the executor already consumes, so the oracle is the +// existing hydration and not a hand-written literal. A literal cannot catch a drift in +// enhanceExecutionSnapshotWithWaitpoints itself; this can. +import { postgresTest } from "@internal/testcontainers"; +import { PostgresRunStore } from "@internal/run-store"; +import type { PrismaClient, Waitpoint } from "@trigger.dev/database"; +import { describe, expect } from "vitest"; +import { seedChildRunsWithOutputs, seedChildRunWithOutput } from "./testFixtures/childRun.js"; +import { enhanceExecutionSnapshotWithWaitpoints } from "../systems/executionSnapshotSystem.js"; +import { buildCompletedWaitpointRecords } from "./completedWaitpointRecords.js"; +import { + createCompletedWaitpointResolver, + createRunOutputsReader, +} from "./completedWaitpointResolver.js"; +import { envelopeSourceFromWaitpointRow } from "./completionEnvelopeSource.js"; +import type { CompletionEnvelopeSource } from "./types.js"; + +const COMPLETED_AT = new Date("2026-08-25T00:00:00.000Z"); +const RUN_ID = "run_0123456789abcdefghijklm"; +const BATCH_ID = "batch_0123456789abcdefghijk"; + +/** + * One waitpoint, in both shapes, from one description. Keeping them in one factory is what + * makes the comparison meaningful: a field added to only one shape shows up as a diff. + */ +function pair(overrides: { + id: string; + type: Waitpoint["type"]; + output?: string | null; + outputType?: string; + outputIsError?: boolean; + completedByTaskRunId?: string | null; + completedByBatchId?: string | null; + completedAfter?: Date | null; + idempotencyKey?: string; + userProvidedIdempotencyKey?: boolean; + inactiveIdempotencyKey?: string | null; +}): { row: Waitpoint; source: CompletionEnvelopeSource } { + const row = { + id: overrides.id, + friendlyId: `waitpoint_${overrides.id}`, + type: overrides.type, + status: "COMPLETED", + completedAt: COMPLETED_AT, + output: overrides.output ?? null, + outputType: overrides.outputType ?? "application/json", + outputIsError: overrides.outputIsError ?? false, + completedByTaskRunId: overrides.completedByTaskRunId ?? null, + completedByBatchId: overrides.completedByBatchId ?? null, + completedAfter: overrides.completedAfter ?? null, + idempotencyKey: overrides.idempotencyKey ?? "internal", + userProvidedIdempotencyKey: overrides.userProvidedIdempotencyKey ?? false, + inactiveIdempotencyKey: overrides.inactiveIdempotencyKey ?? null, + } as unknown as Waitpoint; + + // Through the SHARED mapper the legacy arm uses. A hand-rolled copy here would make a bug in + // that arm invisible to every case below, because the oracle chain would never touch it. + return { row, source: envelopeSourceFromWaitpointRow(row) }; +} + +function snapshot(batchId: string | null) { + return { id: "snap_1", runId: RUN_ID, batchId } as never; +} + +function sortEntries(entries: T[]): T[] { + return [...entries].sort((a, b) => a.id.localeCompare(b.id) || (a.index ?? -1) - (b.index ?? -1)); +} + +/** + * Run one description through both paths and assert the results match. + * + * `deriveFromRun` is the one case where the two paths cannot be identical by construction: + * the row carries the value and the record carries a marker. Feeding the row's own output + * back as the run's output is what makes them comparable, which is exactly the claim the + * variant makes — that TaskRun.output holds the same string. + */ +async function bothPaths( + prisma: PrismaClient, + pairs: ReturnType[], + order: string[], + batchId: string | null = null +) { + const expected = enhanceExecutionSnapshotWithWaitpoints( + snapshot(batchId), + pairs.map((p) => p.row), + order + ).completedWaitpoints; + + const runStore = new PostgresRunStore({ prisma, readOnlyPrisma: prisma }); + + const actual = await createCompletedWaitpointResolver({ + readRunOutputs: createRunOutputsReader(runStore), + })({ + runId: RUN_ID, + ...(batchId ? { batchId } : {}), + pointer: { cycleSeq: 1, count: order.length }, + order, + distinctIds: [...new Set(pairs.map((p) => p.row.id))], + records: buildCompletedWaitpointRecords(pairs.map((p) => p.source)), + }); + + return { expected: sortEntries(expected), actual: sortEntries(actual) }; +} + +describe("the resolver reproduces the existing hydration", () => { + postgresTest("for a single MANUAL waitpoint with an inline output", async ({ prisma }) => { + const { expected, actual } = await bothPaths( + prisma, + [pair({ id: "wp_manual", type: "MANUAL", output: '{"token":1}' })], + [] + ); + + expect(actual).toEqual(expected); + }); + + postgresTest( + "for a MANUAL waitpoint with a user-provided idempotency key", + async ({ prisma }) => { + const { expected, actual } = await bothPaths( + prisma, + [ + pair({ + id: "wp_manual", + type: "MANUAL", + output: '{"token":1}', + idempotencyKey: "user-key", + userProvidedIdempotencyKey: true, + }), + ], + [] + ); + + expect(actual).toEqual(expected); + expect(actual[0]?.idempotencyKey).toBe("user-key"); + } + ); + + postgresTest( + "for an idempotency key the user provided but that went inactive", + async ({ prisma }) => { + const { expected, actual } = await bothPaths( + prisma, + [ + pair({ + id: "wp_manual", + type: "MANUAL", + output: '{"token":1}', + idempotencyKey: "user-key", + userProvidedIdempotencyKey: true, + inactiveIdempotencyKey: "old", + }), + ], + [] + ); + + expect(actual).toEqual(expected); + expect(actual[0]?.idempotencyKey).toBeUndefined(); + } + ); + + postgresTest("for a DATETIME waitpoint", async ({ prisma }) => { + const { expected, actual } = await bothPaths( + prisma, + [ + pair({ + id: "wp_datetime", + type: "DATETIME", + completedAfter: new Date("2026-08-26T00:00:00.000Z"), + }), + ], + [] + ); + + expect(actual).toEqual(expected); + }); + + postgresTest("for a RUN waitpoint outside a batch", async ({ prisma }) => { + const childRunId = await seedChildRunWithOutput(prisma, '{"ok":true}'); + const { expected, actual } = await bothPaths( + prisma, + [ + pair({ + id: "wp_run", + type: "RUN", + output: '{"ok":true}', + completedByTaskRunId: childRunId, + }), + ], + [] + ); + + expect(actual).toEqual(expected); + }); + + // Two deferring records in ONE cycle, which is the shape a batch fan-in produces and the one + // the batched read introduced. Every other case here defers at most once, so a mis-keyed map + // or a swapped output would compare equal against the oracle in all of them. + // + // The outputs differ deliberately, and one run sits at two interleaved positions, so a swap + // between the two runs and a lost second position both show up as a diff. + postgresTest("for two RUN waitpoints deferring to different runs", async ({ prisma }) => { + const [runA, runB] = await seedChildRunsWithOutputs(prisma, ['{"child":"a"}', '{"child":"b"}']); + + const { expected, actual } = await bothPaths( + prisma, + [ + pair({ + id: "wp_run_a", + type: "RUN", + output: '{"child":"a"}', + completedByTaskRunId: runA, + }), + pair({ + id: "wp_run_b", + type: "RUN", + output: '{"child":"b"}', + completedByTaskRunId: runB, + }), + ], + ["wp_run_a", "wp_run_b", "wp_run_a"], + BATCH_ID + ); + + expect(actual).toEqual(expected); + // Stated as well as compared: the oracle agreeing is the assertion, but a reader should not + // have to run it to see that three entries come back and each output landed on its own id. + expect(actual).toHaveLength(3); + expect(actual.filter((w) => w.id === "wp_run_a").map((w) => w.output)).toEqual([ + '{"child":"a"}', + '{"child":"a"}', + ]); + expect(actual.find((w) => w.id === "wp_run_b")?.output).toBe('{"child":"b"}'); + }); + + postgresTest("for a RUN waitpoint read under a batch", async ({ prisma }) => { + const childRunId = await seedChildRunWithOutput(prisma, '{"ok":true}'); + const { expected, actual } = await bothPaths( + prisma, + [ + pair({ + id: "wp_run", + type: "RUN", + output: '{"ok":true}', + completedByTaskRunId: childRunId, + }), + ], + ["wp_run"], + BATCH_ID + ); + + expect(actual).toEqual(expected); + expect(actual[0]?.completedByTaskRun?.batch?.id).toBe(BATCH_ID); + }); + + postgresTest("for a RUN waitpoint whose output is an error", async ({ prisma }) => { + const childRunId = await seedChildRunWithOutput(prisma, '{"message":"boom"}'); + const { expected, actual } = await bothPaths( + prisma, + [ + pair({ + id: "wp_run", + type: "RUN", + output: '{"message":"boom"}', + outputIsError: true, + completedByTaskRunId: childRunId, + }), + ], + [] + ); + + expect(actual).toEqual(expected); + }); + + postgresTest("for a BATCH waitpoint", async ({ prisma }) => { + const { expected, actual } = await bothPaths( + prisma, + [pair({ id: "wp_batch", type: "BATCH", completedByBatchId: BATCH_ID })], + [] + ); + + expect(actual).toEqual(expected); + }); + + postgresTest("for an already-offloaded output", async ({ prisma }) => { + const { expected, actual } = await bothPaths( + prisma, + [ + pair({ + id: "wp_manual", + type: "MANUAL", + output: "store-key-1", + outputType: "application/store", + }), + ], + [] + ); + + expect(actual).toEqual(expected); + }); + + // The case the suite was blind to, and the one the frozen reference orders the other way. The + // oracle emits the ref string; so does this, by a different branch. + postgresTest("for an offloaded RUN success", async ({ prisma }) => { + const childRunId = await seedChildRunWithOutput(prisma, "s3://bucket/key"); + const { expected, actual } = await bothPaths( + prisma, + [ + pair({ + id: "wp_run_ref", + type: "RUN", + output: "s3://bucket/key", + outputType: "application/store", + completedByTaskRunId: childRunId, + }), + ], + [] + ); + + expect(actual).toEqual(expected); + expect(actual[0]?.output).toBe("s3://bucket/key"); + }); + + postgresTest("for one run present at two batch indexes", async ({ prisma }) => { + const childRunId = await seedChildRunWithOutput(prisma, '{"ok":true}'); + const { expected, actual } = await bothPaths( + prisma, + [ + pair({ + id: "wp_run", + type: "RUN", + output: '{"ok":true}', + completedByTaskRunId: childRunId, + }), + ], + ["wp_run", "wp_run"], + BATCH_ID + ); + + expect(actual).toEqual(expected); + expect(actual.map((w) => w.index)).toEqual([0, 1]); + }); + + postgresTest("for an index-less waitpoint sitting beside indexed ones", async ({ prisma }) => { + // Seeded to match the RUN row's own output, which is the parity premise: TaskRun.output + // holds the same string the waitpoint carried. + const childRunId = await seedChildRunWithOutput(prisma, '{"ok":true}'); + const { expected, actual } = await bothPaths( + prisma, + [ + pair({ id: "wp_indexless", type: "MANUAL", output: '{"token":1}' }), + pair({ + id: "wp_run", + type: "RUN", + output: '{"ok":true}', + completedByTaskRunId: childRunId, + }), + ], + ["wp_run"], + BATCH_ID + ); + + expect(actual).toEqual(expected); + expect(actual.find((w) => w.id === "wp_indexless")?.index).toBeUndefined(); + }); + + // The ONE intentional divergence from the oracle. A BATCH waitpoint really is completed with + // an output, but the executor never reads it (sharedRuntimeManager.resolveWaitpoint + // early-returns on type). Pinned so that if that early return ever goes away, this fails and + // says why, instead of the output silently being missing at resume. + postgresTest("deliberately drops a BATCH output, unlike the oracle", async ({ prisma }) => { + const { expected, actual } = await bothPaths( + prisma, + [ + pair({ + id: "wp_batch", + type: "BATCH", + completedByBatchId: BATCH_ID, + output: '{"message":"batch expired"}', + outputIsError: true, + }), + ], + [] + ); + + expect(expected[0]?.output).toBe('{"message":"batch expired"}'); + expect(actual[0]?.output).toBeUndefined(); + expect(actual[0]?.outputIsError).toBe(true); + }); + + postgresTest("for every type at once, under a batch", async ({ prisma }) => { + const childRunId = await seedChildRunWithOutput(prisma, '{"ok":true}'); + const { expected, actual } = await bothPaths( + prisma, + [ + pair({ + id: "wp_run", + type: "RUN", + output: '{"ok":true}', + completedByTaskRunId: childRunId, + }), + pair({ id: "wp_batch", type: "BATCH", completedByBatchId: BATCH_ID }), + pair({ + id: "wp_datetime", + type: "DATETIME", + completedAfter: new Date("2026-08-26T00:00:00.000Z"), + }), + pair({ + id: "wp_manual", + type: "MANUAL", + output: '{"token":1}', + idempotencyKey: "user-key", + userProvidedIdempotencyKey: true, + }), + ], + ["wp_run", "wp_batch", "wp_datetime"], + BATCH_ID + ); + + expect(actual).toEqual(expected); + expect(actual).toHaveLength(4); + }); +}); diff --git a/internal-packages/run-engine/src/engine/waitpointCoordinator/completedWaitpointRecords.test.ts b/internal-packages/run-engine/src/engine/waitpointCoordinator/completedWaitpointRecords.test.ts new file mode 100644 index 00000000000..80c343db33b --- /dev/null +++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/completedWaitpointRecords.test.ts @@ -0,0 +1,173 @@ +import { describe, expect, it } from "vitest"; +import { buildCompletedWaitpointRecords } from "./completedWaitpointRecords.js"; +import type { CompletionEnvelopeSource } from "./types.js"; + +const COMPLETED_AT = new Date("2026-08-25T00:00:00.000Z"); + +function source(overrides: Partial = {}): CompletionEnvelopeSource { + return { + id: "wp_1", + friendlyId: "waitpoint_wp_1", + type: "MANUAL", + completedAt: COMPLETED_AT, + outputType: "application/json", + outputIsError: false, + ...overrides, + }; +} + +describe("buildCompletedWaitpointRecords", () => { + it("emits one record per distinct id", () => { + const records = buildCompletedWaitpointRecords([source(), source()]); + + expect(records).toHaveLength(1); + }); + + it("emits one record for each of several distinct ids", () => { + const records = buildCompletedWaitpointRecords([ + source({ id: "wp_1" }), + source({ id: "wp_2" }), + ]); + + expect(records.map((r) => r.id)).toEqual(["wp_1", "wp_2"]); + }); + + it("writes completedAt as an ISO string", () => { + const [record] = buildCompletedWaitpointRecords([source()]); + + expect(record?.completedAt).toBe("2026-08-25T00:00:00.000Z"); + }); + + it("omits every absent optional field rather than writing undefined", () => { + const [record] = buildCompletedWaitpointRecords([source()]); + + expect("completedByTaskRunId" in record!).toBe(false); + expect("completedByBatchId" in record!).toBe(false); + expect("completedAfter" in record!).toBe(false); + expect("idempotencyKey" in record!).toBe(false); + }); + + it("carries the fields the executor shape needs", () => { + const [record] = buildCompletedWaitpointRecords([ + source({ + completedAfter: new Date("2026-08-26T00:00:00.000Z"), + idempotencyKey: "user-key", + }), + ]); + + expect(record).toMatchObject({ + id: "wp_1", + friendlyId: "waitpoint_wp_1", + type: "MANUAL", + outputType: "application/json", + outputIsError: false, + completedAfter: "2026-08-26T00:00:00.000Z", + idempotencyKey: "user-key", + }); + }); + + describe("the output variant", () => { + it("keeps an already-offloaded value as a ref", () => { + const [record] = buildCompletedWaitpointRecords([ + source({ outputRef: "store-key-1", outputType: "application/store" }), + ]); + + expect(record?.output).toEqual({ ref: "store-key-1" }); + }); + + it("prefers a ref over an inline value when both are somehow present", () => { + const [record] = buildCompletedWaitpointRecords([ + source({ output: '{"ok":true}', outputRef: "store-key-1" }), + ]); + + expect(record?.output).toEqual({ ref: "store-key-1" }); + }); + + // Deliberately a ref, not deriveFromRun, and the opposite of the reference implementation in + // completedWaitpointFreeze.test.ts. Byte-identical either way, and this route stays + // resolvable when the completing run row is gone. + it("routes an offloaded RUN success down the ref branch", () => { + const [record] = buildCompletedWaitpointRecords([ + source({ + type: "RUN", + outputRef: "s3://bucket/key", + outputType: "application/store", + completedByTaskRunId: "run_1", + }), + ]); + + expect(record?.output).toEqual({ ref: "s3://bucket/key" }); + }); + + it("marks a plain RUN output as derivable from the run", () => { + const [record] = buildCompletedWaitpointRecords([ + source({ type: "RUN", output: '{"ok":true}', completedByTaskRunId: "run_1" }), + ]); + + expect(record?.output).toEqual({ deriveFromRun: true }); + }); + + // TaskRun.error is jsonb and does not round-trip to the same string, so a RUN error can + // never be re-read from the run row. + it("keeps a RUN error inline", () => { + const [record] = buildCompletedWaitpointRecords([ + source({ + type: "RUN", + output: '{"message":"boom"}', + outputIsError: true, + completedByTaskRunId: "run_1", + }), + ]); + + expect(record?.output).toEqual({ inline: '{"message":"boom"}' }); + }); + + // The back-reference is onDelete: SetNull, so an orphaned RUN waitpoint has no run row + // left to derive from. + it("keeps an orphaned RUN inline", () => { + const [record] = buildCompletedWaitpointRecords([ + source({ type: "RUN", output: '{"ok":true}' }), + ]); + + expect(record?.output).toEqual({ inline: '{"ok":true}' }); + }); + + it("omits a BATCH output, because the runtime discards it at source", () => { + const [record] = buildCompletedWaitpointRecords([ + source({ type: "BATCH", completedByBatchId: "batch_1", output: '{"ignored":true}' }), + ]); + + expect(record?.output).toBeNull(); + }); + + it("keeps a MANUAL output inline", () => { + const [record] = buildCompletedWaitpointRecords([source({ output: '{"token":1}' })]); + + expect(record?.output).toEqual({ inline: '{"token":1}' }); + }); + + it("keeps a DATETIME output inline", () => { + const [record] = buildCompletedWaitpointRecords([ + source({ type: "DATETIME", output: '{"at":1}' }), + ]); + + expect(record?.output).toEqual({ inline: '{"at":1}' }); + }); + + it("writes null when there is no output at all", () => { + const [record] = buildCompletedWaitpointRecords([source()]); + + expect(record?.output).toBeNull(); + }); + + it("keeps an empty-string output inline, because empty is a value and not an absence", () => { + const [record] = buildCompletedWaitpointRecords([source({ output: "" })]); + + expect(record?.output).toEqual({ inline: "" }); + }); + }); + + it("returns an empty set for no sources", () => { + expect(buildCompletedWaitpointRecords([])).toEqual([]); + }); +}); diff --git a/internal-packages/run-engine/src/engine/waitpointCoordinator/completedWaitpointRecords.ts b/internal-packages/run-engine/src/engine/waitpointCoordinator/completedWaitpointRecords.ts new file mode 100644 index 00000000000..c199046a49b --- /dev/null +++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/completedWaitpointRecords.ts @@ -0,0 +1,70 @@ +import type { CompletedWaitpointRecord, CompletedWaitpointRecordOutput } from "@internal/run-store"; +import type { CompletionEnvelopeSource } from "./types.js"; + +/** + * Turn sourced envelope fields into the frozen record set that rides one wait cycle's key. + * + * One record per DISTINCT id. The cycle's ordered id list carries multiplicity, and the + * resolver expands one record into one entry per position of its id. That list holds only + * batch-indexed ids, because its positions ARE the indexes, so this set — not the list — is + * authoritative for membership. + */ +export function buildCompletedWaitpointRecords( + sources: CompletionEnvelopeSource[] +): CompletedWaitpointRecord[] { + const byId = new Map(); + + for (const source of sources) { + if (byId.has(source.id)) { + continue; + } + + byId.set(source.id, { + id: source.id, + friendlyId: source.friendlyId, + type: source.type, + completedAt: source.completedAt.toISOString(), + outputType: source.outputType, + outputIsError: source.outputIsError, + output: chooseOutput(source), + ...(source.completedByTaskRunId && { completedByTaskRunId: source.completedByTaskRunId }), + ...(source.completedByBatchId && { completedByBatchId: source.completedByBatchId }), + ...(source.completedAfter && { completedAfter: source.completedAfter.toISOString() }), + ...(source.idempotencyKey && { idempotencyKey: source.idempotencyKey }), + }); + } + + return [...byId.values()]; +} + +function chooseOutput(source: CompletionEnvelopeSource): CompletedWaitpointRecordOutput { + // Ref BEFORE the RUN branch, which is the opposite order to the reference implementation in + // completedWaitpointFreeze.test.ts. Both are byte-identical at read time, by that reference's + // own reasoning: an offloaded RUN success has the same ref string in TaskRun.output. This + // order is preferred because it needs no Postgres read to recover a string already in hand, + // and because a deriveFromRun record whose run row is later deleted now refuses rather than + // resolving empty — so routing an offloaded RUN success down the ref branch keeps it + // resolvable when that row is gone. + if (source.outputRef !== undefined) { + return { ref: source.outputRef }; + } + + // A plain RUN output is re-readable from TaskRun.output verbatim. Two RUN cases are not, + // and both must stay inline: an ERROR, because TaskRun.error is jsonb and does not + // round-trip to the same string, and an ORPHAN, because the back-reference is + // onDelete: SetNull so the completing row may be gone. + if (source.type === "RUN" && !source.outputIsError && source.completedByTaskRunId) { + return { deriveFromRun: true }; + } + + // Deliberately dropped, and this is the one place the record set does NOT reproduce the row. + // A BATCH waitpoint IS completed with an output (see batchSystem), but the executor ignores + // it: sharedRuntimeManager.resolveWaitpoint early-returns for type === "BATCH" and never + // reads the body. Carrying it would put bytes in the cycle key that nothing can observe. + if (source.type === "BATCH") { + return null; + } + + // An empty string is a value, not an absence, so this checks undefined only. + return source.output === undefined ? null : { inline: source.output }; +} diff --git a/internal-packages/run-engine/src/engine/waitpointCoordinator/completedWaitpointResolver.runOutput.test.ts b/internal-packages/run-engine/src/engine/waitpointCoordinator/completedWaitpointResolver.runOutput.test.ts new file mode 100644 index 00000000000..078397e6067 --- /dev/null +++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/completedWaitpointResolver.runOutput.test.ts @@ -0,0 +1,255 @@ +// The deriveFromRun branch, against a real TaskRun row. +// +// This branch is the resolver's only Postgres read, so it is the one part that cannot be proved +// by a pure test: the claim is that TaskRun.output holds the same string the waitpoint carried, +// and only a real row can settle that. The pure suite covers everything that does not read. +import { postgresTest } from "@internal/testcontainers"; +import { PostgresRunStore } from "@internal/run-store"; +import type { CompletedWaitpointRecord } from "@internal/run-store"; +import type { PrismaClient } from "@trigger.dev/database"; +import { describe, expect } from "vitest"; +import { + createCompletedWaitpointResolver, + createRunOutputsReader, + UnresolvableWaitpointId, +} from "./completedWaitpointResolver.js"; +import { seedChildRunsWithOutputs, seedChildRunWithOutput } from "./testFixtures/childRun.js"; + +function deriveRecord(completedByTaskRunId: string): CompletedWaitpointRecord { + return { + id: "wp_run", + friendlyId: "waitpoint_wp_run", + type: "RUN", + completedAt: "2026-08-26T00:00:00.000Z", + outputType: "application/json", + outputIsError: false, + output: { deriveFromRun: true }, + completedByTaskRunId, + }; +} + +function resolverFor(prisma: PrismaClient) { + const runStore = new PostgresRunStore({ prisma, readOnlyPrisma: prisma }); + return createCompletedWaitpointResolver({ readRunOutputs: createRunOutputsReader(runStore) }); +} + +/** + * A resolver that records the id set of every batched read and DELEGATES to the real reader, so + * the Postgres read still happens. + * + * Wrapping the collaborator rather than replacing it is deliberate: the assertion is about how + * many reads occur and what they ask for, and neither is observable from the resolved output. + */ +function countingResolver(prisma: PrismaClient) { + const runStore = new PostgresRunStore({ prisma, readOnlyPrisma: prisma }); + const read = createRunOutputsReader(runStore); + const batches: string[][] = []; + + return { + batches, + resolve: createCompletedWaitpointResolver({ + readRunOutputs: async (ids) => { + batches.push(ids); + return read(ids); + }, + }), + }; +} + +describe("the deriveFromRun branch", () => { + postgresTest("reads the completing run's output verbatim", async ({ prisma }) => { + const stored = '{"value":42,"nested":{"a":[1,2,3]}}'; + const runId = await seedChildRunWithOutput(prisma, stored); + + const [entry] = await resolverFor(prisma)({ + runId: "run_parent", + pointer: { cycleSeq: 1, count: 0 }, + order: [], + distinctIds: ["wp_run"], + records: [deriveRecord(runId)], + }); + + // Byte-identical, which is the whole premise of the variant. + expect(entry?.output).toBe(stored); + }); + + postgresTest("carries an offloaded ref through unchanged", async ({ prisma }) => { + const runId = await seedChildRunWithOutput(prisma, "s3://bucket/key"); + + const [entry] = await resolverFor(prisma)({ + runId: "run_parent", + pointer: { cycleSeq: 1, count: 0 }, + order: [], + distinctIds: ["wp_run"], + records: [deriveRecord(runId)], + }); + + expect(entry?.output).toBe("s3://bucket/key"); + }); + + // The run row disappearing between the record write and the read. Postgres does not lose the + // value on the legacy path, so resolving empty here would resolve a triggerAndWait with + // silently wrong data. + postgresTest("refuses when the completing run is gone", async ({ prisma }) => { + const runId = await seedChildRunWithOutput(prisma, '{"value":42}'); + await prisma.taskRun.delete({ where: { id: runId } }); + + const failure = await resolverFor(prisma)({ + runId: "run_parent", + pointer: { cycleSeq: 1, count: 0 }, + order: [], + distinctIds: ["wp_run"], + records: [deriveRecord(runId)], + }).catch((caught: unknown) => caught as UnresolvableWaitpointId); + + expect(failure).toBeInstanceOf(UnresolvableWaitpointId); + expect(failure.reason).toBe("lost-run-output"); + }); + + postgresTest("refuses when the run exists with no output", async ({ prisma }) => { + const runId = await seedChildRunWithOutput(prisma, null); + + const failure = await resolverFor(prisma)({ + runId: "run_parent", + pointer: { cycleSeq: 1, count: 0 }, + order: [], + distinctIds: ["wp_run"], + records: [deriveRecord(runId)], + }).catch((caught: unknown) => caught as UnresolvableWaitpointId); + + expect(failure).toBeInstanceOf(UnresolvableWaitpointId); + expect(failure.reason).toBe("lost-run-output"); + }); + + // One read per record, not one per position, so a run at several batch indexes does not pay a + // query per index. + postgresTest("reads the run once for a record at several indexes", async ({ prisma }) => { + const runId = await seedChildRunWithOutput(prisma, '{"value":42}'); + const { resolve, batches } = countingResolver(prisma); + + const result = await resolve({ + runId: "run_parent", + pointer: { cycleSeq: 1, count: 2 }, + order: ["wp_run", "wp_run"], + distinctIds: ["wp_run"], + records: [deriveRecord(runId)], + }); + + expect(result).toHaveLength(2); + expect(batches).toEqual([[runId]]); + }); + + // The shape a batch fan-in produces. Every deferring record resolves in ONE read, not one + // read each: a per-record read put a serial round trip per child on the resume path, which is + // the cost the record set exists to remove. + postgresTest("reads every deferred run in one batch", async ({ prisma }) => { + const runIds = await seedChildRunsWithOutputs( + prisma, + Array.from({ length: 12 }, (_, i) => `{"value":${i}}`) + ); + const { resolve, batches } = countingResolver(prisma); + + const records = runIds.map((runId, i) => ({ + ...deriveRecord(runId), + id: `wp_run_${i}`, + friendlyId: `waitpoint_wp_run_${i}`, + })); + + const result = await resolve({ + runId: "run_parent", + pointer: { cycleSeq: 1, count: records.length }, + order: records.map((r) => r.id), + distinctIds: records.map((r) => r.id), + records, + }); + + expect(result).toHaveLength(12); + // One batch, holding every distinct run id. + expect(batches).toHaveLength(1); + expect(batches[0]?.slice().sort()).toEqual(runIds.slice().sort()); + // And each output landed on its own waitpoint. + for (const [i, runId] of runIds.entries()) { + const entry = result.find((w) => w.id === `wp_run_${i}`); + expect(entry?.output).toBe(`{"value":${i}}`); + expect(runId).toBeTruthy(); + } + }); + + // An empty string is a VALUE, not an absence. The reader's `row.output !== null` is what keeps + // it: narrowed to a truthy test it would report the run as output-less and throw + // lost-run-output on a run that completed perfectly well. Nothing else pins that, and + // `chooseOutput` already treats empty as a value on the write side. + postgresTest("keeps an empty-string output rather than refusing", async ({ prisma }) => { + const runId = await seedChildRunWithOutput(prisma, ""); + + const [entry] = await resolverFor(prisma)({ + runId: "run_parent", + pointer: { cycleSeq: 1, count: 0 }, + order: [], + distinctIds: ["wp_run"], + records: [deriveRecord(runId)], + }); + + expect(entry?.output).toBe(""); + }); + + // Two waitpoints completed by the SAME run. One id in one batch, and both entries carry it: + // the dedup must not cost the second waitpoint its output. + postgresTest("reads a shared run once and hydrates both waitpoints", async ({ prisma }) => { + const runId = await seedChildRunWithOutput(prisma, '{"shared":true}'); + const { resolve, batches } = countingResolver(prisma); + + const result = await resolve({ + runId: "run_parent", + pointer: { cycleSeq: 1, count: 2 }, + order: ["wp_first", "wp_second"], + distinctIds: ["wp_first", "wp_second"], + records: [ + { ...deriveRecord(runId), id: "wp_first", friendlyId: "waitpoint_wp_first" }, + { ...deriveRecord(runId), id: "wp_second", friendlyId: "waitpoint_wp_second" }, + ], + }); + + expect(batches).toEqual([[runId]]); + expect(result).toHaveLength(2); + expect(result.map((w) => w.output)).toEqual(['{"shared":true}', '{"shared":true}']); + }); + + // A cycle that defers nothing reads nothing, so an all-inline resume pays no Postgres round + // trip at all. + postgresTest("reads nothing when no record defers", async ({ prisma }) => { + const { resolve, batches } = countingResolver(prisma); + + const result = await resolve({ + runId: "run_parent", + pointer: { cycleSeq: 1, count: 0 }, + order: [], + distinctIds: ["wp_inline"], + records: [ + { + ...deriveRecord("run_unused"), + id: "wp_inline", + friendlyId: "waitpoint_wp_inline", + output: { inline: '{"ok":true}' }, + }, + ], + }); + + expect(result[0]?.output).toBe('{"ok":true}'); + expect(batches).toEqual([]); + }); + + postgresTest("throws when a derive record arrives with no reader wired", async ({ prisma }) => { + const runId = await seedChildRunWithOutput(prisma, '{"value":42}'); + + await expect( + createCompletedWaitpointResolver({})({ + runId: "run_parent", + pointer: { cycleSeq: 1, count: 0 }, + order: [], + distinctIds: ["wp_run"], + records: [deriveRecord(runId)], + }) + ).rejects.toThrow(/no run-output reader/); + }); +}); diff --git a/internal-packages/run-engine/src/engine/waitpointCoordinator/completedWaitpointResolver.test.ts b/internal-packages/run-engine/src/engine/waitpointCoordinator/completedWaitpointResolver.test.ts new file mode 100644 index 00000000000..3e5b376a6e5 --- /dev/null +++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/completedWaitpointResolver.test.ts @@ -0,0 +1,332 @@ +import type { CompletedWaitpointRecord } from "@internal/run-store"; +import { BatchId, RunId } from "@trigger.dev/core/v3/isomorphic"; +import { describe, expect, it } from "vitest"; +import { + createCompletedWaitpointResolver, + UnresolvableWaitpointId, + type ResolveArgs, +} from "./completedWaitpointResolver.js"; + +function record(overrides: Partial = {}): CompletedWaitpointRecord { + return { + id: "wp_1", + friendlyId: "waitpoint_wp_1", + type: "MANUAL", + completedAt: "2026-08-25T00:00:00.000Z", + outputType: "application/json", + outputIsError: false, + output: { inline: '{"ok":true}' }, + ...overrides, + }; +} + +type CaseArgs = Omit & { distinctIds?: string[] }; + +/** + * Built with NO run-output reader, deliberately. Every case here carries inline, ref or null + * output, so none reaches the branch that reads Postgres. + * + * Fills `distinctIds` from the records when a case does not name it, because most cases are + * about the expansion rather than the membership. The coverage-check cases set it explicitly, + * since there it IS the subject. + */ +function resolver() { + const resolve = createCompletedWaitpointResolver({}); + return (over: CaseArgs) => + resolve({ ...over, distinctIds: over.distinctIds ?? over.records.map((r) => r.id) }); +} + +const CYCLE = { cycleSeq: 1, count: 0 }; + +describe("the index expansion", () => { + it("emits one entry per position of the id in the order", async () => { + const result = await resolver()({ + runId: "run_1", + pointer: { cycleSeq: 1, count: 2 }, + order: ["wp_1", "wp_1"], + records: [record()], + }); + + expect(result).toHaveLength(2); + expect(result.map((w) => w.index)).toEqual([0, 1]); + }); + + it("gives a run at two batch indexes its two real positions", async () => { + const result = await resolver()({ + runId: "run_1", + pointer: { cycleSeq: 1, count: 3 }, + order: ["wp_other", "wp_1", "wp_1"], + records: [record(), record({ id: "wp_other", friendlyId: "waitpoint_wp_other" })], + }); + + expect(result.filter((w) => w.id === "wp_1").map((w) => w.index)).toEqual([1, 2]); + }); + + // Every wait.for, every single triggerAndWait and every token has no batch index, so it + // is absent from the order. Dropping it here loses the run's results on resume. + it("keeps a record with no position, with an undefined index", async () => { + const result = await resolver()({ + runId: "run_1", + pointer: CYCLE, + order: [], + records: [record()], + }); + + expect(result).toHaveLength(1); + expect(result[0]?.index).toBeUndefined(); + }); + + it("keeps an index-less record alongside an indexed one", async () => { + const result = await resolver()({ + runId: "run_1", + pointer: { cycleSeq: 1, count: 1 }, + order: ["wp_indexed"], + records: [record(), record({ id: "wp_indexed", friendlyId: "waitpoint_wp_indexed" })], + }); + + expect(result).toHaveLength(2); + expect(result.find((w) => w.id === "wp_1")?.index).toBeUndefined(); + expect(result.find((w) => w.id === "wp_indexed")?.index).toBe(0); + }); +}); + +describe("the executor shape", () => { + it("reproduces the scalar fields", async () => { + const [entry] = await resolver()({ + runId: "run_1", + pointer: CYCLE, + order: [], + records: [record({ idempotencyKey: "user-key" })], + }); + + expect(entry).toMatchObject({ + id: "wp_1", + friendlyId: "waitpoint_wp_1", + type: "MANUAL", + completedAt: new Date("2026-08-25T00:00:00.000Z"), + idempotencyKey: "user-key", + output: '{"ok":true}', + outputType: "application/json", + outputIsError: false, + }); + }); + + it("builds completedByTaskRun for a RUN record", async () => { + const childRunId = RunId.fromFriendlyId(RunId.generate().friendlyId); + + const [entry] = await resolver()({ + runId: "run_1", + pointer: CYCLE, + order: [], + records: [record({ type: "RUN", completedByTaskRunId: childRunId, output: null })], + }); + + expect(entry?.completedByTaskRun).toEqual({ + id: childRunId, + friendlyId: RunId.toFriendlyId(childRunId), + }); + }); + + // The cycle is minted once, but a later entry in the resume chain can be read under a + // different batch. The batch shown must be the reading entry's, never the minting one's. + it("takes batch{} from the reading entry's batchId", async () => { + const childRunId = RunId.fromFriendlyId(RunId.generate().friendlyId); + const batchId = BatchId.fromFriendlyId(BatchId.generate().friendlyId); + + const [entry] = await resolver()({ + runId: "run_1", + batchId, + pointer: CYCLE, + order: [], + records: [record({ type: "RUN", completedByTaskRunId: childRunId, output: null })], + }); + + expect(entry?.completedByTaskRun?.batch).toEqual({ + id: batchId, + friendlyId: BatchId.toFriendlyId(batchId), + }); + }); + + it("omits batch{} when the reading entry has no batch", async () => { + const childRunId = RunId.fromFriendlyId(RunId.generate().friendlyId); + + const [entry] = await resolver()({ + runId: "run_1", + pointer: CYCLE, + order: [], + records: [record({ type: "RUN", completedByTaskRunId: childRunId, output: null })], + }); + + expect(entry?.completedByTaskRun?.batch).toBeUndefined(); + }); + + it("builds completedByBatch for a BATCH record", async () => { + const batchId = BatchId.fromFriendlyId(BatchId.generate().friendlyId); + + const [entry] = await resolver()({ + runId: "run_1", + pointer: CYCLE, + order: [], + records: [record({ type: "BATCH", completedByBatchId: batchId, output: null })], + }); + + expect(entry?.completedByBatch).toEqual({ + id: batchId, + friendlyId: BatchId.toFriendlyId(batchId), + }); + }); + + it("carries completedAfter as a Date", async () => { + const [entry] = await resolver()({ + runId: "run_1", + pointer: CYCLE, + order: [], + records: [record({ type: "DATETIME", completedAfter: "2026-08-26T00:00:00.000Z" })], + }); + + expect(entry?.completedAfter).toEqual(new Date("2026-08-26T00:00:00.000Z")); + }); +}); + +describe("the output hydration", () => { + it("returns an inline value as-is", async () => { + const [entry] = await resolver()({ + runId: "run_1", + pointer: CYCLE, + order: [], + records: [record({ output: { inline: '{"v":1}' } })], + }); + + expect(entry?.output).toBe('{"v":1}'); + }); + + it("returns a ref as the output, so the executor resolves it the existing way", async () => { + const [entry] = await resolver()({ + runId: "run_1", + pointer: CYCLE, + order: [], + records: [record({ output: { ref: "store-key-1" }, outputType: "application/store" })], + }); + + expect(entry?.output).toBe("store-key-1"); + }); + + // The deriveFromRun branch is the resolver's only Postgres read, so its cases live in + // completedWaitpointResolver.runOutput.test.ts against a real TaskRun row: the found output, + // the deleted row, the output-less row, the one-read-per-record property, and the unwired + // reader. Faking the read here would assert only that the fake was called. + + it("leaves the output undefined when the record carries none", async () => { + const [entry] = await resolver()({ + runId: "run_1", + pointer: CYCLE, + order: [], + records: [record({ output: null })], + }); + + expect(entry?.output).toBeUndefined(); + }); +}); + +// The id classifier is total and never throws: an unrecognised shape classifies as legacy, +// finds no row, and would otherwise vanish from the resumed run's completed set with no +// error. These are the tests that make that impossible. +describe("the coverage check", () => { + it("throws when the order names an id no half resolved", async () => { + await expect( + resolver()({ + runId: "run_1", + pointer: { cycleSeq: 1, count: 1 }, + order: ["wp_missing"], + records: [record()], + }) + ).rejects.toThrow(UnresolvableWaitpointId); + }); + + it("names the offending id and the reason", async () => { + const error = await resolver()({ + runId: "run_1", + pointer: { cycleSeq: 1, count: 1 }, + order: ["wp_missing"], + records: [record()], + }).catch((caught: unknown) => caught as UnresolvableWaitpointId); + + expect(error.waitpointId).toBe("wp_missing"); + expect(error.reason).toBe("no-source"); + }); + + it("accepts an ordered id that the caller resolved from a row", async () => { + const result = await resolver()({ + runId: "run_1", + pointer: { cycleSeq: 1, count: 1 }, + order: ["wp_legacy"], + records: [record()], + resolvedElsewhere: ["wp_legacy"], + }); + + expect(result.map((w) => w.id)).toEqual(["wp_1"]); + }); + + it("throws when both halves claim the same id", async () => { + const error = await resolver()({ + runId: "run_1", + pointer: CYCLE, + order: [], + records: [record()], + resolvedElsewhere: ["wp_1"], + }).catch((caught: unknown) => caught as UnresolvableWaitpointId); + + expect(error).toBeInstanceOf(UnresolvableWaitpointId); + expect(error.waitpointId).toBe("wp_1"); + expect(error.reason).toBe("two-sources"); + }); + + it("returns only its own half, leaving the legacy half to the caller", async () => { + const result = await resolver()({ + runId: "run_1", + pointer: { cycleSeq: 1, count: 2 }, + order: ["wp_legacy", "wp_1"], + records: [record()], + resolvedElsewhere: ["wp_legacy"], + }); + + expect(result.map((w) => w.id)).toEqual(["wp_1"]); + expect(result[0]?.index).toBe(1); + }); + + // The check must run over the whole membership. An index-less wait is absent from `order` by + // construction, so an order-scoped check returns [] here and the run resumes having silently + // lost its result. + it("throws when an index-less id in the membership has no record", async () => { + const failure = await resolver()({ + runId: "run_1", + pointer: CYCLE, + order: [], + distinctIds: ["wp_indexless"], + records: [], + }).catch((caught: unknown) => caught as UnresolvableWaitpointId); + + expect(failure).toBeInstanceOf(UnresolvableWaitpointId); + expect(failure.waitpointId).toBe("wp_indexless"); + expect(failure.reason).toBe("no-source"); + }); + + it("accepts an index-less id the caller resolved from a row", async () => { + const result = await resolver()({ + runId: "run_1", + pointer: CYCLE, + order: [], + distinctIds: ["wp_legacy"], + records: [], + resolvedElsewhere: ["wp_legacy"], + }); + + expect(result).toEqual([]); + }); + + it("resolves an empty cycle to nothing", async () => { + await expect( + resolver()({ runId: "run_1", pointer: CYCLE, order: [], records: [] }) + ).resolves.toEqual([]); + }); +}); diff --git a/internal-packages/run-engine/src/engine/waitpointCoordinator/completedWaitpointResolver.ts b/internal-packages/run-engine/src/engine/waitpointCoordinator/completedWaitpointResolver.ts new file mode 100644 index 00000000000..f5df6530c34 --- /dev/null +++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/completedWaitpointResolver.ts @@ -0,0 +1,304 @@ +import type { + CompletedWaitpointRecord, + ResolveCompletedWaitpointsArgs, + RunStore, +} from "@internal/run-store"; +import { BatchId, RunId } from "@trigger.dev/core/v3/isomorphic"; +import type { CompletedWaitpoint } from "@trigger.dev/core/v3/schemas"; + +/** + * A waitpoint id that no half of a snapshot can account for, or that both halves claim. + * + * This exists because the id classifier is total and never throws: an unrecognised shape + * classifies as legacy, finds no row, and would otherwise disappear from the resumed run's + * completed set with no error at all. + */ +export type UnresolvableReason = "no-source" | "two-sources" | "lost-run-output"; + +const MESSAGES: Record string> = { + "no-source": (id) => + `Waitpoint ${id} has neither a cycle record nor a fetched row. Refusing to resume without it.`, + "two-sources": (id) => + `Waitpoint ${id} resolved twice, from a cycle record and from a fetched row.`, + "lost-run-output": (id) => + `Waitpoint ${id} defers its output to its completing run, and that run's output is gone. Refusing to resume with an empty output.`, +}; + +export class UnresolvableWaitpointId extends Error { + readonly waitpointId: string; + readonly reason: UnresolvableReason; + + constructor(waitpointId: string, reason: UnresolvableReason) { + super(MESSAGES[reason](waitpointId)); + this.name = "UnresolvableWaitpointId"; + this.waitpointId = waitpointId; + this.reason = reason; + } +} + +export type CompletedWaitpointResolverDeps = { + /** + * Reads TaskRun.output for a SET of completing runs, keyed by run id. + * + * Plural on purpose. A batch parent resumes on every child at once, so a per-id reader made + * the resolver do one round trip per child -- 500 of them, in series, for a 500-wide fan-in, + * where the path this replaces did one chunked read. An id absent from the returned map is an + * absent output, which the caller refuses rather than resolving empty. + * + * Optional, because most cycles carry no `deriveFromRun` record and therefore never need it. + * A cycle that DOES carry one without a reader is a wiring error, not a data condition, so it + * throws rather than resolving empty. + */ + readRunOutputs?(taskRunIds: string[]): Promise>; +}; + +// Bounds one read, for the reason the envelope and waitpoint reads share: a run output can be +// 100KB+, so a wide fan-in read whole can exceed Node's string conversion limits. +const RUN_OUTPUT_CHUNK_SIZE = 100; + +/** + * The production reader: TaskRun.output for the completing runs, through the store so each read + * routes to the run's owning database. + * + * `findRunsByIds` is the store's own grouped replacement for `Promise.all(ids.map(findRun))`, + * and it forces `id` into the projection so the map keys correctly even though this select + * names only `output`. + * + * Takes no read client on purpose. The router reads the owning store's REPLICA when no client is + * passed, and forces its PRIMARY for any client that is not replica-branded -- so accepting one + * would let a caller put a wide `TaskRun.output` read on the writer by reflex. This read does not + * need read-your-writes: the child run committed its output before it completed the waitpoint that + * unblocked this parent, so replica lag cannot hide it. That is the opposite of the envelope read + * in the legacy arm, which reads a waitpoint completed moments earlier and must use the writer. + */ +export function createRunOutputsReader( + runStore: Pick +): (taskRunIds: string[]) => Promise> { + return async (taskRunIds) => { + const outputs = new Map(); + + for (let i = 0; i < taskRunIds.length; i += RUN_OUTPUT_CHUNK_SIZE) { + const chunk = taskRunIds.slice(i, i + RUN_OUTPUT_CHUNK_SIZE); + const rows = await runStore.findRunsByIds(chunk, { select: { output: true } }); + + for (const [id, row] of rows) { + // A row present with a null output is the same absence as a missing row: either way the + // value the waitpoint deferred is gone. Omitting it here keeps one absence rule, so the + // caller's refusal covers both. + if (row.output !== null) { + outputs.set(id, row.output); + } + } + } + + return outputs; + }; +} + +export type ResolveArgs = ResolveCompletedWaitpointsArgs & { + /** Ids the caller resolved from Postgres rows. Read by the coverage check only. */ + resolvedElsewhere?: string[]; +}; + +/** + * Rebuild `CompletedWaitpoint[]` from one wait cycle's records. + * + * Field-for-field equivalent to `enhanceExecutionSnapshotWithWaitpoints`, which is what the + * executor already consumes. It iterates the RECORDS, not the order: the order holds only + * batch-indexed ids, so iterating it would silently drop every index-less wait. + * + * Returns the store-resident half only. A mixed snapshot's legacy half arrives as Postgres + * rows and is expanded by the existing path, and the caller concatenates. Both halves read + * their index from the same order, so the positions agree with no coordination. + */ +export function createCompletedWaitpointResolver(deps: CompletedWaitpointResolverDeps) { + return async function resolveCompletedWaitpoints( + args: ResolveArgs + ): Promise { + const recordIds = new Set(args.records.map((record) => record.id)); + const resolvedElsewhere = new Set(args.resolvedElsewhere ?? []); + + for (const id of resolvedElsewhere) { + if (recordIds.has(id)) { + throw new UnresolvableWaitpointId(id, "two-sources"); + } + } + + // Over the WHOLE membership, not `order`. The order omits every index-less wait, so a + // check scoped to it cannot see an index-less id whose record is missing — which is the + // exact loss this resolver exists to make impossible. + for (const id of new Set([...args.distinctIds, ...args.order])) { + if (!recordIds.has(id) && !resolvedElsewhere.has(id)) { + throw new UnresolvableWaitpointId(id, "no-source"); + } + } + + // Every deferred output in ONE read, before the emit loop. Reading inside the loop meant a + // round trip per record, in series, which is the shape a batch fan-in punishes hardest: the + // wide wait this feature exists to make cheap is exactly the wide wait that paid most. + const runOutputs = await readDeferredOutputs(args.records, deps); + + // Positions once, not once per record. `positionsOf` scanned the whole order for every + // record, so the emit loop was O(records x order) -- a million comparisons for a 1000-wide + // wait, growing with the same input as above. + const positions = positionsById(args.order); + + const out: CompletedWaitpoint[] = []; + + for (const record of args.records) { + const indexes = positions.get(record.id) ?? [undefined]; + // Hydrated once per record, not once per position, so a run at several batch indexes + // resolves from one map entry rather than one per index. + const output = hydrateOutput(record, runOutputs); + + for (const index of indexes) { + out.push({ + id: record.id, + index, + friendlyId: record.friendlyId, + type: record.type, + completedAt: new Date(record.completedAt), + ...(record.idempotencyKey && { idempotencyKey: record.idempotencyKey }), + ...(record.completedByTaskRunId && { + completedByTaskRun: { + id: record.completedByTaskRunId, + friendlyId: RunId.toFriendlyId(record.completedByTaskRunId), + // The reading entry's batch, never the entry that minted the cycle. + ...(args.batchId && { + batch: { id: args.batchId, friendlyId: BatchId.toFriendlyId(args.batchId) }, + }), + }, + }), + ...(record.completedAfter && { completedAfter: new Date(record.completedAfter) }), + ...(record.completedByBatchId && { + completedByBatch: { + id: record.completedByBatchId, + friendlyId: BatchId.toFriendlyId(record.completedByBatchId), + }, + }), + ...(output !== undefined && { output }), + outputType: record.outputType, + outputIsError: record.outputIsError, + }); + } + } + + return out; + }; +} + +// Every id's positions in the order, built in one pass. +// +// An id ABSENT from this map has no position, and the caller emits it once with an undefined +// index -- matching what the existing hydration does for a wait that carried no batch index. +// Absence is how that case is carried, so this never stores an [undefined] entry itself. +function positionsById(order: string[]): Map { + const positions = new Map(); + + for (let i = 0; i < order.length; i++) { + const id = order[i]; + if (id === undefined) { + continue; + } + + const existing = positions.get(id); + if (existing) { + existing.push(i); + } else { + positions.set(id, [i]); + } + } + + return positions; +} + +/** + * The output of every run a record defers to, in one batched read. + * + * Returns an empty map when no record defers, which is the common case: a cycle carrying only + * inline values, refs and BATCH records reads nothing at all. + */ +async function readDeferredOutputs( + records: CompletedWaitpointRecord[], + deps: CompletedWaitpointResolverDeps +): Promise> { + const runIds = new Set(); + let read: NonNullable | undefined; + + for (const record of records) { + const runId = deferredRunIdOf(record); + if (runId === undefined) { + continue; + } + + // Checked here rather than after the pass, so the message names a record that really does + // defer rather than whichever one happened to be first. A missing reader is a wiring fault, + // so every deferring record is equally implicated and the first is as informative as any. + if (!deps.readRunOutputs) { + throw new Error( + `Waitpoint ${record.id} defers its output to run ${runId}, but the resolver was built with no run-output reader.` + ); + } + + read = deps.readRunOutputs; + runIds.add(runId); + } + + // Set exactly when a record deferred, so this is the same condition as an empty `runIds` -- + // and unlike `runIds.size`, it carries the reader with it. + if (read === undefined) { + return new Map(); + } + + return read([...runIds]); +} + +// The run a record defers its output to, or undefined when it carries its own output or has +// nothing to defer to. This is the single definition of "needs a run read", so the pre-pass and +// the hydration cannot disagree about which records those are. +function deferredRunIdOf(record: CompletedWaitpointRecord): string | undefined { + if (record.output === null) { + return undefined; + } + + if ("inline" in record.output || "ref" in record.output) { + return undefined; + } + + return record.completedByTaskRunId ?? undefined; +} + +// Synchronous: every read this needs already happened in readDeferredOutputs. +function hydrateOutput( + record: CompletedWaitpointRecord, + runOutputs: Map +): string | undefined { + if (record.output === null) { + return undefined; + } + + if ("inline" in record.output) { + return record.output.inline; + } + + // A ref is handed back as the output verbatim: the executor already resolves an + // application/store output the same way it does for a Postgres-served snapshot. + if ("ref" in record.output) { + return record.output.ref; + } + + const runId = deferredRunIdOf(record); + if (runId === undefined) { + return undefined; + } + + // Postgres does not lose this: the back-reference nulls on delete but Waitpoint.output stays, + // so the legacy path still emits it. Returning undefined here instead would resolve the + // parent's triggerAndWait successfully with no output, which is silent wrong data. + const output = runOutputs.get(runId); + if (output === undefined) { + throw new UnresolvableWaitpointId(record.id, "lost-run-output"); + } + + return output; +} diff --git a/internal-packages/run-engine/src/engine/waitpointCoordinator/completedWaitpointRoundTrip.test.ts b/internal-packages/run-engine/src/engine/waitpointCoordinator/completedWaitpointRoundTrip.test.ts new file mode 100644 index 00000000000..98c62e966ca --- /dev/null +++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/completedWaitpointRoundTrip.test.ts @@ -0,0 +1,372 @@ +// The seam: a record set written into a real cycle key, read back out, and resolved. +// +// The two halves were covered separately and the join between them was not, so an envelope built +// in a shape the resolver does not expect survives both suites and fails only here. Envelopes come +// from real Postgres rows through the legacy arm; the oracle is the existing hydration over the +// same rows. +// +// The records field is read with a probe because the read API for it does not exist yet -- that +// hydration belongs to the snapshot-store lane. Everything either side of that read is production +// code. +import { createRedisClient } from "@internal/redis"; +import { PostgresRunStore, RedisSnapshotStore, type SnapshotEntryInput } from "@internal/run-store"; +import { Logger } from "@trigger.dev/core/logger"; +import { generateInternalId, generateWaitpointId } from "@trigger.dev/core/v3/isomorphic"; +import type { PrismaClient, Waitpoint } from "@trigger.dev/database"; +import { containerTest } from "@internal/testcontainers"; +import { describe, expect } from "vitest"; +import { enhanceExecutionSnapshotWithWaitpoints } from "../systems/executionSnapshotSystem.js"; +import { setupAuthenticatedEnvironment } from "../tests/setup.js"; +import { buildCompletedWaitpointRecords } from "./completedWaitpointRecords.js"; +import { + createCompletedWaitpointResolver, + createRunOutputsReader, + UnresolvableWaitpointId, +} from "./completedWaitpointResolver.js"; +import { LegacyPostgresWaitpointCoordinator } from "./legacyPostgresCoordinator.js"; + +const COMPLETED_TTL_MS = 72 * 60 * 60 * 1000; + +type Env = Awaited>; + +function entryFor(runId: string, env: Env): SnapshotEntryInput { + return { + id: generateInternalId(), + engine: "V2", + executionStatus: "EXECUTING_WITH_WAITPOINTS", + description: "Run resumed", + runId, + runStatus: "EXECUTING", + createdAt: new Date().toISOString(), + environmentId: env.id, + environmentType: env.type, + projectId: env.project.id, + organizationId: env.organization.id, + }; +} + +// `id` omitted gives Prisma's cuid default, i.e. the legacy format; a minted id gives the store +// format. Both halves of a mixed snapshot come from here so they cannot drift apart. +async function seedWaitpoint( + prisma: PrismaClient, + env: Env, + fields: { + id?: string; + output?: string | null; + outputType?: string; + completedByTaskRunId?: string; + } +): Promise { + const key = `idem_${generateInternalId().slice(-16)}`; + return prisma.waitpoint.create({ + data: { + ...(fields.id ? { id: fields.id } : {}), + friendlyId: `waitpoint_${key}`, + type: fields.completedByTaskRunId ? "RUN" : "MANUAL", + status: "COMPLETED", + completedAt: new Date(), + idempotencyKey: key, + userProvidedIdempotencyKey: false, + output: fields.output ?? null, + outputType: fields.outputType ?? "application/json", + ...(fields.completedByTaskRunId && { completedByTaskRunId: fields.completedByTaskRunId }), + projectId: env.project.id, + environmentId: env.id, + }, + }); +} + +async function seedCompletedRun(prisma: PrismaClient, env: Env, output: string): Promise { + const suffix = generateInternalId().slice(-12); + const run = await prisma.taskRun.create({ + data: { + engine: "V2", + status: "COMPLETED_SUCCESSFULLY", + friendlyId: `run_child${suffix}`, + runtimeEnvironmentId: env.id, + environmentType: env.type, + organizationId: env.organization.id, + projectId: env.project.id, + taskIdentifier: "child-task", + payload: "{}", + payloadType: "application/json", + traceContext: {}, + traceId: `trace_${suffix}`, + spanId: `span_${suffix}`, + queue: "task/child-task", + isTest: false, + taskEventStore: "taskEvent", + depth: 1, + output, + outputType: "application/json", + }, + select: { id: true }, + }); + return run.id; +} + +type Harness = { + runId: string; + store: RedisSnapshotStore; + probe: ReturnType; + coordinator: LegacyPostgresWaitpointCoordinator; + resolve: ReturnType; + runStore: PostgresRunStore; +}; + +function harness(prisma: PrismaClient, redisOptions: never): Harness { + const runStore = new PostgresRunStore({ prisma, readOnlyPrisma: prisma }); + return { + runId: generateInternalId(), + store: new RedisSnapshotStore({ redisOptions, completedTtlMs: COMPLETED_TTL_MS }), + probe: createRedisClient(redisOptions, { onError: () => {} }), + coordinator: new LegacyPostgresWaitpointCoordinator({ + runStore: runStore as never, + prisma, + logger: new Logger("roundtrip", "error"), + }), + resolve: createCompletedWaitpointResolver({ + readRunOutputs: createRunOutputsReader(runStore), + }), + runStore, + }; +} + +// Envelopes from rows, records into the cycle key, id lists back out through the store's read, +// then resolve. Returns the resolver's answer beside the oracle's for the same rows. +async function roundTrip( + h: Harness, + env: Env, + rows: Waitpoint[], + order: string[], + storeFormatIds: string[] +) { + const refs = rows.map((row) => { + const index = order.indexOf(row.id); + return index === -1 ? { id: row.id } : { id: row.id, index }; + }); + // Every position, not just the first: a run at two batch indexes must keep both. + const withRepeats = order.flatMap((id, index) => + refs.some((r) => r.id === id) ? [{ id, index }] : [] + ); + const completedWaitpoints = [ + ...withRepeats, + ...refs.filter((r) => r.index === undefined).map((r) => ({ id: r.id })), + ]; + + // Production envelope build, over real rows, through the arm the engine actually constructs. + const sources = await h.coordinator.readCompletionEnvelopes({ + runId: h.runId, + waitpointIds: storeFormatIds, + }); + const records = buildCompletedWaitpointRecords(sources); + + const appended = await h.store.append({ + entry: entryFor(h.runId, env), + kind: "birth", + isTerminal: false, + cycle: { kind: "new", completedWaitpoints, records }, + }); + expect(appended.outcome).toBe("written"); + + // Back out through the store's own read, not through the objects we just wrote. + const read = await h.store.getLatest(h.runId); + expect(read?.cycle).toBeDefined(); + expect(read?.danglingCycle).toBeFalsy(); + + const raw = await h.probe.hget(`snap:{${h.runId}}:wp:${read!.cycle!.cycleSeq}`, "records"); + const storedRecords = raw ? JSON.parse(raw) : []; + + const legacyIds = rows.map((r) => r.id).filter((id) => !storeFormatIds.includes(id)); + + const actual = await h.resolve({ + runId: h.runId, + pointer: read!.cycle!, + order: read!.completedWaitpointIds?.order ?? [], + distinctIds: read!.completedWaitpointIds?.distinctIds ?? [], + records: storedRecords, + ...(legacyIds.length > 0 && { resolvedElsewhere: legacyIds }), + }); + + const oracle = enhanceExecutionSnapshotWithWaitpoints( + { id: generateInternalId(), runId: h.runId, batchId: null } as never, + rows, + order + ).completedWaitpoints; + + const sort = (xs: T[]) => + [...xs].sort((a, b) => a.id.localeCompare(b.id) || (a.index ?? -1) - (b.index ?? -1)); + + return { + actual: sort(actual), + // The resolver returns its own half only; the legacy half stays with the caller. + expected: sort(oracle.filter((w) => storeFormatIds.includes(w.id))), + storedRecords, + read, + }; +} + +describe("a record set round-trips through the cycle key", () => { + containerTest("for an inline MANUAL output", async ({ prisma, redisOptions }) => { + const h = harness(prisma as never, redisOptions as never); + try { + const env = await setupAuthenticatedEnvironment(prisma as never, "PRODUCTION"); + const row = await seedWaitpoint(prisma as never, env, { + id: generateWaitpointId("MANUAL"), + output: '{"token":1}', + }); + + const { actual, expected, storedRecords } = await roundTrip( + h, + env, + [row], + [row.id], + [row.id] + ); + + expect(storedRecords).toHaveLength(1); + expect(actual).toEqual(expected); + expect(actual[0]?.output).toBe('{"token":1}'); + } finally { + await Promise.all([h.store.quit(), h.probe.quit().catch(() => {})]); + } + }); + + // deriveFromRun is the one place the written record and the resolved entry legitimately + // differ, so it has to survive the real round trip. + containerTest("for a RUN output derived from the run row", async ({ prisma, redisOptions }) => { + const h = harness(prisma as never, redisOptions as never); + try { + const env = await setupAuthenticatedEnvironment(prisma as never, "PRODUCTION"); + const childRunId = await seedCompletedRun(prisma as never, env, '{"child":true}'); + const row = await seedWaitpoint(prisma as never, env, { + id: generateWaitpointId("RUN"), + output: '{"child":true}', + completedByTaskRunId: childRunId, + }); + + const { actual, expected, storedRecords } = await roundTrip( + h, + env, + [row], + [row.id], + [row.id] + ); + + // The record carries a marker, not the value. + expect(storedRecords[0]?.output).toEqual({ deriveFromRun: true }); + // And the resolved entry carries the value, byte-identical to the oracle's. + expect(actual).toEqual(expected); + expect(actual[0]?.output).toBe('{"child":true}'); + } finally { + await Promise.all([h.store.quit(), h.probe.quit().catch(() => {})]); + } + }); + + // Both halves read their index from the same order, so the positions agree with no + // coordination between them. + containerTest( + "for a mixed legacy and store-format snapshot", + async ({ prisma, redisOptions }) => { + const h = harness(prisma as never, redisOptions as never); + try { + const env = await setupAuthenticatedEnvironment(prisma as never, "PRODUCTION"); + const storeRow = await seedWaitpoint(prisma as never, env, { + id: generateWaitpointId("MANUAL"), + output: '{"store":true}', + }); + // No id: Prisma's cuid default, which is the legacy format. + const legacyRow = await seedWaitpoint(prisma as never, env, { output: '{"legacy":true}' }); + + const order = [legacyRow.id, storeRow.id]; + const { actual, expected, read } = await roundTrip(h, env, [storeRow, legacyRow], order, [ + storeRow.id, + ]); + + // The store read carries BOTH ids, because the cycle records the whole membership... + expect(read?.completedWaitpointIds?.order).toEqual(order); + // ...but the resolver answers for its half only, at its real position. + expect(actual).toEqual(expected); + expect(actual).toHaveLength(1); + expect(actual[0]?.id).toBe(storeRow.id); + expect(actual[0]?.index).toBe(1); + } finally { + await Promise.all([h.store.quit(), h.probe.quit().catch(() => {})]); + } + } + ); + + // Repeats live in the order, never in the record set, so this pins that the round trip does + // not collapse them. + containerTest("for one waitpoint at two batch indexes", async ({ prisma, redisOptions }) => { + const h = harness(prisma as never, redisOptions as never); + try { + const env = await setupAuthenticatedEnvironment(prisma as never, "PRODUCTION"); + const row = await seedWaitpoint(prisma as never, env, { + id: generateWaitpointId("RUN"), + output: '{"twice":true}', + }); + + const { actual, expected, storedRecords } = await roundTrip( + h, + env, + [row], + [row.id, row.id], + [row.id] + ); + + expect(storedRecords).toHaveLength(1); + expect(actual).toEqual(expected); + expect(actual.map((w) => w.index)).toEqual([0, 1]); + } finally { + await Promise.all([h.store.quit(), h.probe.quit().catch(() => {})]); + } + }); + + // Losing the records while the id list survives is the shape an eviction leaves behind. + containerTest("and refuses when a member id has no record", async ({ prisma, redisOptions }) => { + const h = harness(prisma as never, redisOptions as never); + try { + const env = await setupAuthenticatedEnvironment(prisma as never, "PRODUCTION"); + const row = await seedWaitpoint(prisma as never, env, { + id: generateWaitpointId("MANUAL"), + output: '{"lost":true}', + }); + + const sources = await h.coordinator.readCompletionEnvelopes({ + runId: h.runId, + waitpointIds: [row.id], + }); + await h.store.append({ + entry: entryFor(h.runId, env), + kind: "birth", + isTerminal: false, + cycle: { + kind: "new", + completedWaitpoints: [{ id: row.id, index: 0 }], + records: buildCompletedWaitpointRecords(sources), + }, + }); + + const read = await h.store.getLatest(h.runId); + // The records field alone is lost; the id list survives. + await h.probe.hdel(`snap:{${h.runId}}:wp:${read!.cycle!.cycleSeq}`, "records"); + + const failure = await h + .resolve({ + runId: h.runId, + pointer: read!.cycle!, + order: read!.completedWaitpointIds?.order ?? [], + distinctIds: read!.completedWaitpointIds?.distinctIds ?? [], + records: [], + }) + .catch((caught: unknown) => caught as UnresolvableWaitpointId); + + expect(failure).toBeInstanceOf(UnresolvableWaitpointId); + expect(failure.waitpointId).toBe(row.id); + expect(failure.reason).toBe("no-source"); + } finally { + await Promise.all([h.store.quit(), h.probe.quit().catch(() => {})]); + } + }); +}); diff --git a/internal-packages/run-engine/src/engine/waitpointCoordinator/completionEnvelopeSource.test.ts b/internal-packages/run-engine/src/engine/waitpointCoordinator/completionEnvelopeSource.test.ts new file mode 100644 index 00000000000..bbdca66b370 --- /dev/null +++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/completionEnvelopeSource.test.ts @@ -0,0 +1,138 @@ +// The row-to-source mapping the legacy arm depends on. It had no direct coverage: the +// equivalence suite reaches the same code through its `pair()` factory, which proves the +// mapping is self-consistent with the record build but never states what the mapping IS. +import type { Waitpoint } from "@trigger.dev/database"; +import { describe, expect, it } from "vitest"; +import { envelopeSourceFromWaitpointRow } from "./completionEnvelopeSource.js"; + +const COMPLETED_AT = new Date("2026-08-25T00:00:00.000Z"); + +function row(overrides: Partial = {}): Waitpoint { + return { + id: "wp_1", + friendlyId: "waitpoint_wp_1", + type: "MANUAL", + status: "COMPLETED", + completedAt: COMPLETED_AT, + output: null, + outputType: "application/json", + outputIsError: false, + completedByTaskRunId: null, + completedByBatchId: null, + completedAfter: null, + idempotencyKey: "internal", + userProvidedIdempotencyKey: false, + inactiveIdempotencyKey: null, + ...overrides, + } as unknown as Waitpoint; +} + +describe("envelopeSourceFromWaitpointRow", () => { + it("carries the scalar fields through", () => { + expect(envelopeSourceFromWaitpointRow(row())).toMatchObject({ + id: "wp_1", + friendlyId: "waitpoint_wp_1", + type: "MANUAL", + completedAt: COMPLETED_AT, + outputType: "application/json", + outputIsError: false, + }); + }); + + it("treats a plain output as an inline value", () => { + const source = envelopeSourceFromWaitpointRow(row({ output: '{"ok":true}' })); + + expect(source.output).toBe('{"ok":true}'); + expect(source.outputRef).toBeUndefined(); + }); + + // The type names it, not the shape. A store reference is an opaque string like any other, so + // reading the string alone cannot tell the two apart. + it("treats an application/store output as a reference", () => { + const source = envelopeSourceFromWaitpointRow( + row({ output: "store-key-1", outputType: "application/store" }) + ); + + expect(source.outputRef).toBe("store-key-1"); + expect(source.output).toBeUndefined(); + }); + + it("keeps an empty-string output, because empty is a value", () => { + expect(envelopeSourceFromWaitpointRow(row({ output: "" })).output).toBe(""); + }); + + it("omits an absent output entirely", () => { + const source = envelopeSourceFromWaitpointRow(row()); + + expect("output" in source).toBe(false); + expect("outputRef" in source).toBe(false); + }); + + describe("the idempotency key", () => { + it("is carried when the user provided it and it is still active", () => { + const source = envelopeSourceFromWaitpointRow( + row({ idempotencyKey: "user-key", userProvidedIdempotencyKey: true }) + ); + + expect(source.idempotencyKey).toBe("user-key"); + }); + + it("is suppressed when the user did not provide it", () => { + const source = envelopeSourceFromWaitpointRow( + row({ idempotencyKey: "internal-key", userProvidedIdempotencyKey: false }) + ); + + expect(source.idempotencyKey).toBeUndefined(); + }); + + it("is suppressed once it goes inactive", () => { + const source = envelopeSourceFromWaitpointRow( + row({ + idempotencyKey: "user-key", + userProvidedIdempotencyKey: true, + inactiveIdempotencyKey: "rotated", + }) + ); + + expect(source.idempotencyKey).toBeUndefined(); + }); + }); + + // The mapper itself is status-blind by design: the legacy arm filters to COMPLETED before + // calling it, so both arms omit a pending waitpoint rather than describing one. This states + // that the mapper is not where that decision lives. + it("does not itself inspect status", () => { + const source = envelopeSourceFromWaitpointRow(row({ status: "PENDING", completedAt: null })); + + expect(source.id).toBe("wp_1"); + expect(source.completedAt).toBeInstanceOf(Date); + }); + + it("carries the RUN and BATCH back-references", () => { + expect( + envelopeSourceFromWaitpointRow(row({ type: "RUN", completedByTaskRunId: "run_child" })) + .completedByTaskRunId + ).toBe("run_child"); + + expect( + envelopeSourceFromWaitpointRow(row({ type: "BATCH", completedByBatchId: "batch_1" })) + .completedByBatchId + ).toBe("batch_1"); + }); + + it("carries completedAfter", () => { + const completedAfter = new Date("2026-08-26T00:00:00.000Z"); + + expect( + envelopeSourceFromWaitpointRow(row({ type: "DATETIME", completedAfter })).completedAfter + ).toEqual(completedAfter); + }); + + // A row read at COMPLETED always has this set. The fallback exists so the shape stays total + // rather than emitting an invalid Date, matching what the snapshot hydration does. + it("falls back to a real date when completedAt is null", () => { + expect(envelopeSourceFromWaitpointRow(row({ completedAt: null })).completedAt).toBeInstanceOf( + Date + ); + }); +}); diff --git a/internal-packages/run-engine/src/engine/waitpointCoordinator/completionEnvelopeSource.ts b/internal-packages/run-engine/src/engine/waitpointCoordinator/completionEnvelopeSource.ts new file mode 100644 index 00000000000..2d1977891a9 --- /dev/null +++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/completionEnvelopeSource.ts @@ -0,0 +1,53 @@ +import type { Waitpoint } from "@trigger.dev/database"; +import type { CompletionEnvelopeSource } from "./types.js"; + +/** + * Map a waitpoint row onto the arm-independent envelope source. + * + * Shared so the legacy arm and the equivalence suite cannot drift: if the suite hand-rolled its + * own copy, a bug in the arm would be invisible to every test that compares against the oracle. + */ +export function envelopeSourceFromWaitpointRow( + row: Pick< + Waitpoint, + | "id" + | "friendlyId" + | "type" + | "completedAt" + | "output" + | "outputType" + | "outputIsError" + | "completedByTaskRunId" + | "completedByBatchId" + | "completedAfter" + | "idempotencyKey" + | "userProvidedIdempotencyKey" + | "inactiveIdempotencyKey" + > +): CompletionEnvelopeSource { + // An already-offloaded value is named by its type, not by its shape, so the type is what + // decides whether the string is a payload or a reference to one. + const isRef = row.outputType === "application/store"; + + return { + id: row.id, + friendlyId: row.friendlyId, + type: row.type, + // A completed waitpoint always has this. The fallback keeps the shape total rather than + // emitting an invalid Date, and mirrors the fallback the snapshot hydration already applies. + completedAt: row.completedAt ?? new Date(), + outputType: row.outputType, + outputIsError: row.outputIsError, + ...(row.output !== null && row.output !== undefined + ? isRef + ? { outputRef: row.output } + : { output: row.output } + : {}), + ...(row.completedByTaskRunId && { completedByTaskRunId: row.completedByTaskRunId }), + ...(row.completedByBatchId && { completedByBatchId: row.completedByBatchId }), + ...(row.completedAfter && { completedAfter: row.completedAfter }), + ...(row.userProvidedIdempotencyKey && !row.inactiveIdempotencyKey && row.idempotencyKey + ? { idempotencyKey: row.idempotencyKey } + : {}), + }; +} diff --git a/internal-packages/run-engine/src/engine/waitpointCoordinator/legacyPostgresCoordinator.ts b/internal-packages/run-engine/src/engine/waitpointCoordinator/legacyPostgresCoordinator.ts index def58f7bc37..009df46a2f2 100644 --- a/internal-packages/run-engine/src/engine/waitpointCoordinator/legacyPostgresCoordinator.ts +++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/legacyPostgresCoordinator.ts @@ -6,6 +6,8 @@ import type { PrismaClient, Waitpoint } from "@trigger.dev/database"; import { boundedIn, Prisma } from "@trigger.dev/database"; import { nanoid } from "nanoid"; import { UnclassifiableWaitpointId } from "../errors.js"; +import { fetchWaitpointEnvelopeRowsInChunks } from "../systems/executionSnapshotSystem.js"; +import { envelopeSourceFromWaitpointRow } from "./completionEnvelopeSource.js"; import type { AssociatedWaitpointData, ClearRunBlockStateParams, @@ -16,6 +18,8 @@ import type { CreateWaitpointResult, RegisterBlocksLocklessParams, RegisterBlocksParams, + CompletionEnvelopeSource, + ReadCompletionEnvelopesParams, RunBlockEdge, WaitpointCoordinator, } from "./types.js"; @@ -82,6 +86,44 @@ export class LegacyPostgresWaitpointCoordinator implements WaitpointCoordinator ); } + /** + * Source the envelope fields from the waitpoint rows. + * + * `runId` is the routing hint, not decoration: the routing store takes it as the third + * argument and reads the run's own store, falling back only for the rare cross-tree token. + * Omitting it fans every read out across every run-ops database, once per resume. + * + * Chunked for the same reason the snapshot hydration chunks: a waitpoint output can be + * 100KB+, and a large fan-in read whole can exceed Node's string limits. `boundedIn` pads + * for plan-cache stability, it does not bound the set. + * + * Projected to the columns the envelope is built from. This read lands on the writer, inside + * the run lock, and the hydration reads the same rows again afterwards for a store-format + * resume that Postgres still serves -- so it takes no more than it uses. + */ + async readCompletionEnvelopes({ + runId, + waitpointIds, + }: ReadCompletionEnvelopesParams): Promise { + if (waitpointIds.length === 0) { + return []; + } + + const rows = await fetchWaitpointEnvelopeRowsInChunks( + this.prisma, + waitpointIds, + this.runStore, + runId + ); + + // COMPLETED only, so both arms honour one omission contract. The store arm cannot return a + // pending waitpoint because a pending one has no completion to read; this arm reads rows by + // id and would otherwise hand back an envelope with completedAt defaulted to now. The + // resolver's coverage check reads an omission as "fail loud", so the two arms disagreeing + // here would turn a pending waitpoint into a resumable one. + return rows.filter((row) => row.status === "COMPLETED").map(envelopeSourceFromWaitpointRow); + } + async registerBlocks({ client, ...edge diff --git a/internal-packages/run-engine/src/engine/waitpointCoordinator/storeCoordinator.test.ts b/internal-packages/run-engine/src/engine/waitpointCoordinator/storeCoordinator.test.ts index f0e9c0c297d..34439fcfa02 100644 --- a/internal-packages/run-engine/src/engine/waitpointCoordinator/storeCoordinator.test.ts +++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/storeCoordinator.test.ts @@ -1835,3 +1835,191 @@ describe("genuine concurrency", () => { } ); }); + +describe("readCompletionEnvelopes", () => { + redisTest("returns the completion and the immutable half together", async ({ redisOptions }) => { + const store = coordinator(redisOptions); + try { + await store.createIfAbsent({ + record: record("w_env", { + idempotencyKey: "user-key", + userProvidedIdempotencyKey: true, + }), + status: "PENDING", + }); + await store.complete({ waitpointId: "w_env", completion: completion() }); + + const envelopes = await store.readCompletionEnvelopes({ + runId: "run_env", + waitpointIds: ["w_env"], + }); + + expect(envelopes).toEqual([ + { + id: "w_env", + friendlyId: "waitpoint_w_env", + type: "MANUAL", + completedAt: new Date(NOW), + outputType: "application/json", + outputIsError: false, + output: '{"ok":true}', + idempotencyKey: "user-key", + }, + ]); + } finally { + await store.quit(); + } + }); + + redisTest( + "carries an offloaded value as a ref, not as an inline value", + async ({ redisOptions }) => { + const store = coordinator(redisOptions); + try { + await store.createIfAbsent({ record: record("w_ref"), status: "PENDING" }); + await store.complete({ + waitpointId: "w_ref", + completion: completion({ + outputType: "application/store", + output: { ref: "store-key-1" }, + }), + }); + + const [envelope] = await store.readCompletionEnvelopes({ + runId: "run_env", + waitpointIds: ["w_ref"], + }); + + expect(envelope?.outputRef).toBe("store-key-1"); + expect(envelope?.output).toBeUndefined(); + } finally { + await store.quit(); + } + } + ); + + // The omission is the contract. A pending waitpoint has no envelope, and defaulting one + // here would hand the resolver a record it must not have. The caller's coverage check is + // what turns the gap into a loud failure. + redisTest("omits a waitpoint that is not completed", async ({ redisOptions }) => { + const store = coordinator(redisOptions); + try { + await store.createIfAbsent({ record: record("w_pending"), status: "PENDING" }); + + const envelopes = await store.readCompletionEnvelopes({ + runId: "run_env", + waitpointIds: ["w_pending"], + }); + + expect(envelopes).toEqual([]); + } finally { + await store.quit(); + } + }); + + redisTest("omits an id that has no record at all", async ({ redisOptions }) => { + const store = coordinator(redisOptions); + try { + const envelopes = await store.readCompletionEnvelopes({ + runId: "run_env", + waitpointIds: ["w_absent"], + }); + + expect(envelopes).toEqual([]); + } finally { + await store.quit(); + } + }); + + redisTest("suppresses an idempotency key the user did not provide", async ({ redisOptions }) => { + const store = coordinator(redisOptions); + try { + await store.createIfAbsent({ + record: record("w_internal", { + idempotencyKey: "internal-key", + userProvidedIdempotencyKey: false, + }), + status: "PENDING", + }); + await store.complete({ waitpointId: "w_internal", completion: completion() }); + + const [envelope] = await store.readCompletionEnvelopes({ + runId: "run_env", + waitpointIds: ["w_internal"], + }); + + expect(envelope?.idempotencyKey).toBeUndefined(); + } finally { + await store.quit(); + } + }); + + redisTest("reads many ids in one pass", async ({ redisOptions }) => { + const store = coordinator(redisOptions); + try { + for (const id of ["w_m1", "w_m2", "w_m3"]) { + await store.createIfAbsent({ record: record(id), status: "PENDING" }); + await store.complete({ waitpointId: id, completion: completion() }); + } + + const envelopes = await store.readCompletionEnvelopes({ + runId: "run_env", + waitpointIds: ["w_m1", "w_m2", "w_m3"], + }); + + expect(envelopes.map((e) => e.id).sort()).toEqual(["w_m1", "w_m2", "w_m3"]); + } finally { + await store.quit(); + } + }); +}); + +// The read is chunked so a large fan-in cannot burst. This asserts correctness ACROSS the chunk +// boundary, which is where an off-by-one in the slice would show: a count either side of 100. +describe("readCompletionEnvelopes chunking", () => { + redisTest("returns every envelope across a chunk boundary", async ({ redisOptions }) => { + const store = coordinator(redisOptions); + try { + const ids = Array.from({ length: 250 }, (_, i) => `w_chunk_${i}`); + for (const id of ids) { + await store.createIfAbsent({ record: record(id), status: "PENDING" }); + await store.complete({ waitpointId: id, completion: completion() }); + } + + const envelopes = await store.readCompletionEnvelopes({ + runId: "run_chunked", + waitpointIds: ids, + }); + + expect(envelopes).toHaveLength(250); + expect(new Set(envelopes.map((e) => e.id)).size).toBe(250); + } finally { + await store.quit(); + } + }); + + redisTest("omits only the incomplete ones in a mixed chunked read", async ({ redisOptions }) => { + const store = coordinator(redisOptions); + try { + const ids = Array.from({ length: 150 }, (_, i) => `w_mix_${i}`); + for (const [i, id] of ids.entries()) { + await store.createIfAbsent({ record: record(id), status: "PENDING" }); + // Leave every third pending, including ids either side of the chunk boundary. + if (i % 3 !== 0) { + await store.complete({ waitpointId: id, completion: completion() }); + } + } + + const envelopes = await store.readCompletionEnvelopes({ + runId: "run_chunked", + waitpointIds: ids, + }); + + const expected = ids.filter((_, i) => i % 3 !== 0).length; + expect(envelopes).toHaveLength(expected); + expect(envelopes.every((e) => e.id.startsWith("w_mix_"))).toBe(true); + } finally { + await store.quit(); + } + }); +}); diff --git a/internal-packages/run-engine/src/engine/waitpointCoordinator/storeCoordinator.ts b/internal-packages/run-engine/src/engine/waitpointCoordinator/storeCoordinator.ts index 723552c57ab..cda1293355d 100644 --- a/internal-packages/run-engine/src/engine/waitpointCoordinator/storeCoordinator.ts +++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/storeCoordinator.ts @@ -10,6 +10,7 @@ import { watcherField, } from "./keys.js"; import { registerWaitpointCommands } from "./scripts.js"; +import type { CompletionEnvelopeSource, ReadCompletionEnvelopesParams } from "./types.js"; /** The values written into a record's `status` field. Uppercase, and never a token. */ export type WaitpointStatus = "PENDING" | "COMPLETED"; @@ -162,6 +163,12 @@ function parseJson(raw: string | undefined): T | undefined { return raw ? (JSON.parse(raw) as T) : undefined; } +/** + * How many envelope reads run concurrently. Matches the chunk the snapshot hydration uses for the + * same shape of read, so a large fan-in bounds Redis client pressure instead of bursting. + */ +const ENVELOPE_READ_CHUNK_SIZE = 100; + export class WaitpointStoreCoordinator { private readonly redis: Redis; private readonly logger: Logger; @@ -505,6 +512,58 @@ export class WaitpointStoreCoordinator { return { pendingIds, deliveredIds, edges }; } + /** + * Source the envelope fields for a run's COMPLETED waitpoints. + * + * Reads `wp:{id}` and nothing else. Both halves live under that one key — `r` holds the + * immutable record, `c` holds the completion — so this needs no run-scoped key. + * + * One command per id, issued concurrently rather than as a pipeline. Each id is its own hash + * tag, so N ids are N cluster slots: a pipeline spanning them is rejected outright under + * cluster mode, and a single-node test server would never surface that. + * + * An id with no record, or a record with no completion, is OMITTED rather than defaulted. + * The omission is the contract: the caller's coverage check turns a gap into a loud failure, + * which a defaulted envelope would hide. + */ + async readCompletionEnvelopes({ + waitpointIds, + }: ReadCompletionEnvelopesParams): Promise { + if (waitpointIds.length === 0) { + return []; + } + + // Chunked, not one Promise.all over the whole set. A 1000-item batch fan-in would otherwise + // launch 1000 concurrent HMGETs at once, and the burst is the cost even though each command + // is small. The bound mirrors the snapshot hydration's own WAITPOINT_CHUNK_SIZE, and stays + // per-command so nothing can span two cluster slots. + const halves: (string | null)[][] = []; + for (let i = 0; i < waitpointIds.length; i += ENVELOPE_READ_CHUNK_SIZE) { + const chunk = waitpointIds.slice(i, i + ENVELOPE_READ_CHUNK_SIZE); + const settled = await Promise.all( + chunk.map((id) => this.redis.hmget(waitpointKeys(id).record, "r", "c")) + ); + halves.push(...settled); + } + + const out: CompletionEnvelopeSource[] = []; + + for (let i = 0; i < waitpointIds.length; i++) { + const id = waitpointIds[i]!; + const fields = halves[i]; + const record = parseJson(fields?.[0] ?? undefined); + const completion = parseJson(fields?.[1] ?? undefined); + + if (!record || !completion) { + continue; + } + + out.push(toEnvelopeSource(id, record, completion)); + } + + return out; + } + /** * Drain one cycle's edges, or clear the run entirely when no edge ids are given. * @@ -536,3 +595,37 @@ export class WaitpointStoreCoordinator { return { outcome: reply[0] as "cleared" | "drained" }; } } + +/** + * Map the store's two halves onto the arm-independent source shape. + * + * The idempotency key is suppressed unless the user provided it, matching the rule the + * snapshot hydration applies today. The store never sets an inactive flag, so + * `userProvidedIdempotencyKey` alone decides it here. + */ +function toEnvelopeSource( + id: string, + record: WaitpointRecordInput, + completion: WaitpointCompletion +): CompletionEnvelopeSource { + const output = completion.output; + const inline = output && "inline" in output ? output.inline : undefined; + const ref = output && "ref" in output ? output.ref : undefined; + + return { + id, + friendlyId: record.friendlyId, + type: record.type, + completedAt: new Date(completion.completedAt), + outputType: completion.outputType, + outputIsError: completion.outputIsError, + ...(inline !== undefined && { output: inline }), + ...(ref !== undefined && { outputRef: ref }), + ...(record.completedByTaskRunId && { completedByTaskRunId: record.completedByTaskRunId }), + ...(record.completedByBatchId && { completedByBatchId: record.completedByBatchId }), + ...(record.completedAfter && { completedAfter: new Date(record.completedAfter) }), + ...(record.userProvidedIdempotencyKey && record.idempotencyKey + ? { idempotencyKey: record.idempotencyKey } + : {}), + }; +} diff --git a/internal-packages/run-engine/src/engine/waitpointCoordinator/testFixtures/childRun.ts b/internal-packages/run-engine/src/engine/waitpointCoordinator/testFixtures/childRun.ts new file mode 100644 index 00000000000..a5974f72ff0 --- /dev/null +++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/testFixtures/childRun.ts @@ -0,0 +1,65 @@ +import type { PrismaClient } from "@trigger.dev/database"; +import { setupAuthenticatedEnvironment } from "../../tests/setup.js"; + +/** + * Completed child runs holding `output`, for the deriveFromRun branch. + * + * The branch's premise is that TaskRun.output holds the same string the waitpoint carried, so a + * test that asserts it needs real rows rather than stand-ins for them. + * + * All the runs share ONE environment, because `setupAuthenticatedEnvironment` hardcodes the + * organization slug and the column is unique: calling it once per run fails on the second. + */ +export async function seedChildRunsWithOutputs( + prisma: PrismaClient, + outputs: (string | null)[], + outputType = "application/json" +): Promise { + const env = await setupAuthenticatedEnvironment(prisma, "PRODUCTION"); + const envSuffix = env.id.slice(-10); + const ids: string[] = []; + + for (const [i, output] of outputs.entries()) { + // Unique per run, not per environment: friendlyId is a unique column and every run here + // shares the one environment. + const suffix = `${envSuffix}${i}`; + + const run = await prisma.taskRun.create({ + data: { + engine: "V2", + status: "COMPLETED_SUCCESSFULLY", + friendlyId: `run_child${suffix}`, + runtimeEnvironmentId: env.id, + environmentType: env.type, + organizationId: env.organization.id, + projectId: env.project.id, + taskIdentifier: "child-task", + payload: "{}", + payloadType: "application/json", + traceContext: {}, + traceId: `trace_${suffix}`, + spanId: `span_${suffix}`, + queue: "task/child-task", + isTest: false, + taskEventStore: "taskEvent", + depth: 1, + ...(output !== null && { output, outputType }), + }, + select: { id: true }, + }); + + ids.push(run.id); + } + + return ids; +} + +/** One child run, for the cases that only need one. */ +export async function seedChildRunWithOutput( + prisma: PrismaClient, + output: string | null, + outputType = "application/json" +): Promise { + const [id] = await seedChildRunsWithOutputs(prisma, [output], outputType); + return id!; +} diff --git a/internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts b/internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts index 412e1ed97ec..026e7deb8b5 100644 --- a/internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts +++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts @@ -17,6 +17,9 @@ import type { ShardKey } from "@trigger.dev/core/v3/isomorphic"; export type WaitpointCoordinator = { clearRunBlockState(params: ClearRunBlockStateParams): Promise<{ count: number }>; readRunBlockState(runId: string): Promise; + readCompletionEnvelopes( + params: ReadCompletionEnvelopesParams + ): Promise; registerBlocks(params: RegisterBlocksParams): Promise<{ pendingCount: number }>; registerBlocksLockless(params: RegisterBlocksLocklessParams): Promise; complete(params: CompleteParams): Promise; @@ -34,6 +37,38 @@ export type WaitpointCoordinator = { }): Promise; }; +export type ReadCompletionEnvelopesParams = { + runId: string; + /** The DISTINCT completed waitpoint ids to source. Result order is not meaningful. */ + waitpointIds: string[]; +}; + +/** + * One completed waitpoint's fields, sourced from whichever arm owns it. + * + * Deliberately NOT the frozen record type. This is the raw material; the record build + * decides which output variant a record carries. Both arms return this same shape, so the + * record build never branches on residency, which is what makes a mixed wait work. + * + * `output` is the literal stored value. `outputRef` is set instead when the value was + * already offloaded to object storage. At most one of the two is set. + */ +export type CompletionEnvelopeSource = { + id: string; + friendlyId: string; + type: "RUN" | "BATCH" | "DATETIME" | "MANUAL"; + completedAt: Date; + outputType: string; + outputIsError: boolean; + output?: string; + outputRef?: string; + completedByTaskRunId?: string; + completedByBatchId?: string; + completedAfter?: Date; + /** Already resolved by the arm: userProvidedIdempotencyKey && !inactiveIdempotencyKey. */ + idempotencyKey?: string; +}; + export type ClearRunBlockStateParams = { runId: string; /** Edge ids to delete. Omit to clear every edge for the run. */ diff --git a/internal-packages/run-store/src/redisSnapshotStore.recordsOnCarryRefusal.test.ts b/internal-packages/run-store/src/redisSnapshotStore.recordsOnCarryRefusal.test.ts new file mode 100644 index 00000000000..ac8abac91e5 --- /dev/null +++ b/internal-packages/run-store/src/redisSnapshotStore.recordsOnCarryRefusal.test.ts @@ -0,0 +1,230 @@ +// A refused carry-forward mints a replacement cycle inside the same call. That replacement must +// carry the records, not just the ids: the resolver's coverage check requires every distinct id to +// resolve through exactly one half, so a cycle holding ids with no records makes a legitimate +// resume fail loud. +import { createRedisClient } from "@internal/redis"; +import { redisTest } from "@internal/testcontainers"; +import { describe, expect } from "vitest"; +import { + RedisSnapshotStore, + type CompletedWaitpointRecord, + type SnapshotEntryInput, +} from "./redisSnapshotStore.js"; + +function entry(over: Partial = {}): SnapshotEntryInput { + return { + id: "snap_1", + engine: "V2", + executionStatus: "RUN_CREATED", + description: "created", + runId: "run_1", + runStatus: "PENDING", + createdAt: "2026-08-21T00:00:00.000Z", + environmentId: "env_1", + environmentType: "PRODUCTION", + projectId: "proj_1", + organizationId: "org_1", + ...over, + }; +} + +function record(id: string, output: string): CompletedWaitpointRecord { + return { + id, + friendlyId: `waitpoint_${id}`, + type: "MANUAL", + completedAt: "2026-08-25T00:00:00.000Z", + outputType: "application/json", + outputIsError: false, + output: { inline: output }, + }; +} + +async function recordsAt( + raw: ReturnType, + cycleSeq: number +): Promise { + const stored = await raw.hget(`snap:{run_1}:wp:${cycleSeq}`, "records"); + return stored ? (JSON.parse(stored) as CompletedWaitpointRecord[]) : undefined; +} + +describe("a refused carry-forward", () => { + redisTest("mints a replacement that carries the records", async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 60_000 }); + const raw = createRedisClient(redisOptions, { onError: () => {} }); + try { + await store.append({ + entry: entry({ id: "snap_1" }), + kind: "birth", + isTerminal: false, + cycle: { + kind: "new", + completedWaitpoints: [{ id: "w_a", index: 0 }], + records: [record("w_a", "first")], + }, + }); + + // Lose everything except the cycle key, as under maxmemory eviction. The carried pointer is + // now untrustworthy, so the store refuses it. + await raw.del("snap:{run_1}:e", "snap:{run_1}:idx", "snap:{run_1}:cur", "snap:{run_1}:seq"); + + const carried = await store.append({ + entry: entry({ id: "snap_2" }), + kind: "birth", + isTerminal: false, + cycle: { + kind: "carryForward", + cycleSeq: 1, + completedWaitpoints: [{ id: "w_b", index: 0 }], + records: [record("w_b", "second")], + }, + }); + + expect(carried).toMatchObject({ outcome: "written", cycleMismatch: true }); + + // The replacement holds the CARRIED records, not the dead incarnation's. + const read = await store.getLatest("run_1"); + const mintedSeq = read?.cycle?.cycleSeq; + expect(mintedSeq).toBeDefined(); + + const records = await recordsAt(raw, mintedSeq!); + expect(records).toHaveLength(1); + expect(records?.[0]?.id).toBe("w_b"); + expect(records?.[0]?.output).toEqual({ inline: "second" }); + } finally { + await Promise.all([store.quit(), raw.quit().catch(() => {})]); + } + }); + + // The reachable production shape. Every copy-forward append (dequeue, checkpoint, attempt) + // re-passes the same refs and carries NO records of its own, and no longer pre-reads them: the + // append script sources the record set from the cycle it is replacing. So a refusal preserves + // the records without the caller having paid a read on every copy-forward that did not refuse. + redisTest("keeps the records when the caller carried refs but none", async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 60_000 }); + const raw = createRedisClient(redisOptions, { onError: () => {} }); + try { + await store.append({ + entry: entry({ id: "snap_1" }), + kind: "birth", + isTerminal: false, + cycle: { + kind: "new", + completedWaitpoints: [{ id: "w_a", index: 0 }], + records: [record("w_a", "first")], + }, + }); + + // Lose the four core keys, keeping the cycle key. This is the shape that makes the store + // refuse the pointer: the seq counter is behind cycleSeqIn while wp:1 still lives. + await raw.del("snap:{run_1}:e", "snap:{run_1}:idx", "snap:{run_1}:cur", "snap:{run_1}:seq"); + + const carried = await store.append({ + entry: entry({ id: "snap_2" }), + kind: "birth", + isTerminal: false, + // No `records`, exactly as a copy-forward append passes it. + cycle: { + kind: "carryForward", + cycleSeq: 1, + completedWaitpoints: [{ id: "w_a", index: 0 }], + }, + }); + + expect(carried).toMatchObject({ outcome: "written", cycleMismatch: true }); + + // The replacement holds the records the refused cycle held, copied inside the script. + const read = await store.getLatest("run_1"); + const records = await recordsAt(raw, read!.cycle!.cycleSeq); + + expect(records).toHaveLength(1); + expect(records?.[0]?.id).toBe("w_a"); + expect(records?.[0]?.output).toEqual({ inline: "first" }); + } finally { + await Promise.all([store.quit(), raw.quit().catch(() => {})]); + } + }); + + // The other half of the refusal: the cycle key itself is gone, so there are no records anywhere + // and the replacement legitimately holds none. It must not inherit a stale set from a re-minted + // cycleSeq whose key survived. + redisTest( + "mints a records-less replacement when the cycle key is gone", + async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 60_000 }); + const raw = createRedisClient(redisOptions, { onError: () => {} }); + try { + await store.append({ + entry: entry({ id: "snap_1" }), + kind: "birth", + isTerminal: false, + cycle: { + kind: "new", + completedWaitpoints: [{ id: "w_a", index: 0 }], + records: [record("w_a", "first")], + }, + }); + + await raw.del( + "snap:{run_1}:e", + "snap:{run_1}:idx", + "snap:{run_1}:cur", + "snap:{run_1}:seq", + "snap:{run_1}:wp:1" + ); + + const carried = await store.append({ + entry: entry({ id: "snap_2" }), + kind: "birth", + isTerminal: false, + cycle: { + kind: "carryForward", + cycleSeq: 1, + completedWaitpoints: [{ id: "w_b", index: 0 }], + }, + }); + + expect(carried).toMatchObject({ outcome: "written", cycleMismatch: true }); + + const read = await store.getLatest("run_1"); + expect(await recordsAt(raw, read!.cycle!.cycleSeq)).toBeUndefined(); + } finally { + await Promise.all([store.quit(), raw.quit().catch(() => {})]); + } + } + ); + + // Without refs there is nothing to mint from, so the entry is written with no pointer. That is + // the older behaviour and it stays: no pointer is safe, a pointer with no records is not. + redisTest("writes no pointer when the caller carried no refs", async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 60_000 }); + const raw = createRedisClient(redisOptions, { onError: () => {} }); + try { + await store.append({ + entry: entry({ id: "snap_1" }), + kind: "birth", + isTerminal: false, + cycle: { + kind: "new", + completedWaitpoints: [{ id: "w_a", index: 0 }], + records: [record("w_a", "first")], + }, + }); + + await raw.del("snap:{run_1}:e", "snap:{run_1}:idx", "snap:{run_1}:cur", "snap:{run_1}:seq"); + + const carried = await store.append({ + entry: entry({ id: "snap_2" }), + kind: "birth", + isTerminal: false, + cycle: { kind: "carryForward", cycleSeq: 1 }, + }); + + expect(carried).toMatchObject({ outcome: "written", cycleMismatch: true }); + const read = await store.getLatest("run_1"); + expect(read?.cycle).toBeUndefined(); + } finally { + await Promise.all([store.quit(), raw.quit().catch(() => {})]); + } + }); +}); diff --git a/internal-packages/run-store/src/redisSnapshotStore.ts b/internal-packages/run-store/src/redisSnapshotStore.ts index 2339ab0dd5e..acde9428877 100644 --- a/internal-packages/run-store/src/redisSnapshotStore.ts +++ b/internal-packages/run-store/src/redisSnapshotStore.ts @@ -117,6 +117,12 @@ export type ResolveCompletedWaitpointsArgs = { pointer: CompletedWaitpointsPointer; /** Index oracle only. A SUBSET of the record ids. Repeats preserved. */ order: string[]; + /** + * Every id the cycle recorded, deduped, including the ids with no batch index. This is the + * membership the resolver's coverage check runs over: `order` omits every index-less wait, + * so a check scoped to it cannot see an id whose record is missing. + */ + distinctIds: string[]; /** The authoritative, complete set. Iterate this, never `order`. */ records: CompletedWaitpointRecord[]; }; @@ -303,6 +309,21 @@ export class RedisSnapshotStore { completedWaitpoints: CompletedWaitpointRef[]; records?: CompletedWaitpointRecord[]; } + | { + /** + * Mint a cycle, and if the caller has no records, inherit the current cycle's when its + * id set is identical. + * + * For the one caller that cannot tell whether this id set continues the previous cycle: + * a head probe that FAILED. It must not carry a pointer it could not verify, and it + * cannot read the records it would need to mint a complete replacement. Minting without + * them would leave ids that resolve from nothing, so the comparison and the copy happen + * here, atomically, where the previous cycle can still be read. + */ + kind: "newInherit"; + completedWaitpoints: CompletedWaitpointRef[]; + records?: CompletedWaitpointRecord[]; + } | { kind: "carryForward"; cycleSeq: number; @@ -335,9 +356,9 @@ export class RedisSnapshotStore { let distinctJson = ""; let records = ""; let orderCount = "0"; - if (args.cycle?.kind === "new") { + if (args.cycle?.kind === "new" || args.cycle?.kind === "newInherit") { const order = deriveOrder(args.cycle.completedWaitpoints); - cycleMode = "new"; + cycleMode = args.cycle.kind === "newInherit" ? "newInherit" : "new"; orderJson = JSON.stringify(order); distinctJson = JSON.stringify(deriveDistinctIds(args.cycle.completedWaitpoints)); records = args.cycle.records ? JSON.stringify(args.cycle.records) : ""; @@ -434,6 +455,11 @@ export class RedisSnapshotStore { } if (orderJson !== "") { // The whole wp: key, not just its order field: records dominates it once populated. + // + // `records` is what the CALLER sent, so this under-reports the two cases where the script + // sources the set itself and the client never sees it: a refused carry-forward, and a mint + // after a failed probe that inherits. Sizing either exactly would cost the read those paths + // exist to avoid, and the set was already measured when the cycle it came from was minted. const cycleBytes = Buffer.byteLength(orderJson, "utf8") + Buffer.byteLength(records, "utf8"); this.metrics?.recordCycleKeyBytes(cycleBytes); if (this.highWater.cycleKeyBytes !== undefined && cycleBytes > this.highWater.cycleKeyBytes) { @@ -780,11 +806,40 @@ export class RedisSnapshotStore { -- The STORE mints cycleSeq, so the sequence is dense by construction and the terminal -- PEXPIRE loop from 1..c is correct. - local function mintCycle() + -- Set equality, NOT string equality. + -- + -- Both sides come from deriveDistinctIds over a Postgres read with no ORDER BY -- the + -- resume reads the run's block edges, the copy-forward reads the snapshot's waitpoint + -- rows -- so the same set of ids arrives in an arbitrary order on each append. Comparing + -- the serialised arrays would therefore disagree for almost every wait holding two or + -- more waitpoints, which is the batch fan-in this inherit exists to protect. + -- + -- Both arrays are already deduped by construction, so equal length plus membership one + -- way is set equality. This matches the decorator's own sameSet, which is deliberately + -- order-insensitive for the same reason. + local function sameDistinctSet(stored, incoming) + if not stored or stored == '' or incoming == '' then return false end + if stored == incoming then return true end + local ok, left = pcall(cjson.decode, stored) + if not ok or type(left) ~= 'table' then return false end + local right = cjson.decode(incoming) + if #left ~= #right then return false end + local seen = {} + for i = 1, #left do seen[left[i]] = true end + for i = 1, #right do + if not seen[right[i]] then return false end + end + return true + end + + -- Takes the record set rather than closing over ARGV, because the two mint sites source it + -- differently: a 'new' cycle uses what the caller sent, and the refusal path below reads it + -- from the cycle it is replacing. + local function mintCycle(recordsJson) local minted = redis.call('HINCRBY', seqKey, 'c', 1) redis.call('HSET', wpKey(minted), 'order', orderJson, 'count', orderCount, 'distinct', distinctJson) - if records ~= '' then - redis.call('HSET', wpKey(minted), 'records', records) + if recordsJson ~= '' then + redis.call('HSET', wpKey(minted), 'records', recordsJson) else -- A new cycle owns the whole key: a lost seq counter can re-mint a cycleSeq whose key -- still holds another cycle's records, and order/count stay mutually consistent so the @@ -795,7 +850,29 @@ export class RedisSnapshotStore { end if cycleMode == 'new' then - cycleSeq = mintCycle() + cycleSeq = mintCycle(records) + elseif cycleMode == 'newInherit' then + -- The caller's head probe failed, so it knows neither whether this id set continues the + -- previous cycle nor what records that cycle holds. Minting fresh is the safe direction, + -- but minting with NO records leaves ids that resolve from nothing, and the next resume + -- refuses the whole cycle rather than losing a result quietly. So inherit them here. + -- + -- Guarded on the distinct SET matching, which is sufficient and is deliberately weaker + -- than the carry test the decorator applies on a successful probe. That test also + -- requires the order to match, because a pointer hands the reader the previous cycle's + -- order; this mints a fresh cycle from the caller's OWN order, so only the records have + -- to be right, and records are keyed by waitpoint id. A differing set is a genuinely new + -- wait and must start with the caller's own records, even when that is none. + -- + -- Read before mintCycle, because mintCycle advances the counter this reads. + local inherited = records + if inherited == '' then + local prev = tonumber(redis.call('HGET', seqKey, 'c') or '0') + if prev > 0 and sameDistinctSet(redis.call('HGET', wpKey(prev), 'distinct'), distinctJson) then + inherited = redis.call('HGET', wpKey(prev), 'records') or '' + end + end + cycleSeq = mintCycle(inherited) elseif cycleMode == 'carry' then -- Attach the CARRIED pointer only if this incarnation actually minted that cycle. seq can -- be evicted while a wp: key survives, so a bare key-exists check would adopt a dead @@ -812,7 +889,22 @@ export class RedisSnapshotStore { -- Only possible when the caller carried the refs. With none there is nothing to mint -- from, and the entry is written with no pointer, which is the older behaviour. if distinctJson ~= '' then - cycleSeq = mintCycle() + -- Source the records HERE, not from the caller. A copy-forward append holds none of + -- its own, and having the caller pre-read them meant one HGET plus a parse on EVERY + -- copy-forward -- attempt start, dequeue, checkpoint -- to serve a branch that needs + -- eviction to reach. Reading in the branch that uses them costs nothing on the common + -- path, keeps the blob inside Redis, and is atomic with the mint, so no reader can + -- observe a replacement cycle whose ids have no records. + -- + -- The surviving cycle key is the source. It is readable in exactly the case that + -- matters: 'minted < cycleSeqIn' means the seq counter was lost while wp: lived. + -- When the cycle key itself is gone ('not c') the records are gone with it and there + -- is nothing to carry, which is what the caller's read also found. + local carried = records + if carried == '' then + carried = redis.call('HGET', wpKey(cycleSeqIn), 'records') or '' + end + cycleSeq = mintCycle(carried) end else cycleSeq = cycleSeqIn diff --git a/internal-packages/run-store/src/taskRunExecutionSnapshotStore.ts b/internal-packages/run-store/src/taskRunExecutionSnapshotStore.ts index c6bc145a91d..008a9dc43b2 100644 --- a/internal-packages/run-store/src/taskRunExecutionSnapshotStore.ts +++ b/internal-packages/run-store/src/taskRunExecutionSnapshotStore.ts @@ -15,6 +15,7 @@ import { Logger } from "@trigger.dev/core/logger"; import { generateInternalId } from "@trigger.dev/core/v3/isomorphic"; import { DelegatingRunStore } from "./delegatingRunStore.js"; import type { + CompletedWaitpointRecord, CompletedWaitpointRef, RedisSnapshotStore, SnapshotEntryInput, @@ -114,6 +115,7 @@ export type StagedAppend = { */ expectedCur?: string; completedWaitpoints?: CompletedWaitpointRef[]; + completedWaitpointRecords?: CompletedWaitpointRecord[]; }; export class TaskRunExecutionSnapshotStore extends DelegatingRunStore { @@ -177,7 +179,8 @@ export class TaskRunExecutionSnapshotStore extends DelegatingRunStore { "runInTransaction", item.entry, item.expectedCur, - item.completedWaitpoints + item.completedWaitpoints, + item.completedWaitpointRecords ); } @@ -420,7 +423,8 @@ export class TaskRunExecutionSnapshotStore extends DelegatingRunStore { "createExecutionSnapshot", entryFromCreateExecutionSnapshot(ctx, input), input.previousSnapshotId, - input.completedWaitpoints + input.completedWaitpoints, + input.completedWaitpointRecords ); return created; } @@ -503,7 +507,8 @@ export class TaskRunExecutionSnapshotStore extends DelegatingRunStore { site: string, entry: SnapshotEntryInput, expectedCur?: string, - completedWaitpoints?: CompletedWaitpointRef[] + completedWaitpoints?: CompletedWaitpointRef[], + completedWaitpointRecords?: CompletedWaitpointRecord[] ): Promise { if (this.staging) { // Inside a transaction the append cannot run until the Postgres side commits, or a rollback @@ -512,6 +517,7 @@ export class TaskRunExecutionSnapshotStore extends DelegatingRunStore { entry, ...(expectedCur !== undefined && { expectedCur }), ...(completedWaitpoints && { completedWaitpoints }), + ...(completedWaitpointRecords && { completedWaitpointRecords }), }); return; } @@ -523,7 +529,11 @@ export class TaskRunExecutionSnapshotStore extends DelegatingRunStore { snapshotId: entry.id, }); - const cycle = await this.#resolveCycle(entry.runId, completedWaitpoints); + const cycle = await this.#resolveCycle( + entry.runId, + completedWaitpoints, + completedWaitpointRecords + ); const result = await this.redis.append({ entry, @@ -569,18 +579,38 @@ export class TaskRunExecutionSnapshotStore extends DelegatingRunStore { * the pointer model exists to remove. So an unchanged id set carries the previous cycleSeq * forward and writes no key. * - * The extra read only happens for an append that actually carries waitpoints, which is the resume - * path rather than the hot path. + * Resolving a cycle adds no read of its own. The id-set comparison reuses the head this method + * already probes, and the refusal path's record set is sourced inside the append script, in the + * branch that uses it. An earlier revision pre-read that set here, which put one HGET and a + * parse of the whole blob on every copy-forward append -- attempt start, dequeue, checkpoint -- + * to serve a branch that needs eviction to reach. Those are the hot paths; the resume path is + * the one that supplies `records` and never read at all. + * + * Three arms: `new` mints from the caller's own records, `newInherit` mints after a probe that + * FAILED and lets the script inherit the previous cycle's records when the id set matches, and + * `carryForward` points at a cycle already minted. * - * `records` is deliberately left unset. The record envelope belongs to the waitpoint lane and - * ships empty in this build, so dual-write never re-versions the entry when it arrives. + * `records` rides the mint arms only. A carryForward passes none: it points at a cycle already + * minted, and if the store refuses that pointer and mints a replacement inside the same call, + * the script reads the record set off the cycle it is replacing. A legacy-only wait supplies + * none anywhere, which is what keeps a Postgres-resident resume byte-identical to before. */ async #resolveCycle( runId: string, - completedWaitpoints?: CompletedWaitpointRef[] + completedWaitpoints?: CompletedWaitpointRef[], + records?: CompletedWaitpointRecord[] ): Promise< - | { kind: "new"; completedWaitpoints: CompletedWaitpointRef[] } - | { kind: "carryForward"; cycleSeq: number; completedWaitpoints: CompletedWaitpointRef[] } + | { + kind: "new" | "newInherit"; + completedWaitpoints: CompletedWaitpointRef[]; + records?: CompletedWaitpointRecord[]; + } + | { + kind: "carryForward"; + cycleSeq: number; + completedWaitpoints: CompletedWaitpointRef[]; + records?: CompletedWaitpointRecord[]; + } | undefined > { if (!completedWaitpoints || completedWaitpoints.length === 0) { @@ -603,20 +633,34 @@ export class TaskRunExecutionSnapshotStore extends DelegatingRunStore { sameOrder(previousIds.order, order) && sameSet(previousIds.distinctIds, distinct) ) { + // A copy-forward carries no records of its own, and needs none: it points at a cycle + // already minted. The script can still refuse that pointer and mint a replacement from + // these refs, and a replacement minted with no records holds ids nothing resolves -- so + // the script sources the record set from the cycle it replaces, atomically with the mint. + // Doing it there rather than here is what keeps every copy-forward append read-free. return { kind: "carryForward", cycleSeq: head.cycle.cycleSeq, completedWaitpoints, + ...(records && { records }), }; } } catch (error) { // A failed probe must not lose the waitpoints. Minting a fresh cycle is the safe direction: // it costs one duplicated record set, where a wrong carryForward would point at another // cycle's ids. + // + // `newInherit` rather than `new`, because a copy-forward caller carries no records of its + // own: the probe is what would have found the previous cycle to read them from. Minting + // plain `new` here writes ids with no records, and the next resume then refuses the cycle + // outright -- a probe failure that recovers turns into a run that cannot resume. The script + // inherits them instead, and only when the id set is identical. this.logger.warn("snapshot cycle probe failed, minting a new cycle", { runId, error }); + + return { kind: "newInherit", completedWaitpoints, ...(records && { records }) }; } - return { kind: "new", completedWaitpoints }; + return { kind: "new", completedWaitpoints, ...(records && { records }) }; } /** diff --git a/internal-packages/run-store/src/taskRunExecutionSnapshotStore.waitpointRecords.test.ts b/internal-packages/run-store/src/taskRunExecutionSnapshotStore.waitpointRecords.test.ts new file mode 100644 index 00000000000..8a80d917a42 --- /dev/null +++ b/internal-packages/run-store/src/taskRunExecutionSnapshotStore.waitpointRecords.test.ts @@ -0,0 +1,500 @@ +// The record set's journey from a caller's input to the wait cycle's key. +// +// The raw store already pins that a records array round-trips through the cycle hash. What is +// untested without this file is the decorator leg: that `completedWaitpointRecords` on a +// snapshot input reaches `cycle.records`, that a mint carries it, and that a copy-forward and +// a legacy-only wait carry none — which is what keeps a Postgres-resident resume unchanged. +import { createRedisClient } from "@internal/redis"; +import { containerTest } from "@internal/testcontainers"; +import { + generateInternalId, + generateWaitpointId, + parseWaitpointId, +} from "@trigger.dev/core/v3/isomorphic"; +import { describe, expect } from "vitest"; +import { PostgresRunStore } from "./PostgresRunStore.js"; +import { RedisSnapshotStore, type CompletedWaitpointRecord } from "./redisSnapshotStore.js"; +import { entryFromCreateRun } from "./snapshotEntry.js"; +import { TaskRunExecutionSnapshotStore } from "./taskRunExecutionSnapshotStore.js"; +import { + buildCreateRunData, + seedSnapshotEnvironment, + seedSnapshotWaitpoints, + type SnapshotFixtureEnv, +} from "./testFixtures/snapshotIdFixture.js"; +import type { RunStore } from "./types.js"; + +const COMPLETED_TTL_MS = 72 * 60 * 60 * 1000; + +function build(prisma: never, redisOptions: never) { + const redis = new RedisSnapshotStore({ redisOptions, completedTtlMs: COMPLETED_TTL_MS }); + const decorated = new TaskRunExecutionSnapshotStore( + new PostgresRunStore({ prisma, readOnlyPrisma: prisma }) as unknown as RunStore, + { + store: redis, + mode: "redis-read", + readPercent: 100, + metrics: { + recordWrite: () => {}, + recordAppendFailed: () => {}, + recordRead: () => {}, + }, + } + ); + return { decorated, redis }; +} + +async function seedRun( + decorated: TaskRunExecutionSnapshotStore, + redis: RedisSnapshotStore, + env: SnapshotFixtureEnv +): Promise { + const runId = generateInternalId(); + const snapshot = { + id: generateInternalId(), + engine: "V2" as const, + executionStatus: "RUN_CREATED" as const, + description: "Run was created", + runStatus: "PENDING" as const, + environmentId: env.id, + environmentType: env.type, + projectId: env.projectId, + organizationId: env.organizationId, + }; + await redis.append({ + entry: entryFromCreateRun({ id: snapshot.id, runId, createdAt: new Date() }, snapshot), + kind: "birth", + isTerminal: false, + }); + await decorated.createRun({ data: buildCreateRunData(runId, env), snapshot }); + return runId; +} + +function resumeInput( + runId: string, + env: SnapshotFixtureEnv, + completedWaitpoints: { id: string; index?: number }[], + completedWaitpointRecords?: CompletedWaitpointRecord[] +) { + return { + id: generateInternalId(), + run: { id: runId, status: "EXECUTING" as const, attemptNumber: 1 }, + snapshot: { executionStatus: "EXECUTING" as const, description: "Run resumed" }, + completedWaitpoints, + ...(completedWaitpointRecords && { completedWaitpointRecords }), + environmentId: env.id, + environmentType: env.type, + projectId: env.projectId, + organizationId: env.organizationId, + }; +} + +function record(id: string, overrides: Partial = {}) { + return { + id, + friendlyId: `waitpoint_${id}`, + type: "MANUAL" as const, + completedAt: "2026-08-25T00:00:00.000Z", + outputType: "application/json", + outputIsError: false, + output: { inline: '{"ok":true}' }, + ...overrides, + } satisfies CompletedWaitpointRecord; +} + +async function readRecords( + probe: ReturnType, + runId: string +): Promise { + const [cycleKey] = await probe.keys(`snap:{${runId}}:wp:*`); + if (!cycleKey) return undefined; + const raw = await probe.hget(cycleKey, "records"); + return raw ? (JSON.parse(raw) as CompletedWaitpointRecord[]) : undefined; +} + +describe("the completed-waitpoint record set", () => { + containerTest( + "a mint writes the records the caller supplied", + async ({ prisma, redisOptions }) => { + const { decorated, redis } = build(prisma as never, redisOptions as never); + const probe = createRedisClient(redisOptions, { onError: () => {} }); + try { + const env = await seedSnapshotEnvironment(prisma); + const runId = await seedRun(decorated, redis, env); + const [wpA, wpB] = await seedSnapshotWaitpoints(prisma, env, 2); + + await decorated.createExecutionSnapshot( + resumeInput( + runId, + env, + [ + { id: wpA!, index: 0 }, + { id: wpB!, index: 1 }, + ], + [record(wpA!), record(wpB!)] + ) + ); + + const records = await readRecords(probe, runId); + + expect(records).toHaveLength(2); + expect(records?.map((r) => r.id).sort()).toEqual([wpA, wpB].sort()); + expect(records?.[0]?.output).toEqual({ inline: '{"ok":true}' }); + } finally { + await Promise.all([redis.quit(), probe.quit().catch(() => {})]); + } + } + ); + + // The inertness guarantee. A wait with no store-resident half supplies no records, and the + // cycle key must then hold none — a Postgres-resident resume is unchanged. + containerTest("a mint with no records supplied writes none", async ({ prisma, redisOptions }) => { + const { decorated, redis } = build(prisma as never, redisOptions as never); + const probe = createRedisClient(redisOptions, { onError: () => {} }); + try { + const env = await seedSnapshotEnvironment(prisma); + const runId = await seedRun(decorated, redis, env); + const [wpA] = await seedSnapshotWaitpoints(prisma, env, 1); + + await decorated.createExecutionSnapshot(resumeInput(runId, env, [{ id: wpA!, index: 0 }])); + + expect(await readRecords(probe, runId)).toBeUndefined(); + } finally { + await Promise.all([redis.quit(), probe.quit().catch(() => {})]); + } + }); + + // One record set per wait cycle, not one per entry in the resume chain. That is the write + // amplification the pointer model exists to remove. + containerTest("a copy-forward writes no second record set", async ({ prisma, redisOptions }) => { + const { decorated, redis } = build(prisma as never, redisOptions as never); + const probe = createRedisClient(redisOptions, { onError: () => {} }); + try { + const env = await seedSnapshotEnvironment(prisma); + const runId = await seedRun(decorated, redis, env); + const [wpA] = await seedSnapshotWaitpoints(prisma, env, 1); + + const waitpoints = [{ id: wpA!, index: 0 }]; + await decorated.createExecutionSnapshot(resumeInput(runId, env, waitpoints, [record(wpA!)])); + await decorated.createExecutionSnapshot(resumeInput(runId, env, waitpoints, [record(wpA!)])); + + const cycleKeys = await probe.keys(`snap:{${runId}}:wp:*`); + + expect(cycleKeys).toHaveLength(1); + expect(await readRecords(probe, runId)).toHaveLength(1); + } finally { + await Promise.all([redis.quit(), probe.quit().catch(() => {})]); + } + }); + + // The shape every copy-forward append actually has: same id set, no records of its own. The + // cycle's records must survive it untouched. + containerTest( + "a records-less copy-forward does not clobber the records", + async ({ prisma, redisOptions }) => { + const { decorated, redis } = build(prisma as never, redisOptions as never); + const probe = createRedisClient(redisOptions, { onError: () => {} }); + try { + const env = await seedSnapshotEnvironment(prisma); + const runId = await seedRun(decorated, redis, env); + const [wpA] = await seedSnapshotWaitpoints(prisma, env, 1); + const waitpoints = [{ id: wpA!, index: 0 }]; + + await decorated.createExecutionSnapshot( + resumeInput(runId, env, waitpoints, [record(wpA!)]) + ); + // No records this time, exactly as dequeue/checkpoint/attempt appends do. + await decorated.createExecutionSnapshot(resumeInput(runId, env, waitpoints)); + + const records = await readRecords(probe, runId); + + expect(await probe.keys(`snap:{${runId}}:wp:*`)).toHaveLength(1); + expect(records).toHaveLength(1); + expect(records?.[0]?.id).toBe(wpA); + } finally { + await Promise.all([redis.quit(), probe.quit().catch(() => {})]); + } + } + ); + + containerTest( + "a record set survives beside a repeat-preserving order", + async ({ prisma, redisOptions }) => { + const { decorated, redis } = build(prisma as never, redisOptions as never); + const probe = createRedisClient(redisOptions, { onError: () => {} }); + try { + const env = await seedSnapshotEnvironment(prisma); + const runId = await seedRun(decorated, redis, env); + const [wpA] = await seedSnapshotWaitpoints(prisma, env, 1); + + const created = await decorated.createExecutionSnapshot( + resumeInput( + runId, + env, + [ + { id: wpA!, index: 0 }, + { id: wpA!, index: 1 }, + ], + [record(wpA!)] + ) + ); + + const ids = await redis.getSnapshotWaitpointIds(runId, created.id); + + // One record, two positions. The record set carries membership, the order carries + // multiplicity. + expect(await readRecords(probe, runId)).toHaveLength(1); + expect(ids.order).toEqual([wpA, wpA]); + } finally { + await Promise.all([redis.quit(), probe.quit().catch(() => {})]); + } + } + ); +}); + +// A cycle-probe failure must not mint an unresolvable cycle. +// +// The probe is how a copy-forward append discovers that its id set continues the previous cycle. +// When it throws, the append still has to write, and minting fresh is the safe direction -- but a +// copy-forward carries no records of its own, so a plain mint writes ids that resolve from +// nothing. The next resume then refuses the whole cycle: one transient probe failure would leave +// a run permanently unable to resume, with the join rows still sitting in Postgres. +// +// Reachable with Redis healthy: getLatest JSON-parses the entry payload, so one corrupt entry +// does it. +describe("a failed cycle probe", () => { + async function recordsAtHead( + redis: RedisSnapshotStore, + probe: ReturnType, + runId: string + ): Promise { + const head = await redis.getLatest(runId); + const cycleSeq = head?.cycle?.cycleSeq; + if (cycleSeq === undefined) return undefined; + const raw = await probe.hget(`snap:{${runId}}:wp:${cycleSeq}`, "records"); + return raw ? (JSON.parse(raw) as CompletedWaitpointRecord[]) : undefined; + } + + // Fails the NEXT probe only, so the append that follows it still runs against a healthy store. + function breakNextProbe(redis: RedisSnapshotStore) { + const original = redis.getLatest.bind(redis); + let broken = true; + redis.getLatest = async (runId: string, opts?: { environmentId?: string }) => { + if (broken) { + broken = false; + throw new Error("probe failed"); + } + return original(runId, opts); + }; + return () => void (redis.getLatest = original); + } + + async function seedStoreWaitpoint( + prisma: never, + env: SnapshotFixtureEnv, + id: string + ): Promise { + await ( + prisma as unknown as { waitpoint: { create: (a: unknown) => Promise } } + ).waitpoint.create({ + data: { + id, + friendlyId: `waitpoint_${id}`, + type: "MANUAL", + status: "COMPLETED", + completedAt: new Date(), + idempotencyKey: `idem_${id.slice(-12)}`, + userProvidedIdempotencyKey: false, + projectId: env.projectId, + environmentId: env.id, + }, + }); + } + + // TWO waitpoints, and the copy-forward re-passes them in the OPPOSITE order. + // + // That is not a contrived permutation: both id lists are derived from Postgres reads with no + // ORDER BY -- the resume reads the run's block edges, the copy-forward reads the snapshot's + // waitpoint rows -- so an arbitrary order on each append is the normal case. A guard comparing + // the serialised id arrays would miss here and mint a record-less cycle, which is the very + // state this path exists to prevent, for every wait holding more than one waitpoint. + // + // A single-id version of this test passes whether the guard compares sets or strings, so it + // cannot hold the property on its own. + containerTest( + "inherits the records when the id set is unchanged but reordered", + async ({ prisma, redisOptions }) => { + const { decorated, redis } = build(prisma as never, redisOptions as never); + const probe = createRedisClient(redisOptions, { onError: () => {} }); + + try { + const env = await seedSnapshotEnvironment(prisma); + const runId = await seedRun(decorated, redis, env); + const first = generateWaitpointId("MANUAL"); + const second = generateWaitpointId("MANUAL"); + await seedStoreWaitpoint(prisma as never, env, first); + await seedStoreWaitpoint(prisma as never, env, second); + + // The resume, which supplies the records and mints cycle 1. + await decorated.createExecutionSnapshot( + resumeInput( + runId, + env, + [ + { id: first, index: 0 }, + { id: second, index: 1 }, + ], + [record(first), record(second)] + ) + ); + + // The copy-forward: same ids, same indexes, arbitrary order, no records of its own. + const restore = breakNextProbe(redis); + try { + await decorated.createExecutionSnapshot( + resumeInput(runId, env, [ + { id: second, index: 1 }, + { id: first, index: 0 }, + ]) + ); + } finally { + restore(); + } + + // A fresh cycle, because an unverified pointer must not be carried... + const cycleKeys = await probe.keys(`snap:{${runId}}:wp:*`); + expect(cycleKeys.length).toBe(2); + + // ...holding BOTH records, so the cycle stays resolvable. + const records = await recordsAtHead(redis, probe, runId); + expect(records?.map((r) => r.id).sort()).toEqual([first, second].sort()); + } finally { + await Promise.all([redis.quit(), probe.quit().catch(() => {})]); + } + } + ); + + // The other direction: a genuinely NEW wait must not inherit the previous cycle's records, or + // the resolver would hand the run a result belonging to a waitpoint it is not waiting on. + containerTest("inherits nothing when the id set differs", async ({ prisma, redisOptions }) => { + const { decorated, redis } = build(prisma as never, redisOptions as never); + const probe = createRedisClient(redisOptions, { onError: () => {} }); + + try { + const env = await seedSnapshotEnvironment(prisma); + const runId = await seedRun(decorated, redis, env); + const [first, second] = await seedSnapshotWaitpoints(prisma, env, 2); + + await decorated.createExecutionSnapshot( + resumeInput(runId, env, [{ id: first!, index: 0 }], [record(first!)]) + ); + + const restore = breakNextProbe(redis); + try { + await decorated.createExecutionSnapshot( + resumeInput(runId, env, [{ id: second!, index: 0 }]) + ); + } finally { + restore(); + } + + expect(await recordsAtHead(redis, probe, runId)).toBeUndefined(); + } finally { + await Promise.all([redis.quit(), probe.quit().catch(() => {})]); + } + }); +}); + +// A copy-forward append reads NO cycle key, whatever the ids look like. +// +// The record set a refusal needs is sourced inside the append script now, so the decorator does +// not pre-read it. That matters because copy-forward appends are the hot paths -- attempt start, +// dequeue, checkpoint -- while the refusal they were preparing for needs eviction to reach. These +// count the reads rather than infer them: the absence of the round trip is the whole point, and it +// is not visible in the resulting entry. +// +// The store-format case is the one that used to pay: it performed the read on every copy-forward. +describe("a copy-forward append reads no cycle key", () => { + function countCycleReads(redis: RedisSnapshotStore) { + const client = (redis as unknown as { redis: Record unknown> }) + .redis; + const original = client.hget.bind(client); + const reads: string[] = []; + client.hget = (...args: never[]) => { + const key = args[0]; + if (typeof key === "string" && key.includes(":wp:")) { + reads.push(key); + } + return original(...args); + }; + return { reads, restore: () => void (client.hget = original) }; + } + + containerTest("with a legacy-only id set", async ({ prisma, redisOptions }) => { + const { decorated, redis } = build(prisma as never, redisOptions as never); + const { reads, restore } = countCycleReads(redis); + + try { + const env = await seedSnapshotEnvironment(prisma); + const runId = await seedRun(decorated, redis, env); + const [wpA] = await seedSnapshotWaitpoints(prisma, env, 1); + const waitpoints = [{ id: wpA!, index: 0 }]; + + // Two appends with the same id set: the second carries the first's cycle forward. + await decorated.createExecutionSnapshot(resumeInput(runId, env, waitpoints)); + await decorated.createExecutionSnapshot(resumeInput(runId, env, waitpoints)); + + expect(wpA && parseWaitpointId(wpA).format).toBe("legacy"); + expect(reads).toEqual([]); + } finally { + restore(); + await redis.quit(); + } + }); + + containerTest("with a store-format id in the cycle", async ({ prisma, redisOptions }) => { + const { decorated, redis } = build(prisma as never, redisOptions as never); + const { reads, restore } = countCycleReads(redis); + + try { + const env = await seedSnapshotEnvironment(prisma); + const runId = await seedRun(decorated, redis, env); + // A real row, but with a store-format id rather than the fixture's cuid. The delegate + // still writes the completed-waitpoint join, so the row has to exist. + const storeId = generateWaitpointId("MANUAL"); + await prisma.waitpoint.create({ + data: { + id: storeId, + friendlyId: `waitpoint_${storeId}`, + type: "MANUAL", + status: "COMPLETED", + completedAt: new Date(), + idempotencyKey: `idem_${storeId.slice(-12)}`, + userProvidedIdempotencyKey: false, + projectId: env.projectId, + environmentId: env.id, + }, + }); + const waitpoints = [{ id: storeId, index: 0 }]; + + await decorated.createExecutionSnapshot( + resumeInput(runId, env, waitpoints, [record(storeId)]) + ); + await decorated.createExecutionSnapshot(resumeInput(runId, env, waitpoints)); + + expect(parseWaitpointId(storeId).format).toBe("b32hexW"); + expect(reads).toEqual([]); + + // And the records the mint wrote are still there, untouched by the copy-forward. + const probe = createRedisClient(redisOptions, { onError: () => {} }); + try { + expect(await readRecords(probe, runId)).toHaveLength(1); + } finally { + await probe.quit().catch(() => {}); + } + } finally { + restore(); + await redis.quit(); + } + }); +}); diff --git a/internal-packages/run-store/src/types.ts b/internal-packages/run-store/src/types.ts index 41fecf90bb5..b805ab91456 100644 --- a/internal-packages/run-store/src/types.ts +++ b/internal-packages/run-store/src/types.ts @@ -13,6 +13,7 @@ import type { } from "@trigger.dev/database"; import type { TaskRunError } from "@trigger.dev/core/v3/schemas"; import type { Residency, ShardKey } from "@trigger.dev/core/v3/isomorphic"; +import type { CompletedWaitpointRecord } from "./redisSnapshotStore.js"; /** * Client accepted by the read methods. Reads route through the replica by @@ -352,6 +353,10 @@ export type CreateExecutionSnapshotInput = { workerId?: string; runnerId?: string; completedWaitpoints?: { id: string; index?: number }[]; + /** One envelope per DISTINCT completed waitpoint id. Owned by the waitpoint lane; the + * snapshot store only carries it into the wait cycle's key. Absent for a legacy-only + * wait, which is what keeps a Postgres-resident resume unchanged. */ + completedWaitpointRecords?: CompletedWaitpointRecord[]; error?: string; };