From 80b3b6d9179bf4daa834f79b319c3fb9329093ea Mon Sep 17 00:00:00 2001 From: imccyu <276526105+imccyu@users.noreply.github.com> Date: Fri, 24 Jul 2026 21:30:32 +0800 Subject: [PATCH] fix(storage,workspace): post-review hardening Review findings applied across the group: - storage hub: stale disposers no longer remove a successor registration; the package now default-exports the Storage service class per the service-package export shape. - json backend: failed publishes roll back the authoritative memory state (a rejected write can no longer resurface via get() or ride the next publish); close() drains in-flight writes and blocks in-flight opens; double-open rejects as a plain caller error instead of malformed-medium. - sqlite backend: loadAll builds records on a null prototype (__proto__ keys round-trip instead of polluting), user_version is stamped only after the schema is fully created, and corrupt record JSON rejects as malformed-medium instead of a bare SyntaxError. - domain form: writes persist before mutating authoritative memory or emitting; DomainChanged is a put/deleted discriminated union. - workspace: attach/detach idempotence decided on the write chain (stale snapshots no longer short-circuit), create() requires a directory, and startup fails loud on duplicate stored paths. Eleven regression tests pin the fixed behaviors. --- packages/storage/domain/src/domain.ts | 16 ++-- packages/storage/domain/src/events.ts | 26 +++-- packages/storage/domain/src/invariant.ts | 33 ++++--- packages/storage/domain/tests/domain.spec.ts | 39 +++++++- .../domain/tests/helpers/memory-backend.ts | 24 ++++- packages/storage/storage-json/src/index.ts | 43 ++++++--- packages/storage/storage-json/src/unit.ts | 45 +++++++-- .../storage-json/tests/json-backend.spec.ts | 42 +++++++- packages/storage/storage-sqlite/src/schema.ts | 10 +- packages/storage/storage-sqlite/src/unit.ts | 21 +++- .../tests/sqlite-backend.spec.ts | 81 ++++++++++++++++ packages/storage/storage/src/index.ts | 16 ++-- packages/storage/storage/src/registry.ts | 6 +- .../storage/storage/tests/registry.spec.ts | 4 +- packages/workspace/workspace/README.md | 4 +- packages/workspace/workspace/src/entity.ts | 95 +++++++++++-------- packages/workspace/workspace/src/index.ts | 29 ++++-- packages/workspace/workspace/src/types.ts | 20 ++-- .../workspace/tests/workspace.spec.ts | 52 ++++++++-- 19 files changed, 470 insertions(+), 136 deletions(-) diff --git a/packages/storage/domain/src/domain.ts b/packages/storage/domain/src/domain.ts index 9f36113dc9..a0796ff628 100644 --- a/packages/storage/domain/src/domain.ts +++ b/packages/storage/domain/src/domain.ts @@ -1,9 +1,11 @@ /** * Runtime of one open domain: authoritative in-memory state, the single * per-domain write chain, and change-event emission. Reads are synchronous - * from memory; every write queues on the chain, mutates memory, awaits - * backend durability, then emits `domain/changed` — so events carry values - * that equal the in-memory state at emission and arrive in write order. + * from memory; every write queues on the chain, awaits backend durability + * FIRST, then mutates memory, then emits `domain/changed` — a rejected + * backend write leaves memory untouched (no divergence between reads and the + * medium), and events carry values that equal the in-memory state at + * emission, in write order. * @module @deepseek-ai/dsh-domain/src/domain */ @@ -175,8 +177,8 @@ export class DomainImpl { return this.globalValue }, set: (value) => this.enqueue(async () => { - this.globalValue = value await this.unit.setGlobal(value) + this.globalValue = value host.emitChanged({ domain: this.name, table: '', key: '', operation: 'put', value }) }), } @@ -271,8 +273,8 @@ class KvTableImpl implements KvTable { put(key: K, value: V): Promise { return this.host.enqueue(async () => { - this.records.set(key, value) await this.host.unit.putRecord(this.tableName, key, value) + this.records.set(key, value) this.emitPut(key, value) }) } @@ -282,8 +284,8 @@ class KvTableImpl implements KvTable { // Existence is decided at this job's chain slot, not at call time: an // earlier queued put of the same key makes this delete observe it. if (!this.records.has(key)) return false - this.records.delete(key) await this.host.unit.deleteRecord(this.tableName, key) + this.records.delete(key) this.host.emitChanged({ domain: this.host.domainName, table: this.tableName, @@ -303,8 +305,8 @@ class KvTableImpl implements KvTable { ) } const next = fn(this.records.get(key) as V) - this.records.set(key, next) await this.host.unit.putRecord(this.tableName, key, next) + this.records.set(key, next) this.emitPut(key, next) return next }) diff --git a/packages/storage/domain/src/events.ts b/packages/storage/domain/src/events.ts index abc5ae46d3..42311e865c 100644 --- a/packages/storage/domain/src/events.ts +++ b/packages/storage/domain/src/events.ts @@ -7,20 +7,32 @@ * @module @deepseek-ai/dsh-domain/src/events */ -/** One durable domain change: a record upsert/delete or a global write. */ -export interface DomainChanged { +/** Shared location fields of one durable domain change. */ +export interface DomainChangedBase { /** Owning domain name. */ readonly domain: string /** Table name; `''` for a global-singleton write. */ readonly table: string /** Record key; `''` for a global-singleton write. */ readonly key: string - /** What happened: `put` covers insert and overwrite; `deleted` is a tombstone. */ - readonly operation: 'put' | 'deleted' - /** The new snapshot; absent for `deleted`. */ - readonly value?: unknown } +/** A record (or the global singleton) was inserted or overwritten. */ +export interface DomainChangedPut extends DomainChangedBase { + readonly operation: 'put' + /** The new snapshot. */ + readonly value: unknown +} + +/** A record was deleted; tombstones carry no value. */ +export interface DomainChangedDeleted extends DomainChangedBase { + readonly operation: 'deleted' + readonly value?: never +} + +/** One durable domain change; a closed union — switch on `operation`. */ +export type DomainChanged = DomainChangedPut | DomainChangedDeleted + declare module 'cordis' { interface Events { /** @@ -28,7 +40,7 @@ declare module 'cordis' { * strictly after the backend acknowledged durability. Events of one * domain arrive in its write-chain order. * @param change - domain, table (`''` for global), key (`''` for global), - * operation discriminant, and the new snapshot (absent for deletions). + * operation discriminant, and on `put` the new snapshot. * @mode emit */ 'domain/changed'(change: DomainChanged): void diff --git a/packages/storage/domain/src/invariant.ts b/packages/storage/domain/src/invariant.ts index f7da02e38b..785d1e4450 100644 --- a/packages/storage/domain/src/invariant.ts +++ b/packages/storage/domain/src/invariant.ts @@ -35,20 +35,25 @@ const install: InvariantInstaller = Object.assign((ctx: Context, fail: Invariant return } const current = domain.table(change.table).get(change.key) - if (change.operation === 'deleted') { - if (current !== undefined) { - return fail( - `domain/changed deletion of '${change.domain}'.'${change.table}'['${change.key}'] ` - + 'emitted while the record is still in memory', - ) - } - return - } - if (current !== change.value) { - return fail( - `domain/changed value for '${change.domain}'.'${change.table}'['${change.key}'] ` - + 'differs from the in-memory record', - ) + switch (change.operation) { + case 'deleted': + if (current !== undefined) { + return fail( + `domain/changed deletion of '${change.domain}'.'${change.table}'['${change.key}'] ` + + 'emitted while the record is still in memory', + ) + } + return + case 'put': + if (current !== change.value) { + return fail( + `domain/changed value for '${change.domain}'.'${change.table}'['${change.key}'] ` + + 'differs from the in-memory record', + ) + } + return + default: + change satisfies never } }, { global: true }) }, { inject: ['storage'] }) diff --git a/packages/storage/domain/tests/domain.spec.ts b/packages/storage/domain/tests/domain.spec.ts index 7d98337861..6fc810ca5a 100644 --- a/packages/storage/domain/tests/domain.spec.ts +++ b/packages/storage/domain/tests/domain.spec.ts @@ -1,7 +1,7 @@ import { describe, expect, it } from 'vitest' import { Context } from 'cordis' import { z } from 'zod' -import { apply as applyStorage } from '@deepseek-ai/dsh-storage' +import Storage from '@deepseek-ai/dsh-storage' import { DomainFacility, defineDomain, domainTable } from '../src/index.ts' import type { Config } from '../src/index.ts' import type { DomainChanged } from '../src/events.ts' @@ -28,7 +28,7 @@ const bareSpec = defineDomain({ /** Boot a context with the storage hub, one memory backend, and a facility over it. */ async function harness(options?: { pool?: MemoryMediaPool; config?: Partial }) { const ctx = new Context() - await ctx.plugin({ apply: applyStorage }) + await ctx.plugin(Storage) const backend = new MemoryStorageBackend(options?.pool) ctx.storage.backend.register('memory', backend) const facility = new DomainFacility(ctx, { backend: 'memory', routes: {}, ...options?.config }) @@ -149,6 +149,41 @@ describe('KvTable writes', () => { }) }) +describe('durability failure', () => { + it('leaves memory untouched and emits nothing when the backend rejects a write', async () => { + const pool = new MemoryMediaPool() + const { facility, changes } = await harness({ pool }) + const domain = await facility.open(spec) + const table = domain.table('items') + await table.put('a', { label: 'x', count: 1 }) + const seen = changes.length + + pool.failNextWrites = 3 + await expect(table.put('a', { label: 'x', count: 99 })).rejects.toThrow(/injected/) + await expect(table.update('a', (c) => ({ ...c, count: c.count + 1 }))).rejects.toThrow(/injected/) + await expect(table.delete('a')).rejects.toThrow(/injected/) + + // Reads still serve the pre-failure record; no events leaked. + 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(seen) + + // The chain survives rejections: the next write lands cleanly with no residue. + await table.update('a', (c) => ({ ...c, count: c.count + 1 })) + expect(table.get('a')).toEqual({ label: 'x', count: 2 }) + }) + + it('keeps serving initial when the first global set fails durability', async () => { + const pool = new MemoryMediaPool() + const { facility } = await harness({ pool }) + const domain = await facility.open(spec) + pool.failNextWrites = 1 + await expect(domain.global.set({ theme: 'dark' })).rejects.toThrow(/injected/) + expect(domain.global.get()).toEqual({ theme: 'plain' }) + expect(pool.media.get('demo')!.global).toBeNull() + }) +}) + describe('global singleton', () => { it('serves initial before first set without materializing, then persists the first set', async () => { const pool = new MemoryMediaPool() diff --git a/packages/storage/domain/tests/helpers/memory-backend.ts b/packages/storage/domain/tests/helpers/memory-backend.ts index 0c398d9ba1..906202becf 100644 --- a/packages/storage/domain/tests/helpers/memory-backend.ts +++ b/packages/storage/domain/tests/helpers/memory-backend.ts @@ -28,13 +28,29 @@ export interface MemoryMedium { * Shared media pool. Construct one and hand it to several * {@link MemoryStorageBackend} instances to simulate reopening the same * medium after a restart; `versions` holds the stamped unit versions and is - * writable by tests to inject a mismatching on-medium version. + * writable by tests to inject a mismatching on-medium version, and + * `failNextWrites` injects write-primitive failures. */ export class MemoryMediaPool { /** Unit name → its records; a missing entry is a never-materialized unit. */ readonly media = new Map() /** Unit name → stamped version; tests may pre-stamp to force `version-mismatch`. */ readonly versions = new Map() + /** + * When positive, that many subsequent write primitives (putRecord / + * deleteRecord / setGlobal) reject without touching the medium, decrementing + * per rejection. Negative-path seam: callers assert their state is + * untouched after a durability failure. + */ + failNextWrites = 0 + + /** Consume one injected failure, throwing in a rejected write's place. */ + consumeInjectedFailure(): void { + if (this.failNextWrites > 0) { + this.failNextWrites -= 1 + throw new Error('injected write failure') + } + } } /** In-memory KV unit over one pooled medium. */ @@ -42,6 +58,7 @@ class MemoryKvUnit implements KvUnit { private closed = false constructor( + private readonly pool: MemoryMediaPool, private readonly medium: MemoryMedium, private readonly descriptor: KvUnitDescriptor, private readonly onClose: () => void, @@ -64,6 +81,7 @@ class MemoryKvUnit implements KvUnit { async putRecord(table: string, key: string, value: unknown): Promise { this.assertOpen() + this.pool.consumeInjectedFailure() let records = this.medium.tables.get(table) if (records === undefined) { records = new Map() @@ -74,11 +92,13 @@ class MemoryKvUnit implements KvUnit { async deleteRecord(table: string, key: string): Promise { this.assertOpen() + this.pool.consumeInjectedFailure() this.medium.tables.get(table)?.delete(key) } async setGlobal(value: unknown): Promise { this.assertOpen() + this.pool.consumeInjectedFailure() this.medium.global = value } @@ -128,7 +148,7 @@ export class MemoryStorageBackend implements StorageBackend { this.pool.media.set(descriptor.name, medium) } this.openUnits.add(descriptor.name) - return new MemoryKvUnit(medium, descriptor, () => this.openUnits.delete(descriptor.name)) + return new MemoryKvUnit(this.pool, medium, descriptor, () => this.openUnits.delete(descriptor.name)) }, } } diff --git a/packages/storage/storage-json/src/index.ts b/packages/storage/storage-json/src/index.ts index caed43e8e0..ca56d366a4 100644 --- a/packages/storage/storage-json/src/index.ts +++ b/packages/storage/storage-json/src/index.ts @@ -37,31 +37,48 @@ export const Config: z = z.object({ /** JSON backend: owns the file-tree root and serves the `kv` facet. */ export class JsonStorageBackend implements StorageBackend { private readonly open = new Map() + // Reserved synchronously at open() entry so a concurrent open of the same + // unit fails, and close() can await opens still in flight. + private readonly opening = new Map>() private closed = false constructor(private readonly root: string) {} readonly kv: KvFacet = { - open: async (descriptor: KvUnitDescriptor): Promise => { - if (this.closed) throw new StorageError('closed', 'json backend is closed') + open: (descriptor: KvUnitDescriptor): Promise => { + if (this.closed) return Promise.reject(new StorageError('closed', 'json backend is closed')) validateDescriptor(descriptor) - if (this.open.has(descriptor.name)) { - throw new StorageError( - 'malformed-medium', - `unit '${descriptor.name}' is already open; a unit has exactly one live handle`, + if (this.open.has(descriptor.name) || this.opening.has(descriptor.name)) { + // Double-open is a caller bug, not a medium condition. + return Promise.reject( + new Error(`unit '${descriptor.name}' is already open; a unit has exactly one live handle`), ) } - await mkdir(this.root, { recursive: true, mode: 0o700 }) - const path = join(this.root, `${descriptor.name}.json`) - const unit = await openJsonUnit(descriptor, path, () => this.open.delete(descriptor.name)) - this.open.set(descriptor.name, unit) - return unit + const opening = this.openUnit(descriptor) + this.opening.set(descriptor.name, opening) + return opening.finally(() => this.opening.delete(descriptor.name)) }, } + private async openUnit(descriptor: KvUnitDescriptor): Promise { + await mkdir(this.root, { recursive: true, mode: 0o700 }) + const path = join(this.root, `${descriptor.name}.json`) + const unit = await openJsonUnit(descriptor, path, () => this.open.delete(descriptor.name)) + if (this.closed) { + // The backend closed while this open was in flight: do not hand out a + // live unit past close(). + await unit.close() + throw new StorageError('closed', 'json backend is closed') + } + this.open.set(descriptor.name, unit) + return unit + } + async close(): Promise { - if (this.closed) return - this.closed = true + if (!this.closed) { + this.closed = true + } + await Promise.allSettled([...this.opening.values()]) for (const unit of [...this.open.values()]) { await unit.close() } diff --git a/packages/storage/storage-json/src/unit.ts b/packages/storage/storage-json/src/unit.ts index d0689c00d2..43a986a637 100644 --- a/packages/storage/storage-json/src/unit.ts +++ b/packages/storage/storage-json/src/unit.ts @@ -40,6 +40,8 @@ export async function openJsonUnit( class JsonKvUnit implements KvUnit { private closed = false + /** In-flight publishes; close() drains them before releasing the unit. */ + private readonly inFlight = new Set>() constructor( private readonly descriptor: KvUnitDescriptor, @@ -59,15 +61,29 @@ class JsonKvUnit implements KvUnit { async putRecord(table: string, key: string, value: unknown): Promise { this.assertOpen() - this.records(table).set(key, value) - await this.publish() + const records = this.records(table) + const hadKey = records.has(key) + const previous = records.get(key) + records.set(key, value) + // Roll back on a failed publish: memory is authoritative, so a rejected + // write must not survive in memory (or ride along with the next publish). + await this.publish().catch(async (error) => { + if (hadKey) records.set(key, previous) + else records.delete(key) + throw error + }) } async deleteRecord(table: string, key: string): Promise { this.assertOpen() - if (this.records(table).delete(key)) { - await this.publish() - } + const records = this.records(table) + if (!records.has(key)) return + const previous = records.get(key) + records.delete(key) + await this.publish().catch(async (error) => { + records.set(key, previous) + throw error + }) } async setGlobal(value: unknown): Promise { @@ -75,13 +91,21 @@ class JsonKvUnit implements KvUnit { if (!this.descriptor.hasGlobal) { throw new Error(`unit '${this.descriptor.name}' does not declare a global slot`) } + const previous = this.state.global this.state.global = value - await this.publish() + await this.publish().catch(async (error) => { + this.state.global = previous + throw error + }) } async close(): Promise { - if (this.closed) return + if (this.closed) { + await Promise.allSettled(this.inFlight) + return + } this.closed = true + await Promise.allSettled(this.inFlight) this.onClose() } @@ -100,6 +124,11 @@ class JsonKvUnit implements KvUnit { } private publish(): Promise { - return writeAtomic(this.path, serialize(this.descriptor.name, this.state)) + const write = writeAtomic(this.path, serialize(this.descriptor.name, this.state)) + this.inFlight.add(write) + // Swallow only on the tracking branch: the caller still awaits `write` + // itself, so rejections stay observed exactly once. + write.catch(() => {}).finally(() => this.inFlight.delete(write)) + return write } } diff --git a/packages/storage/storage-json/tests/json-backend.spec.ts b/packages/storage/storage-json/tests/json-backend.spec.ts index 730e56828b..709e754f3c 100644 --- a/packages/storage/storage-json/tests/json-backend.spec.ts +++ b/packages/storage/storage-json/tests/json-backend.spec.ts @@ -70,11 +70,49 @@ describe('json backend specifics', () => { await backend.close() }) - it('rejects double-open of one unit', async () => { + it('rejects double-open of one unit as a plain caller error', async () => { const root = await freshRoot() const backend = new JsonStorageBackend(root) await backend.kv.open(descriptor) - await expect(backend.kv.open(descriptor)).rejects.toMatchObject({ code: 'malformed-medium' }) + await expect(backend.kv.open(descriptor)).rejects.toThrowError(/already open/) await backend.close() }) + + it('rolls back memory when a publish fails', async () => { + const root = await freshRoot() + const backend = new JsonStorageBackend(root) + const unit = await backend.kv.open(descriptor) + await unit.putRecord('t', 'k', { v: 'committed' }) + // Make the next publish fail: replace the unit file's parent with an + // unwritable directory path via chmod. + const { chmod } = await import('node:fs/promises') + await chmod(root, 0o500) + await expect(unit.putRecord('t', 'k', { v: 'rejected' })).rejects.toThrow() + await expect(unit.putRecord('t', 'k2', { v: 'also rejected' })).rejects.toThrow() + await chmod(root, 0o700) + const snapshot = await unit.loadAll() + expect(snapshot.tables['t']).toEqual({ k: { v: 'committed' } }) + // The next successful publish must not carry rejected writes to disk. + await unit.putRecord('t', 'k3', { v: 'later' }) + const text = await readFile(join(root, 'shape.json'), 'utf8') + expect(text).not.toContain('rejected') + await backend.close() + }) + + it('close drains in-flight writes and blocks in-flight opens', async () => { + const root = await freshRoot() + const backend = new JsonStorageBackend(root) + const unit = await backend.kv.open(descriptor) + const bigWrite = unit.putRecord('t', 'big', { blob: 'x'.repeat(4 * 1024 * 1024) }) + await unit.close() + await expect(bigWrite).resolves.toBeUndefined() + const onDisk = JSON.parse(await readFile(join(root, 'shape.json'), 'utf8')) + expect(onDisk.tables.t.big).toBeDefined() + + const backend2 = new JsonStorageBackend(root) + const opening = backend2.kv.open(descriptor) + const closing = backend2.close() + await expect(opening.then((u) => u.putRecord('t', 'x', {}))).rejects.toMatchObject({ code: 'closed' }) + await closing + }) }) diff --git a/packages/storage/storage-sqlite/src/schema.ts b/packages/storage/storage-sqlite/src/schema.ts index 8aee216774..9c3a0f1d0b 100644 --- a/packages/storage/storage-sqlite/src/schema.ts +++ b/packages/storage/storage-sqlite/src/schema.ts @@ -81,10 +81,6 @@ function configureDatabase(db: DatabaseSync, path: string, journalMode: JournalM `storage database at "${path}" has schema version ${onDisk}, incompatible with this build (${STORAGE_SQLITE_SCHEMA_VERSION})`, ) } - if (onDisk === 0) { - // Stamp fresh databases. - db.exec(`PRAGMA user_version = ${STORAGE_SQLITE_SCHEMA_VERSION}`) - } db.exec(` CREATE TABLE IF NOT EXISTS units ( name TEXT PRIMARY KEY, @@ -97,6 +93,12 @@ function configureDatabase(db: DatabaseSync, path: string, journalMode: JournalM value TEXT NOT NULL ) STRICT `) + if (onDisk === 0) { + // Stamp fresh databases LAST: the stamp asserts the layout is complete, + // so a failure above must leave the medium unstamped (a re-open after + // the obstruction is cleared retries materialization from scratch). + db.exec(`PRAGMA user_version = ${STORAGE_SQLITE_SCHEMA_VERSION}`) + } } /** diff --git a/packages/storage/storage-sqlite/src/unit.ts b/packages/storage/storage-sqlite/src/unit.ts index ea7291f303..c525341e7d 100644 --- a/packages/storage/storage-sqlite/src/unit.ts +++ b/packages/storage/storage-sqlite/src/unit.ts @@ -66,20 +66,35 @@ export class SqliteKvUnit implements KvUnit { this.ensureOpen() const tables: Record> = {} for (const [name, statements] of this.tables) { - const records: Record = {} + // Null prototype: record keys are arbitrary strings, so '__proto__' + // must land as an own property instead of mutating the prototype. + const records: Record = Object.create(null) as Record for (const row of statements.selectAll.all() as unknown as Array<{ key: string; value: string }>) { - records[row.key] = JSON.parse(row.value) + records[row.key] = this.parseValue(row.value, `table '${name}' key '${row.key}'`) } tables[name] = records } let global: unknown = null if (this.globalSelect !== undefined) { const row = this.globalSelect.get(this.descriptor.name) as { value: string } | undefined - if (row !== undefined) global = JSON.parse(row.value) + if (row !== undefined) global = this.parseValue(row.value, 'global slot') } return { tables, global } } + /** Parse one stored value column, mapping bad JSON to `malformed-medium`. */ + private parseValue(text: string, slot: string): unknown { + try { + return JSON.parse(text) + } catch (error) { + throw new StorageError( + 'malformed-medium', + `kv unit '${this.descriptor.name}' holds unparsable JSON at ${slot}`, + { cause: error }, + ) + } + } + async putRecord(table: string, key: string, value: unknown): Promise { this.ensureOpen() this.statementsFor(table).upsert.run(key, JSON.stringify(value)) diff --git a/packages/storage/storage-sqlite/tests/sqlite-backend.spec.ts b/packages/storage/storage-sqlite/tests/sqlite-backend.spec.ts index b7c46bfe89..e952d93f32 100644 --- a/packages/storage/storage-sqlite/tests/sqlite-backend.spec.ts +++ b/packages/storage/storage-sqlite/tests/sqlite-backend.spec.ts @@ -106,4 +106,85 @@ describe('sqlite backend specifics', () => { await backend.close() await expect(backend.kv.open(DESCRIPTOR)).rejects.toMatchObject({ code: 'closed' }) }) + + it('round-trips prototype-polluting keys as own properties', async () => { + const backend = backendAt(':memory:') + const unit = await backend.kv.open(DESCRIPTOR) + await unit.putRecord('records', '__proto__', { evil: true }) + await unit.putRecord('records', 'constructor', { n: 1 }) + const { tables } = await unit.loadAll() + const records = tables['records']! + expect(Object.hasOwn(records, '__proto__')).toBe(true) + expect(records['__proto__']).toEqual({ evil: true }) + expect(records['constructor']).toEqual({ n: 1 }) + expect(Object.getPrototypeOf({})).not.toHaveProperty('evil') + await backend.close() + }) + + it('leaves a failed materialization unstamped so a repaired medium reopens', async () => { + const path = await freshDbPath() + // Obstruct table creation: an index squatting on the unit_globals name + // makes CREATE TABLE IF NOT EXISTS throw AFTER the units table exists. + const setup = new DatabaseSync(path) + setup.exec('CREATE TABLE squatter (x TEXT)') + setup.exec('CREATE INDEX unit_globals ON squatter(x)') + setup.close() + + const broken = backendAt(path) + await expect(broken.kv.open(DESCRIPTOR)).rejects.toThrow(/already an index/) + await broken.close() + + // Clear the obstruction; the medium must still be version 0, not a + // half-materialized database stamped as current. + const repair = new DatabaseSync(path) + expect((repair.prepare('PRAGMA user_version').get() as { user_version: number }).user_version).toBe(0) + repair.exec('DROP INDEX unit_globals') + repair.close() + + const backend = backendAt(path) + const unit = await backend.kv.open(DESCRIPTOR) + await unit.putRecord('records', 'k', { n: 1 }) + await backend.close() + }) + + it('rejects unparsable stored JSON with malformed-medium', async () => { + const path = await freshDbPath() + const backend = backendAt(path) + const unit = await backend.kv.open(DESCRIPTOR) + await unit.putRecord('records', 'good', { n: 1 }) + await unit.setGlobal({ g: 1 }) + await backend.close() + + const db = new DatabaseSync(path) + db.prepare('UPDATE u_specimen_records SET value = ? WHERE key = ?').run('{not json', 'good') + db.close() + + const reopened = backendAt(path) + const damaged = await reopened.kv.open(DESCRIPTOR) + await expect(damaged.loadAll()).rejects.toMatchObject({ + name: 'StorageError', + code: 'malformed-medium', + }) + await reopened.close() + }) + + it('rejects an unparsable global slot with malformed-medium', async () => { + const path = await freshDbPath() + const backend = backendAt(path) + const unit = await backend.kv.open(DESCRIPTOR) + await unit.setGlobal({ g: 1 }) + await backend.close() + + const db = new DatabaseSync(path) + db.prepare('UPDATE unit_globals SET value = ? WHERE unit = ?').run('][', 'specimen') + db.close() + + const reopened = backendAt(path) + const damaged = await reopened.kv.open(DESCRIPTOR) + await expect(damaged.loadAll()).rejects.toMatchObject({ + name: 'StorageError', + code: 'malformed-medium', + }) + await reopened.close() + }) }) diff --git a/packages/storage/storage/src/index.ts b/packages/storage/storage/src/index.ts index ffe8b2f3f6..0e065d00f3 100644 --- a/packages/storage/storage/src/index.ts +++ b/packages/storage/storage/src/index.ts @@ -55,7 +55,10 @@ export class Storage extends Service { } this.forms.set(form, facility) return () => { - this.forms.delete(form) + // Same stale-disposer guard as BackendRegistry.register. + if (this.forms.get(form) === facility) { + this.forms.delete(form) + } } } @@ -77,10 +80,7 @@ export class Storage extends Service { } } -/** - * Mount the storage hub service. - * @param ctx - Plugin context. - */ -export function apply(ctx: Context) { - ctx.plugin(Storage) -} +// Service packages default-export their service class and nothing else +// plugin-shaped (packages/AGENTS.md): mixing a default export with a +// function-plugin `apply` makes the Loader drop the plugin namespace. +export default Storage diff --git a/packages/storage/storage/src/registry.ts b/packages/storage/storage/src/registry.ts index f8d8d41866..136cecc227 100644 --- a/packages/storage/storage/src/registry.ts +++ b/packages/storage/storage/src/registry.ts @@ -28,7 +28,11 @@ export class BackendRegistry { } this.backends.set(name, backend) return () => { - this.backends.delete(name) + // Remove only this registration's contribution: after dispose + re-register, + // a stale disposer firing again must not remove the successor. + if (this.backends.get(name) === backend) { + this.backends.delete(name) + } } } diff --git a/packages/storage/storage/tests/registry.spec.ts b/packages/storage/storage/tests/registry.spec.ts index f5cdc6bac2..cede1284e3 100644 --- a/packages/storage/storage/tests/registry.spec.ts +++ b/packages/storage/storage/tests/registry.spec.ts @@ -1,6 +1,6 @@ import { describe, expect, it } from 'vitest' import { Context } from 'cordis' -import { BackendRegistry, Storage, apply } from '../src/index.ts' +import Storage, { BackendRegistry } from '../src/index.ts' import type { StorageBackend } from '../src/index.ts' const fakeBackend = (): StorageBackend => ({ close: async () => {} }) @@ -27,7 +27,7 @@ describe('BackendRegistry', () => { describe('Storage service', () => { it('mounts on the context and exposes registry plus form mounting', async () => { const ctx = new Context() - await ctx.plugin({ apply }) + await ctx.plugin(Storage) expect(ctx.storage).toBeInstanceOf(Storage) const facility = { marker: true } diff --git a/packages/workspace/workspace/README.md b/packages/workspace/workspace/README.md index 09fc07d4bc..cdd4464cb0 100644 --- a/packages/workspace/workspace/README.md +++ b/packages/workspace/workspace/README.md @@ -6,10 +6,10 @@ Design rationale, the path/uniqueness canon, and the consistency rules live in t ## Shape -- `ctx.workspace.create(path, title?)` — canonicalizes `path` via `fs.realpath` (trailing slashes, `..`, symlinks), rejects a nonexistent directory (the original `ENOENT`) and a canonical path another workspace already owns. Title defaults to `basename(path)`. +- `ctx.workspace.create(path, title?)` — canonicalizes `path` via `fs.realpath` (trailing slashes, `..`, symlinks), rejects a nonexistent path (the original `ENOENT`), a path resolving to anything but a directory, and a canonical path another workspace already owns. Title defaults to `basename(path)`. - `ctx.workspace.get(id)` / `list()` / `resolveByPath(path)` — cache-served lookups; `resolveByPath` is async because it runs the same `realpath` canon first. - `Workspace.attachSession(id)` — idempotent; validates that the session's stored header `cwd`, canonicalized the same way, equals the workspace path. A missing persistence service, unknown session, absent or unresolvable `cwd`, or mismatch rejects without writing (what cannot be validated is not recorded). `detachSession` removes from the account only, never touching the session's own log. -- `Workspace.sessionIds` — the ordered ownership account (array order is display order). Accounted ids whose session no longer exists are filtered from the projection and pruned durably on the next mutation; a medium accounting one session under two workspaces rejects at startup (external edit — the attach check makes it unwritable). +- `Workspace.sessionIds` — the ordered ownership account (array order is display order). Accounted ids whose session no longer exists are filtered from the projection and pruned durably on the next mutation. A medium accounting one session under two workspaces, or claiming one canonical path from two records, rejects at startup (external edit — the write side makes both unreachable). Attach/detach idempotence is decided on the domain write chain, so unawaited concurrent calls settle in call order. - `Workspace.status()` — uncached directory check, `'ok' | 'missing-dir'`; a missing directory never mutates the record. Session persistence is an optional peer resolved with `ctx.get`: absent, attach rejects and projections serve the account unfiltered. diff --git a/packages/workspace/workspace/src/entity.ts b/packages/workspace/workspace/src/entity.ts index de70917cfb..42f38fe385 100644 --- a/packages/workspace/workspace/src/entity.ts +++ b/packages/workspace/workspace/src/entity.ts @@ -45,6 +45,9 @@ export interface WorkspaceEntityHost { readSessionHeader(id: SessionId): Promise } +/** Chain-slot abort sentinel thrown by the update fn when the record needs no change; only `mutate` observes it. */ +const unchangedSentinel = new Error('workspace record unchanged (internal sentinel)') + /** The single {@link Workspace} implementation; constructed only by the registry. */ export class WorkspaceEntity implements Workspace { private record: WorkspaceRecord @@ -81,29 +84,34 @@ export class WorkspaceEntity implements Workspace { } async attachSession(sessionId: SessionId): Promise { - if (this.record.sessionIds.includes(sessionId)) return - const header = await this.host.readSessionHeader(sessionId) - if (header.cwd === undefined) { - throw new Error( - `cannot attach session '${sessionId}' to workspace '${this.record.path}': ` - + 'its stored header carries no cwd to validate against', - ) - } - let cwd: string - try { - cwd = await realpathNormalize(header.cwd) - } catch (error) { - throw new Error( - `cannot attach session '${sessionId}' to workspace '${this.record.path}': ` - + `its cwd '${header.cwd}' does not resolve, so it cannot be validated`, - { cause: error }, - ) - } - if (cwd !== this.record.path) { - throw new Error( - `cannot attach session '${sessionId}' to workspace '${this.record.path}': ` - + `its cwd resolves to '${cwd}'`, - ) + // Validation is skipped when the settled snapshot already accounts the + // id: the cwd fact was checked when it first attached and both inputs + // (stored header cwd, workspace path) are immutable. Membership itself is + // decided on the write chain inside `mutate`, never on this snapshot. + if (!this.record.sessionIds.includes(sessionId)) { + const header = await this.host.readSessionHeader(sessionId) + if (header.cwd === undefined) { + throw new Error( + `cannot attach session '${sessionId}' to workspace '${this.record.path}': ` + + 'its stored header carries no cwd to validate against', + ) + } + let cwd: string + try { + cwd = await realpathNormalize(header.cwd) + } catch (error) { + throw new Error( + `cannot attach session '${sessionId}' to workspace '${this.record.path}': ` + + `its cwd '${header.cwd}' does not resolve, so it cannot be validated`, + { cause: error }, + ) + } + if (cwd !== this.record.path) { + throw new Error( + `cannot attach session '${sessionId}' to workspace '${this.record.path}': ` + + `its cwd resolves to '${cwd}'`, + ) + } } await this.mutate(record => record.sessionIds.includes(sessionId) ? record @@ -111,11 +119,9 @@ export class WorkspaceEntity implements Workspace { } async detachSession(sessionId: SessionId): Promise { - if (!this.record.sessionIds.includes(sessionId)) return - await this.mutate(record => ({ - ...record, - sessionIds: record.sessionIds.filter(id => id !== sessionId), - })) + await this.mutate(record => record.sessionIds.includes(sessionId) + ? { ...record, sessionIds: record.sessionIds.filter(id => id !== sessionId) } + : record) } async status(): Promise<'ok' | 'missing-dir'> { @@ -133,18 +139,31 @@ export class WorkspaceEntity implements Workspace { * `table.update`, stamping `updatedAt` and pruning accounted ids whose * session no longer exists (consistency rule: dead ids are dropped on the * next mutation, whatever that mutation is), then swap the snapshot. + * + * `fn` sees the value current at its chain slot, so membership decisions + * (attach/detach idempotence) are race-free against queued writes; a fn + * signalling no change by returning `current` verbatim aborts the slot + * through the sentinel when pruning also finds nothing, so a no-op neither + * rewrites the medium nor emits a change event. */ private async mutate(fn: (record: WorkspaceRecord) => WorkspaceRecord): Promise { const known = this.host.knownSessionIds() - this.record = await this.host.table().update(this.id, (current) => { - const next = fn(current) - return { - ...next, - sessionIds: known === undefined - ? next.sessionIds - : next.sessionIds.filter(id => known.has(id)), - updatedAt: new Date().toISOString(), - } - }) + let next: WorkspaceRecord + try { + next = await this.host.table().update(this.id, (current) => { + const changed = fn(current) + const sessionIds = known === undefined + ? changed.sessionIds + : changed.sessionIds.filter(id => known.has(id)) + if (changed === current && sessionIds.length === current.sessionIds.length) { + throw unchangedSentinel + } + return { ...changed, sessionIds, updatedAt: new Date().toISOString() } + }) + } catch (error) { + if (error === unchangedSentinel) return + throw error + } + this.record = next } } diff --git a/packages/workspace/workspace/src/index.ts b/packages/workspace/workspace/src/index.ts index be5bb791a8..584c3bde77 100644 --- a/packages/workspace/workspace/src/index.ts +++ b/packages/workspace/workspace/src/index.ts @@ -7,6 +7,7 @@ */ import { randomUUID } from 'node:crypto' +import { stat } from 'node:fs/promises' import { basename } from 'node:path' import { Context, Service } from 'cordis' import type { SessionHeader, SessionId } from '@deepseek-ai/dsh-session' @@ -88,12 +89,22 @@ export class WorkspaceRegistry extends Service { if (persistence !== undefined) { this.known = new Set((await persistence.list()).map(header => header.id)) } - // Rebuild entities, rejecting a double account: one session recorded - // under two workspaces means the medium was edited externally (the attach - // check makes it structurally impossible to write), and hiding it would - // silently pick a winner. + // Rebuild entities, rejecting states the write side makes structurally + // impossible (an external medium edit is the only way in, and hiding it + // would silently pick a winner): one session accounted under two + // workspaces, or two records claiming one canonical path (plain string + // equality — stored paths are already canonical, so no realpath here). const accounted = new Map() + const paths = new Map() for (const [id, record] of this.table.entries()) { + const pathHolder = paths.get(record.path) + if (pathHolder !== undefined) { + throw new Error( + `workspace domain is inconsistent: path '${record.path}' is claimed ` + + `by both workspace '${pathHolder}' and workspace '${id}'`, + ) + } + paths.set(record.path, id) for (const sessionId of record.sessionIds) { const holder = accounted.get(sessionId) if (holder !== undefined) { @@ -110,9 +121,10 @@ export class WorkspaceRegistry extends Service { /** * Create a workspace over an existing directory. The path is canonicalized - * through `fs.realpath` first — a nonexistent directory rejects with the - * original `ENOENT`, and a canonical path already owned by another - * workspace (including a symlink resolving to it) rejects. + * through `fs.realpath` first — a nonexistent path rejects with the + * original `ENOENT`, a path resolving to anything but a directory rejects, + * and a canonical path already owned by another workspace (including a + * symlink resolving to it) rejects. * @param path - Directory the workspace points at; canonicalized before storing. * @param title - Display title; defaults to `basename` of the canonical path. * @returns the created workspace after durability. @@ -120,6 +132,9 @@ export class WorkspaceRegistry extends Service { async create(path: string, title?: string): Promise { const table = this.requireTable() const canonical = await realpathNormalize(path) + if (!(await stat(canonical)).isDirectory()) { + throw new Error(`cannot create a workspace at '${canonical}': path is not a directory`) + } for (const entity of this.entities.values()) { if (entity.path === canonical) { throw new Error(`a workspace for '${canonical}' already exists ('${entity.id}')`) diff --git a/packages/workspace/workspace/src/types.ts b/packages/workspace/workspace/src/types.ts index 197273fbb7..cc5d75653d 100644 --- a/packages/workspace/workspace/src/types.ts +++ b/packages/workspace/workspace/src/types.ts @@ -53,13 +53,15 @@ export interface Workspace { /** * Record a session under this workspace. Idempotent: a session already on - * the account resolves without writing. Otherwise the session's stored - * header is read from session persistence and its `cwd`, normalized through - * the same `fs.realpath` canon as workspace paths, must equal this - * workspace's {@link path} — a missing persistence service, an unknown - * session id, a header without `cwd`, a `cwd` that no longer resolves, or a - * mismatched `cwd` all reject without touching the account (what cannot be - * validated is not recorded). + * the account resolves without writing (membership is decided on the + * domain write chain, so unawaited concurrent attach/detach calls settle + * in call order). For a session not yet on the account, its stored header + * is read from session persistence and its `cwd`, normalized through the + * same `fs.realpath` canon as workspace paths, must equal this workspace's + * {@link path} — a missing persistence service, an unknown session id, a + * header without `cwd`, a `cwd` that no longer resolves, or a mismatched + * `cwd` all reject without touching the account (what cannot be validated + * is not recorded). * @param sessionId - The session to record. * @returns resolution after durability. */ @@ -67,8 +69,8 @@ export interface Workspace { /** * Remove a session from this workspace's account. Idempotent: an id not on - * the account resolves without writing. Never touches the session's own - * stored log. + * the account resolves without writing (decided on the domain write chain, + * like attach). Never touches the session's own stored log. * @param sessionId - The session to remove. * @returns resolution after durability. */ diff --git a/packages/workspace/workspace/tests/workspace.spec.ts b/packages/workspace/workspace/tests/workspace.spec.ts index eb42ec2850..4ff1706b7b 100644 --- a/packages/workspace/workspace/tests/workspace.spec.ts +++ b/packages/workspace/workspace/tests/workspace.spec.ts @@ -1,9 +1,9 @@ import { afterEach, describe, expect, it } from 'vitest' -import { mkdir, mkdtemp, realpath, rm, symlink } from 'node:fs/promises' +import { mkdir, mkdtemp, realpath, rm, symlink, writeFile } from 'node:fs/promises' import { tmpdir } from 'node:os' import { basename, join } from 'node:path' import { Context } from 'cordis' -import { apply as applyStorage } from '@deepseek-ai/dsh-storage' +import Storage from '@deepseek-ai/dsh-storage' import { DomainFacility } from '@deepseek-ai/dsh-domain' import type { DomainChanged } from '@deepseek-ai/dsh-domain' import { SessionId } from '@deepseek-ai/dsh-session' @@ -26,7 +26,7 @@ async function harness(options?: { sessions?: SessionHeader[] | 'absent' }) { const ctx = new Context() - await ctx.plugin({ apply: applyStorage }) + await ctx.plugin(Storage) ctx.storage.backend.register('memory', new MemoryStorageBackend(options?.pool)) ctx.storage.mount('domain', new DomainFacility(ctx, { backend: 'memory', routes: {} })) let listed = options?.sessions === 'absent' ? undefined : options?.sessions ?? [] @@ -106,6 +106,15 @@ describe('WorkspaceRegistry.create', () => { expect(registry.list()).toEqual([]) }) + it('rejects a path resolving to a plain file', async () => { + const dir = await makeDir('has-file') + const file = join(dir, 'plain.txt') + await writeFile(file, 'not a directory') + const { registry } = await harness() + await expect(registry.create(file)).rejects.toThrow(/not a directory/) + expect(registry.list()).toEqual([]) + }) + it('rejects a duplicate path, including a symlink resolving to an existing workspace', async () => { const dir = await makeDir('real') const link = join(base, 'link') @@ -187,6 +196,22 @@ describe('Workspace.attachSession', () => { await workspace.detachSession(SessionId('absent')) expect(changes.length).toBe(written) }) + + it('decides membership at the write-chain slot: unawaited detach then attach re-attaches', async () => { + const dir = await makeDir('race') + const { registry } = await harness({ sessions: [header('s1', dir)] }) + const workspace = await registry.create(dir) + await workspace.attachSession(SessionId('s1')) + // Both fire before either lands. Snapshot-based idempotence would see + // 's1' still on the account and turn the attach into a no-op, losing it; + // chain-slot decisions replay detach → attach in order. (The attach skips + // re-validation off the same stale snapshot — the cwd fact is immutable — + // and enqueues immediately, keeping the chain order deterministic here.) + const detached = workspace.detachSession(SessionId('s1')) + const attached = workspace.attachSession(SessionId('s1')) + await Promise.all([detached, attached]) + expect(workspace.sessionIds).toEqual(['s1']) + }) }) describe('consistency projections', () => { @@ -214,16 +239,29 @@ describe('consistency projections', () => { }) it('rejects startup over a medium accounting one session twice', async () => { - const dir = await makeDir('double') - const pool = pooledRecord('00000000-0000-4000-8000-000000000003', record(dir, ['dup'])) + const dirA = await makeDir('double-a') + const dirB = await makeDir('double-b') + const pool = pooledRecord('00000000-0000-4000-8000-000000000003', record(dirA, ['dup'])) pool.media.get('workspace')!.tables.get('workspaces')! - .set('00000000-0000-4000-8000-000000000004', record(dir, ['dup'])) + .set('00000000-0000-4000-8000-000000000004', record(dirB, ['dup'])) const ctx = new Context() - await ctx.plugin({ apply: applyStorage }) + await ctx.plugin(Storage) ctx.storage.backend.register('memory', new MemoryStorageBackend(pool)) ctx.storage.mount('domain', new DomainFacility(ctx, { backend: 'memory', routes: {} })) await expect(Promise.resolve(ctx.plugin(WorkspaceRegistry))).rejects.toThrow(/accounted/) }) + + it('rejects startup over a medium where two records claim one path', async () => { + const dirA = await makeDir('claimed') + const pool = pooledRecord('00000000-0000-4000-8000-000000000005', record(dirA, [])) + pool.media.get('workspace')!.tables.get('workspaces')! + .set('00000000-0000-4000-8000-000000000006', record(dirA, [])) + const ctx = new Context() + await ctx.plugin(Storage) + ctx.storage.backend.register('memory', new MemoryStorageBackend(pool)) + ctx.storage.mount('domain', new DomainFacility(ctx, { backend: 'memory', routes: {} })) + await expect(Promise.resolve(ctx.plugin(WorkspaceRegistry))).rejects.toThrow(/claimed/) + }) }) describe('Workspace.status', () => {