Своя СУБД на Rust · урок 7: журнал упреждающей записи и восстановление

Кому это. К этому уроку у тебя собран целый движок на Rust. В уроке 6 появилась верхняя половина: каталог, binder, планировщик, операторы Volcano за Box<dyn Operator>. CREATE TABLE, INSERT, SELECT ... WHERE и EXPLAIN уже бегут. Но убей процесс посреди работы, открой базу снова, и часть строк пропадёт.

Зачем именно Rust. Сегодня закрываем половину этой дыры через журнал упреждающей записи и фазу восстановления. На Rust здесь видны три вещи, которых не было в TS. Первая: контрольную сумму мы пишем сами, таблицей CRC-32, которую компилятор считает ещё до запуска программы через const fn. Крейт с CRC не берём: курс собирает движок только на std, без зависимостей. Вторая: момент долговечности назван явно, это File::sync_all, и каждый вызов возвращает io::Result, который ? превращает в наш Error::Io. Третья: журнал одновременно нужен базе и каждому буферному пулу, и Rust не даст раздать его по &mut. Поэтому журнал живёт в Arc<Mutex<WalManager>>, и сегодня станет понятно, почему не Rc<RefCell<…>>.

Важно про объём. Это первый из двух уроков про долговечность. Сегодня механика журнала и redo на самом простом пути: одиночная вставка в режиме autocommit. Транзакций, отката и удаления через версии ещё нет, это следующий урок.

Где мы в движке

┌────────────────────────────────────────────────┐
│  Бинарный протокол + TCP-сервер                  │  урок 11
├────────────────────────────────────────────────┤
│  Исполнитель (Volcano) + EXPLAIN                 │  урок 6
│  Планировщик, binder, каталог                    │  урок 6
│  Парсер SQL (рекурсивный спуск + Pratt)          │  урок 5
├────────────────────────────────────────────────┤
│  Vacuum                                          │  урок 10
│  MVCC (видимость, версии кортежа)                │  урок 8
│  Блокировки (2PL), конкуренция                   │  урок 9
│  Менеджер транзакций (xid, состояния)            │  урок 8
│  WAL (журнал, redo, recovery)                    │  урок 7  ← мы здесь
├────────────────────────────────────────────────┤
│  B+Tree индекс                                   │  урок 4
│  Куча (slotted pages, seq scan)                  │  урок 3
├────────────────────────────────────────────────┤
│  Буферный пул (кэш страниц, вытеснение)           │  урок 2
│  DiskManager (страницы поверх файла)              │  урок 1
└────────────────────────────────────────────────┘

WAL это поперечный слой: он висит сбоку и страхует все изменения страниц. Сегодня проведём его к трём местам: к куче (вставка дописывает запись), к буферному пулу (страница не уходит на диск раньше своей записи) и к Database::open (старт проигрывает журнал).

Что теряется при падении

Вспомни путь вставки. INSERT зовёт heap.insert_tuple, та кладёт байты в страницу через guard.write(), и Drop у PageGuard помечает кадр грязным. На диск страница не идёт: пул сбрасывает её лениво, при вытеснении или в flush_all. Для скорости это правильно, для долговечности смертельно. Между вставкой и сбросом строка живёт только в памяти.

В Rust-порте это видно особенно честно: у Database нет Drop, который сбрасывал бы пул. Выброшенная база ведёт себя как упавший процесс, а штатное завершение это явный db.close(). Поэтому тесты урока имитируют падение одной строкой: открыть новую базу на том же каталоге и выбросить старую.

Идея write-ahead простая. Перед тем как менять данные, опиши изменение маленькой записью в отдельном файле и форсни на диск журнал. Журнал пишется только в конец, это дёшево. Страницы данных пусть сбрасываются лениво, как раньше. Упали? При старте читаем журнал и заново применяем (redo) всё, что журнал успел зафиксировать, а данные нет.

Формат записи

Запись журнала самоописывающая: заголовок фиксированной формы, payload переменной длины и CRC-32 в хвосте. Числа little-endian, как весь формат на диске. Бинарный контракт общий для всех портов воркшопа (examples/our-db/SPEC.md), поэтому журнал, записанный TS-движком, байт в байт совпадает с нашим.

// src/wal.rs
use std::fs::{File, OpenOptions};
use std::io::{Read, Seek, SeekFrom, Write};
use std::path::Path;

use crate::bytes::slice;
use crate::bytes::{get_u16, get_u32, get_u64, put_u16, put_u32, put_u64, read_u16, read_u32};
use crate::error::Result;
use crate::page::PAGE_SIZE;
use crate::tuple::Xid;

/// Заголовок записи до payload: lsn u64, prev_lsn u64, xid u32, type u16, length u32.
pub const WAL_RECORD_HEADER_SIZE: usize = 26;
const CRC_SIZE: usize = 4;

/// Числовые коды типа записи. Зафиксированы навсегда; 6 (бывший DELETE) это дыра.
pub mod kind {
    pub const BEGIN: u16 = 1;
    pub const COMMIT: u16 = 2;
    pub const ABORT: u16 = 3;
    pub const CREATE: u16 = 4;
    pub const INSERT: u16 = 5;
    pub const CHECKPOINT: u16 = 7;
    pub const SET_XMAX: u16 = 8;
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct WalRecord {
    pub lsn: u64,
    pub prev_lsn: u64,
    pub xid: Xid,
    pub kind: u16,
    pub payload: Vec<u8>,
}

lsn это номер записи, монотонный. prev_lsn пригодится в следующем уроке для цепочки записей одной транзакции, сегодня пишем 0. Типы записей сделаны модулем констант, а не enum. Причина в контракте: в файле лежит голый u16, и незнакомый код (запись из будущей версии) не должен ронять разбор. С enum пришлось бы писать TryFrom<u16> и решать, что делать с ошибкой; модуль констант даёт match record.kind { kind::INSERT => … } и ветку _ для всего остального. Номер 6 сознательно пропущен: физического DELETE в журнале не будет, удаление станет записью SET_XMAX.

CRC-32 своими руками

TS-движок берёт crc32 из node:zlib. В std у Rust такой функции нет, а крейт мы не подключаем. Пишем сами. Это CRC-32 в варианте IEEE, тот же, что у zlib, поэтому суммы совпадут с TS-портом.

// src/wal.rs (продолжение)

const CRC_TABLE: [u32; 256] = crc_table();

const fn crc_table() -> [u32; 256] {
    let mut table = [0u32; 256];
    let mut n = 0;
    while n < 256 {
        let mut c = n as u32;
        let mut k = 0;
        while k < 8 {
            c = if c & 1 != 0 {
                0xEDB8_8320 ^ (c >> 1)
            } else {
                c >> 1
            };
            k += 1;
        }
        table[n] = c;
        n += 1;
    }
    table
}

pub fn crc32(data: &[u8]) -> u32 {
    !data.iter().fold(!0u32, |crc, &byte| {
        CRC_TABLE[((crc ^ byte as u32) & 0xFF) as usize] ^ (crc >> 8)
    })
}

Разберём по шагам.

1. 0xEDB8_8320 это порождающий многочлен CRC-32, записанный в отражённом порядке бит. Отражённый, потому что CRC-32 IEEE обрабатывает каждый байт младшим битом вперёд, и тогда сдвиг идёт вправо, а не влево. Подчёркивания в литерале Rust разрешает для читаемости, как в уроке про биты и байты.
2. crc_table проходит 256 возможных байтов и для каждого восемь раз делит на многочлен: если младший бит 1, сдвигаем и делаем XOR с многочленом, иначе просто сдвигаем. Остаток кладём в таблицу.
3. crc_table объявлена const fn, а CRC_TABLE это const. Компилятор выполнит функцию при сборке, и в бинарник попадёт готовый массив из 256 чисел. Ни ленивой инициализации, ни OnceLock, ни расходов при запуске.
4. Внутри const fn нельзя писать for n in 0..256: цикл for разворачивается в вызовы трейта Iterator, а они не константные. Поэтому два while со счётчиками. Это не стиль автора, это ограничение константных вычислений.
5. crc32 начинает с !0u32 (все единицы), на каждом байте берёт строку таблицы по младшему байту текущей суммы и сдвигает остаток. В конце снова !. Начальное и финальное инвертирование входят в стандарт CRC-32.

Проверочное значение этого стандарта общеизвестно: crc32(b"123456789") равно 0xCBF43926. Его и проверяет тест урока, поэтому ошибку в таблице ты увидишь сразу.

Кодек записи

// src/wal.rs (продолжение)

pub fn encode_record(lsn: u64, prev_lsn: u64, xid: Xid, kind: u16, payload: &[u8]) -> Vec<u8> {
    let body_len = WAL_RECORD_HEADER_SIZE + payload.len();
    let mut out = vec![0u8; WAL_RECORD_HEADER_SIZE];
    put_u64(&mut out, 0, lsn);
    put_u64(&mut out, 8, prev_lsn);
    put_u32(&mut out, 16, xid);
    put_u16(&mut out, 20, kind);
    put_u32(&mut out, 22, payload.len() as u32);
    out.extend_from_slice(payload);
    let crc = crc32(&out[..body_len]);
    out.extend_from_slice(&crc.to_le_bytes());
    out
}

/// Разобрать все целые записи. Останавливается на первой оборванной или повреждённой
/// (писатель один, рваная запись всегда последняя). Второе значение это смещение сразу
/// за последней целой записью: туда пойдут будущие дозаписи.
pub fn decode_wal(buf: &[u8]) -> (Vec<WalRecord>, usize) {
    let mut records = Vec::new();
    let mut offset = 0;
    while offset + WAL_RECORD_HEADER_SIZE <= buf.len() {
        let length = get_u32(buf, offset + 22) as usize;
        if length > PAGE_SIZE {
            break;
        }
        let body_end = offset + WAL_RECORD_HEADER_SIZE + length;
        if body_end + CRC_SIZE > buf.len()
            || crc32(&buf[offset..body_end]) != get_u32(buf, body_end)
        {
            break;
        }
        records.push(WalRecord {
            lsn: get_u64(buf, offset),
            prev_lsn: get_u64(buf, offset + 8),
            xid: get_u32(buf, offset + 16),
            kind: get_u16(buf, offset + 20),
            payload: buf[offset + WAL_RECORD_HEADER_SIZE..body_end].to_vec(),
        });
        offset = body_end + CRC_SIZE;
    }
    (records, offset)
}

put_* и get_* это помощники из модуля bytes урока 1 поверх to_le_bytes и from_le_bytes. encode_record собирает заголовок, дописывает payload и считает CRC по телу целиком.

Обрати внимание на тип decode_wal. Он возвращает не Result, а пару: записи и valid_end. Оборванный хвост журнала для нас не ошибка, это нормальное состояние после падения посреди write. Разбор идёт, пока запись цела, и останавливается на первой проблеме: длина больше страницы (мусор в заголовке), запись не помещается до конца буфера (оборвали на полуслове), CRC не сошёлся. Писатель один и пишет только в конец, поэтому испорченной может быть лишь последняя запись. valid_end указывает сразу за последней целой записью: туда WalManager продолжит писать, затерев огрызок.

Индексация buf[offset + 22] внутри get_u32 не может выйти за границу: цикл уже проверил, что заголовок помещается. А buf[..body_end] защищён проверкой body_end + CRC_SIZE > buf.len() в том же if. Порядок условий в || важен: сначала длина, потом CRC, иначе срез паниковал бы на огрызке.

Payload-кодеки

Сегодня нужны два: CREATE (DDL проигрывается при старте, чтобы пересобрать каталог) и INSERT (образ кортежа плюс его адрес).

// src/wal.rs (продолжение)

/// INSERT: relation_id u32, page_id u32, slot u16, образ кортежа.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct InsertPayload {
    pub relation_id: u32,
    pub page_id: u32,
    pub slot: u16,
    pub image: Vec<u8>,
}

impl InsertPayload {
    pub fn encode(&self) -> Vec<u8> {
        let mut out = rid_head(self.relation_id, self.page_id, self.slot);
        out.extend_from_slice(&self.image);
        out
    }

    pub fn decode(payload: &[u8]) -> Result<InsertPayload> {
        Ok(InsertPayload {
            relation_id: read_u32(payload, 0)?,
            page_id: read_u32(payload, 4)?,
            slot: read_u16(payload, 8)?,
            image: slice(payload, 10, payload.len().saturating_sub(10))?.to_vec(),
        })
    }
}

fn rid_head(relation_id: u32, page_id: u32, slot: u16) -> Vec<u8> {
    let mut out = Vec::with_capacity(16);
    out.extend_from_slice(&relation_id.to_le_bytes());
    out.extend_from_slice(&page_id.to_le_bytes());
    out.extend_from_slice(&slot.to_le_bytes());
    out
}

/// CREATE: relation_id u32, текст DDL. Это же одна таблица в снимке CHECKPOINT.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct CheckpointTable {
    pub relation_id: u32,
    pub sql: String,
}

impl CheckpointTable {
    pub fn encode_create(&self) -> Vec<u8> {
        let mut out = self.relation_id.to_le_bytes().to_vec();
        out.extend_from_slice(self.sql.as_bytes());
        out
    }

    pub fn decode_create(payload: &[u8]) -> Result<CheckpointTable> {
        Ok(CheckpointTable {
            relation_id: read_u32(payload, 0)?,
            sql: String::from_utf8_lossy(&payload[4..]).into_owned(),
        })
    }
}

Payload в Rust-порте это структура с парой методов encode и decode, а не четыре свободные функции, как в TS. Так пара «закодировать и разобрать» живёт рядом, а тест на круговой проход пишется одной строкой assert_eq!(InsertPayload::decode(&p.encode())?, p) благодаря derive(PartialEq).

Разница между get_u32 и read_u32 принципиальная. get_* паникует на выходе за границу и годится там, где границу уже проверили (заголовок записи). read_* возвращает Result: payload пришёл из файла, и CRC защищает от случайной порчи, но не от ошибки в чужом коде, который записал короткий payload. Граница доверия проходит по байтам с диска, как в уроке про кодеки.

CREATE-структура названа CheckpointTable, потому что ровно та же пара (номер отношения и DDL) войдёт в снимок каталога в записи CHECKPOINT следующего урока. INSERT несёт адрес (page_id, slot) и готовый образ кортежа. Это физическое логирование: redo просто положит байты обратно.

WalManager: файл, хвост в памяти и sync_all

// src/wal.rs (продолжение)

pub struct WalManager {
    file: File,
    write_offset: u64,
    next_lsn: u64,
    flushed_lsn: u64,
    pending: Vec<u8>,
}

impl WalManager {
    /// Открыть журнал, продолжить нумерацию после существующих записей.
    pub fn open(path: impl AsRef<Path>) -> Result<WalManager> {
        let mut file = OpenOptions::new()
            .read(true)
            .write(true)
            .create(true)
            .truncate(false)
            .open(path)?;
        let (records, valid_end) = decode_wal(&read_file(&mut file)?);
        let max_lsn = records.last().map_or(0, |record| record.lsn);
        Ok(WalManager {
            file,
            write_offset: valid_end as u64,
            next_lsn: max_lsn + 1,
            flushed_lsn: max_lsn,
            pending: Vec::new(),
        })
    }

    /// Прочитать все целые записи журнала с диска (для recovery).
    pub fn read_all(&mut self) -> Result<Vec<WalRecord>> {
        Ok(decode_wal(&read_file(&mut self.file)?).0)
    }

    /// Дописать запись в память, вернуть присвоенный lsn. На диск она пока не идёт.
    pub fn append(&mut self, xid: Xid, kind: u16, payload: &[u8], prev_lsn: u64) -> u64 {
        let lsn = self.next_lsn;
        self.next_lsn += 1;
        self.pending
            .extend_from_slice(&encode_record(lsn, prev_lsn, xid, kind, payload));
        lsn
    }

    /// Сбросить накопленные записи на диск и форсировать (fsync).
    pub fn flush(&mut self) -> Result<()> {
        if !self.pending.is_empty() {
            self.file.seek(SeekFrom::Start(self.write_offset))?;
            self.file.write_all(&self.pending)?;
            self.write_offset += self.pending.len() as u64;
            self.pending.clear();
        }
        self.file.sync_all()?;
        self.flushed_lsn = self.next_lsn - 1;
        Ok(())
    }

    /// Правило write-ahead: журнал durable до указанного lsn.
    pub fn flush_up_to(&mut self, lsn: u64) -> Result<()> {
        if self.flushed_lsn < lsn {
            self.flush()?;
        }
        Ok(())
    }

    /// Наибольший присвоенный lsn (включая ещё не сброшенные записи).
    pub fn current_lsn(&self) -> u64 {
        self.next_lsn - 1
    }
}

fn read_file(file: &mut File) -> Result<Vec<u8>> {
    let mut bytes = Vec::new();
    file.seek(SeekFrom::Start(0))?;
    file.read_to_end(&mut bytes)?;
    Ok(bytes)
}

open сначала читает существующий файл, берёт valid_end и продолжает нумерацию с последней записи: журнал переживает перезапуски. .truncate(false) выглядит лишним, ведь по умолчанию файл и так не усекается. Но clippy требует назвать это явно, когда стоит .create(true): открыть существующий журнал с обрезкой значило бы стереть историю, и намерение должно быть видно в коде. В уроке 8 появится метод checkpoint, который усекает журнал уже осознанно.

Сердце урока в flush. Сначала write_all отдаёт байты ядру. Это ещё не долговечность: данные лежат в страничном кэше ОС и пропадут вместе с питанием. Долговечность наступает в sync_all. Это fsync, и он возвращает io::Result<()>. Любой сбой (диск отвалился, кончилось место) поднимется через ? как Error::Io и дойдёт до клиента, а не потеряется в логах. flushed_lsn обновляется после успешного sync_all: если синхронизация упала, журнал честно считает, что долговечен только старый хвост.

sync_all против sync_data: первый синхронизирует данные и все метаданные файла, второй только данные и те метаданные, без которых их не прочитать (например, новый размер файла). Для журнала, который растёт в конец, хватило бы и sync_data. Эталон берёт sync_all как консервативный выбор, а сравнение вынесено в упражнение.

append ничего не пишет на диск: он копит закодированные записи в pending. Сто вставок подряд дадут сто записей в памяти и один write_all при сбросе. flush_up_to(lsn) это правило write-ahead в одну строку: если просят долговечность до некоторого lsn, а мы туда ещё не дотянулись, форсим.

У всех методов &mut self. Журнал однопоточный и изменяемый, и Rust не даст двум владельцам писать в него одновременно. А писать захотят двое: база (CREATE, фиксация) и каждый буферный пул (write-ahead перед сбросом страницы). Это решим ниже.

Страница помнит свой redo: insert_tuple_at

Redo должен положить кортеж ровно в тот слот, что записан в журнале, а не «куда влезет». У страницы из урока 1 есть insert_tuple, который сам выбирает следующий слот. Для redo нужен адресный вариант.

// src/page.rs (добавляем в impl Page)

    /// Положить кортеж в заданный слот (redo восстановления). Пропущенные слоты
    /// между прежним lower и новым остаются нулями, то есть мёртвыми.
    pub fn insert_tuple_at(&mut self, slot: u16, data: &[u8]) -> Result<()> {
        let needed_lower = HEADER_SIZE + (slot as usize + 1) * SLOT_SIZE;
        let new_lower = needed_lower.max(self.lower() as usize);
        if new_lower + data.len() > self.upper() as usize {
            return Err(invalid(format!(
                "redo: tuple for slot {slot} does not fit the page"
            )));
        }
        let tuple_offset = self.place(data);
        self.set_slot(slot, tuple_offset);
        self.set_lower(new_lower as u16);
        Ok(())
    }

lower поднимается ровно настолько, чтобы покрыть нужный слот, а пропущенные слоты остаются нулями, то есть DEAD_SLOT. Это важно для верности redo: если между двумя вставками в журнале есть пропуск (в следующем уроке так будет с откатанной вставкой), номера соседних слотов не сдвинутся, и RID в индексе не начнут врать. В отличие от TS-версии, наш метод возвращает Result: образ из журнала, который не влезает на страницу, это повреждённые данные, и redo должен остановиться ошибкой, а не тихо испортить страницу.