refactor(credentials,llm): remove speculative mutation and route lifecycle
This commit is contained in:
@@ -27,7 +27,6 @@
|
||||
],
|
||||
"license": "BSD-3-Clause",
|
||||
"peerDependencies": {
|
||||
"@deepseek-ai/dsh-atomic-write": "^0.0.1",
|
||||
"@deepseek-ai/dsh-invariants": "^0.0.1",
|
||||
"@deepseek-ai/dsh-paths": "^0.0.1",
|
||||
"@deepseek-ai/dsh-settings": "^0.0.1",
|
||||
@@ -39,7 +38,6 @@
|
||||
"yaml": "^2.9.0"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@deepseek-ai/dsh-atomic-write": "workspace:^",
|
||||
"@deepseek-ai/dsh-invariants": "workspace:^",
|
||||
"@deepseek-ai/dsh-paths": "workspace:^",
|
||||
"@deepseek-ai/dsh-settings": "workspace:^",
|
||||
|
||||
@@ -10,10 +10,10 @@
|
||||
import { Context, Service } from 'cordis'
|
||||
import z from 'schemastery'
|
||||
import { watch as chokidarWatch } from 'chokidar'
|
||||
import { mkdir, readFile } from 'node:fs/promises'
|
||||
import { randomBytes } from 'node:crypto'
|
||||
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 { withFileLock, writeFileAtomic } from '@deepseek-ai/dsh-atomic-write'
|
||||
import { resolveDshHome } from '@deepseek-ai/dsh-paths'
|
||||
import { Settings, deepEqualJson, type SettingsNamespace } from '@deepseek-ai/dsh-settings'
|
||||
|
||||
@@ -96,6 +96,17 @@ 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 retry constants for the private settings document protocol. */
|
||||
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({
|
||||
@@ -180,11 +191,8 @@ export class SettingsLocal extends Settings {
|
||||
}
|
||||
|
||||
private async persistSection(ns: SettingsNamespace, section: Record<string, unknown>): Promise<void> {
|
||||
// The writer lock's exclusive create needs the parent to exist before
|
||||
// writeFileAtomic gets its own chance to create it.
|
||||
// 0700: the harness home holds user-private documents.
|
||||
await mkdir(dirname(this.spec.filename), { recursive: true, mode: 0o700 })
|
||||
await withFileLock(this.spec.filename, async () => {
|
||||
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
|
||||
@@ -195,16 +203,64 @@ export class SettingsLocal extends Settings {
|
||||
const output = this.spec.format === 'yaml'
|
||||
? this.renderYaml(ns, section)
|
||||
: this.renderJson(ns, section)
|
||||
// 0600: a document that may hold personal values is never world-readable.
|
||||
await writeFileAtomic(this.spec.filename, output, { mode: 0o600, dirMode: 0o700 })
|
||||
const temp = `${this.spec.filename}.${randomBytes(6).toString('hex')}.tmp`
|
||||
// TODO(settings-atomic-durability): Use a replacement that fsyncs the file
|
||||
// and parent directory and preserves owner-only permissions on Windows.
|
||||
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
|
||||
}, {
|
||||
onStaleBreak: (lockPath) => {
|
||||
this.ctx.logger.warn('settings-local: breaking a stale writer lock at %s', lockPath)
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
/** Hold the private cross-process writer lock around one read-render-rename cycle. */
|
||||
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)
|
||||
if (ageMs === undefined) continue
|
||||
if (ageMs > LOCK_STALE_MS) {
|
||||
// TODO(settings-lock-ownership): Replace age-only takeover with ownership-safe
|
||||
// acquisition and release so a slow writer cannot remove a successor's lock.
|
||||
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(resolvePause => setTimeout(resolvePause, 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
|
||||
}
|
||||
}
|
||||
|
||||
override async* [Service.init](): AsyncGenerator<() => Promise<void> | void, void, void> {
|
||||
// The base init loads and publishes; a parse failure there is a boot
|
||||
// failure: an existing-but-invalid document must fail loud, never be
|
||||
|
||||
@@ -16,7 +16,7 @@ export const inject = ['invariants']
|
||||
|
||||
/**
|
||||
* No runtime invariant: this provider's contracts are file round-trip,
|
||||
* watcher timing, and atomic-write behavior — IO effects proven by package
|
||||
* watcher timing, and atomic replacement behavior — IO effects proven by package
|
||||
* tests; the in-process commit relation is owned by `@deepseek-ai/dsh-settings`.
|
||||
*/
|
||||
const install: InvariantInstaller = () => {}
|
||||
|
||||
@@ -17,9 +17,6 @@
|
||||
{
|
||||
"path": "../../../vendor/schemastery"
|
||||
},
|
||||
{
|
||||
"path": "../../util/atomic-write"
|
||||
},
|
||||
{
|
||||
"path": "../../util/paths"
|
||||
},
|
||||
|
||||
@@ -546,34 +546,14 @@ export abstract class Settings extends Service {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Value mirror of the `FiberState` members {@link isUnloading} compares
|
||||
* against: a const enum has no runtime object to import, and the value is
|
||||
* needed at runtime (same rationale as the CLI boot driver's mirror).
|
||||
*/
|
||||
const FIBER_DISPOSED = 4
|
||||
const FIBER_UNLOADING = 5
|
||||
|
||||
/** Whether the consumer's own fiber is tearing down (not just losing the settings service). */
|
||||
function isUnloading(ctx: Context): boolean {
|
||||
const state: number = ctx.fiber.state
|
||||
return state === FIBER_UNLOADING || state === FIBER_DISPOSED
|
||||
}
|
||||
|
||||
/** Hooks a consumer hands to {@link installSettingsSection}. */
|
||||
export interface SettingsSectionHooks<T> {
|
||||
/**
|
||||
* Receive the active configuration source: the resolved settings scope
|
||||
* while one is attached, the composition entry otherwise. Called before
|
||||
* the matching `onChange` at attach and at detach.
|
||||
* while one is attached, the composition entry otherwise.
|
||||
* @param current - thunk returning the currently authoritative value.
|
||||
*/
|
||||
setSource(current: () => T): void
|
||||
/**
|
||||
* Re-judge anything derived from the source — registration-level facts,
|
||||
* memoized resolutions — after an attach, a detach, or a committed change.
|
||||
*/
|
||||
onChange(): void
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -581,8 +561,10 @@ export interface SettingsSectionHooks<T> {
|
||||
* service exists, register `ns` with the consumer's composition entry as the
|
||||
* `base` layer and point the source thunk at the resolved scope; when the
|
||||
* service goes away (disposal, provider reload), fall back to the entry so
|
||||
* the consumer keeps working exactly as composed. The registration rides the
|
||||
* scoped fiber, so no settings service ever mounted means none of this runs.
|
||||
* the consumer keeps working exactly as composed. The returned source is live:
|
||||
* callers read committed changes from `scope.get()` without a change callback.
|
||||
* The registration rides the scoped fiber, so no settings service ever mounted
|
||||
* means none of this runs.
|
||||
* @param ctx - consumer plugin context owning the wiring.
|
||||
* @param ns - the consumer-owned settings namespace.
|
||||
* @param schema - schema resolving the namespace (typically the plugin Config).
|
||||
@@ -600,24 +582,7 @@ export function installSettingsSection<T>(
|
||||
const scope = sctx.settings.register(ns, schema, { base: entry })
|
||||
hooks.setSource(() => scope.get())
|
||||
sctx.effect(() => () => {
|
||||
// This disposer runs for two different reasons. A settings provider
|
||||
// detaching leaves the consumer running, so it must fall back to its
|
||||
// composition entry and re-judge what it derived. The consumer's own
|
||||
// unload runs it too — and there `onChange` would re-register routes
|
||||
// and touch resources the teardown is releasing, so the fallback is
|
||||
// pointless and the notification actively harmful.
|
||||
if (isUnloading(ctx)) return
|
||||
hooks.setSource(() => entry)
|
||||
hooks.onChange()
|
||||
})
|
||||
hooks.onChange()
|
||||
scope.watch(() => {
|
||||
// A stored change landing while the consumer unloads reaches the watcher
|
||||
// before the registration is released, and `onChange` is exactly as
|
||||
// harmful here as in the disposer above: it re-registers routes against
|
||||
// a fiber whose resources are being let go.
|
||||
if (isUnloading(ctx)) return
|
||||
hooks.onChange()
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
@@ -662,98 +662,23 @@ describe('installSettingsSection', () => {
|
||||
const ctx = new Context()
|
||||
const entry = { theme: 'entry' }
|
||||
let current: () => { theme: string } = () => entry
|
||||
let changes = 0
|
||||
installSettingsSection(ctx, settingsNamespace('helper-ns'), HelperSchema, entry, {
|
||||
setSource: (source) => {
|
||||
current = source
|
||||
},
|
||||
onChange: () => {
|
||||
changes += 1
|
||||
},
|
||||
})
|
||||
// No settings service mounted: nothing ran, the entry stays authoritative.
|
||||
expect(current()).toEqual({ theme: 'entry' })
|
||||
expect(changes).toBe(0)
|
||||
|
||||
const fiber = ctx.plugin(MemorySettings, { doc: { 'helper-ns': { theme: 'user' } } })
|
||||
await fiber
|
||||
await vi.waitFor(() => {
|
||||
expect(current()).toEqual({ theme: 'user' })
|
||||
})
|
||||
expect(changes).toBe(1)
|
||||
|
||||
await ctx.settings.update(settingsNamespace('helper-ns'), { theme: 'live' })
|
||||
await vi.waitFor(() => {
|
||||
expect(changes).toBe(2)
|
||||
})
|
||||
expect(current()).toEqual({ theme: 'live' })
|
||||
|
||||
await fiber.dispose()
|
||||
await vi.waitFor(() => {
|
||||
expect(changes).toBe(3)
|
||||
})
|
||||
expect(current()).toEqual({ theme: 'entry' })
|
||||
})
|
||||
|
||||
it('stays silent when the consumer itself unloads', async () => {
|
||||
const { ctx } = await boot({ doc: { 'helper-ns': { theme: 'user' } } })
|
||||
const entry = { theme: 'entry' }
|
||||
let current: () => { theme: string } = () => entry
|
||||
const changes: string[] = []
|
||||
const consumer = ctx.plugin({
|
||||
inject: ['settings'],
|
||||
apply: (child: Context) => {
|
||||
installSettingsSection(child, settingsNamespace('helper-ns'), HelperSchema, entry, {
|
||||
setSource: (source) => {
|
||||
current = source
|
||||
},
|
||||
onChange: () => {
|
||||
changes.push(current().theme)
|
||||
},
|
||||
})
|
||||
},
|
||||
})
|
||||
await consumer
|
||||
await vi.waitFor(() => {
|
||||
expect(changes).toEqual(['user'])
|
||||
})
|
||||
|
||||
// The consumer's own teardown must not re-derive anything: an onChange
|
||||
// here would re-register routes and touch resources being released.
|
||||
await consumer.dispose()
|
||||
await new Promise(resolve => setTimeout(resolve, 20))
|
||||
expect(changes).toEqual(['user'])
|
||||
})
|
||||
|
||||
it('stays silent for a stored change that lands while the consumer unloads', async () => {
|
||||
// The watcher outlives the start of teardown by the width of the unload,
|
||||
// so a document change arriving in that window reaches it. Notifying then
|
||||
// is exactly as harmful as notifying from the disposer.
|
||||
const { ctx, provider } = await boot({ doc: { 'helper-ns': { theme: 'user' } } })
|
||||
const entry = { theme: 'entry' }
|
||||
let current: () => { theme: string } = () => entry
|
||||
const changes: string[] = []
|
||||
const consumer = ctx.plugin({
|
||||
inject: ['settings'],
|
||||
apply: (child: Context) => {
|
||||
installSettingsSection(child, settingsNamespace('helper-ns'), HelperSchema, entry, {
|
||||
setSource: (source) => {
|
||||
current = source
|
||||
},
|
||||
onChange: () => {
|
||||
changes.push(current().theme)
|
||||
},
|
||||
})
|
||||
},
|
||||
})
|
||||
await consumer
|
||||
await vi.waitFor(() => {
|
||||
expect(changes).toEqual(['user'])
|
||||
})
|
||||
|
||||
const unloading = consumer.dispose()
|
||||
provider.pushExternal({ 'helper-ns': { theme: 'racing' } })
|
||||
await unloading
|
||||
expect(changes).toEqual(['user'])
|
||||
})
|
||||
})
|
||||
|
||||
Reference in New Issue
Block a user