MetonaSqlark/tests/v080-kvstore-commit-point.test.ts

/**
 * v0.8.0 回归套件 —— B-6 存储提交点与崩溃语义（KVStore）
 * ============================================================================
 * 本套件覆盖 v0.8.0 迭代 §4「B-6」列出的 KVStore 三处止血，其中两处是
 * **P0 级数据丢失**：
 *
 * 1. `open()` 遇到损坏日志尾部会**清空整个日志**（`truncateLog()` 写空文件）。
 *    实测后果：
 *      写 3 条记录 → 第 4 条撕裂（崩溃）→ 重开后前 3 条可见 →
 *      **再重开一次，数据全部为空**。
 *    即"尾部损坏"被放大成"整库丢失"。修法：截断到最后一条有效记录
 *    （与 `repair()` 同一口径），而不是清空。
 *
 * 2. 自动 checkpoint 失败会把异常抛给 `put()`，但那条写入**已经在 WAL 里**
 *    （持久化成功）。于是用户看到"写入失败"、数据却在盘上 —— 报错与事实相反。
 *    修法：已确认写入不因后台失败而报错，失败记录为 `lastBackgroundError`，
 *    由下一次显式 `checkpoint()` 报告。
 *
 * 故障注入通过 `FaultyBackend` 完成（真实字节级撕裂/掉电，而非"重开测试"）。
 */
import { KVStore } from '../src/engine/kvstore/index';
import { MemoryBackend } from '../src/engine/aria/store/backend';
import { FaultyBackend } from './helpers/faulty-backend';
import { SharedMemoryBackend } from '../src/engine/kvstore/shared_memory_medium';
import { decode } from './helpers/assertions';

const enc = (s: string): ArrayBuffer => new TextEncoder().encode(s).buffer as ArrayBuffer;

// 注意：每个 KVStore 实例都包一个**自己的** FaultyBackend。
// 故障注入状态是"下一次写"这种一次性标记，跨实例共享会互相污染
//（恢复用的实例会意外继承上一步注入的故障）——本套件刻意不提供共享 helper。

describe('[v0.8.0] B-6 ①：损坏日志尾部不得清空整库', () => {
  it('尾部撕裂后重开：已确认写入全部保留（且二次重开仍保留）', async () => {
    const medium = new MemoryBackend();
    const faulty = new FaultyBackend(medium);
    const store = new KVStore(faulty, 0);
    await store.open('b6-tail');

    await store.put('a', enc('1'));
    await store.put('b', enc('2'));
    await store.put('c', enc('3'));

    // 第 4 条记录**只落前 20 字节**（写入过程中掉电 → 撕裂）
    faulty.truncateNextAppendTo(20);
    await store.put('d', enc('4')).catch(() => { /* 撕裂写入的行为由下层决定 */ });

    // ---- 第一次重开：前 3 条必须可见 ----
    const reopened = new KVStore(faulty, 0);
    await reopened.open('b6-tail');
    expect(decode(await reopened.get('a'))).toBe('1');
    expect(decode(await reopened.get('b'))).toBe('2');
    expect(decode(await reopened.get('c'))).toBe('3');

    // ---- 第二次重开：如果第一次重开把日志清空了，这里会全部为空 ----
    // 这是本用例的核心断言：恢复动作本身不得破坏尚未进入快照的数据。
    const reopened2 = new KVStore(faulty, 0);
    await reopened2.open('b6-tail');
    expect(decode(await reopened2.get('a'))).toBe('1');
    expect(decode(await reopened2.get('b'))).toBe('2');
    expect(decode(await reopened2.get('c'))).toBe('3');

    // ---- 第三次重开（幂等）：结果不变 ----
    const reopened3 = new KVStore(faulty, 0);
    await reopened3.open('b6-tail');
    expect(reopened3.size()).toBe(reopened2.size());
    expect(decode(await reopened3.get('b'))).toBe('2');
  });

  it('尾部 bit-flip 后重开：有效前缀保留，损坏记录丢弃', async () => {
    const medium = new MemoryBackend();
    const faulty = new FaultyBackend(medium);
    const store = new KVStore(faulty, 0);
    await store.open('b6-bitflip');
    await store.put('k1', enc('v1'));
    await store.put('k2', enc('v2'));

    // 把最后一条记录载荷里的一个字节改掉（CRC 不再匹配）
    faulty.corruptNextWrite(() => 0xff);
    await store.put('k3', enc('v3')).catch(() => { /* 允许实现拒绝 */ });
    faulty.clearFaults();

    const reopened = new KVStore(new FaultyBackend(medium), 0);
    await reopened.open('b6-bitflip');
    expect(decode(await reopened.get('k1'))).toBe('v1');
    expect(decode(await reopened.get('k2'))).toBe('v2');

    // 再次重开仍然保留（不是"第一次恢复读到了、第二次被清掉"）
    const reopened2 = new KVStore(new FaultyBackend(medium), 0);
    await reopened2.open('b6-bitflip');
    expect(decode(await reopened2.get('k1'))).toBe('v1');
    expect(decode(await reopened2.get('k2'))).toBe('v2');
  });

  it('损坏日志恢复后新写入仍可读（日志未被置为不可用状态）', async () => {
    const medium = new MemoryBackend();
    const faulty = new FaultyBackend(medium);
    const store = new KVStore(faulty, 0);
    await store.open('b6-recover-write');
    await store.put('before', enc('yes'));
    faulty.truncateNextAppendTo(10);
    await store.put('torn', enc('no')).catch(() => {});
    faulty.clearFaults();

    const reopened = new KVStore(new FaultyBackend(medium), 0);
    await reopened.open('b6-recover-write');
    // 修复后应能继续追加（日志被截断到有效长度，后续写入从干净位置开始）
    await reopened.put('after', enc('also-yes'));

    const reopened2 = new KVStore(new FaultyBackend(medium), 0);
    await reopened2.open('b6-recover-write');
    expect(decode(await reopened2.get('before'))).toBe('yes');
    expect(decode(await reopened2.get('after'))).toBe('also-yes');
  });
});

describe('[v0.8.0] B-6 ②：自动 checkpoint 失败不得让已确认写入报错', () => {
  it('自动 checkpoint 的介质写失败：put() 仍然成功，数据在盘上', async () => {
    const medium = new MemoryBackend();
    const faulty = new FaultyBackend(medium);
    // 阈值 1 字节 → 每次写入后都触发自动 checkpoint
    const store = new KVStore(faulty, 1);
    await store.open('b6-autockpt');

    // 快照写失败（checkpoint 的第一步）
    faulty.failNextWrite(1);
    // 关键：put() 必须**成功** —— WAL 追加已完成，数据已持久化
    await expect(store.put('k', enc('v'))).resolves.toBeUndefined();
    expect(faulty.injected.write).toBeGreaterThan(0); // 注入确实生效（避免假绿）

    // 数据真的在：重开后可见
    const reopened = new KVStore(new FaultyBackend(medium), 0);
    await reopened.open('b6-autockpt');
    expect(decode(await reopened.get('k'))).toBe('v');
  });

  it('后台失败由下一次显式 checkpoint() 报告（不静默吞掉）', async () => {
    const medium = new MemoryBackend();
    const faulty = new FaultyBackend(medium);
    const store = new KVStore(faulty, 1);
    await store.open('b6-bg-error');

    faulty.failNextWrite(1);
    await store.put('k', enc('v'));

    // 显式 checkpoint 是用户主动要求"把数据压实到快照"的时机，
    // 此时失败是真实问题，必须报告（KV_BACKGROUND_ERROR）
    await expect(store.checkpoint()).rejects.toMatchObject({ code: 'KV_BACKGROUND_ERROR' });
  });

  it('后台失败报告一次后即清除（不重复污染后续 checkpoint）', async () => {
    const medium = new MemoryBackend();
    const faulty = new FaultyBackend(medium);
    const store = new KVStore(faulty, 1);
    await store.open('b6-bg-once');

    faulty.failNextWrite(1);
    await store.put('k', enc('v'));
    await expect(store.checkpoint()).rejects.toMatchObject({ code: 'KV_BACKGROUND_ERROR' });

    // 第二次 checkpoint 应正常完成（错误已被消费）
    faulty.clearFaults();
    await expect(store.checkpoint()).resolves.toBeUndefined();
  });

  it('WAL 追加本身失败：写入必须报错且不得进入内存索引', async () => {
    const medium = new MemoryBackend();
    const faulty = new FaultyBackend(medium);
    const store = new KVStore(faulty, 0);
    await store.open('b6-wal-fail');

    faulty.failNextAppend(1);
    await expect(store.put('k', enc('v'))).rejects.toMatchObject({ code: 'KV_LOG_ERROR' });
    // 原子性：失败写入不得留在内存索引里
    expect(await store.get('k')).toBeNull();
    expect(store.size()).toBe(0);
  });

  it('掉电静默丢弃 append：写入"看起来成功"但重开后确实不在（无幻影）', async () => {
    const medium = new MemoryBackend();
    const faulty = new FaultyBackend(medium);
    const store = new KVStore(faulty, 0);
    await store.open('b6-dropped');

    await store.put('confirmed', enc('yes'));
    // 下一次 append 被介质静默丢弃（掉电丢失，不抛错）
    faulty.dropNextAppend(1);
    await store.put('lost', enc('no'));
    faulty.clearFaults();

    const reopened = new KVStore(new FaultyBackend(medium), 0);
    await reopened.open('b6-dropped');
    expect(decode(await reopened.get('confirmed'))).toBe('yes');
    // 这条记录确实丢了（这是"掉电丢失"的定义）——本断言的价值是确认
    // **不会复活成脏数据**，且不会影响其它记录
    expect(await reopened.get('lost')).toBeNull();
  });
});

describe('[v0.8.0] B-6 ③：checkpoint 时序保证任何崩溃窗口不丢数据', () => {
  const windows: Array<[string, (f: FaultyBackend) => void]> = [
    ['快照写失败（meta 未更新）', (f) => f.failNextWrite(1)],
    ['meta 写失败（快照已写、日志未截断）', (f) => { f.failNextWrite(1); f.failNextWrite(1); }],
  ];

  it.each(windows)('checkpoint 窗口「%s」后重开：已确认写入仍在', async (_label, inject) => {
    const medium = new MemoryBackend();
    const faulty = new FaultyBackend(medium);
    const store = new KVStore(faulty, 0);
    await store.open(`b6-window-${_label.length}`);
    await store.put('a', enc('1'));
    await store.put('b', enc('2'));

    inject(faulty);
    await store.checkpoint().catch(() => { /* 本用例关注恢复结果，不关注报错形式 */ });
    faulty.clearFaults();

    const reopened = new KVStore(new FaultyBackend(medium), 0);
    await reopened.open(`b6-window-${_label.length}`);
    expect(decode(await reopened.get('a'))).toBe('1');
    expect(decode(await reopened.get('b'))).toBe('2');
  });
});

describe('[v0.8.0] B-6：repair() 与 open() 的恢复口径一致', () => {
  it('repair() 报告丢弃字节数，且不丢有效前缀', async () => {
    const medium = new MemoryBackend();
    const faulty = new FaultyBackend(medium);
    const store = new KVStore(faulty, 0);
    await store.open('b6-repair');
    await store.put('keep', enc('kept'));
    faulty.truncateNextAppendTo(12);
    await store.put('torn', enc('torn')).catch(() => {});
    faulty.clearFaults();

    const reopened = new KVStore(new FaultyBackend(medium), 0);
    await reopened.open('b6-repair');
    const discarded = await reopened.repair();
    expect(discarded).toBeGreaterThanOrEqual(0);
    expect(decode(await reopened.get('keep'))).toBe('kept');
  });
});

// ---------------------------------------------------------------------------
//  B-6 ③：陈旧实例不得覆盖更新实例的提交
// ---------------------------------------------------------------------------

describe('[v0.8.0] B-6 ③：陈旧实例拒绝提交（多实例共享同一库）', () => {
  it('落后实例的 checkpoint 被拒绝，不会抹掉新实例的写入', async () => {
    SharedMemoryBackend.clearRegistry();
    // A 打开并写入
    const a = new KVStore(new SharedMemoryBackend(), 0);
    await a.open('b6-stale');
    await a.put('x', enc('from-A'));

    // B 打开同一库（读到 x）并写入 y，然后关闭
    const b = new KVStore(new SharedMemoryBackend(), 0);
    await b.open('b6-stale');
    await b.put('y', enc('from-B'));
    b.close();

    // 修复前：A.checkpoint() 用 A 的陈旧索引（没有 y）覆盖介质 → y 静默消失，
    // 且**不报任何错**（实测：after A checkpoint -> x: from-A y: null）。
    await expect(a.checkpoint()).rejects.toMatchObject({ code: 'STALE_INSTANCE' });
    expect(a.isStale()).toBe(true);

    // y 必须还在
    const c = new KVStore(new SharedMemoryBackend(), 0);
    await c.open('b6-stale');
    expect(decode(await c.get('x'))).toBe('from-A');
    expect(decode(await c.get('y'))).toBe('from-B');
  });

  it('陈旧实例的后续写入被拒绝（不再产生无法提交的数据）', async () => {
    SharedMemoryBackend.clearRegistry();
    const a = new KVStore(new SharedMemoryBackend(), 0);
    await a.open('b6-stale-write');
    await a.put('x', enc('1'));

    const b = new KVStore(new SharedMemoryBackend(), 0);
    await b.open('b6-stale-write');
    b.close();

    await expect(a.checkpoint()).rejects.toMatchObject({ code: 'STALE_INSTANCE' });
    // 已进入陈旧状态：写入必须显式失败，而不是"写进 WAL 但永远无法提交"
    await expect(a.put('z', enc('2'))).rejects.toMatchObject({ code: 'STALE_INSTANCE' });
  });

  it('重新 open 可恢复（陈旧状态不是永久的）', async () => {
    SharedMemoryBackend.clearRegistry();
    const a = new KVStore(new SharedMemoryBackend(), 0);
    await a.open('b6-stale-recover');
    await a.put('x', enc('1'));
    const b = new KVStore(new SharedMemoryBackend(), 0);
    await b.open('b6-stale-recover');
    await expect(a.checkpoint()).rejects.toMatchObject({ code: 'STALE_INSTANCE' });
    await a.close();

    // 重新 open → 重新领取所有权 → 正常读写与提交
    await a.open('b6-stale-recover');
    expect(a.isStale()).toBe(false);
    await a.put('x2', enc('2'));
    await expect(a.checkpoint()).resolves.toBeUndefined();

    const c = new KVStore(new SharedMemoryBackend(), 0);
    await c.open('b6-stale-recover');
    expect(decode(await c.get('x2'))).toBe('2');
  });

  it('单实例（无竞争）不受影响：所有权检查不误报', async () => {
    SharedMemoryBackend.clearRegistry();
    const store = new KVStore(new SharedMemoryBackend(), 0);
    await store.open('b6-single');
    await store.put('k', enc('v'));
    await expect(store.checkpoint()).resolves.toBeUndefined();
    await expect(store.checkpoint()).resolves.toBeUndefined(); // 幂等
    expect(decode(await store.get('k'))).toBe('v');
  });
});

// ---------------------------------------------------------------------------
//  B-6 介质一致性（shared_memory_medium 的跨实例可见性）
// ---------------------------------------------------------------------------

describe('[v0.8.0] B-6 介质：跨实例写入必须相互可见', () => {
  it('实例 B 的写入对已打开的实例 A 立即可见', async () => {
    SharedMemoryBackend.clearRegistry();
    const a = new KVStore(new SharedMemoryBackend(), 0);
    await a.open('b6-medium-vis');
    await a.put('k1', enc('a1'));
    // A 先读一次（触发缓存填充）
    expect(decode(await a.get('k1'))).toBe('a1');
    // 直接经介质读（KVStore.get 只读内存索引，这里验证的是介质层）
    const bMedium = new SharedMemoryBackend();
    await bMedium.open('b6-medium-vis');
    await bMedium.write('probe', enc('from-B'));

    expect(new TextDecoder().decode((await bMedium.read('probe'))!)).toBe('from-B');
    // A 再读同一个键：必须看到 B 写的内容
    expect(new TextDecoder().decode((await bMedium.read('probe'))!)).toBe('from-B');
  });

  it('删除后不得再读到旧值（缓存必须按删除失效）', async () => {
    SharedMemoryBackend.clearRegistry();
    const m1 = new SharedMemoryBackend();
    await m1.open('b6-medium-del');
    await m1.write('k', enc('v'));
    expect(decode(await m1.read('k'))).toBe('v');

    const m2 = new SharedMemoryBackend();
    await m2.open('b6-medium-del');
    await m2.delete('k');

    // m1 的缓存不得掩盖删除
    expect(await m1.read('k')).toBeNull();
    expect(await m2.read('k')).toBeNull();
  });

  it('append 后 read 必须包含新数据（跨实例）', async () => {
    SharedMemoryBackend.clearRegistry();
    const m1 = new SharedMemoryBackend();
    await m1.open('b6-medium-append');
    await m1.append('k', enc('A'));
    expect(decode(await m1.read('k'))).toBe('A');

    const m2 = new SharedMemoryBackend();
    await m2.open('b6-medium-append');
    await m2.append('k', enc('B'));
    expect(decode(await m1.read('k'))).toBe('AB');
  });

  it('clearRegistry 后同名库是全新存储（测试隔离有效）', async () => {
    SharedMemoryBackend.clearRegistry();
    const m1 = new SharedMemoryBackend();
    await m1.open('b6-medium-epoch');
    await m1.write('k', enc('old'));

    SharedMemoryBackend.clearRegistry();
    const m2 = new SharedMemoryBackend();
    await m2.open('b6-medium-epoch');
    expect(await m2.read('k')).toBeNull();
  });
});