From def5a239976fbc063d4a17fd50f49f2700264c22 Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 21:18:05 -0500 Subject: [PATCH] fix(server): fence status card refresh decisions and retry debt Co-Authored-By: Paperclip --- doc/plans/2026-10-30-status-cards-sleep.md | 5 ++ .../status-card-evaluation-work.test.ts | 82 +++++++++++++++++++ server/src/services/status-cards.ts | 40 ++++++--- 3 files changed, 117 insertions(+), 10 deletions(-) diff --git a/doc/plans/2026-10-30-status-cards-sleep.md b/doc/plans/2026-10-30-status-cards-sleep.md index 3f3b8933cbc..97566de3bc4 100644 --- a/doc/plans/2026-10-30-status-cards-sleep.md +++ b/doc/plans/2026-10-30-status-cards-sleep.md @@ -155,3 +155,8 @@ closed-range evaluations perform one conversion, not another minute walk. Both caches use bounded LRU retention and hold no company/card data. Tests prove eviction, skipped/repeated hours, half-hour DST, date-line changes, rollback, strict boundaries, and actual twenty-card source delivery followed by quiet. + + +Refresh decisions use the card's PostgreSQL row revision as a private compare-and-swap fence. If a card changes during watched-source reads, the owner rereads it and reconsiders the current policy, for at most three attempts. A replacement future interval deadline remains unchanged; a replacement manual policy does not run automatically. The same fence protects no-change, paused-policy and generation claims. Existing issue idempotency, generation ownership and losing-claim cancellation still apply. The row revision is not exposed through card APIs or shared scheduling contracts. + +Company source hints do not preempt a future deadline owned by a failed evaluation (`retry`). Membership and policy deadlines remain eligible for source-driven rechecks, and actual due retries still run. An explicit policy edit can replace the retry through the existing card mutation path. Real PostgreSQL regressions hold the actual search while committing an ordinary policy edit, then verify the saved interval still fires. Persistent search failures stay quiet through issue hints and restart, retry at their saved deadline, and recover through an ordinary policy edit. diff --git a/server/src/__tests__/status-card-evaluation-work.test.ts b/server/src/__tests__/status-card-evaluation-work.test.ts index d012a239efd..7c0f010949d 100644 --- a/server/src/__tests__/status-card-evaluation-work.test.ts +++ b/server/src/__tests__/status-card-evaluation-work.test.ts @@ -588,6 +588,88 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu expect(wakeup).toHaveBeenCalledOnce(); }); + it.each([false, true])("rechecks current policy after a committed edit during held search (changes=%s)", async changed => { + const created = await card(); + if (changed) await issueService(db).create(companyId, { title: "Policy race blocker", status: "blocked", priority: "medium" }); + const entered = Promise.withResolvers(), release = Promise.withResolvers(); + const factory = companySearch.companySearchService; + let hold = true; + const searches = vi.fn(); + vi.spyOn(companySearch, "companySearchService").mockImplementation(owner => { + const service = factory(owner), actual = service.search; + service.search = async (...args) => { + searches(); + const result = await actual(...args); + if (hold) { hold = false; entered.resolve(); await release.promise; } + return result; + }; + return service; + }); + const service = statusCardEvaluationService(db, { wakeup }); + const pass = service.run(new AbortController().signal); + await entered.promise; + const cards = statusCardService(db); + let scheduled!: Date; + try { + scheduled = (await cards.update((await cards.getById(created.id))!, { refreshPolicy: policy("interval") }, actor)).nextEvalAt!; + } finally { release.resolve(); } + await pass; + const current = (await cards.getById(created.id))!; + expect(current.refreshPolicy.mode).toBe("interval"); + expect(current.nextEvalAt!.getTime()).toBe(scheduled.getTime()); + expect(current.nextEvalReason).toBe("refresh"); + expect(searches).toHaveBeenCalledOnce(); + expect(wakeup).not.toHaveBeenCalled(); + const intervalCard = (await cards.getById(created.id))!; + await issueService(db).create(companyId, { title: "Next interval blocker", status: "blocked", priority: "medium" }); + vi.setSystemTime(intervalCard.nextEvalAt!); + await service.run(new AbortController().signal); + expect(wakeup).toHaveBeenCalledOnce(); + }); + + it("keeps future failure retries fenced from ordinary company hints and permits explicit policy recovery", async () => { + const created = await card(); + const factory = companySearch.companySearchService; + let failing = true; + const searches = vi.fn(); + vi.spyOn(companySearch, "companySearchService").mockImplementation(owner => { + const service = factory(owner), actual = service.search; + service.search = async (...args) => { searches(); if (failing) throw new Error("Controlled search unavailable"); return actual(...args); }; + return service; + }); + const service = statusCardEvaluationService(db, { wakeup }); + await service.run(new AbortController().signal); + const failed = (await statusCardService(db).getById(created.id))!; + expect(failed.nextEvalReason).toBe("retry"); + const retryAt = failed.nextEvalAt!.getTime(); + expect(searches).toHaveBeenCalledOnce(); + for (let index = 0; index < 5; index++) { + await issueService(db).create(companyId, { title: `Retry blocker ${index}`, status: "blocked", priority: "medium" }); + service.changed(new Set([`changes:${companyId}`])); + await service.run(new AbortController().signal); + } + expect(searches).toHaveBeenCalledOnce(); + expect((await statusCardService(db).getById(created.id))!.nextEvalAt!.getTime()).toBe(retryAt); + const restarted = statusCardEvaluationService(db, { wakeup }); + restarted.changed(new Set([`changes:${companyId}`])); + await restarted.run(new AbortController().signal); + expect(searches).toHaveBeenCalledOnce(); + expect(await restarted.next()).toBe(retryAt); + vi.setSystemTime(retryAt); + await restarted.run(new AbortController().signal); + expect(searches).toHaveBeenCalledTimes(2); + const cards = statusCardService(db); + failing = false; + await cards.update((await cards.getById(created.id))!, { refreshPolicy: { ...policy(), debounceSeconds: 1 } }, actor); + await restarted.run(new AbortController().signal); + expect(searches).toHaveBeenCalledTimes(3); + const pending = (await cards.getById(created.id))!; + expect(pending.nextEvalReason).toBe("policy"); + vi.setSystemTime(pending.nextEvalAt!); + await restarted.run(new AbortController().signal); + expect(wakeup).toHaveBeenCalledOnce(); + }); + it("retains the durable five-minute retry after evaluation failure across restart and drain", async () => { const created = await card(); await db.update(statusCards).set({ queries: [{ scope: "invalid" } as never], nextEvalAt: new Date() }).where(eq(statusCards.id, created.id)); diff --git a/server/src/services/status-cards.ts b/server/src/services/status-cards.ts index d4660227c28..862d9d64ec4 100644 --- a/server/src/services/status-cards.ts +++ b/server/src/services/status-cards.ts @@ -643,14 +643,17 @@ export function statusCardService( ? membershipAt : interval; } - async function requestRefresh(cardId: string, input: { + async function refreshCurrentCard(cardId: string, input: { full?: boolean; trigger?: "manual" | "interval" | "reactive" | "restore"; actor?: StatusCardActor; now?: Date; - } = {}) { - const card = await getById(cardId); - if (!card) throw notFound("Status card not found"); + } = {}, superseded = false) { + const [source] = await db.select({ card: statusCards, revision: sql`${statusCards}.xmin::text` }) + .from(statusCards).where(eq(statusCards.id, cardId)); + if (!source) throw notFound("Status card not found"); + const { card, revision } = source; + const currentRevision = sql`${statusCards}.xmin::text = ${revision}`; if (card.archivedAt) throw unprocessable("Archived status cards cannot be refreshed"); if (card.queries.length === 0) throw conflict("Compile the status-card query before refreshing it"); if (card.generatingIssueId) { @@ -663,6 +666,10 @@ export function statusCardService( } const now = input.now ?? new Date(); + if (superseded && input.trigger !== undefined && input.trigger !== "manual" && input.trigger !== "restore" + && (card.refreshPolicy.mode === "manual" || (card.refreshPolicy.mode === "interval" && card.nextEvalAt && card.nextEvalAt > now))) { + return { card, generatingIssue: null, alreadyGenerating: false, enqueued: false }; + } let membershipAt: Date | null = null; const snapshot = await executeQueries(card, {}, at => { if (at && (!membershipAt || at < membershipAt)) membershipAt = at; }); const fingerprint = buildStatusCardFingerprint(snapshot); @@ -679,8 +686,9 @@ export function statusCardService( state: "active", nextEvalAt, nextEvalReason: nextEvalAt ? card.refreshPolicy.mode === "reactive" ? "membership" : "refresh" : null, - }).where(eq(statusCards.id, card.id)).returning()); - return { card: next!, generatingIssue: null, alreadyGenerating: false, enqueued: false }; + }).where(and(eq(statusCards.id, card.id), currentRevision)).returning()); + if (!next) return undefined; + return { card: next, generatingIssue: null, alreadyGenerating: false, enqueued: false }; } const hourAgo = new Date(now.getTime() - 60 * 60 * 1000); @@ -712,8 +720,9 @@ export function statusCardService( completedUpdateTimes: recent.filter(row => row.kind !== "compile" && row.finishedAt).map(row => row.startedAt), tokensToday: Number(daily[0]?.tokens ?? 0), }), - }).where(eq(statusCards.id, card.id)).returning()); - return { card: next!, generatingIssue: null, alreadyGenerating: false, enqueued: false }; + }).where(and(eq(statusCards.id, card.id), currentRevision)).returning()); + if (!next) return undefined; + return { card: next, generatingIssue: null, alreadyGenerating: false, enqueued: false }; } const history = await listUpdates(card.id); @@ -765,13 +774,14 @@ export function statusCardService( nextEvalAt, nextEvalReason: nextEvalAt ? card.refreshPolicy.mode === "reactive" ? "membership" : "refresh" : null, failureReason: null, - }).where(and(eq(statusCards.id, card.id), isNull(statusCards.archivedAt), or(isNull(statusCards.generatingIssueId), priorGenerationPredicate))).returning()); + }).where(and(eq(statusCards.id, card.id), isNull(statusCards.archivedAt), currentRevision, or(isNull(statusCards.generatingIssueId), priorGenerationPredicate))).returning()); if (!next) { const winner = await getById(card.id); if (!winner?.generatingIssueId) { if (!TERMINAL_ISSUE_STATUSES.has(generationIssue!.status)) { await issuesSvc().update(generationIssue!.id, { status: "cancelled" }); } + if (winner && !winner.archivedAt) return undefined; throw conflict("Status-card refresh claim was lost"); } if (generationIssue!.id !== winner.generatingIssueId && !TERMINAL_ISSUE_STATUSES.has(generationIssue!.status)) { @@ -794,6 +804,14 @@ export function statusCardService( return { card: next, generatingIssue: generationIssue!, alreadyGenerating: deduplicated, enqueued: true, kind, changes }; } + async function requestRefresh(cardId: string, input: Parameters[1] = {}) { + for (let attempt = 0; attempt < 3; attempt += 1) { + const result = await refreshCurrentCard(cardId, input, attempt > 0); + if (result) return result; + } + throw conflict("Status-card refresh was superseded by concurrent edits"); + } + async function tickDueStatusCards(now = new Date(), reactiveCompanies?: ReadonlySet | "changed", reconstruct = false) { const changedScope = reactiveCompanies === "changed" ? sql`true` : reactiveCompanies?.size ? inArray(statusCards.companyId, [...reactiveCompanies]) : sql`false`; @@ -811,7 +829,9 @@ export function statusCardService( .innerJoin(companies, eq(companies.id, statusCards.companyId)) .where(and(eq(companies.status, "active"), isNull(statusCards.archivedAt), isNull(statusCards.generatingIssueId), or(and(isNotNull(statusCards.nextEvalAt), lte(statusCards.nextEvalAt, now)), - and(reactiveScope, ne(statusCards.state, "error"), ne(statusCards.state, "compiling"), sql`${statusCards.refreshPolicy}->>'mode' = 'reactive'`, sql`jsonb_array_length(${statusCards.queries}) > 0`)))) + and(reactiveScope, + or(isNull(statusCards.nextEvalReason), ne(statusCards.nextEvalReason, "retry"), isNull(statusCards.nextEvalAt), lte(statusCards.nextEvalAt, now)), + ne(statusCards.state, "error"), ne(statusCards.state, "compiling"), sql`${statusCards.refreshPolicy}->>'mode' = 'reactive'`, sql`jsonb_array_length(${statusCards.queries}) > 0`)))) .then(rows => rows.map(row => row.card)); const enqueued: Array<{ cardId: string; generatingIssue: typeof issues.$inferSelect }> = []; let evaluated = 0;