diff --git a/doc/DATABASE.md b/doc/DATABASE.md index d684b1c7e13..18a686ce799 100644 --- a/doc/DATABASE.md +++ b/doc/DATABASE.md @@ -884,6 +884,8 @@ Migration [0338_absurd_vengeance.sql](../packages/db/src/migrations/0338_absurd_ Repository admission loads retained workspace and task sources in bounded batches. Readers share source rows and privacy evaluation for identical principals, but each retained run receives its own trust and responsible-user decision. The snapshot lasts only for one read batch; later admission and publication reload current authority. Keep source-loading query counts bounded as both workspace history and retained reader counts grow. +Each physical-root scan resolves duplicate workspace paths once, with bounded concurrency for distinct paths. Alias results are discarded after that scan so later admission and publication recheck symlink targets. Provenance identities select overlapping root records through the company/root index; source retention writes use bounded batches under the existing company lock, including deny-only ownership retained after a failed copy authorization. + Git-source capture checks the actual common Git storage directory and verified checkout owners. Separately stored metadata does not make its parent a checkout. When a linked worktree cannot identify its primary checkout, the service verifies diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index b82af99d5b5..b4b8437a284 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -674,6 +674,29 @@ Set `SEARCH_EVAL_SCALE=1` to additionally measure a disposable 10,000-task, 30,000-comment dataset. `SEARCH_EVAL_REPORT=/tmp/search-quality.json` saves per-query results and latency measurements; scale measurements are opt-in. +### Repository workspace admission benchmark + +Measure the real repository preparation service with a disposable embedded +PostgreSQL database and synthetic history, without invoking an agent: + +```sh +pnpm perf:workspace-admission +pnpm perf:workspace-admission -- --realpath-delay-ms 2.3 +pnpm perf:workspace-admission -- --distinct-cwds --realpath-delay-ms 2.3 +``` + +Defaults model 8,516 workspaces, 27,971 provenance records and 28 retained +readers on the shared root. The first command measures ordinary local I/O; +the others inject filesystem metadata latency for sensitivity comparisons. +Injected latency does not establish the actual latency of a hosted volume. +`--distinct-cwds` creates separate folders to exercise bounded concurrent I/O. +`--service /absolute/path/to/baseline/server/src/services/execution-workspace-repositories.ts` +compares another checkout, which needs its workspace dependencies installed. +The JSON report times only the service call, counts filesystem lookups and SQL +statements, and verifies retained source identities. Setup and cleanup are +excluded. Source tasks and permissions are synthetic; this is not an end-to-end +agent latency measurement. + ### Recent task ordering The streamlined sidebar keeps five recent tasks per company and account in browser diff --git a/package.json b/package.json index 117c6439ecc..32ed9601455 100644 --- a/package.json +++ b/package.json @@ -3,6 +3,7 @@ "private": true, "type": "module", "scripts": { + "perf:workspace-admission": "node --import ./server/node_modules/tsx/dist/loader.mjs server/scripts/benchmark-workspace-admission.ts", "postinstall": "node scripts/link-plugin-dev-sdk.mjs", "preflight:workspace-links": "node cli/node_modules/tsx/dist/cli.mjs scripts/ensure-workspace-package-links.ts", "dev": "pnpm --filter @paperclipai/server exec tsx ../scripts/dev-runner.ts watch", diff --git a/server/scripts/benchmark-workspace-admission.ts b/server/scripts/benchmark-workspace-admission.ts new file mode 100644 index 00000000000..d48ce416f3c --- /dev/null +++ b/server/scripts/benchmark-workspace-admission.ts @@ -0,0 +1,282 @@ +import assert from "node:assert/strict"; +import { randomUUID, createHash } from "node:crypto"; +import { execFile as execFileCallback } from "node:child_process"; +import { promisify } from "node:util"; +import { mkdtemp, rm, mkdir } from "node:fs/promises"; +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { pathToFileURL } from "node:url"; +import { and, eq, inArray } from "drizzle-orm"; +import { drizzle } from "drizzle-orm/postgres-js"; +import { + companies, + createDb, + executionWorkspaceRepositoryAccess, + executionWorkspaces, + heartbeatRuns, + issues, + startEmbeddedPostgresTestDatabase, +} from "@paperclipai/db"; + +import { createAgentLifecycle } from "../src/services/agent-lifecycle.js"; + +const execFile = promisify(execFileCallback); + +function option(name: string, fallback: string) { + const index = process.argv.indexOf(name); + return index < 0 ? fallback : process.argv[index + 1] ?? fallback; +} + +function latency(name: string, fallback: number) { + const value = Number(option(name, String(fallback))); + if (!Number.isFinite(value) || value < 0) throw new Error(`${name} must be a non-negative number`); + return value; +} + +function count(name: string, fallback: number) { + const value = Number(option(name, String(fallback))); + if (!Number.isSafeInteger(value) || value < 0) throw new Error(`${name} must be a non-negative integer`); + return value; +} + +async function insertBatches>( + insert: (values: T[]) => Promise, + values: T[], +) { + for (let offset = 0; offset < values.length; offset += 500) { + await insert(values.slice(offset, offset + 500)); + } +} + +const workspaceCount = count("--workspaces", 8_516); +const rootWorkspaceCount = count("--root-workspaces", 5_315); +const sourceEntryCount = count("--source-entries", 17_256); +const isolatedSourceEntryCount = count("--isolated-source-entries", 10_630); +const readerEntryCount = count("--readers", 51); +const rootReaderCount = count("--root-readers", 28); +const otherAccessEntryCount = count("--other-access-entries", 34); +const realpathDelayMs = latency("--realpath-delay-ms", 0); +const distinctCwds = process.argv.includes("--distinct-cwds"); +if (workspaceCount < 1 || rootWorkspaceCount < 1 || rootWorkspaceCount > workspaceCount) { + throw new Error("Need at least one workspace; root workspaces must be between one and the workspace total"); +} +if (readerEntryCount < rootReaderCount) { + throw new Error("Root reader count cannot exceed the company-wide reader total"); +} + +const configuredService = option("--service", ""); +const serviceUrl = configuredService + ? pathToFileURL(path.resolve(configuredService)) + : new URL("../src/services/execution-workspace-repositories.ts", import.meta.url); + +let measuring = false; +let realpathCalls = 0; +let realpathIoMs = 0; +let injectedWaitMs = 0; +const actualRealpath = fs.realpath.bind(fs); +fs.realpath = (async (...args: Parameters) => { + if (!measuring) return actualRealpath(...args); + const started = performance.now(); + try { + return await actualRealpath(...args); + } finally { + realpathCalls += 1; + realpathIoMs += performance.now() - started; + if (realpathDelayMs > 0) { + const waitStarted = performance.now(); + await new Promise((resolve) => setTimeout(resolve, realpathDelayMs)); + injectedWaitMs += performance.now() - waitStarted; + } + } +}) as typeof fs.realpath; + +const { executionWorkspaceRepositoryService } = await import(serviceUrl.href); + +const temp = await startEmbeddedPostgresTestDatabase("paperclip-workspace-admission-bench-"); +const db = createDb(temp.connectionString); +const scratchRoot = await fs.realpath(await mkdtemp(path.join(os.tmpdir(), "paperclip-workspace-admission-"))); +const repoRoot = path.join(scratchRoot, "shared-repository"); +const unrelatedRoot = path.join(scratchRoot, "unrelated-workspaces"); + +try { + await mkdir(repoRoot, { recursive: true }); + await mkdir(unrelatedRoot, { recursive: true }); + await execFile("git", ["init", "--quiet", "--initial-branch=main", repoRoot]); + + const companyId = randomUUID(); + const agentId = randomUUID(); + const issueId = randomUUID(); + const workspaceIds = Array.from({ length: workspaceCount }, () => randomUUID()); + const workspaceCwds = workspaceIds.map((_, index) => { + const root = index < rootWorkspaceCount ? repoRoot : unrelatedRoot; + return distinctCwds && index !== 0 ? path.join(root, `workspace-${index}`) : root; + }); + if (distinctCwds) for (let offset = 0; offset < workspaceCwds.length; offset += 64) { + await Promise.all(workspaceCwds.slice(offset, offset + 64).map(cwd => mkdir(cwd, { recursive: true }))); + } + const runIds = Array.from({ length: Math.max(1, rootReaderCount) }, () => randomUUID()); + const now = new Date(); + const rootKey = createHash("sha256").update(repoRoot).digest("hex"); + const unrelatedKey = createHash("sha256").update(unrelatedRoot).digest("hex"); + + await db.insert(companies).values({ id: companyId, name: "Synthetic workspace-admission benchmark", issuePrefix: "WAB" }); + await createAgentLifecycle(db).requestHire(companyId, { + id: agentId, + name: "Synthetic Codie", + role: "engineer", + status: "idle", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }, { runnerResolved: true }); + + await insertBatches((rows) => db.insert(executionWorkspaces).values(rows as never), workspaceIds.map((id, index) => ({ + id, + companyId, + name: `Synthetic workspace ${index}`, + mode: "shared_workspace", + strategyType: "project_primary", + status: "active", + cwd: workspaceCwds[index], + providerType: "local_fs", + }))); + + await db.insert(issues).values({ + id: issueId, + companyId, + title: "Synthetic assigned task", + status: "in_progress", + assigneeAgentId: agentId, + executionWorkspaceId: workspaceIds[0], + responsibleUserId: "local-board", + }); + for (let offset = 0; offset < rootWorkspaceCount; offset += 500) { + await db.update(executionWorkspaces).set({ sourceIssueId: issueId }) + .where(and(eq(executionWorkspaces.companyId, companyId), inArray(executionWorkspaces.id, workspaceIds.slice(offset, offset + 500)))); + } + + if (runIds.length) { + await db.insert(heartbeatRuns).values(runIds.map((id) => ({ + id, + companyId, + agentId, + scopeKind: "issue" as const, + issueId, + nativeIssueId: issueId, + invocationSource: "on_demand", + status: "completed", + responsibleUserId: "local-board", + runtimeMode: "legacy", + contextSnapshot: { issueId }, + startedAt: now, + finishedAt: now, + }))); + } + + const accessRows: Array = []; + for (let index = 0; index < sourceEntryCount; index++) { + const isRootSource = index < rootWorkspaceCount; + accessRows.push({ + companyId, + rootKey: isRootSource ? rootKey : unrelatedKey, + canonicalCwd: isRootSource ? repoRoot : unrelatedRoot, + kind: "source", + scopeId: isRootSource ? workspaceIds[index] : randomUUID(), + }); + } + for (let index = 0; index < isolatedSourceEntryCount; index++) { + accessRows.push({ companyId, rootKey: unrelatedKey, canonicalCwd: unrelatedRoot, kind: "isolated_source", scopeId: randomUUID() }); + } + for (let index = 0; index < readerEntryCount; index++) { + const rootReader = index < rootReaderCount; + accessRows.push({ + companyId, + rootKey: rootReader ? rootKey : unrelatedKey, + canonicalCwd: rootReader ? repoRoot : unrelatedRoot, + kind: "reader", + scopeId: rootReader ? runIds[index] : randomUUID(), + issueId: rootReader ? issueId : randomUUID(), + agentId: rootReader ? agentId : randomUUID(), + responsibleUserId: "local-board", + }); + } + for (let index = 0; index < otherAccessEntryCount; index++) { + accessRows.push({ companyId, rootKey: unrelatedKey, canonicalCwd: unrelatedRoot, kind: "repository", scopeId: randomUUID() }); + } + await insertBatches((rows) => db.insert(executionWorkspaceRepositoryAccess).values(rows as never), accessRows); + + let queryCount = 0; + let provenanceInsertCount = 0; + const measuredDb = drizzle(db.$client, { + schema: db._.fullSchema, + logger: { logQuery(query: string) { + if (!measuring) return; + queryCount++; + if (query.startsWith('insert into "execution_workspace_repository_access"')) provenanceInsertCount++; + } }, + }) as typeof db; + const service = executionWorkspaceRepositoryService(measuredDb); + realpathCalls = 0; + realpathIoMs = 0; + process.stderr.write(`Starting admission benchmark: ${workspaceCount} workspaces, ${rootReaderCount} retained readers, ${realpathDelayMs} ms injected per realpath.\n`); + measuring = true; + const startedAt = performance.now(); + await service.prepareConfiguredRepositorySource({ + companyId, + issueId, + workspaceId: workspaceIds[0], + cwd: repoRoot, + sourceCwd: repoRoot, + agentId, + runId: runIds[0] ?? randomUUID(), + }); + const elapsedMs = Math.round(performance.now() - startedAt); + measuring = false; + process.stderr.write(`Admission service call completed in ${elapsedMs} ms.\n`); + + const retained = await db.select({ kind: executionWorkspaceRepositoryAccess.kind, scopeId: executionWorkspaceRepositoryAccess.scopeId }) + .from(executionWorkspaceRepositoryAccess).where(and(eq(executionWorkspaceRepositoryAccess.companyId, companyId), + eq(executionWorkspaceRepositoryAccess.rootKey, rootKey), eq(executionWorkspaceRepositoryAccess.kind, "isolated_source"))); + const retainedIds = new Set(retained.map(row => row.scopeId)); + assert.equal(retainedIds.size, rootWorkspaceCount, "Every source must retain isolated-copy provenance"); + assert.ok(workspaceIds.slice(0, rootWorkspaceCount).every(id => retainedIds.has(id)), "Source provenance identities must match"); + const metadataRoot = await fs.realpath(path.join(repoRoot, ".git")); + const metadataOwners = await db.select({ scopeId: executionWorkspaceRepositoryAccess.scopeId }) + .from(executionWorkspaceRepositoryAccess).where(and(eq(executionWorkspaceRepositoryAccess.companyId, companyId), + eq(executionWorkspaceRepositoryAccess.rootKey, createHash("sha256").update(metadataRoot).digest("hex")), + eq(executionWorkspaceRepositoryAccess.kind, "source"))); + const metadataOwnerIds = new Set(metadataOwners.map(row => row.scopeId)); + assert.equal(metadataOwnerIds.size, rootWorkspaceCount, "Git metadata must retain every source owner"); + assert.ok(workspaceIds.slice(0, rootWorkspaceCount).every(id => metadataOwnerIds.has(id)), "Git metadata source identities must match"); + const report = { + elapsedMs, + database: { queries: queryCount, provenanceInserts: provenanceInsertCount }, + realpath: { + calls: realpathCalls, + measuredIoMs: Math.round(realpathIoMs), + injectedDelayPerCallMs: realpathDelayMs, + summedInjectedWaitMs: Math.round(injectedWaitMs), + }, + fixture: { + workspaces: workspaceCount, + sharedRootWorkspaces: rootWorkspaceCount, + distinctWorkspaceCwds: new Set(workspaceCwds).size, + repositoryAccessRows: accessRows.length, + sourceRows: sourceEntryCount, + isolatedSourceRows: isolatedSourceEntryCount, + retainedReaders: rootReaderCount, + allCompanyReaders: readerEntryCount, + otherAccessRows: otherAccessEntryCount, + }, + service: serviceUrl.pathname, + note: "Synthetic local service benchmark only; no provider invocation.", + }; + await new Promise((resolve, reject) => { + process.stdout.write(`${JSON.stringify(report, null, 2)}\n`, (error) => error ? reject(error) : resolve()); + }); +} finally { + await temp.cleanup(); + await rm(scratchRoot, { recursive: true, force: true }); +} diff --git a/server/src/__tests__/repository-root-reader-queries.test.ts b/server/src/__tests__/repository-root-reader-queries.test.ts index 4433b2f50f0..5bae7040a4b 100644 --- a/server/src/__tests__/repository-root-reader-queries.test.ts +++ b/server/src/__tests__/repository-root-reader-queries.test.ts @@ -1,5 +1,6 @@ import { createHash, randomUUID } from "node:crypto"; import { mkdtemp, realpath, rm } from "node:fs/promises"; +import fs from "node:fs/promises"; import { tmpdir } from "node:os"; import path from "node:path"; import { and, eq, inArray } from "drizzle-orm"; @@ -113,3 +114,24 @@ it("keeps supplied membership snapshots separate from fresh database snapshots", await db.update(companyMemberships).set({ status: "active" }).where(and(eq(companyMemberships.companyId, companyId), eq(companyMemberships.principalId, userId))); } }); + +it("batches source retention and duplicate path reads without losing owners", async () => { + const sourceIds = Array.from({ length: 501 }, () => randomUUID()); + await db.insert(executionWorkspaces).values(sourceIds.map(id => ({ id, companyId, projectId, cwd, + sourceIssueId: issueId, name: "Shared source", mode: "shared_workspace", strategyType: "operator_directory" }))); + const queries: string[] = []; + const recorded = drizzle(db.$client, { schema: db._.fullSchema, logger: { logQuery: query => queries.push(query) } }) as unknown as typeof db; + const paths = vi.spyOn(fs, "realpath"); + try { + await executionWorkspaceRepositoryService(recorded).guardForPublication(input()); + expect(queries.filter(query => query.startsWith('insert into "execution_workspace_repository_access"')).length).toBeLessThanOrEqual(2); + expect(paths.mock.calls.length).toBeLessThanOrEqual(8); + } finally { + paths.mockRestore(); + } + const retained = await db.select({ scopeId: executionWorkspaceRepositoryAccess.scopeId }).from(executionWorkspaceRepositoryAccess) + .where(and(eq(executionWorkspaceRepositoryAccess.companyId, companyId), + eq(executionWorkspaceRepositoryAccess.rootKey, createHash("sha256").update(cwd).digest("hex")), + eq(executionWorkspaceRepositoryAccess.kind, "source"))); + expect(new Set(retained.map(row => row.scopeId))).toEqual(new Set([workspaceId, ...sourceIds])); +}); diff --git a/server/src/__tests__/task-workspace-source-copy-privacy.test.ts b/server/src/__tests__/task-workspace-source-copy-privacy.test.ts index ad2e87b789b..5c6e7e7a851 100644 --- a/server/src/__tests__/task-workspace-source-copy-privacy.test.ts +++ b/server/src/__tests__/task-workspace-source-copy-privacy.test.ts @@ -691,6 +691,33 @@ exec "$PROBE_REAL_GIT" "$@" await expect(f.service.prepareForAdmission(f.admission)).resolves.toEqual([]); }); + it("rechecks retargeted execution-workspace aliases at admission and publication", async () => { + const f = await fixture(), peer = await standardPeer(f); + const unrelated = path.join(root, `${f.issueId}-unrelated`); await mkdir(unrelated); + const alias = path.join(root, `${f.issueId}-alias`); await symlink(unrelated, alias); + await db.update(executionWorkspaces).set({ cwd: alias }).where(eq(executionWorkspaces.id, peer.workspace.id)); + await db.update(issues).set({ visibility: "private", privacyRootIssueId: peer.issueId, responsibleUserId: peer.userId }) + .where(eq(issues.id, peer.issueId)); + await expect(f.service.prepareForAdmission(f.admission)).resolves.toEqual([]); + await rm(alias); + await symlink(f.admission.cwd, alias); + await expect(assertExecutionWorkspaceRootReadAccess(db, { companyId: f.companyId, cwd: f.admission.cwd, actor: f.actor })) + .rejects.toThrow(); + await expect(f.service.prepareForAdmission(f.admission)).rejects.toThrow(); + await expect(f.service.guardForPublication(f.admission)).rejects.toThrow(); + }); + + it.each(["conflicting_path", "overlapping_invalid_key"] as const)("rejects retained root identity corruption (%s)", async scenario => { + const f = await fixture(); + const canonicalCwd = await realpath(f.admission.cwd); + const unrelated = path.join(root, `${f.issueId}-unrelated`); await mkdir(unrelated); + await db.insert(executionWorkspaceRepositoryAccess).values({ companyId: f.companyId, kind: "source", scopeId: f.workspace.id, + rootKey: createHash("sha256").update(scenario === "conflicting_path" ? canonicalCwd : await realpath(unrelated)).digest("hex"), + canonicalCwd: scenario === "conflicting_path" ? await realpath(unrelated) : path.join(canonicalCwd, "child") }); + await expect(f.service.prepareForAdmission(f.admission)).rejects.toThrow("Repository root identity conflicts"); + await expect(f.service.guardForPublication(f.admission)).rejects.toThrow("Repository root identity conflicts"); + }); + it.each(["revoked", "deleted_source", "deleted_owner", "retained_reader"] as const)("protects already-copied bytes when post-copy authorization fails (%s)", async failure => { const f = await fixture(), exec = promisify(execFileCallback); await exec("git", ["init", "--initial-branch=main", f.admission.cwd]); diff --git a/server/src/services/execution-workspace-repositories.ts b/server/src/services/execution-workspace-repositories.ts index 60acd1f81d8..aeef482fbdf 100644 --- a/server/src/services/execution-workspace-repositories.ts +++ b/server/src/services/execution-workspace-repositories.ts @@ -22,13 +22,34 @@ import { normalizeProjectRepositoryUrl } from "./project-repositories.js"; import { buildGitAuthInvocation, createGitRemoteAuthProvider } from "./git-credentials.js"; import { materializeManagedProjectWorkspace, ensureManagedRepositoriesIgnored } from "./managed-repository-checkout.js"; -import { canonicalRepositoryRoot, repositoryRootsOverlap, legacyRepositoryRootMayMatch, configuredWorkspaceRoot, verifiedIsolatedWorktree, unusedIsolatedCopyTarget, isolatedWorkspaceSourceRoots, type WorkspaceSourceCopyKind } from "./execution-workspace-source-roots.js"; +import { canonicalRepositoryRoot, resolveRepositoryRoots, repositoryRootsOverlap, legacyRepositoryRootMayMatch, configuredWorkspaceRoot, verifiedIsolatedWorktree, unusedIsolatedCopyTarget, isolatedWorkspaceSourceRoots, type WorkspaceSourceCopyKind } from "./execution-workspace-source-roots.js"; export { canonicalRepositoryRoot } from "./execution-workspace-source-roots.js"; const execFile = promisify(execFileCallback); type RepositoryRow = typeof executionWorkspaceRepositories.$inferSelect; type DbTransaction = Parameters[0]>[0]; +function repositoryAccessKeys(entries: Array>) { + return new Set(entries.map(entry => `${entry.kind}:${entry.scopeId}`)); +} + +async function retainRepositorySources(tx: DbTransaction, + root: { companyId: string; canonicalCwd: string; rootKey: string }, + existingKeys: Set, + groups: ReadonlyArray]>, +): Promise { + const additions: Array = []; + for (const [kind, ids] of groups) for (const scopeId of ids) { + const key = `${kind}:${scopeId}`; + if (existingKeys.has(key)) continue; + additions.push({ ...root, kind, scopeId }); + existingKeys.add(key); + } + for (let offset = 0; offset < additions.length; offset += 500) { + await measureRunPhase("workspace.write", () => tx.insert(executionWorkspaceRepositoryAccess).values(additions.slice(offset, offset + 500))); + } +} + 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 = { @@ -69,9 +90,22 @@ async function resolveRepositoryRootSources(tx: Db | DbTransaction, input: { }) { const canonicalCwd = await measureRunPhase("workspace.sources.canonicalize", () => canonicalRepositoryRoot(input.cwd)); const rootKey = createHash("sha256").update(canonicalCwd).digest("hex"); - const companyEntries = await measureRunPhase("workspace.sources.inventory", async () => tx.select().from(executionWorkspaceRepositoryAccess) - .where(eq(executionWorkspaceRepositoryAccess.companyId, input.companyId))); + const companyEntries = await measureRunPhase("workspace.sources.inventory", async () => { + const identities = await tx.selectDistinct({ rootKey: executionWorkspaceRepositoryAccess.rootKey, + canonicalCwd: executionWorkspaceRepositoryAccess.canonicalCwd }).from(executionWorkspaceRepositoryAccess) + .where(eq(executionWorkspaceRepositoryAccess.companyId, input.companyId)); + const rootKeys = [...new Set([rootKey, ...identities.filter(identity => repositoryRootsOverlap(identity.canonicalCwd, canonicalCwd)) + .map(identity => identity.rootKey)])]; + const retained: Array = []; + for (let offset = 0; offset < rootKeys.length; offset += 500) { + retained.push(...await tx.select().from(executionWorkspaceRepositoryAccess).where(and( + eq(executionWorkspaceRepositoryAccess.companyId, input.companyId), + inArray(executionWorkspaceRepositoryAccess.rootKey, rootKeys.slice(offset, offset + 500))))); + } + return retained; + }); const entries = companyEntries.filter(entry => entry.rootKey === rootKey); + const entryKeys = repositoryAccessKeys(entries); const overlappingEntries = companyEntries.filter(entry => repositoryRootsOverlap(entry.canonicalCwd, canonicalCwd)); if (entries.some(entry => entry.canonicalCwd !== canonicalCwd) || overlappingEntries.some(entry => createHash("sha256").update(entry.canonicalCwd).digest("hex") !== entry.rootKey)) { @@ -99,12 +133,10 @@ async function resolveRepositoryRootSources(tx: Db | DbTransaction, input: { const workspaces = await measureRunPhase("workspace.sources.inventory", async () => tx.select({ id: executionWorkspaces.id, cwd: executionWorkspaces.cwd }).from(executionWorkspaces) .where(eq(executionWorkspaces.companyId, input.companyId))); await measureRunPhase("workspace.sources.alias_scan", async () => { + const workspaceRoots = await resolveRepositoryRoots(workspaces.flatMap(workspace => workspace.cwd ? [workspace.cwd] : [])); for (const workspace of workspaces) { if (!workspace.cwd) continue; - const peerRoot = await canonicalRepositoryRoot(workspace.cwd!).catch((error: NodeJS.ErrnoException) => { - if (["ENOTDIR", "EACCES", "EPERM", "ELOOP", "ENAMETOOLONG"].includes(error.code ?? "")) return null; - throw error; - }); + const peerRoot = workspaceRoots.get(workspace.cwd); if (peerRoot && repositoryRootsOverlap(peerRoot, canonicalCwd)) { sourceIds.add(workspace.id); sharedSourceIds.add(workspace.id); addSharedSource(workspace.id, peerRoot); } @@ -126,21 +158,20 @@ async function resolveRepositoryRootSources(tx: Db | DbTransaction, input: { } for (const name of await fs.readdir(namespace)) if (name.startsWith("task-repo-")) publishedNames.add(name); } - const repositoryOrigins = await measureRunPhase("workspace.sources.inventory", async () => tx.select({ id: executionWorkspaceRepositoryAccess.scopeId }).from(executionWorkspaceRepositoryAccess) - .where(and(eq(executionWorkspaceRepositoryAccess.companyId, input.companyId), eq(executionWorkspaceRepositoryAccess.kind, "repository")))); + const repositoryOrigins = inventory.length ? await measureRunPhase("workspace.sources.inventory", async () => tx.select({ id: executionWorkspaceRepositoryAccess.scopeId }).from(executionWorkspaceRepositoryAccess) + .where(and(eq(executionWorkspaceRepositoryAccess.companyId, input.companyId), eq(executionWorkspaceRepositoryAccess.kind, "repository")))) : []; const knownOrigins = new Set(repositoryOrigins.map(origin => origin.id)); const recognizedNames = new Set(); await measureRunPhase("workspace.sources.repository_scan", async () => { + const inventoryRoots = await resolveRepositoryRoots(inventory.flatMap(peer => peer.cwd ? [peer.cwd] : []), true); + const hasReader = entries.some(entry => entry.kind === "reader"); for (const peer of inventory) { const name = path.basename(taskRepositoryRelativePath(peer.identity)); const published = publishedNames.has(name); - const peerRoot = peer.cwd ? await fs.realpath(peer.cwd).catch((error: NodeJS.ErrnoException) => { - if (["ENOENT", "ENOTDIR", "EACCES", "EPERM", "ELOOP", "ENAMETOOLONG"].includes(error.code ?? "")) return null; - throw error; - }) : null; + const peerRoot = peer.cwd ? inventoryRoots.get(peer.cwd)! : null; const matches = sourceIds.has(peer.workspaceId) || (peerRoot !== null && repositoryRootsOverlap(peerRoot, canonicalCwd)) || (published && peerRoot === null && !knownOrigins.has(peer.id)) - || (peerRoot === null && !entries.some(entry => entry.kind === "reader") && peer.cwd && await legacyRepositoryRootMayMatch(peer.cwd, canonicalCwd)); + || (peerRoot === null && !hasReader && peer.cwd && await legacyRepositoryRootMayMatch(peer.cwd, canonicalCwd)); if (matches) { sourceIds.add(peer.workspaceId); if (!isolatedSourceIds.has(peer.workspaceId) || (peerRoot && repositoryRootsOverlap(peerRoot, canonicalCwd))) { @@ -158,7 +189,7 @@ async function resolveRepositoryRootSources(tx: Db | DbTransaction, input: { const isolatedProjectIds = new Set(overlappingEntries.filter(entry => entry.kind === "isolated_project_source").map(entry => entry.scopeId)); if (sourceIds.size) { const owners = await measureRunPhase("workspace.sources.inventory", async () => tx.select({ id: executionWorkspaces.id, projectId: executionWorkspaces.projectId, strategyType: executionWorkspaces.strategyType }).from(executionWorkspaces).where(and( - eq(executionWorkspaces.companyId, input.companyId), or(...[...sourceIds].map(id => eq(executionWorkspaces.id, id)))))); + eq(executionWorkspaces.companyId, input.companyId), inArray(executionWorkspaces.id, [...sourceIds])))); for (const owner of owners) if (owner.projectId) { sourceProjectIds.add(owner.projectId); if (owner.strategyType === "git_worktree" || !sharedSourceIds.has(owner.id) @@ -178,7 +209,7 @@ async function resolveRepositoryRootSources(tx: Db | DbTransaction, input: { for (const entry of overlappingEntries) if (entry.kind === "project_source" && !isolatedProjectIds.has(entry.scopeId)) { sharedProjectIds.add(entry.scopeId); } - return { canonicalCwd, rootKey, entries, overlappingEntries, sourceIds, sharedSourceIds, isolatedSourceIds, sourceProjectIds, sharedProjectIds, isolatedProjectIds }; + return { canonicalCwd, rootKey, entries, entryKeys, overlappingEntries, sourceIds, sharedSourceIds, isolatedSourceIds, sourceProjectIds, sharedProjectIds, isolatedProjectIds }; } async function assertRootProjectAccess(db: Db | DbTransaction, actor: AuthorizationActor, companyId: string, projectIds: Set, @@ -245,16 +276,11 @@ export async function preserveExecutionWorkspaceRootProvenance(tx: DbTransaction }); if (!directory) return; if (!directory.isDirectory()) throw forbidden("Workspace files cannot be detached from unverifiable provenance"); - const { canonicalCwd, rootKey, entries, sharedSourceIds, isolatedSourceIds, sourceProjectIds, sharedProjectIds, isolatedProjectIds } = await measureRunPhase("workspace.sources", () => resolveRepositoryRootSources(tx, { + const { canonicalCwd, rootKey, entryKeys, sharedSourceIds, isolatedSourceIds, sourceProjectIds, sharedProjectIds, isolatedProjectIds } = await measureRunPhase("workspace.sources", () => resolveRepositoryRootSources(tx, { companyId: input.companyId, cwd: input.cwd!, sourceWorkspaceId: input.workspaceId, inspectCurrentAliases: true, })); - for (const [kind, ids] of [["source", sharedSourceIds], ["isolated_source", isolatedSourceIds], ["project_source", sourceProjectIds], ["shared_project_source", sharedProjectIds], ["isolated_project_source", isolatedProjectIds]] as const) { - for (const scopeId of ids) { - if (!entries.some(entry => entry.kind === kind && entry.scopeId === scopeId)) { - await measureRunPhase("workspace.write", () => tx.insert(executionWorkspaceRepositoryAccess).values({ companyId: input.companyId, canonicalCwd, rootKey, kind, scopeId })); - } - } - } + await retainRepositorySources(tx, { companyId: input.companyId, canonicalCwd, rootKey }, entryKeys, + [["source", sharedSourceIds], ["isolated_source", isolatedSourceIds], ["project_source", sourceProjectIds], ["shared_project_source", sharedProjectIds], ["isolated_project_source", isolatedProjectIds]]); } export async function assertExecutionWorkspaceRootReadAccess(db: Db | DbTransaction, input: { @@ -285,12 +311,8 @@ async function retainGitMetadataOwners(tx: DbTransaction, input: { } await measureRunPhase("workspace.readers.projects", () => assertRootProjectsReadAccessForActors(tx, readerActors, input.companyId, projectIds)); await measureRunPhase("workspace.readers.sources", () => 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)) { - await measureRunPhase("workspace.write", () => tx.insert(executionWorkspaceRepositoryAccess).values({ companyId: input.companyId, - canonicalCwd: closure.canonicalCwd, rootKey: closure.rootKey, kind, scopeId })); - } - } + await retainRepositorySources(tx, { companyId: input.companyId, canonicalCwd: closure.canonicalCwd, rootKey: closure.rootKey }, closure.entryKeys, + [["source", workspaceIds], ["project_source", projectIds]]); } export async function assertIsolatedWorkspaceSourceReadAccess(db: Db, input: { @@ -328,6 +350,7 @@ export async function retainIsolatedWorkspaceSourceAccess(db: Db, input: { const existing = await measureRunPhase("workspace.sources.inventory", async () => tx.select().from(executionWorkspaceRepositoryAccess).where(and( eq(executionWorkspaceRepositoryAccess.companyId, input.companyId), eq(executionWorkspaceRepositoryAccess.rootKey, rootKey)))); if (existing.some(entry => entry.canonicalCwd !== canonicalCwd)) throw forbidden("Repository root identity conflicts with its retained access record"); + const existingKeys = repositoryAccessKeys(existing); const workspaceIds = new Set(input.captured.workspaceIds), projectIds = new Set(input.captured.projectIds); let failure = sourceFailure; for (const cwd of roots) { @@ -343,11 +366,8 @@ export async function retainIsolatedWorkspaceSourceAccess(db: Db, input: { try { await measureRunPhase("workspace.readers.projects", () => assertRootProjectAccess(tx, input.actor!, input.companyId, new Set([id]))); } catch (error) { failure ??= error; } } - for (const [kind, ids] of [["isolated_source", workspaceIds], ["isolated_project_source", projectIds]] as const) { - for (const scopeId of ids) if (!existing.some(entry => entry.kind === kind && entry.scopeId === scopeId)) { - await measureRunPhase("workspace.write", () => tx.insert(executionWorkspaceRepositoryAccess).values({ companyId: input.companyId, rootKey, canonicalCwd, kind, scopeId })); - } - } + await retainRepositorySources(tx, { companyId: input.companyId, canonicalCwd, rootKey }, existingKeys, + [["isolated_source", workspaceIds], ["isolated_project_source", projectIds]]); const phase = input.phase ?? "preparing"; let preparingCopyProjectIds: Set | undefined; let verifiedCopyProjectIds: Set | undefined; @@ -554,9 +574,9 @@ export function executionWorkspaceRepositoryService(db: Db) { async function authorizeRootAccess(tx: DbTransaction, input: { companyId: string; cwd: string; sourceWorkspaceId?: string; reader?: RootReader; actor?: AuthorizationActor; issueId?: string; preserveOwnAdmission?: boolean; }) { - const { canonicalCwd, rootKey, entries, overlappingEntries, sourceIds, sharedSourceIds, isolatedSourceIds, sourceProjectIds, sharedProjectIds, isolatedProjectIds } = await measureRunPhase("workspace.sources", () => resolveRepositoryRootSources(tx, input)); + const { canonicalCwd, rootKey, entryKeys, overlappingEntries, sourceIds, sharedSourceIds, isolatedSourceIds, sourceProjectIds, sharedProjectIds, isolatedProjectIds } = await measureRunPhase("workspace.sources", () => resolveRepositoryRootSources(tx, input)); const readers = overlappingEntries.filter(entry => entry.kind === "reader"); - if (input.reader && !entries.some(entry => entry.kind === "reader" && entry.scopeId === input.reader!.runId)) { + if (input.reader && !entryKeys.has(`reader:${input.reader.runId}`)) { const [run] = await measureRunPhase("workspace.sources.inventory", async () => tx.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.id, input.reader!.runId), eq(heartbeatRuns.companyId, input.companyId)))); readers.push({ id: randomUUID(), companyId: input.companyId, rootKey, canonicalCwd, kind: "reader", scopeId: input.reader.runId, issueId: input.reader.issueId, agentId: input.reader.agentId, responsibleUserId: run?.responsibleUserId ?? null, createdAt: new Date(), updatedAt: new Date() }); @@ -600,17 +620,12 @@ export function executionWorkspaceRepositoryService(db: Db) { sourceActors.push(input.actor); } await measureRunPhase("workspace.readers.sources", () => assertTaskWorkspacesReadAccessForActors(tx, sourceActors, input.companyId, sourceIds)); - for (const [kind, ids] of [["source", sharedSourceIds], ["isolated_source", isolatedSourceIds], ["project_source", sourceProjectIds], - ["shared_project_source", sharedProjectIds], ["isolated_project_source", isolatedProjectIds]] as const) { - for (const scopeId of ids) { - if (!entries.some(entry => entry.kind === kind && entry.scopeId === scopeId)) { - await measureRunPhase("workspace.write", () => tx.insert(executionWorkspaceRepositoryAccess).values({ companyId: input.companyId, rootKey, canonicalCwd, kind, scopeId })); - } - } - } + await retainRepositorySources(tx, { companyId: input.companyId, canonicalCwd, rootKey }, entryKeys, + [["source", sharedSourceIds], ["isolated_source", isolatedSourceIds], ["project_source", sourceProjectIds], + ["shared_project_source", sharedProjectIds], ["isolated_project_source", isolatedProjectIds]]); await measureRunPhase("workspace.readers", async () => { for (const reader of readers) { - if (reader.rootKey === rootKey && !entries.some(entry => entry.kind === "reader" && entry.scopeId === reader.scopeId)) { + if (reader.rootKey === rootKey && !entryKeys.has(`reader:${reader.scopeId}`)) { await measureRunPhase("workspace.write", () => tx.insert(executionWorkspaceRepositoryAccess).values(reader)); } } @@ -737,6 +752,7 @@ export function executionWorkspaceRepositoryService(db: Db) { const existing = await measureRunPhase("workspace.sources.inventory", async () => tx.select().from(executionWorkspaceRepositoryAccess).where(and( eq(executionWorkspaceRepositoryAccess.companyId, input.companyId), eq(executionWorkspaceRepositoryAccess.rootKey, rootKey)))); if (existing.some(entry => entry.canonicalCwd !== canonicalCwd)) throw forbidden("Repository root identity conflicts with its retained access record"); + const existingKeys = repositoryAccessKeys(existing); let failure = sourceFailure; for (const cwd of roots) { try { @@ -751,11 +767,8 @@ export function executionWorkspaceRepositoryService(db: Db) { try { await measureRunPhase("workspace.readers.projects", () => assertRootProjectAccess(tx, actor, input.companyId, new Set([id]))); } catch (error) { failure ??= error; } } - for (const [kind, ids] of [["isolated_source", workspaceIds], ["isolated_project_source", projectIds]] as const) { - for (const scopeId of ids) if (!existing.some(entry => entry.kind === kind && entry.scopeId === scopeId)) { - await measureRunPhase("workspace.write", () => tx.insert(executionWorkspaceRepositoryAccess).values({ companyId: input.companyId, rootKey, canonicalCwd, kind, scopeId })); - } - } + await retainRepositorySources(tx, { companyId: input.companyId, canonicalCwd, rootKey }, existingKeys, + [["isolated_source", workspaceIds], ["isolated_project_source", projectIds]]); try { await authorizeRootAccess(tx, { companyId: input.companyId, cwd: input.cwd!, sourceWorkspaceId: workspace.id, actor, issueId: input.issueId }); diff --git a/server/src/services/execution-workspace-source-roots.ts b/server/src/services/execution-workspace-source-roots.ts index 36830eac7cf..33511fc3fae 100644 --- a/server/src/services/execution-workspace-source-roots.ts +++ b/server/src/services/execution-workspace-source-roots.ts @@ -21,6 +21,27 @@ export async function canonicalRepositoryRoot(cwd: string): Promise { } } +export async function resolveRepositoryRoots(cwds: Iterable, requireExisting = false): Promise> { + const paths = [...new Set(cwds)]; + const roots = new Map(); + let next = 0; + const settled = await Promise.allSettled(Array.from({ length: Math.min(16, paths.length) }, async () => { + while (next < paths.length) { + const cwd = paths[next++]; + const root = await (requireExisting ? fs.realpath(cwd) : canonicalRepositoryRoot(cwd)) + .catch((error: NodeJS.ErrnoException) => { + if (["ENOTDIR", "EACCES", "EPERM", "ELOOP", "ENAMETOOLONG"].includes(error.code ?? "") + || (requireExisting && error.code === "ENOENT")) return null; + throw error; + }); + roots.set(cwd, root); + } + })); + const rejected = settled.find(result => result.status === "rejected"); + if (rejected?.status === "rejected") throw rejected.reason; + return roots; +} + export function repositoryRootsOverlap(left: string, right: string): boolean { const contains = (parent: string, child: string) => { const relative = path.relative(parent, child);