MetonaTeam/MetonaSqlark
2
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(A38/A39): 页面化路径真正压缩 + compressLZ4 去除二次复杂度（含测试介质目录语义修正）

Browse Source 
A38 `compression` 在页面化路径上被静默忽略
  压缩只写在"整 value 存一个 backend value"的分支里，而 `save()` 在页面化
  分支**提前 return** —— `pageStorage` 默认自动（OPFS 后端下为 true），
  于是 `compression: true` 在默认配置下完全无效且无任何提示。
  修法：`compression` 传入 `PageSSTableStore`，在**切页之前**整体压缩
  （压缩率优于逐页压缩），加载时对称解压。
  连带修正一个会静默损坏数据的接口问题：`SSTableMeta.totalSize` 的语义是
  "页面里存了多少字节"，加载时按它截断 —— 压缩后必须写**压缩长度**。
  为此 `SSTableStore.save` 改为返回 `{ storedSize }`，两处 flush 流程与
  整 value 路径都用它回填 totalSize（写未压缩长度会让压缩数据被 0 填充撑大）。

  为什么此前没被发现：既有测试只断言"压缩后能读回来"，而"根本没压缩"
  同样能正确读回 —— 断言太弱。新用例改为**结构性断言**：
  开启压缩后落盘字节数必须显著下降（>5×），与实现细节无关。

A39 `compressLZ4` 匹配搜索为 O(n²)
  旧实现逐字节向前扫描最多 65535 个候选位置、每个位置再逐字节比较 ——
  在低压缩率数据上退化为二次复杂度。实测 60KB 伪随机输入耗时 **2345ms**；
  而 SSTable 页/日志段正是几百 KB 到几 MB，属于普通写入路径上的真实卡顿。
  修法：改为 LZ4 标准的 **4 字节哈希链**（`head[]`/`prev[]`，单点最多
  `MAX_CHAIN=32` 次探测）→ 实测 6ms（约 390×）。
  **输出格式完全不变**，既有落盘数据无需迁移；旧实现保留为
  `compressLZ4LinearReference` 并作为测试对照物（证明两者可互解）。
  另加"全字面量"兜底：任何异常都产出合法可解压的流（数据正确性优先于压缩率）。

测试介质修正（同源发现，影响所有 OPFS 多库场景）
  `installOPFSMock` 把 `getDirectoryHandle(name)` 的 `name` **丢弃**，
  所有库共用一棵扁平文件树。实测：`open('db-alpha')` 建表后
  `open('db-beta').getTableNames()` 返回 `["alpha_only"]`。
  真实 OPFS 下 `OPFSBackend.open(name)` 是 `root.getDirectoryHandle(name)`，
  因此 mock 现在实现真实的**目录语义**，并提供 `dir(dbName)` 视图让测试与
  生产代码使用同一个 API（此前的 `listKeys/createFile` 是根目录假 API，
  两个依赖它的用例已改为目录视图）。

验证：新增 tests/engine/aria-compression.test.ts（12 项，含 1MB 大输入与
6 组格式兼容用例）；两处修复都做**变异验证**：回退 A38 的接线 → 页面化压缩
用例失败；回退 A39 到线性实现 → "60KB < 1s" 用例失败（实测 2397ms）。
全量 89 套件 / 1742 测试通过；typecheck、lint、build 零错误/零告警；dist 已重建。
This commit is contained in:
thzxx
2026-09-15 01:52:57 +08:00
parent 85f0f170a4
commit 97b9fa486d
15 changed files with 1327 additions and 517 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/engine/aria-compression.test.ts
+194
View File
@@ -0,0 +1,194 @@
/**
* v0.8.0 回归套件 —— A38 页面化压缩 + A39 compressLZ4 复杂度
* ============================================================================
* A38 **`compression` 在页面化路径上被静默忽略**
* `AriaEngine` 的默认配置是 `pageStorage` 自动（OPFS 后端下为 true），
* 而压缩只写在"整 value 存一个 backend value"的旧分支里，页面化分支
* **提前 return** —— 于是 `compression: true` 在默认配置下完全没有效果，
* 且没有任何提示。本套件的断言方式是**结构性**的：
* 开启压缩后 `SSTableMeta.totalSize`（实际落盘字节数）必须显著变小 ——
* 它与实现细节无关，只反映"数据真的被压缩了"。
*
* * 为什么之前没被发现：既有测试只断言"压缩后能读回来"（往返正确），
* 而"根本没压缩"同样能正确读回 —— 断言太弱，测不出"功能是否生效"。
*
* A39 **`compressLZ4` 匹配搜索是 O(n²)**
* 旧实现逐字节向前扫描最多 65535 个候选位置，每个位置再逐字节比较。
* 在低压缩率数据（伪随机）上退化为二次复杂度：实测 60KB 输入耗时 **2.3 秒**
* （新实现 6ms，约 390×）。SSTable 页/日志段正好是几百 KB 到几 MB，
* 因此这是普通写入路径上的真实卡顿，不是极端场景。
*
* 本套件同时锁定"输出格式未变"：新实现与保留的线性参考实现在同一输入上
* 必须产出**可互相解压**的流（压缩率可略有差异，因为新实现能找到不同但
* 同样合法的匹配）。
*/
import { AriaEngine } from '../../src/engine/aria/index';
import { compressLZ4, compressLZ4LinearReference, decompressLZ4 } from '../../src/engine/aria/compression/lz4';
import { resetOPFSMock } from '../helpers/storage-harness';
beforeEach(() => { resetOPFSMock(); });
const SCHEMA = () => ({
name: 't',
columns: {
id: { type: 'string' as const, primaryKey: true },
blob: { type: 'string' as const },
},
});
/** 高度可压缩的载荷 */
const COMPRESSIBLE = 'x'.repeat(500) + 'y'.repeat(500);
/** 写入 n 行并返回 SSTable 元数据统计 */
async function writeAndMeasure(opts: {
dbName: string;
pageStorage: boolean;
compression: boolean;
rows?: number;
}): Promise<{ storedBytes: number; sstableCount: number; rowCount: number; roundTripOk: boolean }> {
const rows = opts.rows ?? 60;
const engine = new AriaEngine({
storageBackend: 'opfs',
pageStorage: opts.pageStorage,
compression: opts.compression,
memtableSizeThreshold: 2048,
checkpointInterval: 100_000_000,
} as never);
await engine.open(opts.dbName, 1);
await engine.createTable(SCHEMA() as never);
for (let i = 0; i < rows; i++) await engine.insert('t', [{ id: `k${i}`, blob: COMPRESSIBLE }]);
const lsm = (engine as unknown as { lsm: { sstableStore: { listMeta(): Promise<Array<{ totalSize: number }>> } } }).lsm;
const metas = await lsm.sstableStore.listMeta();
const storedBytes = metas.reduce((sum, m) => sum + (m.totalSize ?? 0), 0);
const read = await engine.find('t', { table: 't' });
const roundTripOk = read.length === rows && read.every((r) => r.blob === COMPRESSIBLE);
await engine.close();
return { storedBytes, sstableCount: metas.length, rowCount: read.length, roundTripOk };
}
describe('[v0.8.0] A38 compression 必须真的生效（含页面化路径）', () => {
it('页面化路径：开启压缩后落盘字节数显著下降', async () => {
const off = await writeAndMeasure({ dbName: 'a38-pages-off', pageStorage: true, compression: false });
const on = await writeAndMeasure({ dbName: 'a38-pages-on', pageStorage: true, compression: true });
// 先确认两次实验都写出了 SSTable（否则下面的比值没有意义）
expect(off.sstableCount).toBeGreaterThan(0);
expect(on.sstableCount).toBeGreaterThan(0);
// 数据完整性不受压缩影响
expect(off.roundTripOk).toBe(true);
expect(on.roundTripOk).toBe(true);
// 核心断言：修复前 compression 在页面化路径上被忽略 → 两者字节数相同
expect(on.storedBytes).toBeLessThan(off.storedBytes);
// 高可压缩内容应有的量级（10 倍是保守下界，实测远高于此）
expect(off.storedBytes).toBeGreaterThan(on.storedBytes * 5);
});
it('整 value 路径：压缩同样生效（两条路径行为一致）', async () => {
const off = await writeAndMeasure({ dbName: 'a38-whole-off', pageStorage: false, compression: false });
const on = await writeAndMeasure({ dbName: 'a38-whole-on', pageStorage: false, compression: true });
expect(off.roundTripOk).toBe(true);
expect(on.roundTripOk).toBe(true);
expect(off.storedBytes).toBeGreaterThan(on.storedBytes * 5);
});
it('页面化 + 压缩：重开后数据与索引完整', async () => {
const engine = new AriaEngine({
storageBackend: 'opfs', pageStorage: true, compression: true,
memtableSizeThreshold: 2048, checkpointInterval: 100_000_000,
} as never);
await engine.open('a38-reopen', 1);
await engine.createTable(SCHEMA() as never);
for (let i = 0; i < 40; i++) await engine.insert('t', [{ id: `k${i}`, blob: COMPRESSIBLE }]);
// 不 close（模拟崩溃后重开，覆盖压缩数据的恢复路径）
const engine2 = new AriaEngine({
storageBackend: 'opfs', pageStorage: true, compression: true,
memtableSizeThreshold: 2048, checkpointInterval: 100_000_000,
} as never);
await engine2.open('a38-reopen', 1);
const rows = await engine2.find('t', { table: 't' });
expect(rows).toHaveLength(40);
expect(rows.every((r) => r.blob === COMPRESSIBLE)).toBe(true);
// 单行精确读取（走 SSTable 数据块解析，而不只是全表扫描）
const one = await engine2.find('t', { table: 't', where: { id: 'k7' } });
expect(one).toHaveLength(1);
expect(one[0].blob).toBe(COMPRESSIBLE);
await engine.close().catch(() => { /* 已崩溃语义，忽略 */ });
await engine2.close();
});
it('压缩关闭时数据同样完整（回归护栏）', async () => {
const result = await writeAndMeasure({ dbName: 'a38-nocomp', pageStorage: true, compression: false });
expect(result.roundTripOk).toBe(true);
expect(result.rowCount).toBe(60);
});
});
// ---------------------------------------------------------------------------
// A39：compressLZ4 复杂度与格式兼容
// ---------------------------------------------------------------------------
/** 确定性伪随机（避免测试本身依赖随机源） */
function makeRandom(length: number, seed = 42): Uint8Array {
let s = seed >>> 0;
const out = new Uint8Array(length);
for (let i = 0; i < length; i++) {
s = (Math.imul(s, 1664525) + 1013904223) >>> 0;
out[i] = (s >>> 24) & 0xFF;
}
return out;
}
describe('[v0.8.0] A39 compressLZ4：复杂度与格式兼容', () => {
const cases: Array<[string, Uint8Array]> = [
['空输入', new Uint8Array(0)],
['单字节', new Uint8Array([42])],
['全同字节', new Uint8Array(5000).fill(7)],
['周期序列', new Uint8Array(Array.from({ length: 40_000 }, (_, i) => i % 7))],
['伪随机 8KB', makeRandom(8_000)],
['伪随机 60KB', makeRandom(60_000)],
];
it.each(cases)('%s：往返一致且与线性参考实现可互解', (_label, data) => {
const fast = compressLZ4(data);
const slow = compressLZ4LinearReference(data);
// 新实现自身往返
const fromFast = decompressLZ4(fast);
expect(fromFast.length).toBe(data.length);
expect(Array.from(fromFast)).toEqual(Array.from(data));
// 旧实现的输出，新解压器必须能读（格式兼容：既有落盘数据不需要迁移）
const fromSlow = decompressLZ4(slow);
expect(Array.from(fromSlow)).toEqual(Array.from(data));
// 新实现的输出，旧解压器（同一份代码，此处仅作对称性验证）也必须能读
expect(Array.from(decompressLZ4(fast))).toEqual(Array.from(data));
});
it('低压缩率数据不再退化（60KB 伪随机在 1 秒内完成）', () => {
const data = makeRandom(60_000);
const start = Date.now();
const compressed = compressLZ4(data);
const elapsed = Date.now() - start;
// 修复前该输入耗时约 2.3 秒（逐位置线性扫描 → 二次复杂度）。
// 阈值取 1 秒：即使 CI 慢 10 倍也仍能通过，而回退到旧实现必然失败。
expect(elapsed).toBeLessThan(1000);
expect(decompressLZ4(compressed).length).toBe(data.length);
});
it('大输入（1MB 可压缩内容）在合理时间内完成', () => {
const data = new Uint8Array(1_000_000);
for (let i = 0; i < data.length; i++) data[i] = i % 251;
const start = Date.now();
const compressed = compressLZ4(data);
const elapsed = Date.now() - start;
expect(elapsed).toBeLessThan(3000);
// 周期性内容应被显著压缩
expect(compressed.length).toBeLessThan(data.length);
expect(decompressLZ4(compressed).length).toBe(data.length);
});
});
tests/engine/aria-page-store.test.ts
+3 -1
View File
@@ -260,7 +260,9 @@ describe('AriaEngine — 页面化 SSTable 集成（OPFS）', () => {
await (engine as any).lsm.flush();
await engine.close();
const backendKeys1 = await opfs.listKeys();
// v0.8.0：mock 现在是"真实目录语义"（文件在 `${dbName}/` 下），
// 因此要用与生产代码相同的目录视图查询，而不是根目录列表。
const backendKeys1 = await opfs.dir('page-db-4').listKeys();
expect(backendKeys1.some((k) => k.startsWith('sst_'))).toBe(true);
// 阶段 2：页面模式打开（默认 opfs → 启用），读旧数据 + 写新数据
tests/engine/aria-repair-hardening.test.ts
+8 -5
View File
@@ -71,14 +71,17 @@ describe('AriaEngine — repair 自愈增强', () => {
await engine.createTable(SCHEMA());
await engine.insert('items', [{ id: 'a', val: 1, tag: 'x' }]);
// 制造残留（模拟 createWritable 中断留下的临时文件）
await opfs.createFile('junk.crswap');
await opfs.createFile('junk2.tmp');
expect((await opfs.listKeys()).some((k) => k.endsWith('.crswap'))).toBe(true);
// 制造残留（模拟 createWritable 中断留下的临时文件）。
// v0.8.0：mock 现在是真实目录语义，残留必须落在**该库的目录**里 ——
// repair 的 cleanupStaleFiles 扫的是库目录（root 上的文件不属于任何库）。
const dir = opfs.dir('repair-opfs-1');
await dir.createFile('junk.crswap');
await dir.createFile('junk2.tmp');
expect((await dir.listKeys()).some((k) => k.endsWith('.crswap'))).toBe(true);
await (engine as any).repair();
const after = await opfs.listKeys();
const after = await dir.listKeys();
expect(after.some((k) => k.endsWith('.crswap'))).toBe(false);
expect(after.some((k) => k.endsWith('.tmp'))).toBe(false);
expect(await engine.count('items')).toBe(1);
tests/helpers/storage-harness.ts
+128 -60
View File
@@ -297,18 +297,27 @@ export function resetOPFSMock(dbName = 'mock'): InstalledOPFSMock {
export interface InstalledOPFSMock {
store: TransactionalFileStore;
/** 当前 mock 看到的所有文件名 */
listKeys(): Promise<string[]>;
/**
* 直接创建（或覆盖）一个文件 —— 用于制造"崩溃残留临时文件"等场景，
* 替代旧 API 暴露内部 dir/files 的做法。
* v0.8.0：取**某个库目录**的句柄 —— 测试与生产代码（`OPFSBackend.open`）
* 使用同一个 API，因此"文件放在哪个目录"不可能出现两套理解。
*
* 与之配套的 `listKeys/createFile/...` 是**根目录**视图（`dbName/文件` 形态），
* 只适合"制造残留文件""看整体布局"这类场景；判断某个文件是否存在，
* 应当用 `dir(dbName).hasFile(...)`。
*/
dir(dbName: string): {
listKeys(): Promise<string[]>;
createFile(name: string, content?: ArrayBuffer): Promise<void>;
hasFile(name: string): Promise<boolean>;
readFile(name: string): Promise<ArrayBuffer | null>;
removeFile(name: string): Promise<void>;
};
/** 根目录下的所有文件（含库名前缀） */
listKeys(): Promise<string[]>;
/** 在**根目录**直接创建文件 —— 用于制造"崩溃残留临时文件"等场景 */
createFile(name: string, content?: ArrayBuffer): Promise<void>;
/** 文件是否存在 */
hasFile(name: string): Promise<boolean>;
/** 读取文件内容（返回副本） */
readFile(name: string): Promise<ArrayBuffer | null>;
/** 删除文件 */
removeFile(name: string): Promise<void>;
}
@@ -320,77 +329,136 @@ export interface InstalledOPFSMock {
export function installOPFSMock(dbName = 'mock'): InstalledOPFSMock {
const store = getStore(dbName);
const makeFileHandle = async (name: string, opts?: { create?: boolean }) => {
if (!(await store.has(name)) && !opts?.create) {
throw new Error(`NotFoundError: ${name}`);
}
/** 文件路径 = `${目录名}/${文件名}`（root 目录的目录名为空串） */
const joinPath = (dirName: string, fileName: string): string =>
dirName ? `${dirName}/${fileName}` : fileName;
/**
* 构造某个目录的 FileSystemDirectoryHandle 视图。
*
* v0.8.0：**目录名必须真正参与文件路径**。
*
* `OPFSBackend.open(name)` 会 `root.getDirectoryHandle(name, {create:true})`
* 并把返回的目录当作该库的根 —— 真实 OPFS 因此天然按库名隔离文件。
* 而此前的 mock 把这个 `name` **丢掉**（`async (_name) => dir`），
* 所有库共用同一个扁平名字空间。实测（未修复时）：
* aroma.open('db-alpha') 建表 alpha_only → aroma.open('db-beta')
* → `getTableNames()` 返回 ["alpha_only"]（beta 看到了 alpha 的表）
* 影响面：所有基于本 mock 的"多库/多租户/重启换名"场景都跑在错误语义上。
*/
const makeDir = (dirName: string) => {
const makeFileHandle = async (name: string, opts?: { create?: boolean }) => {
const path = joinPath(dirName, name);
if (!(await store.has(path)) && !opts?.create) {
throw new Error(`NotFoundError: ${path}`);
}
return {
/**
* 真实 OPFS 的 getFile() 返回 File，读的是**已提交**内容（size 与 arrayBuffer 一致）。
* append 路径会读 `existing.size` 来定位追加位置，因此这里保持二者同源。
*/
getFile: async () => {
const content = (await store.read(path)) ?? new ArrayBuffer(0);
return {
size: content.byteLength,
arrayBuffer: async () => content,
};
},
createWritable: async (wOpts?: { keepExistingData?: boolean }) => {
const keepExisting = wOpts?.keepExistingData ?? false;
// OPFS 写语义：keepExistingData:false 时从空缓冲开始（旧 mock 会保留旧字节）
let buffer = keepExisting ? (await store.read(path)) ?? new ArrayBuffer(0) : new ArrayBuffer(0);
return {
write: async (arg: ArrayBuffer | { type: string; position: number; data: ArrayBuffer }) => {
// 注意：不能用 `arg instanceof ArrayBuffer` 判别 —— 跨 realm / 跨 Buffer 实现时
// 会失效（jsdom 与 Node 的 ArrayBuffer 可能不是同一个构造函数），
// 从而把 ArrayBuffer 误当成 {position,data} 分支。改用结构判别。
if (!isPositionedWrite(arg)) {
buffer = copyBuffer(arg as ArrayBuffer);
return;
}
const chunk = arg;
const end = chunk.position + chunk.data.byteLength;
const merged = new Uint8Array(Math.max(end, buffer.byteLength));
merged.set(new Uint8Array(buffer), 0);
merged.set(new Uint8Array(chunk.data), chunk.position);
buffer = merged.buffer as ArrayBuffer;
},
close: async () => {
// 仅在 close 时提交 —— 未 close 的写入对读不可见（旧 mock 会立即可见）
await store.write(path, buffer);
store.commitAll();
},
};
},
};
};
return {
/**
* 真实 OPFS 的 getFile() 返回 File，读的是**已提交**内容（size 与 arrayBuffer 一致）。
* append 路径会读 `existing.size` 来定位追加位置，因此这里保持二者同源。
*/
getFile: async () => {
const content = (await store.read(name)) ?? new ArrayBuffer(0);
return {
size: content.byteLength,
arrayBuffer: async () => content,
};
getFileHandle: makeFileHandle,
entries: async function* () {
const prefix = dirName ? `${dirName}/` : '';
for (const key of await store.listKeys()) {
if (!key.startsWith(prefix)) continue;
const rest = key.slice(prefix.length);
// 只列出本目录的**直接**子项（真实 OPFS 的 entries 语义）
if (rest.includes('/')) continue;
yield [rest];
}
},
createWritable: async (wOpts?: { keepExistingData?: boolean }) => {
const keepExisting = wOpts?.keepExistingData ?? false;
// OPFS 写语义：keepExistingData:false 时从空缓冲开始（旧 mock 会保留旧字节）
let buffer = keepExisting ? (await store.read(name)) ?? new ArrayBuffer(0) : new ArrayBuffer(0);
return {
write: async (arg: ArrayBuffer | { type: string; position: number; data: ArrayBuffer }) => {
// 注意：不能用 `arg instanceof ArrayBuffer` 判别 —— 跨 realm / 跨 Buffer 实现时
// 会失效（jsdom 与 Node 的 ArrayBuffer 可能不是同一个构造函数），
// 从而把 ArrayBuffer 误当成 {position,data} 分支。改用结构判别。
if (!isPositionedWrite(arg)) {
buffer = copyBuffer(arg as ArrayBuffer);
return;
}
const chunk = arg;
const end = chunk.position + chunk.data.byteLength;
const merged = new Uint8Array(Math.max(end, buffer.byteLength));
merged.set(new Uint8Array(buffer), 0);
merged.set(new Uint8Array(chunk.data), chunk.position);
buffer = merged.buffer as ArrayBuffer;
},
close: async () => {
// 仅在 close 时提交 —— 未 close 的写入对读不可见（旧 mock 会立即可见）
await store.write(name, buffer);
store.commitAll();
},
};
removeEntry: async (name: string) => {
await store.delete(joinPath(dirName, name));
store.commitAll();
},
/** 子目录：`OPFSBackend.open(name)` 走这里 */
getDirectoryHandle: async (name: string, _opts?: unknown) => makeDir(joinPath(dirName, name)),
};
};
const dir = {
getFileHandle: makeFileHandle,
entries: async function* () {
for (const key of await store.listKeys()) yield [key];
},
removeEntry: async (name: string) => {
await store.delete(name);
store.commitAll();
},
};
const rootDir = makeDir('');
Object.defineProperty(globalThis, 'navigator', {
value: {
storage: {
getDirectory: async () => ({
getDirectoryHandle: async (_name: string, _opts?: unknown) => dir,
}),
getDirectory: async () => rootDir,
},
},
configurable: true,
writable: true,
});
/**
* 构造"某个库目录"的测试视图。
*
* 与生产代码同路径：`OPFSBackend.open(name)` 拿到的是
* `root.getDirectoryHandle(name)`，本函数走同一个入口。
*/
const dirView = (name: string) => {
const dirName = joinPath('', name);
return {
listKeys: async (): Promise<string[]> => {
const prefix = `${dirName}/`;
return (await store.listKeys())
.filter((k) => k.startsWith(prefix) && !k.slice(prefix.length).includes('/'))
.map((k) => k.slice(prefix.length));
},
createFile: async (fileName: string, content?: ArrayBuffer): Promise<void> => {
await store.write(joinPath(dirName, fileName), content ?? new ArrayBuffer(0));
store.commitAll();
},
hasFile: (fileName: string) => store.has(joinPath(dirName, fileName)),
readFile: (fileName: string) => store.read(joinPath(dirName, fileName)),
removeFile: async (fileName: string): Promise<void> => {
await store.delete(joinPath(dirName, fileName));
store.commitAll();
},
};
};
return {
store,
dir: dirView,
// 辅助方法面向**根目录**（调用方给出的名字即完整相对路径，便于制造残留文件）
listKeys: () => store.listKeys(),
createFile: async (name, content) => {
await store.write(name, content ?? new ArrayBuffer(0));