diff --git a/internal-packages/run-engine/package.json b/internal-packages/run-engine/package.json index 96ace3a0e43..f4cfcd629d1 100644 --- a/internal-packages/run-engine/package.json +++ b/internal-packages/run-engine/package.json @@ -43,7 +43,7 @@ }, "scripts": { "clean": "rimraf dist", - "typecheck": "tsc --noEmit -p tsconfig.build.json", + "typecheck": "tsc --noEmit -p tsconfig.build.json && tsc --noEmit -p tsconfig.freeze-test.json", "test": "vitest --sequence.concurrent=false --no-file-parallelism", "test:coverage": "vitest --sequence.concurrent=false --no-file-parallelism --coverage.enabled", "build": "pnpm run clean && tsc -p tsconfig.build.json", diff --git a/internal-packages/run-engine/src/engine/systems/completedWaitpointFreeze.test.ts b/internal-packages/run-engine/src/engine/systems/completedWaitpointFreeze.test.ts new file mode 100644 index 00000000000..a1af90582b6 --- /dev/null +++ b/internal-packages/run-engine/src/engine/systems/completedWaitpointFreeze.test.ts @@ -0,0 +1,664 @@ +// The freeze's executable definition. It runs the real oracle, +// enhanceExecutionSnapshotWithWaitpoints, against a reference resolver over equivalent +// records, and asserts the two agree field for field. The waitpoint lane owns the +// production resolver; this reference exists so the frozen shapes are checked rather +// than asserted. +import { isDeepStrictEqual } from "node:util"; +import { describe, expect, it } from "vitest"; +import type { Waitpoint } from "@trigger.dev/database"; +import { BatchId, RunId } from "@trigger.dev/core/v3/isomorphic"; +import type { CompletedWaitpoint } from "@trigger.dev/core/v3"; +import type { + CompletedWaitpointRecord, + CompletedWaitpointResolver, + CompletedWaitpointsPointer, + ResolveCompletedWaitpointsArgs, +} from "@internal/run-store"; + +// The frozen key sets, pinned exactly and bidirectionally. Renames, removals, widenings and +// required-to-optional all already break compilation through the usage sites below; an ADDED +// OPTIONAL field does not, and on a jointly-owned frozen type that is the change neither lane +// may make unilaterally. These fail on it. +type Exact = [A] extends [B] ? ([B] extends [A] ? true : never) : never; + +const _recordKeys: Exact< + keyof CompletedWaitpointRecord, + | "id" + | "friendlyId" + | "type" + | "completedAt" + | "outputType" + | "outputIsError" + | "output" + | "completedByTaskRunId" + | "completedByBatchId" + | "completedAfter" + | "idempotencyKey" +> = true; + +const _pointerKeys: Exact = true; + +const _argsKeys: Exact< + keyof ResolveCompletedWaitpointsArgs, + "runId" | "batchId" | "pointer" | "order" | "records" +> = true; +import { enhanceExecutionSnapshotWithWaitpoints } from "./executionSnapshotSystem.js"; + +function makeWaitpoint(overrides: Partial): Waitpoint { + return { + id: "wp_default", + friendlyId: "waitpoint_default", + type: "MANUAL", + status: "COMPLETED", + completedAt: new Date("2026-01-01T00:00:00.000Z"), + idempotencyKey: "idem_generated", + userProvidedIdempotencyKey: false, + inactiveIdempotencyKey: null, + idempotencyKeyExpiresAt: null, + completedByTaskRunId: null, + completedByBatchId: null, + completedAfter: null, + output: null, + outputType: "application/json", + outputIsError: false, + projectId: "proj_1", + environmentId: "env_1", + tags: [], + createdAt: new Date("2026-01-01T00:00:00.000Z"), + updatedAt: new Date("2026-01-01T00:00:00.000Z"), + ...overrides, + } as Waitpoint; +} + +// The WRITE side of the freeze: one Waitpoint row becomes one record. +function toRecord(w: Waitpoint): CompletedWaitpointRecord { + return { + id: w.id, + friendlyId: w.friendlyId, + type: w.type, + completedAt: (w.completedAt ?? new Date()).toISOString(), + outputType: w.outputType, + outputIsError: w.outputIsError, + output: recordOutputFor(w), + completedByTaskRunId: w.completedByTaskRunId ?? undefined, + completedByBatchId: w.completedByBatchId ?? undefined, + completedAfter: w.completedAfter?.toISOString(), + idempotencyKey: + w.userProvidedIdempotencyKey && !w.inactiveIdempotencyKey ? w.idempotencyKey : undefined, + }; +} + +function recordOutputFor(w: Waitpoint): CompletedWaitpointRecord["output"] { + if (w.output === null) return null; + // A RUN success re-derives byte-identically from TaskRun.output. A RUN error cannot, + // because TaskRun.error is jsonb, so it carries inline. + // + // This branch is deliberately BEFORE the application/store branch. An offloaded RUN + // success is still deriveFromRun, and that is correct: completeAttemptSuccess receives + // the same `output` and `outputType` the waitpoint got, so TaskRun.output holds the + // same ref string. The re-read stays byte-identical either way. + if (w.type === "RUN" && !w.outputIsError && w.completedByTaskRunId) + return { deriveFromRun: true }; + if (w.outputType === "application/store") return { ref: w.output }; + return { inline: w.output }; +} + +// The READ side of the freeze. Iterates `records`, never `order`. +async function referenceResolver( + args: ResolveCompletedWaitpointsArgs, + lookupRunOutput: (runId: string) => Promise +): Promise { + const out: CompletedWaitpoint[] = []; + for (const record of args.records) { + const indexes: (number | undefined)[] = []; + for (let i = 0; i < args.order.length; i++) { + if (args.order[i] === record.id) indexes.push(i); + } + if (indexes.length === 0) indexes.push(undefined); + + let output: string | undefined; + if (record.output === null) { + output = undefined; + } else if ("inline" in record.output) { + output = record.output.inline; + } else if ("ref" in record.output) { + output = record.output.ref; + } else if ("deriveFromRun" in record.output) { + output = record.completedByTaskRunId + ? await lookupRunOutput(record.completedByTaskRunId) + : undefined; + } else { + const _never: never = record.output; + throw new Error(`unknown record output variant: ${JSON.stringify(_never)}`); + } + + for (const index of indexes) { + out.push({ + id: record.id, + // Unreachable: the oracle's own loop pushes a non-negative integer or undefined. + // Reproduced because the frozen index-expansion rule names it. + index: index === -1 ? undefined : index, + friendlyId: record.friendlyId, + type: record.type, + completedAt: new Date(record.completedAt), + idempotencyKey: record.idempotencyKey, + completedByTaskRun: record.completedByTaskRunId + ? { + id: record.completedByTaskRunId, + friendlyId: RunId.toFriendlyId(record.completedByTaskRunId), + batch: args.batchId + ? { id: args.batchId, friendlyId: BatchId.toFriendlyId(args.batchId) } + : undefined, + } + : undefined, + completedAfter: record.completedAfter ? new Date(record.completedAfter) : undefined, + completedByBatch: record.completedByBatchId + ? { + id: record.completedByBatchId, + friendlyId: BatchId.toFriendlyId(record.completedByBatchId), + } + : undefined, + output, + outputType: record.outputType, + outputIsError: record.outputIsError, + }); + } + } + return out; +} + +// Proves the frozen hook signature is implementable exactly as declared. The reference +// resolver takes its TaskRun lookup as a second parameter, so the production shape is +// the curried form -- which is what the waitpoint lane will bind to a Prisma client. +// If this assignment stops compiling, the frozen signature has drifted. This file is +// typechecked by tsconfig.freeze-test.json, wired into this package's `typecheck` +// script, so that drift is caught -- vitest's esbuild transform alone would not catch it. +const resolverUnderTest: CompletedWaitpointResolver = (args) => + referenceResolver(args, async () => undefined); + +// The oracle spreads the snapshot, so it needs the two fields the mapping reads. +function makeSnapshot(batchId: string | null) { + return { id: "snap_1", runId: "run_1", batchId, checkpoint: null } as never; +} + +async function assertParity( + waitpoints: Waitpoint[], + order: string[], + batchId: string | null, + runOutputs: Record = {} +) { + const enhanced = enhanceExecutionSnapshotWithWaitpoints(makeSnapshot(batchId), waitpoints, order); + const args: ResolveCompletedWaitpointsArgs = { + runId: "run_1", + batchId: batchId ?? undefined, + pointer: { cycleSeq: 1, count: order.length }, + order, + records: waitpoints.map(toRecord), + }; + // count-carried-forward behaviour (order.length, not the record count) is covered by + // the run-store Redis suite, not here -- this line only constructs `args`, not asserts. + const resolved = await referenceResolver(args, async (id) => runOutputs[id]); + expect(resolved).toEqual(enhanced.completedWaitpoints); + return { enhanced, resolved }; +} + +describe("the frozen record shape", () => { + // Literals, not a mirror of the writer. A field rename or an encoding change must + // fail HERE, because the parity suite cannot see it. + it("pins the RUN record", () => { + expect( + toRecord( + makeWaitpoint({ + id: "wp_run", + friendlyId: "waitpoint_run", + type: "RUN", + completedByTaskRunId: "run_child", + output: '{"value":42}', + }) + ) + ).toEqual({ + id: "wp_run", + friendlyId: "waitpoint_run", + type: "RUN", + completedAt: "2026-01-01T00:00:00.000Z", + outputType: "application/json", + outputIsError: false, + output: { deriveFromRun: true }, + completedByTaskRunId: "run_child", + completedByBatchId: undefined, + completedAfter: undefined, + idempotencyKey: undefined, + }); + }); + + it("pins the BATCH record", () => { + expect( + toRecord( + makeWaitpoint({ + id: "wp_batch", + friendlyId: "waitpoint_batch", + type: "BATCH", + completedByBatchId: "batch_child", + output: "Batch waitpoint completed", + }) + ) + ).toEqual({ + id: "wp_batch", + friendlyId: "waitpoint_batch", + type: "BATCH", + completedAt: "2026-01-01T00:00:00.000Z", + outputType: "application/json", + outputIsError: false, + output: { inline: "Batch waitpoint completed" }, + completedByTaskRunId: undefined, + completedByBatchId: "batch_child", + completedAfter: undefined, + idempotencyKey: undefined, + }); + }); + + it("pins the DATETIME record", () => { + expect( + toRecord( + makeWaitpoint({ + id: "wp_dt", + friendlyId: "waitpoint_dt", + type: "DATETIME", + completedAfter: new Date("2026-02-02T00:00:00.000Z"), + }) + ) + ).toEqual({ + id: "wp_dt", + friendlyId: "waitpoint_dt", + type: "DATETIME", + completedAt: "2026-01-01T00:00:00.000Z", + outputType: "application/json", + outputIsError: false, + output: null, + completedByTaskRunId: undefined, + completedByBatchId: undefined, + completedAfter: "2026-02-02T00:00:00.000Z", + idempotencyKey: undefined, + }); + }); + + it("pins the MANUAL record, with a user idempotency key and an offloaded output", () => { + expect( + toRecord( + makeWaitpoint({ + id: "wp_manual", + friendlyId: "waitpoint_manual", + type: "MANUAL", + idempotencyKey: "idem_user", + userProvidedIdempotencyKey: true, + output: "s3://bucket/key", + outputType: "application/store", + }) + ) + ).toEqual({ + id: "wp_manual", + friendlyId: "waitpoint_manual", + type: "MANUAL", + completedAt: "2026-01-01T00:00:00.000Z", + outputType: "application/store", + outputIsError: false, + output: { ref: "s3://bucket/key" }, + completedByTaskRunId: undefined, + completedByBatchId: undefined, + completedAfter: undefined, + idempotencyKey: "idem_user", + }); + }); + + it("keeps an offloaded RUN success on deriveFromRun, not ref", () => { + // Branch precedence. TaskRun.output holds the same ref string, so the re-read is + // still byte-identical. A later edit that reorders the branches must fail here. + const record = toRecord( + makeWaitpoint({ + id: "wp_run_offloaded", + type: "RUN", + completedByTaskRunId: "run_child", + output: "s3://bucket/key", + outputType: "application/store", + }) + ); + expect(record.output).toEqual({ deriveFromRun: true }); + }); + + it("carries a RUN error inline, never deriveFromRun", () => { + const record = toRecord( + makeWaitpoint({ + id: "wp_run_err", + type: "RUN", + completedByTaskRunId: "run_child", + output: '{"type":"BUILT_IN_ERROR"}', + outputIsError: true, + }) + ); + expect(record.output).toEqual({ inline: '{"type":"BUILT_IN_ERROR"}' }); + }); +}); + +// The pointer's shape is pinned by CompletedWaitpointsPointer and tsconfig.freeze-test.json, +// not by a runtime assertion here -- a value that only echoes its own construction can't fail. + +describe("the completed-waitpoints freeze", () => { + it("expands a repeated id at each of its positions", async () => { + const w = makeWaitpoint({ id: "wp_a", type: "RUN", completedByTaskRunId: "run_child" }); + const { resolved } = await assertParity([w], ["wp_a", "wp_other", "wp_a"], "batch_1"); + expect(resolved.map((r) => r.index)).toEqual([0, 2]); + }); + + it("yields one entry with an undefined index for a record absent from order", async () => { + const w = makeWaitpoint({ id: "wp_absent", type: "MANUAL" }); + const { resolved } = await assertParity([w], ["wp_other"], null); + expect(resolved).toHaveLength(1); + expect(resolved[0]!.index).toBeUndefined(); + }); + + it("resolves a non-batch wait, where order is empty and one record exists", async () => { + // The commonest resume. Postgres's join holds the id while order does not, which is + // why `records` is authoritative and the mint comparison never reads `order`. + // batchId is null here on purpose: a triggerAndWait outside a batch is the shape + // this case is named for, and it exercises the oracle's `batchId ? ... : undefined` + // false branch, which no other case reaches. + // output stays null, so no TaskRun lookup is involved: both halves yield undefined. + const w = makeWaitpoint({ id: "wp_single", type: "RUN", completedByTaskRunId: "run_child" }); + const { resolved } = await assertParity([w], [], null); + expect(resolved).toHaveLength(1); + expect(resolved[0]!.index).toBeUndefined(); + expect(resolved[0]!.completedByTaskRun?.id).toBe("run_child"); + expect(resolved[0]!.completedByTaskRun?.batch).toBeUndefined(); + }); + + it("keys completedByBatch on the id alone, not on the type", async () => { + // The oracle checks completedByBatchId without looking at `type`, at + // executionSnapshotSystem.ts:107-113. A resolver keyed on type would pass every + // other case here and diverge in production. + const w = makeWaitpoint({ + id: "wp_manual_with_batch", + type: "MANUAL", + completedByBatchId: "batch_child", + }); + const { resolved } = await assertParity([w], ["wp_manual_with_batch"], null); + expect(resolved[0]!.completedByBatch?.id).toBe("batch_child"); + }); + + it("carries outputIsError on a non-RUN type", async () => { + const w = makeWaitpoint({ + id: "wp_manual_err", + type: "MANUAL", + output: '{"type":"STRING_ERROR"}', + outputIsError: true, + }); + const { resolved } = await assertParity([w], ["wp_manual_err"], null); + expect(resolved[0]!.outputIsError).toBe(true); + expect(resolved[0]!.output).toBe('{"type":"STRING_ERROR"}'); + }); + + it("resolves through the frozen hook signature", async () => { + // Exercises resolverUnderTest, so the declared CompletedWaitpointResolver type is + // proved implementable at runtime, on top of the compile-time proof at its + // declaration above (checked by tsconfig.freeze-test.json). + const w = makeWaitpoint({ id: "wp_hook", type: "MANUAL" }); + const resolved = await resolverUnderTest({ + runId: "run_1", + batchId: undefined, + pointer: { cycleSeq: 1, count: 1 }, + order: ["wp_hook"], + records: [toRecord(w)], + }); + expect(resolved).toHaveLength(1); + expect(resolved[0]!.id).toBe("wp_hook"); + expect(resolved[0]!.index).toBe(0); + }); + + it("round-trips all four waitpoint types", async () => { + const waitpoints = [ + makeWaitpoint({ id: "wp_run", type: "RUN", completedByTaskRunId: "run_child" }), + makeWaitpoint({ id: "wp_batch", type: "BATCH", completedByBatchId: "batch_child" }), + makeWaitpoint({ + id: "wp_dt", + type: "DATETIME", + completedAfter: new Date("2026-02-02T00:00:00.000Z"), + }), + makeWaitpoint({ id: "wp_manual", type: "MANUAL" }), + ]; + const { resolved } = await assertParity(waitpoints, ["wp_run", "wp_batch"], "batch_1"); + expect(resolved.map((r) => r.type)).toEqual(["RUN", "BATCH", "DATETIME", "MANUAL"]); + }); + + it("applies the idempotency-key rule in all four combinations", async () => { + const combos: Array<[boolean, string | null, string | undefined]> = [ + [true, null, "idem_user"], + [true, "cleared", undefined], + [false, null, undefined], + [false, "cleared", undefined], + ]; + for (const [userProvided, inactive, expected] of combos) { + const w = makeWaitpoint({ + id: "wp_idem", + idempotencyKey: "idem_user", + userProvidedIdempotencyKey: userProvided, + inactiveIdempotencyKey: inactive, + }); + const { resolved } = await assertParity([w], ["wp_idem"], null); + expect(resolved[0]!.idempotencyKey).toBe(expected); + } + }); + + it("forwards completedAfter on a MANUAL waitpoint with a timeout", async () => { + // The plan comment scopes completedAfter to DATETIME. The oracle forwards it for any + // type, so the resolver must too. + const w = makeWaitpoint({ + id: "wp_timeout", + type: "MANUAL", + completedAfter: new Date("2026-03-03T00:00:00.000Z"), + }); + const { resolved } = await assertParity([w], ["wp_timeout"], null); + expect(resolved[0]!.completedAfter).toEqual(new Date("2026-03-03T00:00:00.000Z")); + }); + + it("falls back off deriveFromRun when the completing run was deleted", async () => { + const w = makeWaitpoint({ + id: "wp_orphan", + type: "RUN", + completedByTaskRunId: null, + output: '{"value":42}', + }); + const { resolved } = await assertParity([w], ["wp_orphan"], null); + expect(resolved[0]!.output).toBe('{"value":42}'); + }); + + it("maps every output variant", async () => { + const runSuccess = makeWaitpoint({ + id: "wp_run_ok", + type: "RUN", + completedByTaskRunId: "run_ok", + output: '{"value":42}', + }); + const runError = makeWaitpoint({ + id: "wp_run_err", + type: "RUN", + completedByTaskRunId: "run_err", + output: '{"type":"BUILT_IN_ERROR"}', + outputIsError: true, + }); + const offloaded = makeWaitpoint({ + id: "wp_ref", + type: "MANUAL", + output: "s3://bucket/key", + outputType: "application/store", + }); + const empty = makeWaitpoint({ id: "wp_none", type: "MANUAL", output: null }); + + const { resolved } = await assertParity( + [runSuccess, runError, offloaded, empty], + ["wp_run_ok", "wp_run_err", "wp_ref", "wp_none"], + "batch_1", + // deriveFromRun: the same string TaskRun.output holds verbatim. + { run_ok: '{"value":42}' } + ); + expect(resolved.map((r) => r.output)).toEqual([ + '{"value":42}', + '{"type":"BUILT_IN_ERROR"}', + "s3://bucket/key", + undefined, + ]); + }); +}); + +describe("the exhaustive parity grid", () => { + // Dimensions mirror every Waitpoint column the oracle reads (type, output, outputType, + // outputIsError, completedByTaskRunId, completedByBatchId, completedAfter, + // userProvidedIdempotencyKey, inactiveIdempotencyKey), plus order-membership and the + // reading entry's batchId. A new column the oracle reads must widen a dimension here, + // so the pinned combination count below fails instead of coverage silently shrinking. + const TYPES: Waitpoint["type"][] = ["RUN", "BATCH", "DATETIME", "MANUAL"]; + const OUTPUTS: (string | null)[] = [null, '{"value":42}']; + const OUTPUT_TYPES = ["application/json", "application/store"]; + const OUTPUT_IS_ERRORS = [false, true]; + const TASK_RUN_IDS: (string | null)[] = [null, "run_child"]; + const BATCH_IDS: (string | null)[] = [null, "batch_child"]; + const COMPLETED_AFTERS: (Date | null)[] = [null, new Date("2026-02-02T00:00:00.000Z")]; + const IDEMPOTENCY_COMBOS: Array<[boolean, string | null]> = [ + [false, null], + [false, "cleared"], + [true, null], + [true, "cleared"], + ]; + const ORDER_MEMBERSHIPS = ["absent", "once", "twice"] as const; + const READING_BATCH_IDS: (string | null)[] = [null, "batch_reading_entry"]; + + // Only reached when type is RUN, output is set, outputIsError is false, and + // completedByTaskRunId is "run_child": the deriveFromRun branch. The value matches + // OUTPUTS' non-null entry so a correct resolver is byte-identical to the oracle. + const RUN_OUTPUT_LOOKUP: Record = { run_child: '{"value":42}' }; + + it("agrees with the oracle across every combination", async () => { + type Combo = { + type: Waitpoint["type"]; + output: string | null; + outputType: string; + outputIsError: boolean; + completedByTaskRunId: string | null; + completedByBatchId: string | null; + completedAfter: Date | null; + userProvidedIdempotencyKey: boolean; + inactiveIdempotencyKey: string | null; + orderMembership: (typeof ORDER_MEMBERSHIPS)[number]; + readingBatchId: string | null; + }; + const failures: Array<{ combo: Combo; oracle: unknown; resolver: unknown }> = []; + let cases = 0; + + for (const type of TYPES) { + for (const output of OUTPUTS) { + for (const outputType of OUTPUT_TYPES) { + for (const outputIsError of OUTPUT_IS_ERRORS) { + for (const completedByTaskRunId of TASK_RUN_IDS) { + for (const completedByBatchId of BATCH_IDS) { + for (const completedAfter of COMPLETED_AFTERS) { + for (const [ + userProvidedIdempotencyKey, + inactiveIdempotencyKey, + ] of IDEMPOTENCY_COMBOS) { + for (const orderMembership of ORDER_MEMBERSHIPS) { + for (const readingBatchId of READING_BATCH_IDS) { + cases++; + const combo: Combo = { + type, + output, + outputType, + outputIsError, + completedByTaskRunId, + completedByBatchId, + completedAfter, + userProvidedIdempotencyKey, + inactiveIdempotencyKey, + orderMembership, + readingBatchId, + }; + + const id = "wp_grid"; + const w = makeWaitpoint({ + id, + type, + output, + outputType, + outputIsError, + completedByTaskRunId, + completedByBatchId, + completedAfter, + idempotencyKey: "idem_user", + userProvidedIdempotencyKey, + inactiveIdempotencyKey, + }); + const order = + orderMembership === "absent" + ? ["wp_other"] + : orderMembership === "once" + ? [id] + : [id, id]; + + const enhanced = enhanceExecutionSnapshotWithWaitpoints( + makeSnapshot(readingBatchId), + [w], + order + ); + const args: ResolveCompletedWaitpointsArgs = { + runId: "run_1", + batchId: readingBatchId ?? undefined, + pointer: { cycleSeq: 1, count: order.length }, + order, + records: [toRecord(w)], + }; + const resolved = await referenceResolver( + args, + async (runId) => RUN_OUTPUT_LOOKUP[runId] + ); + + if (!isDeepStrictEqual(resolved, enhanced.completedWaitpoints)) { + failures.push({ + combo, + oracle: enhanced.completedWaitpoints, + resolver: resolved, + }); + } + } + } + } + } + } + } + } + } + } + } + + expect(cases).toBe(6144); + expect( + failures.length, + failures.length > 0 + ? `${failures.length}/${cases} combinations diverged. First: ${JSON.stringify(failures[0], null, 2)}` + : undefined + ).toBe(0); + }); +}); + +describe("the freeze's two deliberate divergences", () => { + it("pins completedAt at write time, where the oracle samples the clock", () => { + // The oracle applies `w.completedAt ?? new Date()`, so a null value changes on every + // read. No deterministic record can match that. The record pins it once instead. + const w = makeWaitpoint({ id: "wp_null_at", completedAt: null }); + const record = toRecord(w); + expect(Math.abs(Date.now() - new Date(record.completedAt).getTime())).toBeLessThan(60_000); + }); + + it("takes batch{} from the reading entry, not the completing run's own batch", async () => { + // A known, deliberate conflation in the oracle. Byte-compatibility requires it. + const w = makeWaitpoint({ id: "wp_run", type: "RUN", completedByTaskRunId: "run_child" }); + const { resolved } = await assertParity([w], ["wp_run"], "batch_reading_entry"); + expect(resolved[0]!.completedByTaskRun?.batch?.id).toBe("batch_reading_entry"); + }); +}); diff --git a/internal-packages/run-engine/src/engine/systems/executionSnapshotSystem.ts b/internal-packages/run-engine/src/engine/systems/executionSnapshotSystem.ts index e79383a8bb6..81c41d2c2ae 100644 --- a/internal-packages/run-engine/src/engine/systems/executionSnapshotSystem.ts +++ b/internal-packages/run-engine/src/engine/systems/executionSnapshotSystem.ts @@ -58,7 +58,7 @@ function enhanceExecutionSnapshot( * Transforms a snapshot (with checkpoint but without waitpoints) into an EnhancedExecutionSnapshot * by combining it with pre-fetched waitpoints. */ -function enhanceExecutionSnapshotWithWaitpoints( +export function enhanceExecutionSnapshotWithWaitpoints( snapshot: ExecutionSnapshotWithCheckpoint, waitpoints: Waitpoint[], completedWaitpointOrder: string[] diff --git a/internal-packages/run-engine/tsconfig.freeze-test.json b/internal-packages/run-engine/tsconfig.freeze-test.json new file mode 100644 index 00000000000..284a132e293 --- /dev/null +++ b/internal-packages/run-engine/tsconfig.freeze-test.json @@ -0,0 +1,19 @@ +// Typechecks completedWaitpointFreeze.test.ts, which tsconfig.build.json otherwise excludes +// (src/**/*.test.ts) and vitest's esbuild transform never checks. This config has no +// "@triggerdotdev/source" customCondition, so it resolves @internal/run-store from its built +// `dist`, not from source -- same as tsconfig.build.json. That means this gate only sees a +// source change in run-store once run-store has been rebuilt, so it MUST be run through turbo +// (`pnpm run typecheck --filter @internal/run-engine`), whose `typecheck` task declares +// `dependsOn: ["^build"]`. Running `tsc -p tsconfig.freeze-test.json` (or `pnpm run typecheck`) +// directly inside this package, against a stale dist/, passes green while the frozen type has +// already drifted in source. Do not "fix" this with customConditions: that pulls +// @trigger.dev/core's source in too, which fails to typecheck here on `lib: ES2020`. +{ + "extends": "./tsconfig.build.json", + "include": ["src/engine/systems/completedWaitpointFreeze.test.ts"], + "exclude": [], + "compilerOptions": { + "composite": false, + "declaration": false + } +} diff --git a/internal-packages/run-store/src/redisSnapshotStore.test.ts b/internal-packages/run-store/src/redisSnapshotStore.test.ts index 369d79e5338..0c7b4c0720a 100644 --- a/internal-packages/run-store/src/redisSnapshotStore.test.ts +++ b/internal-packages/run-store/src/redisSnapshotStore.test.ts @@ -10,6 +10,8 @@ import { isValidFor, RedisSnapshotStore, type SnapshotEntryInput, + type CompletedWaitpointsPointer, + type CompletedWaitpointRecord, } from "./redisSnapshotStore.js"; describe("snapshotKeys", () => { @@ -236,6 +238,145 @@ describe("append", () => { } ); + redisTest( + "round-trips a typed records array through the cycle hash's records field", + async ({ redisOptions }) => { + // The only place CompletedWaitpointRecord[] physically enters Redis. If the writer ever + // serializes a different envelope, this is where that would show up as a broken round trip. + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 }); + const raw = createRedisClient(redisOptions); + const records: CompletedWaitpointRecord[] = [ + { + id: "w_a", + friendlyId: "waitpoint_a", + type: "RUN", + completedAt: "2026-01-01T00:00:00.000Z", + outputType: "application/json", + outputIsError: false, + output: { deriveFromRun: true }, + completedByTaskRunId: "run_child", + }, + ]; + try { + await store.append({ + entry: entry({ id: "snap_1" }), + kind: "birth", + isTerminal: false, + cycle: { + kind: "new", + completedWaitpoints: [{ id: "w_a", index: 0 }], + records, + }, + }); + + const storedRaw = await raw.hget("snap:{run_1}:wp:1", "records"); + expect(JSON.parse(storedRaw!)).toEqual(records); + } finally { + raw.disconnect(); + await store.quit(); + } + } + ); + + redisTest( + "a recordless new cycle clears another cycle's records off a reused key", + async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 }); + const raw = createRedisClient(redisOptions); + try { + await store.append({ + entry: entry({ id: "snap_1" }), + kind: "birth", + isTerminal: false, + cycle: { + kind: "new", + completedWaitpoints: [{ id: "w_a", index: 0 }], + records: [ + { + id: "w_a", + friendlyId: "waitpoint_a", + type: "MANUAL", + completedAt: "2026-01-01T00:00:00.000Z", + outputType: "application/json", + outputIsError: false, + output: { inline: "stale" }, + }, + ], + }, + }); + expect(await raw.hget("snap:{run_1}:wp:1", "records")).not.toBeNull(); + + // Only the counter is lost, as under maxmemory eviction. A birth does not check seq, so + // the next new cycle re-mints cycleSeq 1 onto the surviving key. + await raw.del("snap:{run_1}:seq"); + + await store.append({ + entry: entry({ id: "snap_2" }), + kind: "birth", + isTerminal: false, + cycle: { kind: "new", completedWaitpoints: [{ id: "w_b", index: 0 }] }, + }); + + expect(await raw.hget("snap:{run_1}:wp:1", "order")).toBe(JSON.stringify(["w_b"])); + expect(await raw.hget("snap:{run_1}:wp:1", "records")).toBeNull(); + } finally { + raw.disconnect(); + await store.quit(); + } + } + ); + + redisTest( + "a carry-forward refuses a cycle this incarnation never minted", + async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 60_000 }); + const raw = createRedisClient(redisOptions); + try { + await store.append({ + entry: entry({ id: "snap_1" }), + kind: "birth", + isTerminal: false, + cycle: { + kind: "new", + completedWaitpoints: [{ id: "w_old", index: 0 }], + records: [ + { + id: "w_old", + friendlyId: "waitpoint_old", + type: "MANUAL", + completedAt: "2026-01-01T00:00:00.000Z", + outputType: "application/json", + outputIsError: false, + output: { inline: "stale" }, + }, + ], + }, + }); + + // Lose the whole keyspace except the cycle key, as under maxmemory eviction. + 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 }, + }); + + // Written, flagged, and carrying NO pointer: the dead incarnation's waitpoints must not + // be served to a fresh run under a count that agrees with them. + expect(carried).toMatchObject({ outcome: "written", cycleMismatch: true }); + expect(carried).not.toHaveProperty("cycleSeq"); + const read = await store.getLatest("run_1"); + expect(read?.cycle).toBeUndefined(); + expect(read?.completedWaitpointIds).toBeUndefined(); + } finally { + raw.disconnect(); + await store.quit(); + } + } + ); + redisTest( "reports a duplicate id without overwriting the original entry", async ({ redisOptions }) => { @@ -384,6 +525,129 @@ describe("cycle keys", () => { }); }); +describe("cycle key deletion sweep", () => { + const KEY_SUFFIXES = ["e", "idx", "cur", "seq", "wp:1", "wp:2"] as const; + const INJECTION_POINTS = ["beforeCarryForward", "beforeSecondNewCycle"] as const; + + function powerset(items: readonly T[]): T[][] { + let out: T[][] = [[]]; + for (const item of items) { + out = out.concat(out.map((s) => [...s, item])); + } + return out; + } + + // Mechanized replacement for the two hand-picked eviction regressions above: replays the same + // birth/carryForward/new-cycle/terminal shape once per element of the powerset of key deletions, + // at two points in the sequence, and checks that a cycle key's records never leak into a pointer + // naming a different cycle than the one the key's own order field currently describes. + redisTest( + "records read back for a pointer never mention an id outside that pointer's own order", + async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 60_000 }); + const raw = createRedisClient(redisOptions); + type Violation = { + subset: string[]; + injectionPoint: string; + runId: string; + pointer: CompletedWaitpointsPointer; + recordsFound: unknown; + }; + const violations: Violation[] = []; + let replays = 0; + + try { + for (const injectionPoint of INJECTION_POINTS) { + for (const subset of powerset(KEY_SUFFIXES)) { + replays++; + const runId = `run_sweep_${injectionPoint}_${replays}`; + const base = `snap:{${runId}}`; + const damage = async () => { + if (subset.length > 0) { + await raw.del(...subset.map((s) => `${base}:${s}`)); + } + }; + + await store.append({ + entry: entry({ id: "snap_1", runId }), + kind: "birth", + isTerminal: false, + cycle: { + kind: "new", + completedWaitpoints: [{ id: "w_c1", index: 0 }], + records: [ + { + id: "w_c1", + friendlyId: "waitpoint_c1", + type: "MANUAL", + completedAt: "2026-01-01T00:00:00.000Z", + outputType: "application/json", + outputIsError: false, + output: { inline: "payload_c1" }, + }, + ], + }, + }); + + if (injectionPoint === "beforeCarryForward") await damage(); + await store.append({ + entry: entry({ id: "snap_2", runId }), + kind: "transition", + isTerminal: false, + cycle: { kind: "carryForward", cycleSeq: 1 }, + }); + + if (injectionPoint === "beforeSecondNewCycle") await damage(); + // birth, not transition: both eviction regressions above only reproduce past a birth's + // liveness bypass -- a transition here would just report skippedNoKeyspace once e or seq + // is gone, exempting the replay before the cycle-mint branch ever ran. + await store.append({ + entry: entry({ id: "snap_3", runId }), + kind: "birth", + isTerminal: false, + cycle: { kind: "new", completedWaitpoints: [{ id: "w_c2", index: 0 }] }, + }); + + await store.append({ + entry: entry({ id: "snap_4", runId, executionStatus: "FINISHED" }), + kind: "transition", + isTerminal: true, + }); + + for (const id of ["snap_1", "snap_2", "snap_3", "snap_4"]) { + const read = await store.getById(runId, id); + if (!read?.cycle) continue; + const orderIds = new Set(read.completedWaitpointIds?.order ?? []); + const recordsRaw = await raw.hget(`${base}:wp:${read.cycle.cycleSeq}`, "records"); + if (recordsRaw === null) continue; + const recordIds = (JSON.parse(recordsRaw) as { id: string }[]).map((r) => r.id); + if (recordIds.some((rid) => !orderIds.has(rid))) { + violations.push({ + subset, + injectionPoint, + runId, + pointer: read.cycle, + recordsFound: recordIds, + }); + } + } + } + } + } finally { + raw.disconnect(); + await store.quit(); + } + + if (violations.length > 0) { + throw new Error( + `${violations.length} violation(s) across ${replays} replays. First: ` + + JSON.stringify(violations[0]) + ); + } + } + ); +}); + describe("read-side cycle mismatch", () => { redisTest( "warns and records a metric when a cycle's count disagrees with its order", @@ -1113,6 +1377,53 @@ describe("hash tag and keyPrefix", () => { }); describe("observability", () => { + redisTest("cycle-key bytes cover records, not just order", async ({ redisOptions }) => { + // The plan puts a metric on the wp: KEY size. records dominates that key once + // populated, so measuring order alone understates it by orders of magnitude. + const calls: Array<[string, number]> = []; + const metrics = { + recordAppend: () => {}, + recordEntryBytes: () => {}, + recordCycleKeyBytes: (b: number) => calls.push(["cycleBytes", b]), + recordCycleCount: () => {}, + recordSkippedNoKeyspace: () => {}, + recordCycleMismatch: () => {}, + recordLatency: () => {}, + }; + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 60_000, metrics }); + try { + const records: CompletedWaitpointRecord[] = [ + { + id: "w_a", + friendlyId: "waitpoint_a", + type: "MANUAL", + completedAt: "2026-01-01T00:00:00.000Z", + outputType: "application/json", + outputIsError: false, + output: { inline: "y".repeat(20_000) }, + }, + ]; + const orderJson = JSON.stringify(["w_a"]); + const recordsJson = JSON.stringify(records); + + await store.append({ + entry: entry({ id: "snap_1" }), + kind: "birth", + isTerminal: false, + cycle: { kind: "new", completedWaitpoints: [{ id: "w_a", index: 0 }], records }, + }); + + expect(calls).toEqual([ + [ + "cycleBytes", + Buffer.byteLength(orderJson, "utf8") + Buffer.byteLength(recordsJson, "utf8"), + ], + ]); + } finally { + await store.quit(); + } + }); + redisTest("records sizes and outcomes without ever rejecting", async ({ redisOptions }) => { const calls: unknown[][] = []; const metrics = { @@ -1224,3 +1535,40 @@ describe("observability", () => { } ); }); + +describe("the reserved completedWaitpoints field", () => { + redisTest("append rejects an entry that sets it", async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 60_000 }); + try { + const runId = "run_reserved_throw"; + const pointer: CompletedWaitpointsPointer = { cycleSeq: 1, count: 0 }; + await expect( + store.append({ + entry: { ...entry({ id: "snap_1", runId }), completedWaitpoints: pointer }, + kind: "birth", + isTerminal: false, + }) + ).rejects.toThrow(/reserved/i); + } finally { + await store.quit(); + } + }); + + redisTest("a stored entry never holds the key", async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 60_000 }); + try { + const runId = "run_reserved_absent"; + await store.append({ + entry: entry({ id: "snap_1", runId }), + kind: "birth", + isTerminal: false, + }); + const read = await store.getLatest(runId); + expect(read).not.toBeNull(); + expect(read!.raw).not.toContain("completedWaitpoints"); + expect(read!.entry).not.toHaveProperty("completedWaitpoints"); + } finally { + await store.quit(); + } + }); +}); diff --git a/internal-packages/run-store/src/redisSnapshotStore.ts b/internal-packages/run-store/src/redisSnapshotStore.ts index 7b60843e2b4..2964959c4fe 100644 --- a/internal-packages/run-store/src/redisSnapshotStore.ts +++ b/internal-packages/run-store/src/redisSnapshotStore.ts @@ -6,6 +6,7 @@ import { type Result, } from "@internal/redis"; import { Logger } from "@trigger.dev/core/logger"; +import type { CompletedWaitpoint } from "@trigger.dev/core/v3/schemas"; export type SnapshotKeys = { e: string; idx: string; cur: string; seq: string }; @@ -33,6 +34,88 @@ export function isValidFor(entry: { error?: unknown }): boolean { return !entry.error; } +// --------------------------------------------------------------------------- +// The completed-waitpoints freeze. Frozen jointly with the waitpoint lane. +// Do not change a field here without re-agreeing the contract with that lane. +// --------------------------------------------------------------------------- + +/** + * The once-per-wait-cycle pointer. `cycleSeq` names the snap:{runId}:wp: + * key. `count` is order.length -- NOT the record count -- so it is zero for any + * wait that carries no batch index. + */ +export type CompletedWaitpointsPointer = { + cycleSeq: number; + count: number; +}; + +/** + * A record's output. + * - `inline` holds the literal value. MANUAL and DATETIME are bounded by the offload + * thresholds; error outputs are not (only BUILT_IN_ERROR truncates), so the bound is + * the completion body limit. Postgres holds the same strings, so this is a copy. + * - `ref` holds an application/store reference that was already offloaded. + * - `deriveFromRun` means the resolver reads TaskRun.output for completedByTaskRunId. + * Only a RUN record with outputIsError false AND a non-null completedByTaskRunId uses + * it: TaskRun.output is a String column holding the same string verbatim, so the + * re-read is byte-identical. A RUN error cannot use it, because TaskRun.error is + * jsonb and never round-trips. Waitpoint.completedByTaskRun is onDelete: SetNull, so + * an orphaned RUN waitpoint (the completing run row was deleted) has no run left to + * derive from -- its output carries inline instead. + */ +export type CompletedWaitpointRecordOutput = + | { inline: string } + | { ref: string } + | { deriveFromRun: true } + | null; + +/** + * One completed waitpoint, one per DISTINCT id in a wait cycle. The resolver expands + * this into one CompletedWaitpoint per position of the id in the cycle's order list. + */ +export type CompletedWaitpointRecord = { + id: string; + friendlyId: string; + type: "RUN" | "BATCH" | "DATETIME" | "MANUAL"; + /** ISO. The writer pins it, applying the null fallback once. */ + completedAt: string; + /** Defaults to "application/json" at source. */ + outputType: string; + outputIsError: boolean; + output: CompletedWaitpointRecordOutput; + /** RUN. The resolver derives friendlyId, and batch{} from the READING entry's batchId. */ + completedByTaskRunId?: string; + /** BATCH. The resolver derives friendlyId. */ + completedByBatchId?: string; + /** ISO. Any type may set it: a MANUAL waitpoint with a timeout does. */ + completedAfter?: string; + /** Already resolved: userProvidedIdempotencyKey && !inactiveIdempotencyKey. */ + idempotencyKey?: string; +}; + +/** + * What the store hands the resolver. The store owns the keyspace, so the store reads + * and parses the cycle hash. The resolver never touches Redis and never derives a key. + */ +export type ResolveCompletedWaitpointsArgs = { + runId: string; + /** The batchId of the entry being READ, never the entry that minted the cycle. */ + batchId?: string; + pointer: CompletedWaitpointsPointer; + /** Index oracle only. A SUBSET of the record ids. Repeats preserved. */ + order: string[]; + /** The authoritative, complete set. Iterate this, never `order`. */ + records: CompletedWaitpointRecord[]; +}; + +/** + * This lane owns the signature. The waitpoint lane owns the implementation, which + * lives in run-engine because a deriveFromRun record needs a Postgres read. + */ +export type CompletedWaitpointResolver = ( + args: ResolveCompletedWaitpointsArgs +) => Promise; + export type SnapshotEntryInput = { id: string; engine: "V2"; @@ -53,6 +136,16 @@ export type SnapshotEntryInput = { runnerId?: string; metadata?: unknown; error?: string; + /** + * RESERVED. Always unset. `append` rejects a set value. + * + * The pointer's physical form is the `#c` sidecar field on the `e` hash, + * because the append Lua mints both halves after the client serializes the entry. + * The entry JSON must stay byte-identical to the caller's document, and the Postgres + * snapshot row has no pointer column, so a pointer inside the JSON would stop the two + * documents from being comparable for the dual-write comparator. + */ + completedWaitpoints?: CompletedWaitpointsPointer; }; export type WaitpointIds = { present: boolean; distinctIds: string[]; order: string[] }; @@ -67,7 +160,7 @@ export type SnapshotRead = { isValid: boolean; entry: Record; raw: string; - cycle?: { cycleSeq: number; count: number }; + cycle?: CompletedWaitpointsPointer; completedWaitpointIds?: WaitpointIds; }; @@ -155,9 +248,20 @@ export class RedisSnapshotStore { isTerminal: boolean; expectedCur?: string; cycle?: - | { kind: "new"; completedWaitpoints: CompletedWaitpointRef[]; records?: string } + | { + kind: "new"; + completedWaitpoints: CompletedWaitpointRef[]; + records?: CompletedWaitpointRecord[]; + } | { kind: "carryForward"; cycleSeq: number }; }): Promise { + if (args.entry.completedWaitpoints !== undefined) { + throw new Error( + "completedWaitpoints is a reserved entry field and must stay unset. The pointer's " + + "physical form is the `#c` sidecar field, which the append script mints. " + + "Writing it into the entry JSON breaks byte-comparability with the Postgres row." + ); + } return this.#timed("append", async () => { const k = snapshotKeys(args.entry.runId); const raw = JSON.stringify(args.entry); @@ -172,7 +276,7 @@ export class RedisSnapshotStore { const order = deriveOrder(args.cycle.completedWaitpoints); cycleMode = "new"; orderJson = JSON.stringify(order); - records = args.cycle.records ?? ""; + records = args.cycle.records ? JSON.stringify(args.cycle.records) : ""; orderCount = String(order.length); } else if (args.cycle?.kind === "carryForward") { cycleMode = "carry"; @@ -199,11 +303,17 @@ export class RedisSnapshotStore { args.expectedCur !== undefined ? "1" : "0" )) as string[]; - return this.#interpretAppend(reply, raw, orderJson, args.entry.runId); + return this.#interpretAppend(reply, raw, orderJson, records, args.entry.runId); }); } - #interpretAppend(reply: string[], raw: string, orderJson: string, runId: string): AppendResult { + #interpretAppend( + reply: string[], + raw: string, + orderJson: string, + records: string, + runId: string + ): AppendResult { if (reply[0] === SKIPPED) { this.metrics?.recordSkippedNoKeyspace(); this.metrics?.recordAppend("skippedNoKeyspace", "none"); @@ -224,7 +334,7 @@ export class RedisSnapshotStore { if (cycleMismatch) { this.metrics?.recordCycleMismatch(); } - this.#observeSizes(raw, orderJson, cycleSeq, runId); + this.#observeSizes(raw, orderJson, records, cycleSeq, runId); this.metrics?.recordAppend("written", ttl); return { outcome: "written", @@ -235,14 +345,21 @@ export class RedisSnapshotStore { }; } - #observeSizes(raw: string, orderJson: string, cycleSeq: number, runId: string): void { + #observeSizes( + raw: string, + orderJson: string, + records: string, + cycleSeq: number, + runId: string + ): void { const entryBytes = Buffer.byteLength(raw, "utf8"); this.metrics?.recordEntryBytes(entryBytes); if (this.highWater.entryBytes !== undefined && entryBytes > this.highWater.entryBytes) { this.logger.warn("RedisSnapshotStore entry above high-water mark", { runId, entryBytes }); } if (orderJson !== "") { - const cycleBytes = Buffer.byteLength(orderJson, "utf8"); + // The whole wp: key, not just its order field: records dominates it once populated. + const cycleBytes = Buffer.byteLength(orderJson, "utf8") + Buffer.byteLength(records, "utf8"); this.metrics?.recordCycleKeyBytes(cycleBytes); if (this.highWater.cycleKeyBytes !== undefined && cycleBytes > this.highWater.cycleKeyBytes) { this.logger.warn("RedisSnapshotStore cycle key above high-water mark", { @@ -469,13 +586,22 @@ export class RedisSnapshotStore { redis.call('HSET', wpKey(cycleSeq), 'order', orderJson, 'count', orderCount) if records ~= '' then redis.call('HSET', wpKey(cycleSeq), 'records', records) + else + -- A new cycle owns the whole key: a lost seq counter can re-mint a cycleSeq whose key + -- still holds another cycle's records, and order/count stay mutually consistent so the + -- mismatch check cannot see it. No-op on a fresh key. + redis.call('HDEL', wpKey(cycleSeq), 'records') end elseif cycleMode == 'carry' then - cycleSeq = cycleSeqIn - local c = redis.call('HGET', wpKey(cycleSeq), 'count') - if not c then + -- Attach a pointer only if this incarnation actually minted the cycle. seq can be + -- evicted while a wp: key survives, so a bare key-exists check would adopt a dead + -- incarnation's order and records under a consistent count, invisibly. + local minted = tonumber(redis.call('HGET', seqKey, 'c') or '0') + local c = redis.call('HGET', wpKey(cycleSeqIn), 'count') + if not c or minted < cycleSeqIn then mismatch = 1 else + cycleSeq = cycleSeqIn orderCount = c end end