mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-10 12:37:55 +01:00
Add a versioned online workspace migration command (#19882)
## Thinking Path > - Paperclip preserves the files and recovery state produced by its agents. > - Canonical online migration can preserve selected local workspaces while unrelated work continues. > - Operators sometimes need a newer migration owner before the serving web process can take that release. > - The existing packaged maintenance entry requires stopped writers. > - This pull request adds a finite online entry from one coherent reviewed release tree. > - The entry delegates capture and publication to the existing owners and keeps tool and serving identities separate. ## Linked Issues or Issue Description Refs #19854, #19855 and #19872. The duplicate search found related owner fixes and no separate versioned online entry. **What happened?** The finite online workspace importer was available as a library in the serving image. The packaged all-phase migration entry required maintenance mode. **Expected behavior** Run the reviewed online owner from a separate complete release tree without replacing the web process, stopping unrelated work or changing its release policy. **Deployment mode** A managed instance with its original local volume, database, existing encryption key and shared S3 storage. ## What Changed - Add strict inspect, single, cohort and exact single-journal resume requests with explicit company, workspace, deadline and concurrency bounds. - Verify the private tool root and its own build stamp separately from the serving source commit and deployment. - Reuse the nonmutating full migration-manifest inspector and require the additive controller host-identity column before publication. - Hydrate existing configuration and key bindings, then use the original ESM database pool and canonical storage, source locks and migration owners. - Keep multi-member inspection with the optional canonical cohort preflight in a read-only transaction with denied storage. Releases without that owner refuse cohort operations; single import and resume remain available. - Register the optional runtime bundle entry and document the supported full-tree source mode, resource admission and reconciliation. - Add hash-bound old-writer/reader compatibility and encrypted restore tests. ## Verification Final local revision: `c9345b72f27aee1158047dccd0b1060f9a0be6df`, on master `fbe689938c3bf8c5849d2cd56b9da90029db9389`, macOS arm64 with Node 24.22.2. - Entry and runtime packaging: **34/34 PASS**. - Real PostgreSQL/filesystem compatibility: **6/6 PASS**, disposable database wrapper exit 0 and cleanup complete. The frozen old source lock owner contends with the new entry; unrelated work proceeds. The old authority and snapshot readers decrypt and restore newly published bytes. - Pending and uploaded journal cases use the unchanged old persisted format and preserve operation/snapshot IDs through resume. - Full layered server source and test typecheck: **PASS**. - Module boundaries and `git diff --check`: **PASS**. - Actual source-mode module import under Node 24 succeeds without starting the application. - A separate temporary full-old-producer fixture did not get past module setup: 90 seconds under Vitest and a 15-second direct Node control. Neither is counted as an executed old-producer test or a product failure. The smaller frozen old-writer/reader proof and old-format journal cases are the local evidence. - Hosted resume of the actual saved operation and Linux artifact qualification remain operational acceptance gates. This PR does not claim those ran. - Review regression: a real successful import returned a null operation ID (genuine RED). The repaired entry binds the exact returned snapshot to its succeeded migration journal after commit (GREEN). Committed or superseded snapshots are accepted; a missing readback requires reconciliation rather than a blind retry. - Final affected check: **40/40 PASS** across the entry, packaging and actual PostgreSQL compatibility suites; wrapper exit 0 and cleanup complete. Full layered typecheck and boundaries pass after the user-requested rebase and current Runner rebuild. - A second real postcommit database-read fault also has genuine RED/GREEN proof. The succeeded journal, saved authority and original bytes remain intact, and every readback error now preserves the reconcile-required warning. - Final-head [CI run 42404116001](https://github.com/paperclipai/paperclip/actions/runs/42404116001): **PASS**, with all 59 check contexts terminal (55 success and 4 intentional skips). Fresh Greptile review is **5/5** on `c9345b72f27aee1158047dccd0b1060f9a0be6df`, with no unresolved threads; check and thread pagination are complete. ## Risks - The tool must come from one authenticated complete release tree with its own dependencies. A copied bundle chunk can retain old absolute paths and is not a portable release. - Concurrency is not resource admission. The operator must check memory, CPU and scratch headroom and start with one import or two cohort members. - Schema must be prepared by the existing schema owner. This entry never applies DDL. - Cancellation awaits the canonical owner. An SSH interruption does not prove the process stopped; durable journals must be reconciled before retry. - The entry preserves remote obligations and source files. It does not prove enrollment readiness or zero-downtime operation. ## Model Used OpenAI GPT-6 through Codex, with reasoning, repository tools and code execution. The exact serving snapshot and context-window size are not exposed to this agent. ## 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 - [x] All Paperclip CI gates are green - [x] 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 <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed
1 parent
576d6f6430
commit
5a70999306
8 files changed
+800
-4
No files matched your search
@@ -0,0 +1,121 @@
|
||||
# Versioned online workspace migration entry
|
||||
|
||||
A finite operator can preserve local workspaces with the canonical online
|
||||
migration owners while the web process keeps its current release. It does not
|
||||
start the app, a listener, a scheduler, a provider run or the offline migration
|
||||
service. The web remains on its normal release policy.
|
||||
|
||||
## Release and data identities
|
||||
|
||||
Use one reviewed normal Core release tree, including its own `server`, workspace
|
||||
packages, `node_modules`, Node 24 runtime and `tsx` loader. Verify the artifact's
|
||||
release provenance and extracted file hashes before execution. Keep that tree
|
||||
under a separate private tool root. Do not replace `/app`, mix old server modules
|
||||
with new modules, install over the serving tree, or copy individual patched owner
|
||||
files. Core's canonical storage registry reads the stored instance configuration;
|
||||
this entry does not load a Cloud distribution overlay.
|
||||
|
||||
The request names both the **tool commit** and the **serving source commit and
|
||||
Railway deployment**. They can differ. The tool reads its own release marker or
|
||||
`server/dist/build-info.json`, ignoring an inherited `PAPERCLIP_BUILD_COMMIT` for
|
||||
that check. It checks the current deployment and loopback health separately.
|
||||
Run as the existing application user with the original data-home, database and
|
||||
key bindings. Credentials remain in that process and are never printed.
|
||||
|
||||
The supported private-root mode is the unbundled source entry with the same
|
||||
release's loader and module tree. The compiled entry is also usable when its
|
||||
complete compiled dependency tree is present. Normal builds register
|
||||
`online-workspace-migration.mjs` beside the other runtime entries, but that file
|
||||
is not a self-contained portable artifact: current bundling preserves original
|
||||
module locations and external dependencies. Do not move only its chunks to a
|
||||
new root and treat them as a separate release.
|
||||
|
||||
## Request and command
|
||||
|
||||
Write one regular, mode 0600 JSON request, at most 64 KiB:
|
||||
|
||||
```json
|
||||
{
|
||||
"schema": "paperclip.online-workspace-migration.v1",
|
||||
"toolCommit": "<reviewed full Core commit>",
|
||||
"toolRoot": "/private/operator/core-release",
|
||||
"source": {
|
||||
"commit": "<actual serving full Core commit>",
|
||||
"deploymentId": "<actual Railway deployment UUID>"
|
||||
},
|
||||
"companyId": "<exact company UUID>",
|
||||
"workspaceIds": ["<exact workspace UUID>"],
|
||||
"mode": "inspect",
|
||||
"deadlineMs": 720000,
|
||||
"concurrency": 1
|
||||
}
|
||||
```
|
||||
|
||||
From the tool root, use its own Node 24 and loader:
|
||||
|
||||
```sh
|
||||
PAPERCLIP_ONLINE_WORKSPACE_MIGRATION=true \
|
||||
./node/bin/node --max-old-space-size=2048 --import ./cli/node_modules/tsx/dist/loader.mjs \
|
||||
./server/src/maintenance/online-workspace-migration.ts /private/operator/request.json
|
||||
```
|
||||
|
||||
`inspect` with one member performs bounded read-only canonical inventory. With
|
||||
multiple members it requires the actual cohort API and calls its `apply:false`
|
||||
preflight inside a read-only transaction with storage access forbidden. It does
|
||||
not combine singleton eligibility into cohort proof. The observation holds no
|
||||
source fence and grants no later admission. `single` invokes one canonical
|
||||
importer. `cohort` invokes the existing cohort owner once for the exact
|
||||
full sorted set, one through eight unique workspaces, with the requested bounded
|
||||
concurrency. A release without that owner refuses cohort apply and multi-member inspection.
|
||||
Single inspection, single import and exact single-journal resume remain available. A cohort preflight
|
||||
identifier is not an apply token; its durable journal remains the canonical
|
||||
resume record.
|
||||
|
||||
`resume` requires exactly one workspace plus exact `operationId` and `snapshotId`.
|
||||
It takes the existing complete source-owner scope, checks the saved running,
|
||||
unfinished journal and pending/uploaded snapshot under that scope, and calls the
|
||||
normal importer with the original `createDb` pool. That importer joins the live
|
||||
source scope and retains its full lifecycle, source, finalizer, physical-file and
|
||||
publication checks. No caller resets the journal or supplies a replacement ID.
|
||||
A finished, foreign or changed journal refuses for explicit reconciliation.
|
||||
Cohort retries must use the same full workspace set and the existing cohort owner;
|
||||
a singleton cannot resume a cohort journal.
|
||||
|
||||
## Schema, cancellation and proof
|
||||
|
||||
Before key hydration or any publication, the entry calls the existing
|
||||
nonmutating `inspectMigrations` owner against the complete tool migration
|
||||
manifest. Pending schema refuses. It independently requires the nullable text
|
||||
`native_run_finalizations.controller_host_hash` column. Schema application belongs
|
||||
to the existing schema-prestage owner; this entry never applies DDL.
|
||||
|
||||
Stored runtime configuration and the existing encrypted key are read through
|
||||
their normal owners. Shared S3 storage is required. Canonical migration owns all
|
||||
writes, snapshots, encryption, authenticated restore, locks and recovery recipes.
|
||||
The entry does not assert stopped writers, terminate an agent, clear a hold or
|
||||
change remote authority. An unrelated active run is not a global-idle blocker.
|
||||
An actual scoped writer, foreign alias or unsupported source still refuses.
|
||||
|
||||
Before each apply, use the existing per-operation memory, CPU and temporary-disk
|
||||
resource preflight. A concurrency bound is not resource admission. Budget every
|
||||
logical capture and up to two nested-child verifications. Qualify one single
|
||||
import and a two-member cohort before considering eight, using actual RSS,
|
||||
scratch headroom and elapsed-time evidence. This entry makes no implicit resource
|
||||
safety claim.
|
||||
|
||||
The deadline is at most 720 seconds. Shutdown aborts the active operation and
|
||||
awaits the canonical owner before closing only this database's process-local
|
||||
clients. It does not promise instant cancellation of a dependency that ignores
|
||||
its signal. An interrupted SSH session does not prove that an operator process
|
||||
stopped; reconcile the durable journals before retrying.
|
||||
|
||||
Output includes the separate identities, canonical authority IDs, source digests,
|
||||
outcome and process peak RSS. It excludes repository URLs, local file paths,
|
||||
keys and database URLs. Completed operation IDs come from the exact returned
|
||||
snapshot's succeeded migration journal, including a legitimately superseded
|
||||
snapshot. This readback runs after publication commits. A readback failure
|
||||
reports reconciliation required; it does not prove that publication failed or
|
||||
authorize a blind retry. A successful import is preparatory preservation, not an
|
||||
ephemeral-enrollment or zero-downtime acceptance receipt. Operational proof still
|
||||
requires the reviewed release artifact, fresh source/volume witnesses, resource
|
||||
headroom and actual hosted readback.
|
||||
@@ -72,13 +72,14 @@ function externalizeThirdParty(isBundled, runtimeDir) {
|
||||
}
|
||||
|
||||
/** @param {{ entry: string, outdir: string, maintenanceEntry?: string, ephemeralMigrationEntry?: string,
|
||||
* nativeHarnessPrestageEntry?: string, isBundled?: (specifier: string) => boolean }} options */
|
||||
export async function bundleRuntime({ entry, outdir, maintenanceEntry, ephemeralMigrationEntry, nativeHarnessPrestageEntry, isBundled = (specifier) => specifier.startsWith("@paperclipai/") }) {
|
||||
* nativeHarnessPrestageEntry?: string, onlineWorkspaceMigrationEntry?: string, isBundled?: (specifier: string) => boolean }} options */
|
||||
export async function bundleRuntime({ entry, outdir, maintenanceEntry, ephemeralMigrationEntry, nativeHarnessPrestageEntry, onlineWorkspaceMigrationEntry, isBundled = (specifier) => specifier.startsWith("@paperclipai/") }) {
|
||||
await rm(outdir, { recursive: true, force: true });
|
||||
const result = await build({
|
||||
entryPoints: { index: "paperclip-runtime-entry", server: entry, ...(maintenanceEntry ? { "agent-files": maintenanceEntry } : {}),
|
||||
...(ephemeralMigrationEntry ? { "ephemeral-migration": ephemeralMigrationEntry } : {}),
|
||||
...(nativeHarnessPrestageEntry ? { "native-harness-prestage": nativeHarnessPrestageEntry } : {}) },
|
||||
...(nativeHarnessPrestageEntry ? { "native-harness-prestage": nativeHarnessPrestageEntry } : {}),
|
||||
...(onlineWorkspaceMigrationEntry ? { "online-workspace-migration": onlineWorkspaceMigrationEntry } : {}) },
|
||||
outdir,
|
||||
outExtension: { ".js": ".mjs" },
|
||||
chunkNames: "chunk-[hash]",
|
||||
@@ -109,7 +110,8 @@ async function main() {
|
||||
const summary = await bundleRuntime({ entry: join(serverDir, "dist", "index.js"), outdir: join(serverDir, "dist-runtime"),
|
||||
maintenanceEntry: join(serverDir, "dist", "maintenance", "agent-files.js"),
|
||||
ephemeralMigrationEntry: join(serverDir, "dist", "maintenance", "ephemeral-migration.js"),
|
||||
nativeHarnessPrestageEntry: join(serverDir, "dist", "maintenance", "native-harness-prestage.js") });
|
||||
nativeHarnessPrestageEntry: join(serverDir, "dist", "maintenance", "native-harness-prestage.js"),
|
||||
onlineWorkspaceMigrationEntry: join(serverDir, "dist", "maintenance", "online-workspace-migration.js") });
|
||||
console.log(`[bundle-runtime] ${summary.inputs} inputs -> ${summary.chunks} chunks in ${Math.round(performance.now() - started)} ms: ${summary.outfile}`);
|
||||
}
|
||||
|
||||
|
||||
@@ -98,6 +98,15 @@ describe("bundleRuntime", () => {
|
||||
expect(JSON.parse(stdout)).toEqual({ copied: 42, enrollmentReady: false });
|
||||
});
|
||||
|
||||
it("packages the finite online workspace entry without starting the app", async () => {
|
||||
const onlineWorkspaceMigrationEntry = path.join(root, "server", "dist", "online.js");
|
||||
write(onlineWorkspaceMigrationEntry, 'import { values } from "@paperclipai/fixture"; console.log(JSON.stringify({ migrated: values.doubled, enrollmentReady: false }));');
|
||||
const outdir = path.join(root, "server", "dist-runtime");
|
||||
await bundleRuntime({ entry: path.join(root, "server", "dist", "index.js"), outdir, onlineWorkspaceMigrationEntry });
|
||||
const { stdout } = await execFileAsync(process.execPath, [path.join(outdir, "online-workspace-migration.mjs")], { timeout: 30_000 });
|
||||
expect(JSON.parse(stdout)).toEqual({ migrated: 42, enrollmentReady: false });
|
||||
});
|
||||
|
||||
it("bundles workspace TypeScript and runs with plain node", async () => {
|
||||
const outdir = path.join(root, "server", "dist-runtime");
|
||||
const summary = await bundleRuntime({ entry: path.join(root, "server", "dist", "index.js"), outdir });
|
||||
|
||||
@@ -0,0 +1,8 @@
|
||||
import{AsyncLocalStorage as Qe}from"node:async_hooks";import{withDedicatedDbConnection as et}from"@paperclipai/db";import{sql as ce}from"drizzle-orm";import nr from"node:path";import Wt from"node:path";import{expandHomePrefix as zt,resolveDefaultBackupDir as Gt,resolveDefaultEmbeddedPostgresDir as Ft,resolveDefaultLogsDir as Vt,resolveDefaultSecretsKeyFilePath as Ye,resolveDefaultStorageDir as Ht,resolveHomeAwarePath as Kt,resolvePaperclipConfigPathForInstance as qt,resolvePaperclipHomeDir as Ut,resolvePaperclipInstanceId as Jt,resolvePaperclipInstanceRoot as Yt}from"@paperclipai/shared/home-paths";function Te(){return Ye()}import{setTimeout as Ze}from"node:timers/promises";import{sql as Xe}from"drizzle-orm";async function Ce(e,t,u){if(!Number.isFinite(t)||t<=0)return u();let i=new AbortController,y=Math.max(1,Math.min(3e4,Math.floor(t/3))),R=(async()=>{for(;!i.signal.aborted;)await Ze(y,void 0,{signal:i.signal,ref:!1}),await e.execute(Xe`select 1`)})().catch(()=>{});try{return await u()}finally{i.abort(),await R}}var Oe=new Qe,ge=class extends Error{};function tt(){let e=Oe.getStore();return e?.active?e:void 0}async function rt(e,t,u,i=!1){let y=[...new Set([...(t.issueIds??[]).map(P=>`isolated_repository_source:${t.companyId}:${P}`),...(t.projectIds??[]).map(P=>`managed_project_source:${t.companyId}:${P}`)])].sort();if(y.length>128)throw new Error("workspace_source_scope_limit");let R=tt(),N=async(P,D)=>{for(let a of y){if(P.keys.has(a))continue;let A=D?ce`pg_try_advisory_xact_lock(hashtextextended(${a}, 0))`:ce`pg_advisory_xact_lock(hashtextextended(${a}, 0))`,L=await P.transaction.execute(ce`select ${A} as acquired`);if(D&&!L[0]?.acquired)throw new ge("workspace_source_busy");P.keys.add(a)}};if(R?.db===e){if(new Set([...R.keys,...y]).size>128)throw new Error("workspace_source_scope_limit");if(R.companyId!==t.companyId)throw new Error("workspace_source_company_mismatch");let P=y.filter(a=>!R.keys.has(a)).every(a=>a.startsWith("managed_project_source:")),D=[...R.keys].some(a=>a.startsWith("managed_project_source:"));return await N(R,!P||D),u(R.transaction)}return et(e,P=>P.transaction(async D=>{let a={db:e,companyId:t.companyId,transaction:D,keys:new Set,active:!0};try{await N(a,i);let A=await D.execute(ce`select setting::bigint as idle_timeout_ms from pg_settings where name = 'idle_in_transaction_session_timeout'`);return await Ce(D,Number(A[0]?.idle_timeout_ms),()=>Oe.run(a,()=>u(D)))}finally{a.active=!1}}))}import{z as w}from"zod";var Me=w.object({id:w.string().min(1),relativePath:w.string().min(1),repoUrl:w.string().min(1),ref:w.string().nullable(),pinnedCommit:w.string().optional(),readOnly:w.boolean().optional(),projectId:w.string().uuid().nullable().optional(),projectWorkspaceId:w.string().uuid().nullable().optional()}).strict(),at=w.discriminatedUnion("kind",[w.object({kind:w.literal("empty"),repositories:w.array(Me).optional()}).strict(),w.object({kind:w.literal("git"),repoUrl:w.string().min(1),ref:w.string().nullable(),branchName:w.string().nullable().optional(),branchMode:w.enum(["create","existing"]).optional(),repositories:w.array(Me).optional()}).strict()]),it=w.object({schema:w.literal("paperclip.remote-workspace.v1"),epoch:w.number().int().positive(),snapshotId:w.string().uuid().nullable(),source:at,location:w.object({environmentId:w.string().uuid(),leaseId:w.string().uuid(),providerLeaseId:w.string().min(1),remoteCwd:w.string().min(1)}).strict().nullable(),writerRunId:w.string().uuid().nullable(),dirty:w.boolean()}).strict(),lr=w.object({schema:w.literal("paperclip.remote-workspace-run.v1"),workspaceId:w.string().uuid(),leaseId:w.string().uuid(),providerLeaseId:w.string().min(1),remoteCwd:w.string().min(1),epoch:w.number().int().positive(),state:w.enum(["prepared","saved"]),snapshotId:w.string().uuid().nullable()}).strict();function nt(e){let t=e?.remoteWorkspace;return t==null?null:it.parse(t)}import{createCipheriv as vr,createDecipheriv as _t,createHash as qe,randomBytes as br,randomUUID as St}from"node:crypto";import{constants as xr,createReadStream as ve,createWriteStream as _e}from"node:fs";import{link as kt,lstat as Rr,mkdtemp as vt,open as bt,realpath as Pr,rm as Ke}from"node:fs/promises";import Et from"node:os";import X from"node:path";import{pipeline as Se}from"node:stream/promises";import{Transform as xt}from"node:stream";import{isDeepStrictEqual as Or}from"node:util";import{and as It,eq as ke}from"drizzle-orm";import{executionWorkspaceSnapshots as fe}from"@paperclipai/db";import{createCipheriv as st,createDecipheriv as ct,createHash as We,hkdfSync as mr,randomBytes as ye}from"node:crypto";import{chmodSync as lt,existsSync as Be,linkSync as dt,mkdirSync as pt,readFileSync as ze,statSync as we,unlinkSync as ft,writeFileSync as ut}from"node:fs";import Ge from"node:path";var he=class extends Error{status;details;constructor(t,u,i){super(u),this.status=t,this.details=i}};function G(e,t){return new he(400,e,t)}var ot={access_denied:403,throttled:429,not_found:404,conflict:409,invaliLine truncated
|
||||
`);function F(e){throw new Error(`remote_workspace_snapshot_${e}`)}var Pt=e=>Buffer.from(JSON.stringify([be,e.companyId,e.workspaceId,e.id,e.generation])),Dt=e=>It(ke(fe.id,e.snapshotId),ke(fe.companyId,e.companyId),ke(fe.executionWorkspaceId,e.workspaceId));function At(e){(!e.objectKey||!e.sha256||!e.byteSize||!e.key||!e.manifestDigest)&&F("not_uploaded");let t={schema:be,id:e.id,companyId:e.companyId,workspaceId:e.executionWorkspaceId,generation:e.generation,objectKey:e.objectKey,sha256:e.sha256,byteSize:e.byteSize,key:e.key,manifestDigest:e.manifestDigest};return Ue(t),t}function Ue(e){(e.schema!==be||!Number.isSafeInteger(e.generation)||e.generation<1||!e.objectKey.startsWith(`${e.companyId}/workspace-snapshots/${e.workspaceId}/${e.id}/`)||!/^[a-f0-9]{64}$/.test(e.sha256)||!/^[a-f0-9]{64}$/.test(e.manifestDigest)||!Number.isSafeInteger(e.byteSize)||e.byteSize<Q.length+12+16||!e.key||typeof e.key!="object")&&F("reference_invalid")}async function jt(e,t){let u=qe("sha256"),i=0;for await(let y of ve(e,{signal:t}))u.update(y),i+=y.length;return{sha256:u.digest("hex"),byteSize:i}}async function Tt(e,t){let[u]=await e.select().from(fe).where(Dt(t));return u&&["uploaded","committed","superseded"].includes(u.status)?At(u):null}async function Ct(e){Ue(e);let t=JSON.parse(await He.resolveVersion({material:e.key,externalRef:null}));return(t.schema!==Rt||t.companyId!==e.companyId||t.workspaceId!==e.workspaceId||t.id!==e.id||t.generation!==e.generation||!/^[a-f0-9]{64}$/.test(t.plaintextSha256??"")||!Number.isSafeInteger(t.plaintextBytes)||t.plaintextBytes<0)&&F("key_binding_invalid"),t}async function Ot(e){e.signal?.throwIfAborted();let t=e.reference,u=await Ct(t),i=await vt(X.join(e.temporaryParent??Et.tmpdir(),"pc-workspace-decrypt-"));try{e.signal?.throwIfAborted();let y=X.join(i,"ciphertext"),R=X.join(i,"plaintext"),N=await e.storage.getObject(t.companyId,t.objectKey,{signal:e.signal}),P=0,D=qe("sha256"),a=new xt({transform(K,Ee,te){P+=K.length,D.update(K),te(P>t.byteSize?new Error("remote_workspace_snapshot_object_size_mismatch"):null,K)}});try{await Se(N.stream,a,_e(y,{flags:"wx",mode:384}),{signal:e.signal})}finally{N.stream.destroy()}(P!==t.byteSize||D.digest("hex")!==t.sha256)&&F("object_hash_mismatch");let A=await bt(y,"r"),L=Buffer.alloc(Q.length+12),W=Buffer.alloc(16);try{await A.read(L,0,L.length,0),await A.read(W,0,W.length,t.byteSize-W.length)}finally{await A.close()}L.subarray(0,Q.length).equals(Q)||F("ciphertext_header_invalid");let V=Buffer.from(u.dataKey,"base64");V.length!==32&&F("key_invalid");let _=_t("aes-256-gcm",V,L.subarray(Q.length));_.setAAD(Pt(t)),_.setAuthTag(W);try{await Se(ve(y,{start:L.length,end:t.byteSize-W.length-1,signal:e.signal}),_,_e(R,{flags:"wx",mode:384}),{signal:e.signal})}finally{V.fill(0)}let ee=await jt(R,e.signal);(ee.sha256!==u.plaintextSha256||ee.byteSize!==u.plaintextBytes)&&F("plaintext_hash_mismatch");let H=X.join(X.dirname(e.targetPath),`.workspace-snapshot-${St()}.tmp`);try{await Se(ve(R,{signal:e.signal}),_e(H,{flags:"wx",mode:384}),{signal:e.signal}),e.signal?.throwIfAborted(),await kt(H,e.targetPath)}finally{await Ke(H,{force:!0})}}finally{await Ke(i,{recursive:!0,force:!0})}}async function Mt(e){let t=await import("node:fs/promises"),{createReadStream:u}=await import("node:fs"),i=await import("node:path"),y=await import("node:os"),R=await import("node:crypto"),N=await import("node:child_process"),P="paperclip.remote-workspace-archive.v1",D=Buffer.from(`PCWS1
|
||||
`),a=n=>{throw new Error(`remote_workspace_archive_${n}`)},A=()=>e.signal?.throwIfAborted();A();let L=async n=>{A();let o=R.createHash("sha256"),p=0;for await(let l of u(n,{signal:e.signal}))A(),o.update(l),p+=l.length;return{sha256:o.digest("hex"),byteSize:p}},W=n=>typeof n=="string"&&n.length>0&&!n.includes("\\")&&!/[\u0000-\u001f\u007f]/u.test(n)&&!i.posix.isAbsolute(n)&&n.split("/").every(o=>o!==""&&o!=="."&&o!==".."),V=(n,o)=>{if(typeof o!="string"||!o||o.includes("\\")||/[\u0000-\u001f\u007f]/u.test(o)||i.posix.isAbsolute(o))return!1;let p=i.posix.normalize(i.posix.join(i.posix.dirname(n),o));return p!==".."&&!p.startsWith("../")},_=(n,o,p)=>new Promise((l,f)=>{A();let k=N.execFile("git",["-c","core.fsmonitor=false","-c","pack.threads=1","-c","protocol.allow=never","-c",`core.hooksPath=${y.devNull}`,"-C",n,...o],{signal:e.signal,timeout:3e5,maxBuffer:64*1024*1024,env:{...process.env,GIT_CONFIG_NOSYSTEM:"1",GIT_CONFIG_GLOBAL:y.devNull,GIT_TERMINAL_PROMPT:"0",GIT_OPTIONAL_LOCKS:"0",GIT_DIR:void 0,GIT_COMMON_DIR:void 0,GIT_WORK_TREE:void 0,GIT_INDEX_FILE:void 0,GIT_OBJECT_DIRECTORY:void 0,GIT_ALTERNATE_OBJECT_DIRECTORIES:void 0,GIT_CONFIG_PARAMETERS:void 0,GIT_CONFIG_COUNT:"0",GIT_TEMPLATE_DIR:void 0,GIT_AUTHOR_NAME:"Paperclip",GIT_COMMITTER_NAME:"Paperclip",GIT_AUTHOR_EMAIL:"workspace@paperclip.invalid",GIT_COMMITTER_EMAIL:"workspace@paperclip.invalid",GIT_AUTHOR_DATE:"@0 +0000",GIT_COMMITTER_DATE:"@0 +0000",GIT_NO_LAZY_FETCH:"1"}},(x,d)=>x?f(x):l(d));p!==void 0&&k.stdin?.end(p)}),ee=n=>{if(!n||/[\u0000-\x20\u007f\\]/u.test(n))return!1;if(/^[a-z][a-z0-9+.-]*:\/\//iu.test(n)){let o;try{o=new URL(n)}catch{return!1}return["http:","https:","ssh:","git:"].includes(o.protocol)?((o.password||o.username&&o.protocol!=="ssh:"||o.search||o.hash)&&a("credential_in_git_remote"),!!o.hostname):!1}return/^[a-z]:[\/\\]/iu.test(n)?!1:/^(?:[a-z0-9._-]+@)?(?:[a-z0-9._-]+|\[[a-f0-9:]+\]):[^\s]+$/iu.test(n)},H=async n=>{let o=(r,v)=>_(n,["config","--null",r,"--no-includes","--get-regexp",v]).catch(b=>{if(b.code===1)return"";throw b}),p="^(core\\.(sparsecheckout|sparsecheckoutcone)|index\\.sparse)$",l=await _(n,["config","--local","--no-includes","--bool","--get","extensions.worktreeConfig"]).catch(r=>{if(r.code===1)return"false";throw r}),f=await o("--local","^(remote\\..*\\.(url|pushurl|fetch|promisor|partialclonefilter)|branch\\..*\\.(remote|merge|rebase)|user\\.(name|email)|extensions\\.partialclone|core\\.(sparsecheckout|sparsecheckoutcone)|index\\.sparse)$")+(l.trim()==="true"?await o("--worktree",p):""),k=r=>/^(true|yes|on|1|)$/iu.test(r)?"true":/^(false|no|off|0)$/iu.test(r)?"false":a("unsupported_git_setting"),x=f.split("\0").filter(Boolean).map(r=>{let v=r.indexOf(`
|
||||
`),b=r.slice(0,v),S=r.slice(v+1);return(v<1||/[\u0000-\x20\u007f\\]/u.test(b)||/[\u0000-\u001f\u007f\\]/u.test(S))&&a("unsupported_git_setting"),{key:b,value:S,remote:/^remote\.(.+)\.(url|pushurl|fetch|promisor|partialclonefilter)$/.exec(b),branch:/^branch\.(.+)\.(remote|merge|rebase)$/.exec(b),identity:/^user\.(name|email)$/.test(b),sparse:/^(core\.(sparsecheckout|sparsecheckoutcone)|index\.sparse)$/.test(b)}}),d=x.filter((r,v)=>!r.sparse||x.findLastIndex(b=>b.key===r.key)===v),m=new Set(d.flatMap(r=>r.remote?.[2]==="url"?[r.remote[1]]:[]));for(let r of d)r.remote&&["url","pushurl"].includes(r.remote[2])&&(ee(r.value)||m.delete(r.remote[1]));let c=new Set(d.filter(r=>r.remote?.[2]==="promisor"&&k(r.value)==="true").map(r=>r.remote[1]));[...c].some(r=>!m.has(r))&&a("unsupported_git_setting");let g=new Map;for(let r of d)if(r.branch?.[2]==="remote"){let v=r.branch[1];g.set(v,(g.get(v)??!0)&&(r.value==="."||m.has(r.value)))}let s=[];for(let r of d){if(r.remote){if(!m.has(r.remote[1]))continue;if(r.remote[2]==="promisor"&&(r.value=k(r.value)),r.remote[2]==="partialclonefilter"&&(!c.has(r.remote[1])||!/^(blob:none|blob:limit=[0-9]+[kmg]?|tree:[0-9]+)$/iu.test(r.value))&&a("unsupported_git_setting"),r.remote[2]==="fetch"){let v=r.value.replace(/^[+^]/u,"").split(":");(v.length>2||!v.some(Boolean))&&a("unsupported_git_setting");for(let b of v.filter(Boolean))b.startsWith("-")&&a("unsupported_git_setting"),await _(n,["check-ref-format","--allow-onelevel","--refspec-pattern",b]).catch(()=>a("unsupported_git_setting"))}}else if(r.branch){if(!(g.get(r.branch[1])??m.has("origin")))continue;r.branch[2]==="merge"&&(r.value.startsWith("-")&&a("unsupported_git_setting"),await _(n,["check-ref-format","--allow-onelevel",r.value]).catch(()=>a("unsupported_git_setting"))),r.branch[2]==="rebase"&&!["true","false","merges","interactive","preserve","yes","no","on","off","1","0",""].includes(r.value)&&a("unsupported_git_setting")}else r.sparse?r.value=k(r.value):r.key==="extensions.partialclone"?c.has(r.value)||a("unsupported_git_setting"):r.identity||a("unsupported_git_setting");s.push({key:r.key,value:r.value})}return s},K=async n=>{let o=await _(n,["rev-parse","--path-format=absolute","--show-toplevel"]).catch(async p=>{throw A(),await Ee(n)&&a("invalid_git_repository"),p});await t.realpath(o.trim()).catch(()=>null)!==n&&a("invalid_git_repository")},Ee=async n=>{let o=d=>t.stat(d).then(m=>m.isDirectory(),()=>!1),p=async d=>{A();let m=await t.open(d,t.constants.O_RDONLY|t.constants.O_NONBLOCK).catch(()=>null);if(!m)return null;try{if(!(await m.stat()).isFile())return null;let c=Buffer.alloc(4096),{bytesRead:g}=await m.read(c,0,c.length,0);return c.subarray(0,g).toString("utf8")}finally{await m.close()}},l=i.join(n,".git"),f=l;if((await t.lstat(l).catch(()=>null))?.isFile()){let d=/^gitdir: (.+)$/mu.exec(await p(l)??"");if(!d)return!0;f=i.resolve(n,d[1].trim())}let k=await p(i.join(f,"HEAD"));if(!/^(ref: refs\/\S+|[0-9a-f]{40}|[0-9a-f]{64})\s*$/u.test(k??""))return!0;let x=f;if(await t.lstat(i.join(f,"commondir")).then(()=>!0,()=>!1)){let d=await p(i.join(f,"commondir"));if(!d?.trim())return!0;x=i.resolve(f,d.trim())}return!await o(i.join(x,"objects"))||!await o(i.join(x,"refs"))},te=async n=>{await K(n);let o=(await _(n,["rev-parse","--absolute-git-dir"])).trim();for(let g of["rebase-merge","rebase-apply","sequencer","CHERRY_PICK_HEAD","REVERT_HEAD","MERGE_AUTOSTASH","BISECT_START"])await t.lstat(i.join(o,g)).catch(r=>{if(r.code!=="ENOENT")throw r;return null})&&a("git_operation_in_progress: finish or abort the Git operation before saving the workspace");let p=async(g,s)=>{let r=i.join(o,g),v=await t.lstat(r).catch(j=>{if(j.code!=="ENOENT")throw j;return null});if(!v)return null;(!v.isFile()||v.isSymbolicLink()||v.size>s)&&a("git_operation_in_progress: finish or abort the Git operation before saving the workspace");let b=[],S=0,E=await B(r,async j=>{S+=j.length,S>s&&a("git_operation_in_progress: finish or abort the Git operation before saving the workspace"),b.push(j)});return{name:g,sha256:E.sha256,text:Buffer.concat(b).toString("utf8")}},l=await p("MERGE_HEAD",8192);if(l){for(let S of["MERGE_RR","REBASE_HEAD"])await t.lstat(i.join(o,S)).then(()=>!0,j=>{if(j.code!=="ENOENT")throw j;return!1})&&a("git_operation_in_progress: finish or abort the Git operation before saving the workspace");await _(n,["ls-files","--unmerged","-z"])&&a("git_operation_in_progress: finish or abort the Git operation before saving the workspace");let g=await p("MERGE_MODE",16),s=await p("MERGE_MSG",4*1024*1024),r=await p("ORIG_HEAD",128),v=await p("AUTO_MERGE",128);if(!r||!s||g&&!["","no-ff"].includes(g.text.trim()))return a("git_operation_in_progress: finish or abort the Git operation before saving the workspace");let b=[...l.text.trim().split(`
|
||||
`).map(S=>({object:S,type:"commit"})),{object:r.text.trim(),type:"commit"},...v?[{object:v.text.trim(),type:"tree"}]:[]];for(let{object:S,type:E}of b)(!/^(?:[a-f0-9]{40}|[a-f0-9]{64})$/u.test(S)||await _(n,["rev-parse","--verify",`${S}^{${E}}`]).then(j=>j.trim()).catch(()=>null)!==S)&&a("git_operation_in_progress: finish or abort the Git operation before saving the workspace");return{objects:b,metadata:[["MERGE_HEAD",l],["MERGE_MODE",g],["MERGE_MSG",s],["ORIG_HEAD",r],["AUTO_MERGE",v]].map(([S,E])=>({name:S,sha256:E?.sha256??null}))}}let f=i.join(o,"REBASE_HEAD"),k=await t.lstat(f).catch(g=>{if(g.code!=="ENOENT")throw g;return null});if(!k)return null;(!k.isFile()||k.isSymbolicLink()||k.size>128)&&a("git_operation_in_progress: finish or abort the Git operation before saving the workspace");let x=[],d=0,m=await B(f,async g=>{d+=g.length,d>128&&a("git_operation_in_progress: finish or abort the Git operation before saving the workspace"),x.push(g)}),c=Buffer.concat(x).toString("utf8").trim();return(!/^(?:[a-f0-9]{40}|[a-f0-9]{64})$/u.test(c)||await _(n,["rev-parse","--verify","REBASE_HEAD^{commit}"]).then(g=>g.trim()).catch(()=>null)!==c||await _(n,["ls-files","--unmerged","-z"]))&&a("git_operation_in_progress: finish or abort the Git operation before saving the workspace"),{objects:[{object:c,type:"commit"}],metadata:[{name:"REBASE_HEAD",sha256:m.sha256}]}},B=async(n,o)=>{A();let p=await t.lstat(n,{bigint:!0}),l=await t.open(n,t.constants.O_RDONLY|t.constants.O_NOFOLLOW|t.constants.O_NONBLOCK);try{let f=await l.stat({bigint:!0});(!f.isFile()||p.dev!==f.dev||p.ino!==f.ino||await t.realpath(n)!==n)&&a("changed");let k=R.createHash("sha256"),x=0;A();for await(let c of l.createReadStream({autoClose:!1,signal:e.signal}))A(),k.update(c),x+=c.length,await o?.(c);let d=await l.stat({bigint:!0}),m=await t.lstat(n,{bigint:!0});return(x!==Number(f.size)||await t.realpath(n)!==n||["dev","ino","size","mode","mtimeNs","ctimeNs"].some(c=>f[c]!==d[c]||d[c]!==m[c]))&&a("changed"),{sha256:k.digest("hex"),size:x,mode:Number(f.mode&0o777n)}}finally{await l.close()}},re=async(n,o)=>{let p=await t.lstat(n).catch(f=>{if(f.code!=="ENOENT")throw f;return null});if(!p)return;(!p.isFile()||p.isSymbolicLink())&&a("unsafe_git_link"),await t.mkdir(i.dirname(o),{recursive:!0,mode:448});let l=await t.open(o,"w",384);try{await B(n,f=>l.writeFile(f))}finally{await l.close()}},Je=async n=>{if(!e.gitSourceRoots)return null;(!e.gitSourceRoots.length||e.gitSourceRoots.length>384||e.gitSourceRoots.some(d=>!i.isAbsolute(d)||i.resolve(d)!==d))&&a("online_source_git_requires_shared_fence");let o=async d=>{let m=await t.realpath(d),c=await t.lstat(d,{bigint:!0});return(!c.isDirectory()||c.isSymbolicLink()||!e.gitSourceRoots.some(g=>m===g||m.startsWith(`${g}${i.sep}`)))&&a("online_source_git_requires_shared_fence"),{path:m,device:String(c.dev),inode:String(c.ino)}},p=await o((await _(n,["rev-parse","--absolute-git-dir"])).trim()),l=await o((await _(n,["rev-parse","--path-format=absolute","--git-common-dir"])).trim()),f=[],k=new Set,x=async(d,m)=>{(m>4||k.size>=32)&&a("online_source_git_requires_shared_fence");let c=await o(d);if(k.has(c.path))return;k.add(c.path),await t.lstat(i.join(c.path,"info","http-alternates")).then(()=>!0,E=>{if(E.code!=="ENOENT")throw E;return!1})&&a("online_source_git_requires_shared_fence");let g=i.join(c.path,"info","alternates"),s=await t.lstat(g).catch(E=>{if(E.code!=="ENOENT")throw E;return null});if(!s){f.push(c);return}(!s.isFile()||s.isSymbolicLink()||s.size>64*1024)&&a("online_source_git_requires_shared_fence");let r=[],v=0,b=await B(g,async E=>{v+=E.length,v>64*1024&&a("online_source_git_requires_shared_fence"),r.push(E)}),S=Buffer.concat(r).toString("utf8").split(`
|
||||
`).filter(Boolean);(S.length>32||S.some(E=>/[\0\r]/.test(E)||E.startsWith('"')))&&a("online_source_git_requires_shared_fence"),f.push({...c,alternatesSha256:b.sha256});for(let E of S)await x(i.resolve(c.path,E),m+1)};return await x(i.join(l.path,"objects"),0),{gitDirectory:p,commonDirectory:l,stores:f}},ae=async(n,o)=>(await _(n,["rev-parse","--path-format=absolute","--git-path",o])).trim(),ie=async(n,o)=>{let p=await ae(n,o);return B(p).then(l=>l.sha256).catch(l=>{if(l.code!=="ENOENT")throw l;return null})},xe=async n=>{let o=await ae(n,"objects"),p=[],l=async f=>{let k=i.join(o,f),x=await t.lstat(k);(!x.isDirectory()||x.isSymbolicLink()||await t.realpath(k)!==k)&&a("unsafe_git_link");for(let d of(await t.readdir(k)).sort()){let m=f?`${f}/${d}`:d,c=await t.lstat(i.join(o,m),{bigint:!0});if(c.isSymbolicLink()&&a("unsafe_git_link"),!f&&(d==="pack"||d==="info"||/^[a-f0-9]{2}$/u.test(d))){await l(m);continue}if(f==="info"&&["alternates","http-alternates"].includes(d)&&a("unsupported_git_setting"),f==="info"&&d==="commit-graphs"&&c.isDirectory())continue;/^[a-f0-9]{2}\/[a-f0-9]{38}(?:[a-f0-9]{24})?$/u.test(m)||/^pack\/pack-[a-f0-9]{40}(?:[a-f0-9]{24})?\.(pack|idx|promisor|rev|bitmap|keep)$/u.test(m)?(c.isFile()||a("unsafe_git_link"),p.push({path:m,identity:[c.dev,c.ino,c.size,c.mode,c.mtimeNs,c.ctimeNs].map(String)})):c.isFile()||a("unsupported_git_setting")}};return await l(""),{objects:o,files:p}},ue=[],J=new Set;for(let n of e.exclude??[])W(n)||a("unsupported_filename"),J.add(n);let Ie=n=>e.exclude?.some(o=>n===o||n.startsWith(`${o}/`))||n===".paperclip-runtime"||n.startsWith(".paperclip-runtime/"),me=async(n,o=!1)=>{let p=[],l=new Set,f=!1,k,x=()=>{if(f)throw k;A()},d=async(m,c=!1)=>{let g=m?i.join(n,m):n,s=await t.lstat(g);(!s.isDirectory()||s.isSymbolicLink()||await t.realpath(g)!==g)&&a("changed");let r=(await t.readdir(g)).sort();o&&(e.requireNoGit||e.gitSourceRoots)&&(r.includes("commondir")||r.includes("gitdir")||r.includes("objects")&&r.some(b=>["HEAD","refs","packed-refs"].includes(b))||r.includes("HEAD")&&r.includes("config")||i.basename(g)==="info"&&r.some(b=>["alternates","http-alternates"].includes(b)))&&a("online_source_git_requires_shared_fence");let v=c||r.includes(".git");o&&r.includes(".git")&&!Ie(m?`${m}/.git`:".git")&&(e.requireNoGit&&a("online_source_git_requires_shared_fence"),(await t.lstat(i.join(g,".git"))).isSymbolicLink()&&a("unsafe_git_link"),await K(g));for(let b of r){x();let S=m?`${m}/${b}`:b;if(W(S)||a("unsupported_filename"),Ie(S)){J.add(S);continue}let E=i.join(n,S),j=await t.lstat(E);if(o&&b===".git"){j.isSymbolicLink()&&a("unsafe_git_link"),ue.push(m);continue}if(j.isDirectory()){let h=await t.lstat(b==="node_modules"?i.join(g,"package.json"):i.join(E,"pyvenv.cfg")).catch(()=>null);if(h?.isFile()&&!h.isSymbolicLink()&&!(v?await _(g,["ls-files","-z","--",b]):"")){J.add(S);continue}}if(j.isSocket()){J.add(S);continue}if(j.isDirectory())p.push({path:S,type:"directory",mode:j.mode&511}),await d(S,v);else if(j.isSymbolicLink()){let h=await t.readlink(E);V(S,h)||a("unsafe_symlink"),p.push({path:S,type:"symlink",target:h})}else if(j.isFile()){let h={path:S,type:"file"};p.push(h);let C=B(E).then(T=>{Object.assign(h,T)},T=>{f||(f=!0,k=T)});l.add(C),C.then(()=>l.delete(C)),l.size>=4&&(await Promise.race(l),x())}else a("unsupported_entry")}};try{return await d(""),await Promise.all(l),x(),p}finally{await Promise.all(l)}},Re=n=>R.createHash("sha256").update(JSON.stringify(n)).digest("hex");if(e.kind==="capture"){e.cwd||a("source_missing");let n=await t.lstat(e.cwd);(!n.isDirectory()||n.isSymbolicLink())&&a("source_invalid");let o=await t.realpath(e.cwd);(await t.lstat(o)).isDirectory()||a("source_invalid");let p=await t.realpath(await t.mkdtemp(i.join(y.tmpdir(),"pc-workspace-capture-"))),l=!1,f=!1;try{let k=i.join(p,"tree");await t.mkdir(k,{mode:448});let x=await me(o,!0),d=[...new Set(ue)],m=new Map,c=async(h,C)=>{let T=await Je(h),q=await te(h),Z=(await _(h,["rev-parse","--path-format=absolute","--git-path","index"])).trim(),ne=await L(Z).catch(O=>O.code==="ENOENT"?null:Promise.reject(O)),$=C??await H(h),U=await ie(h,"shallow"),oe=$.some(O=>/^remote\..*\.promisor$/u.test(O.key)&&O.value==="true");return JSON.stringify([await _(h,["status","--porcelain=v2","-z","--untracked-files=no"]),await _(h,["for-each-ref","--format=%(objectname) %(refname)"]),await _(h,["symbolic-ref","-q","HEAD"]).catch(()=>_(h,["rev-parse","HEAD"])),ne,$,U,await ie(h,"info/sparse-checkout"),U||oe?(await xe(h)).files:null,q,T])};for(let h=0;h<d.length;h++){let C=d[h],T=i.join(o,C),q=i.join(p,`repo-${h}.git`),Z=await H(T);m.set(C,await c(T,Z));let ne=await ie(T,"shallow")!==null||Z.some(I=>/^remote\..*\.promisor$/u.test(I.key)&&I.value==="true");if(ne){await _(T,["fsck","--full","--no-dangling"]);let I=(await _(T,["rev-parse","--show-object-format"])).trim();["sha1","sha256"].includes(I)||a("unsupported_git_setting"),await _(o,["init","--bare","--template=",`--object-format=${I}`,q]);let{objects:M,files:se}=a
Line truncated
|
||||
`).filter(Boolean).map(I=>{let[M,se]=I.split(" ");return`update ${se} ${M}
|
||||
`}).join("");De&&await _(O,["update-ref","--stdin"],De);let Ae=await te(T);for(let I of Ae?.metadata??[])await ie(O,I.name)!==I.sha256&&a("changed");for(let{object:I,type:M}of Ae?.objects??[])!await _(O,["cat-file","-e",`${I}^{${M}}`]).then(()=>!0).catch(()=>!1)&&!ne&&await _(O,["-c","protocol.file.allow=always","fetch","--no-tags","--no-recurse-submodules","--no-write-fetch-head","--no-auto-maintenance","--",T,I]),await _(O,["cat-file","-e",`${I}^{${M}}`]);await t.rm(i.join($,"objects","info","alternates"),{force:!0}),await _(O,["fsck","--full","--no-dangling"]),await c(T)!==m.get(C)&&a("changed")}let g=new Map(x.map(h=>[h.path,h])),s=[...x,...(await me(k)).filter(h=>!g.has(h.path))].sort((h,C)=>h.path<C.path?-1:h.path>C.path?1:0),r={schema:P,exclusionPolicy:"paperclip.remote-workspace-exclusions.v1",excluded:[...J].sort(),entries:s},v=Buffer.from(JSON.stringify(r));v.length>64*1024*1024&&a("manifest_too_large");let b=Re(r),S=Buffer.alloc(4);S.writeUInt32BE(v.length);let E=await t.open(e.archivePath,"wx",384);l=!0;try{await E.writeFile(Buffer.concat([D,S,v]));for(let h of s)if(h.type==="file"){let C=await B(i.join(g.has(h.path)?o:k,h.path),T=>E.writeFile(T));(C.size!==h.size||C.sha256!==h.sha256||C.mode!==h.mode)&&a("changed")}}finally{await E.close()}ue.length=0,JSON.stringify(x)!==JSON.stringify(await me(o,!0))&&a("changed");for(let h of d)await c(i.join(o,h))!==m.get(h)&&a("changed");e.manifestPath&&await t.writeFile(e.manifestPath,v,{flag:"wx",mode:384});let j=await L(e.archivePath);return f=!0,{schema:P,manifestDigest:b,...j,entries:s.length}}finally{l&&!f&&await t.rm(e.archivePath,{force:!0}),await t.rm(p,{recursive:!0,force:!0})}}(!e.targetDir||!/^[a-f0-9]{64}$/.test(e.expectedManifestDigest??""))&&a("restore_binding_invalid"),e.maxEntries!==void 0&&(!Number.isSafeInteger(e.maxEntries)||e.maxEntries<0)&&a("restore_binding_invalid"),e.maxManifestBytes!==void 0&&(!Number.isSafeInteger(e.maxManifestBytes)||e.maxManifestBytes<0)&&a("restore_binding_invalid");let Y=await t.open(e.archivePath,"r"),z,Pe=[];try{let n=Buffer.alloc(D.length+4);((await Y.read(n,0,n.length,0)).bytesRead!==n.length||!n.subarray(0,D.length).equals(D))&&a("invalid_header"),A();let o=n.readUInt32BE(D.length);(o>64*1024*1024||e.maxManifestBytes!==void 0&&o>e.maxManifestBytes)&&a("manifest_too_large");let p=Buffer.alloc(o);(await Y.read(p,0,o,n.length)).bytesRead!==o&&a("truncated"),A();let l=JSON.parse(p.toString("utf8"));(l.schema!==P||!Array.isArray(l.entries)||Re(l)!==e.expectedManifestDigest)&&a("manifest_mismatch"),e.maxEntries!==void 0&&l.entries.length>e.maxEntries&&a("entry_limit");let f=new Map,k=0;for(let s of l.entries){A(),(!s||!W(s.path)||f.has(s.path))&&a("invalid_entry");let r=i.posix.dirname(s.path);r!=="."&&f.get(r)!=="directory"&&a("invalid_parent"),s.type==="file"?((!Number.isSafeInteger(s.size)||s.size<0||!/^[a-f0-9]{64}$/.test(s.sha256??""))&&a("invalid_file"),k+=s.size):s.type==="symlink"?V(s.path,s.target)||a("unsafe_symlink"):s.type!=="directory"&&a("invalid_type"),s.type!=="symlink"&&(!Number.isInteger(s.mode)||s.mode<0||s.mode>511)&&a("invalid_mode"),f.set(s.path,s.type)}(!Number.isSafeInteger(k)||(await Y.stat()).size!==n.length+o+k)&&a("size_mismatch");let x=await t.realpath(i.dirname(e.targetDir)),d=await t.lstat(e.targetDir).catch(s=>{if(s.code!=="ENOENT")throw s;return null});d&&(!d.isDirectory()||d.isSymbolicLink()||(await t.readdir(e.targetDir)).length)&&a("target_not_empty"),A(),z=await t.mkdtemp(i.join(x,".workspace-restore-"));let m=n.length+o,c=Buffer.alloc(1024*1024);for(let s of l.entries){A();let r=i.join(z,s.path);if(s.type==="directory")await t.mkdir(r,{mode:448}),Pe.push(r);else if(s.type==="symlink")await t.symlink(s.target,r);else{let v=await t.open(r,"wx",384),b=R.createHash("sha256");try{let S=s.size;for(;S;){A();let E=await Y.read(c,0,Math.min(c.length,S),m);E.bytesRead||a("truncated");let j=c.subarray(0,E.bytesRead);await v.writeFile(j),b.update(j),m+=E.bytesRead,S-=E.bytesRead}b.digest("hex")!==s.sha256&&a("file_hash_mismatch")}finally{await v.close()}await t.chmod(r,s.mode)}}for(let s of l.entries.filter(r=>r.type==="directory").reverse())A(),await t.chmod(i.join(z,s.path),s.mode);let g=await L(e.archivePath);return A(),await t.rename(z,e.targetDir),z=void 0,{schema:P,manifestDigest:e.expectedManifestDigest,...g,entries:l.entries.length}}finally{if(await Y.close(),z){for(let n of Pe)await t.chmod(n,448).catch(()=>{});await t.rm(z,{recursive:!0,force:!0})}}}function Nt(e){return Mt({...e,kind:"restore"})}export{Tt as getWorkspaceSnapshot,Ot as loadWorkspaceSnapshotArchive,nt as readRemoteWorkspaceAuthority,Nt as restoreRemoteWorkspaceArchive,rt as withWorkspaceSourceLocks};
|
||||
@@ -0,0 +1,49 @@
|
||||
{
|
||||
"commit": "e2d490c61bbf12290b35af65f6e12648f5f2af75",
|
||||
"entry": "export {withWorkspaceSourceLocks} from './server/src/services/workspace-source-lock.js';export {readRemoteWorkspaceAuthority} from './packages/shared/src/remote-workspace.js';export {getWorkspaceSnapshot,loadWorkspaceSnapshotArchive} from './server/src/services/remote-workspace-snapshot-storage.js';export {restoreRemoteWorkspaceArchive} from './server/src/services/remote-workspace-snapshot-archive.js';",
|
||||
"sourceFiles": {
|
||||
"server/src/services/remote-workspace-snapshot-archive.ts": "81ef616f81804849d6feb17ae9ff4b2e39ccc78aa8784cce6203a5a8b0ab5f23",
|
||||
"server/src/services/workspace-source-lock.ts": "f01502b1432e896a922fbf0b3526d15b1a987f785afe89c3e8992d05a21c8040",
|
||||
"server/src/services/remote-workspace-snapshot-storage.ts": "0816d4343a47ced33f520589c1b5398831cb02e6b190c21786cdeca01ed27245",
|
||||
"packages/shared/src/remote-workspace.ts": "d4928b2ddd1fb1cfc767aaca474b25796ee3dd5c0d65c9005b051a1c0330dd1b",
|
||||
"server/src/home-paths.ts": "1b36cc65179959c0c4467e80360d5ed8c7fec177abcd1eeaa9c0f523d611585b",
|
||||
"server/src/services/execution-workspace-source-roots.ts": "15027f3782fa6f99c47436d89e81fc770239a55ffc1edb9b44c264ed29e33e04",
|
||||
"server/src/services/native-runtime/native-workspace-lock-keepalive.ts": "29b6b8db45d0c4e8bbbf69bc917f4f02f27bfddea259ca0459948fcd6f66964c",
|
||||
"server/src/services/workspace-git-source-roots.ts": "17810fbdb593f9c330f42f548f286ac9701780395ad908500702f1a4dc2ac79b",
|
||||
"server/src/services/remote-workspace-snapshot-retention.ts": "7cb5d4bbd5a6d20dc9ce992eddfb8a79f015fcee228c359c3d4d240c053de01c",
|
||||
"server/src/secrets/local-encrypted-provider.ts": "b582069603e39102d109c6ce72a29a753b537ceec344285dcb5b26ef33fd94e5",
|
||||
"server/src/services/delivery-work-notifications.ts": "cab5c17b4d5c52955b4eb294d5945b0ac5d721ccb260587a622973e827b73ecf",
|
||||
"server/src/services/remote-execution-termination.ts": "f23a3aa3d4607010b9d5b7bda2c463e5aea44433b2ca459ae09e308ee701378a",
|
||||
"server/src/secrets/types.ts": "3c56de8d21a8d85d9bbfe149c697036e70f01e777a6975d4fb2517a08e049955",
|
||||
"server/src/errors.ts": "b9919cbf75bb358fa8724bb44db62edc3ae32dcb53a2870ce03615a93ade5144",
|
||||
"server/src/services/environment-lease-handoff.ts": "5d4bbba4d619e1608d04861b380a48fa7f4da0a0dce83b7b14ce61ecde38d787"
|
||||
},
|
||||
"bundleSha256": "b668e53a48c6b8cd6c3ce6a8d935144fdedb74de43970607e740f3bfd45bb218",
|
||||
"compiler": {
|
||||
"name": "esbuild",
|
||||
"version": "0.28.2",
|
||||
"format": "esm",
|
||||
"platform": "node",
|
||||
"target": "node22",
|
||||
"minify": true,
|
||||
"legalComments": "none",
|
||||
"virtualModuleSideEffects": false
|
||||
},
|
||||
"externalImports": [
|
||||
"@paperclipai/db",
|
||||
"@paperclipai/shared/home-paths",
|
||||
"drizzle-orm",
|
||||
"node:async_hooks",
|
||||
"node:crypto",
|
||||
"node:fs",
|
||||
"node:fs/promises",
|
||||
"node:os",
|
||||
"node:path",
|
||||
"node:stream",
|
||||
"node:stream/promises",
|
||||
"node:timers/promises",
|
||||
"node:util",
|
||||
"zod"
|
||||
],
|
||||
"scope": "Frozen Core e2d lock and authority/snapshot read/decrypt/restore owners; real current installed database/query/Zod package boundary; no old importer or whole-server compatibility claim."
|
||||
}
|
||||
@@ -0,0 +1,188 @@
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import fs from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { pathToFileURL } from "node:url";
|
||||
import { eq, sql } from "drizzle-orm";
|
||||
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { activityLog, companies, createDb, executionWorkspaceSnapshots, executionWorkspaces,
|
||||
issues, projects, workspaceOperations, type Db } from "@paperclipai/db";
|
||||
import { resolveManagedProjectWorkspaceDir } from "../home-paths.js";
|
||||
import { onlineWorkspaceMigrationReceipt, runOnlineWorkspaceMigration, type OnlineWorkspaceMigrationRequest } from "../maintenance/online-workspace-migration.js";
|
||||
import { createLocalDiskStorageProvider } from "../storage/local-disk-provider.js";
|
||||
import { createStorageService } from "../storage/service.js";
|
||||
import type { StorageService } from "../storage/types.js";
|
||||
import { migrateRemoteWorkspace } from "../services/remote-workspace-migration.js";
|
||||
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
|
||||
|
||||
type OldProtocol = Pick<typeof import("../services/workspace-source-lock.js"), "withWorkspaceSourceLocks">
|
||||
& Pick<typeof import("@paperclipai/shared"), "readRemoteWorkspaceAuthority">
|
||||
& Pick<typeof import("../services/remote-workspace-snapshot-storage.js"), "getWorkspaceSnapshot" | "loadWorkspaceSnapshotArchive">
|
||||
& Pick<typeof import("../services/remote-workspace-snapshot-archive.js"), "restoreRemoteWorkspaceArchive">;
|
||||
const support = await getEmbeddedPostgresTestSupport();
|
||||
const oldCommit = "e2d490c61bbf12290b35af65f6e12648f5f2af75";
|
||||
function deferred() { let resolve!: () => void; const promise = new Promise<void>(ready => { resolve = ready; }); return { promise, resolve }; }
|
||||
|
||||
describe.skipIf(!support.supported)("versioned online migration compatibility with serving e2d", () => {
|
||||
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
let db: ReturnType<typeof createDb>, old: OldProtocol, home: string, storage: StorageService;
|
||||
beforeAll(async () => {
|
||||
const fixture = new URL("./fixtures/online-workspace-e2d-protocol.mjs", import.meta.url);
|
||||
const provenance = JSON.parse(await fs.readFile(new URL("./fixtures/online-workspace-e2d-protocol.provenance.json", import.meta.url), "utf8"));
|
||||
expect(provenance.commit).toBe(oldCommit);
|
||||
expect(provenance.sourceFiles["server/src/services/workspace-source-lock.ts"])
|
||||
.toBe("f01502b1432e896a922fbf0b3526d15b1a987f785afe89c3e8992d05a21c8040");
|
||||
expect(Object.keys(provenance.sourceFiles)).toHaveLength(15);
|
||||
expect(provenance.bundleSha256).toBe("b668e53a48c6b8cd6c3ce6a8d935144fdedb74de43970607e740f3bfd45bb218");
|
||||
expect(createHash("sha256").update(await fs.readFile(fixture)).digest("hex")).toBe(provenance.bundleSha256);
|
||||
old = await import(pathToFileURL(fixture.pathname).href) as OldProtocol;
|
||||
database = await startEmbeddedPostgresTestDatabase("online-migration-compatibility-");
|
||||
db = createDb(database.connectionString, { maxConnections: 1 });
|
||||
await db.execute(sql`set client_min_messages = warning`);
|
||||
}, 90_000);
|
||||
afterAll(async () => { await db?.$client.end({ timeout: 1 }); await database?.cleanup(); });
|
||||
beforeEach(async () => {
|
||||
home = await fs.realpath(await fs.mkdtemp(path.join(os.tmpdir(), "online-migration-compatibility-")));
|
||||
vi.stubEnv("PAPERCLIP_HOME", home);
|
||||
vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "61".repeat(32));
|
||||
const disk = createLocalDiskStorageProvider(path.join(home, "controlled-s3-backend"));
|
||||
storage = createStorageService({ ...disk, id: "s3" });
|
||||
});
|
||||
afterEach(async () => {
|
||||
vi.restoreAllMocks();
|
||||
await db.delete(activityLog); await db.delete(workspaceOperations); await db.delete(executionWorkspaceSnapshots);
|
||||
await db.delete(issues); await db.delete(executionWorkspaces); await db.delete(projects); await db.delete(companies);
|
||||
vi.unstubAllEnvs(); await fs.rm(home, { recursive: true, force: true });
|
||||
});
|
||||
async function fixture(companyId = randomUUID()) {
|
||||
if (!(await db.select().from(companies).where(eq(companies.id, companyId)))[0])
|
||||
await db.insert(companies).values({ id: companyId, name: "Compatibility", issuePrefix: `C${companyId.slice(0, 6)}` });
|
||||
const [project] = await db.insert(projects).values({ companyId, name: "Managed original source" }).returning();
|
||||
const cwd = resolveManagedProjectWorkspaceDir({ companyId, projectId: project.id });
|
||||
await fs.mkdir(cwd, { recursive: true }); await fs.writeFile(path.join(cwd, "saved.txt"), "original user edits\n", { mode: 0o640 });
|
||||
const [issue] = await db.insert(issues).values({ companyId, projectId: project.id, title: "Retained task" }).returning();
|
||||
const [workspace] = await db.insert(executionWorkspaces).values({ companyId, projectId: project.id, sourceIssueId: issue.id,
|
||||
name: "Managed directory", mode: "shared_workspace", strategyType: "project_primary", providerType: "local_fs", cwd, providerRef: cwd,
|
||||
metadata: { retainedHistory: "do not rewrite" } }).returning();
|
||||
await db.update(issues).set({ executionWorkspaceId: workspace.id }).where(eq(issues.id, issue.id));
|
||||
return { companyId, projectId: project.id, workspaceId: workspace.id, cwd };
|
||||
}
|
||||
function request(f: Awaited<ReturnType<typeof fixture>>, extra: Partial<OnlineWorkspaceMigrationRequest> = {}): OnlineWorkspaceMigrationRequest {
|
||||
return { schema: "paperclip.online-workspace-migration.v1", toolCommit: "1".repeat(40), toolRoot: path.join(home, "versioned-tool"),
|
||||
source: { commit: oldCommit, deploymentId: "11111111-1111-4111-8111-111111111111" }, companyId: f.companyId,
|
||||
workspaceIds: [f.workspaceId], mode: "single", deadlineMs: 60_000, concurrency: 1, ...extra };
|
||||
}
|
||||
const apply = (r: OnlineWorkspaceMigrationRequest) => runOnlineWorkspaceMigration({ db, storage, request: r, signal: AbortSignal.timeout(60_000) });
|
||||
const workspace = async (id: string) => (await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, id)))[0];
|
||||
async function oldColdRestore(f: Awaited<ReturnType<typeof fixture>>, expected: string) {
|
||||
const row = await workspace(f.workspaceId), authority = old.readRemoteWorkspaceAuthority(row.metadata)!;
|
||||
const reference = (await old.getWorkspaceSnapshot(db, { ...f, snapshotId: authority.snapshotId! }))!;
|
||||
const parked = path.join(home, "parked-original"); await fs.rename(f.cwd, parked);
|
||||
try {
|
||||
const archivePath = path.join(home, "old-reader.pcws"), targetDir = path.join(home, "old-reader-restored");
|
||||
await old.loadWorkspaceSnapshotArchive({ storage, reference, targetPath: archivePath });
|
||||
await old.restoreRemoteWorkspaceArchive({ archivePath, targetDir, expectedManifestDigest: reference.manifestDigest });
|
||||
expect(await fs.readFile(path.join(targetDir, "saved.txt"), "utf8")).toBe(expected);
|
||||
expect((await fs.stat(path.join(targetDir, "saved.txt"))).mode & 0o777).toBe(0o640);
|
||||
} finally { await fs.rename(parked, f.cwd); }
|
||||
expect((await workspace(f.workspaceId)).metadata?.retainedHistory).toBe("do not rewrite");
|
||||
}
|
||||
it("honors the actual old writer lock, admits an unrelated writer, then lets old readers cold-restore new authority", async () => {
|
||||
const f = await fixture(), other = await fixture(f.companyId), held = deferred(), release = deferred();
|
||||
const writer = old.withWorkspaceSourceLocks(db, { companyId: f.companyId, projectIds: [f.projectId] }, async () => {
|
||||
held.resolve(); await release.promise; await fs.writeFile(path.join(f.cwd, "saved.txt"), "saved by old writer\n");
|
||||
});
|
||||
await held.promise;
|
||||
try {
|
||||
await expect(apply(request(f))).rejects.toThrow("workspace_source_busy");
|
||||
expect(await db.select().from(workspaceOperations)).toEqual([]);
|
||||
expect(await db.select().from(executionWorkspaceSnapshots)).toEqual([]);
|
||||
await old.withWorkspaceSourceLocks(db, { companyId: other.companyId, projectIds: [other.projectId] }, async () => {
|
||||
await fs.writeFile(path.join(other.cwd, "unrelated.txt"), "unrelated writer admitted\n");
|
||||
}, true);
|
||||
expect(await fs.readFile(path.join(other.cwd, "unrelated.txt"), "utf8")).toBe("unrelated writer admitted\n");
|
||||
} finally { release.resolve(); await writer; }
|
||||
const result = await apply(request(f));
|
||||
expect(result).toMatchObject({ applied: true });
|
||||
const [completed] = await db.select().from(workspaceOperations).where(eq(workspaceOperations.executionWorkspaceId, f.workspaceId));
|
||||
expect(completed.status).toBe("succeeded");
|
||||
expect(onlineWorkspaceMigrationReceipt(result).members?.[0]?.operationId).toBe(completed.id);
|
||||
await oldColdRestore(f, "saved by old writer\n");
|
||||
}, 60_000);
|
||||
it("reports a committed publication requiring reconciliation when the independent receipt read fails", async () => {
|
||||
const f = await fixture(), transaction = db.transaction.bind(db);
|
||||
let completedAtFault: string | null = null;
|
||||
vi.spyOn(db, "transaction").mockImplementation((async (action, config) => {
|
||||
if (config?.accessMode === "read only") {
|
||||
const operations = await db.select().from(workspaceOperations).where(eq(workspaceOperations.executionWorkspaceId, f.workspaceId));
|
||||
const completed = operations.find(operation => operation.status === "succeeded");
|
||||
if (completed) {
|
||||
completedAtFault = completed.id;
|
||||
throw new Error("fixture_receipt_read_database_failed");
|
||||
}
|
||||
}
|
||||
return transaction(action, config);
|
||||
}) as Db["transaction"]);
|
||||
try {
|
||||
await expect(apply(request(f))).rejects.toThrow("online_migration_committed_receipt_requires_reconciliation");
|
||||
} finally { vi.restoreAllMocks(); }
|
||||
const [completed] = await db.select().from(workspaceOperations).where(eq(workspaceOperations.executionWorkspaceId, f.workspaceId));
|
||||
expect(completedAtFault).toBe(completed.id);
|
||||
expect(completed.status).toBe("succeeded");
|
||||
expect(completed.finishedAt).not.toBeNull();
|
||||
expect(old.readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)?.snapshotId).toBe(completed.metadata?.snapshotId);
|
||||
expect(await fs.readFile(path.join(f.cwd, "saved.txt"), "utf8")).toBe("original user edits\n");
|
||||
}, 60_000);
|
||||
it.each(["pending", "uploaded"])("resumes the old migration.v1 journal shape with the same operation and snapshot IDs (%s)", async state => {
|
||||
const f = await fixture(), abort = new AbortController();
|
||||
if (state === "pending") {
|
||||
const transaction = db.transaction.bind(db);
|
||||
vi.spyOn(db, "transaction").mockImplementation((async (action, config) => {
|
||||
const result = await transaction(action, config);
|
||||
if ((result as { operation?: { phase?: string } })?.operation?.phase === "remote_workspace_migration")
|
||||
abort.abort(new Error("fixture_old_format_pending_capture"));
|
||||
return result;
|
||||
}) as Db["transaction"]);
|
||||
} else {
|
||||
await db.execute(sql`create function fixture_refuse_migration_publication() returns trigger language plpgsql as $$
|
||||
begin if new.metadata->'remoteWorkspace' is not null then raise exception 'fixture_old_format_uploaded_publication'; end if; return new; end $$`);
|
||||
await db.execute(sql`create trigger fixture_refuse_migration_publication before update on execution_workspaces
|
||||
for each row execute function fixture_refuse_migration_publication()`);
|
||||
}
|
||||
try {
|
||||
await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true, signal: abort.signal })).rejects.toThrow();
|
||||
} finally {
|
||||
if (state === "uploaded") {
|
||||
await db.execute(sql`drop trigger fixture_refuse_migration_publication on execution_workspaces`);
|
||||
await db.execute(sql`drop function fixture_refuse_migration_publication()`);
|
||||
}
|
||||
}
|
||||
vi.restoreAllMocks();
|
||||
const [operation] = await db.select().from(workspaceOperations), [snapshot] = await db.select().from(executionWorkspaceSnapshots);
|
||||
expect(operation.status).toBe("running"); expect(operation.finishedAt).toBeNull();
|
||||
expect(operation.metadata?.nestedWorkspaceScope).toMatchObject({ version: 1, state: state === "pending" ? "pending" : "fixed" });
|
||||
expect(snapshot.status).toBe(state);
|
||||
const before = await workspace(f.workspaceId);
|
||||
expect(before.metadata?.remoteWorkspaceMigration).toMatchObject({ schema: "paperclip.remote-workspace-migration.v1", operationId: operation.id });
|
||||
expect(old.readRemoteWorkspaceAuthority(before.metadata)).toBeNull();
|
||||
await expect(apply(request(f, { mode: "resume", operationId: operation.id, snapshotId: snapshot.id }))).resolves.toMatchObject({ applied: true, snapshotId: snapshot.id });
|
||||
expect((await db.select().from(workspaceOperations)).map(o => o.id)).toEqual([operation.id]);
|
||||
expect((await db.select().from(executionWorkspaceSnapshots)).map(s => s.id)).toEqual([snapshot.id]);
|
||||
await oldColdRestore(f, "original user edits\n");
|
||||
}, 60_000);
|
||||
it.each(["missing", "wrong-type"])("refuses a real incompatible schema before any journal, storage or authority write (%s)", async mode => {
|
||||
const f = await fixture(), put = vi.spyOn(storage, "putFile");
|
||||
await db.execute(sql`alter table native_run_finalizations rename column controller_host_hash to fixture_original_controller_host_hash`);
|
||||
try {
|
||||
if (mode === "wrong-type") await db.execute(sql`alter table native_run_finalizations add column controller_host_hash integer`);
|
||||
await expect(apply(request(f))).rejects.toThrow("online_migration_schema_incompatible");
|
||||
expect(put).not.toHaveBeenCalled(); expect(await db.select().from(workspaceOperations)).toEqual([]);
|
||||
expect(await db.select().from(executionWorkspaceSnapshots)).toEqual([]);
|
||||
expect(old.readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)).toBeNull();
|
||||
} finally {
|
||||
if (mode === "wrong-type") await db.execute(sql`alter table native_run_finalizations drop column controller_host_hash`);
|
||||
await db.execute(sql`alter table native_run_finalizations rename column fixture_original_controller_host_hash to controller_host_hash`);
|
||||
}
|
||||
expect(await fs.readFile(path.join(f.cwd, "saved.txt"), "utf8")).toBe("original user edits\n");
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,158 @@
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import { assertOnlineMigrationManifest, assertOnlineMigrationRuntime, assertOnlineMigrationSchema, onlineWorkspaceMigrationReceipt, parseOnlineWorkspaceMigrationRequest, runOnlineWorkspaceMigration } from "./online-workspace-migration.js";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import * as database from "@paperclipai/db";
|
||||
import fs from "node:fs/promises";
|
||||
import * as buildCommit from "../build-commit.js";
|
||||
import type { StorageService } from "../storage/types.js";
|
||||
|
||||
vi.mock("../services/remote-workspace-migration.js", async importOriginal => {
|
||||
const actual = await importOriginal<Record<string, unknown>>();
|
||||
const owners = { ...actual };
|
||||
owners.migrateRemoteWorkspaceCohort = undefined;
|
||||
return owners;
|
||||
});
|
||||
|
||||
const request = () => ({
|
||||
schema: "paperclip.online-workspace-migration.v1",
|
||||
toolCommit: "a".repeat(40),
|
||||
toolRoot: "/private/reviewed-core",
|
||||
source: { commit: "b".repeat(40), deploymentId: "11111111-1111-4111-8111-111111111111" },
|
||||
companyId: "22222222-2222-4222-8222-222222222222",
|
||||
workspaceIds: ["33333333-3333-4333-8333-333333333333"],
|
||||
mode: "single",
|
||||
deadlineMs: 720_000,
|
||||
concurrency: 1,
|
||||
});
|
||||
|
||||
describe("online workspace migration request", () => {
|
||||
it("keeps the reviewed tool and serving source as separate identities", () => {
|
||||
const parsed = parseOnlineWorkspaceMigrationRequest(request());
|
||||
expect(parsed.toolCommit).not.toBe(parsed.source.commit);
|
||||
expect(parsed.mode).toBe("single");
|
||||
});
|
||||
|
||||
it.each([
|
||||
{ deadlineMs: 720_001 }, { concurrency: 9 }, { toolRoot: "/app" }, { toolRoot: "/app/override" },
|
||||
{ toolRoot: "/" }, { toolRoot: "relative" }, { toolCommit: "head" }, { writersStopped: true },
|
||||
{ mode: "resume" }, { operationId: "44444444-4444-4444-8444-444444444444" },
|
||||
])("refuses unbounded or undeclared requests %j", delta => {
|
||||
expect(() => parseOnlineWorkspaceMigrationRequest({ ...request(), ...delta })).toThrow();
|
||||
});
|
||||
|
||||
it("requires a single exact journal and snapshot for resume", () => {
|
||||
const resume = { ...request(), mode: "resume", operationId: "44444444-4444-4444-8444-444444444444",
|
||||
snapshotId: "55555555-5555-4555-8555-555555555555" };
|
||||
expect(parseOnlineWorkspaceMigrationRequest(resume).operationId).toBe(resume.operationId);
|
||||
expect(() => parseOnlineWorkspaceMigrationRequest({ ...resume, workspaceIds: [...resume.workspaceIds,
|
||||
"66666666-6666-4666-8666-666666666666"] })).toThrow("online_migration_single_workspace_required");
|
||||
expect(() => parseOnlineWorkspaceMigrationRequest({ ...resume, concurrency: 2 })).toThrow();
|
||||
});
|
||||
|
||||
it("refuses duplicate membership and bounds the exact cohort", () => {
|
||||
const cohort = { ...request(), mode: "cohort", concurrency: 2 };
|
||||
expect(() => parseOnlineWorkspaceMigrationRequest({ ...cohort, workspaceIds: [cohort.workspaceIds[0], cohort.workspaceIds[0]] })).toThrow();
|
||||
expect(() => parseOnlineWorkspaceMigrationRequest({ ...cohort, workspaceIds: Array(9).fill(cohort.workspaceIds[0]) })).toThrow();
|
||||
});
|
||||
});
|
||||
|
||||
describe("online workspace migration schema", () => {
|
||||
it.each([{ rows: [] }, { rows: [{ type: "integer", not_null: false }] }, { rows: [{ type: "text", not_null: true }] }])(
|
||||
"refuses missing or incompatible controller identity before accessing storage or owners", async ({ rows }) => {
|
||||
const execute = vi.fn().mockResolvedValue(rows);
|
||||
const storage = new Proxy({}, { get() { throw new Error("storage_must_not_be_used"); } }) as StorageService;
|
||||
await expect(runOnlineWorkspaceMigration({ db: { execute } as unknown as Db, storage,
|
||||
request: parseOnlineWorkspaceMigrationRequest(request()), signal: new AbortController().signal }))
|
||||
.rejects.toThrow("online_migration_schema_incompatible");
|
||||
expect(execute).toHaveBeenCalledTimes(1);
|
||||
},
|
||||
);
|
||||
|
||||
it("accepts the exact additive nullable text column without applying schema", async () => {
|
||||
const execute = vi.fn().mockResolvedValue([{ attname: "controller_host_hash", type: "text", not_null: false }]);
|
||||
await expect(assertOnlineMigrationSchema({ execute } as unknown as Db)).resolves.toBeUndefined();
|
||||
expect(execute).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("honors cancellation before the first query", async () => {
|
||||
const execute = vi.fn();
|
||||
const abort = new AbortController(); abort.abort(new Error("fixture_cancelled"));
|
||||
await expect(runOnlineWorkspaceMigration({ db: { execute } as unknown as Db, storage: {} as StorageService,
|
||||
request: parseOnlineWorkspaceMigrationRequest(request()), signal: abort.signal })).rejects.toThrow("fixture_cancelled");
|
||||
expect(execute).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
describe("online migration manifest and output", () => {
|
||||
it("requires the whole tool migration manifest without applying schema", async () => {
|
||||
const inspect = vi.spyOn(database, "inspectMigrations").mockResolvedValue({ status: "needsMigrations", tableCount: 1,
|
||||
availableMigrations: ["next.sql"], appliedMigrations: [], pendingMigrations: ["next.sql"], journalEntryCount: 0, reason: "pending-migrations" });
|
||||
try {
|
||||
await expect(assertOnlineMigrationManifest("postgres://fixture")).rejects.toThrow("online_migration_schema_manifest_pending");
|
||||
inspect.mockResolvedValue({ status: "upToDate", tableCount: 1, availableMigrations: [], appliedMigrations: [], journalEntryCount: 0 });
|
||||
await expect(assertOnlineMigrationManifest("postgres://fixture")).resolves.toBeUndefined();
|
||||
expect(inspect).toHaveBeenCalledTimes(2);
|
||||
} finally { inspect.mockRestore(); }
|
||||
});
|
||||
|
||||
it("reports authority IDs and a source digest without paths, credentials or repository URLs", () => {
|
||||
const receipt = onlineWorkspaceMigrationReceipt({ companyId: request().companyId, workspaceId: request().workspaceIds[0],
|
||||
snapshotId: "55555555-5555-4555-8555-555555555555", applied: true, replayed: false,
|
||||
localPath: "/private/user-files", source: { kind: "git", repoUrl: "https://private-token@example.test/repo" } });
|
||||
expect(receipt.members?.[0]?.sourceSha256).toMatch(/^[a-f0-9]{64}$/);
|
||||
expect(JSON.stringify(receipt)).not.toContain("private-token");
|
||||
expect(JSON.stringify(receipt)).not.toContain("user-files");
|
||||
const preflight = onlineWorkspaceMigrationReceipt({ companyId: request().companyId, cohortId: "not-an-apply-token",
|
||||
applied: false, eligible: true, workspaceIds: request().workspaceIds });
|
||||
expect(preflight).toMatchObject({ eligible: true, noAdmissionProof: true, workspaceIds: request().workspaceIds });
|
||||
expect(preflight.cohortId).toBeUndefined();
|
||||
expect(preflight.members).toBeUndefined();
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
describe("separate online tool identity", () => {
|
||||
it("uses its own release stamp and exact tool root while checking the old serving source separately", async () => {
|
||||
const parsed = parseOnlineWorkspaceMigrationRequest(request());
|
||||
const originalNode = process.versions.node;
|
||||
Object.defineProperty(process.versions, "node", { value: "24.22.2", configurable: true });
|
||||
vi.stubEnv("RAILWAY_DEPLOYMENT_ID", parsed.source.deploymentId);
|
||||
vi.stubEnv("PAPERCLIP_BUILD_COMMIT", parsed.source.commit);
|
||||
vi.stubEnv("PORT", "3100");
|
||||
const realpath = vi.spyOn(fs, "realpath").mockImplementation(async value => value === parsed.toolRoot
|
||||
? parsed.toolRoot : parsed.toolRoot + "/server/src/maintenance/online-workspace-migration.ts");
|
||||
const stamp = vi.spyOn(buildCommit, "readBuildCommit").mockReturnValue(parsed.toolCommit);
|
||||
const fetch = vi.fn(async (_url: string | URL | Request, _options?: RequestInit) => new Response(JSON.stringify({ status: "ok", commit: parsed.source.commit }), { status: 200 }));
|
||||
vi.stubGlobal("fetch", fetch);
|
||||
try {
|
||||
await expect(assertOnlineMigrationRuntime(parsed)).resolves.toBeUndefined();
|
||||
expect(stamp).toHaveBeenCalledWith(expect.objectContaining({ environmentCommit: null,
|
||||
buildInfoPath: parsed.toolRoot + "/server/dist/build-info.json" }));
|
||||
expect(fetch.mock.calls[0][0]).toBe("http://127.0.0.1:3100/api/health");
|
||||
stamp.mockReturnValue(parsed.source.commit);
|
||||
await expect(assertOnlineMigrationRuntime(parsed)).rejects.toThrow("online_migration_tool_commit_mismatch");
|
||||
stamp.mockReturnValue(parsed.toolCommit);
|
||||
vi.stubEnv("RAILWAY_DEPLOYMENT_ID", "77777777-7777-4777-8777-777777777777");
|
||||
await expect(assertOnlineMigrationRuntime(parsed)).rejects.toThrow("online_migration_source_deployment_changed");
|
||||
realpath.mockResolvedValue("/app/server/src/maintenance/online-workspace-migration.ts");
|
||||
await expect(assertOnlineMigrationRuntime(parsed)).rejects.toThrow("online_migration_tool_root_mismatch");
|
||||
} finally {
|
||||
Object.defineProperty(process.versions, "node", { value: originalNode, configurable: true });
|
||||
realpath.mockRestore(); stamp.mockRestore(); vi.unstubAllGlobals(); vi.unstubAllEnvs();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
describe("optional cohort capability", () => {
|
||||
it.each(["cohort", "inspect"])("refuses %s for multiple members when the reviewed artifact has no cohort owner", async mode => {
|
||||
const execute = vi.fn().mockResolvedValue([{ type: "text", not_null: false }]);
|
||||
const parsed = parseOnlineWorkspaceMigrationRequest({ ...request(), mode, concurrency: 2,
|
||||
workspaceIds: [...request().workspaceIds, "66666666-6666-4666-8666-666666666666"] });
|
||||
await expect(runOnlineWorkspaceMigration({ db: { execute } as unknown as Db,
|
||||
storage: { provider: "s3" } as StorageService, request: parsed, signal: new AbortController().signal }))
|
||||
.rejects.toThrow("online_migration_cohort_unavailable");
|
||||
expect(execute).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,261 @@
|
||||
import fs from "node:fs/promises";
|
||||
import { constants } from "node:fs";
|
||||
import path from "node:path";
|
||||
import { createHash } from "node:crypto";
|
||||
import { fileURLToPath } from "node:url";
|
||||
import { and, eq, sql } from "drizzle-orm";
|
||||
import { closeRegisteredClients, createDb, executionWorkspaces, inspectMigrations, type Db } from "@paperclipai/db";
|
||||
import { z } from "zod";
|
||||
import { readBuildCommit } from "../build-commit.js";
|
||||
import { loadCloudBootstrapConfig, loadConfig } from "../config.js";
|
||||
import { cloudRuntimeConfigService } from "../services/cloud-runtime-config.js";
|
||||
import { setCloudRuntimeConfigSnapshot } from "../services/cloud-runtime-config-snapshot.js";
|
||||
import * as migration from "../services/remote-workspace-migration.js";
|
||||
import { inspectOnlineWorkspaceSource } from "../services/online-workspace-source.js";
|
||||
import { withWorkspaceSourceLocks } from "../services/workspace-source-lock.js";
|
||||
import { getStorageService } from "../storage/index.js";
|
||||
import type { StorageService } from "../storage/types.js";
|
||||
import { initializeNativeHarnessPrestageKey } from "./native-harness-prestage-runtime.js";
|
||||
|
||||
const commit = z.string().regex(/^[a-f0-9]{40}$/);
|
||||
const requestSchema = z.object({
|
||||
schema: z.literal("paperclip.online-workspace-migration.v1"),
|
||||
toolCommit: commit,
|
||||
toolRoot: z.string().min(1).max(4096),
|
||||
source: z.object({ commit, deploymentId: z.string().uuid() }).strict(),
|
||||
companyId: z.string().uuid(),
|
||||
workspaceIds: z.array(z.string().uuid()).min(1).max(8),
|
||||
mode: z.enum(["inspect", "single", "cohort", "resume"]),
|
||||
operationId: z.string().uuid().optional(),
|
||||
snapshotId: z.string().uuid().optional(),
|
||||
deadlineMs: z.number().int().min(1000).max(720_000),
|
||||
concurrency: z.number().int().min(1).max(8),
|
||||
}).strict();
|
||||
|
||||
export type OnlineWorkspaceMigrationRequest = z.infer<typeof requestSchema>;
|
||||
|
||||
export function parseOnlineWorkspaceMigrationRequest(value: unknown): OnlineWorkspaceMigrationRequest {
|
||||
const request = requestSchema.parse(value);
|
||||
request.companyId = request.companyId.toLowerCase();
|
||||
request.source.deploymentId = request.source.deploymentId.toLowerCase();
|
||||
if (request.operationId) request.operationId = request.operationId.toLowerCase();
|
||||
if (request.snapshotId) request.snapshotId = request.snapshotId.toLowerCase();
|
||||
request.workspaceIds = request.workspaceIds.map(id => id.toLowerCase()).sort();
|
||||
if (new Set(request.workspaceIds).size !== request.workspaceIds.length) throw new Error("online_migration_duplicate_workspace");
|
||||
if (!path.isAbsolute(request.toolRoot) || path.normalize(request.toolRoot) !== request.toolRoot
|
||||
|| request.toolRoot === "/" || request.toolRoot === "/app" || request.toolRoot.startsWith("/app/")) {
|
||||
throw new Error("online_migration_private_tool_root_required");
|
||||
}
|
||||
if (["single", "resume"].includes(request.mode) && (request.workspaceIds.length !== 1 || request.concurrency !== 1)) {
|
||||
throw new Error("online_migration_single_workspace_required");
|
||||
}
|
||||
if (request.mode === "resume" ? !request.operationId || !request.snapshotId : request.operationId || request.snapshotId) {
|
||||
throw new Error("online_migration_exact_resume_identity_required");
|
||||
}
|
||||
return request;
|
||||
}
|
||||
|
||||
export async function assertOnlineMigrationManifest(url: string) {
|
||||
const state = await inspectMigrations(url);
|
||||
if (state.status !== "upToDate") throw new Error("online_migration_schema_manifest_pending");
|
||||
}
|
||||
|
||||
export async function assertOnlineMigrationSchema(db: Pick<Db, "execute">) {
|
||||
const rows = await db.execute(sql`select a.attname, format_type(a.atttypid,a.atttypmod) type,
|
||||
a.attnotnull not_null from pg_attribute a
|
||||
where a.attrelid=to_regclass('native_run_finalizations')
|
||||
and a.attname='controller_host_hash' and a.attnum>0 and not a.attisdropped`);
|
||||
if (rows.length !== 1 || rows[0]?.type !== "text" || rows[0]?.not_null !== false) {
|
||||
throw new Error("online_migration_schema_incompatible");
|
||||
}
|
||||
}
|
||||
|
||||
const readOnly = <T>(db: Db, action: (tx: Db) => Promise<T>) => db.transaction(async tx => {
|
||||
await tx.execute(sql`set transaction read only`);
|
||||
await tx.execute(sql`set local statement_timeout='8s'`);
|
||||
await tx.execute(sql`set local lock_timeout='1s'`);
|
||||
return action(tx as unknown as Db);
|
||||
}, { accessMode: "read only" });
|
||||
|
||||
async function assertResumeIdentity(db: Db, request: OnlineWorkspaceMigrationRequest) {
|
||||
const [row] = await db.execute(sql`select w.metadata#>>'{remoteWorkspaceMigration,operationId}' marker,
|
||||
w.metadata#>>'{remoteWorkspace,snapshotId}' authority,
|
||||
o.id operation_id,o.status operation_status,o.finished_at,
|
||||
s.id snapshot_id,s.status snapshot_status from execution_workspaces w
|
||||
left join workspace_operations o on o.company_id=w.company_id and o.execution_workspace_id=w.id
|
||||
and o.id=${request.operationId}::uuid and o.phase='remote_workspace_migration'
|
||||
left join execution_workspace_snapshots s on s.company_id=w.company_id and s.execution_workspace_id=w.id
|
||||
and s.source_operation_id=o.id and s.id=${request.snapshotId}::uuid
|
||||
where w.company_id=${request.companyId}::uuid and w.id=${request.workspaceIds[0]}::uuid`);
|
||||
if (!row || row.marker !== request.operationId || row.operation_id !== request.operationId
|
||||
|| row.snapshot_id !== request.snapshotId || row.authority !== null || row.operation_status !== "running"
|
||||
|| row.finished_at !== null || !["pending", "uploaded"].includes(String(row.snapshot_status))) {
|
||||
throw new Error("online_migration_resume_identity_changed");
|
||||
}
|
||||
}
|
||||
|
||||
async function completedOperationReceipts<T>(db: Db, result: T): Promise<T> {
|
||||
const receipt = result as { applied?: boolean; companyId: string; workspaceId?: string; snapshotId?: string;
|
||||
members?: Array<{ companyId: string; workspaceId: string; snapshotId: string }> };
|
||||
if (receipt.applied !== true) return result;
|
||||
const members = receipt.members ?? [receipt as { companyId: string; workspaceId: string; snapshotId: string }];
|
||||
const completed = await readOnly(db, async tx => {
|
||||
const items = [];
|
||||
for (const member of members) {
|
||||
if (member.companyId !== receipt.companyId || !member.workspaceId || !member.snapshotId) {
|
||||
throw new Error("online_migration_committed_receipt_requires_reconciliation");
|
||||
}
|
||||
const [operation] = await tx.execute(sql`select o.id from execution_workspace_snapshots s
|
||||
join workspace_operations o on o.id=s.source_operation_id and o.company_id=s.company_id
|
||||
and o.execution_workspace_id=s.execution_workspace_id
|
||||
where s.company_id=${member.companyId}::uuid and s.execution_workspace_id=${member.workspaceId}::uuid
|
||||
and s.id=${member.snapshotId}::uuid and s.status in ('committed','superseded')
|
||||
and o.phase='remote_workspace_migration' and o.status='succeeded' and o.finished_at is not null
|
||||
and o.metadata->>'snapshotId'=s.id::text and o.metadata->>'manifestDigest'=s.manifest_digest`);
|
||||
if (!operation) throw new Error("online_migration_committed_receipt_requires_reconciliation");
|
||||
items.push({ ...member, operationId: String(operation.id) });
|
||||
}
|
||||
return items;
|
||||
}).catch(error => {
|
||||
throw new Error("online_migration_committed_receipt_requires_reconciliation", { cause: error });
|
||||
});
|
||||
return (receipt.members ? { ...receipt, members: completed } : { ...receipt, operationId: completed[0].operationId }) as T;
|
||||
}
|
||||
|
||||
export async function runOnlineWorkspaceMigration(input: {
|
||||
db: Db; storage: StorageService; request: OnlineWorkspaceMigrationRequest; signal: AbortSignal;
|
||||
}) {
|
||||
const request = parseOnlineWorkspaceMigrationRequest(input.request);
|
||||
input.signal.throwIfAborted();
|
||||
await assertOnlineMigrationSchema(input.db);
|
||||
if (input.storage.provider !== "s3") throw new Error("online_migration_shared_storage_required");
|
||||
type Cohort = (args: { db: Db; storage: StorageService; companyId: string; workspaceIds: string[];
|
||||
apply: boolean; concurrency: number; signal: AbortSignal }) => Promise<unknown>;
|
||||
const cohortOwner = (migration as typeof migration & { migrateRemoteWorkspaceCohort?: Cohort }).migrateRemoteWorkspaceCohort;
|
||||
if (request.mode === "inspect" && request.workspaceIds.length > 1) {
|
||||
if (!cohortOwner) throw new Error("online_migration_cohort_unavailable");
|
||||
const storage = new Proxy({} as StorageService, { get() { throw new Error("online_migration_inspection_storage_forbidden"); } });
|
||||
return readOnly(input.db, db => cohortOwner({ db, storage, companyId: request.companyId,
|
||||
workspaceIds: request.workspaceIds, apply: false, concurrency: request.concurrency, signal: input.signal }));
|
||||
}
|
||||
if (request.mode === "inspect") {
|
||||
const members = await readOnly(input.db, async db => {
|
||||
const result = [];
|
||||
for (const workspaceId of request.workspaceIds) {
|
||||
input.signal.throwIfAborted();
|
||||
const row = await migration.inventoryRemoteWorkspaceMigration(db, { companyId: request.companyId, workspaceId });
|
||||
result.push({ workspaceId: row.workspaceId, eligible: row.eligible, alreadyMigrated: row.alreadyMigrated,
|
||||
reasons: row.reasons, operationId: row.operationId, snapshotId: row.snapshotId, source: row.source });
|
||||
}
|
||||
return result;
|
||||
});
|
||||
return { companyId: request.companyId, applied: false as const, members };
|
||||
}
|
||||
if (request.mode === "cohort") {
|
||||
if (!cohortOwner) throw new Error("online_migration_cohort_unavailable");
|
||||
const result = await cohortOwner({ db: input.db, storage: input.storage, companyId: request.companyId,
|
||||
workspaceIds: request.workspaceIds, apply: true, concurrency: request.concurrency, signal: input.signal });
|
||||
return completedOperationReceipts(input.db, result);
|
||||
}
|
||||
const apply = async () => {
|
||||
input.signal.throwIfAborted();
|
||||
if (request.mode === "resume") await readOnly(input.db, db => assertResumeIdentity(db, request));
|
||||
const result = await migration.migrateRemoteWorkspace({ db: input.db, storage: input.storage,
|
||||
companyId: request.companyId, workspaceId: request.workspaceIds[0], apply: true, signal: input.signal });
|
||||
if (request.mode === "resume" && result.snapshotId !== request.snapshotId) throw new Error("online_migration_resume_result_changed");
|
||||
return result;
|
||||
};
|
||||
if (request.mode !== "resume") return completedOperationReceipts(input.db, await apply());
|
||||
const [workspace] = await input.db.select().from(executionWorkspaces).where(and(
|
||||
eq(executionWorkspaces.companyId, request.companyId), eq(executionWorkspaces.id, request.workspaceIds[0])));
|
||||
const source = workspace && await inspectOnlineWorkspaceSource(input.db, workspace);
|
||||
if (!source) throw new Error("online_migration_resume_source_unproven");
|
||||
const result = await withWorkspaceSourceLocks(input.db, { companyId: request.companyId,
|
||||
projectIds: source.projectIds, issueIds: source.issueIds }, apply, true);
|
||||
return completedOperationReceipts(input.db, result);
|
||||
}
|
||||
|
||||
export async function assertOnlineMigrationRuntime(request: OnlineWorkspaceMigrationRequest) {
|
||||
if (Number(process.versions.node.split(".")[0]) !== 24) throw new Error("online_migration_node24_required");
|
||||
const root = await fs.realpath(request.toolRoot), entry = await fs.realpath(fileURLToPath(import.meta.url));
|
||||
if (root !== request.toolRoot || !["src", "dist"].some(tree => entry === path.join(root,
|
||||
"server", tree, "maintenance", `online-workspace-migration.${tree === "src" ? "ts" : "js"}`))) {
|
||||
throw new Error("online_migration_tool_root_mismatch");
|
||||
}
|
||||
if (readBuildCommit({ environmentCommit: null, buildCommitPath: path.join(root, ".paperclip-build-commit"),
|
||||
buildInfoPath: path.join(root, "server", "dist", "build-info.json") }) !== request.toolCommit) throw new Error("online_migration_tool_commit_mismatch");
|
||||
if (process.env.RAILWAY_DEPLOYMENT_ID !== request.source.deploymentId) throw new Error("online_migration_source_deployment_changed");
|
||||
const response = await fetch(`http://127.0.0.1:${process.env.PORT || 3100}/api/health`, { signal: AbortSignal.timeout(5000) });
|
||||
const body = response.status === 200 ? await response.json() as { status?: string; commit?: string } : null;
|
||||
if (body?.status !== "ok" || body.commit !== request.source.commit) throw new Error("online_migration_source_health_changed");
|
||||
}
|
||||
|
||||
export function onlineWorkspaceMigrationReceipt(result: unknown) {
|
||||
const row = result as Record<string, unknown>;
|
||||
const member = (value: unknown) => {
|
||||
const item = value as Record<string, unknown>;
|
||||
return { workspaceId: item.workspaceId, snapshotId: item.snapshotId, operationId: item.operationId,
|
||||
eligible: item.eligible, alreadyMigrated: item.alreadyMigrated, reasons: item.reasons,
|
||||
sourceSha256: createHash("sha256").update(JSON.stringify(item.source ?? null)).digest("hex") };
|
||||
};
|
||||
return { companyId: row.companyId, cohortId: row.applied === true ? row.cohortId : undefined,
|
||||
applied: row.applied, replayed: row.replayed, eligible: row.eligible, workspaceIds: row.workspaceIds,
|
||||
noAdmissionProof: row.applied === false,
|
||||
members: Array.isArray(row.members) ? row.members.map(member) : row.workspaceId ? [member(row)] : undefined };
|
||||
}
|
||||
|
||||
export async function runOnlineWorkspaceMigrationEntry(args = process.argv.slice(2)) {
|
||||
if (process.env.PAPERCLIP_ONLINE_WORKSPACE_MIGRATION !== "true" || args.length !== 1) {
|
||||
throw new Error("online_migration_explicit_request_required");
|
||||
}
|
||||
const file = await fs.open(args[0], constants.O_RDONLY | constants.O_NOFOLLOW | constants.O_NONBLOCK);
|
||||
let request: OnlineWorkspaceMigrationRequest;
|
||||
try {
|
||||
const stat = await file.stat();
|
||||
if (!stat.isFile() || stat.size > 65_536 || (stat.mode & 0o077)) throw new Error("online_migration_private_request_required");
|
||||
const bytes = Buffer.alloc(65_537);
|
||||
let length = 0;
|
||||
while (length < bytes.length) {
|
||||
const read = await file.read(bytes, length, bytes.length - length, length);
|
||||
if (!read.bytesRead) break;
|
||||
length += read.bytesRead;
|
||||
}
|
||||
if (length > 65_536) throw new Error("online_migration_request_too_large");
|
||||
request = parseOnlineWorkspaceMigrationRequest(JSON.parse(bytes.subarray(0, length).toString("utf8")));
|
||||
} finally { await file.close(); }
|
||||
let url: string | undefined;
|
||||
const cancellation = new AbortController();
|
||||
const stop = () => cancellation.abort(new Error("online_migration_interrupted"));
|
||||
const deadline = setTimeout(() => cancellation.abort(new Error("online_migration_deadline")), request.deadlineMs);
|
||||
process.once("SIGTERM", stop); process.once("SIGINT", stop);
|
||||
try {
|
||||
await assertOnlineMigrationRuntime(request);
|
||||
if (!process.env.DATABASE_URL) throw new Error("online_migration_explicit_database_required");
|
||||
url = loadCloudBootstrapConfig().databaseUrl;
|
||||
await assertOnlineMigrationManifest(url);
|
||||
cancellation.signal.throwIfAborted();
|
||||
const db = createDb(url);
|
||||
await assertOnlineMigrationSchema(db);
|
||||
const config = await readOnly(db, tx => cloudRuntimeConfigService(tx).inspectStored());
|
||||
if (!config) throw new Error("cloud_runtime_configuration_missing");
|
||||
setCloudRuntimeConfigSnapshot(config);
|
||||
await initializeNativeHarnessPrestageKey(loadConfig().secretsMasterKeyFilePath);
|
||||
await assertOnlineMigrationRuntime(request);
|
||||
const result = await runOnlineWorkspaceMigration({ db, storage: getStorageService(), request, signal: cancellation.signal });
|
||||
console.log(JSON.stringify({ schema: request.schema, toolCommit: request.toolCommit, source: request.source, result: onlineWorkspaceMigrationReceipt(result),
|
||||
enrollmentReady: false, allStartedWorkAwaited: true, maxRssKiB: process.resourceUsage().maxRSS }));
|
||||
return result;
|
||||
} finally {
|
||||
clearTimeout(deadline); process.off("SIGTERM", stop); process.off("SIGINT", stop);
|
||||
if (url) await closeRegisteredClients(url, { scope: "database" });
|
||||
}
|
||||
}
|
||||
|
||||
if (process.env.PAPERCLIP_ONLINE_WORKSPACE_MIGRATION === "true") {
|
||||
await runOnlineWorkspaceMigrationEntry().catch(error => {
|
||||
const reason = error instanceof Error && /^[a-z_]{1,120}(?::[a-z_,]{1,1000})?$/.test(error.message) ? error.message : "online_migration_failed";
|
||||
console.error(JSON.stringify({ schema: "paperclip.online-workspace-migration.v1", state: "blocked", reason, enrollmentReady: false,
|
||||
...(reason === "online_migration_committed_receipt_requires_reconciliation" ? { publicationMayHaveCommitted: true } : {}) }));
|
||||
process.exitCode = 1;
|
||||
});
|
||||
}
|
||||
Reference in new issue
Block a user