4. 实现WAL日志与崩溃恢复

任务说明

实现一个最小可运行的WAL（Write-Ahead Logging）模块，支持日志追加写入、日志回放，并基于WAL实现崩溃恢复：模拟数据库在写入数据页之前先记录日志，在“崩溃”后通过回放日志将数据恢复到崩溃前的一致状态。

核心设计

本案例实现一个单文件数据库存储引擎的核心WAL机制，包含三个部分：

• 日志记录格式：每条日志包含LSN（日志序列号）、事务ID、操作类型（INSERT/UPDATE/DELETE）、键、值以及校验和。
• WAL写入器：所有数据修改先追加到WAL文件（wal.log），并维护当前LSN；数据页（data.db）的修改在日志落盘之后执行。
• 崩溃恢复器：启动时读取WAL，重放所有日志到数据页，实现崩溃前已提交事务的恢复。

代码实现

import os
import struct
import hashlib

# ---------- 日志格式定义 ----------
# 日志记录二进制布局（固定40字节头 + 变长key/value）:
#   LSN        : 8字节无符号整数
#   txn_id     : 4字节无符号整数
#   op_type    : 1字节 (1=INSERT, 2=UPDATE, 3=DELETE)
#   key_len    : 4字节无符号整数
#   value_len  : 4字节无符号整数
#   checksum   : 16字节 (MD5 of payload)
#   key        : key_len 字节
#   value      : value_len 字节 (DELETE时value_len=0)

OP_INSERT = 1
OP_UPDATE = 2
OP_DELETE = 3
HEADER_FMT = ">QIBII16s"
HEADER_SIZE = struct.calcsize(HEADER_FMT)

class WalLog:
    """单条WAL日志记录"""
    def __init__(self, lsn, txn_id, op_type, key, value=b""):
        self.lsn = lsn
        self.txn_id = txn_id
        self.op_type = op_type
        self.key = key
        self.value = value

    def serialize(self):
        payload = self.key + self.value
        checksum = hashlib.md5(payload).digest()
        header = struct.pack(HEADER_FMT, self.lsn, self.txn_id,
                             self.op_type, len(self.key),
                             len(self.value), checksum)
        return header + payload

    @staticmethod
    def deserialize(data):
        header = data[:HEADER_SIZE]
        lsn, txn_id, op_type, key_len, value_len, checksum = \
            struct.unpack(HEADER_FMT, header)
        key_start = HEADER_SIZE
        key_end = key_start + key_len
        value_end = key_end + value_len
        key = data[key_start:key_end]
        value = data[key_end:value_end]
        # 校验和验证
        payload = key + value
        if hashlib.md5(payload).digest() != checksum:
            raise ValueError("Checksum mismatch, log corrupted")
        return WalLog(lsn, txn_id, op_type, key, value)

class WalWriter:
    """WAL写入器：负责日志追加与数据页写入"""
    def __init__(self, wal_path="wal.log", data_path="data.db"):
        self.wal_path = wal_path
        self.data_path = data_path
        self.current_lsn = 0
        self._wal_file = open(wal_path, "ab")
        self._data_file = open(data_path, "r+b" if os.path.exists(data_path) else "w+b")

    def append_log(self, txn_id, op_type, key, value=b""):
        """先写WAL，再写数据页（WAL原则）"""
        self.current_lsn += 1
        log = WalLog(self.current_lsn, txn_id, op_type, key, value)
        # 1. 日志先落盘
        self._wal_file.write(log.serialize())
        self._wal_file.flush()
        os.fsync(self._wal_file.fileno())
        # 2. 再更新数据页
        self._apply_to_data_page(log)
        return self.current_lsn

    def _apply_to_data_page(self, log):
        """将日志操作应用到数据文件（模拟数据页）"""
        # 数据文件布局：每条记录为 [key_len(4B)][key][value_len(4B)][value]
        # 为简化演示，使用内存字典模拟数据页，实际可替换为B+树等结构
        # 此处直接操作文件：读取全部到内存，修改后写回
        data = self._read_all_data()
        if log.op_type == OP_INSERT or log.op_type == OP_UPDATE:
            data[log.key] = log.value
        elif log.op_type == OP_DELETE:
            data.pop(log.key, None)
        self._write_all_data(data)

    def _read_all_data(self):
        """读取数据文件全部内容为字典"""
        self._data_file.seek(0)
        raw = self._data_file.read()
        data = {}
        offset = 0
        while offset < len(raw):
            key_len = struct.unpack(">I", raw[offset:offset+4])[0]
            offset += 4
            key = raw[offset:offset+key_len]
            offset += key_len
            value_len = struct.unpack(">I", raw[offset:offset+4])[0]
            offset += 4
            value = raw[offset:offset+value_len]
            offset += value_len
            data[key] = value
        return data

    def _write_all_data(self, data):
        """将字典写回数据文件"""
        self._data_file.seek(0)
        self._data_file.truncate()
        for key, value in data.items():
            self._data_file.write(struct.pack(">I", len(key)))
            self._data_file.write(key)
            self._data_file.write(struct.pack(">I", len(value)))
            self._data_file.write(value)
        self._data_file.flush()
        os.fsync(self._data_file.fileno())

    def close(self):
        self._wal_file.close()
        self._data_file.close()

class CrashRecovery:
    """崩溃恢复器：启动时扫描WAL并重放日志"""
    def __init__(self, wal_path="wal.log", data_path="data.db"):
        self.wal_path = wal_path
        self.data_path = data_path

    def recover(self):
        """执行崩溃恢复，返回恢复的日志条数"""
        if not os.path.exists(self.wal_path):
            return 0
        recovered = 0
        with open(self.wal_path, "rb") as f:
            while True:
                header = f.read(HEADER_SIZE)
                if len(header) < HEADER_SIZE:
                    break  # 文件末尾或部分写入（模拟崩溃点）
                lsn, txn_id, op_type, key_len, value_len, checksum = \
                    struct.unpack(HEADER_FMT, header)
                payload_len = key_len + value_len
                payload = f.read(payload_len)
                if len(payload) < payload_len:
                    break  # 日志不完整，忽略（崩溃时未写完）
                full_record = header + payload
                try:
                    log = WalLog.deserialize(full_record)
                except ValueError:
                    break  # 校验失败，停止恢复
                # 重放日志到数据页
                self._replay(log)
                recovered += 1
        return recovered

    def _replay(self, log):
        """将单条日志重放到数据文件"""
        # 与WalWriter._apply_to_data_page逻辑相同
        data = self._read_all_data()
        if log.op_type == OP_INSERT or log.op_type == OP_UPDATE:
            data[log.key] = log.value
        elif log.op_type == OP_DELETE:
            data.pop(log.key, None)
        self._write_all_data(data)

    def _read_all_data(self):
        """读取数据文件（与WalWriter相同）"""
        if not os.path.exists(self.data_path):
            return {}
        with open(self.data_path, "rb") as f:
            raw = f.read()
        data = {}
        offset = 0
        while offset < len(raw):
            key_len = struct.unpack(">I", raw[offset:offset+4])[0]
            offset += 4
            key = raw[offset:offset+key_len]
            offset += key_len
            value_len = struct.unpack(">I", raw[offset:offset+4])[0]
            offset += 4
            value = raw[offset:offset+value_len]
            offset += value_len
            data[key] = value
        return data

    def _write_all_data(self, data):
        """写回数据文件（与WalWriter相同）"""
        with open(self.data_path, "wb") as f:
            for key, value in data.items():
                f.write(struct.pack(">I", len(key)))
                f.write(key)
                f.write(struct.pack(">I", len(value)))
                f.write(value)
            f.flush()
            os.fsync(f.fileno())

# ---------- 演示与验证 ----------
def demo_crash_recovery():
    """模拟一次正常写入 + 一次崩溃 + 恢复"""
    # 清理旧文件
    for f in ["wal.log", "data.db"]:
        if os.path.exists(f):
            os.remove(f)

    # 阶段1：正常写入两条记录
    writer = WalWriter()
    writer.append_log(txn_id=1, op_type=OP_INSERT, key=b"user:1", value=b"Alice")
    writer.append_log(txn_id=1, op_type=OP_INSERT, key=b"user:2", value=b"Bob")
    writer.close()
    print("正常写入完成，WAL中日志条数: 2")

    # 阶段2：模拟崩溃——直接删除数据文件（数据页丢失），WAL保留
    os.remove("data.db")
    print("模拟崩溃：数据文件已丢失，WAL日志保留")

    # 阶段3：执行崩溃恢复
    recovery = CrashRecovery()
    count = recovery.recover()
    print(f"崩溃恢复完成，重放日志条数: {count}")

    # 验证恢复结果
    with open("data.db", "rb") as f:
        raw = f.read()
    offset = 0
    recovered_data = {}
    while offset < len(raw):
        key_len = struct.unpack(">I", raw[offset:offset+4])[0]
        offset += 4
        key = raw[offset:offset+key_len]
        offset += key_len
        value_len = struct.unpack(">I", raw[offset:offset+4])[0]
        offset += 4
        value = raw[offset:offset+value_len]
        offset += value_len
        recovered_data[key] = value
    assert recovered_data == {b"user:1": b"Alice", b"user:2": b"Bob"}, \
        f"恢复数据不匹配: {recovered_data}"
    print("验证通过：崩溃前写入的数据已完整恢复")

if __name__ == "__main__":
    demo_crash_recovery()

运行与验证

直接运行上述脚本：

python wal_crash_recovery.py

预期输出：

正常写入完成，WAL中日志条数: 2
模拟崩溃：数据文件已丢失，WAL日志保留
崩溃恢复完成，重放日志条数: 2
验证通过：崩溃前写入的数据已完整恢复

判定标准：程序退出码为0，且最后一行输出“验证通过”即表示WAL写入与崩溃恢复功能正确。若数据文件被删除后能通过WAL完整恢复出user:1=Alice和user:2=Bob两条记录，则任务闭环达成。