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

fix(A3/A4): SAVEPOINT 语义根治 —— 已回滚的行不再复活、陈旧保存点不再吞写入

Browse Source 
A3 已回滚的行在崩溃重启后复活（静默数据错误）
  `ROLLBACK TO <savepoint>` 只改内存快照、**不写 WAL**，而 COMMIT 会把整个 txnId
  标记为已提交，恢复时按"该事务的全部记录"重放 → 被回滚掉的写入被重新应用。
  实测：事务内插入 a、savepoint、插入 b、回滚到 savepoint、提交 →
  实时只剩 a，崩溃重开变成 a+b。

  根治：新增 WALRecordType.SAVEPOINT_ROLLBACK（走既有 data 字段携带
  { replayFromIndex } 边界，二进制格式不变、旧库记录仍可解析）。
  - 写入侧：savepoint() 记录"该事务当时已追加的记录条数"；rollbackToSavepoint()
    先写标记再改内存（与 commit/rollback 的"WAL 领先内存"一致）。
  - 恢复侧：按**事务内**下标计算窗口 —— 保留 [0, keepUpTo)，丢弃
    [keepUpTo, 最后一个标记)，标记之后的记录照常保留。
    注意不能拿全局下标比较：全局数组里混有 txnId=0 的非事务记录（CREATE_TABLE
    等）与其它事务的记录。这个差一错误在实现过程中被测试抓出并修正。
  - 事务内 WAL 记录计数（txnWalRecordCount）在 5 个 appendBatch 站点与 BEGIN
    处维护，事务开始/结束时归零。

A4 跨事务复用的陈旧 savepoint 静默丢弃当前事务的写入
  savepoints 在 commit/rollback 时**从不清空**，且 rollbackToSavepoint 不校验归属。
  实测：上一个事务遗留 savepoint 名 → 新事务 update 后 ROLLBACK TO 该名 + COMMIT，
  写入凭空消失（v=5 被回退成 v=9）。
  根治：事务结束清空 savepoints 与边界表；rollbackToSavepoint 校验
  sp.txnId === currentTxnId，陈旧保存点抛 SAVEPOINT_NOT_FOUND。

新增 tests/v080-savepoint.test.ts（4 个用例）：包含"崩溃重放一致性"、
"普通事务不得误伤"、以及"多个保存点回到最早"的语义护栏。
This commit is contained in:
thzxx
2026-09-14 21:53:46 +08:00
parent 074afd3f1e
commit 89243ef6eb
3 changed files with 270 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/engine/aria/index.ts
+110 -1
View File
@@ -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>` 只回滚内存快照，日志里仍留有 savepoint 之前
// 写入的记录；COMMIT 又把整个 txnId 标记为已提交 —— 于是那些被回滚掉的写入
// 在重启时被重新应用，**已回滚的行复活**（实测：实时只剩 a，崩溃重开变成 a+b）。
//
// 现在 savepoint 回滚会写入 SAVEPOINT_ROLLBACK 记录；恢复时该事务只应用
// **最后一条** SAVEPOINT_ROLLBACK 之后的记录 —— 等价于"回到该保存点"，
// 与实时态严格一致（这样也就不需要 undo 崩溃前已落盘的旧值）。
// 每个事务：{ 保留起点, 丢弃终点 } —— 区间 [start, end) 保留。
//
// 关键：`replayFromIndex` 是**事务内**的记录序号（写入方按事务计数），
// 因此这里必须用"该事务的第几条记录"来比较，不能直接拿全局下标 ——
// 全局下标里还混着 txnId=0 的非事务记录（CREATE_TABLE 等）以及其它事务。
const globalIndexToTxnIndex = new Map<number, number>();
const txnRecordCount = new Map<number, number>();
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<number, { keepUpTo: number; dropFrom: number }>();
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<string, { txnId: number; snapshot: Map<string, Record<string, unknown>> | null }> = new Map();
/**
* v0.8.0: 当前事务在 WAL 中已成功追加的记录条数。
*
* 用途：保存点需要记录"回滚后应保留到哪一条"，否则恢复时无法区分
* "保存点之前的写入"（应保留）与"保存点之后的写入"（应丢弃）。
*/
private txnWalRecordCount = 0;
/** 保存点 → 该保存点时事务的 WAL 记录边界 */
private savepointWalBoundary = new Map<string, number>();
async savepoint(name: string): Promise<void> {
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<void> {
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<string>();
src/engine/aria/types.ts
+22 -1
View File
@@ -121,6 +121,19 @@ export enum WALRecordType {
ROLLBACK = 6,
CREATE_TABLE = 7,
DROP_TABLE = 8,
/**
* v0.8.0: 回滚到保存点。
*
* 为什么必须有这条记录：`ROLLBACK TO <savepoint>` 此前只改内存快照、**不写 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<string, unknown>;
/** 校验和 */
checksum: number;