按 PLAN-v0.7.5.md §B-6 的**完整规格**实施(此前只落地了"降级选项"里的五处止血):
B-6 要求的是 `__aria_manifest` 单一提交点 + LSM 单项改造。完整记录见方案附录 H。
一、单一提交点
- 新增 `src/engine/aria/store/manifest.ts`:`__aria_manifest_<generation>`
(magic + formatVersion + generation + 头部 CRC + 载荷 CRC;先写后验;保留两代)。
载荷 = 页面水位 + 各命名空间 SSTable 元数据 + 表结构 + WAL 起始位置 + 待落盘冻结表意图。
- 顺序固定:**数据落盘 → manifest 提交 → 才允许截断 WAL / 删除旧文件 / 删除旧 SSTable**。
- 恢复只认最后一份 CRC 通过的世代;全部世代无效 → `ARIA_MANIFEST_CORRUPT`
(修复前:裸 JSON meta 解析失败 → `[]` → 静默空库,随后 repair 还会删光活页)。
- 旧格式(__aria_lsm_meta/__aria_schemas/__aria_meta)首次打开自动迁移,旧键保留;
迁移遇到损坏 → `ARIA_LEGACY_META_CORRUPT`。
- 陈旧实例保护(STALE_INSTANCE):认领时一次跨过 MANIFEST_TAKEOVER_STRIDE 个世代,
杜绝"旧实例在途提交落在同一世代号上"(实测第二个实例 open 直接失败)。
二、LSM
- 44 冻结表成为一等状态:失败保留 + 可重试(修复前失败即永久失去落盘机会)。
- 45 `flush()` 先入链再报告后台错误(修复前一次后台失败会让之后每次 flush 直接抛错、
数据永远等不到落盘);被重试修复的失败进 `getBackgroundWarnings()`(可见但不误报失败)。
- 47 `MergeIterator` 胜出来源的补充推迟到下一次 `next()`:提前终止不再多算一条。
- 49 `compacting` 由单 boolean 改为按层集合(跨层触发不再被静默丢弃)。
- 50 compaction 不再"先 splice 整层再合并"(窗口内该层对读者可见);
被取代的 SSTable 进"退休表" + 读者 epoch,等更早读者退出才物理删除。
- 51 底部层原地合并回收墓碑(删除密集场景空间不再无界增长);"整层只剩墓碑" 有专门分支
(修复前会读 `merged[0][0]` 抛 TypeError,compaction 永久失败)。
- 55 flush 与 compaction 拆成两条链,checkpoint 只落 memtable;删除引擎层全部
`prefetch*`/`drainChain` 依赖,改为"快照 + 结构版本乐观重试"
(版本号同时覆盖 levels 与前台 memtable/frozen 的变化)。
- 读路径自洽:介质读故障抛 `ARIA_SSTABLE_READ_FAILED`,不再折叠成"文件不存在"误删元数据。
三、WAL
- LSN 全库单调(manifest 记高水位);按水位删除旧分片(`planKeepFrom` → 提交 → 再删除)。
- **分片号只增不减**:修复前全量截断后重置为 0,会与 manifest 记录的 startSegment 错位,
实测造成两个方向的损坏(删掉的行复活 / 已确认写入丢失,见随机压力套件)。
- 分片空洞(含前缀缺失)显式报 `ARIA_WAL_GAP`,不再静默丢弃尾部。
四、其它
- `sstable.ts` 三份解析循环合并为 `iterEntries()`,越界策略统一。
- `vacuum()` 返回真实压缩层数(修复前硬编码 6 且底部层永不压缩)。
- `close()` 加 try/finally(落盘失败也必须释放后端/锁并复位状态)。
- `getRecoveryReport()`:{droppedSSTables, dataLossSuspected, walGaps, legacyImported,
manifestFallback} —— "自愈了什么、有没有真丢数据"成为可读返回值。
五、验证
- 新增 `tests/v080-b6-single-commit-point.test.ts`(63 项,含 manifest 严格校验表驱动 25 例)。
- 新增 `scripts/mutation-b6.py`:22 项变异验证(把每个修复回退到修复前行为,对应用例必须失败),
全部被拦住 —— 这批用例不是陪跑。
- 常规套件 1935 通过 / 91 套件;覆盖率 90.34 / 82.16 / 94.06 / 93.23(阈值 90/82/94/93);
e2e 14/14;重型套件 4 套件 27 项全绿。
374 lines
16 KiB
Python
374 lines
16 KiB
Python
#!/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())
|