MetonaSqlark/tests/v080-aria-ddl-atomicity.test.ts

/**
 * v0.8.0 回归套件 —— Aria DDL 原子性与崩溃恢复（A41）+ 生命周期守卫（A40）
 * ============================================================================
 * 实测确认的缺陷：**DDL 的 WAL 意图记录写在生效之后**。
 *
 * `createTable` / `dropTable` 的顺序是"改内存 → 落盘 schema → 追加 WAL"，
 * 于是 WAL —— 唯一能在崩溃后重放的结构权威 —— 恰恰是最后才写的：
 *   - `dropTable('other')` 删完 LSM 与 schema 后崩溃（WAL 尚无 DROP 记录）
 *     → 重开 `tables = ["t","other"]`，且 other 的行数据完好。
 *     用户以为删掉了，DROP 被静默撤销。
 *   - `alterTable` **完全不写 WAL**：先改内存 schema、必要时建索引（可能因
 *     存量重复值抛错），最后才 persistSchemas —— 中间抛错就留下
 *     "内存已加列、磁盘没加"的分裂状态；崩溃则结构变更整体丢失。
 *     变异验证：去掉 ALTER 回放后，`ADD tag` + `ADD tag2` 两列在重开后
 *     **都消失**（实测 `after reopen columns: ["id"]`）。
 *
 * 修复方式（结构上唯一正确的顺序）：
 *   **先写 WAL 意图并刷盘 → 再改内存 → 最后落盘 schema**
 * 因为 WAL 回放是幂等的（CREATE 对已存在的表跳过；DROP 对不存在的表是空操作；
 * ALTER 用变更后的完整 schema **覆盖**），先写 WAL 一定能收敛：
 *   - 崩溃于 WAL 之后、生效之前 → 恢复重放，DDL 生效 ✓
 *   - 崩溃于生效之后            → 恢复重放，幂等 ✓
 *
 * 关于 A40 的说明：审计记录的"close() 截断 WAL 并把未提交写入落盘 → 重开后
 * 幽灵行"经探针**未复现**（事务中 close 后重开只看到已提交行；事务中 DDL 抛
 * NOT_SUPPORTED）。本套件把这些已正确的行为固化为护栏，避免将来被误改。
 */
import { describe, it, expect, beforeEach } from '@jest/globals';
import { AriaEngine } from '../src/engine/aria/index';
import { resetOPFSMock } from './helpers/storage-harness';
import type { IStorageBackend } from '../src/engine/aria/store/backend';
import {
  decodeManifest,
  encodeManifest,
  generationFromKey,
  manifestKey,
  type AriaManifest,
} from '../src/engine/aria/store/manifest';

beforeEach(() => { resetOPFSMock(); });

const schema = (name: string, extra: Record<string, unknown> = {}) => ({
  name,
  columns: { id: { type: 'string' as const, primaryKey: true }, ...extra },
} as never);

/**
 * 内存介质 + 持久化**顺序记录** + 崩溃窗口诊断。
 *
 * 为什么需要它：本套件的核心不变量是"WAL 先于 schema 落盘"，那是**顺序**性质，
 * 只有记录顺序才能断言；而 OPFS mock 只暴露最终状态。
 */
class OrderRecordingBackend implements IStorageBackend {
  private files = new Map<string, ArrayBuffer>();
  readonly order: string[] = [];
  async open(): Promise<void> {}
  async close(): Promise<void> {}
  isOpen(): boolean { return true; }
  async read(k: string): Promise<ArrayBuffer | null> { return this.files.get(k) ?? null; }
  async write(k: string, d: ArrayBuffer): Promise<void> {
    this.files.set(k, d.slice(0));
    this.order.push(`write:${k}`);
  }
  async append(k: string, d: ArrayBuffer): Promise<void> {
    const prev = this.files.get(k);
    const merged = new Uint8Array((prev?.byteLength ?? 0) + d.byteLength);
    if (prev) merged.set(new Uint8Array(prev), 0);
    merged.set(new Uint8Array(d), prev?.byteLength ?? 0);
    this.files.set(k, merged.buffer as ArrayBuffer);
    this.order.push(`append:${k}`);
  }
  async writeMany(entries: Record<string, ArrayBuffer>): Promise<void> {
    for (const [k, v] of Object.entries(entries)) await this.write(k, v);
  }
  async delete(k: string): Promise<void> { this.files.delete(k); this.order.push(`delete:${k}`); }
  async deleteMany(keys: string[]): Promise<void> { for (const k of keys) await this.delete(k); }
  async listKeys(): Promise<string[]> { return [...this.files.keys()]; }
  async exists(k: string): Promise<boolean> { return this.files.has(k); }
  async clear(): Promise<void> { this.files.clear(); }

  /**
   * 索引：WAL 记录写入发生在 schema **持久化之前**。
   *
   * v0.8.0（B-6）：schema 不再是独立的 `__aria_schemas` 裸 JSON，
   * 它随 manifest 原子提交 —— 因此"schema 落盘"的落点变成 `__aria_manifest_*`。
   * 断言的**性质**没变（WAL 先于持久化），只是落点换了。
   */
  walPrecedesSchema(): boolean {
    const wal = this.order.findIndex((k) => k.includes('wal'));
    const sch = this.order.findIndex((k) => k.includes('__aria_manifest_'));
    return wal >= 0 && sch >= 0 && wal < sch;
  }
  /** 模拟"该文件没能落盘"（崩溃窗口） */
  dropFile(pattern: string): void {
    for (const k of [...this.files.keys()]) if (k.includes(pattern)) this.files.delete(k);
  }
  /**
   * 当前 manifest 的**语义快照**（解码后的内容 + 世代号）。
   * 用于"DDL 之前"的状态存档：见 `rollbackManifest()`。
   */
  snapshotManifest(): { manifest: AriaManifest; generation: number } {
    const gens = [...this.files.keys()]
      .map((k) => generationFromKey(k))
      .filter((g): g is number => g !== null)
      .sort((a, b) => b - a);
    if (gens.length === 0) throw new Error('no manifest on medium');
    const raw = this.files.get(manifestKey(gens[0]))!;
    const decoded = decodeManifest(new Uint8Array(raw));
    if (!decoded.ok) throw new Error(`manifest decode failed: ${decoded.reason}`);
    return { manifest: decoded.manifest, generation: gens[0] };
  }

  /**
   * 把 manifest 回滚到给定快照（写成一份**更新世代号**的内容，并清掉其它世代）。
   *
   * 语义：模拟"这次 DDL 的 manifest 提交从未落盘"——数据文件与 WAL 保持原样，
   * 于是重开时结构只能靠 **WAL 意图回放** 恢复（这正是本套件要验证的兜底）。
   *
   * 注意：不能简单删掉新世代 —— 世代保留窗口是 2，被回滚到的那一代此时
   * 可能已经被正常清理掉了。因此这里用快照内容重新提交一份新世代。
   */
  rollbackManifest(snapshot: { manifest: AriaManifest; generation: number }): void {
    const gens = [...this.files.keys()]
      .map((k) => generationFromKey(k))
      .filter((g): g is number => g !== null);
    const nextGeneration = Math.max(0, ...gens) + 1;
    const rolled: AriaManifest = {
      ...snapshot.manifest,
      generation: nextGeneration,
      namespaces: snapshot.manifest.namespaces,
      schemas: snapshot.manifest.schemas,
      wal: { ...snapshot.manifest.wal },
      frozen: [],
      owner: { ...snapshot.manifest.owner },
    };
    const bytes = encodeManifest(rolled);
    this.files.set(
      manifestKey(nextGeneration),
      bytes.buffer.slice(bytes.byteOffset, bytes.byteOffset + bytes.byteLength) as ArrayBuffer,
    );
    for (const k of [...this.files.keys()]) {
      const gen = generationFromKey(k);
      if (gen !== null && gen !== nextGeneration) this.files.delete(k);
    }
  }
}

/**
 * 构造一个把注入后端当唯一介质的引擎。
 *
 * 注意：库名由 `open(dbName, version)` 传入，这里不接第二个参数 ——
 * 早期版本接了 `name` 却没用（lint 警告），留着会误导读者以为构造函数需要库名。
 */
function openWith(backend: IStorageBackend): AriaEngine {
  return new AriaEngine({
    storageBackend: 'memory',
    checkpointInterval: 100_000_000,
    testBackend: backend,
  } as never);
}

describe('[v0.8.0] A41 DDL 的 WAL 意图必须先于生效', () => {
  it('createTable：WAL 先于 schema 落盘', async () => {
    const backend = new OrderRecordingBackend();
    const engine = openWith(backend);
    await engine.open('ddl-create', 1);
    backend.order.length = 0;
    await engine.createTable(schema('t'));
    expect(backend.walPrecedesSchema()).toBe(true);
    await engine.close();
  });

  it('dropTable：WAL 先于 schema 落盘', async () => {
    const backend = new OrderRecordingBackend();
    const engine = openWith(backend);
    await engine.open('ddl-drop', 1);
    await engine.createTable(schema('t'));
    await engine.createTable(schema('other'));
    backend.order.length = 0;
    await engine.dropTable('other');
    expect(backend.walPrecedesSchema()).toBe(true);
    await engine.close();
  });

  it('alterTable ADD：WAL 先于 schema 落盘（修复前完全不写 WAL）', async () => {
    const backend = new OrderRecordingBackend();
    const engine = openWith(backend);
    await engine.open('ddl-alter-add', 1);
    await engine.createTable(schema('t'));
    backend.order.length = 0;
    await engine.alterTable('t', 'ADD', { name: 'tag', type: 'string' } as never);
    expect(backend.walPrecedesSchema()).toBe(true);
    await engine.close();
  });

  it('alterTable DROP：WAL 先于 schema 落盘', async () => {
    const backend = new OrderRecordingBackend();
    const engine = openWith(backend);
    await engine.open('ddl-alter-drop', 1);
    await engine.createTable(schema('t', { tag: { type: 'string' } }));
    backend.order.length = 0;
    await engine.alterTable('t', 'DROP', { name: 'tag', type: 'string' } as never);
    expect(backend.walPrecedesSchema()).toBe(true);
    await engine.close();
  });
});

describe('[v0.8.0] A41 DDL 结构变更可崩溃恢复（WAL 兜底）', () => {
  it('ALTER ADD 后 schema 未能落盘就崩溃 → 重开结构仍完整', async () => {
    const backend = new OrderRecordingBackend();
    const engine = openWith(backend);
    await engine.open('ddl-alter-crash', 1);
    // 记录"DDL 之前"的 manifest 状态：崩溃窗口 = 之后的提交都没落盘
    const beforeDdl = backend.snapshotManifest();
    await engine.createTable(schema('t'));
    await engine.alterTable('t', 'ADD', { name: 'tag', type: 'string' } as never);
    await engine.alterTable('t', 'ADD', { name: 'tag2', type: 'string' } as never);

    // 崩溃窗口：两次 ALTER 的 schema 都没能落盘（不 close，否则 close 会重试落盘）
    backend.rollbackManifest(beforeDdl);
    expect(backend.snapshotManifest().generation).toBeGreaterThan(beforeDdl.generation);

    const engine2 = openWith(backend);
    await engine2.open('ddl-alter-crash', 1);
    const cols = Object.keys((await engine2.getTableSchema('t'))!.columns);
    // 变异验证：去掉 ALTER_TABLE 回放后这里只剩 ["id"]
    expect(cols).toContain('tag');
    expect(cols).toContain('tag2');
    await engine2.close();
  });

  it('DROP TABLE 后 schema 未能落盘就崩溃 → 重开表确实已删除', async () => {
    const backend = new OrderRecordingBackend();
    const engine = openWith(backend);
    await engine.open('ddl-drop-crash', 1);
    await engine.createTable(schema('t'));
    await engine.insert('t', [{ id: 'r1' }]);
    await engine.createTable(schema('other'));
    await engine.insert('other', [{ id: 'o1' }]);
    const beforeDrop = backend.snapshotManifest();

    await engine.dropTable('other');
    backend.rollbackManifest(beforeDrop);   // DROP 的持久化提交丢失
    // 不 close（崩溃语义）

    const engine2 = openWith(backend);
    await engine2.open('ddl-drop-crash', 1);
    const tables = await engine2.getTableNames();
    expect(tables).not.toContain('other');   // 修复前 DROP 会被静默撤销
    expect(tables).toContain('t');
    expect((await engine2.find('t', { table: 't' })).map((r) => r.id)).toEqual(['r1']);
    await engine2.close();
  });

  it('CREATE TABLE 后 schema 未能落盘就崩溃 → 重开表仍存在且可用', async () => {
    const backend = new OrderRecordingBackend();
    const engine = openWith(backend);
    await engine.open('ddl-create-crash', 1);
    const beforeCreate = backend.snapshotManifest();
    await engine.createTable(schema('t'));
    await engine.insert('t', [{ id: 'r1' }]);
    backend.rollbackManifest(beforeCreate);

    const engine2 = openWith(backend);
    await engine2.open('ddl-create-crash', 1);
    expect(await engine2.getTableNames()).toContain('t');
    expect((await engine2.find('t', { table: 't' })).map((r) => r.id)).toEqual(['r1']);
    await engine2.close();
  });

  it('DDL 意图记录可重复回放（幂等）', async () => {
    const backend = new OrderRecordingBackend();
    const engine = openWith(backend);
    await engine.open('ddl-idempotent', 1);
    await engine.createTable(schema('t'));
    await engine.alterTable('t', 'ADD', { name: 'tag', type: 'string' } as never);
    await engine.close();

    // 连续重开三次：每次都回放同一份 WAL，结果必须一致（不重复加列、不报错）
    for (let round = 0; round < 3; round++) {
      const e = openWith(backend);
      await e.open('ddl-idempotent', 1);
      const cols = Object.keys((await e.getTableSchema('t'))!.columns);
      expect(cols).toEqual(['id', 'tag']);
      await e.close();
    }
  });
});

describe('[v0.8.0] A40 生命周期守卫（审计结论未复现，固化为护栏）', () => {
  it('事务中 DDL 显式拒绝（NOT_SUPPORTED），不留半状态', async () => {
    const backend = new OrderRecordingBackend();
    const engine = openWith(backend);
    await engine.open('guard-ddl-txn', 1);
    await engine.createTable(schema('t'));
    await engine.beginTransaction();
    await expect(engine.dropTable('t')).rejects.toMatchObject({ code: 'NOT_SUPPORTED' });
    await expect(
      engine.alterTable('t', 'ADD', { name: 'x', type: 'string' } as never),
    ).rejects.toMatchObject({ code: 'NOT_SUPPORTED' });
    await engine.rollbackTransaction();
    expect(await engine.getTableNames()).toContain('t');
    await engine.close();
  });

  it('未提交事务在 close 后不得产生幽灵行', async () => {
    const backend = new OrderRecordingBackend();
    const engine = openWith(backend);
    await engine.open('guard-ghost', 1);
    await engine.createTable(schema('t'));
    await engine.insert('t', [{ id: 'committed' }]);
    await engine.close();

    const engine2 = openWith(backend);
    await engine2.open('guard-ghost', 1);
    await engine2.beginTransaction();
    await engine2.insert('t', [{ id: 'uncommitted' }]);
    await engine2.close();

    const engine3 = openWith(backend);
    await engine3.open('guard-ghost', 1);
    const ids = (await engine3.find('t', { table: 't' })).map((r) => r.id).sort();
    expect(ids).toEqual(['committed']);   // 未提交行不得复活
    await engine3.close();
  });

  it('close 幂等；close 后 rollback 抛 TX_NONE（而不是静默成功）', async () => {
    const backend = new OrderRecordingBackend();
    const engine = openWith(backend);
    await engine.open('guard-close', 1);
    await engine.createTable(schema('t'));
    await engine.close();
    await expect(engine.close()).resolves.toBeUndefined();   // 二次 close 幂等
    await expect(engine.rollbackTransaction()).rejects.toMatchObject({ code: 'TX_NONE' });
  });
});