release: v0.4.3 — 关闭时序与后台任务加固（close 排空、失败不吞错、compaction 竞态修复）+ 提交先 WAL
This commit is contained in:
thzxx
Vendored
+177
-85
@@ -34,7 +34,7 @@ class DatabaseError extends Error {
|
||||
// ---------------------------------------------------------------------------
|
||||
// 版本
|
||||
// ---------------------------------------------------------------------------
|
||||
const VERSION = '0.4.2';
|
||||
const VERSION = '0.4.3';
|
||||
|
||||
/**
|
||||
* metona-sqlark Shared WHERE Matcher — 统一的条件匹配逻辑
|
||||
@@ -1021,6 +1021,13 @@ class IndexedDBEngine {
|
||||
}
|
||||
}
|
||||
async close() {
|
||||
// v0.4.3-fix: 活跃事务时先回滚（避免 commit 对已关闭连接报错）
|
||||
if (this.txActive) {
|
||||
try {
|
||||
await this.rollbackTransaction();
|
||||
}
|
||||
catch { /* 回滚失败不阻塞关闭 */ }
|
||||
}
|
||||
if (this.db) {
|
||||
this.db.onversionchange = null; // 清理监听器
|
||||
this.db.close();
|
||||
@@ -1607,6 +1614,18 @@ class OPFSEngine {
|
||||
this.dbName = '';
|
||||
// 运行时内存缓存（OPFS 文件读写有延迟）
|
||||
this.memoryCache = new MemoryEngine();
|
||||
/**
|
||||
* v0.4.3-fix: 写操作串行队列 — 内存写 + 快照 + 文件持久化整体排队执行，
|
||||
* close() 等待队列排空后再释放目录句柄（避免 close 后挂起写泄漏/读旧数据）。
|
||||
* 前一个操作失败不阻塞后续（错误仍返回给调用方）。
|
||||
*/
|
||||
this.opQueue = Promise.resolve();
|
||||
}
|
||||
/** 将写操作加入串行队列（快照在队列内取，始终最新） */
|
||||
enqueueOp(fn) {
|
||||
const run = this.opQueue.then(fn, fn);
|
||||
this.opQueue = run.then(() => undefined, () => undefined);
|
||||
return run;
|
||||
}
|
||||
// ---- 生命周期 ----
|
||||
async open(dbName, version) {
|
||||
@@ -1620,6 +1639,11 @@ class OPFSEngine {
|
||||
await this.loadExistingTables();
|
||||
}
|
||||
async close() {
|
||||
// v0.4.3-fix: 等待所有挂起写操作完成（否则 close 后写仍在进行 → 重启读旧数据）
|
||||
try {
|
||||
await this.opQueue;
|
||||
}
|
||||
catch { /* 写失败已返回给调用方 */ }
|
||||
this.root = null;
|
||||
this.tablesDir = null;
|
||||
await this.memoryCache.close();
|
||||
@@ -1630,12 +1654,22 @@ class OPFSEngine {
|
||||
// ---- v0.4.2-fix: 自愈 / 重置 / 元数据 ----
|
||||
/** 自愈：重置内存缓存后从 OPFS 重新加载（单文件损坏不影响其他表） */
|
||||
async repair() {
|
||||
// v0.4.3-fix: 先等写队列排空（避免与挂起写竞态）
|
||||
try {
|
||||
await this.opQueue;
|
||||
}
|
||||
catch { /* ignore */ }
|
||||
await this.memoryCache.close();
|
||||
await this.memoryCache.open(this.dbName, 1);
|
||||
await this.loadExistingTables();
|
||||
}
|
||||
/** 清空全部数据与表结构（删除目录内全部文件） */
|
||||
async clearAll() {
|
||||
// v0.4.3-fix: 先等写队列排空
|
||||
try {
|
||||
await this.opQueue;
|
||||
}
|
||||
catch { /* ignore */ }
|
||||
await this.memoryCache.close();
|
||||
await this.memoryCache.open(this.dbName, 1);
|
||||
if (this.tablesDir) {
|
||||
@@ -1670,29 +1704,33 @@ class OPFSEngine {
|
||||
}
|
||||
// ---- 表管理 ----
|
||||
async createTable(schema) {
|
||||
await this.memoryCache.createTable(schema);
|
||||
// v0.4.2-fix: schema 持久化（此前仅写空数据文件 → 空表重启后消失、索引标记丢失）
|
||||
await this.setMeta(`schema_${schema.name}`, JSON.stringify(schema));
|
||||
// OPFS 中表以空 JSON 数组文件形式存在
|
||||
await this.writeTableData(schema.name, []);
|
||||
return this.enqueueOp(async () => {
|
||||
await this.memoryCache.createTable(schema);
|
||||
// v0.4.2-fix: schema 持久化（此前仅写空数据文件 → 空表重启后消失、索引标记丢失）
|
||||
await this.setMeta(`schema_${schema.name}`, JSON.stringify(schema));
|
||||
// OPFS 中表以空 JSON 数组文件形式存在
|
||||
await this.writeTableData(schema.name, []);
|
||||
});
|
||||
}
|
||||
async dropTable(tableName) {
|
||||
await this.memoryCache.dropTable(tableName);
|
||||
// v0.4.2-fix: 清理 schema meta（否则重启恢复幽灵表）
|
||||
if (this.tablesDir) {
|
||||
try {
|
||||
await this.tablesDir.removeEntry(`__metona_schema_${tableName}.meta`);
|
||||
return this.enqueueOp(async () => {
|
||||
await this.memoryCache.dropTable(tableName);
|
||||
// v0.4.2-fix: 清理 schema meta（否则重启恢复幽灵表）
|
||||
if (this.tablesDir) {
|
||||
try {
|
||||
await this.tablesDir.removeEntry(`__metona_schema_${tableName}.meta`);
|
||||
}
|
||||
catch {
|
||||
// 文件不存在则忽略
|
||||
}
|
||||
try {
|
||||
await this.tablesDir.removeEntry(`${tableName}.json`);
|
||||
}
|
||||
catch {
|
||||
// 文件不存在则忽略
|
||||
}
|
||||
}
|
||||
catch {
|
||||
// 文件不存在则忽略
|
||||
}
|
||||
try {
|
||||
await this.tablesDir.removeEntry(`${tableName}.json`);
|
||||
}
|
||||
catch {
|
||||
// 文件不存在则忽略
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
async hasTable(tableName) {
|
||||
if (!this.tablesDir)
|
||||
@@ -1721,20 +1759,24 @@ class OPFSEngine {
|
||||
}
|
||||
/** v0.4.2-fix: 引擎级 ALTER TABLE — 内存 + schema 持久化 + 整表文件重写 */
|
||||
async alterTable(tableName, action, column) {
|
||||
await this.memoryCache.alterTable(tableName, action, column);
|
||||
const schema = await this.memoryCache.getTableSchema(tableName);
|
||||
if (schema)
|
||||
await this.setMeta(`schema_${tableName}`, JSON.stringify(schema));
|
||||
const rows = await this.memoryCache.find(tableName, { table: tableName });
|
||||
await this.writeTableData(tableName, rows);
|
||||
return this.enqueueOp(async () => {
|
||||
await this.memoryCache.alterTable(tableName, action, column);
|
||||
const schema = await this.memoryCache.getTableSchema(tableName);
|
||||
if (schema)
|
||||
await this.setMeta(`schema_${tableName}`, JSON.stringify(schema));
|
||||
const rows = await this.memoryCache.find(tableName, { table: tableName });
|
||||
await this.writeTableData(tableName, rows);
|
||||
});
|
||||
}
|
||||
// ---- CRUD ----
|
||||
async insert(tableName, rows) {
|
||||
const pks = await this.memoryCache.insert(tableName, rows);
|
||||
// 持久化到 OPFS
|
||||
const allRows = await this.memoryCache.find(tableName, { table: tableName });
|
||||
await this.writeTableData(tableName, allRows);
|
||||
return pks;
|
||||
return this.enqueueOp(async () => {
|
||||
const pks = await this.memoryCache.insert(tableName, rows);
|
||||
// 持久化到 OPFS（快照在队列内取，始终最新）
|
||||
const allRows = await this.memoryCache.find(tableName, { table: tableName });
|
||||
await this.writeTableData(tableName, allRows);
|
||||
return pks;
|
||||
});
|
||||
}
|
||||
async find(tableName, query) {
|
||||
return this.memoryCache.find(tableName, query);
|
||||
@@ -1744,50 +1786,62 @@ class OPFSEngine {
|
||||
return this.memoryCache.findStream(tableName, query, onRow);
|
||||
}
|
||||
async update(tableName, query, updates) {
|
||||
const count = await this.memoryCache.update(tableName, query, updates);
|
||||
const allRows = await this.memoryCache.find(tableName, { table: tableName });
|
||||
await this.writeTableData(tableName, allRows);
|
||||
return count;
|
||||
return this.enqueueOp(async () => {
|
||||
const count = await this.memoryCache.update(tableName, query, updates);
|
||||
const allRows = await this.memoryCache.find(tableName, { table: tableName });
|
||||
await this.writeTableData(tableName, allRows);
|
||||
return count;
|
||||
});
|
||||
}
|
||||
async delete(tableName, query) {
|
||||
const count = await this.memoryCache.delete(tableName, query);
|
||||
const allRows = await this.memoryCache.find(tableName, { table: tableName });
|
||||
await this.writeTableData(tableName, allRows);
|
||||
return count;
|
||||
return this.enqueueOp(async () => {
|
||||
const count = await this.memoryCache.delete(tableName, query);
|
||||
const allRows = await this.memoryCache.find(tableName, { table: tableName });
|
||||
await this.writeTableData(tableName, allRows);
|
||||
return count;
|
||||
});
|
||||
}
|
||||
async count(tableName, query) {
|
||||
return this.memoryCache.count(tableName, query);
|
||||
}
|
||||
async clear(tableName) {
|
||||
await this.memoryCache.clear(tableName);
|
||||
await this.writeTableData(tableName, []);
|
||||
return this.enqueueOp(async () => {
|
||||
await this.memoryCache.clear(tableName);
|
||||
await this.writeTableData(tableName, []);
|
||||
});
|
||||
}
|
||||
// ---- 动态索引（v0.3.0） ----
|
||||
async createIndex(tableName, column, unique) {
|
||||
await this.memoryCache.createIndex(tableName, column, unique);
|
||||
// v0.4.2-fix: 索引标记持久化（重启后索引结构恢复）
|
||||
const schema = await this.memoryCache.getTableSchema(tableName);
|
||||
if (schema)
|
||||
await this.setMeta(`schema_${tableName}`, JSON.stringify(schema));
|
||||
return this.enqueueOp(async () => {
|
||||
await this.memoryCache.createIndex(tableName, column, unique);
|
||||
// v0.4.2-fix: 索引标记持久化（重启后索引结构恢复）
|
||||
const schema = await this.memoryCache.getTableSchema(tableName);
|
||||
if (schema)
|
||||
await this.setMeta(`schema_${tableName}`, JSON.stringify(schema));
|
||||
});
|
||||
}
|
||||
async dropIndex(tableName, column, indexName) {
|
||||
await this.memoryCache.dropIndex(tableName, column, indexName);
|
||||
const schema = await this.memoryCache.getTableSchema(tableName);
|
||||
if (schema)
|
||||
await this.setMeta(`schema_${tableName}`, JSON.stringify(schema));
|
||||
return this.enqueueOp(async () => {
|
||||
await this.memoryCache.dropIndex(tableName, column, indexName);
|
||||
const schema = await this.memoryCache.getTableSchema(tableName);
|
||||
if (schema)
|
||||
await this.setMeta(`schema_${tableName}`, JSON.stringify(schema));
|
||||
});
|
||||
}
|
||||
// ---- 事务 ----
|
||||
async beginTransaction() {
|
||||
await this.memoryCache.beginTransaction();
|
||||
}
|
||||
async commitTransaction() {
|
||||
await this.memoryCache.commitTransaction();
|
||||
// 将内存数据刷到 OPFS
|
||||
const tableNames = await this.memoryCache.getTableNames();
|
||||
for (const tableName of tableNames) {
|
||||
const rows = await this.memoryCache.find(tableName, { table: tableName });
|
||||
await this.writeTableData(tableName, rows);
|
||||
}
|
||||
return this.enqueueOp(async () => {
|
||||
await this.memoryCache.commitTransaction();
|
||||
// 将内存数据刷到 OPFS
|
||||
const tableNames = await this.memoryCache.getTableNames();
|
||||
for (const tableName of tableNames) {
|
||||
const rows = await this.memoryCache.find(tableName, { table: tableName });
|
||||
await this.writeTableData(tableName, rows);
|
||||
}
|
||||
});
|
||||
}
|
||||
async rollbackTransaction() {
|
||||
await this.memoryCache.rollbackTransaction();
|
||||
@@ -1800,6 +1854,7 @@ class OPFSEngine {
|
||||
return this.tablesDir;
|
||||
}
|
||||
async writeTableData(tableName, data) {
|
||||
// 由 enqueueOp 串行化调用，此处直接写文件
|
||||
const dir = this.ensureDir();
|
||||
const fileName = `${tableName}.json`;
|
||||
const fileHandle = await dir.getFileHandle(fileName, { create: true });
|
||||
@@ -3274,6 +3329,11 @@ class LSM {
|
||||
this.compacting = false; // 防止重复触发 compaction
|
||||
/** 串行化 flush/compaction 链：保证持久化顺序与 id 分配顺序一致 */
|
||||
this.flushChain = Promise.resolve();
|
||||
/**
|
||||
* v0.4.3-fix: 最近一次后台 flush/compaction 失败。
|
||||
* 后台失败不卡死链（吞错防死锁），但在显式 flush()/close() 时报告（不静默）。
|
||||
*/
|
||||
this.lastBackgroundError = null;
|
||||
this.memtableSizeThreshold = config.memtableSizeThreshold ?? DEFAULT_MEMTABLE_SIZE;
|
||||
this.memtable = new MemTable(this.memtableSizeThreshold);
|
||||
this.levelSizeMultiplier = config.levelSizeMultiplier ?? DEFAULT_LEVEL_SIZE_MULTIPLIER;
|
||||
@@ -3371,11 +3431,26 @@ class LSM {
|
||||
return this.flushChain
|
||||
.then(task)
|
||||
.catch((error) => {
|
||||
// v0.4.3-fix: 记录失败（flush()/close() 时报告），不再完全静默吞错
|
||||
this.lastBackgroundError = error;
|
||||
// eslint-disable-next-line no-console
|
||||
console.warn('[AriaEngine LSM] background flush/compaction failed:', error);
|
||||
// catch 返回 undefined → 链恢复为 resolved，后续任务继续
|
||||
});
|
||||
}
|
||||
/**
|
||||
* v0.4.3-fix: 排空后台链 — 循环等待 flushChain 直到稳定。
|
||||
* 任务完成时可能级联调度新任务（compaction 多级触发），单次 await 等不到。
|
||||
* close/flush/clear 必须等待全部后台任务完成后才能安全关闭底层存储。
|
||||
*/
|
||||
async drainChain() {
|
||||
while (true) {
|
||||
const chain = this.flushChain;
|
||||
await chain;
|
||||
if (this.flushChain === chain)
|
||||
return;
|
||||
}
|
||||
}
|
||||
/** 将指定 Immutable MemTable 刷盘为 SSTable（id 由 store 按命名空间分配） */
|
||||
async flushImmutableAsync(frozen) {
|
||||
const entries = frozen.getAllEntries();
|
||||
@@ -3415,27 +3490,27 @@ class LSM {
|
||||
this.scheduleCompact(0);
|
||||
}
|
||||
}
|
||||
/** 异步调度 compaction，使用 setTimeout 分片执行 */
|
||||
/**
|
||||
* 异步调度 compaction。
|
||||
* v0.4.3-fix: 去掉 setTimeout 分片 — 此前未触发的定时器在 close 后执行，
|
||||
* 用已关闭的 backend 写存储（错误被吞），或 close 后 reopen 时旧闭包引用新 backend 交叉污染。
|
||||
* 现在直接挂在 flushChain 上：串行执行、close 的 drainChain 能等到全部完成。
|
||||
*/
|
||||
scheduleCompact(level) {
|
||||
if (level >= MAX_LSM_LEVELS - 1 || this.compacting)
|
||||
return;
|
||||
this.compacting = true;
|
||||
setTimeout(() => {
|
||||
try {
|
||||
this.flushChain = this.enqueueOnChain(() => this.compactLevelAsync(level));
|
||||
this.flushChain = this.enqueueOnChain(() => this.compactLevelAsync(level).finally(() => {
|
||||
this.compacting = false;
|
||||
// 连续触发：如果 compaction 后仍然超标，继续调度
|
||||
if (this.levels[level].length >= 4) {
|
||||
this.scheduleCompact(level);
|
||||
}
|
||||
finally {
|
||||
this.compacting = false;
|
||||
// 连续触发：如果 compaction 后仍然超标，继续调度
|
||||
if (this.levels[level].length >= 4) {
|
||||
this.scheduleCompact(level);
|
||||
}
|
||||
// 检查下一级是否需要 compaction
|
||||
if (level + 1 < MAX_LSM_LEVELS - 1 && this.levels[level + 1].length >= 4) {
|
||||
this.scheduleCompact(level + 1);
|
||||
}
|
||||
// 检查下一级是否需要 compaction
|
||||
if (level + 1 < MAX_LSM_LEVELS - 1 && this.levels[level + 1].length >= 4) {
|
||||
this.scheduleCompact(level + 1);
|
||||
}
|
||||
}, 0);
|
||||
}));
|
||||
}
|
||||
/** 背压场景下排队 compaction（写入路径调用） */
|
||||
enqueueCompact(level) {
|
||||
@@ -3451,6 +3526,8 @@ class LSM {
|
||||
}
|
||||
}).catch((error) => {
|
||||
this.compacting = false;
|
||||
// v0.4.3-fix: 记录失败（flush()/close() 时报告）
|
||||
this.lastBackgroundError = error;
|
||||
// eslint-disable-next-line no-console
|
||||
console.warn('[AriaEngine LSM] background compaction failed:', error);
|
||||
});
|
||||
@@ -3461,10 +3538,11 @@ class LSM {
|
||||
/**
|
||||
* 预加载指定 key 范围内可能命中的所有 SSTable 到缓存。
|
||||
* 在同步扫描/查找之前调用，保证 loadSSTableReader 不会因缓存未命中而返回 null。
|
||||
* 先等待 flush 链完成：避免 flush 的缓存裁剪与预加载竞争（驱逐刚加载的 SSTable）。
|
||||
* v0.4.3-fix: 等待链稳定（drainChain）— 后台 flush/compaction 在链上动态增长，
|
||||
* 单次 await 后 compaction 仍可能合并 levels，导致扫描时新 meta 缓存未命中而跳块丢数据。
|
||||
*/
|
||||
async prefetchRange(startKey, endKey) {
|
||||
await this.flushChain;
|
||||
await this.drainChain();
|
||||
this.trimCache();
|
||||
const toLoad = [];
|
||||
for (let level = 0; level < MAX_LSM_LEVELS; level++) {
|
||||
@@ -3483,7 +3561,8 @@ class LSM {
|
||||
async prefetchKeys(keys) {
|
||||
if (keys.length === 0)
|
||||
return;
|
||||
await this.flushChain;
|
||||
// v0.4.3-fix: 等待链稳定（同 prefetchRange，防 compaction 竞态丢数据）
|
||||
await this.drainChain();
|
||||
this.trimCache();
|
||||
const toLoad = new Set();
|
||||
for (let level = 0; level < MAX_LSM_LEVELS; level++) {
|
||||
@@ -3674,8 +3753,14 @@ class LSM {
|
||||
}
|
||||
/** 等待所有排队的 flush/compaction 完成，并将剩余数据刷盘 */
|
||||
async flush() {
|
||||
// 等待链上已排队的 flush/compaction
|
||||
await this.flushChain;
|
||||
// v0.4.3-fix: 报告后台失败（消费一次，不永久吞错）
|
||||
if (this.lastBackgroundError !== null) {
|
||||
const error = this.lastBackgroundError;
|
||||
this.lastBackgroundError = null;
|
||||
throw new DatabaseError('AriaEngine background flush/compaction failed (data may be inconsistent)', 'ARIA_BACKGROUND_ERROR', error);
|
||||
}
|
||||
// v0.4.3-fix: 循环等待级联任务（flush 完成可能触发新的 compaction）
|
||||
await this.drainChain();
|
||||
// 若仍有 frozen 数据未刷盘，在链尾追加
|
||||
if (this.immutableMemtable) {
|
||||
const frozen = this.immutableMemtable;
|
||||
@@ -3688,8 +3773,12 @@ class LSM {
|
||||
await this.flushImmutableAsync(frozen);
|
||||
}
|
||||
this.frozenMemtables = [];
|
||||
// 刷盘完成后级联调度可能触发 compaction → 排空到稳定
|
||||
await this.drainChain();
|
||||
}
|
||||
async clear() {
|
||||
// v0.4.3-fix: 清空前排空后台任务（避免 compaction 在清空后写回残留 meta/数据）
|
||||
await this.drainChain();
|
||||
this.memtable.clear();
|
||||
this.immutableMemtable = null;
|
||||
this.frozenMemtables = [];
|
||||
@@ -6304,6 +6393,16 @@ class AriaEngine {
|
||||
async commitTransaction() {
|
||||
if (!this.currentTxnId)
|
||||
throw new DatabaseError('No active transaction', 'TX_NONE');
|
||||
// v0.4.3-fix: 先持久化 WAL COMMIT，再合并快照到 LSM —
|
||||
// 崩溃在 WAL 提交后、快照合并前：恢复时 WAL 重放数据，重启一致；
|
||||
// 崩溃在 WAL 提交前：commitTransaction 尚未返回，事务视为未提交（可回滚）
|
||||
await this.wal.append({
|
||||
type: WALRecordType.COMMIT,
|
||||
txnId: this.currentTxnId,
|
||||
tableName: '',
|
||||
key: '',
|
||||
});
|
||||
await this.wal.flush();
|
||||
if (this.txnSnapshot) {
|
||||
for (const [key, value] of this.txnSnapshot) {
|
||||
if (value.__txn_deleted) {
|
||||
@@ -6315,15 +6414,8 @@ class AriaEngine {
|
||||
}
|
||||
}
|
||||
this.mvcc.commitTransaction(this.currentTxnId);
|
||||
await this.wal.append({
|
||||
type: WALRecordType.COMMIT,
|
||||
txnId: this.currentTxnId,
|
||||
tableName: '',
|
||||
key: '',
|
||||
});
|
||||
this.currentTxnId = null;
|
||||
this.txnSnapshot = null;
|
||||
await this.wal.flush();
|
||||
}
|
||||
async rollbackTransaction() {
|
||||
if (!this.currentTxnId)
|
||||
|
||||
Reference in New Issue
Block a user