mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-02 16:58:52 +01:00
feat(server): verify existing-volume ephemeral enrollment
Run owner migrations in a dedicated maintenance process, verify shared bytes, and export the original signing key only after complete eligibility. Validate with real PostgreSQL/S3 and the bundled executable. Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
16787e4167
commit
dc450aa4bb
11 files changed
+544
-5
No files matched your search
@@ -362,3 +362,44 @@ protects interrupted one-time imports. Deleted-agent objects become eligible
|
||||
after the same grace. Deletion checks the current head under the agent lock and
|
||||
removes all versions. There is no automatic idle scan; operators own scheduling
|
||||
this maintenance during the bounded rollout.
|
||||
|
||||
|
||||
### Existing-volume Cloud enrollment
|
||||
|
||||
The runtime bundle also contains `node server/dist-runtime/ephemeral-migration.mjs`.
|
||||
Cloud starts this dedicated process on the original volume after confirming that
|
||||
all application deployments have stopped. It never starts app routes, queues or
|
||||
agent execution. Set `PAPERCLIP_EPHEMERAL_MIGRATION=true`, a UUID attempt ID in
|
||||
`PAPERCLIP_EPHEMERAL_MIGRATION_ID`, and a fresh 32-byte base64url token in
|
||||
`PAPERCLIP_EPHEMERAL_MIGRATION_TOKEN`. Use the original home and instance ID,
|
||||
external database, shared S3 configuration and existing stable secrets. Durable
|
||||
agent files and shared UI assets must be enabled; ephemeral-server mode stays
|
||||
false until adoption completes.
|
||||
|
||||
The process uses the existing agent-file, run-evidence and skill maintenance
|
||||
owners, follows all pages, verifies stored agent bytes and skill versions, and
|
||||
checks whole-database eligibility. Active execution, missing local bytes, local
|
||||
workspaces and external instructions block enrollment. Those conditions must be
|
||||
resolved by their owning migration paths before detaching the volume.
|
||||
|
||||
`GET /health` is a public maintenance liveness check. `GET /status` and
|
||||
`POST /export` require the attempt token as Bearer authorization. Status includes
|
||||
only the attempt, stamped Core commit, phase, counts and safe error code. Export
|
||||
is available only after all checks succeed; it also returns the original decision
|
||||
signing key for transfer into the stack vault. The process never generates a
|
||||
replacement key. Every response disables caching. Never log or retain the export
|
||||
body outside the vault. A 25-minute deadline blocks export; SIGTERM stops the
|
||||
process, with a ten-second exit deadline.
|
||||
|
||||
Cloud must verify the attempt and image commit, preserve the exact key, and
|
||||
confirm this process is stopped before detaching the volume and starting the
|
||||
normal ephemeral server. Keep tenant routing closed throughout this one-time
|
||||
maintenance. Retain the old volume for recovery evidence; it must not become
|
||||
writable authority again after shared state advances.
|
||||
|
||||
The opt-in `ephemeral-migration.test.ts` uses disposable PostgreSQL and an
|
||||
S3-compatible endpoint. Set `PAPERCLIP_TEST_POSTGRES_ADMIN_URL`,
|
||||
`PAPERCLIP_TEST_S3_ENDPOINT` and normal AWS credentials to run it. It verifies
|
||||
multi-page adoption, filesystem destruction, fresh reads, corrupt object rejection,
|
||||
and missing-byte/active-execution refusals. The default suite does not contact
|
||||
external storage.
|
||||
@@ -71,10 +71,11 @@ function externalizeThirdParty(isBundled, runtimeDir) {
|
||||
};
|
||||
}
|
||||
|
||||
export async function bundleRuntime({ entry, outdir, maintenanceEntry, isBundled = (specifier) => specifier.startsWith("@paperclipai/") }) {
|
||||
export async function bundleRuntime({ entry, outdir, maintenanceEntry, ephemeralMigrationEntry, 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 } : {}) },
|
||||
entryPoints: { index: "paperclip-runtime-entry", server: entry, ...(maintenanceEntry ? { "agent-files": maintenanceEntry } : {}),
|
||||
...(ephemeralMigrationEntry ? { "ephemeral-migration": ephemeralMigrationEntry } : {}) },
|
||||
outdir,
|
||||
outExtension: { ".js": ".mjs" },
|
||||
chunkNames: "chunk-[hash]",
|
||||
@@ -103,7 +104,8 @@ export async function bundleRuntime({ entry, outdir, maintenanceEntry, isBundled
|
||||
async function main() {
|
||||
const started = performance.now();
|
||||
const summary = await bundleRuntime({ entry: join(serverDir, "dist", "index.js"), outdir: join(serverDir, "dist-runtime"),
|
||||
maintenanceEntry: join(serverDir, "dist", "maintenance", "agent-files.js") });
|
||||
maintenanceEntry: join(serverDir, "dist", "maintenance", "agent-files.js"),
|
||||
ephemeralMigrationEntry: join(serverDir, "dist", "maintenance", "ephemeral-migration.js") });
|
||||
console.log(`[bundle-runtime] ${summary.inputs} inputs -> ${summary.chunks} chunks in ${Math.round(performance.now() - started)} ms: ${summary.outfile}`);
|
||||
}
|
||||
|
||||
|
||||
@@ -80,6 +80,15 @@ describe("bundleRuntime", () => {
|
||||
expect(JSON.parse(stdout)).toEqual({ asset: "asset beside its source", phase: "--phase=eligibility" });
|
||||
});
|
||||
|
||||
it("packages standalone enrollment without starting the app", async () => {
|
||||
const ephemeralMigrationEntry = path.join(root, "server", "dist", "migration.js");
|
||||
write(ephemeralMigrationEntry, 'import { values } from "@paperclipai/fixture"; console.log(JSON.stringify({ migrated: values.doubled }));');
|
||||
const outdir = path.join(root, "server", "dist-runtime");
|
||||
await bundleRuntime({ entry: path.join(root, "server", "dist", "index.js"), outdir, ephemeralMigrationEntry });
|
||||
const { stdout } = await execFileAsync(process.execPath, [path.join(outdir, "ephemeral-migration.mjs")], { timeout: 30_000 });
|
||||
expect(JSON.parse(stdout)).toEqual({ migrated: 42 });
|
||||
});
|
||||
|
||||
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,146 @@
|
||||
import fs from "node:fs/promises";
|
||||
import { existsSync } from "node:fs";
|
||||
import { spawn } from "node:child_process";
|
||||
import { createServer } from "node:net";
|
||||
import { fileURLToPath } from "node:url";
|
||||
import { setTimeout as delay } from "node:timers/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { randomBytes, randomUUID } from "node:crypto";
|
||||
import { S3Client, CreateBucketCommand, DeleteBucketCommand, DeleteObjectsCommand, ListObjectsV2Command, PutObjectCommand } from "@aws-sdk/client-s3";
|
||||
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { agents, agentFileRoots, companies, createDb, closeFileStorageDb, heartbeatRuns } from "@paperclipai/db";
|
||||
import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
|
||||
import { migrateEphemeralServerData } from "../maintenance/ephemeral-migration-data.js";
|
||||
import { agentFileStore } from "../services/agent-file-store.js";
|
||||
import { resolveManagedInstructionsRoot } from "../services/agent-instructions.js";
|
||||
import { materializeInstructionBytes } from "../services/agent-instruction-files.js";
|
||||
import { verifyAgentFileObjects } from "../services/agent-file-maintenance.js";
|
||||
import { getStorageService } from "../storage/index.js";
|
||||
import { resolveDecisionSigningSecret } from "../services/decision-signing.js";
|
||||
|
||||
// Explicit real-dependency qualification. Never contacts off-host storage in
|
||||
// the default test path; use a disposable S3-compatible service and PostgreSQL.
|
||||
describe.skipIf(!process.env.PAPERCLIP_TEST_S3_ENDPOINT)("existing-volume enrollment with PostgreSQL and S3", () => {
|
||||
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
let db: ReturnType<typeof createDb>, s3: S3Client;
|
||||
let home: string, companyId: string, bucket: string;
|
||||
beforeAll(async () => {
|
||||
database = await startEmbeddedPostgresTestDatabase("ephemeral-migration-"); db = createDb(database.connectionString);
|
||||
s3 = new S3Client({ endpoint: process.env.PAPERCLIP_TEST_S3_ENDPOINT, region: "us-east-1", forcePathStyle: true });
|
||||
}, 90_000);
|
||||
beforeEach(async () => {
|
||||
companyId = randomUUID(); bucket = `enrollment-${randomUUID()}`;
|
||||
home = await fs.mkdtemp(path.join(os.tmpdir(), "ephemeral-migration-"));
|
||||
await s3.send(new CreateBucketCommand({ Bucket: bucket }));
|
||||
vi.stubEnv("PAPERCLIP_HOME", home); vi.stubEnv("PAPERCLIP_DURABLE_AGENT_FILES", "true");
|
||||
vi.stubEnv("PAPERCLIP_EPHEMERAL_SERVER", "false"); vi.stubEnv("PAPERCLIP_STORAGE_PROVIDER", "s3");
|
||||
vi.stubEnv("PAPERCLIP_STORAGE_S3_BUCKET", bucket); vi.stubEnv("PAPERCLIP_STORAGE_S3_REGION", "us-east-1");
|
||||
vi.stubEnv("PAPERCLIP_STORAGE_S3_ENDPOINT", process.env.PAPERCLIP_TEST_S3_ENDPOINT!);
|
||||
vi.stubEnv("PAPERCLIP_STORAGE_S3_FORCE_PATH_STYLE", "true");
|
||||
await db.insert(companies).values({ id: companyId, name: "Enrollment fixture", issuePrefix: "MIG" });
|
||||
});
|
||||
afterEach(async () => {
|
||||
await db.delete(heartbeatRuns).where(eq(heartbeatRuns.companyId, companyId));
|
||||
await db.delete(agents).where(eq(agents.companyId, companyId));
|
||||
await db.delete(companies).where(eq(companies.id, companyId));
|
||||
let token: string | undefined;
|
||||
do {
|
||||
const page = await s3.send(new ListObjectsV2Command({ Bucket: bucket, ContinuationToken: token }));
|
||||
if (page.Contents?.length) await s3.send(new DeleteObjectsCommand({ Bucket: bucket, Delete: { Objects: page.Contents.map(row => ({ Key: row.Key! })) } }));
|
||||
token = page.NextContinuationToken;
|
||||
} while (token);
|
||||
await s3.send(new DeleteBucketCommand({ Bucket: bucket }));
|
||||
await fs.rm(home, { recursive: true, force: true }); vi.unstubAllEnvs();
|
||||
});
|
||||
afterAll(async () => {
|
||||
if (db) { await closeFileStorageDb(db, 1); await db.$client.end({ timeout: 1 }); }
|
||||
await database?.cleanup(); s3?.destroy();
|
||||
});
|
||||
const migrate = () => migrateEphemeralServerData(db, { signal: AbortSignal.timeout(120_000), phase: () => {} });
|
||||
async function agent(index = 0) {
|
||||
const row = { id: randomUUID(), companyId, name: `Migrated ${index}`, adapterConfig: {} };
|
||||
const root = resolveManagedInstructionsRoot(row);
|
||||
await materializeInstructionBytes(root, "AGENTS.md", Buffer.from(`Original ${index}`));
|
||||
await db.insert(agents).values({ ...row, adapterConfig: { instructionsBundleMode: "managed", instructionsRootPath: root } });
|
||||
return { ...row, root };
|
||||
}
|
||||
it("migrates every page, verifies object bytes, and reads after deleting the entire predecessor filesystem", async () => {
|
||||
const rows = [];
|
||||
for (let index = 0; index < 26; index++) rows.push(await agent(index));
|
||||
const binary = Buffer.alloc(2 * 1024 * 1024, 0x81);
|
||||
await materializeInstructionBytes(rows[0]!.root, "nested/data.bin", binary);
|
||||
await materializeInstructionBytes(rows[0]!.root, "empty.txt", Buffer.alloc(0));
|
||||
expect(await migrate()).toMatchObject({ eligibility: { eligible: true }, counts: { agentFiles: 26, verifiedAgentObjects: 28 } });
|
||||
await fs.rm(home, { recursive: true, force: true });
|
||||
home = await fs.mkdtemp(path.join(os.tmpdir(), "ephemeral-successor-")); vi.stubEnv("PAPERCLIP_HOME", home);
|
||||
vi.stubEnv("PAPERCLIP_EPHEMERAL_SERVER", "true"); vi.stubEnv("PAPERCLIP_DURABLE_AGENT_FILES", "false");
|
||||
// The service's normal board actor uses instance-operator authority.
|
||||
const actor = { type: "board" as const, source: "local_implicit" as const, userId: "local-board" };
|
||||
const store = agentFileStore(db);
|
||||
for (let index = 0; index < rows.length; index++) {
|
||||
expect(await store.read(companyId, rows[index]!.id, "AGENTS.md", actor)).toEqual(Buffer.from(`Original ${index}`));
|
||||
}
|
||||
expect(await store.read(companyId, rows[0]!.id, "nested/data.bin", actor)).toEqual(binary);
|
||||
const [head] = await db.select().from(agentFileRoots).where(eq(agentFileRoots.agentId, rows[0]!.id));
|
||||
const file = head!.entries.find(entry => entry.kind === "file" && entry.objectKey !== null)!;
|
||||
if (file.kind !== "file" || !file.objectKey) throw new Error("Missing object");
|
||||
await s3.send(new PutObjectCommand({ Bucket: bucket, Key: file.objectKey, Body: "corrupt" }));
|
||||
await expect(verifyAgentFileObjects(db, getStorageService(), AbortSignal.timeout(30_000))).rejects.toThrow(/agent_file_object_(size|digest)_mismatch/);
|
||||
}, 120_000);
|
||||
const entry = fileURLToPath(new URL("../../dist-runtime/ephemeral-migration.mjs", import.meta.url));
|
||||
it.skipIf(!existsSync(entry))("runs the actual bundled maintenance entry with plain Node and exits cleanly", async () => {
|
||||
await agent();
|
||||
vi.stubEnv("PAPERCLIP_DECISION_SIGNING_SECRET", "");
|
||||
const original = resolveDecisionSigningSecret();
|
||||
const listener = createServer();
|
||||
await new Promise<void>(resolve => listener.listen(0, "127.0.0.1", resolve));
|
||||
const address = listener.address();
|
||||
if (!address || typeof address === "string") throw new Error("Missing listener");
|
||||
await new Promise<void>(resolve => listener.close(() => resolve()));
|
||||
const id = randomUUID(), token = randomBytes(32).toString("base64url");
|
||||
const stamp = JSON.parse(await fs.readFile(new URL("../../dist/build-info.json", import.meta.url), "utf8"));
|
||||
const child = spawn(process.execPath, [entry], { env: { ...process.env, DATABASE_URL: database.connectionString,
|
||||
PAPERCLIP_DATABASE_LISTEN_URL: database.connectionString, PAPERCLIP_SECRETS_MASTER_KEY: randomBytes(32).toString("hex"),
|
||||
PAPERCLIP_SHARED_UI_ASSETS: "true", PAPERCLIP_EPHEMERAL_MIGRATION: "true", PAPERCLIP_EPHEMERAL_MIGRATION_ID: id,
|
||||
PAPERCLIP_EPHEMERAL_MIGRATION_TOKEN: token, PORT: String(address.port) }, stdio: ["ignore", "pipe", "pipe"] });
|
||||
let diagnostic = "";
|
||||
child.stderr.on("data", chunk => { diagnostic = (diagnostic + String(chunk)).slice(-8_192); });
|
||||
child.stdout.resume();
|
||||
const exited = new Promise<{ code: number | null; signal: NodeJS.Signals | null }>((resolve, reject) => {
|
||||
child.once("error", reject); child.once("exit", (code, signal) => resolve({ code, signal }));
|
||||
});
|
||||
const origin = `http://127.0.0.1:${address.port}`;
|
||||
const headers = { authorization: `Bearer ${token}` };
|
||||
try {
|
||||
let status: Record<string, unknown> | undefined;
|
||||
for (const until = Date.now() + 45_000; Date.now() < until;) {
|
||||
if (child.exitCode !== null) throw new Error(`Maintenance exited: ${diagnostic}`);
|
||||
status = await fetch(`${origin}/status`, { headers, signal: AbortSignal.timeout(1_000) }).then(r => r.json() as Promise<Record<string, unknown>>).catch(() => undefined);
|
||||
if (status?.state === "ready" || status?.state === "blocked") break;
|
||||
await delay(100);
|
||||
}
|
||||
expect(status).toMatchObject({ id, commit: stamp.commit, state: "ready", eligibility: { eligible: true } });
|
||||
expect(JSON.stringify(status)).not.toContain(original);
|
||||
const exported = await fetch(`${origin}/export`, { method: "POST", headers }).then(r => r.json());
|
||||
expect(exported).toMatchObject({ id, commit: stamp.commit, decisionSigningSecret: original });
|
||||
expect((await fetch(`${origin}/api/companies`, { headers })).status).toBe(404);
|
||||
child.kill("SIGTERM");
|
||||
expect(await Promise.race([exited, delay(12_000).then(() => { throw new Error("Maintenance did not exit gracefully"); })])).toEqual({ code: 0, signal: null });
|
||||
} finally {
|
||||
if (child.exitCode === null && child.signalCode === null) child.kill("SIGKILL");
|
||||
await exited;
|
||||
}
|
||||
}, 60_000);
|
||||
|
||||
it("refuses active executions and missing predecessor bytes without publishing an empty head", async () => {
|
||||
const row = await agent();
|
||||
const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId: row.id, status: "running" }).returning();
|
||||
await expect(migrate()).rejects.toThrow("active_execution_requires_drain");
|
||||
await db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, run!.id));
|
||||
await fs.rm(home, { recursive: true, force: true });
|
||||
await expect(migrate()).rejects.toThrow();
|
||||
expect(await db.select().from(agentFileRoots).where(eq(agentFileRoots.agentId, row.id))).toEqual([]);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,96 @@
|
||||
import { inArray } from "drizzle-orm";
|
||||
import { heartbeatRuns, type Db } from "@paperclipai/db";
|
||||
import { getStorageService } from "../storage/index.js";
|
||||
import { migrateAgentFiles, verifyAgentFileObjects } from "../services/agent-file-maintenance.js";
|
||||
import { readEphemeralServerDataEligibility } from "../services/ephemeral-server-contract.js";
|
||||
import { migrateRunEvidence } from "../services/run-evidence-migration.js";
|
||||
import { skillInstallationMaintenance } from "../services/skill-installation-maintenance.js";
|
||||
import { skillPackageMaintenance } from "../services/skill-package-maintenance.js";
|
||||
|
||||
/** Runs only in the maintenance process after Cloud has confirmed all legacy
|
||||
* deployments stopped. Each owner retains its own migration/publication rules;
|
||||
* a restart repeats idempotent pages, never reimports over durable authority. */
|
||||
export async function migrateEphemeralServerData(db: Db, input: {
|
||||
signal: AbortSignal; phase: (name: string) => void;
|
||||
}) {
|
||||
const { signal, phase } = input;
|
||||
const counts: Record<string, number> = {};
|
||||
const [active] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns)
|
||||
.where(inArray(heartbeatRuns.status, ["running", "queued"])).limit(1);
|
||||
if (active) throw new Error("active_execution_requires_drain");
|
||||
const before = await readEphemeralServerDataEligibility(db);
|
||||
const unsupported = before.blockers.filter(blocker => ![
|
||||
"agent_files_require_migration", "run_logs_require_migration", "provider_traces_require_migration", "skills_require_migration",
|
||||
].includes(blocker));
|
||||
if (unsupported.length) throw new Error(unsupported[0]);
|
||||
|
||||
phase("agent_files");
|
||||
let after: string | undefined;
|
||||
do {
|
||||
signal.throwIfAborted();
|
||||
const page = await migrateAgentFiles(db, { apply: true, writersStopped: true, after, limit: 25 });
|
||||
if (!page.pageReady) throw new Error("agent_file_migration_blocked");
|
||||
counts.agentFiles = (counts.agentFiles ?? 0) + page.results.length;
|
||||
after = page.next ?? undefined;
|
||||
} while (after);
|
||||
|
||||
phase("run_evidence");
|
||||
// Separate cursors avoid restarting an exhausted kind when the other kind
|
||||
// still has pages. A blocked item fails enrollment instead of being skipped.
|
||||
for (const kind of ["log", "trace"] as const) {
|
||||
after = undefined;
|
||||
do {
|
||||
signal.throwIfAborted();
|
||||
const page = await migrateRunEvidence(db, { apply: true, kinds: [kind], limit: 50, signal,
|
||||
...(kind === "log" ? { afterRunId: after } : { afterTraceId: after }) });
|
||||
if (page.items.some(item => item.state !== "migrated")) throw new Error(`${kind}_migration_blocked`);
|
||||
counts[kind] = (counts[kind] ?? 0) + page.items.length;
|
||||
after = kind === "log" ? page.nextRunId : page.nextTraceId;
|
||||
} while (after);
|
||||
}
|
||||
|
||||
const packages = skillPackageMaintenance(db);
|
||||
const skills = skillInstallationMaintenance(db);
|
||||
for (const [name, migrate] of [["skill_packages", packages.migrateSnapshots], ["skill_versions", packages.migrateVersions]] as const) {
|
||||
phase(name);
|
||||
after = undefined;
|
||||
for (;;) {
|
||||
signal.throwIfAborted();
|
||||
const page: Awaited<ReturnType<typeof migrate>> = await migrate({ apply: true, after, limit: 50, signal });
|
||||
counts[name] = (counts[name] ?? 0) + page.inspected;
|
||||
if (page.inspected < 50 || !page.after) break;
|
||||
after = page.after;
|
||||
}
|
||||
}
|
||||
phase("skill_installations");
|
||||
after = undefined;
|
||||
for (;;) {
|
||||
signal.throwIfAborted();
|
||||
const page = await skills.installations({ apply: true, writersStopped: true, after, limit: 50, signal });
|
||||
if (page.results.some(row => !["converted", "unchanged"].includes(row.status))) throw new Error("skill_installation_migration_blocked");
|
||||
counts.skillInstallations = (counts.skillInstallations ?? 0) + page.inspected;
|
||||
if (page.inspected < 50 || !page.after) break;
|
||||
after = page.after;
|
||||
}
|
||||
|
||||
phase("verify_agent_objects");
|
||||
const storage = getStorageService();
|
||||
if (storage.provider !== "s3") throw new Error("shared_object_storage_required");
|
||||
counts.verifiedAgentObjects = await verifyAgentFileObjects(db, storage, signal);
|
||||
phase("verify_skill_versions");
|
||||
after = undefined;
|
||||
for (;;) {
|
||||
signal.throwIfAborted();
|
||||
const page = await skills.verifyVersions({ after, limit: 50, signal });
|
||||
if (page.results.some(row => !["remote_storage", "database_release_snapshot", "database_snapshot", "expired_git_history"].includes(row.status))) {
|
||||
throw new Error("skill_version_verification_failed");
|
||||
}
|
||||
counts.verifiedSkillVersions = (counts.verifiedSkillVersions ?? 0) + page.inspected;
|
||||
if (page.inspected < 50 || !page.after) break;
|
||||
after = page.after;
|
||||
}
|
||||
phase("eligibility");
|
||||
const eligibility = await readEphemeralServerDataEligibility(db);
|
||||
if (!eligibility.eligible) throw new Error("final_data_eligibility_failed");
|
||||
return { counts, eligibility };
|
||||
}
|
||||
@@ -0,0 +1,62 @@
|
||||
import { randomBytes, randomUUID } from "node:crypto";
|
||||
import { afterEach, expect, it } from "vitest";
|
||||
import { startEphemeralMigrationService, type EphemeralMigrationResult } from "./ephemeral-migration-service.js";
|
||||
|
||||
let service: Awaited<ReturnType<typeof startEphemeralMigrationService>> | undefined;
|
||||
afterEach(async () => { await service?.close(); service = undefined; });
|
||||
const result = (): EphemeralMigrationResult => ({ counts: { files: 2 }, eligibility: { eligible: true, blockers: [] }, decisionSigningSecret: randomBytes(32).toString("base64") });
|
||||
async function start(migrate: Parameters<typeof startEphemeralMigrationService>[0]["migrate"], timeoutMs?: number) {
|
||||
const id = randomUUID(), token = randomBytes(32).toString("base64url"), commit = "a".repeat(40);
|
||||
service = await startEphemeralMigrationService({ id, commit, token, port: 0, host: "127.0.0.1", migrate, timeoutMs });
|
||||
const address = service.server.address();
|
||||
if (!address || typeof address === "string") throw new Error("Missing listener");
|
||||
const request = (pathname: string, method = "GET", authorized = true) => fetch(`http://127.0.0.1:${address.port}${pathname}`, {
|
||||
method, headers: authorized ? { authorization: `Bearer ${token}` } : {},
|
||||
});
|
||||
return { request, id, commit };
|
||||
}
|
||||
|
||||
it("keeps the original key out of health/status and exports only after authenticated successful migration", async () => {
|
||||
const original = result();
|
||||
let finish!: (value: EphemeralMigrationResult) => void;
|
||||
const { request, id, commit } = await start(() => new Promise(resolve => { finish = resolve; }));
|
||||
expect((await request("/health", "GET", false)).status).toBe(200);
|
||||
expect((await request("/status", "GET", false)).status).toBe(401);
|
||||
expect((await request("/export", "POST", false)).status).toBe(401);
|
||||
expect((await request("/export", "POST")).status).toBe(409);
|
||||
expect((await request("/api/companies")).status).toBe(404);
|
||||
finish(original); await service!.finished;
|
||||
const status = await request("/status");
|
||||
expect(status.headers.get("cache-control")).toBe("no-store");
|
||||
const body = await status.json();
|
||||
expect(body).toMatchObject({ id, commit, state: "ready", counts: original.counts });
|
||||
expect(JSON.stringify(body)).not.toContain(original.decisionSigningSecret);
|
||||
expect(await (await request("/export", "POST")).json()).toMatchObject({ id, commit, decisionSigningSecret: original.decisionSigningSecret });
|
||||
});
|
||||
|
||||
it("fails closed without exposing owner errors or credentials", async () => {
|
||||
const { request } = await start(async () => { throw new Error("provider rejected secret-value at https://private.invalid"); });
|
||||
await service!.finished;
|
||||
expect(await (await request("/status")).json()).toMatchObject({ state: "blocked", failure: "owner_migration_failed" });
|
||||
expect((await request("/export", "POST")).status).toBe(409);
|
||||
});
|
||||
|
||||
it("rejects incomplete eligibility even when a key was recovered", async () => {
|
||||
const { request } = await start(async () => ({ ...result(), eligibility: { eligible: false, blockers: ["legacy_data"] } }));
|
||||
await service!.finished;
|
||||
expect((await request("/export", "POST")).status).toBe(409);
|
||||
expect(service!.receipt().failure).toBe("incomplete_migration_receipt");
|
||||
});
|
||||
|
||||
it("publishes a blocked deadline immediately and never exports a late completion", async () => {
|
||||
let finish!: (value: EphemeralMigrationResult) => void;
|
||||
let aborted!: Promise<void>;
|
||||
const { request } = await start(signal => {
|
||||
aborted = new Promise(resolve => signal.addEventListener("abort", () => resolve(), { once: true }));
|
||||
return new Promise(resolve => { finish = resolve; });
|
||||
}, 20);
|
||||
await aborted;
|
||||
expect(service!.receipt()).toMatchObject({ state: "blocked", failure: "maintenance_deadline_exceeded" });
|
||||
finish(result()); await service!.finished;
|
||||
expect((await request("/export", "POST")).status).toBe(409);
|
||||
});
|
||||
@@ -0,0 +1,77 @@
|
||||
import { timingSafeEqual } from "node:crypto";
|
||||
import { createServer } from "node:http";
|
||||
|
||||
export interface EphemeralMigrationResult {
|
||||
counts: Record<string, number>;
|
||||
eligibility: { eligible: boolean; blockers: string[] };
|
||||
decisionSigningSecret: string;
|
||||
}
|
||||
|
||||
/** A dedicated maintenance process, never mounted on the ordinary app router.
|
||||
* Cloud supplies a fresh attempt token and keeps tenant routing closed. Only
|
||||
* the authenticated export transfers the original key to its durable vault. */
|
||||
export async function startEphemeralMigrationService(input: {
|
||||
id: string; commit: string; token: string; port: number; host?: string; timeoutMs?: number;
|
||||
migrate: (signal: AbortSignal, phase: (name: string) => void) => Promise<EphemeralMigrationResult>;
|
||||
}) {
|
||||
if (!/^[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}$/i.test(input.id) || !/^[a-f0-9]{40}$/i.test(input.commit)
|
||||
|| !/^[A-Za-z0-9_-]{43,128}$/.test(input.token)) throw new Error("Invalid maintenance identity or credential");
|
||||
const authorization = Buffer.from(`Bearer ${input.token}`);
|
||||
const cancellation = new AbortController();
|
||||
let state: "running" | "ready" | "blocked" = "running", phase = "configuration", failure: string | undefined;
|
||||
let result: EphemeralMigrationResult | undefined;
|
||||
const startedAt = new Date().toISOString();
|
||||
let completedAt: string | undefined;
|
||||
const receipt = () => ({ schema: "paperclip.ephemeral-migration.v1", id: input.id, commit: input.commit,
|
||||
state, phase, startedAt, completedAt, failure, counts: result?.counts, eligibility: result?.eligibility });
|
||||
const server = createServer((request, response) => {
|
||||
response.setHeader("cache-control", "no-store");
|
||||
response.setHeader("content-type", "application/json");
|
||||
const send = (code: number, body: unknown) => { response.writeHead(code); response.end(JSON.stringify(body)); };
|
||||
if (request.method === "GET" && request.url === "/health") return send(200, { status: "maintenance" });
|
||||
const supplied = Buffer.from(request.headers.authorization ?? "");
|
||||
if (supplied.length !== authorization.length || !timingSafeEqual(supplied, authorization)) return send(401, { error: "unauthorized" });
|
||||
if (request.method === "GET" && request.url === "/status") return send(200, receipt());
|
||||
if (request.method === "POST" && request.url === "/export") {
|
||||
if (state !== "ready" || !result) return send(409, receipt());
|
||||
return send(200, { ...receipt(), decisionSigningSecret: result.decisionSigningSecret });
|
||||
}
|
||||
send(404, { error: "not_found" });
|
||||
});
|
||||
server.requestTimeout = 10_000;
|
||||
server.headersTimeout = 10_000;
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
server.once("error", reject);
|
||||
server.listen(input.port, input.host ?? "0.0.0.0", () => { server.off("error", reject); resolve(); });
|
||||
});
|
||||
cancellation.signal.addEventListener("abort", () => {
|
||||
state = "blocked";
|
||||
failure = cancellation.signal.reason instanceof Error ? cancellation.signal.reason.message : "maintenance_stopped";
|
||||
completedAt = new Date().toISOString();
|
||||
}, { once: true });
|
||||
const deadline = setTimeout(() => cancellation.abort(new Error("maintenance_deadline_exceeded")), input.timeoutMs ?? 25 * 60_000);
|
||||
const finished = (async () => {
|
||||
try {
|
||||
const migrated = await input.migrate(cancellation.signal, next => { phase = next; });
|
||||
cancellation.signal.throwIfAborted();
|
||||
if (!migrated.eligibility.eligible || migrated.decisionSigningSecret.trim().length < 32) throw new Error("incomplete_migration_receipt");
|
||||
result = migrated; state = "ready";
|
||||
} catch (error) {
|
||||
// Owner codes are safe diagnostics. Never echo arbitrary SDK errors,
|
||||
// credential-bearing URLs, file contents or stack traces through HTTP.
|
||||
const code = error instanceof Error ? error.message : "owner_migration_failed";
|
||||
failure = /^[a-z_]{1,80}$/.test(code) ? code : "owner_migration_failed";
|
||||
state = "blocked";
|
||||
} finally { clearTimeout(deadline); completedAt = new Date().toISOString(); }
|
||||
})();
|
||||
return { server, finished, receipt,
|
||||
async close() {
|
||||
cancellation.abort(new Error("maintenance_stopped"));
|
||||
clearTimeout(deadline);
|
||||
server.closeIdleConnections();
|
||||
server.closeAllConnections();
|
||||
await new Promise<void>((resolve, reject) => server.close(error => error ? reject(error) : resolve()));
|
||||
await finished;
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,42 @@
|
||||
import { closeRegisteredClients, createDb } from "@paperclipai/db";
|
||||
import { readBuildCommit } from "../build-commit.js";
|
||||
import { loadConfig } from "../config.js";
|
||||
import { cloudRuntimeConfigService } from "../services/cloud-runtime-config.js";
|
||||
import { setCloudRuntimeConfigSnapshot } from "../services/cloud-runtime-config-snapshot.js";
|
||||
import { isCloudManagedInstance } from "../services/cloud-instance.js";
|
||||
import { resolveDecisionSigningSecret } from "../services/decision-signing.js";
|
||||
import { assertEphemeralServerConfig } from "../services/ephemeral-server-contract.js";
|
||||
import { migrateEphemeralServerData } from "./ephemeral-migration-data.js";
|
||||
import { startEphemeralMigrationService } from "./ephemeral-migration-service.js";
|
||||
|
||||
// Cloud starts this image entry only after its provider has confirmed the web
|
||||
// and every legacy worker stopped. It never starts schedulers or app routes.
|
||||
if (process.env.PAPERCLIP_EPHEMERAL_MIGRATION !== "true") throw new Error("Explicit maintenance mode is required");
|
||||
const url = process.env.DATABASE_URL;
|
||||
const commit = readBuildCommit();
|
||||
if (!url || !commit) throw new Error("Maintenance requires an explicit database and stamped image");
|
||||
const service = await startEphemeralMigrationService({
|
||||
id: process.env.PAPERCLIP_EPHEMERAL_MIGRATION_ID ?? "", token: process.env.PAPERCLIP_EPHEMERAL_MIGRATION_TOKEN ?? "",
|
||||
commit, port: Number(process.env.PORT ?? 3100),
|
||||
async migrate(signal, phase) {
|
||||
const db = createDb(url, { maxConnections: 2 });
|
||||
try {
|
||||
if (isCloudManagedInstance()) {
|
||||
const config = await cloudRuntimeConfigService(db).inspectStored();
|
||||
if (!config) throw new Error("cloud_runtime_configuration_missing");
|
||||
setCloudRuntimeConfigSnapshot(config);
|
||||
}
|
||||
// Never generate a replacement key over a missing volume. The original
|
||||
// key is exported only after every owner confirms durable eligibility.
|
||||
const decisionSigningSecret = resolveDecisionSigningSecret({ allowCreate: false });
|
||||
assertEphemeralServerConfig(loadConfig(), { ...process.env, PAPERCLIP_EPHEMERAL_SERVER: "true",
|
||||
PAPERCLIP_DECISION_SIGNING_SECRET: decisionSigningSecret });
|
||||
const result = await migrateEphemeralServerData(db, { signal, phase });
|
||||
return { ...result, decisionSigningSecret };
|
||||
} finally { await closeRegisteredClients(url, { scope: "database" }); }
|
||||
},
|
||||
});
|
||||
for (const signal of ["SIGINT", "SIGTERM"] as const) process.once(signal, () => {
|
||||
const deadline = setTimeout(() => process.exit(1), 10_000);
|
||||
void service.close().then(() => { clearTimeout(deadline); process.exit(0); }, () => process.exit(1));
|
||||
});
|
||||
@@ -1,3 +1,4 @@
|
||||
import { createHash } from "node:crypto";
|
||||
import fs from "node:fs/promises";
|
||||
import path from "node:path";
|
||||
import { and, asc, eq, gt, inArray } from "drizzle-orm";
|
||||
@@ -96,3 +97,34 @@ export async function collectAgentFileObjects(db: Db, provider: Pick<StorageProv
|
||||
}
|
||||
return { retained, eligible, deleted, next: page.continuationToken ?? null };
|
||||
}
|
||||
|
||||
/** Verify all authoritative bytes before the legacy volume can be detached.
|
||||
* This runs only with writers stopped, using bounded pages and streams. */
|
||||
export async function verifyAgentFileObjects(db: Db, storage: StorageService, signal: AbortSignal): Promise<number> {
|
||||
if (storage.provider !== "s3") throw new Error("shared_object_storage_required");
|
||||
let after: string | undefined;
|
||||
let verified = 0;
|
||||
for (;;) {
|
||||
signal.throwIfAborted();
|
||||
const heads = await db.select().from(agentFileRoots).where(after ? gt(agentFileRoots.agentId, after) : undefined)
|
||||
.orderBy(asc(agentFileRoots.agentId)).limit(25);
|
||||
for (const head of heads) for (const file of head.entries) {
|
||||
if (file.kind !== "file") continue;
|
||||
const hash = createHash("sha256");
|
||||
let bytes = 0;
|
||||
if (file.objectKey !== null) {
|
||||
const object = await storage.getObject(head.companyId, file.objectKey, { signal });
|
||||
for await (const chunk of object.stream) {
|
||||
signal.throwIfAborted(); bytes += chunk.length;
|
||||
if (bytes > file.size) throw new Error("agent_file_object_size_mismatch");
|
||||
hash.update(chunk);
|
||||
}
|
||||
}
|
||||
if (bytes !== file.size || hash.digest("hex") !== file.sha256) throw new Error("agent_file_object_digest_mismatch");
|
||||
verified++;
|
||||
}
|
||||
if (heads.length < 25) break;
|
||||
after = heads.at(-1)!.agentId;
|
||||
}
|
||||
return verified;
|
||||
}
|
||||
@@ -0,0 +1,30 @@
|
||||
import { mkdtemp, rm } from "node:fs/promises";
|
||||
import { existsSync } from "node:fs";
|
||||
import { tmpdir } from "node:os";
|
||||
import path from "node:path";
|
||||
import { afterEach, beforeEach, expect, it, vi } from "vitest";
|
||||
import { resolveDefaultSecretsKeyFilePath } from "../home-paths.js";
|
||||
import { resolveDecisionSigningSecret, signDecisionSpec, verifyDecisionSpec } from "./decision-signing.js";
|
||||
|
||||
let home: string;
|
||||
beforeEach(async () => {
|
||||
home = await mkdtemp(path.join(tmpdir(), "decision-migration-"));
|
||||
vi.stubEnv("PAPERCLIP_HOME", home);
|
||||
vi.stubEnv("PAPERCLIP_DECISION_SIGNING_SECRET", "");
|
||||
});
|
||||
afterEach(async () => { vi.unstubAllEnvs(); await rm(home, { recursive: true, force: true }); });
|
||||
it("does not manufacture a replacement key when the legacy volume is missing", () => {
|
||||
const key = path.join(path.dirname(resolveDefaultSecretsKeyFilePath()), "decision-signing.key");
|
||||
expect(() => resolveDecisionSigningSecret({ allowCreate: false })).toThrow();
|
||||
expect(existsSync(key)).toBe(false);
|
||||
});
|
||||
it("preserves decision signatures after the original key moves to environment storage", async () => {
|
||||
const original = resolveDecisionSigningSecret();
|
||||
const spec = { action: "approve", id: "retained-decision" };
|
||||
const signature = signDecisionSpec(spec);
|
||||
expect(resolveDecisionSigningSecret({ allowCreate: false })).toBe(original);
|
||||
await rm(home, { recursive: true, force: true });
|
||||
vi.stubEnv("PAPERCLIP_DECISION_SIGNING_SECRET", original);
|
||||
expect(resolveDecisionSigningSecret({ allowCreate: false })).toBe(original);
|
||||
expect(verifyDecisionSpec(spec, signature)).toBe(true);
|
||||
});
|
||||
@@ -120,7 +120,7 @@ function loadOrCreateGeneratedSecret(): string {
|
||||
}
|
||||
}
|
||||
|
||||
export function resolveDecisionSigningSecret(): string {
|
||||
export function resolveDecisionSigningSecret(options: { allowCreate?: boolean } = {}): string {
|
||||
const fromEnv = process.env.PAPERCLIP_DECISION_SIGNING_SECRET?.trim();
|
||||
if (fromEnv) {
|
||||
if (fromEnv.length < MIN_SECRET_LENGTH) {
|
||||
@@ -130,7 +130,9 @@ export function resolveDecisionSigningSecret(): string {
|
||||
}
|
||||
return fromEnv;
|
||||
}
|
||||
return loadOrCreateGeneratedSecret();
|
||||
return options.allowCreate === false
|
||||
? readGeneratedSecret(resolveGeneratedSecretFilePath())
|
||||
: loadOrCreateGeneratedSecret();
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
Reference in new issue
Block a user