196 lines
8.2 KiB
TypeScript
196 lines
8.2 KiB
TypeScript
/**
|
||
* AriaEngine — SSTable CRC-32 损坏检测集成测试
|
||
*
|
||
* 覆盖：
|
||
* 1. 打开时整文件 CRC 校验失败 → 损坏 SSTable 被清理（自愈），打开不阻塞
|
||
* 2. 运行期预加载 CRC 校验失败 → 损坏文件不缓存并被清理
|
||
* 3. 旧版无校验文件（checksum=0）在 LSM 层仍正常加载
|
||
*/
|
||
import { AriaEngine } from '../../src/engine/aria/index';
|
||
import { createSchema } from '../../src/table/schema';
|
||
|
||
import { installOPFSMock } from '../helpers/opfs-mock';
|
||
|
||
beforeEach(() => { installOPFSMock(new Map()); });
|
||
|
||
let idbCounter = 0;
|
||
function uniqueDB(): string {
|
||
return `crc-${Date.now()}-${++idbCounter}-${Math.random().toString(36).slice(2, 8)}`;
|
||
}
|
||
|
||
const SCHEMA = () => createSchema('users', {
|
||
id: { type: 'string', primaryKey: true },
|
||
name: { type: 'string' },
|
||
age: { type: 'number' },
|
||
});
|
||
|
||
/** 篡改指定 pg_ 页面文件的一个字节 */
|
||
async function corruptKey(engine: AriaEngine, key: string, byteOffset: number): Promise<void> {
|
||
const backend = (engine as any).backend;
|
||
const raw = await backend.read(key);
|
||
expect(raw).not.toBeNull();
|
||
const buf = new Uint8Array(raw as ArrayBuffer);
|
||
buf[Math.min(byteOffset, buf.length - 1)] ^= 0xff;
|
||
await backend.write(key, buf.buffer as ArrayBuffer);
|
||
}
|
||
|
||
/** 列出全部 SSTable 的页面文件 key（页面化存储：pg_ 前缀） */
|
||
async function listSSTKeys(engine: AriaEngine): Promise<string[]> {
|
||
const backend = (engine as any).backend;
|
||
const keys = (await backend.listKeys()) as string[];
|
||
return keys.filter((k) => k.startsWith('pg_'));
|
||
}
|
||
|
||
/** 读取主 LSM 的 SSTable meta 列表 */
|
||
async function listSSTMetas(engine: AriaEngine): Promise<{ id: number; pageIds?: number[] }[]> {
|
||
const backend = (engine as any).backend;
|
||
const raw = await backend.read('__aria_lsm_meta');
|
||
if (!raw) return [];
|
||
return JSON.parse(new TextDecoder().decode(raw)) as { id: number; pageIds?: number[] }[];
|
||
}
|
||
|
||
describe('AriaEngine — SSTable CRC 损坏检测（集成）', () => {
|
||
it('打开时 CRC 损坏的 SSTable 被清理，其余数据可读', async () => {
|
||
const engine = new AriaEngine({ storageBackend: 'opfs', memtableSizeThreshold: 64 * 1024 * 1024 });
|
||
const dbName = uniqueDB();
|
||
await engine.open(dbName, 1);
|
||
await engine.createTable(SCHEMA());
|
||
|
||
// 写入 200 行（触发多次 flush → 多个 SSTable）
|
||
for (let batch = 0; batch < 3; batch++) {
|
||
const rows = [] as Record<string, unknown>[];
|
||
for (let i = 0; i < 50; i++) {
|
||
rows.push({ id: `u-${batch * 50 + i}`, name: `User${batch * 50 + i}`, age: batch * 50 + i });
|
||
}
|
||
await engine.insert('users', rows);
|
||
await (engine as any).lsm.flush();
|
||
}
|
||
await engine.close();
|
||
|
||
// 重新打开，篡改第一个 SSTable 文件的数据区
|
||
const engine2 = new AriaEngine({ storageBackend: 'opfs', memtableSizeThreshold: 64 * 1024 * 1024 });
|
||
await engine2.open(dbName, 1);
|
||
const sstKeys = await listSSTKeys(engine2);
|
||
expect(sstKeys.length).toBeGreaterThanOrEqual(2);
|
||
await corruptKey(engine2, sstKeys[0], 64);
|
||
|
||
// 再次打开：损坏文件应被跳过并清理，打开不抛错
|
||
await engine2.close();
|
||
const engine3 = new AriaEngine({ storageBackend: 'opfs', memtableSizeThreshold: 64 * 1024 * 1024 });
|
||
await engine3.open(dbName, 1);
|
||
|
||
// 剩余未损坏文件的数据应可查询
|
||
const remaining = await engine3.find('users', { table: 'users' });
|
||
expect(remaining.length).toBeGreaterThan(0);
|
||
expect(remaining.length).toBeLessThan(200);
|
||
|
||
// 损坏文件已被清理（页面 + meta 移除）
|
||
const afterKeys = await listSSTKeys(engine3);
|
||
expect(afterKeys).not.toContain(sstKeys[0]);
|
||
const metas = await listSSTMetas(engine3);
|
||
// 被篡改页面所属的 SSTable meta 应被移除
|
||
const victimMetas = await listSSTMetas(engine2);
|
||
const victimMeta = victimMetas.find((m) => m.pageIds?.includes(Number(sstKeys[0].slice(3))));
|
||
expect(metas.some((m) => m.id === victimMeta?.id)).toBe(false);
|
||
|
||
await engine3.close();
|
||
});
|
||
|
||
it('打开时损坏全部 SSTable → 库仍可打开，数据为空但不崩溃', async () => {
|
||
const engine = new AriaEngine({ storageBackend: 'opfs', memtableSizeThreshold: 64 * 1024 * 1024 });
|
||
const dbName = uniqueDB();
|
||
await engine.open(dbName, 1);
|
||
await engine.createTable(SCHEMA());
|
||
for (let batch = 0; batch < 2; batch++) {
|
||
const rows = [] as Record<string, unknown>[];
|
||
for (let i = 0; i < 50; i++) {
|
||
rows.push({ id: `u-${batch * 50 + i}`, name: `User${batch * 50 + i}`, age: batch * 50 + i });
|
||
}
|
||
await engine.insert('users', rows);
|
||
await (engine as any).lsm.flush();
|
||
}
|
||
await engine.close();
|
||
|
||
const engine2 = new AriaEngine({ storageBackend: 'opfs', memtableSizeThreshold: 64 * 1024 * 1024 });
|
||
await engine2.open(dbName, 1);
|
||
for (const key of await listSSTKeys(engine2)) {
|
||
await corruptKey(engine2, key, 16);
|
||
}
|
||
await engine2.close();
|
||
|
||
const engine3 = new AriaEngine({ storageBackend: 'opfs', memtableSizeThreshold: 64 * 1024 * 1024 });
|
||
// 不应抛 ARIA_OPEN_ERROR
|
||
await engine3.open(dbName, 1);
|
||
const rows3 = await engine3.find('users', { table: 'users' });
|
||
expect(rows3.length).toBe(0);
|
||
await engine3.close();
|
||
});
|
||
|
||
it('运行期 CRC 损坏 → 预加载不缓存损坏文件并清理（自愈）', async () => {
|
||
const engine = new AriaEngine({ storageBackend: 'opfs', memtableSizeThreshold: 64 * 1024 * 1024 });
|
||
const dbName = uniqueDB();
|
||
await engine.open(dbName, 1);
|
||
await engine.createTable(SCHEMA());
|
||
for (let batch = 0; batch < 2; batch++) {
|
||
const rows = [] as Record<string, unknown>[];
|
||
for (let i = 0; i < 50; i++) {
|
||
rows.push({ id: `u-${batch * 50 + i}`, name: `User${batch * 50 + i}`, age: batch * 50 + i });
|
||
}
|
||
await engine.insert('users', rows);
|
||
await (engine as any).lsm.flush();
|
||
}
|
||
|
||
// 运行期：直接篡改后端中的某个 SSTable 文件（不触发打开校验路径）
|
||
const sstKeys = await listSSTKeys(engine);
|
||
expect(sstKeys.length).toBeGreaterThanOrEqual(2);
|
||
const victim = sstKeys[sstKeys.length - 1]; // 篡改最新（memtable 已 flush 后的文件）
|
||
await corruptKey(engine, victim, 100);
|
||
|
||
// 缓存里已有该文件（flush 时缓存）→ 清 LSM 层与 BufferPool 页面缓存模拟运行期磁盘损坏
|
||
const lsm = (engine as any).lsm;
|
||
lsm.sstableCache.clear();
|
||
await (engine as any).bufferPool.clear();
|
||
|
||
// 查询触发 prefetchRange → preloadSSTable 发现 CRC 失败 → 清理 + 不缓存
|
||
const remaining = await engine.find('users', { table: 'users' });
|
||
expect(remaining.length).toBeLessThan(100);
|
||
expect(lsm.sstableCache.has(Number(victim.slice(3)))).toBe(false);
|
||
const metas = await lsm.sstableStore.listMeta();
|
||
expect(metas.some((m: { id: number }) => m.id === Number(victim.slice(3)))).toBe(false);
|
||
|
||
await engine.close();
|
||
});
|
||
|
||
it('旧版无校验文件（checksum=0）在 LSM 中正常加载', async () => {
|
||
const engine = new AriaEngine({ storageBackend: 'opfs', memtableSizeThreshold: 64 * 1024 * 1024 });
|
||
const dbName = uniqueDB();
|
||
await engine.open(dbName, 1);
|
||
await engine.createTable(SCHEMA());
|
||
await engine.insert('users', [
|
||
{ id: 'a', name: 'Alice', age: 30 },
|
||
{ id: 'b', name: 'Bob', age: 25 },
|
||
]);
|
||
await (engine as any).lsm.flush();
|
||
await engine.close();
|
||
|
||
const engine2 = new AriaEngine({ storageBackend: 'opfs', memtableSizeThreshold: 64 * 1024 * 1024 });
|
||
await engine2.open(dbName, 1);
|
||
const sstKeys = await listSSTKeys(engine2);
|
||
expect(sstKeys.length).toBe(1);
|
||
const backend = (engine2 as any).backend;
|
||
const raw = await backend.read(sstKeys[0]);
|
||
const buf = new Uint8Array(raw as ArrayBuffer);
|
||
new DataView(buf.buffer).setUint32(buf.byteLength - 4, 0, false); // checksum 清零 → 旧版
|
||
await backend.write(sstKeys[0], buf.buffer as ArrayBuffer);
|
||
await engine2.close();
|
||
|
||
// 重开：checksum=0 跳过校验，数据完整可读
|
||
const engine3 = new AriaEngine({ storageBackend: 'opfs', memtableSizeThreshold: 64 * 1024 * 1024 });
|
||
await engine3.open(dbName, 1);
|
||
const rows3 = await engine3.find('users', { table: 'users' });
|
||
expect(rows3.length).toBe(2);
|
||
expect(rows3.map((r) => r.id).sort()).toEqual(['a', 'b']);
|
||
await engine3.close();
|
||
});
|
||
});
|