1. Проблема: задача зарезервирована, но не выполнена

В предыдущем уроке мы реализовали Python-воркер, который резервирует задачу в PostgreSQL, выполняет её и отмечает завершение.

task = claim_task()

if task:
    process_task(task)
    mark_as_completed(task["id"])

Но мы обнаружили проблему: что произойдёт, если Python-процесс завершится после успешного резервирования, но до завершения задачи?

Представим последовательность событий:

1. Worker A резервирует задачу #42.
2. PostgreSQL фиксирует статус processing.
3. Worker A аварийно завершается.
4. Другой воркер ищет задачи со статусом pending.

Задача #42 остаётся в processing. Её транзакция уже завершилась, блокировка освобождена, но новый воркер не выбирает эту строку.

Проблема: без отдельного механизма восстановления задача может остаться необработанной.

В этом уроке мы научимся возвращать такие задачи в обработку и защищать PostgreSQL от изменений, выполняемых устаревшим воркером.

2. Почему нельзя просто сбрасывать processing в pending?

На первый взгляд достаточно периодически выполнять:

UPDATE backend_training.tasks
SET status = 'pending'
WHERE status = 'processing';

Но это опасное решение.

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

Если вернуть такую задачу в pending, другой воркер сможет получить её повторно. В результате два воркера могут одновременно выполнять одну логическую операцию.

Нам нужен способ отличать актуальное резервирование от потенциально устаревшего.

Что такое lease?

Lease — ограниченное по времени право воркера обрабатывать задачу.

Например, Worker A резервирует задачу #42 на 10 секунд.

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

Но здесь есть важное ограничение: истечение lease не доказывает, что старый воркер перестал работать.

Поэтому одного срока аренды недостаточно. Нам понадобится ещё и идентификатор конкретной попытки обработки.

3. Расширяем знакомую таблицу tasks

Мы продолжим использовать таблицу backend_training.tasks из предыдущих уроков.

Добавим два столбца:

attempts будет увеличиваться при каждом новом резервировании, даже если задача уже имела статус processing.

Сохраняем прежние статусы: pending, processing, completed.

3.1. Подготовьте лабораторный проект

Скачайте прикреплённый ZIP-архив четвёртого урока. Распакуйте его в отдельную папку, не заменяя файлы предыдущей лабораторной.

lesson_04/
├── README_guide.txt
├── 01_migrate.sql
├── 02_reset.sql
├── requirements.txt
├── db.py
├── models.py
├── repository.py
└── worker.py

Установите зависимости, следуя README_guide.txt.

python3 -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements.txt

Настройте переменную окружения DATABASE_URL для своей учебной базы.

export DATABASE_URL='postgresql+psycopg://USER:PASSWORD@localhost:5432/backend_training_lab'

Замените USER и PASSWORD реквизитами вашей базы. При необходимости используйте URL-кодирование специальных символов пароля.

3.2. Выполните миграцию

Убедитесь, что все учебные воркеры остановлены, а открытые транзакции завершены.

Откройте 01_migrate.sql в DBeaver и выполните его в базе backend_training_lab.

Миграция добавляет:

ALTER TABLE backend_training.tasks
    ADD COLUMN IF NOT EXISTS lease_until timestamptz;

ALTER TABLE backend_training.tasks
    ADD COLUMN IF NOT EXISTS attempts integer
    NOT NULL DEFAULT 0;

timestamptz позволяет хранить момент времени с учётом временных зон.

attempts изначально равен нулю, потому что задача ещё не резервировалась.

Важно: выполняйте полный файл 01_migrate.sql, а не только приведённый фрагмент. В нём также предусмотрена проверка имени учебной базы.

3.3. Восстановите исходные данные

Выполните 02_reset.sql.

Внимание: этот файл удаляет все строки учебной таблицы и заново создаёт три задачи. Не используйте его в рабочей БД.

Проверьте результат:

SELECT id, type, status, attempts, lease_until
FROM backend_training.tasks
ORDER BY id;

4. Обновляем ORM-модель Task

В файле models.py появились два новых атрибута.

lease_until: Mapped[datetime | None] = mapped_column(
    DateTime(timezone=True),
    nullable=True,
)

attempts: Mapped[int] = mapped_column(
    Integer,
    nullable=False,
    default=0,
)

lease_until может быть None, если активной аренды нет.

attempts содержит номер последнего резервирования задачи.

Остальные поля и подключение к PostgreSQL остаются знакомыми из предыдущего урока.

5. Обновляем claim_task()

Раньше claim_task() выбирала только задачи в состоянии pending.

Теперь она должна также находить задачи со статусом processing, у которых истёк срок аренды.

5.1. Новое условие выборки

stmt = (
    select(Task)
    .where(
        or_(
            Task.status == "pending",
            and_(
                Task.status == "processing",
                Task.lease_until <= func.clock_timestamp(),
            ),
        )
    )
    .order_by(Task.id)
    .limit(1)
    .with_for_update(skip_locked=True)
)

Теперь подходят две категории задач:

• Новые задачи со статусом pending.
• Задачи со статусом processing, у которых lease_until уже наступил.

FOR UPDATE SKIP LOCKED продолжает защищать конкурентное резервирование на уровне блокировок PostgreSQL.

При этом ORDER BY id задаёт порядок кандидатов, но не гарантирует строгий порядок выполнения задач при конкуренции.

5.2. Обновляем данные при резервировании

После получения строковой блокировки функция изменяет три значения:

task.status = "processing"
task.attempts += 1

task.lease_until = session.scalar(
    text(
        "SELECT clock_timestamp() "
        "+ (:seconds * INTERVAL '1 second')"
    ),
    {"seconds": lease_seconds},
)

При каждом резервировании номер попытки увеличивается на единицу.

Например, первое резервирование задачи #42 создаёт попытку №1.

После истечения аренды повторное резервирование создаёт попытку №2.

5.3. Что возвращает функция?

claim = {
    "id": task.id,
    "type": task.type,
    "attempts": task.attempts,
    "lease_until": task.lease_until,
}

Теперь воркер получает не только данные задачи, но и номер своей попытки.

Этот номер потребуется, когда воркер попытается отметить задачу как выполненную.

5.4. Когда выполняется COMMIT?

Как и в третьем уроке, резервирование происходит внутри:

with SessionFactory.begin() as session:
    ...
    session.flush()

session.flush() отправляет изменения в PostgreSQL, но не фиксирует транзакцию.

COMMIT выполняется при успешном выходе из блока SessionFactory.begin().

Таким образом, изменение статуса, увеличение attempts и установка lease_until фиксируются одной транзакцией.

6. Защищаем завершение задачи от старого воркера

Предположим, Worker A зарезервировал задачу #42 с attempts = 1.

Затем его lease истёк, и Worker B повторно зарезервировал ту же задачу с attempts = 2.

Но Worker A не завершился. Он продолжил работу и пытается выполнить:

mark_as_completed(42)

Если функция проверяет только идентификатор и статус задачи, старый воркер может вмешаться в состояние более новой попытки.

Поэтому мы изменим сигнатуру функции:

mark_as_completed(task_id, attempts)

Новая версия проверяет не только задачу, но и номер актуальной попытки.

6.1. Реализация mark_as_completed()

Рассмотрим функцию из repository.py:

class StaleClaimError(RuntimeError):
    """Попытка больше не вправе завершить задачу."""


def mark_as_completed(
    task_id: int,
    attempts: int,
) -> None:

    if attempts < 1:
        raise ValueError(
            "Номер попытки должен быть положительным"
        )

    with SessionFactory.begin() as session:
        stmt = (
            update(Task)
            .where(
                Task.id == task_id,
                Task.status == "processing",
                Task.attempts == attempts,
                Task.lease_until > func.clock_timestamp(),
            )
            .values(
                status="completed",
                lease_until=None,
            )
            .returning(Task.id)
        )

        updated_id = session.scalar(stmt)

        if updated_id is None:
            raise StaleClaimError(
                f"Невозможно завершить задачу #{task_id} "
                f"попыткой {attempts}: "
                "задача уже завершена, lease истёк "
                "или начата новая попытка"
            )

Здесь используются четыре условия.

Только если все условия выполнены, PostgreSQL изменяет статус задачи на completed.

Если подходящей строки нет, UPDATE ... RETURNING не возвращает идентификатор.

Тогда приложение выбрасывает StaleClaimError.

Важный момент: проверка номера попытки и изменение статуса выполняются одним SQL-оператором.

Это защищает переход состояния от гонки, которая возникла бы при отдельном SELECT для проверки и последующем безусловном UPDATE.

7. Обновлённый Python-воркер

Основная последовательность обработки осталась практически неизменной:

task = claim_task(lease_seconds=10)

if task:
    process_task(task)

    mark_as_completed(
        task["id"],
        task["attempts"],
    )

Теперь claim_task() возвращает номер попытки, а mark_as_completed() использует его для проверки актуальности.

В файле worker.py появились два учебных режима.

Отличие от третьего урока: ранее режим --hold удерживал открытую транзакцию.

Теперь --hold-after-claim приостанавливает Python-воркер уже после успешного COMMIT.

Это позволяет исследовать не строковую блокировку, а истечение срока аренды при всё ещё работающем процессе.

8. Практическая работа

Проведём три эксперимента.

Перед каждым экспериментом обязательно завершайте предыдущие воркеры и открытые учебные транзакции, а затем выполняйте 02_reset.sql, если инструкция требует сброса.

Эксперимент №1. Сбой после COMMIT

Цель: проверить, что аварийное завершение Python-процесса не отменяет уже зафиксированное резервирование.

Выполните 02_reset.sql.

Затем запустите:

python worker.py A --crash-after-claim --lease-seconds 10

Воркер зарезервирует задачу #42 и намеренно завершится с кодом 17.

Перед проверкой предположите: какой статус будет у #42 сразу после сбоя? Отменится ли ранее выполненный COMMIT?

Сразу после завершения процесса выполните в DBeaver:

SELECT
    id,
    status,
    attempts,
    lease_until,
    clock_timestamp() >= lease_until AS lease_expired
FROM backend_training.tasks
WHERE id = 42;

Ожидаем:

• status = 'processing';
• attempts = 1;
• lease_until содержит время окончания аренды.

Если проверка выполнена до истечения десяти секунд, lease_expired будет равен false.

После истечения срока он станет true.

Объяснение: claim_task() завершила транзакцию до аварийного выхода Python-процесса.

Поэтому PostgreSQL сохранил статус и номер попытки, а завершение процесса не откатило уже выполненный COMMIT.

Эксперимент №2. Повторное резервирование

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

Продолжайте с состоянием, полученным в эксперименте №1. Сбрасывать таблицу не нужно.

Проверяйте срок аренды:

SELECT
    clock_timestamp() >= lease_until AS expired
FROM backend_training.tasks
WHERE id = 42;

Дождитесь, пока expired станет true.

Перед запуском Worker B ответьте: какую задачу он получит? Каким станет значение attempts?

Запустите:

python worker.py B --lease-seconds 10

Worker B должен повторно зарезервировать задачу #42.

На этот раз attempts будет равен 2.

После успешного выполнения программы проверьте таблицу:

SELECT id, status, attempts, lease_until
FROM backend_training.tasks
ORDER BY id;

Ожидаем:

Объяснение: Worker B выбрал задачу #42, потому что её прежний lease истёк.

При новом резервировании PostgreSQL увеличил номер попытки и установил новый срок аренды.

После завершения обработки Worker B изменил статус задачи на completed.

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

Эксперимент №3. Устаревший Worker A

Цель: проверить, что воркер со старым номером попытки не может завершить повторно зарезервированную задачу.

Завершите предыдущие процессы и выполните 02_reset.sql.

Откройте два терминала.

Терминал A:

python worker.py A --hold-after-claim --lease-seconds 10

Worker A зарезервирует задачу #42 с attempts = 1 и остановится в ожидании Enter.

Не нажимайте Enter.

В отличие от лабораторной третьего урока, здесь транзакция резервирования уже завершена через COMMIT.

Дождитесь истечения аренды:

SELECT
    id,
    attempts,
    clock_timestamp() >= lease_until AS expired
FROM backend_training.tasks
WHERE id = 42;

Когда expired = true, перейдите ко второму терминалу.

Терминал B:

python worker.py B --lease-seconds 10

Worker B повторно зарезервирует задачу #42 с attempts = 2.

Дождитесь успешного завершения Worker B.

Перед продолжением Worker A предположите: сможет ли он изменить статус #42 с номером попытки 1?

Вернитесь в терминал A и нажмите Enter.

Worker A выполнит имитацию обработки, после чего попытается вызвать:

mark_as_completed(42, 1)

Но эта попытка больше не является актуальной.

Ожидается сообщение вида:

[A] Старая попытка отклонена:
Невозможно завершить задачу #42 попыткой 1:
задача уже завершена, lease истёк
или начата новая попытка

Программа Worker A завершится с кодом 2.

Теперь проверьте состояние PostgreSQL:

SELECT id, status, attempts, lease_until
FROM backend_training.tasks
WHERE id = 42;

В описанном сценарии задача должна иметь статус completed и attempts = 2.

Worker A не смог изменить её состояние.

Объяснение: старый воркер сохранил номер попытки 1, а в PostgreSQL актуальной уже является попытка 2.

Поэтому условие Task.attempts == attempts для Worker A не выполняется.

Обратите внимание: в данном эксперименте задача уже завершена Worker B. Поэтому отказ также связан с тем, что её статус больше не равен processing.

python worker.py A --hold-after-claim --lease-seconds 10

python worker.py B --hold-after-claim --lease-seconds 30

9. Что наша реализация действительно гарантирует?

Мы реализовали две разные защиты.

Защита №1. Конкурентное резервирование

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

Изменение статуса, номера попытки и срока аренды происходит внутри одной транзакции.

Это защищает от конфликтующего резервирования одной строки одновременно двумя транзакциями, использующими наш протокол.

Защита №2. Завершение актуальной попытки

mark_as_completed() проверяет номер попытки и действующий lease.

Если резервирование уже устарело, функция не разрешает отметить задачу как выполненную.

Но существует ещё одна проблема.

Что если старый воркер уже выполнил внешнюю операцию?

В эксперименте №3 Worker A вывел:

Обработка #42 завершилась

Но после этого PostgreSQL отклонил изменение статуса задачи.

Если бы вместо time.sleep(1) выполнялся реальный HTTP-запрос, внешний сервис мог бы уже создать документ, отправить уведомление или провести операцию.

При этом Worker B мог бы выполнить такой же запрос повторно.

Итог: проверка номера попытки защищает состояние нашей базы данных, но сама по себе не предотвращает повторные внешние побочные эффекты.

Для этого нужны дополнительные механизмы, например идемпотентность внешней операции.

10. Ограничения решения

Мы создали работающий учебный механизм восстановления задач, но он ещё не является полноценной production-очередью.

Дополнительно нужно учитывать, что SKIP LOCKED может пропускать строки с конфликтующими блокировками, а ORDER BY id не обеспечивает строгую очередность завершения задач.

Также текущая миграция рассчитана на учебную таблицу. Если обновлять уже работающую систему, необходимо отдельно продумать состояние существующих задач processing, у которых lease_until ещё равен NULL.

В нашей лабораторной эта ситуация устраняется полным сбросом учебных данных. В production так поступать нельзя.

11. Проверка понимания

Перед итоговым тестом попробуйте самостоятельно ответить на вопросы.

1. Почему истечение lease не означает, что старый воркер завершился?
2. Почему при повторном резервировании необходимо увеличивать attempts?
3. Почему проверку номера попытки нужно выполнять непосредственно в SQL-запросе обновления?
4. Может ли Worker A выполнить внешний HTTP-запрос после того, как его lease истёк?
5. Что произойдёт, если обработка задачи длится дольше срока аренды, а продление lease не реализовано?

Для каждого ответа постарайтесь объяснить не только ожидаемый результат, но и конкретный механизм PostgreSQL, который его определяет.

12. Итоги урока

Мы расширили знакомый механизм конкурентной обработки фоновых задач.

Теперь наше приложение умеет:

• Резервировать новые задачи.
• Повторно резервировать задачи с истёкшим lease.
• Увеличивать номер попытки при каждом резервировании.
• Отклонять завершение устаревшей попытки.
• Разделять защиту состояния PostgreSQL и внешних побочных эффектов.

Главный инженерный вывод:

В распределённой обработке недостаточно знать, что задача имеет статус processing.

Нужно также понимать, до какого момента действует резервирование и какая попытка обработки является актуальной.

При этом lease и номер попытки не гарантируют однократное выполнение внешней операции.

Надёжность фоновых обработчиков строится из нескольких независимых механизмов, каждый из которых решает собственную задачу.

Дополнительные материалы

• PostgreSQL 16 — SELECT, FOR UPDATE и SKIP LOCKED
• PostgreSQL 16 — функции времени
• SQLAlchemy 2.x — Session Basics
• SQLAlchemy 2.x — UPDATE и RETURNING

Для дополнительного изучения механизмов строковых блокировок можно обратиться к книге Е. Рогова «PostgreSQL 16 изнутри», раздел 13.5 «Блокировка без ожидания».