diff --git a/server/src/__tests__/ownership-reconciliation-report.test.ts b/server/src/__tests__/ownership-reconciliation-report.test.ts index 402706d8841..bf4a4b371b4 100644 --- a/server/src/__tests__/ownership-reconciliation-report.test.ts +++ b/server/src/__tests__/ownership-reconciliation-report.test.ts @@ -34,7 +34,7 @@ import { applyOwnershipReportBackstops, registerOwnershipReconciliationWork } fr import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js"; import { notifyNativeFinalizationWork } from "../services/native-finalization-notifications.js"; import { instanceSettingsService } from "../services/instance-settings.js"; -import { listUndispatchedQueuedRuns } from "../services/queued-run-dispatch-liveness.js"; +import { listUndispatchedQueuedRuns, readUndispatchedQueuedRuns } from "../services/queued-run-dispatch-liveness.js"; import { releaseLapsedCoordinatorOwner } from "../services/workspace-recovery-stop-proof.js"; import { subscribeDatabaseWork } from "@paperclipai/db/work-signals"; import { @@ -1072,6 +1072,38 @@ describe("report-only ownership reconciliation", () => { } }); + it("reads undispatched queued runs in bounded slices that resume from a cursor, and skips an agent with a scheduled retry", async () => { + const created = [30, 29, 28].map(minutesAgo); + const runs = await db.insert(heartbeatRuns).values(created.map(createdAt => + ({ companyId, agentId, status: "queued", invocationSource: "automation", createdAt }))).returning(); + const first = await readUndispatchedQueuedRuns(db, { pageSize: 1, maxRuns: 2 }); + expect(first.runs.map(run => run.runId)).toEqual([runs[0]!.id, runs[1]!.id]); + expect(first.next).not.toBeNull(); + const rest = await readUndispatchedQueuedRuns(db, { pageSize: 1, maxRuns: 2, after: first.next }); + expect(rest).toEqual({ runs: [expect.objectContaining({ runId: runs[2]!.id })], next: null }); + await db.insert(heartbeatRuns).values({ companyId, agentId, status: "scheduled_retry", invocationSource: "automation", + scheduledRetryAt: new Date(Date.now() + 60_000) }); + expect(await listUndispatchedQueuedRuns(db)).toEqual([]); + }); + + it("keeps the earliest grace deadline when the queued-run check reaches its cap, and rescans once it ends", async () => { + const createdAt = minutesAgo(1); + await db.insert(heartbeatRuns).values(Array.from({ length: 2_000 }, () => + ({ companyId, agentId, status: "queued", invocationSource: "automation", createdAt }))); + const dueAt = createdAt.getTime() + 5 * 60_000; + const state = createOwnershipReportState(); + const queuedRuns = (report: OwnershipReport) => report.violations.filter(violation => violation.kind === "queued_run"); + const first = await runOwnershipReconciliationPass(db, { state, record: false }); + expect(queuedRuns(first)).toEqual([]); + expect(first.retryNeeded).toBe(true); + expect(first.nextDueAt?.getTime()).toBeLessThanOrEqual(dueAt); + const tail = await runOwnershipReconciliationPass(db, { state, record: false }); + expect(queuedRuns(tail)).toEqual([]); + expect(tail.nextDueAt?.getTime()).toBeLessThanOrEqual(dueAt); + const overdue = await runOwnershipReconciliationPass(db, { state, record: false, now: new Date(dueAt + 1) }); + expect(queuedRuns(overdue)).toHaveLength(2_000); + }, 60_000); + it("signals queued-run dispatch when finalization work settles or a finished coordinator's lapsed owner is released", async () => { const issueId = randomUUID(); const runId = randomUUID(); diff --git a/server/src/services/queued-run-dispatch-liveness.ts b/server/src/services/queued-run-dispatch-liveness.ts index 8e15d888ced..fe5d2ca09d1 100644 --- a/server/src/services/queued-run-dispatch-liveness.ts +++ b/server/src/services/queued-run-dispatch-liveness.ts @@ -18,7 +18,8 @@ export type UndispatchedQueuedRun = { createdAt: Date; }; -type PageCursor = { at: string; runId: string }; +export type UndispatchedQueuedRunCursor = { at: string; runId: string }; +type PageCursor = UndispatchedQueuedRunCursor; async function listUndispatchedQueuedRunPage(db: Db, companyId: string | undefined, createdAfter: Date | null | undefined, after: PageCursor | null, limit: number): @@ -29,7 +30,7 @@ async function listUndispatchedQueuedRunPage(db: Db, companyId: string | undefin agentId: heartbeatRuns.agentId, issueId: sql`coalesce(${heartbeatRuns.issueId}::text, ${heartbeatRuns.contextSnapshot}->>'issueId')`, createdAt: heartbeatRuns.createdAt, - cursorAt: sql`${heartbeatRuns.createdAt}::text`, + cursorAt: sql`to_char(${heartbeatRuns.createdAt} at time zone 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.US"Z"')`, }).from(heartbeatRuns) .innerJoin(companies, eq(companies.id, heartbeatRuns.companyId)) .innerJoin(agents, and(eq(agents.id, heartbeatRuns.agentId), eq(agents.companyId, heartbeatRuns.companyId))) @@ -43,7 +44,7 @@ async function listUndispatchedQueuedRunPage(db: Db, companyId: string | undefin after ? sql`(${heartbeatRuns.createdAt}, ${heartbeatRuns.id}) > (${after.at}::timestamptz, ${after.runId}::uuid)` : undefined, sql`not exists (select 1 from ${heartbeatRuns} live where live.company_id = ${heartbeatRuns.companyId} and live.agent_id = ${heartbeatRuns.agentId} - and live.status = 'running')`, + and live.status in ('running', 'scheduled_retry'))`, )) .orderBy(asc(heartbeatRuns.createdAt), asc(heartbeatRuns.id)) .limit(limit); @@ -55,15 +56,26 @@ async function listUndispatchedQueuedRunPage(db: Db, companyId: string | undefin export async function listUndispatchedQueuedRuns(db: Db, options: { companyId?: string; createdAfter?: Date | null; pageSize?: number } = {}): Promise { + return (await readUndispatchedQueuedRuns(db, options)).runs; +} + +/** Reads from `after` until the set ends or `maxRuns` rows were read. `next` + * is where a later read continues, or null when the set was read to its end. */ +export async function readUndispatchedQueuedRuns(db: Db, options: { companyId?: string; createdAfter?: Date | null; + pageSize?: number; after?: UndispatchedQueuedRunCursor | null; maxRuns?: number } = {}): + Promise<{ runs: UndispatchedQueuedRun[]; next: UndispatchedQueuedRunCursor | null }> { const pageSize = options.pageSize ?? UNDISPATCHED_PAGE_SIZE; + const maxRuns = options.maxRuns ?? Number.POSITIVE_INFINITY; const runs: UndispatchedQueuedRun[] = []; - let after: PageCursor | null = null; + let after: PageCursor | null = options.after ?? null; for (;;) { - const page = await listUndispatchedQueuedRunPage(db, options.companyId, options.createdAfter, after, pageSize); + const want = Math.min(pageSize, maxRuns - runs.length); + const page = await listUndispatchedQueuedRunPage(db, options.companyId, options.createdAfter, after, want); runs.push(...page.map(({ cursorAt: _cursorAt, ...run }) => run)); - if (page.length < pageSize) return runs; + if (page.length < want) return { runs, next: null }; const last = page[page.length - 1]!; after = { at: last.cursorAt, runId: last.runId }; + if (runs.length >= maxRuns) return { runs, next: after }; } } diff --git a/server/src/services/recovery/ownership-invariant-report.ts b/server/src/services/recovery/ownership-invariant-report.ts index 0287180fc2c..ebf821f2da5 100644 --- a/server/src/services/recovery/ownership-invariant-report.ts +++ b/server/src/services/recovery/ownership-invariant-report.ts @@ -15,7 +15,8 @@ import { deliveredQueuedCommentIds } from "../queued-comment-delivery.js"; import { logger } from "../../middleware/logger.js"; import { getServerProcessStartedAt } from "../../server-info.js"; import { persistActivity } from "../activity-log.js"; -import { listUndispatchedQueuedRuns, QUEUED_RUN_NOT_DISPATCHED, readQueuedRunExecutionWindow } from "../queued-run-dispatch-liveness.js"; +import { QUEUED_RUN_NOT_DISPATCHED, readQueuedRunExecutionWindow, readUndispatchedQueuedRuns, + type UndispatchedQueuedRunCursor } from "../queued-run-dispatch-liveness.js"; import { currentProcessProbe, proveOwnerGone, type OwnerProof, type ProcessProbe } from "../owner-proof.js"; import { LEGACY_OWNER_REASONS, recordVersionAttestations } from "./owner-gone-attestation.js"; import { @@ -103,7 +104,8 @@ const COVERAGE_GAPS: Record = { I8_fast_turns: "Measured per run by the speed-budget check in the run log, not by this pass.", I7_scope: "I7 measures agent-assigned in_progress issues. todo, in_review and blocked issues are not measured yet.", record_cap: "Each kind and the I7 check read at most 2,000 records per pass; a larger kind is reported as truncated. " - + "The I3 and I4 scans read ten pages per pass and continue from a cursor on the next pass.", + + "The I3 and I4 scans read ten pages per pass and continue from a cursor on the next pass. The I7 queued-run " + + "check reads at most 2,000 undispatched queued runs per pass and continues from a cursor on the next pass.", wake_coverage: "Writers of workspace runtime services, remote workspace operations, snapshots, restore points, " + "git cache intents, provider artifacts and harness backups publish no work signal. Those kinds are read at " + "startup and on any other report wake. Phase 2 adds a signal when each kind moves to the shared settle path.", @@ -127,11 +129,16 @@ export type OwnershipReportState = { admissionRunCursor: { at: string; id: string } | null; /** True when an I3 scan stopped with more rows to read. */ admissionBacklog: boolean; + /** Keyset cursor of the I7 queued-run check (created_at, id); null starts over. */ + queuedRunCursor: UndispatchedQueuedRunCursor | null; + /** Earliest grace deadline seen in the earlier slices of the current queued-run scan. */ + queuedRunDueAt: Date | null; }; export function createOwnershipReportState(): OwnershipReportState { return { goneLastPass: new Set(), reported: new Set(), intentCursor: null, unwritten: new Map(), intentBacklog: false, - admissionWakeCursor: null, admissionRunCursor: null, admissionBacklog: false }; + admissionWakeCursor: null, admissionRunCursor: null, admissionBacklog: false, queuedRunCursor: null, + queuedRunDueAt: null }; } const TERMINAL_RUN_STATUSES = ["succeeded", "failed", "cancelled", "timed_out", "interrupted"]; @@ -644,10 +651,18 @@ async function checkVisibleNextAction(db: Db, now: Date, limit: number): Promise /** A queued run counts as an issue's next action only while dispatch can still * start it. One that stays queued past the grace window while its ready agent * runs nothing has lost its dispatch signal (invariant 7). */ -async function checkQueuedRunsDispatched(db: Db, now: Date, limit: number): Promise<{ violations: InvariantViolation[]; nextDueAt: Date | null }> { +async function checkQueuedRunsDispatched(db: Db, state: OwnershipReportState, now: Date, limit: number): + Promise<{ violations: InvariantViolation[]; nextDueAt: Date | null }> { const window = await readQueuedRunExecutionWindow(db); - if (!window.dispatches) return { violations: [], nextDueAt: null }; - const runs = await listUndispatchedQueuedRuns(db, { pageSize: limit, createdAfter: window.createdAfter }); + if (!window.dispatches) { + state.queuedRunCursor = null; state.queuedRunDueAt = null; + return { violations: [], nextDueAt: null }; + } + if (state.queuedRunDueAt && state.queuedRunDueAt <= now) { state.queuedRunCursor = null; state.queuedRunDueAt = null; } + const { runs, next } = await readUndispatchedQueuedRuns(db, { pageSize: limit, createdAfter: window.createdAfter, + after: state.queuedRunCursor, maxRuns: MAX_RECORDS_PER_KIND }); + state.queuedRunCursor = next; + const carried = state.queuedRunDueAt; const cutoff = now.getTime() - QUEUED_RUN_DISPATCH_GRACE_MS; const violations: InvariantViolation[] = []; let nextDueAt: Date | null = null; @@ -659,6 +674,8 @@ async function checkQueuedRunsDispatched(db: Db, now: Date, limit: number): Prom nextDueAt ??= new Date(run.createdAt.getTime() + QUEUED_RUN_DISPATCH_GRACE_MS); } } + if (carried && (!nextDueAt || carried < nextDueAt)) nextDueAt = carried; + state.queuedRunDueAt = next ? nextDueAt : null; return { violations, nextDueAt }; } @@ -777,7 +794,7 @@ export async function runOwnershipReconciliationPass(db: Db, options: { return result.violations; }, async () => { - const result = await checkQueuedRunsDispatched(db, now, limit); + const result = await checkQueuedRunsDispatched(db, state, now, limit); if (result.nextDueAt && (!nextDueAt || result.nextDueAt < nextDueAt)) nextDueAt = result.nextDueAt; return result.violations; }, @@ -805,7 +822,7 @@ export async function runOwnershipReconciliationPass(db: Db, options: { // Only a first proof needs a second pass; a record already reported for I2 // is not a reason to keep scanning. - if (state.intentBacklog || state.admissionBacklog) retryNeeded = true; + if (state.intentBacklog || state.admissionBacklog || state.queuedRunCursor) retryNeeded = true; const awaitingSecondProof = [...goneThisPass].filter(key => !state.goneLastPass.has(key)).length; state.goneLastPass = goneThisPass; // Violations whose activity row failed last time are written again, even