Skip to content

Commit aee3f07

Browse files
d-csd-cs
committed
fix(run-store): give a snapshot one identity and one instant across both stores
Three defects, all of which passed the existing suites because no test drove a snapshot that actually carried waitpoints, and because the parity suite compared createdAt against a value it had just read back from the row. The decorator never passed a cycle to the append, so no wp:<cycleSeq> key was written for any snapshot and the completed-waitpoint side of Redis was permanently empty. It now mints a cycle when the id set differs from the current head and carries the previous cycleSeq forward when it does not, so a resume writes the record set once and the copy-forwards that follow write no key at all. The since-window hydration returned an empty completedWaitpointOrder. That column is not the join: the engine reads it off the head row as the oracle that gives each completed waitpoint its position in a batch, so an empty order resumed every batched triggerAndWait with an undefined index. Seven of the eight write sites stamped the entry from the app clock while Postgres stamped its own column default, so the two stores held different instants for one snapshot. The decorator now supplies createdAt, and an equal updatedAt, at every site, and the standalone path supplies it too rather than reading the row back. Beyond making the field comparable, this aligns the since-window: the cursor is resolved from one store and applied in the other, and two different instants misfilter that window. The parity suite gains an independent clock-provenance guard, and a case proving an absent instant still takes the database default, which is what keeps the store's behaviour unchanged while the decorator is off.
1 parent 8119466 commit aee3f07

16 files changed

Lines changed: 1269 additions & 175 deletions

‎internal-packages/run-engine/src/engine/tests/snapshotStoreChaos.test.ts‎

Lines changed: 400 additions & 0 deletions
Large diffs are not rendered by default.

‎internal-packages/run-engine/src/engine/tests/snapshotStoreReadGate.test.ts‎

Lines changed: 43 additions & 37 deletions
Original file line numberDiff line numberDiff line change
@@ -87,7 +87,12 @@ describe("snapshot store read gate", () => {
8787
await engine.completeRunAttempt({
8888
runId: run.id,
8989
snapshotId: attempt.snapshot.id,
90-
completion: { ok: true, id: run.id, output: `{"done":true}`, outputType: "application/json" },
90+
completion: {
91+
ok: true,
92+
id: run.id,
93+
output: `{"done":true}`,
94+
outputType: "application/json",
95+
},
9196
});
9297

9398
const finished = await prisma.taskRun.findFirstOrThrow({ where: { id: run.id } });
@@ -186,9 +191,7 @@ describe("snapshot store read gate", () => {
186191
.catch((error: unknown) => ({ threw: (error as Error).constructor.name }));
187192

188193
const postgresOnly = buildDecoratedStore({ prisma, redisOptions, mode: "off" });
189-
const engineOff = new RunEngine(
190-
engineOptions(prisma, redisOptions, postgresOnly) as never
191-
);
194+
const engineOff = new RunEngine(engineOptions(prisma, redisOptions, postgresOnly) as never);
192195
let viaPostgres: unknown;
193196
try {
194197
viaPostgres = await engineOff
@@ -260,39 +263,42 @@ describe("snapshot store read gate", () => {
260263
}
261264
);
262265

263-
containerTest("falls back to Postgres for a pre-cutover run", async ({ prisma, redisOptions }) => {
264-
// A run created while the dial was off has no keyspace. Turning reads on must not lose it.
265-
const off = buildDecoratedStore({ prisma, redisOptions, mode: "off" });
266-
const engineOff = new RunEngine(engineOptions(prisma, redisOptions, off) as never);
267-
268-
let runId: string;
269-
let environment: any;
270-
try {
271-
environment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
272-
await setupBackgroundWorker(engineOff, environment, "gate-task");
273-
const run = await engineOff.trigger(triggerArgs("gate-task", environment, 5), prisma);
274-
runId = run.id;
275-
await setTimeout(500);
276-
} finally {
277-
await engineOff.quit();
278-
await off.quit();
279-
}
266+
containerTest(
267+
"falls back to Postgres for a pre-cutover run",
268+
async ({ prisma, redisOptions }) => {
269+
// A run created while the dial was off has no keyspace. Turning reads on must not lose it.
270+
const off = buildDecoratedStore({ prisma, redisOptions, mode: "off" });
271+
const engineOff = new RunEngine(engineOptions(prisma, redisOptions, off) as never);
280272

281-
const on = buildDecoratedStore({
282-
prisma,
283-
redisOptions,
284-
mode: "redis-read",
285-
readPercent: 100,
286-
});
287-
const engineOn = new RunEngine(engineOptions(prisma, redisOptions, on) as never);
288-
try {
289-
const data = await engineOn.getRunExecutionData({ runId });
290-
assertNonNullable(data);
291-
expect(data.snapshot.executionStatus).toBe("QUEUED");
292-
expect(on.reads.some((r) => r.source === "postgres")).toBe(true);
293-
} finally {
294-
await engineOn.quit();
295-
await on.quit();
273+
let runId: string;
274+
let environment: any;
275+
try {
276+
environment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
277+
await setupBackgroundWorker(engineOff, environment, "gate-task");
278+
const run = await engineOff.trigger(triggerArgs("gate-task", environment, 5), prisma);
279+
runId = run.id;
280+
await setTimeout(500);
281+
} finally {
282+
await engineOff.quit();
283+
await off.quit();
284+
}
285+
286+
const on = buildDecoratedStore({
287+
prisma,
288+
redisOptions,
289+
mode: "redis-read",
290+
readPercent: 100,
291+
});
292+
const engineOn = new RunEngine(engineOptions(prisma, redisOptions, on) as never);
293+
try {
294+
const data = await engineOn.getRunExecutionData({ runId });
295+
assertNonNullable(data);
296+
expect(data.snapshot.executionStatus).toBe("QUEUED");
297+
expect(on.reads.some((r) => r.source === "postgres")).toBe(true);
298+
} finally {
299+
await engineOn.quit();
300+
await on.quit();
301+
}
296302
}
297-
});
303+
);
298304
});

‎internal-packages/run-store/scripts/generateDelegatingRunStore.ts‎

Lines changed: 7 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -176,10 +176,14 @@ export class DelegatingRunStore implements RunStore {
176176
${readonlyProperties
177177
// Indexed access rather than the written type, so the getter needs no import of its own and
178178
// follows the interface if that type is ever changed.
179-
.map((p) => ` get ${p.name}(): RunStore["${p.name}"] {\n return this.delegate.${p.name};\n }`)
179+
.map(
180+
(p) => ` get ${p.name}(): RunStore["${p.name}"] {\n return this.delegate.${p.name};\n }`
181+
)
180182
.join("\n\n")}${readonlyProperties.length > 0 ? "\n\n" : ""}${unique
181-
.map((n) => ` ${n}(...args: any[]): any {\n return (this.delegate as any).${n}(...args);\n }`)
182-
.join("\n\n")}
183+
.map(
184+
(n) => ` ${n}(...args: any[]): any {\n return (this.delegate as any).${n}(...args);\n }`
185+
)
186+
.join("\n\n")}
183187
}
184188
`
185189
);

‎internal-packages/run-store/src/PostgresRunStore.snapshotWrites.test.ts‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -150,7 +150,9 @@ describe("PostgresRunStore snapshotWrites flag", () => {
150150
{ select: { id: true } }
151151
);
152152

153-
expect((await prisma.taskRun.findFirstOrThrow({ where: { id: run.id } })).status).toBe("EXPIRED");
153+
expect((await prisma.taskRun.findFirstOrThrow({ where: { id: run.id } })).status).toBe(
154+
"EXPIRED"
155+
);
154156
expect(await prisma.taskRunExecutionSnapshot.count({ where: { runId: run.id } })).toBe(0);
155157
});
156158

@@ -226,7 +228,9 @@ describe("PostgresRunStore snapshotWrites flag", () => {
226228
},
227229
});
228230

229-
expect((await prisma.taskRun.findFirstOrThrow({ where: { id: run.id } })).status).toBe("DEQUEUED");
231+
expect((await prisma.taskRun.findFirstOrThrow({ where: { id: run.id } })).status).toBe(
232+
"DEQUEUED"
233+
);
230234
expect(await prisma.taskRunExecutionSnapshot.count({ where: { runId: run.id } })).toBe(0);
231235
});
232236

‎internal-packages/run-store/src/PostgresRunStore.ts‎

Lines changed: 18 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -744,6 +744,8 @@ export class PostgresRunStore implements RunStore {
744744

745745
const snapshotCreate = {
746746
id: params.snapshot.id,
747+
createdAt: params.snapshot.createdAt,
748+
updatedAt: params.snapshot.createdAt,
747749
engine: params.snapshot.engine,
748750
executionStatus: params.snapshot.executionStatus,
749751
description: params.snapshot.description,
@@ -831,6 +833,8 @@ export class PostgresRunStore implements RunStore {
831833

832834
const snapshotCreate = {
833835
id: params.snapshot.id,
836+
createdAt: params.snapshot.createdAt,
837+
updatedAt: params.snapshot.createdAt,
834838
engine: params.snapshot.engine,
835839
executionStatus: params.snapshot.executionStatus,
836840
description: params.snapshot.description,
@@ -942,6 +946,8 @@ export class PostgresRunStore implements RunStore {
942946
costInCents: data.costInCents,
943947
...this.#nestedSnapshot({
944948
id: data.snapshot.id,
949+
createdAt: data.snapshot.createdAt,
950+
updatedAt: data.snapshot.createdAt,
945951
executionStatus: data.snapshot.executionStatus,
946952
description: data.snapshot.description,
947953
runStatus: data.snapshot.runStatus,
@@ -1147,6 +1153,8 @@ export class PostgresRunStore implements RunStore {
11471153
error: data.error as Prisma.InputJsonValue,
11481154
...this.#nestedSnapshot({
11491155
id: data.snapshot.id,
1156+
createdAt: data.snapshot.createdAt,
1157+
updatedAt: data.snapshot.createdAt,
11501158
engine: data.snapshot.engine,
11511159
executionStatus: data.snapshot.executionStatus,
11521160
description: data.snapshot.description,
@@ -1277,6 +1285,8 @@ export class PostgresRunStore implements RunStore {
12771285
maxAttempts: data.maxAttempts ?? undefined,
12781286
...this.#nestedSnapshot({
12791287
id: data.snapshot.id,
1288+
createdAt: data.snapshot.createdAt,
1289+
updatedAt: data.snapshot.createdAt,
12801290
engine: "V2",
12811291
executionStatus: "PENDING_EXECUTING",
12821292
description: "Run was dequeued for execution",
@@ -1382,6 +1392,8 @@ export class PostgresRunStore implements RunStore {
13821392
error: data.error as Prisma.InputJsonValue,
13831393
...this.#nestedSnapshot({
13841394
id: data.snapshot.id,
1395+
createdAt: data.snapshot.createdAt,
1396+
updatedAt: data.snapshot.createdAt,
13851397
engine: data.snapshot.engine,
13861398
executionStatus: data.snapshot.executionStatus,
13871399
description: data.snapshot.description,
@@ -1454,6 +1466,8 @@ export class PostgresRunStore implements RunStore {
14541466
...(data.snapshot &&
14551467
this.#nestedSnapshot({
14561468
id: data.snapshot.id,
1469+
createdAt: data.snapshot.createdAt,
1470+
updatedAt: data.snapshot.createdAt,
14571471
engine: "V2",
14581472
executionStatus: data.snapshot.executionStatus ?? "DELAYED",
14591473
description:
@@ -1984,6 +1998,7 @@ export class PostgresRunStore implements RunStore {
19841998
): Promise<Prisma.TaskRunExecutionSnapshotGetPayload<{ include: { checkpoint: true } }>> {
19851999
const {
19862000
id,
2001+
createdAt,
19872002
run,
19882003
snapshot,
19892004
previousSnapshotId,
@@ -2014,7 +2029,7 @@ export class PostgresRunStore implements RunStore {
20142029
);
20152030
}
20162031

2017-
const now = new Date();
2032+
const now = createdAt ?? new Date();
20182033
return {
20192034
id,
20202035
engine: "V2",
@@ -2049,6 +2064,8 @@ export class PostgresRunStore implements RunStore {
20492064
const newSnapshot = await prisma.taskRunExecutionSnapshot.create({
20502065
data: {
20512066
id,
2067+
createdAt,
2068+
updatedAt: createdAt,
20522069
engine: "V2",
20532070
executionStatus: snapshot.executionStatus,
20542071
description: snapshot.description,

‎internal-packages/run-store/src/redisSnapshotStore.sinceCreatedAt.test.ts‎

Lines changed: 23 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -24,8 +24,7 @@ function entry(runId: string, id: string, createdAt: string): SnapshotEntryInput
2424
};
2525
}
2626

27-
const at = (seconds: number) =>
28-
new Date(Date.UTC(2026, 0, 1, 0, 0, seconds)).toISOString();
27+
const at = (seconds: number) => new Date(Date.UTC(2026, 0, 1, 0, 0, seconds)).toISOString();
2928

3029
async function seed(
3130
store: RedisSnapshotStore,
@@ -42,26 +41,29 @@ async function seed(
4241
}
4342

4443
describe("getSinceCreatedAt", () => {
45-
redisTest("returns only entries newer than the cursor, oldest first", async ({ redisOptions }) => {
46-
const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: COMPLETED_TTL_MS });
47-
try {
48-
const runId = "run_window";
49-
await seed(
50-
store,
51-
runId,
52-
[0, 1, 2, 3, 4].map((n) => ({ id: `snap_${n}`, createdAt: at(n) }))
53-
);
54-
55-
const result = await store.getSinceCreatedAt(runId, at(1));
56-
57-
expect(result.kind).toBe("hit");
58-
if (result.kind !== "hit") return;
59-
// Ascending, matching what the engine hands its caller after its own reverse().
60-
expect(result.entries.map((e) => e.id)).toEqual(["snap_2", "snap_3", "snap_4"]);
61-
} finally {
62-
await store.quit();
44+
redisTest(
45+
"returns only entries newer than the cursor, oldest first",
46+
async ({ redisOptions }) => {
47+
const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: COMPLETED_TTL_MS });
48+
try {
49+
const runId = "run_window";
50+
await seed(
51+
store,
52+
runId,
53+
[0, 1, 2, 3, 4].map((n) => ({ id: `snap_${n}`, createdAt: at(n) }))
54+
);
55+
56+
const result = await store.getSinceCreatedAt(runId, at(1));
57+
58+
expect(result.kind).toBe("hit");
59+
if (result.kind !== "hit") return;
60+
// Ascending, matching what the engine hands its caller after its own reverse().
61+
expect(result.entries.map((e) => e.id)).toEqual(["snap_2", "snap_3", "snap_4"]);
62+
} finally {
63+
await store.quit();
64+
}
6365
}
64-
});
66+
);
6567

6668
redisTest("misses when the run has no keyspace", async ({ redisOptions }) => {
6769
const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: COMPLETED_TTL_MS });

‎internal-packages/run-store/src/runStoreMethodNames.ts‎

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -78,6 +78,4 @@ export const RUN_STORE_METHOD_NAMES = [
7878
] as const;
7979

8080
// Data properties the base exposes as getters over the delegate, not as forwarders.
81-
export const RUN_STORE_PROPERTY_NAMES = [
82-
"primaryReadClient",
83-
] as const;
81+
export const RUN_STORE_PROPERTY_NAMES = ["primaryReadClient"] as const;

0 commit comments

Comments
 (0)