Atomic metadata writes: unique tmp path per writer so concurrent pty tag / state mutations cant corrupt the file (closes #22) · compounding.tech/pty@260bb67

+3

CHANGELOG.md

··· 2 2 3 3 ## Unreleased 4 4 5 + ### Fixes 6 + - **Concurrent writers can no longer corrupt the session metadata file.** Reported against `pty tag` but the flaw spanned four write sites: `writeMetadata` (sessions.ts), supervisor state save, `pty supervisor reset`, and the event-log retention/truncation path. All four shared the same bug — `target + ".tmp"` is a single fixed path per target, so two concurrent writers truncate-and-overwrite each other's tmp file, and whichever `rename` runs second (or first, depending on timing) can land a half-written JSON blob into the target. Fix: new `atomicWriteFileSync` / `atomicWriteFile` helpers in `sessions.ts` that use a unique tmp path per writer (`<target>.tmp.<pid>.<rand>`), `writeFileSync` into that, then `rename` into place. POSIX same-filesystem `rename` is atomic; readers always see either the old file or the new one, never an intermediate. All four call sites now route through the helper. Event-log truncation (previously `writeFile` in place over a 1000-line rollover) also goes through tmp+rename, so `pty events` readers never see a mid-rewrite file. Contract: last-write-wins under concurrent writers — lost updates are possible, corruption is not. Large (> 4KB) `user.*` / `state.set` event payloads can still interleave at the POSIX `O_APPEND` level; the module documents this and recommends keeping large blobs in state rather than events. 7 + 5 8 ## 0.10.0 (2026-04-24) 6 9 7 10 ### Nesting prevention (client-inside-a-client refusal)

+2 -3

src/cli.ts

··· 22 22 allRefs, 23 23 readMetadata, 24 24 writeMetadata, 25 + atomicWriteFileSync, 25 26 getSessionDir, 26 27 getState, getStateKey, setState, deleteState, listStateKeys, 27 28 type SessionInfo, ··· 2379 2380 state.sessions[name].restartWindowStart = 0; 2380 2381 state.sessions[name].nextBackoffMs = 1000; 2381 2382 state.sessions[name].failed = false; 2382 - const tmp = statePath + ".tmp"; 2383 - fs.writeFileSync(tmp, JSON.stringify(state, null, 2)); 2384 - fs.renameSync(tmp, statePath); 2383 + atomicWriteFileSync(statePath, JSON.stringify(state, null, 2)); 2385 2384 } 2386 2385 } catch {} 2387 2386

+21 -4

src/events.ts

··· 1 1 import * as fs from "node:fs"; 2 2 import * as fsp from "node:fs/promises"; 3 - import { getEventsPath, getSessionDir, ensureSessionDir } from "./sessions.ts"; 3 + import { 4 + getEventsPath, getSessionDir, ensureSessionDir, 5 + atomicWriteFileSync, atomicWriteFile, 6 + } from "./sessions.ts"; 4 7 5 8 export const EventType = { 6 9 BELL: "bell", ··· 196 199 /** Synchronous twin of `appendEvent` — lets synchronous metadata-mutation 197 200 * helpers (setDisplayName, updateTags, setState, deleteState) emit their 198 201 * change events inline without forcing their signatures to go async. 199 - * Uses the same retention path with a sync stat fast-path. */ 202 + * Uses the same retention path with a sync stat fast-path. 203 + * 204 + * Concurrency note: `fs.appendFileSync` issues a single `write()` with 205 + * `O_APPEND`, which POSIX guarantees is atomic for payloads up to 206 + * `PIPE_BUF` bytes (typically 4096 on Linux/macOS). All built-in events 207 + * are well under that. If a caller passes a `user.*` event or 208 + * `state.set` with a > 4KB payload, concurrent appends could interleave 209 + * — keep large payloads out of the event stream and in state. */ 200 210 export function appendEventSync(name: string, event: EventRecord): void { 201 211 ensureSessionDir(); 202 212 const filePath = getEventsPath(name); ··· 212 222 const content = fs.readFileSync(filePath, "utf-8"); 213 223 const lines = content.trimEnd().split("\n"); 214 224 if (lines.length >= MAX_LINES) { 215 - fs.writeFileSync(filePath, lines.slice(-KEEP_LINES).join("\n") + "\n"); 225 + // Tmp+rename so readers never see a half-written rewrite. Concurrent 226 + // appenders after our read but before our rename will have their 227 + // lines lost (their append goes to the old inode that we unlink 228 + // via rename); that's a "concurrent writes overwrite each other" 229 + // case, not corruption — readers always see a valid JSONL file. 230 + atomicWriteFileSync(filePath, lines.slice(-KEEP_LINES).join("\n") + "\n"); 216 231 } 217 232 } catch { 218 233 // File might have been concurrently removed — ignore. ··· 294 309 const content = await fsp.readFile(filePath, "utf-8"); 295 310 const lines = content.trimEnd().split("\n"); 296 311 if (lines.length >= MAX_LINES) { 297 - await fsp.writeFile(filePath, lines.slice(-KEEP_LINES).join("\n") + "\n"); 312 + // Tmp+rename so readers never see a half-written rewrite — same 313 + // reasoning as `maybeTruncateSync`. 314 + await atomicWriteFile(filePath, lines.slice(-KEEP_LINES).join("\n") + "\n"); 298 315 } 299 316 } 300 317

+43 -4

src/sessions.ts

··· 1 1 import * as fs from "node:fs"; 2 + import * as fsp from "node:fs/promises"; 2 3 import * as path from "node:path"; 3 4 import * as os from "node:os"; 4 5 import * as net from "node:net"; ··· 120 121 return status === "exited" || status === "vanished"; 121 122 } 122 123 124 + /** Atomic file publish: write to a unique per-writer tmp file in the 125 + * same directory, then rename over the target. Readers see either 126 + * the old file or the new one, never a half-written intermediate. 127 + * Concurrent writers do NOT coordinate — the last rename wins — but 128 + * they can't corrupt each other's tmp files because each writer uses 129 + * its own unique path. Same-filesystem rename on POSIX is atomic. */ 130 + export function atomicWriteFileSync(target: string, content: string): void { 131 + const tmp = `${target}.tmp.${process.pid}.${randomHex(8)}`; 132 + try { 133 + fs.writeFileSync(tmp, content); 134 + fs.renameSync(tmp, target); 135 + } catch (e) { 136 + // If writeFileSync or renameSync fails, try to clean up the tmp. 137 + // Silent — the original target is still intact either way. 138 + try { fs.unlinkSync(tmp); } catch {} 139 + throw e; 140 + } 141 + } 142 + 143 + /** Async twin of `atomicWriteFileSync` for code paths that are already 144 + * async (EventWriter, etc). Same semantics, same guarantees. */ 145 + export async function atomicWriteFile(target: string, content: string): Promise<void> { 146 + const tmp = `${target}.tmp.${process.pid}.${randomHex(8)}`; 147 + try { 148 + await fsp.writeFile(tmp, content); 149 + await fsp.rename(tmp, target); 150 + } catch (e) { 151 + try { await fsp.unlink(tmp); } catch {} 152 + throw e; 153 + } 154 + } 155 + 156 + function randomHex(bytes: number): string { 157 + // Small inline hex generator — keeping sessions.ts free of a `node:crypto` 158 + // import for this tiny helper. Not cryptographic; just needs low 159 + // collision probability across concurrent writers in the same dir. 160 + let out = ""; 161 + for (let i = 0; i < bytes; i++) out += Math.floor(Math.random() * 256).toString(16).padStart(2, "0"); 162 + return out; 163 + } 164 + 123 165 export function writeMetadata(name: string, metadata: SessionMetadata): void { 124 166 ensureSessionDir(); 125 - const target = getMetadataPath(name); 126 - const tmp = target + ".tmp"; 127 - fs.writeFileSync(tmp, JSON.stringify(metadata, null, 2)); 128 - fs.renameSync(tmp, target); 167 + atomicWriteFileSync(getMetadataPath(name), JSON.stringify(metadata, null, 2)); 129 168 } 130 169 131 170 /** Set or clear the displayName on an existing session. Atomic read-modify-write.

+2 -3

src/supervisor.ts

··· 3 3 import { 4 4 getSessionDir, ensureSessionDir, readMetadata, cleanupAll, 5 5 acquireLock, releaseLock, updateTags, 6 + atomicWriteFileSync, 6 7 type SessionMetadata, 7 8 } from "./sessions.ts"; 8 9 import { spawnDaemon } from "./spawn.ts"; ··· 449 450 }; 450 451 } 451 452 const filePath = path.join(getSupervisorDir(), "state.json"); 452 - const tmp = filePath + ".tmp"; 453 453 try { 454 - fs.writeFileSync(tmp, JSON.stringify(state, null, 2)); 455 - fs.renameSync(tmp, filePath); 454 + atomicWriteFileSync(filePath, JSON.stringify(state, null, 2)); 456 455 } catch { 457 456 // Non-fatal 458 457 }

+322

tests/atomic-writes.test.ts

··· 1 + // Concurrent writer corruption regression. Originally reported against 2 + // `pty tag` — two calls racing on the same `<name>.json.tmp` path could 3 + // leave a corrupted JSON file behind. Same bug class lived in the 4 + // supervisor state save, the `pty supervisor reset` path, and the 5 + // event-log truncation path (which rewrote in place, so readers could 6 + // see a half-written file mid-truncation). 7 + // 8 + // These tests spawn N child processes that race on the same session and 9 + // assert the persisted file is ALWAYS valid JSON / valid JSONL. Last- 10 + // write-wins is acceptable per the fix's stated contract; corruption 11 + // is not. 12 + 13 + import { describe, it, expect, afterEach, afterAll } from "vitest"; 14 + import * as fs from "node:fs"; 15 + import * as os from "node:os"; 16 + import * as path from "node:path"; 17 + import { fileURLToPath } from "node:url"; 18 + import { spawn, spawnSync } from "node:child_process"; 19 + import { 20 + atomicWriteFileSync, atomicWriteFile, 21 + updateTags, setState, readMetadata, 22 + } from "../src/sessions.ts"; 23 + import { appendEventSync, readRecentEvents } from "../src/events.ts"; 24 + 25 + const __dirname = path.dirname(fileURLToPath(import.meta.url)); 26 + const nodeBin = process.execPath; 27 + const cliPath = path.join(__dirname, "..", "dist", "cli.js"); 28 + const serverModule = path.join(__dirname, "..", "dist", "server.js"); 29 + 30 + const testRoot = fs.mkdtempSync(path.join(os.tmpdir(), "pty-atomic-")); 31 + afterAll(() => { 32 + fs.rmSync(testRoot, { recursive: true, force: true, maxRetries: 3, retryDelay: 100 }); 33 + }); 34 + 35 + let bgPids: number[] = []; 36 + let sessionDirs: string[] = []; 37 + 38 + function makeSessionDir(): string { 39 + const dir = fs.mkdtempSync(path.join(testRoot, "d-")); 40 + sessionDirs.push(dir); 41 + return dir; 42 + } 43 + 44 + let nameCounter = 0; 45 + function uniqueName(): string { 46 + return `at${++nameCounter}-${Math.random().toString(36).slice(2, 6)}`; 47 + } 48 + 49 + async function startDaemon(sessionDir: string, name: string): Promise<number> { 50 + const config = JSON.stringify({ 51 + name, command: "cat", args: [], displayCommand: "cat", 52 + cwd: os.tmpdir(), rows: 24, cols: 80, 53 + }); 54 + const child = spawn(nodeBin, [serverModule], { 55 + detached: true, 56 + stdio: ["ignore", "ignore", "pipe"], 57 + env: { ...process.env, PTY_SERVER_CONFIG: config, PTY_SESSION_DIR: sessionDir }, 58 + }); 59 + let stderr = ""; 60 + child.stderr?.on("data", (d: Buffer) => { stderr += d.toString(); }); 61 + let exitCode: number | null = null; 62 + child.on("exit", (code) => { exitCode = code; }); 63 + (child.stderr as any)?.unref?.(); 64 + child.unref(); 65 + const socketPath = path.join(sessionDir, `${name}.sock`); 66 + const start = Date.now(); 67 + while (Date.now() - start < 5000) { 68 + if (exitCode !== null) throw new Error(`Daemon exited: ${stderr}`); 69 + try { 70 + fs.statSync(socketPath); 71 + await new Promise((r) => setTimeout(r, 100)); 72 + bgPids.push(child.pid!); 73 + return child.pid!; 74 + } catch {} 75 + await new Promise((r) => setTimeout(r, 50)); 76 + } 77 + throw new Error("Timeout waiting for daemon"); 78 + } 79 + 80 + afterEach(() => { 81 + for (const pid of bgPids) { try { process.kill(pid, "SIGTERM"); } catch {} } 82 + bgPids = []; 83 + for (const dir of sessionDirs) { 84 + try { 85 + for (const e of fs.readdirSync(dir)) { try { fs.unlinkSync(path.join(dir, e)); } catch {} } 86 + } catch {} 87 + } 88 + sessionDirs = []; 89 + }); 90 + 91 + describe("atomicWriteFileSync / atomicWriteFile helpers", () => { 92 + it("produces a file readable by a concurrent reader throughout a rewrite cycle", async () => { 93 + const dir = makeSessionDir(); 94 + const target = path.join(dir, "test.json"); 95 + atomicWriteFileSync(target, JSON.stringify({ version: 0 })); 96 + 97 + // Writer loop: rewrites the file 200x in quick succession. 98 + // Reader loop: opens and JSON.parses the file; must never fail. 99 + let writerDone = false; 100 + const writerErrors: string[] = []; 101 + const readerErrors: string[] = []; 102 + 103 + const writer = (async () => { 104 + try { 105 + for (let i = 0; i < 200; i++) { 106 + atomicWriteFileSync(target, JSON.stringify({ version: i + 1 })); 107 + await new Promise((r) => setImmediate(r)); 108 + } 109 + } catch (e: any) { writerErrors.push(e.message); } 110 + finally { writerDone = true; } 111 + })(); 112 + 113 + const reader = (async () => { 114 + while (!writerDone) { 115 + try { 116 + const content = fs.readFileSync(target, "utf-8"); 117 + JSON.parse(content); // must not throw 118 + } catch (e: any) { readerErrors.push(e.message); } 119 + await new Promise((r) => setImmediate(r)); 120 + } 121 + })(); 122 + 123 + await Promise.all([writer, reader]); 124 + expect(writerErrors).toEqual([]); 125 + expect(readerErrors).toEqual([]); 126 + }, 15000); 127 + 128 + it("async twin produces the same guarantee under Promise.all writers", async () => { 129 + const dir = makeSessionDir(); 130 + const target = path.join(dir, "test-async.json"); 131 + await atomicWriteFile(target, JSON.stringify({ version: 0 })); 132 + 133 + let writerDone = false; 134 + const readerErrors: string[] = []; 135 + 136 + const writer = (async () => { 137 + for (let i = 0; i < 50; i++) { 138 + await Promise.all( 139 + Array.from({ length: 4 }, (_, k) => 140 + atomicWriteFile(target, JSON.stringify({ batch: i, writer: k })), 141 + ), 142 + ); 143 + } 144 + writerDone = true; 145 + })(); 146 + 147 + const reader = (async () => { 148 + while (!writerDone) { 149 + try { 150 + const content = fs.readFileSync(target, "utf-8"); 151 + JSON.parse(content); 152 + } catch (e: any) { readerErrors.push(e.message); } 153 + await new Promise((r) => setImmediate(r)); 154 + } 155 + })(); 156 + 157 + await Promise.all([writer, reader]); 158 + expect(readerErrors).toEqual([]); 159 + }, 15000); 160 + 161 + it("cleans up tmp files after every successful write (no leftover *.tmp.*)", async () => { 162 + const dir = makeSessionDir(); 163 + const target = path.join(dir, "clean.json"); 164 + for (let i = 0; i < 50; i++) { 165 + atomicWriteFileSync(target, JSON.stringify({ i })); 166 + } 167 + const entries = fs.readdirSync(dir); 168 + const leftover = entries.filter((e) => e.includes(".tmp.")); 169 + expect(leftover).toEqual([]); 170 + }); 171 + }); 172 + 173 + describe("concurrent pty tag via multiple CLI processes", () => { 174 + it("leaves the metadata file valid JSON after 10 racing taggers", async () => { 175 + const dir = makeSessionDir(); 176 + const name = uniqueName(); 177 + await startDaemon(dir, name); 178 + 179 + // Spawn 10 concurrent `pty tag` processes, each setting a distinct key. 180 + const N = 10; 181 + const procs: ReturnType<typeof spawn>[] = []; 182 + for (let i = 0; i < N; i++) { 183 + procs.push(spawn(nodeBin, [cliPath, "tag", name, `k${i}=${i}`], { 184 + env: { ...process.env, PTY_SESSION_DIR: dir }, 185 + stdio: "ignore", 186 + })); 187 + } 188 + await Promise.all(procs.map((p) => new Promise<void>((resolve) => p.on("exit", () => resolve())))); 189 + 190 + // File must be valid JSON. Last-write-wins means we may have lost some 191 + // updates, but the file itself MUST parse. 192 + const metaPath = path.join(dir, `${name}.json`); 193 + const content = fs.readFileSync(metaPath, "utf-8"); 194 + expect(() => JSON.parse(content)).not.toThrow(); 195 + 196 + // No leftover tmp files from crashed or racing writers. 197 + const entries = fs.readdirSync(dir); 198 + const leftover = entries.filter((e) => e.includes(".tmp.")); 199 + expect(leftover).toEqual([]); 200 + }, 30000); 201 + 202 + it("in-process Promise.all of setState calls leaves metadata valid", async () => { 203 + const dir = makeSessionDir(); 204 + const name = uniqueName(); 205 + await startDaemon(dir, name); 206 + process.env.PTY_SESSION_DIR = dir; 207 + 208 + // 50 concurrent setState calls in the same process — stress-tests the 209 + // atomic helper under the sync Promise.all-of-sync-functions pattern. 210 + await Promise.all( 211 + Array.from({ length: 50 }, (_, i) => 212 + Promise.resolve().then(() => setState(name, `k${i}`, i)), 213 + ), 214 + ); 215 + 216 + const meta = readMetadata(name); 217 + expect(meta).not.toBeNull(); 218 + // All 50 should have landed — Node's single-threaded eventloop + sync 219 + // setState means writes serialize, no losses. 220 + expect(Object.keys(meta!.state ?? {}).sort()).toHaveLength(50); 221 + }, 15000); 222 + }); 223 + 224 + describe("event log truncation vs concurrent reader", () => { 225 + it("reader never sees a half-written file during truncation", async () => { 226 + const dir = makeSessionDir(); 227 + const name = uniqueName(); 228 + process.env.PTY_SESSION_DIR = dir; 229 + 230 + // Prime the log well past MAX_LINES so truncation is guaranteed to run. 231 + // emitUserEvent uses validateUserEventType; we go straight to 232 + // appendEventSync via a minimal UserEvent to skip that path. 233 + for (let i = 0; i < 1200; i++) { 234 + appendEventSync(name, { 235 + session: name, 236 + type: `user.prime` as const, 237 + ts: new Date().toISOString(), 238 + data: { i }, 239 + } as any); 240 + } 241 + 242 + // Reader loop: hammer the events file while a second writer loop keeps 243 + // triggering truncation. Reader must ALWAYS see valid JSONL. 244 + let writerDone = false; 245 + const readerErrors: string[] = []; 246 + 247 + const writer = (async () => { 248 + for (let i = 0; i < 500; i++) { 249 + appendEventSync(name, { 250 + session: name, 251 + type: `user.more` as const, 252 + ts: new Date().toISOString(), 253 + data: { i }, 254 + } as any); 255 + if (i % 25 === 0) await new Promise((r) => setImmediate(r)); 256 + } 257 + writerDone = true; 258 + })(); 259 + 260 + const reader = (async () => { 261 + while (!writerDone) { 262 + try { 263 + const events = readRecentEvents(name); 264 + // readRecentEvents drops malformed lines silently, but if the file 265 + // was half-written we'd see a truncated final line missing its 266 + // newline or a line that doesn't JSON.parse. Verify every 267 + // returned event has a `type` field (the weakest postcondition). 268 + for (const e of events) { 269 + if (typeof e.type !== "string") { 270 + readerErrors.push("event without .type"); 271 + break; 272 + } 273 + } 274 + } catch (e: any) { 275 + readerErrors.push(e.message); 276 + } 277 + await new Promise((r) => setImmediate(r)); 278 + } 279 + })(); 280 + 281 + await Promise.all([writer, reader]); 282 + expect(readerErrors).toEqual([]); 283 + 284 + // Final file must also be valid. 285 + const content = fs.readFileSync(path.join(dir, `${name}.events.jsonl`), "utf-8"); 286 + const finalLines = content.trimEnd().split("\n").filter(Boolean); 287 + for (const l of finalLines) { 288 + expect(() => JSON.parse(l)).not.toThrow(); 289 + } 290 + }, 30000); 291 + }); 292 + 293 + describe("concurrent updateTags from separate processes", () => { 294 + it("no corruption when 20 processes race (direct child-process fan-out)", async () => { 295 + const dir = makeSessionDir(); 296 + const name = uniqueName(); 297 + await startDaemon(dir, name); 298 + 299 + const N = 20; 300 + const procs: ReturnType<typeof spawn>[] = []; 301 + for (let i = 0; i < N; i++) { 302 + procs.push(spawn(nodeBin, [cliPath, "tag", name, `race${i}=${i}`], { 303 + env: { ...process.env, PTY_SESSION_DIR: dir }, 304 + stdio: "ignore", 305 + })); 306 + } 307 + await Promise.all(procs.map((p) => new Promise<void>((resolve) => p.on("exit", () => resolve())))); 308 + 309 + const metaPath = path.join(dir, `${name}.json`); 310 + const content = fs.readFileSync(metaPath, "utf-8"); 311 + const parsed = JSON.parse(content); 312 + expect(typeof parsed).toBe("object"); 313 + expect(parsed).not.toBeNull(); 314 + 315 + // Per the contract: last-write-wins, so SOME subset of the 20 keys 316 + // will survive. We can't predict which, but (a) the file parses and 317 + // (b) at least one update landed (otherwise the atomic helper is 318 + // broken in a different way). 319 + const raceKeys = Object.keys(parsed.tags ?? {}).filter((k) => k.startsWith("race")); 320 + expect(raceKeys.length).toBeGreaterThan(0); 321 + }, 30000); 322 + });

Configure Feed

Configure Feed