diff --git a/doc/S3-BULK-STORAGE.md b/doc/S3-BULK-STORAGE.md new file mode 100644 index 00000000000..d8475cb61e0 --- /dev/null +++ b/doc/S3-BULK-STORAGE.md @@ -0,0 +1,17 @@ +# S3 bulk storage operations + +Cloud runtimes (`PAPERCLIP_K8S_IN_CLUSTER=true`) default attachments to the configured S3 provider. Explicit `PAPERCLIP_STORAGE_PROVIDER` or `config.json` storage settings still win. Completed run logs use the same bucket/settings by default and are stored under `/run-logs/`; dedicated `RUN_LOG_S3_*` variables remain an override. + +Encrypted state snapshots keep durable state at `instance-state//...`. Manifest entries with retention are written separately at `retention/-days/instance-state//...`, so transcript/run-log retention cannot delete Codex sqlite snapshots. The approved transcript and run-log retention is 90 days. + +Apply the lifecycle policy with: + +```sh +PAPERCLIP_STORAGE_S3_BUCKET=paperclip \ +PAPERCLIP_STORAGE_S3_PREFIX=production \ +scripts/apply-s3-bulk-lifecycle.sh +``` + +The script uses `aws s3api put-bucket-lifecycle-configuration`. Set `AWS_ENDPOINT_URL` for an S3-compatible provider. Re-running it is idempotent but replaces the bucket lifecycle configuration, so merge additional bucket rules into the script payload before applying on a shared bucket. + +Litestream was evaluated for Codex sqlite continuous point-in-time restore. It improves recovery point objectives, but adds a runtime binary/sidecar, per-database replica configuration, monitoring, restore orchestration, and ongoing version/security maintenance. P4 ships scheduled `sqlite3 .backup` copies without that dependency; adopting Litestream remains an operations decision for a later cloud phase. diff --git a/packages/shared/src/state-manifest.test.ts b/packages/shared/src/state-manifest.test.ts index 9d9673dcbbe..0740e9d28a1 100644 --- a/packages/shared/src/state-manifest.test.ts +++ b/packages/shared/src/state-manifest.test.ts @@ -22,5 +22,7 @@ describe("STATE_MANIFEST", () => { } expect(STATE_MANIFEST.find((entry) => entry.id === "secrets_master_key")?.resolve({ homeDir, instanceId: "live" })[0]) .toBe(path.join(homeDir, "instances", "live", "secrets", "master.key")); + expect(STATE_MANIFEST.find((entry) => entry.id === "codex_agent_home")?.resolve({ homeDir, instanceId: "live" })) + .toContain(path.join(homeDir, "instances", "live", "companies", "*", "agents", "*", "codex-home")); }); }); diff --git a/packages/shared/src/state-manifest.ts b/packages/shared/src/state-manifest.ts index 171b303b93d..af3deb5eb5c 100644 --- a/packages/shared/src/state-manifest.ts +++ b/packages/shared/src/state-manifest.ts @@ -39,7 +39,7 @@ export const STATE_MANIFEST: readonly StateClassEntry[] = [ { id: "local_backup_staging", resolve: (ctx) => [instance(ctx, "data", "backups"), instance(ctx, "data", "instance-backups")], disposition: "cache", redact: "forbid", consistency: "plain", retention: { days: 30 } }, { id: "agent_instructions_bundle", resolve: (ctx) => [instance(ctx, "companies", "*", "agents", "*", "instructions")], disposition: "git", redact: "secret_refs", consistency: "plain" }, { id: "secrets_master_key", resolve: (ctx) => [resolveDefaultSecretsKeyFilePath(ctx)], disposition: "s3_secret", redact: "forbid", consistency: "plain" }, - { id: "codex_agent_home", resolve: (ctx) => [instance(ctx, "companies", "*", "codex-home")], disposition: "s3_bulk", redact: "forbid", consistency: "sqlite_backup" }, + { id: "codex_agent_home", resolve: (ctx) => [instance(ctx, "companies", "*", "agents", "*", "codex-home"), instance(ctx, "companies", "*", "codex-home")], disposition: "s3_bulk", redact: "forbid", consistency: "sqlite_backup" }, { id: "runtime_materializations", resolve: (ctx) => [instance(ctx, "companies", "*", "acp-engine"), instance(ctx, "companies", "*", "acpx-local"), instance(ctx, "companies", "*", "claude-prompt-cache")], disposition: "cache", redact: "secret_refs", consistency: "plain" }, { id: "skill_bundles", resolve: (ctx) => [instance(ctx, "skills", "*")], disposition: "git", redact: "secret_refs", consistency: "plain" }, { id: "project_repositories", resolve: (ctx) => [instance(ctx, "projects", "*", "*", "*")], disposition: "git", redact: "forbid", consistency: "plain" }, diff --git a/scripts/apply-s3-bulk-lifecycle.sh b/scripts/apply-s3-bulk-lifecycle.sh new file mode 100755 index 00000000000..b4747c621e7 --- /dev/null +++ b/scripts/apply-s3-bulk-lifecycle.sh @@ -0,0 +1,22 @@ +#!/usr/bin/env bash +set -euo pipefail + +: "${PAPERCLIP_STORAGE_S3_BUCKET:?PAPERCLIP_STORAGE_S3_BUCKET is required}" +prefix="${PAPERCLIP_STORAGE_S3_PREFIX:-}" +prefix="${prefix#/}" +prefix="${prefix%/}" +base="${prefix:+${prefix}/}" +config="$(mktemp)" +trap 'rm -f "$config"' EXIT +cat >"$config" < { expect(await fs.readFile(source, "utf8")).toBe("before"); expect(JSON.parse(await fs.readFile(path.join(homeDir, "markers", "state-snapshot.success.json"), "utf8"))).toMatchObject({ objectKey: result.objectKey }); }); + + it("stores nested transcript glob matches in a retention-specific object", async () => { + const homeDir = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-retention-test-")); + tempDirs.push(homeDir); + const transcript = path.join(homeDir, "claude", "projects", "company", "agent", "session.jsonl"); + await fs.mkdir(path.dirname(transcript), { recursive: true }); + await fs.writeFile(transcript, "transcript"); + const objects = new Map(); + const provider: StorageProvider = { + id: "s3", + async putObject(input) { + const chunks: Buffer[] = []; + if (Buffer.isBuffer(input.body)) chunks.push(input.body); + else for await (const chunk of input.body) chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)); + objects.set(input.objectKey, Buffer.concat(chunks)); + }, + async getObject(input) { const body = objects.get(input.objectKey); if (!body) throw new Error("missing"); return { stream: Readable.from(body), contentLength: body.length }; }, + async headObject(input) { const body = objects.get(input.objectKey); return { exists: Boolean(body), contentLength: body?.length }; }, + async deleteObject(input) { objects.delete(input.objectKey); }, + }; + const manifest: StateClassEntry[] = [{ id: "transcripts", resolve: () => [path.join(homeDir, "claude", "projects", "**", "*.jsonl")], disposition: "s3_bulk", redact: "secret_refs", consistency: "plain", retention: { days: 90 } }]; + const service = createInstanceStateSnapshotService({ storageProvider: provider, encryptionProvider: createAesStateSnapshotEncryptionProvider(Buffer.alloc(32, 9)), context: { homeDir, instanceId: "test" }, manifest }); + + const result = await service.runSnapshot(); + + expect(result.retentionObjects).toHaveLength(1); + expect(result.retentionObjects[0]).toMatchObject({ retentionDays: 90, entryCount: 1 }); + expect(result.retentionObjects[0]?.objectKey).toContain("retention/90-days/"); + expect(objects.has(result.retentionObjects[0]!.objectKey)).toBe(true); + }); }); diff --git a/server/src/__tests__/storage-cloud-defaults.test.ts b/server/src/__tests__/storage-cloud-defaults.test.ts new file mode 100644 index 00000000000..dbbbb9ce044 --- /dev/null +++ b/server/src/__tests__/storage-cloud-defaults.test.ts @@ -0,0 +1,11 @@ +import { describe, expect, it } from "vitest"; +import { resolveStorageProviderDefault } from "../config.js"; + +describe("cloud storage defaults", () => { + it("defaults cloud runtimes to S3 without overriding explicit configuration", () => { + expect(resolveStorageProviderDefault({ inCloudRuntime: true })).toBe("s3"); + expect(resolveStorageProviderDefault({ inCloudRuntime: false })).toBe("local_disk"); + expect(resolveStorageProviderDefault({ inCloudRuntime: true, fileProvider: "local_disk" })).toBe("local_disk"); + expect(resolveStorageProviderDefault({ inCloudRuntime: false, envProvider: "s3" })).toBe("s3"); + }); +}); diff --git a/server/src/config.ts b/server/src/config.ts index 90d6cbf6b14..9d762c75b36 100644 --- a/server/src/config.ts +++ b/server/src/config.ts @@ -89,6 +89,14 @@ export interface Config { telemetryEnabled: boolean; } +export function resolveStorageProviderDefault(input: { + envProvider?: StorageProvider | null; + fileProvider?: StorageProvider; + inCloudRuntime?: boolean; +}): StorageProvider { + return input.envProvider ?? input.fileProvider ?? (input.inCloudRuntime ? "s3" : "local_disk"); +} + function detectTailnetBindHost(): string | undefined { const explicit = process.env.PAPERCLIP_TAILNET_BIND_HOST?.trim(); if (explicit) return explicit; @@ -134,7 +142,11 @@ export function loadConfig(): Config { storageProviderFromEnvRaw && STORAGE_PROVIDERS.includes(storageProviderFromEnvRaw as StorageProvider) ? (storageProviderFromEnvRaw as StorageProvider) : null; - const storageProvider: StorageProvider = storageProviderFromEnv ?? fileStorage?.provider ?? "local_disk"; + const storageProvider = resolveStorageProviderDefault({ + envProvider: storageProviderFromEnv, + fileProvider: fileStorage?.provider, + inCloudRuntime: process.env.PAPERCLIP_K8S_IN_CLUSTER === "true", + }); const storageLocalDiskBaseDir = resolveHomeAwarePath( process.env.PAPERCLIP_STORAGE_LOCAL_DIR ?? fileStorage?.localDisk?.baseDir ?? diff --git a/server/src/index.ts b/server/src/index.ts index 14fb2027e0c..4278b353685 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -82,6 +82,7 @@ import { isLoopbackHost, rewriteLoopbackUrlPort } from "./url-utils.js"; import { createPluginWorkerManager } from "./services/plugin-worker-manager.js"; import { createStorageServiceFromConfig } from "./storage/index.js"; import { createStorageProviderFromConfig } from "./storage/provider-registry.js"; +import { configureRunLogS3Defaults } from "./services/run-log-store.js"; import { createAesStateSnapshotEncryptionProvider, createInstanceStateSnapshotService } from "./services/instance-state-snapshot.js"; import { printStartupBanner } from "./startup-banner.js"; import { getBoardClaimWarningUrl, initializeBoardClaimChallenge } from "./board-claim.js"; @@ -641,6 +642,14 @@ export async function startServer(): Promise { const uiMode = config.uiDevMiddleware ? "vite-dev" : config.serveUi ? "static" : "none"; const storageService = createStorageServiceFromConfig(config); const storageProvider = createStorageProviderFromConfig(config); + configureRunLogS3Defaults({ + enabled: process.env.PAPERCLIP_K8S_IN_CLUSTER === "true" && config.storageProvider === "s3", + bucket: config.storageS3Bucket, + region: config.storageS3Region, + endpoint: config.storageS3Endpoint, + prefix: config.storageS3Prefix, + forcePathStyle: config.storageS3ForcePathStyle, + }); const stateSnapshotInstanceId = resolvePaperclipInstanceId(); const stateSnapshotKeyRaw = process.env.PAPERCLIP_STATE_SNAPSHOT_KEY?.trim(); const stateSnapshotKey = stateSnapshotKeyRaw diff --git a/server/src/services/instance-state-snapshot.ts b/server/src/services/instance-state-snapshot.ts index 822226ef3b6..ae3579cc3e6 100644 --- a/server/src/services/instance-state-snapshot.ts +++ b/server/src/services/instance-state-snapshot.ts @@ -66,6 +66,19 @@ async function expandPattern(pattern: string): Promise { const segments = pattern.slice(parsed.root.length).split(path.sep); let current = [parsed.root]; for (const segment of segments) { + if (segment === "**") { + const directories: string[] = []; + const visit = async (directory: string): Promise => { + if (!(await exists(directory))) return; + directories.push(directory); + for (const entry of await fs.readdir(directory, { withFileTypes: true })) { + if (entry.isDirectory()) await visit(path.join(directory, entry.name)); + } + }; + for (const base of current) await visit(base); + current = directories; + continue; + } if (!segment.includes("*")) { current = current.map((base) => path.join(base, segment)); continue; @@ -104,6 +117,7 @@ export type InstanceStateSnapshotResult = { sizeBytes: number; sha256: string; entryCount: number; + retentionObjects: Array<{ objectKey: string; retentionDays: number; sizeBytes: number; sha256: string; entryCount: number }>; startedAt: string; finishedAt: string; }; @@ -125,30 +139,50 @@ export function createInstanceStateSnapshotService(opts: { const startedAt = new Date(); const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-state-snapshot-")); try { - const stageDir = path.join(tempDir, "stage"); - let entryCount = 0; - for (const entry of manifest) { - if (entry.disposition === "db" || entry.disposition === "cache" || entry.disposition === "ephemeral") continue; - for (const pattern of entry.resolve(context)) { - for (const source of await expandPattern(pattern)) { - const root = source === homeRoot || source.startsWith(homeRoot + path.sep) ? homeRoot : userRoot; - const rootName = root === homeRoot ? "paperclip-home" : "user-home"; - await copyEntry(entry, source, path.join(stageDir, rootName, path.relative(root, source))); - entryCount += 1; + const timestamp = startedAt.toISOString().replace(/[:.]/g, "-"); + const eligible = manifest.filter((entry) => entry.disposition !== "db" && entry.disposition !== "cache" && entry.disposition !== "ephemeral"); + const writeArchive = async (entries: readonly StateClassEntry[], objectKey: string, label: string) => { + const stageDir = path.join(tempDir, `stage-${label}`); + let entryCount = 0; + for (const entry of entries) { + for (const pattern of entry.resolve(context)) { + for (const source of await expandPattern(pattern)) { + const root = source === homeRoot || source.startsWith(homeRoot + path.sep) ? homeRoot : userRoot; + const rootName = root === homeRoot ? "paperclip-home" : "user-home"; + await copyEntry(entry, source, path.join(stageDir, rootName, path.relative(root, source))); + entryCount += 1; + } } } + await fs.mkdir(stageDir, { recursive: true }); + await fs.writeFile(path.join(stageDir, "snapshot-manifest.json"), JSON.stringify({ version: 1, createdAt: startedAt.toISOString(), entries: entries.map(({ resolve: _resolve, ...entry }) => entry) }, null, 2)); + const archivePath = path.join(tempDir, `${label}.tar.gz`); + await execFileAsync("tar", ["-czf", archivePath, "-C", stageDir, "."]); + const encryptedPath = `${archivePath}.enc`; + await opts.encryptionProvider.encrypt(archivePath, encryptedPath); + const encryptedStat = await fs.stat(encryptedPath); + const sha256 = createHash("sha256"); + for await (const chunk of createReadStream(encryptedPath)) sha256.update(chunk); + await opts.storageProvider.putObject({ objectKey, body: createReadStream(encryptedPath), contentType: "application/vnd.paperclip.state-snapshot", contentLength: encryptedStat.size }); + return { objectKey, sizeBytes: encryptedStat.size, sha256: sha256.digest("hex"), entryCount }; + }; + + const objectKey = `instance-state/${context.instanceId ?? "default"}/${timestamp}.tar.gz.enc`; + const durable = await writeArchive(eligible.filter((entry) => !entry.retention?.days), objectKey, "durable"); + const retentionGroups = new Map(); + for (const entry of eligible) { + const days = entry.retention?.days; + if (!days) continue; + const group = retentionGroups.get(days) ?? []; + group.push(entry); + retentionGroups.set(days, group); } - await fs.writeFile(path.join(stageDir, "snapshot-manifest.json"), JSON.stringify({ version: 1, createdAt: startedAt.toISOString(), entries: manifest.map(({ resolve: _resolve, ...entry }) => entry) }, null, 2)); - const archivePath = path.join(tempDir, "snapshot.tar.gz"); - await execFileAsync("tar", ["-czf", archivePath, "-C", stageDir, "."]); - const encryptedPath = `${archivePath}.enc`; - await opts.encryptionProvider.encrypt(archivePath, encryptedPath); - const encryptedStat = await fs.stat(encryptedPath); - const objectKey = `instance-state/${context.instanceId ?? "default"}/${startedAt.toISOString().replace(/[:.]/g, "-")}.tar.gz.enc`; - const sha256 = createHash("sha256"); - for await (const chunk of createReadStream(encryptedPath)) sha256.update(chunk); - await opts.storageProvider.putObject({ objectKey, body: createReadStream(encryptedPath), contentType: "application/vnd.paperclip.state-snapshot", contentLength: encryptedStat.size }); - const result = { objectKey, sizeBytes: encryptedStat.size, sha256: sha256.digest("hex"), entryCount, startedAt: startedAt.toISOString(), finishedAt: new Date().toISOString() }; + const retentionObjects: InstanceStateSnapshotResult["retentionObjects"] = []; + for (const [retentionDays, entries] of retentionGroups) { + const retained = await writeArchive(entries, `retention/${retentionDays}-days/instance-state/${context.instanceId ?? "default"}/${timestamp}.tar.gz.enc`, `retention-${retentionDays}`); + retentionObjects.push({ ...retained, retentionDays }); + } + const result = { ...durable, entryCount: durable.entryCount + retentionObjects.reduce((sum, item) => sum + item.entryCount, 0), retentionObjects, startedAt: startedAt.toISOString(), finishedAt: new Date().toISOString() }; await fs.mkdir(markerDir, { recursive: true }); await fs.writeFile(path.join(markerDir, "state-snapshot.success.json"), JSON.stringify(result)); await fs.rm(path.join(markerDir, "state-snapshot.failure"), { force: true }); diff --git a/server/src/services/run-log-store.ts b/server/src/services/run-log-store.ts index 90f7f975fe7..5fc4049538e 100644 --- a/server/src/services/run-log-store.ts +++ b/server/src/services/run-log-store.ts @@ -387,25 +387,44 @@ export function createDurableRunLogStore(options: DurableRunLogStoreOptions): Ru // NOT redirect the product's workspace/file storage (smaller blast radius). // Unset RUN_LOG_S3_BUCKET -> no mirror -> local-only (safe degrade). Creds come // from the standard AWS_ACCESS_KEY_ID/AWS_SECRET_ACCESS_KEY chain. +type RunLogS3Defaults = { + enabled: boolean; + bucket: string; + region: string; + endpoint?: string; + prefix?: string; + forcePathStyle: boolean; +}; + +let runLogS3Defaults: RunLogS3Defaults | undefined; + +export function configureRunLogS3Defaults(defaults: RunLogS3Defaults | undefined): void { + runLogS3Defaults = defaults; + cachedStore = null; +} + function resolveRunLogS3(): DurableRunLogStoreOptions["s3"] { - const bucket = process.env.RUN_LOG_S3_BUCKET?.trim(); + const bucket = + process.env.RUN_LOG_S3_BUCKET?.trim() || + (runLogS3Defaults?.enabled ? runLogS3Defaults.bucket : undefined); if (!bucket) return undefined; const provider = createS3StorageProvider({ bucket, - region: process.env.RUN_LOG_S3_REGION?.trim() || "us-east-1", - endpoint: process.env.RUN_LOG_S3_ENDPOINT?.trim() || undefined, + region: process.env.RUN_LOG_S3_REGION?.trim() || runLogS3Defaults?.region || "us-east-1", + endpoint: process.env.RUN_LOG_S3_ENDPOINT?.trim() || runLogS3Defaults?.endpoint, prefix: undefined, // prefixing is handled by keyPrefix below (kept off the provider) forcePathStyle: process.env.RUN_LOG_S3_FORCE_PATH_STYLE ? process.env.RUN_LOG_S3_FORCE_PATH_STYLE === "true" - : true, // Cubbit (and most S3-compatible endpoints) need path-style + : (runLogS3Defaults?.forcePathStyle ?? true), }); // Opt-in in-flight tail mirroring: at most one partial upload per interval // per active run, so a crash loses at most one interval's tail. Unset/0 // keeps the historical finalize-only mirroring. const inflightSeconds = Number.parseFloat(process.env.RUN_LOG_S3_INFLIGHT_MIRROR_SECONDS ?? ""); + const defaultPrefix = [runLogS3Defaults?.prefix, "run-logs"].filter(Boolean).join("/"); return { provider, - keyPrefix: process.env.RUN_LOG_S3_PREFIX?.trim() || "run-logs", + keyPrefix: process.env.RUN_LOG_S3_PREFIX?.trim() || defaultPrefix || "run-logs", inflightMirrorMs: Number.isFinite(inflightSeconds) && inflightSeconds > 0 ? Math.round(inflightSeconds * 1000) : undefined, };