diff --git a/doc/plans/2026-10-30-issue-recovery-sleep.md b/doc/plans/2026-10-30-issue-recovery-sleep.md index 18670cc66b8..f191d5dca42 100644 --- a/doc/plans/2026-10-30-issue-recovery-sleep.md +++ b/doc/plans/2026-10-30-issue-recovery-sleep.md @@ -1,14 +1,18 @@ # Issue recovery without periodic scans -Four existing owners reconstruct their current obligations once at startup. Committing source writes invalidate only their company. The coordinator keeps scopes, per-issue deadlines, and per-entry failures in memory. Its pending check does not query the database. +The issue recovery owners reconstruct their current obligations once at startup. Committing source writes invalidate only their company. The coordinator keeps scopes, per-issue deadlines, and per-entry failures in memory. Its pending check does not query the database. | Owner | Durable truth and producer boundary | Deadline or event | Exit | | --- | --- | --- | --- | | Stranded assignment | Visible assigned todo/in-progress/review issues; issue create/update/checkout/release; run lifecycle and execution control; agent/company/budget/configuration; interaction, approval and subtree hold changes | A committed eligibility change. A transient continuation uses the latest finish plus the existing exponential delay. | An execution path, durable wait, escalation, or current ineligibility. | +| Orphan blocker assignment | Agent-created unassigned todo/blocked blockers with a nonterminal dependent; issue, relation, agent and company changes | Once per changed company, independent of assignment candidates. | Assigned, no longer blocking, or creator cannot execute until its next eligibility change. | +| Active recovery actions | Active/escalated action records, including actions on completed sources; action, issue, run and wait changes | Once per changed company, independent of assignment candidates. | Resolved, restored path, explicit hold or a separately owned retry. | | Resolved dependency | Blocked assigned issues and company-scoped blocker relations; blocker status or relation changes through issue service and import; run/control and pause gate changes | The committing change. A failed enqueue retains that issue with error backoff. | Existing idempotent wake, live path, unresolved dependency or deliberate self-block. | | Task watchdog | Active issue watchdogs and their current source subtree; normal watchdog upsert/disable; issue/comment/document/work-product and wait changes | Exact source creation plus the existing 15-second first-run grace. An incomplete scan retains the watchdog with error backoff. | Live path, open/reviewed reusable review, disabled configuration, or ineligible source. | | Stale issue lock | Existing checkout/execution run references; issue and run lifecycle changes | A committing ownership change, or the fresh owning controller's saved lease expiry. | No stale reference, a valid owner, or the existing guarded terminalization and lock clear. | +Reconstruction uses keyset pages of at most100rows. Repeated company hints coalesce reconstruction for30seconds. Already-known entry deadlines preempt that spacing without an early probe on each comment. Scheduling suppression is read once per pass. A failed read keeps scoped backoff debt. The two subsidiary company repairs do not run inside the per-issue loop. Legacy direct aggregate API calls retain both repairs. + The owner methods accept company and issue/watchdog identity. A narrowed lock check still reads other same-company references before using terminal-issue authority. A stale terminal alias cannot end an active issue's shared run. The controller lease owner supplies its fresh expiry; the existing write predicate revalidates the observed boot ID and database-clock expiry. Producer coverage uses the existing run-dispatch, execution-control, issue-review-state, native status reassessment and external conversation settlement boundaries. Direct gaps have explicit signals inside their owning transaction: document writes/import, attachment removal, blocker import, recovery-action adoption/mutation, conversation pause release, Telegram failure comment replacement, native artifact title/metadata updates and company deletion. Metadata-only comment classification that does not change timestamp, visibility or execution eligibility needs no additional watchdog activation. Approval links and decisions use their existing issue-review/continuation effects. Work-product create/update/delete/import uses its existing native status boundary. @@ -16,3 +20,5 @@ Producer coverage uses the existing run-dispatch, execution-control, issue-revie The new owners replace only these four startup and periodic calls. Deferred comments, silent active runs and residual orphan/resource recovery retain their existing backstop until the next slice. Every previously migrated native-owner skip remains unchanged. A controlled process test prints an execution proof but cannot apply a task disposition. The test operator completes that task through issue service. If the existing continuation policy already saved a disposition-repair retry, the test preserves it until its stored deadline. Only after all provider work has joined does it advance the clock; dispatch then revalidates the completed task and cancels the retry. The final quiet proof requires no registered deadlines, no tracked in-flight work and no owner SQL over twenty former polling minutes. It does not claim a saved future retry is already drained. + +Active action repair has no separate grace or lease countdown. Its delayed disposition repair first writes a scheduled-retry run, then mirrors the date into action wakePolicy.retryAt and timeoutAt. The dispatch owner owns that saved run deadline. Other skipped gates wait for their existing execution, wait, pause, budget, or finalizer producer; the company action worker does not add a polling fallback. diff --git a/server/src/__tests__/company-portability.test.ts b/server/src/__tests__/company-portability.test.ts index fdf00b828af..f6dbbf552b6 100644 --- a/server/src/__tests__/company-portability.test.ts +++ b/server/src/__tests__/company-portability.test.ts @@ -4168,6 +4168,7 @@ describe("company portability", () => { const insertedRelationValues: Array> = []; const monitorUpdates: Array> = []; const db = { + transaction: async (write: (tx: unknown) => Promise) => write(db), insert: () => ({ values: (rows: Array>) => ({ onConflictDoNothing: async () => { diff --git a/server/src/__tests__/issue-recovery-deadlines.test.ts b/server/src/__tests__/issue-recovery-deadlines.test.ts index 4faac4ab535..b3d65f4dbcc 100644 --- a/server/src/__tests__/issue-recovery-deadlines.test.ts +++ b/server/src/__tests__/issue-recovery-deadlines.test.ts @@ -2,14 +2,18 @@ import { idleWorkSnapshot } from "../services/task-admission.js"; import { randomUUID } from "node:crypto"; import { and, eq, sql } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; -import { agents, authUsers, companies, companyMemberships, createDb, heartbeatRuns, issues, issueWatchdogs } from "@paperclipai/db"; +import { agents, authUsers, companies, companyMemberships, createDb, heartbeatRuns, issueRecoveryActions, issueRelations, issues, issueWatchdogs } from "@paperclipai/db"; import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; import { heartbeatService } from "../services/heartbeat.js"; import { issueService } from "../services/issues.js"; +import { issueRecoveryActionService } from "../services/issue-recovery-actions.js"; +import { notifyIssueRecovery } from "../services/issue-recovery-notifications.js"; import { taskWatchdogService } from "../services/task-watchdogs.js"; import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +vi.setConfig({ testTimeout: 90_000 }); + const external = resolveExternalTestDatabaseUrl("PAPERCLIP_ISSUE_RECOVERY_TEST_DATABASE_URL"); const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); describe.skipIf(!support.supported)("issue recovery production producers and deadlines", () => { @@ -81,13 +85,14 @@ describe.skipIf(!support.supported)("issue recovery production producers and dea expect(wake, detail).toBeNull(); expect(extraDeadlines.map(next => next())).toEqual(extraDeadlines.map(() => null)); expect(idleWorkSnapshot().active).toBe(0); - }, { timeout: 10_000 }); + }, { timeout: 40_000 }); } async function quiet() { await settled(); const reads = vi.spyOn(db, "select"); const executes = vi.spyOn(db, "execute"); - vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); - try { await vi.advanceTimersByTimeAsync(20 * 60_000); expect(reads).not.toHaveBeenCalled(); expect(executes).not.toHaveBeenCalled(); } + const selected = reads.mock.calls.length, executed = executes.mock.calls.length; + if (!vi.isFakeTimers()) vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); + try { await vi.advanceTimersByTimeAsync(20 * 60_000); expect(reads).toHaveBeenCalledTimes(selected); expect(executes).toHaveBeenCalledTimes(executed); } finally { vi.useRealTimers(); reads.mockRestore(); executes.mockRestore(); } } async function finish(issueId: string) { @@ -95,12 +100,10 @@ describe.skipIf(!support.supported)("issue recovery production producers and dea await heartbeat.drainActiveRunExecutions(); const saved = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.status, "scheduled_retry"), sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`)); + expect(await db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.status} in ('queued', 'running')`)).toHaveLength(0); + if (!vi.isFakeTimers()) vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); if (saved.length > 0) { const due = Math.max(...saved.map(run => run.scheduledRetryAt!.getTime())); - expect(await db.select().from(heartbeatRuns).where(and( - sql`${heartbeatRuns.status} in ('queued', 'running')`, - sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`))).toHaveLength(0); - vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); await vi.advanceTimersByTimeAsync(Math.max(0, due - Date.now()) + 1); await vi.waitFor(async () => { const remaining = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.status, "scheduled_retry"), @@ -120,20 +123,85 @@ describe.skipIf(!support.supported)("issue recovery production producers and dea const run = runs.find(row => row.status === "running" || row.status === "succeeded"); expect(run).toBeDefined(); runId = run!.id; }, { timeout: 15_000 }); - await heartbeat.drainActiveRunExecutions(); + await vi.waitFor(async () => { + await heartbeat.drainActiveRunExecutions(); + const [current] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId!)); + expect(current.status, JSON.stringify({ error: current.error, code: current.errorCode, stderr: current.stderrExcerpt })).toBe("succeeded"); + }, { timeout: 15_000 }); const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId!)); - expect(run.status, JSON.stringify({ error: run.error, code: run.errorCode, stderr: run.stderrExcerpt, result: run.resultJson })).toBe("succeeded"); expect(run.stdoutExcerpt).toContain("recovery-proof"); + expect(run.stdoutExcerpt).toContain("recovery-proof"); return run; } it("keeps empty, configured idle, and terminal history quiet", async () => { await fixture(); await start(); await quiet(); }); + it("creates a blocking child through the ordinary root service transaction", async () => { + const f = await fixture(); + const parent = await issueService(db).create(f.companyId, { title: "Parent", status: "todo" }); + const child = await issueService(db).createChild(parent.id, { title: "Blocking child", status: "todo", blockParentUntilDone: true }); + expect(child.parentBlockerAdded).toBe(true); + expect(await db.select().from(issueRelations).where(and(eq(issueRelations.companyId, f.companyId), + eq(issueRelations.issueId, child.issue.id), eq(issueRelations.relatedIssueId, parent.id)))).toHaveLength(1); + }); + it.each(["startup", "committed"])("restores blocker-only and completed-action debt with no assignment candidates (%s)", async mode => { + const f = await fixture(); + if (mode === "committed") { await start(); await quiet(); } + const blocker = await issueService(db).create(f.companyId, { title: "Orphan blocker", status: "blocked", createdByAgentId: f.agentId }); + const dependent = await issueService(db).create(f.companyId, { title: "Blocked dependent", status: "blocked", blockedByIssueIds: [blocker.id] }); + const completedSource = await issueService(db).create(f.companyId, { title: "Completed source", status: "done" }); + const action = await issueRecoveryActionService(db).upsertSourceScoped({ companyId: f.companyId, sourceIssueId: completedSource.id, + kind: "stranded_assigned_issue", ownerType: "agent", ownerAgentId: f.agentId, cause: "stranded_assigned_issue", + fingerprint: "source-terminal", nextAction: "Restore work", wakePolicy: { type: "wake_owner" }, evidence: {} }); + if (mode === "startup") await start(); + await vi.waitFor(async () => { + expect((await db.select().from(issues).where(eq(issues.id, blocker.id)))[0].assigneeAgentId).toBe(f.agentId); + expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, action.id)))[0].status).toBe("resolved"); + }, { timeout: 40_000 }); + await finish(blocker.id); await finish(dependent.id); await quiet(); + }); + it("bounds subsidiary scans for100 assignment candidates and coalesces repeated company writes", async () => { + const f = await fixture(), foreign = await fixture(); + await db.update(agents).set({ lifecycleState: "preparing" }).where(eq(agents.id, f.agentId)); + await db.insert(issues).values(Array.from({ length: 100 }, (_, index) => ({ companyId: f.companyId, + title: `Waiting setup ${index}`, status: "todo", assigneeAgentId: f.agentId }))); + const [source] = await db.insert(issues).values({ companyId: f.companyId, title: "Done source", status: "done" }).returning(); + const action = await issueRecoveryActionService(db).upsertSourceScoped({ companyId: f.companyId, sourceIssueId: source.id, + kind: "stranded_assigned_issue", ownerType: "agent", ownerAgentId: f.agentId, cause: "stranded_assigned_issue", + fingerprint: "query-budget", nextAction: "Restore work", wakePolicy: { type: "wake_owner" }, evidence: {} }); + const scans = { blockers: 0, actions: 0, loads: 0 }; + const originalSelect = db.select.bind(db); + vi.spyOn(db, "select").mockImplementation(((fields?: Record) => { + const query = originalSelect(fields as never); const from = query.from.bind(query); + query.from = ((table: unknown) => { + if (table === issues && fields?.id === issues.id && fields.companyId === issues.companyId && Object.keys(fields).length === 2) scans.loads++; + if (table === issueRelations && fields?.createdByAgentId === issues.createdByAgentId) scans.blockers++; + if (table === issueRecoveryActions && fields?.action === issueRecoveryActions) scans.actions++; + return from(table as never); + }) as typeof query.from; + return query; + }) as typeof db.select); + await start(); await settled(); + expect(scans.actions).toBe(1); expect(scans.blockers).toBe(0); expect(scans.loads).toBeLessThanOrEqual(8); + expect((await originalSelect().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, action.id)))[0].status).toBe("resolved"); + const [foreignSource] = await db.insert(issues).values({ companyId: foreign.companyId, title: "Foreign done", status: "done" }).returning(); + const [foreignAction] = await db.insert(issueRecoveryActions).values({ companyId: foreign.companyId, sourceIssueId: foreignSource.id, + kind: "stranded_assigned_issue", ownerType: "board", cause: "stranded_assigned_issue", fingerprint: "foreign-query-budget", nextAction: "Inspect" }).returning(); + const before = { ...scans }; + for (let index = 0; index < 100; index++) await db.transaction(tx => notifyIssueRecovery(tx, { companyId: f.companyId })); + await new Promise(resolve => setTimeout(resolve, 100)); + expect(scans.loads).toBe(before.loads); + await settled(); + expect(scans.actions).toBe(before.actions); expect(scans.blockers).toBe(before.blockers); + expect(scans.loads).toBeLessThanOrEqual(before.loads + 4); + expect((await originalSelect().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, foreignAction.id)))[0].status).toBe("active"); + expect(await originalSelect().from(heartbeatRuns)).toHaveLength(0); await quiet(); + }); it("dispatches an ordinary new assignment after commit and returns to quiet after completion", async () => { const f = await fixture(); await start(); await quiet(); const issue = await issueService(db).create(f.companyId, { title: "Recover assignment", status: "todo", assigneeAgentId: f.agentId }); await completed(issue.id); await finish(issue.id); await quiet(); - }, 45_000); + }, 90_000); it("reacts to blocker resolution once and dispatches the durable dependency wake", async () => { const f = await fixture(); const blocker = await issueService(db).create(f.companyId, { title: "Blocker", status: "todo" }); @@ -147,7 +215,7 @@ describe.skipIf(!support.supported)("issue recovery production producers and dea await finish(dependent.id); await quiet(); expect(await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), sql`${heartbeatRuns.contextSnapshot}->>'wakeReason' = 'issue_blockers_resolved'`))).toHaveLength(1); - }, 45_000); + }, 90_000); it("uses ordinary watchdog configuration and subtree closure to dispatch a reusable review", async () => { const f = await fixture(); const source = await issueService(db).create(f.companyId, { title: "Finished subtree", status: "done" }); @@ -163,8 +231,9 @@ describe.skipIf(!support.supported)("issue recovery production producers and dea expect(await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), sql`${heartbeatRuns.contextSnapshot}->>'wakeReason' = 'task_watchdog_stopped_subtree'`))).toHaveLength(1); await finish(reviewId!); - await taskWatchdogService(db).disableForIssue(f.companyId, source.id); await quiet(); - }, 45_000); + await taskWatchdogService(db).disableForIssue(f.companyId, source.id); + if (!vi.isFakeTimers()) vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); await quiet(); + }, 90_000); it("bounds ordinary writes to their company and reconstructs foreign stale debt after restart", async () => { const selected = await fixture(); await start(); await quiet(); const foreign = await fixture(); @@ -198,19 +267,19 @@ describe.skipIf(!support.supported)("issue recovery production producers and dea })).rejects.toThrow("rollback control"); await quiet(); expect(await db.select().from(issues).where(eq(issues.title, "Nested rollback"))).toHaveLength(0); - }, 60_000); + }, 90_000); it("retains the first committed assignment during drain and executes on resume", async () => { const f = await fixture(); await start(); await quiet(); draining = true; const issue = await issueService(db).create(f.companyId, { title: "Drained assignment", status: "todo", assigneeAgentId: f.agentId }); await new Promise(resolve => setTimeout(resolve, 100)); const reads = vi.spyOn(db, "select"); - vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); + if (!vi.isFakeTimers()) vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); try { await vi.advanceTimersByTimeAsync(90_000); expect(reads).not.toHaveBeenCalled(); } finally { vi.useRealTimers(); reads.mockRestore(); } expect(await db.select().from(heartbeatRuns)).toHaveLength(0); draining = false; worker.wake(); dispatch.wake(); await completed(issue.id); await finish(issue.id); await quiet(); - }, 45_000); + }, 90_000); it("retains a fresh subtree's exact first-run grace then disarms on disable", async () => { const f = await fixture(); const source = await issueService(db).create(f.companyId, { title: "Fresh unassigned subtree", status: "todo" }); @@ -218,7 +287,8 @@ describe.skipIf(!support.supported)("issue recovery production producers and dea await taskWatchdogService(db).upsertForIssue(f.companyId, source.id, { agentId: f.agentId }); await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBe(source.createdAt.getTime() + 15_000), { timeout: 5_000 }); expect(await db.select().from(issues).where(eq(issues.originKind, "task_watchdog"))).toHaveLength(0); - await taskWatchdogService(db).disableForIssue(f.companyId, source.id); await quiet(); + await taskWatchdogService(db).disableForIssue(f.companyId, source.id); + if (!vi.isFakeTimers()) vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); await quiet(); }); it("waits for a live foreign controller's exact stored lease before clearing a terminal lock", async () => { const f = await fixture(); @@ -232,6 +302,7 @@ describe.skipIf(!support.supported)("issue recovery production producers and dea await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBe(lease.getTime()), { timeout: 3_000 }); expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0].status).toBe("running"); await vi.waitFor(async () => expect((await db.select().from(issues).where(eq(issues.id, source.id)))[0].executionRunId).toBeNull(), { timeout: 12_000 }); - expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0].status).toBe("succeeded"); await quiet(); - }, 30_000); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0].status).toBe("succeeded"); + if (!vi.isFakeTimers()) vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); await quiet(); + }, 90_000); }); diff --git a/server/src/__tests__/issue-recovery-work-lifecycle.test.ts b/server/src/__tests__/issue-recovery-work-lifecycle.test.ts index 4d0647f4406..52f465daa8e 100644 --- a/server/src/__tests__/issue-recovery-work-lifecycle.test.ts +++ b/server/src/__tests__/issue-recovery-work-lifecycle.test.ts @@ -1,6 +1,6 @@ import { PgDialect } from "drizzle-orm/pg-core"; import { afterEach, describe, expect, it, vi } from "vitest"; -import { issueWatchdogs } from "@paperclipai/db"; +import { issueRecoveryActions, issueWatchdogs } from "@paperclipai/db"; import { installDatabaseWorkSignals } from "../../../packages/db/src/work-signals.js"; import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; import { notifyIssueRecovery } from "../services/issue-recovery-notifications.js"; @@ -12,29 +12,34 @@ afterEach(async () => { for (const stop of stops.splice(0)) await stop(); vi.use function fixture() { vi.useFakeTimers(); vi.setSystemTime(1_000_000); const dialect = new PgDialect(); - const rows = { issues: [] as Array<{ id: string; companyId: string }>, watchdogs: [] as Array<{ id: string; companyId: string }> }; + const rows = { issues: [] as Array<{ id: string; companyId: string }>, watchdogs: [] as Array<{ id: string; companyId: string }>, blockers: [] as Array<{ id: string; companyId: string }>, actions: [] as Array<{ id: string; companyId: string }> }; const execute = vi.fn(async (query: any) => { if (dialect.sqlToQuery(query).sql.includes("pg_current_xact_id")) return [{ xid: "42" }]; throw new Error("unexpected SQL"); }); - const load = vi.fn((table: unknown) => table === issueWatchdogs ? rows.watchdogs : rows.issues); + const load = vi.fn((table: unknown, distinct = false) => distinct + ? table === issueRecoveryActions ? rows.actions : rows.blockers + : table === issueWatchdogs ? rows.watchdogs : rows.issues); + const query = (distinct = false) => ({ from: (table: unknown) => ({ where: () => ({ orderBy: () => ({ limit: async () => load(table, distinct) }) }) }) }); const db: any = installDatabaseWorkSignals({ execute, - select: () => ({ from: (table: unknown) => ({ where: async () => load(table) }) }), - transaction: async (callback: (tx: any) => Promise) => callback({ execute, select: db.select, transaction: db.transaction }), + select: () => query(), selectDistinct: () => query(true), + transaction: async (callback: (tx: any) => Promise) => callback({ execute, select: db.select, selectDistinct: db.selectDistinct, transaction: db.transaction }), }); const state = { draining: false, standby: false, suppressed: false, deadline: null as number | null, fail: false, incomplete: false }; const check = vi.fn(async (scope: { onDeferredDeadline(at: number): void }) => { if (state.fail) throw new Error("database unavailable"); if (state.deadline !== null && state.deadline > Date.now()) scope.onDeferredDeadline(state.deadline); }); - const owner = { stranded: check, dependencies: async (scope: any) => { await check(scope); return { enqueueFailed: 0 }; }, + const orphanBlockers = vi.fn(async () => {}), recoveryActions = vi.fn(async () => {}); + const suppression = vi.fn(async () => state.suppressed); + const owner = { orphanBlockers, recoveryActions, stranded: check, dependencies: async (scope: any) => { await check(scope); return { enqueueFailed: 0 }; }, locks: check, watchdogs: async (scope: any) => { await check(scope); return { incomplete: state.incomplete ? 1 : 0 }; }, - schedulingSuppressed: async () => state.suppressed }; + schedulingSuppressed: suppression }; const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !state.standby, canReconcile: () => !state.standby, onError: vi.fn() }); const worker = registerIssueRecoveryWork(db, coordinator, owner, () => !state.draining && !state.standby); stops.push(() => coordinator.stop()); - return { db, rows, state, check, load, coordinator, worker, + return { db, rows, state, check, load, coordinator, worker, orphanBlockers, recoveryActions, suppression, notify: () => db.transaction((tx: any) => notifyIssueRecovery(tx, { companyId: "selected" })), settle: () => vi.advanceTimersByTimeAsync(1_000) }; } @@ -45,6 +50,51 @@ describe("issue recovery obligation lifecycle", () => { await vi.advanceTimersByTimeAsync(20 * 60_000); expect(f.load).toHaveBeenCalledTimes(loaded); expect(f.check).not.toHaveBeenCalled(); }); + it("reads scheduling suppression once per pass instead of once per issue", async () => { + const f = fixture(); + f.rows.issues.push(...Array.from({ length: 25 }, (_, index) => ({ id: `issue-${index}`, companyId: "selected" }))); + await f.worker.ready; + expect(f.check).toHaveBeenCalledTimes(75); + expect(f.suppression).toHaveBeenCalledTimes(6); + expect(f.coordinator.nextWakeAt()).toBeNull(); + }); + it("backs off a failed suppression read without probing an empty or failed owner repeatedly", async () => { + const f = fixture(); f.suppression.mockRejectedValueOnce(new Error("settings unavailable")); + await f.worker.ready; const reads = f.suppression.mock.calls.length; + expect(f.coordinator.nextWakeAt()).toBe(1_030_000); + await vi.advanceTimersByTimeAsync(29_999); expect(f.suppression).toHaveBeenCalledTimes(reads); + await vi.advanceTimersByTimeAsync(1); expect(f.suppression).toHaveBeenCalledTimes(reads + 1); + expect(f.coordinator.nextWakeAt()).toBeNull(); + }); + it("normalizes a wide scope with cyclic SQL objects before publishing work hints", async () => { + const f = fixture(); await f.worker.ready; + const wide = { companyId: "selected", endpointId: issueRecoveryActions.id, table: issueRecoveryActions }; + await f.db.transaction((tx: object) => notifyIssueRecovery(tx, wide)); await f.settle(); + expect(f.coordinator.nextWakeAt()).toBeNull(); + const loads = f.load.mock.calls.length; + await vi.advanceTimersByTimeAsync(20 * 60_000); expect(f.load).toHaveBeenCalledTimes(loads); + }); + it("runs orphan blocker and action duties without assignment entries", async () => { + const f = fixture(); f.rows.blockers.push({ id: "selected", companyId: "selected" }); + f.rows.actions.push({ id: "selected", companyId: "selected" }); await f.worker.ready; + expect(f.check).not.toHaveBeenCalled(); + expect(f.orphanBlockers).toHaveBeenCalledExactlyOnceWith({ companyId: "selected" }); + expect(f.recoveryActions).toHaveBeenCalledExactlyOnceWith({ companyId: "selected" }); + expect(f.coordinator.nextWakeAt()).toBeNull(); + }); + it("coalesces repeated company reconstruction without delaying an existing lease deadline", async () => { + const f = fixture(); f.rows.issues.push({ id: "issue", companyId: "selected" }); + f.state.deadline = 1_015_000; await f.worker.ready; + await f.notify(); await f.settle(); const loaded = f.load.mock.calls.length, checked = f.check.mock.calls.length; + for (let index = 0; index < 100; index++) { await f.notify(); await vi.advanceTimersByTimeAsync(10); } + expect(f.load).toHaveBeenCalledTimes(loaded); expect(f.check).toHaveBeenCalledTimes(checked); + await vi.advanceTimersByTimeAsync(13_000); + expect(f.state.deadline).toBe(1_015_000); + expect(f.coordinator.nextWakeAt()).toBe(1_031_000); + await vi.advanceTimersByTimeAsync(16_000); + expect(f.load.mock.calls.length).toBeGreaterThan(loaded); + expect(f.coordinator.nextWakeAt()).toBeNull(); + }); it("retains exact grace and lease deadlines without early owner probes", async () => { const f = fixture(); f.rows.issues.push({ id: "issue", companyId: "selected" }); f.rows.watchdogs.push({ id: "watchdog", companyId: "selected" }); f.state.deadline = 1_015_000; @@ -85,7 +135,7 @@ describe("issue recovery obligation lifecycle", () => { const f = fixture(); f.rows.issues.push({ id: "issue", companyId: "selected" }); await f.worker.ready; let release!: () => void; const held = new Promise(resolve => { release = resolve; }); f.check.mockImplementationOnce(() => held); - await f.notify(); await f.settle(); await f.notify(); release(); await f.settle(); await f.settle(); + await f.notify(); await f.settle(); await f.notify(); release(); await f.settle(); await vi.advanceTimersByTimeAsync(30_000); expect(f.coordinator.nextWakeAt()).toBeNull(); expect(f.check.mock.calls.length).toBeGreaterThan(3); }); diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 1d9ac79f520..8a9fdbe5686 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -2,6 +2,8 @@ import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db"; /** Topics for existing durable queues; register intent before their writes. */ export const DELIVERY_QUEUES = { + orphanBlockers: "orphan-blockers", + activeRecoveryActions: "active-recovery-actions", strandedAssignments: "stranded-assignments", resolvedDependencies: "resolved-dependencies", taskWatchdogs: "task-watchdogs", diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 42dc9928ad5..068532357ea 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -4245,7 +4245,9 @@ export function heartbeatService( registerIssueStateRecoveryWork: (coordinator: Parameters[1], canRun: () => boolean) => registerIssueRecoveryWork(db, coordinator, { - stranded: scope => reconcileStrandedAssignedIssues({ ...scope, skipExternalConversationSettlement: true }), + stranded: scope => reconcileStrandedAssignedIssues({ ...scope, skipExternalConversationSettlement: true, skipRelatedRecovery: true }), + orphanBlockers: recovery.reconcileUnassignedBlockingIssues, + recoveryActions: recovery.reconcileActiveRecoveryActions, dependencies: reconcileResolvedDependencyWakes, locks: sweepStaleIssueLocks, watchdogs: reconcileTaskWatchdogs, diff --git a/server/src/services/issue-recovery-notifications.ts b/server/src/services/issue-recovery-notifications.ts index db04ca801ba..3c974228d86 100644 --- a/server/src/services/issue-recovery-notifications.ts +++ b/server/src/services/issue-recovery-notifications.ts @@ -3,8 +3,8 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notificatio export type IssueRecoveryScope = { companyId: string }; export async function notifyIssueRecovery(transaction: object, scope?: IssueRecoveryScope) { - const hint = scope ? JSON.stringify(scope) : "*"; - for (const queue of [DELIVERY_QUEUES.strandedAssignments, DELIVERY_QUEUES.resolvedDependencies, + const hint = scope ? JSON.stringify({ companyId: scope.companyId }) : "*"; + for (const queue of [DELIVERY_QUEUES.orphanBlockers, DELIVERY_QUEUES.activeRecoveryActions, DELIVERY_QUEUES.strandedAssignments, DELIVERY_QUEUES.resolvedDependencies, DELIVERY_QUEUES.taskWatchdogs, DELIVERY_QUEUES.staleIssueLocks]) { await notifyDeliveryWork(transaction, queue, hint); } diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 5a569f4ce1e..34234b01c98 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -9677,27 +9677,30 @@ export function issueService(db: Db) { }); if (blockParentUntilDone) { - const existingBlockers = await db - .select({ blockerIssueId: issueRelations.issueId }) - .from(issueRelations) - .where( - and( - eq(issueRelations.companyId, parent.companyId), - eq(issueRelations.relatedIssueId, parent.id), - eq(issueRelations.type, "blocks"), - ), + await db.transaction(async tx => { + const existingBlockers = await tx + .select({ blockerIssueId: issueRelations.issueId }) + .from(issueRelations) + .where( + and( + eq(issueRelations.companyId, parent.companyId), + eq(issueRelations.relatedIssueId, parent.id), + eq(issueRelations.type, "blocks"), + ), + ); + await syncBlockedByIssueIds( + parent.id, + parent.companyId, + [ + ...new Set([ + ...existingBlockers.map((row) => row.blockerIssueId), + child.id, + ]), + ], + { agentId: actorAgentId ?? null, userId: actorUserId ?? null }, + tx, ); - await syncBlockedByIssueIds( - parent.id, - parent.companyId, - [ - ...new Set([ - ...existingBlockers.map((row) => row.blockerIssueId), - child.id, - ]), - ], - { agentId: actorAgentId ?? null, userId: actorUserId ?? null }, - ); + }); [child] = await withIssueRelationSummaries( parent.companyId, [child], diff --git a/server/src/services/native-runtime/native-workspace-finalizer.test.ts b/server/src/services/native-runtime/native-workspace-finalizer.test.ts index 59270a2bb65..53063f345e9 100644 --- a/server/src/services/native-runtime/native-workspace-finalizer.test.ts +++ b/server/src/services/native-runtime/native-workspace-finalizer.test.ts @@ -131,7 +131,7 @@ describe("native workspace finalization recovery", () => { const external = resolveExternalTestDatabaseUrl("PAPERCLIP_NATIVE_WORKSPACE_FINALIZER_TEST_DATABASE_URL"); if (!external) temporary = await startEmbeddedPostgresTestDatabase("paperclip-native-workspace-finalizer-"); connectionString = external ?? temporary!.connectionString; - db = createDb(connectionString); + db = createDb(connectionString, { maxConnections: 6 }); await db.insert(companies).values([ { id: companyId, name: "Native workspace owner", issuePrefix: "NWO" }, @@ -452,7 +452,7 @@ describe("native workspace finalization recovery", () => { const seeded = await seedRun({ executionWorkspaceId: randomUUID(), title: "Live copyback" }); const live = await withNativeWorkspaceFinalizationOwnership({ db, companyId, runId: seeded.runId }, async (ownership) => { await ownership.assertHeld(); - expect(await resumeNativeWorkspaceFinalization({ db: createDb(connectionString), runId: seeded.runId })).toBeNull(); + expect(await resumeNativeWorkspaceFinalization({ db: createDb(connectionString, { maxConnections: 6 }), runId: seeded.runId })).toBeNull(); expect(await db.select().from(workspaceOperations).where(eq(workspaceOperations.heartbeatRunId, seeded.runId))).toEqual([]); return "copied"; }); @@ -476,7 +476,7 @@ describe("native workspace finalization recovery", () => { try { await entered; await db.execute(sql`select pg_terminate_backend(pid) from pg_locks where locktype = 'advisory' and database = (select oid from pg_database where datname = current_database()) and pid <> pg_backend_pid()`); - const second = await withNativeWorkspaceFinalizationOwnership({ db: createDb(connectionString), companyId, runId: seeded.runId }, async () => "must not run"); + const second = await withNativeWorkspaceFinalizationOwnership({ db: createDb(connectionString, { maxConnections: 6 }), companyId, runId: seeded.runId }, async () => "must not run"); expect(second).toEqual({ acquired: false, reason: "busy" }); const [held] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seeded.runId)); expect(held.runnerProfileJson?.nativeWorkspaceFinalizationOwner).toBeTruthy(); @@ -489,7 +489,7 @@ describe("native workspace finalization recovery", () => { it("recovers its own joined copyback receipt after the application pool disconnects during cleanup", async () => { const seeded = await seedRun({ executionWorkspaceId: randomUUID(), title: "Reconnect joined copyback" }); - const interruptedDb = createDb(connectionString); + const interruptedDb = createDb(connectionString, { maxConnections: 6 }); let joined = false; await expect(withNativeWorkspaceFinalizationOwnership({ db: interruptedDb, companyId, runId: seeded.runId }, async () => { await fs.writeFile(path.join(workspaceRoot, "joined-copyback.txt"), "durable work"); @@ -507,7 +507,7 @@ describe("native workspace finalization recovery", () => { .toEqual({ acquired: false, reason: "owner_unverified" }); } await db.update(heartbeatRuns).set({ runnerProfileJson: beforeReconnect.runnerProfileJson }).where(eq(heartbeatRuns.id, seeded.runId)); - expect(await withNativeWorkspaceFinalizationOwnership({ db: createDb(connectionString), companyId, runId: seeded.runId }, async () => "resumed")) + expect(await withNativeWorkspaceFinalizationOwnership({ db: createDb(connectionString, { maxConnections: 6 }), companyId, runId: seeded.runId }, async () => "resumed")) .toEqual({ acquired: true, value: "resumed" }); }); diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 7888b3e23fb..4b2ec5c0672 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -2203,7 +2203,7 @@ export function recoveryService( return Boolean(budgetBlock); } - async function reconcileUnassignedBlockingIssues() { + async function reconcileUnassignedBlockingIssues(opts?: { companyId?: string }) { const candidates = await db .select({ id: issues.id, @@ -2217,6 +2217,8 @@ export function recoveryService( .where( and( eq(issueRelations.type, "blocks"), + eq(issueRelations.companyId, issues.companyId), + opts?.companyId ? eq(issues.companyId, opts.companyId) : undefined, inArray(issues.status, ["todo", "blocked"]), isNull(issues.assigneeAgentId), isNull(issues.assigneeUserId), @@ -3501,7 +3503,7 @@ export function recoveryService( return Boolean(run || wake); } - async function reconcileActiveRecoveryActions() { + async function reconcileActiveRecoveryActions(opts?: { companyId?: string }) { const rows = await db .select({ action: issueRecoveryActions, @@ -3515,7 +3517,8 @@ export function recoveryService( eq(issues.companyId, issueRecoveryActions.companyId), ), ) - .where(inArray(issueRecoveryActions.status, ["active", "escalated"])); + .where(and(inArray(issueRecoveryActions.status, ["active", "escalated"]), + opts?.companyId ? eq(issueRecoveryActions.companyId, opts.companyId) : undefined)); const result = { requeued: 0, @@ -4532,6 +4535,7 @@ export function recoveryService( onDeferredDeadline?: (at: number) => void; issueCreatedAtGte?: Date | null; skipExternalConversationSettlement?: boolean; + skipRelatedRecovery?: boolean; }) { const candidates = await db .select(recoveryIssueFields) @@ -5780,16 +5784,18 @@ export function recoveryService( } } - const orphanBlockerRecovery = await reconcileUnassignedBlockingIssues(); - result.orphanBlockersAssigned = orphanBlockerRecovery.assigned; - result.skipped += orphanBlockerRecovery.skipped; - result.issueIds.push(...orphanBlockerRecovery.issueIds); + if (!opts?.skipRelatedRecovery) { + const orphanBlockerRecovery = await reconcileUnassignedBlockingIssues(opts); + result.orphanBlockersAssigned = orphanBlockerRecovery.assigned; + result.skipped += orphanBlockerRecovery.skipped; + result.issueIds.push(...orphanBlockerRecovery.issueIds); - const activeRecovery = await reconcileActiveRecoveryActions(); - result.continuationRequeued += activeRecovery.requeued; - result.escalated += activeRecovery.escalated; - result.skipped += activeRecovery.skipped; - result.issueIds.push(...activeRecovery.issueIds); + const activeRecovery = await reconcileActiveRecoveryActions(opts); + result.continuationRequeued += activeRecovery.requeued; + result.escalated += activeRecovery.escalated; + result.skipped += activeRecovery.skipped; + result.issueIds.push(...activeRecovery.issueIds); + } result.issueIds = [...new Set(result.issueIds)]; return result; @@ -6563,6 +6569,8 @@ export function recoveryService( recordWatchdogDecision, scanSilentActiveRuns, reconcileStrandedAssignedIssues, + reconcileUnassignedBlockingIssues, + reconcileActiveRecoveryActions, reconcileLegacyContinuation, legacyRepairDispatchBlock, sweepStaleIssueLocks, diff --git a/server/src/services/recovery/work.ts b/server/src/services/recovery/work.ts index 93679e80423..8bf7b75c7f4 100644 --- a/server/src/services/recovery/work.ts +++ b/server/src/services/recovery/work.ts @@ -1,6 +1,6 @@ import { registerTaskWatchdogWork } from "../task-watchdog-work.js"; -import { and, eq, inArray, isNotNull, isNull, or } from "drizzle-orm"; -import { issues, type Db } from "@paperclipai/db"; +import { and, asc, eq, gt, inArray, isNotNull, isNull, or, sql } from "drizzle-orm"; +import { issueRecoveryActions, issueRelations, issues, type Db } from "@paperclipai/db"; import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; import { DELIVERY_QUEUES, type DeliveryQueue } from "../delivery-work-notifications.js"; import type { IssueRecoveryScope } from "../issue-recovery-notifications.js"; @@ -8,12 +8,15 @@ import { registerScopedDeadlineWork } from "../scoped-deadline-work.js"; type Entry = { id: string; companyId: string; dueAt: number }; type Scope = { companyId: string; issueId: string; onDeferredDeadline: (at: number) => void }; +const PAGE_SIZE = 100; export function registerIssueRecoveryWork( db: Db, coordinator: ReturnType, owner: { stranded(scope: Scope): Promise; + orphanBlockers(scope: { companyId: string }): Promise; + recoveryActions(scope: { companyId: string }): Promise; dependencies(scope: Scope): Promise<{ enqueueFailed: number }>; locks(scope: Scope): Promise; watchdogs(scope: { companyId: string; watchdogId: string; onDeferredDeadline: (at: number) => void }): Promise<{ incomplete: number }>; @@ -22,20 +25,32 @@ export function registerIssueRecoveryWork( canRun: () => boolean, ) { function register(queue: DeliveryQueue, statuses: string[] | null, check: (scope: Scope) => Promise) { + const cursors = new Map(); + let enabled = false; return registerScopedDeadlineWork(coordinator, { queue, + reconstructionIntervalMs: 30_000, + beforePass: async () => { enabled = !(await owner.schedulingSuppressed()); }, decodeScope: hint => hint === "*" ? undefined : JSON.parse(hint) as IssueRecoveryScope, matches: (entry, scope) => !scope || entry.companyId === scope.companyId, load: async scope => { + const key = scope?.companyId ?? "*"; + if (!enabled) { cursors.delete(key); return []; } + const cursor = cursors.get(key); const rows = await db.select({ id: issues.id, companyId: issues.companyId }).from(issues).where(and( scope ? eq(issues.companyId, scope.companyId) : undefined, + cursor ? gt(issues.id, cursor) : undefined, statuses ? inArray(issues.status, statuses) : or(isNotNull(issues.checkoutRunId), isNotNull(issues.executionRunId)), statuses ? isNull(issues.hiddenAt) : undefined, - )); - return rows.map(row => ({ ...row, dueAt: Date.now() })); + statuses ? isNull(issues.assigneeUserId) : undefined, + statuses ? or(isNotNull(issues.assigneeAgentId), eq(issues.status, "in_review")) : undefined, + )).orderBy(asc(issues.id)).limit(PAGE_SIZE); + const complete = rows.length < PAGE_SIZE; + if (complete) cursors.delete(key); else cursors.set(key, rows.at(-1)!.id); + return { rows: rows.map(row => ({ ...row, dueAt: Date.now() })), complete }; }, check: async entry => { - if (await owner.schedulingSuppressed()) return null; + if (!enabled) return null; let next: number | null = null; await check({ companyId: entry.companyId, issueId: entry.id, onDeferredDeadline: at => { next = next === null ? at : Math.min(next, at); } }); @@ -43,7 +58,45 @@ export function registerIssueRecoveryWork( }, }, canRun); } + function registerCompany(queue: DeliveryQueue, kind: "blockers" | "actions", check: (scope: { companyId: string }) => Promise) { + const cursors = new Map(); + let enabled = false; + return registerScopedDeadlineWork(coordinator, { + queue, + reconstructionIntervalMs: 30_000, + beforePass: async () => { enabled = !(await owner.schedulingSuppressed()); }, + decodeScope: hint => hint === "*" ? undefined : JSON.parse(hint) as IssueRecoveryScope, + matches: (entry, scope) => !scope || entry.companyId === scope.companyId, + load: async scope => { + const key = scope?.companyId ?? "*"; + if (!enabled) { cursors.delete(key); return []; } + const cursor = cursors.get(key); + const rows = kind === "actions" + ? await db.selectDistinct({ id: issueRecoveryActions.companyId, companyId: issueRecoveryActions.companyId }) + .from(issueRecoveryActions).where(and(inArray(issueRecoveryActions.status, ["active", "escalated"]), + scope ? eq(issueRecoveryActions.companyId, scope.companyId) : undefined, + cursor ? gt(issueRecoveryActions.companyId, cursor) : undefined)) + .orderBy(asc(issueRecoveryActions.companyId)).limit(PAGE_SIZE) + : await db.selectDistinct({ id: issues.companyId, companyId: issues.companyId }).from(issues).where(and( + inArray(issues.status, ["todo", "blocked"]), isNull(issues.assigneeAgentId), isNull(issues.assigneeUserId), + isNotNull(issues.createdByAgentId), + scope ? eq(issues.companyId, scope.companyId) : undefined, + cursor ? gt(issues.companyId, cursor) : undefined, + sql`exists (select 1 from ${issueRelations} r inner join ${issues} dependent + on dependent.id = r.related_issue_id and dependent.company_id = r.company_id + where r.company_id = ${issues.companyId} and r.issue_id = ${issues.id} + and r.type = 'blocks' and dependent.status not in ('done', 'cancelled'))`, + )).orderBy(asc(issues.companyId)).limit(PAGE_SIZE); + const complete = rows.length < PAGE_SIZE; + if (complete) cursors.delete(key); else cursors.set(key, rows.at(-1)!.id); + return { rows: rows.map(row => ({ ...row, dueAt: Date.now() })), complete }; + }, + check: async entry => { if (enabled) await check({ companyId: entry.companyId }); return null; }, + }, canRun); + } const assignments = register(DELIVERY_QUEUES.strandedAssignments, ["todo", "in_progress", "in_review"], owner.stranded); + const orphanBlockers = registerCompany(DELIVERY_QUEUES.orphanBlockers, "blockers", owner.orphanBlockers); + const recoveryActions = registerCompany(DELIVERY_QUEUES.activeRecoveryActions, "actions", owner.recoveryActions); const dependencies = register(DELIVERY_QUEUES.resolvedDependencies, ["blocked"], async scope => { const result = await owner.dependencies(scope); if (result.enqueueFailed > 0) throw new Error("Resolved dependency wake remains undelivered"); @@ -52,7 +105,8 @@ export function registerIssueRecoveryWork( const watchdogs = registerTaskWatchdogWork(db, coordinator, { reconcile: owner.watchdogs, enabled: async () => !(await owner.schedulingSuppressed()), }, canRun); - return { ready: Promise.all([assignments.ready, dependencies.ready, locks.ready, watchdogs.ready]), - wake() { assignments.invalidate(); dependencies.invalidate(); locks.invalidate(); watchdogs.invalidate(); }, - async stop() { await Promise.all([assignments.stop(), dependencies.stop(), locks.stop(), watchdogs.stop()]); } }; + const workers = [assignments, orphanBlockers, recoveryActions, dependencies, locks, watchdogs]; + return { ready: Promise.all(workers.map(worker => worker.ready)), + wake() { for (const worker of workers) worker.invalidate(); }, + async stop() { await Promise.all(workers.map(worker => worker.stop())); } }; } diff --git a/server/src/services/scoped-deadline-work.ts b/server/src/services/scoped-deadline-work.ts index b46841e2a91..dbb9fe3df95 100644 --- a/server/src/services/scoped-deadline-work.ts +++ b/server/src/services/scoped-deadline-work.ts @@ -12,6 +12,8 @@ export function registerScopedDeadlineWork( queue: DeliveryQueue; batchSize?: number; minimumPassIntervalMs?: number; + reconstructionIntervalMs?: number; + beforePass?(): Promise; decodeScope(hint: string): Scope | undefined; matches(entry: Entry, scope?: Scope): boolean; load(scope?: Scope, signal?: AbortSignal): Promise; @@ -25,6 +27,7 @@ export function registerScopedDeadlineWork( const entries = new Map(); const failures = new Map(); const scopeFailures = new Map(); + const reconstructAfter = new Map(); let earliest = 0; let lastCompany: string | null = null; function failed(map: typeof failures, id: string) { @@ -34,13 +37,14 @@ export function registerScopedDeadlineWork( function dirtyScopes() { return [...dirty].filter(hint => hint !== "*" || !scopeFailures.has(hint)) .filter(hint => !scans.has(hint) || scopeFailures.has(hint)) + .filter(hint => (reconstructAfter.get(hint) ?? 0) <= Date.now() || scopeFailures.has(hint)) .map(hint => { try { return options.decodeScope(hint); } catch { return undefined; } }); } function next() { if (!canRun()) return dirty.size > 0 || entries.size > 0 ? Date.now() + 30_000 : null; let nextAt: number | null = null; const take = (at: number) => { nextAt = nextAt === null ? at : Math.min(nextAt, at); }; - for (const hint of dirty) take(Math.max(earliest, scopeFailures.get(hint)?.retryAt ?? 0)); + for (const hint of dirty) take(Math.max(earliest, reconstructAfter.get(hint) ?? 0, scopeFailures.get(hint)?.retryAt ?? 0)); const unavailable = dirtyScopes(); for (const entry of entries.values()) if (!unavailable.some(scope => options.matches(entry, scope))) take(Math.max(earliest, entry.dueAt, failures.get(entry.id)?.retryAt ?? 0)); @@ -54,9 +58,18 @@ export function registerScopedDeadlineWork( } }, run: async signal => { if (!canRun() || Date.now() < earliest) return; + if ((options.reconstructionIntervalMs ?? 0) > 0) for (const [hint, until] of reconstructAfter) { + if (until <= Date.now()) reconstructAfter.delete(hint); + } + const ready = [...dirty].filter(hint => Math.max(reconstructAfter.get(hint) ?? 0, scopeFailures.get(hint)?.retryAt ?? 0) <= Date.now()); + if (ready.length === 0 && ![...entries.values()].some(entry => Math.max(entry.dueAt, failures.get(entry.id)?.retryAt ?? 0) <= Date.now())) return; earliest = Date.now() + (options.minimumPassIntervalMs ?? 1000); + try { await options.beforePass?.(); } catch (error) { + for (const hint of ready) failed(scopeFailures, hint); + for (const entry of entries.values()) if (entry.dueAt <= Date.now()) failed(failures, entry.id); + throw error; + } const errors: unknown[] = []; - const ready = [...dirty].filter(hint => (scopeFailures.get(hint)?.retryAt ?? 0) <= Date.now()); const requested = ready.includes("*") ? ["*", ...ready.filter(hint => hint !== "*")].slice(0, 25) : ready.slice(0, 25); @@ -81,6 +94,9 @@ export function registerScopedDeadlineWork( entries.delete(entry.id); failures.delete(entry.id); } scans.delete(hint); + if ((options.reconstructionIntervalMs ?? 0) > 0) { + reconstructAfter.set(hint, Date.now() + options.reconstructionIntervalMs!); + } if ((versions.get(hint) ?? 0) !== scan.version) dirty.add(hint); else versions.delete(hint); if (!scope) scopeFailures.clear(); else scopeFailures.delete(hint); } catch (error) { diff --git a/server/src/services/task-watchdog-work.ts b/server/src/services/task-watchdog-work.ts index cde0a7eca43..98794472da2 100644 --- a/server/src/services/task-watchdog-work.ts +++ b/server/src/services/task-watchdog-work.ts @@ -1,4 +1,4 @@ -import { and, eq } from "drizzle-orm"; +import { and, asc, eq, gt } from "drizzle-orm"; import { issueWatchdogs, type Db } from "@paperclipai/db"; import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; @@ -16,16 +16,29 @@ export function registerTaskWatchdogWork( }, canRun: () => boolean, ) { + const cursors = new Map(); + let enabled = false; return registerScopedDeadlineWork(coordinator, { queue: DELIVERY_QUEUES.taskWatchdogs, + reconstructionIntervalMs: 30_000, + beforePass: async () => { enabled = await owner.enabled(); }, decodeScope: hint => hint === "*" ? undefined : JSON.parse(hint) as IssueRecoveryScope, matches: (entry, scope) => !scope || entry.companyId === scope.companyId, - load: async scope => (await db.select({ id: issueWatchdogs.id, companyId: issueWatchdogs.companyId }) - .from(issueWatchdogs).where(and(eq(issueWatchdogs.status, "active"), - scope ? eq(issueWatchdogs.companyId, scope.companyId) : undefined))) - .map(row => ({ ...row, dueAt: Date.now() })), + load: async scope => { + const key = scope?.companyId ?? "*"; + if (!enabled) { cursors.delete(key); return []; } + const cursor = cursors.get(key); + const rows = await db.select({ id: issueWatchdogs.id, companyId: issueWatchdogs.companyId }) + .from(issueWatchdogs).where(and(eq(issueWatchdogs.status, "active"), + scope ? eq(issueWatchdogs.companyId, scope.companyId) : undefined, + cursor ? gt(issueWatchdogs.id, cursor) : undefined)) + .orderBy(asc(issueWatchdogs.id)).limit(100); + const complete = rows.length < 100; + if (complete) cursors.delete(key); else cursors.set(key, rows.at(-1)!.id); + return { rows: rows.map(row => ({ ...row, dueAt: Date.now() })), complete }; + }, check: async entry => { - if (!(await owner.enabled())) return null; + if (!enabled) return null; let next: number | null = null; const result = await owner.reconcile({ companyId: entry.companyId, watchdogId: entry.id, onDeferredDeadline: at => { next = next === null ? at : Math.min(next, at); } });