From ec75547a2e3bdf665091da59eb15dcfca7cacf44 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Mon, 9 Nov 2026 16:49:25 -0600 Subject: [PATCH] perf(workspaces): validate cohort members concurrently (#20387) ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Online workspace migration preserves each member under shared source locks. > - A cohort must validate every member before it reserves or captures state. > - Two read phases checked each member in sequence. > - An eight-member operation spent about eight minutes in those phases before capture. > - This pull request runs the bounded readers together and joins all results. > - Each source check, lock, and write order stays in place. ## Linked Issues or Issue Description **What happened?** An eight-member cohort spent 4 minutes 13 seconds in preflight, then 2 minutes 20 seconds in serial reservation, then 4 minutes 10 seconds in the read checks after finalizer locks. Capture started only after these phases. The two read phases invoked the same checks in sequence for each member. **Expected behavior** Run these read checks together within the existing eight-member limit. Wait for every started reader before throwing or releasing the source or finalizer fence. Keep reservation and publication serial. **Steps to reproduce** 1. Import a shared-source cohort with several members. 2. Delay one member's preflight or recovery inspection. 3. Observe that the old loop does not start the next reader until that inspection ends. 4. Fail one reader while another is still running. The fence must remain held until both readers finish. **Paperclip version or commit** Reproduced baseline: `facd409073c3c149421fd8f5af86c571fc28f294`. The branch now includes `9c634e42e449e91f3935cf59036981852f94a65e` and its merged queue-owner change. The queue hunks are unchanged. **Deployment mode** Server with PostgreSQL and online workspace migration. Related: [cohort migration in #19873](https://github.com/paperclipai/paperclip/pull/19873), [indexed workspace lookups in #20372](https://github.com/paperclipai/paperclip/pull/20372), and [queued source-lock admission in #20370](https://github.com/paperclipai/paperclip/pull/20370). The duplicate search found no open change for these two read phases. ## What Changed - Run cohort preflight member checks with `Promise.allSettled`. The existing input limit bounds this phase to eight readers. - Run the member checks after finalizer lock acquisition the same way. - Throw the first rejected result in the original member order after every reader has joined. - Add one fault fixture for both phases. It checks real source/finalizer locks, peer joining, retained local data, and the absence of capture or publication. ## Verification - Against the unchanged baseline, the two new phase cases failed because only one reader started instead of two. These were behavioral failures, not setup failures. - Focused real-PostgreSQL cohort checks selected 13 cases. The first run passed 10 and failed three existing controls. The two new reader-join cases passed. Eight unrelated cases were not selected. - The first run's existing capture overlap check observed one active uploader instead of two. Two existing fence/commit controls reached their 15-second test limits. Server typechecking ran at the same time. This does not establish the cause of those failures. - One retry ran only those same three controls after typechecking finished. All three passed with unchanged source, assertions, and deadlines. Eighteen cases were not selected. The earlier failures remain accounted for; they are not classified as infrastructure faults or flakes. - `pnpm --filter @paperclipai/server typecheck` and `git diff --check` passed. - These focused counts were recorded before the non-overlapping queue-owner rebase. On final head `06158d905c4716bf44b1d028f96c706ac60e0e11`, both new reader-fence cases passed again. Nineteen cases were not selected in this recheck. - Focused commands: - `pnpm --filter @paperclipai/server exec vitest run src/services/remote-workspace-cohort-migration.test.ts -t 'joins|captures two shared-Git|before reservation|source fingerprints|member journal|outer commit|fence connection' --maxWorkers=1` - Retry only: `pnpm --filter @paperclipai/server exec vitest run src/services/remote-workspace-cohort-migration.test.ts -t 'captures two shared-Git|actual outer commit|actual fence connection' --maxWorkers=1` - Checks used Node 24.22.2, pnpm 9.15.4, and disposable PostgreSQL 18.1. No hosted operation used this patch. End-to-end improvement is unverified. - Public commands, input limits, journal state, and recovery contracts do not change. No command documentation update is needed. ## Risks - Concurrent reads increase database and filesystem load while source locks are held. The existing cohort input limit bounds each phase to eight readers. Initial source discovery already uses the same bound. - A failed phase waits for started peers. This retains the fence until their work ends. Existing query and caller limits remain in place. - Serial reservation and publication remain. Every freshness, fingerprint, recovery, and publication check still runs. - The local tests prove overlap and joining. They do not establish a hosted speedup. Database contention can limit the benefit. ## Model Used OpenAI Codex, GPT-6 family. The exact runtime model ID and context window are not exposed in this session. Used reasoning, code editing, shell tools, and real PostgreSQL tests. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge Co-authored-by: Paperclip --- .../remote-workspace-cohort-migration.test.ts | 34 +++++++++++++++++++ .../services/remote-workspace-migration.ts | 10 +++--- 2 files changed, 40 insertions(+), 4 deletions(-) diff --git a/server/src/services/remote-workspace-cohort-migration.test.ts b/server/src/services/remote-workspace-cohort-migration.test.ts index 7f5c932933a..25fe135d197 100644 --- a/server/src/services/remote-workspace-cohort-migration.test.ts +++ b/server/src/services/remote-workspace-cohort-migration.test.ts @@ -313,6 +313,40 @@ describe.skipIf(!support.supported)("bounded workspace cohort migration", () => expect.soft(await db.select().from(workspaceOperations)).toHaveLength(1); await assertLocal(f); }); + it.each(["preflight", "recovery"] as const)("joins concurrent %s validation readers before releasing the fence", async phase => { + const f = await fixture(), held = await recoverable(f), inspect = onlineWorkspaceSource.inspectOnlineWorkspaceSource; + const counts = new Map(), selected = new Set(f.leaves.map(row => row.id)); + const phaseCall = phase === "preflight" ? 2 : 4; + let entered = 0, settled = false, bothEntered!: () => void, release!: () => void, firstFailed!: () => void; + const both = new Promise(resolve => { bothEntered = resolve; }), peer = new Promise(resolve => { release = resolve; }); + const failed = new Promise(resolve => { firstFailed = resolve; }), timer = setTimeout(() => bothEntered(), 5000); + vi.spyOn(onlineWorkspaceSource, "inspectOnlineWorkspaceSource").mockImplementation(async (...args) => { + const id = args[1].id, call = (counts.get(id) ?? 0) + 1; counts.set(id, call); + if (selected.has(id) && call === phaseCall) { + if (++entered === 1) { await both; firstFailed(); throw new Error("first_validation_failure"); } + bothEntered(); await peer; + } + return inspect(...args); + }); + const put = vi.spyOn(storage, "putFile"), other = createDb(database.connectionString, { maxConnections: 1 }); + const pending = run(f).then(value => ({ value }), error => ({ error })).finally(() => { settled = true; }); + try { + await both; await failed; + expect(entered).toBe(2); + expect(settled).toBe(false); + await expect(withWorkspaceSourceLocks(other, { companyId: f.companyId, projectIds: [f.projectId] }, async () => true, true)) + .rejects.toBeInstanceOf(IsolatedRepositorySourceBusy); + if (phase === "recovery") await other.transaction(async tx => + expect(await tryLockNativeWorkspaceFinalization(tx as unknown as Db, { companyId: f.companyId, runId: held.runId })).toBe(false)); + expect(put).not.toHaveBeenCalled(); + } finally { clearTimeout(timer); bothEntered(); release(); await pending; await other.$client.end({ timeout: 1 }); } + expect(await pending).toMatchObject({ error: { message: "first_validation_failure" } }); + await expect(withWorkspaceSourceLocks(db, { companyId: f.companyId, projectIds: [f.projectId] }, async () => true, true)).resolves.toBe(true); + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, held.runId)))[0]).toEqual(held.owner); + await assertLocal(f); + expect(await db.select().from(workspaceOperations)).toHaveLength(phase === "preflight" ? 0 : 2); + expect(await db.select().from(activityLog)).toHaveLength(0); + }); it.each(["failure", "cancel"] as const)("joins active capture peers and keeps the fence on %s", async mode => { const f = await fixture(), controller = new AbortController(); let entered = 0, firstFailed!: () => void, release!: () => void; const both = new Promise(resolve => { firstFailed = resolve; }), peer = new Promise(resolve => { release = resolve; }); diff --git a/server/src/services/remote-workspace-migration.ts b/server/src/services/remote-workspace-migration.ts index 730ea633014..bb69c0b91f7 100644 --- a/server/src/services/remote-workspace-migration.ts +++ b/server/src/services/remote-workspace-migration.ts @@ -407,12 +407,13 @@ async function lockMigrationRecovery(lock: MigrationTransaction, plans: Migratio sources: online.flatMap(({ reserved, onlineSource }) => [...reserved.legacyRecoverySources, ...onlineSource!.companions.flatMap(proof => proof.sources)]), holds: online.flatMap(({ reserved, onlineSource }) => [...reserved.onlineRecoveryHolds, ...onlineSource!.companions.flatMap(proof => proof.holds)]) })) throw new Error("remote_workspace_migration_blocked:remote_copyback_owned"); - for (const { input, onlineSource, reserved } of online) { + const readers = await Promise.allSettled(online.map(async ({ input, onlineSource, reserved }) => { const current = await inspect(input.db, { ...input, onlineSource, operationId: reserved.operation!.id }, cohort); if (current.reasons.length || canonical(current.legacyRecoverySources) !== canonical(reserved.legacyRecoverySources) || canonical(current.onlineRecoveryHolds) !== canonical(reserved.onlineRecoveryHolds)) throw new Error("remote_workspace_migration_recovery_source_changed"); - } + })); + for (const reader of readers) if (reader.status === "rejected") throw reader.reason; } function createMigrationChildVerifier(input: MigrationInput, onlineSource: OnlineWorkspaceSource | undefined, sourceLock: MigrationTransaction) { @@ -735,12 +736,13 @@ export async function migrateRemoteWorkspaceCohort(input: { || new Set(cohort.members.map(member => member.operationId)).size !== workspaceIds.length || pending.some(marker => canonical(marker) !== canonical(cohort))) throw new Error("workspace_cohort_membership_changed"); const preflight = async (db: ReadDb) => { - for (const member of members) { + const readers = await Promise.allSettled(members.map(async member => { const current = await inspect(db, { ...member.input, onlineSource: member.onlineSource }, cohort); if (current.reasons.length) throw new Error(`remote_workspace_migration_blocked:${current.reasons.join(",")}`); if (fingerprint(current.workspace) !== fingerprint(initial.find(workspace => workspace.id === member.input.workspaceId)!)) throw new Error("workspace_cohort_source_changed"); - } + })); + for (const reader of readers) if (reader.status === "rejected") throw reader.reason; }; if (!input.apply) { await preflight(input.db); return { companyId, cohortId: cohort.id, applied: false, eligible: true, workspaceIds }; } const publications: Parameters[0][] = [];