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

/**
 * AriaEngine WAL + MVCC 单元测试
 */
import { WAL, type WALStore } from '../../src/engine/aria/wal/log';
import { WALRecordType, type WALRecord } from '../../src/engine/aria/types';
import { CheckpointManager, type Flushable } from '../../src/engine/aria/wal/checkpoint';
import { MVCCManager } from '../../src/engine/aria/transaction/mvcc';

// ===================================================================
//  WAL 存储 Mock
// ===================================================================
class MockWALStore implements WALStore {
  chunks: Uint8Array[] = [];
  async append(data: Uint8Array) { this.chunks.push(data); }
  async readAll(): Promise<Uint8Array> {
    const total = this.chunks.reduce((s, c) => s + c.byteLength, 0);
    const combined = new Uint8Array(total);
    let off = 0;
    for (const c of this.chunks) { combined.set(c, off); off += c.byteLength; }
    return combined;
  }
  async truncate() { this.chunks = []; }
  async exists() { return this.chunks.length > 0; }
}

// ===================================================================
//  WAL 测试
// ===================================================================
describe('AriaEngine — WAL', () => {
  it('append 记录后可恢复', async () => {
    const store = new MockWALStore();
    const wal = new WAL(store, true, 'full');

    wal.append({ type: WALRecordType.INSERT, txnId: 1, tableName: 'users', key: '1', data: { name: 'Alice' } });
    wal.append({ type: WALRecordType.INSERT, txnId: 1, tableName: 'users', key: '2', data: { name: 'Bob' } });

    const records: WALRecord[] = [];
    await wal.recover((r) => records.push(r));
    expect(records).toHaveLength(2);
    expect(records[0].tableName).toBe('users');
    expect(records[0].key).toBe('1');
  });

  it('batch 模式缓冲后 flush', async () => {
    const store = new MockWALStore();
    const wal = new WAL(store, true, 'batch');

    wal.append({ type: WALRecordType.UPDATE, txnId: 2, tableName: 'items', key: 'a', data: { v: 1 } });
    wal.append({ type: WALRecordType.DELETE, txnId: 2, tableName: 'items', key: 'b' });

    // 未 flush 前无法恢复
    let records: WALRecord[] = [];
    await wal.recover((r) => records.push(r));
    expect(records).toHaveLength(0);

    await wal.flush();
    records = [];
    await wal.recover((r) => records.push(r));
    expect(records).toHaveLength(2);
  });

  it('none 模式不记录', async () => {
    const store = new MockWALStore();
    const wal = new WAL(store, false, 'none');
    wal.append({ type: WALRecordType.INSERT, txnId: 3, tableName: 'x', key: 'y', data: {} });
    expect(store.chunks).toHaveLength(0);
  });

  it('checkpoint 清空 WAL', async () => {
    const store = new MockWALStore();
    const wal = new WAL(store, true, 'full');

    wal.append({ type: WALRecordType.CREATE_TABLE, txnId: 0, tableName: 't', key: '' });
    expect(await store.exists()).toBe(true);

    await wal.checkpoint();
    expect(await store.exists()).toBe(false);
  });

  it('多种记录类型编解码', async () => {
    const store = new MockWALStore();
    const wal = new WAL(store, true, 'full');

    wal.append({ type: WALRecordType.BEGIN, txnId: 100, tableName: '', key: '' });
    wal.append({ type: WALRecordType.INSERT, txnId: 100, tableName: 'users', key: '1', data: { x: 'hello' } });
    wal.append({ type: WALRecordType.UPDATE, txnId: 100, tableName: 'users', key: '1', data: { x: 'world' } });
    wal.append({ type: WALRecordType.COMMIT, txnId: 100, tableName: '', key: '' });

    const records: WALRecord[] = [];
    await wal.recover((r) => records.push(r));
    expect(records).toHaveLength(4);
    expect(records[0].type).toBe(WALRecordType.BEGIN);
    expect(records[1].type).toBe(WALRecordType.INSERT);
    expect(records[2].type).toBe(WALRecordType.UPDATE);
    expect(records[3].type).toBe(WALRecordType.COMMIT);
  });

  it('getBufferedCount 返回缓冲数', () => {
    const wal = new WAL(new MockWALStore(), true, 'batch');
    expect(wal.getBufferedCount()).toBe(0);
    wal.append({ type: WALRecordType.INSERT, txnId: 0, tableName: 't', key: 'k' });
    expect(wal.getBufferedCount()).toBe(1);
  });
});

// ===================================================================
//  Checkpoint 测试（使用安全 Mock，避免 null 引用导致 CI 卡死）
// ===================================================================
describe('AriaEngine — CheckpointManager', () => {
  class MockFlushable implements Flushable { flushed = false; async flushAll() { this.flushed = true; } }
  class MockLSM { flushed = false; async flush() { this.flushed = true; } }
  class MockWAL { checkpointed = false; async checkpoint() { this.checkpointed = true; } async flush() {} }

  it('tick 未达间隔不触发 checkpoint', async () => {
    const lsm = new MockLSM();
    const wal = new MockWAL();
    const flushable = new MockFlushable();
    const cm = new CheckpointManager(lsm as any, wal as any, flushable, 100);
    await cm.tick();
    await cm.tick();
    expect(flushable.flushed).toBe(false);
    expect(lsm.flushed).toBe(false);
  });

  it('tick 达到间隔触发 checkpoint', async () => {
    const lsm = new MockLSM();
    const wal = new MockWAL();
    const cm = new CheckpointManager(lsm as any, wal as any, null, 2);
    await cm.tick();
    expect(lsm.flushed).toBe(false);
    await cm.tick();
    expect(lsm.flushed).toBe(true);
    expect(wal.checkpointed).toBe(true);
  });

  it('checkpoint 直接触发', async () => {
    const lsm = new MockLSM();
    const wal = new MockWAL();
    const cm = new CheckpointManager(lsm as any, wal as any, null, 100);
    await cm.checkpoint();
    expect(lsm.flushed).toBe(true);
    expect(wal.checkpointed).toBe(true);
  });
});

// ===================================================================
//  MVCC 测试
// ===================================================================
describe('AriaEngine — MVCC', () => {
  let mvcc: MVCCManager;

  beforeEach(() => { mvcc = new MVCCManager(); });

  it('beginTransaction 分配唯一 ID', () => {
    const id1 = mvcc.beginTransaction();
    const id2 = mvcc.beginTransaction();
    expect(id1).not.toBe(id2);
  });

  it('commit 清理本事务版本并清理事务登记（v0.6.3）', () => {
    const txnId = mvcc.beginTransaction();
    mvcc.writeVersion('users', '1', { name: 'Alice' }, txnId);
    mvcc.commitTransaction(txnId);
    // v0.6.3: 已提交版本随 commit 清理（快照读取已移除，版本链仅作事务内 undo）
    const store = (mvcc as unknown as { versionStore: Map<string, unknown> }).versionStore;
    expect(store.has('users.1')).toBe(false);
  });

  it('rollback 移除写入的版本', () => {
    const txnId = mvcc.beginTransaction();
    mvcc.writeVersion('users', '1', { name: 'Temp' }, txnId);
    mvcc.rollbackTransaction(txnId);
    const store = (mvcc as unknown as { versionStore: Map<string, unknown> }).versionStore;
    expect(store.has('users.1')).toBe(false);
  });

  it('writeVersion 记录版本链（undo log 语义）', () => {
    const txnId = mvcc.beginTransaction();
    mvcc.writeVersion('users', '1', { name: 'Alice', age: 30 }, txnId);
    mvcc.writeVersion('users', '1', { name: 'Alice', age: 31 }, txnId);
    const store = (mvcc as unknown as { versionStore: Map<string, { prevVersion: unknown }[]> }).versionStore;
    const versions = store.get('users.1')!;
    expect(versions).toHaveLength(2);
    expect(versions[1].prevVersion).not.toBeNull(); // 版本链
  });

  it('deleteVersion 写入墓碑版本（未提交可见，commit 清理）', () => {
    const txnId = mvcc.beginTransaction();
    mvcc.deleteVersion('users', '1', txnId);
    const store = (mvcc as unknown as { versionStore: Map<string, { data: Record<string, unknown> }[]> }).versionStore;
    const versions = store.get('users.1')!;
    expect((versions[0].data as unknown as { __mvcc_tombstone: boolean }).__mvcc_tombstone).toBe(true);
    mvcc.commitTransaction(txnId);
    expect(store.has('users.1')).toBe(false);
  });

  it('活跃事务可并发登记', () => {
    const t1 = mvcc.beginTransaction();
    const t2 = mvcc.beginTransaction();
    mvcc.writeVersion('users', '1', { v: 1 }, t1);
    mvcc.writeVersion('users', '2', { v: 2 }, t2);
    // 两个事务的写入各自独立记录
    const store = (mvcc as unknown as { versionStore: Map<string, unknown[]> }).versionStore;
    expect(store.has('users.1')).toBe(true);
    expect(store.has('users.2')).toBe(true);
  });

  it('gc 清理过旧版本（未提交版本链保留最新 N 个）', () => {
    // 创建很多未提交版本后 gc（活跃事务 undo 链）
    for (let i = 0; i < 200; i++) {
      const txnId = mvcc.beginTransaction();
      mvcc.writeVersion('users', '1', { ver: i }, txnId);
    }
    mvcc.gc(100);
    // gc 后版本保留最新 100 个
    const store = (mvcc as unknown as { versionStore: Map<string, unknown[]> }).versionStore;
    expect(store.get('users.1')!.length).toBe(100);
    // gc 后仍可继续写入
    const txnId = mvcc.beginTransaction();
    mvcc.writeVersion('users', '1', { ver: 999 }, txnId);
    expect(store.get('users.1')!.length).toBe(101);
  });
});