mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-23 17:33:51 +02:00
fix(server): retry heartbeat run read checks after a dropped connection (#16857)
`canActorReadHeartbeatRun` now wraps the issue existence read and the `issue:read` authorization decision in separate `retryIdempotentDatabaseOperation` calls, so one pooled-connection drop (`CONNECTION_CLOSED`) no longer fails heartbeat run, activity, live-event and issue-source reads. Adds a regression test and a note in doc/DATABASE.md. Refs #15248. Refs #15300. Refs #14773. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
9179d8dcc3
commit
7fed88e78d
3 files changed
+109
-7
No files matched your search
@@ -301,6 +301,12 @@ completed lookups and the separate liveness backfill are not replayed. Projectio
|
||||
callers that use a transaction retain single-attempt reads. The opt-in is only
|
||||
for a pooled database handle outside a transaction.
|
||||
|
||||
The heartbeat run read check uses the same bounded retries. Routes call it with
|
||||
the pooled database handle. Its issue existence lookup and its `issue:read`
|
||||
authorization decision retry separately, so a disconnect in the decision does not
|
||||
replay the completed lookup. Both steps only read. Missing issues, denied
|
||||
decisions, and other errors keep their usual behavior.
|
||||
|
||||
## Tool profile reads
|
||||
|
||||
Tool profile list, detail, and new-tool review queries select only the catalog
|
||||
|
||||
@@ -0,0 +1,91 @@
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import type { AuthorizationActor, AuthorizationDecision } from "../services/authorization.js";
|
||||
import { canActorReadHeartbeatRun } from "../services/heartbeat-run-privacy.js";
|
||||
|
||||
const companyId = "00000000-0000-4000-8000-000000000001";
|
||||
const issueId = "00000000-0000-4000-8000-000000000003";
|
||||
const actor = { type: "board", userId: "user-1", source: "session" } as AuthorizationActor;
|
||||
const run = { companyId, scopeKind: "issue" as const, issueId };
|
||||
|
||||
function disconnected(code = "CONNECTION_CLOSED") {
|
||||
return new Error("Failed query", {
|
||||
cause: Object.assign(new Error(`write ${code}`), { code }),
|
||||
});
|
||||
}
|
||||
|
||||
function fixture(options: { issueFailures?: Error[]; decideFailures?: Error[]; missing?: boolean; allowed?: boolean } = {}) {
|
||||
let issueReads = 0;
|
||||
const select = vi.fn(() => ({
|
||||
from: () => ({
|
||||
where: () => {
|
||||
// A failed query object stays failed. Recovery must build another one.
|
||||
const attempt = issueReads++;
|
||||
const result = Promise.resolve().then(() => {
|
||||
const error = options.issueFailures?.[attempt];
|
||||
if (error) throw error;
|
||||
return options.missing ? [] : [{ id: issueId }];
|
||||
});
|
||||
return { then: result.then.bind(result) };
|
||||
},
|
||||
}),
|
||||
}));
|
||||
let decisions = 0;
|
||||
const decide = vi.fn(async () => {
|
||||
const error = options.decideFailures?.[decisions++];
|
||||
if (error) throw error;
|
||||
return { allowed: options.allowed ?? true } as AuthorizationDecision;
|
||||
});
|
||||
return { db: { select } as unknown as Db, access: { decide }, select, decide };
|
||||
}
|
||||
|
||||
describe("heartbeat run read decision connection recovery", () => {
|
||||
it("rebuilds the issue existence read after a transient disconnect", async () => {
|
||||
const test = fixture({ issueFailures: [disconnected(), disconnected("CONNECT_TIMEOUT")] });
|
||||
|
||||
await expect(canActorReadHeartbeatRun(test.db, test.access, actor, run)).resolves.toBe(true);
|
||||
expect(test.select).toHaveBeenCalledTimes(3);
|
||||
expect(test.decide).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("retries only the issue:read decision when it disconnects", async () => {
|
||||
const test = fixture({ decideFailures: [disconnected()], allowed: false });
|
||||
|
||||
await expect(canActorReadHeartbeatRun(test.db, test.access, actor, run)).resolves.toBe(false);
|
||||
expect(test.select).toHaveBeenCalledTimes(1);
|
||||
expect(test.decide).toHaveBeenCalledTimes(2);
|
||||
expect(test.decide).toHaveBeenLastCalledWith(expect.objectContaining({
|
||||
actor,
|
||||
action: "issue:read",
|
||||
resource: expect.objectContaining({ type: "issue", companyId, issueId }),
|
||||
}));
|
||||
});
|
||||
|
||||
it("returns the final error after three disconnected attempts", async () => {
|
||||
const errors = [disconnected(), disconnected(), disconnected()];
|
||||
const test = fixture({ decideFailures: errors });
|
||||
|
||||
await expect(canActorReadHeartbeatRun(test.db, test.access, actor, run)).rejects.toBe(errors[2]);
|
||||
expect(test.decide).toHaveBeenCalledTimes(3);
|
||||
});
|
||||
|
||||
it("does not retry other errors", async () => {
|
||||
const error = new Error("permission denied for table issues");
|
||||
const test = fixture({ issueFailures: [error] });
|
||||
|
||||
await expect(canActorReadHeartbeatRun(test.db, test.access, actor, run)).rejects.toBe(error);
|
||||
expect(test.select).toHaveBeenCalledTimes(1);
|
||||
expect(test.decide).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("keeps missing issues and company-scoped runs unchanged", async () => {
|
||||
const missing = fixture({ missing: true });
|
||||
await expect(canActorReadHeartbeatRun(missing.db, missing.access, actor, run)).resolves.toBe(false);
|
||||
expect(missing.decide).not.toHaveBeenCalled();
|
||||
|
||||
const company = fixture();
|
||||
await expect(canActorReadHeartbeatRun(company.db, company.access, actor, { companyId, scopeKind: "company" }))
|
||||
.resolves.toBe(true);
|
||||
expect(company.select).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
@@ -2,6 +2,7 @@ import type { Db } from "@paperclipai/db";
|
||||
import { issues, heartbeatRuns } from "@paperclipai/db";
|
||||
import { and, eq } from "drizzle-orm";
|
||||
import { canActorReadExecutionWorkspace, type AuthorizationActor, type AuthorizationDecision } from "./authorization.js";
|
||||
import { retryIdempotentDatabaseOperation } from "../database-retry.js";
|
||||
|
||||
type RunIssueBinding = {
|
||||
companyId: string;
|
||||
@@ -30,28 +31,32 @@ export async function canActorReadHeartbeatRun(
|
||||
run: RunIssueBinding,
|
||||
): Promise<boolean> {
|
||||
if (run.scopeKind === "company") return true;
|
||||
if (!run.issueId) return false;
|
||||
const issueId = run.issueId;
|
||||
if (!issueId) return false;
|
||||
|
||||
const issueExists = await db
|
||||
// Callers pass the pooled Db, not a transaction. The existence check and
|
||||
// the issue:read decision only read, so each rebuilds after a transient
|
||||
// disconnect without replaying the other.
|
||||
const issueExists = await retryIdempotentDatabaseOperation(() => db
|
||||
.select({ id: issues.id })
|
||||
.from(issues)
|
||||
.where(and(eq(issues.id, run.issueId), eq(issues.companyId, run.companyId)))
|
||||
.then((rows) => rows.length > 0);
|
||||
.where(and(eq(issues.id, issueId), eq(issues.companyId, run.companyId)))
|
||||
.then((rows) => rows.length > 0));
|
||||
if (!issueExists) return false;
|
||||
|
||||
const decision = await access.decide({
|
||||
const decision = await retryIdempotentDatabaseOperation(() => access.decide({
|
||||
actor,
|
||||
action: "issue:read",
|
||||
resource: {
|
||||
type: "issue",
|
||||
companyId: run.companyId,
|
||||
issueId: run.issueId,
|
||||
issueId,
|
||||
projectId: null,
|
||||
parentIssueId: null,
|
||||
assigneeAgentId: null,
|
||||
assigneeUserId: null,
|
||||
},
|
||||
});
|
||||
}));
|
||||
return decision.allowed;
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user