From 17fa71867578e5716409dc1cf5c87e7c74a15425 Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Mon, 5 Oct 2026 10:58:56 -0700 Subject: [PATCH] Retry transient disconnects in scoped read queries (#15248) Apply existing bounded transient-disconnect retries to audited dashboard, heartbeat-list, and base issue reads. Rebuild each query per attempt and preserve authorization, company scope, enrichment, and final error propagation. Validation: exact-head Greptile 5/5, passing CI, no unresolved review threads, and a clean merge. Detailed verification and limitations are recorded in the pull request. Co-Authored-By: Paperclip --- doc/DATABASE.md | 14 +- .../dashboard-disconnect-retry.test.ts | 62 ++++-- .../service-read-disconnect-retry.test.ts | 183 ++++++++++++++++++ server/src/services/dashboard.ts | 8 +- server/src/services/heartbeat.ts | 57 +++--- server/src/services/issues.ts | 9 +- 6 files changed, 278 insertions(+), 55 deletions(-) create mode 100644 server/src/__tests__/service-read-disconnect-retry.test.ts diff --git a/doc/DATABASE.md b/doc/DATABASE.md index b6258c21acb..772e97668e5 100644 --- a/doc/DATABASE.md +++ b/doc/DATABASE.md @@ -177,12 +177,14 @@ idempotent actor synchronization operations, not arbitrary transactions. A persistent outage still fails the request after the bounded retries; each connection attempt remains subject to the configured database connect timeout. -The dashboard's company lookup, task counts, pending approval count, and -monthly spend each retry these connection errors at most twice. Each callback -is read-only and rebuilds its query for each attempt. A failed read -does not replay completed reads or the budget workflow. Missing companies, -authentication errors, and other database errors propagate without retry. -This does not enable general SQL replay. +The dashboard's company lookup, agent and task counts, pending approval count, +monthly spend, and run activity aggregate each retry these connection errors +at most twice. The heartbeat run list and base issue lookup by UUID or identifier +use the same bounded retries. Each callback is read-only and rebuilds its query +for each attempt. A failed read does not replay completed reads, the budget +workflow, or issue label and watchdog enrichment. Missing resources, +authentication errors, and other database errors retain their usual behavior. +This does not enable general SQL replay or retry a full request. ## Execution identity row locks diff --git a/server/src/__tests__/dashboard-disconnect-retry.test.ts b/server/src/__tests__/dashboard-disconnect-retry.test.ts index 16ad286261c..64a2f3c0b9f 100644 --- a/server/src/__tests__/dashboard-disconnect-retry.test.ts +++ b/server/src/__tests__/dashboard-disconnect-retry.test.ts @@ -18,9 +18,9 @@ function disconnected(code = "CONNECTION_CLOSED") { } function fixture(failures: Error[] = [], options: { - laterError?: Error; missingCompany?: boolean; readFailures?: Record; + activityFailures?: Error[]; } = {}) { const reads: string[] = []; const companyParameters: unknown[][] = []; @@ -47,7 +47,6 @@ function fixture(failures: Error[] = [], options: { if (failure) throw failure; return options.missingCompany ? [] : [{ id: companyId, budgetMonthlyCents: 10_000 }]; } - if (options.laterError) throw options.laterError; if (name === "agents") return [{ status: "idle", count: 2 }]; if (name === "issues") return [{ status: "in_progress", count: 1 }]; if (name === "approvals") return [{ count: 3 }]; @@ -59,9 +58,15 @@ function fixture(failures: Error[] = [], options: { return query; }, })); - const execute = vi.fn().mockResolvedValue([]); + const activityQueries: SQL[] = []; + const execute = vi.fn(async (query: SQL) => { + const failure = options.activityFailures?.[activityQueries.length]; + activityQueries.push(query); + if (failure) throw failure; + return [{ date: new Date().toISOString().slice(0, 10), status: "succeeded", count: 2 }]; + }); const db = { select, execute } as unknown as Db; - return { service: dashboardService(db), reads, companyParameters, parameters, select, execute }; + return { service: dashboardService(db), reads, companyParameters, parameters, select, execute, activityQueries }; } beforeEach(() => { @@ -121,14 +126,6 @@ describe("dashboard read connection recovery", () => { expect(overview).not.toHaveBeenCalled(); }); - it("does not replay later summary reads when they lose a connection", async () => { - const error = disconnected(); - const test = fixture([], { laterError: error }); - await expect(test.service.summary(companyId)).rejects.toBe(error); - expect(test.reads).toEqual(["companies", "agents"]); - expect(overview).not.toHaveBeenCalled(); - }); - it("does not replay the budget workflow when it loses a connection", async () => { const error = disconnected(); overview.mockRejectedValue(error); @@ -139,12 +136,13 @@ describe("dashboard read connection recovery", () => { }); const tables = ["companies", "agents", "issues", "approvals", "cost_events"]; - it.each(["issues", "approvals", "cost_events"])("retries only the failed %s read", async (table) => { + it.each(["agents", "issues", "approvals", "cost_events"])("retries only the failed %s read", async (table) => { const test = fixture([], { readFailures: { [table]: [disconnected()] } }); await expect(test.service.summary(companyId)).resolves.toMatchObject({ companyId }); const prefix = tables.slice(0, tables.indexOf(table) + 1); const suffix = tables.slice(tables.indexOf(table) + 1); expect(test.reads).toEqual([...prefix, table, ...suffix]); + expect(test.select).toHaveBeenCalledTimes(tables.length + 1); const attempts = test.parameters.get(table)!; expect(attempts).toHaveLength(2); expect(attempts[0]).toContain(companyId); @@ -153,7 +151,7 @@ describe("dashboard read connection recovery", () => { expect(overview).toHaveBeenCalledExactlyOnceWith(companyId); }); - it.each(["issues", "approvals", "cost_events"])("stops after the %s read exhausts its budget", async (table) => { + it.each(["agents", "issues", "approvals", "cost_events"])("stops after the %s read exhausts its budget", async (table) => { const failures = [disconnected(), disconnected(), disconnected()]; const test = fixture([], { readFailures: { [table]: failures } }); await expect(test.service.summary(companyId)).rejects.toBe(failures[2]); @@ -162,4 +160,40 @@ describe("dashboard read connection recovery", () => { expect(test.execute).not.toHaveBeenCalled(); expect(overview).not.toHaveBeenCalled(); }); + + it("rebuilds only the run activity query through two transient disconnects", async () => { + const test = fixture([], { activityFailures: [disconnected(), disconnected()] }); + const result = await test.service.summary(companyId); + expect(result.runActivity.at(-1)).toMatchObject({ succeeded: 2, total: 2 }); + expect(test.reads).toEqual(tables); + expect(test.select).toHaveBeenCalledTimes(tables.length); + expect(test.execute).toHaveBeenCalledTimes(3); + expect(new Set(test.activityQueries).size).toBe(3); + const queries = test.activityQueries.map((query) => dialect.sqlToQuery(query)); + expect(queries[0].params).toContain(companyId); + expect(queries[1]).toEqual(queries[0]); + expect(queries[2]).toEqual(queries[0]); + expect(overview).toHaveBeenCalledExactlyOnceWith(companyId); + }); + + it("propagates the last run activity failure without entering the budget workflow", async () => { + const failures = [disconnected(), disconnected(), disconnected()]; + const test = fixture([], { activityFailures: failures }); + await expect(test.service.summary(companyId)).rejects.toBe(failures[2]); + expect(test.reads).toEqual(tables); + expect(test.execute).toHaveBeenCalledTimes(3); + expect(overview).not.toHaveBeenCalled(); + }); + + it.each(["agents", "activity"])("does not retry a non-transient %s error", async (read) => { + const error = disconnected("42501"); + const test = fixture([], read === "agents" + ? { readFailures: { agents: [error] } } + : { activityFailures: [error] }); + await expect(test.service.summary(companyId)).rejects.toBe(error); + expect(test.reads.filter((table) => table === "agents")).toHaveLength(1); + expect(test.execute).toHaveBeenCalledTimes(read === "activity" ? 1 : 0); + expect(overview).not.toHaveBeenCalled(); + }); + }); diff --git a/server/src/__tests__/service-read-disconnect-retry.test.ts b/server/src/__tests__/service-read-disconnect-retry.test.ts new file mode 100644 index 00000000000..6ae2515e589 --- /dev/null +++ b/server/src/__tests__/service-read-disconnect-retry.test.ts @@ -0,0 +1,183 @@ +import type { Request, Response } from "express"; +import { getTableName, type SQL } from "drizzle-orm"; +import { PgDialect } from "drizzle-orm/pg-core"; +import type { Db } from "@paperclipai/db"; +import { describe, expect, it, vi } from "vitest"; +import { heartbeatService } from "../services/heartbeat.js"; +import { issueService } from "../services/issues.js"; +import { getAccessibleResource } from "../routes/authz.js"; + +const companyId = "00000000-0000-4000-8000-000000000001"; +const agentId = "00000000-0000-4000-8000-000000000002"; +const issueId = "00000000-0000-4000-8000-000000000003"; +const dialect = new PgDialect(); + +function disconnected(code = "CONNECTION_CLOSED") { + return new Error("Failed query", { + cause: Object.assign(new Error(`write ${code}`), { code }), + }); +} + +function fixture(options: { + failures?: Record; + missing?: boolean; + encoding?: string; +} = {}) { + const attempts: Array<{ + table: string; + params: unknown[]; + fields: Record; + order: string[]; + limit: number | undefined; + }> = []; + const row = { id: issueId, companyId, identifier: "TEST-1", externalConversationState: "waiting" }; + const select = vi.fn((fields: Record = {}) => ({ + from(table: Parameters[0]) { + const name = getTableName(table); + let condition: SQL; + let order: SQL[] = []; + let limit: number | undefined; + let result: Promise | undefined; + const query = { + where(value: SQL) { condition = value; return query; }, + innerJoin() { return query; }, + orderBy(...values: SQL[]) { order = values; return query; }, + limit(value: number) { limit = value; return query; }, + then(resolve: (value: unknown[]) => unknown, reject: (error: unknown) => unknown) { + // A failed query object stays failed. Recovery must build another one. + result ??= Promise.resolve().then(() => { + const attempt = attempts.filter((read) => read.table === name).length; + attempts.push({ table: name, fields, params: dialect.sqlToQuery(condition).params, + order: order.map((value) => dialect.sqlToQuery(value).sql), limit }); + const error = options.failures?.[name]?.[attempt]; + if (error) throw error; + if (options.missing) return []; + if (name === "issues") return [row]; + if (name === "issue_labels") return [{ issueId, label: { id: "label-1", name: "Ready" } }]; + if (name === "issue_watchdogs") return []; + if (name === "heartbeat_runs") return [{ + id: "run-1", companyId, agentId, status: "succeeded", + contextIssueId: issueId, contextWakeReason: "issue_assigned", resultSummary: "Finished", + }]; + throw new Error(`Unexpected read: ${name}`); + }); + return result.then(resolve, reject); + }, + }; + return query; + }, + })); + const execute = vi.fn().mockResolvedValue([{ server_encoding: options.encoding ?? "UTF8" }]); + const db = { select, execute } as unknown as Db; + return { db, row, attempts, select, execute }; +} + +describe("heartbeat list connection recovery", () => { + it.each([ + { summary: false, agent: agentId, limit: 5, encoding: "UTF8" }, + { summary: true, agent: undefined, limit: undefined, encoding: "UTF8" }, + { summary: false, agent: undefined, limit: undefined, encoding: "SQL_ASCII" }, + ])("rebuilds the scoped list with its projection and limit (%j)", async (options) => { + const test = fixture({ encoding: options.encoding, + failures: { heartbeat_runs: [disconnected(), disconnected("CONNECTION_ENDED")] } }); + await expect(heartbeatService(test.db).list(companyId, options.agent, options.limit, options)) + .resolves.toEqual([{ + id: "run-1", companyId, agentId, status: "succeeded", + contextSnapshot: { issueId, wakeReason: "issue_assigned" }, + resultJson: options.summary || options.encoding === "SQL_ASCII" ? null : { summary: "Finished" }, + }]); + expect(test.select).toHaveBeenCalledTimes(3); + expect(test.execute).toHaveBeenCalledTimes(1); + expect(test.attempts).toHaveLength(3); + for (const attempt of test.attempts) { + expect(attempt.params).toEqual(options.agent ? [companyId, agentId] : [companyId]); + expect(attempt.limit).toBe(options.limit); + expect(attempt.order).toEqual(['"heartbeat_runs"."created_at" desc']); + expect("resultSummary" in attempt.fields).toBe(!options.summary && options.encoding === "UTF8"); + expect(attempt.fields).toEqual(test.attempts[0].fields); + } + }); + + it("propagates the final list error after three attempts", async () => { + const failures = [disconnected(), disconnected(), disconnected()]; + const test = fixture({ failures: { heartbeat_runs: failures } }); + await expect(heartbeatService(test.db).list(companyId)).rejects.toBe(failures[2]); + expect(test.select).toHaveBeenCalledTimes(3); + expect(test.execute).toHaveBeenCalledTimes(1); + }); + + it.each([new Error("CONNECTION_CLOSED"), disconnected("28P01"), disconnected("42501")])( + "does not retry other list errors", async (error) => { + const test = fixture({ failures: { heartbeat_runs: [error] } }); + await expect(heartbeatService(test.db).list(companyId)).rejects.toBe(error); + expect(test.select).toHaveBeenCalledTimes(1); + }, + ); + + it("returns an empty list without retries", async () => { + const test = fixture({ missing: true }); + await expect(heartbeatService(test.db).list(companyId)).resolves.toEqual([]); + expect(test.select).toHaveBeenCalledTimes(1); + }); +}); + +describe.each([ + { name: "UUID", read: (db: Db) => issueService(db).getById(issueId), parameter: issueId }, + { name: "identifier", read: (db: Db) => issueService(db).getByIdentifier("test-1"), parameter: "TEST-1" }, + { name: "identifier through getById", read: (db: Db) => issueService(db).getById(" test-1 "), parameter: "TEST-1" }, +])("issue lookup connection recovery ($name)", ({ read, parameter }) => { + it("rebuilds the base lookup and enriches only the successful result", async () => { + const test = fixture({ failures: { issues: [disconnected(), disconnected("CONNECT_TIMEOUT")] } }); + await expect(read(test.db)).resolves.toEqual({ ...test.row, + labels: [{ id: "label-1", name: "Ready" }], labelIds: ["label-1"], watchdog: null }); + expect(test.select).toHaveBeenCalledTimes(5); + expect(test.attempts.map((attempt) => attempt.table)) + .toEqual(["issues", "issues", "issues", "issue_labels", "issue_watchdogs"]); + for (const attempt of test.attempts.slice(0, 3)) { + expect(attempt.params).toEqual([parameter]); + expect(attempt.fields).toHaveProperty("externalConversationState"); + } + expect(test.attempts[3].params).toEqual([issueId]); + expect(test.attempts[4].params).toEqual([companyId, issueId, "active"]); + }); + + it("propagates the final lookup error without enrichment", async () => { + const failures = [disconnected(), disconnected(), disconnected()]; + const test = fixture({ failures: { issues: failures } }); + await expect(read(test.db)).rejects.toBe(failures[2]); + expect(test.attempts.map((attempt) => attempt.table)).toEqual(["issues", "issues", "issues"]); + }); + + it("returns null for a missing issue without retries or enrichment", async () => { + const test = fixture({ missing: true }); + await expect(read(test.db)).resolves.toBeNull(); + expect(test.select).toHaveBeenCalledTimes(1); + }); + + it("preserves non-transient database errors", async () => { + const error = disconnected("42501"); + const test = fixture({ failures: { issues: [error] } }); + await expect(read(test.db)).rejects.toBe(error); + expect(test.select).toHaveBeenCalledTimes(1); + }); + + it.each(["issue_labels", "issue_watchdogs"])("does not replay the lookup when %s enrichment fails", async (table) => { + const error = disconnected(); + const test = fixture({ failures: { [table]: [error] } }); + await expect(read(test.db)).rejects.toBe(error); + expect(test.attempts.map((attempt) => attempt.table)).toEqual(["issues", "issue_labels", "issue_watchdogs"]); + }); +}); + +describe("issue access after lookup recovery", () => { + it.each([true, false])("retains the same not-found response for missing and inaccessible issues (missing=%s)", async (missing) => { + const test = fixture({ missing, failures: { issues: [disconnected()] } }); + const req = { method: "GET", actor: { type: "agent", companyId: "other-company" } } as Request; + const res = { status: vi.fn().mockReturnThis(), json: vi.fn() } as unknown as Response; + await expect(getAccessibleResource(req, res, issueService(test.db).getById(issueId), "Issue not found")) + .resolves.toBeNull(); + expect(res.status).toHaveBeenCalledExactlyOnceWith(404); + expect(res.json).toHaveBeenCalledExactlyOnceWith({ error: "Issue not found" }); + expect(test.attempts.filter((attempt) => attempt.table === "issues")).toHaveLength(2); + }); +}); diff --git a/server/src/services/dashboard.ts b/server/src/services/dashboard.ts index f4d7f7e4ed9..57ae04dac7e 100644 --- a/server/src/services/dashboard.ts +++ b/server/src/services/dashboard.ts @@ -38,11 +38,11 @@ export function dashboardService(db: Db) { if (!company) throw notFound("Company not found"); - const agentRows = await db + const agentRows = await retryIdempotentDatabaseOperation(() => db .select({ status: agents.status, count: sql`count(*)` }) .from(agents) .where(eq(agents.companyId, companyId)) - .groupBy(agents.status); + .groupBy(agents.status)); const taskRows = await retryIdempotentDatabaseOperation(() => db .select({ status: issues.status, count: sql`count(*)` }) @@ -109,7 +109,7 @@ export function dashboardService(db: Db) { // created after the run it retries, so ancestors of an out-of-window // child are themselves out of window and invisible to the membership // test below. Unbounded, the seed walks every run the company ever had. - const runActivityRows = (await db.execute(sql` + const runActivityRows = (await retryIdempotentDatabaseOperation(() => db.execute(sql` WITH RECURSIVE recovered_runs(id) AS ( SELECT parent.id FROM ${heartbeatRuns} AS child @@ -134,7 +134,7 @@ export function dashboardService(db: Db) { WHERE run.company_id = ${companyId} AND run.created_at >= ${runActivityStart.toISOString()}::timestamptz GROUP BY date, run.status, run.error_code, recovered - `)) as unknown as Iterable<{ + `))) as unknown as Iterable<{ date: string; status: string; error_code: string | null; diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 1c5cf0518f8..1b52026be75 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1,3 +1,4 @@ +import { retryIdempotentDatabaseOperation } from "../database-retry.js"; import { externalObjectService } from "./external-objects.js"; import { isAiAuthenticationBlocked } from "./ai-auth-failure.js"; import { CHAT_COMPLETION_WAKE_REASON, prepareChatCompletionTurn, chatCompletionInstruction, isCompletedOnboardingHandoffWake } from "./chat-completion-delivery.js"; @@ -30014,37 +30015,39 @@ export function heartbeatService( ) => { const safeForLegacyEncoding = await hasUnsafeTextProjectionDatabase(); const summary = options.summary === true; - const query = db - .select( - summary - ? { - ...heartbeatRunSummaryListColumns, - ...heartbeatRunListContextColumns, - } - : safeForLegacyEncoding + const rows = await retryIdempotentDatabaseOperation(async () => { + const query = db + .select( + summary ? { - ...heartbeatRunListColumns, - error: sql`NULL`.as("error"), + ...heartbeatRunSummaryListColumns, ...heartbeatRunListContextColumns, } - : { - ...heartbeatRunListColumns, - ...heartbeatRunListContextColumns, - ...heartbeatRunListResultColumns, - }, - ) - .from(heartbeatRuns) - .where( - agentId - ? and( - eq(heartbeatRuns.companyId, companyId), - eq(heartbeatRuns.agentId, agentId), - ) - : eq(heartbeatRuns.companyId, companyId), - ) - .orderBy(desc(heartbeatRuns.createdAt)); + : safeForLegacyEncoding + ? { + ...heartbeatRunListColumns, + error: sql`NULL`.as("error"), + ...heartbeatRunListContextColumns, + } + : { + ...heartbeatRunListColumns, + ...heartbeatRunListContextColumns, + ...heartbeatRunListResultColumns, + }, + ) + .from(heartbeatRuns) + .where( + agentId + ? and( + eq(heartbeatRuns.companyId, companyId), + eq(heartbeatRuns.agentId, agentId), + ) + : eq(heartbeatRuns.companyId, companyId), + ) + .orderBy(desc(heartbeatRuns.createdAt)); - const rows = limit ? await query.limit(limit) : await query; + return limit ? await query.limit(limit) : await query; + }); return rows.map((row) => { const { contextIssueId, diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 839b2945b4c..d9fec8106fe 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -1,6 +1,7 @@ import { recordChatHandoff, recordChatCompletion, existingChatCompletionReply, acknowledgeChatCompletionReply } from "./chat-completion-delivery.js"; import { mirrorSlackBoardComment, slackBoardReplyBindings } from "./slack-board-messages.js"; import { assertAgentRunWriteAllowed } from "../agent-run-cancellation.js"; +import { retryIdempotentDatabaseOperation } from "../database-retry.js"; import { externalConversationStateSql, nonIdleSlackIssueCondition, resumeSlackConversation } from "./slack-conversation-state.js"; import { documentService } from "./documents.js"; import { parseTaskSearch, taskSearchCtes, taskSearchScore } from "./task-search.js"; @@ -6662,22 +6663,22 @@ export function issueService(db: Db) { } async function getIssueByUuid(id: string) { - const row = await db + const row = await retryIdempotentDatabaseOperation(() => db .select({ ...getTableColumns(issues), externalConversationState: externalConversationStateSql() }) .from(issues) .where(eq(issues.id, id)) - .then((rows) => rows[0] ?? null); + .then((rows) => rows[0] ?? null)); if (!row) return null; const [enriched] = await withIssueLabels(db, [row]); return enriched; } async function getIssueByIdentifier(identifier: string) { - const row = await db + const row = await retryIdempotentDatabaseOperation(() => db .select({ ...getTableColumns(issues), externalConversationState: externalConversationStateSql() }) .from(issues) .where(eq(issues.identifier, identifier.toUpperCase())) - .then((rows) => rows[0] ?? null); + .then((rows) => rows[0] ?? null)); if (!row) return null; const [enriched] = await withIssueLabels(db, [row]); return enriched;