Своя СУБД на Effect · урок 9: конкуренция, блокировки и дедлоки

Кому это. В прошлом уроке у движка появились транзакции и MVCC-версии. DELETE стал журналируемой пометкой xmax, видимость считает isVisible по снимку, COMMIT форсит журнал, ROLLBACK идёт по undo-логу. Но движок был честно однопользовательский: одна транзакция за раз, плюс autocommit. Снимок мы заложили правильно, но множество active всегда оставалось пустым: других открытых транзакций просто не существовало.

Зачем именно Effect. Сегодня даём транзакциям сосуществовать. Появится db.begin (теперь это Effect<Transaction>), который открывает явную транзакцию и возвращает хэндл; несколько таких хэндлов могут чередовать операторы. Тогда заработает snapshot isolation: транзакция A видит стабильный снимок, пока B рядом коммитит. Чтобы запись осталась корректной, добавим менеджер блокировок с эксклюзивным локом строки, проверку write-write конфликтов и детект дедлоков через граф ожидания.

Тут Effect снова решает задачу, для которой он создан. Транзакционная память и TxRef это учебниковый инструмент для менеджера блокировок: владельцы, держащиеся ресурсы и граф ожидания живут в трёх TxRef, а acquire и release это атомарные блоки Effect.tx, которые читают и пишут эти TxRef без явных локов. Дальше будет важная тонкость про то, как падение внутри транзакции мешает накопить граф ожидания, и про ловушку типов в Effect.flatMap со switch по дискриминанту. Это и есть основной Effect-нарратив урока: транзакционная память как инструмент, а граница транзакции как место выбора между атомарностью и типизированной ошибкой.

Важно про объём. Почти весь фундамент готов с прошлого урока: версии, видимость, снимок, менеджер транзакций. Сегодня мы не переписываем их, а снимаем ограничение и достраиваем сбоку: новый файл lock-manager.ts, пара методов в операторах, хэндл транзакции в Database. Среда у нас по-прежнему однофайберная: настоящего параллелизма нет, тесты сами чередуют операторы на одном файбере. Поэтому стратегия fail-fast: LockWaitError валит только текущий statement, транзакция остаётся открытой, её ребро ожидания живёт в графе для будущего поиска цикла. В настоящем многофайберном движке acquire блокировал бы файбер через Effect.txRetry, но логика конфликтов от этого не меняется.

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

┌────────────────────────────────────────────────┐
│  Бинарный протокол + 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
└────────────────────────────────────────────────┘

Один новый слой между транзакциями и MVCC: менеджер блокировок. Он не трогает чтение (оно идёт по снимку без локов), но встаёт на пути записи: перед тем как пометить строку удалённой или вставить новую версию, оператор берёт эксклюзивный лок строки.

Две стороны конкуренции

У безопасной конкуренции две стороны, и важно их не путать.

Первая это чтение. Под MVCC чтение вообще не берёт локов. Каждая транзакция работает над снимком, взятым в момент begin, и видит ту версию строки, что была актуальна на тот момент. Параллельная транзакция может вставлять, менять и удалять строки, коммитить их, а наш читатель этого не заметит: его снимок зафиксирован. Это и есть “читатели не блокируют писателей, писатели не блокируют читателей”. Вся механика для этого уже написана в прошлом уроке, нам осталось её включить.

Вторая это запись. Тут MVCC недостаточно: если две транзакции меняют одну строку, кто-то должен проиграть, иначе одно изменение потеряется (lost update). Решение в два слоя. Эксклюзивный лок строки сериализует одновременных писателей: пока A держит строку, B на ней ждёт. А write-write конфликт (строку уже изменила транзакция, зафиксировавшаяся после моего снимка) ловится проверкой xmax и откатывается. Вместе это даёт корректную запись поверх snapshot isolation.

Снимок наконец наполняется

Начнём с приятного: чтобы заработала изоляция снимков, в коде менеджера транзакций менять нечего. Вспомни captureSnapshot из прошлого урока:

// src/transaction.ts (этот код уже написан в уроке 8, читаем заново)

  /** Снимок: активны все открытые транзакции кроме самой себя; 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 };
  };

Он всегда собирал active из всех открытых транзакций. В прошлом уроке открытая транзакция была одна (она сама), поэтому после activeXids.delete(self) множество оставалось пустым. Стоит открыть две транзакции одновременно, и снимок второй из них честно запишет первую в active. Дальше committedAsOf уже умеет это учитывать:

// src/snapshot.ts (тоже из урока 8, ключевая строка)

export function committedAsOf(x: Xid, snapshot: Snapshot, status: CommitStatus): boolean {
  return x !== 0 && status.get(x) === 'committed' && x < snapshot.xmax && !snapshot.active.has(x);
}

!snapshot.active.has(x) это и есть изоляция: даже если транзакция x уже зафиксировалась, но на момент моего снимка она была активной, я считаю её невидимой. Её изменения не пролезут в мой согласованный снимок. Так A, начавшая раньше B, не увидит то, что B вставила и закоммитила позже: если A взяла снимок до старта B, то B вообще за границей xmax; если после, то B попала в active. Вся изоляция снимков это одна строчка, написанная заранее. Нам осталось дать транзакциям сосуществовать.

Менеджер блокировок на транзакционной памяти

Теперь сторона записи. Заведи новый файл src/lock-manager.ts. Это сердце сегодняшнего урока, и он на удивление компактный. Состояние держим в трёх TxRef:

// src/lock-manager.ts
import { Effect, TxRef } from 'effect';

import { DeadlockError, LockWaitError } from '~/engine/errors.ts';
import type { Xid } from '~/tuple.ts';

export interface LockManager {
  /** Взять эксклюзивный лок на ресурс. Повторный захват своим владельцем это no-op. */
  readonly acquire: (
    xid: Xid,
    resource: string,
  ) => Effect.Effect<void, LockWaitError | DeadlockError>;
  /** Освободить все локи и рёбра ожидания транзакции (на commit/abort). */
  readonly release: (xid: Xid) => Effect.Effect<void>;
}

type OwnerMap = ReadonlyMap<string, Xid>;
type HeldMap = ReadonlyMap<Xid, ReadonlySet<string>>;
type WaitsForMap = ReadonlyMap<Xid, ReadonlySet<Xid>>;

type AcquireResult = 'acquired' | 'wait' | 'deadlock';

export const makeLockManager: Effect.Effect<LockManager> = Effect.gen(function* () {
  const ownersRef = yield* TxRef.make<OwnerMap>(new Map());
  const heldRef = yield* TxRef.make<HeldMap>(new Map());
  const waitsForRef = yield* TxRef.make<WaitsForMap>(new Map());

Три структуры данных. ownersRef отвечает на вопрос “кто держит этот ресурс”. heldRef это обратный индекс “какими ресурсами владеет транзакция”, чтобы release отпустил всё разом одной транзакцией. waitsForRef это граф ожидания: ребро A -> B значит “A ждёт ресурс, занятый B”. Все три карты иммутабельные ReadonlyMap, и каждое обновление пишет в TxRef новую копию (copy-on-write). Это удобнее, чем мутировать на месте: транзакция видит чистые значения и сама управляет атомарностью.

Главная функция всего файла это acquireTx. Она возвращает не void, а дискриминант исхода, и сразу станет ясно, почему.

// src/lock-manager.ts (продолжение тела makeLockManager)

  const acquireTx = (xid: Xid, resource: string): Effect.Effect<AcquireResult> =>
    Effect.tx(
      Effect.gen(function* () {
        const owners = yield* TxRef.get(ownersRef);
        const owner = owners.get(resource);
        if (owner === undefined || owner === xid) {
          yield* TxRef.set(ownersRef, new Map(owners).set(resource, xid));
          yield* TxRef.update(heldRef, (held) => addHeld(held, xid, resource));
          return 'acquired';
        }
        const updated = yield* TxRef.modify(waitsForRef, (graph) => {
          const next = addEdge(graph, xid, owner);
          return [next, next];
        });
        if (hasCycle(updated, xid)) {
          yield* TxRef.update(waitsForRef, (graph) => removeEdge(graph, xid, owner));
          return 'deadlock';
        }
        return 'wait';
      }),
    );

Прочитай медленно. Внутри Effect.tx живёт обычный Effect.gen: это и есть тело транзакции, а Effect.tx только задаёт её границу. Свободный ресурс или свой же повторный захват: ставим себя владельцем, дописываем heldRef, возвращаем 'acquired'. Чужой ресурс это конфликт: добавляем ребро ожидания xid -> owner и сразу проверяем граф. TxRef.modify берёт функцию, которая возвращает пару “новое состояние ячейки, результат вызова”, поэтому мы возвращаем [next, next] и получаем обновлённую карту, чтобы hasCycle смотрел уже на новое состояние. Цикл значит дедлок: убираем ребро, чтобы граф не врал после отката, возвращаем 'deadlock'. Иначе ребро остаётся жить в графе, и мы возвращаем 'wait'.

Тут пора остановиться и проговорить главную тонкость порта: почему acquireTx не падает через Effect.fail внутри транзакции, а возвращает дискриминант. Транзакция атомарна: либо все её записи в TxRef применились, либо ни одна. Если упасть внутри Effect.tx, рантайм откатит все записи этой транзакции, включая TxRef.update(waitsForRef, ...) с добавленным ребром. Но это ребро должно пережить неуспешный acquire: его увидит будущий acquire другой транзакции, который замкнёт цикл. Иначе говоря: ребро это не временная переменная, это часть состояния менеджера. Если мы откатим запись, будущая B не найдёт дедлока, и две транзакции зависнут в “ожидании”, которого никто не разрешит.

Решение разводит ответственность: транзакция держит атомарность над состоянием, типизированная ошибка живёт снаружи, после коммита.

// src/lock-manager.ts (граница транзакции)

  const acquire = (
    xid: Xid,
    resource: string,
  ): Effect.Effect<void, LockWaitError | DeadlockError> =>
    acquireTx(xid, resource).pipe(
      Effect.flatMap((result): Effect.Effect<void, LockWaitError | DeadlockError> => {
        switch (result) {
          case 'acquired': {
            return Effect.void;
          }
          case 'deadlock': {
            return Effect.fail(
              new DeadlockError({ message: `deadlock: транзакция ${xid} откатывается` }),
            );
          }
          case 'wait': {
            return Effect.fail(new LockWaitError({ message: `ресурс ${resource} занят` }));
          }
        }
      }),
    );

acquireTx уже отдаёт обычный Effect: граница Effect.tx закрылась, транзакция зафиксирована. Поэтому снаружи мы спокойно роняем нужную ошибку через Effect.fail: запись ребра уже применилась, ничего не откатится.

И ещё одна ловушка типов, на которую легко наступить. Effect.flatMap сужает тип ошибки по callback’у: TS видит switch со ветками, дающими разные E (never для 'acquired', DeadlockError для 'deadlock', LockWaitError для 'wait'), и пытается их объединить. Без явной аннотации возврата TS может либо взять только одну ветку (если она первая), либо сузить тип до конкретного варианта. Явная аннотация (result): Effect.Effect<void, LockWaitError | DeadlockError> => фиксирует объединение в одном месте. Это та же история, что с Effect.result в runAutocommitDml урока 8: на сшивке веток TS нужна подсказка, иначе он схлопнет тип.

Теперь release и поиск цикла. Освобождение тоже идёт транзакцией, потому что трогает все три TxRef сразу.

// src/lock-manager.ts (release плюс приватные операции)

  const release = (xid: Xid): Effect.Effect<void> =>
    Effect.tx(
      Effect.gen(function* () {
        const held = yield* TxRef.get(heldRef);
        const resources = held.get(xid);
        if (resources !== undefined) {
          yield* TxRef.update(ownersRef, (owners) => {
            const next = new Map(owners);
            for (const resource of resources) {
              if (next.get(resource) === xid) {
                next.delete(resource);
              }
            }
            return next;
          });
          yield* TxRef.update(heldRef, (current) => {
            const next = new Map(current);
            next.delete(xid);
            return next;
          });
        }
        yield* TxRef.update(waitsForRef, (graph) => dropWaiter(graph, xid));
      }),
    );

  return { acquire, release };
});

// -------- чистые операции над снимками карт (copy-on-write) --------

function addHeld(held: HeldMap, xid: Xid, resource: string): HeldMap {
  const next = new Map(held);
  const resources = new Set(next.get(xid));
  resources.add(resource);
  next.set(xid, resources);
  return next;
}

function addEdge(graph: WaitsForMap, waiter: Xid, holder: Xid): WaitsForMap {
  const next = new Map(graph);
  const targets = new Set(next.get(waiter));
  targets.add(holder);
  next.set(waiter, targets);
  return next;
}

function removeEdge(graph: WaitsForMap, waiter: Xid, holder: Xid): WaitsForMap {
  const next = new Map(graph);
  const targets = new Set(next.get(waiter));
  targets.delete(holder);
  if (targets.size === 0) {
    next.delete(waiter);
  } else {
    next.set(waiter, targets);
  }
  return next;
}

/** Удалить транзакцию из графа: и её рёбра, и входящие рёбра к ней. */
function dropWaiter(graph: WaitsForMap, xid: Xid): WaitsForMap {
  const next = new Map(graph);
  next.delete(xid);
  for (const [waiter, targets] of next) {
    if (targets.has(xid)) {
      const trimmed = new Set(targets);
      trimmed.delete(xid);
      next.set(waiter, trimmed);
    }
  }
  return next;
}

/** Есть ли цикл в графе ожидания, достижимый из start (поиск в глубину). */
function hasCycle(graph: WaitsForMap, start: Xid): boolean {
  const visited = new Set<Xid>();
  const stack: Xid[] = [start];
  while (stack.length > 0) {
    const node = stack.pop()!;
    const targets = graph.get(node);
    if (targets === undefined) {
      continue;
    }
    for (const next of targets) {
      if (next === start) {
        return true;
      }
      if (!visited.has(next)) {
        visited.add(next);
        stack.push(next);
      }
    }
  }
  return false;
}

release забирает все ресурсы транзакции из ownersRef, выкидывает её из heldRef и из waitsForRef (и как ожидающего, и как того, кого ждали). Всё это одной транзакцией, поэтому другой acquire не успеет увидеть полу-освобождённое состояние. hasCycle это поиск в глубину по WaitsForMap, чистая функция над снимком графа. Чистые операции addHeld/addEdge/removeEdge/dropWaiter копируют Map и Set целиком: это и есть copy-on-write, благодаря которому транзакция видит только иммутабельные значения.

Здесь стоит сказать ещё про одну сторону однопоточной модели. В многопоточном движке acquireTx на занятом ресурсе вернул бы 'wait', и мы бы дальше сделали Effect.txRetry вместо Effect.fail(LockWaitError). Effect.txRetry это главный трюк транзакционной памяти: транзакция подпишется на все TxRef, что она прочитала, и переиграется, как только хоть один из них изменится (то есть когда кто-то отпустит ресурс через release). Получился бы настоящий блокирующий файбер. У нас файбер один на всё, блокировать некого, поэтому fail-fast: LockWaitError валит текущий statement, транзакция остаётся открытой, её ребро ожидания продолжает жить. Когда вызывающий продолжит чередовать операторы и приведёт к взаимному ожиданию, hasCycle это заметит. То есть LockWaitError это “сейчас не вышло, попробуй позже или откатись”, а DeadlockError это “так не выйдет никогда, откатываю тебя”.

Ошибки конкуренции

Менеджер бросает две новые ошибки, а оператор записи третью. Добавь их в src/engine/errors.ts рядом с EngineError и UniquenessError из прошлых уроков. Все три это Data.TaggedError, как и весь остальной канал ошибок.

// src/engine/errors.ts (добавляем три ошибки)

import { Data } from 'effect';

/**
 * Конфликт сериализации (урок 9): write-write конфликт под snapshot isolation. Строку уже
 * изменила другая транзакция, невидимая моему снимку. Транзакция откатывается.
 */
export class SerializationError extends Data.TaggedError('SerializationError')<{
  readonly message: string;
}> {}

/**
 * Лок занят другой транзакцией без цикла ожидания (урок 9). В многофайберном движке
 * транзакция бы заблокировалась через Effect.txRetry; здесь падает только текущий statement,
 * транзакция остаётся открытой, её ребро ожидания живёт.
 */
export class LockWaitError extends Data.TaggedError('LockWaitError')<{
  readonly message: string;
}> {}

/** Обнаружен цикл в графе ожидания (урок 9): запросившая транзакция это жертва. */
export class DeadlockError extends Data.TaggedError('DeadlockError')<{
  readonly message: string;
}> {}

Три тега, три разных пути обработки: тест проверяет, что write-write конфликт это именно SerializationError, что занятый лок без цикла это LockWaitError, а цикл это DeadlockError. На границе с пользователем (например, в протоколе урока 11) мы превратим их в SQLSTATE: 40001 для сериализации, 55P03 для ожидания, 40P01 для дедлока. Постгрес тут наш проводник.