Skip to content

Commit a16f0d8

Browse files
committed
fix(run-store,run-engine): contain a sweep failure to the keyspace that caused it
The two rule bodies issue their own Redis commands and none of them were guarded, so an error from any one of them left the whole pass. The scan restarts from the same place each time, so collection then stopped for every run until a human removed the key. Observed against a single malformed keyspace: seven consecutive passes collected nothing. A bad keyspace now costs itself one pass. The count is reported so that containment is not mistaken for nothing going wrong.
1 parent 3585eaa commit a16f0d8

3 files changed

Lines changed: 86 additions & 10 deletions

File tree

‎internal-packages/run-engine/src/engine/types.ts‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -64,6 +64,10 @@ export const SNAPSHOT_SWEEP_COUNT_FIELDS = [
6464
"deleted",
6565
"skipped",
6666
"pendingDeletion",
67+
// Keyspaces the pass could not read. Reported because containment without a number is
68+
// indistinguishable from nothing going wrong: a run that fails every pass leaks forever and the
69+
// only other evidence is one log line per pass.
70+
"failed",
6771
"nodes",
6872
"partial",
6973
] as const;

‎internal-packages/run-store/src/snapshotOrphanSweeper.test.ts‎

Lines changed: 51 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -466,4 +466,55 @@ describe("SnapshotOrphanSweeper", () => {
466466
await Promise.all([store.quit(), sweeper.quit(), probe.quit().catch(() => {})]);
467467
}
468468
});
469+
containerTest(
470+
"one wrong-typed keyspace costs that run its pass, not every run's",
471+
async ({ prisma, redisOptions }) => {
472+
// A single malformed keyspace used to abort the WHOLE pass: the Redis commands in the two
473+
// rule bodies were uncaught, so a WRONGTYPE propagated out of sweep() and every later pass
474+
// hit the same key. Collection stopped for every run until a human deleted it.
475+
const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: COMPLETED_TTL_MS });
476+
const runStore = new PostgresRunStore({ prisma, readOnlyPrisma: prisma });
477+
const sweeper = new SnapshotOrphanSweeper({
478+
redisOptions,
479+
runStore: runStore as unknown as RunStore,
480+
completedTtlMs: COMPLETED_TTL_MS,
481+
orphanAgeMs: ORPHAN_AGE_MS,
482+
confirmOrphanAfterMs: 0,
483+
});
484+
const probe = createRedisClient(redisOptions, { onError: () => {} });
485+
try {
486+
const env = await seedSnapshotEnvironment(prisma);
487+
488+
// A healthy terminal run that rule 1 must still expire.
489+
const healthyId = generateInternalId();
490+
await store.append({
491+
entry: birthEntry(healthyId, env, new Date()),
492+
kind: "birth",
493+
isTerminal: false,
494+
});
495+
await prisma.taskRun.create({
496+
data: { ...buildCreateRunData(healthyId, env), status: "COMPLETED_SUCCESSFULLY" },
497+
});
498+
499+
// A keyspace whose `seq` is a string where the code expects a hash. Discovered by the scan
500+
// (it matches on `:e`), then every command against `seq` raises WRONGTYPE.
501+
const brokenId = generateInternalId();
502+
const brokenKeys = snapshotKeys(brokenId);
503+
await probe.hset(brokenKeys.e, "s_x", JSON.stringify({ id: "s_x", runId: brokenId }));
504+
await probe.set(brokenKeys.seq, "not-a-hash");
505+
506+
const result = await sweeper.sweep();
507+
508+
// The pass completes and reports the failure rather than throwing it.
509+
expect(result.failed).toBe(1);
510+
expect(result.partial).toBe(false);
511+
512+
// And the healthy run was still collected, which is the whole point.
513+
expect(result.expired).toBe(1);
514+
expect(await probe.pttl(snapshotKeys(healthyId).e)).toBeGreaterThan(0);
515+
} finally {
516+
await Promise.all([store.quit(), sweeper.quit(), probe.quit().catch(() => {})]);
517+
}
518+
}
519+
);
469520
});

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

Lines changed: 31 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -80,6 +80,13 @@ export type SweepResult = {
8080
* on an observed sweep pass, so the observation has to carry its own coverage.
8181
*/
8282
nodes: number;
83+
/**
84+
* Keyspaces whose own Redis commands raised, so the pass acted on neither rule for them. Contained
85+
* per run on purpose: a single malformed keyspace used to abort the whole pass, and because the
86+
* scan restarts from the same place every time, collection then stopped for EVERY run until a
87+
* human removed the key. One bad keyspace now costs itself one pass.
88+
*/
89+
failed: number;
8390
/** True when the pass stopped early on its deadline or abort signal, so coverage is incomplete. */
8491
partial: boolean;
8592
};
@@ -200,6 +207,7 @@ export class SnapshotOrphanSweeper {
200207
deleted: 0,
201208
skipped: 0,
202209
pendingDeletion: 0,
210+
failed: 0,
203211
nodes: 0,
204212
partial: false,
205213
};
@@ -295,18 +303,31 @@ export class SnapshotOrphanSweeper {
295303
for (const runId of runIds) {
296304
const run = rows.get(runId);
297305

298-
if (!run) {
299-
await this.#applyRuleTwo(runId, dryRun, result);
300-
continue;
301-
}
306+
// Per run, not per batch. The rule bodies below issue their own Redis commands, and an error
307+
// from any of them used to leave the whole pass. The run-row lookup above has had its own
308+
// guard from the start; this gives the Redis half the same treatment, for the same reason.
309+
try {
310+
if (!run) {
311+
await this.#applyRuleTwo(runId, dryRun, result);
312+
continue;
313+
}
302314

303-
if (!FINAL.has(run.status)) {
304-
// A live run. A SUSPENDED run can legitimately wait for weeks, so this is never touched.
305-
result.skipped += 1;
306-
continue;
307-
}
315+
if (!FINAL.has(run.status)) {
316+
// A live run. A SUSPENDED run can legitimately wait for weeks, so this is never touched.
317+
result.skipped += 1;
318+
continue;
319+
}
308320

309-
await this.#applyRuleOne(runId, dryRun, result);
321+
await this.#applyRuleOne(runId, dryRun, result);
322+
} catch (error) {
323+
// Never rethrown. A keyspace this pass cannot read is a keyspace the NEXT pass can try, and
324+
// the alternative is losing collection for every other run as well.
325+
this.#logger.error("SnapshotOrphanSweeper skipped a keyspace after a failed command", {
326+
runId,
327+
error,
328+
});
329+
result.failed += 1;
330+
}
310331
}
311332
}
312333

0 commit comments

Comments
 (0)