fix(server-s3): transcript snapshot copies asynchronously (F42 follow-up)
copyIfNewer read/wrote a member file of up to 256 MiB with readSync/writeSync, blocking the event loop for every user. It now copies with fs.promises 1 MiB at a time from the runAs.openSafeAsync handle (owner/type check kept), keeps the 256 MiB member cap and zero-run holes, writes a temp file and renames it into place, and joins a copy already in flight for the same destination. A failed copy is logged and never rejects; snapshotTranscript is async and all three import callers await it.
This commit is contained in:
+46
-22
@@ -58,22 +58,22 @@ const SWEEP_YIELD_EVERY_FILES = 100;
|
||||
* route prefers the live file and falls back to this snapshot.
|
||||
*
|
||||
* Re-snapshots when the source has grown (a live session that gained turns
|
||||
* since the last import). Best-effort and non-fatal.
|
||||
* since the last import). Best-effort and non-fatal: resolves, never rejects.
|
||||
*/
|
||||
function snapshotTranscript(sourceJsonlPath, sessionId) {
|
||||
async function snapshotTranscript(sourceJsonlPath, sessionId) {
|
||||
try {
|
||||
const srcMain = path.resolve(sourceJsonlPath);
|
||||
const snapDir = getTranscriptSnapshotDir();
|
||||
const destMain = path.join(snapDir, `${sessionId}.jsonl`);
|
||||
if (path.resolve(destMain) !== srcMain) {
|
||||
copyIfNewer(srcMain, destMain);
|
||||
await copyIfNewer(srcMain, destMain);
|
||||
}
|
||||
|
||||
// Subagent transcripts live under `<sessionId>/subagents/agent-*.jsonl`.
|
||||
for (const subPath of findSessionSubagents(sourceJsonlPath)) {
|
||||
const destSub = path.join(snapDir, sessionId, "subagents", path.basename(subPath));
|
||||
if (path.resolve(destSub) === path.resolve(subPath)) continue;
|
||||
copyIfNewer(subPath, destSub);
|
||||
await copyIfNewer(subPath, destSub);
|
||||
}
|
||||
|
||||
// Workflow-tool inner-agent transcripts live nested at
|
||||
@@ -83,7 +83,7 @@ function snapshotTranscript(sourceJsonlPath, sessionId) {
|
||||
for (const sub of findSessionWorkflowSubagents(sourceJsonlPath)) {
|
||||
const destSub = path.join(snapDir, sessionId, "subagents", sub.rel);
|
||||
if (path.resolve(destSub) === path.resolve(sub.abs)) continue;
|
||||
copyIfNewer(sub.abs, destSub);
|
||||
await copyIfNewer(sub.abs, destSub);
|
||||
}
|
||||
} catch {
|
||||
/* non-fatal: metadata import already succeeded */
|
||||
@@ -97,45 +97,69 @@ function snapshotTranscript(sourceJsonlPath, sessionId) {
|
||||
* file is a regular file that member owns — `statSync`/`copyFileSync` would
|
||||
* follow a link the member planted to another account's transcript, with the
|
||||
* dashboard's CAP_DAC_READ_SEARCH, into a snapshot the member may then read.
|
||||
*
|
||||
* Asynchronous, 1 MiB per read/write, so a big copy yields to the event loop
|
||||
* between chunks. One copy per `dest` at a time: an overlapping sync joins the
|
||||
* one in flight. Written to a temp file renamed over `dest`, so a reader never
|
||||
* sees half a snapshot. Never rejects — an unexpected failure is logged.
|
||||
*/
|
||||
const SNAPSHOT_MEMBER_MAX = 256 * 1024 * 1024;
|
||||
const ZERO_MB = Buffer.alloc(1 << 20);
|
||||
const copiesInFlight = new Map();
|
||||
function copyIfNewer(src, dest) {
|
||||
let fd;
|
||||
let p = copiesInFlight.get(dest);
|
||||
if (!p) {
|
||||
p = copyNow(src, dest)
|
||||
.catch((err) => console.warn(`[import] snapshot of ${src} failed: ${err.message}`))
|
||||
.finally(() => copiesInFlight.delete(dest));
|
||||
copiesInFlight.set(dest, p);
|
||||
}
|
||||
return p;
|
||||
}
|
||||
|
||||
async function copyNow(src, dest) {
|
||||
const runAs = require("../server/lib/run-as");
|
||||
let fh;
|
||||
try {
|
||||
fd = require("../server/lib/run-as").openSafe(src);
|
||||
fh = await runAs.openSafeAsync(src);
|
||||
} catch {
|
||||
return; // source vanished mid-import, or is not its member's own file
|
||||
}
|
||||
try {
|
||||
const srcSize = fs.fstatSync(fd).size;
|
||||
const srcSize = (await fh.stat()).size;
|
||||
let destSize = -1;
|
||||
try {
|
||||
destSize = fs.statSync(dest).size;
|
||||
destSize = (await fs.promises.stat(dest)).size;
|
||||
} catch {
|
||||
/* dest missing */
|
||||
}
|
||||
if (destSize >= srcSize) return; // snapshot already at least as complete
|
||||
// S3: a member's file is copied into the HOST's disk, so only up to a cap,
|
||||
// and a run of zeros stays a hole — a sparse file costs its member nothing.
|
||||
// ponytail: sync copy of ≤ 256 MiB; go async if member transcripts get that big.
|
||||
if (srcSize > SNAPSHOT_MEMBER_MAX && require("../server/lib/run-as").isMemberFile(src)) return;
|
||||
fs.mkdirSync(path.dirname(dest), { recursive: true });
|
||||
const out = fs.openSync(dest, "w", 0o600);
|
||||
// ponytail: one 1 MiB buffer per copy and copies run one after another per
|
||||
// sync; parallelise across files only if snapshot latency ever matters.
|
||||
if (srcSize > SNAPSHOT_MEMBER_MAX && runAs.isMemberFile(src)) return;
|
||||
await fs.promises.mkdir(path.dirname(dest), { recursive: true });
|
||||
const tmp = `${dest}.${process.pid}.${require("crypto").randomBytes(4).toString("hex")}.tmp`;
|
||||
const out = await fs.promises.open(tmp, "wx", 0o600);
|
||||
try {
|
||||
const buf = Buffer.allocUnsafe(ZERO_MB.length);
|
||||
let pos = 0;
|
||||
for (let n; pos < srcSize; pos += n) {
|
||||
n = fs.readSync(fd, buf, 0, Math.min(buf.length, srcSize - pos), pos);
|
||||
({ bytesRead: n } = await fh.read(buf, 0, Math.min(buf.length, srcSize - pos), pos));
|
||||
if (n <= 0) break;
|
||||
if (!buf.subarray(0, n).equals(ZERO_MB.subarray(0, n))) fs.writeSync(out, buf, 0, n, pos);
|
||||
if (!buf.subarray(0, n).equals(ZERO_MB.subarray(0, n))) await out.write(buf, 0, n, pos);
|
||||
}
|
||||
fs.ftruncateSync(out, pos);
|
||||
} finally {
|
||||
fs.closeSync(out);
|
||||
await out.truncate(pos);
|
||||
await out.close();
|
||||
await fs.promises.rename(tmp, dest);
|
||||
} catch (err) {
|
||||
await out.close().catch(() => {});
|
||||
await fs.promises.rm(tmp, { force: true });
|
||||
throw err;
|
||||
}
|
||||
} finally {
|
||||
fs.closeSync(fd);
|
||||
await fh.close();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1872,7 +1896,7 @@ async function importAllSessions(dbModule) {
|
||||
// Snapshot transcripts into the dashboard's data dir so they outlive
|
||||
// Claude Code's cleanupPeriodDays pruning (default 30 days).
|
||||
for (const session of batch) {
|
||||
snapshotTranscript(session._sourceJsonlPath, session.sessionId);
|
||||
await snapshotTranscript(session._sourceJsonlPath, session.sessionId);
|
||||
}
|
||||
// Link Workflow-tool inner agents to their run. Workflow runs write a
|
||||
// per-run journal + nested agent transcripts that emit NO hooks, so an
|
||||
@@ -2010,7 +2034,7 @@ async function syncDefaultProjects(dbModule, options = {}) {
|
||||
continue;
|
||||
|
||||
const result = importSession(dbModule, session);
|
||||
snapshotTranscript(session._sourceJsonlPath, session.sessionId);
|
||||
await snapshotTranscript(session._sourceJsonlPath, session.sessionId);
|
||||
// A brand-new session is always a real change. An existing one only
|
||||
// counts when importSession actually wrote new events (not skipped).
|
||||
if (!existed || !result.skipped) {
|
||||
@@ -2538,7 +2562,7 @@ async function importFromDirectory(dbModule, rootDir, options = {}) {
|
||||
// outside the DB transaction since it's filesystem I/O.
|
||||
for (const session of parsedSessions) {
|
||||
if (session._sourceJsonlPath) {
|
||||
snapshotTranscript(session._sourceJsonlPath, session.sessionId);
|
||||
await snapshotTranscript(session._sourceJsonlPath, session.sessionId);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user