From b89d6adb39466721f36d1871feaffb5674cd0f7c Mon Sep 17 00:00:00 2001 From: Dotta Date: Tue, 21 Jul 2026 02:38:02 +0000 Subject: [PATCH] fix: stream and consistently copy state snapshots --- server/src/services/instance-state-snapshot.ts | 13 ++++++++++--- server/src/storage/local-disk-provider.ts | 5 ++++- 2 files changed, 14 insertions(+), 4 deletions(-) diff --git a/server/src/services/instance-state-snapshot.ts b/server/src/services/instance-state-snapshot.ts index 193674263e6..822226ef3b6 100644 --- a/server/src/services/instance-state-snapshot.ts +++ b/server/src/services/instance-state-snapshot.ts @@ -87,6 +87,11 @@ async function expandPattern(pattern: string): Promise { async function copyEntry(entry: StateClassEntry, source: string, destination: string): Promise { const stat = await fs.stat(source); await fs.mkdir(path.dirname(destination), { recursive: true }); + if (entry.consistency === "sqlite_backup" && stat.isDirectory()) { + await fs.mkdir(destination, { recursive: true }); + for (const child of await fs.readdir(source)) await copyEntry(entry, path.join(source, child), path.join(destination, child)); + return; + } if (entry.consistency === "sqlite_backup" && stat.isFile() && /\.(sqlite|sqlite3|db)$/i.test(source)) { await execFileAsync("sqlite3", [source, `.backup '${destination.replace(/'/g, "''")}'`]); return; @@ -138,10 +143,12 @@ export function createInstanceStateSnapshotService(opts: { await execFileAsync("tar", ["-czf", archivePath, "-C", stageDir, "."]); const encryptedPath = `${archivePath}.enc`; await opts.encryptionProvider.encrypt(archivePath, encryptedPath); - const body = await fs.readFile(encryptedPath); + const encryptedStat = await fs.stat(encryptedPath); const objectKey = `instance-state/${context.instanceId ?? "default"}/${startedAt.toISOString().replace(/[:.]/g, "-")}.tar.gz.enc`; - await opts.storageProvider.putObject({ objectKey, body, contentType: "application/vnd.paperclip.state-snapshot", contentLength: body.length }); - const result = { objectKey, sizeBytes: body.length, sha256: createHash("sha256").update(body).digest("hex"), entryCount, startedAt: startedAt.toISOString(), finishedAt: new Date().toISOString() }; + 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() }; 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/storage/local-disk-provider.ts b/server/src/storage/local-disk-provider.ts index 56b35495452..8046fb212e5 100644 --- a/server/src/storage/local-disk-provider.ts +++ b/server/src/storage/local-disk-provider.ts @@ -1,4 +1,6 @@ import { createReadStream, promises as fs } from "node:fs"; +import { createWriteStream } from "node:fs"; +import { pipeline } from "node:stream/promises"; import path from "node:path"; import type { StorageProvider, GetObjectResult, HeadObjectResult } from "./types.js"; import { notFound, badRequest } from "../errors.js"; @@ -47,7 +49,8 @@ export function createLocalDiskStorageProvider(baseDir: string): StorageProvider await fs.mkdir(dir, { recursive: true }); const tempPath = `${targetPath}.tmp-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`; - await fs.writeFile(tempPath, input.body); + if (Buffer.isBuffer(input.body)) await fs.writeFile(tempPath, input.body); + else await pipeline(input.body, createWriteStream(tempPath)); await fs.rename(tempPath, targetPath); },