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..ea360bc0b72 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"; @@ -173,7 +173,7 @@ async function getSnapshotWaitpointIdsWithPresence( * This is necessary because waitpoints can have large outputs (100KB+), * and fetching many at once can exceed Node.js string limits. */ -async function fetchWaitpointsInChunks( +export async function fetchWaitpointsInChunks( prisma: PrismaClientOrTransaction, waitpointIds: string[], runStore?: RunStore, @@ -449,6 +449,7 @@ export class ExecutionSnapshotSystem { workerId, runnerId, completedWaitpoints, + completedWaitpointRecords, error, }: { run: { id: string; status: TaskRunStatus; attemptNumber?: number | null }; @@ -470,6 +471,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 +494,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 3dbed999445..36738cfa983 100644 --- a/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts +++ b/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts @@ -1,4 +1,6 @@ import { timeoutError } from "@trigger.dev/core/v3"; +import { parseWaitpointId } from "@trigger.dev/core/v3/isomorphic"; +import type { CompletedWaitpointRecord } from "@internal/run-store"; import type { PrismaClientOrTransaction, TaskRun, @@ -10,7 +12,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"; @@ -601,6 +604,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, { @@ -623,6 +633,7 @@ export class WaitpointSystem { id: b.waitpoint.id, index: b.batchIndex ?? undefined, })), + ...(completedWaitpointRecords && { completedWaitpointRecords }), } ); @@ -668,6 +679,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({ @@ -682,6 +698,7 @@ export class WaitpointSystem { id: b.waitpoint.id, index: b.batchIndex ?? undefined, })), + ...(completedWaitpointRecords && { completedWaitpointRecords }), checkpointId: snapshot.checkpointId ?? undefined, }); @@ -728,6 +745,42 @@ export class WaitpointSystem { return this.coordinator.mintAssociatedWaitpointData({ projectId, environmentId }); } + /** + * 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. + */ + async #completedWaitpointRecordsFor( + runId: string, + blockingWaitpoints: RunBlockEdge[] + ): Promise { + const storeFormatIds = [ + ...new Set( + blockingWaitpoints + .map((b) => b.waitpoint.id) + .filter((id) => parseWaitpointId(id).format === "b32hexW") + ), + ]; + + if (storeFormatIds.length === 0) { + return undefined; + } + + 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..f1e6cc33d41 --- /dev/null +++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/completedWaitpointEquivalence.test.ts @@ -0,0 +1,382 @@ +// 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 { seedChildRunWithOutput } from "./testFixtures/childRun.js"; +import { enhanceExecutionSnapshotWithWaitpoints } from "../systems/executionSnapshotSystem.js"; +import { buildCompletedWaitpointRecords } from "./completedWaitpointRecords.js"; +import { + createCompletedWaitpointResolver, + createRunOutputReader, +} 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({ + readRunOutput: createRunOutputReader(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); + }); + + 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..0d05396ceae --- /dev/null +++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/completedWaitpointResolver.runOutput.test.ts @@ -0,0 +1,142 @@ +// 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, + createRunOutputReader, + UnresolvableWaitpointId, +} from "./completedWaitpointResolver.js"; +import { 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({ readRunOutput: createRunOutputReader(runStore) }); +} + +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 runStore = new PostgresRunStore({ prisma, readOnlyPrisma: prisma }); + const reads: string[] = []; + const reader = createRunOutputReader(runStore); + + // Counts calls and DELEGATES to the real reader, so the Postgres read still happens. This + // wraps the collaborator rather than replacing it: the assertion is about how many reads + // occur, which is not observable from the resolved output alone. + const result = await createCompletedWaitpointResolver({ + readRunOutput: async (id) => { + reads.push(id); + return reader(id); + }, + })({ + runId: "run_parent", + pointer: { cycleSeq: 1, count: 2 }, + order: ["wp_run", "wp_run"], + distinctIds: ["wp_run"], + records: [deriveRecord(runId)], + }); + + expect(result).toHaveLength(2); + expect(reads).toEqual([runId]); + }); + + 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..2ae08afbb49 --- /dev/null +++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/completedWaitpointResolver.ts @@ -0,0 +1,198 @@ +import type { + CompletedWaitpointRecord, + ReadClient, + 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. Returns undefined when the row is gone. + * + * 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. + */ + readRunOutput?(taskRunId: string): Promise; +}; + +/** + * The production reader: TaskRun.output for the completing run, through the store so the read + * routes to the run's owning database. + */ +export function createRunOutputReader( + runStore: Pick, + client?: ReadClient +): (taskRunId: string) => Promise { + return async (taskRunId) => { + const run = await runStore.findRun({ id: taskRunId }, { select: { output: true } }, client); + return run?.output ?? undefined; + }; +} + +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"); + } + } + + const out: CompletedWaitpoint[] = []; + + for (const record of args.records) { + const indexes = positionsOf(record.id, args.order); + // Hydrated once per record, not once per position, so a run at several batch indexes + // costs one read rather than one per index. + const output = await hydrateOutput(record, deps); + + 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; + }; +} + +// An id with no position yields one entry with an undefined index, matching what the +// existing hydration does for a wait that carried no batch index. +function positionsOf(waitpointId: string, order: string[]): (number | undefined)[] { + const indexes: (number | undefined)[] = []; + + for (let i = 0; i < order.length; i++) { + if (order[i] === waitpointId) { + indexes.push(i); + } + } + + return indexes.length === 0 ? [undefined] : indexes; +} + +async function hydrateOutput( + record: CompletedWaitpointRecord, + deps: CompletedWaitpointResolverDeps +): Promise { + 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; + } + + if (!record.completedByTaskRunId) { + return undefined; + } + + if (!deps.readRunOutput) { + throw new Error( + `Waitpoint ${record.id} defers its output to run ${record.completedByTaskRunId}, but the resolver was built with no run-output reader.` + ); + } + + // 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 = await deps.readRunOutput(record.completedByTaskRunId); + if (output === undefined) { + throw new UnresolvableWaitpointId(record.id, "lost-run-output"); + } + + return output; +} 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 d1e48fa4f8d..46eea8d740c 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 { fetchWaitpointsInChunks } 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,35 @@ 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. + */ + async readCompletionEnvelopes({ + runId, + waitpointIds, + }: ReadCompletionEnvelopesParams): Promise { + if (waitpointIds.length === 0) { + return []; + } + + const rows = await fetchWaitpointsInChunks(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..b422bc43ac7 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,141 @@ 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(); + } + }); +}); diff --git a/internal-packages/run-engine/src/engine/waitpointCoordinator/storeCoordinator.ts b/internal-packages/run-engine/src/engine/waitpointCoordinator/storeCoordinator.ts index 723552c57ab..42be33724f6 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"; @@ -505,6 +506,49 @@ 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 []; + } + + const halves = await Promise.all( + waitpointIds.map((id) => this.redis.hmget(waitpointKeys(id).record, "r", "c")) + ); + + 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 +580,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..57d08393a65 --- /dev/null +++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/testFixtures/childRun.ts @@ -0,0 +1,43 @@ +import type { PrismaClient } from "@trigger.dev/database"; +import { setupAuthenticatedEnvironment } from "../../tests/setup.js"; + +/** + * A completed child run 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 a real row rather than a stand-in for one. + */ +export async function seedChildRunWithOutput( + prisma: PrismaClient, + output: string | null, + outputType = "application/json" +): Promise { + const env = await setupAuthenticatedEnvironment(prisma, "PRODUCTION"); + const suffix = env.id.slice(-10); + + 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 }, + }); + + return run.id; +} diff --git a/internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts b/internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts index 8a50abb7d1c..8611a361b42 100644 --- a/internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts +++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts @@ -16,6 +16,9 @@ import type { PrismaClientOrTransaction, Waitpoint } from "@trigger.dev/database 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; @@ -31,6 +34,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..c928271b52b --- /dev/null +++ b/internal-packages/run-store/src/redisSnapshotStore.recordsOnCarryRefusal.test.ts @@ -0,0 +1,180 @@ +// 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, so this is the case a refusal + // actually meets. Before the decorator read the surviving cycle's records, this minted a + // replacement holding ids with no records, permanently. + 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")], + }, + }); + + // What the decorator now does for a records-less carry: read the surviving cycle's + // records and carry those into the refusal branch. + const carried = await store.getCycleRecords("run_1", 1); + expect(carried).toHaveLength(1); + + await raw.del("snap:{run_1}:e", "snap:{run_1}:idx", "snap:{run_1}:cur", "snap:{run_1}:seq"); + + await store.append({ + entry: entry({ id: "snap_2" }), + kind: "birth", + isTerminal: false, + cycle: { + kind: "carryForward", + cycleSeq: 1, + completedWaitpoints: [{ id: "w_a", index: 0 }], + records: carried, + }, + }); + + 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"); + } 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..25518a75b9e 100644 --- a/internal-packages/run-store/src/redisSnapshotStore.ts +++ b/internal-packages/run-store/src/redisSnapshotStore.ts @@ -17,6 +17,12 @@ export function snapshotKeys(runId: string): SnapshotKeys { return { e: `${base}:e`, idx: `${base}:idx`, cur: `${base}:cur`, seq: `${base}:seq` }; } +// The per-cycle key. Shares the {runId} tag with the four core keys, and the append scripts +// derive the same name in Lua from KEYS[1]; this is its only TypeScript-side spelling. +export function cycleKey(runId: string, cycleSeq: number): string { + return `snap:{${runId}}:wp:${cycleSeq}`; +} + export type CompletedWaitpointRef = { id: string; index?: number }; // Reproduces PostgresRunStore.#createExecutionSnapshot's completedWaitpointOrder derivation exactly: @@ -117,6 +123,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[]; }; @@ -496,6 +508,25 @@ export class RedisSnapshotStore { }); } + /** + * The record set a cycle already holds, if any. + * + * Read on one path only: a copy-forward append that carries no records of its own. A + * copy-forward legitimately has none, because it only points at a cycle that was already + * minted. But the append script can REFUSE an untrustworthy pointer and mint a replacement + * from the carried refs, and a replacement minted with no records holds ids that nothing can + * resolve. So the caller reads the surviving cycle's records and carries those. + */ + async getCycleRecords( + runId: string, + cycleSeq: number + ): Promise { + return this.#timed("getCycleRecords", async () => { + const raw = await this.redis.hget(cycleKey(runId, cycleSeq), "records"); + return raw ? (JSON.parse(raw) as CompletedWaitpointRecord[]) : undefined; + }); + } + // A miss is not an error. It is the coexistence path: a pre-cutover snapshot id, expired history, // or an org not yet enabled. The caller falls back to Postgres. async getSince( diff --git a/internal-packages/run-store/src/taskRunExecutionSnapshotStore.ts b/internal-packages/run-store/src/taskRunExecutionSnapshotStore.ts index c6bc145a91d..9094e617023 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, @@ -572,15 +582,28 @@ export class TaskRunExecutionSnapshotStore extends DelegatingRunStore { * The extra read only happens for an append that actually carries waitpoints, which is the resume * path rather than the hot path. * - * `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 every arm that can mint. A carryForward normally writes no key, but the + * store may refuse the pointer and mint a replacement inside the same call, and that + * replacement needs the records or the resolver's coverage check rejects the cycle later. + * A legacy-only wait supplies none at all, 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"; + completedWaitpoints: CompletedWaitpointRef[]; + records?: CompletedWaitpointRecord[]; + } + | { + kind: "carryForward"; + cycleSeq: number; + completedWaitpoints: CompletedWaitpointRef[]; + records?: CompletedWaitpointRecord[]; + } | undefined > { if (!completedWaitpoints || completedWaitpoints.length === 0) { @@ -603,10 +626,17 @@ export class TaskRunExecutionSnapshotStore extends DelegatingRunStore { sameOrder(previousIds.order, order) && sameSet(previousIds.distinctIds, distinct) ) { + // A copy-forward carries no records of its own, and does not need any: it points at a + // cycle already minted. But the script may refuse the pointer and mint a replacement + // from these refs, and a replacement minted with no records holds ids that nothing + // resolves. So carry the surviving cycle's records for that branch. + const carried = records ?? (await this.#recordsForCycle(runId, head.cycle.cycleSeq)); + return { kind: "carryForward", cycleSeq: head.cycle.cycleSeq, completedWaitpoints, + ...(carried && { records: carried }), }; } } catch (error) { @@ -616,7 +646,26 @@ export class TaskRunExecutionSnapshotStore extends DelegatingRunStore { this.logger.warn("snapshot cycle probe failed, minting a new cycle", { runId, error }); } - return { kind: "new", completedWaitpoints }; + return { kind: "new", completedWaitpoints, ...(records && { records }) }; + } + + // Never fatal. Failing to read the records only loses the refusal branch's ability to mint a + // complete replacement, which is where it started; a throw here would fail an append that + // would otherwise have succeeded. + async #recordsForCycle( + runId: string, + cycleSeq: number + ): Promise { + try { + return await this.redis.getCycleRecords(runId, cycleSeq); + } catch (error) { + this.logger.warn("reading a cycle's records failed, carrying none", { + runId, + cycleSeq, + error, + }); + return undefined; + } } /** 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..61f83705953 --- /dev/null +++ b/internal-packages/run-store/src/taskRunExecutionSnapshotStore.waitpointRecords.test.ts @@ -0,0 +1,249 @@ +// 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 } 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(() => {})]); + } + } + ); +}); diff --git a/internal-packages/run-store/src/types.ts b/internal-packages/run-store/src/types.ts index 9ea39473e5b..7e7a548db35 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 } 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; };