diff --git a/docs/LEDGER.md b/docs/LEDGER.md new file mode 100644 index 0000000..3ba04ae --- /dev/null +++ b/docs/LEDGER.md @@ -0,0 +1,96 @@ +# Durable Request Ledger + +Append-only, single-host request ledger for idempotent work with crash recovery. + +## State machine + +``` +ACCEPTED → RUNNING → SUCCEEDED + → FAILED + → UNKNOWN → SUCCEEDED (via reconcile only) +``` + +- Terminal states `SUCCEEDED` and `FAILED` are immutable. +- `UNKNOWN` means the process may have already performed side effects; do not auto-retry. +- `reconcile(id, result)` is the only path from `UNKNOWN` to `SUCCEEDED` and requires a receipt (`result !== undefined`). + +## Durability guarantees and limits + +- Each event is one JSON line in `ledger.jsonl` with monotonic `seq` and a `crc` (first 16 hex chars of SHA-256 over the event body without `crc`). +- With `fsync: true` (default), an append calls `handle.sync()` before memory is updated and the call resolves. A failed write leaves memory unchanged. +- With `fsync: false`, a sudden power loss may drop the most recent events. +- Designed for a **local filesystem on a single host**. Network filesystems are not recommended (lock and `rename` semantics vary). +- Payload bodies are not stored; only a SHA-256 fingerprint is kept. +- Line size cap: 256 KiB. `result` / `error` JSON size cap: 64 KiB (`LedgerLimitError`, no silent truncation). + +## Locking + +- `Ledger.open(dir)` creates a `ledger.lock/` directory with non-recursive `fs.mkdir` (atomic exclusive create) and writes `owner.json` as `{ pid, token, startedAt }`. +- While the event loop is alive, an `unref` timer every `staleMs / 2` refreshes the lock directory mtime via `fs.utimes`. Default `staleMs` is 15000 (overridable for tests). +- Another process reclaims only when: + 1. `process.kill(owner.pid, 0)` yields `ESRCH` **and** mtime age is greater than **`staleMs / 2`** (at least one missed heartbeat; avoids false `ESRCH` stealing a freshly heartbeated lock); or + 2. the owner still appears alive (success or `EPERM`) **and** mtime is older than **`4 × staleMs`** (margin for heartbeat delay under load). +- Reclaim remembers the inspected `owner.json` token, `rename`s to `lockDir + '.stale-' + `, and **deletes only if the moved directory’s token still matches**. If it differs (another process already installed a new lock that was renamed by mistake), restore when `ledger.lock` is empty, otherwise leave the moved directory in place and throw `LedgerLockedError` (never delete a foreign lock). +- `close()` stops the heartbeat and removes the lock directory only when `owner.json` token matches. +- If the heartbeat finds the directory gone or a foreign token, the instance is marked compromised and further writes throw `LedgerLockedError('lock lost')`. + +### Lock limits + +- If the owning process blocks the event loop longer than **`4 × staleMs`** (60s with the default), or heartbeat delay pushes mtime past the reclaim threshold while the pid still looks alive, another process may reclaim the lock. The displaced owner becomes **compromised** on the next heartbeat/write and subsequent writes are rejected. +- After a crash: reclaim proceeds once `ESRCH` is observed and mtime shows a missed heartbeat (`> staleMs / 2`). If the pid was reused by an unrelated live process, wait up to **`4 × staleMs`** for mtime-based reclaim. +- A theoretical race where reclaim overlaps the owner’s last durable write is not prevented by distributed consensus; append `seq`/`crc` replay can **detect** torn history but does not prevent the overlap. This ledger assumes a **single-host local filesystem**, not a networked multi-writer store. +- Single-host local filesystem only; network filesystems are not recommended. + +## RequestCoordinator + ledger + +- `new RequestCoordinator()` (no ledger) keeps the v0.1.0 in-memory API: `reconcile` is synchronous and returns `undefined`. +- `new RequestCoordinator({ ledger })` is durable: `reconcile(id, result)` returns a **Promise** and updates in-memory state only after the ledger write succeeds (failure rejects and leaves memory unchanged). +- If `execute` succeeds but persisting `succeed` fails, `run` throws `LedgerWriteError` (does not return the execute result). After restart that request becomes `UNKNOWN`. + +## Recovery on open + +1. Replay all lines; verify `seq` continuity and `crc`. +2. If the **last** line is incomplete (JSON parse failure from a partial write), truncate back to the last good byte offset and set `ledger.recovery = { truncatedBytes }`. +3. If a **middle** line is broken, or any line has a bad `crc` / `seq` gap, throw `LedgerCorruptError` (no silent skip). +4. Requests left in `RUNNING` are moved to `UNKNOWN` via a `recovered` event (side effects may already have happened). +5. `ACCEPTED` requests stay `ACCEPTED`. + +## API sketch + +```js +import { Ledger } from '../src/ledger.mjs'; +import { RequestCoordinator } from '../src/request-coordinator.mjs'; + +const ledger = await Ledger.open(dir, { fsync: true }); +const coord = new RequestCoordinator({ ledger }); + +const result = await coord.run('job-1', { op: 'create' }, async (payload) => { + return { ok: true }; +}); + +// After a crash that left UNKNOWN (durable reconcile is async): +// await coord.reconcile('job-1', { ok: true }); + +await ledger.close(); +``` + +Without `{ ledger }`, `RequestCoordinator` stays in-memory and matches the v0.1.0 API. + +## Optional compact + +`compact()` rewrites the log to the minimal event sequence for current records using a temp file, `fsync`, and atomic `rename`, while the lock is held. + +## Limits (summary) + +| Topic | Behavior | +| --- | --- | +| Power loss + `fsync: false` | Recent events may be lost | +| Network FS | Not recommended | +| `UNKNOWN` | Manual `reconcile` only; no automatic retry | +| Multi-process | One writer via `ledger.lock/` + heartbeat | +| Blocked event loop > `staleMs` | Lock may be reclaimed | +| Privacy | Fingerprints only; no raw payloads on disk | + +## Korean summary (optional) + +추가 전용 JSONL 원장으로 요청 상태를 내구성 있게 기록합니다. 잠금은 `ledger.lock/` 디렉터리와 mtime 하트비트이며, 산 프로세스는 `4×staleMs` 이후, ESRCH는 하트비트 1회 분(`staleMs/2`) 초과 시에만 인수합니다. stale rename은 owner token이 같을 때만 삭제합니다. diff --git a/package.json b/package.json index b9cabd7..bc2cfda 100644 --- a/package.json +++ b/package.json @@ -9,5 +9,18 @@ }, "scripts": { "test": "node --test" + }, + "files": [ + "src", + "docs", + "examples", + "README.md", + "LICENSE", + "SECURITY.md", + "CONTRIBUTING.md" + ], + "exports": { + "./ledger": "./src/ledger.mjs", + "./request-coordinator": "./src/request-coordinator.mjs" } } diff --git a/src/ledger.mjs b/src/ledger.mjs new file mode 100644 index 0000000..29678df --- /dev/null +++ b/src/ledger.mjs @@ -0,0 +1,622 @@ +import { createHash, randomBytes } from 'node:crypto'; +import path from 'node:path'; +import { RequestConflictError } from './request-coordinator.mjs'; + +const MAX_LINE = 256 * 1024; +const MAX_BLOB = 64 * 1024; +const DEFAULT_STALE_MS = 15_000; + +export class LedgerStateError extends Error { + constructor(message) { + super(message); + this.name = 'LedgerStateError'; + } +} +export class LedgerCorruptError extends Error { + constructor(message) { + super(message); + this.name = 'LedgerCorruptError'; + } +} +export class LedgerLockedError extends Error { + constructor(message = 'Ledger is locked by another process') { + super(message); + this.name = 'LedgerLockedError'; + } +} +export class LedgerLimitError extends Error { + constructor(message) { + super(message); + this.name = 'LedgerLimitError'; + } +} + +function fingerprint(payload) { + return createHash('sha256').update(JSON.stringify(payload)).digest('hex'); +} + +function eventCrc(body) { + return createHash('sha256').update(JSON.stringify(body)).digest('hex').slice(0, 16); +} + +function sleep(ms) { + return new Promise((r) => setTimeout(r, ms)); +} + +/** True only when process.kill(pid, 0) reports ESRCH (process is gone). */ +function pidDefinitelyDead(pid) { + if (!Number.isInteger(pid) || pid <= 0) return true; + try { + process.kill(pid, 0); + return false; + } catch (e) { + return !!(e && e.code === 'ESRCH'); + } +} + +/** Live (or EPERM) owners are stale only after this multiple of staleMs. */ +const STALE_LIVE_MULTIPLIER = 4; + +function cloneRecord(r) { + if (!r) return null; + const out = { + id: r.id, + state: r.state, + fingerprint: r.fingerprint, + seq: r.seq, + updatedAt: r.updatedAt, + }; + if (r.result !== undefined) { + out.result = structuredClone + ? structuredClone(r.result) + : JSON.parse(JSON.stringify(r.result)); + } + if (r.error) out.error = { ...r.error }; + if (r.reason !== undefined) out.reason = r.reason; + return out; +} + +function serializeBlob(value, label) { + let text; + try { + text = JSON.stringify(value); + } catch { + throw new LedgerLimitError(`${label} is not JSON-serializable`); + } + if (text === undefined) throw new TypeError(`${label} required`); + if (Buffer.byteLength(text, 'utf8') > MAX_BLOB) { + throw new LedgerLimitError(`${label} exceeds ${MAX_BLOB} bytes`); + } + return value; +} + +function errorShape(error) { + const out = { + name: String(error?.name || 'Error'), + message: String(error?.message || error || 'Error'), + }; + if (error && error.code !== undefined) out.code = error.code; + serializeBlob(out, 'error'); + return out; +} + +/** + * Append-only durable request ledger (local filesystem). + * Single-host local disk recommended; network filesystems are not advised. + */ +export class Ledger { + #dir; + #file; + #lockDir; + #token; + #fsync; + #fs; + #staleMs; + #heartbeat = null; + #compromised = false; + #index = new Map(); + #seq = 0; + #queue = Promise.resolve(); + #closed = false; + recovery = null; + + constructor(dir, { fsync, token, fs, recovery, staleMs }) { + this.#dir = dir; + this.#file = path.join(dir, 'ledger.jsonl'); + this.#lockDir = path.join(dir, 'ledger.lock'); + this.#fsync = fsync; + this.#token = token; + this.#fs = fs; + this.#staleMs = staleMs; + this.recovery = recovery; + } + + static async open(dir, { + fsync = true, + lockTimeoutMs = 0, + staleMs = DEFAULT_STALE_MS, + deps = {}, + } = {}) { + const fs = deps.fs || await import('node:fs/promises'); + await fs.mkdir(dir, { recursive: true }); + const token = await acquireLock(dir, { + lockTimeoutMs, + staleMs, + beforeStaleRename: deps.beforeStaleRename, + }, fs); + const ledger = new Ledger(dir, { fsync, token, fs, recovery: null, staleMs }); + ledger.#startHeartbeat(); + try { + await ledger.#replay(); + return ledger; + } catch (e) { + try { + await ledger.close(); + } catch { + /* ignore */ + } + throw e; + } + } + + get(id) { + return cloneRecord(this.#index.get(id)); + } + + list({ state } = {}) { + let rows = [...this.#index.values()]; + if (state) rows = rows.filter((r) => r.state === state); + return rows.map(cloneRecord); + } + + accept(id, payload) { + return this.#enqueued(async () => { + this.#ensureWritable(); + if (typeof id !== 'string' || !id.trim()) throw new TypeError('Non-empty request ID required'); + const fp = fingerprint(payload); + const prior = this.#index.get(id); + if (prior) { + if (prior.fingerprint !== fp) throw new RequestConflictError(id); + return { created: false, record: cloneRecord(prior) }; + } + const record = await this.#appendAndApply({ + type: 'accept', + id, + fingerprint: fp, + }); + return { created: true, record }; + }); + } + + markRunning(id) { + return this.#enqueued(async () => { + this.#ensureWritable(); + const cur = this.#require(id); + if (cur.state !== 'ACCEPTED') throw new LedgerStateError(`Cannot markRunning from ${cur.state}`); + return this.#appendAndApply({ type: 'running', id }); + }); + } + + succeed(id, result) { + return this.#enqueued(async () => { + this.#ensureWritable(); + if (result === undefined) throw new TypeError('Receipt required'); + const cur = this.#require(id); + if (cur.state !== 'RUNNING') throw new LedgerStateError(`Cannot succeed from ${cur.state}`); + const safe = serializeBlob(result, 'result'); + return this.#appendAndApply({ type: 'succeed', id, result: safe }); + }); + } + + fail(id, error) { + return this.#enqueued(async () => { + this.#ensureWritable(); + const cur = this.#require(id); + if (cur.state !== 'RUNNING') throw new LedgerStateError(`Cannot fail from ${cur.state}`); + return this.#appendAndApply({ type: 'fail', id, error: errorShape(error) }); + }); + } + + markUnknown(id, reason) { + return this.#enqueued(async () => { + this.#ensureWritable(); + const cur = this.#require(id); + if (cur.state !== 'RUNNING') throw new LedgerStateError(`Cannot markUnknown from ${cur.state}`); + return this.#appendAndApply({ type: 'unknown', id, reason: String(reason ?? 'unknown') }); + }); + } + + reconcile(id, result) { + return this.#enqueued(async () => { + this.#ensureWritable(); + if (result === undefined) throw new TypeError('Receipt required'); + const cur = this.#require(id); + if (cur.state !== 'UNKNOWN') throw new LedgerStateError(`Cannot reconcile from ${cur.state}`); + const safe = serializeBlob(result, 'result'); + return this.#appendAndApply({ type: 'reconcile', id, result: safe }); + }); + } + + async compact() { + return this.#enqueued(async () => { + this.#ensureWritable(); + const events = []; + let seq = 0; + for (const rec of this.#index.values()) { + seq += 1; + events.push(this.#buildEvent({ type: 'accept', id: rec.id, fingerprint: rec.fingerprint }, seq)); + if (rec.state === 'ACCEPTED') continue; + seq += 1; + events.push(this.#buildEvent({ type: 'running', id: rec.id }, seq)); + if (rec.state === 'RUNNING') continue; + seq += 1; + if (rec.state === 'SUCCEEDED') { + events.push(this.#buildEvent({ type: 'succeed', id: rec.id, result: rec.result }, seq)); + } else if (rec.state === 'FAILED') { + events.push(this.#buildEvent({ type: 'fail', id: rec.id, error: rec.error }, seq)); + } else if (rec.state === 'UNKNOWN') { + events.push(this.#buildEvent({ type: 'unknown', id: rec.id, reason: rec.reason || 'unknown' }, seq)); + } + } + const body = events.map((e) => JSON.stringify(e)).join('\n') + (events.length ? '\n' : ''); + const tmp = path.join(this.#dir, `ledger.${this.#token}.tmp`); + const handle = await this.#fs.open(tmp, 'w'); + try { + await handle.writeFile(body, 'utf8'); + if (this.#fsync) await handle.sync(); + } finally { + await handle.close(); + } + await this.#fs.rename(tmp, this.#file); + this.#seq = seq; + return this.list({}); + }); + } + + async close() { + if (this.#closed) return; + await this.#queue.catch(() => {}); + this.#closed = true; + if (this.#heartbeat) { + clearInterval(this.#heartbeat); + this.#heartbeat = null; + } + try { + const raw = await this.#fs.readFile(path.join(this.#lockDir, 'owner.json'), 'utf8'); + const holder = JSON.parse(raw); + if (holder.token === this.#token) { + await this.#fs.rm(this.#lockDir, { recursive: true, force: true }); + } + } catch { + /* ignore */ + } + } + + #startHeartbeat() { + const interval = Math.max(50, Math.floor(this.#staleMs / 2)); + this.#heartbeat = setInterval(() => { + this.#touchLock().catch(() => { + this.#compromised = true; + }); + }, interval); + this.#heartbeat.unref(); + } + + async #touchLock() { + if (this.#closed || this.#compromised) return; + try { + const ownerPath = path.join(this.#lockDir, 'owner.json'); + const raw = await this.#fs.readFile(ownerPath, 'utf8'); + const holder = JSON.parse(raw); + if (holder.token !== this.#token) { + this.#compromised = true; + return; + } + const now = new Date(); + await this.#fs.utimes(this.#lockDir, now, now); + } catch { + this.#compromised = true; + } + } + + #ensureWritable() { + if (this.#closed) throw new Error('Ledger is closed'); + if (this.#compromised) throw new LedgerLockedError('lock lost'); + } + + #require(id) { + const cur = this.#index.get(id); + if (!cur) throw new LedgerStateError(`Unknown request id: ${id}`); + return cur; + } + + #enqueued(fn) { + const run = this.#queue.then(fn, fn); + this.#queue = run.then(() => {}, () => {}); + return run; + } + + #buildEvent(fields, seq = this.#seq + 1) { + const body = { + v: 1, + seq, + ts: new Date().toISOString(), + ...fields, + }; + const crc = eventCrc(body); + const event = { ...body, crc }; + const line = JSON.stringify(event); + if (Buffer.byteLength(line, 'utf8') > MAX_LINE) { + throw new LedgerLimitError(`event line exceeds ${MAX_LINE} bytes`); + } + return event; + } + + async #appendAndApply(fields) { + const event = this.#buildEvent(fields); + await this.#writeLine(event); + this.#seq = event.seq; + this.#apply(event); + return cloneRecord(this.#index.get(event.id)); + } + + async #writeLine(event) { + const line = JSON.stringify(event) + '\n'; + const handle = await this.#fs.open(this.#file, 'a'); + try { + await handle.writeFile(line, 'utf8'); + if (this.#fsync) await handle.sync(); + } finally { + await handle.close(); + } + } + + #apply(event) { + const now = event.ts; + if (event.type === 'accept') { + this.#index.set(event.id, { + id: event.id, + state: 'ACCEPTED', + fingerprint: event.fingerprint, + seq: event.seq, + updatedAt: now, + }); + return; + } + const cur = this.#index.get(event.id); + if (!cur) throw new LedgerCorruptError(`event for missing id ${event.id}`); + if (event.type === 'running') { + cur.state = 'RUNNING'; + } else if (event.type === 'succeed' || event.type === 'reconcile') { + cur.state = 'SUCCEEDED'; + cur.result = event.result; + delete cur.error; + delete cur.reason; + } else if (event.type === 'fail') { + cur.state = 'FAILED'; + cur.error = event.error; + delete cur.result; + delete cur.reason; + } else if (event.type === 'unknown' || event.type === 'recovered') { + cur.state = 'UNKNOWN'; + cur.reason = event.reason || 'recovered'; + delete cur.result; + } else { + throw new LedgerCorruptError(`unknown event type ${event.type}`); + } + cur.seq = event.seq; + cur.updatedAt = now; + } + + async #replay() { + let text = ''; + try { + text = await this.#fs.readFile(this.#file, 'utf8'); + } catch (e) { + if (e && e.code === 'ENOENT') { + this.#seq = 0; + return; + } + throw e; + } + if (!text) { + this.#seq = 0; + return; + } + const rawLines = text.split('\n'); + const lines = []; + for (let i = 0; i < rawLines.length; i++) { + const line = rawLines[i]; + if (i === rawLines.length - 1 && line === '') continue; + lines.push(line); + } + + let expected = 1; + let goodEnd = 0; + let consumed = 0; + for (let i = 0; i < lines.length; i++) { + const line = lines[i]; + const isLast = i === lines.length - 1; + let event; + try { + event = JSON.parse(line); + } catch (e) { + if (isLast) { + this.recovery = { truncatedBytes: text.length - goodEnd }; + await this.#fs.truncate(this.#file, goodEnd); + break; + } + throw new LedgerCorruptError(`corrupt ledger at line ${i + 1}: ${e.message}`); + } + try { + if (!event || typeof event !== 'object') throw new Error('not object'); + const { crc, ...rest } = event; + if (crc !== eventCrc(rest)) throw new Error('bad crc'); + if (event.seq !== expected) throw new Error(`seq gap want ${expected} got ${event.seq}`); + } catch (e) { + throw new LedgerCorruptError(`corrupt ledger at line ${i + 1}: ${e.message}`); + } + this.#apply(event); + this.#seq = event.seq; + expected += 1; + consumed += Buffer.byteLength(line, 'utf8') + 1; + goodEnd = consumed; + } + + const running = [...this.#index.values()].filter((r) => r.state === 'RUNNING'); + for (const rec of running) { + await this.#appendAndApply({ type: 'recovered', id: rec.id, reason: 'interrupted' }); + } + } +} + +async function readOwner(lockDir, fs) { + try { + return JSON.parse(await fs.readFile(path.join(lockDir, 'owner.json'), 'utf8')); + } catch { + return null; + } +} + +async function createLockDir(lockDir, token, fs) { + await fs.mkdir(lockDir); + const owner = { + pid: process.pid, + token, + startedAt: new Date().toISOString(), + }; + await fs.writeFile(path.join(lockDir, 'owner.json'), JSON.stringify(owner), 'utf8'); + const now = new Date(); + try { + await fs.utimes(lockDir, now, now); + } catch { + /* ignore */ + } + return token; +} + +async function acquireLock(dir, { lockTimeoutMs, staleMs, beforeStaleRename }, fs) { + const lockDir = path.join(dir, 'ledger.lock'); + const token = randomBytes(16).toString('hex'); + const started = Date.now(); + let spins = 0; + const liveStaleAfter = STALE_LIVE_MULTIPLIER * staleMs; + const esrchGrace = staleMs / 2; + + for (;;) { + if (++spins > 10_000) throw new LedgerLockedError('lock acquire spin limit'); + try { + await createLockDir(lockDir, token, fs); + return token; + } catch (e) { + if (e.code !== 'EEXIST') throw e; + + let st = null; + try { + st = await fs.stat(lockDir); + } catch (se) { + if (se && se.code === 'ENOENT') continue; + throw se; + } + + const age = Date.now() - st.mtimeMs; + const owner = await readOwner(lockDir, fs); + const rememberedToken = owner && typeof owner.token === 'string' ? owner.token : ''; + const dead = owner ? pidDefinitelyDead(owner.pid) : true; + // ① ESRCH after at least one missed heartbeat (age > staleMs/2) → reclaim. + // (fresh mtime + ESRCH alone is ignored: hosts may report false ESRCH.) + // ② Live/EPERM → only after 4×staleMs mtime age. + const reclaimable = (dead && age > esrchGrace) || (!dead && age > liveStaleAfter); + + if (!reclaimable) { + if (!lockTimeoutMs || lockTimeoutMs <= 0) throw new LedgerLockedError(); + if (Date.now() - started >= lockTimeoutMs) throw new LedgerLockedError(); + await sleep(50); + continue; + } + + if (typeof beforeStaleRename === 'function') { + await beforeStaleRename({ lockDir, rememberedToken, staleMs }); + } + + const staleName = `${lockDir}.stale-${token}`; + try { + await fs.rename(lockDir, staleName); + } catch (re) { + if (re && re.code === 'ENOENT') continue; + throw re; + } + + const movedOwner = await readOwner(staleName, fs); + const movedToken = movedOwner && typeof movedOwner.token === 'string' ? movedOwner.token : ''; + + // Only delete if we moved the same stale lock we inspected (token match). + if (movedToken !== rememberedToken) { + let restored = false; + let restoreErr = null; + try { + await fs.stat(lockDir); + // Original name occupied — leave the moved dir in place; do not delete. + } catch (se) { + if (se && se.code === 'ENOENT') { + try { + await fs.rename(staleName, lockDir); + restored = true; + } catch (re2) { + restoreErr = re2; + } + } else { + restoreErr = se; + } + } + if (restored) { + throw new LedgerLockedError('stale rename raced; restored foreign lock'); + } + if (restoreErr) { + throw new LedgerLockedError( + `stale rename raced; restore failed: ${restoreErr.message || restoreErr}`, + ); + } + throw new LedgerLockedError('stale rename raced; left moved lock in place'); + } + + let movedStat = null; + try { + movedStat = await fs.stat(staleName); + } catch { + movedStat = null; + } + const movedAge = movedStat ? Date.now() - movedStat.mtimeMs : Infinity; + const movedDead = movedOwner ? pidDefinitelyDead(movedOwner.pid) : true; + const stillReclaimable = + (movedDead && movedAge > esrchGrace) || (!movedDead && movedAge > liveStaleAfter); + + if (!stillReclaimable) { + let lockExists = true; + try { + await fs.stat(lockDir); + } catch (se) { + if (se && se.code === 'ENOENT') lockExists = false; + else throw se; + } + if (!lockExists) { + try { + await fs.rename(staleName, lockDir); + } catch (re2) { + throw new LedgerLockedError( + `stale criteria failed; restore failed: ${re2.message || re2}`, + ); + } + } + throw new LedgerLockedError(); + } + + try { + await fs.rm(staleName, { recursive: true, force: true }); + } catch { + /* ignore */ + } + // Loop and mkdir. + } + } +} diff --git a/src/request-coordinator.mjs b/src/request-coordinator.mjs index 9d193b1..9426eff 100644 --- a/src/request-coordinator.mjs +++ b/src/request-coordinator.mjs @@ -6,23 +6,65 @@ export class RequestConflictError extends Error { export class UnknownOutcomeError extends Error { constructor(id, cause) { super(`Outcome unknown for request ${id}`, { cause }); this.name = 'UnknownOutcomeError'; } } +export class LedgerWriteError extends Error { + constructor(message, options) { super(message, options); this.name = 'LedgerWriteError'; } +} function fingerprint(payload) { return createHash('sha256').update(JSON.stringify(payload)).digest('hex'); } + +function reviveError(shape) { + const error = new Error(shape?.message || 'Error'); + error.name = shape?.name || 'Error'; + if (shape && shape.code !== undefined) error.code = shape.code; + return error; +} + /** - * In-memory reference coordinator. Not a durable queue. + * In-memory reference coordinator. Optional durable ledger via `{ ledger }`. * execute(payload) must be supplied by the caller. Do not retry UNKNOWN automatically. */ export class RequestCoordinator { #records = new Map(); + #ledger = null; + + constructor(options = {}) { + this.#ledger = options.ledger ?? null; + if (this.#ledger) { + for (const rec of this.#ledger.list({})) { + this.#hydrate(rec); + } + } + } + + #hydrate(rec) { + if (!rec) return; + const row = { + state: rec.state, + fingerprint: rec.fingerprint, + result: rec.result, + error: rec.error ? reviveError(rec.error) : undefined, + reason: rec.reason, + }; + const prev = this.#records.get(rec.id); + if (prev?.promise) row.promise = prev.promise; + this.#records.set(rec.id, row); + } + get(id) { const r = this.#records.get(id); return r && { id, state: r.state, fingerprint: r.fingerprint, result: r.result }; } + async run(id, payload, execute) { if (typeof id !== 'string' || !id.trim()) throw new TypeError('Non-empty request ID required'); if (typeof execute !== 'function') throw new TypeError('execute must be a function'); const hash = fingerprint(payload); + + if (this.#ledger) { + return this.#runDurable(id, payload, hash, execute); + } + const prior = this.#records.get(id); if (prior) { if (prior.fingerprint !== hash) throw new RequestConflictError(id); @@ -48,11 +90,119 @@ export class RequestCoordinator { })(); return record.promise; } - /** Deliberate reconciliation, not an implicit retry. */ + + async #runDurable(id, payload, hash, execute) { + const existing = this.#ledger.get(id); + if (existing) { + if (existing.fingerprint !== hash) throw new RequestConflictError(id); + if (existing.state === 'SUCCEEDED') { + this.#hydrate(existing); + return existing.result; + } + if (existing.state === 'FAILED') { + this.#hydrate(existing); + throw reviveError(existing.error); + } + if (existing.state === 'UNKNOWN') { + this.#hydrate(existing); + throw new UnknownOutcomeError(id); + } + } + + const local = this.#records.get(id); + if (local?.promise && local.fingerprint === hash) { + return local.promise; + } + + const accepted = await this.#ledger.accept(id, payload); + this.#hydrate(this.#ledger.get(id)); + + if (!accepted.created) { + const prior = this.#records.get(id); + if (prior.fingerprint !== hash) throw new RequestConflictError(id); + if (prior.state === 'UNKNOWN') throw new UnknownOutcomeError(id); + if (prior.state === 'FAILED') throw prior.error || reviveError(this.#ledger.get(id).error); + if (prior.state === 'SUCCEEDED') return prior.result; + if (prior.state === 'RUNNING' && prior.promise) return prior.promise; + if (prior.state === 'RUNNING') throw new UnknownOutcomeError(id); + } + + const record = this.#records.get(id) || { state: 'ACCEPTED', fingerprint: hash }; + this.#records.set(id, record); + if (record.promise) return record.promise; + + record.promise = (async () => { + let executed = false; + let execResult; + try { + const cur = this.#ledger.get(id); + if (!cur) throw new Error('Missing ledger record'); + if (cur.state === 'SUCCEEDED') { + this.#hydrate(cur); + return cur.result; + } + if (cur.state === 'FAILED') { + this.#hydrate(cur); + throw reviveError(cur.error); + } + if (cur.state === 'UNKNOWN') { + this.#hydrate(cur); + throw new UnknownOutcomeError(id); + } + if (cur.state === 'ACCEPTED') { + await this.#ledger.markRunning(id); + this.#hydrate(this.#ledger.get(id)); + } + execResult = await execute(payload); + executed = true; + if (execResult === undefined) throw new Error('Missing execution receipt'); + try { + await this.#ledger.succeed(id, execResult); + } catch (writeErr) { + throw new LedgerWriteError('Failed to persist success receipt', { cause: writeErr }); + } + this.#hydrate(this.#ledger.get(id)); + return execResult; + } catch (error) { + if (error instanceof LedgerWriteError) throw error; + if (error instanceof RequestConflictError || error instanceof UnknownOutcomeError) throw error; + if (error?.name === 'LedgerStateError' || error?.name === 'LedgerLockedError') throw error; + if (executed) { + // execute succeeded but we are in an unexpected path — do not return success + throw new LedgerWriteError('Failed after execute without durable success', { cause: error }); + } + try { + const st = this.#ledger.get(id)?.state; + if (error?.outcomeUnknown === true && st === 'RUNNING') { + await this.#ledger.markUnknown(id, error.message || 'unknown'); + this.#hydrate(this.#ledger.get(id)); + } else if (st === 'RUNNING') { + await this.#ledger.fail(id, error); + this.#hydrate(this.#ledger.get(id)); + } + } catch { + // Ledger record of the execute failure failed; still surface the execute error. + } + throw error; + } + })(); + return record.promise; + } + + /** + * Deliberate reconciliation, not an implicit retry. + * Without a ledger: sync, returns undefined (v0.1.0). + * With a ledger: returns a Promise; memory updates only after the durable write succeeds. + */ reconcile(id, result) { const record = this.#records.get(id); if (!record || record.state !== 'UNKNOWN') throw new Error('Only UNKNOWN requests can be reconciled'); if (result === undefined) throw new Error('Receipt required'); + if (this.#ledger) { + return this.#ledger.reconcile(id, result).then((rec) => { + this.#hydrate(rec); + }); + } record.result = result; record.state = 'SUCCEEDED'; } diff --git a/test/durable-coordinator.test.mjs b/test/durable-coordinator.test.mjs new file mode 100644 index 0000000..f999360 --- /dev/null +++ b/test/durable-coordinator.test.mjs @@ -0,0 +1,138 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import fs from 'node:fs/promises'; +import path from 'node:path'; +import os from 'node:os'; +import { Ledger } from '../src/ledger.mjs'; +import { + RequestCoordinator, + RequestConflictError, + UnknownOutcomeError, + LedgerWriteError, +} from '../src/request-coordinator.mjs'; + +const tmp = async () => fs.mkdtemp(path.join(os.tmpdir(), 'dur-coord-')); + +test('memory coordinator without ledger still matches 0.1.0 behavior', async () => { + const c = new RequestCoordinator(); + let count = 0; + const exec = async () => { count++; return { ok: true }; }; + await Promise.all([c.run('a', { x: 1 }, exec), c.run('a', { x: 1 }, exec)]); + assert.equal(count, 1); +}); + +test('durable: succeed persists; restart returns result without execute', async () => { + const dir = await tmp(); + const ledger = await Ledger.open(dir); + const c = new RequestCoordinator({ ledger }); + let count = 0; + const result = await c.run('job', { p: 1 }, async () => { count++; return { done: true }; }); + assert.deepEqual(result, { done: true }); + assert.equal(count, 1); + await ledger.close(); + + const ledger2 = await Ledger.open(dir); + const c2 = new RequestCoordinator({ ledger: ledger2 }); + let count2 = 0; + const again = await c2.run('job', { p: 1 }, async () => { count2++; return { nope: true }; }); + assert.deepEqual(again, { done: true }); + assert.equal(count2, 0); + await ledger2.close(); +}); + +test('durable: UNKNOWN and FAILED replay; conflict; same-id concurrent execute once', async () => { + const dir = await tmp(); + const ledger = await Ledger.open(dir); + const c = new RequestCoordinator({ ledger }); + await assert.rejects( + c.run('u', {}, async () => { const e = new Error('lost'); e.outcomeUnknown = true; throw e; }), + ); + await assert.rejects(c.run('u', {}, async () => ({ x: 1 })), UnknownOutcomeError); + await c.reconcile('u', { fixed: true }); + assert.deepEqual(await c.run('u', {}, async () => ({ x: 1 })), { fixed: true }); + + await assert.rejects( + c.run('f', {}, async () => { const e = new Error('nope'); e.code = 'EFAIL'; throw e; }), + ); + await assert.rejects(c.run('f', {}, async () => ({})), (err) => { + assert.equal(err.message, 'nope'); + assert.equal(err.code, 'EFAIL'); + return true; + }); + + await c.run('ok', { a: 1 }, async () => ({ ok: true })); + await assert.rejects(c.run('ok', { a: 2 }, async () => ({ ok: true })), RequestConflictError); + + let n = 0; + const exec = async () => { n++; await new Promise((r) => setTimeout(r, 20)); return { n }; }; + const parts = await Promise.all(Array.from({ length: 50 }, () => c.run('many', { k: 1 }, exec))); + assert.equal(n, 1); + assert.ok(parts.every((p) => p.n === 1)); + await ledger.close(); +}); + +test('durable: crash while RUNNING then open yields UNKNOWN', async () => { + const dir = await tmp(); + const ledger = await Ledger.open(dir); + const c = new RequestCoordinator({ ledger }); + let stop = false; + const pending = c.run('mid', { z: 1 }, async () => { + while (!stop) await new Promise((r) => setTimeout(r, 20)); + const err = new Error('interrupted'); + err.outcomeUnknown = true; + throw err; + }); + await new Promise((r) => setTimeout(r, 40)); + assert.equal(c.get('mid').state, 'RUNNING'); + // simulate process loss: close without finishing + await ledger.close(); + stop = true; + void pending.catch(() => {}); + await new Promise((r) => setTimeout(r, 40)); + + const ledger2 = await Ledger.open(dir); + assert.equal(ledger2.get('mid').state, 'UNKNOWN'); + const c2 = new RequestCoordinator({ ledger: ledger2 }); + await assert.rejects(c2.run('mid', { z: 1 }, async () => ({ x: 1 })), UnknownOutcomeError); + await ledger2.close(); +}); + +test('durable: succeed write failure yields LedgerWriteError then UNKNOWN on reopen', async () => { + const dir = await tmp(); + let failNextAppend = false; + const real = await import('node:fs/promises'); + const deps = { + fs: { + ...real, + open: async (p, flags) => { + const handle = await real.open(p, flags); + if (String(p).endsWith('ledger.jsonl') && String(flags).includes('a')) { + return { + writeFile: async (data) => { + if (failNextAppend) throw new Error('injected succeed write failure'); + return handle.writeFile(data); + }, + sync: async () => handle.sync(), + close: async () => handle.close(), + }; + } + return handle; + }, + }, + }; + const ledger = await Ledger.open(dir, { fsync: true, deps }); + const c = new RequestCoordinator({ ledger }); + await assert.rejects( + c.run('w', { q: 1 }, async () => { + failNextAppend = true; + return { ok: true }; + }), + LedgerWriteError, + ); + failNextAppend = false; + await ledger.close(); + + const ledger2 = await Ledger.open(dir); + assert.equal(ledger2.get('w').state, 'UNKNOWN'); + await ledger2.close(); +}); diff --git a/test/helpers/ledger-child.mjs b/test/helpers/ledger-child.mjs new file mode 100644 index 0000000..fa3768b --- /dev/null +++ b/test/helpers/ledger-child.mjs @@ -0,0 +1,56 @@ +#!/usr/bin/env node +// Child helper for multi-process lock tests. +// Args: [lockTimeoutMs] [staleMs] +import { Ledger, LedgerLockedError } from '../../src/ledger.mjs'; + +const [dir, mode, timeoutStr, staleStr] = process.argv.slice(2); + +// When the test runner executes this file with no args, exit cleanly. +if (!dir || !mode) { + process.exit(0); +} + +const lockTimeoutMs = Number(timeoutStr || 0); +const staleMs = staleStr !== undefined && staleStr !== '' ? Number(staleStr) : 15_000; + +try { + if (mode === 'hold') { + const ledger = await Ledger.open(dir, { fsync: true, lockTimeoutMs: 0, staleMs }); + process.stdout.write(JSON.stringify({ ok: true, pid: process.pid }) + '\n'); + await new Promise(() => {}); // hold until killed + await ledger.close(); + } else if (mode === 'try') { + try { + const ledger = await Ledger.open(dir, { fsync: true, lockTimeoutMs, staleMs }); + process.stdout.write(JSON.stringify({ ok: true, pid: process.pid }) + '\n'); + await ledger.close(); + process.exit(0); + } catch (e) { + process.stdout.write(JSON.stringify({ ok: false, name: e.name, message: e.message }) + '\n'); + process.exit(e instanceof LedgerLockedError ? 2 : 1); + } + } else if (mode === 'steal') { + // Compete to open a (possibly stale) lock; winner holds until killed. + const watchdog = setTimeout(() => { + process.stdout.write(JSON.stringify({ ok: false, name: 'Timeout', message: 'steal watchdog' }) + '\n'); + process.exit(3); + }, 5000); + try { + const ledger = await Ledger.open(dir, { fsync: true, lockTimeoutMs, staleMs }); + clearTimeout(watchdog); + process.stdout.write(JSON.stringify({ ok: true, pid: process.pid }) + '\n'); + await new Promise(() => {}); + await ledger.close(); + } catch (e) { + clearTimeout(watchdog); + process.stdout.write(JSON.stringify({ ok: false, name: e.name, message: e.message }) + '\n'); + process.exit(e instanceof LedgerLockedError ? 2 : 1); + } + } else { + process.stderr.write('unknown mode\n'); + process.exit(1); + } +} catch (e) { + process.stdout.write(JSON.stringify({ ok: false, name: e?.name, message: e?.message }) + '\n'); + process.exit(1); +} diff --git a/test/ledger-recovery.test.mjs b/test/ledger-recovery.test.mjs new file mode 100644 index 0000000..2509031 --- /dev/null +++ b/test/ledger-recovery.test.mjs @@ -0,0 +1,305 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import fs from 'node:fs/promises'; +import path from 'node:path'; +import os from 'node:os'; +import { spawn } from 'node:child_process'; +import { fileURLToPath } from 'node:url'; +import { + Ledger, + LedgerCorruptError, + LedgerLockedError, +} from '../src/ledger.mjs'; + +const tmp = async () => fs.mkdtemp(path.join(os.tmpdir(), 'ledger-rec-')); +const __dirname = path.dirname(fileURLToPath(import.meta.url)); +const childPath = path.join(__dirname, 'helpers', 'ledger-child.mjs'); + +function spawnChild(args) { + return spawn(process.execPath, [childPath, ...args], { stdio: ['ignore', 'pipe', 'pipe'] }); +} + +function readFirstLine(child, timeoutMs = 10000) { + return new Promise((resolve, reject) => { + let buf = ''; + const timer = setTimeout(() => reject(new Error('child timeout')), timeoutMs); + child.stdout.on('data', (d) => { + buf += d; + if (buf.includes('\n')) { + clearTimeout(timer); + resolve(buf.split('\n')[0]); + } + }); + child.on('error', (e) => { + clearTimeout(timer); + reject(e); + }); + }); +} + +function readExit(child, timeoutMs = 15000) { + return new Promise((resolve, reject) => { + let buf = ''; + const timer = setTimeout(() => { + reject(new Error(`child exit timeout buf=${buf} code=${child.exitCode}`)); + }, timeoutMs); + const finish = (code) => { + clearTimeout(timer); + resolve({ buf: buf.trim(), code }); + }; + child.stdout.on('data', (d) => { buf += d; }); + child.stderr.on('data', (d) => { buf += d; }); + if (child.exitCode !== null) { + finish(child.exitCode); + return; + } + child.on('exit', finish); + }); +} + +test('restart: SUCCEEDED not re-executed; RUNNING becomes UNKNOWN', async () => { + const dir = await tmp(); + const L = await Ledger.open(dir); + await L.accept('ok', { a: 1 }); + await L.markRunning('ok'); + await L.succeed('ok', { value: 7 }); + await L.accept('run', { b: 2 }); + await L.markRunning('run'); + await L.close(); + + const L2 = await Ledger.open(dir); + assert.equal(L2.get('ok').state, 'SUCCEEDED'); + assert.deepEqual(L2.get('ok').result, { value: 7 }); + assert.equal(L2.get('run').state, 'UNKNOWN'); + assert.ok(L2.get('run').reason); + await L2.close(); +}); + +test('partial last line recovered by truncate; middle corruption rejected', async () => { + const dir = await tmp(); + const L = await Ledger.open(dir); + await L.accept('a', { n: 1 }); + await L.accept('b', { n: 2 }); + await L.close(); + const file = path.join(dir, 'ledger.jsonl'); + const good = await fs.readFile(file, 'utf8'); + await fs.writeFile(file, good + '{"v":1,"seq":3,partial'); + const L2 = await Ledger.open(dir); + assert.ok(L2.recovery?.truncatedBytes > 0); + assert.equal(L2.get('a').state, 'ACCEPTED'); + assert.equal(L2.get('b').state, 'ACCEPTED'); + await L2.close(); + + const lines = good.trim().split('\n'); + const brokenMiddle = [lines[0], '{not-json', lines[1]].join('\n') + '\n'; + await fs.writeFile(file, brokenMiddle); + await assert.rejects(() => Ledger.open(dir), LedgerCorruptError); +}); + +test('crc tamper and seq gap detected', async () => { + const dir = await tmp(); + const L = await Ledger.open(dir); + await L.accept('x', { z: 1 }); + await L.close(); + const file = path.join(dir, 'ledger.jsonl'); + const row = JSON.parse((await fs.readFile(file, 'utf8')).trim()); + row.crc = '0000000000000000'; + await fs.writeFile(file, JSON.stringify(row) + '\n'); + await assert.rejects(() => Ledger.open(dir), LedgerCorruptError); + + const dir2 = await tmp(); + const L3 = await Ledger.open(dir2); + await L3.accept('y', {}); + await L3.close(); + const f2 = path.join(dir2, 'ledger.jsonl'); + const e = JSON.parse((await fs.readFile(f2, 'utf8')).trim()); + e.seq = 5; + const { createHash } = await import('node:crypto'); + const { crc, ...rest } = e; + e.crc = createHash('sha256').update(JSON.stringify(rest)).digest('hex').slice(0, 16); + await fs.writeFile(f2, JSON.stringify(e) + '\n'); + await assert.rejects(() => Ledger.open(dir2), LedgerCorruptError); +}); + +test('two real processes: only one open succeeds', async () => { + const dir = await tmp(); + // Large staleMs so heartbeat (staleMs/2) tolerates load jitter. + const staleMs = 2000; + const holder = spawnChild([dir, 'hold', '0', String(staleMs)]); + const firstLine = await readFirstLine(holder, 8000); + assert.equal(JSON.parse(firstLine).ok, true); + + const challenged = await new Promise((resolve, reject) => { + const challenger = spawnChild([dir, 'try', '0', String(staleMs)]); + let buf = ''; + const timer = setTimeout(() => { + try { challenger.kill('SIGKILL'); } catch { /* ignore */ } + reject(new Error(`challenger hang buf=${buf}`)); + }, 8000); + challenger.stdout.on('data', (d) => { buf += d; }); + challenger.stderr.on('data', (d) => { buf += d; }); + challenger.on('exit', (code) => { + clearTimeout(timer); + try { + resolve({ ...JSON.parse(buf.trim().split('\n')[0]), code }); + } catch (e) { + reject(new Error(`bad challenger output: ${buf} code=${code}`)); + } + }); + }); + assert.equal(challenged.ok, false); + assert.equal(challenged.name, 'LedgerLockedError'); + await assert.rejects(() => Ledger.open(dir, { staleMs, lockTimeoutMs: 0 }), LedgerLockedError); + + let holderExited = holder.exitCode !== null; + holder.on('exit', () => { holderExited = true; }); + try { + holder.kill('SIGKILL'); + } catch { + try { holder.kill(); } catch { /* ignore */ } + } + for (let i = 0; i < 50 && !holderExited && holder.exitCode === null; i++) { + await new Promise((r) => setTimeout(r, 100)); + } + assert.ok(holderExited || holder.exitCode !== null, 'holder should exit after SIGKILL'); +}); + +test('dead owner lock is taken over via ESRCH without waiting 4x staleMs', async () => { + const dir = await tmp(); + const staleMs = 300; + const holder = spawnChild([dir, 'hold', '0', String(staleMs)]); + const firstLine = await readFirstLine(holder, 8000); + assert.equal(JSON.parse(firstLine).ok, true); + + let holderExited = holder.exitCode !== null; + holder.on('exit', () => { holderExited = true; }); + try { + holder.kill('SIGKILL'); + } catch { + try { holder.kill(); } catch { /* ignore */ } + } + for (let i = 0; i < 50 && !holderExited && holder.exitCode === null; i++) { + await new Promise((r) => setTimeout(r, 100)); + } + assert.ok(holderExited || holder.exitCode !== null, 'holder should exit after SIGKILL'); + // ESRCH reclaim requires age > staleMs/2 (missed heartbeat); still far below 4×staleMs. + await new Promise((r) => setTimeout(r, staleMs / 2 + 100)); + + const L = await Ledger.open(dir, { staleMs }); + assert.ok(L); + await L.close(); +}); + +test('live pid: mtime between staleMs and 4x is locked; beyond 4x is reclaimable', async () => { + const dir = await tmp(); + const staleMs = 200; + const lockDir = path.join(dir, 'ledger.lock'); + await fs.mkdir(lockDir); + // Use this process pid so the owner is definitely alive. + await fs.writeFile( + path.join(lockDir, 'owner.json'), + JSON.stringify({ pid: process.pid, token: 'livetoken', startedAt: new Date().toISOString() }), + ); + + const mid = new Date(Date.now() - staleMs * 3); + await fs.utimes(lockDir, mid, mid); + await assert.rejects(() => Ledger.open(dir, { staleMs, lockTimeoutMs: 0 }), LedgerLockedError); + + const old = new Date(Date.now() - staleMs * 5); + await fs.utimes(lockDir, old, old); + const L = await Ledger.open(dir, { staleMs }); + assert.ok(L); + await L.close(); +}); + +test('heartbeat keeps lock; external remove compromises writer', async () => { + const dir = await tmp(); + const staleMs = 400; + const L = await Ledger.open(dir, { staleMs }); + await L.accept('a', { n: 1 }); + + // Live owner longer than staleMs but under 4×staleMs; another open must fail. + await new Promise((r) => setTimeout(r, staleMs + 200)); + await assert.rejects(() => Ledger.open(dir, { staleMs, lockTimeoutMs: 0 }), LedgerLockedError); + + // External delete of lock dir → subsequent writes rejected. + await fs.rm(path.join(dir, 'ledger.lock'), { recursive: true, force: true }); + await new Promise((r) => setTimeout(r, Math.floor(staleMs / 2) + 50)); + await assert.rejects(() => L.accept('b', { n: 2 }), (err) => { + assert.equal(err.name, 'LedgerLockedError'); + assert.match(String(err.message), /lock lost/i); + return true; + }); + await L.close(); +}); + +test('exactly one of six processes wins a stale lock reclaim', async () => { + const dir = await tmp(); + const staleMs = 300; + // Create a stale lock directory with a dead pid and old mtime. + const lockDir = path.join(dir, 'ledger.lock'); + await fs.mkdir(lockDir); + await fs.writeFile( + path.join(lockDir, 'owner.json'), + JSON.stringify({ pid: 1_000_000_007, token: 'deadtoken', startedAt: new Date(0).toISOString() }), + ); + const old = new Date(Date.now() - staleMs * 3); + await fs.utimes(lockDir, old, old); + + const kids = Array.from({ length: 6 }, () => spawnChild([dir, 'steal', '0', String(staleMs)])); + const results = await Promise.all(kids.map(async (k) => { + try { + const line = await readFirstLine(k, 8000); + return JSON.parse(line); + } catch { + return { ok: false }; + } + })); + const wins = results.filter((r) => r.ok === true); + assert.equal(wins.length, 1, JSON.stringify(results)); + for (const k of kids) { + try { + k.kill('SIGKILL'); + } catch { + try { k.kill(); } catch { /* ignore */ } + } + } + for (let i = 0; i < 50; i++) { + if (kids.every((k) => k.exitCode !== null)) break; + await new Promise((r) => setTimeout(r, 100)); + } +}); + +test('stale rename race: challenger must not delete winner lock', async () => { + const dir = await tmp(); + const staleMs = 200; + const lockDir = path.join(dir, 'ledger.lock'); + await fs.mkdir(lockDir); + const staleOwner = { + pid: 1_000_000_007, + token: 'stale-original', + startedAt: new Date(0).toISOString(), + }; + await fs.writeFile(path.join(lockDir, 'owner.json'), JSON.stringify(staleOwner)); + const old = new Date(Date.now() - staleMs * 5); + await fs.utimes(lockDir, old, old); + + let winner = null; + const loser = Ledger.open(dir, { + staleMs, + lockTimeoutMs: 0, + deps: { + beforeStaleRename: async () => { + // A wins first: reclaim and install a fresh lock before B renames. + winner = await Ledger.open(dir, { staleMs, lockTimeoutMs: 0 }); + }, + }, + }); + + await assert.rejects(() => loser, LedgerLockedError); + assert.ok(winner); + const owner = JSON.parse(await fs.readFile(path.join(lockDir, 'owner.json'), 'utf8')); + assert.notEqual(owner.token, 'stale-original'); + assert.equal(owner.pid, process.pid); + await winner.close(); +}); diff --git a/test/ledger.test.mjs b/test/ledger.test.mjs new file mode 100644 index 0000000..f8f0313 --- /dev/null +++ b/test/ledger.test.mjs @@ -0,0 +1,148 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import fs from 'node:fs/promises'; +import path from 'node:path'; +import os from 'node:os'; +import { + Ledger, + LedgerStateError, + LedgerLimitError, +} from '../src/ledger.mjs'; +import { RequestConflictError } from '../src/request-coordinator.mjs'; + +const tmp = async () => fs.mkdtemp(path.join(os.tmpdir(), 'ledger-')); + +test('idempotent accept and conflict fingerprint', async () => { + const dir = await tmp(); + const L = await Ledger.open(dir); + const a = await L.accept('r1', { n: 1 }); + assert.equal(a.created, true); + assert.equal(a.record.state, 'ACCEPTED'); + const b = await L.accept('r1', { n: 1 }); + assert.equal(b.created, false); + assert.equal(b.record.id, 'r1'); + await assert.rejects(() => L.accept('r1', { n: 2 }), RequestConflictError); + await assert.rejects(() => L.accept('', {}), TypeError); + await assert.rejects(() => L.accept(null, {}), TypeError); + await L.close(); +}); + +test('transition rules exhaustive', async () => { + const dir = await tmp(); + const L = await Ledger.open(dir); + await L.accept('t1', { x: 1 }); + await L.markRunning('t1'); + await L.succeed('t1', { ok: true }); + await assert.rejects(() => L.markRunning('t1'), LedgerStateError); + await assert.rejects(() => L.fail('t1', new Error('x')), LedgerStateError); + await assert.rejects(() => L.succeed('t1', { again: true }), LedgerStateError); + + await L.accept('t2', {}); + await L.markRunning('t2'); + await L.fail('t2', Object.assign(new Error('boom'), { code: 'E' })); + await assert.rejects(() => L.reconcile('t2', { x: 1 }), LedgerStateError); + + await L.accept('t3', {}); + await L.markRunning('t3'); + await L.markUnknown('t3', 'lost'); + await assert.rejects(() => L.fail('t3', new Error('no')), LedgerStateError); + await L.reconcile('t3', { confirmed: true }); + assert.equal(L.get('t3').state, 'SUCCEEDED'); + await L.close(); +}); + +test('result undefined and size limit', async () => { + const dir = await tmp(); + const L = await Ledger.open(dir); + await L.accept('u1', {}); + await L.markRunning('u1'); + await assert.rejects(() => L.succeed('u1', undefined), /receipt|undefined|required/i); + const big = 'x'.repeat(65 * 1024); + await assert.rejects(() => L.succeed('u1', { big }), LedgerLimitError); + await L.close(); +}); + +test('write failure leaves memory unchanged', async () => { + const dir = await tmp(); + let failNext = false; + const real = await import('node:fs/promises'); + const deps = { + fs: { + ...real, + open: async (p, flags) => { + const handle = await real.open(p, flags); + if (String(p).endsWith('ledger.jsonl') && String(flags).includes('a')) { + return { + writeFile: async (data) => { + if (failNext) throw new Error('injected write failure'); + return handle.writeFile(data); + }, + sync: async () => handle.sync(), + close: async () => handle.close(), + }; + } + return handle; + }, + }, + }; + const L = await Ledger.open(dir, { fsync: true, deps }); + await L.accept('w1', { a: 1 }); + failNext = true; + const before = L.get('w1'); + await assert.rejects(() => L.markRunning('w1'), /injected write failure/); + assert.deepEqual(L.get('w1'), before); + assert.equal(L.get('w1').state, 'ACCEPTED'); + failNext = false; + await L.markRunning('w1'); + assert.equal(L.get('w1').state, 'RUNNING'); + await L.close(); +}); + +test('seq continuous under concurrent accepts', async () => { + const dir = await tmp(); + const L = await Ledger.open(dir); + await Promise.all(Array.from({ length: 100 }, (_, i) => L.accept(`id-${i}`, { i }))); + const listed = L.list({}); + assert.equal(listed.length, 100); + const text = await fs.readFile(path.join(dir, 'ledger.jsonl'), 'utf8'); + const seqs = text.trim().split('\n').map((l) => JSON.parse(l).seq); + assert.deepEqual(seqs, Array.from({ length: 100 }, (_, i) => i + 1)); + await L.close(); +}); + +test('compact preserves logical state', async () => { + const dir = await tmp(); + const L = await Ledger.open(dir); + await L.accept('c1', { v: 1 }); + await L.markRunning('c1'); + await L.succeed('c1', { done: true }); + await L.accept('c2', { v: 2 }); + await L.compact(); + assert.equal(L.get('c1').state, 'SUCCEEDED'); + assert.equal(L.get('c2').state, 'ACCEPTED'); + await L.close(); + const L2 = await Ledger.open(dir); + assert.equal(L2.get('c1').state, 'SUCCEEDED'); + assert.deepEqual(L2.get('c1').result, { done: true }); + assert.equal(L2.get('c2').state, 'ACCEPTED'); + await L2.close(); +}); + +test('builtin imports only (node:); ledger has no net or child_process', async () => { + const src = await fs.readFile(new URL('../src/ledger.mjs', import.meta.url), 'utf8'); + const coord = await fs.readFile(new URL('../src/request-coordinator.mjs', import.meta.url), 'utf8'); + for (const text of [src, coord]) { + const imports = [...text.matchAll(/from\s+['"]([^'"]+)['"]/g)].map((m) => m[1]); + for (const spec of imports) { + assert.ok(spec.startsWith('node:') || spec.startsWith('./') || spec.startsWith('../'), spec); + if (!spec.startsWith('node:') && !spec.startsWith('.')) assert.fail(`external import: ${spec}`); + } + assert.equal(imports.some((s) => s.startsWith('node:') === false && !s.startsWith('.')), false); + } + assert.equal(/\bfrom\s+['"](?!\.|node:)[^'"]+['"]/.test(src), false); + assert.equal(/\bfrom\s+['"](?!\.|node:)[^'"]+['"]/.test(coord), false); + assert.equal(/node:net\b/.test(src), false); + assert.equal(/node:child_process\b/.test(src), false); + assert.equal(/\bfrom\s+['"]node:net['"]/.test(src), false); + assert.equal(/\bfrom\s+['"]node:child_process['"]/.test(src), false); +});