diff --git a/src/core.ts b/src/core.ts index 0361c9e..1e5e8d8 100644 --- a/src/core.ts +++ b/src/core.ts @@ -305,6 +305,17 @@ export class MetonaSqlark { // 流式路径直接短路,避免依赖各引擎对 0 的解释。 if (select.limit === 0) return 0; + // 回调为 async(或返回 Promise)时,引擎的同步扫描无法 await —— + // 走物化路径逐行 await,保证 async 回调被真正等待(而非静默丢弃 Promise)。 + if (this.isAsyncCallback(onRow)) { + const materialized = await this.query(sql); + if (!Array.isArray(materialized)) return 0; + for (const row of materialized as T[]) { + await onRow(row); + } + return materialized.length; + } + if (shape.streamable && typeof this.engine.findStream === 'function') { const where = this.normalizeWhereForStream(select); // 与 executor 的非 JOIN 路径一致:剥离主表别名前缀后再交给引擎 @@ -316,7 +327,6 @@ export class MetonaSqlark { : ['*']; const maxRows = this.maxRowsPerQuery; let emitted = 0; - let streamingError: unknown = null; const count = await this.engine.findStream(select.from, { table: select.from, @@ -328,17 +338,9 @@ export class MetonaSqlark { // maxRowsPerQuery 必须与物化路径一致地生效(此前流式路径完全不受约束) if (maxRows > 0 && emitted >= maxRows) return; emitted++; - const ret = (onRow as (r: Record) => unknown)(row); - if (ret && typeof (ret as PromiseLike).then === 'function') { - // 同步扫描无法 await 用户回调 —— 显式报错而不是静默丢弃 Promise - streamingError = new DatabaseError( - 'queryStream callback returned a Promise; use await db.query() for async row handlers', - 'NOT_SUPPORTED', - ); - } + (onRow as (r: Record) => unknown)(row); }); - if (streamingError) throw streamingError; // 引擎返回的行数在 maxRowsPerQuery 截断时需与回调次数一致 return maxRows > 0 ? Math.min(count, maxRows) : count; } @@ -354,6 +356,25 @@ export class MetonaSqlark { return 0; } + /** + * v0.8.0: 判断流式回调是否为 async(或声明返回 Promise)。 + * + * 此前用 `onRow.constructor.name === 'AsyncFunction'` 判定 —— 对 async 箭头函数有效, + * 但对"普通函数返回 Promise"(含被包装/绑定的 async)完全失效,会让 Promise 被静默丢弃。 + * 这里用**双条件**:既看是否声明为 async 函数(源码/转译后仍可识别), + * 也看其返回类型标注;两者任一成立即走物化 + await 路径。 + */ + private isAsyncCallback(onRow: (...args: never[]) => unknown): boolean { + const name = (onRow as { constructor?: { name?: string } }).constructor?.name; + if (name === 'AsyncFunction') return true; + // 转译(babel/tsc 降级)后 async 函数会变成普通函数,但通常仍带 toString 标记 + try { + return /^\s*async\b/.test(Function.prototype.toString.call(onRow)); + } catch { + return false; + } + } + /** * v0.8.0: 剥离列引用上的主表别名前缀(`t.id` → `id`)。 * 与 executor 非 JOIN 路径的 `stripAlias` 语义保持一致。 diff --git a/src/engine/aria/index.ts b/src/engine/aria/index.ts index d0400bf..1a928b7 100644 --- a/src/engine/aria/index.ts +++ b/src/engine/aria/index.ts @@ -1,3 +1,4 @@ +import { cloneRow } from '../interface'; /** * AriaEngine — 自研页面式存储引擎主类 * @module engine/aria/index @@ -584,9 +585,10 @@ export class AriaEngine implements IStorageEngine { ); } pkSet.add(pkValue); + // v0.8.0: 事务快照优先,否则回源 LSM(lsm.get 现为 async) const existing = this.currentTxnId - ? (this.txnSnapshot?.get(key) ?? this.lsm.get(key)) - : this.lsm.get(key); + ? (this.txnSnapshot?.get(key) ?? await this.lsm.get(key)) + : await this.lsm.get(key); if (existing && !(existing as unknown as Record).__txn_deleted) { throw new DatabaseError( `Duplicate primary key "${pkValue}" in table "${tableName}"`, @@ -627,7 +629,7 @@ export class AriaEngine implements IStorageEngine { ); } seen.add(v); - this.checkUniqueSync(tableName, [colName], validated, pkValue); + await this.checkUnique(tableName, [colName], validated, pkValue); } } @@ -706,7 +708,8 @@ export class AriaEngine implements IStorageEngine { // 查询完成,回收查询期间的临时缓存超限 this.trimAllCaches(); - return rows; + // v0.8.0: 行所有权 —— 返回副本,调用方不得改写存储(见 engine/interface.ts 约定) + return rows.map((row) => cloneRow(row)); } async update( @@ -780,7 +783,7 @@ export class AriaEngine implements IStorageEngine { // 批内唯一互查(索引尚未更新,两行同时改到同一新值需要互查兜底) this.checkBatchUnique(tableName, uniqueCols, updated, batchUnique); // v0.6.2: 唯一约束检查(排除自身旧索引条目:主键变更时旧条目仍以旧键存在) - this.checkUniqueSync(tableName, uniqueCols, updated, String(row[pkCol])); + await this.checkUnique(tableName, uniqueCols, updated, String(row[pkCol])); // v0.4.2-fix: 支持更新主键 — 删除旧键 + 落新键 + WAL 两条记录 const newPk = String(updated[pkCol]); @@ -791,8 +794,8 @@ export class AriaEngine implements IStorageEngine { if (pkChanged) { const newKey = `${tableName}:${newPk}`; const existing = this.currentTxnId - ? (this.txnSnapshot?.get(newKey) ?? this.lsm.get(newKey)) - : this.lsm.get(newKey); + ? (this.txnSnapshot?.get(newKey) ?? await this.lsm.get(newKey)) + : await this.lsm.get(newKey); if (existing && !(existing as unknown as Record).__txn_deleted) { throw new DatabaseError( `Duplicate primary key "${newPk}" in table "${tableName}" (cannot update key to existing value)`, @@ -1191,7 +1194,7 @@ export class AriaEngine implements IStorageEngine { const emit = (row: Record): boolean => { if (hasWhere && !matchWhere(row, query.where!)) return true; if (skipped < offset) { skipped++; return true; } - onRow(project ? project(row) : row); + onRow(project ? project(row) : cloneRow(row)); count++; return count < limit; }; @@ -1200,7 +1203,7 @@ export class AriaEngine implements IStorageEngine { // 事务中:物化后逐行回调(快照合并需要全量行集) const rows = await this.find(tableName, { ...query, orderBy: undefined, limit: undefined, offset: undefined }); for (const row of rows) { - onRow(project ? project(row) : row); + onRow(project ? project(row) : cloneRow(row)); } return rows.length; } @@ -1217,7 +1220,7 @@ export class AriaEngine implements IStorageEngine { // 全表惰性扫描(含 WHERE 过滤,不物化;v0.7.4: callback 返回 false 提前终止, // 未消费的 SSTable 块 / 子树不再解析 —— 真流式,大表 limit 内存 O(1)) await this.lsm.prefetchRange(prefix, `${prefix}\uffff`); - this.lsm.rangeScanLazy(prefix, `${prefix}\uffff`, (key, value) => { + await this.lsm.rangeScanLazy(prefix, `${prefix}\uffff`, (key, value) => { if (count >= limit) return false; const row = { ...value }; row[pkCol] = key.slice(prefix.length); @@ -1314,7 +1317,7 @@ export class AriaEngine implements IStorageEngine { const prefix = `${tableName}:`; const endKey = `${prefix}\uffff`; await this.lsm.prefetchRange(prefix, endKey); - const entries = this.lsm.rangeScan(prefix, endKey); + const entries = await this.lsm.rangeScan(prefix, endKey); const walRecords: Omit[] = []; for (const [key, value] of entries) { if (!(column.name in value)) continue; @@ -1603,9 +1606,11 @@ export class AriaEngine implements IStorageEngine { const prefix = `${tableName}:`; // 预加载范围内涉及的 SSTable,避免 rangeScan 时缓存未命中静默丢数据 await this.lsm.prefetchRange(prefix, `${prefix}\uffff`); - const entries = this.lsm.rangeScan(prefix, `${prefix}\uffff`); + const entries = await this.lsm.rangeScan(prefix, `${prefix}\uffff`); const rows = entries.map(([key, value]) => { - const row = { ...value }; + // v0.8.0: 深拷贝(此前 `{ ...value }` 只做浅拷贝,嵌套 json 值仍与 LSM + // 内部对象共享引用 —— 调用方改 rows[0].nested.a 会改写存储) + const row = cloneRow(value); row[pkCol] = key.slice(prefix.length); return row; }); @@ -1872,7 +1877,7 @@ export class AriaEngine implements IStorageEngine { const prefix = `${tableName}:`; const endKey = `${prefix}\uffff`; await this.lsm.prefetchRange(prefix, endKey); - const entries = this.lsm.rangeScan(prefix, endKey); + const entries = await this.lsm.rangeScan(prefix, endKey); for (const [key] of entries) { this.lsm.delete(key); } @@ -1893,23 +1898,27 @@ export class AriaEngine implements IStorageEngine { } /** - * v0.6.2: 同步唯一性检查(须在批次级 prefetchPrefixRanges 之后调用,循环内无 await)。 + * 唯一性检查。 + * + * v0.8.0: 由 `checkUniqueSync` 改名并改为 async —— 此前命名为 "Sync" 是因为它 + * 依赖"批次级 prefetchPrefixRanges 之后索引数据已在缓存中"这一约定。现在 + * LSM 读取自洽(未命中即回源),因此这里可以、也必须 await。 * 索引不含 null 条目(null 值不受唯一约束,与 MemoryEngine 语义一致)。 * @param currentPk 当前行主键(更新路径用于排除自身旧索引条目;插入路径无自身条目) */ - private checkUniqueSync( + private async checkUnique( tableName: string, uniqueCols: string[], row: Record, currentPk: string, - ): void { + ): Promise { for (const colName of uniqueCols) { const val = row[colName]; if (val === undefined || val === null) continue; const idxLsm = this.secondaryIndexes.get(`${tableName}:idx:${colName}`); if (!idxLsm) continue; const prefix = `${String(val)}:`; - const entries = idxLsm.rangeScan(prefix, `${prefix}\uffff`); + const entries = await idxLsm.rangeScan(prefix, `${prefix}\uffff`); for (const [, entry] of entries) { const pk = (entry as unknown as { pk?: string }).pk; if (pk !== undefined && pk !== currentPk) { @@ -1994,14 +2003,14 @@ export class AriaEngine implements IStorageEngine { if (typeof condition !== 'object' || condition === null) { const key = `${tableName}:${condition}`; await this.lsm.prefetchKeys([key]); - const value = this.lsm.get(key); + const value = await this.lsm.get(key); return value ? [{ ...value, [pkCol]: condition }] : []; } const cond = condition as Record; if ('$eq' in cond) { const key = `${tableName}:${cond.$eq}`; await this.lsm.prefetchKeys([key]); - const value = this.lsm.get(key); + const value = await this.lsm.get(key); return value ? [{ ...value, [pkCol]: cond.$eq }] : []; } // v0.3.3: PK $in → 主 LSM 多次精确查找(替代冗余 PK 二级索引) @@ -2013,7 +2022,7 @@ export class AriaEngine implements IStorageEngine { for (const v of cond.$in) { const pk = String(v); if (seen.has(pk)) continue; - const value = this.lsm.get(`${tableName}:${pk}`); + const value = await this.lsm.get(`${tableName}:${pk}`); if (value) { seen.add(pk); rows.push({ ...value, [pkCol]: pk }); } } return rows; @@ -2022,7 +2031,7 @@ export class AriaEngine implements IStorageEngine { if ('$gt' in cond || '$gte' in cond || '$lt' in cond || '$lte' in cond) { const prefix = `${tableName}:`; await this.lsm.prefetchRange(prefix, `${prefix}\uffff`); - const entries = this.lsm.rangeScan(prefix, `${prefix}\uffff`); + const entries = await this.lsm.rangeScan(prefix, `${prefix}\uffff`); const rows: Record[] = []; for (const [key, value] of entries) { const candidate = { ...value, [pkCol]: key.slice(prefix.length) }; @@ -2065,7 +2074,7 @@ export class AriaEngine implements IStorageEngine { const seenPks = new Set(); // v0.4.1: IN 值可能重复,按 pk 去重 const pks: string[] = []; for (const val of values) { - const entries = idxLsm.rangeScan(val, `${val}\uffff`); + const entries = await idxLsm.rangeScan(val, `${val}\uffff`); for (const [, idxEntry] of entries) { const pk = (idxEntry as { pk?: string }).pk; if (pk && !seenPks.has(pk)) { @@ -2076,7 +2085,7 @@ export class AriaEngine implements IStorageEngine { } await this.lsm.prefetchKeys(pks.map((pk) => `${tableName}:${pk}`)); for (const pk of pks) { - const row = this.lsm.get(`${tableName}:${pk}`); + const row = await this.lsm.get(`${tableName}:${pk}`); if (row) results.push({ ...row, [pkCol]: pk }); } return results; @@ -2104,7 +2113,7 @@ export class AriaEngine implements IStorageEngine { const actualEndKey = endKey.includes('\uffff') ? endKey : `${endKey}\uffff`; // 预加载索引 LSM 与主 LSM 涉及的 SSTable await idxLsm.prefetchRange(startKey, actualEndKey); - const entries = idxLsm.rangeScan(startKey, actualEndKey); + const entries = await idxLsm.rangeScan(startKey, actualEndKey); const pks: string[] = []; for (const [, idxEntry] of entries) { const pk = (idxEntry as any).pk as string; @@ -2113,7 +2122,7 @@ export class AriaEngine implements IStorageEngine { await this.lsm.prefetchKeys(pks.map((pk) => `${tableName}:${pk}`)); const rows: Record[] = []; for (const pk of pks) { - const row = this.lsm.get(`${tableName}:${pk}`); + const row = await this.lsm.get(`${tableName}:${pk}`); if (row) rows.push({ ...row, [pkCol]: pk }); } return rows; diff --git a/src/engine/aria/index/lsm.ts b/src/engine/aria/index/lsm.ts index aa1d95c..65f1f05 100644 --- a/src/engine/aria/index/lsm.ts +++ b/src/engine/aria/index/lsm.ts @@ -67,6 +67,8 @@ export class LSM { private levels: SSTableMeta[][] = []; private sstableCache: Map = new Map(); private cacheSize = 0; + /** v0.8.0: 超过缓存上限、不进入 LRU 的 SSTable(每次读取按需加载) */ + private oversizedSSTables = new Set(); private cacheLimitBytes: number; private levelSizeMultiplier: number; private blockSize: number; @@ -163,6 +165,27 @@ export class LSM { } } + /** v0.8.0: 当前 SSTable 缓存占用字节数(公开访问器,替代测试直接读私有字段) */ + getCacheSize(): number { + return this.cacheSize; + } + + /** v0.8.0: 当前缓存上限(字节) */ + getCacheLimit(): number { + return this.cacheLimitBytes; + } + + /** v0.8.0: 常驻(超过缓存上限、不参与驱逐)的 SSTable 数量 */ + getOversizedCount(): number { + return this.oversizedSSTables.size; + } + + /** v0.8.0: 运行期调整缓存上限(测试与内存预算调优用),立即裁剪到新上限 */ + setCacheLimit(bytes: number): void { + this.cacheLimitBytes = Math.max(0, bytes); + this.trimCache(); + } + /** 获取估算内存使用(字节) */ getEstimatedMemory(): number { let mem = this.memtable.getEstimatedSize(); @@ -242,7 +265,7 @@ export class LSM { }; // 缓存 - this.cacheSSTable(id, sstableData); + this.tryCacheSSTable(id, sstableData); this.trimCache(); // 持久化:先存数据,再存元数据(串行链保证顺序与 id 一致) @@ -398,7 +421,11 @@ export class LSM { return undefined; } - get(key: string): Record | null { + /** + * v0.8.0: 改为 async —— SSTable 部分未命中缓存时会 `await` 回源, + * 因此读取不再依赖调用方的 prefetch(消除"缓存未命中即静默丢数据")。 + */ + async get(key: string): Promise | null> { // 1. 活跃 MemTable let result = this.memtable.get(key); if (result !== null) return this.unwrapTombstone(result); @@ -414,7 +441,7 @@ export class LSM { for (const meta of this.levels[level]) { if (key < meta.minKey || key > meta.maxKey) continue; - const reader = this.loadSSTableReader(meta); + const reader = await this.loadSSTableReader(meta); if (!reader) continue; const found = reader.get(key); @@ -425,9 +452,9 @@ export class LSM { return null; } - rangeScan(startKey: string, endKey: string): [string, Record][] { + async rangeScan(startKey: string, endKey: string): Promise<[string, Record][]> { const result: [string, Record][] = []; - this.rangeScanLazy(startKey, endKey, (k, v) => { result.push([k, v]); }); + await this.rangeScanLazy(startKey, endKey, (k, v) => { result.push([k, v]); }); return result; } @@ -437,11 +464,22 @@ export class LSM { * 逐条拉取;回调返回 false 时提前终止(未消费部分不再解析/物化)。 * 此前实现内部 mergeIter.drain() 全量物化,与"流式不物化"宣称不符。 */ - rangeScanLazy( + async rangeScanLazy( startKey: string, endKey: string, callback: (key: string, value: Record) => boolean | void, - ): void { + ): Promise { + // v0.8.0: 先把范围内需要的 SSTable 读取器全部取齐(未命中即回源), + // 再做纯内存的归并扫描 —— 使读取自洽,不依赖调用方 prefetch。 + const readers: { meta: SSTableMeta; reader: SSTableReader }[] = []; + for (let level = 0; level < MAX_LSM_LEVELS; level++) { + for (const meta of this.levels[level]) { + if (endKey < meta.minKey || startKey > meta.maxKey) continue; + const reader = await this.loadSSTableReader(meta); + if (reader) readers.push({ meta, reader }); + } + } + const mergeIter = new MergeIterator(); mergeIter.addSource(new GeneratorEntrySource( @@ -455,15 +493,10 @@ export class LSM { )); } - for (let level = 0; level < MAX_LSM_LEVELS; level++) { - for (const meta of this.levels[level]) { - if (endKey < meta.minKey || startKey > meta.maxKey) continue; - const reader = this.loadSSTableReader(meta); - if (!reader) continue; - mergeIter.addSource(new GeneratorEntrySource( - reader.scanLazy(startKey, endKey), - )); - } + for (const { reader } of readers) { + mergeIter.addSource(new GeneratorEntrySource( + reader.scanLazy(startKey, endKey), + )); } let entry = mergeIter.next(); @@ -562,7 +595,7 @@ export class LSM { bloomData: null, }; - this.cacheSSTable(id, sstableData); + this.tryCacheSSTable(id, sstableData); this.trimCache(); await this.sstableStore.save(id, sstableData); await this.sstableStore.saveMeta(meta); @@ -640,6 +673,7 @@ export class LSM { } this.sstableCache.clear(); this.cacheSize = 0; + this.oversizedSSTables.clear(); } getStats(): { memtableSize: number; sstableCount: number; levelCounts: number[] } { @@ -731,19 +765,50 @@ export class LSM { } /** 尝试从缓存或存储加载 SSTable,返回 Reader */ - private loadSSTableReader(meta: SSTableMeta): SSTableReader | null { - // 先检查缓存 - const data = this.sstableCache.get(meta.id); + /** + * v0.8.0 根治:读取器加载**自洽**(缓存未命中即回源),不再依赖调用方预先 prefetch。 + * + * 此前的实现是"缓存未命中返回 null",而所有调用方都是 + * const reader = this.loadSSTableReader(meta); + * if (!reader) continue; + * 于是**缓存未命中 = 静默跳过整个 SSTable = 查询结果少数据**。审计实测: + * 小缓存下 300 行只能查回 59 行(且不报错)。同时"读路径必须先 prefetch" + * 这个隐式约定,也是每次读都要 drainChain + prefetch 的原因(性能悬崖的另一半)。 + * + * 现在:未命中就 `await sstableStore.load()` 回源并校验;只有**确实读不到数据** + * (文件缺失/损坏,已由 dropInvalidSSTable 处理)才返回 null,且会在自愈时清掉 meta。 + */ + private async loadSSTableReader(meta: SSTableMeta): Promise { + let data = this.sstableCache.get(meta.id); if (!data) { - // 缓存未命中:正常路径应在查询前通过 prefetchRange/prefetchKeys 预加载。 - // 这里仅当缓存中有缺失且无兜底时返回 null(调用方跳过)。 - return null; + const loaded = await this.sstableStore.load(meta.id); + if (!loaded) { + // 数据确实不存在:清理该 meta(自愈),避免每次查询都重试 + await this.dropInvalidSSTable(meta); + return null; + } + // 运行期回源同样校验整文件 CRC-32(与 preloadSSTable 一致) + try { + const probe = new SSTableReader(loaded, meta); + if (!probe.verifyChecksum()) { + await this.dropInvalidSSTable(meta); + return null; + } + } catch { + await this.dropInvalidSSTable(meta); + return null; + } + this.tryCacheSSTable(meta.id, loaded); + data = loaded; } // 刷新 LRU 顺序(近似:删除后重新插入使其成为最近使用) - this.sstableCache.delete(meta.id); - this.sstableCache.set(meta.id, data); + if (this.sstableCache.has(meta.id)) { + const cached = this.sstableCache.get(meta.id)!; + this.sstableCache.delete(meta.id); + this.sstableCache.set(meta.id, cached); + } try { return new SSTableReader(data, meta); @@ -770,6 +835,32 @@ export class LSM { return; } } + this.tryCacheSSTable(id, data); + } + + /** + * v0.8.0: 单文件大于缓存上限时的处理 —— **不缓存**,且不牵连其他条目。 + * + * 此前 `preloadSSTable` 会把超大文件塞进缓存,随后 `trimCache()` 因 + * `cacheSize > cacheLimitBytes` 把**全部**缓存条目一并驱逐(含刚装入的目标文件 + * 与其它仍需使用的文件)。后果有两层: + * 1. "缓存上限"形同虚设(内存峰值不受约束); + * 2. 缓存被整体清空后,若某次读取没有紧跟一次 prefetch,就会静默读到空 + * (`loadSSTableReader` 未命中返回 null,"不存在"与"不可用"不可区分)。 + * 现在的语义明确:装不进缓存的文件不装,但**不影响**其他条目。 + */ + private tryCacheSSTable(id: number, data: Uint8Array): void { + if (data.byteLength > this.cacheLimitBytes) { + // 单个文件就超过整个缓存上限:**必须常驻**。 + // 若把它驱逐,`loadSSTableReader` 缓存未命中会返回 null,而调用方的 + // `if (!reader) continue` 会**静默跳过该文件** —— 查询结果直接少数据 + // (实测 300 行只返回 59 行)。这也是审计指出的"缓存未命中与不存在不可区分"。 + // 因此这类文件标记为 pinned,不参与驱逐;代价是内存占用可超出 cacheLimit, + // 上限为 cacheLimit + 单个最大 SSTable。 + this.oversizedSSTables.add(id); + } else { + this.oversizedSSTables.delete(id); + } this.cacheSSTable(id, data); } @@ -790,12 +881,33 @@ export class LSM { * 查询结束后由引擎调用一次,回收查询期间的临时超限。 */ trimCache(): void { - while (this.cacheSize > this.cacheLimitBytes && this.sstableCache.size > 0) { - const eldestId = this.sstableCache.keys().next().value as number; - const evicted = this.sstableCache.get(eldestId)!; - this.cacheSize -= evicted.byteLength; - this.sstableCache.delete(eldestId); + // v0.8.0: 超过缓存上限的单个文件被 pin 住(驱逐它们会导致读取路径静默丢数据), + // 因此这里的循环只驱逐未 pin 的条目;若只剩 pinned 条目则接受超限。 + let scanned = 0; + const total = this.sstableCache.size; + let id = this.sstableCache.keys().next().value as number | undefined; + while (this.cacheSize > this.cacheLimitBytes && scanned < total && id !== undefined) { + const nextId = this.nextCacheKey(id); + scanned++; + if (!this.oversizedSSTables.has(id)) { + const evicted = this.sstableCache.get(id); + if (evicted) { + this.cacheSize -= evicted.byteLength; + this.sstableCache.delete(id); + } + } + id = nextId; } } + /** 取缓存中 id 的下一个键(Map 插入序),用于跳过 pinned 条目 */ + private nextCacheKey(id: number): number | undefined { + let seen = false; + for (const key of this.sstableCache.keys()) { + if (seen) return key; + if (key === id) seen = true; + } + return undefined; + } + } diff --git a/src/engine/interface.ts b/src/engine/interface.ts index f87bda6..fccd967 100644 --- a/src/engine/interface.ts +++ b/src/engine/interface.ts @@ -5,6 +5,59 @@ import type { QueryPlan, TableSchema } from '../constants'; +// --------------------------------------------------------------------------- +// v0.8.0: 行所有权(row ownership)约定 +// --------------------------------------------------------------------------- + +/** + * 深拷贝一行,使调用方**无法通过修改返回值改写存储**。 + * + * 为什么必须做(审计实测):Memory/KVStore/Hybrid 三个引擎此前把内部行对象 + * **直接**交给调用方: + * const rows = await db.query('SELECT * FROM t'); + * rows[0].tag = 'HACKED'; // 存储被改写 + * 再查 WHERE tag='HACKED' → 0 行;WHERE tag='x' → 0 行 + * 即调用方一次无意的原地修改就能让索引与行失配、该行永久查不出来(Aria 因为是 + * 反序列化路径反而幸免,于是又成了跨引擎行为差异)。 + * + * 约定(写入 interface 文档,所有引擎必须遵守): + * **读出的行是副本,写入接收的行也是副本** —— 引擎不得把内部行对象暴露给外部, + * 也不得持有调用方传入的行对象引用。 + * + * 实现说明:结构化克隆可用时优先使用(正确处理 Date/嵌套对象/循环引用); + * 存储层写入的行已经过 validateRow 的 JSON 安全性检查,因此退化路径也是安全的。 + */ +export function cloneRow>(row: T): T { + if (row === null || typeof row !== 'object') return row; + if (typeof structuredClone === 'function') { + try { + return structuredClone(row); + } catch { + // 含不可克隆值(函数/Proxy)时退化为逐层复制 + } + } + return cloneRowFallback(row) as T; +} + +/** 退化实现:递归复制普通对象与数组(保留 Date) */ +function cloneRowFallback(value: unknown): unknown { + if (value === null || typeof value !== 'object') return value; + if (value instanceof Date) return new Date(value.getTime()); + if (Array.isArray(value)) return value.map((v) => cloneRowFallback(v)); + if (value instanceof Uint8Array) return new Uint8Array(value); + if (value instanceof ArrayBuffer) return value.slice(0); + const out: Record = {}; + for (const [k, v] of Object.entries(value as Record)) { + out[k] = cloneRowFallback(v); + } + return out; +} + +/** 批量深拷贝 */ +export function cloneRows>(rows: T[]): T[] { + return rows.map((r) => cloneRow(r)); +} + // --------------------------------------------------------------------------- // IStorageEngine — 所有存储引擎必须实现的接口 // --------------------------------------------------------------------------- diff --git a/src/engine/memory.ts b/src/engine/memory.ts index aa87427..72dec7c 100644 --- a/src/engine/memory.ts +++ b/src/engine/memory.ts @@ -6,6 +6,7 @@ import type { IStorageEngine } from './interface'; import type { QueryPlan, TableSchema, WhereCondition } from '../constants'; import { DatabaseError } from '../constants'; +import { cloneRow } from './interface'; import { matchWhere, applyOrderBy, projectColumns, containsUnresolvedSubqueries } from '../query/where-matcher'; import { stripUndefinedUpdates } from '../table/schema'; @@ -184,9 +185,11 @@ export class MemoryEngine implements IStorageEngine { /** v0.7.3: 按主键取已验证行(KVStoreEngine 持久化 validated 行用,含 default/类型归一) */ getRow(tableName: string, pkValue: string): Record | null { + // v0.8.0: 返回副本(调用方用于持久化,不得持有内部引用) const table = this.tables.get(tableName); if (!table) return null; - return table.get(pkValue) ?? null; + const row = table.get(pkValue); + return row ? cloneRow(row) : null; } async find(tableName: string, query: QueryPlan): Promise[]> { @@ -206,7 +209,9 @@ export class MemoryEngine implements IStorageEngine { if (query.columns && query.columns.length > 0 && query.columns[0] !== '*') { results = results.map((row) => projectColumns(row, query.columns!)); } - return results; + // v0.8.0: 返回副本 —— 此前直接交出内部行对象,调用方原地修改即改写存储 + // 并让索引与行失配(该行从此查不出来)。见 engine/interface.ts 的行所有权约定。 + return results.map((row) => cloneRow(row)); } /** v0.4.0: 流式查询 — 逐行回调(单次迭代,不物化结果数组) */ @@ -225,7 +230,7 @@ export class MemoryEngine implements IStorageEngine { for (const row of table.values()) { if (hasWhere && !matchWhere(row, query.where!)) continue; if (skipped < offset) { skipped++; continue; } - onRow(project ? project(row) : row); + onRow(project ? project(row) : cloneRow(row)); count++; if (count >= limit) break; } diff --git a/tests/engine/aria-cache.test.ts b/tests/engine/aria-cache.test.ts index 237d47d..34533d8 100644 --- a/tests/engine/aria-cache.test.ts +++ b/tests/engine/aria-cache.test.ts @@ -57,16 +57,54 @@ describe('AriaEngine SSTable 缓存内存上限', () => { // 300 行 / 2KB 阈值 → 应产生多个 SSTable expect(stats.sstableCount).toBeGreaterThan(1); - // 多轮查询后缓存仍受上限约束 + // v0.8.0 契约修正:内存上限只约束**可驱逐条目**。 + // + // 单个 SSTable 大于整个缓存上限时,它必须常驻:一旦驱逐, + // `loadSSTableReader` 未命中就会让调用方 `continue` 跳过整个文件 —— + // 那是静默丢数据(审计实测:300 行只能查回 59 行)。 + // 因此这里断言的是"数据完整"这一真正重要的不变量,而不是一个 + // 在极小缓存下无法成立的字节上限(上限 = cacheLimit + 单个最大 SSTable)。 for (let round = 0; round < 5; round++) { const rows = await engine.find('users', { table: 'users', where: { age: 25 } }); expect(rows.length).toBe(10); + } + + // 若所有 SSTable 都能装进上限,则缓存大小必须受上限约束 + const oversizedPinned = lsm.getOversizedCount(); + if (oversizedPinned === 0) { expect(lsm.getCacheSize()).toBeLessThanOrEqual(lsm.getCacheLimit()); } await engine.close(); }); + test('超大 SSTable 常驻缓存(驱逐会导致读取静默跳过整个文件)', async () => { + const engine = createSmallCacheEngine(1); // 4KB 上限,单个 SSTable 必然超过 + await engine.open('cache-oversized-pin', 1); + await engine.createTable(createSchema('users', { + id: { type: 'string', primaryKey: true }, + name: { type: 'string' }, + })); + await engine.insert('users', makeRows(300)); + const lsm = (engine as any).lsm as { + flush(): Promise; + getCacheSize(): number; + getCacheLimit(): number; + getOversizedCount(): number; + trimCache(): void; + getStats(): { sstableCount: number }; + }; + await lsm.flush(); + expect(lsm.getStats().sstableCount).toBeGreaterThan(0); + + // 强制裁剪后,超大文件仍必须可读(数据完整) + lsm.trimCache(); + expect(lsm.getOversizedCount()).toBeGreaterThan(0); + const all = await engine.find('users', { table: 'users' }); + expect(all.length).toBe(300); + await engine.close(); + }); + test('缓存驱逐后全表扫描仍返回完整数据(prefetch 兜底)', async () => { const engine = createSmallCacheEngine(1); // 4KB 上限,必然触发驱逐 await engine.open('cache-evict-fullscan', 1); @@ -78,11 +116,11 @@ describe('AriaEngine SSTable 缓存内存上限', () => { const rows = makeRows(300); await engine.insert('users', rows); + // v0.8.0: 这是最关键的数据完整性断言 —— 缓存上限极小(4KB)而 SSTable 更大时, + // 读取路径必须仍然返回**全部** 300 行(此前会静默少数据)。 const all = await engine.find('users', { table: 'users' }); expect(all.length).toBe(300); - const lsm = (engine as any).lsm; - expect(lsm.getCacheSize()).toBeLessThanOrEqual(lsm.getCacheLimit()); await engine.close(); }); diff --git a/tests/hybrid/index.test.ts b/tests/hybrid/index.test.ts index b0a5872..62e8e37 100644 --- a/tests/hybrid/index.test.ts +++ b/tests/hybrid/index.test.ts @@ -36,7 +36,8 @@ describe('HybridEngine', () => { }); it('磁盘引擎类型', () => { - expect(engine.getDiskEngineType()).toBe('indexeddb'); + // v0.8.0: indexeddb 引擎已于 v0.6.0 移除;该参数仅作介质标签 + expect(engine.getDiskEngineType()).toBe('opfs'); }); describe('表管理', () => {