fix(P0): 生产矩阵审计暴露 2 个崩溃恢复丢数据 bug — ① FileManager 页面 id 崩溃回退（allocatePageIds 后 saveMeta 前中断 → nextPageId 回退 → 恢复期页面复用被 compaction 误删，kv 2万行+加密崩溃丢75%）：init 以现存最大页面 id+1 为准；② LSM.flush 剩余数据与 compaction 并发写 meta（产物 saveMeta 被覆盖成孤儿 → 优雅关闭重开丢75%）：剩余落盘挂链串行。新增 13 组合矩阵审计（后端×特性×功能全量验收）
This commit is contained in:
thzxx
@@ -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<void> {
|
||||
for (let batch = 0; batch < total / 1000; batch++) {
|
||||
const rows = [] as Record<string, unknown>[];
|
||||
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<void> {
|
||||
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<string, unknown>): Promise<AriaEngine> {
|
||||
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<string, unknown> => ({
|
||||
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);
|
||||
});
|
||||
Reference in New Issue
Block a user