/** * 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 { crc32 } from '../crc32'; // --------------------------------------------------------------------------- // WAL 存储接口 // --------------------------------------------------------------------------- export interface WALStore { /** 追加 WAL 记录 */ append(data: Uint8Array): Promise; /** 读取所有 WAL 记录 */ readAll(): Promise; /** 截断 WAL(checkpoint 后清理) */ truncate(): Promise; /** 检查 WAL 是否存在 */ exists(): Promise; } // --------------------------------------------------------------------------- // WAL // --------------------------------------------------------------------------- export class WAL { private lsn = 0; private store: WALStore; private enabled: boolean; private buffer: Uint8Array[] = []; private syncMode: 'full' | 'batch' | 'none'; /** v0.3.3: 未 checkpoint 的 WAL 累计字节数(full/batch/none 通用) */ private bufferedBytes = 0; 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): Promise { if (!this.enabled) return; this.lsn++; const fullRecord: WALRecord = { ...record, lsn: this.lsn, checksum: 0, // 稍后计算 }; const bytes = this.encodeRecord(fullRecord); if (this.syncMode === 'full') { // v0.6.3-fix: 写入失败必须抛给调用方 —— 此前仅 console.warn 吞错: // 内存已提交而 WAL 缺失,崩溃即丢且调用方无感知。 await this.store.append(bytes); this.bufferedBytes += bytes.byteLength; } else if (this.syncMode === 'batch') { this.buffer.push(bytes); this.bufferedBytes += bytes.byteLength; } // 'none' mode: 不写 WAL } /** 批量追加多条 WAL 记录(组提交:合并为一次底层写入,v0.3.1) */ async appendBatch(records: Omit[]): Promise { if (!this.enabled || records.length === 0) return; const chunks: Uint8Array[] = []; for (const record of records) { this.lsn++; chunks.push(this.encodeRecord({ ...record, lsn: this.lsn, checksum: 0 })); } const combined = this.mergeChunks(chunks); if (this.syncMode === 'full') { // v0.6.3-fix: 同 append —— 批量写入失败抛给调用方(不再吞错) await this.store.append(combined); this.bufferedBytes += combined.byteLength; } else if (this.syncMode === 'batch') { this.buffer.push(combined); this.bufferedBytes += combined.byteLength; } // 'none' mode: 不写 WAL } /** 批量刷新缓冲的 WAL 记录 */ async flush(): Promise { if (!this.enabled || this.buffer.length === 0) return; const combined = this.mergeChunks(this.buffer); await this.store.append(combined); this.buffer = []; } /** 合并多个字节块为一个连续缓冲区 */ private mergeChunks(chunks: Uint8Array[]): Uint8Array { if (chunks.length === 1) return chunks[0]; const totalLen = chunks.reduce((sum, b) => sum + b.byteLength, 0); const combined = new Uint8Array(totalLen); let offset = 0; for (const buf of chunks) { combined.set(buf, offset); offset += buf.byteLength; } return combined; } // ======================================================================= // 恢复 // ======================================================================= /** 从 WAL 恢复未提交的事务数据 */ async recover( applyRecord: (record: WALRecord) => void, ): Promise { 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 { if (!this.enabled) return; await this.flush(); await this.store.truncate(); this.lsn = 0; this.bufferedBytes = 0; } // ======================================================================= // 统计 // ======================================================================= getBufferedCount(): number { return this.buffer.length; } /** v0.3.3: 未 checkpoint 的 WAL 累计字节数(full/batch/none 通用) */ getBufferedBytes(): number { return this.bufferedBytes; } // ----------------------------------------------------------------------- // 编解码 // ----------------------------------------------------------------------- /** 旧版弱滚动校验(v0.4.4 及更早写入的 WAL 记录使用,双算法探测兼容) */ private legacyChecksum(data: Uint8Array): number { let crc = 0; for (let i = 0; i < data.length; i++) { crc = ((crc << 5) - crc + data[i]) | 0; } return crc >>> 0; } 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; // v0.4.5: 标准 CRC-32 校验(此前为弱滚动校验,误检率更高) const u8 = new Uint8Array(buf, 0, offset); const crc = crc32(u8); view.setUint32(offset, crc, 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 | 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-32,失败再尝试旧版弱滚动校验(兼容旧库 WAL 记录) const storedCrc = view.getUint32(offset, false); offset += 4; const recordBytes = data.slice(recordStart, offset - 4); const computedNew = crc32(recordBytes); const computedLegacy = this.legacyChecksum(recordBytes); if ((computedNew >>> 0) !== storedCrc && (computedLegacy >>> 0) !== storedCrc) { // CRC 不匹配,跳过此损坏记录(长度字段链完整时后续好记录仍可恢复, // 行为由 aria-wal-crc 测试锁定) // eslint-disable-next-line no-console 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; } }