diff --git a/Dockerfile b/Dockerfile index 19a5542..79eb65e 100644 --- a/Dockerfile +++ b/Dockerfile @@ -138,7 +138,21 @@ COPY --chown=node:node session.bashrc /app/ ENV PORT=7860 \ DATA_DIR=/data \ PUBLIC_DIR=/app/public \ - DISABLE_AUTOUPDATER=1 + DISABLE_AUTOUPDATER=1 \ + UV_THREADPOOL_SIZE=16 + +# Why 16 and not libuv's default 4: /data is a FUSE bucket, and moving the +# re-pin walks off the event loop moved them onto that pool instead — one +# outstanding op per watching pane, and panes launched together stay in phase, +# so their beats land together. Measured on the bucket, one stat on an unrelated +# file while N panes walk (p95 / worst): +# +# pool 4: N=4 0.2ms / 8ms N=8 99.6ms / 177ms +# pool 16: N=4 0.2ms / 5ms N=8 0.3ms / 89ms +# +# At 4 the pool is oversubscribed by the walks and everything else fs-shaped in +# the process queues behind them. The isolated worst-case outliers are the mount +# hiccuping and survive any pool size — it's the p95 that this fixes. # Snapshot the env var NAMES present at build time. HF injects Space secrets and # variables only at runtime, so anything in the runtime env that's absent here diff --git a/server/package.json b/server/package.json index fe66089..01cf4c1 100644 --- a/server/package.json +++ b/server/package.json @@ -14,7 +14,7 @@ "start": "node src/index.js", "dev": "node --watch src/index.js", "test:ui": "node terminal-ui.test.mjs", - "test": "node test/hidden.test.mjs && node test/spawn-group.test.mjs && node test/revive.test.mjs && node test/repin.test.mjs && node test/opencode-resume.test.mjs && node test/terminal-modes.test.mjs && node test/trace-tail.test.mjs && node test/trace-window.test.mjs && node migration.test.mjs && node resize.test.mjs" + "test": "node test/hidden.test.mjs && node test/slowfs.test.mjs && node test/spawn-group.test.mjs && node test/revive.test.mjs && node test/repin.test.mjs && node test/codex-repin.test.mjs && node test/opencode-resume.test.mjs && node test/terminal-modes.test.mjs && node test/trace-tail.test.mjs && node test/trace-window.test.mjs && node migration.test.mjs && node resize.test.mjs" }, "engines": { "node": ">=20.19" diff --git a/server/src/index.js b/server/src/index.js index 381d0e9..3aa0871 100644 --- a/server/src/index.js +++ b/server/src/index.js @@ -36,6 +36,11 @@ import { shareSession, shareNamespace, findTrace, shareAccess, grantAccess, revo importBundle, listBundles, SHAREABLE_CLIS } from './share.js'; import * as backup from './backup.js'; import * as runstate from './runstate.js'; +import { installSlowFsProbe } from './slowfs.js'; + +// Before anything else touches the mount: a sync fs call to /data is ~85ms of +// frozen event loop here, and nothing else in the stack can see it. See slowfs.js. +installSlowFsProbe(); ensureDirs(); refreshVersions(); diff --git a/server/src/runner.js b/server/src/runner.js index 460b287..358f976 100644 --- a/server/src/runner.js +++ b/server/src/runner.js @@ -705,21 +705,35 @@ function codexSessionsRoot() { } // Rollout files touched since `sinceMs`, newest first. -function codexRolloutsSince(sinceMs) { +// +// Async for the same reason as claudeTranscriptsSince, and it is the same bug: +// this is a readdir per day-directory plus a stat per rollout, and although +// CODEX_HOME is on local disk, its `sessions` child is a SYMLINK onto the FUSE +// bucket — `stat -f` says fuseblk, and `find` without -L will tell you the +// directory is empty, which is how this hid. A stat there costs ~85ms, so the +// sync version froze every pane in the Space for 400-750ms on the REPIN_MS beat, +// once per codex session. Measured: stalls >250ms at 2.5/min, worst 3.9s. +// +// Sequential rather than Promise.all, exactly as in claudeTranscriptsSince: +// parallel FUSE stats would saturate the 4-thread libuv pool and push every +// other fs operation in the process behind them, and nothing here is waiting on +// the result. +// Exported for server/test/codex-repin.test.mjs. +export async function codexRolloutsSince(sinceMs) { const out = []; - const walk = (dir, depth) => { + const walk = async (dir, depth) => { if (depth > 5) return; let ents = []; - try { ents = fs.readdirSync(dir, { withFileTypes: true }); } catch { return; } + try { ents = await fsp.readdir(dir, { withFileTypes: true }); } catch { return; } for (const e of ents) { const p = path.join(dir, e.name); - if (e.isDirectory()) walk(p, depth + 1); + if (e.isDirectory()) await walk(p, depth + 1); else if (e.name.startsWith('rollout-') && e.name.endsWith('.jsonl')) { - try { const m = fs.statSync(p).mtimeMs; if (m >= sinceMs) out.push({ p, m }); } catch {} + try { const m = (await fsp.stat(p)).mtimeMs; if (m >= sinceMs) out.push({ p, m }); } catch {} } } }; - walk(codexSessionsRoot(), 0); + await walk(codexSessionsRoot(), 0); return out.sort((a, b) => b.m - a.m); } @@ -763,31 +777,41 @@ async function transcriptHead(p) { } catch { return null; } finally { if (fh) { try { await fh.close(); } catch {} } } } -function firstLine(p) { - const fd = fs.openSync(p, 'r'); +// Async for the same reason as the walk above: these rollouts are on the bucket, +// and reading up to a megabyte of one synchronously blocked the loop carrying +// every session's PTY. Opened INSIDE the try, like transcriptHead: a rollout can +// rotate away between the stat that found it and this open. +async function firstLine(p) { + let fh = null; try { + fh = await fsp.open(p, 'r'); const CHUNK = 65536, MAX = 1024 * 1024; let buf = Buffer.alloc(0); for (let pos = 0; pos < MAX; pos += CHUNK) { const b = Buffer.alloc(CHUNK); - const n = fs.readSync(fd, b, 0, CHUNK, pos); + const { bytesRead: n } = await fh.read(b, 0, CHUNK, pos); buf = Buffer.concat([buf, b.subarray(0, n)]); const nl = buf.indexOf(0x0a); if (nl >= 0) return buf.toString('utf8', 0, nl); if (n < CHUNK) break; // EOF } return buf.toString('utf8'); - } finally { fs.closeSync(fd); } + } finally { if (fh) { try { await fh.close(); } catch {} } } } -function tryCaptureCodexId(sessionId, workdir, sinceMs) { - const claimed = new Set(list().filter((s) => s.id !== sessionId && s.codexSessionId).map((s) => s.codexSessionId)); - const pinned = (list().find((s) => s.id === sessionId) || {}).codexSessionId; - for (const c of codexRolloutsSince(sinceMs)) { +// Read fresh at every use, never snapshotted: this awaits a bucket walk and then +// a file read per candidate, so a claim or a pin taken before those resolve is a +// stale view of the world by the time it is acted on. Same lesson as the claude +// scan — see the note above the re-pin in scheduleClaudeCapture. +async function tryCaptureCodexId(sessionId, workdir, sinceMs, stillOurs = () => true) { + const claimedByOthers = () => + new Set(list().filter((s) => s.id !== sessionId && s.codexSessionId).map((s) => s.codexSessionId)); + const currentPin = () => (list().find((s) => s.id === sessionId) || {}).codexSessionId; + for (const c of await codexRolloutsSince(sinceMs)) { const m = c.p.match(/rollout-.*-([0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})\.jsonl$/); - if (!m || claimed.has(m[1])) continue; + if (!m || claimedByOthers().has(m[1])) continue; let meta; - try { meta = JSON.parse(firstLine(c.p)); } catch { continue; } + try { meta = JSON.parse(await firstLine(c.p)); } catch { continue; } const mp = (meta && meta.payload) || {}; if (!cwdUnderWorkdir(mp.cwd, workdir)) continue; // A sibling's ongoing conversation in the same folder gets fresh writes @@ -799,9 +823,23 @@ function tryCaptureCodexId(sessionId, workdir, sinceMs) { // aren't this agent's conversation, so pinning one would break resume and // the Overview digest. if (mp.thread_source === 'subagent' || (mp.source && mp.source.subagent)) continue; + // Re-read now that the walk and the head read have both resolved: this + // rollout may since have been claimed by the session it actually belongs to, + // and taking it anyway would strand that session on a conversation it cannot + // reclaim. + if (claimedByOthers().has(m[1])) continue; + const pinned = currentPin(); // The watcher re-runs for the life of the pane, so the usual outcome is // "still the same conversation" — don't rewrite sessions.json for that. if (m[1] === pinned) return true; + // Both of these were checked before the walk, and the walk plus the head + // read above have been awaiting the bucket for ~a second since. In that + // window a relaunch spawns a new host whose watch owns the pin from then on, + // and a sibling can go live in this folder — precisely the case + // folderIsShared refuses to guess at. Writing on the strength of the + // pre-walk answer lets a disposed chain overwrite a correct pin or claim a + // sibling's rollout. Same re-check the claude re-pin does before its write. + if (!stillOurs() || folderIsShared(sessionId, workdir, 'codex')) return false; if (pinned) console.warn(`[codex] re-pinning ${sessionId}: ${pinned} -> ${m[1]} (conversation was replaced)`); update(sessionId, { codexSessionId: m[1], codexRollout: c.p }); return true; @@ -809,15 +847,33 @@ function tryCaptureCodexId(sessionId, workdir, sinceMs) { return false; } +// How long a beat waits for its rollout scan before giving up on it and +// rearming. Generous: a cold walk of a real sessions tree runs ~1s, and timing +// one out early would only add a redundant walk, not fix anything. +const SCAN_TIMEOUT_MS = 30_000; + +// Resolves true if `p` is still pending after `ms`, false if it settled first. +// Never rejects and never leaves the process alive on the timer — the caller +// wants to carry on, not to be told about a failure. `p` must already carry its +// own .catch: once the race is lost, nothing else is watching it. +function raceTimeout(p, ms) { + return new Promise((resolve) => { + const t = setTimeout(() => resolve(true), ms); + if (t.unref) t.unref(); + p.then(() => { clearTimeout(t); resolve(false); }, () => { clearTimeout(t); resolve(false); }); + }); +} + // A pin captured before subagents were filtered out (or one whose rollout was // rotated away) may point at a guardian/missing rollout — clear it so we // re-capture the real conversation on this launch. -function pinIsStale(session) { +async function pinIsStale(session) { if (!session.codexSessionId) return false; const p = session.codexRollout; - if (!p || !fs.existsSync(p)) return true; + if (!p) return true; + try { await fsp.stat(p); } catch { return true; } // gone, or the bucket says so try { - const mp = (JSON.parse(firstLine(p)) || {}).payload || {}; + const mp = (JSON.parse(await firstLine(p)) || {}).payload || {}; return mp.thread_source === 'subagent' || !!(mp.source && mp.source.subagent); } catch { return false; } } @@ -828,32 +884,103 @@ function pinIsStale(session) { // for as long as the pane is alive and follow the newest rollout this folder // produces. tryCaptureCodexId only writes when the id actually changes. function scheduleCodexCapture(session, workdir) { - if (session.codexSessionId && pinIsStale(session)) { - session = update(session.id, { codexSessionId: undefined, codexRollout: undefined }) || session; - } const prev = codexCapturing.get(session.id); if (prev) clearTimeout(prev); const since = Date.now() - 2000; + // The host this watch belongs to. A relaunch spawns a new host and a new + // watch, and that one owns the pin from then on — same identity guard the + // claude watcher uses, and it matters for the same reason now that a tick + // awaits and a relaunch can land mid-scan. + const host = hosts.get(session.id); + const stillOurs = () => hosts.get(session.id) === host; let warnedShared = false; + let checkedStale = false; + let scanInFlight = false; + let warnedSlow = false; - const tick = () => { - if (!isRunning(session.id)) { codexCapturing.delete(session.id); return; } + const tick = async () => { + // Clearing a stale pin used to run inline at schedule time, where its + // existsSync + whole-first-line read sat on the launch path. Both touch the + // bucket, so do it on the first beat instead — and read the session fresh, + // since the one passed in was captured before any of this awaited. + // + // Deliberately ABOVE the isRunning gate. Inline it ran once per launch, + // whatever the pane did next; behind the gate a pane that exits inside the + // first beat skipped it, and as this is now the only clear site in the tree + // the bad pin survived to the next launch. Nothing clears this timer on + // exit — only a relaunch does — so the beat still arrives and this still + // runs exactly once, with stillOurs() keeping a relaunch's fresh pin safe. + if (!checkedStale) { + checkedStale = true; + const live = list().find((s) => s.id === session.id); + if (live && live.codexSessionId && await pinIsStale(live) && stillOurs()) { + update(session.id, { codexSessionId: undefined, codexRollout: undefined }); + } + } + if (!isRunning(session.id)) { codexCapturing.delete(session.id); return false; } if (folderIsShared(session.id, workdir, 'codex')) { if (!warnedShared) { warnedShared = true; console.warn(`[codex] ${session.id}: folder shared with another live session — not following thread resets here`); } - } else { - tryCaptureCodexId(session.id, workdir, since); + } else if (stillOurs()) { + // A walk that never comes back must not take the watcher with it. On a + // wedged mount fsp.stat neither resolves nor rejects, so without this the + // tick never settles, `run`'s .then never fires, rearm never runs, and + // this pane silently stops following thread resets for good. + // + // The timeout only lets the BEAT continue — the walk itself cannot be + // cancelled and keeps its libuv thread. So scanInFlight makes sure we + // never stack a second walk on top of a stuck one: with a 4-thread pool, + // one walk per beat on a hung mount would consume the pool within a + // minute and wedge every other fs call in the process. + if (scanInFlight) { + if (!warnedSlow) { + warnedSlow = true; + console.warn(`[codex] ${session.id}: rollout scan still outstanding — skipping this beat`); + } + } else { + scanInFlight = true; + const scan = tryCaptureCodexId(session.id, workdir, since, stillOurs) + .catch((e) => { console.warn(`[codex] ${session.id}: rollout scan failed (${e && e.message})`); }) + .finally(() => { scanInFlight = false; }); + if (await raceTimeout(scan, SCAN_TIMEOUT_MS)) { + console.warn(`[codex] ${session.id}: rollout scan past ${SCAN_TIMEOUT_MS}ms — rearming without it`); + } + } } - const t = setTimeout(tick, REPIN_MS); - if (t.unref) t.unref(); - codexCapturing.set(session.id, t); + return true; }; - const t0 = setTimeout(tick, 5000); // rollout appears ~instantly - if (t0.unref) t0.unref(); - codexCapturing.set(session.id, t0); + // One rearm per tick, whatever the tick did. A tick that throws logs and keeps + // the watcher alive: dropping it would stop following thread resets for the + // rest of the pane's life, which is the failure this mechanism exists for. + const run = () => tick() + .catch((e) => { + console.warn(`[codex] ${session.id}: repin tick failed (${e && e.message}) — retrying next beat`); + return isRunning(session.id); + }) + .then((again) => { if (again) rearm(); else codexCapturing.delete(session.id); }); + + let armed = null; + function rearm(ms = REPIN_MS) { + // Not the pane's watch any more: either a relaunch during an in-flight walk + // started a fresh one — which already owns the map entry, and arming here + // would leave BOTH chains beating with only one reachable by clearTimeout — + // or the pane exited and nothing replaced it. + if (!stillOurs()) { + // Drop OUR entry on the way out; a fired Timeout still retains its + // callback, and with it this closure and the disposed host. Only ours: a + // newer watch's timer has to survive untouched. + if (codexCapturing.get(session.id) === armed) codexCapturing.delete(session.id); + return; + } + armed = setTimeout(run, ms); + if (armed.unref) armed.unref(); + codexCapturing.set(session.id, armed); + } + + rearm(5000); // rollout appears ~instantly } // opencode has no per-conversation handle we can pass on launch, so we can't @@ -1160,10 +1287,14 @@ export function codexRolloutForBreadcrumb(crumb) { // SessionStart permits transcript_path=null. Resolve that case by the exact id // encoded in Codex's rollout filename — deterministic even in a shared folder. -export function codexRolloutForId(id) { +// Async because the walk it uses is: this landed on main while the walk was +// being moved off the event loop, and the two met at the rebase. Keeping it +// synchronous would mean a second sync walk of the whole rollout tree — the +// exact stall this branch exists to remove. +export async function codexRolloutForId(id) { if (!UUID.test(id || '')) return null; const suffix = `-${id}.jsonl`; - return codexRolloutsSince(0).find((item) => path.basename(item.p).endsWith(suffix))?.p || null; + return (await codexRolloutsSince(0)).find((item) => path.basename(item.p).endsWith(suffix))?.p || null; } // Register the SessionStart hook in $CLAUDE_CONFIG_DIR/settings.json. Merge, @@ -1237,7 +1368,7 @@ function exactPinFacts(session, host, workdir, pinField) { // Apply one event from the pane's own adapter. This path never asks which file // or database row is newest: the reported id is the lookup key, and local state // is used only to validate that exact key before persisting it. -function applyBreadcrumb(session, host, workdir, crumb) { +async function applyBreadcrumb(session, host, workdir, crumb) { let facts; let patch; if (session.cli === 'claude') { @@ -1264,7 +1395,13 @@ function applyBreadcrumb(session, host, workdir, crumb) { if (session.cli === 'codex') { const reported = crumb?.payload?.transcript_path; const rollout = codexRolloutForBreadcrumb(crumb) - || (reported == null ? codexRolloutForId(crumb?.payload?.session_id) : null); + || (reported == null ? await codexRolloutForId(crumb?.payload?.session_id) : null); + // That fallback awaits a walk of the rollout tree, and the pane's own exit + // handler calls in here and then deletes the host on the next line. If a + // relaunch got a new host in the meantime, this crumb describes a launch + // that is over and writing it would overwrite the new one's pin — the same + // re-check tryCaptureCodexId and the claude re-pin both make. + if (hosts.get(session.id) !== host) return { repin: null, why: 'relaunched during rollout lookup' }; if (!rollout) return { repin: null, why: reported == null ? 'rollout not available yet' : 'invalid transcript_path', @@ -1300,7 +1437,7 @@ function applyBreadcrumb(session, host, workdir, crumb) { return verdict; } -function consumeBreadcrumb(session, host, workdir, force = false) { +async function consumeBreadcrumb(session, host, workdir, force = false) { const fresh = takeBreadcrumb(session.id, session.cli); let pending = host.pendingExactBreadcrumb; if (fresh) { @@ -1309,7 +1446,7 @@ function consumeBreadcrumb(session, host, workdir, force = false) { } if (!pending || (!fresh && !force && Date.now() < pending.nextAt)) return; try { - const verdict = applyBreadcrumb(session, host, workdir, pending.crumb); + const verdict = await applyBreadcrumb(session, host, workdir, pending.crumb); if (verdict.retry && pending.attempts < BREADCRUMB_RETRIES) { host.pendingExactBreadcrumb = { crumb: pending.crumb, @@ -1336,12 +1473,20 @@ function scheduleBreadcrumbCapture(session, workdir) { if (prev) clearTimeout(prev); const host = hosts.get(session.id); let armed = null; - const tick = () => { + // One rearm per tick, after the beat finishes rather than alongside it: the + // codex fallback inside now awaits a walk of the rollout tree, and arming on + // a fixed interval regardless would stack beats on a slow mount. Same shape + // as the codex and claude watchers. + const tick = async () => { + if (hosts.get(session.id) !== host) { + if (breadcrumbCapturing.get(session.id) === armed) breadcrumbCapturing.delete(session.id); + return; + } + try { await consumeBreadcrumb(session, host, workdir); } catch { /* consumeBreadcrumb logs its own */ } if (hosts.get(session.id) !== host) { if (breadcrumbCapturing.get(session.id) === armed) breadcrumbCapturing.delete(session.id); return; } - consumeBreadcrumb(session, host, workdir); armed = setTimeout(tick, BREADCRUMB_MS); if (armed.unref) armed.unref(); breadcrumbCapturing.set(session.id, armed); @@ -1778,7 +1923,12 @@ export function ensureRunning(session, cols = 120, rows = 34) { // Do one final local read before releasing this launch's nonce. In // particular, `/clear` followed immediately by quit must still persist the // conversation that was on screen when the pane ended. - consumeBreadcrumb(session, host, workdir, true); + // Not awaited — onExit is synchronous and the teardown below must not wait + // on a bucket walk. The write it may perform re-checks host ownership + // first, so landing after the hosts.delete below (or after a relaunch) is + // safe: it either still owns the pin or declines to touch it. + consumeBreadcrumb(session, host, workdir, true) + .catch((e) => console.warn(`[${session.cli}] ${session.id}: final breadcrumb read failed (${e && e.message})`)); hosts.delete(session.id); if (host.gridTimer) { clearTimeout(host.gridTimer); host.gridTimer = null; } if (host.traceHistoryTimer) { clearTimeout(host.traceHistoryTimer); host.traceHistoryTimer = null; } diff --git a/server/src/slowfs.js b/server/src/slowfs.js new file mode 100644 index 0000000..6a14992 --- /dev/null +++ b/server/src/slowfs.js @@ -0,0 +1,75 @@ +// Warn when a synchronous fs call blocks the event loop, and name the caller. +// +// On the Space, DATA_DIR / HOME / CLAUDE_CONFIG_DIR are on the FUSE bucket, +// where a statSync costs ~85ms against 0.01ms on local disk. Node runs JS on one +// thread, so one such call freezes every pane the server is carrying for that +// long. This has bitten us three times now — claudeTranscriptsSince, then +// codexRolloutsSince — and it is hard to spot because **the kernel's pressure +// accounting cannot see it**: FUSE parks the caller in wait_event_interruptible +// (state S), not io_schedule, so the wait is never counted as iowait and +// /proc/pressure/io reads 0.00 while the loop is stuck. Hence a tripwire here. +// +// Silent on a normal filesystem. AM_SLOWFS_MS=0 disables it. +import fs from 'node:fs'; + +const THRESHOLD_MS = Number(process.env.AM_SLOWFS_MS ?? 50); +// One line per call site per window: a bucket stall must not become a log storm. +const REPEAT_MS = 10_000; + +const lastLogged = new Map(); // call site -> { at, suppressed } + +// The first frame outside this file that names real source. Internal callback +// frames ("at Array.forEach ()") point at nothing actionable. +function callSite(skip) { + const holder = {}; + try { Error.captureStackTrace(holder, skip); } catch { return '?'; } + for (const raw of String(holder.stack || '').split('\n').slice(1)) { + const line = raw.trim(); + if (!line.startsWith('at ') || line.includes('slowfs.js') || line.includes('node:')) continue; + if (!line.includes('file:') && !line.includes('/')) continue; + return line.slice(3); + } + return '?'; +} + +export function installSlowFsProbe({ thresholdMs = THRESHOLD_MS } = {}) { + if (!Number.isFinite(thresholdMs) || thresholdMs <= 0) return false; + + // Every *Sync method, rather than a hand-kept list that drifts as callers move. + for (const name of Object.keys(fs).filter((k) => k.endsWith('Sync'))) { + const original = fs[name]; + if (typeof original !== 'function') continue; + + function slowFsWrapper(...args) { + const t0 = performance.now(); + try { + return original.apply(this, args); + } finally { + // This sits in front of every sync fs call in the process, including + // ones inside catch blocks: it must never throw and never swallow. + try { + const ms = performance.now() - t0; + if (ms >= thresholdMs) warn(name, args[0], ms, slowFsWrapper); + } catch { /* diagnostics must not break the app */ } + } + } + // realpathSync.native and friends hang off the function itself. + Object.assign(slowFsWrapper, original); + Object.defineProperty(slowFsWrapper, 'name', { value: name }); + fs[name] = slowFsWrapper; + } + console.warn(`[slowfs] watching synchronous fs; logging calls over ${thresholdMs}ms`); + return true; +} + +function warn(method, target, ms, skip) { + const site = callSite(skip); + const key = `${method} ${site}`; + const prev = lastLogged.get(key); + const now = Date.now(); + if (prev && now - prev.at < REPEAT_MS) { prev.suppressed++; return; } + const extra = prev?.suppressed ? ` (+${prev.suppressed} more since last line)` : ''; + lastLogged.set(key, { at: now, suppressed: 0 }); + // target may be a Buffer, URL or fd; String() covers all three well enough. + console.warn(`[slowfs] ${ms.toFixed(0)}ms ${method} ${String(target)} — at ${site}${extra}`); +} diff --git a/server/test/codex-repin.test.mjs b/server/test/codex-repin.test.mjs new file mode 100644 index 0000000..35e200c --- /dev/null +++ b/server/test/codex-repin.test.mjs @@ -0,0 +1,80 @@ +// The codex rollout walk, after being made async. +// +// It walks $CODEX_HOME/sessions, whose `sessions` child is a symlink onto the +// FUSE bucket on the Space. Synchronously that froze every pane for 400-750ms on +// every REPIN_MS beat, once per codex session. Run with: +// node test/codex-repin.test.mjs +import fs from 'node:fs'; +import path from 'node:path'; +import os from 'node:os'; + +const TMP = fs.mkdtempSync(path.join(os.tmpdir(), 'codex-repin-')); +process.env.CODEX_HOME = path.join(TMP, 'codex-home'); +process.env.DATA_DIR = path.join(TMP, 'data'); +fs.mkdirSync(process.env.DATA_DIR, { recursive: true }); + +const runner = await import('../src/runner.js'); + +let pass = 0, fail = 0; +const check = (name, got, want) => { + const ok = got === want; + ok ? pass++ : fail++; + console.log(` ${ok ? 'PASS' : 'FAIL'} ${name}${ok ? '' : `\n got ${got} want ${want}`}`); +}; + +const NOW = 1770000000000; +const SESSIONS = path.join(process.env.CODEX_HOME, 'sessions'); + +// Codex lays rollouts out under sessions/YYYY/MM/DD/, so the walk has to descend +// three levels before it ever sees a file. +function rollout(uuid, { day = '28', month = '07', mtimeMs }) { + const dir = path.join(SESSIONS, '2026', month, day); + fs.mkdirSync(dir, { recursive: true }); + const p = path.join(dir, `rollout-2026-${month}-${day}T00-00-00-${uuid}.jsonl`); + fs.writeFileSync(p, `${JSON.stringify({ payload: { cwd: '/w', timestamp: '2026-07-28T00:00:00Z' } })}\n`); + fs.utimesSync(p, new Date(mtimeMs), new Date(mtimeMs)); + return p; +} + +const U = (n) => `0000000${n}-0000-4000-8000-000000000000`; + +console.log('an absent root is not an error'); +check('returns empty, does not throw', (await runner.codexRolloutsSince(0)).length, 0); + +console.log('\nfinds rollouts under the YYYY/MM/DD tree'); +const old = rollout(U(1), { mtimeMs: NOW - 60_000 }); +const mid = rollout(U(2), { mtimeMs: NOW - 30_000 }); +const New = rollout(U(3), { month: '08', day: '01', mtimeMs: NOW }); + +let got = await runner.codexRolloutsSince(0); +check('finds all three across months', got.length, 3); +check('newest first', got[0].p, New); +check('oldest last', got[2].p, old); + +console.log('\nfilters on mtime'); +got = await runner.codexRolloutsSince(NOW - 45_000); +check('drops the one older than sinceMs', got.length, 2); +check('keeps the newest', got[0].p, New); +check('keeps the middle', got[1].p, mid); + +console.log('\nignores anything that is not a rollout'); +fs.writeFileSync(path.join(SESSIONS, '2026', '07', '28', 'notes.jsonl'), '{}\n'); +fs.writeFileSync(path.join(SESSIONS, '2026', '07', '28', 'rollout-x.txt'), 'x\n'); +check('still only the rollouts', (await runner.codexRolloutsSince(0)).length, 3); + +console.log('\nis actually async — the whole point of the change'); +const returned = runner.codexRolloutsSince(0); +check('returns a promise', typeof returned.then, 'function'); +await returned; + +// A rollout nested past the depth cap must not be walked forever. +console.log('\nrespects the depth cap'); +let deep = SESSIONS; +for (let i = 0; i < 8; i++) deep = path.join(deep, `d${i}`); +fs.mkdirSync(deep, { recursive: true }); +fs.writeFileSync(path.join(deep, `rollout-deep-${U(9)}.jsonl`), '{}\n'); +check('too-deep rollout ignored', (await runner.codexRolloutsSince(0)).length, 3); + +console.log(`\n${pass} passed, ${fail} failed`); +fs.rmSync(TMP, { recursive: true, force: true }); +process.exit(fail ? 1 : 0); diff --git a/server/test/repin.test.mjs b/server/test/repin.test.mjs index 93be9a9..ab7b8d4 100644 --- a/server/test/repin.test.mjs +++ b/server/test/repin.test.mjs @@ -169,7 +169,8 @@ check('Codex exact id accepted', runner.breadcrumbVerdict(codexCrumb, 's1', { cli: 'codex', runId: RUN, workdir: WORKDIR, pinned: null, claimed: new Set(), pidTrusted: true, }).repin, CODEX_ID); check('Codex rollout path retained', runner.codexRolloutForBreadcrumb(codexCrumb), rollout); -check('Codex null transcript resolves by exact id', runner.codexRolloutForId(CODEX_ID), rollout); +// Awaited: the id lookup walks the rollout tree, which is async now. +check('Codex null transcript resolves by exact id', await runner.codexRolloutForId(CODEX_ID), rollout); const CODEX_ALIAS = path.join(TMP, 'codex-alias'); fs.mkdirSync(CODEX_ALIAS); fs.symlinkSync(path.join(CODEX, 'sessions'), path.join(CODEX_ALIAS, 'sessions')); diff --git a/server/test/slowfs.test.mjs b/server/test/slowfs.test.mjs new file mode 100644 index 0000000..1522883 --- /dev/null +++ b/server/test/slowfs.test.mjs @@ -0,0 +1,69 @@ +// The sync-fs tripwire: it must name the caller of a slow call without ever +// changing what that call does. Run with: node test/slowfs.test.mjs +// +// Timing is deliberately not asserted on — a test needing a real 50ms stat would +// want a FUSE mount or a sleep, and both are flaky. The threshold drops to ~0 +// instead, so every call is "slow" and the same path runs: wrap, time, attribute. +import fs from 'node:fs'; +import path from 'node:path'; +import os from 'node:os'; + +const TMP = fs.mkdtempSync(path.join(os.tmpdir(), 'slowfs-')); +const FILE = path.join(TMP, 'hello.txt'); +fs.writeFileSync(FILE, 'contents\n'); + +let pass = 0, fail = 0; +const check = (name, got, want) => { + const ok = got === want; + ok ? pass++ : fail++; + console.log(` ${ok ? 'PASS' : 'FAIL'} ${name}${ok ? '' : `\n got ${got} want ${want}`}`); +}; + +const { installSlowFsProbe } = await import('../src/slowfs.js'); + +console.log('disabled unless a positive threshold is set'); +const untouched = fs.statSync; +check('0 disables', installSlowFsProbe({ thresholdMs: 0 }), false); +check('negative disables', installSlowFsProbe({ thresholdMs: -1 }), false); +check('NaN disables', installSlowFsProbe({ thresholdMs: Number('nope') }), false); +check('fs left unwrapped when disabled', fs.statSync, untouched); + +// Capture the warnings rather than exporting state from the module purely to +// be tested: the log line IS the product here. +const logs = []; +const realWarn = console.warn; +console.warn = (...a) => { logs.push(a.join(' ')); }; + +check('installs', installSlowFsProbe({ thresholdMs: 0.0000001 }), true); + +console.log('\ninstalled: every sync call still behaves exactly as before'); +check('readFileSync returns content', fs.readFileSync(FILE, 'utf8'), 'contents\n'); +check('existsSync true', fs.existsSync(FILE), true); +check('existsSync false', fs.existsSync(path.join(TMP, 'nope')), false); +check('statSync size', fs.statSync(FILE).size, 9); +check('readdirSync finds the file', fs.readdirSync(TMP).includes('hello.txt'), true); +// Wrapping replaces the function object, so properties hanging off it (notably +// realpathSync.native) would vanish without the Object.assign. +check('realpathSync.native survives wrapping', typeof fs.realpathSync.native, 'function'); + +// A wrapper that swallowed errors would be worse than the bug it hunts. +let threw = null; +try { fs.readFileSync(path.join(TMP, 'missing'), 'utf8'); } catch (e) { threw = e.code; } +check('errors still propagate', threw, 'ENOENT'); + +console.log('\nit attributes the call to the caller, not to itself'); +const joined = logs.join('\n'); +check('something was logged', logs.length > 0, true); +check('names this test file', joined.includes('slowfs.test.mjs'), true); +check('does not blame slowfs.js', joined.includes('slowfs.js'), false); +check('names the method', joined.includes('readFileSync'), true); + +console.log('\na stalled mount must not become a log storm'); +logs.length = 0; +for (let i = 0; i < 5; i++) fs.existsSync(FILE); // one line => one call site +check('repeats collapse to a single line', logs.filter((l) => l.includes('existsSync')).length, 1); + +console.warn = realWarn; +console.log(`\n${pass} passed, ${fail} failed`); +fs.rmSync(TMP, { recursive: true, force: true }); +process.exit(fail ? 1 : 0);