mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-27 19:35:13 +01:00
feat: add cloud S3 bulk storage defaults
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
a323bdf0af
commit
84c03f2dee
10 files changed
+184
-28
No files matched your search
@@ -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 `<storage-prefix>/run-logs/`; dedicated `RUN_LOG_S3_*` variables remain an override.
|
||||
|
||||
Encrypted state snapshots keep durable state at `instance-state/<instance>/...`. Manifest entries with retention are written separately at `retention/<days>-days/instance-state/<instance>/...`, 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.
|
||||
@@ -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"));
|
||||
});
|
||||
});
|
||||
@@ -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" },
|
||||
|
||||
Executable
+22
@@ -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" <<JSON
|
||||
{"Rules":[{"ID":"paperclip-run-logs-90d","Status":"Enabled","Filter":{"Prefix":"${base}run-logs/"},"Expiration":{"Days":90}},{"ID":"paperclip-retained-snapshots-90d","Status":"Enabled","Filter":{"Prefix":"${base}retention/90-days/"},"Expiration":{"Days":90}}]}
|
||||
JSON
|
||||
|
||||
if [[ "${PAPERCLIP_LIFECYCLE_DRY_RUN:-false}" == "true" ]]; then
|
||||
cat "$config"
|
||||
exit 0
|
||||
fi
|
||||
|
||||
aws s3api put-bucket-lifecycle-configuration \
|
||||
--bucket "$PAPERCLIP_STORAGE_S3_BUCKET" \
|
||||
--lifecycle-configuration "file://$config"
|
||||
@@ -48,4 +48,34 @@ describe("instance state snapshots", () => {
|
||||
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<string, Buffer>();
|
||||
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);
|
||||
});
|
||||
});
|
||||
@@ -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");
|
||||
});
|
||||
});
|
||||
+13
-1
@@ -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 ??
|
||||
|
||||
@@ -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<StartedServer> {
|
||||
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
|
||||
|
||||
@@ -66,6 +66,19 @@ async function expandPattern(pattern: string): Promise<string[]> {
|
||||
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<void> => {
|
||||
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<number, StateClassEntry[]>();
|
||||
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 });
|
||||
|
||||
@@ -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,
|
||||
};
|
||||
|
||||
Reference in new issue
Block a user