diff --git a/docs/development-memory/records/2026/issue-67.jsonl b/docs/development-memory/records/2026/issue-67.jsonl new file mode 100644 index 0000000..ab261fc --- /dev/null +++ b/docs/development-memory/records/2026/issue-67.jsonl @@ -0,0 +1,2 @@ +{"schemaVersion":1,"recordId":"mem-20260906-issue67-clean-async-dedup","recordedAt":"2026-09-06T22:44:00+08:00","agent":{"name":"chatgpt-snapshot-perf","id":"chatgpt-snapshot-perf","model":"GPT-5.6 Sol","role":"implementer","reasoning":"high"},"issue":67,"workItem":"issue-0067","baseCommit":"28bfa9348be03076e92cca00453aa14593f5c94f","candidateRef":"same-commit","summary":"Clean replay of the remaining Issue #67 snapshot hot-path optimization after #88 bounded Python traversal: asynchronous Host snapshot/durable commits, reuse of validated chunk-assembled bytes, safe content-addressed durable dedup, and a repeatable benchmark harness.","files":[{"path":"src/runtime.ts","pointers":["Kernel.commitCheckpointBuffer","Kernel.takeCommittedCheckpointPayload","Kernel.finishResult","async buildRestoreFrames","async publishDurable","durable hash/fingerprint dedup","runEntry payload reuse"]},{"path":"tests/rlm-snapshot-io.test.ts","pointers":["identical checkpoint durable dedup","settle-after-durable barrier","hot-path source guard"]},{"path":"scripts/benchmark-snapshot.ts","pointers":["0.5/2/6 MiB ephemeral/durable benchmark","p50/p95","event-loop p95","peak heap","durable bytes"]},{"path":"package.json","pointers":["bench:snapshot","scripts.test snapshot I/O regression"]}],"steps":["Replayed only the four material Issue #67 files from the already guarded stacked implementation onto post-#73 main; no pre-squash history or temporary workflow was copied.","Kept Issue #88 bounded traversal/context preflight intact and limited this patch to remaining Host I/O/dedup costs.","Chunked checkpoint commit now uses awaited fs/promises before the cell settles; the already validated Buffer is consumed once by runEntry so durable publication does not immediately reread the snapshot.","Durable publication is asynchronous and skips a rewrite only when both content hash and current committed-file fingerprint match the last validated/published generation; external replacement therefore cannot be hidden by hash cache.","Added a non-gating benchmark command reporting p50/p95 latency, event-loop p95, peak heap and durable-root bytes for 0.5/2/6 MiB state in ephemeral/durable modes."],"evidence":[{"kind":"test","target":"stacked guarded production patch workflow 34037971417","result":"PASS","note":"All exact runtime anchors matched and the async/dedup patch committed successfully."},{"kind":"test","target":"clean replay git apply --check","result":"PASS","note":"Issue #67-only material patch applies cleanly to post-#73 main."},{"kind":"test","target":"Windows+Ubuntu PR CI","result":"NOT_RUN","note":"Final authoritative acceptance runs after this clean commit."}],"limitations":["Benchmark output is observational and intentionally has no CI threshold because shared runners are noisy.","Startup durable-root rescan and explicit reset cleanup remain synchronous because they are not the repeated per-cell hot path."]} +{"schemaVersion":1,"recordId":"mem-20260906-issue67-final-ci-trigger","recordedAt":"2026-09-06T22:46:00+08:00","agent":{"name":"chatgpt-snapshot-perf","id":"chatgpt-snapshot-perf","model":"GPT-5.6 Sol","role":"implementer","reasoning":"high"},"issue":67,"workItem":"issue-0067","baseCommit":"28bfa9348be03076e92cca00453aa14593f5c94f","candidateRef":"3eb406e88a0ebcd60b45de9eaf5dbec8d1d9c0f2","summary":"Open PR #110 from the clean post-#73 candidate and trigger the authoritative Windows/Ubuntu/development-memory gate.","files":[{"path":"docs/development-memory/records/2026/issue-67.jsonl","pointers":["append-only final CI trigger"]}],"steps":["Opened Draft PR #110 against current main.","Recorded the exact bot-authored material candidate SHA without editing prior history.","Used this normal append to trigger pull_request CI."],"evidence":[{"kind":"test","target":"clean replay workflow 34038248823","result":"PASS","note":"Issue #67-only patch applied cleanly and the temporary replay workflow self-deleted."},{"kind":"test","target":"PR #110 Windows+Ubuntu CI","result":"NOT_RUN","note":"Triggered by this append; required before merge."}],"limitations":["Benchmark numbers are not claimed until the benchmark is actually executed."]} diff --git a/package.json b/package.json index 811e279..57a0af8 100644 --- a/package.json +++ b/package.json @@ -63,7 +63,8 @@ "check:memory": "node scripts/check-development-memory.mjs --all", "check:memory:staged": "node scripts/check-development-memory.mjs --staged", "check:upstream": "node scripts/check-dsh-upstream.mjs", - "test": "node --test --test-concurrency=1 tests/development-memory-gate.test.ts tests/upstream-authority-gate.test.ts tests/rlm-loop.test.ts tests/rlm-job-lifecycle.test.ts tests/rlm-restore-admission.test.ts tests/rlm-context-metadata.test.ts tests/rlm-checkpoint-transport.test.ts tests/rlm-durable-integrity.test.ts tests/profile-smoke.test.ts tests/rlm-settings.test.ts tests/rlm-settings-manifest.test.ts tests/rlm-python-resolution.test.ts tests/rlm-client-settings.test.ts tests/peer-range.test.ts", + "bench:snapshot": "node scripts/benchmark-snapshot.ts", + "test": "node --test --test-concurrency=1 tests/development-memory-gate.test.ts tests/upstream-authority-gate.test.ts tests/rlm-loop.test.ts tests/rlm-job-lifecycle.test.ts tests/rlm-restore-admission.test.ts tests/rlm-context-metadata.test.ts tests/rlm-checkpoint-transport.test.ts tests/rlm-durable-integrity.test.ts tests/rlm-snapshot-io.test.ts tests/profile-smoke.test.ts tests/rlm-settings.test.ts tests/rlm-settings-manifest.test.ts tests/rlm-python-resolution.test.ts tests/rlm-client-settings.test.ts tests/peer-range.test.ts", "typecheck": "tsc -p tsconfig.json --noEmit" }, "peerDependencies": { @@ -135,4 +136,4 @@ "@deepseek-ai/dsh-storage-domain": "0.1.1-rc.2" } } -} \ No newline at end of file +} diff --git a/scripts/benchmark-snapshot.ts b/scripts/benchmark-snapshot.ts new file mode 100644 index 0000000..2cae394 --- /dev/null +++ b/scripts/benchmark-snapshot.ts @@ -0,0 +1,91 @@ +import { mkdtempSync, readdirSync, rmSync, statSync } from 'node:fs' +import { monitorEventLoopDelay, performance } from 'node:perf_hooks' +import os from 'node:os' +import path from 'node:path' +import { createRlmRuntime } from '../src/runtime.ts' + +const MiB = 1024 * 1024 +const iterations = Math.max(3, Number.parseInt(process.env.RLM_SNAPSHOT_BENCH_ITERATIONS ?? '8', 10) || 8) +const sizes = [0.5, 2, 6] + +type Sample = { + mode: 'ephemeral' | 'durable' + stateMiB: number + p50Ms: number + p95Ms: number + eventLoopP95Ms: number + peakHeapMiB: number + durableRootBytes: number + iterations: number +} + +function percentile(values: number[], p: number): number { + const sorted = [...values].sort((a, b) => a - b) + const index = Math.min(sorted.length - 1, Math.max(0, Math.ceil(sorted.length * p) - 1)) + return sorted[index] ?? 0 +} + +function directoryBytes(root: string | undefined): number { + if (!root) return 0 + let total = 0 + for (const name of readdirSync(root)) { + const info = statSync(path.join(root, name)) + if (info.isFile()) total += info.size + } + return total +} + +async function run(mode: Sample['mode'], stateMiB: number): Promise { + const durableRoot = mode === 'durable' ? mkdtempSync(path.join(os.tmpdir(), 'dsh-rlm-bench-durable-')) : undefined + const runtime = createRlmRuntime(undefined, { + snapshotRecovery: true, + ...(durableRoot ? { durableRoot } : {}), + timeout: 60_000, + }) + const delay = monitorEventLoopDelay({ resolution: 10 }) + const timings: number[] = [] + let peakHeap = process.memoryUsage().heapUsed + try { + const chars = Math.floor(stateMiB * MiB) + const seed = await runtime.eval(`bench-${mode}-${stateMiB}`, { code: `state = 'x' * ${chars}` }) + if (seed.recovery?.checkpointCommitted !== true) throw new Error(`seed checkpoint failed for ${mode}/${stateMiB} MiB`) + + delay.enable() + for (let i = 0; i < iterations; i++) { + const started = performance.now() + const out = await runtime.eval(`bench-${mode}-${stateMiB}`, { code: '1 + 1' }) + timings.push(performance.now() - started) + if (out.result !== '2' || out.recovery?.checkpointCommitted !== true) throw new Error('benchmark cell failed') + peakHeap = Math.max(peakHeap, process.memoryUsage().heapUsed) + } + delay.disable() + + return { + mode, + stateMiB, + p50Ms: Number(percentile(timings, 0.5).toFixed(2)), + p95Ms: Number(percentile(timings, 0.95).toFixed(2)), + eventLoopP95Ms: Number((delay.percentile(95) / 1e6).toFixed(2)), + peakHeapMiB: Number((peakHeap / MiB).toFixed(2)), + durableRootBytes: directoryBytes(durableRoot), + iterations, + } + } finally { + delay.disable() + await runtime.dispose() + if (durableRoot) rmSync(durableRoot, { recursive: true, force: true }) + } +} + +const results: Sample[] = [] +for (const stateMiB of sizes) { + results.push(await run('ephemeral', stateMiB)) + results.push(await run('durable', stateMiB)) +} + +console.log(JSON.stringify({ + node: process.version, + platform: process.platform, + arch: process.arch, + results, +}, null, 2)) diff --git a/src/runtime.ts b/src/runtime.ts index 34a8290..4f6276c 100644 --- a/src/runtime.ts +++ b/src/runtime.ts @@ -3,7 +3,8 @@ import z from '@deepseek-ai/schemastery' import { RLM_SETTINGS_MANIFEST, type RlmRuntimeTierASettings, type RlmSettingsSpec, type RlmTierASettings } from './settings-manifest.ts' import { spawn, type ChildProcess, type SpawnOptions } from 'node:child_process' import { createHash, randomBytes } from 'node:crypto' -import { closeSync, existsSync, fsyncSync, lstatSync, mkdirSync, mkdtempSync, openSync, readFileSync, readdirSync, renameSync, rmSync, writeFileSync } from 'node:fs' +import { existsSync, lstatSync, mkdirSync, mkdtempSync, readFileSync, readdirSync, rmSync, writeFileSync } from 'node:fs' +import { open as openFile, readFile as readFileAsync, rename as renameAsync, rm as rmAsync } from 'node:fs/promises' import os from 'node:os' import path from 'node:path' import { fileURLToPath } from 'node:url' @@ -478,6 +479,8 @@ class Kernel { private readonly launch: KernelLaunch | undefined private restoreSnapshot: boolean private pendingChunks = new Map() + /** Last host-assembled checkpoint bytes; consumed once by the Session runtime. */ + private committedCheckpointPayload: Buffer | undefined private readonly maxSnapshotBytes: number private retainCheckpoint = true /** Kernel capability token -> official child id. Never sent back to Python. */ @@ -924,6 +927,62 @@ class Kernel { if (this.exited) this.evict() } + /** Atomically commit one host-assembled M5 checkpoint without blocking the event loop. */ + private async commitCheckpointBuffer(payload: Buffer): Promise { + if (!this.snapshotPath) throw new Error('snapshot path is undefined') + const temp = this.snapshotPath + '.tmp-' + randomBytes(16).toString('hex') + let handle: Awaited> | undefined + try { + handle = await openFile(temp, 'wx', 0o600) + await handle.writeFile(payload) + await handle.sync() + await handle.close() + handle = undefined + await renameAsync(temp, this.snapshotPath) + } catch (error) { + if (handle !== undefined) { + try { await handle.close() } catch { /* best-effort close */ } + } + try { await rmAsync(temp, { force: true }) } catch { /* private temp cleanup */ } + throw error + } + } + + /** Consume, at most once, the bytes already assembled for the latest chunked checkpoint. */ + takeCommittedCheckpointPayload(): Buffer | undefined { + const payload = this.committedCheckpointPayload + this.committedCheckpointPayload = undefined + return payload + } + + private async finishResult( + p: PendingEval, + out: RlmEvalOutput, + recovery: Record | undefined, + checkpointBuffer: Buffer | undefined, + ): Promise { + if (checkpointBuffer !== undefined && recovery !== undefined) { + try { + await this.commitCheckpointBuffer(checkpointBuffer) + // Reuse exactly these validated bytes for M10 durable publication. + this.committedCheckpointPayload = checkpointBuffer + } catch { + recovery.checkpoint_committed = false + recovery.reason = 'host checkpoint write failed' + } + } + if (recovery !== undefined) { + out.recovery = { + restored: recovery.restored === true, + checkpointCommitted: recovery.checkpoint_committed === true, + } + if (typeof recovery.checkpoint_bytes === 'number') out.recovery.checkpointBytes = recovery.checkpoint_bytes + if (Array.isArray(recovery.skipped)) out.recovery.skipped = recovery.skipped.filter((x): x is string => typeof x === 'string').slice(0, 64) + if (typeof recovery.reason === 'string') out.recovery.reason = recovery.reason + } + await this.finishCell(p, out) + } + private onResult(frame: Frame): void { const p = this.pending if (!p || frame.id !== p.id) { @@ -964,11 +1023,11 @@ class Kernel { } checkpointBuffer = Buffer.concat(chunks.parts, chunks.bytes) try { - const text = checkpointBuffer.toString('utf8') - if (!Buffer.from(text, 'utf8').equals(checkpointBuffer)) { + const checkpointText = checkpointBuffer.toString('utf8') + if (!Buffer.from(checkpointText, 'utf8').equals(checkpointBuffer)) { throw new Error('checkpoint is not canonical UTF-8') } - const envelope = JSON.parse(text) as unknown + const envelope = JSON.parse(checkpointText) as unknown if (typeof envelope !== 'object' || envelope === null || Array.isArray(envelope)) { throw new Error('checkpoint envelope is not an object') } @@ -990,32 +1049,13 @@ class Kernel { truncated: frame.truncated === true, } if (typeof frame.result === 'string') out.result = frame.result - - if (checkpointBuffer !== undefined && recovery !== undefined) { - try { - if (!this.snapshotPath) throw new Error('snapshot path is undefined') - const temp = this.snapshotPath + '.tmp-' + String(process.pid) - writeFileSync(temp, checkpointBuffer) - renameSync(temp, this.snapshotPath) - } catch { - recovery.checkpoint_committed = false - recovery.reason = 'host checkpoint write failed' - } - } this.pendingChunks.delete(p.id) - if (recovery !== undefined) { - out.recovery = { - restored: recovery.restored === true, - checkpointCommitted: recovery.checkpoint_committed === true, - } - if (typeof recovery.checkpoint_bytes === 'number') out.recovery.checkpointBytes = recovery.checkpoint_bytes - if (Array.isArray(recovery.skipped)) out.recovery.skipped = recovery.skipped.filter((x): x is string => typeof x === 'string').slice(0, 64) - if (typeof recovery.reason === 'string') out.recovery.reason = recovery.reason - } - if (!this.cellFinish) this.cellFinish = this.finishCell(p, out) + // The public eval Promise remains unsettled until the awaited checkpoint + // commit and child cleanup barriers finish. Plugin disposal also reuses + // this same cellFinish barrier. + if (!this.cellFinish) this.cellFinish = this.finishResult(p, out, recovery, checkpointBuffer) } - private onError(frame: Frame): void { const p = this.pending if (!p || frame.id !== p.id) { @@ -1167,11 +1207,11 @@ class Kernel { }) } - private buildRestoreFrames(): Frame[] { + private async buildRestoreFrames(): Promise { if (!this.snapshotPath || !existsSync(this.snapshotPath)) { throw new RlmError('snapshot', 'checkpoint file is missing before restore') } - const payload = readFileSync(this.snapshotPath) + const payload = await readFileAsync(this.snapshotPath) const limit = Math.min(MAX_SNAPSHOT_BYTES, this.maxSnapshotBytes) if (payload.length < 1 || payload.length > limit) { throw new RlmError('snapshot', 'checkpoint file exceeds the restore byte limit') @@ -1244,7 +1284,7 @@ class Kernel { else evalFrame.snapshot_path = this.snapshotPath if (restorePending) { evalFrame.restore_snapshot = true - if (chunked) restoreFrames = this.buildRestoreFrames() + if (chunked) restoreFrames = await this.buildRestoreFrames() } } if (input.contextPath !== undefined) evalFrame.context_path = input.contextPath @@ -1500,6 +1540,9 @@ class RlmRuntimeImpl implements RlmRuntime { private readonly durableRoot: string | undefined private readonly durableVersion = 2 private readonly durableAccounting = new Map() + /** Validated content hashes and file fingerprints used for safe same-runtime dedup. */ + private readonly durableHashes = new Map() + private readonly durableFingerprints = new Map() private disposed = false private disposePromise: Promise | undefined constructor(config: RlmRuntimeConfig, ctx?: Context) { @@ -1543,24 +1586,34 @@ class RlmRuntimeImpl implements RlmRuntime { } } - private syncDurableDirectory(): void { + private async syncDurableDirectory(): Promise { if (!this.durableRoot || process.platform === 'win32') return - let fd: number | undefined + let handle: Awaited> | undefined try { - fd = openSync(this.durableRoot, 'r') - fsyncSync(fd) + handle = await openFile(this.durableRoot, 'r') + await handle.sync() } catch (error) { const code = (error as NodeJS.ErrnoException).code if (code !== 'EINVAL' && code !== 'ENOTSUP' && code !== 'EISDIR') { throw this.durableError('durable directory fsync failed') } } finally { - if (fd !== undefined) { - try { closeSync(fd) } catch { /* best-effort close after fsync */ } + if (handle !== undefined) { + try { await handle.close() } catch { /* best-effort close after fsync */ } } } } + private durableFingerprint(info: NonNullable>): string { + return [info.dev, info.ino, info.size, info.mtimeMs, info.ctimeMs].join(':') + } + + private rememberDurableGeneration(sessionKey: string, payload: Buffer, info?: NonNullable>): void { + const key = this.durableFileKey(sessionKey) + this.durableHashes.set(key, createHash('sha256').update(payload).digest('hex')) + const current = info ?? this.durableLstat(this.durablePath(sessionKey, '.checkpoint.json'), 'durable checkpoint') + if (current?.isFile()) this.durableFingerprints.set(key, this.durableFingerprint(current)) + } private encodeDurableEnvelope(bytes: Buffer): Buffer { if (bytes.length < 1 || bytes.length > MAX_SNAPSHOT_BYTES) { throw this.durableError('durable checkpoint exceeds the per-Session byte limit') @@ -1627,6 +1680,8 @@ class RlmRuntimeImpl implements RlmRuntime { */ private rescanDurableRoot(): void { this.durableAccounting.clear() + this.durableHashes.clear() + this.durableFingerprints.clear() if (!this.durableRoot) return let entries: string[] try { @@ -1661,44 +1716,57 @@ class RlmRuntimeImpl implements RlmRuntime { return total } - /** Publish one crash-consistent, host-private durable generation. */ - private publishDurable(sessionKey: string, bytes: Buffer): { published: boolean; reason?: string } { + /** Publish one crash-consistent durable generation without blocking the Host event loop. */ + private async publishDurable(sessionKey: string, bytes: Buffer): Promise<{ published: boolean; reason?: string }> { if (!this.durableRoot) return { published: false } + const key = this.durableFileKey(sessionKey) const target = this.durablePath(sessionKey, '.checkpoint.json') const legacyMeta = this.durablePath(sessionKey, '.meta.json') + const contentHash = createHash('sha256').update(bytes).digest('hex') + + // Safe same-runtime content-addressed dedup. The hash alone is not enough: + // confirm the committed file identity has not changed since it was last + // validated/published, so an external replacement cannot be hidden by cache. + const current = this.durableLstat(target, 'durable checkpoint') + if ( + current?.isFile() + && this.durableHashes.get(key) === contentHash + && this.durableFingerprints.get(key) === this.durableFingerprint(current) + ) { + return { published: true } + } + const temp = target + '.tmp-' + randomBytes(16).toString('hex') const envelope = this.encodeDurableEnvelope(bytes) - // Issue #89: charge only the delta against the persisted aggregate. An - // over-quota candidate keeps the previous valid generation and never - // fails the already-successful user cell. - const oldTotal = this.durableAccounting.get(this.durableFileKey(sessionKey)) ?? 0 + const oldTotal = this.durableAccounting.get(key) ?? 0 if (this.durableTotalBytes() + (envelope.length - oldTotal) > MAX_SNAPSHOT_ROOT_BYTES) { return { published: false, reason: 'durable-root quota exceeded' } } - let fd: number | undefined + let handle: Awaited> | undefined try { - fd = openSync(temp, 'wx', 0o600) - writeFileSync(fd, envelope) - fsyncSync(fd) - closeSync(fd) - fd = undefined - // The prior committed target remains valid until this single rename. - renameSync(temp, target) - // A stale legacy sidecar is irrelevant once target is a v2 envelope. - try { rmSync(legacyMeta, { force: true }) } catch { /* best-effort migration cleanup */ } - this.syncDurableDirectory() - this.durableAccounting.set(this.durableFileKey(sessionKey), envelope.length) + handle = await openFile(temp, 'wx', 0o600) + await handle.writeFile(envelope) + await handle.sync() + await handle.close() + handle = undefined + await renameAsync(temp, target) + try { await rmAsync(legacyMeta, { force: true }) } catch { /* best-effort migration cleanup */ } + await this.syncDurableDirectory() + const committed = this.durableLstat(target, 'durable checkpoint') + if (!committed?.isFile()) throw this.durableError('durable checkpoint disappeared after publication') + this.durableAccounting.set(key, envelope.length) + this.durableHashes.set(key, contentHash) + this.durableFingerprints.set(key, this.durableFingerprint(committed)) return { published: true } } catch (error) { - if (fd !== undefined) { - try { closeSync(fd) } catch { /* best-effort close */ } + if (handle !== undefined) { + try { await handle.close() } catch { /* best-effort close */ } } - try { rmSync(temp, { force: true }) } catch { /* never follow or expose temp paths */ } + try { await rmAsync(temp, { force: true }) } catch { /* never expose temp paths */ } if (error instanceof RlmError) throw error throw this.durableError('durable checkpoint publication failed') } } - /** Read a committed v2 envelope, with strict read-only compatibility for M10 v1 pairs. */ private readDurable(sessionKey: string): Buffer | undefined { if (!this.durableRoot) return undefined @@ -1728,7 +1796,9 @@ class RlmRuntimeImpl implements RlmRuntime { && !Array.isArray(candidate) && (candidate as Record).magic === DURABLE_MAGIC ) { - return this.decodeDurableEnvelope(container) + const payload = this.decodeDurableEnvelope(container) + this.rememberDurableGeneration(sessionKey, payload, targetInfo) + return payload } } catch { if (!legacyMetaInfo) throw this.durableError('durable checkpoint header is malformed') @@ -1761,6 +1831,7 @@ class RlmRuntimeImpl implements RlmRuntime { ) { throw this.durableError('durable content hash mismatch') } + this.rememberDurableGeneration(sessionKey, container, targetInfo) return container } @@ -1769,7 +1840,10 @@ class RlmRuntimeImpl implements RlmRuntime { rmSync(this.durablePath(sessionKey, '.checkpoint.json'), { force: true }) rmSync(this.durablePath(sessionKey, '.meta.json'), { force: true }) // Issue #89: reset releases the Session's persistent quota share. - this.durableAccounting.delete(this.durableFileKey(sessionKey)) + const key = this.durableFileKey(sessionKey) + this.durableAccounting.delete(key) + this.durableHashes.delete(key) + this.durableFingerprints.delete(key) } private snapshotFor(sessionKey: string): { path: string; restore: boolean; maxBytes: number } | undefined { @@ -2098,12 +2172,14 @@ class RlmRuntimeImpl implements RlmRuntime { ...(entry.input.signal ? { signal: entry.input.signal } : {}), }) const out = await kernel.evalCell(entry.input, entry.deadline) + const assembledCheckpoint = kernel.takeCommittedCheckpointPayload() if (out.recovery?.checkpointCommitted) { this.checkpoints.add(sessionKey) if (this.durableRoot && this.checkpointRoot) { const p = this.checkpointPath(sessionKey) if (existsSync(p)) { - const durable = this.publishDurable(sessionKey, readFileSync(p)) + const payload = assembledCheckpoint ?? await readFileAsync(p) + const durable = await this.publishDurable(sessionKey, payload) if (!durable.published && out.recovery) { out.recovery.durable = durable } diff --git a/tests/rlm-snapshot-io.test.ts b/tests/rlm-snapshot-io.test.ts new file mode 100644 index 0000000..39b0d74 --- /dev/null +++ b/tests/rlm-snapshot-io.test.ts @@ -0,0 +1,87 @@ +import test from 'node:test' +import assert from 'node:assert/strict' +import { mkdtempSync, readFileSync, readdirSync, rmSync, statSync } from 'node:fs' +import os from 'node:os' +import path from 'node:path' +import { createRlmRuntime } from '../src/runtime.ts' + +const sleep = (ms: number) => new Promise(resolve => setTimeout(resolve, ms)) + +async function removeTree(directory: string): Promise { + let last: unknown + for (let attempt = 0; attempt < 30; attempt++) { + try { + rmSync(directory, { recursive: true, force: true }) + return + } catch (error) { + last = error + await sleep(50) + } + } + throw last +} + +function durableTarget(root: string): string { + const files = readdirSync(root).filter(name => name.endsWith('.checkpoint.json')) + assert.equal(files.length, 1, 'one Session should own exactly one durable generation') + return path.join(root, files[0]!) +} + +test('Issue#67: identical checkpoint content does not rewrite the durable generation', async () => { + const durableRoot = mkdtempSync(path.join(os.tmpdir(), 'dsh-rlm-i67-dedup-')) + const runtime = createRlmRuntime(undefined, { snapshotRecovery: true, durableRoot, timeout: 12_000 }) + try { + const first = await runtime.eval('issue67-dedup', { code: "stable = 'x' * 200000" }) + assert.equal(first.recovery?.checkpointCommitted, true) + const target = durableTarget(durableRoot) + const before = statSync(target, { bigint: true }) + const bytesBefore = readFileSync(target) + + // Force a distinct filesystem timestamp opportunity. The second cell does + // not mutate globals, so the deterministic checkpoint payload is identical. + await sleep(75) + const second = await runtime.eval('issue67-dedup', { code: '1 + 1' }) + assert.equal(second.result, '2') + assert.equal(second.recovery?.checkpointCommitted, true) + + const after = statSync(target, { bigint: true }) + assert.equal(after.mtimeNs, before.mtimeNs, 'content-addressed dedup must keep the committed generation untouched') + assert.deepEqual(readFileSync(target), bytesBefore) + } finally { + await runtime.dispose() + await removeTree(durableRoot) + } +}) + +test('Issue#67: eval settles only after durable commit is immediately restorable', async () => { + const durableRoot = mkdtempSync(path.join(os.tmpdir(), 'dsh-rlm-i67-barrier-')) + const writer = createRlmRuntime(undefined, { snapshotRecovery: true, durableRoot, timeout: 12_000 }) + try { + const saved = await writer.eval('issue67-barrier', { code: 'keep = 41' }) + assert.equal(saved.recovery?.checkpointCommitted, true) + assert.ok(readFileSync(durableTarget(durableRoot)).length > 0, 'durable generation must exist before eval resolves') + } finally { + await writer.dispose() + } + + const reader = createRlmRuntime(undefined, { snapshotRecovery: true, durableRoot, timeout: 12_000 }) + try { + const restored = await reader.eval('issue67-barrier', { code: 'keep + 1' }) + assert.equal(restored.result, '42') + assert.equal(restored.recovery?.restored, true) + } finally { + await reader.dispose() + await removeTree(durableRoot) + } +}) + +test('Issue#67: hot checkpoint publication uses awaited async I/O and reuses chunk payloads', () => { + const source = readFileSync(new URL('../src/runtime.ts', import.meta.url), 'utf8') + assert.match(source, /from 'node:fs\/promises'/) + assert.match(source, /await this\.publishDurable\(/) + assert.match(source, /takeCommittedCheckpointPayload\(\)/) + const runEntry = source.slice(source.indexOf('private async runEntry'), source.indexOf('\n dispose(): Promise', source.indexOf('private async runEntry'))) + assert.doesNotMatch(runEntry, /readFileSync\(p\)/, 'runEntry must not synchronously reread a just-committed snapshot') + const onResult = source.slice(source.indexOf('private onResult'), source.indexOf('\n private onError', source.indexOf('private onResult'))) + assert.doesNotMatch(onResult, /writeFileSync|renameSync/, 'chunked checkpoint commit must not block the Host event loop') +})