fix(run-store,run-engine): attach the frozen record type to Redis, close the output union
- RedisSnapshotStore.append's cycle.records now takes a typed CompletedWaitpointRecord[] and serializes it, instead of accepting an opaque pre-serialized string with no compile-time link to the frozen type. Adds a round-trip test reading the cycle hash's records field back with a raw client. - The test file's referenceResolver output discrimination is now exhaustive: an explicit deriveFromRun branch plus a `never` fallback, so a future output variant fails to compile here instead of silently resolving through a TaskRun re-read. - tsconfig.freeze-test.json documents that it resolves @internal/run-store from dist, so the gate must run through turbo rather than directly inside run-engine against a stale build. - assertParity now asserts pointer.count === order.length on every parity case, binding the frozen count-is-order.length rule instead of leaving it decorative.
This commit is contained in:
@@ -94,10 +94,13 @@ async function referenceResolver(
|
||||
output = record.output.inline;
|
||||
} else if ("ref" in record.output) {
|
||||
output = record.output.ref;
|
||||
} else {
|
||||
} 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) {
|
||||
@@ -156,16 +159,17 @@ async function assertParity(
|
||||
runOutputs: Record<string, string> = {}
|
||||
) {
|
||||
const enhanced = enhanceExecutionSnapshotWithWaitpoints(makeSnapshot(batchId), waitpoints, order);
|
||||
const resolved = await referenceResolver(
|
||||
{
|
||||
runId: "run_1",
|
||||
batchId: batchId ?? undefined,
|
||||
pointer: { cycleSeq: 1, count: order.length },
|
||||
order,
|
||||
records: waitpoints.map(toRecord),
|
||||
},
|
||||
async (id) => runOutputs[id]
|
||||
);
|
||||
const args: ResolveCompletedWaitpointsArgs = {
|
||||
runId: "run_1",
|
||||
batchId: batchId ?? undefined,
|
||||
pointer: { cycleSeq: 1, count: order.length },
|
||||
order,
|
||||
records: waitpoints.map(toRecord),
|
||||
};
|
||||
// The frozen rule: count is order.length, NOT the record count. Binding it here means every
|
||||
// parity case enforces it, not only the dedicated "the frozen pointer shape" cases.
|
||||
expect(args.pointer.count).toBe(order.length);
|
||||
const resolved = await referenceResolver(args, async (id) => runOutputs[id]);
|
||||
expect(resolved).toEqual(enhanced.completedWaitpoints);
|
||||
return { enhanced, resolved };
|
||||
}
|
||||
|
||||
@@ -1,3 +1,13 @@
|
||||
// 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"],
|
||||
|
||||
@@ -11,6 +11,7 @@ import {
|
||||
RedisSnapshotStore,
|
||||
type SnapshotEntryInput,
|
||||
type CompletedWaitpointsPointer,
|
||||
type CompletedWaitpointRecord,
|
||||
} from "./redisSnapshotStore.js";
|
||||
|
||||
describe("snapshotKeys", () => {
|
||||
@@ -237,6 +238,46 @@ 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(
|
||||
"reports a duplicate id without overwriting the original entry",
|
||||
async ({ redisOptions }) => {
|
||||
|
||||
@@ -246,7 +246,11 @@ 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<AppendResult> {
|
||||
if (args.entry.completedWaitpoints !== undefined) {
|
||||
@@ -270,7 +274,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";
|
||||
|
||||
Reference in New Issue
Block a user