diff --git a/changelog.d/quiesce-writes-first.fixed.md b/changelog.d/quiesce-writes-first.fixed.md index ddb8939af..423521c92 100644 --- a/changelog.d/quiesce-writes-first.fixed.md +++ b/changelog.d/quiesce-writes-first.fixed.md @@ -1,4 +1,4 @@ - `quiesce()` and `resume()` now answer only once the host-mode record agrees with the host, or fail, apart from calls that overlap. - **`quiesce()`** writes the record before it marks the host quiesced. When that write failed, `quiesce()` rejected with the host quiesced in memory while its record said it was serving, so a restart booted serving, and a second `quiesce()` resolved without writing the record. Now a failed write leaves the host serving, as it was, and the next `quiesce()` quiesces and writes the record. - **`resume()`** still serves before it clears the record, so a failed clear leaves the record quiesced. A second `resume()` used to resolve without clearing it, so the next restart booted quiesced; it now clears the record. While a surgery holds the store, that clear is refused as `resume()` is on a quiesced host, and a `resume()` arriving while another is still flushing leaves the clear to that one. - - **Without a recovery file,** the store is the record, and a failed sync of the store after the write was only logged, so `quiesce()` resolved before its record was durable. It now fails `quiesce()` or `resume()`. The write is already in the store's log, though, so after a `quiesce()` that failed this way the record can still reach disk, and a restart can boot quiesced unless a `resume()` clears it first. With a recovery file, the file is the record, and a failed sync is still only logged. + - **Without a recovery file,** the store is the record, and a failed sync of the store after the write was only logged, so `quiesce()` resolved before its record was durable. It now fails `quiesce()` or `resume()`. The write is already in the store's log, though, so after a `quiesce()` that failed this way the record can still reach disk, and a restart can boot quiesced unless a `resume()` clears it first. With a recovery file, the file is the record, and a failed sync of the store is still only logged. diff --git a/changelog.d/recovery-file-durability.fixed.md b/changelog.d/recovery-file-durability.fixed.md new file mode 100644 index 000000000..b814ca080 --- /dev/null +++ b/changelog.d/recovery-file-durability.fixed.md @@ -0,0 +1 @@ +- The host-mode and deferred-writes recovery files are now synced to disk before they replace the record, and their directory and its parent after the rename. On Windows, and where the filesystem refuses a directory's sync with `EINVAL` or `EBADF`, only the file is synced. A write that fails before the rename removes its `.tmp` file instead of leaving it beside the record, logs one it can't remove, and fails with an error that names the record and says it was left as it was. A directory sync that fails for any other reason raises an ops alert naming the record, since the new record is already in place. diff --git a/src/framework.ts b/src/framework.ts index 8351c2a97..9b7962fca 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -2,7 +2,10 @@ import { ToolPresentation, presentationTools, isPresentationTool, renderCatalogu import { dirname, join } from 'node:path'; import { INLINE_WITHHELD_TEXT, REFERENCE_LIMITS, classifyBlock, isInlineContradiction, isReferenceExpired, referenceRegistry, referenceStubOrNull, referenceTypeLabel, sanitizeLabel, unfetchableReason } from './mcpl/references.js'; import { ReferenceFetcher, DEFAULT_FETCH_MAX_BYTES, EAGER_FETCH_TIMEOUT_MS } from './mcpl/reference-fetcher.js'; -import { appendFileSync, existsSync, mkdirSync, readFileSync, renameSync, writeFileSync } from 'node:fs'; +import { + appendFileSync, closeSync, existsSync, fsyncSync, mkdirSync, openSync, readFileSync, renameSync, unlinkSync, + writeFileSync, +} from 'node:fs'; import { JsStore } from '@animalabs/chronicle'; import type { Membrane, ContentBlock, NormalizedRequest, NormalizedResponse, YieldingStream, DetailedUsage, ToolResult as MembraneToolResult, ToolResultContentBlock } from '@animalabs/membrane'; import { MembraneError } from '@animalabs/membrane'; @@ -5353,13 +5356,74 @@ export class AgentFramework { return landed; } - /** Atomic JSON write for the recovery files (tmp + rename, 0600) — the - * same idiom as the Discord awareness outbox. */ + /** Atomic, durable JSON write for the recovery files. The document goes to + * a temp file at 0600, which is synced and renamed over the record, so + * the new record's data is asked to reach the disk before the rename + * publishes it. Then the directory is synced, for the record's entry in + * it, and its parent, for the directory's own entry, which is new when a + * write, this one or an earlier one, had to make the directory; levels + * above are left to the filesystem. Directories aren't synced on win32, + * which syncs the file alone, or where the filesystem can't sync one + * (syncRecoveryDirectory). A failure before the rename removes the temp + * file, if it was made, and throws an error saying the record was left as + * it was, with the original as its cause and its code kept. A directory + * sync that fails, other than where the filesystem can't sync one, is + * raised as an ops alert, not thrown: the new record is in place, and the + * callers go on to record it as written. */ private writeRecoveryFile(path: string, document: unknown): void { - mkdirSync(dirname(path), { recursive: true }); + const directory = dirname(path); const temporary = `${path}.${process.pid}.${randomUUID()}.tmp`; - writeFileSync(temporary, `${JSON.stringify(document, null, 2)}\n`, { mode: 0o600 }); - renameSync(temporary, path); + let opened = false; + try { + mkdirSync(directory, { recursive: true }); + const fd = openSync(temporary, 'w', 0o600); + opened = true; + try { + writeFileSync(fd, `${JSON.stringify(document, null, 2)}\n`); + fsyncSync(fd); + } finally { + closeSync(fd); + } + renameSync(temporary, path); + } catch (err) { + if (opened) { + try { + unlinkSync(temporary); + } catch (unlinkErr) { + console.error(`[host-mode] couldn't remove ${temporary} after a failed write:`, unlinkErr); + } + } + const cause = err as NodeJS.ErrnoException; + throw Object.assign(new Error(`${path} was left as it was: ${cause.message}`, { cause }), cause.code ? { code: cause.code } : {}); + } + this.syncRecoveryDirectory(directory, path); + this.syncRecoveryDirectory(dirname(directory), path); + } + + /** Syncs the directory a recovery file is in, or that directory's parent, + * so its entries reach the disk. win32 is left out. So is a sync refused + * with EINVAL or EBADF, the answer where the filesystem or system can't + * sync a directory: that refusal would come again on every write, and + * PostgreSQL's fsync_fname_ext lets the same two go. Any other failure is + * raised as an ops alert naming the record. */ + private syncRecoveryDirectory(directory: string, record: string): void { + if (process.platform === 'win32') return; + try { + const fd = openSync(directory, 'r'); + try { + fsyncSync(fd); + } catch (err) { + const code = (err as NodeJS.ErrnoException).code; + if (code !== 'EINVAL' && code !== 'EBADF') throw err; + } finally { + closeSync(fd); + } + } catch (err) { + const reason = err instanceof Error ? err.message : String(err); + this.opsAlert('recovery-sync', 'framework', + `syncing ${directory} failed, so ${record} may not survive a crash: ${reason}`, + { data: { record, directory, code: (err as NodeJS.ErrnoException).code } }); + } } private readRecoveryFile(path: string): unknown { diff --git a/test/host-quiesce.test.ts b/test/host-quiesce.test.ts index 81207f91f..0eb1f518f 100644 --- a/test/host-quiesce.test.ts +++ b/test/host-quiesce.test.ts @@ -489,7 +489,7 @@ test("a resume() during another resume()'s flush leaves the record to it, and it } }); -test("store only, a host-mode change whose sync fails rejects; with a recovery file, the file is the record and a failed sync is only logged", async () => { +test("store only, a host-mode change whose sync fails rejects; with a recovery file, the file is the record and a failed sync of the store is only logged", async () => { const logged = captureErrors(); try { const slot = await hostModeFixture('store only'); diff --git a/test/recovery-file-durability.test.ts b/test/recovery-file-durability.test.ts new file mode 100644 index 000000000..c81cf1620 --- /dev/null +++ b/test/recovery-file-durability.test.ts @@ -0,0 +1,301 @@ +/** + * The recovery files (host mode, deferred writes) are written as a temp file + * synced and renamed over the record, with the directory and its parent + * synced after the rename. These drive both records through the framework, + * recording the node:fs calls each write makes. + */ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import fs, { existsSync, mkdirSync, mkdtempSync, readdirSync, readFileSync, rmSync, statSync, writeFileSync } from 'node:fs'; +import { syncBuiltinESMExports } from 'node:module'; +import { tmpdir } from 'node:os'; +import { dirname, join } from 'node:path'; + +import { AgentFramework } from '../src/index.js'; +import { MockMembrane } from './helpers/mock-membrane.js'; + +async function withFramework( + fn: (framework: AgentFramework, recovery: string, dir: string) => Promise, + config: { hostModePath?: (dir: string) => string } = {}, +): Promise { + const dir = mkdtempSync(join(tmpdir(), 'recovery-durability-')); + const framework = await AgentFramework.create({ + storePath: join(dir, 'store'), + membrane: new MockMembrane().asMembrane(), + agents: [{ name: 'agent', model: 'test-model', systemPrompt: 'test' }], + modules: [], + ...(config.hostModePath ? { hostModePath: config.hostModePath(dir) } : {}), + }); + try { + framework.start(); + await fn(framework, join(dir, 'store', 'recovery'), dir); + } finally { + await framework.stop(); + rmSync(dir, { recursive: true, force: true }); + } +} + +interface FsCall { call: 'open' | 'fsync' | 'close' | 'rename' | 'unlink'; path?: string; to?: string; fd?: number } + +/** Runs `fn` with node:fs's openSync, fsyncSync, closeSync, renameSync and unlinkSync recorded in order, each fd + * named by the path it was opened for. `fail` may return an error for an fsync or an unlink to throw instead. */ +async function recordingFs( + fn: () => Promise, + fail?: (call: FsCall) => Error | undefined, +): Promise { + const calls: FsCall[] = []; + const opened = new Map(); + const original = { + openSync: fs.openSync, fsyncSync: fs.fsyncSync, closeSync: fs.closeSync, renameSync: fs.renameSync, unlinkSync: fs.unlinkSync, + }; + fs.openSync = ((path: fs.PathLike, ...rest: unknown[]) => { + const fd = (original.openSync as (...args: unknown[]) => number)(path, ...rest); + opened.set(fd, String(path)); + calls.push({ call: 'open', path: String(path), fd }); + return fd; + }) as typeof fs.openSync; + fs.fsyncSync = ((fd: number) => { + const call: FsCall = { call: 'fsync', path: opened.get(fd), fd }; + calls.push(call); + const error = fail?.(call); + if (error) throw error; + original.fsyncSync(fd); + }) as typeof fs.fsyncSync; + fs.closeSync = ((fd: number) => { + calls.push({ call: 'close', path: opened.get(fd), fd }); + original.closeSync(fd); + }) as typeof fs.closeSync; + fs.renameSync = ((from: fs.PathLike, to: fs.PathLike) => { + calls.push({ call: 'rename', path: String(from), to: String(to) }); + original.renameSync(from, to); + }) as typeof fs.renameSync; + fs.unlinkSync = ((path: fs.PathLike) => { + const call: FsCall = { call: 'unlink', path: String(path) }; + calls.push(call); + const error = fail?.(call); + if (error) throw error; + original.unlinkSync(path); + }) as typeof fs.unlinkSync; + syncBuiltinESMExports(); + try { + await fn(); + } finally { + Object.assign(fs, original); + syncBuiltinESMExports(); + } + return calls; +} + +const fsError = (code: string, syscall: string) => Object.assign(new Error(`${code}: i/o error, ${syscall}`), { code, syscall }); +const isTemporaryOf = (record: string, path: string | undefined) => + path !== undefined && path.startsWith(`${record}.`) && path.endsWith('.tmp'); + +/** The rename that put a record in place, and the temp file it renamed. */ +function recordRename(calls: FsCall[], record: string): { at: number; temporary: string } { + const at = calls.findIndex((c) => c.call === 'rename' && c.to === record); + assert.ok(at >= 0, `a rename onto ${record}: ${JSON.stringify(calls)}`); + const temporary = calls[at].path!; + assert.ok(isTemporaryOf(record, temporary), temporary); + return { at, temporary }; +} + +/** The temp file synced before its rename onto `record`, and `directory` synced after it. */ +function assertSyncedAround(calls: FsCall[], record: string, directory: string): void { + const rename = recordRename(calls, record); + const tempSync = calls.findIndex((c) => c.call === 'fsync' && c.path === rename.temporary); + assert.ok(tempSync >= 0 && tempSync < rename.at, `the temp file synced before the rename: ${JSON.stringify(calls)}`); + assert.ok( + calls.some((c, i) => i > rename.at && c.call === 'fsync' && c.path === directory), + `${directory} synced after the rename: ${JSON.stringify(calls)}`, + ); +} + +/** Every descriptor opened on a path `ours` names is closed before its number is opened again. */ +function assertEachClosed(calls: FsCall[], ours: (path: string | undefined) => boolean): void { + calls.forEach((c, i) => { + if (c.call !== 'open' || !ours(c.path)) return; + const next = calls.findIndex((d, j) => j > i && d.fd === c.fd && (d.call === 'close' || d.call === 'open')); + assert.ok(next >= 0 && calls[next]!.call === 'close', `${c.path} (fd ${c.fd}) closed: ${JSON.stringify(calls)}`); + }); +} + +test('quiesce syncs the host-mode record\'s temp file before renaming it over the record, and the directory and its parent after, closing each and publishing the record at 0600', async () => { + await withFramework(async (framework, recovery) => { + const record = join(recovery, 'host-mode.json'); + const calls = await recordingFs(() => framework.quiesce({ reason: 'test' })); + assertSyncedAround(calls, record, recovery); + assertSyncedAround(calls, record, dirname(recovery)); + assertEachClosed(calls, (path) => isTemporaryOf(record, path) || path === recovery || path === dirname(recovery)); + assert.equal(statSync(record).mode & 0o777, 0o600); + assert.equal(JSON.parse(readFileSync(record, 'utf8')).quiesced, true); + }); +}); + +test('each context write deferred while quiesced rewrites the deferred-writes record, synced the same way', async () => { + await withFramework(async (framework, recovery) => { + await framework.quiesce({ reason: 'test' }); + const add = (framework as unknown as { addMessage(participant: string, content: unknown[]): unknown }).addMessage.bind(framework); + const record = join(recovery, 'deferred-writes.json'); + const calls = await recordingFs(async () => { + add('user', [{ type: 'text', text: 'held' }]); + add('user', [{ type: 'text', text: 'held too' }]); + }); + const renames = calls.flatMap((c, i) => (c.call === 'rename' && c.to === record ? [i] : [])); + assert.equal(renames.length, 2, `a rewrite for each deferred write: ${JSON.stringify(calls)}`); + renames.forEach((_at, k) => { + // This rewrite's calls: from the rename before it to the one after. + const write = calls.slice(k === 0 ? 0 : renames[k - 1]! + 1, renames[k + 1] ?? calls.length); + assertSyncedAround(write, record, recovery); + assertSyncedAround(write, record, dirname(recovery)); + }); + const pending = (JSON.parse(readFileSync(record, 'utf8')) as { pending: Array<{ content: Array<{ text: string }> }> }).pending; + assert.deepEqual(pending.map((entry) => entry.content[0]!.text), ['held', 'held too']); + }); +}); + +test('a directory a failed write had to make is synced into its parent by the next write', async () => { + await withFramework(async (framework, _recovery, dir) => { + const fresh = join(dir, 'fresh'); + const record = join(fresh, 'host-mode.json'); + assert.equal(existsSync(fresh), false); + await recordingFs( + () => framework.quiesce({ reason: 'first' }).catch(() => undefined), + (c) => (c.call === 'fsync' && isTemporaryOf(record, c.path) ? fsError('EIO', 'fsync') : undefined), + ); + assert.equal(existsSync(fresh), true, 'the failed write made the directory'); + const calls = await recordingFs(() => framework.quiesce({ reason: 'second' })); + assertSyncedAround(calls, record, fresh); + assertSyncedAround(calls, record, dir); + }, { hostModePath: (dir) => join(dir, 'fresh', 'host-mode.json') }); +}); + +test('a write whose directory can\'t be made says the record was left as it was, and tries to remove no temp file', async () => { + await withFramework(async (framework, _recovery, dir) => { + const blocker = join(dir, 'blocker'); + writeFileSync(blocker, 'a file where the record\'s directory goes'); + const record = join(blocker, 'host-mode.json'); + let error: unknown; + const calls = await recordingFs(() => framework.quiesce({ reason: 'test' }).catch((e: unknown) => { error = e; })); + assert.ok(error instanceof Error, String(error)); + const cause = error.cause as NodeJS.ErrnoException; + assert.ok(typeof cause?.code === 'string', String(cause)); + assert.equal(error.message, `${record} was left as it was: ${cause.message}`); + assert.equal((error as NodeJS.ErrnoException).code, cause.code); + assert.ok( + !calls.some((c) => (c.call === 'open' || c.call === 'unlink') && isTemporaryOf(record, c.path)), + `no temp file opened or removed: ${JSON.stringify(calls)}`, + ); + }, { hostModePath: (dir) => join(dir, 'blocker', 'host-mode.json') }); +}); + +test('a host-mode record that can\'t be replaced leaves no temp file behind, and quiesce says the record was left as it was', async () => { + await withFramework(async (framework, recovery) => { + const record = join(recovery, 'host-mode.json'); + // A directory where the record goes, so the rename fails. + mkdirSync(record, { recursive: true }); + let error: unknown; + const calls = await recordingFs(() => framework.quiesce({ reason: 'test' }).catch((e: unknown) => { error = e; })); + recordRename(calls, record); + assert.ok(error instanceof Error, String(error)); + assert.match(error.message, new RegExp(`^${record.replace(/[.*+?^${}()|[\]\\]/g, '\\$&')} was left as it was: EISDIR`)); + assert.equal((error as NodeJS.ErrnoException).code, 'EISDIR'); + assert.deepEqual(readdirSync(recovery).filter((name) => name.endsWith('.tmp')), []); + }); +}); + +test('a quiesce whose temp file can\'t be synced closes it and rejects with the host still serving, the record as it was and no temp file', async () => { + await withFramework(async (framework, recovery) => { + const record = join(recovery, 'host-mode.json'); + await framework.quiesce({ reason: 'first' }); + await framework.resume(); + const before = readFileSync(record, 'utf8'); + let error: unknown; + const calls = await recordingFs( + () => framework.quiesce({ reason: 'test' }).catch((e: unknown) => { error = e; }), + (c) => (c.call === 'fsync' && isTemporaryOf(record, c.path) ? fsError('EIO', 'fsync') : undefined), + ); + assert.ok(error instanceof Error, String(error)); + assert.equal(error.message, `${record} was left as it was: EIO: i/o error, fsync`); + assert.ok(!calls.some((c) => c.call === 'rename' && c.to === record), `no rename: ${JSON.stringify(calls)}`); + assertEachClosed(calls, (path) => isTemporaryOf(record, path)); + assert.equal(framework.getHostModeStatus().quiesced, false, 'the host still serving'); + assert.equal(readFileSync(record, 'utf8'), before); + assert.deepEqual(readdirSync(recovery).filter((name) => name.endsWith('.tmp')), []); + }); +}); + +test('a temp file that can\'t be removed after a failed write is named in the log', async (t) => { + const logged: string[] = []; + t.mock.method(console, 'error', (...args: unknown[]) => { logged.push(args.map(String).join(' ')); }); + await withFramework(async (framework, recovery) => { + const record = join(recovery, 'host-mode.json'); + let error: unknown; + const calls = await recordingFs( + () => framework.quiesce({ reason: 'test' }).catch((e: unknown) => { error = e; }), + (c) => (isTemporaryOf(record, c.path) && (c.call === 'fsync' || c.call === 'unlink') ? fsError(c.call === 'fsync' ? 'EIO' : 'EACCES', c.call) : undefined), + ); + assert.ok(error instanceof Error, String(error)); + const unlink = calls.find((c) => c.call === 'unlink' && isTemporaryOf(record, c.path)); + assert.ok(unlink, JSON.stringify(calls)); + assert.ok(logged.some((line) => line.includes(`couldn't remove ${unlink.path} after a failed write`)), logged.join('\n')); + }); +}); + +test('on win32 the temp file is still synced, and no directory is opened to sync it', async () => { + await withFramework(async (framework, recovery) => { + const platform = Object.getOwnPropertyDescriptor(process, 'platform')!; + Object.defineProperty(process, 'platform', { ...platform, value: 'win32' }); + let calls: FsCall[]; + try { + calls = await recordingFs(() => framework.quiesce({ reason: 'test' })); + } finally { + Object.defineProperty(process, 'platform', platform); + } + const rename = recordRename(calls, join(recovery, 'host-mode.json')); + assert.ok(calls.some((c, i) => i < rename.at && c.call === 'fsync' && c.path === rename.temporary), JSON.stringify(calls)); + assert.ok( + !calls.some((c) => c.call === 'open' && (c.path === recovery || c.path === dirname(recovery))), + `neither directory opened: ${JSON.stringify(calls)}`, + ); + }); +}); + +for (const which of ['directory', 'parent'] as const) { + test(`a sync of the record's ${which} that fails after the rename leaves the new record in place and raises an ops alert naming it`, async () => { + await withFramework(async (framework, recovery) => { + const alerts: Array<{ kind: string; agentName: string; message: string; data?: Record }> = []; + (framework as unknown as { opsAlert: unknown }).opsAlert = (kind: string, agentName: string, message: string, opts?: { data?: Record }) => { + alerts.push({ kind, agentName, message, data: opts?.data }); + }; + const record = join(recovery, 'host-mode.json'); + const failing = which === 'directory' ? recovery : dirname(recovery); + await recordingFs(() => framework.quiesce({ reason: 'test' }), (c) => (c.call === 'fsync' && c.path === failing ? fsError('EIO', 'fsync') : undefined)); + assert.equal(JSON.parse(readFileSync(record, 'utf8')).quiesced, true); + assert.deepEqual(alerts, [{ + kind: 'recovery-sync', + agentName: 'framework', + message: `syncing ${failing} failed, so ${record} may not survive a crash: EIO: i/o error, fsync`, + data: { record, directory: failing, code: 'EIO' }, + }]); + }); + }); +} + +for (const code of ['EINVAL', 'EBADF'] as const) { + test(`directory syncs refused with ${code}, as where the filesystem can't sync a directory, are let go with no ops alert and the new record in place`, async () => { + await withFramework(async (framework, recovery) => { + const alerts: unknown[] = []; + (framework as unknown as { opsAlert: unknown }).opsAlert = (...args: unknown[]) => { alerts.push(args); }; + const record = join(recovery, 'host-mode.json'); + const directories = [recovery, dirname(recovery)]; + const calls = await recordingFs( + () => framework.quiesce({ reason: 'test' }), + (c) => (c.call === 'fsync' && directories.includes(c.path!) ? fsError(code, 'fsync') : undefined), + ); + for (const directory of directories) assertSyncedAround(calls, record, directory); + assertEachClosed(calls, (path) => path !== undefined && directories.includes(path)); + assert.equal(JSON.parse(readFileSync(record, 'utf8')).quiesced, true); + assert.deepEqual(alerts, []); + }); + }); +}