From e9343d8501e67474814ba11b8ad215a579bac28c Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Tue, 10 Nov 2026 15:00:50 -0800 Subject: [PATCH] fix(heartbeat): release task Retries held by a drain as soon as it ends (#20640) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - When a person clicks Retry on a stopped task, Paperclip saves the request as a held wake and dispatches it right away, unless admission is held. Admission is held during a task drain, for example while a deployment restarts the server > - A Retry held by a drain is released later by the deferred-receipt worker. That worker shares one queue with every other held wake: comment waits, execution waits, and interrupted queues > - On a busy instance that queue holds hundreds of passive waits that come due every 30 seconds. A Retry is checked in due-time order behind all of them. Ending a drain does not wake the worker either > - In a real case, a Retry clicked during a deployment drain was still held 40 minutes later, after the drain expired and the server restarted. It ran only when the person clicked Retry again > - This pull request checks operator Retries ahead of passive waits, and wakes the worker as soon as a drain stops or expires > - The benefit is that a Retry held by a drain starts within one worker pass after the drain ends ## Linked Issues or Issue Description **What happened?** A person clicked Retry on a blocked task 14 seconds after a deployment's task drain started. The Retry was saved as `deferred_issue_execution`, and the immediate dispatch returned early because of the drain. The drain had a 20-minute TTL, and the deployment restarted the server about 6 minutes in. The held Retry still did not start, and its `updated_at` never changed. Forty minutes later the person clicked Retry again, and the new request started a run and merged with the held one. That instance had 390 held wakes: 271 execution waits and 116 others, against 6 task Retries. The execution waits come due every 30 seconds. Measured on the instance, the deferred-receipt worker took about 7 minutes to get through that backlog once. **Expected behavior** A Retry that a drain held starts soon after the drain stops or expires, or after the server restarts, whatever the size of the passive backlog. **Steps to reproduce** 1. Start a non-idle task drain (`POST /instance/task-drain`). 2. Click Retry on a stopped task. The receipt is `deferred_issue_execution`. 3. Stop the drain. On `master`, the deferred-receipt worker only notices on its next 30-second re-check. It then checks the Retry in due-time order behind every passive wait that came due before it. **Paperclip version or commit** `master` at 8148fc0b1ec. ## What Changed - `server/src/services/scoped-deadline-work.ts`: - New optional `priority(entry)`. Due entries are sorted by priority first, then by due time and id as before. - New `refresh(hint)`. It clears a scope's reconstruction delay and invalidates it. Failure backoff is kept. - New optional `onStop` hook, called when the coordinator stops the worker. - `server/src/services/heartbeat/deferred-receipt-work.ts`: - A `task_retry` receipt that a person requested is an operator Retry, with priority 1. Passive receipts keep priority 0, so their order does not change. - Each deferred scan loads operator Retries first, with their own query, then pages by id as before. A Retry is therefore in memory in the first pass even beyond the first 100-row page. A receipt loaded twice merges by id. - New optional `owner.subscribeRefresh`. The work calls `refresh()` on all three of its queues when it fires, and releases the subscription from `onStop`, so the coordinator's own stop removes it. - `server/src/services/heartbeat.ts`: passes `subscribeRefresh` that listens to task-drain changes and expiry. Before this change, a deploy drain left the worker waiting for its 30-second re-check, behind the scan's reconstruction delay. - `server/src/__tests__/deferred-execution-receipt-work.test.ts`, four new tests: - A real Retry, requested during a deploy drain, starts a run (to `succeeded`) within 5 seconds after the drain stops. - An operator Retry whose id sorts after 120 passive execution waits (more than one load page) is resumed first, in the first worker pass. - A held Retry inserted with no notification right after a full scan is resumed as soon as the drain refresh fires. - Stopping only the coordinator removes the drain listener. ## Verification - `npx vitest run src/__tests__/deferred-execution-receipt-work.test.ts -t "held by a deploy drain|ahead of a backlog|ordinary committed task Retry"`: 3 pass. The third is the existing ordinary Retry dispatch test. - Red checks, removing one piece at a time: - Without the drain subscription, the drain test fails: the Retry is still `deferred_issue_execution` with no run after 5 seconds. - Without the priority, or without the prioritized load, the backlog test fails: the first resumed request is the passive `00000000-…` wake. - With `invalidate()` instead of `refresh()`, the post-scan test fails after 5 seconds. - Without the `onStop` release, the listener test fails. - Full `deferred-execution-receipt-work.test.ts`: 50 pass. - `chat-channels.integration.test.ts -t "still batches several replies in one finding thread"`: it timed out once on CI and passes 2 of 2 times locally with this change. - `scoped-deadline-work`, `heartbeat-task-drain`, `heartbeat-task-drain-admission-release`, `cloud-control-task-drain`, `task-admission-deadline`, `onboarding-first-task-admission` and `delivery-work-coordinator` suites: 164 pass. - `pnpm run typecheck` in `server` (layered): clean. ## Risks - Operator Retries are a small set; there were 6 on the instance above. They now take the first slots of a worker pass, so passive waits can move by at most that many places. Passive waits keep their own order. - A Retry that keeps failing its check still gets the existing failure backoff, so it cannot take every pass. - A drain start also wakes the worker. That pass ends at once, because the worker's run gate still reads the drain. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`) via Claude Code, with tool use (shell, file editing, vitest runs against embedded PostgreSQL, read-only production log and database analysis). ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- .../deferred-execution-receipt-work.test.ts | 81 ++++++++++++++++++- server/src/services/heartbeat.ts | 7 +- .../heartbeat/deferred-receipt-work.ts | 34 ++++++-- server/src/services/scoped-deadline-work.ts | 20 +++-- 4 files changed, 127 insertions(+), 15 deletions(-) diff --git a/server/src/__tests__/deferred-execution-receipt-work.test.ts b/server/src/__tests__/deferred-execution-receipt-work.test.ts index 76fe58ed5dd..c74076e0568 100644 --- a/server/src/__tests__/deferred-execution-receipt-work.test.ts +++ b/server/src/__tests__/deferred-execution-receipt-work.test.ts @@ -21,7 +21,7 @@ import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.j import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; import { notifyDeferredExecutionEligibility, notifyDeferredExecutionWork, notifyNativeStatusWakeWork, notifyTaskRetrySettlement } from "../services/deferred-execution-notifications.js"; import { registerDeferredReceiptWork } from "../services/heartbeat/deferred-receipt-work.js"; -import { heartbeatService } from "../services/heartbeat.js"; +import { heartbeatService, startTaskDrain, stopTaskDrain } from "../services/heartbeat.js"; import { requestTaskRetry } from "../services/task-retry.js"; import { issueService } from "../services/issues.js"; import { environmentService } from "../services/environments.js"; @@ -376,6 +376,85 @@ describe("deferred receipt durable dispatch", () => { expect(await db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.contextSnapshot}->>'taskRetryRequestId' = ${receipt.requestId}`)).toHaveLength(1); await quiet(); }); + it("dispatches a task Retry held by a deploy drain as soon as the drain stops", async () => { + const f = await fixture(); + await heartbeat.registerDeferredReceiptWork(coordinator, () => !drain).ready; + await heartbeat.registerRunDispatchWork(coordinator, () => !drain).ready; + await heartbeat.registerScheduledRetryCancellationWork(coordinator, () => !drain).ready; + startTaskDrain({ ttlMs: 20 * 60_000 }); + let requestId: string; + try { + const receipt = await requestTaskRetry({ db, companyId: f.companyId, issueId: f.issue.id, actorId: f.actorId, + observedRunId: f.source.id, idempotencyKey: randomUUID() }); + expect(receipt.accepted).toBe(true); + requestId = receipt.requestId!; + await heartbeat.resumeTaskRetry(f.companyId, requestId); + await vi.advanceTimersByTimeAsync(1000); + const [held] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, requestId)); + expect(held).toMatchObject({ status: "deferred_issue_execution", runId: null }); + } finally { stopTaskDrain(); } + let runId: string | undefined; + await waitForReal(async () => { + const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, requestId)); + expect(wake.runId, JSON.stringify(wake)).toBeTruthy(); runId = wake.runId!; + }, { timeout: 5_000 }); + await waitForReal(async () => { + await heartbeat.drainActiveRunExecutions(); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId!)); + expect(run.status, JSON.stringify({ error: run.error, code: run.errorCode })).toBe("succeeded"); + }, { timeout: 15_000 }); + await finish(f.issue.id); + await quiet(); + }); + it("dispatches an operator Retry ahead of a backlog of passive waits", async () => { + const f = await fixture(); + await db.insert(agentWakeupRequests).values(Array.from({ length: 120 }, (_, index) => ({ + id: `00000000-0000-4000-8000-${String(index).padStart(12, "0")}`, companyId: f.companyId, agentId: f.agent.id, + source: "automation", status: "deferred_issue_execution", requestedByActorType: "system", + payload: { executionWait: { reason: "recovery" } }, updatedAt: new Date(Date.now() - 60_000), + }))); + const [retry] = await db.insert(agentWakeupRequests).values({ id: "ffffffff-ffff-4fff-bfff-ffffffffffff", + companyId: f.companyId, agentId: f.agent.id, source: "on_demand", status: "deferred_issue_execution", + reason: "task_retry", requestedByActorType: "user", requestedByActorId: f.actorId, + payload: { issueId: f.issue.id } }).returning(); + const order: string[] = []; + const resume = vi.fn(async (scope: { requestId?: string }) => { + order.push(scope.requestId!); + await db.update(agentWakeupRequests).set({ status: "skipped" }).where(eq(agentWakeupRequests.id, scope.requestId!)); + }); + await registerDeferredReceiptWork(db, coordinator, { resume, settleTaskChat: vi.fn(), createNativeBatch: () => vi.fn(), + schedulingSuppressed: async () => false }, () => !drain).ready; + expect(order[0]).toBe(retry.id); + await waitForReal(() => expect(order).toHaveLength(121), { timeout: 30_000 }); + await quiet(); + }); + it("reloads a held Retry without a pending notification as soon as a drain ends, right after a full scan", async () => { + const f = await fixture(); + let refresh: (() => void) | undefined; + const resume = vi.fn(async (scope: { requestId?: string }) => { + await db.update(agentWakeupRequests).set({ status: "skipped" }).where(eq(agentWakeupRequests.id, scope.requestId!)); + }); + await registerDeferredReceiptWork(db, coordinator, { resume, settleTaskChat: vi.fn(), createNativeBatch: () => vi.fn(), + schedulingSuppressed: async () => false, subscribeRefresh: listener => { refresh = listener; return () => {}; } }, + () => !drain).ready; + const [retry] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agent.id, + source: "on_demand", status: "deferred_issue_execution", reason: "task_retry", requestedByActorType: "user", + requestedByActorId: f.actorId, payload: { issueId: f.issue.id } }).returning(); + refresh!(); + await waitForReal(() => expect(resume).toHaveBeenCalledWith(expect.objectContaining({ requestId: retry.id })), + { timeout: 5_000 }); + await quiet(); + }); + it("removes its task-drain refresh listener when the coordinator stops", async () => { + let refresh: (() => void) | undefined; + const unsubscribe = vi.fn(); + await registerDeferredReceiptWork(db, coordinator, { resume: vi.fn(), settleTaskChat: vi.fn(), + createNativeBatch: () => vi.fn(), schedulingSuppressed: async () => false, + subscribeRefresh: listener => { refresh = listener; return unsubscribe; } }, () => !drain).ready; + expect(refresh).toBeTypeOf("function"); + await coordinator.stop(); + expect(unsubscribe).toHaveBeenCalledTimes(1); + }); async function blockedRetry(gate: "budget" | "project budget" | "paused agent", conversation = false) { const f = await fixture(), lifecycle = createAgentLifecycle(db), budgets = budgetService(db); if (conversation) await db.update(issues).set({ status: "in_review", conversationAgentId: f.agent.id, diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index f04965ec2cc..7724e412126 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -561,7 +561,8 @@ import { resolveAndRetainRunTrustPreset } from "./run-trust-preset.js"; import { effectiveProjectWorkspacePolicy, sourcelessProjectWorkspacePolicy } from "./project-workspace-provider.js"; import type { PluginWorkerManager } from "./plugin-worker-manager.js"; -import { computeTaskDrain, applyTaskDrain, startTaskDrain, stopTaskDrain, readTaskDrain } from "./task-admission.js"; +import { computeTaskDrain, applyTaskDrain, startTaskDrain, stopTaskDrain, readTaskDrain, subscribeTaskDrainChanges, + subscribeTaskDrainExpiry } from "./task-admission.js"; const execFile = promisify(execFileCallback); const UNSUCCESSFUL_HEARTBEAT_RUN_TERMINAL_STATUSES = [ @@ -4515,6 +4516,10 @@ export function heartbeatService( settleTaskChat: scope => reconcileTaskChatRetryReceipts(db, { ...scope, settleOnly: true }), createNativeBatch: heartbeatQueue.createNativeStatusWakeBatch, schedulingSuppressed: async () => (await getSchedulingSuppression(true)).suppressed, + subscribeRefresh: refresh => { + const unsubscribe = [subscribeTaskDrainChanges(refresh), subscribeTaskDrainExpiry(refresh)]; + return () => { for (const stop of unsubscribe) stop(); }; + }, }, () => canRun() && readTaskDrain(new Date()) === null), readIdleWatchdogSchedule: taskWatchdogs.readIdleSchedule, registerIssueStateRecoveryWork: (coordinator: Parameters[1], canRun: () => boolean) => diff --git a/server/src/services/heartbeat/deferred-receipt-work.ts b/server/src/services/heartbeat/deferred-receipt-work.ts index 5a3e3bddd64..bbf49a50739 100644 --- a/server/src/services/heartbeat/deferred-receipt-work.ts +++ b/server/src/services/heartbeat/deferred-receipt-work.ts @@ -7,7 +7,7 @@ import type { DeferredExecutionScope, DeferredReceiptRecoveryOptions } from "../ import { registerScopedDeadlineWork } from "../scoped-deadline-work.js"; import { dependencyResolvedWakePayloadSql } from "../issue-dependency-wakeups.js"; -type Entry = { id: string; companyId: string; dueAt: number; immediate: boolean }; +type Entry = { id: string; companyId: string; dueAt: number; immediate: boolean; operatorRetry?: boolean }; type NativeIntentOptions = DeferredExecutionScope & { onDeferredDeadline(at: number): void; onError(error: unknown): void; @@ -21,9 +21,15 @@ export function registerDeferredReceiptWork( settleTaskChat(scope: DeferredExecutionScope): Promise; createNativeBatch(): (scope: NativeIntentOptions) => Promise; schedulingSuppressed(): Promise; + subscribeRefresh?(refresh: () => void): () => void; }, canRun: () => boolean, ) { + let unsubscribeRefresh: (() => void) | undefined; + function releaseRefresh() { + unsubscribeRefresh?.(); + unsubscribeRefresh = undefined; + } function register(kind: "deferred" | "task_chat" | "native") { const cursors = new Map(); let enabled = false; @@ -45,23 +51,28 @@ export function registerDeferredReceiptWork( // receipts already checked; actual wait/claim deadlines still replace. mergeEntry: (pending, loaded) => pending.immediate && loaded.immediate ? { ...loaded, dueAt: Math.min(pending.dueAt, loaded.dueAt) } : loaded, + priority: entry => entry.operatorRetry ? 1 : 0, + onStop: releaseRefresh, load: async scope => { const key = scope ? JSON.stringify({ companyId: scope.companyId, requestId: scope.requestId }) : "*"; if (!enabled) { cursors.delete(key); return []; } const cursor = cursors.get(key); - const rows = await db.select({ id: agentWakeupRequests.id, companyId: agentWakeupRequests.companyId, + const operatorRetry = sql`coalesce(${agentWakeupRequests.reason}, '') = 'task_retry' + and coalesce(${agentWakeupRequests.requestedByActorType}, '') = 'user'`; + const columns = { id: agentWakeupRequests.id, companyId: agentWakeupRequests.companyId, status: agentWakeupRequests.status, updatedAt: agentWakeupRequests.updatedAt, claimedAt: agentWakeupRequests.claimedAt, error: agentWakeupRequests.error, + operatorRetry: sql`${operatorRetry}`, delayed: sql`${agentWakeupRequests.payload}->'queuedCommentInterrupt' is not null or ${agentWakeupRequests.payload}->'executionWait' is not null`, interruptedLegacy: sql`exists (select 1 from heartbeat_runs r where r.company_id = ${agentWakeupRequests.companyId} and r.agent_id = ${agentWakeupRequests.agentId} and r.runtime_mode = 'legacy' and r.status = 'cancelled' and r.result_json->>'queuedCommentInterruptQueueId' = ${agentWakeupRequests.id}::text)`, - }).from(agentWakeupRequests).where(and( + }; + const eligible = and( scope ? eq(agentWakeupRequests.companyId, scope.companyId) : undefined, scope?.requestId ? eq(agentWakeupRequests.id, scope.requestId) : undefined, - cursor ? gt(agentWakeupRequests.id, cursor) : undefined, kind === "native" ? and( eq(agentWakeupRequests.requestedByActorType, "system"), eq(agentWakeupRequests.requestedByActorId, "native-status-committer"), @@ -80,17 +91,25 @@ export function registerDeferredReceiptWork( and r.runtime_mode = 'legacy' and r.status = 'cancelled' and r.result_json->>'queuedCommentInterruptQueueId' = ${agentWakeupRequests.id}::text)`, )), - )).orderBy(asc(agentWakeupRequests.id)).limit(100); + ); + const prioritized = kind === "deferred" && !cursor + ? await db.select(columns).from(agentWakeupRequests).where(and(eligible, operatorRetry)) + .orderBy(asc(agentWakeupRequests.id)).limit(100) + : []; + const rows = await db.select(columns).from(agentWakeupRequests).where(and(eligible, + cursor ? gt(agentWakeupRequests.id, cursor) : undefined)) + .orderBy(asc(agentWakeupRequests.id)).limit(100); const complete = rows.length < 100; if (complete) cursors.delete(key); else cursors.set(key, rows.at(-1)!.id); const now = Date.now(); - return { rows: rows.map(row => { + return { rows: [...prioritized, ...rows].map(row => { const deadline = kind === "native" && row.status === "claimed" && row.claimedAt ? row.claimedAt.getTime() + 60_000 : kind === "native" && row.error ? row.updatedAt.getTime() + 30_000 : kind === "deferred" && row.delayed && !row.interruptedLegacy ? row.updatedAt.getTime() + 30_000 : null; const immediate = deadline === null || deadline <= now; - return { id: row.id, companyId: row.companyId, dueAt: immediate ? now : deadline, immediate }; + return { id: row.id, companyId: row.companyId, dueAt: immediate ? now : deadline, immediate, + operatorRetry: kind === "deferred" && row.operatorRetry }; }), complete }; }, check: async entry => { @@ -132,6 +151,7 @@ export function registerDeferredReceiptWork( }, canRun); } const workers = [register("deferred"), register("task_chat"), register("native")]; + unsubscribeRefresh = owner.subscribeRefresh?.(() => { for (const worker of workers) worker.refresh(); }); 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 b6613ac54c1..117632b52c4 100644 --- a/server/src/services/scoped-deadline-work.ts +++ b/server/src/services/scoped-deadline-work.ts @@ -32,7 +32,9 @@ export function registerScopedDeadlineWork( load(scope?: Scope, signal?: AbortSignal): Promise; // Owners may retain pending queue priority while refreshing eligibility. mergeEntry?(pending: Entry, loaded: Entry): Entry; + priority?(entry: Entry): number; check(entry: Entry, signal?: AbortSignal): Promise; + onStop?(): void; }, canRun: () => boolean, ) { @@ -73,7 +75,7 @@ export function registerScopedDeadlineWork( const worker = coordinator.register(options.queue, { retryMs: 30_000, pausePassiveMaintenance: true, - onStop: () => { stopped = true; revision++; }, + onStop: () => { stopped = true; revision++; options.onStop?.(); }, onSettled: hints => { revision++; for (const hint of hints ?? ["*"]) { dirty.add(hint); versions.set(hint, (versions.get(hint) ?? 0) + 1); } }, @@ -134,7 +136,8 @@ export function registerScopedDeadlineWork( const unavailable = dirtyScopes(); const due = [...entries.values()].filter(entry => !unavailable.some(scope => options.matches(entry, scope)) && Math.max(entry.dueAt, failures.get(entry.id)?.retryAt ?? 0) <= Date.now()) - .sort((a, b) => a.dueAt - b.dueAt || a.id.localeCompare(b.id)); + .sort((a, b) => (options.priority?.(b) ?? 0) - (options.priority?.(a) ?? 0) + || a.dueAt - b.dueAt || a.id.localeCompare(b.id)); const batch = takeCompanyWorkBatch(due, options.batchSize ?? 25, lastCompany); lastCompany = batch.lastCompany; // An owner may share more than one resource (for example a task and an // endpoint). Merge intersecting keys before dispatch so waiting siblings @@ -275,6 +278,11 @@ export function registerScopedDeadlineWork( nextRunAt: next, nextRunAtOnError: next, }); + function invalidate(hint = "*") { + revision++; + dirty.add(hint); versions.set(hint, (versions.get(hint) ?? 0) + 1); + worker.wake(); + } return { ...worker, reconciliationSnapshot: (): DeadlineReconciliationSnapshot => ({ revision: `${ownerRevision}:${revision}`, @@ -282,10 +290,10 @@ export function registerScopedDeadlineWork( && failures.size === 0 && scopeFailures.size === 0, }), wake() { revision++; worker.wake(); }, - invalidate(hint = "*") { - revision++; - dirty.add(hint); versions.set(hint, (versions.get(hint) ?? 0) + 1); - worker.wake(); + invalidate, + refresh(hint = "*") { + reconstructAfter.delete(hint); + invalidate(hint); }, }; }