Урок №8. Пакетное резервирование задач и квоты приоритетов

Что изучаем: как за одну короткую транзакцию зарезервировать несколько задач, распределить места между HIGH, MEDIUM и LOW и не потерять свободные места, если одной из категорий нет.

В уроке №7 мы выбирали по одной задаче. Теперь представим воркер, который может взять сразу четыре. Как сделать так, чтобы HIGH сохраняли преимущество, но MEDIUM и LOW тоже могли попадать в пачку?

1. Проблема: почему одного LIMIT недостаточно

Допустим, в очереди десять HIGH, шесть MEDIUM и четыре LOW. Запрос со строгим приоритетом:

SELECT id
FROM backend_training.lesson08_tasks
WHERE status = 'pending'
ORDER BY priority DESC, created_at ASC, id ASC
FOR UPDATE SKIP LOCKED
LIMIT 4;

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

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

Ожидаемый состав: две HIGH, одна MEDIUM и одна LOW. Квоты — правило нашего приложения, а не встроенный планировщик PostgreSQL.

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

2. Подготовка лабораторной

Скачайте ZIP с материалами урока №8. Мы продолжим работать с существующей БД backend_training_lab, в схеме backend_training. Лабораторная использует отдельную таблицу lesson08_tasks и не затрагивает таблицы прошлых уроков.

В терминале перейдите в каталог lesson08_batch_quotas:

pip install -r requirements.txt
psql -d backend_training_lab -f schema.sql

Настройте переменную DATABASE_URL так же, как в предыдущих уроках. При необходимости пример есть в .env.example, но этот файл не загружается автоматически.

export DATABASE_URL='postgresql+psycopg://USER:PASSWORD@localhost:5432/backend_training_lab'
python -m lesson08.demo batch

Все опыты показывают четыре этапа: ДЕЙСТВИЕ → ОЖИДАЕМ → ПОЛУЧИЛИ → ПОЧЕМУ. Сначала сформулируйте свой прогноз, затем запустите эксперимент и сравните.

3. Эксперимент №1 — резервируем пачку

Ситуация. Есть пять задач: LOW #42 и #46, HIGH #43 и #45, MEDIUM #44. Worker A может зарезервировать не больше трёх задач за раз.

Ваша гипотеза: какие три ID окажутся в processing после успешной транзакции? Будет ли более старая LOW #42 выбрана раньше HIGH?

python -m lesson08.demo batch

Строгий приоритет: [43, 45, 44]
ВСЯ ПАЧКА: [43, 45, 44] (всего 3)

4. Эксперимент №2 — распределяем места по квотам

Ситуация. В очереди HIGH #101, #102, #103; MEDIUM #201, #202; LOW #301, #302. Пачка рассчитана на четыре места, квоты — 2/1/1.

Ваша гипотеза: займёт ли HIGH #103 место, предназначенное LOW, если LOW-задачи доступны?

python -m lesson08.demo quotas

Квота HIGH   (2 места): [101, 102]
Квота MEDIUM (1 место): [201]
Квота LOW    (1 место): [301]
Fallback (свободные места): []
ВСЯ ПАЧКА: [101, 102, 201, 301] (всего 4)

5. Эксперимент №3 — fallback: что делать с пустой квотой?

Ситуация. Доступны HIGH #101–104, MEDIUM #201, а LOW-задач нет. Квоты по-прежнему 2/1/1.

Ваша гипотеза: сколько задач вернёт воркер — три или четыре? Какая категория займёт пустующее место?

python -m lesson08.demo fallback

Квота HIGH   (2 места): [101, 102]
Квота MEDIUM (1 место): [201]
Квота LOW    (1 место): []
Fallback (свободные места): [103]
ВСЯ ПАЧКА: [101, 102, 201, 103] (всего 4)

Рис. 1. Формирование пачки задач: заполнение квот и перераспределение свободных мест через fallback.

Нажмите на изображение, чтобы увеличить.

✕ Закрыть

6. Как это устроено в коде

Посмотрите файл lesson08/repository.py. Для резервирования используются два запроса: CLAIM_PRIORITY_SQL берёт задачи конкретной категории, а CLAIM_ANY_SQL выбирает любую доступную задачу по строгому приоритету — для первого опыта и fallback.

Основной SQL-запрос для одной категории:

WITH selected AS (
    SELECT id
    FROM backend_training.lesson08_tasks
    WHERE status = 'pending'
      AND priority = :priority
    ORDER BY created_at ASC, id ASC
    LIMIT :limit
    FOR UPDATE SKIP LOCKED
)
UPDATE backend_training.lesson08_tasks AS task
SET status = 'processing',
    attempts = task.attempts + 1,
    lease_until = clock_timestamp()
        + (:lease_seconds * INTERVAL '1 second'),
    claimed_by = :worker_name
FROM selected
WHERE task.id = selected.id
RETURNING task.id, task.priority,
          task.attempts, task.created_at;

Разбор: подзапрос selected находит не более :limit подходящих строк и блокирует их. SKIP LOCKED пропускает строки, занятые другой транзакцией. Затем UPDATE переводит выбранные задачи в processing, увеличивает attempts и записывает владельца и срок аренды (lease_until). RETURNING отдаёт данные обновлённых строк.

В актуальной версии лабораторной мы не повторяем три одинаковых вызова Python вручную. Квоты перебираются циклом:

claims_by_priority = {}

for priority in sorted(quotas, reverse=True):
    claims_by_priority[priority] = _claim_rows(
        connection,
        worker_name=worker_name,
        priority=priority,
        limit=quotas[priority],
        lease_seconds=lease_seconds,
    )

missing = batch_size - sum(
    len(claims) for claims in claims_by_priority.values()
)

extra = _claim_rows(
    connection,
    worker_name=worker_name,
    limit=missing,
    lease_seconds=lease_seconds,
)

Сортировка sorted(quotas, reverse=True) перебирает приоритеты от HIGH к LOW. Переменная missing показывает, сколько мест не удалось заполнить по квотам. Вызов без параметра priority запускает fallback. Если missing=0, вспомогательная функция ничего не запрашивает.

Обратите внимание: цикл убирает дублирование Python-кода, но не превращает несколько SQL-операций в одну. Это осознанный выбор ради ясности. Все вызовы выполняются в рамках одной транзакции.

7. Эксперимент №4 — два воркера и откат целой пачки

Ситуация. В очереди ровно восемь задач: четыре HIGH, две MEDIUM и две LOW. Worker A и Worker B работают через разные соединения с PostgreSQL.

Перед запуском ответьте себе: если Worker A зарезервировал четыре задачи, но ещё не сделал COMMIT, сможет ли Worker B взять другую четвёрку? Что произойдёт с первой пачкой после ROLLBACK Worker A?

python -m lesson08.demo concurrent

Заметьте: в этом сценарии не требуется запускать два процесса Python параллельно. Достаточно двух открытых соединений и перекрывающихся по времени транзакций. Это именно проверка конкурентного доступа на уровне PostgreSQL.

8. Где проходит граница транзакции?

with engine.begin() as connection:
    batch = claim_batch_in_transaction(
        connection,
        worker_name="Worker A",
        batch_size=4,
        quotas={3: 2, 2: 1, 1: 1},
    )
# Здесь COMMIT уже выполнен.
# Только теперь начинаем внешнюю обработку.

engine.begin() фиксирует изменения при нормальном завершении блока и откатывает их при исключении. Поэтому несколько SQL-запросов могут образовывать одну атомарную операцию резервирования пачки.

Мы не держим транзакцию открытой во время вызова внешнего API. Сначала резервируем и фиксируем пачку, затем обрабатываем задачи. Если воркер упадёт после COMMIT, обычный ROLLBACK уже ничего не вернёт — для восстановления понадобится механизм lease из урока №4.

Резервирование без пересечения пачек не означает exactly-once обработку. Внешний API может получить повторный запрос при сбоях и повторных попытках — эту проблему мы разбирали в уроке №6.

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

• Квоты работают на одну пачку. Они не гарантируют точные доли обработанных задач за минуту или час среди нескольких воркеров.
• SKIP LOCKED пропускает занятые строки. Фактический состав пачки зависит от текущих блокировок; это подходит для очередей, но не даёт согласованного снимка всех доступных задач.
• Короткая транзакция важна. Чем дольше мы резервируем пачку, тем дольше удерживаем блокировки строк.
• Несколько запросов — компромисс. Цикл понятнее, но при большом числе приоритетов создаёт больше обращений к БД. Оптимизацию следует обосновывать измерениями.
• Квоты не заменяют другие правила. Здесь нет глобального rate limit, SLA, aging и гарантии минимального времени ожидания.

10. Проверьте понимание

1. Почему fallback выполняется после попыток заполнить все квоты, а не сразу после HIGH?

2. Почему после ROLLBACK Worker A значение attempts у его задач снова равно нулю?

3. Гарантирует ли квота LOW=1, что любая LOW-задача будет обработана за определённое время? Объясните своё решение.

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

Источники и материалы

• Е. Рогов, «PostgreSQL 16 изнутри»: глава 8, раздел о пакетной обработке; глава 13, раздел 13.5 «Блокировка без ожидания».
• PostgreSQL 16 — SELECT, FOR UPDATE и SKIP LOCKED.
• PostgreSQL 16 — UPDATE и RETURNING.
• SQLAlchemy 2.0 — транзакции и engine.begin().