From 0991d9bda1093b30a9bb2d71bfb85c0ee1014743 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Mon, 9 Nov 2026 21:02:02 -0600 Subject: [PATCH] fix(workspaces): protect migration sources and add guarded recovery service (#20443) ## Thinking Path > - Paperclip manages agents and their task workspaces. > - Online migration saves files before it publishes remote workspace authority. > - A saved migration journal can survive the importing process. > - Git cache expiration did not count that journal as a consumer. > - Expiration could remove a source and change its fingerprint while its import was incomplete. > - This PR protects unfinished imports and exposes a guarded internal service transition for an already affected cohort. > - New imports can preserve the current sources while all old recovery data stays retained. ## Linked Issues or Issue Description **What happened?** Git cache expiration could archive and remove a workspace after an import process ended but before its saved cohort finished. The saved source fingerprint then differed from the current archived workspace. Exact cohort resume correctly refused that changed source. Reopening the workspace cannot recreate the old metadata or fingerprint. **Expected behavior** An unfinished migration remains a durable consumer after process loss. If an older server already expired some sources, one canonical transition can fail the exact unpublished cohort without losing its snapshots, objects, source history, or recovery obligations. New attempts use the current verified sources. **Steps to reproduce** 1. Start a cohort import of eight Git worktrees with local changes. 2. Interrupt it after four snapshot uploads. 3. Let an older server expire two idle source worktrees. 4. Resume the saved cohort. It refuses the changed archived sources. Related: #20429 preserves exact cohort identity when independent companion imports change only the scope proof. #20424 preserves prepared recovery state for archived sources. This PR handles completed source expiration, not a scope-only change. Public searches found related expiration work and no matching compensation owner. ## What Changed - Count saved migration markers and unfinished migration journals as Git expiration consumers. Keep failed migration retention holds as consumers too. - Expose `supersedeExpiredRemoteWorkspaceMigrationCohort` as an internal service API for a privileged operator with a real database handle and reviewed company/cohort inputs. It is called directly by the PostgreSQL/Git test and reviewed one-time administrative qualification. It adds no CLI mode or HTTP route. The existing CLI still refuses a damaged cohort; this PR does not claim automatic CLI recovery. - Require an exact unpublished cohort of up to eight members. It requires completed expiration and an unexpired restore recipe for each missing member. The other members must retain their saved source fingerprints. - Recheck current source and recovery state under source, lifecycle, finalizer, and row locks. One transaction fails all old journals, writes audit records, and removes only their exact matched active markers. - Keep old cohort proofs, operations, snapshots, and object references. Genuine failure receipts use the existing failed-save retention hold. This transition does not report migration success. - Reuse the existing bounded source-lock queue when `sourceLockWaitMs` is set. Keep fast refusal by default. Cancellation targets only this waiter, and current checks still run after admission. - Record predecessor IDs on new attempts and reserve a snapshot generation above the existing ledger maximum. New success does not retire the old hold. - Document the owning transition and retention rules. ## Verification - The two new journal-consumer cases failed under the old expiration predicate and passed under the new predicate. - The full new recovery case passed in 21.12 seconds with real PostgreSQL locks, Git repositories, encrypted local object storage, and archive restores. It covers four retained uploads, two real expirations, unchanged old resume refusal, authority change refusal, transaction rollback, a held native finalizer, idempotence, eight new imports, dirty/untracked file restores, and a retention pass 30 days later with no object deletion. - Server TypeScript and module-boundary checks passed. - On `7cec8c915acfad2d08874eaf26abaa852f560219`, the queued-admission regression failed before forwarding the option (24.07 seconds), then passed (22.25 seconds). It verifies a real held project lock, cancellation that preserves the holder and all journals, release into full compensation, retention, and eight restored fresh imports. Current server TypeScript and module/diff checks passed. The only production follow-up changes are the optional queue input and its existing-owner call. - The earlier `737a0df3a71b10fe48d7bbfe7ffd8ae296b4852c` CI run had one Runner support failure: the unchanged credential-free Python/Pi calibration exceeded its outer 25-second subprocess deadline. That job passed 6,926 other tests and skipped four. The timeout is retained as a nonpass. The changed logical head requires fresh CI and review; no test deadline was raised. - The affected cohort, Git expiration, and eligibility suites passed all 148 cases in 228.41 seconds on `aa589f3f447e36d0277c32f8b44cbe5e84a673be`. Merged current master `a9d0946e64`; the only conflict retained both independent added test blocks. The complete merged cohort suite passed all 30 cases in 70.00 seconds on `737a0df3a71b10fe48d7bbfe7ffd8ae296b4852c`, including upstream larger-cohort controls. Final-head server TypeScript, incremental preflight, module-boundary, and diff checks passed. Unchanged expiration/eligibility evidence remains valid. Fresh `809ab704e619812d8f1bf38cf5de9c65893861cc` CI and review are pending. - On `809ab704e619812d8f1bf38cf5de9c65893861cc`, the complete recovery fixture passed in 21.38 seconds after adding one valid archived recipe with no restore point. It refuses with `migration_supersession_expiration_unverified` and keeps all workspaces, journals, and snapshots unchanged. The same case then verifies queue cancellation/release, full compensation, retained history, fresh imports, and restored bytes. Server TypeScript passed. This follow-up changes only the test and service-scope documentation; product bytes remain identical to qualified `7cec8c915acfad2d08874eaf26abaa852f560219`. - Earlier fixture attempts did not pass. They exposed the selection of a held source and an incorrect expected refusal string. The fixture now selects eligible sources and asserts the canonical authority refusal. Production guards were not relaxed. - Live qualification of the exact owning product bytes completed in 230.419 seconds. All eight old operations became failed with auditable supersession lineage. All eight old snapshot rows remained byte-for-byte identical, and their object references remain retained. The existing source-lock queue waited 1.151 seconds for a normal finalizer handoff. The three complete source/recovery passes took about 218 seconds including their transaction, which explains why the earlier 180-second total budget was insufficient for that measured workload. No source proof or publication guard was skipped. Compensation made no provider or object-storage calls. - Earlier live qualifications remain nonpasses: the 180-second attempt did not finish; its instrumented successor completed source checks and refused a busy source lock before mutation. The successful bounded-queue attempt retained those failure records. Fresh archived/online imports are a separate subsequent step and are not claimed complete here. - The local checks do not claim browser acceptance, app deployment, fleet enrollment, or a completed zero-downtime cutover. ## Risks - This owner mutates migration bookkeeping. It preserves all-or-nothing row comparisons and current source/recovery locks. A changed journal, authority, writer, finalizer, expired recipe, or partial transaction refuses the transition. - Historical snapshots and objects remain pinned after new success. This intentionally retains storage. Release requires a separate reviewed action. - An older importer cannot use the successor generation contract. Use the containing canonical owner for recovery. Normal app delivery must also carry the expiration guard; a private maintenance graph does not patch a running server's cleanup worker. - The migration owner crosses 1,000 lines. The bounded compensation transition remains beside its private source inspection, fingerprint, fence, reservation, and publication contracts. Moving it now would require exporting or duplicating those safety owners. No unrelated refactor or new framework is included. ## Model Used OpenAI GPT-6 through Codex, with reasoning, source editing, Git, and local execution tools. The exact serving model ID and context-window size are not exposed in this session. ## 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 --- doc/DATABASE.md | 27 ++++ ...t-workspace-expiration-eligibility.test.ts | 15 +- .../src/services/git-workspace-expiration.ts | 8 +- .../remote-workspace-cohort-migration.test.ts | 146 +++++++++++++++++- .../services/remote-workspace-migration.ts | 136 +++++++++++++++- 5 files changed, 322 insertions(+), 10 deletions(-) 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); +}