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][] = [];