Блокировки и как использовать их с jOOQ

Опирается на правила: PG-L-001 … PG-L-080, антипаттерны PG-L-090 … PG-L-095 из PostgreSQL Style Guide → раздел Блокировки.

Важно знать

UPDATE/DELETE автоматически берут row-level lock. Параллельный UPDATE ждёт, SELECT — нет (MVCC).

SELECT FOR UPDATE — read-modify-write защита. Только в @Transactional.

FOR UPDATE SKIP LOCKED — distributed очередь задач (outbox-relay).

FOR UPDATE NOWAIT — fail-fast.

Pessimistic vs Optimistic — optimistic через version для read-heavy, pessimistic для money.

Advisory lock — pg_advisory_xact_lock(bigint) для глобальных операций (scheduled job, миграции).

Deadlock — две TX в разном порядке; лечится упорядочением (по min/max id).

CannotAcquireLockException + @Retryable — норма в high-conc OLTP.

lock_timeout для критичных операций (ALTER TABLE).

PostgreSQL имеет три уровня блокировок: table (DDL), row (SELECT FOR UPDATE), advisory (произвольный ключ). UCP формулирует — что и когда брать, как через jOOQ, какие ловушки.

Базовая модель MVCC

PG-L-001: UPDATE/DELETE берут row-level lock автоматически.

TX1: UPDATE orders SET status='PAID' WHERE id=42;
TX2: UPDATE orders SET status='CANCELLED' WHERE id=42;   -- ждёт TX1
TX2: SELECT status FROM orders WHERE id=42;              -- НЕ ждёт, видит старую (committed)

Чтение никогда не ждёт запись через MVCC — поэтому фантомы между чтением и решением «надо ли менять» обычны, и SELECT FOR UPDATE существует именно для этого.

SELECT FOR UPDATE

PG-L-010..011: главный инструмент.

BEGIN;
SELECT stock FROM product WHERE id = 100 FOR UPDATE;   -- блокирует строку
-- решение: stock=5, заказ на 3, ок
UPDATE product SET stock = stock - 3 WHERE id = 100;
COMMIT;

Без FOR UPDATE два параллельных запроса прочитают stock=5, оба решат «можем», оба сделают UPDATE — продадут 6 при остатке 5.

Только в транзакции. Без @Transactional — лок отпускается сразу, проверка бесполезна.

Варианты FOR

PG-L-012: 95% — FOR UPDATE.

Команда	Когда
`FOR UPDATE`	Любой UPDATE-after-SELECT
`FOR NO KEY UPDATE`	UPDATE не PK/UK полей (HOT update friendly)
`FOR SHARE`	«Эта строка не должна измениться, пока я работаю» (редко)
`FOR KEY SHARE`	Удержать строку для FK-проверки

SKIP LOCKED — очередь задач

PG-L-020..021: distributed worker pattern.

BEGIN;
SELECT id, payload
FROM task_queue
WHERE status = 'PENDING'
ORDER BY created_at
LIMIT 1
FOR UPDATE SKIP LOCKED;

UPDATE task_queue SET status = 'PROCESSING' WHERE id = ?;
COMMIT;
-- обработка
UPDATE task_queue SET status = 'DONE' WHERE id = ?;

Несколько worker-ов одновременно SELECT FOR UPDATE SKIP LOCKED — каждый получит свою строку, никто не ждёт.

Используется в outbox-relay (см. Kafka → outbox publishing).

NOWAIT — fail-fast

PG-L-030:

SELECT * FROM order_doc WHERE id = ? FOR UPDATE NOWAIT;

Если строка занята — error, не ждёт. Полезно для API-endpoints («попробуйте через минуту»).

jOOQ

PG-L-040..041:

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

// FOR UPDATE SKIP LOCKED
List<TaskRecord> tasks = ctx
    .selectFrom(TASK_QUEUE)
    .where(TASK_QUEUE.STATUS.eq("PENDING"))
    .orderBy(TASK_QUEUE.CREATED_AT)
    .limit(10)
    .forUpdate()
    .skipLocked()
    .fetch();

// FOR UPDATE NOWAIT
.forUpdate().noWait()

// FOR NO KEY UPDATE
.forNoKeyUpdate()

@Transactional обязателен — иначе jOOQ закрывает connection, лок отпускается.

@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();
}

Pessimistic vs Optimistic

PG-L-050..051:

Pessimistic — `SELECT FOR UPDATE`

Берёшь лок заранее.
Простой код.
При высокой конкуренции — очередь TX.

Optimistic — через `version`-колонку

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

-- READ
SELECT id, status, version FROM order_doc WHERE id = ?;

-- UPDATE с проверкой версии
UPDATE order_doc
SET status = 'PAID', version = version + 1
WHERE id = ? AND version = ?;   -- если изменилось — 0 rows affected

jOOQ:

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 + " modified concurrently");
}

Выбор: optimistic для read-heavy (конфликты раз в 1000), pessimistic для money/high-write (десяток конкурентов на строку → optimistic = «10 retry в цикле»).

Advisory locks

PG-L-060..061: блокировка на произвольный bigint, не на строку.

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

-- try-вариант
SELECT pg_try_advisory_xact_lock(12345);   -- true/false

-- два аргумента: namespace + id
SELECT pg_advisory_xact_lock(1001, tenant_id);

Применения:

Запуск scheduled-job в одном экземпляре из кластера.
Глобальная очередь.
Предотвращение двух одновременных миграций.

jOOQ:

@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 {} already running on another instance, skip", jobKey);
    }
}

Deadlock

PG-L-070..072: две TX в разном порядке.

TX1: UPDATE orders WHERE id=1   -- блокирует строку 1
TX2: UPDATE orders WHERE id=2   -- блокирует строку 2
TX1: UPDATE orders WHERE id=2   -- ждёт TX2
TX2: UPDATE orders WHERE id=1   -- ждёт TX1
-- DEADLOCK

PG детектит через deadlock_timeout (default 1s), убивает одну TX с ошибкой 40P01.

Лечится упорядочением:

// ✗ — порядок зависит от ввода
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);

Spring retry на CannotAcquireLockException:

@Retryable(
    retryFor = CannotAcquireLockException.class,
    maxAttempts = 3,
    backoff = @Backoff(delay = 50, multiplier = 2)
)
@Transactional
public void transferMoney(...) { ... }

1-3 retry с back-off обычно решают.

lock_timeout

PG-L-080: критичные операции.

SET LOCAL lock_timeout = '5s';
UPDATE order_doc ... ;

Особенно важно для миграций (ALTER TABLE берёт ACCESS EXCLUSIVE):

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

Без lock_timeout миграция может застрять и за ней все новые запросы (lock queue).

Что запрещено

Антипаттерн	Правило	Что взамен
`SELECT FOR UPDATE` без транзакции	`PG-L-090`	`@Transactional`
`FOR UPDATE` со сложным WHERE без индекса	`PG-L-091`	EXPLAIN, индекс
Длинная TX с локом	`PG-L-092`	лок ближе к UPDATE, отпускать сразу
Pessimistic на каждое чтение	`PG-L-093`	optimistic для read-heavy
Локи в разном порядке	`PG-L-094`	упорядочить по PK
`FOR UPDATE` без `LIMIT` на большой таблице	`PG-L-095`	`LIMIT N` для очередей
Read-modify-write без FOR UPDATE	`PG-L-010`	`FOR UPDATE`
RR/SERIALIZABLE вместо `SELECT FOR UPDATE`	`PG-L-050`	pessimistic явный
Игнорировать `CannotAcquireLockException`	`PG-L-072`	`@Retryable`
`ALTER TABLE` без `lock_timeout`	`PG-L-080`	`SET LOCAL lock_timeout`

Куда дальше

PG → Блокировки — нормативные формулировки.
Уровни изоляции и аномалии — когда FOR UPDATE недостаточно.
Spring @Transactional — propagation и isolation.
Миграции без даунтайма — lock_timeout для ALTER.
Kafka → outbox publishing — FOR UPDATE SKIP LOCKED.
Distributed → idempotency — связь с advisory locks.