From cd4caaab266ecd857563be32e7aec9974fc26e4a Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 18:49:20 -0500 Subject: [PATCH] fix: batch status-card source notifications Coalesce committed company hints in a fixed one-second window without flooring real deadlines. Verify one hundred ordinary issue/comment writes, company isolation, deadline preemption, retained hints and settled SQL quietness. Await actual evaluation I/O in fake-clock acceptance instead of jumping across connection timers. Co-Authored-By: Paperclip --- doc/plans/2026-10-30-status-cards-sleep.md | 12 ++ .../status-card-evaluation-work.test.ts | 130 +++++++++++++++++- server/src/services/status-card-evaluation.ts | 4 + 3 files changed, 139 insertions(+), 7 deletions(-) diff --git a/doc/plans/2026-10-30-status-cards-sleep.md b/doc/plans/2026-10-30-status-cards-sleep.md index dbf229cdace..5b98962698b 100644 --- a/doc/plans/2026-10-30-status-cards-sleep.md +++ b/doc/plans/2026-10-30-status-cards-sleep.md @@ -123,3 +123,15 @@ five minutes keeps its remaining time without extension. This bounds migration compatibility delay while preserving prior crash/failure recovery. After the first successful evaluation all newly saved deadlines have explicit ownership. No recurring compatibility scan is added. + +Source notifications use the shared deadline helper's fixed one-second batching +window. The first hint after idle can reconcile immediately. Later hints merge +by company until that window ends. The window does not slide with each new +write, so a sustained source stream cannot postpone reconciliation forever. +One second keeps reactive cards prompt (the removed heartbeat default was 30s) +and bounds burst SQL. No minimum spacing is set: an already-known policy, +expiry, or retry deadline can preempt batching. The window alone creates no +scheduled work after settlement. Real issue/comment producer acceptance measures +due/next queries and actual searches under 100 writes, preserves both companies' +hints, dispatches another company's saved deadline inside the batching window, +and checks SQL quietness after authorized results. diff --git a/server/src/__tests__/status-card-evaluation-work.test.ts b/server/src/__tests__/status-card-evaluation-work.test.ts index 4e00e326855..6bb466dea89 100644 --- a/server/src/__tests__/status-card-evaluation-work.test.ts +++ b/server/src/__tests__/status-card-evaluation-work.test.ts @@ -1,5 +1,7 @@ import { budgetService } from "../services/budgets.js"; import { documentService } from "../services/documents.js"; +import * as companySearch from "../services/company-search.js"; +import { setImmediate as immediate } from "node:timers/promises"; import { randomUUID } from "node:crypto"; import { eq, sql } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; @@ -24,6 +26,7 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu let draining = false; let companyId: string, agentId: string; const errors: unknown[] = []; + const evaluations = new Set>(); let compiledQuery = companySearchQuerySchema.parse({ scope: "issues", status: ["blocked"] }); const actor = { agentId: null, userId: "board-user" }; const policy = (mode: "manual" | "reactive" | "interval" = "reactive") => statusCardRefreshPolicySchema.parse({ @@ -60,16 +63,29 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu await db.delete(instanceSettings); await db.delete(agents); await db.delete(companyMemberships); await db.delete(companies); }); afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); }); - async function start() { + async function start(service = statusCardEvaluationService(db, { wakeup })) { coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining, canReconcile: () => true, onError: error => errors.push(error) }); - await registerStatusCardEvaluationWork(coordinator, statusCardEvaluationService(db, { wakeup }), () => !draining).ready; + await registerStatusCardEvaluationWork(coordinator, { ...service, run: signal => { + const work = service.run(signal); + evaluations.add(work); + void work.then(() => evaluations.delete(work), () => evaluations.delete(work)); + return work; + } }, () => !draining).ready; } async function settled() { - await vi.waitFor(async () => { - await vi.advanceTimersByTimeAsync(30_000); - expect(idleWorkSnapshot().active).toBe(0); - }, { timeout: 5000 }); + // Do not leap across PostgreSQL connect timers while real I/O is pending. + // Await actual domain work, then require several settled event-loop turns + // so a transient zero hold cannot stand in for transaction reconciliation. + for (let turn = 0; turn < 3; turn++) { + await vi.waitFor(async () => { + await Promise.allSettled([...evaluations]); + await vi.advanceTimersByTimeAsync(500); + await Promise.allSettled([...evaluations]); + expect(idleWorkSnapshot().active).toBe(0); + }, { timeout: 5000 }); + await immediate(); + } } async function quiet() { await settled(); @@ -128,11 +144,111 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu expect(errors).toEqual([]); }); + it("coalesces one hundred committed source writes without delaying another company's saved deadline", async () => { + const reactive = await card(); + const originalCompany = companyId, originalAgent = agentId; + companyId = randomUUID(); agentId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Other", issuePrefix: `O${companyId.slice(0, 6)}` }); + await db.insert(agents).values({ id: agentId, companyId, name: "Other summarizer", role: "general", status: "idle", adapterType: "process", adapterConfig: {} }); + const other = await card("interval"); + const otherIssue = await issueService(db).create(companyId, { title: "Other interval blocker", status: "blocked", priority: "medium" }); + const otherReactive = await card(); + companyId = originalCompany; agentId = originalAgent; + const service = statusCardEvaluationService(db, { wakeup }); + const originalRun = service.run; + let running: Promise | undefined; + const run = vi.spyOn(service, "run").mockImplementation(signal => { + const result = originalRun(signal); running = result; return result; + }); + await start(service); await settled(); + await vi.advanceTimersByTimeAsync(1000); + // Give each committed producer and any actual domain work time to settle. + // This prevents incidental overlap from hiding per-write reconstruction. + const flush = async () => { + for (let i = 0; i < 10; i++) { + await running; await immediate(); + } + }; + await flush(); + const select = vi.spyOn(db, "select"); + const factory = companySearch.companySearchService; + const searchedCompanies: string[] = []; + vi.spyOn(companySearch, "companySearchService").mockImplementation(owner => { + const search = factory(owner); + const actual = search.search; + search.search = async (...args) => { searchedCompanies.push(args[0]); return actual(...args); }; + return search; + }); + run.mockClear(); + const burstStarted = Date.now(); + for (let i = 0; i < 50; i++) { + const issue = await issueService(db).create(companyId, { title: `Burst blocker ${i}`, status: "blocked", priority: "medium" }); + await flush(); + await issueService(db).addComment(issue.id, `Human update ${i}`, { userId: "board-user" }); + await flush(); + } + await flush(); + // One eager reconciliation after idle and one bounded batched continuation; + // the producers themselves are not counted as background SQL. + const dueReads = () => select.mock.calls.filter(([fields]) => fields && "card" in fields && fields.card === statusCards).length; + const nextReads = () => select.mock.calls.filter(([fields]) => fields && "at" in fields && fields.at === statusCards.nextEvalAt).length; + const elapsedMs = Date.now() - burstStarted; + const maximum = Math.ceil(elapsedMs / 1000) + 2; + console.info("status-card burst", { elapsedMs, maximum, runs: run.mock.calls.length, due: dueReads(), next: nextReads(), searches: searchedCompanies.length }); + expect(run.mock.calls.length).toBeLessThanOrEqual(maximum); + expect(dueReads()).toBeLessThanOrEqual(maximum); expect(nextReads()).toBeLessThanOrEqual(maximum); + expect(searchedCompanies.length).toBeLessThanOrEqual(maximum); + expect(searchedCompanies.length).toBeGreaterThan(0); + expect(new Set(searchedCompanies)).toEqual(new Set([originalCompany])); + expect(wakeup).not.toHaveBeenCalled(); + // Re-register with an existing durable deadline inside the coalescing + // window. A company hint must not make that real deadline wait one second. + await coordinator!.stop(); coordinator = undefined; + const dueAt = Date.now() + 500; + await db.update(statusCards).set({ nextEvalAt: new Date(dueAt), nextEvalReason: "refresh" }).where(eq(statusCards.id, other.id)); + await start(service); await flush(); + await issueService(db).addComment(otherIssue.id, "Another company's coalesced human update", { userId: "board-user" }); + await issueService(db).addComment((await db.select({ id: issues.id }).from(issues).where(eq(issues.companyId, originalCompany)))[0]!.id, + "More changes while another company is due", { userId: "board-user" }); + await flush(); + await vi.advanceTimersByTimeAsync(500); await flush(); + await vi.waitFor(() => expect(wakeup).toHaveBeenCalledOnce()); + expect(Date.now()).toBeLessThan(dueAt + 500); + expect(wakeup.mock.calls[0]![0]).not.toBe(originalAgent); + const generated = (await statusCardService(db).getById(other.id))!; + expect(generated.generatingIssueId).toBeTruthy(); + expect((await statusCardService(db).getById(reactive.id))!.pendingChangeCount).toBeGreaterThan(0); + expect((await statusCardService(db).getById(otherReactive.id))!.pendingChangeCount).toBeGreaterThan(0); + expect(searchedCompanies).toContain(other.companyId); + // Complete all ordinary results, then prove there is no recurring batch + // timer once the requested work has settled. + companyId = other.companyId; agentId = other.agentId!; + await writeback(other.id, generated.generatingIssueId!); + await statusCardService(db).update((await statusCardService(db).getById(other.id))!, { archived: true }, actor); + companyId = originalCompany; agentId = originalAgent; + await vi.advanceTimersByTimeAsync(120_000); await settled(); await flush(); + await vi.waitFor(async () => expect((await statusCardService(db).getById(reactive.id))!.generatingIssueId).toBeTruthy()); + const pending = (await statusCardService(db).getById(reactive.id))!; + expect(pending.generatingIssueId).toBeTruthy(); + await writeback(reactive.id, pending.generatingIssueId!); + const otherPending = (await statusCardService(db).getById(otherReactive.id))!; + expect(otherPending.generatingIssueId).toBeTruthy(); + companyId = other.companyId; agentId = other.agentId!; + await writeback(otherReactive.id, otherPending.generatingIssueId!); + companyId = originalCompany; agentId = originalAgent; + select.mockRestore(); + await quiet(); expect(errors).toEqual([]); + }, 30_000); + it("cancels archived or disabled future work and replaces an earlier interval deadline", async () => { await start(); let created = await card("interval"); await settled(); const old = coordinator!.nextWakeAt(); created = await statusCardService(db).update(created, { refreshPolicy: { ...policy("interval"), intervalMinutes: 1 } }, actor); - await settled(); expect(coordinator!.nextWakeAt()).toBeLessThan(old!); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(100); + await Promise.allSettled([...evaluations]); + expect(coordinator!.nextWakeAt()).toBeLessThan(old!); + }); await statusCardService(db).update(created, { archived: true }, actor); await quiet(); expect(wakeup).not.toHaveBeenCalled(); expect(errors).toEqual([]); }); diff --git a/server/src/services/status-card-evaluation.ts b/server/src/services/status-card-evaluation.ts index 3e33c0d0885..e6086535fc9 100644 --- a/server/src/services/status-card-evaluation.ts +++ b/server/src/services/status-card-evaluation.ts @@ -90,5 +90,9 @@ export function registerStatusCardEvaluationWork( ) { return registerDeadlineWork(coordinator, DELIVERY_QUEUES.statusCardEvaluation, { run: service.run, next: service.next, onSettled: service.changed, canRun, retryMs: 30_000, + // Batch source hints for at most one second after reconciliation. Do not + // use minimumSpacingMs: known policy/retry deadlines must preempt batching. + // This window alone arms no timer when the feature has no pending work. + notificationCoalesceMs: 1_000, }); }