MetonaSqlark/tests/engine/aria-repair-hardening.test.ts

/**
 * AriaEngine — repair 自愈增强 + 随机操作压力测试
 *
 * 覆盖：
 * 1. repair 清理孤儿页面（meta 未引用的 pg_ 文件）
 * 2. repair 清理 OPFS 残留临时文件
 * 3. WAL 空洞 + repair → 截断清空（不再重放错位数据）
 * 4. 随机操作压力：insert/update/delete + 模拟崩溃（不 checkpoint 断开）→ 重开全量验证
 * 5. 随机操作 + 页面化 + 模拟崩溃 → 重开验证
 * 6. repair 幂等
 */
import { AriaEngine } from '../../src/engine/aria/index';
import { createSchema } from '../../src/table/schema';
import { resetOPFSMock } from '../helpers/storage-harness';

beforeEach(() => { resetOPFSMock(); });

let idbCounter = 0;
function uniqueDB(): string {
  return `rph-${Date.now()}-${++idbCounter}-${Math.random().toString(36).slice(2, 8)}`;
}

const SCHEMA = () => createSchema('items', {
  id: { type: 'string', primaryKey: true },
  val: { type: 'number' },
  tag: { type: 'string', index: true },
});

// ===================================================================
//  repair 增强
// ===================================================================
describe('AriaEngine — repair 自愈增强', () => {
  it('清理孤儿页面（meta 未引用的 pg_ 文件）', async () => {
    const dbName = uniqueDB();
    const engine = new AriaEngine({
      storageBackend: 'opfs',
      pageStorage: true, // 强制页面化（IDB 上也启用，便于断言页面文件）
      memtableSizeThreshold: 64 * 1024 * 1024,
      checkpointInterval: 100000,
    });
    await engine.open(dbName, 1);
    await engine.createTable(SCHEMA());
    await engine.insert('items', [{ id: 'a', val: 1, tag: 'x' }]);
    await (engine as any).lsm.flush();

    const backend = (engine as any).backend;
    const keys = await backend.listKeys();
    const pgKeys = keys.filter((k: string) => k.startsWith('pg_'));
    expect(pgKeys.length).toBeGreaterThan(0);

    // 伪造孤儿页面（模拟崩溃中断的删除流程）
    await backend.write('pg_999999', new Uint8Array(4096).buffer);
    await backend.write('pg_999998', new Uint8Array(4096).buffer);
    expect((await backend.listKeys()).filter((k: string) => k.startsWith('pg_99999'))).toHaveLength(2);

    await (engine as any).repair();

    // 孤儿页面被清理，正常页面保留
    const after = await backend.listKeys();
    expect(after).not.toContain('pg_999999');
    expect(after).not.toContain('pg_999998');
    expect(await engine.count('items')).toBe(1);
    await engine.close();
  });

  it('清理 OPFS 残留临时文件（.crswap/.tmp）', async () => {
    // 用 OPFS mock 后端验证 cleanupStaleFiles 被调用
    const opfs = resetOPFSMock();
    const engine = new AriaEngine({ storageBackend: 'opfs', checkpointInterval: 100000 });
    await engine.open('repair-opfs-1', 1);
    await engine.createTable(SCHEMA());
    await engine.insert('items', [{ id: 'a', val: 1, tag: 'x' }]);

    // 制造残留（模拟 createWritable 中断留下的临时文件）
    await opfs.createFile('junk.crswap');
    await opfs.createFile('junk2.tmp');
    expect((await opfs.listKeys()).some((k) => k.endsWith('.crswap'))).toBe(true);

    await (engine as any).repair();

    const after = await opfs.listKeys();
    expect(after.some((k) => k.endsWith('.crswap'))).toBe(false);
    expect(after.some((k) => k.endsWith('.tmp'))).toBe(false);
    expect(await engine.count('items')).toBe(1);
    await engine.close();
  });

  it('WAL 空洞 + repair → 截断清空（不再重放错位数据）', async () => {
    const dbName = uniqueDB();
    const engine = new AriaEngine({ storageBackend: 'opfs', checkpointInterval: 100000 });
    await engine.open(dbName, 1);
    await engine.createTable(SCHEMA());
    // 数据落盘 + checkpoint 清 WAL
    await engine.insert('items', [{ id: 'a', val: 1, tag: 'x' }]);
    await (engine as any).lsm.flush();
    await (engine as any).wal.checkpoint();
    // 新数据只进 WAL
    await engine.insert('items', [{ id: 'b', val: 2, tag: 'y' }]);
    const backend = (engine as any).backend;
    const keys = await backend.listKeys();
    const walSeg = keys.find((k: string) => k.startsWith('__wal_'));
    // 制造空洞：删掉当前分片（数据仍在内存）
    await backend.delete(walSeg);

    await (engine as any).repair();

    // repair 后 WAL 已清空，无空洞残留
    const after = await backend.listKeys();
    expect(after.some((k: string) => k.startsWith('__wal_'))).toBe(false);
    // 在线自愈：memtable 中的 'b'（已确认写入）随 flush 落盘 → 数据完整不丢
    const rows = await engine.find('items', { table: 'items' });
    expect(rows).toHaveLength(2);
    expect(rows.some((r) => r.id === 'a')).toBe(true);
    expect(rows.some((r) => r.id === 'b')).toBe(true);
    await engine.close();
  });

  it('repair 幂等（连续调用无副作用）', async () => {
    const dbName = uniqueDB();
    const engine = new AriaEngine({ storageBackend: 'opfs', checkpointInterval: 100000 });
    await engine.open(dbName, 1);
    await engine.createTable(SCHEMA());
    await engine.insert('items', [{ id: 'a', val: 1, tag: 'x' }]);
    await (engine as any).repair();
    await (engine as any).repair();
    await (engine as any).repair();
    expect(await engine.count('items')).toBe(1);
    await engine.close();
  });
});

// ===================================================================
//  随机操作压力 + 模拟崩溃
// ===================================================================
describe('AriaEngine — 随机操作压力 + 模拟崩溃', () => {
  it('500 随机操作（insert/update/delete）→ 模拟崩溃 → 重开验证全部已确认写入', async () => {
    const dbName = uniqueDB();
    const engine = new AriaEngine({
      storageBackend: 'opfs',
      memtableSizeThreshold: 8 * 1024,
      checkpointInterval: 50,
      walSyncMode: 'full',
    });
    await engine.open(dbName, 1);
    await engine.createTable(SCHEMA());

    let seed = 12345;
    const rand = () => { seed = (seed * 1103515245 + 12345) & 0x7fffffff; return seed / 0x7fffffff; };

    // 已确认写入集合（模拟应用层已收到成功返回）
    const confirmed = new Map<string, { val: number; tag: string }>();
    for (let i = 0; i < 500; i++) {
      const r = rand();
      const id = `k-${Math.floor(rand() * 100)}`;
      if (r < 0.6) {
        // insert/upsert（随机 id 可能碰撞 → 已存在则跳过，confirmed 保持首个值）
        const row = { id, val: Math.floor(rand() * 1000), tag: `t${Math.floor(rand() * 5)}` };
        try {
          await engine.insert('items', [row]);
          confirmed.set(id, row);
        } catch (e) {
          if ((e as { code?: string }).code !== 'DUPLICATE_KEY') throw e;
        }
      } else if (r < 0.8) {
        // update
        const upd = { val: Math.floor(rand() * 1000) };
        await engine.update('items', { table: 'items', where: { id } }, upd);
        const cur = confirmed.get(id);
        if (cur) confirmed.set(id, { ...cur, ...upd });
      } else {
        // delete
        await engine.delete('items', { table: 'items', where: { id } });
        confirmed.delete(id);
      }
    }

    // 模拟崩溃：不 close（不 checkpoint），直接断开 backend
    await (engine as any).backend.close();
    (engine as any).opened = false;

    // 重开：WAL 重放 + SSTable 加载
    const engine2 = new AriaEngine({ storageBackend: 'opfs', checkpointInterval: 50, walSyncMode: 'full' });
    await engine2.open(dbName, 1);
    const rows = await engine2.find('items', { table: 'items' });
    const byId = new Map(rows.map((r) => [r.id, r]));
    // 全部已确认写入可见
    expect(byId.size).toBe(confirmed.size);
    for (const [id, expected] of confirmed) {
      expect(byId.has(id)).toBe(true);
      expect(byId.get(id)!.val).toBe(expected.val);
    }
    await engine2.close();
  });

  it('随机操作 + 页面化 + 模拟崩溃 → 重开验证', async () => {
    resetOPFSMock();
    const dbName = `rand-opfs-${Date.now()}-${Math.random().toString(36).slice(2, 6)}`;
    const engine = new AriaEngine({
      storageBackend: 'opfs',
      memtableSizeThreshold: 8 * 1024,
      checkpointInterval: 50,
      walSyncMode: 'full',
    });
    await engine.open(dbName, 1);
    await engine.createTable(SCHEMA());

    let seed = 999;
    const rand = () => { seed = (seed * 1103515245 + 12345) & 0x7fffffff; return seed / 0x7fffffff; };
    const confirmed = new Map<string, { val: number; tag: string }>();
    for (let i = 0; i < 300; i++) {
      const r = rand();
      const id = `p-${Math.floor(rand() * 80)}`;
      if (r < 0.6) {
        const row = { id, val: Math.floor(rand() * 500), tag: `tag${Math.floor(rand() * 4)}` };
        try {
          await engine.insert('items', [row]);
          confirmed.set(id, row);
        } catch (e) {
          if ((e as { code?: string }).code !== 'DUPLICATE_KEY') throw e;
        }
      } else if (r < 0.8) {
        const upd = { val: Math.floor(rand() * 500) };
        await engine.update('items', { table: 'items', where: { id } }, upd);
        const cur = confirmed.get(id);
        if (cur) confirmed.set(id, { ...cur, ...upd });
      } else {
        await engine.delete('items', { table: 'items', where: { id } });
        confirmed.delete(id);
      }
    }

    // 模拟崩溃
    await (engine as any).backend.close();
    (engine as any).opened = false;

    const engine2 = new AriaEngine({ storageBackend: 'opfs', checkpointInterval: 50, walSyncMode: 'full' });
    await engine2.open(dbName, 1);
    const rows = await engine2.find('items', { table: 'items' });
    const byId = new Map(rows.map((r) => [r.id, r]));
    expect(byId.size).toBe(confirmed.size);
    for (const [id, expected] of confirmed) {
      expect(byId.get(id)!.val).toBe(expected.val);
    }
    // 索引查询也验证
    const tagged = await engine2.find('items', { table: 'items', where: { tag: 'tag0' } });
    const expectedTagged = Array.from(confirmed.values()).filter((v) => v.tag === 'tag0').length;
    expect(tagged.length).toBe(expectedTagged);
    await engine2.close();
  });
});