From ea4aa185b0f3bc30be1c5fc160cdfd0eb433fe41 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Thu, 15 Oct 2026 07:33:03 -0500 Subject: [PATCH] perf: batch retained-reader authorization reads (#16424) Batch retained-reader project checks through the existing multi-actor read API. Share raw database reads within one call, retain actor-specific privacy and supplied membership data, and avoid loading large saved-run payloads for identity checks. Keep current-reader writes, distinct saved-run policies, company root fences and later-call revocation checks. Add query-count and access-control regressions. All 255 affected cases and full repository typecheck/build pass. Hosted response-time acceptance remains separate. Co-Authored-By: Paperclip --- ...026-10-15-retained-reader-query-batches.md | 42 +++++++ .../repository-root-reader-queries.test.ts | 115 ++++++++++++++++++ server/src/services/authorization.ts | 6 +- .../execution-workspace-repositories.ts | 33 +++-- 4 files changed, 186 insertions(+), 10 deletions(-) create mode 100644 doc/plans/2026-10-15-retained-reader-query-batches.md create mode 100644 server/src/__tests__/repository-root-reader-queries.test.ts diff --git a/doc/plans/2026-10-15-retained-reader-query-batches.md b/doc/plans/2026-10-15-retained-reader-query-batches.md new file mode 100644 index 00000000000..7528af7feda --- /dev/null +++ b/doc/plans/2026-10-15-retained-reader-query-batches.md @@ -0,0 +1,42 @@ +# Bound retained-reader authorization database work + +## Measured problem + +A synthetic local service fixture uses 383 workspace paths and 64 retained runs +with the same agent and responsible user. It exercises the production publication +guard against disposable PostgreSQL. It does not represent a live tenant or a +Product E2E evaluation. + +The original guard made 1,431 queries. A declared 4ms round-trip delay in a +loopback proxy increased one admission to 15.19 seconds and three concurrent +admissions to 54.69 seconds. The company root fence serializes those admissions. +With this change, one admission makes 680 queries and takes 7.13 seconds under +the same proxy; three take 20.93 seconds. The zero-injected-delay samples took +359ms before and 146ms after. Timing samples are controlled local evidence, not +an agent-response or hosted latency guarantee. + +## Change boundary + +Retained readers still need individual saved-run identity and execution-policy +checks. Current readers still need the existing write checks. Existing company +root fences, row locks, source privacy and publication ordering stay in place. + +Batch retained-reader project reads through the existing multi-actor read API. +Share only database read promises within one call. Keep actor-supplied membership +snapshots and per-actor privacy results separate. Key raw reads by their company, +principal and run as applicable. Each later call reads current grants again. +Select only the run identity and saved policy fields needed by the guard, +including both native and legacy issue-binding fallbacks. + +## Verification and remaining proof + +The query-count regression must fail on the original code without changing its +800-query bound. Distinct reader identities and stored policies must still fail +closed. Grant revocation must affect the next call. Supplied membership snapshots +must match individual decisions in both actor orders without leaking into database +snapshots. Run the complete affected authorization and source-trust suites. + +Full repository typecheck and build plus fresh current-head CI and review precede +merge. A real hosted warm agent reply remains the acceptance test for the separate +under-30-second objective. Neither this service benchmark nor a fresh local +provider run establishes that hosted result. diff --git a/server/src/__tests__/repository-root-reader-queries.test.ts b/server/src/__tests__/repository-root-reader-queries.test.ts new file mode 100644 index 00000000000..4433b2f50f0 --- /dev/null +++ b/server/src/__tests__/repository-root-reader-queries.test.ts @@ -0,0 +1,115 @@ +import { createHash, randomUUID } from "node:crypto"; +import { mkdtemp, realpath, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { and, eq, inArray } from "drizzle-orm"; +import { drizzle } from "drizzle-orm/postgres-js"; +import { afterAll, beforeAll, expect, it, vi } from "vitest"; +import { agents, authUsers, companies, companyMemberships, createDb, executionWorkspaceRepositoryAccess, + executionWorkspaces, heartbeatRuns, issues, projects, projectAccessMembers, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS } from "@paperclipai/db"; +import { executionWorkspaceRepositoryService } from "../services/execution-workspace-repositories.js"; +import { authorizationService, type AuthorizationActor } from "../services/authorization.js"; +import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; + +let database: Awaited>; +let db: ReturnType; +let cwd: string; +const companyId = randomUUID(), agentId = randomUUID(), userId = randomUUID(), issueId = randomUUID(), projectId = randomUUID(); +const runIds = Array.from({ length: 64 }, () => randomUUID()); +let workspaceId: string; + +beforeAll(async () => { + vi.stubEnv("PAPERCLIP_ISSUE_PRIVACY_MODE", "enforce"); + database = await startEmbeddedPostgresTestDatabase("root-reader-query-"); + db = createDb(database.connectionString); + cwd = await realpath(await mkdtemp(path.join(tmpdir(), "root-reader-query-"))); + await db.insert(companies).values({ id: companyId, name: "Retained readers", issuePrefix: "RRQ" }); + await db.insert(authUsers).values({ id: userId, name: "Sponsor", email: `${userId}@example.test`, emailVerified: true, createdAt: new Date(), updatedAt: new Date() }); + await db.insert(agents).values({ id: agentId, companyId, name: "Worker", adapterType: "process" }); + await db.insert(companyMemberships).values([ + { companyId, principalType: "user", principalId: userId, membershipRole: "operator", status: "active" }, + { companyId, principalType: "agent", principalId: agentId, membershipRole: "member", status: "active" }, + ]); + await db.insert(projects).values({ id: projectId, companyId, name: "Source" }); + const [workspace] = await db.insert(executionWorkspaces).values({ companyId, projectId, cwd, name: "Task source", mode: "shared_workspace", strategyType: "operator_directory" }).returning(); + workspaceId = workspace.id; + await db.insert(issues).values({ id: issueId, companyId, projectId, title: "Task", assigneeAgentId: agentId, executionWorkspaceId: workspaceId }); + await db.insert(heartbeatRuns).values(runIds.map(id => ({ id, companyId, agentId, responsibleUserId: userId, + status: "succeeded", contextSnapshot: { issueId }, resultJson: { unusedPayload: "x".repeat(65536) } }))); + const rootKey = createHash("sha256").update(cwd).digest("hex"); + await db.insert(executionWorkspaceRepositoryAccess).values(runIds.map(scopeId => ({ companyId, rootKey, + canonicalCwd: cwd, kind: "reader", scopeId, issueId, agentId, responsibleUserId: userId }))); +}, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); + +afterAll(async () => { + vi.unstubAllEnvs(); + await db?.$client.end({ timeout: 0 }); + await database?.cleanup(); + if (cwd) await rm(cwd, { recursive: true, force: true }); +}); + +const input = () => ({ companyId, issueId, workspaceId, cwd, agentId, runId: runIds[0] }); + +it("bounds repeated-principal queries without loading retained run payloads", async () => { + const queries: string[] = []; + const recorded = drizzle(db.$client, { schema: db._.fullSchema, logger: { logQuery: query => queries.push(query) } }) as unknown as typeof db; + await executionWorkspaceRepositoryService(recorded).guardForPublication(input()); + expect(queries.length).toBeLessThan(800); + const runs = queries.filter(query => query.includes('from "heartbeat_runs"')); + expect(runs.length).toBeGreaterThan(0); + expect(runs.every(query => !query.includes('"result_json"') && !query.includes('"runner_profile_json"'))).toBe(true); +}); + +it("keeps each retained run identity and policy distinct for the same principals", async () => { + const service = executionWorkspaceRepositoryService(db); + const invalidId = runIds[31]; + try { + await db.update(executionWorkspaceRepositoryAccess).set({ responsibleUserId: randomUUID() }).where(eq(executionWorkspaceRepositoryAccess.scopeId, invalidId)); + await expect(service.guardForPublication(input())).rejects.toThrow("identity"); + await db.update(executionWorkspaceRepositoryAccess).set({ responsibleUserId: userId }).where(eq(executionWorkspaceRepositoryAccess.scopeId, invalidId)); + await db.update(heartbeatRuns).set({ contextSnapshot: { issueId, executionPolicy: { trustPreset: "unsupported" } } }).where(eq(heartbeatRuns.id, invalidId)); + await expect(service.guardForPublication(input())).rejects.toThrow(); + } finally { + await db.update(executionWorkspaceRepositoryAccess).set({ responsibleUserId: userId }).where(eq(executionWorkspaceRepositoryAccess.scopeId, invalidId)); + await db.update(heartbeatRuns).set({ contextSnapshot: { issueId } }).where(eq(heartbeatRuns.id, invalidId)); + } + await expect(service.guardForPublication(input())).resolves.toBeUndefined(); +}); + +it("reloads project grants on every publication attempt", async () => { + const service = executionWorkspaceRepositoryService(db); + await db.update(projects).set({ visibility: "private" }).where(eq(projects.id, projectId)); + const grants = await db.insert(projectAccessMembers).values([ + { companyId, projectId, subjectType: "agent", subjectId: agentId }, + { companyId, projectId, subjectType: "user", subjectId: userId }, + ]).returning(); + try { + await expect(service.guardForPublication(input())).resolves.toBeUndefined(); + await db.delete(projectAccessMembers).where(and(eq(projectAccessMembers.projectId, projectId), eq(projectAccessMembers.subjectType, "user"))); + await expect(service.guardForPublication(input())).rejects.toThrow(); + } finally { + await db.delete(projectAccessMembers).where(inArray(projectAccessMembers.id, grants.map(grant => grant.id))); + await db.update(projects).set({ visibility: "open" }).where(eq(projects.id, projectId)); + } +}); + +it("keeps supplied membership snapshots separate from fresh database snapshots", async () => { + const actor: AuthorizationActor = { type: "agent", source: "agent_jwt", companyId, agentId, + runId: runIds[0], onBehalfOfUserId: userId }; + const supplied = { ...actor, onBehalfOfMemberships: [{ companyId, membershipRole: "operator", status: "active" }] }; + const service = authorizationService(db); + const resource = { type: "company" as const, companyId }; + await db.update(companyMemberships).set({ status: "inactive" }).where(and(eq(companyMemberships.companyId, companyId), eq(companyMemberships.principalId, userId))); + try { + for (const candidates of [[supplied, actor], [actor, supplied]]) { + const decisions = await service.decideReadsForActors(candidates, [resource]); + for (let index = 0; index < candidates.length; index++) { + expect(decisions[index][0]).toEqual(await service.decide({ actor: { ...candidates[index] }, action: "company_scope:read", resource })); + expect(decisions[index][0].code).toBe(candidates[index] === supplied + ? "RESPONSIBLE_USER_UNAUTHORIZED" : "RESPONSIBLE_USER_UNAVAILABLE"); + } + } + } finally { + await db.update(companyMemberships).set({ status: "active" }).where(and(eq(companyMemberships.companyId, companyId), eq(companyMemberships.principalId, userId))); + } +}); diff --git a/server/src/services/authorization.ts b/server/src/services/authorization.ts index ce5810ece00..7f57d0bce93 100644 --- a/server/src/services/authorization.ts +++ b/server/src/services/authorization.ts @@ -755,7 +755,8 @@ function createAuthorizationService(db: Db | DbTransaction, readSnapshot?: ReadA async function decideReadsForActors(actors: AuthorizationActor[], resources: ReadAuthorizationResource[]): Promise { if (!actors.length) return []; - const snapshots: ReadAuthorizationSnapshot[] = actors.map(() => ({ issues: new Map(), projects: new Map(), memo: new Map() })); + const memo = new Map>(); + const snapshots: ReadAuthorizationSnapshot[] = actors.map(() => ({ issues: new Map(), projects: new Map(), memo })); const batchActors: ResponsibleUserActorWithMemo[] = actors.map(actor => ({ ...actor, __responsibleUserSnapshotMemo: undefined })); const privacyKey = (actor: AuthorizationActor) => JSON.stringify([actor.type, actor.agentId, actor.userId, actor.onBehalfOfUserId]); const privacyActors = [...new Map(batchActors.map(actor => [privacyKey(actor), actor])).values()]; @@ -892,7 +893,8 @@ function createAuthorizationService(db: Db | DbTransaction, readSnapshot?: ReadA return promise; } - const promise = loadResponsibleUserSnapshot(input.companyId, input.userId); + const promise = batchRead(`responsible:${input.companyId}:${input.userId}`, + () => loadResponsibleUserSnapshot(input.companyId, input.userId)); actorWithMemo.__responsibleUserSnapshotMemo.set(key, promise); void promise.catch(() => { if (actorWithMemo.__responsibleUserSnapshotMemo?.get(key) === promise) { diff --git a/server/src/services/execution-workspace-repositories.ts b/server/src/services/execution-workspace-repositories.ts index 27c762d112c..47d94e6e80e 100644 --- a/server/src/services/execution-workspace-repositories.ts +++ b/server/src/services/execution-workspace-repositories.ts @@ -29,6 +29,11 @@ type DbTransaction = Parameters[0]>[0]; const controllerAcquisitionMessage = "New controller repository acquisition requires a trusted task. Existing verified public checkouts remain available."; const lowTrustRepositoryMessage = "Low-trust repository reuse permits anonymous public URLs only; new acquisition is unavailable. Use an authorized configured source for private repositories."; +const rootReaderRunColumns = { + id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId, + responsibleUserId: heartbeatRuns.responsibleUserId, nativeIssueId: heartbeatRuns.nativeIssueId, issueId: heartbeatRuns.issueId, + contextSnapshot: sql>`jsonb_build_object('issueId', ${heartbeatRuns.contextSnapshot}->'issueId', 'taskId', ${heartbeatRuns.contextSnapshot}->'taskId', 'executionPolicy', ${heartbeatRuns.contextSnapshot}->'executionPolicy')`, +}; function resolveAcquisitionSource(request: PrepareWorkspaceRepository["repository"], available: ProjectRepository[], anonymousOnly: boolean) { if (anonymousOnly && request.kind === "catalog") throw forbidden(lowTrustRepositoryMessage); @@ -180,8 +185,17 @@ async function assertRootProjectAccess(db: Db | DbTransaction, actor: Authorizat } } +async function assertRootProjectsReadAccessForActors(db: Db | DbTransaction, actors: AuthorizationActor[], companyId: string, projectIds: Set) { + if (!actors.length || !projectIds.size) return; + const decisions = await authorizationService(db).decideReadsForActors(actors, + [...projectIds].map(projectId => ({ type: "project" as const, companyId, projectId }))); + if (decisions.some(batch => batch.some(decision => !decision.allowed))) { + throw forbidden("Workspace source project is no longer accessible"); + } +} + async function retainedRootReader(tx: DbTransaction, companyId: string, reader: Pick) { - const [run] = await tx.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.id, reader.scopeId), eq(heartbeatRuns.companyId, companyId))); + const [run] = await tx.select(rootReaderRunColumns).from(heartbeatRuns).where(and(eq(heartbeatRuns.id, reader.scopeId), eq(heartbeatRuns.companyId, companyId))); const [agent] = reader.agentId ? await tx.select({ id: agents.id }).from(agents).where(and(eq(agents.id, reader.agentId), eq(agents.companyId, companyId))) : []; const [issue] = reader.issueId ? await tx.select().from(issues).where(and(eq(issues.id, reader.issueId), eq(issues.companyId, companyId))) : []; if (!run || !agent || !issue || run.agentId !== reader.agentId || run.responsibleUserId !== reader.responsibleUserId @@ -260,8 +274,8 @@ async function retainGitMetadataOwners(tx: DbTransaction, input: { for (const reader of closure.overlappingEntries.filter(entry => entry.kind === "reader")) { const { actor } = await retainedRootReader(tx, input.companyId, reader); readerActors.push(actor); - await assertRootProjectAccess(tx, actor, input.companyId, projectIds); } + await assertRootProjectsReadAccessForActors(tx, readerActors, input.companyId, projectIds); await assertTaskWorkspacesReadAccessForActors(tx, readerActors, input.companyId, workspaceIds); for (const [kind, ids] of [["source", workspaceIds], ["project_source", projectIds]] as const) { for (const scopeId of ids) if (!closure.entries.some(entry => entry.kind === kind && entry.scopeId === scopeId)) { @@ -356,8 +370,8 @@ export async function retainIsolatedWorkspaceSourceAccess(db: Db, input: { const { actor, issue } = await retainedRootReader(tx, input.companyId, reader); readerActors.push(actor); await assertRootProjectTrust(tx, { companyId: input.companyId, actor, issueId: issue.id, projectIds: target.sourceProjectIds }); - await assertRootProjectAccess(tx, actor, input.companyId, target.sourceProjectIds); } + await assertRootProjectsReadAccessForActors(tx, readerActors, input.companyId, target.sourceProjectIds); await assertTaskWorkspacesReadAccessForActors(tx, readerActors, input.companyId, target.sourceIds); } catch (error) { failure ??= error; } if (!failure && verifiedCopyProjectIds) { @@ -385,7 +399,7 @@ async function assertRootProjectTrust(db: Db | DbTransaction, input: { if (!restricted.length) return; const incompatible = () => forbidden("Physical workspace source trust is incompatible with the task's execution policy"); const runId = input.actor.runId; - const [run] = runId ? await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.id, runId), eq(heartbeatRuns.companyId, input.companyId))) : []; + const [run] = runId ? await db.select(rootReaderRunColumns).from(heartbeatRuns).where(and(eq(heartbeatRuns.id, runId), eq(heartbeatRuns.companyId, input.companyId))) : []; const creation = !input.issueId && input.actor.type === "agent" && runId ? input.taskCreation : undefined; const runIssueId = run?.nativeIssueId ?? run?.contextSnapshot?.issueId ?? run?.contextSnapshot?.taskId ?? run?.issueId; const issueId = input.issueId ?? (creation && typeof runIssueId === "string" ? runIssueId : null); @@ -471,7 +485,7 @@ export function executionWorkspaceRepositoryService(db: Db) { const [project] = issue.projectId ? await reader.select().from(projects).where(and(eq(projects.id, issue.projectId), eq(projects.companyId, issue.companyId))) : []; const [workspaceSourceProject] = workspace?.projectId ? await reader.select().from(projects).where(and(eq(projects.id, workspace.projectId), eq(projects.companyId, issue.companyId))) : []; const runId = input.runId ?? issue.executionRunId; - const [run] = runId ? await reader.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.id, runId), eq(heartbeatRuns.companyId, issue.companyId))) : []; + const [run] = runId ? await reader.select(rootReaderRunColumns).from(heartbeatRuns).where(and(eq(heartbeatRuns.id, runId), eq(heartbeatRuns.companyId, issue.companyId))) : []; if (runId && (!run || (input.runId && run.agentId !== agentId) || (run.nativeIssueId ?? run.contextSnapshot?.issueId ?? run.contextSnapshot?.taskId ?? run.issueId) !== issue.id)) { throw forbidden("Task repository run identity is unavailable"); } @@ -545,12 +559,14 @@ export function executionWorkspaceRepositoryService(db: Db) { const sourceProjects = sourceIds.size ? await tx.selectDistinct({ projectId: executionWorkspaces.projectId }).from(executionWorkspaces).where(and( eq(executionWorkspaces.companyId, input.companyId), inArray(executionWorkspaces.id, [...sourceIds]))) : []; const sourceActors: AuthorizationActor[] = []; + const projectReaders: AuthorizationActor[] = []; for (const reader of readers) { const { actor, issue } = await retainedRootReader(tx, input.companyId, reader); await assertRootProjectTrust(tx, { companyId: input.companyId, projectIds: sourceProjectIds, actor, issueId: issue.id, sourceWorkspaceId: input.reader?.runId === reader.scopeId ? input.sourceWorkspaceId : undefined }); - await assertRootProjectAccess(tx, actor, input.companyId, sourceProjectIds, - input.reader?.runId === reader.scopeId ? { sharedProjectIds, issueId: issue.id } : undefined); + if (input.reader?.runId === reader.scopeId) { + await assertRootProjectAccess(tx, actor, input.companyId, sourceProjectIds, { sharedProjectIds, issueId: issue.id }); + } else projectReaders.push(actor); if (ownAdmissionOnly) { const [ownSource] = await tx.select({ id: executionWorkspaces.id }).from(executionWorkspaces).where(and( eq(executionWorkspaces.companyId, input.companyId), eq(executionWorkspaces.id, input.sourceWorkspaceId!), @@ -566,6 +582,7 @@ export function executionWorkspaceRepositoryService(db: Db) { } } } + await assertRootProjectsReadAccessForActors(tx, projectReaders, input.companyId, sourceProjectIds); if (input.actor) { if (!input.issueId) throw forbidden("Workspace write authorization requires a task"); await assertRootProjectTrust(tx, { ...input, projectIds: sourceProjectIds, actor: input.actor }); @@ -676,7 +693,7 @@ export function executionWorkspaceRepositoryService(db: Db) { throw forbidden("Task repository publication binding is unavailable"); } await requiresAnonymousAcquisition(tx, { issue, workspace, agentId: input.agentId, runId: input.runId }); - const [run] = await tx.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.id, input.runId), eq(heartbeatRuns.companyId, input.companyId))); + const [run] = await tx.select({ responsibleUserId: heartbeatRuns.responsibleUserId }).from(heartbeatRuns).where(and(eq(heartbeatRuns.id, input.runId), eq(heartbeatRuns.companyId, input.companyId))); const { actor } = await retainedRootReader(tx, input.companyId, { scopeId: input.runId, issueId: input.issueId, agentId: input.agentId, responsibleUserId: run?.responsibleUserId ?? null, });