diff --git a/CHANGELOG.md b/CHANGELOG.md index 8f4f211..8e37b50 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,19 @@ All notable changes to MetonaSqlark will be documented in this file. ### Fixed +- **页面 id 崩溃回退导致静默丢数据(P0)** — FileManager `allocatePageIds` 在分配页面后、 + `saveMeta` 持久化 nextPageId 前崩溃 → `__aria_meta` 停留在旧水位 → 恢复时页面 id 复用 + (覆盖崩溃前旧页面)→ 恢复期 compaction 按 meta.pageIds 删除"旧"SSTable 时误删被复用的 + 新数据页面 → 崩溃恢复后静默丢 75%(kv 后端 2 万行 + 加密 + 双索引场景复现 5228/20000)。 + 修复:`init` 以"现存最大页面 id + 1"为准(单调不回退,绝不复用已存在页面) +- **LSM.flush 剩余数据与 compaction 并发写 meta(P0)** — `flush()` 对 immutable/memtable + 的剩余落盘直接 await(不在链上),与链上 compaction 并发写 SSTable meta:compaction 产物 + `saveMeta` 后,memtable flush 的 `saveMeta` 读到中间态列表(含 compaction 产物)→ 覆盖产物 + 引用 → compaction 产物变孤儿 → 主表/索引数据静默丢失(kv 后端优雅关闭后重开丢 75%)。 + 修复:剩余落盘挂链串行(memtable flush 在 compaction 之后),meta 无竞态 +- **生产矩阵审计**(13 个组合全量验收):后端(opfs/kv/memory)× 特性(页面化/加密/压缩/ + WAL 开关/同步模式)× 功能(CRUD/双索引/事务/主键变更/级联删除/崩溃恢复)—— + 新增 `tests/engine/aria-matrix-audit.test.ts`,暴露并修复上述 2 个 P0 - **二级索引范围扫描漏读尾部数据(P0)** — SSTable 索引块记录的是"块内最后一个 key" (builder 约定),读端 `locateBlockLE` 却按块首 key 语义二分:范围查询 endKey 落在 某块尾部时,下一块(尾 key 越界但首 key 仍在范围内)被排除 → 二级索引条件查询 diff --git a/src/engine/aria/index/lsm.ts b/src/engine/aria/index/lsm.ts index 9b10df2..5cb50db 100644 --- a/src/engine/aria/index/lsm.ts +++ b/src/engine/aria/index/lsm.ts @@ -573,15 +573,21 @@ export class LSM { } // v0.4.3-fix: 循环等待级联任务(flush 完成可能触发新的 compaction) await this.drainChain(); - // 若仍有 frozen 数据未刷盘,在链尾追加 + // v0.6.1-fix(P0): 剩余数据 flush 必须挂链串行 —— 此前直接 await 执行, + // 与链上 compaction 并发写 SSTable meta:compaction 产物 saveMeta 后, + // memtable flush 的 saveMeta 读到中间态列表(含 compaction 产物)→ 覆盖产物 + // 引用 → compaction 产物变孤儿 → 索引/主表数据静默丢失(优雅关闭后重开丢 75%)。 + // 挂链后按序执行(memtable flush 在 compaction 之后),meta 无竞态。 if (this.immutableMemtable) { const frozen = this.immutableMemtable; - await this.flushImmutableAsync(frozen); + this.flushChain = this.enqueueOnChain(() => this.flushImmutableAsync(frozen)); } if (this.memtable.getEntryCount() > 0) { this.freezeMemtable(); const frozen = this.immutableMemtable; - if (frozen) await this.flushImmutableAsync(frozen); + if (frozen) { + this.flushChain = this.enqueueOnChain(() => this.flushImmutableAsync(frozen)); + } } this.frozenMemtables = []; // 刷盘完成后级联调度可能触发 compaction → 排空到稳定 diff --git a/src/engine/aria/store/file_manager.ts b/src/engine/aria/store/file_manager.ts index 3d91c87..d9cedcd 100644 --- a/src/engine/aria/store/file_manager.ts +++ b/src/engine/aria/store/file_manager.ts @@ -28,11 +28,24 @@ export class FileManager implements PageIO { async init(dbName: string): Promise { this.dbName = dbName; const meta = await this.backend.read('__aria_meta'); + let nextPageId = 1; if (meta && meta instanceof ArrayBuffer && meta.byteLength >= 4) { - const view = new DataView(meta); - this.nextPageId = view.getUint32(0, false); - } else { - this.nextPageId = 1; + nextPageId = new DataView(meta).getUint32(0, false); + } + // v0.6.1-fix(P0): 崩溃一致性 —— allocatePageIds 的 saveMeta 可能在分配 + // 页面后被崩溃中断(nextPageId 回退)。恢复时若按回退值继续分配, + // 页面 id 复用会覆盖旧页面;随后 compaction 按 meta.pageIds 删除"旧"SSTable + // 时误删被复用的新数据 → 崩溃恢复后静默丢数据(kv 后端 2 万行丢 75%)。 + // 修复:以"现存最大页面 id + 1"为准(单调不回退,绝不复用已存在页面)。 + const keys = await this.backend.listKeys(); + for (const k of keys) { + if (k.startsWith('pg_')) { + const id = Number.parseInt(k.slice(3), 10); + if (!Number.isNaN(id) && id + 1 > nextPageId) nextPageId = id + 1; + } + } + this.nextPageId = nextPageId; + if (!meta || !(meta instanceof ArrayBuffer) || meta.byteLength < 4 || nextPageId !== new DataView(meta as ArrayBuffer).getUint32(0, false)) { await this.saveMeta(); } this.metaLoaded = true; diff --git a/tests/engine/aria-matrix-audit.test.ts b/tests/engine/aria-matrix-audit.test.ts new file mode 100644 index 0000000..5bc82a7 --- /dev/null +++ b/tests/engine/aria-matrix-audit.test.ts @@ -0,0 +1,317 @@ +/** + * v0.6.1 生产矩阵审计 — 模式 × 功能全组合验收 + * + * 后端(opfs/kv/memory)× 特性(页面化/加密/压缩/WAL开关/同步模式)× + * 功能(CRUD/二级索引/事务/崩溃恢复/主键变更一致性) + * + * 目标:全部组合生产级可用。任一组合同等断言 —— 数据完整、索引一致、崩溃恢复不丢。 + */ +import { AriaEngine } from '../../src/engine/aria/index'; +import { createSchema } from '../../src/table/schema'; +import { SharedMemoryBackend } from '../../src/engine/kvstore/shared_memory_medium'; +import { installOPFSMock } from '../helpers/opfs-mock'; + +let counter = 0; +function uniqueDB(): string { + return `mx-${Date.now()}-${++counter}-${Math.random().toString(36).slice(2, 6)}`; +} + +const SCHEMA = () => createSchema('big', { + id: { type: 'string', primaryKey: true }, + val: { type: 'number' }, + tag: { type: 'string', index: true }, + grp: { type: 'number', index: true }, + name: { type: 'string' }, +}); + +/** 写入 total 行(tag/grp 双索引) */ +async function seed(engine: AriaEngine, table: string, total: number): Promise { + for (let batch = 0; batch < total / 1000; batch++) { + const rows = [] as Record[]; + for (let i = 0; i < 1000; i++) { + const idx = batch * 1000 + i; + rows.push({ id: `k${idx}`, val: idx, tag: `t${idx % 10}`, grp: idx % 5, name: `U${idx}` }); + } + await engine.insert(table, rows); + } +} + +/** 双索引完整性断言 */ +async function assertIndexes(engine: AriaEngine, table: string, total: number): Promise { + expect(await engine.count(table)).toBe(total); + for (let t = 0; t < 10; t++) { + expect(await engine.find(table, { table, where: { tag: `t${t}` } })).toHaveLength(total / 10); + } + for (let g = 0; g < 5; g++) { + expect(await engine.find(table, { table, where: { grp: g } })).toHaveLength(total / 5); + } +} + +/** 崩溃恢复:关闭后端 → 新引擎重开同库 */ +async function reopen(dbName: string, config: Record): Promise { + const engine2 = new AriaEngine(config as never); + await engine2.open(dbName, 1); + return engine2; +} + +beforeEach(() => { + SharedMemoryBackend.clearRegistry(); + installOPFSMock(new Map()); +}); + +describe('生产矩阵审计 — 后端 × 核心功能', () => { + const coreConfig = (backend: string): Record => ({ + storageBackend: backend, + memtableSizeThreshold: 256 * 1024, + checkpointInterval: 5000, + walSyncMode: 'full', + }); + + it('kv × 3 万行 + 双索引 + 事务 + 崩溃恢复', async () => { + const dbName = uniqueDB(); + const engine = new AriaEngine(coreConfig('kv') as never); + await engine.open(dbName, 1); + await engine.createTable(SCHEMA()); + await seed(engine, 'big', 30000); + // 事务:提交一批更新 + await engine.beginTransaction(); + for (let i = 0; i < 100; i++) { + await engine.update('big', { table: 'big', where: { id: `k${i}` } }, { name: `Tx${i}` }); + } + await engine.commitTransaction(); + await assertIndexes(engine, 'big', 30000); + const row = await engine.find('big', { table: 'big', where: { id: 'k50' } }); + expect(row[0].name).toBe('Tx50'); + // 事务回滚不产生残留 + await engine.beginTransaction(); + for (let i = 1000; i < 1100; i++) { + await engine.update('big', { table: 'big', where: { id: `k${i}` } }, { name: `Rx${i}` }); + } + await engine.rollbackTransaction(); + expect((await engine.find('big', { table: 'big', where: { id: 'k1050' } }))[0].name).toBe('U1050'); + // 崩溃恢复 + await (engine as any).backend.close(); + (engine as any).opened = false; + const engine2 = await reopen(dbName, coreConfig('kv')); + await assertIndexes(engine2, 'big', 30000); + expect((await engine2.find('big', { table: 'big', where: { id: 'k50' } }))[0].name).toBe('Tx50'); + await engine2.close(); + }, 180000); + + it('opfs × 3 万行 + 双索引 + 崩溃恢复', async () => { + const dbName = uniqueDB(); + const engine = new AriaEngine(coreConfig('opfs') as never); + await engine.open(dbName, 1); + await engine.createTable(SCHEMA()); + await seed(engine, 'big', 30000); + await assertIndexes(engine, 'big', 30000); + await (engine as any).backend.close(); + (engine as any).opened = false; + const engine2 = await reopen(dbName, coreConfig('opfs')); + await assertIndexes(engine2, 'big', 30000); + await engine2.close(); + }, 180000); + + it('memory × 2 万行 + 双索引 + 事务回滚', async () => { + const engine = new AriaEngine(coreConfig('memory') as never); + await engine.open(uniqueDB(), 1); + await engine.createTable(SCHEMA()); + await seed(engine, 'big', 20000); + await engine.beginTransaction(); + await engine.update('big', { table: 'big', where: { id: 'k7' } }, { tag: 't9' }); + await engine.rollbackTransaction(); + await assertIndexes(engine, 'big', 20000); + // 更新后索引一致(无脏数据)——k7 从 t7 移到 t9 + await engine.update('big', { table: 'big', where: { id: 'k7' } }, { tag: 't9' }); + expect(await engine.find('big', { table: 'big', where: { tag: 't9' } })).toHaveLength(2001); + expect(await engine.find('big', { table: 'big', where: { tag: 't7' } })).toHaveLength(1999); + await engine.close(); + }, 120000); +}); + +describe('生产矩阵审计 — 特性组合', () => { + it('kv × 加密 × 压缩 × 页面化(默认开)→ 2 万行 + 崩溃恢复', async () => { + const dbName = uniqueDB(); + const cfg = { + storageBackend: 'kv', memtableSizeThreshold: 256 * 1024, checkpointInterval: 5000, + walSyncMode: 'full', compression: true, encryption: { password: 'matrix-secret' }, + }; + const engine = new AriaEngine(cfg as never); + await engine.open(dbName, 1); + await engine.createTable(SCHEMA()); + await seed(engine, 'big', 20000); + await assertIndexes(engine, 'big', 20000); + await (engine as any).backend.close(); + (engine as any).opened = false; + const engine2 = await reopen(dbName, cfg); + await assertIndexes(engine2, 'big', 20000); + await engine2.close(); + // 错误密码必须拒绝打开 + const bad = new AriaEngine({ ...cfg, encryption: { password: 'wrong' } } as never); + await expect(bad.open(dbName, 1)).rejects.toThrow(); + }, 180000); + + it('kv × pageStorage:false(整 value)→ 2 万行 + 崩溃恢复', async () => { + const dbName = uniqueDB(); + const cfg = { + storageBackend: 'kv', memtableSizeThreshold: 256 * 1024, checkpointInterval: 5000, + walSyncMode: 'full', pageStorage: false, + }; + const engine = new AriaEngine(cfg as never); + await engine.open(dbName, 1); + await engine.createTable(SCHEMA()); + await seed(engine, 'big', 20000); + await assertIndexes(engine, 'big', 20000); + await (engine as any).backend.close(); + (engine as any).opened = false; + const engine2 = await reopen(dbName, cfg); + await assertIndexes(engine2, 'big', 20000); + await engine2.close(); + }, 180000); + + it('opfs × 加密 × 压缩 × 页面化全开 → 2 万行 + 崩溃恢复', async () => { + const dbName = uniqueDB(); + const cfg = { + storageBackend: 'opfs', memtableSizeThreshold: 256 * 1024, checkpointInterval: 5000, + walSyncMode: 'full', compression: true, encryption: { password: 'matrix-secret' }, + pageStorage: true, + }; + const engine = new AriaEngine(cfg as never); + await engine.open(dbName, 1); + await engine.createTable(SCHEMA()); + await seed(engine, 'big', 20000); + await assertIndexes(engine, 'big', 20000); + await (engine as any).backend.close(); + (engine as any).opened = false; + const engine2 = await reopen(dbName, cfg); + await assertIndexes(engine2, 'big', 20000); + await engine2.close(); + }, 180000); + + it('opfs × walEnabled:false → 写入 + 优雅关闭后重开完整', async () => { + const dbName = uniqueDB(); + const cfg = { + storageBackend: 'opfs', memtableSizeThreshold: 128 * 1024, checkpointInterval: 2000, + walEnabled: false, pageStorage: false, + }; + const engine = new AriaEngine(cfg as never); + await engine.open(dbName, 1); + await engine.createTable(SCHEMA()); + await seed(engine, 'big', 20000); + await assertIndexes(engine, 'big', 20000); + // 优雅关闭:close 等待后台链 + checkpoint 落盘 → 重开完整 + await engine.close(); + const engine2 = await reopen(dbName, cfg); + await assertIndexes(engine2, 'big', 20000); + await engine2.close(); + }, 180000); + + it('kv × walEnabled:false → 写入 + 优雅关闭后重开完整', async () => { + const dbName = uniqueDB(); + const cfg = { + storageBackend: 'kv', memtableSizeThreshold: 128 * 1024, checkpointInterval: 2000, + walEnabled: false, pageStorage: false, + }; + const engine = new AriaEngine(cfg as never); + await engine.open(dbName, 1); + await engine.createTable(SCHEMA()); + await seed(engine, 'big', 20000); + await assertIndexes(engine, 'big', 20000); + await engine.close(); + const engine2 = await reopen(dbName, cfg); + await assertIndexes(engine2, 'big', 20000); + await engine2.close(); + }, 180000); + + it('kv × walSyncMode:batch → 优雅关闭后重开完整(崩溃保底已 checkpoint 数据)', async () => { + const dbName = uniqueDB(); + const cfg = { + storageBackend: 'kv', memtableSizeThreshold: 256 * 1024, checkpointInterval: 3000, + walSyncMode: 'batch', + }; + const engine = new AriaEngine(cfg as never); + await engine.open(dbName, 1); + await engine.createTable(SCHEMA()); + await seed(engine, 'big', 20000); + await assertIndexes(engine, 'big', 20000); + await engine.close(); + const engine2 = await reopen(dbName, cfg); + await assertIndexes(engine2, 'big', 20000); + await engine2.close(); + }, 180000); + + it('opfs × walSyncMode:none → 优雅关闭后重开完整(checkpoint 兜底)', async () => { + const dbName = uniqueDB(); + const cfg = { + storageBackend: 'opfs', memtableSizeThreshold: 128 * 1024, checkpointInterval: 3000, + walSyncMode: 'none', pageStorage: false, + }; + const engine = new AriaEngine(cfg as never); + await engine.open(dbName, 1); + await engine.createTable(SCHEMA()); + await seed(engine, 'big', 20000); + await assertIndexes(engine, 'big', 20000); + await engine.close(); + const engine2 = await reopen(dbName, cfg); + await assertIndexes(engine2, 'big', 20000); + await engine2.close(); + }, 180000); +}); + +describe('生产矩阵审计 — 主键变更索引一致性(三后端)', () => { + it('kv × 主键变更:新旧索引无脏数据', async () => { + const engine = new AriaEngine({ storageBackend: 'kv', memtableSizeThreshold: 256 * 1024, checkpointInterval: 100000 } as never); + await engine.open(uniqueDB(), 1); + await engine.createTable(SCHEMA()); + await engine.insert('big', [{ id: 'a1', val: 1, tag: 'x', grp: 1, name: 'A' }]); + // 主键 a1 → b1(tag/grp 不变) + await engine.update('big', { table: 'big', where: { id: 'a1' } }, { id: 'b1' }); + expect(await engine.count('big')).toBe(1); + expect(await engine.find('big', { table: 'big', where: { tag: 'x' } })).toHaveLength(1); + expect(await engine.find('big', { table: 'big', where: { id: 'a1' } })).toHaveLength(0); + await engine.close(); + }, 60000); + + it('opfs × 主键变更 + 崩溃恢复:索引一致', async () => { + const dbName = uniqueDB(); + const engine = new AriaEngine({ storageBackend: 'opfs', memtableSizeThreshold: 256 * 1024, checkpointInterval: 100000, pageStorage: false } as never); + await engine.open(dbName, 1); + await engine.createTable(SCHEMA()); + for (let i = 0; i < 5000; i++) { + await engine.insert('big', [{ id: `k${i}`, val: i, tag: `t${i % 10}`, grp: i % 5, name: `U${i}` }]); + } + for (let i = 0; i < 5000; i += 10) { + await engine.update('big', { table: 'big', where: { id: `k${i}` } }, { id: `p${i}` }); + } + await (engine as any).backend.close(); + (engine as any).opened = false; + const engine2 = new AriaEngine({ storageBackend: 'opfs', memtableSizeThreshold: 256 * 1024, checkpointInterval: 100000, pageStorage: false } as never); + await engine2.open(dbName, 1); + expect(await engine2.count('big')).toBe(5000); + for (let t = 0; t < 10; t++) { + expect(await engine2.find('big', { table: 'big', where: { tag: `t${t}` } })).toHaveLength(500); + } + await engine2.close(); + }, 120000); + + it('memory × 级联删除 + 索引清理', async () => { + const engine = new AriaEngine({ storageBackend: 'memory', memtableSizeThreshold: 256 * 1024 } as never); + await engine.open(uniqueDB(), 1); + await engine.createTable(createSchema('parent', { + id: { type: 'string', primaryKey: true }, + tag: { type: 'string', index: true }, + })); + await engine.createTable(createSchema('child', { + id: { type: 'string', primaryKey: true }, + pid: { type: 'string', references: 'parent.id', onDelete: 'CASCADE', index: true }, + })); + for (let i = 0; i < 2000; i++) { + await engine.insert('parent', [{ id: `p${i}`, tag: `t${i % 10}` }]); + await engine.insert('child', [{ id: `c${i}`, pid: `p${i}` }]); + } + await engine.delete('parent', { table: 'parent', where: { id: 'p5' } }); + expect(await engine.count('child')).toBe(1999); + expect(await engine.find('child', { table: 'child', where: { pid: 'p5' } })).toHaveLength(0); + await engine.close(); + }, 60000); +}); diff --git a/tests/zz-dbg.test.ts b/tests/zz-dbg.test.ts new file mode 100644 index 0000000..e72f9c1 --- /dev/null +++ b/tests/zz-dbg.test.ts @@ -0,0 +1,24 @@ +import { AriaEngine } from '../src/engine/aria/index'; +import { createSchema } from '../src/table/schema'; + +it('dbg: call fk directly', async () => { + const engine = new AriaEngine({ storageBackend: 'memory' } as never); + await engine.open('dbg-fk2', 1); + await engine.createTable(createSchema('parent', { + id: { type: 'string', primaryKey: true }, + })); + await engine.createTable(createSchema('child', { + id: { type: 'string', primaryKey: true }, + pid: { type: 'string', references: 'parent:id', onDelete: 'CASCADE' }, + })); + await engine.insert('parent', [{ id: 'p1' }]); + await engine.insert('child', [{ id: 'c1', pid: 'p1' }]); + const wal: unknown[] = []; + const n = await (engine as any).applyForeignKeyRules('parent', 'p1', wal, new Set()); + console.log('fk direct result:', n, 'wal records:', wal.length); + console.log('child count after fk:', await engine.count('child')); + // 手动 CASCADE 验证 + const c = await (engine as any).lsm.get('child:c1'); + console.log('child:c1 in lsm:', c ? 'YES' : 'NO'); + await engine.close(); +});