From 099a079254a19879a022bbe42d95bfb357022504 Mon Sep 17 00:00:00 2001 From: Daniel Sutton Date: Mon, 24 Aug 2026 11:16:00 +0100 Subject: [PATCH] test(run-store,run-engine): sweep keyspace damage, pin the frozen key sets Two bugs where a surviving cycle key no longer belonged to the cycle its pointer named were both found by hand. Replaces hand-picked damage cases with a sweep: the powerset of deletions over the five key shapes, across two injection points, 128 replays, asserting that records read back for a pointer never mention an id outside that pointer's own order. Reverting the HDEL in the new-cycle branch makes it report 56 violations. Also pins the frozen key sets exactly and bidirectionally. Renames, removals, widenings and required-to-optional already broke compilation through the usage sites; an added optional field did not, and on a jointly-owned frozen type that is the change neither lane may make alone. --- .../systems/completedWaitpointFreeze.test.ts | 29 +++++ .../run-store/src/redisSnapshotStore.test.ts | 123 ++++++++++++++++++ 2 files changed, 152 insertions(+) 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 f3eab876a..a1af90582 100644 --- a/internal-packages/run-engine/src/engine/systems/completedWaitpointFreeze.test.ts +++ b/internal-packages/run-engine/src/engine/systems/completedWaitpointFreeze.test.ts @@ -11,8 +11,37 @@ 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 { diff --git a/internal-packages/run-store/src/redisSnapshotStore.test.ts b/internal-packages/run-store/src/redisSnapshotStore.test.ts index 39004b81a..0c7b4c072 100644 --- a/internal-packages/run-store/src/redisSnapshotStore.test.ts +++ b/internal-packages/run-store/src/redisSnapshotStore.test.ts @@ -525,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",