MetonaSqlark/tests/engine/aria-wal-segment.test.ts

/**
 * AriaEngine — SegmentedWALStore 分片 WAL 存储测试
 *
 * 覆盖：
 * 1. append → readAll 往返（分片切换、字节顺序）
 * 2. 空洞检测：序号不连续 → 空洞后的分片丢弃
 * 3. 旧格式 __wal_N + __wal_count 兼容读取
 * 4. truncate / exists
 * 5. 后端无 append 接口时回退 read+write（正确性一致）
 * 6. AriaEngine 集成：高频写入跨重启数据完整（分片 + WAL 重放）
 */
import { SegmentedWALStore } from '../../src/engine/aria/wal/segmented_store';
import { MemoryBackend } from '../../src/engine/aria/store/backend';
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 `seg-${Date.now()}-${++idbCounter}-${Math.random().toString(36).slice(2, 8)}`;
}

const enc = (s: string) => new Uint8Array(new TextEncoder().encode(s));

describe('AriaEngine — SegmentedWALStore 单元', () => {
  it('append → readAll 往返，多批次字节顺序一致', async () => {
    const backend = new MemoryBackend();
    await backend.open('seg-unit-1');
    const store = new SegmentedWALStore(backend, 1024);

    expect(await store.exists()).toBe(false);
    await store.append(enc('AAA'));
    await store.append(enc('BBB'));
    await store.append(enc('CCC'));
    expect(await store.exists()).toBe(true);

    const all = await store.readAll();
    expect(new TextDecoder().decode(all)).toBe('AAABBBCCC');

    // 后端是分片文件（单个 key）
    const keys = await backend.listKeys();
    expect(keys).toEqual(['__wal_000000.bin']);
    await backend.close();
  });

  it('超过分片阈值自动切换分片', async () => {
    const backend = new MemoryBackend();
    await backend.open('seg-unit-2');
    const store = new SegmentedWALStore(backend, 16); // 每分片 16 字节

    await store.append(enc('AAAAAAAAAA')); // 10B → 分片 0
    await store.append(enc('BBBBBBBBBB')); // 10B → 分片 1（0 已 10B+10B > 16）
    await store.append(enc('CC'));         // 2B  → 分片 1（10+2 <= 16）

    const all = await store.readAll();
    expect(new TextDecoder().decode(all)).toBe('AAAAAAAAAABBBBBBBBBBCC');
    const keys = await backend.listKeys().then((ks) => ks.sort());
    expect(keys).toEqual(['__wal_000000.bin', '__wal_000001.bin']);
    await backend.close();
  });

  it('空洞检测：分片序号不连续 → 空洞后的分片整体丢弃', async () => {
    const backend = new MemoryBackend();
    await backend.open('seg-unit-3');
    const store = new SegmentedWALStore(backend, 16);

    await store.append(enc('SEG0-SEG0-S')); // 10B → 分片 0
    await store.append(enc('SEG1-SEG1-S')); // 10B → 分片 1（0 已 10B+10B > 16）
    await store.append(enc('S2'));          // 2B  → 分片 1（10+2 <= 16）

    // 模拟 truncate 部分完成：删除分片 1（空洞）
    await backend.delete('__wal_000001.bin');

    const all = await store.readAll();
    // 空洞在分片 1 → 1 及之后丢弃，只保留分片 0
    expect(new TextDecoder().decode(all)).toBe('SEG0-SEG0-S');
    await backend.close();
  });

  it('空洞在开头（分片 0 缺失）→ 全部丢弃（保守截断）', async () => {
    const backend = new MemoryBackend();
    await backend.open('seg-unit-4');
    const store = new SegmentedWALStore(backend, 16);
    await store.append(enc('X0'));
    await store.append(enc('X1'));
    await backend.delete('__wal_000000.bin');

    const all = await store.readAll();
    expect(all.byteLength).toBe(0);
    await backend.close();
  });

  it('旧格式 __wal_N + __wal_count 兼容读取（迁移前数据）', async () => {
    const backend = new MemoryBackend();
    await backend.open('seg-unit-5');
    // 构造旧格式数据
    await backend.write('__wal_0', enc('LEGACY0').buffer);
    await backend.write('__wal_1', enc('LEGACY1').buffer);
    await backend.write('__wal_count', enc('2').buffer);

    const store = new SegmentedWALStore(backend, 1024);
    expect(await store.exists()).toBe(true);
    const all = await store.readAll();
    expect(new TextDecoder().decode(all)).toBe('LEGACY0LEGACY1');

    // 迁移中追加新格式：新记录写入分片文件（不与旧键冲突）
    await store.append(enc('NEW'));
    const keys = await backend.listKeys().then((ks) => ks.sort());
    expect(keys).toContain('__wal_000000.bin');
    expect(keys).toContain('__wal_0');

    // truncate 清空新旧全部
    await store.truncate();
    expect(await store.exists()).toBe(false);
    expect((await backend.listKeys()).length).toBe(0);
    await backend.close();
  });

  it('truncate 清空分片并重置状态', async () => {
    const backend = new MemoryBackend();
    await backend.open('seg-unit-6');
    const store = new SegmentedWALStore(backend, 16);
    await store.append(enc('AAAAAA'));
    await store.append(enc('BBBBBB'));
    await store.truncate();
    expect(await store.exists()).toBe(false);
    expect((await backend.listKeys()).length).toBe(0);

    // 截断后继续追加从分片 0 重新开始
    await store.append(enc('CCC'));
    const all = await store.readAll();
    expect(new TextDecoder().decode(all)).toBe('CCC');
    await backend.close();
  });

  it('后端无 append 接口 → 回退 read+write 语义一致', async () => {
    // MemoryBackend 没有 append 接口 → 走回退路径
    const backend = new MemoryBackend();
    await backend.open('seg-unit-7');
    const store = new SegmentedWALStore(backend, 1024);
    await store.append(enc('PART1'));
    await store.append(enc('PART2'));
    const all = await store.readAll();
    expect(new TextDecoder().decode(all)).toBe('PART1PART2');
    await backend.close();
  });
});

// ===================================================================
//  AriaEngine 集成 — 分片 WAL 跨重启
// ===================================================================
describe('AriaEngine — 分片 WAL 集成', () => {
  it('高频写入（多条 WAL 记录）→ close → reopen 数据完整', async () => {
    const dbName = uniqueDB();
    const engine = new AriaEngine({
      storageBackend: 'opfs',
      checkpointInterval: 100000,
      walSyncMode: 'full',
    });
    await engine.open(dbName, 1);
    await engine.createTable(createSchema('logs', {
      id: { type: 'string', primaryKey: true },
      msg: { type: 'string' },
    }));
    const total = 300;
    for (let i = 0; i < total; i++) {
      await engine.insert('logs', [{ id: `log-${i}`, msg: `message ${i}` }]);
    }
    // 不 flush 不 checkpoint，全部留在 WAL → 模拟崩溃后重放
    await (engine as any).backend.close();
    (engine as any).opened = false;

    const engine2 = new AriaEngine({
      storageBackend: 'opfs',
      checkpointInterval: 100000,
      walSyncMode: 'full',
    });
    await engine2.open(dbName, 1);
    const rows = await engine2.find('logs', { table: 'logs' });
    expect(rows).toHaveLength(total);
    expect(rows.some((r) => r.id === 'log-299')).toBe(true);
    await engine2.close();
  });

  it('分片结构落盘验证（backend 中是分片文件而非旧单记录键）', async () => {
    const dbName = uniqueDB();
    const engine = new AriaEngine({
      storageBackend: 'opfs',
      checkpointInterval: 100000,
      walSyncMode: 'full',
    });
    await engine.open(dbName, 1);
    await engine.createTable(createSchema('t', {
      id: { type: 'string', primaryKey: true },
    }));
    await engine.insert('t', [{ id: '1' }, { id: '2' }, { id: '3' }]);

    const backend = (engine as any).backend;
    const keys = await backend.listKeys();
    const walKeys = keys.filter((k: string) => k.startsWith('__wal_'));
    // 新格式：__wal_000000.bin；无旧格式单记录键与 count 键
    expect(walKeys).toContain('__wal_000000.bin');
    expect(walKeys.some((k: string) => /^__wal_\d+$/.test(k) && !k.endsWith('.bin'))).toBe(false);
    expect(walKeys).not.toContain('__wal_count');
    await engine.close();
  });

  it('WAL 分片部分残留（空洞）→ 重开恢复已落盘数据，不丢已 checkpoint 数据', async () => {
    const dbName = uniqueDB();
    const engine = new AriaEngine({
      storageBackend: 'opfs',
      checkpointInterval: 100000,
      walSyncMode: 'full',
    });
    await engine.open(dbName, 1);
    await engine.createTable(createSchema('t', {
      id: { type: 'string', primaryKey: true },
    }));
    // 第一批：flush 落盘 + checkpoint 清 WAL
    await engine.insert('t', [{ id: 'a' }, { id: 'b' }]);
    await (engine as any).lsm.flush();
    await (engine as any).wal.checkpoint();
    // 第二批：只进 WAL（不 flush）
    await engine.insert('t', [{ id: 'c' }, { id: 'd' }]);

    // 模拟 WAL 分片文件被外部删掉（空洞场景）
    const backend = (engine as any).backend;
    await backend.delete('__wal_000000.bin');

    await engine.close();
    const engine2 = new AriaEngine({ storageBackend: 'opfs', checkpointInterval: 100000 });
    await engine2.open(dbName, 1);
    // 已落盘的 a/b 必须保留；c/d 在 WAL 中且文件被删 → 恢复不到（可接受的保守丢弃）
    const rows = await engine2.find('t', { table: 't' });
    expect(rows.some((r) => r.id === 'a')).toBe(true);
    expect(rows.some((r) => r.id === 'b')).toBe(true);
    await engine2.close();
  });

  it('旧格式 WAL 升级迁移：旧键重放后 checkpoint 清空', async () => {
    const dbName = uniqueDB();
    // 手工构造旧格式 WAL 库：直接写旧键（模拟 v0.4.4 库崩溃现场）
    const engine = new AriaEngine({
      storageBackend: 'opfs',
      checkpointInterval: 100000,
      walSyncMode: 'full',
    });
    await engine.open(dbName, 1);
    await engine.createTable(createSchema('t', {
      id: { type: 'string', primaryKey: true },
    }));
    // 写入一批数据但只保留在 WAL（新格式）→ close 前手工转成旧格式
    await engine.insert('t', [{ id: 'x' }]);
    const backend = (engine as any).backend;
    const raw = await backend.read('__wal_000000.bin');
    expect(raw).not.toBeNull();
    // 清掉新格式，写成旧格式（单记录键）
    await backend.delete('__wal_000000.bin');
    await backend.write('__wal_0', raw);
    await backend.write('__wal_count', new TextEncoder().encode('1').buffer);
    // 同时清掉内存里的 WAL 状态，模拟"重启"
    await (engine as any).backend.close();
    (engine as any).opened = false;

    const engine2 = new AriaEngine({ storageBackend: 'opfs', checkpointInterval: 100000 });
    await engine2.open(dbName, 1);
    const rows = await engine2.find('t', { table: 't' });
    expect(rows).toHaveLength(1);
    expect(rows[0].id).toBe('x');
    // 恢复后 checkpoint 清空旧键
    const keys = await (engine2 as any).backend.listKeys();
    expect(keys.some((k: string) => k.startsWith('__wal_'))).toBe(false);
    await engine2.close();
  });
});