MetonaSqlark/tests/engine/aria-prod-load.test.ts

/**
 * AriaEngine — 生产负载验证（正确性优先，规模可完成）
 *
 * 覆盖 README 宣称的核心能力在生产负载下的正确性：
 * 1. 5 万行写入（多级 Compaction）→ 完整查询 → 崩溃恢复
 * 2. 高频更新/删除（Compaction 回收墓碑）→ 重启后一致
 * 3. 大 value（100KB×50）→ 编码/压缩/恢复
 * 4. 混合操作 + 崩溃 → 已确认写入零丢失
 * 5. 大量删除（90%）+ Compaction → 重启无残留
 * 6. kv 后端 5 万行（页面化路径）
 *
 * 注：10 万级 kv 后端性能专项见 CHANGELOG v0.6.1 待办（KVStore 日志增长优化）。
 */
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 `pload-${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 },
  name: { type: 'string' },
});

beforeEach(() => {
  SharedMemoryBackend.clearRegistry();
  installOPFSMock(new Map());
});

describe('AriaEngine — 生产负载验证', () => {
  it('5 万行写入（多级 Compaction + 页面化）→ 完整查询 → 崩溃恢复（OPFS 后端）', async () => {
    const dbName = uniqueDB();
    const engine = new AriaEngine({
      storageBackend: 'opfs',
      memtableSizeThreshold: 256 * 1024,
      checkpointInterval: 2000,
      walSyncMode: 'full',
    });
    await engine.open(dbName, 1);
    await engine.createTable(SCHEMA());

    const TOTAL = 50000;
    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}`, name: `User${idx}` });
      }
      await engine.insert('big', rows);
    }
    expect(await engine.count('big')).toBe(TOTAL);

    // 多级 compaction（VACUUM 语义）
    await (engine as any).lsm.flush();
    for (let level = 0; level < 4; level++) {
      await (engine as any).lsm.compactLevel(level);
    }
    const stats = (engine as any).lsm.getStats();
    expect((stats.levelCounts as number[]).reduce((a: number, b: number) => a + b, 0)).toBeGreaterThanOrEqual(1);

    // 完整查询
    expect((await engine.find('big', { table: 'big' }))).toHaveLength(TOTAL);

    // 崩溃恢复
    await (engine as any).backend.close();
    (engine as any).opened = false;
    const engine2 = new AriaEngine({
      storageBackend: 'opfs',
      memtableSizeThreshold: 256 * 1024,
      checkpointInterval: 2000,
      walSyncMode: 'full',
    });
    await engine2.open(dbName, 1);
    expect(await engine2.count('big')).toBe(TOTAL);
    for (const id of ['k0', 'k25000', 'k49999']) {
      expect(await engine2.find('big', { table: 'big', where: { id } })).toHaveLength(1);
    }
    // 索引（重启重建）
    expect(await engine2.find('big', { table: 'big', where: { tag: 't5' } })).toHaveLength(5000);
    await engine2.close();
  }, 600000);

  it('高频更新/删除（Compaction 回收墓碑）→ 重启后一致', async () => {
    const dbName = uniqueDB();
    const engine = new AriaEngine({
      storageBackend: 'opfs',
      memtableSizeThreshold: 128 * 1024,
      checkpointInterval: 1000,
      walSyncMode: 'full',
    });
    await engine.open(dbName, 1);
    await engine.createTable(SCHEMA());

    const rows = [] as Record<string, unknown>[];
    for (let i = 0; i < 10000; i++) rows.push({ id: `k${i}`, val: i, tag: `t${i % 5}` });
    await engine.insert('big', rows);

    let seed = 7;
    const rand = () => { seed = (seed * 1103515245 + 12345) & 0x7fffffff; return seed / 0x7fffffff; };
    for (let i = 0; i < 2000; i++) {
      const id = `k${Math.floor(rand() * 10000)}`;
      if (rand() < 0.5) {
        await engine.update('big', { table: 'big', where: { id } }, { val: Math.floor(rand() * 1e9) });
      } else {
        await engine.delete('big', { table: 'big', where: { id } });
      }
    }
    await (engine as any).lsm.flush();
    await (engine as any).lsm.compactLevel(0);

    const count = await engine.count('big');
    expect(count).toBeGreaterThan(0);
    await engine.close();

    const engine2 = new AriaEngine({
      storageBackend: 'opfs',
      memtableSizeThreshold: 128 * 1024,
      checkpointInterval: 1000,
      walSyncMode: 'full',
    });
    await engine2.open(dbName, 1);
    expect(await engine2.count('big')).toBe(count);
    await engine2.close();
  }, 600000);

  it('大 value（100KB × 50）压缩写入/恢复完整', async () => {
    const dbName = uniqueDB();
    const engine = new AriaEngine({
      storageBackend: 'opfs',
      compression: true,
      memtableSizeThreshold: 512 * 1024,
      checkpointInterval: 5000,
      walSyncMode: 'full',
    });
    await engine.open(dbName, 1);
    await engine.createTable(createSchema('docs', {
      id: { type: 'string', primaryKey: true },
      body: { type: 'string' },
    }));
    const chunk = '这是大段生产数据内容。'.repeat(5000); // ~100KB
    for (let i = 0; i < 50; i++) {
      await engine.insert('docs', [{ id: `d${i}`, body: chunk }]);
    }
    await engine.close();

    const engine2 = new AriaEngine({
      storageBackend: 'opfs',
      compression: true,
      memtableSizeThreshold: 512 * 1024,
      checkpointInterval: 5000,
      walSyncMode: 'full',
    });
    await engine2.open(dbName, 1);
    expect(await engine2.count('docs')).toBe(50);
    const one = await engine2.find('docs', { table: 'docs', where: { id: 'd25' } });
    expect((one[0].body as string).length).toBe(chunk.length);
    await engine2.close();
  }, 600000);

  it('混合操作 + 崩溃：已确认写入零丢失（20000 操作）', async () => {
    const dbName = uniqueDB();
    const engine = new AriaEngine({
      storageBackend: 'opfs',
      memtableSizeThreshold: 128 * 1024,
      checkpointInterval: 1000,
      walSyncMode: 'full',
    });
    await engine.open(dbName, 1);
    await engine.createTable(SCHEMA());

    let seed = 123;
    const rand = () => { seed = (seed * 1103515245 + 12345) & 0x7fffffff; return seed / 0x7fffffff; };
    const confirmed = new Map<string, { val: number; tag: string }>();
    for (let i = 0; i < 20000; i++) {
      const r = rand();
      const id = `k${Math.floor(rand() * 10000)}`;
      if (r < 0.5) {
        const row = { id, val: Math.floor(rand() * 1e9), tag: `t${Math.floor(rand() * 5)}` };
        try {
          await engine.insert('big', [row]);
          confirmed.set(id, row);
        } catch (e) {
          if ((e as { code?: string }).code !== 'DUPLICATE_KEY') throw e;
        }
      } else if (r < 0.8) {
        const val = Math.floor(rand() * 1e9);
        await engine.update('big', { table: 'big', where: { id } }, { val });
        if (confirmed.has(id)) confirmed.set(id, { ...confirmed.get(id)!, val });
      } else {
        await engine.delete('big', { table: 'big', where: { id } });
        confirmed.delete(id);
      }
    }

    await (engine as any).backend.close();
    (engine as any).opened = false;

    const engine2 = new AriaEngine({
      storageBackend: 'opfs',
      memtableSizeThreshold: 128 * 1024,
      checkpointInterval: 1000,
      walSyncMode: 'full',
    });
    await engine2.open(dbName, 1);
    expect(await engine2.count('big')).toBe(confirmed.size);
    let sampled = 0;
    for (const [id, expected] of confirmed) {
      if (sampled++ > 1000) break;
      const rows = await engine2.find('big', { table: 'big', where: { id } });
      expect(rows).toHaveLength(1);
      expect(rows[0].val).toBe(expected.val);
    }
    await engine2.close();
  }, 600000);

  it('大量删除（90% 行）+ Compaction → 重启无残留（墓碑清理）', async () => {
    const dbName = uniqueDB();
    const engine = new AriaEngine({
      storageBackend: 'opfs',
      memtableSizeThreshold: 128 * 1024,
      checkpointInterval: 1000,
      walSyncMode: 'full',
    });
    await engine.open(dbName, 1);
    await engine.createTable(SCHEMA());
    const rows = [] as Record<string, unknown>[];
    for (let i = 0; i < 10000; i++) rows.push({ id: `k${i}`, val: i, tag: `t${i % 5}` });
    await engine.insert('big', rows);

    const toDelete = [] as string[];
    for (let i = 1000; i < 10000; i++) toDelete.push(`k${i}`);
    await engine.delete('big', { table: 'big', where: { id: { $in: toDelete } } });
    expect(await engine.count('big')).toBe(1000);

    await (engine as any).lsm.flush();
    for (let l = 0; l < 4; l++) await (engine as any).lsm.compactLevel(l);
    await engine.close();

    const engine2 = new AriaEngine({
      storageBackend: 'opfs',
      memtableSizeThreshold: 128 * 1024,
      checkpointInterval: 1000,
      walSyncMode: 'full',
    });
    await engine2.open(dbName, 1);
    expect(await engine2.count('big')).toBe(1000);
    const all = await engine2.find('big', { table: 'big' });
    expect(all.every((r) => Number(String(r.id).slice(1)) < 1000)).toBe(true);
    await engine2.close();
  }, 600000);

  it('kv 后端 5 万行（页面化路径）：写入 → 崩溃 → 恢复完整', async () => {
    const dbName = uniqueDB();
    const engine = new AriaEngine({
      storageBackend: 'kv',
      memtableSizeThreshold: 512 * 1024,
      checkpointInterval: 5000,
      walSyncMode: 'full',
    });
    await engine.open(dbName, 1);
    await engine.createTable(SCHEMA());

    const TOTAL = 50000;
    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}`, name: `User${idx}` });
      }
      await engine.insert('big', rows);
    }
    expect(await engine.count('big')).toBe(TOTAL);
    await (engine as any).backend.close();
    (engine as any).opened = false;

    const engine2 = new AriaEngine({
      storageBackend: 'kv',
      memtableSizeThreshold: 512 * 1024,
      checkpointInterval: 5000,
      walSyncMode: 'full',
    });
    await engine2.open(dbName, 1);
    expect(await engine2.count('big')).toBe(TOTAL);
    expect(await engine2.find('big', { table: 'big', where: { tag: 't3' } })).toHaveLength(5000);
    await engine2.close();
  }, 600000);

  it('10 万行 kv 后端（含索引）：完整查询 + 崩溃恢复（v0.6.1-perf 回归）', async () => {
    const dbName = uniqueDB();
    const engine = new AriaEngine({
      storageBackend: 'kv',
      memtableSizeThreshold: 512 * 1024,
      checkpointInterval: 30000,
      walSyncMode: 'full',
    });
    await engine.open(dbName, 1);
    await engine.createTable(SCHEMA());

    const TOTAL = 100000;
    const t0 = Date.now();
    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}`, name: `User${idx}` });
      }
      await engine.insert('big', rows);
    }
    const insertMs = Date.now() - t0;
    // 性能护栏：修复前 353s（batch 32 起每批 8~11s 性能悬崖），
    // 修复后本机 ~12.5s。CI（debian runner 慢 2~3 倍、重型套件串行）下
    // 健康耗时约 30~80s；护栏放宽到 240s —— 仍能拦截性能悬崖回归（353s >> 240s），
    // 不误报健康慢环境。
    console.log(`10万行 kv 插入耗时: ${insertMs}ms`);
    expect(insertMs).toBeLessThan(240000);
    expect(await engine.count('big')).toBe(TOTAL);

    // 全部 10 个 tag 索引查询完整
    for (let t = 0; t < 10; t++) {
      const viaIdx = await engine.find('big', { table: 'big', where: { tag: `t${t}` } });
      expect(viaIdx.length).toBe(10000);
    }

    // 崩溃恢复
    await (engine as any).backend.close();
    (engine as any).opened = false;
    const engine2 = new AriaEngine({
      storageBackend: 'kv',
      memtableSizeThreshold: 512 * 1024,
      checkpointInterval: 30000,
      walSyncMode: 'full',
    });
    await engine2.open(dbName, 1);
    expect(await engine2.count('big')).toBe(TOTAL);
    expect(await engine2.find('big', { table: 'big', where: { tag: 't7' } })).toHaveLength(10000);
    await engine2.close();
  }, 600000);

  it('10 万行 opfs 后端（含索引）：完整查询（v0.6.1-perf 回归）', async () => {
    const engine = new AriaEngine({
      storageBackend: 'opfs',
      memtableSizeThreshold: 512 * 1024,
      checkpointInterval: 30000,
      walSyncMode: 'full',
    });
    await engine.open(uniqueDB(), 1);
    await engine.createTable(SCHEMA());

    const TOTAL = 100000;
    const t0 = Date.now();
    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}`, name: `User${idx}` });
      }
      await engine.insert('big', rows);
    }
    const insertMs = Date.now() - t0;
    // 同上：CI 慢环境护栏放宽（本机 ~25s；悬崖回归仍会被拦截）
    console.log(`10万行 opfs 插入耗时: ${insertMs}ms`);
    expect(insertMs).toBeLessThan(300000);
    expect(await engine.count('big')).toBe(TOTAL);
    for (let t = 0; t < 10; t++) {
      const viaIdx = await engine.find('big', { table: 'big', where: { tag: `t${t}` } });
      expect(viaIdx.length).toBe(10000);
    }
    await engine.close();
  }, 600000);
});