feat(ssh): stage, commit and abort a dormant migration against its destination (#16741 T8-c3)

src/main/ssh/orcad-catalog-durable-mutation.ts

@@ -0,0 +1,23 @@
+import type { OrcadMigrationCatalogState } from '../../shared/orcad-migration-manifest'
+
+export async function resolveDurableOrcadCatalogMutation(
+  mutate: () => Promise<OrcadMigrationCatalogState>,
+  read: () => Promise<OrcadMigrationCatalogState>,
+  accepted: (state: OrcadMigrationCatalogState) => boolean
+) {
+  try {
+    return await mutate()
+  } catch (mutationError) {
+    let observed: OrcadMigrationCatalogState
+    try {
+      observed = await read()
+    } catch {
+      throw mutationError
+    }
+    if (!accepted(observed)) {
+      throw mutationError
+    }
+  }
+  // State reads may reflect unflushed mutations; an idempotent acknowledgment proves durability.
+  return mutate()
+}

src/main/ssh/orcad-migration-cutover-coordinator.test.ts

@@ -0,0 +1,284 @@
+import { mkdtempSync, rmSync } from 'node:fs'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { afterEach, describe, expect, it, vi } from 'vitest'
+import { getManagedOrcadOwnerEnvironmentId } from '../../shared/managed-orcad-ssh-owner'
+import type {
+  OrcadMigrationCatalogState,
+  OrcadMigrationManifest
+} from '../../shared/orcad-migration-manifest'
+import type { SshTarget } from '../../shared/ssh-types'
+import { closeTestStores, createSqliteTestStore } from '../persistence-test-harness'
+import { Store } from '../persistence/loading-store/store'
+import {
+  abortOrcadMigrationCutover,
+  commitOrcadMigrationDestination,
+  stageOrcadMigrationDestination,
+  type OrcadMigrationCutoverContext
+} from './orcad-migration-cutover-coordinator'
+import { ORCAD_MIGRATION_DESTINATION_UNSUPPORTED } from './orcad-migration-catalog-client'
+import { listOrcadMigrationSourceCutovers } from './orcad-migration-cutover-journal'
+import { fenceOrcadMigrationSource } from './orcad-migration-source-fence'
+import { SshTargetOrcadClaims } from './ssh-target-orcad-claims'
+
+const TARGET: SshTarget = {
+  id: 'ssh-prod',
+  label: 'Production',
+  host: 'prod.example.com',
+  port: 22,
+  username: 'deploy',
+  generation: 2
+}
+
+const directories: string[] = []
+afterEach(async () => {
+  await closeTestStores()
+  for (const directory of directories.splice(0)) {
+    rmSync(directory, { recursive: true, force: true })
+  }
+})
+
+/** An in-memory destination with the T6-9 semantics: idempotent stage, receipt-keyed commit. */
+function fakeDestination() {
+  let state: 'absent' | 'staged' | 'committed' = 'absent'
+  const view = (manifest: OrcadMigrationManifest): OrcadMigrationCatalogState => {
+    const base = { migrationId: manifest.migrationId, manifestSha256: manifest.manifestSha256 }
+    if (state === 'committed') {
+      return {
+        ...base,
+        state,
+        receipt: {
+          version: 1,
+          migrationId: manifest.migrationId,
+          manifestSha256: manifest.manifestSha256,
+          source: manifest.source,
+          importedAt: '2026-10-01T00:00:00.000Z',
+          repositoryIds: [],
+          projectGroupIds: [],
+          folderWorkspaceIds: []
+        }
+      }
+    }
+    return state === 'staged'
+      ? { ...base, state, stagedAt: '2026-10-01T00:00:00.000Z', snapshotUploads: [] }
+      : { ...base, state }
+  }
+  const destination = {
+    commits: 0,
+    readState: vi.fn(async (manifest: OrcadMigrationManifest) => view(manifest)),
+    stage: vi.fn(async (manifest: OrcadMigrationManifest) => {
+      if (state === 'absent') {
+        state = 'staged'
+      }
+      return view(manifest)
+    }),
+    commit: vi.fn(async (manifest: OrcadMigrationManifest) => {
+      if (state === 'staged') {
+        state = 'committed'
+        destination.commits += 1
+      }
+      return view(manifest)
+    }),
+    abort: vi.fn(async (manifest: OrcadMigrationManifest) => {
+      const aborted = state === 'staged'
+      if (aborted) {
+        state = 'absent'
+      }
+      return { ...view(manifest), aborted, ...(aborted ? {} : { durableAbsent: true as const }) }
+    }),
+    stageChunk: vi.fn()
+  }
+  return destination
+}
+
+async function setup() {
+  const userDataPath = mkdtempSync(join(tmpdir(), 'orcad-cutover-coordinator-'))
+  directories.push(userDataPath)
+  const store = createSqliteTestStore(Store, { dataFile: join(userDataPath, 'orca-data.json') })
+  store.addSshTarget(TARGET)
+  store.addRepo({
+    id: 'repo-1',
+    path: '/srv/app',
+    displayName: 'App',
+    badgeColor: '#737373',
+    addedAt: 1,
+    kind: 'git',
+    connectionId: TARGET.id
+  })
+  const claims = new SshTargetOrcadClaims(store)
+  const fenced = await fenceOrcadMigrationSource({
+    userDataPath,
+    store,
+    claims,
+    targetId: TARGET.id,
+    destinationEnvironmentId: 'env-1',
+    destinationName: 'Managed',
+    terminalProof: { verdict: 'exited', provenPtyIds: [] },
+    hasDirectSshAuthority: () => false
+  })
+  if (fenced.outcome !== 'fenced') {
+    throw new Error('expected a fence')
+  }
+  const destination = fakeDestination()
+  const context: OrcadMigrationCutoverContext = { userDataPath, store, claims, destination }
+  const journal = () => listOrcadMigrationSourceCutovers(userDataPath)[0]
+  const owner = () => getManagedOrcadOwnerEnvironmentId(store.getSshTarget(TARGET.id)?.owner)
+  return { context, destination, store, migrationId: fenced.cutover.migrationId, journal, owner }
+}
+
+describe('migration cutover coordinator', () => {
+  it('stages then commits once, journaling each phase the destination proves', async () => {
+    const h = await setup()
+    await expect(stageOrcadMigrationDestination(h.context, h.migrationId)).resolves.toMatchObject({
+      phase: 'destination-staged'
+    })
+    expect(h.journal()?.phase).toBe('destination-staged')
+    await expect(commitOrcadMigrationDestination(h.context, h.migrationId)).resolves.toMatchObject({
+      phase: 'destination-committed'
+    })
+    expect(h.journal()?.phase).toBe('destination-committed')
+    await commitOrcadMigrationDestination(h.context, h.migrationId)
+    expect(h.destination.commits).toBe(1)
+  })
+
+  it('reads a lost commit reply back from the destination instead of failing or retrying blind', async () => {
+    const h = await setup()
+    await stageOrcadMigrationDestination(h.context, h.migrationId)
+    const commit = h.destination.commit.getMockImplementation()!
+    h.destination.commit.mockImplementationOnce(async (manifest) => {
+      await commit(manifest)
+      throw new Error('socket closed')
+    })
+    await expect(commitOrcadMigrationDestination(h.context, h.migrationId)).resolves.toMatchObject({
+      phase: 'destination-committed'
+    })
+    expect(h.destination.commits).toBe(1)
+  })
+
+  it('keeps the journal at staged when the commit reply and the re-read are both lost', async () => {
+    const h = await setup()
+    await stageOrcadMigrationDestination(h.context, h.migrationId)
+    h.destination.commit.mockRejectedValueOnce(new Error('socket closed'))
+    h.destination.readState.mockRejectedValueOnce(new Error('socket closed'))
+    await expect(commitOrcadMigrationDestination(h.context, h.migrationId)).rejects.toThrow(
+      'socket closed'
+    )
+    expect(h.journal()?.phase).toBe('destination-staged')
+    expect(h.owner()).toBe('env-1')
+  })
+
+  it.each([
+    [
+      'a source row changed',
+      (store: Store) => store.updateRepo('repo-1', { displayName: 'Renamed' }),
+      'orcad_migration_source_changed'
+    ],
+    [
+      'a terminal started on the source',
+      (store: Store) =>
+        store.upsertSshRemotePtyLease({ targetId: TARGET.id, ptyId: 'late', state: 'attached' }),
+      'orcad_migration_source_terminals_live'
+    ]
+  ])('refuses to stage or commit after %s', async (_label, change, code) => {
+    const h = await setup()
+    change(h.store)
+    await expect(stageOrcadMigrationDestination(h.context, h.migrationId)).rejects.toThrow(code)
+    expect(h.destination.stage).not.toHaveBeenCalled()
+  })
+
+  it('refuses a commit whose source changed after staging', async () => {
+    const h = await setup()
+    await stageOrcadMigrationDestination(h.context, h.migrationId)
+    h.store.updateRepo('repo-1', { displayName: 'Renamed' })
+    await expect(commitOrcadMigrationDestination(h.context, h.migrationId)).rejects.toThrow(
+      'orcad_migration_source_changed'
+    )
+    expect(h.destination.commit).not.toHaveBeenCalled()
+  })
+
+  it('aborts a stage and releases the fence only once the destination proves it absent', async () => {
+    const h = await setup()
+    await stageOrcadMigrationDestination(h.context, h.migrationId)
+    await expect(abortOrcadMigrationCutover(h.context, h.migrationId)).resolves.toEqual({
+      outcome: 'released',
+      evidence: 'catalog-absent'
+    })
+    expect(h.owner()).toBeNull()
+    expect(h.journal()).toBeUndefined()
+  })
+
+  it('keeps the fence when the abort answer is lost and the destination cannot be read', async () => {
+    const h = await setup()
+    await stageOrcadMigrationDestination(h.context, h.migrationId)
+    h.destination.abort.mockRejectedValueOnce(new Error('socket closed'))
+    h.destination.readState.mockResolvedValueOnce({
+      state: 'absent',
+      migrationId: h.migrationId,
+      manifestSha256: h.journal()!.manifestSha256
+    })
+    h.destination.readState.mockRejectedValueOnce(new Error('socket closed'))
+    h.destination.abort.mockRejectedValueOnce(new Error('socket closed'))
+    await expect(abortOrcadMigrationCutover(h.context, h.migrationId)).rejects.toThrow(
+      'socket closed'
+    )
+    expect(h.owner()).toBe('env-1')
+    expect(h.journal()?.phase).toBe('destination-staged')
+  })
+
+  it('never releases a committed destination, even when the journal lags behind it', async () => {
+    const h = await setup()
+    await stageOrcadMigrationDestination(h.context, h.migrationId)
+    await h.destination.commit(h.journal()!.manifest)
+    await expect(abortOrcadMigrationCutover(h.context, h.migrationId)).resolves.toMatchObject({
+      outcome: 'refused',
+      code: 'orcad_migration_committed_source_cannot_be_released'
+    })
+    expect(h.journal()?.phase).toBe('destination-committed')
+    expect(h.owner()).toBe('env-1')
+    await expect(abortOrcadMigrationCutover(h.context, h.migrationId)).resolves.toMatchObject({
+      outcome: 'refused'
+    })
+  })
+
+  it('releases on an unsupported destination only before anything was staged', async () => {
+    const unsupported = new Error(ORCAD_MIGRATION_DESTINATION_UNSUPPORTED)
+    const fresh = await setup()
+    fresh.destination.readState.mockRejectedValueOnce(unsupported)
+    await expect(abortOrcadMigrationCutover(fresh.context, fresh.migrationId)).resolves.toEqual({
+      outcome: 'released',
+      evidence: 'destination-unsupported'
+    })
+    expect(fresh.owner()).toBeNull()
+
+    const staged = await setup()
+    await stageOrcadMigrationDestination(staged.context, staged.migrationId)
+    staged.destination.readState.mockRejectedValueOnce(unsupported)
+    await expect(abortOrcadMigrationCutover(staged.context, staged.migrationId)).rejects.toThrow(
+      ORCAD_MIGRATION_DESTINATION_UNSUPPORTED
+    )
+    expect(staged.owner()).toBe('env-1')
+  })
+
+  it('treats any other read failure at abort as unverifiable and keeps the fence', async () => {
+    const h = await setup()
+    h.destination.readState.mockRejectedValueOnce(new Error('socket closed'))
+    await expect(abortOrcadMigrationCutover(h.context, h.migrationId)).rejects.toThrow(
+      'socket closed'
+    )
+    expect(h.owner()).toBe('env-1')
+  })
+
+  it('refuses a destination answer for another manifest', async () => {
+    const h = await setup()
+    h.destination.stage.mockResolvedValueOnce({
+      state: 'staged',
+      migrationId: h.migrationId,
+      manifestSha256: 'f'.repeat(64),
+      stagedAt: '2026-10-01T00:00:00.000Z'
+    })
+    await expect(stageOrcadMigrationDestination(h.context, h.migrationId)).rejects.toThrow(
+      'orcad_migration_destination_state_mismatch'
+    )
+    expect(h.journal()?.phase).toBe('source-fenced')
+  })
+})