fix(memory-jobs): reject pending lineage reruns

This commit is contained in:
Dotta committed 2026-04-21 16:00:48 -05:00
1 parent 071d20974b
commit 33b2b798b1
2 files changed
+52 -1

No files matched your search

@@ -107,6 +107,49 @@ describeEmbeddedPostgres("memory job store", () => {
});
});
it("rejects reruns when a lineage already has a queued retry", async () => {
const companyId = randomUUID();
const store = memoryJobStore(db);
await insertCompany(db, companyId);
const original = await store.enqueue({
companyId,
bindingId: randomUUID(),
bindingKey: "primary",
operationType: "capture",
sourceKind: "manual",
});
await store.cancel({
companyId,
jobId: original.id,
now: new Date("2026-04-21T18:00:00.000Z"),
resultSummary: "Cancelled by test",
});
const firstRetry = await store.rerun(companyId, original.id, {
now: new Date("2026-04-21T18:01:00.000Z"),
});
await expect(store.rerun(companyId, original.id)).rejects.toMatchObject({
status: 409,
message: `Memory job ${original.id} already has a queued or running rerun attempt`,
});
await store.cancel({
companyId,
jobId: firstRetry.id,
now: new Date("2026-04-21T18:02:00.000Z"),
resultSummary: "Retry cancelled by test",
});
await expect(store.rerun(companyId, original.id)).resolves.toMatchObject({
retryOfJobId: original.id,
attemptNumber: 3,
status: "queued",
});
});
it("derives stuck effective state from expired running leases in list and detail queries", async () => {
const companyId = randomUUID();
const now = new Date("2026-04-21T19:00:00.000Z");
+9 -1
View File
@@ -696,7 +696,10 @@ export function memoryJobStore(db: Db) {
);
const lineageRows = await txDb
.select({ attemptNumber: memoryExtractionJobs.attemptNumber })
.select({
attemptNumber: memoryExtractionJobs.attemptNumber,
status: memoryExtractionJobs.status,
})
.from(memoryExtractionJobs)
.where(
and(
@@ -708,6 +711,11 @@ export function memoryJobStore(db: Db) {
),
);
const pendingAttempt = lineageRows.find((row) => row.status === "queued" || row.status === "running");
if (pendingAttempt) {
throw conflict(`Memory job ${retryRootId} already has a queued or running rerun attempt`);
}
const nextAttemptNumber = lineageRows.reduce((maxAttempt, row) => {
return Math.max(maxAttempt, row.attemptNumber);
}, 0) + 1;