From 507dd25a3a3c8857cebe0e885b4c6d6ea47fdfa6 Mon Sep 17 00:00:00 2001 From: imccyu <276526105+imccyu@users.noreply.github.com> Date: Sat, 25 Jul 2026 00:01:21 +0800 Subject: [PATCH] =?UTF-8?q?fix(storage,workspace):=20review-bot=20findings?= =?UTF-8?q?=20=E2=80=94=20emit=20isolation,=20domain=20ownership,=20null?= =?UTF-8?q?=20globals?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - domain/changed emission is isolated from the write path: an observer throwing synchronously can no longer turn a committed (durable + in-memory) write into a rejection; the failure is logged and later writes proceed. - Domain lifecycle belongs to the opening consumer: Domain gains an idempotent close() (drain, unit close, reservation release), the facility stops registering effects on its own context and instead closes any still-open domains on unmount; WorkspaceRegistry holds its domain through its own effect, so disposing and re-mounting the consumer no longer wedges on already-open. - defineDomain rejects a global schema accepting null at declaration time: JSON null is the medium's absence sentinel, so a nullable global could never round-trip; failing loud at the spec keeps set(null) unrepresentable. Regression tests cover all three (hostile listener, close/reopen and consumer re-mount, nullable-global rejection). --- packages/storage/domain/README.md | 2 +- packages/storage/domain/src/domain.ts | 49 +++++++++++++---- packages/storage/domain/src/index.ts | 52 ++++++++++++------ packages/storage/domain/src/spec.ts | 18 +++++-- packages/storage/domain/tests/domain.spec.ts | 54 +++++++++++++++++-- packages/workspace/workspace/src/index.ts | 3 ++ .../workspace/tests/workspace.spec.ts | 18 +++++++ 7 files changed, 160 insertions(+), 36 deletions(-) diff --git a/packages/storage/domain/README.md b/packages/storage/domain/README.md index 38793ad9c7..df1d99644a 100644 --- a/packages/storage/domain/README.md +++ b/packages/storage/domain/README.md @@ -1,6 +1,6 @@ # @deepseek-ai/dsh-domain -Domain data form for the DeepSeek Harness storage hub: mounts `ctx.storage.domain`, opening schema-validated KV domains over configured storage backends. A domain is declared once with `defineDomain` (zod record schemas, `z.infer`-derived types), opened through `DomainFacility.open`, and served from authoritative in-memory state — reads are synchronous, writes serialize on one per-domain chain, reach durability on the routed backend first, then update memory and emit `domain/changed`. +Domain data form for the DeepSeek Harness storage hub: mounts `ctx.storage.domain`, opening schema-validated KV domains over configured storage backends. A domain is declared once with `defineDomain` (zod record schemas, `z.infer`-derived types), opened through `DomainFacility.open`, and served from authoritative in-memory state — reads are synchronous, writes serialize on one per-domain chain, reach durability on the routed backend first, then update memory and emit `domain/changed`. The opening consumer owns the handle's lifecycle and releases it with `Domain.close()` (idempotent; typically its own `ctx.effect` disposer); domains still open when the plugin unmounts are closed by the facility. Design rationale, open semantics, and the storage/domain layer split live in the [Agent Note](../../../.agents/notes/proposed/architecture/2026-07-24-domain-kv-storage-and-workspace.zh.md). diff --git a/packages/storage/domain/src/domain.ts b/packages/storage/domain/src/domain.ts index 227553c7ce..10c71524a9 100644 --- a/packages/storage/domain/src/domain.ts +++ b/packages/storage/domain/src/domain.ts @@ -106,6 +106,16 @@ export interface Domain { * @returns the typed table handle. */ table(name: N): KvTable, TableValueOf> + + /** + * Close this domain: reject new writes immediately, drain already-queued + * writes (their events still emit), release the backend unit, then free + * the domain name for a later open. Idempotent — repeated calls share one + * teardown. The consumer owns this call (typically as its own `ctx.effect` + * disposer); the facility closes any domain left open when it unmounts. + * @returns resolution after the unit is released. + */ + close(): Promise } /** Internal seam handing table handles their domain-owned write machinery. */ @@ -137,9 +147,9 @@ export class DomainImpl { /** Tail of the write chain; every link settles (rejections are observed by the caller's slice). */ private chain: Promise = Promise.resolve() - /** Set when dispose begins: new writes reject while already-queued writes drain. */ + /** Set when close begins: new writes reject while already-queued writes drain. */ private disposing = false - /** Set when dispose finishes (chain drained, unit closed): reads reject from here on. */ + /** Set when close finishes (chain drained, unit closed): reads reject from here on. */ private closed = false private disposal?: Promise @@ -152,6 +162,8 @@ export class DomainImpl { * the spec, so the entry set IS the table set. * @param globalValue - Validated stored global, or the spec's `initial` * when the medium held none; `undefined` when the spec declares no global. + * @param onClosed - Facility hook run once after teardown completes; frees + * the domain name for a later open. */ constructor( private readonly ctx: Context, @@ -159,6 +171,7 @@ export class DomainImpl { private readonly unit: KvUnit, records: Map>, globalValue: unknown, + private readonly onClosed: () => void, ) { this.name = spec.name const host: TableHost = { @@ -166,7 +179,7 @@ export class DomainImpl { unit, enqueue: job => this.enqueue(job), assertReadable: () => { this.assertReadable() }, - emitChanged: (change) => { this.ctx.emit('domain/changed', change) }, + emitChanged: (change) => { this.emitChanged(change) }, } for (const [table, tableRecords] of records) { this.tables.set(table, new KvTableImpl(host, table, tableRecords)) @@ -181,7 +194,7 @@ export class DomainImpl { set: value => this.enqueue(async () => { await this.unit.setGlobal(value) this.globalValue = value - host.emitChanged({ domain: this.name, table: '', key: '', operation: 'put', value }) + this.emitChanged({ domain: this.name, table: '', key: '', operation: 'put', value }) }), } } @@ -211,22 +224,40 @@ export class DomainImpl { /** * Close this domain: reject new writes immediately, drain already-queued - * writes (their events still emit), then close the unit. Idempotent — - * repeated calls share one teardown. + * writes (their events still emit), close the unit, then free the name via + * the facility hook. Idempotent — repeated calls share one teardown. * @returns resolution after the unit is released. */ - dispose(): Promise { - this.disposal ??= this.runDispose() + close(): Promise { + this.disposal ??= this.runClose() return this.disposal } - private async runDispose(): Promise { + private async runClose(): Promise { this.disposing = true // Chain links never reject (each is settled via then(noop, noop)), so // this await is a pure drain barrier. await this.chain await this.unit.close() this.closed = true + this.onClosed() + } + + /** + * Dispatch one post-durability change notification, containing observer + * failures: the write is already committed (medium and memory both hold + * the new state), so a throwing listener must not retroactively reject it. + */ + private emitChanged(change: DomainChanged): void { + try { + this.ctx.emit('domain/changed', change) + } catch (error) { + // Swallows synchronous observer exceptions only: emit dispatches + // listeners inline and nothing else runs in the try. The event is a + // notification, not a transaction participant — the commit point has + // passed, so containment (with a log) is the only correct outcome. + this.ctx.logger.warn(`domain '${this.name}': domain/changed listener failed: ${String(error)}`) + } } private enqueue(job: () => Promise): Promise { diff --git a/packages/storage/domain/src/index.ts b/packages/storage/domain/src/index.ts index fafa97ebae..e0dc09214f 100644 --- a/packages/storage/domain/src/index.ts +++ b/packages/storage/domain/src/index.ts @@ -81,8 +81,12 @@ export class DomainFacility { * (`facet-unsupported`); open the unit projected from the spec (backend * `version-mismatch`/`malformed-medium` pass through); load and validate * every stored record against the spec's zod schemas (`invalid-record` - * with the offending table and key); construct the domain and register its - * disposal effect (drain the write chain, close the unit). + * with the offending table and key); construct the domain. + * + * Lifecycle: the CALLER owns the returned handle and closes it via + * `Domain.close()` (typically as its own `ctx.effect` disposer) — the + * facility does not tie the domain to any consumer fiber. Domains still + * open when the facility unmounts are closed by the plugin disposer. * @param spec - The domain declaration, typically from `defineDomain`. * @returns the opened domain handle, typed by the spec. */ @@ -119,20 +123,15 @@ export class DomainFacility { : snapshot.global === null ? globalSpec.initial : parseRecord(spec.name, '', '', () => globalSpec.schema.parse(snapshot.global)) - const domain = new DomainImpl(this.ctx, spec, unit, tables, globalValue) - // The open-domain table entry is itself the effect: registration and - // the drain-then-unlist teardown live in one closure. - this.ctx.effect(() => { - this.domains.set(spec.name, domain) - return async () => { - // Drain before unlisting: writes landing during the drain still - // emit domain/changed, and the domain must stay resolvable (the - // package invariant cross-checks each event) until fully closed. - await domain.dispose() - this.domains.delete(spec.name) - this.reserved.delete(spec.name) - } + // The onClosed hook runs strictly after teardown completes: writes + // landing during the drain still emit domain/changed, and the domain + // stays resolvable (the package invariant cross-checks each event) + // until fully closed — only then does the name free up for reopening. + const domain: DomainImpl = new DomainImpl(this.ctx, spec, unit, tables, globalValue, () => { + this.domains.delete(spec.name) + this.reserved.delete(spec.name) }) + this.domains.set(spec.name, domain) // The single type-erasure point: DomainImpl is the untyped runtime, // Domain the spec-typed view; the unknown hop is required because // S's conditional global-handle type stays unresolved here. @@ -142,7 +141,7 @@ export class DomainFacility { throw error } } catch (error) { - // Any failure means the effect never registered (nothing can throw + // Any failure means the domain never registered (nothing can throw // after it), so releasing the name reservation is unconditional. this.reserved.delete(spec.name) throw error @@ -159,6 +158,16 @@ export class DomainFacility { get(name: string): DomainImpl | undefined { return this.domains.get(name) } + + /** + * Close every domain still open on this facility. The unmount path for + * consumers that never called `Domain.close()` themselves; closing is + * idempotent, so double-closing an already-closed domain is harmless. + * @returns resolution after every unit is released. + */ + async closeAll(): Promise { + await Promise.all([...this.domains.values()].map(domain => domain.close())) + } } /** Run one zod parse, translating failure to `invalid-record` with its location. */ @@ -181,5 +190,14 @@ function parseRecord(domain: string, table: string, key: string, parse: () => * @param config - Validated plugin config. */ export function apply(ctx: Context, config: Config) { - ctx.effect(() => ctx.storage.mount('domain', new DomainFacility(ctx, config))) + const facility = new DomainFacility(ctx, config) + ctx.effect(() => { + const unmount = ctx.storage.mount('domain', facility) + return async () => { + // Close leftovers before unmounting: draining writes still emit + // domain/changed, whose invariant resolves the facility through the hub. + await facility.closeAll() + unmount() + } + }) } diff --git a/packages/storage/domain/src/spec.ts b/packages/storage/domain/src/spec.ts index df833bebe1..b50af8f25b 100644 --- a/packages/storage/domain/src/spec.ts +++ b/packages/storage/domain/src/spec.ts @@ -65,10 +65,14 @@ export function domainTable(schema: ZodType): DomainTabl } /** - * Identity helper that pins a spec's literal types and validates its names. - * Misconfiguration fails loud: a domain or table name outside `UNIT_NAME_RE` - * or a version that is not a non-negative integer throws here, at the owning - * package's module load, before any medium is touched. + * Identity helper that pins a spec's literal types and validates its shape. + * Misconfiguration fails loud at the owning package's module load, before any + * medium is touched: a domain or table name outside `UNIT_NAME_RE`, a version + * that is not a non-negative integer, or a global schema that accepts `null` + * all throw. The `null` rejection guards round-tripping: backends store the + * global as opaque JSON with `null` as the "never written" sentinel, so a + * nullable global would be indistinguishable from an absent one on reopen + * (a stored `null` silently reverts to `initial`). * @param spec - The domain declaration. * @returns the same spec, narrowed to its literal type. */ @@ -84,6 +88,12 @@ export function defineDomain(spec: S): S { throw new Error(`domain '${spec.name}' table name '${table}' must match ${UNIT_NAME_RE}`) } } + if (spec.global !== undefined && spec.global.schema.safeParse(null).success) { + throw new Error( + `domain '${spec.name}' global schema must not accept null: ` + + 'null is the medium\'s "never written" sentinel, so a stored null could not round-trip', + ) + } return spec } diff --git a/packages/storage/domain/tests/domain.spec.ts b/packages/storage/domain/tests/domain.spec.ts index ebff665c08..8d02cd50e5 100644 --- a/packages/storage/domain/tests/domain.spec.ts +++ b/packages/storage/domain/tests/domain.spec.ts @@ -48,6 +48,15 @@ describe('defineDomain', () => { name: 'ok', version: 1, tables: { 'Bad Table': domainTable(itemSchema) }, })).toThrow(/table name/) }) + + it('rejects a global schema that accepts null (the never-written sentinel)', () => { + expect(() => defineDomain({ + name: 'ok', + version: 1, + global: { schema: settingsSchema.nullable(), initial: null }, + tables: {}, + })).toThrow(/must not accept null/) + }) }) describe('DomainFacility.open', () => { @@ -262,21 +271,56 @@ describe('global singleton', () => { }) }) -describe('disposal', () => { - it('drains queued writes, closes the unit, then rejects reads and writes', async () => { +describe('close and lifecycle', () => { + it('close drains queued writes, then rejects reads and writes, and frees the name', async () => { const pool = new MemoryMediaPool() - const { ctx, facility } = await harness({ pool }) + const { facility } = await harness({ pool }) const domain = await facility.open(spec) const table = domain.table('items') const pending = Promise.all([ table.put('a', { label: 'x', count: 1 }), table.put('b', { label: 'y', count: 2 }), ]) - await ctx.fiber.dispose() // effect disposer: drain chain, close unit - await pending // queued before dispose → still landed + await Promise.all([domain.close(), domain.close()]) // idempotent + await pending // queued before close → still landed // Durability is the drain contract: both queued writes reached the medium. expect([...pool.media.get('demo')!.tables.get('items')!.keys()].sort()).toEqual(['a', 'b']) await expect(table.put('c', { label: 'z', count: 3 })).rejects.toMatchObject({ code: 'closed' }) expect(() => table.get('a')).toThrow(/closed/) + // The name is free again: reopening sees the drained state. + const reopened = await facility.open(spec) + expect([...reopened.table('items').keys()].sort()).toEqual(['a', 'b']) + }) + + it('facility unmount closes domains the consumer never closed', async () => { + const ctx = new Context() + await ctx.plugin(Storage) + ctx.storage.backend.register('memory', new MemoryStorageBackend()) + const DomainPlugin = await import('../src/index.ts') + const fiber = await ctx.plugin(DomainPlugin, { backend: 'memory' }) + const domain = await ctx.storage.domain.open(bareSpec) + const table = domain.table('rows') + await table.put('a', { label: 'x', count: 1 }) + await fiber.dispose() + await expect(table.put('b', { label: 'y', count: 2 })).rejects.toMatchObject({ code: 'closed' }) + expect(() => ctx.storage.form('domain')).toThrow(/not mounted/) + }) + + it('contains a throwing domain/changed listener without rejecting the committed write', async () => { + const pool = new MemoryMediaPool() + const { ctx, facility, changes } = await harness({ pool }) + const domain = await facility.open(spec) + const table = domain.table('items') + ctx.on('domain/changed', () => { + throw new Error('hostile observer') + }) + await expect(table.put('a', { label: 'x', count: 1 })).resolves.toBeUndefined() + // Commit survived intact on both planes, and well-behaved listeners + // (registered before the thrower) still observed the event. + expect(table.get('a')).toEqual({ label: 'x', count: 1 }) + expect(pool.media.get('demo')!.tables.get('items')!.get('a')).toEqual({ label: 'x', count: 1 }) + expect(changes).toHaveLength(1) + // The chain is unpoisoned: subsequent writes proceed normally. + await expect(table.delete('a')).resolves.toBe(true) }) }) diff --git a/packages/workspace/workspace/src/index.ts b/packages/workspace/workspace/src/index.ts index 584c3bde77..6eded589d7 100644 --- a/packages/workspace/workspace/src/index.ts +++ b/packages/workspace/workspace/src/index.ts @@ -84,6 +84,9 @@ export class WorkspaceRegistry extends Service { /** Open the domain and rebuild the entity cache before the service is published as active. */ protected async [Service.init](): Promise { const domain = await this.ctx.storage.domain.open(workspaceDomainSpec) + // This registry owns the domain handle it opened: closing on fiber + // disposal frees the domain name, so a re-plugged registry can reopen it. + this.ctx.effect(() => () => domain.close(), 'workspace.domainClose') this.table = domain.table('workspaces') const persistence = this.ctx.get('sessionPersistence') if (persistence !== undefined) { diff --git a/packages/workspace/workspace/tests/workspace.spec.ts b/packages/workspace/workspace/tests/workspace.spec.ts index 360103f7a0..fd8610b801 100644 --- a/packages/workspace/workspace/tests/workspace.spec.ts +++ b/packages/workspace/workspace/tests/workspace.spec.ts @@ -185,6 +185,24 @@ describe('WorkspaceRegistry.create', () => { const registry = new WorkspaceRegistry(ctx) await expect(registry.create(dir)).rejects.toThrow(/not started/) }) + + it('closes its domain on fiber disposal so a re-plugged registry reopens it', async () => { + const dir = await makeDir('replug') + const ctx = new Context() + await ctx.plugin(Storage) + ctx.storage.backend.register('memory', new MemoryStorageBackend()) + ctx.storage.mount('domain', new DomainFacility(ctx, { backend: 'memory', routes: {} })) + const fiber = ctx.plugin(WorkspaceRegistry) + await fiber + const first = await ctx.workspace.create(dir) + await fiber.dispose() + // The registry's effect closed the domain, freeing the name: a second + // plugin of the same registry must reopen it (not already-open) and see + // the durable record. + await ctx.plugin(WorkspaceRegistry) + const reloaded = await ctx.workspace.resolveByPath(dir) + expect(reloaded?.id).toBe(first.id) + }) }) describe('Workspace.attachSession', () => {