diff --git a/doc/DATABASE.md b/doc/DATABASE.md index af376af072d..89db9c4acdc 100644 --- a/doc/DATABASE.md +++ b/doc/DATABASE.md @@ -161,6 +161,33 @@ cohort proof for all members. Cancellation or a changed journal rolls back the whole refresh. Partial reservations still require the original aggregate proof. Archives, recovery obligations and publication checks retain their existing owners. +Git cache expiration treats a saved migration marker or unfinished migration +journal as a durable consumer. It checks this under the lifecycle lock. A failed +migration with `lastFailure` and no `retiredSave` also keeps its source and +snapshot retention hold after process loss. + +If an older server already expired members of an unpublished cohort, +`supersedeExpiredRemoteWorkspaceMigrationCohort` accepts the exact saved cohort +and the exact expired member IDs. This is an internal service API for a privileged +operator with a real database handle and reviewed company/cohort inputs. It does +not add a CLI mode or an HTTP route. An operator calls the exported owner directly; +its test and the reviewed one-time administrative qualification use that path. +It requires completed Git expiration and an +unexpired restore-point recipe for each missing member. Other members must retain +their saved source fingerprints. Fresh source, lifecycle, finalizer and row locks +protect one transaction that fails every old journal and removes only its matched +active marker. It does not report migration success or change task state. The +old cohort proof, operations, snapshots and object references remain unchanged. +The failure receipt retains those objects through the existing failed-save hold. +The optional `sourceLockWaitMs` uses the existing bounded source-lock queue. +The default still refuses a busy source immediately. Cancellation targets this +waiter only. All source and recovery checks run again after lock admission. + +New canonical imports use the current archived recipe or current online source. +They record the old operation and supersession IDs and reserve a snapshot +generation above the existing ledger maximum. A successful new import does not +retire the old hold. Release of those historical objects requires separate review. + ### Original local Git sources for archived workspaces When an archived runtime branch has an exact successful deletion receipt, its diff --git a/server/src/__tests__/git-workspace-expiration-eligibility.test.ts b/server/src/__tests__/git-workspace-expiration-eligibility.test.ts index 95a98b017a6..d0ecb32cf3b 100644 --- a/server/src/__tests__/git-workspace-expiration-eligibility.test.ts +++ b/server/src/__tests__/git-workspace-expiration-eligibility.test.ts @@ -1,7 +1,7 @@ import { randomUUID } from "node:crypto"; import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; import { eq, sql } from "drizzle-orm"; -import { agents, companies, createDb, executionWorkspaces, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations, projects, projectWorkspaces } from "@paperclipai/db"; +import { agents, companies, createDb, executionWorkspaces, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations, projects, projectWorkspaces, workspaceOperations } from "@paperclipai/db"; import { isGitWorkspaceExpirationEligible } from "../services/git-workspace-expiration.js"; import { createSystemRecovery } from "../services/system-recovery-state.js"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; @@ -34,6 +34,19 @@ describeDatabase("Git expiration eligibility", () => { return { companyId, agentId, issueId, workspace, now, old, eligible }; } + it.each(["running", "failed-hold"])("keeps a saved migration consumer after its process is gone (%s)", async state => { + const f = await fixture(); + expect(await f.eligible()).toBe(true); + const [operation] = await db.insert(workspaceOperations).values({ companyId: f.companyId, + executionWorkspaceId: f.workspace.id, issueId: f.issueId, phase: "remote_workspace_migration", + status: state === "running" ? "running" : "failed", + metadata: state === "running" ? { schema: "paperclip.remote-workspace-migration.v1" } + : { lastFailure: { code: "source_expired_during_import" } } }).returning(); + expect(await f.eligible()).toBe(false); + await db.delete(workspaceOperations).where(eq(workspaceOperations.id, operation.id)); + expect(await f.eligible()).toBe(true); + }); + async function alias(f: Awaited>, kind: string) { const companyId = randomUUID(); await db.insert(companies).values({ id: companyId, name: "Other owner", issuePrefix: `EO${companyId.slice(0, 6)}` }); diff --git a/server/src/services/git-workspace-expiration.ts b/server/src/services/git-workspace-expiration.ts index 0b9157fa4d6..2c64fc50fca 100644 --- a/server/src/services/git-workspace-expiration.ts +++ b/server/src/services/git-workspace-expiration.ts @@ -5,7 +5,7 @@ import { randomBytes } from "node:crypto"; import { acquireGitExpirationLocks } from "./git-workspace-expiration-locks.js"; import { and, eq, inArray, ne, sql } from "drizzle-orm"; import { executionWorkspaces, issues, heartbeatRuns, nativeRunFinalizations, environmentLeases, - workspaceRuntimeServices, issueRecoveryActions, projectWorkspaces, withDedicatedDbConnection, type Db } from "@paperclipai/db"; + workspaceRuntimeServices, issueRecoveryActions, projectWorkspaces, workspaceOperations, withDedicatedDbConnection, type Db } from "@paperclipai/db"; import { createIssueGitRemoteAuthProvider } from "./git-credentials.js"; import type { StorageService } from "../storage/types.js"; import { TASK_DIRECTORY_RETENTION_KEY } from "./isolated-task-directory-snapshot.js"; @@ -166,7 +166,7 @@ export async function gitWorkspaceLastActivityAt(db: Db | Tx, workspace: Workspa export async function isGitWorkspaceExpirationEligible(tx: Tx, workspace: Workspace, now: Date, idleCutoff?: Date, options: { closedSiblingsAllowed?: boolean } = {}) { - if (workspace.metadata?.reopenPendingConsumption === true) return false; + if (workspace.metadata?.reopenPendingConsumption === true || workspace.metadata?.remoteWorkspaceMigration != null) return false; const siblingBlocks = options.closedSiblingsAllowed ? sql`AND (other.status NOT IN ('archived', 'cleanup_failed') OR other.closed_at IS NULL OR other.metadata->${TASK_DIRECTORY_RETENTION_KEY}->>'removed' = 'false')` @@ -208,6 +208,10 @@ export async function isGitWorkspaceExpirationEligible(tx: Tx, workspace: Worksp AND (l.execution_workspace_id = ${workspace.id} OR l.heartbeat_run_id IN (SELECT id FROM relevant_runs)) AND (l.status = 'pending_cleanup' OR l.cleanup_status = 'failed' OR l.metadata ? 'nativeWorkspaceExportResume' OR (l.released_at IS NULL AND NOT coalesce(l.status = 'retained' AND l.lease_policy = 'reuse_by_environment' AND l.cleanup_status = 'success', false)))) + AND NOT EXISTS (SELECT 1 FROM ${workspaceOperations} o WHERE o.company_id = ${workspace.companyId} + AND o.execution_workspace_id = ${workspace.id} AND o.phase = 'remote_workspace_migration' + AND (o.status IN ('running', 'pending') OR o.metadata->>'uncertain' = 'true' + OR (o.status = 'failed' AND o.metadata ? 'lastFailure' AND o.metadata->'retiredSave' IS NULL))) AND NOT EXISTS (SELECT 1 FROM ${workspaceRuntimeServices} WHERE company_id = ${workspace.companyId} AND (execution_workspace_id = ${workspace.id} OR cwd = ${workspace.cwd} OR left(cwd, length(${workspace.cwd}) + 1) = ${workspace.cwd} || '/') AND status <> 'stopped') diff --git a/server/src/services/remote-workspace-cohort-migration.test.ts b/server/src/services/remote-workspace-cohort-migration.test.ts index 719defc7b99..abf97f246f7 100644 --- a/server/src/services/remote-workspace-cohort-migration.test.ts +++ b/server/src/services/remote-workspace-cohort-migration.test.ts @@ -4,16 +4,17 @@ import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; import { promisify } from "node:util"; +import { pathToFileURL } from "node:url"; import { eq, inArray, sql } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; -import { activityLog, agents, companies, createDb, environmentLeases, environments, heartbeatRuns, nativeRunFinalizations, executionWorkspaceSnapshots, executionWorkspaces, issues, projects, projectWorkspaces, workspaceOperations, type Db } from "@paperclipai/db"; +import { activityLog, agents, companies, createDb, environmentLeases, environments, heartbeatRuns, nativeRunFinalizations, executionWorkspaceSnapshots, executionWorkspaceRestorePoints, executionWorkspaces, issues, projects, projectWorkspaces, workspaceOperations, type Db } from "@paperclipai/db"; import { readRemoteWorkspaceAuthority } from "@paperclipai/shared"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js"; import { resolveManagedProjectWorkspaceDir } from "../home-paths.js"; import { createLocalDiskStorageProvider } from "../storage/local-disk-provider.js"; import { createStorageService } from "../storage/service.js"; import type { StorageService } from "../storage/types.js"; -import { migrateRemoteWorkspace, migrateRemoteWorkspaceCohort } from "./remote-workspace-migration.js"; +import { migrateRemoteWorkspace, migrateRemoteWorkspaceCohort, supersedeExpiredRemoteWorkspaceMigrationCohort } from "./remote-workspace-migration.js"; import { inspectOnlineWorkspaceSource } from "./online-workspace-source.js"; import { IsolatedRepositorySourceBusy, withWorkspaceSourceLocks } from "./workspace-source-lock.js"; import { getWorkspaceSnapshot, loadWorkspaceSnapshotArchive } from "./remote-workspace-snapshot-storage.js"; @@ -22,6 +23,8 @@ import { subscribeCompanyLiveEvents } from "./live-events.js"; import { restoreRemoteWorkspaceArchive } from "./remote-workspace-snapshot-archive.js"; import * as workspaceArchive from "./remote-workspace-snapshot-archive.js"; import * as onlineWorkspaceSource from "./online-workspace-source.js"; +import { expireReconstructableGitWorkspace } from "./git-workspace-expiration.js"; +import { remoteWorkspaceSnapshotRetentionService } from "./remote-workspace-snapshot-retention.js"; const support = await getEmbeddedPostgresTestSupport(), execute = promisify(execFile); const git = async (cwd: string, ...args: string[]) => (await execute("git", ["-c", `core.hooksPath=${os.devNull}`, "-C", cwd, ...args], { @@ -40,7 +43,7 @@ describe.skipIf(!support.supported)("bounded workspace cohort migration", () => }); afterEach(async () => { vi.restoreAllMocks(); - await db.delete(activityLog); await db.delete(workspaceOperations); await db.delete(executionWorkspaceSnapshots); + await db.delete(activityLog); await db.delete(workspaceOperations); await db.delete(executionWorkspaceSnapshots); await db.delete(executionWorkspaceRestorePoints); await db.execute(sql`drop trigger if exists cohort_publish_probe on execution_workspaces`); await db.execute(sql`drop function if exists cohort_publish_probe()`); await db.execute(sql`drop sequence if exists cohort_publish_attempts`); await db.execute(sql`drop trigger if exists cohort_commit_fault on workspace_operations`); await db.execute(sql`drop function if exists cohort_commit_fault()`); @@ -49,7 +52,7 @@ describe.skipIf(!support.supported)("bounded workspace cohort migration", () => await db.delete(issues); await db.delete(agents); await db.delete(executionWorkspaces); await db.delete(projectWorkspaces); await db.delete(projects); await db.delete(companies); vi.unstubAllEnvs(); await fs.rm(home, { recursive: true, force: true }); }); - async function fixture(repoName?: string) { + async function fixture(repoName?: string, leafCount = 2) { const companyId = randomUUID(), projectId = randomUUID(), root = resolveManagedProjectWorkspaceDir({ companyId, projectId, repoName }); await db.insert(companies).values({ id: companyId, name: "Cohort", issuePrefix: `C${companyId.slice(0, 6)}` }); await db.insert(projects).values({ id: projectId, companyId, name: "Shared Git" }); @@ -57,7 +60,7 @@ describe.skipIf(!support.supported)("bounded workspace cohort migration", () => await git(root, "init", "-qb", "main"); await git(root, "add", "."); await git(root, "commit", "-qm", "Original history"); const repoUrl = "https://example.invalid/cohort.git"; await git(root, "remote", "add", "origin", repoUrl); const members = []; - for (const name of ["base", "one", "two"]) { + for (const name of ["base", "one", "two", ...Array.from({ length: leafCount - 2 }, (_, index) => `leaf-${index}`)]) { const cwd = name === "base" ? root : path.join(root, ".paperclip", "worktrees", name); if (name !== "base") await git(root, "worktree", "add", "-qb", name, cwd); const [issue] = await db.insert(issues).values({ companyId, projectId, title: name }).returning(); @@ -223,6 +226,139 @@ describe.skipIf(!support.supported)("bounded workspace cohort migration", () => await assertLocal(f); expect(await db.select().from(workspaceOperations)).toHaveLength(mode === "foreign" ? 3 : 2); }); + it("compensates an exact unpublished cohort after Git expiration without losing its data or history", async () => { + const f = await fixture("cohort", 8), bare = path.join(home, "upstream", "cohort.git"), old = new Date(Date.now() - 4 * 86_400_000); + await fs.mkdir(path.dirname(bare), { recursive: true }); await git(f.root, "clone", "--bare", f.root, bare); + const repoUrl = pathToFileURL(bare).href; await git(f.root, "remote", "set-url", "origin", repoUrl); + await db.update(executionWorkspaces).set({ repoUrl, metadata: { createdByRuntime: true, gitBranchOwnershipVersion: 1 } }); + await db.update(issues).set({ status: "done", completedAt: old, updatedAt: old }); + const held = await recoverable({ ...f, leaves: [f.leaves[3]] }), put = storage.putFile.bind(storage); let uploads = 0; + const interrupted = vi.spyOn(storage, "putFile").mockImplementation(async input => { + if (++uploads === 5) throw new Error("cohort_after_four_uploads"); return put(input); + }); + await expect(run(f, { concurrency: 1 })).rejects.toThrow("cohort_after_four_uploads"); interrupted.mockRestore(); + const initialOperations = (await db.select().from(workspaceOperations)).filter(row => row.phase === "remote_workspace_migration"); + expect(initialOperations).toHaveLength(8); + const cohort = initialOperations[0].metadata!.cohortMigration as { id: string; members: Array<{ workspaceId: string; operationId: string }> }; + const expirable = cohort.members.filter(member => member.workspaceId !== f.leaves[3].id); + const expiredWorkspaceIds = [expirable[0].workspaceId, expirable.at(-1)!.workspaceId], now = new Date(Date.now() + 3 * 86_400_000); + const expireOptions = { now, expireBase: false, dryRun: false, signal: AbortSignal.timeout(60_000), + restorePoint: { storage, notBefore: old, keyDurable: true } }; + const [protectedRow] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, expiredWorkspaceIds[0])); + const restarted = createDb(database.connectionString, { maxConnections: 1 }); + try { expect(await expireReconstructableGitWorkspace(restarted, protectedRow, expireOptions)) + .toMatchObject({ expired: false, skipReason: "live_or_unresolved_consumer" }); } + finally { await restarted.$client.end({ timeout: 1 }); } + for (const workspaceId of expiredWorkspaceIds) { + const [before] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, workspaceId)); + const marker = before.metadata!.remoteWorkspaceMigration, metadata = { ...before.metadata }; delete metadata.remoteWorkspaceMigration; + const operation = initialOperations.find(row => row.executionWorkspaceId === workspaceId)!; + await db.update(workspaceOperations).set({ status: "failed" }).where(eq(workspaceOperations.id, operation.id)); + const [unprotected] = await db.update(executionWorkspaces).set({ metadata }).where(eq(executionWorkspaces.id, workspaceId)).returning(); + const expired = await expireReconstructableGitWorkspace(db, unprotected, expireOptions); + expect(expired.skipReason).toBe(null); expect(expired).toMatchObject({ expired: true, restorePoint: true }); + const [archived] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, workspaceId)); + await db.update(executionWorkspaces).set({ metadata: { ...archived.metadata, remoteWorkspaceMigration: marker } }).where(eq(executionWorkspaces.id, workspaceId)); + await db.update(workspaceOperations).set({ status: "running" }).where(eq(workspaceOperations.id, operation.id)); + await expect(fs.access(before.cwd!)).rejects.toMatchObject({ code: "ENOENT" }); + } + await expect(run(f)).rejects.toThrow("workspace_cohort_online_source_required"); + const oldOperations = await db.select().from(workspaceOperations), oldSnapshots = await db.select().from(executionWorkspaceSnapshots); + expect(oldSnapshots.filter(row => row.status === "uploaded")).toHaveLength(4); + const tasks = await db.select().from(issues), runs = await db.select().from(heartbeatRuns), leases = await db.select().from(environmentLeases); + const finalizers = await db.select().from(nativeRunFinalizations); + const compensate = (options: Partial[0]> = {}) => + supersedeExpiredRemoteWorkspaceMigrationCohort({ db, companyId: f.companyId, cohort, + expiredWorkspaceIds, signal: AbortSignal.timeout(60_000), ...options }); + const [missingProof] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, expiredWorkspaceIds[0])); + const withoutRestorePoint = { ...(missingProof.metadata!.gitCacheRestoration as Record), schema: "paperclip.git-cache-restoration.v1" }; + delete (withoutRestorePoint as Record).restorePoint; + await db.update(executionWorkspaces).set({ metadata: { ...missingProof.metadata, gitCacheRestoration: withoutRestorePoint } }) + .where(eq(executionWorkspaces.id, missingProof.id)); + const unverifiedRows = await rows(f); + await expect(compensate()).rejects.toThrow("migration_supersession_expiration_unverified"); + expect(await rows(f)).toEqual(unverifiedRows); expect(await db.select().from(workspaceOperations)).toEqual(oldOperations); + expect(await db.select().from(executionWorkspaceSnapshots)).toEqual(oldSnapshots); + await db.update(executionWorkspaces).set({ metadata: missingProof.metadata }).where(eq(executionWorkspaces.id, missingProof.id)); + const [authorityRow] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, cohort.members[2].workspaceId)); + await db.update(executionWorkspaces).set({ metadata: { ...authorityRow.metadata, remoteWorkspace: { + schema: "paperclip.remote-workspace.v1", epoch: 1, source: initialOperations[0].metadata!.source, + snapshotId: oldSnapshots.find(row => row.status === "uploaded")!.id, writerRunId: null, location: null, dirty: false } } }) + .where(eq(executionWorkspaces.id, authorityRow.id)); + await expect(compensate()).rejects.toThrow("online_workspace_source_scope_unproven"); + await db.update(executionWorkspaces).set({ metadata: authorityRow.metadata }).where(eq(executionWorkspaces.id, authorityRow.id)); + const beforeRows = await rows(f), last = cohort.members.at(-1)!; + await db.execute(sql.raw(`create function cohort_commit_fault() returns trigger language plpgsql as $$ begin if new.id='${last.operationId}'::uuid and new.status='failed' then raise exception 'cohort_compensation_fault'; end if; return new; end $$`)); + await db.execute(sql`create trigger cohort_commit_fault before update on workspace_operations for each row execute function cohort_commit_fault()`); + await expect(compensate()).rejects.toThrow(); + expect(await db.select().from(workspaceOperations)).toEqual(oldOperations); expect(await rows(f)).toEqual(beforeRows); + await db.execute(sql`drop trigger cohort_commit_fault on workspace_operations`); await db.execute(sql`drop function cohort_commit_fault()`); + const other = createDb(database.connectionString, { maxConnections: 1 }); + let receipt!: Awaited>; + const queued = () => db.execute(sql`select exists(select 1 from pg_locks where locktype='advisory' and not granted + and database=(select oid from pg_database where datname=current_database())) as waiting`); + try { + await other.transaction(async tx => { + expect(await tryLockNativeWorkspaceFinalization(tx as unknown as Db, { companyId: f.companyId, runId: held.runId })).toBe(true); + await expect(compensate()).rejects.toThrow("remote_copyback_owned"); + }); + await withWorkspaceSourceLocks(other, { companyId: f.companyId, projectIds: [f.projectId] }, async holder => { + const abort = new AbortController(), pending = compensate({ sourceLockWaitMs: 10_000, signal: abort.signal }); + const outcome = pending.then(value => ({ value, error: null }), error => ({ value: null, error })); + try { + await expect.poll(async () => (await queued())[0].waiting, { timeout: 8_000 }).toBe(true); + abort.abort(new Error("compensation_queue_cancelled")); + expect((await outcome).error?.message).toBe("compensation_queue_cancelled"); + expect((await holder.execute(sql`select exists(select 1 from pg_locks where pid=pg_backend_pid() + and locktype='advisory' and granted) as held`))[0].held).toBe(true); + expect(await db.select().from(workspaceOperations)).toEqual(oldOperations); + } finally { abort.abort(); await outcome; } + }); + let pending: ReturnType | undefined; + try { + await withWorkspaceSourceLocks(other, { companyId: f.companyId, projectIds: [f.projectId] }, async () => { + pending = compensate({ sourceLockWaitMs: 10_000 }); + void pending.catch(() => undefined); + await expect.poll(async () => (await queued())[0].waiting, { timeout: 8_000 }).toBe(true); + }); + } finally { if (pending) receipt = await pending; } + } finally { await other.$client.end({ timeout: 1 }); } + expect(receipt).toMatchObject({ superseded: true, migrationCompleted: false, replayed: false }); + await expect(compensate()).resolves.toMatchObject({ ...receipt, replayed: true }); + const failed = await db.select().from(workspaceOperations); + for (const operation of failed) expect(operation).toMatchObject({ status: "failed", metadata: { + ...oldOperations.find(row => row.id === operation.id)!.metadata, lastFailure: { schema: "paperclip.remote-workspace-migration-supersession.v1", + code: "source_expired_during_import", cohortId: cohort.id, supersessionId: receipt.supersessionId } } }); + expect(await db.select().from(executionWorkspaceSnapshots)).toEqual(oldSnapshots); + expect(await db.select().from(issues)).toEqual(tasks); expect(await db.select().from(heartbeatRuns)).toEqual(runs); + expect(await db.select().from(environmentLeases)).toEqual(leases); expect(await db.select().from(nativeRunFinalizations)).toEqual(finalizers); + for (const workspaceId of expiredWorkspaceIds) await migrateRemoteWorkspace({ db, storage, companyId: f.companyId, workspaceId, + apply: true, signal: AbortSignal.timeout(60_000) }); + await expect(run(f, { workspaceIds: cohort.members.map(member => member.workspaceId).filter(id => !expiredWorkspaceIds.includes(id)) })) + .resolves.toMatchObject({ applied: true }); + const currentOperations = await db.select().from(workspaceOperations); + for (const member of cohort.members) { + const successor = currentOperations.find(row => row.executionWorkspaceId === member.workspaceId && row.status === "succeeded")!; + expect(successor.id).not.toBe(member.operationId); + expect(successor.metadata?.migrationSupersession).toMatchObject({ operationId: member.operationId, cohortId: cohort.id, + supersessionId: receipt.supersessionId, generation: oldSnapshots.some(row => row.executionWorkspaceId === member.workspaceId) ? 2 : 1 }); + } + const deleted = vi.spyOn(storage, "deleteObject"); + const retention = await remoteWorkspaceSnapshotRetentionService(db, { storage, now: () => new Date(Date.now() + 30 * 86_400_000) }).tick(); + expect(retention.pinned).toBeGreaterThanOrEqual(oldSnapshots.length); expect(deleted).not.toHaveBeenCalled(); + expect((await db.select().from(workspaceOperations)).filter(row => oldOperations.some(old => old.id === row.id))).toEqual(failed); + await fs.rename(f.root, f.root + "-original"); + for (const workspace of f.leaves) { + const [saved] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, workspace.id)); + const ref = (await getWorkspaceSnapshot(db, { companyId: f.companyId, workspaceId: saved.id, + snapshotId: readRemoteWorkspaceAuthority(saved.metadata)!.snapshotId! }))!; + const archivePath = path.join(home, `${saved.id}.pcws`), targetDir = path.join(home, `${saved.id}-restored`); + await loadWorkspaceSnapshotArchive({ storage, reference: ref, targetPath: archivePath }); + await restoreRemoteWorkspaceArchive({ archivePath, targetDir, expectedManifestDigest: ref.manifestDigest }); + expect(await fs.readFile(path.join(targetDir, "saved.txt"), "utf8")).toBe(`${workspace.name} dirty tracked work\n`); + expect(await fs.readFile(path.join(targetDir, "untracked.txt"), "utf8")).toBe(`${workspace.name} untracked work\n`); + } + }, 120_000); it.each(["fresh", "resume"] as const)("imports a larger same-root cohort with bounded work (%s)", async mode => { const f = await fixture(), selected = await aliasMembers(f, f.parent, 10), workspaceIds = selected.map(row => row.id); for (const child of f.leaves) await migrateRemoteWorkspace({ db, storage, companyId: f.companyId, workspaceId: child.id, diff --git a/server/src/services/remote-workspace-migration.ts b/server/src/services/remote-workspace-migration.ts index b2b8c4df0cf..f470f8d2d93 100644 --- a/server/src/services/remote-workspace-migration.ts +++ b/server/src/services/remote-workspace-migration.ts @@ -29,8 +29,13 @@ import { inspectOnlineWorkspaceSource, requiresOnlineWorkspaceSource, type Onlin import { withWorkspaceSourceLocks, withIndependentWorkspaceSourceLocks, IsolatedRepositorySourceBusy } from "./workspace-source-lock.js"; import { inspectOnlineLegacyWorkspaceRecoverySource, inspectOnlineLegacyWorkspaceRecoveryHold, lockOnlineLegacyWorkspaceRecoverySources, ONLINE_LEGACY_WORKSPACE_PRESERVATION, type OnlineLegacyWorkspaceRecoveryHold } from "./online-legacy-workspace-recovery.js"; import { inspectNativeWorkspaceMigration, nativeWorkspaceMigrationMarker } from "./native-runtime/native-workspace-migration.js"; +import { readGitCacheExpiration, GIT_CACHE_EXPIRATION_REASON } from "./git-workspace-restoration.js"; const SCHEMA = "paperclip.remote-workspace-migration.v1"; +const SUPERSESSION = "paperclip.remote-workspace-migration-supersession.v1"; +const supersessionSchema = z.object({ schema: z.literal(SUPERSESSION), companyId: z.string().uuid(), workspaceId: z.string().uuid(), + cohortId: z.string().uuid(), operationId: z.string().uuid(), supersessionId: z.string().uuid(), snapshotIds: z.array(z.string().uuid()).max(8) }).strict(); +const successorSchema = supersessionSchema.extend({ generation: z.number().int().min(1).max(2_147_483_647) }); type Workspace = typeof executionWorkspaces.$inferSelect; type ReadDb = Pick; type OnlineRecoveryHold = OnlineLegacyWorkspaceRecoveryHold & { actionId: string }; @@ -347,7 +352,7 @@ type MigrationInput = { }; type MigrationTransaction = Parameters[0]>[0]; -async function resolveMigrationSource(input: MigrationInput) { +async function resolveMigrationSource(input: Pick) { const [sourceWorkspace] = !input.writersStopped ? await input.db.select().from(executionWorkspaces).where(and( eq(executionWorkspaces.companyId, input.companyId), eq(executionWorkspaces.id, input.workspaceId))) : []; const onlineSource = sourceWorkspace && !readRemoteWorkspaceAuthority(sourceWorkspace.metadata) @@ -408,10 +413,19 @@ async function reserveMigration(input: MigrationInput, onlineSource?: OnlineWork if (canonical(operation.metadata?.onlineRecoveryHolds ?? []) !== canonical(current.onlineRecoveryHolds)) throw new Error("remote_workspace_migration_source_changed"); return { workspace, authority: null, operation, source, sourceFingerprint, issueBinding, retainedDirectory: current.retainedDirectory, archivedGitSource: current.archivedGitSource, legacyRecoverySources: current.legacyRecoverySources, finalizedRecoveryHolds: current.finalizedRecoveryHolds, onlineRecoveryHolds: current.onlineRecoveryHolds, archivedRepositoryIntents: current.archivedRepositoryIntents }; } + let migrationSupersession; + if (workspace.metadata?.remoteWorkspaceMigrationSupersession != null) { + const predecessor = supersessionSchema.parse(workspace.metadata.remoteWorkspaceMigrationSupersession); + if (predecessor.companyId !== workspace.companyId || predecessor.workspaceId !== workspace.id) throw new Error("migration_supersession_lineage_changed"); + const [latest] = await tx.select({ generation: sql`coalesce(max(${executionWorkspaceSnapshots.generation}),0)` }).from(executionWorkspaceSnapshots) + .where(and(eq(executionWorkspaceSnapshots.companyId, workspace.companyId), eq(executionWorkspaceSnapshots.executionWorkspaceId, workspace.id))); + migrationSupersession = successorSchema.parse({ ...predecessor, generation: Number(latest.generation) + 1 }); + } const [operation] = await tx.insert(workspaceOperations).values({ ...(cohort ? { id: cohort.members.find(member => member.workspaceId === input.workspaceId)!.operationId } : {}), companyId: workspace.companyId, executionWorkspaceId: workspace.id, issueId: workspace.sourceIssueId, cwd: workspace.cwd ?? workspace.providerRef, phase: "remote_workspace_migration", status: "running", metadata: { schema: SCHEMA, sourceFingerprint, source, sourceIssueBinding: current.issueBinding, ...(cohort ? { cohortMigration: cohort } : {}), + ...(migrationSupersession ? { migrationSupersession } : {}), ...(nativeSessionContinuity ? { nativeSessionContinuity } : {}), nestedWorkspaceScope: { version: 1, state: "pending" }, ...(current.finalizedRecoveryHolds.length ? { finalizedRecoveryHolds: current.finalizedRecoveryHolds } : {}), @@ -488,8 +502,11 @@ async function captureMigration(input: MigrationInput, reserved: ReservedMigrati if (!refreshed) throw new Error("remote_workspace_migration_operation_changed"); operation = refreshed; } + const predecessor = operation.metadata?.migrationSupersession == null ? null : successorSchema.parse(operation.metadata.migrationSupersession); + if (predecessor && (predecessor.companyId !== input.companyId || predecessor.workspaceId !== input.workspaceId)) + throw new Error("migration_supersession_lineage_changed"); const snapshot = await beginWorkspaceSnapshot(input.db, { companyId: input.companyId, workspaceId: input.workspaceId, - operationId: operation.id, generation: 1, epoch: 1 }); + operationId: operation.id, generation: predecessor?.generation ?? 1, epoch: 1 }); const scratch = await fs.realpath(await fs.mkdtemp(path.join(os.tmpdir(), "pc-workspace-migration-"))); try { input.signal?.throwIfAborted(); @@ -932,3 +949,118 @@ export async function migrateRemoteWorkspaceCohort(input: { for (const publication of publications) publishActivity(publication); return result; } + + +export async function supersedeExpiredRemoteWorkspaceMigrationCohort(input: { + db: Db; companyId: string; cohort: unknown; expiredWorkspaceIds: readonly string[]; signal: AbortSignal; sourceLockWaitMs?: number; +}) { + const companyId = z.string().uuid().parse(input.companyId).toLowerCase(), cohort = cohortSchema.parse(input.cohort); + const expired = z.array(z.string().uuid()).min(1).max(8).parse(input.expiredWorkspaceIds).map(id => id.toLowerCase()).sort(); + const ids = cohort.members.map(member => member.workspaceId), operationIds = cohort.members.map(member => member.operationId); + if (cohort.companyId !== companyId || ids.length > 8 || new Set(ids).size !== ids.length + || new Set(operationIds).size !== ids.length || canonical(ids) !== canonical([...ids].sort()) + || new Set(expired).size !== expired.length || expired.some(id => !ids.includes(id))) throw new Error("migration_supersession_membership_changed"); + input.signal.throwIfAborted(); + const readRows = async (db: ReadDb) => ({ + workspaces: await db.select().from(executionWorkspaces).where(and(eq(executionWorkspaces.companyId, companyId), inArray(executionWorkspaces.id, ids))).orderBy(executionWorkspaces.id), + operations: await db.select().from(workspaceOperations).where(and(eq(workspaceOperations.companyId, companyId), inArray(workspaceOperations.id, operationIds))).orderBy(workspaceOperations.id), + snapshots: await db.select().from(executionWorkspaceSnapshots).where(and(eq(executionWorkspaceSnapshots.companyId, companyId), inArray(executionWorkspaceSnapshots.sourceOperationId, operationIds))).orderBy(executionWorkspaceSnapshots.id), + }); + const initial = await readRows(input.db); + if (initial.workspaces.length !== ids.length || initial.operations.length !== ids.length + || initial.operations.some(operation => operation.phase !== "remote_workspace_migration" + || cohort.members.find(member => member.operationId === operation.id)?.workspaceId !== operation.executionWorkspaceId + || canonical(operation.metadata?.cohortMigration) !== canonical(cohort)) + || initial.snapshots.some(snapshot => !["pending", "uploaded"].includes(snapshot.status) + || cohort.members.find(member => member.operationId === snapshot.sourceOperationId)?.workspaceId !== snapshot.executionWorkspaceId)) + throw new Error("migration_supersession_journal_changed"); + const previous = object(initial.operations[0].metadata?.lastFailure); + const receipt = (supersessionId: string, replayed: boolean) => ({ superseded: true, migrationCompleted: false, replayed, + companyId, cohortId: cohort.id, supersessionId, members: cohort.members.map(member => ({ ...member, + snapshotIds: initial.snapshots.filter(snapshot => snapshot.sourceOperationId === member.operationId).map(snapshot => snapshot.id) })) }); + if (initial.operations.every(operation => operation.status === "failed" && operation.finishedAt !== null + && object(operation.metadata?.lastFailure).schema === SUPERSESSION + && object(operation.metadata?.lastFailure).code === "source_expired_during_import" + && object(operation.metadata?.lastFailure).operationId === operation.id + && object(operation.metadata?.lastFailure).workspaceId === operation.executionWorkspaceId + && object(operation.metadata?.lastFailure).companyId === companyId + && object(operation.metadata?.lastFailure).cohortId === cohort.id + && object(operation.metadata?.lastFailure).supersessionId === previous.supersessionId + && canonical(object(operation.metadata?.lastFailure).expiredWorkspaceIds) === canonical(expired) + && !Object.hasOwn(operation.metadata ?? {}, "retiredSave")) && typeof previous.supersessionId === "string") return receipt(previous.supersessionId, true); + if (initial.operations.some(operation => operation.status !== "running" || operation.finishedAt !== null || operation.metadata?.uncertain === true + || Object.hasOwn(operation.metadata ?? {}, "lastFailure") || Object.hasOwn(operation.metadata ?? {}, "retiredSave"))) + throw new Error("migration_supersession_journal_changed"); + const observe = async (db: ReadDb) => { + const rows = await readRows(db), members = []; + if (JSON.stringify(rows) !== JSON.stringify(initial)) throw new Error("migration_supersession_source_changed"); + for (const workspace of rows.workspaces) { + input.signal.throwIfAborted(); + const operation = rows.operations.find(row => row.executionWorkspaceId === workspace.id)!, marker = object(workspace.metadata?.remoteWorkspaceMigration); + if (readRemoteWorkspaceAuthority(workspace.metadata) || marker.schema !== SCHEMA || marker.operationId !== operation.id + || marker.sourceFingerprint !== operation.metadata?.sourceFingerprint || canonical(marker.source) !== canonical(operation.metadata?.source) + || workspace.metadata?.remoteWorkspaceMigrationSupersession != null || operation.issueId !== workspace.sourceIssueId + || operation.cwd !== (workspace.cwd ?? workspace.providerRef)) throw new Error("migration_supersession_source_changed"); + const memberInput = { db: db as Db, companyId, workspaceId: workspace.id, signal: input.signal }; + const onlineSource = await resolveMigrationSource(memberInput), current = await inspect(db, { ...memberInput, onlineSource }, cohort); + if (current.reasons.length) throw new Error(`remote_workspace_migration_blocked:${current.reasons.join(",")}`); + if (expired.includes(workspace.id)) { + const expiration = readGitCacheExpiration(workspace), recipe = current.archivedGitSource?.recipe; + if (workspace.status !== "archived" || workspace.cleanupReason !== GIT_CACHE_EXPIRATION_REASON + || !expiration?.completed || !expiration.workspaceRemoved || !recipe?.restorePoint || recipe.restoredAt + || fingerprint(workspace) === marker.sourceFingerprint) throw new Error("migration_supersession_expiration_unverified"); + } else if (!onlineSource || fingerprint(workspace) !== marker.sourceFingerprint) throw new Error("migration_supersession_source_changed"); + members.push({ workspace, operation, onlineSource, current }); + } + return members; + }; + const observed = await observe(input.db), projectIds = [...new Set(observed.flatMap(member => member.onlineSource?.projectIds + ?? (member.workspace.projectId ? [member.workspace.projectId] : [])))].sort(); + const issueIds = [...new Set(observed.flatMap(member => member.onlineSource?.issueIds ?? [member.workspace.sourceIssueId!]))].sort(); + const lifecycleIds = [...new Set(observed.flatMap(member => [member.workspace.id, ...(member.onlineSource?.workspaceIds ?? [])]))].sort(); + const publications: Parameters[0][] = [], supersessionId = randomUUID(); + await withIndependentWorkspaceSourceLocks(input.db, { companyId, projectIds, issueIds }, async lock => { + for (const id of lifecycleIds) if (!await tryAcquireExecutionWorkspaceLifecycleLock(lock, id)) throw new IsolatedRepositorySourceBusy("workspace_source_busy"); + const current = await observe(input.db); + if (canonical(current) !== canonical(observed)) throw new Error("migration_supersession_source_changed"); + const sources = current.flatMap(member => [...member.current.legacyRecoverySources, + ...(member.onlineSource?.companions.flatMap(proof => proof.sources) ?? [])]); + const holds = current.flatMap(member => [...member.current.onlineRecoveryHolds, + ...(member.onlineSource?.companions.flatMap(proof => proof.holds) ?? [])]); + if (!await lockOnlineLegacyWorkspaceRecoverySources(lock as unknown as Db, { companyId, sources, holds })) + throw new Error("remote_workspace_migration_blocked:remote_copyback_owned"); + await lock.transaction(async tx => { + await lockMigrationParents(tx, companyId, null, [...observed.map(member => member.workspace.sourceIssueId!), + ...sources.map(source => source.issueId)]); + await tx.select().from(executionWorkspaces).where(and(eq(executionWorkspaces.companyId, companyId), inArray(executionWorkspaces.id, ids))).orderBy(executionWorkspaces.id).for("update", { noWait: true }); + await tx.select().from(workspaceOperations).where(and(eq(workspaceOperations.companyId, companyId), inArray(workspaceOperations.id, operationIds))).orderBy(workspaceOperations.id).for("update", { noWait: true }); + for (const snapshot of initial.snapshots) await tx.select().from(executionWorkspaceSnapshots).where(and( + eq(executionWorkspaceSnapshots.companyId, companyId), eq(executionWorkspaceSnapshots.id, snapshot.id))).for("update", { noWait: true }); + const checked = await observe(tx); + if (canonical(checked) !== canonical(current)) throw new Error("migration_supersession_source_changed"); + const at = new Date(); + for (const member of checked) { + input.signal.throwIfAborted(); + if (member.current.archivedGitSource) await lockArchivedGitMigrationSources(tx, member.workspace, member.current.archivedGitSource); + const snapshotIds = initial.snapshots.filter(snapshot => snapshot.sourceOperationId === member.operation.id).map(snapshot => snapshot.id); + const lineage = { schema: SUPERSESSION, companyId, workspaceId: member.workspace.id, cohortId: cohort.id, + operationId: member.operation.id, supersessionId, snapshotIds }; + const [operation] = await tx.update(workspaceOperations).set({ status: "failed", finishedAt: at, updatedAt: at, + metadata: { ...member.operation.metadata, lastFailure: { ...lineage, code: "source_expired_during_import", at: at.toISOString(), expiredWorkspaceIds: expired } } }).where(and( + eq(workspaceOperations.companyId, companyId), eq(workspaceOperations.executionWorkspaceId, member.workspace.id), eq(workspaceOperations.id, member.operation.id), + eq(workspaceOperations.phase, "remote_workspace_migration"), eq(workspaceOperations.status, "running"), sql`${workspaceOperations.finishedAt} is null`, eq(workspaceOperations.metadata, member.operation.metadata!))).returning(); + const metadata: Record = { ...member.workspace.metadata, remoteWorkspaceMigrationSupersession: lineage }; delete metadata.remoteWorkspaceMigration; + const [workspace] = await tx.update(executionWorkspaces).set({ metadata, updatedAt: at }).where(and( + eq(executionWorkspaces.companyId, companyId), eq(executionWorkspaces.id, member.workspace.id), eq(executionWorkspaces.metadata, member.workspace.metadata!))).returning(); + if (!operation || !workspace) throw new Error("migration_supersession_compare_and_swap_failed"); + publications.push((await persistActivity(tx as unknown as Db, { companyId, actorType: "system", actorId: "remote-workspace-migration", + action: "workspace.migration_failed", entityType: "execution_workspace", entityId: workspace.id, + details: { code: "source_expired_during_import", cohortId: cohort.id, operationId: operation.id, supersessionId, snapshotIds } })).publication); + } + await notifyRemoteWorkspaceSnapshots(tx, `cohort:${cohort.id}`); + input.signal.throwIfAborted(); + }); + }, true, input.sourceLockWaitMs === undefined ? undefined : { waitMs: input.sourceLockWaitMs, signal: input.signal }); + for (const publication of publications) publishActivity(publication); + return receipt(supersessionId, false); +}