diff --git a/docs/SERVER-MODE.md b/docs/SERVER-MODE.md index 445bdba9..5a78471d 100644 --- a/docs/SERVER-MODE.md +++ b/docs/SERVER-MODE.md @@ -574,7 +574,9 @@ a pane really is. Each case below is refused or bounded, with a test in a whole member file is read only up to a cap (1 MiB for a login or a settings file, 16 MiB for an index or a workflow journal, 256 MiB for a SQLite copy, 64 MiB otherwise); a member transcript over 256 MiB is neither - parsed nor snapshotted, and the snapshot keeps a sparse file's holes. + parsed nor snapshotted, and the snapshot keeps a sparse file's holes. The + snapshot copy is asynchronous (1 MiB at a time), so a big one never stalls + the dashboard for other users, and it lands by rename — never half-written. - **A request that makes a route throw** answers 500 instead of stopping the server (a member folder of a gone user crashed `POST /api/lanes/worktree`). - **Who a pane runs as.** A non-admin owns no workspace outside their home diff --git a/docs/plans/server-mode/S3-PLAN.md b/docs/plans/server-mode/S3-PLAN.md index 7f6c1a01..9b8de09c 100644 --- a/docs/plans/server-mode/S3-PLAN.md +++ b/docs/plans/server-mode/S3-PLAN.md @@ -272,7 +272,7 @@ Review ran 2026-10-08 (four lenses: root/sudo, cross-member, secrets, local-mode - F39 `execIn` and `write-as` (every sync sudo into a member) stop after 5 s (`MEMBER_EXEC`) — a FIFO `include.path` or a FIFO write target hung git and write-as; - F40 transcripts and journals are read with `readLines` (chunked, a line over 64 MiB is skipped, never built) — readline crashed the server on a one-line 512 MB sparse file (import-history, sessions, codex-usage); - F41 `readFileSafe*` read a member's file only up to a cap (`max`; 1 MiB logins/settings, 16 MiB indexes/journals, 256 MiB SQLite copies, 64 MiB default), and transcript-cache skips a member transcript over 256 MiB; - - F42 the transcript snapshot skips a member file over 256 MiB and writes zero runs as holes — a sparse file was copied out as real blocks into the host's disk; + - F42 the transcript snapshot skips a member file over 256 MiB and writes zero runs as holes — a sparse file was copied out as real blocks into the host's disk (follow-up: the copy is asynchronous, 1 MiB per read/write, so a big one no longer blocks the event loop; one copy per snapshot file at a time, written to a temp file and renamed into place); - F43 an async route's rejection goes to `next(err)` (Express `Layer.handle_request`), and `refuseS3Folder` uses `ownerForRead` — `POST /api/lanes/worktree` with a gone member's folder crashed the server; - F44 a workspace outside the users root may be owned only by an admin (live role: `assertOwnerFits` at create, assign and every spawn), and demoting/disabling a user ends their host panes — a developer-owned host folder started host-account panes; - F45 a pane's cwd must be its workspace's account's (`ECWDACCOUNT`, checked before anything is written for the pane); diff --git a/scripts/import-history.js b/scripts/import-history.js index 5c4c0c26..4aebde97 100644 --- a/scripts/import-history.js +++ b/scripts/import-history.js @@ -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 `/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); } } diff --git a/server/__tests__/server-s3-review3.test.js b/server/__tests__/server-s3-review3.test.js index a4bea6a5..6c494f3e 100644 --- a/server/__tests__/server-s3-review3.test.js +++ b/server/__tests__/server-s3-review3.test.js @@ -229,7 +229,7 @@ describe("a member's planted link is never followed with the dashboard's rights" fs.mkdirSync(path.join(proj, sid, "subagents"), { recursive: true }); fs.writeFileSync(path.join(proj, `${sid}.jsonl`), '{"own":1}\n'); fs.writeFileSync(path.join(proj, sid, "subagents", "agent-own.jsonl"), '{"own":2}\n'); - snapshotTranscript(path.join(proj, `${sid}.jsonl`), sid); + await snapshotTranscript(path.join(proj, `${sid}.jsonl`), sid); const snap = getTranscriptSnapshotDir(); assert.equal(fs.readFileSync(path.join(snap, `${sid}.jsonl`), "utf8"), '{"own":1}\n'); assert.ok(fs.existsSync(path.join(snap, sid, "subagents", "agent-own.jsonl"))); diff --git a/server/__tests__/server-s3-review5.test.js b/server/__tests__/server-s3-review5.test.js index 959342d5..dcc677b3 100644 --- a/server/__tests__/server-s3-review5.test.js +++ b/server/__tests__/server-s3-review5.test.js @@ -7,7 +7,8 @@ * database); a stopped or flooding member tmux server and a hung sudo never * block the dashboard nor blind it to other members' panes; a one-line 512 MB * file, a huge member file and a sparse transcript neither crash the server - * nor fill the host's disk; an async route that throws answers 500; a + * nor fill the host's disk, and the snapshot copy yields to the event loop + * and runs once per file; an async route that throws answers 500; a * non-admin owns no host-folder workspace and a pane's cwd is its * workspace's account's; a pane token works only from its pane's uid; an * analyst's GET path stays in their own folder; a drop goes to the pane's @@ -370,13 +371,83 @@ describe("a member's huge or sparse files", () => { fs.rmSync(t); }); - it("the transcript snapshot never fills the host's disk from a sparse file", () => { + it("the transcript snapshot copy yields to the event loop between chunks", async () => { + const { snapshotTranscript } = require("../../scripts/import-history"); + const { getTranscriptSnapshotDir } = require("../lib/claude-home"); + const proj = path.join(MIA, ".claude", "projects", "-p"); + const sid = "11111111-2222-4333-8444-777777777777"; + const src = path.join(proj, `${sid}.jsonl`); + fs.mkdirSync(proj, { recursive: true }); + fs.writeFileSync(src, Buffer.alloc(64 * 1024 * 1024, 0x61)); + let ticks = 0; + let maxGap = 0; + let last = performance.now(); + let done = false; + const tick = () => { + const now = performance.now(); + maxGap = Math.max(maxGap, now - last); + last = now; + ticks += 1; + if (!done) setImmediate(tick); + }; + setImmediate(tick); + try { + await snapshotTranscript(src, sid); + } finally { + done = true; + } + assert.ok(ticks >= 8, `only ${ticks} event-loop ticks ran during the copy`); + assert.ok(maxGap < 50, `the event loop was blocked for ${maxGap.toFixed(1)} ms`); + const copy = path.join(getTranscriptSnapshotDir(), `${sid}.jsonl`); + assert.ok(fs.readFileSync(copy).equals(fs.readFileSync(src))); + fs.rmSync(src); + fs.rmSync(copy); + }); + + it("two overlapping snapshots copy a file once, and a failed copy only logs", async () => { + const { snapshotTranscript } = require("../../scripts/import-history"); + const { getTranscriptSnapshotDir } = require("../lib/claude-home"); + const proj = path.join(MIA, ".claude", "projects", "-p"); + const sid = "11111111-2222-4333-8444-888888888888"; + const src = path.join(proj, `${sid}.jsonl`); + fs.mkdirSync(path.join(proj, sid, "subagents"), { recursive: true }); + fs.writeFileSync(src, '{"a":1}\n'.repeat(1 << 18)); + fs.writeFileSync(path.join(proj, sid, "subagents", "agent-x.jsonl"), '{"b":1}\n'); + // The subagent's snapshot folder cannot be made: its parent is a file. + fs.mkdirSync(getTranscriptSnapshotDir(), { recursive: true }); + fs.writeFileSync(path.join(getTranscriptSnapshotDir(), sid), ""); + const realRename = fs.promises.rename; + const realWarn = console.warn; + let renames = 0; + const warned = []; + fs.promises.rename = (...a) => { + renames += 1; + return realRename(...a); + }; + console.warn = (...a) => warned.push(a.join(" ")); + try { + await Promise.all([snapshotTranscript(src, sid), snapshotTranscript(src, sid)]); + } finally { + fs.promises.rename = realRename; + console.warn = realWarn; + } + assert.equal(renames, 1); + const copy = path.join(getTranscriptSnapshotDir(), `${sid}.jsonl`); + assert.ok(fs.readFileSync(copy).equals(fs.readFileSync(src))); + assert.ok(warned.some((w) => w.includes("agent-x.jsonl")), JSON.stringify(warned)); + assert.deepEqual( + fs.readdirSync(getTranscriptSnapshotDir()).filter((f) => f.endsWith(".tmp")), + [] + ); + }); + + it("the transcript snapshot never fills the host's disk from a sparse file", async () => { const { snapshotTranscript } = require("../../scripts/import-history"); const { getTranscriptSnapshotDir } = require("../lib/claude-home"); const proj = path.join(MIA, ".claude", "projects", "-p"); const huge = "11111111-2222-4333-8444-555555555555"; sparse(path.join(proj, `${huge}.jsonl`), 300 * 1024 * 1024); - snapshotTranscript(path.join(proj, `${huge}.jsonl`), huge); + await snapshotTranscript(path.join(proj, `${huge}.jsonl`), huge); assert.equal(fs.existsSync(path.join(getTranscriptSnapshotDir(), `${huge}.jsonl`)), false); // Under the cap: copied, and its hole stays a hole. const holey = "11111111-2222-4333-8444-666666666666"; @@ -384,7 +455,7 @@ describe("a member's huge or sparse files", () => { head: '{"a":1}\n', tail: '{"z":1}\n', }); - snapshotTranscript(path.join(proj, `${holey}.jsonl`), holey); + await snapshotTranscript(path.join(proj, `${holey}.jsonl`), holey); const copy = path.join(getTranscriptSnapshotDir(), `${holey}.jsonl`); const st = fs.statSync(copy); assert.equal(st.size, 32 * 1024 * 1024); diff --git a/server/__tests__/workflow-transcript-resolution.test.js b/server/__tests__/workflow-transcript-resolution.test.js index cb039877..d8720dc7 100644 --- a/server/__tests__/workflow-transcript-resolution.test.js +++ b/server/__tests__/workflow-transcript-resolution.test.js @@ -235,7 +235,7 @@ describe("snapshotTranscript (durable nested-layout fallback)", () => { ]); }); - it("preserves the nested layout into the snapshot so the read route resolves it", () => { + it("preserves the nested layout into the snapshot so the read route resolves it", async () => { // Build an independent source session, snapshot it, then resolve from the // snapshot dir exactly as the route's third fallback does. const srcMain = path.join(ENC_DIR, `${SESSION_SNAP}.jsonl`); @@ -250,15 +250,15 @@ describe("snapshotTranscript (durable nested-layout fallback)", () => { writeFile(srcMain, jsonl([{ type: "summary", summary: "snap" }])); writeFile(srcNested, transcriptJsonl(PROMPT_TEXT, LONG_RESULT)); - snapshotTranscript(srcMain, SESSION_SNAP); + await snapshotTranscript(srcMain, SESSION_SNAP); const snapHit = getSnapshotSubagentTranscriptPath(SESSION_SNAP, WF_ID, RUN1); assert.ok(snapHit, "snapshot resolver should find the preserved nested transcript"); assert.equal(fs.readFileSync(snapHit, "utf8"), fs.readFileSync(srcNested, "utf8")); }); - it("does not throw for a session with no transcripts", () => { - assert.doesNotThrow(() => snapshotTranscript(path.join(ENC_DIR, "nope.jsonl"), "nope")); + it("does not throw for a session with no transcripts", async () => { + await assert.doesNotReject(snapshotTranscript(path.join(ENC_DIR, "nope.jsonl"), "nope")); }); });