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('getLSN 跟踪序列号', () => {
    const store = new MockWALStore();
    const wal = new WAL(store, true, 'full');
    expect(wal.getLSN()).toBe(0);
    wal.append({ type: WALRecordType.BEGIN, txnId: 10, tableName: '', key: '' });
    expect(wal.getLSN()).toBe(1);
    wal.append({ type: WALRecordType.COMMIT, txnId: 10, tableName: '', key: '' });
    expect(wal.getLSN()).toBe(2);
  });

  it('isEnabled 反映配置', () => {
    expect(new WAL(new MockWALStore(), true).isEnabled()).toBe(true);
    expect(new WAL(new MockWALStore(), false).isEnabled()).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(cm.getOpCount()).toBe(2);
    expect(flushable.flushed).toBe(false);
    expect(lsm.flushed).toBe(false);
  });

  it('setInterval 修改间隔后 tick 触发 checkpoint', async () => {
    const lsm = new MockLSM();
    const wal = new MockWAL();
    const cm = new CheckpointManager(lsm as any, wal as any, null, 1000);
    cm.setInterval(2);
    await cm.tick();
    await cm.tick();
    expect(cm.getOpCount()).toBe(0); // reset after checkpoint
    expect(lsm.flushed).toBe(true);
    expect(wal.checkpointed).toBe(true);
  });

  it('forceCheckpoint 强制触发', async () => {
    const lsm = new MockLSM();
    const wal = new MockWAL();
    const cm = new CheckpointManager(lsm as any, wal as any, null, 100);
    await cm.forceCheckpoint();
    expect(cm.getOpCount()).toBe(0);
    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);
    expect(mvcc.isActive(id1)).toBe(true);
    expect(mvcc.isActive(id2)).toBe(true);
  });

  it('commit 后 isActive 返回 false', () => {
    const txnId = mvcc.beginTransaction();
    mvcc.commitTransaction(txnId);
    expect(mvcc.isActive(txnId)).toBe(false);
  });

  it('rollback 后 isActive 返回 false', () => {
    const txnId = mvcc.beginTransaction();
    mvcc.rollbackTransaction(txnId);
    expect(mvcc.isActive(txnId)).toBe(false);
  });

  it('writeVersion + readVersion 往返', () => {
    const txnId = mvcc.beginTransaction();
    mvcc.writeVersion('users', '1', { name: 'Alice', age: 30 }, txnId);
    const val = mvcc.readVersion('users', '1', txnId);
    expect(val).not.toBeNull();
    expect(val!.name).toBe('Alice');
  });

  it('未提交版本对其他事务不可见', () => {
    const txn1 = mvcc.beginTransaction();
    mvcc.writeVersion('users', '1', { name: 'Alice' }, txn1);

    const txn2 = mvcc.beginTransaction();
    const val = mvcc.readVersion('users', '1', txn2);
    expect(val).toBeNull(); // txn1's write not yet committed
  });

  it('commit 后新事务可见', () => {
    const txn1 = mvcc.beginTransaction();
    mvcc.writeVersion('users', '1', { name: 'Alice' }, txn1);
    mvcc.commitTransaction(txn1);

    const txn2 = mvcc.beginTransaction();
    const val = mvcc.readVersion('users', '1', txn2);
    expect(val).not.toBeNull();
    expect(val!.name).toBe('Alice');
  });

  it('rollback 移除写入的版本', () => {
    const txnId = mvcc.beginTransaction();
    mvcc.writeVersion('users', '1', { name: 'Temp' }, txnId);
    mvcc.rollbackTransaction(txnId);

    const txn2 = mvcc.beginTransaction();
    expect(mvcc.readVersion('users', '1', txn2)).toBeNull();
  });

  it('deleteVersion 创建墓碑', () => {
    const txnId = mvcc.beginTransaction();
    mvcc.writeVersion('users', '1', { name: 'Alice' }, txnId);
    mvcc.commitTransaction(txnId);
    // delete
    const txn2 = mvcc.beginTransaction();
    mvcc.deleteVersion('users', '1', txn2);
    mvcc.commitTransaction(txn2);
    // 删除后读取
    const txn3 = mvcc.beginTransaction();
    const val = mvcc.readVersion('users', '1', txn3);
    expect(val).not.toBeNull();
    expect((val! as any).__mvcc_tombstone).toBe(true);
  });

  it('getLatestCommittedVersions 返回最新已提交', () => {
    const txnId = mvcc.beginTransaction();
    mvcc.writeVersion('users', '1', { name: 'Alice' }, txnId);
    mvcc.writeVersion('users', '2', { name: 'Bob' }, txnId);
    mvcc.commitTransaction(txnId);

    const result = mvcc.getLatestCommittedVersions('users');
    expect(result['1'].name).toBe('Alice');
    expect(result['2'].name).toBe('Bob');
  });

  it('clearTable 清理指定表', () => {
    const txnId = mvcc.beginTransaction();
    mvcc.writeVersion('users', '1', { name: 'A' }, txnId);
    mvcc.writeVersion('users', '2', { name: 'B' }, txnId);
    mvcc.writeVersion('orders', 'o1', { amt: 100 }, txnId);
    mvcc.commitTransaction(txnId);

    mvcc.clearTable('users');
    const txn2 = mvcc.beginTransaction();
    expect(mvcc.readVersion('users', '1', txn2)).toBeNull();
    expect(mvcc.readVersion('users', '2', txn2)).toBeNull();
    expect(mvcc.readVersion('orders', 'o1', txn2)).not.toBeNull();
  });

  it('getActiveTxnCount 返回活跃事务数', () => {
    expect(mvcc.getActiveTxnCount()).toBe(0);
    mvcc.beginTransaction();
    mvcc.beginTransaction();
    expect(mvcc.getActiveTxnCount()).toBe(2);
  });

  it('gc 清理过旧版本', () => {
    // 创建很多版本后 gc
    for (let i = 0; i < 200; i++) {
      const txnId = mvcc.beginTransaction();
      mvcc.writeVersion('users', '1', { ver: i }, txnId);
      mvcc.commitTransaction(txnId);
    }
    mvcc.gc(100);
    // gc 后应可继续操作
    const txnId = mvcc.beginTransaction();
    const val = mvcc.readVersion('users', '1', txnId);
    expect(val).not.toBeNull();
  });
});