From 1679bed1a4ad1d5ffc0b71de5982cea18c8ab813 Mon Sep 17 00:00:00 2001 From: Krasimir Kralev <263465593+KrasimirKralev@users.noreply.github.com> Date: Thu, 4 Jun 2026 00:05:28 +0300 Subject: [PATCH] fix(state): write state atomically Fixes #108 and #109. - Replace direct state writes with same-directory atomic temp-file writes. - Preserve existing file modes and create new state files as 0600. - Clean up temp files on failed replacement paths. - Add state/SDK regression coverage plus changelog entry. Proof: - pnpm run typecheck - node --test dist/test/state.test.js dist/test/resume.test.js dist/test/multi_approval_resume.test.js dist/test/approve_preview.test.js - pnpm run lint - pnpm run test - built SDK write/read proof preserved 0600 across replacement - autoreview clean: no accepted/actionable findings Co-authored-by: Krasimir Kralev --- CHANGELOG.md | 1 + src/commands/stdlib/state.ts | 4 +- src/sdk/primitives/state.ts | 44 +++++++++++- src/state/store.ts | 40 ++++++++++- test/state.test.ts | 136 ++++++++++++++++++++++++++++++++++- 5 files changed, 219 insertions(+), 6 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8d3aeeb..6948a2c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,7 @@ All notable changes to Lobster will be documented in this file. ## Unreleased +- Write Lobster state files atomically while preserving restricted file modes, preventing truncated resume/session state after process termination. Thanks to [@KrasimirKralev](https://github.com/KrasimirKralev) (Issues [#108](https://github.com/openclaw/lobster/issues/108), [#109](https://github.com/openclaw/lobster/issues/109), PR [#110](https://github.com/openclaw/lobster/pull/110)). - Fix `timeout_ms` + `retry` so per-attempt timeouts retry as documented while external workflow cancellation still stops immediately. Thanks to [@KrasimirKralev](https://github.com/KrasimirKralev) (PR [#106](https://github.com/openclaw/lobster/pull/106)). ## 2026.5.22 diff --git a/src/commands/stdlib/state.ts b/src/commands/stdlib/state.ts index e92e378..eacfba7 100644 --- a/src/commands/stdlib/state.ts +++ b/src/commands/stdlib/state.ts @@ -1,6 +1,6 @@ import { promises as fsp } from "node:fs"; -import { defaultStateDir, keyToPath } from "../../state/store.js"; +import { defaultStateDir, keyToPath, writeFileAtomic } from "../../state/store.js"; export const stateGetCommand = { name: "state.get", @@ -70,7 +70,7 @@ export const stateSetCommand = { const filePath = keyToPath(stateDir, key); await fsp.mkdir(stateDir, { recursive: true }); - await fsp.writeFile(filePath, JSON.stringify(value, null, 2) + "\n", "utf8"); + await writeFileAtomic(filePath, JSON.stringify(value, null, 2) + "\n"); return { output: asStream([value]) }; }, diff --git a/src/sdk/primitives/state.ts b/src/sdk/primitives/state.ts index 154bd08..c4658db 100644 --- a/src/sdk/primitives/state.ts +++ b/src/sdk/primitives/state.ts @@ -15,10 +15,50 @@ * .pipe(stateSet('my-key')); */ +import { randomBytes } from "node:crypto"; import { promises as fsp } from "node:fs"; import os from "node:os"; import path from "node:path"; +/** + * Write a file atomically (stage to a sibling temp file, fsync, then rename). + * `rename(2)` is atomic on a single filesystem, so a concurrent reader or a + * crash never observes a truncated/partial file. Plain `fsp.writeFile` + * truncates the target up front, leaving a corruption window on SIGKILL/OOM/ + * power loss. New state files are private by default; existing file modes are + * preserved across replacement. Kept local to keep the SDK self-contained. + * @param {string} filePath + * @param {string} data + */ +async function writeFileAtomic(filePath, data) { + const dir = path.dirname(filePath); + const tmpPath = path.join( + dir, + `.${path.basename(filePath)}.${randomBytes(6).toString("hex")}.tmp`, + ); + let mode = 0o600; + let handle; + let cleanup = true; + try { + try { + mode = (await fsp.stat(filePath)).mode & 0o777; + } catch (err) { + if (err?.code !== "ENOENT") throw err; + } + handle = await fsp.open(tmpPath, "wx", mode); + await handle.writeFile(data, "utf8"); + await handle.sync(); + await handle.close(); + handle = undefined; + await fsp.chmod(tmpPath, mode); + await fsp.rename(tmpPath, filePath); + cleanup = false; + } finally { + if (handle) await handle.close().catch(() => {}); + if (cleanup) await fsp.rm(tmpPath, { force: true }).catch(() => {}); + } +} + /** * Get the state directory * @param {Object} ctx @@ -116,7 +156,7 @@ export function stateSet(key) { const filePath = keyToPath(stateDir, key); await fsp.mkdir(stateDir, { recursive: true }); - await fsp.writeFile(filePath, JSON.stringify(value, null, 2) + "\n", "utf8"); + await writeFileAtomic(filePath, JSON.stringify(value, null, 2) + "\n"); // Pass through the value return { @@ -174,5 +214,5 @@ export async function writeState(key, value, ctx = {}) { const filePath = keyToPath(stateDir, key); await fsp.mkdir(stateDir, { recursive: true }); - await fsp.writeFile(filePath, JSON.stringify(value, null, 2) + "\n", "utf8"); + await writeFileAtomic(filePath, JSON.stringify(value, null, 2) + "\n"); } diff --git a/src/state/store.ts b/src/state/store.ts index 9b6410a..8d787d7 100644 --- a/src/state/store.ts +++ b/src/state/store.ts @@ -33,6 +33,44 @@ export function stableStringify(value) { }); } +/** + * Write a file atomically: stage to a sibling temp file, fsync, then rename + * over the target. `rename(2)` is atomic on a single filesystem, so a reader + * (or a crash) never observes a truncated/partial file — it sees either the + * complete old content or the complete new content. Plain `fsp.writeFile` + * truncates the target up front, leaving a corruption window on SIGKILL/OOM/ + * power loss. New state files are private by default; existing file modes are + * preserved across replacement. The temp file is removed on any failed path. + */ +export async function writeFileAtomic(filePath, data) { + const dir = path.dirname(filePath); + const tmpPath = path.join( + dir, + `.${path.basename(filePath)}.${randomBytes(6).toString("hex")}.tmp`, + ); + let mode = 0o600; + let handle; + let cleanup = true; + try { + try { + mode = (await fsp.stat(filePath)).mode & 0o777; + } catch (err) { + if (err?.code !== "ENOENT") throw err; + } + handle = await fsp.open(tmpPath, "wx", mode); + await handle.writeFile(data, "utf8"); + await handle.sync(); + await handle.close(); + handle = undefined; + await fsp.chmod(tmpPath, mode); + await fsp.rename(tmpPath, filePath); + cleanup = false; + } finally { + if (handle) await handle.close().catch(() => {}); + if (cleanup) await fsp.rm(tmpPath, { force: true }).catch(() => {}); + } +} + export async function readStateJson({ env, key }) { const stateDir = defaultStateDir(env); const filePath = keyToPath(stateDir, key); @@ -51,7 +89,7 @@ export async function writeStateJson({ env, key, value }) { const filePath = keyToPath(stateDir, key); await fsp.mkdir(stateDir, { recursive: true }); - await fsp.writeFile(filePath, JSON.stringify(value, null, 2) + "\n", "utf8"); + await writeFileAtomic(filePath, JSON.stringify(value, null, 2) + "\n"); } export async function deleteStateJson({ env, key }) { diff --git a/test/state.test.ts b/test/state.test.ts index ae9e465..455b08b 100644 --- a/test/state.test.ts +++ b/test/state.test.ts @@ -2,9 +2,11 @@ import test from "node:test"; import assert from "node:assert/strict"; import os from "node:os"; import path from "node:path"; -import { mkdtempSync } from "node:fs"; +import { mkdtempSync, promises as fsp } from "node:fs"; import { createDefaultRegistry } from "../src/commands/registry.js"; import { runPipeline } from "../src/runtime.js"; +import { stateSet, readState, writeState } from "../src/sdk/primitives/state.js"; +import { writeStateJson, readStateJson, writeFileAtomic } from "../src/state/store.js"; function streamOf(items) { return (async function* () { @@ -73,3 +75,135 @@ test("state.get returns null for missing key", async () => { assert.deepEqual(output.items, [null]); }); + +// --- Atomic-write behavior proofs (issues #108, #109) --- +// +// Plain fsp.writeFile truncates the target before writing, so a concurrent +// reader (or a crash mid-write) can observe an empty/partial file and fail to +// JSON.parse it. These tests drive many large writes while reading in parallel +// and assert the reader NEVER sees a truncated value. They fail against the +// pre-fix non-atomic writeFile and pass with writeFileAtomic (stage + rename). + +test("writeStateJson is atomic: concurrent reads never observe truncated state (#108)", async () => { + const tmp = mkdtempSync(path.join(os.tmpdir(), "lobster-atomic-store-")); + const env = { LOBSTER_STATE_DIR: tmp }; + const key = "pipeline-resume"; + const payload = "x".repeat(256 * 1024); // large enough that writeFile is not instantaneous + + await writeStateJson({ env, key, value: { payload, n: 0 } }); + + let readErrors = 0; + let partialReads = 0; + const reader = (async () => { + for (let i = 0; i < 500; i++) { + try { + const v = await readStateJson({ env, key }); + if (!v || v.payload !== payload) partialReads++; + } catch { + readErrors++; // JSON.parse on truncated content throws SyntaxError + } + } + })(); + const writer = (async () => { + for (let n = 1; n <= 150; n++) { + await writeStateJson({ env, key, value: { payload, n } }); + } + })(); + await Promise.all([reader, writer]); + + assert.equal(readErrors, 0, "reader must never hit a parse/IO error mid-write"); + assert.equal(partialReads, 0, "reader must never observe truncated/empty state"); + + const leftovers = (await fsp.readdir(tmp)).filter((f) => f.includes(".tmp")); + assert.deepEqual(leftovers, [], "atomic write must not leave temp files behind"); +}); + +test("writeFileAtomic creates private files and preserves existing modes", async () => { + const tmp = mkdtempSync(path.join(os.tmpdir(), "lobster-atomic-mode-")); + const freshPath = path.join(tmp, "fresh.json"); + const existingPath = path.join(tmp, "existing.json"); + + await writeFileAtomic(freshPath, '{"ok":true}\n'); + assert.equal((await fsp.stat(freshPath)).mode & 0o777, 0o600); + + await fsp.writeFile(existingPath, '{"old":true}\n', { mode: 0o640 }); + await fsp.chmod(existingPath, 0o640); + await writeFileAtomic(existingPath, '{"ok":true}\n'); + assert.equal((await fsp.stat(existingPath)).mode & 0o777, 0o640); +}); + +test("writeFileAtomic removes temp files when replacement fails", async () => { + const tmp = mkdtempSync(path.join(os.tmpdir(), "lobster-atomic-cleanup-")); + const targetDir = path.join(tmp, "state.json"); + await fsp.mkdir(targetDir); + + await assert.rejects(() => writeFileAtomic(targetDir, '{"ok":true}\n')); + const leftovers = (await fsp.readdir(tmp)).filter((f) => f.includes(".tmp")); + assert.deepEqual(leftovers, []); +}); + +test("SDK stateSet/readState is atomic under concurrent reads (#109)", async () => { + const tmp = mkdtempSync(path.join(os.tmpdir(), "lobster-atomic-sdk-")); + const ctx = { env: { LOBSTER_STATE_DIR: tmp } }; + const key = "sdk-state"; + const payload = "y".repeat(256 * 1024); + + const writeOnce = async (n: number) => { + const prim = stateSet(key); + const input = (async function* () { + yield { payload, n }; + })(); + const res = await prim.run({ input, ctx }); + for await (const _ of res.output) { + void _; + } + }; + + await writeOnce(0); + + let readErrors = 0; + let partialReads = 0; + const reader = (async () => { + for (let i = 0; i < 500; i++) { + try { + const v = await readState(key, ctx); + if (!v || v.payload !== payload) partialReads++; + } catch { + readErrors++; + } + } + })(); + const writer = (async () => { + for (let n = 1; n <= 120; n++) { + await writeOnce(n); + } + })(); + await Promise.all([reader, writer]); + + assert.equal(readErrors, 0, "SDK reader must never hit a parse/IO error mid-write"); + assert.equal(partialReads, 0, "SDK reader must never observe truncated/empty state"); +}); + +test("SDK writeState preserves restricted state-file mode", async () => { + const tmp = mkdtempSync(path.join(os.tmpdir(), "lobster-atomic-sdk-mode-")); + const ctx = { env: { LOBSTER_STATE_DIR: tmp } }; + const filePath = path.join(tmp, "sdk-state.json"); + await fsp.mkdir(tmp, { recursive: true }); + await fsp.writeFile(filePath, '{"old":true}\n', { mode: 0o600 }); + await fsp.chmod(filePath, 0o600); + + await writeState("sdk-state", { ok: true }, ctx); + + assert.equal((await fsp.stat(filePath)).mode & 0o777, 0o600); + assert.deepEqual(await readState("sdk-state", ctx), { ok: true }); +}); + +test("SDK writeState removes temp files when replacement fails", async () => { + const tmp = mkdtempSync(path.join(os.tmpdir(), "lobster-atomic-sdk-cleanup-")); + const ctx = { env: { LOBSTER_STATE_DIR: tmp } }; + await fsp.mkdir(path.join(tmp, "sdk-state.json")); + + await assert.rejects(() => writeState("sdk-state", { ok: true }, ctx)); + const leftovers = (await fsp.readdir(tmp)).filter((f) => f.includes(".tmp")); + assert.deepEqual(leftovers, []); +});