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.
This commit is contained in:
@@ -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.
|
||||
|
||||
@@ -45,6 +45,9 @@ export interface WorkspaceEntityHost {
|
||||
readSessionHeader(id: SessionId): Promise<SessionHeader>
|
||||
}
|
||||
|
||||
/** 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<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<string>((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<string, WorkspaceId>()
|
||||
const paths = new Map<string, WorkspaceId>()
|
||||
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<Workspace> {
|
||||
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}')`)
|
||||
|
||||
@@ -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.
|
||||
*/
|
||||
|
||||
@@ -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', () => {
|
||||
|
||||
Reference in New Issue
Block a user