fix(settings-local): one operation chain, read-modify-write under a writer lock, and diff-shaped YAML edits
Review round three found the provider's write path could destroy state it never observed: - Watcher reloads and document writes ran on two independent promise chains, and a write rendered the whole next document from the cached text. An external edit still inside the debounce window (or missed outright) was overwritten, and the follow-up reload no-oped because the post-rename content matched the cache — the edit vanished without a trace. Reloads and writes now share one operation chain, and every write starts by reconciling the on-disk text into the seam before rendering, so unobserved sibling sections survive and publish first. An unparsable on-disk document fails the write loud instead of being overwritten. - The initial load raced the watcher's own setup: a change written between that read and the watcher becoming active never fired an event. The watcher's ready signal now queues one reconcile, closing the gap. - Two processes sharing a harness home rendered from independent caches, last writer winning. Writes now hold a wx-created <file>.lock sibling around the read-render-rename cycle with bounded backoff, a crashed- holder stale takeover, and a deadline failure; readers stay lock-free because the rename commit is atomic. - renderYaml replaced the whole namespace node, dropping every comment inside the section. The next section now lands as a leaf-level diff (set changed values, delete removed keys), so comments, anchors, and formatting survive on every untouched node and on the key of every changed pair; arrays still replace wholesale when unequal.
This commit is contained in:
Yichen Jiang
@@ -9,11 +9,11 @@ import { Context, Service } from 'cordis'
|
||||
import z from 'schemastery'
|
||||
import { watch as chokidarWatch } from 'chokidar'
|
||||
import { randomBytes } from 'node:crypto'
|
||||
import { mkdir, readFile, rename, rm, writeFile } from 'node:fs/promises'
|
||||
import { mkdir, readFile, rename, rm, stat, writeFile } from 'node:fs/promises'
|
||||
import { dirname, extname, join, resolve } from 'node:path'
|
||||
import { Document, parseDocument } from 'yaml'
|
||||
import { resolveDshHome } from '@deepseek-ai/dsh-paths'
|
||||
import { Settings, type SettingsNamespace } from '@deepseek-ai/dsh-settings'
|
||||
import { Settings, deepEqualJson, type SettingsNamespace } from '@deepseek-ai/dsh-settings'
|
||||
|
||||
/** Plugin config: file location and hot-reload behavior. */
|
||||
export interface Config {
|
||||
@@ -64,11 +64,53 @@ export function resolveSpec(config: Config): ResolvedSpec {
|
||||
}
|
||||
}
|
||||
|
||||
/** Whether a parsed YAML value is a map for diffing purposes. */
|
||||
function isMapLike(value: unknown): value is Record<string, unknown> {
|
||||
return typeof value === 'object' && value !== null && !Array.isArray(value)
|
||||
}
|
||||
|
||||
/**
|
||||
* Apply the difference between one node's stored and next value as minimal
|
||||
* `setIn`/`deleteIn` edits, recursing through maps, so every untouched node —
|
||||
* and the key node of every changed pair — keeps its comments, anchors, and
|
||||
* formatting. Non-map values (arrays and scalars) replace wholesale when
|
||||
* unequal, taking any comments inside them along.
|
||||
*/
|
||||
function patchNode(document: Document, path: readonly string[], current: unknown, next: unknown): void {
|
||||
if (isMapLike(current) && isMapLike(next)) {
|
||||
for (const key of Object.keys(current)) {
|
||||
if (!(key in next)) document.deleteIn([...path, key])
|
||||
}
|
||||
for (const [key, value] of Object.entries(next)) {
|
||||
patchNode(document, [...path, key], current[key], value)
|
||||
}
|
||||
return
|
||||
}
|
||||
if (!deepEqualJson(current, next)) document.setIn([...path], next)
|
||||
}
|
||||
|
||||
/** Whether a filesystem error means absence; every non-ENOENT failure must surface. */
|
||||
function isENOENT(error: unknown): boolean {
|
||||
return (error as NodeJS.ErrnoException | null)?.code === 'ENOENT'
|
||||
}
|
||||
|
||||
/** Whether an exclusive create failed because the path already exists. */
|
||||
function isEEXIST(error: unknown): boolean {
|
||||
return (error as NodeJS.ErrnoException | null)?.code === 'EEXIST'
|
||||
}
|
||||
|
||||
/**
|
||||
* Writer-lock protocol constants. These are robustness invariants of the
|
||||
* cross-process write protocol, not deployment tunables: a holder rewrites one
|
||||
* small document in milliseconds, so contention resolves well inside the
|
||||
* retry deadline, and a lock older than the stale age can only belong to a
|
||||
* crashed holder.
|
||||
*/
|
||||
const LOCK_RETRY_INITIAL_MS = 20
|
||||
const LOCK_RETRY_MAX_MS = 200
|
||||
const LOCK_TIMEOUT_MS = 2_000
|
||||
const LOCK_STALE_MS = 5_000
|
||||
|
||||
/** File-backed settings provider (`settings.yaml`/`.json`). */
|
||||
export class SettingsLocal extends Settings {
|
||||
static Config: z<Config> = z.object({
|
||||
@@ -85,10 +127,13 @@ export class SettingsLocal extends Settings {
|
||||
* this cache are no-ops, which is also the self-write suppression.
|
||||
*/
|
||||
private text: string | undefined
|
||||
/** Serializes watcher-triggered reloads so reads never interleave. */
|
||||
private refreshTask: Promise<void> = Promise.resolve()
|
||||
/** Serializes whole-document writes across namespace queues; settled tail. */
|
||||
private persistChain: Promise<void> = Promise.resolve()
|
||||
/**
|
||||
* Single exclusive operation chain: watcher reloads and document writes run
|
||||
* one at a time in queue order (settled tail), so a write can never render
|
||||
* from text a concurrent reload is busy replacing, and a reload can never
|
||||
* read a half-committed write.
|
||||
*/
|
||||
private operations: Promise<void> = Promise.resolve()
|
||||
/** Set at dispose: refuse new watcher events and let in-flight work no-op. */
|
||||
private closed = false
|
||||
|
||||
@@ -125,32 +170,107 @@ export class SettingsLocal extends Settings {
|
||||
|
||||
protected persist(ns: SettingsNamespace, section: Record<string, unknown>): Promise<void> {
|
||||
// One document backs every namespace, so writes from different namespace
|
||||
// queues must serialize here: each render must see the text the previous
|
||||
// write committed, or the loser's section silently vanishes from disk.
|
||||
// The stored tail is settled on both outcomes, so chaining needs no catch.
|
||||
const task = this.persistChain.then(() => this.persistSection(ns, section))
|
||||
this.persistChain = task.then(() => undefined, () => undefined)
|
||||
// queues serialize with each other and with watcher reloads on the one
|
||||
// operation chain: each render must see the text the previous operation
|
||||
// committed, or a sibling section silently vanishes from disk.
|
||||
return this.enqueue(() => this.persistSection(ns, section))
|
||||
}
|
||||
|
||||
/** Queue one exclusive document operation behind every earlier one. */
|
||||
private enqueue<T>(operation: () => Promise<T>): Promise<T> {
|
||||
const task = this.operations.then(operation)
|
||||
this.operations = task.then(() => undefined, () => undefined)
|
||||
return task
|
||||
}
|
||||
|
||||
/** Queue a reload; only an invariant violation escaping a commit can reject it. */
|
||||
private queueRefresh(): void {
|
||||
void this.enqueue(() => this.refresh()).catch((error: unknown) => {
|
||||
// Only an invariant violation escaping the commit path can reject a
|
||||
// refresh; keep the operation queue alive and surface it as an error so
|
||||
// one poisoned commit cannot silently end hot reloading forever.
|
||||
this.ctx.logger.error('settings-local: reload commit failed at %s', this.spec.filename)
|
||||
this.ctx.logger.error(error)
|
||||
})
|
||||
}
|
||||
|
||||
private async persistSection(ns: SettingsNamespace, section: Record<string, unknown>): Promise<void> {
|
||||
const output = this.spec.format === 'yaml'
|
||||
? this.renderYaml(ns, section)
|
||||
: this.renderJson(ns, section)
|
||||
await mkdir(dirname(this.spec.filename), { recursive: true })
|
||||
// Exclusive-create (`wx`) a random-suffix sibling: the open refuses to
|
||||
// follow any planted symlink at a guessable temp path, and the fresh inode
|
||||
// carries owner-only permissions that survive the rename — a document that
|
||||
// may hold personal values is never world-readable and never a symlink.
|
||||
const temp = `${this.spec.filename}.${randomBytes(6).toString('hex')}.tmp`
|
||||
try {
|
||||
await writeFile(temp, output, { mode: 0o600, flag: 'wx' })
|
||||
await rename(temp, this.spec.filename)
|
||||
} catch (error) {
|
||||
await rm(temp, { force: true })
|
||||
throw error
|
||||
await this.withWriterLock(async () => {
|
||||
// Read-modify-write: fold in any on-disk state this process has not
|
||||
// observed yet — an external edit still inside the watcher debounce
|
||||
// window, a change the watcher missed, or another process's write — so
|
||||
// the render below can never resurrect a stale document. An unparsable
|
||||
// on-disk document fails the write loud instead of silently overwriting
|
||||
// a user's manual edit.
|
||||
await this.reconcileFromDisk()
|
||||
const output = this.spec.format === 'yaml'
|
||||
? this.renderYaml(ns, section)
|
||||
: this.renderJson(ns, section)
|
||||
// Exclusive-create (`wx`) a random-suffix sibling: the open refuses to
|
||||
// follow any planted symlink at a guessable temp path, and the fresh inode
|
||||
// carries owner-only permissions that survive the rename — a document that
|
||||
// may hold personal values is never world-readable and never a symlink.
|
||||
const temp = `${this.spec.filename}.${randomBytes(6).toString('hex')}.tmp`
|
||||
try {
|
||||
await writeFile(temp, output, { mode: 0o600, flag: 'wx' })
|
||||
await rename(temp, this.spec.filename)
|
||||
} catch (error) {
|
||||
await rm(temp, { force: true })
|
||||
throw error
|
||||
}
|
||||
this.text = output
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Hold the cross-process writer lock around one read-render-rename cycle.
|
||||
* The lock is a `wx`-created sibling (`<file>.lock`); the rename-based
|
||||
* commit keeps readers lock-free, so only writers contend. A lock older
|
||||
* than {@link LOCK_STALE_MS} is a crashed holder and is broken with a
|
||||
* warning; a live holder past {@link LOCK_TIMEOUT_MS} fails the write.
|
||||
*/
|
||||
private async withWriterLock<T>(operation: () => Promise<T>): Promise<T> {
|
||||
const lockPath = `${this.spec.filename}.lock`
|
||||
const deadline = Date.now() + LOCK_TIMEOUT_MS
|
||||
let delay = LOCK_RETRY_INITIAL_MS
|
||||
for (;;) {
|
||||
try {
|
||||
await writeFile(lockPath, `${process.pid}\n`, { mode: 0o600, flag: 'wx' })
|
||||
break
|
||||
} catch (error) {
|
||||
if (!isEEXIST(error)) throw error
|
||||
}
|
||||
const ageMs = await this.lockAgeMs(lockPath)
|
||||
// The holder released between the failed create and the stat: the lock
|
||||
// is free right now, so retry without burning backoff or deadline.
|
||||
if (ageMs === undefined) continue
|
||||
if (ageMs > LOCK_STALE_MS) {
|
||||
this.ctx.logger.warn('settings-local: breaking a stale writer lock at %s', lockPath)
|
||||
await rm(lockPath, { force: true })
|
||||
continue
|
||||
}
|
||||
if (Date.now() >= deadline) {
|
||||
throw new Error(`settings-local: timed out waiting for the writer lock at ${lockPath}`)
|
||||
}
|
||||
await new Promise(resolve => setTimeout(resolve, delay))
|
||||
delay = Math.min(delay * 2, LOCK_RETRY_MAX_MS)
|
||||
}
|
||||
try {
|
||||
return await operation()
|
||||
} finally {
|
||||
await rm(lockPath, { force: true })
|
||||
}
|
||||
}
|
||||
|
||||
/** Age of the writer lock, or `undefined` when it vanished after a failed create. */
|
||||
private async lockAgeMs(lockPath: string): Promise<number | undefined> {
|
||||
try {
|
||||
return Date.now() - (await stat(lockPath)).mtimeMs
|
||||
} catch (error) {
|
||||
if (!isENOENT(error)) throw error
|
||||
return undefined
|
||||
}
|
||||
this.text = output
|
||||
}
|
||||
|
||||
override async* [Service.init](): AsyncGenerator<() => Promise<void> | void, void, void> {
|
||||
@@ -168,13 +288,14 @@ export class SettingsLocal extends Settings {
|
||||
})
|
||||
watcher.on('all', () => {
|
||||
if (this.closed) return
|
||||
this.refreshTask = this.refreshTask.then(() => this.refresh()).catch((error: unknown) => {
|
||||
// Only an invariant violation escaping the commit path can reject a
|
||||
// refresh; keep the reload queue alive and surface it as an error so
|
||||
// one poisoned commit cannot silently end hot reloading forever.
|
||||
this.ctx.logger.error('settings-local: reload commit failed at %s', this.spec.filename)
|
||||
this.ctx.logger.error(error)
|
||||
})
|
||||
this.queueRefresh()
|
||||
})
|
||||
watcher.on('ready', () => {
|
||||
// The base init's load raced the watcher's own setup: a change written
|
||||
// between that read and the watcher becoming active never fires an
|
||||
// event. One reconcile at ready closes the gap.
|
||||
if (this.closed) return
|
||||
this.queueRefresh()
|
||||
})
|
||||
watcher.on('error', (error) => {
|
||||
this.ctx.logger.warn('settings-local: watcher error on %s', this.spec.filename)
|
||||
@@ -182,10 +303,10 @@ export class SettingsLocal extends Settings {
|
||||
})
|
||||
yield async () => {
|
||||
// Quiesce: stop accepting events, close the watcher, then wait out any
|
||||
// queued or in-flight refresh so nothing publishes after disposal.
|
||||
// queued or in-flight operation so nothing publishes after disposal.
|
||||
this.closed = true
|
||||
await watcher.close()
|
||||
await this.refreshTask
|
||||
await this.operations
|
||||
}
|
||||
}
|
||||
|
||||
@@ -212,46 +333,62 @@ export class SettingsLocal extends Settings {
|
||||
* Re-read the document after a watcher event. Unchanged content (including
|
||||
* this provider's own writes) is a no-op; an unreadable or unparsable
|
||||
* document keeps the last good sections and warns — a live hot-reload must
|
||||
* never take the process down.
|
||||
* never take the process down. An invariant violation escaping a commit is
|
||||
* not a reload failure and propagates to the queue's error surface.
|
||||
*/
|
||||
private async refresh(): Promise<void> {
|
||||
if (this.closed) return
|
||||
let text: string
|
||||
try {
|
||||
await this.reconcileFromDisk()
|
||||
} catch (error) {
|
||||
if ((error as { code?: unknown } | null)?.code === 'INVARIANT') throw error
|
||||
this.ctx.logger.warn('settings-local: reload failed at %s; keeping the last good document', this.spec.filename)
|
||||
this.ctx.logger.warn(error)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Compare the on-disk text against the cache and publish any difference
|
||||
* into the seam. Absence publishes the empty document; an unreadable or
|
||||
* unparsable file throws, so each caller picks its policy — a reload warns
|
||||
* and keeps the last good document, a write fails loud.
|
||||
*/
|
||||
private async reconcileFromDisk(): Promise<void> {
|
||||
let text: string | undefined
|
||||
try {
|
||||
text = await readFile(this.spec.filename, 'utf8')
|
||||
} catch (error) {
|
||||
if (!isENOENT(error)) {
|
||||
this.ctx.logger.warn('settings-local: reload failed at %s; keeping the last good document', this.spec.filename)
|
||||
this.ctx.logger.warn(error)
|
||||
return
|
||||
}
|
||||
if (this.text === undefined || this.isClosed()) return
|
||||
if (!isENOENT(error)) throw error
|
||||
text = undefined
|
||||
}
|
||||
if (text === this.text || this.isClosed()) return
|
||||
if (text === undefined) {
|
||||
this.text = undefined
|
||||
this.publish({})
|
||||
return
|
||||
}
|
||||
if (text === this.text || this.isClosed()) return
|
||||
let doc: Record<string, unknown>
|
||||
try {
|
||||
doc = this.parse(text)
|
||||
} catch (error) {
|
||||
this.ctx.logger.warn('settings-local: reload failed at %s; keeping the last good document', this.spec.filename)
|
||||
this.ctx.logger.warn(error)
|
||||
return
|
||||
}
|
||||
const doc = this.parse(text)
|
||||
this.text = text
|
||||
this.publish(doc)
|
||||
}
|
||||
|
||||
/** Render the next YAML text by patching one namespace in the comment-preserving document. */
|
||||
/**
|
||||
* Render the next YAML text by patching one namespace in the
|
||||
* comment-preserving document. The next section lands as a leaf-level diff
|
||||
* against the stored one — only changed values set, only removed keys
|
||||
* delete — so comments inside the section survive edits to their siblings,
|
||||
* not just comments outside it.
|
||||
*/
|
||||
private renderYaml(ns: SettingsNamespace, section: Record<string, unknown>): string {
|
||||
if (this.text === undefined) {
|
||||
return new Document({ [ns]: section }).toString()
|
||||
}
|
||||
// this.text only ever caches content that parsed successfully, so this
|
||||
// re-parse (for the mutable comment-preserving tree) cannot fail.
|
||||
// re-parse (for the mutable comment-preserving tree) cannot fail, and
|
||||
// parse() already rejected any non-map root.
|
||||
const document = parseDocument(this.text)
|
||||
document.set(ns, section)
|
||||
const root: unknown = document.toJS()
|
||||
patchNode(document, [ns], isMapLike(root) ? root[ns] : undefined, section)
|
||||
return document.toString()
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user