From c2ff3eb92113fb16713b270e34932636956ff16b Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 21:49:38 -0500 Subject: [PATCH] fix: wake retry scheduling only when retry state changes Separate retry-schedule notifications from ordinary queued-run eligibility. Preserve exact deadlines and prove native producer commit/rollback behavior without global retry scans on ordinary agent or issue edits. Co-Authored-By: Paperclip --- doc/plans/2026-10-30-run-dispatch-sleep.md | 9 ++++ .../heartbeat-run-dispatch-lifecycle.test.ts | 19 ++++---- .../heartbeat-run-dispatch-work.test.ts | 48 ++++++++++++++++++- server/src/services/heartbeat/retries.ts | 6 +-- .../native-provider-capacity-retry.ts | 4 +- .../native-safe-replacement.test.ts | 10 ++++ .../native-runtime/native-safe-replacement.ts | 2 + server/src/services/recovery/service.ts | 6 +-- .../services/run-dispatch-notifications.ts | 7 ++- ...eduled-retry-cancellation-notifications.ts | 4 +- 10 files changed, 94 insertions(+), 21 deletions(-) diff --git a/doc/plans/2026-10-30-run-dispatch-sleep.md b/doc/plans/2026-10-30-run-dispatch-sleep.md index e5cd0a9e014..8ad35c128fc 100644 --- a/doc/plans/2026-10-30-run-dispatch-sleep.md +++ b/doc/plans/2026-10-30-run-dispatch-sleep.md @@ -12,6 +12,15 @@ invalidate retries notify the retry worker in their transaction. Promotion uses the existing gate and compare-and-set, and its committed write wakes ordinary queued dispatch. A queued run behind occupied capacity waits for the owning run's lifecycle transition. Configuration and authority changes also wake it. +Retry-schedule notifications are separate from ordinary queue eligibility. +`notifyRunDispatch` wakes only queued dispatch; ordinary issue, agent, budget +and company edits cannot repeatedly scan the global retry schedule. +`notifyScheduledRetryDispatch` wakes retry reconciliation as well as dispatch +for actual retry creation, rescheduling and cancellation. Its producers include +bounded and operator-requested retries, recovery retries, native capacity and +safe-replacement successors, and the existing retry-cancellation transition +boundary. Shared runtime changes do not impose a spacing floor on a saved deadline. + An idle queue has no recurring read or timer-driven dispatch attempt. Notifications batch for one second and carry the changed company to queued dispatch. Known retry deadlines still preempt batching. Cancellation-owned retries stay with their cancellation worker and do not retain a promotion deadline. The supported writer contract is service writes through the shared database diff --git a/server/src/__tests__/heartbeat-run-dispatch-lifecycle.test.ts b/server/src/__tests__/heartbeat-run-dispatch-lifecycle.test.ts index f3398d828f8..a9e120387c8 100644 --- a/server/src/__tests__/heartbeat-run-dispatch-lifecycle.test.ts +++ b/server/src/__tests__/heartbeat-run-dispatch-lifecycle.test.ts @@ -3,7 +3,7 @@ import { PgDialect } from "drizzle-orm/pg-core"; import { afterEach, describe, expect, it, vi } from "vitest"; import { installDatabaseWorkSignals } from "../../../packages/db/src/work-signals.js"; import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; -import { notifyRunDispatch } from "../services/run-dispatch-notifications.js"; +import { notifyRunDispatch, notifyScheduledRetryDispatch } from "../services/run-dispatch-notifications.js"; import { registerRunDispatchWork } from "../services/heartbeat/dispatch-work.js"; const cleanup: Array<() => Promise> = []; @@ -38,6 +38,7 @@ function fixture(input: { due?: number; suppressed?: boolean; draining?: boolean }, () => !state.draining && !state.standby); cleanup.push(() => coordinator.stop()); return { db, state, promote, next, dispatch, coordinator, worker, + notifyRetry: () => db.transaction(tx => notifyScheduledRetryDispatch(tx)), notify: (scope?: { companyId: string }) => db.transaction(tx => notifyRunDispatch(tx, scope)), settle: () => vi.advanceTimersByTimeAsync(1_000) }; } @@ -61,9 +62,9 @@ describe("run dispatch obligations", () => { }); it("rearms earlier work and cancels a future deadline on a committed producer change", async () => { const f = fixture({ due: 1_600_000 }); await f.worker.ready; - f.state.due = 1_005_000; await f.notify(); await f.settle(); + f.state.due = 1_005_000; await f.notifyRetry(); await f.settle(); expect(f.coordinator.nextWakeAt()).toBe(1_005_000); - f.state.due = null; await f.notify(); await f.settle(); + f.state.due = null; await f.notifyRetry(); await f.settle(); expect(f.coordinator.nextWakeAt()).toBeNull(); const before = f.next.mock.calls.length; await vi.advanceTimersByTimeAsync(20 * 60_000); expect(f.next).toHaveBeenCalledTimes(before); @@ -76,7 +77,7 @@ describe("run dispatch obligations", () => { await f.notify(source); await vi.advanceTimersByTimeAsync(10); } - expect(f.promote.mock.calls.length).toBeLessThanOrEqual(2); + expect(f.promote).toHaveBeenCalledOnce(); expect(f.dispatch.mock.calls.length).toBeLessThanOrEqual(2); await f.settle(); expect(f.dispatch).toHaveBeenLastCalledWith({ companyId: "company-a" }); @@ -87,7 +88,7 @@ describe("run dispatch obligations", () => { }); it("remembers the first producer arriving during drain without owner SQL", async () => { const f = fixture(); await f.worker.ready; f.state.draining = true; - await f.notify(); await f.settle(); await vi.advanceTimersByTimeAsync(90_000); + await f.notifyRetry(); await f.settle(); await vi.advanceTimersByTimeAsync(90_000); expect(f.promote).toHaveBeenCalledOnce(); expect(f.next).toHaveBeenCalledOnce(); expect(idleWorkSnapshot().active).toBe(0); f.state.draining = false; await vi.advanceTimersByTimeAsync(30_000); @@ -96,7 +97,7 @@ describe("run dispatch obligations", () => { }); it("releases a failed producer hold during drain and retains recovery for resume", async () => { const f = fixture(); await f.worker.ready; - f.state.fail = true; await f.notify(); await f.settle(); + f.state.fail = true; await f.notifyRetry(); await f.settle(); expect(f.promote).toHaveBeenCalledTimes(2); f.state.draining = true; await vi.advanceTimersByTimeAsync(90_000); expect(f.promote).toHaveBeenCalledTimes(2); expect(f.next).toHaveBeenCalledOnce(); @@ -109,8 +110,8 @@ describe("run dispatch obligations", () => { let release!: () => void; const held = new Promise(resolve => { release = resolve; }); f.promote.mockImplementationOnce(() => held); - await f.notify(); await f.settle(); - await f.notify(); await f.settle(); + await f.notifyRetry(); await f.settle(); + await f.notifyRetry(); await f.settle(); expect(f.promote).toHaveBeenCalledTimes(2); release(); await f.settle(); await vi.advanceTimersByTimeAsync(30_000); @@ -121,7 +122,7 @@ describe("run dispatch obligations", () => { await vi.advanceTimersByTimeAsync(90_000); expect(f.promote).not.toHaveBeenCalled(); f.state.standby = false; await vi.advanceTimersByTimeAsync(30_000); expect(f.promote).toHaveBeenCalledOnce(); expect(f.state.due).toBeNull(); - f.state.due = Date.now() + 60_000; await f.notify(); await f.settle(); await f.worker.stop(); + f.state.due = Date.now() + 60_000; await f.notifyRetry(); await f.settle(); await f.worker.stop(); await vi.advanceTimersByTimeAsync(90_000); expect(f.promote).toHaveBeenCalledTimes(2); }); }); diff --git a/server/src/__tests__/heartbeat-run-dispatch-work.test.ts b/server/src/__tests__/heartbeat-run-dispatch-work.test.ts index bcc8ed23972..1c0e5005a04 100644 --- a/server/src/__tests__/heartbeat-run-dispatch-work.test.ts +++ b/server/src/__tests__/heartbeat-run-dispatch-work.test.ts @@ -2,9 +2,11 @@ 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, companies, costEvents, createDb, databaseWorkPending, heartbeatRuns } from "@paperclipai/db"; +import { agents, companies, costEvents, createDb, databaseWorkPending, heartbeatRuns, issues } from "@paperclipai/db"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { issueService } from "../services/issues.js"; +import { materializeNativeProviderCapacityRetry } from "../services/native-runtime/native-provider-capacity-retry.js"; import { agentService } from "../services/agents.js"; import { budgetService } from "../services/budgets.js"; import { heartbeatService } from "../services/heartbeat.js"; @@ -24,9 +26,14 @@ describe.skipIf(!support.supported)("run dispatch production wiring (controlled let reads: ReturnType; let draining = false; const errors: unknown[] = []; + const retryQueries: string[] = []; beforeAll(async () => { if (!external) database = await startEmbeddedPostgresTestDatabase("run-dispatch-work-"); db = createDb(external ?? database!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); + db.$client.options.debug = (_connection, query) => { + if (query.startsWith("select") && query.includes('"heartbeat_runs"') && + query.includes("startupCancellation") && query.includes('"scheduled_retry_at"')) retryQueries.push(query); + }; await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select pg_sleep(0.01)`))); }, 90_000); beforeEach(async () => { @@ -88,6 +95,45 @@ describe.skipIf(!support.supported)("run dispatch production wiring (controlled await fixture(); await worker!.stop(); await coordinator!.stop(); await start(); await quiet(); }); + it.each(["agent", "issue"] as const)("does not scan global scheduled retries for repeated ordinary %s edits", async kind => { + const f = await fixture(); + const [issue] = await db.insert(issues).values({ companyId: f.companyId, title: "Edit source", status: "backlog" }).returning(); + await start(); await quiet(); + const before = retryQueries.length; + expect(before).toBeGreaterThan(0); // The passive observer saw real startup retry SQL. + for (let index = 0; index < 12; index++) { + if (kind === "agent") await agentService(db).update(f.agentId, { name: `Edited agent ${index}` }); + else await issueService(db).update(issue.id, { title: `Edited issue ${index}` }); + await new Promise(resolve => setTimeout(resolve, 100)); + } + await quiet(); + expect(retryQueries).toHaveLength(before); + }); + it.each([false, true])("observes a native capacity retry only after its transaction settles (rollback=%s)", async rollback => { + const f = await fixture(); + const [issue] = await db.insert(issues).values({ companyId: f.companyId, title: "Capacity retry", status: "in_progress", + assigneeAgentId: f.agentId }).returning(); + await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: issue.id }).where(eq(heartbeatRuns.id, f.runId)); + await start(); await quiet(); + let retryId = ""; + const writing = db.transaction(async tx => { + retryId = await materializeNativeProviderCapacityRetry({ tx: tx as unknown as typeof db, ...f, issueId: issue.id }); + expect(databaseWorkPending(db, DELIVERY_QUEUES.scheduledRetryDispatch)).toBe(true); + expect(await status(retryId)).toBeUndefined(); + if (rollback) throw new Error("native capacity rollback"); + }); + if (rollback) { + await expect(writing).rejects.toThrow("native capacity rollback"); + await quiet(); expect(await status(retryId)).toBeUndefined(); + } else { + await writing; + const retry = await status(retryId); + expect(retry.status).toBe("scheduled_retry"); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(retry.scheduledRetryAt!.getTime()), { timeout: 5_000 }); + await heartbeat.cancelRun(retryId, "fixture teardown", { suppressImmediateRecovery: true }); + await quiet(); + } + }); it("ordinary retry creation wakes the worker, respects the exact deadline, dispatches and settles", async () => { const f = await fixture(); await start(); await quiet(); const result = await heartbeat.scheduleBoundedRetry(f.runId, { delayMs: 4_000, random: () => 0 }); diff --git a/server/src/services/heartbeat/retries.ts b/server/src/services/heartbeat/retries.ts index 32333c0ebec..9eda1347ffa 100644 --- a/server/src/services/heartbeat/retries.ts +++ b/server/src/services/heartbeat/retries.ts @@ -1,4 +1,4 @@ -import { notifyRunDispatch } from "../run-dispatch-notifications.js"; +import { notifyScheduledRetryDispatch } from "../run-dispatch-notifications.js"; import { deriveTaskKeyWithHeartbeatFallback } from "./run-state.js"; import { infrastructureRetryDelayMs } from "../system-recovery-state.js"; import { @@ -998,7 +998,7 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen retryContextSnapshot.previousRunId = continuation.previousRunId; } - await notifyRunDispatch(tx, { companyId: run.companyId }); + await notifyScheduledRetryDispatch(tx, { companyId: run.companyId }); const wakeupRequest = await tx .insert(agentWakeupRequests) .values({ @@ -1756,7 +1756,7 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen }; const updated = await db.transaction(async (tx) => { - await notifyRunDispatch(tx, { companyId: scheduled.run.companyId }); + await notifyScheduledRetryDispatch(tx, { companyId: scheduled.run.companyId }); const row = await tx .update(heartbeatRuns) .set({ diff --git a/server/src/services/native-runtime/native-provider-capacity-retry.ts b/server/src/services/native-runtime/native-provider-capacity-retry.ts index 83314ff9c51..99575395f52 100644 --- a/server/src/services/native-runtime/native-provider-capacity-retry.ts +++ b/server/src/services/native-runtime/native-provider-capacity-retry.ts @@ -1,4 +1,4 @@ -import { notifyRunDispatch } from "../run-dispatch-notifications.js"; +import { notifyScheduledRetryDispatch } from "../run-dispatch-notifications.js"; import { randomUUID } from "node:crypto"; import { and, eq } from "drizzle-orm"; import { agentWakeupRequests, heartbeatRuns, type Db } from "@paperclipai/db"; @@ -50,7 +50,7 @@ export async function materializeNativeProviderCapacityRetry(input: { scheduledRetryAt: dueAt.toISOString(), executionRetryAccounting: accountingForScheduledRetry(run, NATIVE_PROVIDER_OVERLOADED_CODE, attempt), }; - await notifyRunDispatch(input.tx, { companyId: run.companyId }); + await notifyScheduledRetryDispatch(input.tx, { companyId: run.companyId }); const [wake] = await input.tx.insert(agentWakeupRequests).values({ companyId: run.companyId, agentId: run.agentId, source: "automation", triggerDetail: "system", reason: NATIVE_PROVIDER_OVERLOADED_CODE, status: "queued", runId: retryId, diff --git a/server/src/services/native-runtime/native-safe-replacement.test.ts b/server/src/services/native-runtime/native-safe-replacement.test.ts index 422b23dba6b..491bae17b2b 100644 --- a/server/src/services/native-runtime/native-safe-replacement.test.ts +++ b/server/src/services/native-runtime/native-safe-replacement.test.ts @@ -1,3 +1,4 @@ +import { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; import { getExecutionBlocker } from "../execution-blocker.js"; import { createRunDispatch, deriveCommentId } from "../../modules/run-dispatch/index.js"; import { buildExecutionContinuation } from "../execution-continuation.js"; @@ -577,7 +578,12 @@ const support = externalDatabaseUrl }); it("persists exactly one linked successor under competing sweepers and restarts", async () => { const source = await seed(2); + const owner = heartbeatService(db); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } }); + const worker = owner.registerRunDispatchWork(coordinator, () => true); + await worker.ready; const now = new Date(); + try { await Promise.all([ reconcileSafeNativeReplacements(db, now), reconcileSafeNativeReplacements(db, now), @@ -608,6 +614,10 @@ const support = externalDatabaseUrl .where(eq(nativeRunFinalizations.runId, source.runId)) )[0]?.failureDetail?.successorRunId, ).toBe(children[0]!.id); + await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBe(children[0]!.scheduledRetryAt!.getTime()), { timeout: 5_000 }); + } finally { + await worker.stop(); await coordinator.stop(); await owner.drainActiveRunExecutions(); + } }); it("retries status publication after a crash without dispatching provider work", async () => { await deliverExecutionStatuses(db); diff --git a/server/src/services/native-runtime/native-safe-replacement.ts b/server/src/services/native-runtime/native-safe-replacement.ts index da485883cb0..12b556fa779 100644 --- a/server/src/services/native-runtime/native-safe-replacement.ts +++ b/server/src/services/native-runtime/native-safe-replacement.ts @@ -1,3 +1,4 @@ +import { notifyScheduledRetryDispatch } from "../run-dispatch-notifications.js"; import { notifyNativeStatusReassessment } from "../native-status-reassessment-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-notifications.js"; import { notifyExecutionRunWork } from "../execution-control-notifications.js"; @@ -410,6 +411,7 @@ export async function reconcileSafeNativeReplacements( }) .returning(); await notifyExecutionRunWork(tx, { companyId: run.companyId }); + await notifyScheduledRetryDispatch(tx, { companyId: run.companyId }); await tx.insert(heartbeatRuns).values({ id: successorRunId, companyId: run.companyId, diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index fffc2699dbf..e3596791942 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -1,4 +1,4 @@ -import { notifyRunDispatch } from "../run-dispatch-notifications.js"; +import { notifyScheduledRetryDispatch } from "../run-dispatch-notifications.js"; import { notifyNativeEnvironmentSettlement } from "../native-environment-settlement-notifications.js"; import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js"; import { notifyNativeStatusReassessment } from "../native-status-reassessment-notifications.js"; @@ -2717,7 +2717,7 @@ export function recoveryService( const now = new Date(); const retryAt = readProviderQuotaRetryAt(input.latestRun, now); return database.transaction(async (tx) => { - await notifyRunDispatch(tx, { companyId: input.issue.companyId }); + await notifyScheduledRetryDispatch(tx, { companyId: input.issue.companyId }); const wakeup = await tx .insert(agentWakeupRequests) .values({ @@ -3318,7 +3318,7 @@ export function recoveryService( created = Boolean(enqueuedRun); } else { scheduledRun = await db.transaction(async (tx) => { - await notifyRunDispatch(tx, { companyId: input.issue.companyId }); + await notifyScheduledRetryDispatch(tx, { companyId: input.issue.companyId }); const wakeup = await tx .insert(agentWakeupRequests) .values({ diff --git a/server/src/services/run-dispatch-notifications.ts b/server/src/services/run-dispatch-notifications.ts index e62b8c62105..276edacab23 100644 --- a/server/src/services/run-dispatch-notifications.ts +++ b/server/src/services/run-dispatch-notifications.ts @@ -4,6 +4,11 @@ export type RunDispatchScope = { companyId: string }; export async function notifyRunDispatch(transaction: object, scope?: RunDispatchScope) { const hint = scope ? JSON.stringify({ companyId: scope.companyId }) : "*"; - await notifyDeliveryWork(transaction, DELIVERY_QUEUES.scheduledRetryDispatch, hint); await notifyDeliveryWork(transaction, DELIVERY_QUEUES.queuedRunDispatch, hint); } + +/** Only a retry row or deadline change needs to re-read the saved retry schedule. */ +export async function notifyScheduledRetryDispatch(transaction: object, scope?: RunDispatchScope) { + await notifyRunDispatch(transaction, scope); + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.scheduledRetryDispatch); +} diff --git a/server/src/services/scheduled-retry-cancellation-notifications.ts b/server/src/services/scheduled-retry-cancellation-notifications.ts index 24e57de66cb..945bb40e14b 100644 --- a/server/src/services/scheduled-retry-cancellation-notifications.ts +++ b/server/src/services/scheduled-retry-cancellation-notifications.ts @@ -1,9 +1,9 @@ -import { notifyRunDispatch } from "./run-dispatch-notifications.js"; +import { notifyScheduledRetryDispatch } from "./run-dispatch-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; export type ScheduledRetryCancellationScope = { companyId: string; runId?: string }; export async function notifyScheduledRetryCancellation(transaction: object, scope?: ScheduledRetryCancellationScope) { - await notifyRunDispatch(transaction, scope); + await notifyScheduledRetryDispatch(transaction, scope); await notifyDeliveryWork(transaction, DELIVERY_QUEUES.scheduledRetryCancellation, scope ? JSON.stringify(scope) : "*"); }