diff --git a/src/engine/aria/index.ts b/src/engine/aria/index.ts index 1a928b7..4e76988 100644 --- a/src/engine/aria/index.ts +++ b/src/engine/aria/index.ts @@ -219,9 +219,64 @@ export class AriaEngine implements IStorageEngine { if (r.type === WALRecordType.COMMIT) committedTxns.add(r.txnId); if (r.type === WALRecordType.ROLLBACK) committedTxns.delete(r.txnId); } + + // v0.8.0 根治：确定每个事务的**回放起始下标**。 + // + // `ROLLBACK TO ` 只回滚内存快照，日志里仍留有 savepoint 之前 + // 写入的记录；COMMIT 又把整个 txnId 标记为已提交 —— 于是那些被回滚掉的写入 + // 在重启时被重新应用，**已回滚的行复活**（实测：实时只剩 a，崩溃重开变成 a+b）。 + // + // 现在 savepoint 回滚会写入 SAVEPOINT_ROLLBACK 记录；恢复时该事务只应用 + // **最后一条** SAVEPOINT_ROLLBACK 之后的记录 —— 等价于"回到该保存点"， + // 与实时态严格一致（这样也就不需要 undo 崩溃前已落盘的旧值）。 + // 每个事务：{ 保留起点, 丢弃终点 } —— 区间 [start, end) 保留。 + // + // 关键：`replayFromIndex` 是**事务内**的记录序号（写入方按事务计数）， + // 因此这里必须用"该事务的第几条记录"来比较，不能直接拿全局下标 —— + // 全局下标里还混着 txnId=0 的非事务记录（CREATE_TABLE 等）以及其它事务。 + const globalIndexToTxnIndex = new Map(); + const txnRecordCount = new Map(); + for (let i = 0; i < allRecords.length; i++) { + const r = allRecords[i]; + if (r.txnId === 0) continue; + const n = txnRecordCount.get(r.txnId) ?? 0; + globalIndexToTxnIndex.set(i, n); + txnRecordCount.set(r.txnId, n + 1); + } + + // 边界语义（必须显式定义，否则差一错误就在这里）： + // 写入侧记录的 `replayFromIndex = N` 表示"保存点建立时，本事务已成功追加了 N 条记录"， + // 即事务的第 0..N-1 条记录必须保留（BEGIN 是第 0 条）。 + // 因此恢复侧应保留的**事务内下标区间**是 [0, N)，丢弃 [N, 标记位置)。 + // 等价地：只丢弃"事务内下标 >= N"且"在最后一个标记之前"的记录。 + const txnReplayWindow = new Map(); + for (let i = 0; i < allRecords.length; i++) { + const r = allRecords[i]; + if (r.type !== WALRecordType.SAVEPOINT_ROLLBACK) continue; + const declared = (r.data as { replayFromIndex?: number } | undefined)?.replayFromIndex; + const keepUpTo = typeof declared === 'number' ? declared : 0; + const txnIdx = globalIndexToTxnIndex.get(i) ?? 0; + const prev = txnReplayWindow.get(r.txnId); + // 多个保存点回滚：保留上界取**最早**的（回到最早的保存点）， + // 丢弃起点取**最后一个**标记的事务内下标。 + txnReplayWindow.set(r.txnId, { + keepUpTo: prev ? Math.min(prev.keepUpTo, keepUpTo) : keepUpTo, + dropFrom: txnIdx, + }); + } + // 第二遍：仅应用 txnId==0（非事务）或已提交事务的数据 - for (const r of allRecords) { + for (let i = 0; i < allRecords.length; i++) { + const r = allRecords[i]; if (r.txnId === 0 || committedTxns.has(r.txnId)) { + if (r.type === WALRecordType.SAVEPOINT_ROLLBACK) continue; // 标记记录，无数据 + // 事务内落在 [start, end) 之外的记录：属于被 savepoint 回滚掉的部分，丢弃 + const win = txnReplayWindow.get(r.txnId); + if (win !== undefined) { + const txnIdx = globalIndexToTxnIndex.get(i) ?? 0; + // 保留 [0, keepUpTo)；丢弃 [keepUpTo, dropFrom)；标记之后的记录照常保留 + if (txnIdx >= win.keepUpTo && txnIdx < win.dropFrom) continue; + } if (r.type === WALRecordType.DROP_TABLE) { // v0.3.3: DROP_TABLE 回放（异步：需预加载 SSTable 后清除残留数据） await this.applyDropTableRecovery(r.tableName); @@ -660,6 +715,7 @@ export class AriaEngine implements IStorageEngine { } await this.wal.appendBatch(walRecords); + this.txnWalRecordCount += walRecords.length; this.opCounter += rows.length; this.checkMemoryBudget(); @@ -856,6 +912,7 @@ export class AriaEngine implements IStorageEngine { } await this.wal.appendBatch(walRecords); + this.txnWalRecordCount += walRecords.length; this.opCounter += count; await this.checkpointManager.tick(); @@ -1033,6 +1090,7 @@ export class AriaEngine implements IStorageEngine { } await this.wal.appendBatch(walRecords); + this.txnWalRecordCount += walRecords.length; this.opCounter += count; await this.checkpointManager.tick(); @@ -1263,6 +1321,7 @@ export class AriaEngine implements IStorageEngine { this.updateSecondaryIndexes(tableName, String(row[pkCol]), null, row); } await this.wal.appendBatch(walRecords); + this.txnWalRecordCount += walRecords.length; this.opCounter += rows.length; await this.checkpointManager.tick(); this.tryGC(); @@ -1336,6 +1395,7 @@ export class AriaEngine implements IStorageEngine { }); } await this.wal.appendBatch(walRecords); + this.txnWalRecordCount += walRecords.length; this.opCounter += walRecords.length; await this.checkpointManager.tick(); this.trimAllCaches(); @@ -1448,6 +1508,8 @@ export class AriaEngine implements IStorageEngine { if (this.currentTxnId) throw new DatabaseError('Transaction already in progress', 'TX_ACTIVE'); this.currentTxnId = this.mvcc.beginTransaction(); this.txnSnapshot = new Map(); + // v0.8.0: 事务内 WAL 记录计数从 0 开始（保存点边界依赖它） + this.txnWalRecordCount = 0; // v0.7.3-fix: WAL BEGIN 写失败回滚内存事务状态 —— 此前 append 抛错（full 模式） // 时 currentTxnId 已设置 → TX_ACTIVE 永久泄漏（后续无法开始新事务）。 @@ -1459,6 +1521,7 @@ export class AriaEngine implements IStorageEngine { tableName: '', key: '', }); + this.txnWalRecordCount = 1; // BEGIN 本身占一条 } catch (error) { this.mvcc.rollbackTransaction(this.currentTxnId); this.currentTxnId = null; @@ -1493,6 +1556,13 @@ export class AriaEngine implements IStorageEngine { this.mvcc.commitTransaction(this.currentTxnId); + // v0.8.0 根治：事务结束时清空 savepoint 表。 + // 此前 commit/rollback **都不清理** savepoints，于是： + // 1. 上一个事务的名字仍被占用 → 新事务 SAVEPOINT 同名报 "already exists"； + // 2. 新事务 ROLLBACK TO 陈旧名字会拿到旧事务快照，把当前事务的写入静默替换。 + this.savepoints.clear(); + this.savepointWalBoundary.clear(); + this.txnWalRecordCount = 0; this.currentTxnId = null; this.txnSnapshot = null; } @@ -1526,6 +1596,10 @@ export class AriaEngine implements IStorageEngine { this.mvcc.rollbackTransaction(txnId); this.txnSnapshot = null; + // v0.8.0：事务结束清空 savepoint（理由同 commitTransaction） + this.savepoints.clear(); + this.savepointWalBoundary.clear(); + this.txnWalRecordCount = 0; this.currentTxnId = null; // v0.3.3: 事务内直接写入了二级索引 LSM，回滚后全量重建受影响表的索引 @@ -1540,6 +1614,17 @@ export class AriaEngine implements IStorageEngine { private savepoints: Map> | null }> = new Map(); + /** + * v0.8.0: 当前事务在 WAL 中已成功追加的记录条数。 + * + * 用途：保存点需要记录"回滚后应保留到哪一条"，否则恢复时无法区分 + * "保存点之前的写入"（应保留）与"保存点之后的写入"（应丢弃）。 + */ + private txnWalRecordCount = 0; + + /** 保存点 → 该保存点时事务的 WAL 记录边界 */ + private savepointWalBoundary = new Map(); + async savepoint(name: string): Promise { if (!this.currentTxnId) throw new DatabaseError('No active transaction for savepoint', 'TX_NONE'); if (this.savepoints.has(name)) throw new DatabaseError(`Savepoint "${name}" already exists`, 'SAVEPOINT_EXISTS'); @@ -1548,11 +1633,35 @@ export class AriaEngine implements IStorageEngine { txnId: this.currentTxnId, snapshot: this.txnSnapshot ? new Map(this.txnSnapshot) : null, }); + // v0.8.0: 记录边界（该事务已追加的记录条数）—— 见 txnWalRecordCount 说明 + this.savepointWalBoundary.set(`${this.currentTxnId}:${name}`, this.txnWalRecordCount); } async rollbackToSavepoint(name: string): Promise { const sp = this.savepoints.get(name); if (!sp) throw new DatabaseError(`Savepoint "${name}" not found`, 'SAVEPOINT_NOT_FOUND'); + // v0.8.0 根治：savepoint 归属校验。 + // 此前只按名字查找，而 savepoints 在 commit/rollback 时**从不清空** —— + // 上一个事务遗留的 savepoint 会让当前事务的写入被旧事务快照静默替换 + // （实测：本事务 update 后 ROLLBACK TO 陈旧名 + COMMIT，写入凭空消失）。 + if (sp.txnId !== this.currentTxnId) { + throw new DatabaseError( + `Savepoint "${name}" belongs to a different transaction (stale savepoint)`, + 'SAVEPOINT_NOT_FOUND', + ); + } + // v0.8.0: 先写 WAL 标记再改内存（与 commit/rollback 的"WAL 领先内存"一致）—— + // 否则崩溃恢复会把该事务在 savepoint 之前写入的记录重新应用，已回滚的行复活。 + const boundary = this.savepointWalBoundary.get(`${this.currentTxnId}:${name}`) ?? 0; + await this.wal.append({ + type: WALRecordType.SAVEPOINT_ROLLBACK, + txnId: this.currentTxnId!, + tableName: '', + key: '', + data: { replayFromIndex: boundary }, + }); + this.txnWalRecordCount++; + this.savepointWalBoundary.set(`${this.currentTxnId}:${name}`, boundary); // v0.6.3-fix: 记录当前快照涉及的表（回滚后重建索引）—— 此前事务内直写 // 索引 LSM，savepoint 回滚只还原快照 → savepoint 之后的索引条目残留 const affectedTables = new Set(); diff --git a/src/engine/aria/types.ts b/src/engine/aria/types.ts index 68b905d..fdd4116 100644 --- a/src/engine/aria/types.ts +++ b/src/engine/aria/types.ts @@ -121,6 +121,19 @@ export enum WALRecordType { ROLLBACK = 6, CREATE_TABLE = 7, DROP_TABLE = 8, + /** + * v0.8.0: 回滚到保存点。 + * + * 为什么必须有这条记录：`ROLLBACK TO ` 此前只改内存快照、**不写 WAL**， + * 而 COMMIT 会把整个 txnId 标记为已提交，恢复时按"该事务的全部记录"重放 —— + * 于是被 savepoint 回滚掉的行在崩溃重启后**复活**（实测：实时只剩 a， + * 崩溃重开变成 a+b）。反向也有问题：跨事务复用的陈旧 savepoint 会让当前事务的 + * 写入被上一事务的快照静默替换。 + * + * 记录语义：该事务在此之前的写入都应被丢弃 —— 恢复时只应用该事务**最后一条** + * SAVEPOINT_ROLLBACK 之后的记录。 + */ + SAVEPOINT_ROLLBACK = 9, } /** 单条 WAL 记录 */ @@ -135,7 +148,15 @@ export interface WALRecord { tableName: string; /** 主键值 */ key: string; - /** 操作数据（INSERT/UPDATE 时有效） */ + /** + * 操作数据（INSERT/UPDATE/DELETE 时有效）。 + * + * v0.8.0: SAVEPOINT_ROLLBACK 记录借用该字段携带边界信息： + * `{ replayFromIndex: N }` 表示"创建该保存点时，本事务已成功追加了 N 条记录"。 + * 恢复时据此丢弃 [N, 该标记) 区间的写入 —— 等价于回到该保存点， + * 同时**保留** N 之前的写入（这是"回到保存点"而非"回滚整个事务"的关键）。 + * 复用既有字段意味着 WAL 二进制格式无需变更，旧库记录仍可解析。 + */ data?: Record; /** 校验和 */ checksum: number; diff --git a/tests/v080-savepoint.test.ts b/tests/v080-savepoint.test.ts new file mode 100644 index 0000000..c4a008d --- /dev/null +++ b/tests/v080-savepoint.test.ts @@ -0,0 +1,138 @@ +/** + * v0.8.0 回归 —— Aria SAVEPOINT 语义根治 + * + * 覆盖两个此前会导致**静默数据错误**的缺陷： + * + * A3 已回滚的行在崩溃重启后复活 + * `ROLLBACK TO ` 此前只改内存快照、不写 WAL，而 COMMIT 会把整个 + * txnId 标记为已提交，恢复时按"该事务的全部记录"重放 —— 于是被回滚掉的写入 + * 被重新应用。实测：实时只剩 a，崩溃重开变成 a+b。 + * 修复：savepoint 回滚写入 SAVEPOINT_ROLLBACK 记录（携带 replayFromIndex + * 边界），恢复时只应用"保存点之后、最后一个标记之前"的写入。 + * + * A4 跨事务复用的陈旧 savepoint 静默丢弃当前事务的写入 + * savepoints 在 commit/rollback 时从不清空，且 rollbackToSavepoint 不校验归属。 + * 实测：新事务 update 后 ROLLBACK TO 上一事务遗留的名字 + COMMIT，写入凭空消失。 + * 修复：事务结束清空 savepoint；回滚前校验 txnId 归属。 + */ +import { MetonaSqlark } from '../src/core'; +import { AriaEngine } from '../src/engine/aria/index'; + +describe('[v0.8.0] A4 跨事务 savepoint 泄漏', () => { + test('陈旧 savepoint 名不可重用，当前事务写入不被替换', async () => { + const db = await MetonaSqlark.create({ name: 'v080-sp-stale', mode: 'aria', diskEngine: 'memory' } as never); + await db.defineTable('t', { id: { type: 'string', primaryKey: true }, v: { type: 'number' } }); + await db.query("INSERT INTO t VALUES ('x', 0)"); + + await db.query('BEGIN'); + await db.query("UPDATE t SET v = 9 WHERE id='x'"); + await db.query('SAVEPOINT sp'); + await db.query('COMMIT'); + expect((await db.query("SELECT v FROM t WHERE id='x'") as Array<{ v: number }>)[0].v).toBe(9); + + await db.query('BEGIN'); + await db.query("UPDATE t SET v = 5 WHERE id='x'"); + // 上一事务的 savepoint 随事务结束失效 + await expect(db.query('ROLLBACK TO sp')).rejects.toMatchObject({ code: 'SAVEPOINT_NOT_FOUND' }); + await db.query('COMMIT'); + // 关键：本事务的写入必须保留（此前会被旧快照静默替换回 9） + expect((await db.query("SELECT v FROM t WHERE id='x'") as Array<{ v: number }>)[0].v).toBe(5); + await db.close(); + }); +}); + +describe('[v0.8.0] A3 savepoint 回滚的崩溃一致性', () => { + /** kv 后端介质按库名存于全局注册表且 close 不清数据 → 新引擎 open 同名库 = 崩溃重启 */ + const CFG = { storageBackend: 'kv' as const, walSyncMode: 'full' as const, checkpointInterval: 999999 }; + + test('回滚到保存点后提交：崩溃重放不得复活已回滚的行', async () => { + const eng = new AriaEngine(CFG) as unknown as { + open(n: string, v: number): Promise; + createTable(s: unknown): Promise; + beginTransaction(): Promise; + insert(t: string, r: unknown[]): Promise; + savepoint(n: string): Promise; + rollbackToSavepoint(n: string): Promise; + commitTransaction(): Promise; + find(t: string, q: unknown): Promise>>; + close(): Promise; + }; + await eng.open('v080-sp-crash', 1); + await eng.createTable({ name: 't', columns: { id: { type: 'string', primaryKey: true } } }); + await eng.beginTransaction(); + await eng.insert('t', [{ id: 'a' }]); + await eng.savepoint('s'); + await eng.insert('t', [{ id: 'b' }]); + await eng.rollbackToSavepoint('s'); + await eng.commitTransaction(); + const live = (await eng.find('t', { table: 't' })).map((r) => r.id).sort(); + + // 崩溃：不调用 close()（close 会 checkpoint 截断 WAL，掩盖重放路径） + const eng2 = new AriaEngine(CFG) as unknown as typeof eng; + await eng2.open('v080-sp-crash', 1); + const after = (await eng2.find('t', { table: 't' })).map((r) => r.id).sort(); + + expect(live).toEqual(['a']); + expect(after).toEqual(live); // 不得复活 b + await eng2.close(); + }); + + test('无 savepoint 的普通事务崩溃重放仍然完整（不得误伤）', async () => { + const eng = new AriaEngine(CFG) as unknown as { + open(n: string, v: number): Promise; + createTable(s: unknown): Promise; + beginTransaction(): Promise; + insert(t: string, r: unknown[]): Promise; + commitTransaction(): Promise; + find(t: string, q: unknown): Promise>>; + close(): Promise; + }; + await eng.open('v080-sp-crash2', 1); + await eng.createTable({ name: 't', columns: { id: { type: 'string', primaryKey: true } } }); + await eng.beginTransaction(); + await eng.insert('t', [{ id: 'a' }]); + await eng.insert('t', [{ id: 'b' }]); + await eng.commitTransaction(); + const live = (await eng.find('t', { table: 't' })).map((r) => r.id).sort(); + + const eng2 = new AriaEngine(CFG) as unknown as typeof eng; + await eng2.open('v080-sp-crash2', 1); + const after = (await eng2.find('t', { table: 't' })).map((r) => r.id).sort(); + expect(live).toEqual(['a', 'b']); + expect(after).toEqual(live); + await eng2.close(); + }); + + test('多个保存点：回到最早保存点，其后写入全部丢弃', async () => { + const eng = new AriaEngine(CFG) as unknown as { + open(n: string, v: number): Promise; + createTable(s: unknown): Promise; + beginTransaction(): Promise; + insert(t: string, r: unknown[]): Promise; + savepoint(n: string): Promise; + rollbackToSavepoint(n: string): Promise; + commitTransaction(): Promise; + find(t: string, q: unknown): Promise>>; + close(): Promise; + }; + await eng.open('v080-sp-multi', 1); + await eng.createTable({ name: 't', columns: { id: { type: 'string', primaryKey: true } } }); + await eng.beginTransaction(); + await eng.insert('t', [{ id: 'a' }]); + await eng.savepoint('s1'); + await eng.insert('t', [{ id: 'b' }]); + await eng.savepoint('s2'); + await eng.insert('t', [{ id: 'c' }]); + await eng.rollbackToSavepoint('s1'); + await eng.commitTransaction(); + const live = (await eng.find('t', { table: 't' })).map((r) => r.id).sort(); + + const eng2 = new AriaEngine(CFG) as unknown as typeof eng; + await eng2.open('v080-sp-multi', 1); + const after = (await eng2.find('t', { table: 't' })).map((r) => r.id).sort(); + + expect(live).toEqual(['a']); + expect(after).toEqual(['a']); + await eng2.close(); + }); +});