fix(storage,workspace): review-bot findings — emit isolation, domain ownership, null globals

- 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).
2026-07-25 00:01:21 +08:00
parent 1bddf5269a
commit 507dd25a3a
7 changed files with 160 additions and 36 deletions
--- 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).

--- a/packages/storage/domain/src/domain.ts
+++ b/packages/storage/domain/src/domain.ts
@@ -106,6 +106,16 @@ export interface Domain<S extends DomainSpec> {
   * @returns the typed table handle.
   */
  table<N extends keyof S['tables'] & string>(name: N): KvTable<TableKeyOf<S, N>, TableValueOf<S, N>>
+
+  /**
+   * 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<void>
 }

 /** 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<void> = 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<void>

@@ -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<string, Map<string, unknown>>,
    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<void> {
-    this.disposal ??= this.runDispose()
+  close(): Promise<void> {
+    this.disposal ??= this.runClose()
    return this.disposal
  }

-  private async runDispose(): Promise<void> {
+  private async runClose(): Promise<void> {
    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<T>(job: () => Promise<T>): Promise<T> {
--- 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<S> 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<void> {
+    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<T>(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()
+    }
+  })
 }
--- a/packages/storage/domain/src/spec.ts
+++ b/packages/storage/domain/src/spec.ts
@@ -65,10 +65,14 @@ export function domainTable<K extends string, V>(schema: ZodType<V>): 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<S extends DomainSpec>(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
 }

--- 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<string, Item>(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)
  })
 })