diff --git a/doc/plans/2026-11-07-online-workspace-migration-entry.md b/doc/plans/2026-11-07-online-workspace-migration-entry.md index 666ad87a57d..25858e7107b 100644 --- a/doc/plans/2026-11-07-online-workspace-migration-entry.md +++ b/doc/plans/2026-11-07-online-workspace-migration-entry.md @@ -96,6 +96,15 @@ The entry does not assert stopped writers, terminate an agent, clear a hold or change remote authority. An unrelated active run is not a global-idle blocker. An actual scoped writer, foreign alias or unsupported source still refuses. +Programmatic administrative calls to `migrateRemoteWorkspace` and +`migrateRemoteWorkspaceCohort` can pass `sourceLockWaitMs` with their existing +`signal` to join the source-lock queue. The budget covers acquisition of the +complete sorted source scope, rather than restarting for each key. Timeout or +abort rolls back that admission before any journal reservation. This opt-in uses +an independent source transaction and preserves the lifecycle and publication +checks; activity is published after commit. Omitted options retain try-only +admission. This is a canonical API option, not an additional JSON entry field. + Before each apply, use the existing per-operation memory, CPU and temporary-disk resource preflight. A concurrency bound is not resource admission. Budget every logical capture and up to two nested-child verifications. Qualify one single diff --git a/server/src/services/remote-workspace-migration.test.ts b/server/src/services/remote-workspace-migration.test.ts index d50b58dcf3e..409cb4170ab 100644 --- a/server/src/services/remote-workspace-migration.test.ts +++ b/server/src/services/remote-workspace-migration.test.ts @@ -7,7 +7,7 @@ import { promisify } from "node:util"; import { createServer, type Server } from "node:http"; import { pathToFileURL } from "node:url"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; -import { and, eq } from "drizzle-orm"; +import { and, eq, sql } from "drizzle-orm"; import { activityLog, agents, environmentLeases, heartbeatRuns, nativeRunFinalizations, nativeRunResults, completionContracts, applyPendingMigrations, closeRegisteredClients, companies, createDb, executionWorkspaceSnapshots, executionWorkspaceRepositories, executionWorkspaceRuntimeLeases, executionWorkspaces, issueRecoveryActions, issues, projects, projectWorkspaces, workspaceOperations, workspaceRuntimeServices } from "@paperclipai/db"; import { readRemoteWorkspaceAuthority } from "@paperclipai/shared"; @@ -16,7 +16,7 @@ import { createStorageService } from "../storage/service.js"; import type { StorageService } from "../storage/types.js"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js"; import { migrateWorkspaceTrees } from "./remote-workspace-migration-batch.js"; -import { inventoryRemoteWorkspaceMigration, migrateRemoteWorkspace } from "./remote-workspace-migration.js"; +import { inventoryRemoteWorkspaceMigration, migrateRemoteWorkspace, migrateRemoteWorkspaceCohort } from "./remote-workspace-migration.js"; import { getWorkspaceSnapshot, loadWorkspaceSnapshotArchive } from "./remote-workspace-snapshot-storage.js"; import { restoreRemoteWorkspaceArchive } from "./remote-workspace-snapshot-archive.js"; import * as workspaceArchive from "./remote-workspace-snapshot-archive.js"; @@ -35,7 +35,7 @@ import { managedProjectCheckoutName } from "./managed-project-checkout-name.js"; import { captureGitWorkspaceContents, readManagedCheckoutContents } from "./git-workspace-contents.js"; import { isolatedRemoteSourcePath } from "./isolated-remote-workspace-source.js"; import { verifiedGitCacheSource } from "./git-workspace-restoration.js"; -import { withWorkspaceSourceLocks } from "./workspace-source-lock.js"; +import { withIndependentWorkspaceSourceLocks, withWorkspaceSourceLocks } from "./workspace-source-lock.js"; import { inspectOnlineWorkspaceSource } from "./online-workspace-source.js"; const execute = promisify(execFile); @@ -76,6 +76,84 @@ describe.skipIf(!support.supported)("remote workspace migration", () => { return { companyId, workspaceId, issueId: issue!.id, cwd }; } async function workspace(id: string) { return (await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, id)))[0]!; } + async function queuedFixture() { + const original = await seed(); + vi.stubEnv("PAPERCLIP_HOME", await fs.realpath(root)); + const cwd = path.join(resolvePaperclipInstanceRoot(), "isolated-workspaces", original.companyId, original.issueId); + await fs.mkdir(path.dirname(cwd), { recursive: true }); await fs.rename(original.cwd, cwd); + await db.update(executionWorkspaces).set({ cwd, providerRef: cwd }).where(eq(executionWorkspaces.id, original.workspaceId)); + return { ...original, cwd }; + } + async function sourceWriter(companyId: string, issueIds: string[]) { + let entered!: () => void, release!: () => void; + const ready = new Promise(resolve => { entered = resolve; }); + const held = new Promise(resolve => { release = resolve; }); + const done = withIndependentWorkspaceSourceLocks(db, { companyId, issueIds }, async () => { entered(); await held; }); + await ready; + return { release, done }; + } + const queuedBackends = async () => db.execute(sql`select pid from pg_stat_activity where datname=current_database() + and wait_event_type='Lock' and query like 'select pg_advisory_xact_lock%'`); + it.each(["single", "cohort"] as const)("commits a queued %s import after the finite source writer releases", async kind => { + const f = await queuedFixture(), writer = await sourceWriter(f.companyId, [f.issueId]); + let settled = false; + const importing = (kind === "single" + ? migrateRemoteWorkspace({ db, storage, ...f, apply: true, sourceLockWaitMs: 5000 }) + : migrateRemoteWorkspaceCohort({ db, storage, companyId: f.companyId, workspaceIds: [f.workspaceId], + apply: true, signal: AbortSignal.timeout(10_000), sourceLockWaitMs: 5000 })) + .then(value => ({ value }), error => ({ error })).finally(() => { settled = true; }); + try { + await expect.poll(async () => (await queuedBackends()).length, { timeout: 2000 }).toBe(1); + expect(settled).toBe(false); + expect(await db.select().from(workspaceOperations)).toHaveLength(0); + writer.release(); await writer.done; + expect(await importing).toMatchObject({ value: { applied: true } }); + expect(await db.select().from(workspaceOperations)).toEqual([expect.objectContaining({ status: "succeeded" })]); + expect(await db.select().from(executionWorkspaceSnapshots)).toEqual([expect.objectContaining({ status: "committed" })]); + expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)).not.toBeNull(); + expect(await fs.readFile(path.join(f.cwd, "saved.txt"), "utf8")).toBe("original saved work\n"); + } finally { writer.release(); await writer.done; await importing; } + }); + it.each(["timeout", "abort"] as const)("leaves the original and journal intact on queued admission %s", async mode => { + const f = await queuedFixture(), before = await workspace(f.workspaceId); + const writer = await sourceWriter(f.companyId, [f.issueId]), abort = new AbortController(), reason = new Error("cancel queued import"); + const importing = migrateRemoteWorkspace({ db, storage, ...f, apply: true, sourceLockWaitMs: mode === "timeout" ? 300 : 5000, + signal: abort.signal }).then(value => ({ value }), error => ({ error })); + try { + await expect.poll(async () => (await queuedBackends()).length, { timeout: 2000 }).toBe(1); + const cancelledAt = performance.now(); + if (mode === "abort") abort.abort(reason); + expect(await importing).toMatchObject({ error: mode === "abort" ? reason : expect.objectContaining({ message: "workspace_source_busy" }) }); + expect(performance.now() - cancelledAt).toBeLessThan(2000); + expect(await workspace(f.workspaceId)).toEqual(before); + expect(await db.select().from(workspaceOperations)).toHaveLength(0); + expect(await db.select().from(executionWorkspaceSnapshots)).toHaveLength(0); + expect(await fs.readFile(path.join(f.cwd, "saved.txt"), "utf8")).toBe("original saved work\n"); + expect(await queuedBackends()).toHaveLength(0); + } finally { writer.release(); await writer.done; await importing; } + await expect(withIndependentWorkspaceSourceLocks(db, { companyId: f.companyId, issueIds: [f.issueId] }, async () => true, true)).resolves.toBe(true); + }); + it("bounds the whole source queue and releases keys acquired before a later timeout or abort", async () => { + for (const mode of ["timeout", "abort"] as const) { + const companyId = randomUUID(), issueIds = [randomUUID(), randomUUID()].sort(); + const first = await sourceWriter(companyId, [issueIds[0]]), second = await sourceWriter(companyId, [issueIds[1]]); + const abort = new AbortController(), reason = new Error("cancel partial admission"), startedAt = performance.now(); + const waiting = withIndependentWorkspaceSourceLocks(db, { companyId, issueIds }, async () => "unexpected", true, + { waitMs: mode === "timeout" ? 600 : 5000, signal: abort.signal }).then(value => ({ value }), error => ({ error })); + try { + await expect.poll(async () => (await queuedBackends()).length).toBe(1); + if (mode === "timeout") await new Promise(resolve => setTimeout(resolve, 350)); + first.release(); await first.done; + await expect.poll(async () => (await queuedBackends()).length).toBe(1); + await expect.poll(() => withIndependentWorkspaceSourceLocks(db, { companyId, issueIds: [issueIds[0]] }, async () => false, true) + .catch(() => true)).toBe(true); + if (mode === "abort") abort.abort(reason); + expect(await waiting).toMatchObject({ error: mode === "abort" ? reason : expect.objectContaining({ message: "workspace_source_busy" }) }); + expect(performance.now() - startedAt).toBeLessThan(mode === "timeout" ? 850 : 2000); + await expect(withIndependentWorkspaceSourceLocks(db, { companyId, issueIds: [issueIds[0]] }, async () => true, true)).resolves.toBe(true); + } finally { first.release(); second.release(); await Promise.all([first.done, second.done, waiting]); } + } + }); async function pendingRepository(f: Awaited>) { const [repository] = await db.insert(executionWorkspaceRepositories).values({ companyId: f.companyId, executionWorkspaceId: f.workspaceId, requestedByIssueId: f.issueId, repositoryIdentity: "https://example.invalid/pending.git", diff --git a/server/src/services/remote-workspace-migration.ts b/server/src/services/remote-workspace-migration.ts index e166f001567..730ea633014 100644 --- a/server/src/services/remote-workspace-migration.ts +++ b/server/src/services/remote-workspace-migration.ts @@ -325,7 +325,7 @@ export async function inventoryRemoteWorkspaceMigration(db: Db, input: { company type MigrationInput = { db: Db; storage: StorageService; companyId: string; workspaceId: string; apply?: boolean; source?: RemoteWorkspaceSource; - writersStopped?: boolean; signal?: AbortSignal; + writersStopped?: boolean; signal?: AbortSignal; sourceLockWaitMs?: number; }; type MigrationTransaction = Parameters[0]>[0]; @@ -640,7 +640,8 @@ export async function migrateRemoteWorkspace(input: MigrationInput) { } finally { await childVerifier?.close(); if (scratch) await removeTaskDirectoryTree(scratch); } }; if (onlineSource) { - const result = await withWorkspaceSourceLocks(input.db, { + const sourceLocks = input.sourceLockWaitMs === undefined ? withWorkspaceSourceLocks : withIndependentWorkspaceSourceLocks; + const result = await sourceLocks(input.db, { companyId: input.companyId, issueIds: onlineSource.issueIds, projectIds: onlineSource.projectIds, }, async lock => { @@ -649,7 +650,7 @@ export async function migrateRemoteWorkspace(input: MigrationInput) { throw new IsolatedRepositorySourceBusy("workspace_source_busy"); } return migrate(lock); - }, true); + }, true, input.sourceLockWaitMs === undefined ? undefined : { waitMs: input.sourceLockWaitMs, signal: input.signal }); if (committedPublication) publishActivity(committedPublication); return result; } @@ -664,7 +665,7 @@ const cohortSchema = z.object({ export async function migrateRemoteWorkspaceCohort(input: { db: Db; storage: StorageService; companyId: string; workspaceIds: readonly string[]; - apply?: boolean; concurrency?: number; signal: AbortSignal; + apply?: boolean; concurrency?: number; signal: AbortSignal; sourceLockWaitMs?: number; }) { const companyId = z.string().uuid().parse(input.companyId).toLowerCase(); const workspaceIds = z.array(z.string().uuid()).min(1).max(8).parse(input.workspaceIds).map(id => id.toLowerCase()).sort(); @@ -803,7 +804,7 @@ export async function migrateRemoteWorkspaceCohort(input: { if (errors.length) throw new AggregateError(failure ? [failure, ...errors] : errors, "workspace_cohort_cleanup_failed"); if (!failure) signal.throwIfAborted(); } - }, true); + }, true, input.sourceLockWaitMs === undefined ? undefined : { waitMs: input.sourceLockWaitMs, signal }); for (const publication of publications) publishActivity(publication); return result; } diff --git a/server/src/services/workspace-source-lock.ts b/server/src/services/workspace-source-lock.ts index 52f3aa0269b..59fa5a334e0 100644 --- a/server/src/services/workspace-source-lock.ts +++ b/server/src/services/workspace-source-lock.ts @@ -11,6 +11,47 @@ type SourceTransaction = Parameters[0]>[0]; type SourceLockScope = { db: Db; companyId: string; keys: Set; transaction: SourceTransaction; active: boolean }; const sourceLocks = new AsyncLocalStorage(); export class IsolatedRepositorySourceBusy extends Error {} +export type WorkspaceSourceLockWait = { waitMs: number; signal?: AbortSignal }; + +async function acquireQueuedSourceLocks(db: Db, transaction: SourceTransaction, keys: string[], wait: WorkspaceSourceLockWait) { + wait.signal?.throwIfAborted(); + const deadline = performance.now() + wait.waitMs; + const [settings] = await transaction.execute(sql`select current_setting('lock_timeout') as lock_timeout, + pg_backend_pid() as pid, backend_start::text from pg_stat_activity where pid=pg_backend_pid()`); + let cancelling: Promise | undefined; + const cancel = () => { + cancelling ??= withDedicatedDbConnection(db, async control => { + const timer = setTimeout(() => { void control.$client.end({ timeout: 0 }); }, 1000); + try { + await control.execute(sql`select pg_cancel_backend(pid) from pg_stat_activity + where pid=${Number(settings.pid)} and backend_start=${String(settings.backend_start)}::timestamptz + and datname=current_database() and usename=current_user`); + } finally { clearTimeout(timer); } + }).catch(() => undefined); + }; + wait.signal?.addEventListener("abort", cancel, { once: true }); + try { + for (const key of keys) { + wait.signal?.throwIfAborted(); + const remaining = Math.ceil(deadline - performance.now()); + if (remaining <= 0) throw new IsolatedRepositorySourceBusy("workspace_source_busy"); + await transaction.execute(sql`select set_config('lock_timeout', ${`${remaining}ms`}, true)`); + await transaction.execute(sql`select pg_advisory_xact_lock(hashtextextended(${key}, 0))`); + } + wait.signal?.throwIfAborted(); + if (performance.now() >= deadline) throw new IsolatedRepositorySourceBusy("workspace_source_busy"); + await transaction.execute(sql`select set_config('lock_timeout', ${String(settings.lock_timeout)}, true)`); + wait.signal?.throwIfAborted(); + } catch (error) { + wait.signal?.throwIfAborted(); + if ((error as { cause?: { code?: string }; code?: string }).cause?.code === "55P03" + || (error as { code?: string }).code === "55P03") throw new IsolatedRepositorySourceBusy("workspace_source_busy"); + throw error; + } finally { + wait.signal?.removeEventListener("abort", cancel); + await cancelling; + } +} // Timers and detached promises keep the async context that created them, so // work can see a scope whose transaction has committed and whose dedicated @@ -26,8 +67,8 @@ function heldScope(): SourceLockScope | undefined { * Joining that scope would tie these locks to the other transaction. */ export function withIndependentWorkspaceSourceLocks(db: Db, input: { companyId: string; issueIds?: readonly string[]; projectIds?: readonly string[] }, - action: (transaction: SourceTransaction) => Promise, tryOnly = false): Promise { - return sourceLocks.exit(() => withWorkspaceSourceLocks(db, input, action, tryOnly)); + action: (transaction: SourceTransaction) => Promise, tryOnly = false, wait?: WorkspaceSourceLockWait): Promise { + return sourceLocks.exit(() => withWorkspaceSourceLocks(db, input, action, tryOnly, wait)); } export function withIsolatedRepositorySourceLock(db: Db, companyId: string, issueId: string, @@ -37,13 +78,16 @@ export function withIsolatedRepositorySourceLock(db: Db, companyId: string, i export async function withWorkspaceSourceLocks(db: Db, input: { companyId: string; issueIds?: readonly string[]; projectIds?: readonly string[] }, - action: (transaction: SourceTransaction) => Promise, tryOnly = false): Promise { + action: (transaction: SourceTransaction) => Promise, tryOnly = false, wait?: WorkspaceSourceLockWait): Promise { + if (wait && (!Number.isSafeInteger(wait.waitMs) || wait.waitMs < 1 || wait.waitMs > 2_147_483_647)) + throw new Error("workspace_source_wait_invalid"); const keys = [...new Set([ ...(input.issueIds ?? []).map(id => `isolated_repository_source:${input.companyId}:${id}`), ...(input.projectIds ?? []).map(id => `managed_project_source:${input.companyId}:${id}`), ])].sort(); if (keys.length > 128) throw new Error("workspace_source_scope_limit"); const held = heldScope(); + if (wait && held?.db === db) throw new Error("workspace_source_wait_requires_independent_scope"); const acquire = async (scope: SourceLockScope, nonblocking: boolean) => { for (const key of keys) { if (scope.keys.has(key)) continue; @@ -65,7 +109,10 @@ export async function withWorkspaceSourceLocks(db: Db, return withDedicatedDbConnection(db, connection => connection.transaction(async transaction => { const scope: SourceLockScope = { db, companyId: input.companyId, transaction, keys: new Set(), active: true }; try { - await acquire(scope, tryOnly); + if (wait) { + await acquireQueuedSourceLocks(db, transaction, keys, wait); + for (const key of keys) scope.keys.add(key); + } else await acquire(scope, tryOnly); const rows = await transaction.execute(sql`select setting::bigint as idle_timeout_ms from pg_settings where name = 'idle_in_transaction_session_timeout'`); return await withWorkspaceLockKeepalive(transaction, Number(rows[0]?.idle_timeout_ms), () => sourceLocks.run(scope, () => action(transaction)));