284 lines
8.6 KiB
TypeScript
284 lines
8.6 KiB
TypeScript
/**
|
||
* AriaEngine WAL — Write-Ahead Log
|
||
* @module engine/aria/wal/log
|
||
*
|
||
* 崩溃恢复前的写操作持久化日志。
|
||
*
|
||
* WAL 文件格式:
|
||
* ┌──────────┬──────────────┬──────────┐
|
||
* │ Record 1│ Record 2 │ ... │
|
||
* │ 4B LSN │ │ │
|
||
* │ 1B type │ │ │
|
||
* │ 4B txnId│ │ │
|
||
* │ 2B tblLen│ │ │
|
||
* │ N table│ │ │
|
||
* │ 2B keyLen│ │ │
|
||
* │ N key │ │ │
|
||
* │ 4B jsonLen│ │ │
|
||
* │ N json │ │ │
|
||
* │ 4B CRC │ │ │
|
||
* └──────────┴──────────────┴──────────┘
|
||
*/
|
||
|
||
import { WALRecordType, type WALRecord } from '../types';
|
||
import type { BufferPool } from '../buffer/pool';
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// WAL 存储接口
|
||
// ---------------------------------------------------------------------------
|
||
|
||
export interface WALStore {
|
||
/** 追加 WAL 记录 */
|
||
append(data: Uint8Array): Promise<void>;
|
||
/** 读取所有 WAL 记录 */
|
||
readAll(): Promise<Uint8Array>;
|
||
/** 截断 WAL（checkpoint 后清理） */
|
||
truncate(): Promise<void>;
|
||
/** 检查 WAL 是否存在 */
|
||
exists(): Promise<boolean>;
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// WAL
|
||
// ---------------------------------------------------------------------------
|
||
|
||
export class WAL {
|
||
private lsn = 0;
|
||
private store: WALStore;
|
||
private enabled: boolean;
|
||
private buffer: Uint8Array[] = [];
|
||
private syncMode: 'full' | 'batch' | 'none';
|
||
|
||
constructor(store: WALStore, enabled: boolean = true, syncMode: 'full' | 'batch' | 'none' = 'batch') {
|
||
this.store = store;
|
||
this.enabled = enabled;
|
||
this.syncMode = syncMode;
|
||
}
|
||
|
||
// =======================================================================
|
||
// 写入
|
||
// =======================================================================
|
||
|
||
/** 追加一条 WAL 记录（full 模式同步等待写入完成） */
|
||
async append(record: Omit<WALRecord, 'lsn' | 'checksum'>): Promise<void> {
|
||
if (!this.enabled) return;
|
||
|
||
this.lsn++;
|
||
const fullRecord: WALRecord = {
|
||
...record,
|
||
lsn: this.lsn,
|
||
checksum: 0, // 稍后计算
|
||
};
|
||
|
||
const bytes = this.encodeRecord(fullRecord);
|
||
|
||
if (this.syncMode === 'full') {
|
||
try {
|
||
await this.store.append(bytes);
|
||
} catch {
|
||
// eslint-disable-next-line no-console
|
||
console.warn('[AriaEngine WAL] Failed to append record');
|
||
}
|
||
} else if (this.syncMode === 'batch') {
|
||
this.buffer.push(bytes);
|
||
}
|
||
// 'none' mode: 不写 WAL
|
||
}
|
||
|
||
/** 批量刷新缓冲的 WAL 记录 */
|
||
async flush(): Promise<void> {
|
||
if (!this.enabled || this.buffer.length === 0) return;
|
||
|
||
const totalLen = this.buffer.reduce((sum, b) => sum + b.byteLength, 0);
|
||
const combined = new Uint8Array(totalLen);
|
||
let offset = 0;
|
||
for (const buf of this.buffer) {
|
||
combined.set(buf, offset);
|
||
offset += buf.byteLength;
|
||
}
|
||
|
||
await this.store.append(combined);
|
||
this.buffer = [];
|
||
}
|
||
|
||
// =======================================================================
|
||
// 恢复
|
||
// =======================================================================
|
||
|
||
/** 从 WAL 恢复未提交的事务数据 */
|
||
async recover(
|
||
applyRecord: (record: WALRecord) => void,
|
||
): Promise<number> {
|
||
if (!this.enabled) return 0;
|
||
|
||
const exists = await this.store.exists();
|
||
if (!exists) return 0;
|
||
|
||
const data = await this.store.readAll();
|
||
if (data.byteLength === 0) return 0;
|
||
|
||
const records = this.decodeAllRecords(data);
|
||
for (const record of records) {
|
||
applyRecord(record);
|
||
}
|
||
|
||
this.lsn = records.length > 0 ? records[records.length - 1].lsn : 0;
|
||
return records.length;
|
||
}
|
||
|
||
// =======================================================================
|
||
// Checkpoint
|
||
// =======================================================================
|
||
|
||
/** Checkpoint 后清空 WAL */
|
||
async checkpoint(): Promise<void> {
|
||
if (!this.enabled) return;
|
||
await this.flush();
|
||
await this.store.truncate();
|
||
this.lsn = 0;
|
||
}
|
||
|
||
// =======================================================================
|
||
// 统计
|
||
// =======================================================================
|
||
|
||
isEnabled(): boolean {
|
||
return this.enabled;
|
||
}
|
||
|
||
getLSN(): number {
|
||
return this.lsn;
|
||
}
|
||
|
||
getBufferedCount(): number {
|
||
return this.buffer.length;
|
||
}
|
||
|
||
// -----------------------------------------------------------------------
|
||
// 编解码
|
||
// -----------------------------------------------------------------------
|
||
|
||
private encodeRecord(record: WALRecord): Uint8Array {
|
||
const encoder = new TextEncoder();
|
||
const tableBytes = encoder.encode(record.tableName);
|
||
const keyBytes = encoder.encode(record.key);
|
||
const jsonStr = record.data ? JSON.stringify(record.data) : '';
|
||
const jsonBytes = encoder.encode(jsonStr);
|
||
|
||
const size =
|
||
4 + // LSN
|
||
1 + // type
|
||
4 + // txnId
|
||
2 + tableBytes.length + // table
|
||
2 + keyBytes.length + // key
|
||
4 + jsonBytes.length + // json
|
||
4; // CRC
|
||
|
||
const buf = new ArrayBuffer(size);
|
||
const view = new DataView(buf);
|
||
let offset = 0;
|
||
|
||
view.setUint32(offset, record.lsn, false);
|
||
offset += 4;
|
||
view.setUint8(offset, record.type);
|
||
offset += 1;
|
||
view.setUint32(offset, record.txnId, false);
|
||
offset += 4;
|
||
|
||
view.setUint16(offset, tableBytes.length, false);
|
||
offset += 2;
|
||
new Uint8Array(buf).set(tableBytes, offset);
|
||
offset += tableBytes.length;
|
||
|
||
view.setUint16(offset, keyBytes.length, false);
|
||
offset += 2;
|
||
new Uint8Array(buf).set(keyBytes, offset);
|
||
offset += keyBytes.length;
|
||
|
||
view.setUint32(offset, jsonBytes.length, false);
|
||
offset += 4;
|
||
new Uint8Array(buf).set(jsonBytes, offset);
|
||
offset += jsonBytes.length;
|
||
|
||
// 简单 CRC
|
||
let crc = 0;
|
||
const u8 = new Uint8Array(buf, 0, offset);
|
||
for (let i = 0; i < u8.length; i++) {
|
||
crc = ((crc << 5) - crc + u8[i]) | 0;
|
||
}
|
||
view.setUint32(offset, crc >>> 0, false);
|
||
|
||
return new Uint8Array(buf);
|
||
}
|
||
|
||
private decodeAllRecords(data: Uint8Array): WALRecord[] {
|
||
const records: WALRecord[] = [];
|
||
const view = new DataView(data.buffer, data.byteOffset, data.byteLength);
|
||
let offset = 0;
|
||
|
||
while (offset + 15 <= data.byteLength) {
|
||
try {
|
||
const recordStart = offset;
|
||
const lsn = view.getUint32(offset, false);
|
||
offset += 4;
|
||
const type = view.getUint8(offset) as WALRecordType;
|
||
offset += 1;
|
||
const txnId = view.getUint32(offset, false);
|
||
offset += 4;
|
||
|
||
const tableLen = view.getUint16(offset, false);
|
||
offset += 2;
|
||
if (offset + tableLen > data.byteLength) break;
|
||
const tableName = new TextDecoder().decode(data.slice(offset, offset + tableLen));
|
||
offset += tableLen;
|
||
|
||
const keyLen = view.getUint16(offset, false);
|
||
offset += 2;
|
||
if (offset + keyLen > data.byteLength) break;
|
||
const key = new TextDecoder().decode(data.slice(offset, offset + keyLen));
|
||
offset += keyLen;
|
||
|
||
const jsonLen = view.getUint32(offset, false);
|
||
offset += 4;
|
||
if (offset + jsonLen > data.byteLength) break;
|
||
let recordData: Record<string, unknown> | undefined;
|
||
if (jsonLen > 0) {
|
||
const json = new TextDecoder().decode(data.slice(offset, offset + jsonLen));
|
||
try {
|
||
recordData = JSON.parse(json);
|
||
} catch { /* ok */ }
|
||
}
|
||
offset += jsonLen;
|
||
|
||
// 验证 CRC（跨记录数据计算，不含 CRC 自身）
|
||
const storedCrc = view.getUint32(offset, false);
|
||
offset += 4;
|
||
const recordBytes = data.slice(recordStart, offset - 4);
|
||
let computedCrc = 0;
|
||
for (let i = 0; i < recordBytes.length; i++) {
|
||
computedCrc = ((computedCrc << 5) - computedCrc + recordBytes[i]) | 0;
|
||
}
|
||
if ((computedCrc >>> 0) !== storedCrc) {
|
||
// CRC 不匹配，跳过此损坏记录
|
||
console.warn(`[AriaEngine WAL] CRC mismatch at record LSN=${lsn}, skipping`);
|
||
continue;
|
||
}
|
||
|
||
records.push({
|
||
lsn,
|
||
type,
|
||
txnId,
|
||
tableName,
|
||
key,
|
||
data: recordData,
|
||
checksum: storedCrc,
|
||
});
|
||
} catch {
|
||
break;
|
||
}
|
||
}
|
||
|
||
return records;
|
||
}
|
||
}
|