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 <noreply@paperclip.ing>

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
 

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<string, Error[]>;
+  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();
+  });
+
 });

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<string, Error[]>;
+  missing?: boolean;
+  encoding?: string;
+} = {}) {
+  const attempts: Array<{
+    table: string;
+    params: unknown[];
+    fields: Record<string, unknown>;
+    order: string[];
+    limit: number | undefined;
+  }> = [];
+  const row = { id: issueId, companyId, identifier: "TEST-1", externalConversationState: "waiting" };
+  const select = vi.fn((fields: Record<string, unknown> = {}) => ({
+    from(table: Parameters<typeof getTableName>[0]) {
+      const name = getTableName(table);
+      let condition: SQL;
+      let order: SQL[] = [];
+      let limit: number | undefined;
+      let result: Promise<unknown[]> | 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);
+  });
+});

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<number>`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<number>`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;

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<string | null>`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<string | null>`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,

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;