Своя СУБД на Effect · урок 8: транзакции, откат и видимость

Кому это. В прошлом уроке у движка появился журнал упреждающей записи. INSERT и CREATE переживают падение: старт проигрывает журнал. Но мы честно оставили дыру: настоящих транзакций нет, ROLLBACK движок не знает, а DELETE работает на честном слове и падение не переживает.

Зачем именно Effect. Сегодня закрываем дыру целиком. Появятся транзакции (BEGIN/COMMIT/ROLLBACK), кортеж научится версиям через xmin/xmax, и видимость строки начнёт решать движок. Это MVCC, на которой стоит Postgres. И тут Effect снова покажет ту же идиому, что в журнале урока 7: смешанный API. TransactionManager будет реализовывать Journal-интерфейс, и часть его методов синхронные (begin, logInsert, logSetXmax, flushableLsn), а часть это Effect (commit и abort, потому что они зовут wal.flush и applyUndo через IO). Граница та же: память синхронно, диск через Effect. Плюс applyUndo это callback-Effect в abort, типичный приём, когда менеджеру нужно “что-то сделать с ресурсом, но что именно решит вызывающий”. И всё это под тем же Scope, что WAL и каталог: добавление транзакций ничего не меняет в управлении ресурсами, только в логике.

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

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

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

Два новых слоя: менеджер транзакций (выдаёт xid, ведёт состояния, undo-лог) и MVCC (видимость по снимку). Они пронизывают исполнитель: каждый скан фильтрует версии по видимости, удаление пишет версию вместо стирания.

Идея версий

Главная мысль в одном абзаце. У кортежа из урока 3 уже есть два поля заголовка, которые до сих пор ничего не значили: xmin и xmax. xmin это xid транзакции, создавшей версию. xmax это xid транзакции, удалившей версию, или 0 если версия жива. Удаление больше не стирает слот: оно ставит xmax. Обновление это удаление старой версии плюс вставка новой. А видна ли строка конкретной транзакции, решает функция видимости. Так одно удаление становится журналируемой пометкой, которую можно воспроизвести при recovery и откатить при rollback. Дыра прошлого урока закрывается ровно этим.

Версии в странице: setXmax

Сначала научим страницу читать и писать xmax кортежа на месте. В уроке 1 у Page были аксессоры заголовка страницы; добавим пару для поля xmax кортежа в слоте. Чистый код, без Effect.

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

/**
 * Записать xmax кортежа в слоте на месте (MVCC-удаление). Слот остаётся живым;
 * видимость скрывает версию после фиксации удалившей транзакции. Заголовок кортежа
 * лежит по смещению слота, xmax это u32 LE по TupleOffset.xmax.
 */
setXmax(slot: number, xmax: number): void {
  this.buffer.writeUInt32LE(xmax >>> 0, this.getSlot(slot) + TupleOffset.xmax);
}

/** Прочитать xmax кортежа в слоте. */
getXmax(slot: number): number {
  return this.buffer.readUInt32LE(this.getSlot(slot) + TupleOffset.xmax);
}

Важно: setXmax не трогает сам слот, только заголовок кортежа по его смещению. Слот остаётся живым (isSlotLive по-прежнему true), поэтому seq scan кортеж увидит, и дальше уже видимость решит, показывать ли его. Это принципиальное отличие от deleteTuple из урока 3: тот зануляет слот (физическое удаление), а setXmax оставляет версию на месте, лишь помечая её удалённой.

Видимость: снимок и правило

Теперь сердце MVCC. Положи в src/snapshot.ts. Чистая логика без ввода-вывода, поэтому совпадает с TS-треком.

// src/snapshot.ts
import type { Tuple, Xid } from '~/tuple.ts';

export type TxnState = 'in-progress' | 'committed' | 'aborted';

/**
 * Реестр статусов транзакций для проверки видимости. Незнакомый xid считается откатанным.
 * floor это водяной знак: ниже него xid считается зафиксированным без явной записи (после
 * checkpoint урока 7, аналог обрезания CLOG в Postgres).
 */
export class CommitStatus {
  private readonly states = new Map<Xid, TxnState>();
  private floorXid: Xid = 0;

  constructor(entries: Iterable<readonly [Xid, TxnState]> = []) {
    for (const [xid, state] of entries) {
      this.states.set(xid, state);
    }
  }

  /** Статус транзакции. Явная запись приоритетна; всё ниже floor это committed. */
  get(xid: Xid): TxnState {
    const explicit = this.states.get(xid);
    if (explicit !== undefined) {
      return explicit;
    }
    return xid !== 0 && xid < this.floorXid ? 'committed' : 'aborted';
  }

  set(xid: Xid, state: TxnState): void {
    this.states.set(xid, state);
  }

  get floor(): Xid {
    return this.floorXid;
  }

  /** Поднять floor (после checkpoint или recovery): xid ниже считаются committed. */
  raiseFloorTo(watermark: Xid): void {
    if (watermark > this.floorXid) {
      this.floorXid = watermark;
    }
  }

  /** Перенять floor и явные статусы из другого реестра (после recovery). */
  absorb(other: CommitStatus): void {
    this.raiseFloorTo(other.floor);
    for (const [xid, state] of other.entries()) {
      this.states.set(xid, state);
    }
  }

  entries(): Iterable<readonly [Xid, TxnState]> {
    return this.states.entries();
  }
}

CommitStatus отвечает на вопрос “зафиксирована ли транзакция xid”. Обычно по явной записи; незнакомый xid считается откатанным консервативно. Тонкость это floor: после checkpoint старые commit-записи журнала усекаются, и тогда любой xid ниже водяного знака считается зафиксированным без явной записи. Это спасает от бесконечного роста реестра, как обрезание CLOG в Postgres. Дальше сам снимок и видимость.

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

export interface Snapshot {
  /** Самый старый активный xid (понадобится vacuum в уроке 10). */
  readonly xmin: Xid;
  /** Верхняя граница (эксклюзивно): xid >= xmax ещё не начинались на момент снимка. */
  readonly xmax: Xid;
  /** Активные (незафиксированные) транзакции на момент снимка, кроме самой себя. */
  readonly active: ReadonlySet<Xid>;
}

/** Зафиксирована ли транзакция x к моменту снимка (committed-as-of-snapshot). */
export function committedAsOf(x: Xid, snapshot: Snapshot, status: CommitStatus): boolean {
  return x !== 0 && status.get(x) === 'committed' && x < snapshot.xmax && !snapshot.active.has(x);
}

/**
 * Видна ли версия кортежа транзакции self под её снимком. Собственные записи (xmin или
 * xmax === self) транзакция видит сразу. Иначе смотрим на зафиксированность к снимку.
 */
export function isVisible(
  tuple: Tuple,
  snapshot: Snapshot,
  self: Xid,
  status: CommitStatus,
): boolean {
  const inserterVisible = tuple.xmin === self || committedAsOf(tuple.xmin, snapshot, status);
  if (!inserterVisible) {
    return false;
  }
  if (tuple.xmax === 0) {
    return true;
  }
  const deleted = tuple.xmax === self || committedAsOf(tuple.xmax, snapshot, status);
  return !deleted;
}

Снимок это три числа. Сегодня active всегда пуст: транзакция одна. В уроке про конкуренцию он наполнится.

isVisible это правило в пять строк, и его стоит прочитать медленно. Версия видна, если выполнены два условия. Первое: её создатель виден, то есть либо это моя собственная вставка (xmin === self), либо создатель зафиксировался к моему снимку. Если создатель не виден (его транзакция откатилась или ещё не зафиксировалась), версия невидима сразу. Второе: версия не удалена видимым образом. xmax === 0 значит жива. Иначе версия удалена, если удаливший это я сам (xmax === self) или он зафиксировался к снимку. committedAsOf собирает воедино: транзакция зафиксирована к снимку, если её статус committed, она началась до снимка (x < xmax) и не была активной на момент снимка.

Покрути это вживую. Виджет показывает одну транзакцию на таймлайне: BEGIN, INSERT, DELETE, и три исхода: COMMIT, ROLLBACK, падение. Видно, как xmin, xmax и статус транзакции вместе решают, видна ли строка в каждом исходе.

Менеджер транзакций

Теперь тот, кто выдаёт xid, ведёт состояния, цепочку записей журнала и undo-лог. И тут Effect-порт делает осознанный выбор подписей. Эталон Journal теперь требует второй метод, logSetXmax, чтобы куча могла журналировать MVCC-удаление. Переезжаем интерфейс в transaction.ts (с прошлого урока он жил в wal.ts рядом с AutocommitJournal, но теперь его место рядом с TransactionManager).

// src/transaction.ts
import { Effect } from 'effect';

import { WalRecordType } from '~/constants.ts';
import { EngineError } from '~/engine/errors.ts';
import type { BufferPoolError, IoError } from '~/errors.ts';
import { CommitStatus, type Snapshot } from '~/snapshot.ts';
import type { Xid } from '~/tuple.ts';
import { encodeInsertPayload, encodeSetXmaxPayload, type Wal } from '~/wal.ts';

/**
 * Канал логирования для слоя кучи. Куча зовёт эти методы из-под закреплённой страницы и
 * штампует возвращённый lsn в page.lsn. Синхронные: запись копится в pending WAL, на
 * диск пока не идёт; на диск форсит commit. Если бы это был Effect, его пришлось бы
 * yield-ить из чистой работы со страницей, и каждая вставка тянула бы Effect.gen в
 * BTreeNode.leafInsertAt по соседству. Тот же приём, что у wal.append из урока 7.
 */
export interface Journal {
  logInsert(xid: Xid, relationId: number, pageId: number, slot: number, image: Buffer): bigint;
  logSetXmax(xid: Xid, relationId: number, pageId: number, slot: number, xmax: Xid): bigint;
}

/** Undo вставки: пометить слот мёртвым. */
export interface InsertUndo {
  readonly kind: 'insert';
  readonly relationId: number;
  readonly pageId: number;
  readonly slot: number;
}

/** Undo MVCC-удаления: вернуть xmax кортежа в 0. */
export interface SetXmaxUndo {
  readonly kind: 'setXmax';
  readonly relationId: number;
  readonly pageId: number;
  readonly slot: number;
}

export type UndoEntry = InsertUndo | SetXmaxUndo;

/**
 * Применить одну undo-запись (откат): операция движка над страницей. Effect-callback,
 * потому что pool.withPin это acquireRelease и applyUndo внутри abort идёт через IO.
 */
export type ApplyUndo = (entry: UndoEntry) => Effect.Effect<void, IoError | BufferPoolError>;

interface ActiveTransaction {
  readonly xid: Xid;
  readonly beginLsn: bigint;
  readonly snapshot: Snapshot;
  prevLsn: bigint;
  readonly undoLog: UndoEntry[];
}

const EMPTY_PAYLOAD = Buffer.alloc(0);

export interface TransactionManager extends Journal {
  readonly commitStatus: CommitStatus;
  readonly watermark: Xid;
  readonly openXids: () => Xid[];
  readonly snapshotFor: (xid: Xid) => Snapshot;
  readonly readSnapshot: () => { xid: Xid; snapshot: Snapshot };
  readonly resumeFrom: (nextXid: number, recovered: CommitStatus) => void;
  readonly begin: () => Xid;
  readonly commit: (xid: Xid) => Effect.Effect<void, IoError>;
  readonly abort: (
    xid: Xid,
    applyUndo: ApplyUndo,
  ) => Effect.Effect<void, IoError | BufferPoolError>;
  readonly flushableLsn: () => bigint;
}

Прочитай интерфейс. Это идиома порта в чистом виде: begin/logInsert/logSetXmax/snapshotFor/readSnapshot/flushableLsn/openXids/watermark синхронные (пишут в Map в замыкании и в pending WAL; их зовёт куча из-под закреплённой страницы и canFlush-хук пула), commit/abort это Effect (потому что commit делает wal.flush через IO, а abort прогоняет цепочку applyUndo через pool.withPin). Граница как у журнала: память синхронно, диск через Effect. Дальше сам менеджер.

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

/** Завести менеджер транзакций над журналом. Чистая конструкция, без ввода-вывода. */
export const makeTransactionManager = (wal: Wal): TransactionManager => {
  let nextXid = 1;
  const active = new Map<Xid, ActiveTransaction>();
  const status = new CommitStatus();

  const require = (xid: Xid): ActiveTransaction => {
    const txn = active.get(xid);
    if (txn === undefined) {
      throw new EngineError({ message: `нет открытой транзакции ${xid}` });
    }
    return txn;
  };

  /** Снимок: активны все открытые транзакции кроме самой себя; xmax это следующий xid. */
  const captureSnapshot = (self: Xid): Snapshot => {
    const activeXids = new Set<Xid>(active.keys());
    activeXids.delete(self);
    let xmin = self;
    for (const xid of activeXids) {
      if (xid < xmin) {
        xmin = xid;
      }
    }
    return { xmin, xmax: self, active: activeXids };
  };

  const begin = (): Xid => {
    const xid = nextXid;
    nextXid += 1;
    const beginLsn = wal.append(xid, WalRecordType.begin, EMPTY_PAYLOAD, 0n);
    const snapshot = captureSnapshot(xid);
    active.set(xid, { xid, beginLsn, snapshot, prevLsn: beginLsn, undoLog: [] });
    status.set(xid, 'in-progress');
    return xid;
  };

  const commit = (xid: Xid): Effect.Effect<void, IoError> =>
    Effect.gen(function* () {
      const txn = require(xid);
      wal.append(xid, WalRecordType.commit, EMPTY_PAYLOAD, txn.prevLsn);
      yield* wal.flush; // durability: COMMIT durable на диске до возврата
      status.set(xid, 'committed');
      active.delete(xid);
    });

  const abort = (xid: Xid, applyUndo: ApplyUndo): Effect.Effect<void, IoError | BufferPoolError> =>
    Effect.gen(function* () {
      const txn = require(xid);
      // undo в обратном порядке: каждый шаг это Effect, поэтому forEach по массиву
      for (let index = txn.undoLog.length - 1; index >= 0; index--) {
        yield* applyUndo(txn.undoLog[index]!);
      }
      wal.append(xid, WalRecordType.abort, EMPTY_PAYLOAD, txn.prevLsn);
      status.set(xid, 'aborted');
      active.delete(xid);
    });

  const logInsert: Journal['logInsert'] = (xid, relationId, pageId, slot, image) => {
    const txn = require(xid);
    const lsn = wal.append(
      xid,
      WalRecordType.insert,
      encodeInsertPayload(relationId, pageId, slot, image),
      txn.prevLsn,
    );
    txn.prevLsn = lsn;
    txn.undoLog.push({ kind: 'insert', relationId, pageId, slot });
    return lsn;
  };

  const logSetXmax: Journal['logSetXmax'] = (xid, relationId, pageId, slot, xmax) => {
    const txn = require(xid);
    const lsn = wal.append(
      xid,
      WalRecordType.setXmax,
      encodeSetXmaxPayload(relationId, pageId, slot, xmax),
      txn.prevLsn,
    );
    txn.prevLsn = lsn;
    txn.undoLog.push({ kind: 'setXmax', relationId, pageId, slot });
    return lsn;
  };

  /**
   * Граница no-steal: минимум (beginLsn - 1) по всем открытым транзакциям, иначе текущий
   * максимум. Страница с lsn выше границы тронута незафиксированной транзакцией и на диск
   * не сбрасывается.
   */
  const flushableLsn = (): bigint => {
    if (active.size === 0) {
      return wal.currentLsn();
    }
    let min: bigint | undefined;
    for (const txn of active.values()) {
      const candidate = txn.beginLsn - 1n;
      if (min === undefined || candidate < min) {
        min = candidate;
      }
    }
    return min!;
  };

  return {
    commitStatus: status,
    get watermark(): Xid {
      return nextXid;
    },
    openXids: () => [...active.keys()],
    snapshotFor: (xid) => require(xid).snapshot,
    readSnapshot: () => ({ xid: nextXid, snapshot: captureSnapshot(nextXid) }),
    resumeFrom: (resumeXid, recovered) => {
      nextXid = Math.max(nextXid, resumeXid);
      status.absorb(recovered);
    },
    begin,
    commit,
    abort,
    logInsert,
    logSetXmax,
    flushableLsn,
  };
};

Это много кода, но он распадается на ясные части. begin присваивает xid, синхронно пишет в журнал запись BEGIN (через wal.append, которое синхронно копит в pending), снимает снимок и заводит активную транзакцию с пустым undo-логом. commit пишет COMMIT и через yield* wal.flush форсит журнал: вот она, долговечность фиксации, после commit изменения переживут падение. abort идёт по undo-логу в обратном порядке через yield* applyUndo(entry): каждый шаг это Effect (внутри он дёргает pool.withPin), поэтому весь цикл идёт под Effect.gen.

logInsert и logSetXmax это реализация Journal: куча зовёт их синхронно, изнутри своего Effect.scoped под закреплённой страницей. Каждая дописывает запись в журнал, продляет цепочку prev_lsn и кладёт запись в undo-лог. flushableLsn это граница no-steal, к ней вернёмся в Database.

И makeTransactionManager это значение, а не сервис: одна база, один менеджер, состояние в замыкании. Никакого Scope: ресурсов менеджер не держит, только память. WAL под Scope, потому что у него файловый дескриптор; менеджер транзакций нет.

Заметь маленькую находку: commitStatus это поле, а не геттер. Это удобно: проверка видимости (isVisible) принимает CommitStatus, и тот же объект мутируется внутри менеджера через status.set. Снаружи он один и тот же, и операторы исполнителя видят актуальное состояние без перечитывания.

Heap +markDeleted: MVCC-удаление

Теперь куча умеет удалять через версию. Добавь в src/heap-file.ts метод рядом с insertTuple и deleteTuple.

// src/heap-file.ts (добавляем markDeleted и расширяем journal до Journal с двумя методами)
import type { Journal } from '~/transaction.ts'; // переезжаем импорт из wal.ts

// в интерфейсе HeapFile добавь:
//   readonly markDeleted: (rid: Rid, xid: Xid) => Effect.Effect<void, IoError | BufferPoolError>;

const markDeleted = (rid: Rid, xid: Xid): Effect.Effect<void, IoError | BufferPoolError> =>
  Effect.scoped(
    Effect.gen(function* () {
      const pinned = yield* pool.withPin(rid.pageId);
      // защита примитива: на мёртвом слоте getSlot вернёт DEAD_SLOT (0), и setXmax
      // записал бы xmax в заголовок страницы. Исполнитель удаляет только видимые
      // строки, сюда мёртвый слот не доходит, но напрямую портить заголовок нельзя.
      if (!pinned.page.isSlotLive(rid.slot)) {
        return yield* Effect.die(new Error(`markDeleted: слот ${rid.pageId}:${rid.slot} уже мёртв`));
      }
      pinned.page.setXmax(rid.slot, xid);
      if (journal !== undefined) {
        pinned.page.lsn = journal.logSetXmax(xid, relationId, rid.pageId, rid.slot, xid);
      }
      yield* pinned.markDirty;
    }),
  );

Симметрично штампу pinned.page.lsn = journal.logInsert(...) из урока 7: меняем кортеж на странице через setXmax, синхронно журналируем через logSetXmax и сразу штампуем page.lsn, потом markDirty. Всё под одним Effect.scoped: pin снимется финалайзером по выходу. Старый deleteTuple (физическое стирание из урока 3) остаётся как примитив, но движок его больше не зовёт.