MetonaSqlark/scripts/mutation-b6.py

#!/usr/bin/env python3
"""B-6 变异验证：把每个修复回退到修复前的行为，对应用例必须失败。

用法：python3 scripts/mutation-b6.py
任何一条"回退后测试仍然通过"都会以非零退出码报出来（说明测试只是陪跑）。
"""
import io
import os
import re
import shlex
import signal
import subprocess
import sys

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
LSM = 'src/engine/aria/index/lsm.ts'
MANIFEST = 'src/engine/aria/store/manifest.ts'
MERGE = 'src/engine/aria/index/merge_iterator.ts'
ENGINE = 'src/engine/aria/index.ts'
SEGSTORE = 'src/engine/aria/wal/segmented_store.ts'
CHECKPOINT = 'src/engine/aria/wal/checkpoint.ts'
SSTABLE = 'src/engine/aria/index/sstable.ts'

B6 = 'tests/v080-b6-single-commit-point.test.ts'
STRESS = 'tests/engine/aria-repair-hardening.test.ts'

MUTATIONS = [
    dict(
        name='47 MergeIterator 立即补充（提前终止多算 1 条）',
        file=MERGE,
        old="""    // 胜出来源的补充推迟到下一次 next()（消费者只取 N 条 → 底层只产出 N 条）
    this.pendingRefill = first.sourceIndex;""",
        new="""    // [MUTATION] 立即补充（修复前行为）
    this.seedFromSource(first.sourceIndex);""",
        test=B6, pattern='流式扫描提前终止不多算',
    ),
    dict(
        name='49 compacting 单 boolean（跨层触发被丢弃）',
        file=LSM,
        old="""  private scheduleCompact(level: number): void {
    if (level > MAX_LSM_LEVELS - 1 || this.compacting.has(level)) return;
    this.compacting.add(level);""",
        new="""  private scheduleCompact(level: number): void {
    if (level > MAX_LSM_LEVELS - 1 || this.compacting.size > 0) return;   // [MUTATION] 单 boolean 语义
    this.compacting.add(level);""",
        test=B6, pattern='按层 compaction 状态',
    ),
    dict(
        name='50 compaction 先 splice 整层（窗口内该层不可见）',
        file=LSM,
        old="""    const selected = [...this.levels[level]];
    const selectedIds = new Set(selected.map((m) => m.id));""",
        new="""    const selected = this.levels[level].splice(0, this.levels[level].length);   // [MUTATION] 旧行为
    const selectedIds = new Set(selected.map((m) => m.id));""",
        test=B6, pattern='compaction 期间该层对读者始终可见',
    ),
    dict(
        name='51 底部层不回收墓碑',
        file=LSM,
        old="""    const merged = isBottomLevel
      ? mergedRaw.filter(([, value]) => !(value as unknown as Record<string, unknown>).__tombstone)
      : mergedRaw;""",
        new="""    const merged = mergedRaw;   // [MUTATION] 不回收墓碑""",
        test=B6, pattern='底部层合并回收墓碑',
    ),
    dict(
        name='44 冻结表失败后不重试',
        file=LSM,
        old="""  private async retryPendingFlushes(): Promise<void> {
    let rounds = 0;""",
        new="""  private async retryPendingFlushes(): Promise<void> {
    return;   // [MUTATION] 无重试路径
    let rounds = 0;""",
        test=B6, pattern='持续失败 → flush 明确报错且数据不丢',
    ),
    dict(
        name='45 后台错误检查放回"入链之前"',
        file=LSM,
        old="""  async flush(): Promise<void> {
    this.enqueuePendingMemtables();
    const reported = this.consumeBackgroundError();
    await this.drainChain();""",
        new="""  async flush(): Promise<void> {
    // [MUTATION] 旧行为：先检查/报告后台错误 → 本次 flush 被整个跳过
    const early = this.consumeBackgroundError();
    if (early !== null) {
      throw new DatabaseError('background error (legacy behaviour)', 'ARIA_BACKGROUND_ERROR', early);
    }
    this.enqueuePendingMemtables();
    const reported = this.consumeBackgroundError();
    await this.drainChain();""",
        test=B6, pattern='后台错误不得让本次 flush 白做',
    ),
    dict(
        name='manifest 全世代损坏 → 返回空状态（静默空库）',
        file=MANIFEST,
        old="""    if (generations.length > 0) {
      // 文件存在但一代都读不出来：这是元数据损坏，不是空库
      throw new DatabaseError(""",
        new="""    if (false && generations.length > 0) {   // [MUTATION] 静默当空库
      throw new DatabaseError(""",
        test=B6, pattern='全部世代都损坏',
    ),
    dict(
        name='陈旧实例不设防（静默覆盖新世代）',
        file=MANIFEST,
        old="""    if (!opts.allowTakeover && newestForeign > this.generation) {""",
        new="""    if (false && !opts.allowTakeover && newestForeign > this.generation) {   // [MUTATION] 不设防""",
        test=B6, pattern='陈旧实例提交被拒绝',
    ),
    dict(
        name='旧格式 meta 损坏 → 返回空（静默空库）',
        file=ENGINE,
        old="""      } catch (error) {
        throw new DatabaseError(
          `Legacy SSTable metadata "${key}" is corrupt and cannot be migrated: ${(error as Error).message}`,
          'ARIA_LEGACY_META_CORRUPT',
          error,
        );
      }""",
        new="""      } catch (error) {
        continue;   // [MUTATION] 静默跳过损坏的旧 meta
      }""",
        test=B6, pattern='旧格式 meta 损坏',
    ),
    dict(
        name='介质读故障被当成"文件不存在"（误删 meta）',
        file=LSM,
        old="""      } catch (error) {
        throw new DatabaseError(
          `AriaEngine failed to read SSTable id=${meta.id} from storage: ${(error as Error).message}`,
          'ARIA_SSTABLE_READ_FAILED',
          error,
        );
      }""",
        new="""      } catch (error) {
        await this.dropInvalidSSTable(meta, 'read failure treated as missing');   // [MUTATION] 旧行为
        return null;
      }""",
        test=B6, pattern='介质读故障 ≠ 文件缺失',
    ),
    dict(
        name='WAL 分片空洞静默丢弃尾部',
        file='src/engine/aria/wal/log.ts',
        old="""    if (gaps.length > 0 && !opts.allowGaps) {""",
        new="""    if (false && gaps.length > 0 && !opts.allowGaps) {   // [MUTATION] 静默丢弃""",
        test=B6, pattern='活跃区间内缺失分片',
    ),
    dict(
        name='WAL 分片号复用（旧世代排在新记录之后）',
        file=SEGSTORE,
        old="""      if (leftover.length === 0) {
        this.currentSegment = Math.max(maxSeq + 1, keepFrom);""",
        new="""      if (leftover.length === 0) {
        this.currentSegment = 0;   // [MUTATION] 复用分片号""",
        test=STRESS, pattern='随机操作压力',
    ),
    dict(
        name='冻结表意图不做"写入是否真的丢了"校验',
        file=ENGINE,
        old="""    const summary = intents.map((i) => `${i.ns}#${i.id}(${i.entryCount} 项)`).join(', ');
    throw new DatabaseError(""",
        new="""    const summary = intents.map((i) => `${i.ns}#${i.id}(${i.entryCount} 项)`).join(', ');
    if (summary) return;   // [MUTATION] 不校验
    throw new DatabaseError(""",
        test=B6, pattern='ARIA_WRITE_LOST',
    ),
    dict(
        name='提交中的冻结表仍写进意图（回退时误报 WRITE_LOST）',
        file=LSM,
        old="""      // 正在提交的表由当前这次 manifest 提交负责（见 FrozenTable.committing）
      if (frozen.committing) continue;""",
        new="""      // [MUTATION] 提交中的表也写进意图""",
        test=B6, pattern='回退上一代并标记 manifestFallback',
    ),
    dict(
        name='checkpoint 等完整 compaction（写路径被拖住）',
        file=CHECKPOINT,
        old="""    if (this.flushable && typeof this.flushable.flushMemtables === 'function') {""",
        new="""    if (false && this.flushable && typeof this.flushable.flushMemtables === 'function') {   // [MUTATION] 旧行为""",
        test=B6, pattern='checkpoint 不再等完整 compaction',
    ),
    dict(
        name='collectReaders 不做结构版本重试（扫描漏掉并发发布的 SSTable）',
        file=LSM,
        old="""      if (version === this.structureVersion) return readers;
      this.readStructureRetries++;
    }""",
        new="""      return readers;   // [MUTATION] 不校验结构版本
    }""",
        test=B6, pattern='并发 flush 与 compaction 交错时',
    ),
    dict(
        name='点查不做结构版本重试（get 的校验失效）',
        file=LSM,
        old="""        if (version !== this.structureVersion) {
          this.readStructureRetries++;
          continue; // 期间有结构变化（新产物发布 / 前台冻结）→ 重来
        }""",
        new="""        if (false) {
          this.readStructureRetries++;
          continue;
        }""",
        test=B6, pattern='并发 flush 与 compaction 交错时',
    ),
    dict(
        name='冻结 memtable 不更新结构版本（点查读到旧值/漏数据）',
        file=LSM,
        old="""    // v0.8.0：**前台变化也算结构变化**。读路径的"结构版本一致"必须覆盖""",
        new="""    this.structureVersion--;   // [MUTATION] 抵消下面的自增
    // v0.8.0：**前台变化也算结构变化**。读路径的"结构版本一致"必须覆盖""",
        test=B6, pattern='并发 flush 与 compaction 交错时',
    ),
    dict(
        name='sstable 越界策略不统一（scanAll 另有一套实现：越界即整体放弃）',
        file=SSTABLE,
        old="""  scanAll(callback: (key: string, value: Record<string, unknown>) => void): void {
    for (const [key, value] of this.iterEntries(0, this.indexEntries.length - 1)) {
      callback(key, value);
    }
  }""",
        new="""  scanAll(callback: (key: string, value: Record<string, unknown>) => void): void {
    // [MUTATION] 修复前的状态：这里另有一份解析循环，越界策略与 iterEntries 不同
    const lenSize = this.lenFieldSize();
    for (let bi = 0; bi < this.indexEntries.length; bi++) {
      const blockData = this.getBlockData(this.indexEntries[bi]);
      if (!blockData) continue;
      const blockView = new DataView(blockData.buffer, blockData.byteOffset, blockData.byteLength);
      const count = blockView.getUint32(0, false);
      let offset = 4;
      for (let i = 0; i < count; i++) {
        if (offset + lenSize > blockData.byteLength) return;
        const keyLen = blockView.getUint32(offset, false);
        offset += lenSize;
        if (offset + keyLen + lenSize > blockData.byteLength) return;
        const key = new TextDecoder().decode(blockData.slice(offset, offset + keyLen));
        offset += keyLen;
        const valLen = blockView.getUint32(offset, false);
        offset += lenSize;
        if (offset + valLen > blockData.byteLength) return;
        const valBytes = blockData.slice(offset, offset + valLen);
        offset += valLen;
        try { callback(key, JSON.parse(new TextDecoder().decode(valBytes))); } catch { /* skip */ }
      }
    }
  }""",
        test=B6, pattern='三个读取路径对同一个损坏文件给出一致结论',
    ),
    dict(
        name='vacuum 硬编码返回 6（报告与事实无关）',
        file=ENGINE,
        old="""    let compactedLevels = 0;
    const isBottom = (level: number): boolean => level === MAX_LSM_LEVELS - 1;""",
        new="""    let compactedLevels = 6;   // [MUTATION] 旧行为：与事实无关的数字
    const isBottom = (level: number): boolean => level === MAX_LSM_LEVELS - 1;""",
        test=B6, pattern='vacuum 返回',
    ),
    dict(
        name='close() 落盘失败后不释放后端/锁（无 try/finally）',
        file=ENGINE,
        old="""    let failure: unknown = null;
    try {
      await this.flushAllLsms();""",
        new="""    let failure: unknown = null;
    if (true) {   // [MUTATION] 旧行为：失败即中断，收尾逻辑不再执行
      await this.flushAllLsms();
      await this.bufferPool.flushAll();
      await this.advanceWalCheckpoint();
      await this.backend.close();
      if (this.dbLock) { await this.dbLock.release(); this.dbLock = null; }
      this.schemas.clear();
      this.opened = false;
      return;
    }
    try {
      await this.flushAllLsms();""",
        test=B6, pattern='close\\(\\) 在落盘失败时',
    ),
    dict(
        name='恢复报告不聚合 LSM 侧丢弃（数据丢失不可见）',
        file=ENGINE,
        old="""    for (const lsm of this.allLsms()) {
      const r = lsm.getRecoveryReport();
      for (const d of r.droppedSSTables) {
        dropped.push({ namespace: r.namespace, ...d });
      }
      if (r.dataLossSuspected) dataLoss = true;
    }""",
        new="""    // [MUTATION] 不聚合 LSM 报告""",
        test=B6, pattern='恢复报告聚合',
    ),
]


CURRENT = {'path': None, 'content': None}


def _restore_current(*_args):
    """收到信号时也要把源码恢复原状（否则中断会留下变异源码）。"""
    if CURRENT['path'] and CURRENT['content'] is not None:
        io.open(CURRENT['path'], 'w', encoding='utf-8').write(CURRENT['content'])
        print(f"\n[restore] {CURRENT['path']} 已恢复", flush=True)
    sys.exit(130)


for _sig in ('SIGINT', 'SIGTERM', 'SIGHUP'):
    try:
        signal.signal(getattr(signal, _sig), _restore_current)
    except (AttributeError, ValueError):  # pragma: no cover - 平台差异
        pass


def run(cmd):
    return subprocess.run(cmd, cwd=ROOT, shell=True, capture_output=True, text=True)


def main():
    only = sys.argv[1] if len(sys.argv) > 1 else None
    results = []
    for m in MUTATIONS:
        if only and only not in m['name']:
            continue
        path = os.path.join(ROOT, m['file'])
        original = io.open(path, encoding='utf-8').read()
        if m['old'] not in original:
            results.append((m['name'], 'SKIP(锚点未找到)'))
            print(f"[skip] {m['name']}: 锚点未找到", flush=True)
            continue
        mutated = original.replace(m['old'], m['new'], 1)
        CURRENT['path'] = path
        CURRENT['content'] = original
        io.open(path, 'w', encoding='utf-8').write(mutated)
        try:
            # 注意：用 shlex.quote 而不是 repr —— repr 会把正则里的反斜杠再转义一次，
            # 于是 jest 收到 `\\(` 这种模式、匹配不到任何用例，脚本会把"没跑用例"
            # 误判成"测试仍然通过"。
            cmd = (f"npx jest {shlex.quote(m['test'])} -t {shlex.quote(m['pattern'])} "
                   f"--testPathIgnorePatterns='/node_modules/' 2>&1 | tail -80")
            out = run(cmd).stdout
            # 注意：不能只搜 "0 total" —— jest 的 "Snapshots: 0 total" 也会命中，
            # 于是把"跑了用例且失败"误判成"用例未匹配"。
            no_tests = bool(re.search(r'^Tests:\s+0 total', out, re.M)) or ('No tests found' in out)
            failed = ('✕' in out) or ('Tests:' in out and 'failed' in out) or ('●' in out)
            if no_tests:
                status = 'SKIP(用例未匹配)'
            elif failed:
                status = 'FAIL(如预期)'
            else:
                status = 'PASS(!!)'
            results.append((m['name'], status))
            msg = {
                'FAIL(如预期)': '测试失败（变异被拦住）',
                'PASS(!!)': '测试仍然通过！',
                'SKIP(用例未匹配)': '用例未匹配（模式写错了）',
            }[status]
            print(f"[{'ok' if status == 'FAIL(如预期)' else 'BAD'}] {m['name']} → {msg}", flush=True)
        finally:
            io.open(path, 'w', encoding='utf-8').write(original)
            CURRENT['path'] = None
            CURRENT['content'] = None
    print('\n==== 变异验证汇总 ====')
    bad = [r for r in results if r[1] != 'FAIL(如预期)']
    for name, status in results:
        print(f'  {status:14s} {name}')
    if bad:
        print(f'\n{len(bad)} 条变异没有被测试拦住 —— 那些用例只是陪跑。')
        return 1
    print('\n全部变异都被对应用例拦住。')
    return 0


if __name__ == '__main__':
    sys.exit(main())