From 41992f4383f0a0013471aa2bc0f7c7c9cd844856 Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 19:54:00 -0500 Subject: [PATCH] fix: release reconciled decision-target holds during drain Co-Authored-By: Paperclip --- .../__tests__/decision-target-work.test.ts | 39 +++++++++++++++++-- server/src/services/decision-target-work.ts | 3 +- 2 files changed, 37 insertions(+), 5 deletions(-) diff --git a/server/src/__tests__/decision-target-work.test.ts b/server/src/__tests__/decision-target-work.test.ts index f9ec92d1068..626af25b8ee 100644 --- a/server/src/__tests__/decision-target-work.test.ts +++ b/server/src/__tests__/decision-target-work.test.ts @@ -271,8 +271,33 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes it("retains a first target change through drain without passive SQL", async () => { const row = await create(); await start(); await quiet(); draining = true; await change("cancelled"); - const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(120_000); - expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(30_000); expect(idleWorkSnapshot().active).toBe(0); }); + expect(recovery.hasDecisionTargetWork()).toBe(true); + expect(coordinator!.nextWakeAt()).not.toBeNull(); + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "update"), executes = vi.spyOn(db, "execute"); + await vi.advanceTimersByTimeAsync(120_000); + expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); expect(executes).not.toHaveBeenCalled(); + expect(idleWorkSnapshot().active).toBe(0); + reads.mockRestore(); writes.mockRestore(); executes.mockRestore(); + draining = false; await expired(row.id); await quiet(); expect(wake).toHaveBeenCalledTimes(1); + }); + + it("keeps a real uncommitted target write held during drain and releases only after commit", async () => { + const row = await create(); await start(); await quiet(); draining = true; + let release!: () => void, entered!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const ready = new Promise(resolve => { entered = resolve; }); + const writing = db.transaction(async tx => { + await issueService(db).update(targetId, { status: "cancelled", actorUserId: userId }, tx); + entered(); await gate; + }); + try { + await ready; await vi.advanceTimersByTimeAsync(120_000); + expect(idleWorkSnapshot().active).toBeGreaterThan(0); + expect((await service().get(row.id))?.status).toBe("open"); expect(wake).not.toHaveBeenCalled(); + } finally { release(); await writing; } + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(30_000); expect(idleWorkSnapshot().active).toBe(0); }); + expect(recovery.hasDecisionTargetWork()).toBe(true); draining = false; await expired(row.id); await quiet(); expect(wake).toHaveBeenCalledTimes(1); }); @@ -286,8 +311,14 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes const row = await create(); await start(); await quiet(); await failExpiry(); await change("cancelled"); await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(30_000); expect(errors.length).toBeGreaterThan(0); }); draining = true; - const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(120_000); - expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(30_000); expect(idleWorkSnapshot().active).toBe(0); }); + expect(recovery.hasDecisionTargetWork()).toBe(true); + expect(coordinator!.nextWakeAt()).not.toBeNull(); + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "update"), executes = vi.spyOn(db, "execute"); + await vi.advanceTimersByTimeAsync(120_000); + expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); expect(executes).not.toHaveBeenCalled(); + expect(idleWorkSnapshot().active).toBe(0); + reads.mockRestore(); writes.mockRestore(); executes.mockRestore(); await db.execute(sql`drop trigger decision_target_expire_fail on decisions`); draining = false; await expired(row.id); await quiet(); expect(wake).toHaveBeenCalledTimes(1); for (const error of errors) expect(error).toMatchObject({ cause: { message: "expiry unavailable" } }); errors.length = 0; diff --git a/server/src/services/decision-target-work.ts b/server/src/services/decision-target-work.ts index 219c9ad6f2b..0a285fa86ec 100644 --- a/server/src/services/decision-target-work.ts +++ b/server/src/services/decision-target-work.ts @@ -12,6 +12,7 @@ export function registerDecisionTargetWork( pausePassiveMaintenance: true, onSettled: service.decisionTargetsChanged, run: signal => canRun() ? service.sweepInvalidTargets(new Date(), signal) : Promise.resolve(), - hasPending: async () => service.hasDecisionTargetWork(), + hasPending: async () => canRun() && service.hasDecisionTargetWork(), + nextRunAt: () => service.hasDecisionTargetWork() ? Date.now() + 30_000 : null, }); }