/** * v0.4.3 回归测试 * 覆盖: * - P0: close 后后台 compaction 在 backend 关闭后执行(残留任务污染/吞错) * - P0: flush 报告后台失败(不再静默吞错) * - P1: commit 先持久化 WAL 再合并快照(WAL 失败时数据一致性) * - P1: OPFS close 等待挂起写完成 */ import { AriaEngine } from '../src/engine/aria/index'; import { createSchema } from '../src/table/schema'; import { LSM } from '../src/engine/aria/index/lsm'; import { OPFSEngine } from '../src/engine/opfs'; import 'fake-indexeddb/auto'; let idbCounter = 0; function uniqueDB(): string { return `r43-${Date.now()}-${++idbCounter}-${Math.random().toString(36).slice(2, 8)}`; } // =================================================================== // P0: close 与后台 compaction // =================================================================== describe('P0 — close 后无残留后台任务', () => { it('close 等待后台 compaction 完成(backend 关闭后无残留写)', async () => { const engine = new AriaEngine({ storageBackend: 'memory', // 小缓存 + 小 memtable:flush 文件超缓存上限被驱逐 → compaction 缓存未命中 → 触发调度 bufferPoolPages: 4, memtableSizeThreshold: 32 * 1024, checkpointInterval: 100000, }); await engine.open(uniqueDB(), 1); await engine.createTable(createSchema('t', { id: { type: 'string', primaryKey: true }, v: { type: 'number' }, data: { type: 'string' }, })); for (let i = 0; i < 400; i++) { await engine.insert('t', [{ id: `k${String(i).padStart(4, '0')}`, v: i, data: 'x'.repeat(200) }]); } await (engine as any).lsm.flush(); const backend = (engine as any).backend; // 立即 close:修复前 setTimeout compaction 在 backend.close() 后才执行 → 残留写 await engine.close(); // 给残留 setTimeout 执行机会 await new Promise((r) => setTimeout(r, 100)); const keysAfterClose = await (backend as any).listKeys(); // 修复前:close 后 compaction 写入 → store 残留 sst_* 文件 expect(keysAfterClose).toHaveLength(0); }); it('close 后立即 reopen 不被旧后台任务污染', async () => { const dbName = uniqueDB(); const engine = new AriaEngine({ storageBackend: 'indexeddb', bufferPoolPages: 4, memtableSizeThreshold: 32 * 1024, checkpointInterval: 100000, }); await engine.open(dbName, 1); await engine.createTable(createSchema('t', { id: { type: 'string', primaryKey: true }, v: { type: 'number' }, data: { type: 'string' }, })); for (let i = 0; i < 300; i++) { await engine.insert('t', [{ id: `k${String(i).padStart(4, '0')}`, v: i, data: 'x'.repeat(200) }]); } await (engine as any).lsm.flush(); // 立即 close + 立即 reopen(修复前:旧任务的闭包引用新 backend → 交叉写) await engine.close(); await engine.open(dbName, 1); // 新库数据必须完整(不被旧任务破坏) expect(await engine.count('t')).toBe(300); await engine.close(); }); }); // =================================================================== // P0: 后台失败可见性(不吞错) // =================================================================== describe('P0 — flush 报告后台失败', () => { class FailingStore { failNext = true; saved = 0; deleted = 0; metas: { id: number; level: number }[] = []; async save(id: number, _data: Uint8Array): Promise { if (this.failNext) { this.failNext = false; throw new Error('disk full (simulated)'); } this.saved++; } async load(_id: number): Promise { return null; } async delete(_id: number): Promise { this.deleted++; } async allocateId(): Promise { return ++this.saved; } async listMeta(): Promise<{ id: number; level: number }[]> { return this.metas; } async saveMeta(meta: { id: number; level: number }): Promise { this.metas.push(meta); } async deleteMeta(id: number): Promise { this.metas = this.metas.filter((m) => m.id !== id); } } it('后台 flush 失败后 flush() 抛 DatabaseError(ARIA_BACKGROUND_ERROR)', async () => { const store = new FailingStore(); const lsm = new LSM({ memtableSizeThreshold: 64, sstableStore: store as any, }); // 触发 freeze + 后台 flush(save 抛错 → 记录 lastBackgroundError) for (let i = 0; i < 200; i++) { lsm.put(`k${i}`, { v: i }); } await new Promise((r) => setTimeout(r, 50)); // flush 必须报告后台失败(修复前静默吞错) await expect(lsm.flush()).rejects.toMatchObject({ code: 'ARIA_BACKGROUND_ERROR' }); // 再次 flush:错误已消费,正常完成 await expect(lsm.flush()).resolves.toBeUndefined(); }); it('后台 compaction 失败后 flush() 报告(链不卡死)', async () => { const store = new FailingStore(); const lsm = new LSM({ memtableSizeThreshold: 32, sstableStore: store as any, }); for (let i = 0; i < 500; i++) { lsm.put(`k${i}`, { v: i }); } await new Promise((r) => setTimeout(r, 100)); // 链不卡死:flush 要么成功要么报告错误(不能永久 pending) const result = await Promise.race([ lsm.flush().then(() => 'ok', (e) => `err:${(e as any).code}`), new Promise((r) => setTimeout(() => r('pending'), 500)), ]); expect(result).not.toBe('pending'); }); }); // =================================================================== // P1: commit 顺序与 OPFS close // =================================================================== describe('P1 — 提交顺序与 close 等待', () => { it('commit 先持久化 WAL 再合并快照(WAL 始终领先)', async () => { // 通过顺序断言:事务 INSERT 的 WAL 记录必须在快照合并可见之前已落盘 const engine = new AriaEngine({ storageBackend: 'memory' }); await engine.open(uniqueDB(), 1); await engine.createTable(createSchema('t', { id: { type: 'string', primaryKey: true } })); await engine.beginTransaction(); await engine.insert('t', [{ id: '1' }]); // 快照合并前:WAL 已含事务记录(batch 模式 flush 时机校验) await engine.commitTransaction(); // 崩溃恢复路径:WAL 完整则恢复数据 const backend = (engine as any).backend; const walKeys = (await backend.listKeys()).filter((k) => k.startsWith('__wal_')); expect(walKeys.length).toBeGreaterThan(0); await engine.close(); }); it('OPFS close 等待挂起写完成(文件完整)', async () => { // mock OPFS(跨实例共享) const files = new Map(); const dirMock = { getDirectoryHandle: async (_n: string, _o?: any) => dirMock as any, getFileHandle: async (name: string, opts?: any) => { if (opts?.create) { return { createWritable: async () => ({ write: async (d: string) => { // 模拟慢 I/O await new Promise((r) => setTimeout(r, 30)); files.set(name, d); }, close: async () => {}, }), }; } if (!files.has(name)) throw new Error('Not found'); return { getFile: async () => ({ text: async () => files.get(name)!, arrayBuffer: async () => new ArrayBuffer(0) }) }; }, removeEntry: async (n: string) => { files.delete(n); }, }; (dirMock as any).entries = () => ({ [Symbol.asyncIterator]: async function* () { for (const [k] of files) yield [k]; }, }); const nav = (globalThis as any).navigator || {}; nav.storage = { getDirectory: async () => dirMock }; (globalThis as any).navigator = nav; const engine = new OPFSEngine(); await engine.open('opfs-close', 1); await engine.createTable(createSchema('t', { id: { type: 'string', primaryKey: true } })); // 写入不等待(写 I/O 30ms 挂起) const p = engine.insert('t', [{ id: '1' }]); // 立即 close:必须等待挂起写完成 await engine.close(); await p; // 文件完整(修复前 close 不等写 → 可能读到旧/空文件) expect(files.get('t.json')).toContain('"1"'); }); });