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

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

Зачем именно Effect. Убей процесс посреди работы и перезапусти. Что осталось на диске? Часть данных лежит только в буферном пуле, в памяти, и при падении испаряется. Сегодня мы закрываем половину этой дыры через журнал упреждающей записи и фазу восстановления. И тут Effect показывает третью грань управления ресурсами. У файлового слоя (урок 1) был один ресурс под Scope. У буферного пула (урок 2) был мьютекс поверх состояния в замыкании. У движка (урок 6) ресурсы таблиц собирались в Scope базы. Теперь сбоку встаёт ещё один файл, журнал, и он тоже живёт под Scope: makeWal берёт дескриптор через Effect.acquireRelease и оставляет закрытие финалайзеру, ровно как DiskManager. А его операции смешанные: те, что трогают только память (append, currentLsn), синхронные, потому что их зовёт синхронный хук пула; те, что ходят на диск (flush, readAll), это Effect. Это типичный паттерн порта на Effect: не всё подряд лезет в Effect, граница идёт по реальным ресурсам.

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

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

┌────────────────────────────────────────────────┐
│  Бинарный протокол + 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 это поперечный слой: он не лежит между двумя другими, он висит сбоку и страхует все изменения страниц. Сегодня мы заведём его и проведём к двум местам: к слою кучи (вставка дописывает запись в журнал) и к буферному пулу (страница не уходит на диск раньше своей журнальной записи). А на старте makeDatabase добавится фаза восстановления.

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

Вспомни путь вставки. INSERT зовёт heap.insertTuple, та кладёт кортеж в страницу буферного пула и помечает её грязной через pinned.markDirty. Грязная страница на диск не идёт сразу: пул сбрасывает её лениво, при вытеснении или при flushAll. Это правильно для скорости (диск медленный, в память пишем часто), но смертельно для долговечности: между вставкой и сбросом страница живёт только в памяти. Питание выключили, процесс убили, и кортеж исчез, хотя клиенту мы уже сказали “готово”.

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

Константы журнала

Запись журнала это самоописывающий блок: заголовок фиксированной формы, переменный payload и контрольная сумма. Добавь в src/constants.ts (числа на диске little-endian, как и весь формат).

// src/constants.ts (добавляем)

/**
 * Константы WAL (см. SPEC.md, раздел "Запись WAL"). Журнал это последовательность
 * самоописывающих записей переменной длины, числа little-endian.
 */

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

/** Смещения полей заголовка записи WAL. */
export const WalOffset = {
  lsn: 0,
  prevLsn: 8,
  xid: 16,
  type: 20,
  length: 22,
} as const;

/**
 * Типы записей WAL. Сегодня нужны create и insert; begin/commit/abort/setXmax/checkpoint
 * появятся в следующем уроке вместе с транзакциями. Тип 6 (физический DELETE) сознательно
 * пропущен: удаление будет журналироваться через setXmax, поэтому номер 6 не занимаем.
 */
export const WalRecordType = {
  begin: 1,
  commit: 2,
  abort: 3,
  create: 4,
  insert: 5,
  checkpoint: 7,
  setXmax: 8,
} as const;

lsn это номер записи, монотонный. prev_lsn пригодится в следующем уроке для цепочки записей одной транзакции (сегодня пишем 0). xid это номер транзакции; сегодня все вставки идут под фиктивным xid ENGINE_XID=1 из урока 6. Мы сразу заводим полный набор типов: формат байтов это контракт, и менять его задним числом дороже.

Кодек записи

Положи в src/wal.ts. Кодек чистый, без ввода-вывода, поэтому совпадает с TS-треком.

// src/wal.ts
import { constants } from 'node:fs';
import { type FileHandle, open } from 'node:fs/promises';
import { crc32 } from 'node:zlib';

import { Effect, type Scope } from 'effect';

import { PAGE_SIZE, WAL_RECORD_HEADER_SIZE, WalOffset, WalRecordType } from '~/constants.ts';
import { IoError } from '~/errors.ts';

export interface WalRecord {
  readonly lsn: bigint;
  readonly prevLsn: bigint;
  readonly xid: number;
  readonly type: number;
  readonly payload: Buffer;
}

const CRC_SIZE = 4;

export function encodeRecord(
  lsn: bigint,
  previousLsn: bigint,
  xid: number,
  type: number,
  payload: Buffer,
): Buffer {
  const body = Buffer.alloc(WAL_RECORD_HEADER_SIZE + payload.byteLength);
  body.writeBigUInt64LE(lsn, WalOffset.lsn);
  body.writeBigUInt64LE(previousLsn, WalOffset.prevLsn);
  body.writeUInt32LE(xid, WalOffset.xid);
  body.writeUInt16LE(type, WalOffset.type);
  body.writeUInt32LE(payload.byteLength, WalOffset.length);
  payload.copy(body, WAL_RECORD_HEADER_SIZE);

  const full = Buffer.alloc(body.byteLength + CRC_SIZE);
  body.copy(full, 0);
  full.writeUInt32LE(crc32(body) >>> 0, body.byteLength);
  return full;
}

encodeRecord собирает запись: заголовок, payload, и в хвост CRC32 по всему телу. Контрольная сумма ловит оборванную или побитую запись: если процесс убили посреди write, последняя запись окажется неполной, CRC не сойдётся, и мы её отбросим.

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

/**
 * Разобрать все целые записи. Останавливается на первой оборванной или повреждённой
 * (один последовательный писатель, поэтому рваная запись всегда последняя). validEnd
 * это смещение сразу за последней целой записью, туда пойдут будущие дозаписи.
 */
export function decodeWal(buffer: Buffer): { records: WalRecord[]; validEnd: number } {
  const records: WalRecord[] = [];
  let offset = 0;
  while (offset + WAL_RECORD_HEADER_SIZE <= buffer.byteLength) {
    const length = buffer.readUInt32LE(offset + WalOffset.length);
    if (length > PAGE_SIZE) {
      break;
    }
    const bodyEnd = offset + WAL_RECORD_HEADER_SIZE + length;
    const recordEnd = bodyEnd + CRC_SIZE;
    if (recordEnd > buffer.byteLength) {
      break;
    }
    const body = buffer.subarray(offset, bodyEnd);
    if (crc32(body) >>> 0 !== buffer.readUInt32LE(bodyEnd)) {
      break;
    }
    records.push({
      lsn: buffer.readBigUInt64LE(offset + WalOffset.lsn),
      prevLsn: buffer.readBigUInt64LE(offset + WalOffset.prevLsn),
      xid: buffer.readUInt32LE(offset + WalOffset.xid),
      type: buffer.readUInt16LE(offset + WalOffset.type),
      payload: Buffer.from(buffer.subarray(offset + WAL_RECORD_HEADER_SIZE, bodyEnd)),
    });
    offset = recordEnd;
  }
  return { records, validEnd: offset };
}

decodeWal идёт по буферу и останавливается на первой проблемной записи: длина больше страницы (мусор), запись не помещается до конца файла (оборвали на полуслове), CRC не сошёлся. Поскольку писатель один и пишет только в конец, испорченной может быть лишь последняя запись. validEnd это смещение конца последней целой записи: с него makeWal продолжит писать, затерев огрызок.

Payload-кодеки

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

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

const RELID_SIZE = 4;
const PAGE_ID_SIZE = 4;
const SLOT_SIZE_BYTES = 2;

export function encodeCreatePayload(relationId: number, sql: string): Buffer {
  const head = Buffer.alloc(RELID_SIZE);
  head.writeUInt32LE(relationId, 0);
  return Buffer.concat([head, Buffer.from(sql, 'utf8')]);
}

export function decodeCreatePayload(payload: Buffer): { relationId: number; sql: string } {
  return { relationId: payload.readUInt32LE(0), sql: payload.toString('utf8', RELID_SIZE) };
}

export function encodeInsertPayload(
  relationId: number,
  pageId: number,
  slot: number,
  image: Buffer,
): Buffer {
  const head = Buffer.alloc(RELID_SIZE + PAGE_ID_SIZE + SLOT_SIZE_BYTES);
  head.writeUInt32LE(relationId, 0);
  head.writeUInt32LE(pageId, RELID_SIZE);
  head.writeUInt16LE(slot, RELID_SIZE + PAGE_ID_SIZE);
  return Buffer.concat([head, image]);
}

export function decodeInsertPayload(payload: Buffer): {
  relationId: number;
  pageId: number;
  slot: number;
  image: Buffer;
} {
  const imageStart = RELID_SIZE + PAGE_ID_SIZE + SLOT_SIZE_BYTES;
  return {
    relationId: payload.readUInt32LE(0),
    pageId: payload.readUInt32LE(RELID_SIZE),
    slot: payload.readUInt16LE(RELID_SIZE + PAGE_ID_SIZE),
    image: Buffer.from(payload.subarray(imageStart)),
  };
}

CREATE несёт id отношения и DDL-строку (тот канонический SQL, который печатает prettyPrint из урока 5). INSERT несёт id, адрес (pageId, slot) и сам образ сериализованного кортежа. Образ это физическое логирование: redo просто положит байты обратно.

makeWal: журнал как ресурс под Scope

Вот первая важная Effect-сторона урока. Журнал это ещё один файловый ресурс, такой же, как у DiskManager в уроке 1. Его дескриптор тоже надо взять при открытии и закрыть при выходе из scope. Идиома та же: Effect.acquireRelease плюс scoped-Effect.

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

export interface Wal {
  /** Дописать запись в память, вернуть присвоенный lsn (на диск пока не идёт). */
  readonly append: (xid: number, type: number, payload: Buffer, previousLsn: bigint) => bigint;
  /** Наибольший присвоенный lsn (включая ещё не сброшенные). Синхронно для хуков. */
  readonly currentLsn: () => bigint;
  /** Сбросить накопленные записи на диск и форсировать (fsync). */
  readonly flush: Effect.Effect<void, IoError>;
  /** Правило write-ahead: гарантировать, что журнал durable до указанного lsn. */
  readonly flushUpTo: (lsn: bigint) => Effect.Effect<void, IoError>;
  /** Прочитать все целые записи журнала (для recovery). */
  readonly readAll: Effect.Effect<WalRecord[], IoError>;
}

/** Завести журнал в Scope: дескриптор закроется финалайзером. Продолжает нумерацию lsn. */
export const makeWal = (path: string): Effect.Effect<Wal, IoError, Scope.Scope> =>
  Effect.gen(function* () {
    const handle: FileHandle = yield* Effect.acquireRelease(
      Effect.tryPromise({
        try: () => open(path, constants.O_RDWR | constants.O_CREAT, 0o644),
        catch: (cause) => new IoError({ operation: 'open', cause }),
      }),
      (openHandle) =>
        Effect.promise(async () => {
          try {
            await openHandle.close();
          } catch {
            // best-effort
          }
        }),
    );

    const readWholeFile = Effect.tryPromise({
      try: async () => {
        const stats = await handle.stat();
        const buffer = Buffer.alloc(stats.size);
        if (stats.size > 0) {
          await handle.read(buffer, 0, stats.size, 0);
        }
        return buffer;
      },
      catch: (cause) => new IoError({ operation: 'read', cause }),
    });

    const initial = decodeWal(yield* readWholeFile);
    const last = initial.records.at(-1);
    const maxLsn = last === undefined ? 0n : last.lsn;

    // изменяемое состояние журнала в замыкании
    let writeOffset = initial.validEnd;
    let nextLsn = maxLsn + 1n;
    let flushedLsn = maxLsn;
    let pending: Buffer[] = [];

    const append = (xid: number, type: number, payload: Buffer, previousLsn: bigint): bigint => {
      const lsn = nextLsn;
      nextLsn += 1n;
      pending.push(encodeRecord(lsn, previousLsn, xid, type, payload));
      return lsn;
    };

    const currentLsn = (): bigint => nextLsn - 1n;

    const flush: Effect.Effect<void, IoError> = Effect.tryPromise({
      try: async () => {
        if (pending.length > 0) {
          const batch = Buffer.concat(pending);
          await handle.write(batch, 0, batch.byteLength, writeOffset);
          writeOffset += batch.byteLength;
          pending = [];
        }
        await handle.sync();
        flushedLsn = nextLsn - 1n;
      },
      catch: (cause) => new IoError({ operation: 'wal-flush', cause }),
    });

    const flushUpTo = (lsn: bigint): Effect.Effect<void, IoError> =>
      flushedLsn < lsn ? flush : Effect.void;

    const readAll: Effect.Effect<WalRecord[], IoError> = readWholeFile.pipe(
      Effect.map((buffer) => decodeWal(buffer).records),
    );

    return { append, currentLsn, flush, flushUpTo, readAll };
  });

Тут видна вторая важная Effect-сторона: смешанная подпись. Прочитай интерфейс Wal ещё раз. append это синхронная функция, currentLsn тоже синхронный, а flush, flushUpTo и readAll это Effect. Почему так?

Потому что append ничего на диск не пишет: он добавляет байты в массив pending в памяти. И currentLsn это просто чтение поля. Это две операции, которые мы будем звать из синхронных хуков: куча после успешной вставки штампует page.lsn = journal.logInsert(...) прямо внутри своего синхронного pinned.page.insertTuple, а буферный пул в canFlush сверяет page.lsn против flushableLsn тоже синхронно. Если бы append был Effect, его пришлось бы yield-ить, а это значило бы тащить Effect.gen глубоко внутрь чистой работы со страницей. Поэтому идиома: методы, трогающие только память, синхронные; методы, ходящие на диск, это Effect. Граница не “вообще всё в Effect”, граница ровно по ресурсам.

makeWal сама на старте читает существующий файл (если был), берёт validEnd для дозаписи и продолжает нумерацию lsn с последней записи: журнал переживает перезапуски, нумерация не сбрасывается. flush сбрасывает накопленное одним write, потом fsync через handle.sync(), потом обновляет flushedLsn. flushUpTo(lsn) это write-ahead в одну строку: если просят durable до некоторого lsn, а мы туда ещё не дотянулись, форсим.

И заметь: Effect.acquireRelease плюс закрытие финалайзером дают то же свойство, что у DiskManager в уроке 1. Когда makeDatabase завершится (scope закроется), дескриптор журнала закроется автоматически, без wal.close(). Это и есть третья грань Scope в нашем движке: рядом с файлами куч и индексов появляется ещё один файл, и он закрывается тем же механизмом.

Журналируем вставку: Journal и AutocommitJournal

Слой кучи должен описывать каждую вставку в журнале. Чтобы куча не зависела от конкретного журнала, опишем узкий контракт.

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

/**
 * Контракт журналирования для слоя данных. Сегодня умеет одно: записать вставку и
 * вернуть её lsn (куча штампует им page.lsn). В следующем уроке этот интерфейс расширит
 * менеджер транзакций (добавит логирование удаления и откат).
 */
export interface Journal {
  logInsert(xid: number, relationId: number, pageId: number, slot: number, image: Buffer): bigint;
}

/** Минимальный autocommit-журнал: каждая вставка это запись WAL, prev_lsn = 0 (без отката). */
export class AutocommitJournal implements Journal {
  constructor(private readonly wal: Wal) {}

  logInsert(xid: number, relationId: number, pageId: number, slot: number, image: Buffer): bigint {
    return this.wal.append(
      xid,
      WalRecordType.insert,
      encodeInsertPayload(relationId, pageId, slot, image),
      0n,
    );
  }
}

Journal это интерфейс, а не сервис: куче нужна способность “записать вставку и вернуть lsn”, а кто и как делает, ей всё равно. Метод синхронный, потому что wal.append синхронный (запись копится в памяти, на диск пока не идёт). Сегодня единственная реализация это AutocommitJournal: каждая вставка дописывает INSERT-запись с prev_lsn = 0 и xid = ENGINE_XID. Цепочка записей и откат нужны транзакциям, поэтому появятся в следующем уроке, когда журнал возьмёт под крыло менеджер транзакций (он тоже реализует Journal, но умнее).

Наращиваем Page: insertTupleAt для адресного redo

При восстановлении redo должен положить кортеж обратно ровно в тот слот, что записан в журнале, а не “куда влезет”. У страницы из урока 1 был insertTuple, который сам выбирает следующий слот, но redo нужен адресный вариант. Добавь в src/page.ts.

// src/page.ts (добавляем метод в класс Page)

/**
 * Положить кортеж в конкретный слот по логическому номеру (используется при redo
 * восстановления). В отличие от insertTuple адресует слот напрямую: байты ложатся у
 * upper, slot указывает на них, lower поднимается до покрытия slot+1. Пропущенные
 * промежуточные слоты остаются нулями (мёртвыми), поэтому пропуск записи при redo не
 * сдвигает номера соседних слотов.
 */
insertTupleAt(slot: number, data: Buffer): void {
  const tupleOffset = this.upper - data.byteLength;
  data.copy(this.buffer, tupleOffset);
  this.upper = tupleOffset;
  this.setSlot(slot, tupleOffset);
  const neededLower = HEADER_SIZE + (slot + 1) * SLOT_SIZE;
  if (neededLower > this.lower) {
    this.lower = neededLower;
  }
}

Разница с insertTuple в том, что номер слота задан снаружи. Это нужно для верности redo: если между двумя вставками в журнале есть пропуск (в следующем уроке так будет с откатанной вставкой), redo обязан сохранить исходные номера слотов, иначе RID сдвинутся и индекс начнёт врать. insertTupleAt поднимает lower ровно настолько, чтобы покрыть указанный слот, оставляя пропущенные слоты мёртвыми.