Блокировки в PostgreSQL: SELECT FOR UPDATE, SKIP LOCKED и advisory locks

← назад к разделу

Когда два запроса одновременно читают одну строку, принимают решение и оба пытаются её изменить — можно получить противоречивый результат. Блокировки решают именно эту проблему: они позволяют читать строку «с намерением изменить», и никто другой не тронет её, пока вы не закончите.

Как PostgreSQL блокирует строки по умолчанию

UPDATE и DELETE автоматически берут блокировку на каждую затронутую строку. Это происходит незаметно.

-- TX1 обновляет строку
UPDATE orders SET status = 'PAID' WHERE id = 42;

-- TX2 пытается обновить ту же строку одновременно
UPDATE orders SET status = 'CANCELLED' WHERE id = 42;  -- ждёт TX1

При этом обычный SELECT строку не блокирует и не ждёт — это основа MVCC (Multi-Version Concurrency Control): чтение видит «зафиксированную» версию строки и никогда не блокируется записью. На уровне по умолчанию READ COMMITTED эта версия определяется моментом начала запроса, а не транзакции: два одинаковых SELECT внутри одной транзакции вполне могут увидеть разные данные, если между ними кто-то успел зафиксировать изменения.

Проблема появляется, когда нужно прочитать, принять решение и обновить — и всё это как одна атомарная операция.

SELECT FOR UPDATE — читаю и собираюсь изменить

Представьте склад: два менеджера одновременно смотрят остаток товара (10 штук) и оба решают «можно продать 8». Оба делают UPDATE. В итоге продано 16 при остатке 10.

SELECT FOR UPDATE блокирует строки уже на этапе чтения:

BEGIN;

SELECT stock FROM product WHERE id = 100 FOR UPDATE;
-- строка заблокирована; второй запрос будет ждать

UPDATE product SET stock = stock - 8 WHERE id = 100;

COMMIT;
-- блокировка снята

Пока первая транзакция не завершится, параллельный SELECT FOR UPDATE той же строки будет ждать; обычный SELECT — нет, он увидит старое значение.

Важно: FOR UPDATE работает только внутри транзакции. Без явно открытой транзакции (@Transactional, BeginTx, with conn.transaction()) запрос уходит в режиме автокоммита — там каждый запрос сам себе транзакция, она заканчивается вместе с SELECT, и блокировка снимается ровно там же. Защиты нет.

Ту же гонку видно без базы: строка — обычное поле, FOR UPDATE — блокировка перед чтением.

import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;

public class StockDemo {
    static int stock;
    static final AtomicInteger sold = new AtomicInteger();
    static final ReentrantLock row = new ReentrantLock();

    public static void main(String[] args) throws Exception {
        System.out.println("без блокировки: " + round(false));
        System.out.println("с блокировкой:  " + round(true));
    }

    static String round(boolean forUpdate) throws Exception {
        stock = 10;
        sold.set(0);
        Thread first = new Thread(() -> sell(8, forUpdate));
        Thread second = new Thread(() -> sell(8, forUpdate));
        first.start();
        second.start();
        first.join();
        second.join();
        return "продано " + sold.get() + ", остаток " + stock;
    }

    static void sell(int amount, boolean forUpdate) {
        if (forUpdate) row.lock();
        try {
            int seen = stock;
            pause();
            if (seen >= amount) {
                stock = seen - amount;
                sold.addAndGet(amount);
            }
        } finally {
            if (forUpdate) row.unlock();
        }
    }

    static void pause() {
        try {
            Thread.sleep(50);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

package main

import (
	"fmt"
	"sync"
	"sync/atomic"
	"time"
)

var (
	stock int
	sold  atomic.Int64
	row   sync.Mutex
)

func main() {
	fmt.Println("без блокировки:", round(false))
	fmt.Println("с блокировкой: ", round(true))
}

func round(forUpdate bool) string {
	stock = 10
	sold.Store(0)
	var wg sync.WaitGroup
	for range 2 {
		wg.Go(func() { sell(8, forUpdate) })
	}
	wg.Wait()
	return fmt.Sprintf("продано %d, остаток %d", sold.Load(), stock)
}

func sell(amount int, forUpdate bool) {
	if forUpdate {
		row.Lock()
		defer row.Unlock()
	}
	seen := stock
	time.Sleep(50 * time.Millisecond)
	if seen >= amount {
		stock = seen - amount
		sold.Add(int64(amount))
	}
}

let stock = 0;
let sold = 0;
let busy = Promise.resolve();
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
const lock = (body) => (busy = busy.then(body));

async function sell(amount) {
  const seen = stock;
  await sleep(50);
  if (seen >= amount) {
    stock = seen - amount;
    sold += amount;
  }
}

async function round(forUpdate) {
  stock = 10;
  sold = 0;
  const attempt = forUpdate ? () => lock(() => sell(8)) : () => sell(8);
  await Promise.all([attempt(), attempt()]);
  return `продано ${sold}, остаток ${stock}`;
}

async function main() {
  console.log("без блокировки:", await round(false));
  console.log("с блокировкой: ", await round(true));
}
main();

import threading
import time

stock = 0
sold = 0
row = threading.Lock()


def sell(amount, for_update):
    global stock, sold
    if for_update:
        row.acquire()
    try:
        seen = stock
        time.sleep(0.05)
        if seen >= amount:
            stock = seen - amount
            sold += amount
    finally:
        if for_update:
            row.release()


def round_of_sales(for_update):
    global stock, sold
    stock, sold = 10, 0
    threads = [threading.Thread(target=sell, args=(8, for_update)) for _ in range(2)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    return f"продано {sold}, остаток {stock}"


print("без блокировки:", round_of_sales(False))
print("с блокировкой: ", round_of_sales(True))

Первая строка вывода — продано 16 там, где на складе было 10: оба потока прочитали одно значение и оба решили, что товара хватает. С блокировкой второй поток дождался первого, увидел уменьшенный остаток и продажу не сделал. SELECT FOR UPDATE делает в базе ровно это: задерживает чтение до тех пор, пока предыдущая транзакция не закончит.

Варианты блокировки

В большинстве случаев достаточно FOR UPDATE: строка блокируется полностью для любого UPDATE. Остальные три встречаются в конкретных ситуациях. FOR NO KEY UPDATE берут, когда обновляют только не-ключевые поля, и знать его стоит из-за побочного эффекта обычного FOR UPDATE: тот блокирует и ключ, поэтому вставки в соседние таблицы, которые ссылаются на строку внешним ключом, встают в очередь за вашим UPDATE; с NO KEY проверка внешнего ключа проходит. FOR SHARE говорит «строка не должна измениться, пока я работаю», и другие могут взять такой же FOR SHARE, но не FOR UPDATE. FOR KEY SHARE минимальная блокировка, её берёт сама база при проверке внешних ключей, руками её пишут редко.

Почему это вообще работает, стоит сказать прямо, иначе блокировка выглядит магией. На уровне по умолчанию READ COMMITTED запрос с FOR UPDATE, отстояв очередь за строкой, перечитывает её в новой версии — той, которую записал предыдущий владелец, — и заново проверяет по ней условие WHERE. Поэтому второй менеджер видит уже остаток 2, а не 10: он читает не свой снимок, а свежую строку. По той же причине строка, переставшая подходить под условие, из выдачи просто исчезнет.

На REPEATABLE READ и выше так нельзя: транзакция обязана видеть один снимок, а свежая версия в него не помещается. Поэтому вместо ожидания и перечитывания она получит ошибку could not serialize access due to concurrent update (SQLSTATE 40001) и должна повторить работу целиком — про повтор в статье про уровни изоляции.

SKIP LOCKED — очередь задач из таблицы

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

FOR UPDATE SKIP LOCKED пропускает уже заблокированные строки вместо того чтобы ждать:

BEGIN;

SELECT id, payload
FROM task_queue
WHERE status = 'PENDING'
ORDER BY created_at
LIMIT 1
FOR UPDATE SKIP LOCKED;
-- каждый обработчик получит свою строку

-- обновляем ровно тот id, который вернул SELECT выше
UPDATE task_queue SET status = 'PROCESSING' WHERE id = :id;

COMMIT;

Несколько обработчиков могут выполнять этот запрос одновременно — каждый захватит свою строку, никто не будет ждать остальных. Это стандартный паттерн для outbox-relay и распределённых очередей.

Главное про такую очередь — то, чего в примере не видно: транзакция держится всё время обработки задачи. Строка заблокирована от SELECT … FOR UPDATE до COMMIT, а между ними стоит сама работа — вызов внешней службы, генерация документа, отправка письма.

У этого есть плюс, ради которого схему и берут: обработчик упал, соединение оборвалось — транзакция откатилась сама, блокировка снялась, задача снова видна как PENDING и достанется другому. Никаких зависших «в обработке» задач, которые надо чинить руками.

И есть цена. Задача на три минуты означает транзакцию на три минуты: занятое соединение из пула, удерживаемый горизонт видимости и растущий мусор в таблицах, который не может вычистить VACUUM. Для задач в секунды это нормально; для долгих — нет.

Альтернатива для долгих задач — захват статусом с арендой. Обработчик короткой транзакцией помечает задачу своей (UPDATE … SET status = 'PROCESSING', owner = :me, lease_until = now() + interval '1 minute' WHERE id = :id AND status = 'PENDING'), коммитит и работает уже без открытой транзакции, периодически продлевая аренду. Упал — аренда истекла, и задачу заберёт другой: сторож возвращает в PENDING всё, у чего lease_until в прошлом. Платой становится то, от чего избавляла первая схема: задача может выполниться дважды (владелец «завис», аренда истекла, а он потом ожил), поэтому обработка обязана быть идемпотентной.

Правило выбора простое: работа укладывается в секунды и идемпотентной её делать дорого — SKIP LOCKED с транзакцией на всё время; работа долгая или ходит во внешние службы — аренда со сроком и продлением.

NOWAIT — лучше ошибка, чем ожидание

Иногда ждать нежелательно: пользователь ожидает мгновенного ответа, и если строка занята — лучше сразу сообщить об этом.

FOR UPDATE NOWAIT не ждёт — сразу бросает ошибку, если строка заблокирована:

SELECT * FROM orders WHERE id = 'ord-01' FOR UPDATE NOWAIT;
-- если строка занята — ошибка, а не ожидание

Приложение ловит исключение и возвращает пользователю «попробуйте через несколько секунд».

Отдельная тема — FOR UPDATE в запросе с соединением. Блокируются строки всех таблиц, попавших в выдачу, и это редко то, что нужно: заблокировать хотели заказ, а заблокировали заодно и покупателя. Указать цель явно позволяет форма FOR UPDATE OF:

SELECT o.id, c.email
FROM orders o
JOIN customer c ON c.id = o.customer_id
WHERE o.id = 'ord-01'
FOR UPDATE OF o;

И сразу ошибка, на которую натыкаются с LEFT JOIN: FOR UPDATE cannot be applied to the nullable side of an outer join. Логика простая — блокировать нечего, строки правой таблицы может не быть вовсе. Лечится тем же OF: назовите таблицу, строки которой действительно есть.

Блокировки в коде приложения

В коде приложения блокировка живёт в тексте запроса, у каждого языка свой драйвер:

// FOR UPDATE
ProductRecord product = dsl
    .selectFrom(PRODUCT)
    .where(PRODUCT.ID.eq(productId))
    .forUpdate()
    .fetchOne();

// FOR UPDATE SKIP LOCKED — очередь задач
List<TaskRecord> tasks = dsl
    .selectFrom(TASK_QUEUE)
    .where(TASK_QUEUE.STATUS.eq("PENDING"))
    .orderBy(TASK_QUEUE.CREATED_AT)
    .limit(10)
    .forUpdate()
    .skipLocked()
    .fetch();

// FOR UPDATE NOWAIT
dsl.selectFrom(ORDER_DOC)
    .where(ORDER_DOC.ID.eq(orderId))
    .forUpdate()
    .noWait()
    .fetchOne();

// FOR UPDATE
var product Product
err := tx.QueryRow(ctx,
	`SELECT id, stock FROM product WHERE id = $1 FOR UPDATE`, productID).
	Scan(&product.ID, &product.Stock)

// FOR UPDATE SKIP LOCKED — очередь задач
rows, err := tx.Query(ctx,
	`SELECT id, payload FROM task_queue WHERE status = 'PENDING'
	 ORDER BY created_at LIMIT 10 FOR UPDATE SKIP LOCKED`)

// FOR UPDATE NOWAIT
err = tx.QueryRow(ctx,
	`SELECT id, status FROM order_doc WHERE id = $1 FOR UPDATE NOWAIT`, orderID).
	Scan(&order.ID, &order.Status)

// FOR UPDATE
const { rows: [product] } = await client.query(
  'SELECT id, stock FROM product WHERE id = $1 FOR UPDATE', [productId]);

// FOR UPDATE SKIP LOCKED — очередь задач
const { rows: tasks } = await client.query(
  `SELECT id, payload FROM task_queue WHERE status = 'PENDING'
   ORDER BY created_at LIMIT 10 FOR UPDATE SKIP LOCKED`);

// FOR UPDATE NOWAIT
const { rows: [order] } = await client.query(
  'SELECT id, status FROM order_doc WHERE id = $1 FOR UPDATE NOWAIT', [orderId]);

# FOR UPDATE
product = conn.execute(
    "SELECT id, stock FROM product WHERE id = %s FOR UPDATE", (product_id,)
).fetchone()

# FOR UPDATE SKIP LOCKED — очередь задач
tasks = conn.execute(
    """SELECT id, payload FROM task_queue WHERE status = 'PENDING'
       ORDER BY created_at LIMIT 10 FOR UPDATE SKIP LOCKED"""
).fetchall()

# FOR UPDATE NOWAIT
order = conn.execute(
    "SELECT id, status FROM order_doc WHERE id = %s FOR UPDATE NOWAIT", (order_id,)
).fetchone()

Так forUpdate().skipLocked() выглядит в реальном коде — метод, которым outbox-relay вычитывает очередную пачку ещё не отправленных событий:

@Transactional
public int relayBatch() {
    List<OutboxRecord> batch = dsl
        .selectFrom(OUTBOX)
        .where(OUTBOX.PUBLISHED_AT.isNull())
        .orderBy(OUTBOX.OCCURRED_AT.asc())
        .limit(batchSize)
        .forUpdate()
        .skipLocked()
        .fetch();

    if (batch.isEmpty()) {
        return 0;
    }

    OffsetDateTime now = OffsetDateTime.ofInstant(dateTimeService.now(), ZoneOffset.UTC);
    int published = 0;
    for (OutboxRecord record : batch) {
        externalEventPublisher.publish(
            record.getId(),
            record.getAggregateType(),
            record.getAggregateId(),
            record.getEventType(),
            record.getEventVersion(),
            record.getPayload().data(),
            record.getOccurredAt().toInstant());

        dsl.update(OUTBOX)
            .set(OUTBOX.PUBLISHED_AT, now)
            .where(OUTBOX.ID.eq(record.getId()))
            .execute();
        published++;
    }
    return published;
}

Настоящий код проекта: order OutboxRelay#relayBatch

К блокировке здесь относится только .forUpdate().skipLocked() в начале метода — дальше идут публикация событий и отметка времени отправки.

Ещё пример — внутри транзакции, без которой блокировка не живёт:

@Transactional
public void reserveStock(long productId, int quantity) {
    var product = dsl.selectFrom(PRODUCT)
        .where(PRODUCT.ID.eq(productId))
        .forUpdate()
        .fetchOne();

    if (product == null) throw new ProductNotFoundException(productId);
    if (product.getStock() < quantity) throw new InsufficientStockException(productId);

    dsl.update(PRODUCT)
       .set(PRODUCT.STOCK, PRODUCT.STOCK.minus(quantity))
       .where(PRODUCT.ID.eq(productId))
       .execute();
}

func (r *StockRepo) Reserve(ctx context.Context, productID int64, quantity int) error {
	return pgx.BeginFunc(ctx, r.pool, func(tx pgx.Tx) error {
		var stock int
		err := tx.QueryRow(ctx, `SELECT stock FROM product WHERE id = $1 FOR UPDATE`, productID).Scan(&stock)
		if errors.Is(err, pgx.ErrNoRows) {
			return ErrProductNotFound
		}
		if err != nil {
			return err
		}
		if stock < quantity {
			return ErrInsufficientStock
		}
		_, err = tx.Exec(ctx, `UPDATE product SET stock = stock - $2 WHERE id = $1`, productID, quantity)
		return err
	})
}

async function reserveStock(pool, productId, quantity) {
  const client = await pool.connect();
  try {
    await client.query('BEGIN');
    const { rows } = await client.query(
      'SELECT stock FROM product WHERE id = $1 FOR UPDATE', [productId]);
    if (rows.length === 0) throw new ProductNotFoundError(productId);
    if (rows[0].stock < quantity) throw new InsufficientStockError(productId);
    await client.query('UPDATE product SET stock = stock - $2 WHERE id = $1', [productId, quantity]);
    await client.query('COMMIT');
  } catch (e) {
    await client.query('ROLLBACK');
    throw e;
  } finally {
    client.release();
  }
}

def reserve_stock(conn, product_id: int, quantity: int) -> None:
    with conn.transaction():
        row = conn.execute(
            "SELECT stock FROM product WHERE id = %s FOR UPDATE", (product_id,)
        ).fetchone()
        if row is None:
            raise ProductNotFound(product_id)
        if row[0] < quantity:
            raise InsufficientStock(product_id)
        conn.execute(
            "UPDATE product SET stock = stock - %s WHERE id = %s", (quantity, product_id)
        )

Pessimistic vs Optimistic: когда что выбрать

SELECT FOR UPDATE — это pessimistic подход: «предполагаю конфликт, блокирую заранее». Он прост в коде, но при большом числе параллельных запросов образует очередь.

Optimistic подход предполагает, что конфликты редки, и проверяет это только в момент записи. Для этого добавляют колонку version:

ALTER TABLE orders ADD COLUMN version bigint NOT NULL DEFAULT 0;

Читаем вместе с версией:

живой пример

SELECT id, status, total_amount FROM orders WHERE id = 'ord-01';

Пишем с проверкой версии:

UPDATE orders
SET status = 'PAID', version = version + 1
WHERE id = 'ord-01' AND version = :version;
-- если кто-то изменил строку — version другая, 0 rows affected

В коде это один UPDATE с условием на версию и проверка числа изменённых строк:

int updated = dsl.update(ORDER_DOC)
    .set(ORDER_DOC.STATUS, "PAID")
    .set(ORDER_DOC.VERSION, ORDER_DOC.VERSION.plus(1))
    .where(ORDER_DOC.ID.eq(orderId)
        .and(ORDER_DOC.VERSION.eq(originalVersion)))
    .execute();

if (updated == 0) {
    throw new OptimisticLockException("order " + orderId + " изменён параллельно");
}

tag, err := pool.Exec(ctx,
	`UPDATE order_doc SET status = 'PAID', version = version + 1
	 WHERE id = $1 AND version = $2`, orderID, originalVersion)
if err != nil {
	return err
}
if tag.RowsAffected() == 0 {
	return fmt.Errorf("order %d изменён параллельно: %w", orderID, ErrOptimisticLock)
}

const result = await pool.query(
  `UPDATE order_doc SET status = 'PAID', version = version + 1
   WHERE id = $1 AND version = $2`, [orderId, originalVersion]);

if (result.rowCount === 0) {
  throw new OptimisticLockError(`order ${orderId} изменён параллельно`);
}

cur = conn.execute(
    """UPDATE order_doc SET status = 'PAID', version = version + 1
       WHERE id = %s AND version = %s""",
    (order_id, original_version),
)
if cur.rowcount == 0:
    raise OptimisticLockError(f"order {order_id} изменён параллельно")

Вся защита — в этом нуле. Откуда он берётся, видно на обычном объекте: двое прочитали версию 7, первый записал, второму условие version = 7 уже не подходит.

public class VersionDemo {
    record Order(String status, long version) {}

    static Order stored = new Order("NEW", 7);

    static int update(String status, long expectedVersion) {
        if (stored.version() != expectedVersion) {
            return 0;
        }
        stored = new Order(status, expectedVersion + 1);
        return 1;
    }

    public static void main(String[] args) {
        long readByFirst = stored.version();
        long readBySecond = stored.version();

        System.out.println("первый: обновлено строк " + update("PAID", readByFirst));
        System.out.println("второй: обновлено строк " + update("CANCELLED", readBySecond));
        System.out.println("в таблице: " + stored);
    }
}

package main

import "fmt"

type Order struct {
	Status  string
	Version int64
}

var stored = Order{"NEW", 7}

func update(status string, expectedVersion int64) int {
	if stored.Version != expectedVersion {
		return 0
	}
	stored = Order{status, expectedVersion + 1}
	return 1
}

func main() {
	readByFirst := stored.Version
	readBySecond := stored.Version

	fmt.Println("первый: обновлено строк", update("PAID", readByFirst))
	fmt.Println("второй: обновлено строк", update("CANCELLED", readBySecond))
	fmt.Printf("в таблице: Order{status=%s, version=%d}\n", stored.Status, stored.Version)
}

let stored = { status: "NEW", version: 7 };

function update(status, expectedVersion) {
  if (stored.version !== expectedVersion) return 0;
  stored = { status, version: expectedVersion + 1 };
  return 1;
}

const readByFirst = stored.version;
const readBySecond = stored.version;

console.log("первый: обновлено строк", update("PAID", readByFirst));
console.log("второй: обновлено строк", update("CANCELLED", readBySecond));
console.log("в таблице:", stored);

from dataclasses import dataclass


@dataclass(frozen=True)
class Order:
    status: str
    version: int


stored = Order("NEW", 7)


def update(status, expected_version):
    global stored
    if stored.version != expected_version:
        return 0
    stored = Order(status, expected_version + 1)
    return 1


read_by_first = stored.version
read_by_second = stored.version

print("первый: обновлено строк", update("PAID", read_by_first))
print("второй: обновлено строк", update("CANCELLED", read_by_second))
print("в таблице:", stored)

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

Когда что использовать:

• Optimistic — когда конфликты редки (read-heavy): документы, профили, справочники. Меньше нагрузки на базу, выше пропускная способность.
• Pessimistic — когда конфликты часты: финансовые операции, остатки на складе, любые «горячие» строки с высокой конкуренцией.

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

И правило, которое в примере выше соблюдено, но не названо: version увеличивают в том же UPDATE, что и данные, и обязательно с условием по старой версии. Отдельный запрос «сначала проверим версию, потом обновим» не даёт ничего: между двумя запросами помещается чужая транзакция, а вся защита держится ровно на том, что проверка и запись — одна атомарная операция.

Advisory locks — блокировка на произвольный ключ

Иногда нужно заблокировать не строку, а какую-то операцию целиком — например, чтобы задание по расписанию запускалось только на одном экземпляре приложения в кластере.

Advisory lock — это блокировка на произвольное число (bigint). PostgreSQL хранит её в памяти, а не на строке:

-- транзакционный advisory lock (снимается при COMMIT/ROLLBACK)
SELECT pg_advisory_xact_lock(12345);

-- попытка взять lock без ожидания (возвращает true/false)
SELECT pg_try_advisory_xact_lock(12345);

-- с двумя числами: пространство имён + идентификатор
SELECT pg_advisory_xact_lock(1001, :tenant_id);

Типичные применения — одиночный запуск задания по расписанию при нескольких экземплярах приложения и защита от параллельного старта двух одинаковых миграций.

В коде приложения:

@Transactional
public void runIfNotAlreadyRunning(long jobKey, Runnable job) {
    Boolean acquired = dsl.select(
        DSL.function("pg_try_advisory_xact_lock", Boolean.class, DSL.val(jobKey))
    ).fetchOne(0, Boolean.class);

    if (Boolean.TRUE.equals(acquired)) {
        job.run();
    } else {
        log.info("job {} уже запущен на другом экземпляре, пропускаю", jobKey);
    }
}

func runIfNotAlreadyRunning(ctx context.Context, pool *pgxpool.Pool, jobKey int64, job func() error) error {
	return pgx.BeginFunc(ctx, pool, func(tx pgx.Tx) error {
		var acquired bool
		if err := tx.QueryRow(ctx, `SELECT pg_try_advisory_xact_lock($1)`, jobKey).Scan(&acquired); err != nil {
			return err
		}
		if !acquired {
			slog.Info("job уже запущен на другом экземпляре, пропускаю", "job", jobKey)
			return nil
		}
		return job()
	})
}

async function runIfNotAlreadyRunning(pool, jobKey, job) {
  const client = await pool.connect();
  try {
    await client.query('BEGIN');
    const { rows } = await client.query('SELECT pg_try_advisory_xact_lock($1) AS acquired', [jobKey]);
    if (rows[0].acquired) {
      await job(client);
    } else {
      console.info(`job ${jobKey} уже запущен на другом экземпляре, пропускаю`);
    }
    await client.query('COMMIT');
  } catch (e) {
    await client.query('ROLLBACK');
    throw e;
  } finally {
    client.release();
  }
}

def run_if_not_already_running(conn, job_key: int, job) -> None:
    with conn.transaction():
        acquired = conn.execute(
            "SELECT pg_try_advisory_xact_lock(%s)", (job_key,)
        ).fetchone()[0]
        if acquired:
            job(conn)
        else:
            log.info("job %s уже запущен на другом экземпляре, пропускаю", job_key)

Три вещи про advisory-блокировки, без которых их применяют неправильно.

Сессионная форма и пул соединений. Кроме транзакционного pg_advisory_xact_lock есть сессионный pg_advisory_lock: он держится, пока соединение живо, и снимается только явным pg_advisory_unlock или разрывом соединения. С пулом это почти всегда ошибка: соединение возвращается в пул вместе с висящей блокировкой, и следующий, кому оно достанется, унаследует чужой замок или, наоборот, снимет его своим unlock. В приложении с пулом берут транзакционную форму — она снимается на COMMIT сама.

Пространство ключей общее на всю базу. Никаких «своих» областей у приложения нет: число 12345, взятое двумя разными заданиями, — это одна и та же блокировка, и второе задание будет молча ждать первого без всякой связи между ними. Поэтому ключ не выдумывают, а считают из имени: SELECT pg_advisory_xact_lock(hashtext('nightly-report')). Совпадения хешей маловероятны, зато имя видно в коде. Вариант с двумя числами (pg_advisory_xact_lock(1001, :tenant_id)) решает ту же задачу явно: первое число — область, второе — объект.

Для «одиночного запуска задания» уже есть готовое. Планировщики внутри приложения умеют распределённый замок: ShedLock у Java-стека, gocron с замком в базе или Redis у Go — замок держится в отдельной таблице, у него есть минимальное и максимальное время удержания, и он переживает падение экземпляра, а не только его соединение. Писать advisory-блокировку руками стоит там, где нужен замок на произвольный ключ данных — обработку конкретного клиента, конкретный файл, — а не там, где нужен «только один экземпляр задания».

Deadlock — взаимная блокировка

Под нагрузкой один-два процента переводов падают с ошибкой 40P01 deadlock detected, а на стенде это не воспроизводится: две транзакции взяли по одной строке и ждут строку друг друга. Это deadlock, взаимная блокировка:

TX1: блокирует строку с id=1
TX2: блокирует строку с id=2
TX1: пытается заблокировать id=2 — ждёт TX2
TX2: пытается заблокировать id=1 — ждёт TX1
→ тупик

Обнаруживает это PostgreSQL не мгновенно. deadlock_timeout (по умолчанию 1 секунда) — это не «через сколько прервут», а «сколько ждать, прежде чем заподозрить неладное»: транзакция сначала честно стоит в очереди секунду, и только потом сервер идёт смотреть, не замкнулось ли ожидание в кольцо. Замкнулось — одну из транзакций прерывают с ошибкой 40P01.

Главная причина: транзакции берут блокировки в разном порядке.

Решение — всегда блокировать строки в одном порядке, например по возрастанию id:

// Плохо: порядок зависит от аргументов
var from = lockAccount(fromAccountId);
var to   = lockAccount(toAccountId);

// Хорошо: всегда сначала меньший id
long firstId  = Math.min(fromAccountId, toAccountId);
long secondId = Math.max(fromAccountId, toAccountId);
var first  = lockAccount(firstId);
var second = lockAccount(secondId);

// Плохо: порядок зависит от аргументов
from := lockAccount(ctx, tx, fromAccountID)
to := lockAccount(ctx, tx, toAccountID)

// Хорошо: всегда сначала меньший id
first := lockAccount(ctx, tx, min(fromAccountID, toAccountID))
second := lockAccount(ctx, tx, max(fromAccountID, toAccountID))

// Плохо: порядок зависит от аргументов
const from = await lockAccount(client, fromAccountId);
const to = await lockAccount(client, toAccountId);

// Хорошо: всегда сначала меньший id
const first = await lockAccount(client, Math.min(fromAccountId, toAccountId));
const second = await lockAccount(client, Math.max(fromAccountId, toAccountId));

# Плохо: порядок зависит от аргументов
src = lock_account(conn, from_account_id)
dst = lock_account(conn, to_account_id)

# Хорошо: всегда сначала меньший id
first = lock_account(conn, min(from_account_id, to_account_id))
second = lock_account(conn, max(from_account_id, to_account_id))

Дополнительно — повтор транзакции, если взаимная блокировка всё-таки случилась. Здесь важно не ошибиться с кодом ошибки, и на этом ошибаются чаще всего. PostgreSQL сообщает о взаимной блокировке кодом 40P01, а соседний 40001 — это неудача сериализации на REPEATABLE READ и SERIALIZABLE; оба имеет смысл повторить с паузой. А истёкший lock_timeout и NOWAIT (код 55P03) — не повод для слепого повтора: строка занята, и об этом честнее сказать вызывающему. Повтор по кодам семейства 40 выглядит так:

@Retryable(
    retryFor = PessimisticLockingFailureException.class, // сюда попадают и 40P01, и 40001
    maxAttempts = 3,
    backoff = @Backoff(delay = 50, multiplier = 2, random = true)
)
@Transactional
public void transferMoney(...) { ... }

func transferMoney(ctx context.Context, pool *pgxpool.Pool, from, to, amount int64) error {
	delay := 50 * time.Millisecond
	for attempt := 1; ; attempt++ {
		err := pgx.BeginFunc(ctx, pool, func(tx pgx.Tx) error {
			return transfer(ctx, tx, from, to, amount)
		})
		var pgErr *pgconn.PgError
		if err == nil || !errors.As(err, &pgErr) || attempt == 3 {
			return err
		}
		if pgErr.Code != "40P01" && pgErr.Code != "40001" {
			return err
		}
		time.Sleep(delay + time.Duration(rand.IntN(50))*time.Millisecond)
		delay *= 2
	}
}

async function transferMoney(pool, from, to, amount) {
  let delay = 50;
  for (let attempt = 1; ; attempt++) {
    try {
      return await inTransaction(pool, (client) => transfer(client, from, to, amount));
    } catch (e) {
      const retryable = e.code === '40P01' || e.code === '40001';
      if (!retryable || attempt === 3) throw e;
      await new Promise((r) => setTimeout(r, delay + Math.random() * 50));
      delay *= 2;
    }
  }
}

import random
import time
from psycopg.errors import TransactionRollback


def transfer_money(conn, from_id, to_id, amount, attempts=3):
    delay = 0.05
    for attempt in range(1, attempts + 1):
        try:
            with conn.transaction():
                return transfer(conn, from_id, to_id, amount)
        except TransactionRollback:
            if attempt == attempts:
                raise
            time.sleep(delay + random.random() * 0.05)
            delay *= 2

Одного-трёх повторов с небольшой паузой хватает почти всегда.

lock_timeout — ограничение времени ожидания

Если транзакция не может получить блокировку за отведённое время — лучше получить ошибку, чем бесконечно висеть в очереди. По умолчанию lock_timeout = 0, то есть ограничения нет вовсе: ждать можно сколько угодно. Поэтому в большинстве баз он и не настроен — сам по себе он не появится.

SET LOCAL действует только до конца транзакции, так что ставить его надо внутри BEGIN … COMMIT, иначе он не даст ничего:

BEGIN;
SET LOCAL lock_timeout = '5s';
UPDATE orders SET ... WHERE id = 'ord-01';
COMMIT;

Это особенно важно для миграций. ALTER TABLE берёт ACCESS EXCLUSIVE — самую тяжёлую блокировку. Без ограничения времени миграция может часами ждать активных запросов, а за ней выстроится очередь новых:

BEGIN;
SET LOCAL lock_timeout = '3s';
ALTER TABLE orders ADD COLUMN processed_at timestamptz;
COMMIT;

Частые ошибки

• FOR UPDATE без индекса по WHERE — PostgreSQL не «повышает» блокировку до табличной, как это делают некоторые другие базы: заблокированы будут ровно подходящие строки. Но чтобы их найти, придётся прочитать всю таблицу, а заодно поблокировать всё, что попалось по пути и подошло под условие. Проверяйте план через EXPLAIN.
• Длинная транзакция с блокировкой — строка заблокирована на всё время обработки. Берите блокировку как можно ближе к UPDATE.
• FOR UPDATE без LIMIT на большой таблице — для очередей задач всегда добавляйте LIMIT.

Глубже: upsert: INSERT … ON CONFLICT и MERGEрасширенное

«Вставить, а если уже есть, обновить» руками пишется как SELECT, потом INSERT или UPDATE, и между ними влезает вторая транзакция: обе не нашли строку, обе вставили, вторая упала на уникальном индексе. PostgreSQL решает это одной командой, которая делает проверку и запись атомарно, под блокировкой строки:

INSERT INTO stock (product_id, quantity)
VALUES ('p-01', 10)
ON CONFLICT (product_id) DO UPDATE
SET quantity = stock.quantity + EXCLUDED.quantity
RETURNING product_id, quantity;

Цель конфликта в скобках это колонки уникального индекса или имя ограничения; без уникального индекса ON CONFLICT не работает, и это правильно: без него «уже есть» не определено. EXCLUDED обозначает строку, которую пытались вставить, stock.quantity текущее значение в таблице. DO NOTHING вместо DO UPDATE молча пропускает дубликат, и это готовая защита от повторной доставки события: ключ события уникален, второй раз он не вставится.

MERGE, появившийся в PostgreSQL 15, делает то же, но сопоставляет целую таблицу-источник с целевой и умеет разные действия для разных случаев: WHEN MATCHED THEN UPDATE, WHEN NOT MATCHED THEN INSERT, WHEN MATCHED AND … THEN DELETE. Он удобен для сверок и загрузок, где источник это запрос, а не одна строка. Для «одна строка, одна проверка» короче и надёжнее ON CONFLICT: у MERGE нет той же защиты от параллельной вставки одинаковых ключей, при гонке он может упасть на уникальном индексе, и его повторяют.

Глубже: ключ идемпотентности: повтор без второго списаниярасширенное

Клиент отправил POST /orders, ответ потерялся в сети, клиент повторил запрос. Без защиты второй запрос создаст второй заказ. Защита живёт в базе: клиент присылает с запросом ключ попытки (UUID в заголовке Idempotency-Key), а сервер хранит его в таблице с уникальным индексом.

CREATE TABLE idempotency_key (
    key         uuid PRIMARY KEY,
    response    jsonb NOT NULL,
    created_at  timestamptz NOT NULL DEFAULT now()
);

Порядок в транзакции: вставить ключ, выполнить операцию, сохранить ответ, зафиксировать. Повтор с тем же ключом упирается в первичный ключ; сервис ловит ошибку уникальности (или делает INSERT ... ON CONFLICT DO NOTHING и смотрит, вставилось ли), читает сохранённый ответ и отдаёт его без второго списания. Если первый запрос ещё выполняется, вставка ждёт на его блокировке строки и после его коммита видит ответ, а после отката получает возможность выполнить операцию заново. Это и есть причина, по которой ключ должен быть в той же транзакции, что и операция: иначе ключ сохранится, а заказ нет.

Ключи копятся, и их чистят по сроку: сутки или неделя, столько, сколько клиент может повторять. Тест фазы спрашивает про уникальный индекс «заказ плюс попытка», это тот же приём с составным ключом. Как это оформляют в REST, разобрано в разделе про API.

Глубже: массовая вставка: пакеты, unnest и COPYрасширенное

Сто тысяч строк по одному INSERT это сто тысяч обращений к базе по сети, минуты вместо секунд. Три способа быстрее, по нарастанию.

Многострочный INSERT кладёт сотни строк одной командой: INSERT INTO t VALUES (...), (...), ...; у драйверов для этого есть пакетная отправка: addBatch и executeBatch в JDBC (с reWriteBatchedInserts=true в строке подключения драйвер сам склеивает пакет в многострочную команду), pgx.Batch в Go, executemany в psycopg. Порция в 500–1000 строк обычно оптимум: больше не ускоряет, а команда растёт.

unnest передаёт массивы параметров и разворачивает их в строки одной командой с фиксированным числом параметров, что удобно с подготовленными запросами:

INSERT INTO order_items (order_id, product_id, quantity)
SELECT * FROM unnest($1::uuid[], $2::uuid[], $3::int[]);

COPY самый быстрый путь: данные идут потоком в формате CSV или бинарном, без разбора SQL; в JDBC это CopyManager, в pgx — CopyFrom, в psycopg — cursor.copy(). Миллионы строк за секунды, но без ON CONFLICT и с одной ошибкой на весь поток: обычно грузят в промежуточную таблицу, а оттуда переносят INSERT ... SELECT.

Про размер транзакции. Одна транзакция на миллион строк держит блокировки и не даёт автоочистке убрать старые версии, а откат в конце теряет всё. Одна транзакция на строку медленна из-за фиксации каждой. Середина, транзакция на порцию в несколько тысяч строк с возможностью продолжить с места остановки, это обычный компромисс для загрузок.

Глубже: кто кого держит: pg_stat_activity и pg_blocking_pidsрасширенное

Запрос висит, и непонятно, ждёт ли он замок или просто долгий. Ответ в pg_stat_activity: у каждой сессии там колонки state (active или idle in transaction) и wait_event_type, и значение Lock в последней означает ожидание блокировки.

живой пример

SELECT pid, state, wait_event_type, wait_event, now() - xact_start AS in_tx, left(query, 60) AS query
FROM pg_stat_activity
WHERE state <> 'idle'
ORDER BY xact_start;

Кто держит, отвечает функция pg_blocking_pids(pid): она возвращает идентификаторы сессий, из-за которых ждёт указанная. Одним запросом получают пары «ждёт и держит»:

живой пример

SELECT w.pid AS waiting, pg_blocking_pids(w.pid) AS blocked_by, left(w.query, 60) AS waiting_query
FROM pg_stat_activity w
WHERE cardinality(pg_blocking_pids(w.pid)) > 0;

Чаще всего держатель оказывается в состоянии idle in transaction: приложение открыло транзакцию, изменило строку и ушло звать внешний сервис или просто забыло закоммитить. Такую сессию снимают pg_terminate_backend(pid), а причину ищут в коде. Сами замки видны в pg_locks (granted = false у ожидающих), но для повседневной диагностики хватает двух запросов выше. Что делать, если ждать не хочется вовсе, NOWAIT и lock_timeout из этой статьи.

Коротко

• UPDATE/DELETE берут row-level блокировку автоматически, обычный SELECT — нет (MVCC); SELECT FOR UPDATE блокирует строку уже на чтении, работает только внутри транзакции и на READ COMMITTED после ожидания перечитывает новую версию строки, заново проверяя WHERE, — на REPEATABLE READ вместо этого прилетит 40001.
• SKIP LOCKED пропускает занятые строки — так разбирают очередь несколько обработчиков, но транзакция держится всё время обработки: падение возвращает задачу само, а долгая задача просит аренду со сроком и идемпотентность. NOWAIT вместо ожидания сразу даёт ошибку.
• Optimistic (version-колонка) — для редких конфликтов; pessimistic (FOR UPDATE) — для финансов и горячих строк.
• Advisory lock — блокировка на произвольный ключ, в приложении с пулом только транзакционная и с ключом от hashtext('имя'): пространство ключей общее на всю базу, а «один экземпляр задания» проще закрыть готовым замком планировщика (ShedLock, gocron с замком).
• Deadlock лечится упорядочиванием блокировок по id; повтор транзакции по кодам 40P01 и 40001 закрывает редкие случаи.
• lock_timeout обязателен для ALTER TABLE — без него миграция может заблокировать весь трафик.
• Upsert это INSERT … ON CONFLICT (уникальный ключ) DO UPDATE с EXCLUDED; DO NOTHING пропускает повтор, MERGE для сверок с таблицей-источником.
• Ключ идемпотентности живёт в таблице с уникальным ключом в той же транзакции, что операция; повтор отдаёт сохранённый ответ.
• Массовая вставка: многострочный INSERT или пакет драйвера по 500–1000 строк, unnest для массивов, COPY для миллионов; транзакция на порцию.
• Кто кого держит: pg_stat_activity с wait_event_type = Lock и pg_blocking_pids(pid); виновник обычно idle in transaction.

Что пощупать

Обе блокировки строки из статьи стоят в практикуме remodov/marketplace-system в рабочих местах: SELECT … FOR UPDATE держит остаток товара, пока его резервируют сто покупателей, а FOR UPDATE SKIP LOCKED позволяет нескольким копиям сервиса заказов разбирать одну таблицу исходящих, не мешая друг другу.

Код: product, outbox.

Код на Go: repository.go, service.go.

Код на Node: product.repository.ts, product.service.ts.

Код на Python: repository.py, service.py.

Сделаем сами

Ветка step-05-stock-and-locking.

На Go: ветка step-05-stock-and-locking.

На Node: ветка step-05-stock-and-locking.

На Python: ветка step-05-stock-and-locking.

Что почитать дальше

• Уровни изоляции и аномалии — когда SELECT FOR UPDATE недостаточно и нужен SERIALIZABLE.
• Транзакции из кода — propagation, isolation и как транзакция взаимодействует с блокировками, на примере Java-стека.
• Миграции без простоя — lock_timeout и безопасный ALTER TABLE под нагрузкой.