feat(storage): support compatible whole-unit versions (#6957)

* feat(storage): support compatible whole-unit versions

* fix(storage): contain failed writes

* test(storage): cover lock rejection and per-record version isolation

* docs(storage): align version rules with review findings

* fix(storage): unify versioned batch writes and deletion recovery
This commit is contained in:
imccyu
2026-10-26 13:22:25 +08:00
committed by GitHub
32 changed files with 799 additions and 289 deletions
@@ -12,17 +12,17 @@
en: 34e5bde1535a2bb1
zh: 559deb25b2ec50ba
/agent-note-domain-kv-storage-capability-seam-and-the-workspace-entity/proposal/dsh-storage-the-storage-hub:
en: 191afc01a150e4a0
zh: da3fbc33875a4f60
en: 5eb0b18c2adf5111
zh: 856ee7b1464b8bfc
/agent-note-domain-kv-storage-capability-seam-and-the-workspace-entity/proposal/dsh-storage-json:
en: a18ec1e534a5488b
zh: 2f5beb445e30554b
/agent-note-domain-kv-storage-capability-seam-and-the-workspace-entity/proposal/dsh-storage-sqlite:
en: 6a26dd7ef357db74
zh: 951b31136e2415f9
en: 1a4bebbbf52db7a2
zh: 425c32b0dc17b49e
/agent-note-domain-kv-storage-capability-seam-and-the-workspace-entity/proposal/dsh-domain-the-domain-data-form:
en: 2ee24f36cd93f7dd
zh: 45c83bd745de41f9
en: b99e07359dc78dc8
zh: 4af881d0b6d5b628
/agent-note-domain-kv-storage-capability-seam-and-the-workspace-entity/proposal/future-work-session-side-deletion-design-settled-not-implemented-this-phase:
en: a3824326587b6539
zh: 9736e3060bd57651
@@ -44,9 +44,9 @@ A backend is one **medium owner** (a file-tree root / one db file) exposing prim
The backend contract (asserted clause by clause by the shared conformance suite, one suite for both backends):
1. `open` creates when the medium holds nothing (lazy materialization allowed: may defer to the first write, but `loadAll` must immediately serve empty tables); loads when the medium exists.
2. A stored version ≠ descriptor.version → `StorageError('version-mismatch')`; no migration, no rebuild.
2. Unit-wide checks accept `descriptor.version` and the owner's `compatibleVersions`; other stamps reject with `StorageError('version-mismatch')`. Ordinary compatible reads preserve saved stamps and values. [Domain versioning](../../../../docs/subsystems/storage.md#domain-versioning) defines write stamping and JSON per-record exceptions.
3. Durability: after a write primitive resolves, a process crash followed by a re-open must observe the write in `loadAll`.
4. The backend does not promise write ordering within a unit — **the caller serializes**; the backend only guarantees each single call is atomic (JSON whole-file replace / SQLite single statement).
4. The backend does not promise write ordering within a unit — **the caller serializes**; whole-unit mutations atomically publish their data and current stamp (JSON whole-file replacement / SQLite transaction for batch deletion or advancing the stamp); per-record batches can be partially applied if interrupted before resolution.
5. `deleteRecord` is idempotent; `putRecord` overwrites.
6. Any string key / any JSON value is safe (keys never reach file paths, a structural property).
7. `close` is idempotent; any operation after close → `StorageError('closed')`.
@@ -88,8 +88,8 @@ CREATE TABLE IF NOT EXISTS "u_<unit>_<table>" (
key TEXT PRIMARY KEY, value TEXT NOT NULL) STRICT; -- value = record JSON document
```
- Unit versions live in `units` rows; a descriptor mismatch → `version-mismatch`. Row granularity is document-per-row, preserving precise per-key durable updates (the path left open for high-frequency point-update tables like the session sidecar); when query needs appear, JSON1 reads the value column directly.
- Write primitives are single statements and thus atomic; no cross-statement transactions needed (the domain layer has no cross-table transactions, see the out-of-scope list).
- Unit versions live in `units` rows; an unaccepted version rejects with `version-mismatch`. Row granularity is document-per-row, preserving precise per-key durable updates (the path left open for high-frequency point-update tables like the session sidecar); when query needs appear, JSON1 reads the value column directly.
- Current-version single-record writes use single statements. Batch deletes and the first mutation of an accepted older unit use one transaction; the first mutation commits its data and current stamp together. This does not add cross-table domain transactions.
### `dsh-domain`: the domain data form
@@ -117,6 +117,7 @@ export interface DomainTableSpec<K extends string, V> { readonly valueSchema: Zo
export interface DomainSpec {
readonly name: string // ^[a-z][a-z0-9_]*$
readonly version: number
readonly compatibleVersions?: readonly number[]
readonly global?: DomainGlobalSpec<unknown>
readonly tables: Record<string, DomainTableSpec<string, unknown>>
}
@@ -157,8 +158,8 @@ Rules:
- **Single-level mapping**: key → record, no nested tables; hierarchical needs use composite keys or fields inside the value. The two backends stay isomorphic as a result (one JSON object level ↔ one SQLite row).
- **Records are plain data**: immutable, directly JSON-serializable POJOs; values returned by `get`/`entries` must not be mutated in place (TypeScript readonly projection, no runtime freezing). Behavior-carrying domain objects belong to consumer packages.
- **Serialized writes**: one promise chain per domain; `put`/`delete`/`update`/`global.set` all queue on it; `update`'s fn runs on the chain, so concurrency cannot interleave. No active-record (pulling out a mutable object that auto-persists — uncontrollable persist timing, in conflict with the whole-unit atomic-rewrite model).
- **Version fails loud**: a stored version differing from the spec throws outright; no migration, no rebuild (the data is not regenerable; pre-release rejects old formats).
- **Serialized writes**: one promise chain per domain; `put`/`delete`/`deleteMany`/`update`/`global.set` all queue on it; `update`'s fn runs on the chain, so concurrency cannot interleave. No active-record (pulling out a mutable object that auto-persists — uncontrollable persist timing, in conflict with the whole-unit atomic-rewrite model).
- **Version compatibility**: the domain owner declares readable older versions through `compatibleVersions` and decodes historical values with its current schemas. Storage does not convert record values; [domain versioning](../../../../docs/subsystems/storage.md#domain-versioning) defines stamp publication and mixed-generation requirements.
- **Change events**: after each write's durability resolves, emit `domain/changed` (`@mode emit`), one per record, no old value (matching the repository's "new snapshot + operation discriminant" convention, template `goal/changed`); the payload `DomainChanged` is a put/deleted discriminated union — domain + table + key (both `''` for global changes) + operation, with the put branch carrying the new snapshot value and the deleted branch carrying none (`packages/storage/storage-domain/src/events.ts`). This is next phase's RPC push-frame event source. The error vocabulary is `DomainError`, codes: `already-open` / `facet-unsupported` / `invalid-record` (with `{ table, key }`) / `missing-key` / `closed`.
### Future work: session-side deletion (design settled, not implemented this phase)
@@ -44,9 +44,9 @@ host 侧唯一的持久化面是 session 事件日志(`packages/session/sessio
后端约定(共享约定测试逐条断言,两后端同套件):
1. `open` 对不存在的介质创建(懒物化允许:可延迟到首写,但 `loadAll` 立即可用返回空表);对已存在介质载入。
2. 介质上版本 ≠ descriptor.version → `StorageError('version-mismatch')`,不迁移不重建。
2. 单元级检查接受 `descriptor.version` 和拥有方声明的 `compatibleVersions`;其他版本戳以 `StorageError('version-mismatch')` 拒绝。普通兼容读取保留已存版本戳和值。[领域版本](../../../../docs/subsystems/storage.zh.md#domain-versioning)定义写入盖戳和 JSON 逐记录例外。
3. 持久性:写原语 resolve 后进程崩溃再 open,`loadAll` 必须反映该写入。
4. 后端不承诺 unit 内写并发序——**调用方负责串行**;后端只保证单次调用原子(JSON 整文件替换 / SQLite 单语句)。
4. 后端不承诺 unit 内写并发序——**调用方负责串行**;整单元修改原子发布其数据和当前版本戳(JSON 整文件替换 / SQLite 在批量删除或推进版本戳时使用事务);逐记录批次在完成前中断时可能只应用一部分。
5. `deleteRecord` 幂等;`putRecord` 覆写。
6. 任意字符串 key / 任意 JSON 值安全(key 不进文件路径,结构性质)。
7. `close` 幂等;close 后任何操作 → `StorageError('closed')`。
@@ -88,8 +88,8 @@ CREATE TABLE IF NOT EXISTS "u_<unit>_<table>" (
key TEXT PRIMARY KEY, value TEXT NOT NULL) STRICT; -- value = record JSON document
```
- unit 版本存 `units` 行,descriptor 不符 → `version-mismatch`。行粒度 document-per-row,保住按 key 精确落盘更新(为 session sidecar 这类高频点更新大表留路);查询需求出现时 JSON1 直查 value 列。
- 写原语单语句即原子,无跨语句事务需求(domain 层无跨表事务,见不做清单)。
- unit 版本存 `units` 行,未接受的版本以 `version-mismatch` 拒绝。行粒度 document-per-row,保住按 key 精确落盘更新(为 session sidecar 这类高频点更新大表留路);查询需求出现时 JSON1 直查 value 列。
- 当前版本的单记录写入使用单条语句。批量删除和对已接受旧单元的首次修改使用一个事务;首次修改一同提交数据和当前版本戳。这不增加跨表领域事务。
### `dsh-domain`:领域数据形式
@@ -117,6 +117,7 @@ export interface DomainTableSpec<K extends string, V> { readonly valueSchema: Zo
export interface DomainSpec {
readonly name: string // ^[a-z][a-z0-9_]*$
readonly version: number
readonly compatibleVersions?: readonly number[]
readonly global?: DomainGlobalSpec<unknown>
readonly tables: Record<string, DomainTableSpec<string, unknown>>
}
@@ -157,8 +158,8 @@ export interface KvTable<K extends string, V> {
- **一级 mapping**:key → 记录,不做嵌套表;层级需求用复合 key 或值内字段。两后端因此同构(JSON object 一层 ↔ SQLite 一行)。
- **记录是纯数据**:可直接 JSON 序列化的不可变 POJO;`get`/`entries` 返回值不得原地改(TypeScript readonly 投影,不做运行时冻结)。带行为的领域对象属于消费方包。
- **写串行**:域内一条 promise 链,`put`/`delete`/`update`/`global.set` 全排队;`update` 的 fn 在链上执行,并发不交错。不做 active-record(取出可变对象自动落盘——落盘时机不可控,与整域原子覆写冲突)。
- **版本 fail loud**:盘上版本与 spec 不符直接报错,不迁移不重建(数据不可再生,pre-release 拒绝旧格式)。
- **写串行**:域内一条 promise 链,`put`/`delete`/`deleteMany`/`update`/`global.set` 全排队;`update` 的 fn 在链上执行,并发不交错。不做 active-record(取出可变对象自动落盘——落盘时机不可控,与整域原子覆写冲突)。
- **版本兼容**:领域拥有方通过 `compatibleVersions` 声明可读的旧版本,并用当前 schema 解码历史值。Storage 不转换记录值;[领域版本](../../../../docs/subsystems/storage.zh.md#domain-versioning)定义版本戳发布和混合代际要求。
- **变更事件**:每次写落盘 resolve 后 emit `domain/changed`(`@mode emit`),逐条发、不带旧值(对齐仓库"新快照 + 操作判别"惯例,范本 `goal/changed`);payload `DomainChanged` 是 put/deleted 判别联合——域名 + 表名 + key(global 变更两者为 `''`)+ operation,put 支带新快照 value、deleted 支无 value(`packages/storage/storage-domain/src/events.ts`)。此为下期 RPC 推帧的事件源。错误词汇 `DomainError`,码表:`already-open` / `facet-unsupported` / `invalid-record`(带 `{ table, key }`)/ `missing-key` / `closed`。
### Future work:session 侧删除(设计定案,本期不实施)
+9 -6
View File
@@ -9,14 +9,17 @@
en: b11e06e7c0e0fd13
zh: 4585f26ff416cb41
/storage/the-backend-contract:
en: 225e25b9b4b08865
zh: 53bd3554da610607
en: 50afeec4e9fd3e08
zh: e8d2892c83ad3441
/storage/declaring-a-domain:
en: 775cce941500f01f
zh: 78acbc67f48574cb
en: fa4613279da360c5
zh: 09e7a7d63bd1774a
/storage/domain-versioning:
en: 37e40cc29f3bb898
zh: bbd55e732bd4b83b
/storage/the-open-domain:
en: b69b25bc06f1c052
zh: 8aca58718da520c7
en: c0787f6a496b4c46
zh: f219b16ec96d6cb6
/storage/the-domain-facility-ctx-storagedomain:
en: deed74fcd4506ef4
zh: 8e3e6e5d2dc2e989
+26 -9
View File
@@ -44,7 +44,7 @@ interface StorageBackend {
}
```
A backend owns one medium (a file-tree root, a database file) and exposes optional operation groups; `kv` is the only shipped group. `KvFacet.open(descriptor)` opens one named unit — `KvUnitDescriptor` carries the name, current format version, optional compatible record versions, table names, and whether a global singleton slot exists — and returns a `KvUnit` with `loadAll`, `putRecord`, `deleteRecord`, `deleteRecords`, `setGlobal`, and `close`. Unit and table names must match `UNIT_NAME_RE` (safe as a file name and as a SQL identifier segment); record keys are opaque strings in the `single` layout and match `[a-zA-Z0-9_-]+` path segments in the `per-record` layout. A unit does not serialize concurrent writes — ordering belongs to the caller — but each single call is atomic on the medium; durability holds for single-record writes and whole-file publishes, while a `per-record` batch deletion relies on the filesystem persisting directory-entry updates. `deleteRecords` applies a whole batch in one call: it resolves once every listed record is absent from the readable state and the backend accepted the batch, and a rejection normally leaves the batch unapplied; a backend that cannot establish or restore the batch's state rejects with `unknown-state`, after which every later operation on that unit rejects with the same error until `close()` releases it (calls after close reject `closed`). A `single`-layout failure after the atomic replacement (a directory-fsync error) can leave the new file on the medium while the caller sees a rejection; the unit keeps its memory consistent and the next publish rewrites the file. A `single` medium stamped with a different version rejects `version-mismatch`; a `per-record` document stamped outside the accepted set reads as absent. A medium that cannot be parsed as the unit rejects `malformed-medium`. [`backend.ts`](../../packages/storage/storage/src/backend.ts) is the normative clause-by-clause contract, and the shared conformance suite in [`tests/contract.ts`](../../packages/storage/storage/tests/contract.ts) checks the shared clauses against each backend. The [json backend](../../packages/storage/storage-json/README.md) republishes one whole human-readable file per unit atomically; the [sqlite backend](../../packages/storage/storage-sqlite/README.md) stores one document per row in one database for frequently updated data.
A backend owns one medium (a file-tree root, a database file) and exposes optional operation groups; `kv` is the only shipped group. `KvFacet.open(descriptor)` opens one named unit — `KvUnitDescriptor` carries the name, current format version, optional compatible versions, table names, and whether a global singleton slot exists — and returns a `KvUnit` with `loadAll`, `putRecord`, `deleteRecord`, `deleteRecords`, `setGlobal`, and `close`. Unit and table names must match `UNIT_NAME_RE` (safe as a file name and as a SQL identifier segment); record keys are opaque strings in the `single` layout and match `[a-zA-Z0-9_-]+` path segments in the `per-record` layout. A unit does not serialize concurrent writes — ordering belongs to the caller — whole-document and row-store calls are atomic on the medium, while a per-record batch interrupted before resolution can be partially applied. Resolved writes are durable; per-record deletions sync directory changes on POSIX. `deleteRecords` applies a whole batch in one call: it resolves once every listed record is absent from the readable state and the backend accepted the batch, and a rejection normally leaves the batch unapplied; a backend that cannot establish or restore the batch's state rejects with `unknown-state`, after which every later operation on that unit rejects with the same error until `close()` releases it (calls after close reject `closed`). A `single`-layout failure after the atomic replacement (a directory-fsync error) can leave the new file on the medium while the caller sees a rejection; close the backend and its domains, then reopen to inspect saved state before retrying. A `single` medium stamped outside the current and declared compatible versions rejects `version-mismatch`; a `per-record` document stamped outside the accepted set reads as absent. A medium that cannot be parsed as the unit rejects `malformed-medium`. [`backend.ts`](../../packages/storage/storage/src/backend.ts) is the normative clause-by-clause contract, and the shared conformance suite in [`tests/contract.ts`](../../packages/storage/storage/tests/contract.ts) checks the shared clauses against each backend. The [json backend](../../packages/storage/storage-json/README.md) republishes one whole human-readable file per unit atomically; the [sqlite backend](../../packages/storage/storage-sqlite/README.md) stores one document per row in one database for frequently updated data.
## Declaring a domain
@@ -58,20 +58,20 @@ interface DomainSpec {
/** Current domain format version; reads enforce it according to the selected layout. */
readonly version: number
/**
* Medium layout for the backend unit: `single` (the default) stores the
* Document layout for the backend unit: `single` (the default) stores the
* whole unit as one document; `per-record` stores each record as its own
* document, for units whose records are large, sparse, or individually
* disposable — the projection cache — and scopes version checks per record
* (an unaccepted record document is discarded, never migrated).
* disposable — the projection cache — with per-record version checks.
* Row stores retain unit-wide versioning for either layout request.
*/
readonly layout?: 'single' | 'per-record'
/**
* Older domain versions whose stored records the current record schemas
* also accept (the declaring owner vouches for that, typically by
* declaring the fields older records lack as optional). `per-record` backends
* read documents stamped with a listed version instead of discarding them,
* and accept a legacy whole-unit file so stamped for the one-time
* bootstrap; writes always stamp {@link version}.
* declaring the fields older records lack as optional). Ordinary reads admit
* listed versions without restamping saved data; writes stamp {@link version}
* while leaving untouched records as saved. Current schemas must therefore
* accept mixed generations even under the current unit stamp.
*/
readonly compatibleVersions?: readonly number[]
/**
@@ -94,6 +94,23 @@ interface DomainSpec {
`defineDomain(spec)` pins the spec's literal types and fails loud at the owner'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 (`null` is the medium's "never written" sentinel, so a stored nullable global could not round-trip). `domainTable<K, V>(schema)` declares one table with a phantom compile-time key type (typically a [branded id](core.md#branded-ids)); `descriptorOf(spec)` projects the backend-facing unit descriptor.
<a id="domain-versioning"></a>
## Domain versioning
`version` names the current read/write format. `compatibleVersions` lists older versions whose records and global value the owner can decode with its current schemas; only nonnegative integers below `version` are allowed. Without that declaration, reads stay exact-version. Acceptance does not weaken record validation or permit a newer unknown version. Each row's saved fields must determine its meaning unambiguously; a unit stamp cannot distinguish different meanings of identical fields in mixed rows.
Ordinary compatible opens and reads do not restamp or normalize saved values. Whole-unit JSON publishes the current stamp and changed data in one atomic file replacement; SQLite commits its first updated unit stamp and mutation in one transaction; batch deletes always use one transaction. Deleting missing records or an empty batch does not advance the stamp. The stamp identifies an accepted format, not completion of value normalization; readers must accept mixed generations. A backend with a unit-wide stamp rejects an unsupported unit version.
Serialization failure before publication or successful SQLite rollback preserves the previous data and stamp. An I/O failure after JSON replacement, or a failure of SQLite rollback itself, can leave the saved outcome unknown despite rejection. Stop using and close the backend and its domains, then reopen to inspect saved state before deciding whether to retry. The backend READMEs define these failure cases.
JSON's `per-record` layout checks and stamps each document independently; an unsupported document reads as absent. Its existing whole-unit bootstrap is an exception to read-only loading: it copies accepted historical values into current-stamped documents without converting them. An existing declared table directory, even when empty, or a declared global document suppresses bootstrap. SQLite retains unit-wide version checks regardless of the layout request. See the [JSON backend](../../packages/storage/storage-json/README.md) and [SQLite backend](../../packages/storage/storage-sqlite/README.md) for layout and recovery behavior.
For a compatible successor, bump `version`, keep all still-readable stamps in `compatibleVersions`, and extend the same schema's decoding/defaults. Consumers normalize values when rewriting selected records; Storage does not assign application record versions. Verify historical and skipped upgrades, mixed values, reads without writes, failure rollback, restart, other-unit isolation, and unknown-version refusal. Table removal and layout changes require their own retention and upgrade policy; a compatible stamp alone does not define either.
A v2 unit may contain untouched v1 values. A v3 reader accepting stamp 2 must still decode those values; accepting stamp 1 is also necessary for users who skipped v2 entirely. Removing old decoding requires rewriting and validating every affected record, including those already under a current unit stamp. An incompatible change needs an explicit, recoverable upgrade before opening with the new schema; the storage domain has no value-migration runner.
Keep a consistent backup before writing a new format if return to an older reader is required: that reader can refuse the new stamp after the first mutation. Lowering a saved stamp by hand does not reverse value conversion. Domain compatibility does not migrate SQLite's physical schema or JSON envelope formats, and changing a backend route does not transfer stored data. Session-format migrations and SQLite physical-layout versioning have separate owners and do not migrate domain values.
## The open domain
```ts type-equiv
@@ -123,7 +140,7 @@ interface Domain<S extends DomainSpec> {
}
```
Reads are synchronous from authoritative in-memory state: `KvTable` exposes `get`/`entries`/`keys`/`size` (snapshot iterators that stay stable while queued writes land), and the global handle's `get()` serves the spec's `initial` until the first `set` materializes the slot on the medium. Every write — `put`, `delete`, `deleteMany`, `update`, `global.set` — queues on one per-domain chain and reaches backend durability first, then mutates memory, then emits `domain/changed`; a rejected backend write leaves memory untouched, so reads never diverge from the medium; a backend that reports `unknown-state` instead quarantines the domain, and its reads and later writes reject until `close()` releases it. `update(key, fn)` is an atomic read-modify-write at its chain slot (a missing key rejects `missing-key`); `delete` of an absent key resolves `false` with no write and no event. `deleteMany(keys)` removes the present keys as one backend batch in one chain slot, resolves the removed count, and emits one `domain/changed` per removed key; a rejected backend write leaves every record in place and emits nothing; an `unknown-state` rejection quarantines the domain. Returned records are the stored objects themselves, not copies — replace via `put`/`update`, never mutate in place.
Reads are synchronous from authoritative in-memory state: `KvTable` exposes `get`/`entries`/`keys`/`size` (snapshot iterators that stay stable while queued writes land), and the global handle's `get()` serves the spec's `initial` until the first `set` materializes the slot on the medium. Every write — `put`, `delete`, `deleteMany`, `update`, `global.set` — queues on one per-domain chain and reaches backend durability first, then mutates memory, then emits `domain/changed`; a rejected backend write does not publish a domain change; an unknown durable outcome requires reopening before further use of that snapshot; a backend that reports `unknown-state` instead quarantines the domain, and its reads and later writes reject until `close()` releases it. `update(key, fn)` is an atomic read-modify-write at its chain slot (a missing key rejects `missing-key`); `delete` of an absent key resolves `false` with no write and no event. `deleteMany(keys)` removes the present keys as one backend batch in one chain slot, resolves the removed count, and emits one `domain/changed` per removed key; a rejected backend write leaves every record in place and emits nothing; an `unknown-state` rejection quarantines the domain. Returned records are the stored objects themselves, not copies — replace via `put`/`update`, never mutate in place.
## The domain facility: `ctx.storageDomain`
+26 -9
View File
@@ -44,7 +44,7 @@ interface StorageBackend {
}
```
一个后端拥有一个介质(一棵文件树的根目录、一个数据库文件),并提供可选的操作组;`kv` 是唯一已交付的操作组。`KvFacet.open(descriptor)` 打开一个具名 unit——`KvUnitDescriptor` 携带名称、当前格式版本、可选的兼容记录版本、表名清单,以及是否存在全局单例 slot——并返回提供 `loadAll`、`putRecord`、`deleteRecord`、`deleteRecords`、`setGlobal` 和 `close` 的 `KvUnit`。unit 名与表名必须匹配 `UNIT_NAME_RE`(既可安全用作文件名,也可安全用作 SQL 标识符片段);记录键在 `single` 布局中是不透明字符串,在 `per-record` 布局中则是匹配 `[a-zA-Z0-9_-]+` 的路径片段。unit 不对并发写入做串行化——顺序由调用方负责——但每次单独调用在介质上都是原子的;单记录写入与整文件发布在 resolve 后即已持久,`per-record` 批量删除则依赖文件系统对目录项更新的持久化。`deleteRecords` 以一次调用应用整批删除:所有列出的记录都从可读状态消失且后端接受该批后才 resolve,拒绝时通常整批未应用;无法建立或恢复批次状态的批量删除会以 `unknown-state` 拒绝,之后该 unit 的所有操作都以同一错误拒绝,直到 `close()` 释放它(关闭后的调用以 `closed` 拒绝)。`single` 布局在原子替换之后失败(目录 fsync 错误)时,新文件可能已落在介质上而调用方看到拒绝;unit 的内存保持一致,下一次发布会重写该文件。`single` 介质上记录的版本不同时拒绝 `version-mismatch`;`per-record` 文档的版本在接受集合之外时读作不存在。无法按该 unit 解析的介质拒绝 `malformed-medium`。[`backend.ts`](../../packages/storage/storage/src/backend.ts) 是逐条款的规范性约定,[`tests/contract.ts`](../../packages/storage/storage/tests/contract.ts) 中的共享一致性套件会针对每个后端检查共享条款。[json 后端](../../packages/storage/storage-json/README.zh.md)以原子方式为每个 unit 整文件重新发布一份人类可读文件;[sqlite 后端](../../packages/storage/storage-sqlite/README.zh.md)在单个数据库中每行存储一份文档,用于频繁更新的数据。
一个后端拥有一个介质(一棵文件树的根目录、一个数据库文件),并提供可选的操作组;`kv` 是唯一已交付的操作组。`KvFacet.open(descriptor)` 打开一个具名 unit——`KvUnitDescriptor` 携带名称、当前格式版本、可选的兼容版本、表名清单,以及是否存在全局单例 slot——并返回提供 `loadAll`、`putRecord`、`deleteRecord`、`deleteRecords`、`setGlobal` 和 `close` 的 `KvUnit`。unit 名与表名必须匹配 `UNIT_NAME_RE`(既可安全用作文件名,也可安全用作 SQL 标识符片段);记录键在 `single` 布局中是不透明字符串,在 `per-record` 布局中则是匹配 `[a-zA-Z0-9_-]+` 的路径片段。unit 不对并发写入做串行化——顺序由调用方负责——整文档与行存储调用在介质上是原子的,而逐记录批次在完成前中断时可能只应用一部分。写入完成后即已持久化;逐记录删除会在 POSIX 上同步目录变更。`deleteRecords` 以一次调用应用整批删除:所有列出的记录都从可读状态消失且后端接受该批后才 resolve,拒绝时通常整批未应用;无法建立或恢复批次状态的批量删除会以 `unknown-state` 拒绝,之后该 unit 的所有操作都以同一错误拒绝,直到 `close()` 释放它(关闭后的调用以 `closed` 拒绝)。`single` 布局在原子替换之后失败(目录 fsync 错误)时,新文件可能已落在介质上而调用方看到拒绝;必须关闭后端及其领域,再重新打开以检查已存状态,然后决定是否重试。`single` 介质上的版本不在当前版本和已声明兼容版本内时拒绝 `version-mismatch`;`per-record` 文档的版本在接受集合之外时读作不存在。无法按该 unit 解析的介质拒绝 `malformed-medium`。[`backend.ts`](../../packages/storage/storage/src/backend.ts) 是逐条款的规范性约定,[`tests/contract.ts`](../../packages/storage/storage/tests/contract.ts) 中的共享一致性套件会针对每个后端检查共享条款。[json 后端](../../packages/storage/storage-json/README.zh.md)以原子方式为每个 unit 整文件重新发布一份人类可读文件;[sqlite 后端](../../packages/storage/storage-sqlite/README.zh.md)在单个数据库中每行存储一份文档,用于频繁更新的数据。
## 声明领域
@@ -58,20 +58,20 @@ interface DomainSpec {
/** Current domain format version; reads enforce it according to the selected layout. */
readonly version: number
/**
* Medium layout for the backend unit: `single` (the default) stores the
* Document layout for the backend unit: `single` (the default) stores the
* whole unit as one document; `per-record` stores each record as its own
* document, for units whose records are large, sparse, or individually
* disposable — the projection cache — and scopes version checks per record
* (an unaccepted record document is discarded, never migrated).
* disposable — the projection cache — with per-record version checks.
* Row stores retain unit-wide versioning for either layout request.
*/
readonly layout?: 'single' | 'per-record'
/**
* Older domain versions whose stored records the current record schemas
* also accept (the declaring owner vouches for that, typically by
* declaring the fields older records lack as optional). `per-record` backends
* read documents stamped with a listed version instead of discarding them,
* and accept a legacy whole-unit file so stamped for the one-time
* bootstrap; writes always stamp {@link version}.
* declaring the fields older records lack as optional). Ordinary reads admit
* listed versions without restamping saved data; writes stamp {@link version}
* while leaving untouched records as saved. Current schemas must therefore
* accept mixed generations even under the current unit stamp.
*/
readonly compatibleVersions?: readonly number[]
/**
@@ -94,6 +94,23 @@ interface DomainSpec {
`defineDomain(spec)` 固定 spec 的字面量类型,并在拥有方的模块加载时、任何介质被触碰之前就明确报错:领域名或表名不匹配 `UNIT_NAME_RE`、版本不是非负整数、global schema 接受 `null`,这些都会抛出(`null` 是介质的「从未写入」哨兵值,可空的 global 一旦存储就无法往返还原)。`domainTable<K, V>(schema)` 声明一张表,其键类型是仅存在于编译期的 phantom 类型(通常是[品牌化 id](core.zh.md#branded-ids));`descriptorOf(spec)` 投影出面向后端的 unit 描述符。
<a id="domain-versioning"></a>
## 领域版本
`version` 表示当前读写格式。`compatibleVersions` 列出拥有方可用当前 schema 解码其记录和全局值的旧版本;只允许小于 `version` 的非负整数。不声明兼容版本时,读取仍要求精确版本。接受旧版本不放宽记录校验,也不允许更新的未知版本。每行已存字段必须能无歧义地确定其含义;单元版本戳无法在混合行中区分相同字段的不同含义。
普通兼容打开和读取不会重新盖戳或规范化已存值。整单元 JSON 用一次原子文件替换发布当前版本戳和修改后的数据;SQLite 将首次单元版本更新与修改放在同一事务中提交;批量删除始终使用一个事务。删除不存在的记录或空批次不会推进版本戳。版本戳标识可接受格式,不代表值规范化已经完成;读取器必须接受混合代际。采用单元级版本戳的后端会拒绝不支持的单元版本。
发布前的序列化失败或成功的 SQLite 回滚会保留原有数据与版本戳。JSON 替换后的 I/O 失败或 SQLite 回滚本身失败时,即使操作被拒绝,已存结果也可能未知。必须停止使用并关闭后端及其领域,然后重新打开检查已存状态,再决定是否重试。各后端 README 定义了这些失败情况。
JSON 的 `per-record` 布局独立检查和标记每份文档;不支持的文档读取为缺失。其已有整单元引导是只读加载的例外:它将接受的历史值复制到当前版本文档中,不转换其中的值。已有的已声明表目录(即使为空)或已声明全局文档会禁止引导。SQLite 无论布局请求为何都保持单元级版本检查。布局与恢复行为见 [JSON 后端](../../packages/storage/storage-json/README.zh.md)和 [SQLite 后端](../../packages/storage/storage-sqlite/README.zh.md)。
兼容后继版本需提升 `version`、在 `compatibleVersions` 中保留仍可读的版本戳,并扩展同一 schema 的解码和默认规则。消费者在重写选定记录时规范化值;Storage 不分配应用记录版本。验证历史版本与跨版本升级、混合值、不写入的读取、失败回滚、重启、其他单元隔离和未知版本拒绝。删表与布局变更各自需要保留及升级策略;兼容版本戳本身不定义这些策略。
v2 单元可能包含未改动的 v1 值。接受版本戳 2 的 v3 读取器仍须解码这些值;对于跳过 v2 的用户,还必须接受版本戳 1。移除旧解码逻辑前必须重写并验证全部受影响记录,包括已处于当前单元版本戳下的记录。不兼容变更需要在用新 schema 打开前执行明确且可恢复的升级;存储领域没有值迁移执行器。
如果需要回到旧读取器,必须在写入新格式前保留一致的备份:首次修改后,旧读取器可能拒绝新的版本戳。手工降低已存版本戳不会逆转值转换。领域兼容性不迁移 SQLite 物理 schema 或 JSON 外层格式,更改后端路由也不会转移已存数据。Session 格式迁移与 SQLite 物理布局版本各有拥有方,不迁移领域值。
## 打开的领域
```ts type-equiv
@@ -123,7 +140,7 @@ interface Domain<S extends DomainSpec> {
}
```
读取是同步的,来自权威的内存态:`KvTable` 暴露 `get`/`entries`/`keys`/`size`(快照迭代器,在排队写入落地期间保持稳定),global 句柄的 `get()` 在第一次 `set` 将 slot 物化到介质之前一直返回 spec 的 `initial`。每次写入——`put`、`delete`、`deleteMany`、`update`、`global.set`——都在同一条逐领域写链上排队,先在后端完成持久化,再更新内存,最后发出 `domain/changed`;后端写入被拒时内存原样不动,因此读取绝不会偏离介质;若后端报告 `unknown-state`,领域会被隔离,读取与后续写入都会拒绝,直到 `close()` 释放它。`update(key, fn)` 在其写链 slot 上是一次原子的读-改-写(键缺失时拒绝 `missing-key`);`delete` 一个不存在的键 resolve 为 `false`,不产生写入也不产生事件。`deleteMany(keys)` 在一个写链 slot 内把存在的键作为一次后端批量删除移除,resolve 为移除数量,并为每个被移除的键发一次 `domain/changed`;后端写入被拒时所有记录保持原样,也不发事件;`unknown-state` 拒绝会隔离整个领域。返回的记录就是存储的对象本身,不是副本——请经 `put`/`update` 整体替换,绝不要就地修改。
读取是同步的,来自权威的内存态:`KvTable` 暴露 `get`/`entries`/`keys`/`size`(快照迭代器,在排队写入落地期间保持稳定),global 句柄的 `get()` 在第一次 `set` 将 slot 物化到介质之前一直返回 spec 的 `initial`。每次写入——`put`、`delete`、`deleteMany`、`update`、`global.set`——都在同一条逐领域写链上排队,先在后端完成持久化,再更新内存,最后发出 `domain/changed`;后端写入被拒时不会发布领域变更;持久化结果未知时必须重新打开后才能继续使用该快照;若后端报告 `unknown-state`,领域会被隔离,读取与后续写入都会拒绝,直到 `close()` 释放它。`update(key, fn)` 在其写链 slot 上是一次原子的读-改-写(键缺失时拒绝 `missing-key`);`delete` 一个不存在的键 resolve 为 `false`,不产生写入也不产生事件。`deleteMany(keys)` 在一个写链 slot 内把存在的键作为一次后端批量删除移除,resolve 为移除数量,并为每个被移除的键发一次 `domain/changed`;后端写入被拒时所有记录保持原样,也不发事件;`unknown-state` 拒绝会隔离整个领域。返回的记录就是存储的对象本身,不是副本——请经 `put`/`update` 整体替换,绝不要就地修改。
## 领域 facility:`ctx.storageDomain`
@@ -36,8 +36,8 @@
en: 1cc8b33834457c38
zh: c549afc7c4960e9e
/deepseek-ai-dsh-storage-domain/understand-the-implementation/design-concept:
en: 4dffdacca3d2cd9c
zh: b41875243c4a1c0b
en: c1d0f46c6cd3ffb2
zh: e111921534389435
/deepseek-ai-dsh-storage-domain/understand-the-implementation/open-sequence:
en: 052da5d6c45205ca
zh: 724c875e3c431fee
@@ -63,8 +63,8 @@
en: 0e7d1c06d17c542f
zh: ead94d355f161be1
/deepseek-ai-dsh-storage-domain/known-limitations-and-deferred-work:
en: bd03348ad87d9210
zh: f50d6f19bcac93b9
en: fc50f5990ce953c9
zh: b30fdc41199a9702
/deepseek-ai-dsh-storage-domain/known-limitations-and-deferred-work/dev-note:
en: e67f633dc6fe7773
zh: 3bb0c20c93f16f98
+3 -3
View File
@@ -85,7 +85,7 @@ The domain layer is a single implementation, not an abstracted seam: consumers d
### Design concept
- **The spec object is the single source of truth.** `defineDomain` pins the spec's literal types and validates its fields at the owning package's module load, before any medium is touched. Record schemas are zod, so `z.infer` avoids duplicating consumer types; plugin `Config` stays schemastery.
- **Memory is authoritative; the medium is the durable projection.** Reads are synchronous from validated in-memory state. Every write queues on one per-domain write chain: backend durability first, then memory mutation, then `domain/changed` — a rejected backend write leaves memory untouched, so reads never diverge from the medium; a backend that reports `unknown-state` quarantines the domain, whose reads and later writes then reject until it is closed.
- **Memory is authoritative; the medium is the durable projection.** Reads are synchronous from validated in-memory state. Every write queues on one per-domain write chain: backend durability first, then memory mutation, then `domain/changed`. A rejected backend write does not publish a domain change; an `unknown-state` rejection quarantines reads and later writes until close. Any unknown durable outcome requires closing and reopening the backend and domain before further use.
- **One write chain per domain.** `put`, `delete`, `deleteMany`, `update`, and `global.set` all queue on it; `update`'s transform runs at its chain slot, so concurrent updates never interleave. Records are plain immutable data — returned values are the stored objects themselves and must not be mutated in place.
- **Writes emit after the commit point.** `domain/changed` is a notification, not a transaction participant: a throwing listener is contained with a logged warning rather than rejecting the already-durable write.
@@ -144,8 +144,8 @@ Independent: domain reads and writes never touch request prefixes, so nothing he
These limits define when the domain layer is a poor fit or needs special operational care. They are current package constraints, not a task backlog.
- **Single-process change visibility** — `domain/changed` is an in-process event; a second host process or a reconnecting GUI observes no changes until the cross-process revision pattern lands ([Agent Note](../../../.agents/notes/proposed/architecture/2026-07-24-domain-kv-storage-and-workspace.md)).
- **No cross-table transactions, secondary indexes, or multi-segment keys** — each write touches one record; these extensions are deferred in the Agent Note's out-of-scope list.
- **No data migration** — a domain whose stored version differs from its spec rejects at open (`version-mismatch`); changing a schema requires migrating the stored data by hand.
- **No cross-table transactions, secondary indexes, or multi-segment keys** — each write targets one table or the global slot; these extensions are deferred in the Agent Note's out-of-scope list.
- **No incompatible-value migration runner** — `compatibleVersions` admits older data for the current decoder; consumers convert values when rewriting selected records. A current stamp can still contain historical rows. See [domain versioning](../../../docs/subsystems/storage.md#domain-versioning) for backend differences, upgrades, downgrade refusal, and retiring old readers.
<a id="dev-note"></a>
### Dev Note
+3 -3
View File
@@ -85,7 +85,7 @@ domain.table('workspaces').update(id, (r) => ({ ...r, path: newPath }))
### 设计理念
- **spec 对象是唯一真源。** `defineDomain` 固定 spec 的字面类型,并在所属包的模块加载时、任何介质被触碰之前校验其字段。记录 schema 使用 zod,因此 `z.infer` 可避免重复定义消费方类型;插件 `Config` 仍由 schemastery 负责。
- **内存具有最终决定权;介质是持久投影。** 读取同步取自经过校验的内存状态。每次写入都在每个领域一条的写入链上排队:先到达后端持久状态,再变更内存,然后发出 `domain/changed`——被拒绝的后端写入不会触碰内存,因此读取永远不会与介质分叉;若后端报告 `unknown-state`,领域会被隔离,其读取与后续写入都会拒绝,直到领域关闭。
- **内存具有最终决定权;介质是持久投影。** 读取同步取自经过校验的内存状态。每次写入都在每个领域一条的写入链上排队:先到达后端持久状态,再变更内存,然后发出 `domain/changed`。后端写入被拒绝时不发布领域变更;`unknown-state` 拒绝会隔离读取和后续写入,直至关闭。持久化结果未知时,必须先关闭并重新打开后端与领域,才能继续使用。
- **每个领域一条写入链。** `put`、`delete`、`deleteMany`、`update` 与 `global.set` 都在其上排队;`update` 的变换在链上自己的槽位运行,因此并发更新绝不会交错。记录是普通不可变数据——返回值就是已存对象本身,绝不能原地修改。
- **写入在提交点之后发出。** `domain/changed` 是通知,不是事务参与者:监听器抛出异常时,系统会隔离该异常并记录警告,而不会让已经持久的写入被拒绝。
@@ -144,8 +144,8 @@ domain.table('workspaces').update(id, (r) => ({ ...r, path: newPath }))
这些限制说明领域层何时不合适,或何时需要特别的运维注意。它们是当前包约束,不是任务积压。
- **变更只在单进程内可见**——`domain/changed` 是进程内事件;在跨进程修订模式落地前,第二个主机进程或重新连接的 GUI 无法观察变更([Agent Note](../../../.agents/notes/proposed/architecture/2026-07-24-domain-kv-storage-and-workspace.zh.md))。
- **没有跨表事务、二级索引或多段键**——每次写入只触碰一条记录;这些扩展列在 Agent Note 的范围外清单中。
- **没有数据迁移**——领域的已存版本与 spec 不同时,打开操作会被拒绝(`version-mismatch`);修改 schema 需要手工迁移已存数据。
- **没有跨表事务、二级索引或多段键**——每次写入针对一张表或全局槽位;这些扩展列在 Agent Note 的范围外清单中。
- **没有不兼容值迁移执行器**——`compatibleVersions` 允许当前解码器读取旧数据;消费方在重写选定记录时转换其值。当前版本戳下仍可能包含历史行。后端差异、升级、降级拒绝与移除旧读取器参阅[领域版本](../../../docs/subsystems/storage.zh.md#domain-versioning)。
<a id="dev-note"></a>
### 开发备注
+11 -14
View File
@@ -352,12 +352,7 @@ class KvTableImpl<K extends string, V> implements KvTable<K, V> {
if (!this.records.has(key)) return false
await this.host.unit.deleteRecord(this.tableName, key)
this.records.delete(key)
this.host.emitChanged({
domain: this.host.domainName,
table: this.tableName,
key,
operation: 'deleted',
})
this.emitDeleted(key)
return true
})
}
@@ -372,14 +367,7 @@ class KvTableImpl<K extends string, V> implements KvTable<K, V> {
// Memory reaches the committed batch state before the first listener runs,
// so every `domain/changed` observes the same removed set as the medium.
for (const key of present) this.records.delete(key)
for (const key of present) {
this.host.emitChanged({
domain: this.host.domainName,
table: this.tableName,
key,
operation: 'deleted',
})
}
for (const key of present) this.emitDeleted(key)
return present.length
})
}
@@ -409,4 +397,13 @@ class KvTableImpl<K extends string, V> implements KvTable<K, V> {
value,
})
}
private emitDeleted(key: K): void {
this.host.emitChanged({
domain: this.host.domainName,
table: this.tableName,
key,
operation: 'deleted',
})
}
}
+7 -7
View File
@@ -38,20 +38,20 @@ export interface DomainSpec {
/** Current domain format version; reads enforce it according to the selected layout. */
readonly version: number
/**
* Medium layout for the backend unit: `single` (the default) stores the
* Document layout for the backend unit: `single` (the default) stores the
* whole unit as one document; `per-record` stores each record as its own
* document, for units whose records are large, sparse, or individually
* disposable — the projection cache — and scopes version checks per record
* (an unaccepted record document is discarded, never migrated).
* disposable — the projection cache — with per-record version checks.
* Row stores retain unit-wide versioning for either layout request.
*/
readonly layout?: 'single' | 'per-record'
/**
* Older domain versions whose stored records the current record schemas
* also accept (the declaring owner vouches for that, typically by
* declaring the fields older records lack as optional). `per-record` backends
* read documents stamped with a listed version instead of discarding them,
* and accept a legacy whole-unit file so stamped for the one-time
* bootstrap; writes always stamp {@link version}.
* declaring the fields older records lack as optional). Ordinary reads admit
* listed versions without restamping saved data; writes stamp {@link version}
* while leaving untouched records as saved. Current schemas must therefore
* accept mixed generations even under the current unit stamp.
*/
readonly compatibleVersions?: readonly number[]
/**
@@ -37,9 +37,9 @@ export class MemoryMediaPool {
/** Unit name → stamped version; tests may pre-stamp to force `version-mismatch`. */
readonly versions = new Map<string, number>()
/**
* When positive, that many subsequent write primitives (putRecord /
* deleteRecord / setGlobal) reject without touching the medium, decrementing
* per rejection. Negative path: callers assert their state is
* When positive, that many subsequent mutations reject without touching
* the medium, decrementing per rejection. A missing-key delete is a no-op
* and leaves this counter unchanged. Callers assert their state is
* untouched after a durability failure.
*/
failNextWrites = 0
@@ -88,26 +88,29 @@ class MemoryKvUnit implements KvUnit {
this.medium.tables.set(table, records)
}
records.set(key, value)
this.pool.versions.set(this.descriptor.name, this.descriptor.version)
}
async deleteRecord(table: string, key: string): Promise<void> {
this.assertOpen()
this.pool.consumeInjectedFailure()
this.medium.tables.get(table)?.delete(key)
deleteRecord(table: string, key: string): Promise<void> {
return this.deleteRecords(table, [key])
}
async deleteRecords(table: string, keys: readonly string[]): Promise<void> {
this.assertOpen()
this.pool.consumeInjectedFailure()
const records = this.medium.tables.get(table)
if (records === undefined) return
for (const key of new Set(keys)) records.delete(key)
const present = [...new Set(keys)].filter(key => records.has(key))
if (present.length === 0) return
this.pool.consumeInjectedFailure()
for (const key of present) records.delete(key)
this.pool.versions.set(this.descriptor.name, this.descriptor.version)
}
async setGlobal(value: unknown): Promise<void> {
this.assertOpen()
this.pool.consumeInjectedFailure()
this.medium.global = value
this.pool.versions.set(this.descriptor.name, this.descriptor.version)
}
async close(): Promise<void> {
@@ -144,7 +147,7 @@ export class MemoryStorageBackend implements StorageBackend {
const stamped = this.pool.versions.get(descriptor.name)
if (stamped === undefined) {
this.pool.versions.set(descriptor.name, descriptor.version)
} else if (stamped !== descriptor.version) {
} else if (stamped !== descriptor.version && !descriptor.compatibleVersions?.includes(stamped)) {
throw new StorageError(
'version-mismatch',
`memory unit '${descriptor.name}' is stamped v${stamped}, descriptor wants v${descriptor.version}`,
+10 -10
View File
@@ -9,8 +9,8 @@
en: b04256279a328d19
zh: b0913489b49403b1
/deepseek-ai-dsh-storage-json/summary:
en: 9eb9c035b03acd09
zh: 2adec96e42a054e1
en: ba875a866989d7d8
zh: 92e94db1d3d3577e
/deepseek-ai-dsh-storage-json/table-of-contents:
en: d152484eb41ac6b4
zh: 09388d293f9be9cb
@@ -18,20 +18,20 @@
en: 87ddc847b7024cac
zh: 6e97d0f216587d5b
/deepseek-ai-dsh-storage-json/use-this-package/when-to-choose-it:
en: 28ebb528cbe13f31
zh: 7e3744e9d080d2bb
en: 40e60fa139e0d6f6
zh: 0c4ce58c26ee4bf7
/deepseek-ai-dsh-storage-json/use-this-package/configuration:
en: 7144aa3b5af76874
zh: dfc528d027f86037
/deepseek-ai-dsh-storage-json/use-this-package/observable-behavior:
en: 75b7e8c8ae917439
zh: 30ff8fc276621b7a
en: 79d83481397cb10b
zh: e160178363a57ba6
/deepseek-ai-dsh-storage-json/understand-the-implementation:
en: b7d222309b3e5438
zh: 807c7314ee341d94
/deepseek-ai-dsh-storage-json/understand-the-implementation/design-concept:
en: 6cd51dcd7e255715
zh: 8c4514893f89064a
en: 975a4b005d9f8d0d
zh: 71127701477ae358
/deepseek-ai-dsh-storage-json/understand-the-implementation/file-formats:
en: b3de0a697db4f916
zh: 56b26f7823392bd5
@@ -57,8 +57,8 @@
en: c13c59d13a5f1651
zh: 5aab838ae3ff5324
/deepseek-ai-dsh-storage-json/known-limitations-and-deferred-work:
en: 07d386decf38f247
zh: c3e866b4d8e3372a
en: 073ca8d8c9bb8123
zh: 088ca8438f11ce6c
/deepseek-ai-dsh-storage-json/known-limitations-and-deferred-work/dev-note:
en: dd40dd4cb9109aea
zh: 3fd79d2242eccadb
+9 -7
View File
@@ -9,7 +9,7 @@ English | [中文](README.zh.md)
## Summary
`dsh-storage-json` stores domain data as readable JSON under a configured root and registers as backend `json`. Its default `single` layout keeps one complete `<unit>.json` file per unit; its `per-record` layout keeps one version-stamped document per record. Both layouts publish each changed file atomically, while the domain layer orders calls. Choose it when operators need inspectable files and the selected layout fits the write volume; choose SQLite for larger or highly concurrent data. The backend is host-side only and contributes no prompt, tool, or schema.
`dsh-storage-json` stores domain data as readable JSON under a configured root and registers as backend `json`. Its default `single` layout keeps one complete `<unit>.json` file per unit; its `per-record` layout keeps one version-stamped document per record. Both layouts publish each changed file atomically, while the domain layer orders calls. Choose it when operators need inspectable files and the selected layout fits the write volume; choose SQLite for larger, frequently updated units. The backend is host-side only and contributes no prompt, tool, or schema.
## Table of Contents
@@ -29,7 +29,7 @@ Use this package when a composition needs readable, editable JSON storage. Route
### When to choose it
Choose the default `single` layout for small units that benefit from one complete, pretty-printed file. Choose `per-record` when point writes should replace only one record document. Choose the SQLite backend when data is large, writes are frequent, or multiple records need transactional updates.
Choose the default `single` layout for small units that benefit from one complete, pretty-printed file. Choose `per-record` for disposable records whose point writes should replace only one document; unsupported or unreadable documents read as absent. Choose SQLite for frequent point updates in large units. The domain API has no multi-record transaction operation on either backend.
### Configuration
@@ -53,9 +53,11 @@ The generated [configuration catalog](../../../docs/config-catalog.md#deepseek-a
### Observable behavior
A missing `single` file or `per-record` directory opens as an empty unit and materializes on the first write. In `single`, malformed content rejects with `malformed-medium`, and a different stored version rejects with `version-mismatch`. In `per-record`, each malformed or unreadable document, and each document whose version is outside the descriptor's current and compatible versions, reads as an absent record, so one bad document does not reject the unit. In `per-record`, record keys must match `[a-zA-Z0-9_-]+`; an unsafe key rejects before any file operation, while the `single` layout keeps keys opaque. Single-record writes and `single`-layout publishes are durable once resolved; a `per-record` batch deletion depends on the filesystem persisting directory-entry updates (see Known Limitations). A batch that cannot establish or restore its state rejects with `unknown-state`, after which every later operation except `close()` rejects; calls after close reject with `closed`.
Without a legacy bootstrap, a missing `single` file or `per-record` directory opens as an empty unit and materializes on the first write. In `single`, malformed content rejects with `malformed-medium`, and a stored version outside the current and `compatibleVersions` set rejects with `version-mismatch`. An accepted older file stays unchanged on read; the next atomic write stamps the current version and preserves untouched values in declared tables. Deleting missing keys or an empty batch does not restamp. In `per-record`, each malformed or unreadable document, and each document whose version is outside the accepted set, reads as an absent record, so one bad document does not reject the unit. Version refusal does not delete the document. Per-record keys must match `[a-zA-Z0-9_-]+`; an unsafe key rejects before any file operation. Resolved writes are durable; `per-record` deletions sync their directory changes on POSIX before resolving. A batch interrupted before resolution can be partially applied. A batch that cannot restore its readable or durable state rejects with `unknown-state`, after which every later operation except `close()` rejects; calls after close reject with `closed`.
An empty `per-record` tree can initialize its declared tables from a valid `<root>/<unit>.json` whole-unit document only when the source unit name matches and its version is current or declared compatible. The backend leaves that source file unchanged and stamps migrated records with the current version. A source version outside the accepted set leaves the new tree empty. Any document path in a declared table, or a declared `global.json`, suppresses this initialization for the complete unit, even if that document is unreadable or stale.
An unmaterialized `per-record` tree can initialize its declared tables from a valid `<root>/<unit>.json` whole-unit document only when the source unit name matches and its version is current or declared compatible. This bootstrap writes during loading: it leaves the source file unchanged and copies stored values into current-stamped documents without converting them. A source version outside the accepted set leaves the new tree empty. Any declared table directory, even when empty, or a declared `global.json` suppresses this initialization for the complete unit. Deleted records therefore remain absent while the retained source file stays unchanged.
A synchronous serialization failure restores the open `single` unit's previous memory and permits retry on that handle. An I/O failure can occur after file replacement, so a rejected write does not prove that disk data or its stamp stayed unchanged. Stop using and close the backend and its domains, then reopen to reload saved state before deciding whether to retry. See [domain versioning](../../../docs/subsystems/storage.md#domain-versioning) for mixed records and downgrade behavior.
-----
@@ -69,8 +71,8 @@ The two layouts share atomic publication but assign state ownership differently.
### Design concept
- **`single` keeps memory authoritative.** Each write changes the in-memory unit, serializes its complete state, and atomically replaces `<unit>.json`. A failed publish restores the prior in-memory value. A batch delete is one such publish: the whole batch is applied or restored together.
- **`per-record` keeps the directory authoritative.** Each put or delete changes one `<unit>/<table>/<key>.json` document, and `loadAll()` rereads the tree. A batch delete renames the batch's documents out of the readable set, restores them if a rename fails, and unlinks the staged copies afterwards. If that restoration fails, the unit rejects with `unknown-state` and every later operation on it except `close()` rejects. Each document stamps the unit version and carries one record value.
- **`single` keeps memory authoritative.** Each write changes the in-memory unit, serializes its complete state, and atomically replaces `<unit>.json`. A batch delete publishes once. Serialization failures restore every affected in-memory value; uncertain I/O outcomes require reopening instead of continuing from that snapshot.
- **`per-record` keeps the directory authoritative.** Each put or delete changes one `<unit>/<table>/<key>.json` document, and `loadAll()` rereads the tree. A batch delete renames the batch's documents out of the readable set, syncs the directory on POSIX, then unlinks the staged copies. A failed staging or sync restores the documents and syncs the restored directory. If that restoration or sync fails, the unit rejects with `unknown-state` and every later operation on it except `close()` rejects. Each document stamps the unit version and carries one record value.
- **Publication is durable per call.** A write uses a temporary file, fsync, atomic `rename()` replacement, and a parent-directory fsync on POSIX. The domain layer's write chain supplies ordering across calls.
### File formats
@@ -140,7 +142,7 @@ These limits define when this backend is a poor fit or needs special operational
- **`single` rewrites the whole unit** — each write republishes the complete unit file; use `per-record` or route the domain to SQLite when this cost is too high.
- **No cross-process write locking** — two processes writing the same unit can interleave replacements; writes to the same file use last-completion wins.
- **Windows rename without explicit write-through** — durability relies on libuv's `rename()` (`MoveFileExW` with replacement); the stricter Win32 write-through publish helper from the session-log backend is planned to move down here when the `log` facet lands.
- **`per-record` batch delete is not directory-fsynced** — the batch's renames and unlinks rely on the filesystem persisting directory-entry updates; a crash can restore a deleted document under its original key or leave `.deleting` residue. `single` publishes fsync the parent directory.
- **No crash-atomic `per-record` batch** — a crash before deletion resolves can leave a partial batch and `.deleting` residue. Completed deletions sync the staged directory changes on POSIX; cleanup residue is unreadable.
<a id="dev-note"></a>
### Dev Note
+9 -7
View File
@@ -9,7 +9,7 @@ kind: "package-reference"
## 概述
`dsh-storage-json` 在配置的根目录下把领域数据存为可读 JSON,并注册为后端 `json`。默认的 `single` 布局为每个单元保存一份完整的 `<unit>.json` 文件;`per-record` 布局为每条记录保存一份带版本戳的文档。两种布局都以原子方式发布每个变更文件,领域层负责安排调用顺序。当运维方需要可检查文件且所选布局适合写入量时选择它;对于更大或高并发的数据则选择 SQLite。本后端只面向宿主侧,不贡献提示词、工具或 schema。
`dsh-storage-json` 在配置的根目录下把领域数据存为可读 JSON,并注册为后端 `json`。默认的 `single` 布局为每个单元保存一份完整的 `<unit>.json` 文件;`per-record` 布局为每条记录保存一份带版本戳的文档。两种布局都以原子方式发布每个变更文件,领域层负责安排调用顺序。当运维方需要可检查文件且所选布局适合写入量时选择它;对于更大且频繁更新的单元则选择 SQLite。本后端只面向宿主侧,不贡献提示词、工具或 schema。
## 目录
@@ -29,7 +29,7 @@ kind: "package-reference"
### 何时选择
小型单元需要一份完整、美化打印的文件时,选择默认的 `single` 布局。定点写入只应替换一份记录文档时,选择 `per-record`。当数据量大、写入频繁或多条记录需要事务更新时,选择 SQLite 后端。
小型单元需要一份完整、美化打印的文件时,选择默认的 `single` 布局。可丢弃记录的定点写入只需替换一份文档时,选择 `per-record`;不支持或不可读的文档会读作不存在。大型单元需要频繁定点更新时,选择 SQLite。领域 API 在两种后端上都没有多记录事务操作。
### 配置
@@ -53,9 +53,11 @@ kind: "package-reference"
### 可观察行为
缺失的 `single` 文件或 `per-record` 目录会作为空单元打开,并在第一次写入时物化。在 `single` 中,畸形内容以 `malformed-medium` 拒绝,不同的已存版本以 `version-mismatch` 拒绝。在 `per-record` 中,每份畸形或不可读的文档,以及版本不在描述符的当前版本和兼容版本内的文档,都读作记录不存在,因此单个坏文档不会使单元被拒绝。在 `per-record` 中记录键必须匹配 `[a-zA-Z0-9_-]+`,不安全的键在任何文件操作前被拒绝;`single` 布局保持键不透明。单记录写入与 `single` 布局的发布在 resolve 后即已持久;`per-record` 的批量删除依赖文件系统对目录项更新的持久化(见已知限制)。无法建立或恢复状态的批量删除以 `unknown-state` 拒绝,之后除 `close()` 外对它的所有操作都会拒绝;关闭后的调用以 `closed` 拒绝。
没有旧文件引导时,缺失的 `single` 文件或 `per-record` 目录会作为空单元打开,并在第一次写入时物化。在 `single` 中,畸形内容以 `malformed-medium` 拒绝,已存版本不在当前版本和 `compatibleVersions` 集合内时以 `version-mismatch` 拒绝。读取接受的旧文件时保持原状;下一次原子写入使用当前版本戳,保留已声明表中未改动记录的原值。删除不存在的键或空批次不会重新盖戳。在 `per-record` 中,每份畸形或不可读的文档,以及版本不在接受集合内的文档,都读作记录不存在,因此单个坏文档不会使单元被拒绝。版本拒绝不会删除文档。逐记录布局的键必须匹配 `[a-zA-Z0-9_-]+`;不安全的键在任何文件操作前被拒绝。写入完成后即已持久化;`per-record` 删除在完成前会在 POSIX 上同步目录变更。在完成前中断的批次可能只应用一部分。无法恢复可读或持久状态的批量删除以 `unknown-state` 拒绝,之后除 `close()` 外的所有操作都会拒绝;关闭后的调用以 `closed` 拒绝。
只有当源单元名称匹配,且源版本为当前版本或已声明的兼容版本时,空的 `per-record` 目录树才可以从有效的 `<root>/<unit>.json` 整单元文档初始化其已声明表。后端保持该源文件不变,并为迁移的记录写入当前版本戳。接受集合之外的源版本会使新目录树保持为空。已声明表中只要存在任意文档路径,或存在已声明的 `global.json`,就会对整个单元禁止该初始化,即使该文档不可读或版本陈旧。
只有当源单元名称匹配,且源版本为当前版本或已声明的兼容版本时,尚未物化的 `per-record` 目录树才可以从有效的 `<root>/<unit>.json` 整单元文档初始化其已声明表。该引导会在加载时写入:它保持源文件不变,并把已存值复制到当前版本戳的文档中,不转换其中的值。接受集合之外的源版本会使新目录树保持为空。只要存在任意已声明表的目录(即使为空),或已声明的 `global.json`,就会对整个单元禁止该初始化。因此删除的记录会保持缺失,同时保留的源文件不变。
同步序列化失败会恢复已打开 `single` 单元先前的内存状态,允许在同一句柄上重试。I/O 失败可能发生在文件替换之后,因此写入被拒绝并不能证明磁盘数据或版本戳保持不变。必须停止使用并关闭后端及其领域,然后重新打开以加载已存状态,再决定是否重试。混合记录与降级行为参阅[领域版本](../../../docs/subsystems/storage.zh.md#domain-versioning)。
-----
@@ -69,8 +71,8 @@ kind: "package-reference"
### 设计理念
- **`single` 以内存为权威状态。** 每次写入都会更改内存单元、序列化其完整状态,并以原子方式替换 `<unit>.json`。发布失败会恢复先前的内存值。批量删除就是这样一次发布:整批一起应用或一起恢复。
- **`per-record` 以目录为权威状态。** 每次 put 或 delete 都会更改一个 `<unit>/<table>/<key>.json` 文档,`loadAll()` 则重新读取目录树。每份文档都带有单元版本戳与一条记录值。批量删除先把整批文档改名移出可读集合,任一次改名失败就恢复,之后删除暂存副本;恢复本身失败时该 unit 以 `unknown-state` 拒绝,之后除 `close()` 外对它的所有操作都会拒绝。
- **`single` 以内存为权威状态。** 每次写入都会更改内存单元、序列化其完整状态,并以原子方式替换 `<unit>.json`。批量删除只发布一次。序列化失败会恢复所有受影响的内存值;I/O 结果不确定时必须重新打开,不能继续使用该快照。
- **`per-record` 以目录为权威状态。** 每次 put 或 delete 都会更改一个 `<unit>/<table>/<key>.json` 文档,`loadAll()` 则重新读取目录树。每份文档都带有单元版本戳与一条记录值。批量删除先把整批文档改名移出可读集合,在 POSIX 上同步目录,然后删除暂存副本。暂存或同步失败时会恢复文档,并同步恢复后的目录;恢复或同步失败时该 unit 以 `unknown-state` 拒绝,之后除 `close()` 外对它的所有操作都会拒绝。
- **每次调用都持久发布。** 写入过程使用临时文件、fsync、原子 `rename()` 替换,并在 POSIX 上 fsync 父目录。领域层写入链负责安排跨调用的顺序。
### 文件格式
@@ -140,7 +142,7 @@ kind: "package-reference"
- **`single` 会重写整个单元**——每次写入都重新发布完整单元文件;当此成本过高时,使用 `per-record` 或把领域路由到 SQLite。
- **没有跨进程写锁**——两个进程写入同一单元时可能交错执行替换;对同一文件的写入以最后完成者为准。
- **Windows rename 没有显式 write-through**——持久性依赖 libuv 的 `rename()`(`MoveFileExW` 并启用替换);`log` 分面落地时,计划把会话日志后端更严格的 Win32 write-through 发布辅助函数下移到此处。
- **`per-record` 批量删除不做目录 fsync**——批量删除的改名与删除依赖文件系统对目录项更新的持久化;崩溃可能让被删记录以原键重新出现,或留下 `.deleting` 残留。`single` 发布会 fsync 父目录。
- **`per-record` 批次不保证崩溃原子性**——删除完成前崩溃可能留下部分批次与 `.deleting` 残留。已完成的删除会在 POSIX 上同步暂存后的目录变更;清理残留不可读。
<a id="dev-note"></a>
### 开发备注
+6 -2
View File
@@ -39,9 +39,13 @@ export async function writeAtomic(path: string, data: string): Promise<void> {
}
}
/** fsync a POSIX directory so a just-renamed entry is crash-durable. */
/**
* Persist directory-entry changes on POSIX; skip this unsupported operation on Windows.
* @param path - Absolute directory path.
* @returns resolution after the directory is synced, or immediately on Windows.
*/
/* v8 ignore start -- Windows rejects O_RDONLY directory opens; POSIX coverage exercises this. */
async function fsyncDirectory(path: string): Promise<void> {
export async function fsyncDirectory(path: string): Promise<void> {
if (process.platform === 'win32') return
const handle = await open(path, 'r')
try {
+3 -3
View File
@@ -41,7 +41,7 @@ export function serialize(name: string, state: UnitState): string {
/**
* Parse file content into unit state, validating shape and version.
* @param text - Raw file content.
* @param descriptor - Expected identity; version mismatch rejects.
* @param descriptor - Expected identity and accepted current or compatible versions.
* @returns the parsed state.
*/
export function parse(text: string, descriptor: KvUnitDescriptor): UnitState {
@@ -63,10 +63,10 @@ export function parse(text: string, descriptor: KvUnitDescriptor): UnitState {
throw new StorageError('malformed-medium', `unit '${descriptor.name}': missing or foreign unit header`)
}
const version = (unit as Record<string, unknown>)['version'] as number
if (version !== descriptor.version) {
if (version !== descriptor.version && !descriptor.compatibleVersions?.includes(version)) {
throw new StorageError(
'version-mismatch',
`unit '${descriptor.name}': stored version ${version} != expected ${descriptor.version}`,
`unit '${descriptor.name}': stored version ${version} is not accepted by version ${descriptor.version}`,
)
}
// `typeof` alone admits an array: reading one as a table map reports every
@@ -6,9 +6,9 @@
* put or delete touches one durable document, and a batch delete renames its
* documents out of the readable set before unlinking them. The domain layer owns the live
* in-memory tables (seeded by the open-time `loadAll`) and serializes writes
* through its write chain, so this unit never mutates memory and needs no
* rollback — a failed write simply leaves both the file and the domain's
* memory unchanged.
* through its write chain. A rejected write leaves domain memory unchanged;
* a failed deletion restores its staged documents or quarantines the unit
* with `unknown-state` when restoration cannot be established.
*
* Per-record contract: a record document that is malformed or stamped with a
* version outside the accepted set (the descriptor's current version plus
@@ -17,13 +17,13 @@
* the record instead of migrating it. Record keys become path segments, so
* they must be path-safe (`[a-zA-Z0-9_-]+`); an unsafe key rejects at write.
*
* Legacy bootstrap: when the new tree has no document path, a legacy
* whole-unit file `<root>/<name>.json` (the pre-per-record layout) seeds
* Legacy bootstrap: when the new tree has no declared table directory or
* global document, a whole-unit file `<root>/<name>.json` seeds
* per-record documents, provided its stored unit version is in the accepted
* set — a legacy file stamped with any other version is left alone and reads
* as the empty unit. Any new document path, including one whose contents are
* unreadable or stale, suppresses the bootstrap for the whole unit. The
* legacy file is never changed or deleted.
* as the empty unit. An existing declared table directory, even when empty,
* or a declared global document suppresses the bootstrap for the whole unit.
* The legacy file is never changed or deleted.
* @module @deepseek-ai/dsh-storage-json/src/per-record-unit
*/
@@ -32,7 +32,7 @@ import { dirname, join } from 'node:path'
import type { Dirent } from 'node:fs'
import { StorageError } from '@deepseek-ai/dsh-storage'
import type { KvUnit, KvUnitDescriptor } from '@deepseek-ai/dsh-storage'
import { writeAtomic } from './atomic.ts'
import { fsyncDirectory, writeAtomic } from './atomic.ts'
import { parseRecord, serializeRecord } from './format.ts'
import type { UnitState } from './format.ts'
@@ -82,13 +82,14 @@ async function loadPerRecordState(descriptor: KvUnitDescriptor, dir: string): Pr
// Missing directory = empty unit; the legacy bootstrap below still runs
// (the fresh-upgrade shape is exactly an absent new tree).
}
const hasNewDocuments = entries === undefined
const hasNewState = entries === undefined
? false
: (await Promise.all(entries.map(async (entry) => {
if (entry.isDirectory()) {
const records = state.tables.get(entry.name)
if (records !== undefined) {
return loadTableRecords(records, versions, join(dir, entry.name))
await loadTableRecords(records, versions, join(dir, entry.name))
return true
}
}
if (entry.name === 'global.json' && descriptor.hasGlobal) {
@@ -98,7 +99,7 @@ async function loadPerRecordState(descriptor: KvUnitDescriptor, dir: string): Pr
}
return false
}))).some(Boolean)
if (!hasNewDocuments) await bootstrapLegacyUnit(descriptor, dir, state)
if (!hasNewState) await bootstrapLegacyUnit(descriptor, dir, state)
return state
}
@@ -108,7 +109,7 @@ function acceptedStamps(descriptor: KvUnitDescriptor): readonly number[] {
}
/**
* Bootstrap an empty per-record tree from a legacy whole-unit file
* Bootstrap an unmaterialized per-record tree from a legacy whole-unit file
* (`<root>/<name>.json`, the pre-per-record layout). Every declared-table
* record is copied into a current-version document, while the legacy file is
* retained unchanged. A missing, foreign (another unit's name), malformed,
@@ -159,12 +160,9 @@ async function bootstrapLegacyUnit(descriptor: KvUnitDescriptor, dir: string, st
/**
* Read one declared table's record documents into `records`.
* @returns whether the directory contains any `.json` document path,
* independently of key safety, readability, or stored version.
*/
async function loadTableRecords(records: Map<string, unknown>, versions: readonly number[], dir: string): Promise<boolean> {
async function loadTableRecords(records: Map<string, unknown>, versions: readonly number[], dir: string): Promise<void> {
const files = await readdir(dir, { withFileTypes: true })
const hasDocuments = files.some(file => file.name.endsWith('.json'))
const loaded = await Promise.all(files.map(async (file) => {
if (!file.name.endsWith('.json')) return
const key = file.name.slice(0, -'.json'.length)
@@ -175,7 +173,6 @@ async function loadTableRecords(records: Map<string, unknown>, versions: readonl
for (const record of loaded) {
if (record !== undefined) records.set(...record)
}
return hasDocuments
}
/** Read one record document; a foreign (unreadable or stale) one reads as absent. */
@@ -189,9 +186,9 @@ async function readRecord(path: string, versions: readonly number[]): Promise<un
/**
* One opened `per-record`-layout unit. Stateless by design: the directory is
* the medium, the domain layer owns the live memory, and each method here is
* a single durable file operation. Write ordering belongs to the caller (the
* domain layer's write chain), exactly like the `single`-layout unit.
* the medium and the domain layer owns the live memory. Replacements publish
* one document; deletions stage and sync their documents before cleanup.
* Write ordering belongs to the caller (the domain layer's write chain).
*/
export class PerRecordJsonUnit implements KvUnit {
private closed = false
@@ -225,10 +222,8 @@ export class PerRecordJsonUnit implements KvUnit {
}
/** Durably delete one record. Idempotent: a missing key is a no-op. */
async deleteRecord(table: string, key: string): Promise<void> {
this.assertOpen()
assertSafeKey(this.descriptor.name, key)
await this.tracked(rm(join(this.tableDir(table), `${key}.json`), { force: true }))
deleteRecord(table: string, key: string): Promise<void> {
return this.deleteRecords(table, [key])
}
/**
@@ -236,18 +231,18 @@ export class PerRecordJsonUnit implements KvUnit {
* readable set first; only after the whole batch is staged are the staged
* copies unlinked. A failure while staging restores the documents it moved,
* so a rejection normally leaves the batch unapplied. If that restoration
* itself fails, the readable state is unknown: the unit rejects with
* itself fails, the readable or durable state is unknown: the unit rejects with
* `unknown-state` and every later operation on it rejects, while `close`
* still releases it. A staged document a crash leaves behind is not readable
* (`open` reads `*.json` documents only).
* still releases it. Staged directory changes are synced before cleanup on
* POSIX, so a resolved deletion remains absent after a crash. Cleanup residue
* is not readable (`open` reads `*.json` documents only).
*/
async deleteRecords(table: string, keys: readonly string[]): Promise<void> {
this.assertOpen()
const dir = this.tableDir(table)
const unique = [...new Set(keys)]
for (const key of unique) assertSafeKey(this.descriptor.name, key)
// One tracked promise covers staging, restoration, and cleanup, so close()
// cannot release the unit while any part of the batch is still running.
// Track staging, restoration, and cleanup together so close() drains them.
await this.tracked(this.removeDocuments(dir, unique))
}
@@ -270,6 +265,7 @@ export class PerRecordJsonUnit implements KvUnit {
}
staged.push({ from, to })
}
if (staged.length > 0) await fsyncDirectory(dir)
} catch (error: unknown) {
const restoreFailures: unknown[] = []
for (const { from, to } of staged) {
@@ -279,11 +275,17 @@ export class PerRecordJsonUnit implements KvUnit {
restoreFailures.push(restoreError)
}
}
if (staged.length > 0) {
try {
await fsyncDirectory(dir)
} catch (restoreError: unknown) {
restoreFailures.push(restoreError)
}
}
if (restoreFailures.length === 0) throw error
this.unusable = new StorageError(
'unknown-state',
`unit '${this.descriptor.name}' could not restore ${restoreFailures.length} staged document(s)`
+ ' after a failed batch delete; its readable state is unknown',
`unit '${this.descriptor.name}' could not restore a failed batch delete; its state is unknown`,
{ cause: new AggregateError([error, ...restoreFailures], 'batch delete failed and restoration failed') },
)
throw this.unusable
@@ -347,11 +349,10 @@ export class PerRecordJsonUnit implements KvUnit {
}
/** Durably replace one document, creating its parent directory. */
private writeDocument(path: string, value: unknown): Promise<void> {
return (async () => {
await mkdir(dirname(path), { recursive: true, mode: 0o700 })
await writeAtomic(path, serializeRecord(this.descriptor.version, value))
})()
private async writeDocument(path: string, value: unknown): Promise<void> {
const text = serializeRecord(this.descriptor.version, value)
await mkdir(dirname(path), { recursive: true, mode: 0o700 })
await writeAtomic(path, text)
}
/** Track one durable write so close() drains it. */
@@ -47,6 +47,8 @@ export async function openSingleUnit(
tables: new Map(descriptor.tables.map(table => [table, new Map<string, unknown>()])),
}
: parse(text, descriptor)
// Opening leaves the file untouched; the next atomic publish uses the current stamp.
state.version = descriptor.version
return new SingleJsonUnit(descriptor, path, state, onClose)
}
@@ -87,16 +89,8 @@ class SingleJsonUnit implements KvUnit {
})
}
async deleteRecord(table: string, key: string): Promise<void> {
this.assertOpen()
const records = this.records(table)
if (!records.has(key)) return
const previous = records.get(key)
records.delete(key)
await this.publish().catch((error: unknown) => {
records.set(key, previous)
throw error
})
deleteRecord(table: string, key: string): Promise<void> {
return this.deleteRecords(table, [key])
}
async deleteRecords(table: string, keys: readonly string[]): Promise<void> {
@@ -156,7 +150,7 @@ class SingleJsonUnit implements KvUnit {
return records
}
private publish(): Promise<void> {
private async publish(): Promise<void> {
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`
@@ -1,7 +1,7 @@
import { mkdir, mkdtemp, readFile, readdir, rename, rm, writeFile } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { afterAll, describe, expect, it } from 'vitest'
import { afterAll, describe, expect, it, onTestFinished } from 'vitest'
import { Context } from '@deepseek-ai/cordis'
import Storage, { storageBackendServiceKey } from '@deepseek-ai/dsh-storage'
import { runKvBackendContract } from '../../storage/tests/contract.ts'
@@ -83,9 +83,13 @@ describe('json backend specifics', () => {
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' })
await unit.setGlobal({ g: 'committed' })
onTestFinished(() => backend.close())
const original = await backend.kv.open(descriptor)
await original.putRecord('t', 'k', { v: 'committed' })
await original.putRecord('t', 'kept', { v: 'also committed' })
await original.setGlobal({ g: 'committed' })
await original.close()
const unit = await backend.kv.open({ ...descriptor, version: 2, compatibleVersions: [1] })
const path = join(root, 'shape.json')
const backup = join(root, 'shape.committed.json')
// A directory at the publish target rejects atomic replacement on every host.
@@ -94,17 +98,19 @@ describe('json backend specifics', () => {
await expect(unit.putRecord('t', 'k', { v: 'rejected' })).rejects.toThrow()
await expect(unit.putRecord('t', 'k2', { v: 'also rejected' })).rejects.toThrow()
await expect(unit.deleteRecord('t', 'k')).rejects.toThrow()
await expect(unit.deleteRecords('t', ['k'])).rejects.toThrow()
await expect(unit.deleteRecords('t', ['k', 'kept', 'missing', 'k'])).rejects.toThrow()
await expect(unit.setGlobal({ g: 'rejected' })).rejects.toThrow()
await rm(path, { recursive: true })
await rename(backup, path)
const snapshot = await unit.loadAll()
expect(snapshot.tables['t']).toEqual({ k: { v: 'committed' } })
expect(snapshot.tables['t']).toEqual({ k: { v: 'committed' }, kept: { v: 'also committed' } })
expect(snapshot.global).toEqual({ g: 'committed' })
expect(JSON.parse(await readFile(path, 'utf8'))).toMatchObject({ unit: { version: 1 } })
// The next successful publish must not carry rejected writes to disk.
await unit.putRecord('t', 'k3', { v: 'later' })
const text = await readFile(path, 'utf8')
expect(text).not.toContain('rejected')
expect(JSON.parse(text)).toMatchObject({ unit: { version: 2 } })
await backend.close()
})
@@ -470,16 +476,33 @@ describe('per-record layout', () => {
it('reads per-record documents stamped with a declared compat version and stamps writes current', async () => {
const root = await freshRoot()
const backend = new JsonStorageBackend(root)
onTestFinished(() => backend.close())
const compat = { ...descriptor, compatibleVersions: [1] }
await mkdir(join(root, 'recs', 't'), { recursive: true })
await writeFile(recordPath(root, 'oldrec'), JSON.stringify({ version: 1, record: { v: 'old' } }), 'utf8')
const original = JSON.stringify({ version: 1, record: { v: 'old' } })
await writeFile(recordPath(root, 'oldrec'), original, 'utf8')
const untouched = JSON.stringify({ version: 1, record: { v: 'keep' } })
await writeFile(recordPath(root, 'untouched'), untouched, 'utf8')
await writeFile(recordPath(root, 'ancient'), JSON.stringify({ version: 0, record: { v: 'no' } }), 'utf8')
const globalPath = join(root, 'recs', 'global.json')
const global = JSON.stringify({ version: 1, record: 'old global' })
await writeFile(globalPath, global, 'utf8')
const unit = await backend.kv.open(compat)
// Version 1 is declared compat and served; version 0 is not and discards.
expect(await unit.loadAll()).toEqual({ tables: { t: { oldrec: { v: 'old' } } }, global: null })
expect(await unit.loadAll()).toEqual({ tables: { t: { oldrec: { v: 'old' }, untouched: { v: 'keep' } } }, global: 'old global' })
await unit.deleteRecord('t', 'missing')
await expect(unit.putRecord('t', 'oldrec', { toJSON: () => { throw new Error('serialization failed') } }))
.rejects.toThrow('serialization failed')
expect(await readFile(recordPath(root, 'oldrec'), 'utf8')).toBe(original)
await unit.deleteRecords('t', ['oldrec', 'missing', 'oldrec'])
await expect(readFile(recordPath(root, 'oldrec'), 'utf8')).rejects.toMatchObject({ code: 'ENOENT' })
await unit.putRecord('t', 'oldrec', { v: 'new' })
expect(JSON.parse(await readFile(recordPath(root, 'oldrec'), 'utf8')))
.toEqual({ version: 2, record: { v: 'new' } })
expect(await readFile(globalPath, 'utf8')).toBe(global)
await unit.setGlobal('new global')
expect(JSON.parse(await readFile(globalPath, 'utf8'))).toEqual({ version: 2, record: 'new global' })
expect(await readFile(recordPath(root, 'untouched'), 'utf8')).toBe(untouched)
await backend.close()
})
@@ -1,12 +1,13 @@
import { mkdtemp, rm } from 'node:fs/promises'
import { mkdtemp, readFile, readdir, rm, writeFile } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { basename, join } from 'node:path'
import { afterEach, expect, it, vi } from 'vitest'
import { afterEach, expect, it, onTestFinished, vi } from 'vitest'
import { JsonStorageBackend } from '../src/index.ts'
/** Rename hook installed by a case; returning true means the case handled the call. */
const gates = vi.hoisted(() => ({
renameOverride: null as ((from: string, to: string) => Promise<boolean>) | null,
syncOverride: null as ((path: string) => Promise<boolean>) | null,
}))
vi.mock('node:fs/promises', async (importOriginal) => {
@@ -24,6 +25,18 @@ vi.mock('node:fs/promises', async (importOriginal) => {
}
})
vi.mock('../src/atomic.ts', async (importOriginal) => {
const actual = await importOriginal<typeof import('../src/atomic.ts')>()
return {
...actual,
fsyncDirectory: async (path: string): Promise<void> => {
const override = gates.syncOverride
if (override !== null && await override(path)) return
await actual.fsyncDirectory(path)
},
}
})
const DESCRIPTOR = { name: 'batch', version: 1, layout: 'per-record' as const, tables: ['t'], hasGlobal: true }
const roots: string[] = []
@@ -35,9 +48,118 @@ async function freshRoot(): Promise<string> {
afterEach(async () => {
gates.renameOverride = null
gates.syncOverride = null
for (const root of roots.splice(0)) await rm(root, { recursive: true, force: true })
})
it.each(['deleteRecord', 'deleteRecords'] as const)('restores documents after a directory-sync failure in %s, then retries', async (operation) => {
const root = await freshRoot()
const backend = new JsonStorageBackend(root)
onTestFinished(() => backend.close())
const unit = await backend.kv.open(DESCRIPTOR)
await unit.putRecord('t', 'a', { v: 1 })
await unit.putRecord('t', 'b', { v: 2 })
const path = join(root, 'batch', 't', 'a.json')
const before = await readFile(path, 'utf8')
const failure = new Error('directory sync failed')
let syncs = 0
gates.syncOverride = async () => {
if (syncs++ === 0) throw failure
return false
}
const remove = () => operation === 'deleteRecord'
? unit.deleteRecord('t', 'a')
: unit.deleteRecords('t', ['a', 'b'])
await expect(remove()).rejects.toBe(failure)
expect(syncs).toBe(2)
expect(await readFile(path, 'utf8')).toBe(before)
expect(await unit.loadAll()).toEqual({ tables: { t: { a: { v: 1 }, b: { v: 2 } } }, global: null })
await remove()
expect(await unit.loadAll()).toEqual({ tables: { t: operation === 'deleteRecord' ? { b: { v: 2 } } : {} }, global: null })
})
it('quarantines a deletion whose restored directory cannot be synced', async () => {
const root = await freshRoot()
const backend = new JsonStorageBackend(root)
onTestFinished(() => backend.close())
const unit = await backend.kv.open(DESCRIPTOR)
await unit.putRecord('t', 'a', { v: 1 })
const failure = new Error('directory sync denied')
gates.syncOverride = () => Promise.reject(failure)
await expect(unit.deleteRecords('t', ['a'])).rejects.toMatchObject({
code: 'unknown-state',
cause: { errors: [failure, failure] },
})
await expect(unit.loadAll()).rejects.toMatchObject({ code: 'unknown-state' })
await expect(unit.deleteRecords('t', [])).rejects.toMatchObject({ code: 'unknown-state' })
await unit.close()
await expect(unit.loadAll()).rejects.toMatchObject({ code: 'closed' })
})
it('does not sync or create directories for empty and missing-key deletions', async () => {
const root = await freshRoot()
const backend = new JsonStorageBackend(root)
onTestFinished(() => backend.close())
const unit = await backend.kv.open(DESCRIPTOR)
gates.syncOverride = () => Promise.reject(new Error('unexpected directory sync'))
await unit.deleteRecords('t', [])
await unit.deleteRecords('t', ['missing', 'missing'])
await unit.deleteRecord('t', 'missing')
expect(await readdir(root)).toEqual([])
})
it('leaves a failed first staging operation unapplied without syncing', async () => {
const root = await freshRoot()
const backend = new JsonStorageBackend(root)
onTestFinished(() => backend.close())
const unit = await backend.kv.open(DESCRIPTOR)
await unit.putRecord('t', 'a', { v: 1 })
const failure = new Error('staging denied')
gates.renameOverride = () => Promise.reject(failure)
gates.syncOverride = () => Promise.reject(new Error('unexpected directory sync'))
await expect(unit.deleteRecords('t', ['a'])).rejects.toBe(failure)
expect(await unit.loadAll()).toEqual({ tables: { t: { a: { v: 1 } } }, global: null })
})
it('leaves an unmaterialized tree eligible for bootstrap after serialization fails', async () => {
const root = await freshRoot()
const backend = new JsonStorageBackend(root)
onTestFinished(() => backend.close())
const unit = await backend.kv.open(DESCRIPTOR)
await expect(unit.putRecord('t', 'broken', { toJSON: () => { throw new Error('serialization failed') } }))
.rejects.toThrow('serialization failed')
expect(await readdir(root)).toEqual([])
await writeFile(join(root, 'batch.json'), JSON.stringify({
unit: { name: 'batch', version: 1 }, global: null, tables: { t: { saved: { v: 1 } } },
}))
expect((await unit.loadAll()).tables['t']).toEqual({ saved: { v: 1 } })
})
it.each(['deleteRecord', 'deleteRecords'] as const)('keeps an emptied tree authoritative after %s, including reopen', async (operation) => {
const root = await freshRoot()
const legacyPath = join(root, 'batch.json')
const legacy = JSON.stringify({ unit: { name: 'batch', version: 0 }, global: null, tables: { t: { a: { v: 1 }, b: { v: 2 } } } })
await writeFile(legacyPath, legacy)
const descriptor = { ...DESCRIPTOR, compatibleVersions: [0] }
const backend = new JsonStorageBackend(root)
onTestFinished(() => backend.close())
const unit = await backend.kv.open(descriptor)
expect((await unit.loadAll()).tables['t']).toEqual({ a: { v: 1 }, b: { v: 2 } })
if (operation === 'deleteRecord') {
await unit.deleteRecord('t', 'a')
await unit.deleteRecord('t', 'b')
} else {
await unit.deleteRecords('t', ['a', 'b'])
}
expect(await unit.loadAll()).toEqual({ tables: { t: {} }, global: null })
await backend.close()
const reopened = new JsonStorageBackend(root)
onTestFinished(() => reopened.close())
expect(await (await reopened.kv.open(descriptor)).loadAll()).toEqual({ tables: { t: {} }, global: null })
expect(await readFile(legacyPath, 'utf8')).toBe(legacy)
})
/** A filesystem denial the batch must treat as an inability to restore. */
function denied(): NodeJS.ErrnoException {
return Object.assign(new Error('denied'), { code: 'EACCES' })
@@ -9,8 +9,8 @@
en: ec2b5805a2c77212
zh: 319135f3461ed09b
/deepseek-ai-dsh-storage-sqlite/summary:
en: 9cdc451cf6c54bb8
zh: d3c86bf676052688
en: b09aef281e7fa55d
zh: 22999e12a66ff22c
/deepseek-ai-dsh-storage-sqlite/table-of-contents:
en: d152484eb41ac6b4
zh: 09388d293f9be9cb
@@ -18,20 +18,20 @@
en: 98de508e10f22c1e
zh: 1e7d0cadc10cba8d
/deepseek-ai-dsh-storage-sqlite/use-this-package/when-to-choose-it:
en: 8e1ba279af55e197
zh: 7dfad7470534de6c
en: f5b6290fe1e6929b
zh: dead924928ace851
/deepseek-ai-dsh-storage-sqlite/use-this-package/configuration:
en: e21da50c3c40bcb8
zh: 0ed58b20e4a32ce9
/deepseek-ai-dsh-storage-sqlite/use-this-package/observable-behavior:
en: eaf8cbb16e426ebe
zh: 7e089537588adb05
en: b0df4ed07bf7ce0a
zh: 18da8edf449cfd92
/deepseek-ai-dsh-storage-sqlite/understand-the-implementation:
en: 6867b52cbb94e572
zh: 187d5d61980c8fe7
en: 8f003bbcea33bf50
zh: c79bf8f71d345070
/deepseek-ai-dsh-storage-sqlite/understand-the-implementation/design-concept:
en: 895f7ff7b86a72f2
zh: 8db4c7702ee4145e
en: 46b7d90df396de82
zh: b59445936f921779
/deepseek-ai-dsh-storage-sqlite/understand-the-implementation/open-sequence:
en: 50d85e8d8e8b5bc1
zh: 2a7067bd7fb47b35
@@ -57,8 +57,8 @@
en: c13c59d13a5f1651
zh: 5aab838ae3ff5324
/deepseek-ai-dsh-storage-sqlite/known-limitations-and-deferred-work:
en: c24fe7a01bc44ed3
zh: 69638d0692bdb637
en: 528ca77480d5a868
zh: 36fa03755b1a14cf
/deepseek-ai-dsh-storage-sqlite/known-limitations-and-deferred-work/dev-note:
en: e67f633dc6fe7773
zh: 3bb0c20c93f16f98
+11 -8
View File
@@ -9,7 +9,7 @@ English | [中文](README.zh.md)
## Summary
`dsh-storage-sqlite` is a storage backend that hosts every routed unit in one SQLite database file, storing each record as one JSON document per row, registered as backend `sqlite`. A single record update touches exactly one row, which is what makes this the right medium for high-frequency, point-sized writes. Choose it when a domain's data changes often or the deployment prefers one queryable database; choose the JSON backend when the data should be readable as plain files. The backend is host-side only: it contributes no prompt, tool, or schema, so the model and the agent loop never see it.
`dsh-storage-sqlite` is a storage backend that hosts every routed unit in one SQLite database file, storing each record as one JSON document per row, registered as backend `sqlite`. A single record update replaces one data row, which is what makes this the right medium for high-frequency, point-sized writes. Choose it when a domain's data changes often or the deployment prefers one queryable database; choose the JSON backend when the data should be readable as plain files. The backend is host-side only: it contributes no prompt, tool, or schema, so the model and the agent loop never see it.
## Table of Contents
@@ -29,7 +29,7 @@ Use this package when a composition keeps frequently updated domain data in one
### When to choose it
Choose it when writes are frequent and point-sized — each key maps to exactly one row, so updating one record touches one row instead of rewriting a whole file. Choose the JSON backend when humans inspect or edit the stored data as plain files. The synchronous `node:sqlite` driver blocks the JavaScript thread for the duration of each single-statement call, which is fine at domain-data scale but worth accounting for at high write rates.
Choose it when writes are frequent and point-sized — each key maps to one data row, so updating a record avoids rewriting the whole unit. The first mutation of an accepted older unit also updates its version metadata in the same transaction. Choose JSON when humans inspect or edit stored data as plain files. The synchronous `node:sqlite` driver blocks the JavaScript thread for each write call, including that initial transaction; account for this at high write rates.
### Configuration
@@ -54,7 +54,9 @@ Two fields: the database path and the journal mode. `:memory:` opens an in-proce
### Observable behavior
Missing directories and database files are created owner-only (`0o700`/`0o600`); an existing database keeps its modes. A unit whose stored format version differs from its descriptor rejects `version-mismatch`, and a database stamped with a physical layout version other than the current one rejects outright — no migration, pre-release stance. Failures carry stable `StorageError` codes (`version-mismatch`, `malformed-medium`, `unknown-state`, `closed`); single-record writes and committed batches are durable once resolved. A batch that cannot open or roll back its transaction rejects with `unknown-state`, after which every later operation except `close()` rejects, and calls after close reject `closed`.
Missing directories and database files are created owner-only (`0o700`/`0o600`); an existing database keeps its modes. A unit accepts its descriptor's current version or `compatibleVersions`, otherwise rejects `version-mismatch`. Accepted older units retain their stamp on read; the first mutation commits data and the current stamp in one transaction. Deleting missing keys or an empty batch leaves the stamp unchanged. Untouched record values and other units remain unchanged; the backend does not convert application values. A database stamped with a physical layout version other than the current one rejects outright; domain compatibility does not migrate the database layout. Storage checks use `StorageError` codes; underlying SQLite failures propagate, and writes are durable once resolved.
Every nonempty batch delete uses one transaction, including its version update when needed. A failed transaction preserves data and stamp when SQLite rolls back successfully, including an automatic rollback. A failed `BEGIN` leaves the batch unapplied and permits retry on the same handle. If rollback itself fails, the backend retains a `StorageError` with code `unknown-state` whose cause is an `AggregateError` containing both failures, and rejects further unit reads, writes, and pending or new opens. The transaction outcome is unknown. Close the backend and its domains, then create a new backend to inspect saved state before deciding whether to retry. See [domain versioning](../../../docs/subsystems/storage.md#domain-versioning) for mixed records and downgrade behavior.
-----
@@ -64,14 +66,14 @@ Missing directories and database files are created owner-only (`0o700`/`0o600`);
<details>
<summary>Implementation internals — click to expand</summary>
The backend is a document-per-row layout over one `node:sqlite` connection, designed so a per-key update is a single prepared statement.
The backend is a document-per-row layout over one `node:sqlite` connection. Per-key writes reuse prepared statements; a version upgrade adds the metadata update and transaction.
### Design concept
- **Document per row.** Each unit table becomes a physical STRICT table `u_<unit>_<table> (key TEXT PRIMARY KEY, value TEXT)` whose `value` column holds the record's JSON text; the global singleton lives in a shared `unit_globals` table. One key update touches exactly one row — the reason to route a high-churn domain here.
- **Statement atomicity.** Every single-record write is one prepared statement; a batch delete runs its deletes in one explicit transaction, so the batch commits or rolls back together. A batch that cannot open or roll back its transaction rejects with `unknown-state`, after which every later operation on the unit except `close()` rejects; write ordering stays the caller's responsibility (the domain layer's write chain).
- **Document per row.** Each unit table becomes a physical STRICT table `u_<unit>_<table> (key TEXT PRIMARY KEY, value TEXT)` whose `value` column holds the record's JSON text; the global singleton lives in a shared `unit_globals` table. A key update replaces one data row; advancing the unit stamp also updates its metadata.
- **Atomic writes.** Current-version single-record writes use one prepared statement. Batch deletes and the first mutation of an accepted older unit share one transaction helper; the first mutation includes its current version stamp. Successful rollback preserves both data and stamp. Write ordering stays the caller's responsibility (the domain layer's write chain).
- **Names validated before DDL.** Unit and table names must match `UNIT_NAME_RE` before they reach DDL, so no external input is ever interpolated into SQL identifiers.
- **Versions fail loud.** The physical layout version lives in `PRAGMA user_version` (fresh databases stamp it last); unit format versions live in the `units` table. Any other stamped value rejects — no migrations.
- **Separate version checks.** The physical layout version lives in `PRAGMA user_version` (fresh databases stamp it last) and must match exactly. Unit versions live in `units` and also accept the owner's declared `compatibleVersions`; record conversion belongs to the owner.
### Open sequence
@@ -125,9 +127,10 @@ None — the backend never touches live request prefixes.
These limits define when this backend is a poor fit or needs special operational care. They are current package constraints, not a task backlog.
- **Synchronous driver blocks the event loop** — each write is a synchronous `DatabaseSync` call; the block lasts a single statement, which is acceptable at domain-data scale.
- **Synchronous driver blocks the event loop** — writes use synchronous `DatabaseSync` calls; batch deletes and an accepted older unit's first mutation hold their transaction on the JavaScript thread.
- **No busy-wait or retry policy** — a competing connection holding a write lock rejects the operation immediately instead of waiting; the domain layer's write chain serializes writes within one process, and cross-process coordination is out of scope.
- **Only the current physical layout version opens** — any other stamped `user_version` is rejected rather than migrated (pre-release stance).
- **Version checks remain unit-wide** — a `per-record` layout request does not add per-record stamps or JSON's absent-document recovery policy to this backend.
- **Open sequence duplicated with the query provider** — `openDatabase` and `session-query-sqlite` both enforce SQLite file ownership, but each package owns a distinct application identity and schema; no shared medium helper couples them.
<a id="dev-note"></a>
+11 -8
View File
@@ -9,7 +9,7 @@ kind: "package-reference"
## 概述
`dsh-storage-sqlite` 是一个存储后端:把每个已路由单元托管在同一个 SQLite 数据库文件中,每条记录按行存储一份 JSON 文档,注册为后端 `sqlite`。单条记录更新恰好触碰一行,这正是它适合高频定点写入的原因。当领域数据变动频繁、或部署偏好单一可查询数据库时选择它;当数据需要以纯文本文件形式可读时选择 JSON 后端。本后端只面向宿主侧:它不贡献提示词、工具或 schema,因此模型与 agent loop(智能体循环)永远不会看到它。
`dsh-storage-sqlite` 是一个存储后端:把每个已路由单元托管在同一个 SQLite 数据库文件中,每条记录按行存储一份 JSON 文档,注册为后端 `sqlite`。单条记录更新替换一条数据行,这正是它适合高频定点写入的原因。当领域数据变动频繁、或部署偏好单一可查询数据库时选择它;当数据需要以纯文本文件形式可读时选择 JSON 后端。本后端只面向宿主侧:它不贡献提示词、工具或 schema,因此模型与 agent loop(智能体循环)永远不会看到它。
## 目录
@@ -29,7 +29,7 @@ kind: "package-reference"
### 何时选择
当写入频繁且为定点更新时选择它——每个键恰好映射到一行,因此更新一条记录只触碰一行,而不是重写整个文件。当人类需要以纯文本文件查看或编辑已存数据时选择 JSON 后端。同步的 `node:sqlite` 驱动会在每条单语句调用期间阻塞 JavaScript 线程,这在领域数据规模下可以接受,但高写入率时值得纳入考量。
当写入频繁且为定点更新时选择它——每个键映射到一条数据行,因此更新记录无需重写整个单元。对接受的旧单元首次修改时,还会在同一事务中更新其版本元数据。当人类需要以纯文本文件查看或编辑已存数据时选择 JSON。同步的 `node:sqlite` 驱动会在每次写入调用期间阻塞 JavaScript 线程,包括该首次事务;高写入率时须纳入考量。
### 配置
@@ -54,7 +54,9 @@ kind: "package-reference"
### 可观察行为
缺失的目录与数据库文件会以仅所有者可访问的权限创建(`0o700`/`0o600`);已有数据库保持其既有模式。已存格式版本与描述符不同的单元拒绝 `version-mismatch`,盖有非当前物理布局版本的数据库会直接拒绝——不做迁移,预发布立场。失败携带稳定的 `StorageError` 代码(`version-mismatch`、`malformed-medium`、`unknown-state`、`closed`);单记录写入与已提交批次一旦 resolve 即已持久。无法开启或回滚事务的批量删除以 `unknown-state` 拒绝,之后除 `close()` 外对该 unit 的所有操作都会拒绝;关闭后的调用以 `closed` 拒绝。
缺失的目录与数据库文件会以仅所有者可访问的权限创建(`0o700`/`0o600`);已有数据库保持其既有模式。单元接受描述符的当前版本或 `compatibleVersions`,否则拒绝 `version-mismatch`。读取接受的旧单元时保留其版本戳;首次修改将数据与当前版本戳放在同一事务中提交。删除不存在的键或空批次时版本戳保持不变。未改动记录的值与其他单元保持不变;后端不转换应用值。盖有非当前物理布局版本的数据库会直接拒绝;领域兼容性不迁移数据库布局。Storage 检查使用 `StorageError` 代码;底层 SQLite 失败会向上传递,写入操作完成后即已持久化。
每个非空批量删除都使用一个事务,需要时在该事务内更新版本戳。事务失败时,只要 SQLite 成功回滚,数据与版本戳就保持原状,包括自动回滚的情况。`BEGIN` 失败会使整个批次保持未应用,允许在同一句柄上重试。如果回滚本身失败,后端会保留代码为 `unknown-state` 的 `StorageError`,其 cause 是包含两次失败的 `AggregateError`,并拒绝后续单元读写,以及正在等待或新发起的打开操作。此时事务结果未知。须关闭后端及其领域,再创建新后端检查已存状态,然后决定是否重试。混合记录与降级行为参阅[领域版本](../../../docs/subsystems/storage.zh.md#domain-versioning)。
-----
@@ -64,14 +66,14 @@ kind: "package-reference"
<details>
<summary>实现细节——点击展开</summary>
本后端是在单个 `node:sqlite` 连接之上的文档按行布局,设计目标是让每次按键更新都是一条预处理语句。
本后端是在单个 `node:sqlite` 连接之上的文档按行布局。按键写入复用预处理语句;版本升级会增加元数据更新与事务。
### 设计理念
- **每行一份文档。** 每个单元表都变成一张物理 STRICT 表 `u_<unit>_<table> (key TEXT PRIMARY KEY, value TEXT)`,其 `value` 列保存记录的 JSON 文本;全局单例存放在共享的 `unit_globals` 表中。一个键的更新恰好触碰一行——这就是把高频变更领域路由到这里的原因。
- **语句原子性。** 每个单记录写入都是一条预处理语句;批量删除在一个显式事务中执行其删除,因此整批一起提交或回滚。无法开启或回滚事务的批量删除以 `unknown-state` 拒绝,之后除 `close()` 外对该 unit 的所有操作都会拒绝;写入顺序仍由调用方负责(领域层的写入链)。
- **每行一份文档。** 每个单元表都变成一张物理 STRICT 表 `u_<unit>_<table> (key TEXT PRIMARY KEY, value TEXT)`,其 `value` 列保存记录的 JSON 文本;全局单例存放在共享的 `unit_globals` 表中。一个键的更新会替换一条数据行;推进单元版本戳时还会更新其元数据。
- **原子写入。** 当前版本的单记录写入使用一条预处理语句。批量删除和对接受的旧单元首次修改共用一个事务助手;首次修改包含当前版本戳。成功回滚会保留数据与版本戳。写入顺序仍由调用方负责(领域层的写入链)。
- **名称在 DDL 之前校验。** 单元名与表名在进入 DDL 之前必须匹配 `UNIT_NAME_RE`,因此任何外部输入都不会被插值进 SQL 标识符。
- **版本明确报错。** 物理布局版本存放在 `PRAGMA user_version`(全新数据库最后盖戳);单元格式版本存放在 `units` 表中。任何其他已标记值都会被拒绝——不做迁移。
- **版本分别检查。** 物理布局版本存放在 `PRAGMA user_version`(全新数据库最后盖戳),必须精确匹配。单元版本存放在 `units` 中,也接受拥有方声明的 `compatibleVersions`;记录转换由拥有方负责。
### 打开顺序
@@ -125,9 +127,10 @@ kind: "package-reference"
这些限制说明本后端何时不合适,或何时需要特别的运维注意。它们是当前包约束,不是任务积压。
- **同步驱动阻塞事件循环**——每次写入都是一次同步 `DatabaseSync` 调用;阻塞只持续一条语句,在领域数据规模下可以接受。
- **同步驱动阻塞事件循环**——写入使用同步 `DatabaseSync` 调用;批量删除与对接受的旧单元首次修改会在 JavaScript 线程上持有事务。
- **没有忙等待或重试策略**——持有写锁的竞争连接会立即拒绝操作,而不是等待;领域层的写入链在单进程内串行化写入,跨进程协调不在范围内。
- **只打开当前的物理布局版本**——任何其他已标记的 `user_version` 都会被拒绝而不是迁移(预发布立场)。
- **版本检查保持单元级**——`per-record` 布局请求不会让本后端增加逐记录版本戳或 JSON 的缺失文档恢复策略。
- **打开顺序与查询提供方重复**——`openDatabase` 与 `session-query-sqlite` 都强制执行 SQLite 文件所有权约束,但两个包各自拥有不同的应用标识与 schema;没有共享介质辅助模块将二者耦合。
<a id="dev-note"></a>
+19 -5
View File
@@ -50,7 +50,9 @@ export const Config: z<Config> = z.object({
/**
* The SQLite {@link StorageBackend}. Owns one `DatabaseSync` connection and
* the open-unit table; `kv.open` validates names, enforces the per-unit
* version stamp in `units`, and ensures the unit's record tables.
* version stamp in `units`, and ensures the unit's record tables. A failed
* rollback makes all reads, writes, and opens reject the retained failure;
* close the backend and recreate it before further use.
*/
export class SqliteStorageBackend implements StorageBackend {
/** The key-value facet; the only shape this backend serves. */
@@ -60,6 +62,7 @@ export class SqliteStorageBackend implements StorageBackend {
/** Open (or still-opening) units by name; presence is the double-open guard. */
private readonly units = new Map<string, Promise<SqliteKvUnit>>()
private closing: Promise<void> | undefined
private failure: StorageError | undefined
/**
* @param config - Validated plugin configuration.
@@ -76,6 +79,7 @@ export class SqliteStorageBackend implements StorageBackend {
if (this.closing !== undefined) {
return Promise.reject(new StorageError('closed', 'sqlite storage backend is closed'))
}
if (this.failure !== undefined) return Promise.reject(this.failure)
if (!UNIT_NAME_RE.test(descriptor.name)) {
return Promise.reject(new Error(`kv unit name '${descriptor.name}' violates ${UNIT_NAME_RE}`))
}
@@ -97,12 +101,13 @@ export class SqliteStorageBackend implements StorageBackend {
private async materializeUnit(descriptor: KvUnitDescriptor): Promise<SqliteKvUnit> {
const db = await this.ready
this.ensureUsable()
const row = db.prepare('SELECT version FROM units WHERE name = ?').get(descriptor.name) as
| { version: number }
| undefined
if (row === undefined) {
db.prepare('INSERT INTO units (name, version) VALUES (?, ?)').run(descriptor.name, descriptor.version)
} else if (row.version !== descriptor.version) {
} else if (row.version !== descriptor.version && !descriptor.compatibleVersions?.includes(row.version)) {
throw new StorageError(
'version-mismatch',
`kv unit '${descriptor.name}' is stamped version ${row.version} on the medium, incompatible with descriptor version ${descriptor.version}`,
@@ -117,9 +122,18 @@ export class SqliteStorageBackend implements StorageBackend {
) STRICT
`)
}
return new SqliteKvUnit(db, descriptor, () => {
this.units.delete(descriptor.name)
})
return new SqliteKvUnit(
db,
descriptor,
row?.version ?? descriptor.version,
() => { this.units.delete(descriptor.name) },
() => { this.ensureUsable() },
(error) => { this.failure = error },
)
}
private ensureUsable(): void {
if (this.failure !== undefined) throw this.failure
}
/**
+59 -38
View File
@@ -1,10 +1,10 @@
/**
* One opened SQLite KV unit: prepared per-table statements over the
* `u_<unit>_<table>` record tables plus this unit's row in the shared
* `unit_globals` table. Each single-record primitive is one statement, so
* SQLite's per-statement atomicity applies; a batch delete runs its deletes in
* one explicit transaction. There is no write queue — write ordering is the
* caller's responsibility per the KV contract.
* `unit_globals` table. The first mutation of an accepted older version commits
* its data and current version stamp in one transaction. Batch deletes also
* use one transaction; other writes remain single statements. Write ordering
* belongs to the caller per the KV contract.
* @module @deepseek-ai/dsh-storage-sqlite/unit
*/
@@ -29,20 +29,26 @@ export class SqliteKvUnit implements KvUnit {
private readonly tables = new Map<string, TableStatements>()
private readonly globalUpsert: StatementSync | undefined
private readonly globalSelect: StatementSync | undefined
private needsVersionStamp: boolean
private closed = false
/** Set when a batch transaction could not be opened or rolled back. */
private unusable: StorageError | undefined
/**
* @param db - Open database handle owned by the backend (never closed here).
* @param descriptor - Validated descriptor whose record tables already exist.
* @param storedVersion - Accepted unit version currently saved on the medium.
* @param onClose - Backend callback releasing this unit's open-name slot.
* @param ensureBackendUsable - Rejects use of a connection whose rollback failed.
* @param onRollbackFailure - Retains the failure on the backend before this call rejects.
*/
constructor(
private readonly db: DatabaseSync,
private readonly descriptor: KvUnitDescriptor,
storedVersion: number,
private readonly onClose: () => void,
private readonly ensureBackendUsable: () => void,
private readonly onRollbackFailure: (error: StorageError) => void,
) {
this.needsVersionStamp = storedVersion !== descriptor.version
for (const table of descriptor.tables) {
// Both name segments are validated against UNIT_NAME_RE by the backend,
// so the physical identifier is safe to interpolate into statement text.
@@ -101,47 +107,28 @@ export class SqliteKvUnit implements KvUnit {
putRecord(table: string, key: string, value: unknown): Promise<void> {
return this.settle(() => {
this.statementsFor(table).upsert.run(key, JSON.stringify(value))
const statement = this.statementsFor(table).upsert
const text = JSON.stringify(value)
this.write(() => statement.run(key, text).changes)
})
}
deleteRecord(table: string, key: string): Promise<void> {
return this.settle(() => {
this.statementsFor(table).remove.run(key)
const statement = this.statementsFor(table).remove
this.write(() => statement.run(key).changes)
})
}
deleteRecords(table: string, keys: readonly string[]): Promise<void> {
return this.settle(() => {
const remove = this.statementsFor(table).remove
try {
this.db.exec('BEGIN IMMEDIATE')
} catch (error: unknown) {
// An unopened transaction can only mean the connection is not in the
// state this unit expects; later statements would run in an unknown one.
this.unusable = new StorageError(
'unknown-state',
`kv unit '${this.descriptor.name}' could not open a batch transaction; its readable state is unknown`,
{ cause: error },
)
throw this.unusable
}
try {
for (const key of new Set(keys)) remove.run(key)
this.db.exec('COMMIT')
} catch (error: unknown) {
try {
this.db.exec('ROLLBACK')
} catch (rollbackError: unknown) {
this.unusable = new StorageError(
'unknown-state',
`kv unit '${this.descriptor.name}' batch delete failed and could not roll back; its readable state is unknown`,
{ cause: new AggregateError([error, rollbackError], 'batch delete failed and rollback failed') },
)
throw this.unusable
}
throw error
}
if (keys.length === 0) return
this.write(() => {
let changes = 0
for (const key of new Set(keys)) changes += Number(remove.run(key).changes)
return changes
}, true)
})
}
@@ -150,10 +137,44 @@ export class SqliteKvUnit implements KvUnit {
if (this.globalUpsert === undefined) {
throw new Error(`kv unit '${this.descriptor.name}' declared no global slot`)
}
this.globalUpsert.run(this.descriptor.name, JSON.stringify(value))
const statement = this.globalUpsert
const text = JSON.stringify(value)
this.write(() => statement.run(this.descriptor.name, text).changes)
})
}
/** Commit batches and version updates atomically; missing-key deletes leave the stamp unchanged. */
private write(operation: () => number | bigint, transactional = false): void {
if (!this.needsVersionStamp && !transactional) {
operation()
return
}
this.db.exec('BEGIN IMMEDIATE')
try {
const changed = operation() > 0
if (this.needsVersionStamp && changed) {
this.db.prepare('UPDATE units SET version = ? WHERE name = ?').run(this.descriptor.version, this.descriptor.name)
}
this.db.exec('COMMIT')
if (changed) this.needsVersionStamp = false
} catch (error) {
if (this.db.isTransaction) {
try {
this.db.exec('ROLLBACK')
} catch (rollbackError) {
const failure = new StorageError(
'unknown-state',
`kv unit '${this.descriptor.name}' write and rollback failed; the backend is unusable and must be closed and reopened`,
{ cause: new AggregateError([error, rollbackError], 'write and rollback failed', { cause: error }) },
)
this.onRollbackFailure(failure)
throw failure
}
}
throw error
}
}
close(): Promise<void> {
if (!this.closed) {
this.closed = true
@@ -181,7 +202,7 @@ export class SqliteKvUnit implements KvUnit {
if (this.closed) {
throw new StorageError('closed', `kv unit '${this.descriptor.name}' is closed`)
}
if (this.unusable !== undefined) throw this.unusable
this.ensureBackendUsable()
}
private statementsFor(table: string): TableStatements {
@@ -5,18 +5,24 @@ import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { DatabaseSync } from 'node:sqlite'
import Storage, { storageBackendServiceKey } from '@deepseek-ai/dsh-storage'
import type { KvUnitDescriptor } from '@deepseek-ai/dsh-storage'
import type { KvUnit, KvUnitDescriptor } from '@deepseek-ai/dsh-storage'
import { runKvBackendContract } from '../../storage/tests/contract.ts'
import * as StorageSqlite from '../src/index.ts'
import { Config, SqliteStorageBackend, STORAGE_SQLITE_SCHEMA_VERSION } from '../src/index.ts'
/** Mirror the loader: resolve schemastery defaults before construction. */
function backendAt(path: string): SqliteStorageBackend {
return new SqliteStorageBackend(new Config({ path }))
const backend = new SqliteStorageBackend(new Config({ path }))
backends.push(backend)
return backend
}
const backends: SqliteStorageBackend[] = []
const dirs: string[] = []
afterEach(async () => { for (const d of dirs.splice(0)) await rm(d, { recursive: true, force: true }) })
afterEach(async () => {
for (const backend of backends.splice(0)) await backend.close()
for (const d of dirs.splice(0)) await rm(d, { recursive: true, force: true })
})
async function freshDbPath(): Promise<string> {
const dir = await mkdtemp(join(tmpdir(), 'dsh-storage-sqlite-'))
@@ -41,6 +47,13 @@ const DESCRIPTOR: KvUnitDescriptor = {
hasGlobal: true,
}
/** Exercise either a record write or a batch delete on the opened unit. */
function mutateUnit(unit: KvUnit, operation: 'putRecord' | 'deleteRecords'): Promise<void> {
return operation === 'putRecord'
? unit.putRecord('records', 'new', { n: 2 })
: unit.deleteRecords('records', ['saved', 'missing', 'saved'])
}
describe('sqlite backend specifics', () => {
it('opens an in-memory database', async () => {
const backend = backendAt(':memory:')
@@ -50,21 +63,26 @@ describe('sqlite backend specifics', () => {
await backend.close()
})
it('rolls back a batch delete whose transaction fails', async () => {
it.each(['ABORT', 'ROLLBACK'])('restores a batch delete after SQLite %s', async (action) => {
const path = await freshDbPath()
const backend = backendAt(path)
const unit = await backend.kv.open(DESCRIPTOR)
await unit.putRecord('records', 'a', { n: 1 })
await unit.putRecord('records', 'b', { n: 2 })
const db = new DatabaseSync(path)
db.exec('CREATE TRIGGER block_b BEFORE DELETE ON "u_specimen_records"'
+ " WHEN old.key = 'b' BEGIN SELECT RAISE(ABORT, 'blocked'); END")
await expect(unit.deleteRecords('records', ['a', 'b'])).rejects.toThrow(/blocked/)
db.exec('DROP TRIGGER block_b')
db.close()
// The transaction rolled back, so the key deleted before the failing row is back.
expect((await unit.loadAll()).tables['records']).toEqual({ a: { n: 1 }, b: { n: 2 } })
await backend.close()
try {
db.exec('CREATE TRIGGER block_b BEFORE DELETE ON "u_specimen_records"'
+ ` WHEN old.key = 'b' BEGIN SELECT RAISE(${action}, 'blocked'); END`)
await expect(unit.deleteRecords('records', ['a', 'b'])).rejects.toThrow(/blocked/)
// The transaction rolled back, so the key deleted before the failing row is back.
expect((await unit.loadAll()).tables['records']).toEqual({ a: { n: 1 }, b: { n: 2 } })
db.exec('DROP TRIGGER block_b')
await unit.deleteRecords('records', ['a', 'b'])
expect((await unit.loadAll()).tables['records']).toEqual({})
} finally {
await backend.close()
db.close()
}
})
it('reports unknown-state when a failed batch cannot roll back', async () => {
@@ -98,23 +116,26 @@ describe('sqlite backend specifics', () => {
await backend.close()
})
it('reports unknown-state when a batch cannot open its transaction', async () => {
it('leaves a batch unapplied when BEGIN fails, then retries on the same unit', async () => {
const path = await freshDbPath()
const backend = backendAt(path)
const unit = await backend.kv.open(DESCRIPTOR)
await unit.putRecord('records', 'a', { n: 1 })
// oxlint-disable-next-line typescript/unbound-method -- Delegated through Reflect.apply with the spy's receiver.
const original = DatabaseSync.prototype.exec
const beginError = new Error('begin denied')
const execSpy = vi.spyOn(DatabaseSync.prototype, 'exec').mockImplementation(function (this: DatabaseSync, sql: string) {
if (sql.startsWith('BEGIN')) throw new Error('begin denied')
if (sql.startsWith('BEGIN')) throw beginError
Reflect.apply(original, this, [sql])
})
try {
await expect(unit.deleteRecords('records', ['a'])).rejects.toMatchObject({ code: 'unknown-state' })
await expect(unit.deleteRecord('records', 'a')).rejects.toMatchObject({ code: 'unknown-state' })
await expect(unit.deleteRecords('records', ['a'])).rejects.toBe(beginError)
expect((await unit.loadAll()).tables['records']).toEqual({ a: { n: 1 } })
} finally {
execSpy.mockRestore()
}
await unit.deleteRecords('records', ['a'])
expect((await unit.loadAll()).tables['records']).toEqual({})
await expect(unit.close()).resolves.toBeUndefined()
await backend.close()
})
@@ -155,6 +176,125 @@ describe('sqlite backend specifics', () => {
await backend.close()
})
it.each([
['single', 'putRecord'], ['single', 'deleteRecords'],
['per-record', 'putRecord'], ['per-record', 'deleteRecords'],
] as const)('preserves the unit stamp when BEGIN is busy with layout %s and %s, then retries', async (layout, operation) => {
const path = await freshDbPath()
const first = backendAt(path)
await (await first.kv.open(DESCRIPTOR)).putRecord('records', 'saved', { n: 1 })
await first.close()
const observer = new DatabaseSync(path)
const backend = backendAt(path)
try {
const unit = await backend.kv.open({ ...DESCRIPTOR, version: 2, compatibleVersions: [1], layout })
await unit.deleteRecord('records', 'missing')
expect(observer.prepare('SELECT version FROM units WHERE name = ?').get(DESCRIPTOR.name)).toEqual({ version: 1 })
observer.exec('BEGIN IMMEDIATE')
await expect(mutateUnit(unit, operation)).rejects.toMatchObject({ errcode: 5 })
expect(observer.isTransaction).toBe(true)
expect(observer.prepare('SELECT version FROM units WHERE name = ?').get(DESCRIPTOR.name)).toEqual({ version: 1 })
expect((await unit.loadAll()).tables['records']).toEqual({ saved: { n: 1 } })
observer.exec('ROLLBACK')
await mutateUnit(unit, operation)
expect(observer.prepare('SELECT version FROM units WHERE name = ?').get(DESCRIPTOR.name)).toEqual({ version: 2 })
expect((await unit.loadAll()).tables['records']).toEqual(operation === 'putRecord' ? { saved: { n: 1 }, new: { n: 2 } } : {})
await backend.close()
const oldReader = backendAt(path)
await expect(oldReader.kv.open({ ...DESCRIPTOR, layout })).rejects.toMatchObject({ code: 'version-mismatch' })
await oldReader.close()
} finally {
if (observer.isTransaction) observer.exec('ROLLBACK')
await backend.close()
observer.close()
}
})
it.each([
['ABORT', 'putRecord'], ['ABORT', 'deleteRecords'],
['ROLLBACK', 'putRecord'], ['ROLLBACK', 'deleteRecords'],
] as const)('preserves data and stamp on %s during the first %s version update, then retries', async (action, operation) => {
const path = await freshDbPath()
const first = backendAt(path)
await (await first.kv.open(DESCRIPTOR)).putRecord('records', 'saved', { n: 1 })
await first.close()
const observer = new DatabaseSync(path)
const backend = backendAt(path)
try {
const physicalVersion = observer.prepare('PRAGMA user_version').get()
observer.exec(`CREATE TRIGGER reject_unit_version BEFORE UPDATE ON units BEGIN SELECT RAISE(${action}, 'version write failed'); END`)
const unit = await backend.kv.open({ ...DESCRIPTOR, version: 2, compatibleVersions: [1] })
const savedVersion = () => observer.prepare('SELECT version FROM units WHERE name = ?').get(DESCRIPTOR.name)
expect(savedVersion()).toEqual({ version: 1 })
await expect(mutateUnit(unit, operation)).rejects.toThrow('version write failed')
expect(savedVersion()).toEqual({ version: 1 })
expect((await unit.loadAll()).tables['records']).toEqual({ saved: { n: 1 } })
observer.exec('DROP TRIGGER reject_unit_version')
await mutateUnit(unit, operation)
expect(savedVersion()).toEqual({ version: 2 })
expect(observer.prepare('PRAGMA user_version').get()).toEqual(physicalVersion)
expect((await unit.loadAll()).tables['records']).toEqual(operation === 'putRecord' ? { saved: { n: 1 }, new: { n: 2 } } : {})
} finally {
await backend.close()
observer.close()
}
})
it.each(['putRecord', 'deleteRecords'] as const)('rejects all backend use after a failed %s rollback and preserves both errors', async (operation) => {
const path = await freshDbPath()
const siblingDescriptor = { ...DESCRIPTOR, name: 'sibling' }
const first = backendAt(path)
await (await first.kv.open(DESCRIPTOR)).putRecord('records', 'saved', { n: 1 })
await (await first.kv.open(siblingDescriptor)).close()
await first.close()
const observer = new DatabaseSync(path)
const backend = backendAt(path)
try {
observer.exec("CREATE TRIGGER reject_unit_version BEFORE UPDATE ON units BEGIN SELECT RAISE(ABORT, 'version write failed'); END")
const unit = await backend.kv.open({ ...DESCRIPTOR, version: 2, compatibleVersions: [1] })
const sibling = await backend.kv.open(siblingDescriptor)
// oxlint-disable-next-line typescript/unbound-method -- replay the original method with the intercepted connection as this
const originalExec = DatabaseSync.prototype.exec
const rollbackError = new Error('rollback failed')
const rollback = vi.spyOn(DatabaseSync.prototype, 'exec').mockImplementation(function (this: DatabaseSync, sql) {
if (sql === 'ROLLBACK') throw rollbackError
originalExec.call(this, sql)
})
try {
const pending = backend.kv.open({ ...DESCRIPTOR, name: 'pending' })
const failure: unknown = await mutateUnit(unit, operation).catch((error: unknown) => error)
expect(failure).toMatchObject({
name: 'StorageError',
code: 'unknown-state',
cause: {
name: 'AggregateError',
cause: { message: 'version write failed' },
errors: [{ message: 'version write failed' }, rollbackError],
},
})
await expect(sibling.putRecord('records', 'acknowledged', { n: 3 })).rejects.toBe(failure)
await expect(sibling.loadAll()).rejects.toBe(failure)
await expect(unit.loadAll()).rejects.toBe(failure)
await expect(pending).rejects.toBe(failure)
await expect(backend.kv.open({ ...DESCRIPTOR, name: 'later' })).rejects.toBe(failure)
expect(observer.prepare('SELECT * FROM u_sibling_records').all()).toEqual([])
} finally {
rollback.mockRestore()
}
await backend.close()
expect(observer.prepare('SELECT version FROM units WHERE name = ?').get(DESCRIPTOR.name)).toEqual({ version: 1 })
expect(observer.prepare('SELECT key FROM u_specimen_records').all()).toEqual([{ key: 'saved' }])
expect(observer.prepare('SELECT * FROM u_sibling_records').all()).toEqual([])
} finally {
await backend.close()
observer.close()
}
})
it('rejects invalid unit and table names before touching the medium', async () => {
const backend = backendAt(':memory:')
await expect(backend.kv.open({ ...DESCRIPTOR, name: 'Bad-Name' })).rejects.toThrow(/violates/)
+2 -2
View File
@@ -36,8 +36,8 @@
en: bbdc4833afb85dad
zh: 7494de01d5a4aecf
/deepseek-ai-dsh-storage/understand-the-implementation/the-backend-contract:
en: f03b47b38dee85dc
zh: 0fe0fc1e580d6ab9
en: bbbacd2373dc0cfa
zh: 6aa58b25feaf6a9a
/deepseek-ai-dsh-storage/understand-the-implementation/source-map:
en: 506e29aac47a1913
zh: b71dcbf4080e2c49
+1 -1
View File
@@ -76,7 +76,7 @@ The hub is a pure registration table with two faces, designed so backends and da
### The backend contract
[`src/backend.ts`](src/backend.ts) is the normative contract for backend implementers, checked for the shared clauses by the shared conformance suite in `tests/contract.ts`. A backend owns exactly one medium and exposes optional data-shape facets; `kv` is the only facet, and opening a unit yields a versioned, globally-singleton schema handle whose single calls are atomic, with durability scoped per layout. Unit and table names must match `UNIT_NAME_RE`; record keys are opaque in the `single` layout and path-safe segments in the `per-record` layout. The unit does not serialize concurrent writes — ordering belongs to the caller — and a stored version differing from the descriptor rejects `version-mismatch` (no migration).
[`src/backend.ts`](src/backend.ts) is the normative contract for backend implementers; the shared conformance suite in `tests/contract.ts` checks whole-unit behavior on both shipped backends. A backend owns exactly one medium and exposes optional data-shape facets; `kv` is the only facet. Opening a unit yields a handle for opaque JSON values whose atomicity and durability follow the selected layout. Unit and table names must match `UNIT_NAME_RE`. Whole-unit keys stay opaque; JSON per-record keys become path-safe file segments. The caller orders concurrent writes. Reads accept the current or declared `compatibleVersions`; unsupported unit stamps reject `version-mismatch`, while unsupported JSON per-record documents read as absent. Mutations atomically stamp the current version with their data; a missing-key delete does not restamp. Domains own value conversion, and backend implementers must preserve these semantics; see [domain versioning](../../../docs/subsystems/storage.md#domain-versioning) for mixed values, failure recovery, and upgrades.
### Source map
+1 -1
View File
@@ -76,7 +76,7 @@ kind: "package-reference"
### 后端约定
[`src/backend.ts`](src/backend.ts) 是后端实现者的规范性约定,由 `tests/contract.ts` 中的共享一致性套件检查共享条款。一个后端只拥有一种介质,并暴露可选的数据形状分面;`kv` 是唯一的分面,打开单元即可获得一个带版本、全局单例的 schema 句柄,其每次调用均具备原子性,持久性保证按布局区分。单元名与表名必须匹配 `UNIT_NAME_RE`;记录键在 `single` 布局中不透明,在 `per-record` 布局中则是路径安全片段。单元不对并发写入做串行化——顺序由调用方负责——介质上记录的版本与描述符不同时拒绝 `version-mismatch`(不做迁移)。
[`src/backend.ts`](src/backend.ts) 是后端实现者的规范性约定;`tests/contract.ts` 中的共享一致性套件检查两个已交付后端的整单元行为。一个后端只拥有一种介质,并暴露可选的数据形状分面;`kv` 是唯一的分面。打开单元即可获得一个把 JSON 视为不透明值的句柄,其原子性与持久性按所选布局界定。单元名与表名必须匹配 `UNIT_NAME_RE`。整单元布局的键保持不透明;JSON 逐记录布局的键会成为路径安全的文件路径片段。调用方负责安排并发写入的顺序。读取接受当前版本或已声明的 `compatibleVersions`;不支持的单元版本戳以 `version-mismatch` 拒绝,不支持的 JSON 逐记录文档则读作不存在。修改将当前版本戳与数据原子提交;删除不存在的键不会重新盖戳。领域负责值转换,后端实现者必须保持这些语义;混合值、失败恢复与升级参阅[领域版本](../../../docs/subsystems/storage.zh.md#domain-versioning)。
### 源码地图
+32 -26
View File
@@ -31,10 +31,11 @@ export interface KvFacet {
/**
* Open one unit, creating it when the medium holds no trace of it yet
* (materialization may defer to the first write, but {@link KvUnit.loadAll}
* must immediately serve the empty shape). A version already stamped on the
* medium that differs from `descriptor.version` rejects with
* `version-mismatch`; a medium that cannot be parsed as this unit rejects
* with `malformed-medium`. Opening the same unit name twice without closing
* must immediately serve the empty shape). Backends with a unit-wide stamp
* reject an existing version outside `descriptor.version` and
* `compatibleVersions` with `version-mismatch`; document-based per-record
* layouts read an unaccepted document as absent. A medium that cannot be
* parsed as this unit rejects with `malformed-medium`. Opening the same unit name twice without closing
* is a caller bug and rejects.
* @param descriptor - Static identity and shape of the unit to open.
* @returns the opened unit.
@@ -46,29 +47,30 @@ export interface KvFacet {
export interface KvUnitDescriptor {
/** Unit name; must match {@link UNIT_NAME_RE}. Also the file-name / SQL-identifier segment. */
readonly name: string
/** Unit format version; a non-negative integer stamped on the medium at first materialization. */
/** Current unit format version; a non-negative integer stamped on initial materialization and subsequent mutations. */
readonly version: number
/** Table names; each must match {@link UNIT_NAME_RE}. */
readonly tables: readonly string[]
/** Whether this unit carries the global singleton slot. */
readonly hasGlobal: boolean
/**
* Medium layout. `single` (the default) keeps the whole unit in one
* Document layout. `single` (the default) keeps the whole unit in one
* document; `per-record` keeps each record in its own document, so a unit
* whose records are large or sparse never rewrites the rest on one write,
* and an unaccepted version stamp discards only that record instead of
* rejecting the whole unit. Backends that only serve one layout accept the
* other's units as foreign documents.
* and an unaccepted document reads as absent instead of
* rejecting the whole unit. Row stores retain unit-wide versioning for
* either layout request.
*/
readonly layout?: 'single' | 'per-record'
/**
* Older unit versions whose stored records are also readable under the
* declaring owner's current record schemas (the owner vouches for that —
* typically by declaring the fields old records lack as optional). Reads of
* a `per-record` unit accept documents stamped with any listed version, and
* the legacy whole-unit bootstrap accepts a legacy file stamped with one;
* writes always stamp {@link version}. `single`-layout reads stay
* exact-version.
* declaring owner's current schemas (the owner vouches for that —
* typically by declaring the fields old records lack as optional). Both
* layouts accept listed versions without restamping ordinary reads. Whole-unit
* mutations atomically stamp {@link version} with their data; per-record
* replacements stamp their document. Untouched records keep their saved
* values, so the current schemas must accept mixed generations.
* The legacy per-record bootstrap also accepts listed versions.
*/
readonly compatibleVersions?: readonly number[]
}
@@ -77,11 +79,11 @@ export interface KvUnitDescriptor {
* One opened unit. Values are opaque JSON to this layer: no schema, no
* events, no domain meaning. The unit does NOT serialize concurrent writes —
* write ordering is the caller's responsibility (the domain layer runs one
* write chain per unit); each single call is atomic on the medium. Durability
* holds for single-record writes and whole-file publishes. In the
* `per-record` layout batch deletions rely on the filesystem persisting
* directory-entry updates, so a crash can restore a deleted document. Any call
* after {@link close} rejects with `closed`.
* write chain per unit). Whole-document and row-store calls are atomic on the
* medium; a per-record batch interrupted before resolution can be partially
* applied. Resolved writes are durable; per-record deletions sync directory
* changes on POSIX before resolving. Any call after {@link close} rejects with
* `closed`.
*/
export interface KvUnit {
/**
@@ -103,7 +105,8 @@ export interface KvUnit {
putRecord(table: string, key: string, value: unknown): Promise<void>
/**
* Delete one record durably. Idempotent: a missing key is a no-op.
* Delete one record durably. Idempotent: a missing key is a no-op and does
* not advance the version stamp.
* @param table - Declared table name.
* @param key - Record key.
* @returns resolution after durability.
@@ -111,17 +114,20 @@ export interface KvUnit {
deleteRecord(table: string, key: string): Promise<void>
/**
* Delete several records durably as one batch. Idempotent: missing keys are
* no-ops. Resolves once every listed record is absent from the readable
* Delete several records durably as one batch. Idempotent: missing keys and
* empty batches are no-ops and do not advance the version stamp. Resolves
* once every listed record is absent from the readable
* state and the backend accepted the batch. A rejection normally leaves the
* batch unapplied, so no listed record was removed. A backend that cannot
* establish or restore the batch's state rejects with `unknown-state`
* instead: the readable state is then unknown, every later operation on that
* instead: the readable or durable state is then unknown, every later operation on that
* unit rejects with the same error until {@link close} releases it, and
* calls after close reject `closed`. On the `single` layout a
* directory-fsync failure can arrive after the atomic replacement: the new
* file may already hold the batch while the caller sees a rejection, and the
* next publish rewrites it. A backend whose medium can apply a whole batch
* file may already hold the batch while the caller sees a rejection. Close
* the backend and its domains, then reopen to inspect saved state before
* retrying. Per-record batches are not crash-atomic before resolution.
* A backend whose medium can apply a whole batch
* in one operation (the `single` document, a row store) must do that instead
* of repeating {@link deleteRecord} per key.
* @param table - Declared table name.
+138 -2
View File
@@ -89,14 +89,14 @@ export function runKvBackendContract(label: string, create: () => Promise<KvBack
await reopened.close()
})
it('rejects a version mismatch on reopen without touching the data', async () => {
it.each([2, 4])('rejects version %s on reopen without touching the data', async (version) => {
const harness = await create()
const unit = await harness.backend.kv!.open(DESCRIPTOR)
await unit.putRecord('alpha', 'k', { v: 1 })
await harness.backend.close()
const reopened = await harness.reopen()
await expect(reopened.kv!.open({ ...DESCRIPTOR, version: 4 })).rejects.toMatchObject({
await expect(reopened.kv!.open({ ...DESCRIPTOR, version })).rejects.toMatchObject({
name: 'StorageError',
code: 'version-mismatch',
})
@@ -106,6 +106,142 @@ export function runKvBackendContract(label: string, create: () => Promise<KvBack
await reopened.close()
})
it('reads an accepted older version without changing its stamp or records', async () => {
const harness = await create()
const original = await harness.backend.kv!.open(DESCRIPTOR)
await original.putRecord('alpha', 'k', { v: 1 })
await original.setGlobal({ counter: 1 })
await harness.backend.close()
const current = { ...DESCRIPTOR, version: 4, compatibleVersions: [3] }
const reader = await harness.reopen()
const unit = await reader.kv!.open(current)
expect(await unit.loadAll()).toEqual({ tables: { alpha: { k: { v: 1 } }, beta: {} }, global: { counter: 1 } })
await reader.close()
const unchanged = await harness.reopen()
expect((await (await unchanged.kv!.open(DESCRIPTOR)).loadAll()).tables['alpha']).toEqual({ k: { v: 1 } })
await unchanged.close()
})
it.each(['deleteRecord', 'deleteRecords', 'emptyBatch'] as const)('keeps the older stamp for a no-op %s', async (operation) => {
const harness = await create()
const original = await harness.backend.kv!.open(DESCRIPTOR)
await original.putRecord('alpha', 'saved', { v: 1 })
await harness.backend.close()
const writer = await harness.reopen()
const unit = await writer.kv!.open({ ...DESCRIPTOR, version: 4, compatibleVersions: [3] })
if (operation === 'deleteRecord') await unit.deleteRecord('alpha', 'missing')
else await unit.deleteRecords('alpha', operation === 'emptyBatch' ? [] : ['missing', 'missing'])
await writer.close()
const unchanged = await harness.reopen()
expect((await (await unchanged.kv!.open(DESCRIPTOR)).loadAll()).tables['alpha'])
.toEqual({ saved: { v: 1 } })
await unchanged.close()
})
it('restores the open snapshot after serialization failure and retries on the same unit', async () => {
const harness = await create()
await (await harness.backend.kv!.open(DESCRIPTOR)).putRecord('alpha', 'saved', { v: 1 })
await harness.backend.close()
const current = { ...DESCRIPTOR, version: 4, compatibleVersions: [3] }
const writer = await harness.reopen()
const unit = await writer.kv!.open(current)
await expect(unit.putRecord('alpha', 'broken', { toJSON: () => { throw new Error('serialization failed') } }))
.rejects.toThrow('serialization failed')
expect((await unit.loadAll()).tables['alpha']).toEqual({ saved: { v: 1 } })
await unit.putRecord('alpha', 'retry', { v: 2 })
await writer.close()
const restarted = await harness.reopen()
expect((await (await restarted.kv!.open(current)).loadAll()).tables['alpha'])
.toEqual({ saved: { v: 1 }, retry: { v: 2 } })
await restarted.close()
})
it.each(['putRecord', 'deleteRecord', 'deleteRecords', 'setGlobal'] as const)('%s commits the current stamp with its data', async (operation) => {
const harness = await create()
const original = await harness.backend.kv!.open(DESCRIPTOR)
await original.putRecord('alpha', 'k', { v: 1 })
await original.putRecord('beta', 'untouched', { saved: true })
await original.setGlobal({ counter: 1 })
const otherDescriptor = { ...DESCRIPTOR, name: 'other_unit' }
await (await harness.backend.kv!.open(otherDescriptor)).putRecord('alpha', 'saved', { v: 1 })
await harness.backend.close()
const current = { ...DESCRIPTOR, version: 4, compatibleVersions: [3] }
const writer = await harness.reopen()
const unit = await writer.kv!.open(current)
await expect(unit.putRecord('alpha', 'k', { toJSON: () => { throw new Error('serialization failed') } }))
.rejects.toThrow('serialization failed')
expect((await unit.loadAll()).tables['alpha']).toEqual({ k: { v: 1 } })
await writer.close()
const afterFailure = await harness.reopen()
expect((await (await afterFailure.kv!.open(DESCRIPTOR)).loadAll()).tables['alpha']).toEqual({ k: { v: 1 } })
await afterFailure.close()
const retry = await harness.reopen()
const writable = await retry.kv!.open(current)
await writable.deleteRecord('alpha', 'missing')
await writable.deleteRecords('alpha', [])
if (operation === 'putRecord') await writable.putRecord('alpha', 'k', { v: 2 })
else if (operation === 'deleteRecord') await writable.deleteRecord('alpha', 'k')
else if (operation === 'deleteRecords') await writable.deleteRecords('alpha', ['k', 'missing', 'k'])
else await writable.setGlobal({ counter: 2 })
await retry.close()
const restarted = await harness.reopen()
await expect(restarted.kv!.open(DESCRIPTOR)).rejects.toMatchObject({ code: 'version-mismatch' })
const saved = await (await restarted.kv!.open(current)).loadAll()
expect(saved.tables['beta']).toEqual({ untouched: { saved: true } })
expect(saved.tables['alpha']).toEqual(
operation === 'deleteRecord' || operation === 'deleteRecords' ? {} : { k: { v: operation === 'putRecord' ? 2 : 1 } },
)
expect(saved.global).toEqual({ counter: operation === 'setGlobal' ? 2 : 1 })
expect((await (await restarted.kv!.open(otherDescriptor)).loadAll()).tables['alpha'])
.toEqual({ saved: { v: 1 } })
await restarted.close()
})
it.each([
{ name: 'v1 → v3', writeV2: false },
{ name: 'v1 → v2 → v3', writeV2: true },
])('reads historical values through $name', async ({ writeV2 }) => {
const harness = await create()
const v1 = { ...DESCRIPTOR, version: 1 }
const v2 = { ...DESCRIPTOR, version: 2, compatibleVersions: [1] }
const v3 = { ...DESCRIPTOR, compatibleVersions: writeV2 ? [2] : [1] }
const original = await harness.backend.kv!.open(v1)
await original.putRecord('alpha', 'untouched', { label: 'Keep' })
await original.putRecord('alpha', 'edited', { label: 'Edit' })
await harness.backend.close()
if (writeV2) {
const second = await harness.reopen()
const middle = await second.kv!.open(v2)
await middle.putRecord('alpha', 'edited', { title: 'Edit' })
await second.close()
}
const third = await harness.reopen()
const latest = await third.kv!.open(v3)
expect((await latest.loadAll()).tables['alpha']).toEqual({
untouched: { label: 'Keep' }, edited: writeV2 ? { title: 'Edit' } : { label: 'Edit' },
})
await latest.putRecord('alpha', 'edited', { title: 'Edit', addedInV3: true })
await third.close()
const restarted = await harness.reopen()
await expect(restarted.kv!.open(v2)).rejects.toMatchObject({ code: 'version-mismatch' })
expect((await (await restarted.kv!.open(v3)).loadAll()).tables['alpha']).toMatchObject({
untouched: { label: 'Keep' }, edited: { addedInV3: true },
})
await restarted.close()
})
it('rejects operations after unit close, and close is idempotent', async () => {
const { backend } = await create()
const unit = await backend.kv!.open(DESCRIPTOR)