Як побудувати надійну черга завдань на PostgreSQL: конкурентність, збої воркерів і масштабування

Майже кожен програмний продукт рано чи пізно потребує фонової обробки задач: розсилку листів, обробку платежів, синхронізацію даних чи генерацію звітів. Інженерам доводиться або підтримувати вже готову систему такого типу, або будувати власну.

PostgreSQL дозволяє зібрати чергу завдань напрочуд швидко: створюєте таблицю з задачами, воркери опитують її на наявність задач у статусі очікування, оновлюють статус — і система формально працює.

Проблеми починаються тоді, коли додаються нові воркери, один з них падає посередині обробки задачі або таблиця розростається до великих обсягів — саме тоді керувати системою стає значно складніше.

У матеріалі автор поетапно розбирає найпростішу реалізацію черги та показує, як кожен новий тип збою підказує, що потрібно додати далі. Відштовхуємося від базової таблиці задач:

CREATE TABLE queue_jobs (
  id SERIAL PRIMARY KEY,
  queue TEXT NOT NULL,
  payload JSONB NOT NULL,
  status TEXT NOT NULL DEFAULT 'pending',
  created_at TIMESTAMP DEFAULT NOW(),
  ...
);

Далі воркери вибирають задачі з черги та проводять їх через машину станів: pending, processing, completed, failed тощо.

І тут виникає перше запитання: як воркерам у розподіленій системі забирати задачі з черги без гонитви за одним і тим самим записом?

Перша гонитва

Перша race condition з'являється, коли кілька воркерів одночасно намагаються обробити ту саму задачу.

a) Наївний підхід

Найпростіша стратегія — дозволити будь-якому воркеру прочитати той самий рядок зі статусом pending до того, як інший воркер його змінить.

SELECT * FROM queue_jobs WHERE status = 'pending' ORDER BY id LIMIT 1;

Якщо запустити цю наївну стратегію та подивитися на трасування подій, виявляється: з чотирма воркерами та черги з 8 задач у стані очікування всі воркери в результаті беруться за одну й ту саму задачу одночасно, що дає 24 дублювані виконання.

Total duplicate executions: (no. of workers - 1) * jobs

b) Стратегія з блокуванням

SELECT * FROM queue_jobs WHERE status = 'pending' ORDER BY id LIMIT 1 FOR UPDATE;

Дублювання зникає, але тепер воркери шикуються в черзі позаду одного заблокованого рядка. Обмеження в тому, що конкурентні транзакції захоплення можуть чекати на той самий рядок: воркер B може простоювати, поки воркер A завершує свою транзакцію захоплення, навіть коли існують інші вільні задачі.

t=0 Queue initialized with 8 pending jobs.
t=1 Worker A locks J1 inside its short claim transaction. Workers B, C, D wait on that row.
t=2 Worker A commits J1: pending → running. The row lock is released; handler processing continues outside the transaction.
...
t=17 Worker D commits J8: pending → running. The row lock is released; handler processing continues outside the transaction.
t=18 4 handlers complete. Their row locks had already been released when each claim transaction committed.
t=18 All jobs are complete.

c) Стратегія Skip Locked

SELECT * FROM queue_jobs WHERE status = 'pending' ORDER BY id FOR UPDATE SKIP LOCKED LIMIT 1;

У цьому варіанті всі 8 задач обробляються воркерами без подвійних захоплень і без зайвої роботи.

t=0 Queue initialized with 8 pending jobs.
t=1 4 workers claim 4 different rows. Locked rows are skipped instead of waited on.
t=2 4 jobs complete in parallel.
t=3 4 workers claim 4 different rows. Locked rows are skipped instead of waited on.
t=4 4 jobs complete in parallel.
t=4 All jobs are complete.

Важливо, що блокування тримається лише під час захоплення задачі, а не під час її обробки. Саме тому перехід зі статусу pending у running і саме блокування мають відбуватися в межах однієї короткої транзакції, після коміту якої рядок розблоковується, а обробник вже виконує роботу поза межами транзакції.

Автор провів навантажувальний тест на 100 тисяч задач, і система впоралася без жодного дубльованого захоплення та без жодної втраченої задачі. Результати тесту доступні у повній версії матеріалу.

Якщо відійти від самої черги й подумати про можливі збої воркерів у розподіленій системі, перше, що спадає на думку — раптове завершення роботи воркера або втрата з ним зв'язку.

Коли воркер падає

Якщо воркер помирає, потрібен механізм, який це виявить і коректно обробить — інакше задача може застигнути у статусі running назавжди. Оскільки збої неминучі, треба продумати стратегію відновлення.

Впровадження лізингу (lease)

Лізинг — це механізм, за яким воркер захоплює задачу і тримає її протягом визначеного часу. Якщо воркер помирає до завершення лізингу, після його закінчення задачу може перехопити інший воркер, і система знову робить її доступною.

CREATE TABLE queue_job_state (
  job_id INT PRIMARY KEY,
  locked_by TEXT,
  locked_until TIMESTAMP,
  ...
);

На цьому етапі стан черги виносять в окрему таблицю, щоб відслідковувати статус кожної задачі незалежно. Пізніше автор поясняє, чому в підсумку використано три таблиці; тут перелік стовпців скорочено для стислості, а повну схему можна подивитися в оригінальному матеріалі.

Демонстрація показує, як черга відновилася після критичного збою воркера й завершила задачу з другої спроби: у реальному тесті воркера A вбивали командою SIGKILL, його лізинг спливав, задачу перехоплював і завершував воркер B.

Проте в механізмі закінчення лізингу залишається пролом: гіпотетично воркер A може “прокинутися” вже після того, як його роботу забрав воркер B. Це може призвести до того, що воркер A перезапише оновлення статусу воркера B за принципом «останній запис виграє».

Токен фенсингу

У демонстрації без токена фенсингу видно, як воркер A прокидається після того, як воркер B уже перехопив задачу. Якщо ж увімкнути токени фенсингу, стає видно, як вони блокують перезапис статусу воркером A.

Основна ідея — стовпець lease_version, який дозволяє виявляти застарілі записи. Коли воркер A прокидається вже після того, як воркер B перехопив задачу, його версія лізингу не збігається з поточним поколінням володіння, тож умовний UPDATE не зачіпає жодного рядка, і застарілий запис просто відхиляється.

CREATE TABLE queue_state (
  lease_version INTEGER NOT NULL DEFAULT 0 CHECK (lease_version >= 0),
);
...

Ідемпотентність — необхідна умова

Токен фенсингу захищає лише від зміни внутрішнього стану колишнім власником після передачі задачі іншому воркеру. Він не рятує від зовнішніх побічних ефектів — наприклад, повторного платежу чи повторного надсилання листа. Уявіть: воркер A встиг відправити лист-підтвердження перед смертю, а воркер B, перехопивши задачу, надсилає його ще раз. У випадку платежів це може обернутися ще гірше. Оскільки обробники задач часто звертаються до зовнішніх сервісів, якими ми не керуємо, потрібно обов'язково робити обробники листів, платежів і вебхуків ідемпотентними.

Повторні спроби можуть перевантажити систему

Повторні спроби — простий, але дієвий спосіб боротьби з тимчасовими збоями. Проте наївне їх застосування може призвести до того, що система витрачатиме всі ресурси на нескінченні невдалі повторення. У прикладі з листами: якщо сервіс розсилки тимчасово недоступний, необмежені повторні спроби можуть перетворитися на справжній шторм запитів, який споживає ресурси.

Повторна спроба з фіксованою затримкою — простий підхід, коли система пробує знову через постійний проміжок часу. У нашому прикладі задачу переносять на майбутню дату й повторюють після заданої затримки.

Проблема фіксованої затримки в тому, що вона не враховує причину збою. Якщо сервіс розсилки недоступний, система продовжуватиме повторювати спроби з тим самим інтервалом, що знову призводить до шторму запитів. Кращий підхід — експоненційна затримка з джитером (розсіюванням), коли система рознесе повторні спроби в часі, щоб не перевантажувати сервіс.

У наведеному прикладі фіксована затримка давала пік 1000 повторних спроб за одну секунду, тоді як повний джитер розсіював навантаження до піку в 58 спроб і розтягував їх приблизно на 89 секунд замість двох.

Незмінне навантаження варто тримати окремо від «гарячого» стану

У підсумковій системі використано три таблиці: queue_jobs, queue_attempts та queue_job_state. Залежно від масштабу можна обійтися й однією таблицею, але тут навмисно розділено незмінне навантаження (payload) і «гарячий» стан. Головна причина — змішування високочастотних оновлень статусу з незмінним навантаженням у межах однієї таблиці може негативно вплинути на продуктивність, про що йтиметься далі. А поки — розподіл ролей:

• queue_jobs — переважно незмінна таблиця, зберігає оригінальне навантаження задачі
• queue_job_state — зберігає статуси задач і переходи між станами
• queue_attempts — зберігає історію спроб виконання

Загалом трьохтабличний дизайн дає два ключові переваги:

• Зменшення розбухання таблиць: розділення великого незмінного навантаження і часто оновлюваного стану скорочує ширину «гарячих» рядків, знижує churn у купі даних, тиск на кеш, обсяг роботи vacuum і розмір індексів, які використовують воркери. Оскільки навантаження задачі може бути великим, оновлення лише поля статусу впливає на ефективність використання пам'яті.
• Прискорення сканування індексів: воркерам постійно потрібно скановати стан черги, щоб знайти задачі в очікуванні. Завдяки виділеній таблиці queue_job_state індекси, якими користуються воркери, можуть залишатися компактнішими й зосередженими саме на «гарячому» стані захоплення, що підвищує ефективність запитів захоплення.

Розбір шляху захоплення задачі через індекс

Шлях захоплення задачі через індекс — критичний для всієї системи, тож автор дослідив його окремо. Під час тестування конкурентності, повторних спроб, застарілих воркерів і збоїв виникло питання: скільки коштує воркерам захоплення задач у міру зростання таблиці? Для відповіді запускалися повні мутуючі запити з EXPLAIN(ANALYZE, BUFFERS) на дедалі більших наборах даних. Результати виявилися неочікуваними.

Ось скільки коштувало знайти й захопити наступні 20 задач у міру зростання черги за тестом runClaimPlanExperiment:

- 1K pending → 1.58 ms
- 10K pending → 5.11 ms
- 100K pending → 53.57 ms
- 1M total / 100K pending → 51.85 ms

function getClaimExplainSql(): string {
  return `
    EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON)
    WITH candidates AS (
      SELECT state.job_id
      FROM queue_job_state AS state
      WHERE state.queue = ANY($1::text[])
        AND state.status = 'pending'
        AND state.run_at

Причина проблеми — не в SKIP LOCKED і не в тому, що PostgreSQL нехтував частковим індексом queue_job_state_claim_idx. Проблема в тому, що при 1 мільйоні рядків загалом і 100 тисячах у стані очікування план виконання проходив через bitmap Index Scan, який живив Bitmap Heap Scan, а потім сортував усі знайдені кандидати, перш ніж застосувати LIMIT 20.

# Очікувана поведінка
ordered index
  ↓
J1 J2 J3 ... J20
  ↓
STOP

# Реальна поведінка
100K matching rows
  ↓
sort
  ↓
LIMIT 20

Запит справді використовував queue_job_state_claim_idx, але не як упорядкований top-N скан індексу, як планувалося. Тож автор поставив контрольований експеримент, змінивши умову:

state.queue = ANY($1::text[])

state.queue = $1::text

Заміна умови рівності скоротила час виконання запиту з 51,85 мс до 5,307 мс — приблизно у 9,8 раза швидше. Версія з рівністю дійсно застосовувала впорядкований Index Scan на queue_job_state_claim_idx.

{
  "dataset": "1m-total-100k-pending",
  "predicate": "state.queue = $1::text",
  "planningTimeMs": 3.953,
  "executionTimeMs": 5.307,
  "claimIndexUsed": true,
  "claimScanNodeType": "Index Scan",
  "indexesUsed": [
    "queue_job_state_claim_idx",
    "queue_job_state_pkey",
    "queue_jobs_id_queue_unique"
  ]
}

Найдорожчою частиною виявився саме етап відбору кандидатів у запиті захоплення задачі, який зводиться до:

SELECT state.job_id
FROM queue_job_state AS state
WHERE state.queue = ANY($1::text[])
  AND state.status = 'pending'
  AND state.run_at

Найцікавіше — саме падіння часу на етапі відбору кандидатів: 0,323 мс проти 50,281 мс, тобто приблизно у 155 разів швидше. Ця цифра пояснює, чому версія з умовою рівності настільки швидша.

Щоб зрозуміти, як це впливає на прикладний код, варто пригадати, що індекс фактично впорядкований за:

queue priority DESC run_at job_id

Якщо PostgreSQL знає, що queue = 'payments', він може відразу перейти до однієї безперервної секції впорядкованого індексу і йти прямо до перших 20 придатних рядків. Натомість умова

queue = ANY('{payments,email,reports}')

змушує PostgreSQL сканувати кілька діапазонів індексу для кожного з трьох значень queue.

Оптимізатор не може застосувати той самий простий упорядкований top-N обхід. У вимірюваному випадку це змушувало його зібрати набагато більший набір кандидатів і відсортувати його.

На основі цього спостереження команда змінила API воркерів. Спочатку воно виглядало так:

claimJobs({
  queues: ["payments", "email", "reports"],
});

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

claimJobs({
  queue: "payments",
});

Той самий ефект для воркера, що обробляє кілька черг, можна отримати на рівні застосунку — наприклад, через round-robin або зважений розподіл:

for (const queueName of queues) {
  const jobs = await queue.claimJobs({
    queue: queueName,
    limit: availableCapacity,
  });
}

Ключове архітектурне рішення: кожна операція захоплення задачі в базі даних має націлюватися на одну назву черги за раз. Бенчмарк також показав, наскільки важлива форма предикату — сама наявність індексу в схемі не гарантує, що PostgreSQL зможе застосувати його саме для того шаблону доступу, який мали на увазі розробники.

Складання воркера в єдину систему

Тепер, знаючи всі складові, можна зібрати їх разом. Сама логіка воркера — найменш цікава частина всієї системи черги: це код, який просто обробляє задачі. Загалом послідовність виглядає так:

• Воркер запускається
• Обирає черзу для обробки
• Захоплює пакет задач через FOR UPDATE SKIP LOCKED
• Встановлює лізинг і збільшує версію лізингу
• Обробляє задачі з обмеженою конкурентністю
• Надсилає heartbeat під час довгих обчислень
• У разі успіху завершує задачу з використанням токенів фенсингу
• У разі помилки, яку можна повторити, застосовує експоненційну затримку з джитером
• У разі помилки, яку не можна повторити, позначає задачу як мертву

Висновок

PostgreSQL надає майже всі примітиви, необхідні для побудови надійної черги: транзакції, блокування на рівні рядків, стійке зберігання даних та індекси. Проте, як показали розглянуті сценарії збоїв, керувати такою системою в цілому може бути непросто.

Автор наголошує, що замість власної реалізації можна обрати SQS, черги на основі Redis, Kafka чи готову бібліотеку черг для PostgreSQL. Але розуміння цих сценаріїв збоїв робить такі готові системи набагато прозорішими та допомагає розпізнати, коли черга, що начебто працює, насправді може виявитися ненадійною.

Джерело: Master.dev Blog