fix(B-6 ③ + 测试介质): 陈旧实例拒绝提交覆盖 + SharedMemoryBackend 跨实例可见性
③ 陈旧实例的 checkpoint 会静默抹掉新实例的写入（P0）
修复前多个 KVStore 实例同时打开同一库时，snapshot/meta 是全库共享的，而每个
实例各有内存索引与 seq —— 落后实例的一次 checkpoint 会用它的旧索引覆盖介质：
A.open → A.put(x) ; B.open(读到 x) → B.put(y) → B.close()
→ A.checkpoint() // A 的索引里没有 y
→ 重开：x 在、**y 消失**，且 A.checkpoint() **没有报任何错**
修法：meta 增加 `owner`（实例 id），open() 时领取所有权；checkpoint 前比对
owner —— 不是自己即判为过期，抛 `STALE_INSTANCE` 并拒绝提交（而不是覆盖）；
此后该实例的写入也显式失败（否则只会写进永远无法提交的 WAL）。
重新 open 即可恢复（陈旧状态不是永久的）。
**测试介质本身的缺陷（同源发现，影响此前的所有多实例验证）**
`SharedMemoryBackend` 每个实例各有一份私有拼接缓存，且只在**自己的**
write/append 时失效 → 实例 A 读过的键永远命中 A 的缓存，**看不到** B 之后的
写入。介质行为退化为"每个实例各有一份快照"，于是所有"两实例共享介质"的
崩溃/多标签页用例都跑在错误的介质上（这正是上面那条 bug 起初查不出来的原因：
守卫读 owner 时拿到的是自己的旧值）。
修法：拼接缓存按库名**共享**（挂在注册表条目上），任何实例的 write/append/
delete 都清掉该库的共享缓存；删除用 `null` 哨兵，杜绝"删了还能读到旧值"。
同时保持既有 close 契约（close 后读返回 null、写删清空不抛错且不持久化，
跨实例持久化语义不变）—— `clearRegistry()` 换代，避免跨用例数据泄漏。
验证：tests/v080-kvstore-commit-point.test.ts 扩到 19 项（含 4 项陈旧实例 +
4 项介质可见性）；全量 87 套件 / 1686 测试通过（含 Aria 10 万行生产负载）；
typecheck(src+tests) 与 lint 零错误。
This commit is contained in:
thzxx
+110
-6
@@ -39,6 +39,20 @@ const DEFAULT_CHECKPOINT_THRESHOLD = 16 * 1024 * 1024;
|
||||
interface KVStoreMeta {
|
||||
/** 当前日志水位（快照内嵌；无快照时 0） */
|
||||
seq: number;
|
||||
/**
|
||||
* v0.8.0（B-6）：最后一次提交（checkpoint）的实例 id。
|
||||
*
|
||||
* 为什么需要：多个 KVStore 实例可能同时打开同一个库（多标签页、或测试里
|
||||
* 共享介质）。每个实例各有自己的内存索引与 `seq`，而 snapshot/meta 是
|
||||
* **全库共享**的 —— 于是落后实例的一次 checkpoint 会用它的旧索引覆盖介质，
|
||||
* 把其它实例已提交的写入**静默抹掉**。实测（本提交的用例锁定）：
|
||||
* A.open → A.put(x) ; B.open(=读到 x) → B.put(y) → B.close()
|
||||
* → A.checkpoint() // A 的索引里没有 y
|
||||
* → 重开：x 在、**y 消失**，且 A.checkpoint() 没有报任何错
|
||||
* 有了 owner 之后，checkpoint 前先比对 meta.owner：不是自己就说明
|
||||
* 介质已被别的实例接管，此时**拒绝提交**并显式报错，而不是覆盖。
|
||||
*/
|
||||
owner?: string;
|
||||
}
|
||||
|
||||
function defaultMedium(): IStorageBackend {
|
||||
@@ -67,6 +81,16 @@ export class KVStore {
|
||||
private opQueue: Promise<unknown> = Promise.resolve();
|
||||
/** 最近一次后台操作失败（checkpoint 时报告） */
|
||||
private lastBackgroundError: unknown = null;
|
||||
/**
|
||||
* v0.8.0（B-6）：本实例的提交所有权标识。
|
||||
*
|
||||
* 每次 `open()` 领取一个新的随机 id 并写入 meta（见 `claimOwnership`）。
|
||||
* checkpoint 前比对 `meta.owner`：不等于自己 → 介质已被更新的实例接管，
|
||||
* 本实例的索引可能落后，**不得**再提交快照（否则会静默抹掉对方的写入）。
|
||||
*/
|
||||
private instanceId = '';
|
||||
/** 是否已被更新的实例接管（进入该状态后所有写入与提交都被拒绝） */
|
||||
private stale = false;
|
||||
|
||||
constructor(medium?: IStorageBackend, checkpointThreshold: number = DEFAULT_CHECKPOINT_THRESHOLD) {
|
||||
this.medium = medium ?? defaultMedium();
|
||||
@@ -87,7 +111,12 @@ export class KVStore {
|
||||
this.dbName = dbName;
|
||||
// v0.7.4: 打开时同样清理后台错误状态（防 close/reopen 残留）
|
||||
this.lastBackgroundError = null;
|
||||
this.stale = false;
|
||||
await this.medium.open(dbName);
|
||||
// v0.8.0（B-6）：领取提交所有权（写回 meta.owner）。
|
||||
// 打开是唯一"接管"介质的时机；此前的实例之后会在 checkpoint 时发现
|
||||
// owner 已变而拒绝提交，从而不会用陈旧索引覆盖本实例的数据。
|
||||
await this.claimOwnership();
|
||||
this.index = new Map();
|
||||
this.seq = 0;
|
||||
this.logBytes = 0;
|
||||
@@ -176,6 +205,8 @@ export class KVStore {
|
||||
// v0.7.4: 清理后台错误状态 —— 此前跨 close/reopen 残留，
|
||||
// 重开后首次 checkpoint 会抛出上一次生命周期的旧错误
|
||||
this.lastBackgroundError = null;
|
||||
this.instanceId = '';
|
||||
this.stale = false;
|
||||
this.opened = false;
|
||||
}
|
||||
|
||||
@@ -269,6 +300,8 @@ export class KVStore {
|
||||
/** checkpoint：快照 → meta → 截断日志（时序保证任何崩溃窗口不丢数据） */
|
||||
async checkpoint(): Promise<void> {
|
||||
await this.enqueue(async () => {
|
||||
// v0.8.0（B-6）：提交前确认所有权 —— 拒绝用陈旧索引覆盖介质
|
||||
await this.assertOwnership();
|
||||
// 报告上次后台失败
|
||||
if (this.lastBackgroundError !== null) {
|
||||
const error = this.lastBackgroundError;
|
||||
@@ -280,9 +313,8 @@ export class KVStore {
|
||||
// 1. 写快照（COW 原子）
|
||||
const snapBytes = encodeSnapshot(this.seq, this.index);
|
||||
await this.medium.write(SNAPSHOT_KEY, snapBytes.buffer as ArrayBuffer);
|
||||
// 2. 写 meta（指向新水位）
|
||||
const meta: KVStoreMeta = { seq: this.seq };
|
||||
await this.medium.write(META_KEY, new TextEncoder().encode(JSON.stringify(meta)).buffer);
|
||||
// 2. 写 meta（指向新水位 + 续期所有权）
|
||||
await this.writeMeta();
|
||||
// 3. 截断日志（meta 已更新 → 截断安全）
|
||||
await this.truncateLog();
|
||||
});
|
||||
@@ -345,6 +377,15 @@ export class KVStore {
|
||||
deletes: string[],
|
||||
appends: Record<string, ArrayBuffer> = {},
|
||||
): Promise<void> {
|
||||
// v0.8.0（B-6）：已被接管 → 拒绝写入（否则写入只会进 WAL 而永远无法提交，
|
||||
// 或者在下一次 checkpoint 时把对方的提交覆盖掉）
|
||||
if (this.stale) {
|
||||
throw new DatabaseError(
|
||||
'KVStore instance has been superseded by another instance on the same database;'
|
||||
+ ' re-open the database to continue',
|
||||
'STALE_INSTANCE',
|
||||
);
|
||||
}
|
||||
this.seq++;
|
||||
const record = encodeLogRecord(this.seq, puts, deletes, appends);
|
||||
try {
|
||||
@@ -414,12 +455,15 @@ export class KVStore {
|
||||
*/
|
||||
private async autoCheckpoint(): Promise<void> {
|
||||
try {
|
||||
await this.assertOwnership();
|
||||
await this.medium.write(SNAPSHOT_KEY, encodeSnapshot(this.seq, this.index).buffer as ArrayBuffer);
|
||||
const meta: KVStoreMeta = { seq: this.seq };
|
||||
await this.medium.write(META_KEY, new TextEncoder().encode(JSON.stringify(meta)).buffer);
|
||||
await this.writeMeta();
|
||||
await this.truncateLog();
|
||||
} catch (error) {
|
||||
this.lastBackgroundError = error;
|
||||
// 陈旧实例的自动 checkpoint 失败**不**计入 lastBackgroundError：
|
||||
// 那会把它伪装成"介质故障"，而真实原因是本实例已被接管，用户需要的是
|
||||
// 立即的、明确的 STALE_INSTANCE 错误（由 assertOwnership 在显式路径给出）。
|
||||
if (!this.stale) this.lastBackgroundError = error;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -473,6 +517,66 @@ export class KVStore {
|
||||
}
|
||||
}
|
||||
|
||||
// =======================================================================
|
||||
// 提交所有权（v0.8.0 / B-6）
|
||||
// =======================================================================
|
||||
|
||||
/** 生成实例 id（无需密码学强度，只要同期实例间不碰撞） */
|
||||
private newInstanceId(): string {
|
||||
const rand = Math.random().toString(36).slice(2);
|
||||
return `kv-${Date.now().toString(36)}-${rand}`;
|
||||
}
|
||||
|
||||
/** 读取 meta（缺失或损坏时返回 null） */
|
||||
private async readMeta(): Promise<KVStoreMeta | null> {
|
||||
try {
|
||||
const raw = await this.medium.read(META_KEY);
|
||||
if (!raw || raw.byteLength === 0) return null;
|
||||
return JSON.parse(new TextDecoder().decode(raw)) as KVStoreMeta;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/** 写入 meta（水位 + 所有权续期） */
|
||||
private async writeMeta(): Promise<void> {
|
||||
const meta: KVStoreMeta = { seq: this.seq, owner: this.instanceId };
|
||||
await this.medium.write(META_KEY, new TextEncoder().encode(JSON.stringify(meta)).buffer);
|
||||
}
|
||||
|
||||
/** open() 时领取所有权：写入本实例 id，使此前打开的实例在提交时发现自己已过期 */
|
||||
private async claimOwnership(): Promise<void> {
|
||||
this.instanceId = this.newInstanceId();
|
||||
await this.writeMeta();
|
||||
}
|
||||
|
||||
/**
|
||||
* 提交前校验所有权。
|
||||
*
|
||||
* 只有当介质上的 owner 存在、且**不是**本实例时才算过期：
|
||||
* - owner 缺失（旧版本库 / meta 被清理）→ 视为无主，允许提交；
|
||||
* - owner === 自己 → 正常。
|
||||
* 一旦判定过期，本实例进入 `stale` 状态：后续写入与提交都显式报错 ——
|
||||
* 显式失败远好于静默丢数据（这是本 wound 的全部意义）。
|
||||
*/
|
||||
private async assertOwnership(): Promise<void> {
|
||||
const meta = await this.readMeta();
|
||||
if (meta?.owner && this.instanceId && meta.owner !== this.instanceId) {
|
||||
this.stale = true;
|
||||
throw new DatabaseError(
|
||||
'KVStore instance has been superseded by another instance on the same database'
|
||||
+ ' (refusing to commit a stale snapshot, which would discard the newer instance\'s writes);'
|
||||
+ ' re-open the database to continue',
|
||||
'STALE_INSTANCE',
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/** 本实例是否已被更新实例接管（诊断用） */
|
||||
isStale(): boolean {
|
||||
return this.stale;
|
||||
}
|
||||
|
||||
/** 确定日志中有效字节长度（从 0 开始连续解析到第一条损坏/残缺记录） */
|
||||
private findValidLogLength(log: Uint8Array): number {
|
||||
let offset = 0;
|
||||
|
||||
@@ -7,54 +7,127 @@
|
||||
* （模拟"磁盘持久化"语义——重新 open 同名库可读到上次写入的数据）。
|
||||
*
|
||||
* 仅用于测试与 Node 环境；浏览器使用 OPFS 介质（KVStore 默认自动选择）。
|
||||
*
|
||||
* ============================================================================
|
||||
* v0.8.0（B-6）：**介质本身必须有"多个实例共享同一份字节"的语义**
|
||||
* ============================================================================
|
||||
* 修复前每个实例各有一个私有 `materialized` 拼接缓存，而缓存只在**自己的**
|
||||
* write/append 时失效。于是实例 A 读过的键会一直被 A 的缓存命中，**看不到**
|
||||
* 实例 B 之后的写入 —— 介质行为退化为"每个实例各有一份快照"。
|
||||
*
|
||||
* 这个缺陷直接损害了本项目的验证能力：所有"两个实例共享介质"的崩溃/多标签页
|
||||
* 用例都在一个错误的介质上运行。实测：
|
||||
* A.open(db) → A 读 meta（缓存 A 的 owner）
|
||||
* B.open(db) → B 写 meta（新 owner）
|
||||
* A.checkpoint() → A 读 meta 仍得到**自己**的 owner
|
||||
* → "陈旧实例拒绝提交"的所有权守卫看起来失效，实际是介质没把 B 的写入
|
||||
* 反映给 A。
|
||||
*
|
||||
* 现在缓存**按库名共享**（挂在注册表条目上），并且失效是"对这一份存储"生效的：
|
||||
* - 任何实例的 write/append/delete 都会清掉该库的共享缓存；
|
||||
* - 存储为空时写回 `null` 哨兵，使"删除后仍读到旧值"不可能发生。
|
||||
*/
|
||||
|
||||
import type { IStorageBackend } from '../aria/store/backend';
|
||||
|
||||
/** 全局注册表：dbName → key → chunk 列表（跨实例共享，模拟持久化） */
|
||||
const registry = new Map<string, Map<string, ArrayBuffer[]>>();
|
||||
/** 一份"磁盘"：chunk 列表 + 拼接缓存（按库名共享，跨实例可见） */
|
||||
interface DbStore {
|
||||
/** key → chunk 列表（append O(1)，read 时一次性拼接） */
|
||||
chunks: Map<string, ArrayBuffer[]>;
|
||||
/** 惰性拼接缓存：key → 拼接结果（`null` 表示"该键不存在"） */
|
||||
materialized: Map<string, ArrayBuffer | null>;
|
||||
}
|
||||
|
||||
/** 全局注册表：dbName（+ 库 id）→ 共享存储（跨实例共享，模拟持久化） */
|
||||
const registry = new Map<string, DbStore>();
|
||||
|
||||
/**
|
||||
* 库 id 计数器：用于 `clearRegistry()` 后让"同名但属于新会话"的库拿到
|
||||
* **新的** DbStore，而不是复用上一会话残留的对象引用。
|
||||
*/
|
||||
let registryEpoch = 0;
|
||||
/** 当前会话里 dbName → 本次会话的 DbStore（换代后失效） */
|
||||
let liveStores = new Map<string, { epoch: number; store: DbStore }>();
|
||||
|
||||
function storeFor(name: string): DbStore {
|
||||
const live = liveStores.get(name);
|
||||
if (live && live.epoch === registryEpoch) return live.store;
|
||||
|
||||
const existing = registry.get(name);
|
||||
if (existing) {
|
||||
// 同一会话内复用（多个 backend 实例共享同一份字节）
|
||||
liveStores.set(name, { epoch: registryEpoch, store: existing });
|
||||
return existing;
|
||||
}
|
||||
const created: DbStore = { chunks: new Map(), materialized: new Map() };
|
||||
registry.set(name, created);
|
||||
liveStores.set(name, { epoch: registryEpoch, store: created });
|
||||
return created;
|
||||
}
|
||||
|
||||
export class SharedMemoryBackend implements IStorageBackend {
|
||||
private dbName = '';
|
||||
/** 存储：key → chunk 列表（append O(1)，read 时一次性拼接缓存） */
|
||||
private chunks: Map<string, ArrayBuffer[]> = new Map();
|
||||
/** 惰性拼接缓存（read 后缓存，write/append 失效） */
|
||||
private materialized: Map<string, ArrayBuffer> = new Map();
|
||||
private store: DbStore | null = null;
|
||||
|
||||
/** 清空全局注册表（测试隔离用） */
|
||||
/**
|
||||
* 清空全局注册表（测试隔离用）。
|
||||
*
|
||||
* 同时让"当前会话"失效：后续 open 会为同名库创建全新的存储 ——
|
||||
* 否则 `clearRegistry()` 之后新建的实例可能仍指向上一用例的 DbStore
|
||||
*（注册表被清空但对象引用还活在旧实例里），出现跨用例数据泄漏。
|
||||
*/
|
||||
static clearRegistry(): void {
|
||||
registry.clear();
|
||||
registryEpoch += 1;
|
||||
liveStores = new Map();
|
||||
}
|
||||
|
||||
async open(name: string): Promise<void> {
|
||||
this.dbName = name;
|
||||
if (!registry.has(name)) {
|
||||
registry.set(name, new Map());
|
||||
}
|
||||
// 从注册表恢复 chunks（持久化语义）
|
||||
this.chunks = registry.get(name)! as unknown as Map<string, ArrayBuffer[]>;
|
||||
this.materialized = new Map();
|
||||
this.store = storeFor(name);
|
||||
}
|
||||
|
||||
/** close 不清除数据（持久化语义：重开同名库数据仍在） */
|
||||
async close(): Promise<void> {
|
||||
this.chunks = new Map();
|
||||
this.materialized = new Map();
|
||||
this.store = null;
|
||||
}
|
||||
|
||||
isOpen(): boolean {
|
||||
return this.dbName !== '';
|
||||
return this.dbName !== '' && this.store !== null;
|
||||
}
|
||||
|
||||
/**
|
||||
* 取当前共享存储。
|
||||
*
|
||||
* v0.8.0：`close()` 之后**不再**惰性重新绑定到共享存储 —— 那会让
|
||||
* "close 后读不到数据"变成"close 后又读到了"（实测该行为被
|
||||
* `tests/production-abnormal.test.ts` 的 close 契约用例逮到）。
|
||||
*
|
||||
* 正确语义（与既有契约一致，见该用例注释）：
|
||||
* - close 后读 → `null`（本实例已不持有数据）；
|
||||
* - close 后写/删/清空 → **不抛错**，但作用在游离内存上、**不持久化**
|
||||
* - 跨实例持久化语义不受影响：另一个实例 open 同名库仍能读到 close 前的数据。
|
||||
*/
|
||||
private db(): DbStore {
|
||||
if (!this.store) {
|
||||
// 游离存储：仅供 close 后的调用"安全落地"，不进入注册表
|
||||
this.store = { chunks: new Map(), materialized: new Map() };
|
||||
}
|
||||
return this.store;
|
||||
}
|
||||
|
||||
async read(key: string): Promise<ArrayBuffer | null> {
|
||||
if (this.materialized.has(key)) return this.materialized.get(key)!;
|
||||
const list = this.chunks.get(key);
|
||||
if (!list || list.length === 0) return null;
|
||||
const db = this.db();
|
||||
if (db.materialized.has(key)) return db.materialized.get(key) ?? null;
|
||||
const list = db.chunks.get(key);
|
||||
if (!list || list.length === 0) {
|
||||
db.materialized.set(key, null);
|
||||
return null;
|
||||
}
|
||||
if (list.length === 1) {
|
||||
this.materialized.set(key, list[0]);
|
||||
db.materialized.set(key, list[0]);
|
||||
return list[0];
|
||||
}
|
||||
const total = list.reduce((s2, c) => s2 + c.byteLength, 0);
|
||||
const total = list.reduce((sum, c) => sum + c.byteLength, 0);
|
||||
const combined = new Uint8Array(total);
|
||||
let off = 0;
|
||||
for (const c of list) {
|
||||
@@ -62,56 +135,61 @@ export class SharedMemoryBackend implements IStorageBackend {
|
||||
off += c.byteLength;
|
||||
}
|
||||
const buf = combined.buffer as ArrayBuffer;
|
||||
this.materialized.set(key, buf);
|
||||
db.materialized.set(key, buf);
|
||||
return buf;
|
||||
}
|
||||
|
||||
async write(key: string, data: ArrayBuffer): Promise<void> {
|
||||
this.chunks.set(key, [data]);
|
||||
this.materialized.set(key, data);
|
||||
const db = this.db();
|
||||
db.chunks.set(key, [data]);
|
||||
db.materialized.set(key, data);
|
||||
}
|
||||
|
||||
async append(key: string, data: ArrayBuffer): Promise<void> {
|
||||
// O(1) 追加：只记录 chunk，read 时惰性拼接
|
||||
const list = this.chunks.get(key);
|
||||
if (list) {
|
||||
list.push(data);
|
||||
} else {
|
||||
this.chunks.set(key, [data]);
|
||||
}
|
||||
this.materialized.delete(key);
|
||||
// O(1) 追加：只记录 chunk，read 时惰性拼接。
|
||||
// 缓存按库共享，因此这里清掉的缓存对所有实例都生效 —— 这正是
|
||||
// "另一个实例的写入必须对自己可见"这一语义的实现点。
|
||||
const db = this.db();
|
||||
const list = db.chunks.get(key);
|
||||
if (list) list.push(data);
|
||||
else db.chunks.set(key, [data]);
|
||||
db.materialized.delete(key);
|
||||
}
|
||||
|
||||
async writeMany(entries: Record<string, ArrayBuffer>): Promise<void> {
|
||||
// 同步批量写入 = 原子（JS 单线程，无中间 await 点）
|
||||
const db = this.db();
|
||||
for (const [key, data] of Object.entries(entries)) {
|
||||
this.chunks.set(key, [data]);
|
||||
this.materialized.set(key, data);
|
||||
db.chunks.set(key, [data]);
|
||||
db.materialized.set(key, data);
|
||||
}
|
||||
}
|
||||
|
||||
async delete(key: string): Promise<void> {
|
||||
this.chunks.delete(key);
|
||||
this.materialized.delete(key);
|
||||
const db = this.db();
|
||||
db.chunks.delete(key);
|
||||
db.materialized.set(key, null);
|
||||
}
|
||||
|
||||
async deleteMany(keys: string[]): Promise<void> {
|
||||
const db = this.db();
|
||||
for (const key of keys) {
|
||||
this.chunks.delete(key);
|
||||
this.materialized.delete(key);
|
||||
db.chunks.delete(key);
|
||||
db.materialized.set(key, null);
|
||||
}
|
||||
}
|
||||
|
||||
async listKeys(): Promise<string[]> {
|
||||
return Array.from(this.chunks.keys());
|
||||
return Array.from(this.db().chunks.keys());
|
||||
}
|
||||
|
||||
async exists(key: string): Promise<boolean> {
|
||||
return this.chunks.has(key);
|
||||
return this.db().chunks.has(key);
|
||||
}
|
||||
|
||||
async clear(): Promise<void> {
|
||||
this.chunks.clear();
|
||||
this.materialized.clear();
|
||||
const db = this.db();
|
||||
db.chunks.clear();
|
||||
db.materialized.clear();
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user