Черга задач на MongoDB

Claim через find_one_and_update, лізи з heartbeat, ретраї з backoff і рекламація прострочених задач.

Документ описує, як влаштована черга задач пайплайна gdrive-sync: claim через атомарний find_one_and_update, лізи з heartbeat, ретраї з backoff, рекламація прострочених задач і чому все це працює без окремого брокера.

Основні файли:

packages/gdrive-sync/src/gdrive_sync/state.py — TaskStatus, TaskQueue (claim / heartbeat / release / fail / reclaim) — вся механіка черги.
packages/gdrive-sync/src/gdrive_sync/scanner.py — продюсер: перетворює зміни в Drive на задачі (file_upsert_op), рекламує прострочені лізи.
packages/gdrive-sync/src/gdrive_sync/worker.py — консюмер черги синку.
packages/gdrive-sync/src/gdrive_sync/analysis/worker.py — консюмер черги аналізу.
packages/gdrive-sync/src/gdrive_sync/db.py — колекції та індекси.

1. Загальна ідея

Черга — це звичайні Mongo-колекції, без Redis/RabbitMQ/SQS. Одна й та сама механіка обслуговує дві колекції: gdrive__files (черга синку; _id = Drive file id) і video_analysis__tasks (черга аналізу; _id = sha256 контенту). TaskQueue параметризується колекцією, тож логіка існує в одному місці.

Третя колекція сканера, gdrive__posters, чергою не є: у неї немає ні status, ні attempts, ні ліз, і жоден воркер її не клеймить — воркер синку лише читає з неї постер відео, яке саме обробляє. Саме тому постери живуть окремо, а не прапорцем у gdrive__files: KEDA-тригер масштабує воркерів за буквальним {"status": "pending"} у черзі синку, і документ, який ніколи не стане роботою, підняв би зайвий под.

Чому Mongo, а не брокер:

Задача і стан — один документ. Черга тут не «повідомлення», а поточний стан файлу (статус, спроби, помилка, артефакти). Окремий брокер означав би двофазну синхронізацію двох сховищ.
Атомарний claim безкоштовно. find_one_and_update — атомарна операція: два воркери фізично не можуть забрати одну задачу.
Продюсер ідемпотентний. Сканер робить upsert-и: повторна доставка тієї самої зміни з Drive не породжує дублікат задачі — просто оновлює документ.

2. Документ задачі

Поля, якими оперує черга (решта — метадані файлу):

Поле	Призначення
`status`	`TaskStatus`: `pending`, `processing`, `done`, `failed`, `skipped_too_large`, `skipped_by_content`, `tombstone`
`attempts`	Кількість claim-ів; ліміт — `max_attempts` (дефолт 5)
`lease_until`	Подвійна роль, див. §4
`worker_id`	Хто тримає лізу: `hostname:pid:uuid8`
`error`	Останній текст помилки, обрізаний до 2000 символів з кінця
`updated_at`	Час останньої зміни, завжди `$$NOW` (серверний годинник)

Стани та переходи:

                    ┌──────────────────────────────────────────────┐
                    │                (claim, атомарно)             │
   scanner upsert   ▼                                              │
  ──────────► [pending] ──claim──► [processing] ──finalize──► [done]
                    ▲                    │
                    │                    ├─ transient error ──► pending (backoff-ліза)
   revival          │                    ├─ attempts вичерпані ─► [failed]
  (tombstone/skip)──┘                    ├─ ліза протухла ──────► pending (reclaim сканером)
                                         └─ фатальна помилка ───► [failed]

  окремі гілки сканера/воркера:
    upsert у skip-шляху ──► [skipped_by_content]   (черга не бачить)
    файл завеликий ───────► [skipped_too_large]    (термінальний до ручного втручання)
    видалений у Drive ────► [tombstone]            (TTL-очистка за 30 днів)

3. Claim: атомарне взяття задачі

TaskQueue.claim(worker_id) — серце черги, один find_one_and_update:

{
    "attempts": {"$lt": self.max_attempts},
    "$or": [
        {"status": PENDING, "lease_until": None},  # свіжа задача
        {"status": PENDING, "lease_until": {"$lte": now}},  # backoff минув
        {"status": PROCESSING, "lease_until": {"$lt": now}},  # ліза протухла — можна вкрасти
    ],
}

Update (aggregation pipeline) в тому ж виклику: status=processing, worker_id, нова ліза, attempts + 1, updated_at=$$NOW; ReturnDocument.AFTER повертає вже клеймлений документ.

Ключові наслідки:

Двом воркерам одна задача не дістанеться: find_one_and_update атомарний на рівні документа.
Третя гілка $or — це вбудований self-healing: воркер, що впав з задачею, нічого не мусить прибирати за собою — коли ліза протухне, задачу забере будь-хто (інший воркер напряму або сканер через reclaim, §6).
attempts інкрементується при claim-і, не при помилці: зависла й украдена задача теж «витрачає» спробу, тож отруйна задача не крутиться вічно.
Воркер крутить while task := claim(): process(task) і виходить, коли черга порожня — KEDA скейлить кількість подів за розміром черги, до нуля включно.

4. `lease_until`: одне поле, дві ролі

Для processing — ліза (оренда): «задача моя до цього моменту». Поки ліза жива, інші її не чіпають; протухла — задачу можна забрати.
Для pending — “not before”-мітка: задачу не можна клеймити до цього часу. Так реалізовано retry-backoff (§5) і коротку паузу після revision race без окремих полів чи відкладених черг.

lease_until для запису завжди рахується на сервері через lease_expr(ttl) = $dateAdd від $$NOW — годинник воркера в записі не бере участі (див. §7).

5. Життя під лізою: heartbeat і три способи відпустити задачу

Heartbeat. Обробка файлу може тривати довше за TTL лізи (дефолт 900 с). TaskQueue.leased() — context manager, який на час обробки запускає фонову корутину: кожні ttl/3 секунд вона продовжує лізу (heartbeat(task_id, worker_id)). Фільтр heartbeat-а — {_id, worker_id, status: processing}: якщо задачу тим часом украли (reclaim або інший воркер), matched_count == 0, heartbeat зупиняється і не воскрешає чужу тепер задачу — це той самий worker_id-guard.

Відпустити задачу можна трьома способами:

release_for_retry(id, attempts, error) — транзієнтна помилка (мережа, S3 тощо): status=pending + backoff-ліза: delay = min(base · 2^(attempts−1), cap) · uniform(1.0, 1.5) (дефолт: base 60 с, cap 3600 с; jitter розсинхронізовує повтори).
release_not_an_attempt(id) — revision race: файл змінився під час завантаження (md5 не збігся). Це не помилка нікого: attempts відкочується на −1, задача повертається в pending з короткою «not before»-лізою, щоб не крутитись у гарячому циклі, поки сканер не запише новий md5.
fail(id, error) — фатальна помилка (бите відео, GeminiFatalError) або вичерпані спроби: status=failed, ліза знімається. failed — термінальний до ручного втручання оператора.

Успішне завершення — не метод черги: воркер сам фіналізує документ (worker._finalize: status=done, sha256, probe, зняття лізи).

6. Reclaim: прибирання за впалими воркерами

Раз на цикл сканер викликає reclaim_expired() для обох черг, два update_many у строгому порядку:

processing + протухла ліза + attempts >= max_attempts → failed (задача згоріла всі спроби — селити її назад у чергу немає сенсу);
решта processing з протухлою лізою → pending (ліза й worker_id знімаються).

Порядок важливий: спочатку «поховати» вичерпані, потім повернути живі — інакше вичерпана задача на мить стала б pending і претендувала на claim. Reclaim — це підстраховка третьої гілки claim-у (§3): навіть якщо воркерів зараз нуль, стан черги прибереться і метрики стануть чесними.

7. Серверний годинник і aggregation-pipeline updates

Всі записи статусів ідуть aggregation pipeline update-ами ([{"$set": {...}}], квадратні дужки — принципові):

"updated_at": "$$NOW" і lease_expr беруть час Mongo-сервера. Лізова логіка порівнює часи, записані різними подами; clock skew між подами зламав би її, серверний годинник — ні.
$ifNull дозволяє умовну ініціалізацію (first_seen_at) без $setOnInsert, який конфліктує з $set того ж поля (ConflictingUpdateOperators).
Порівняння в фільтрах claim/reclaim використовують _utcnow() воркера — це безпечно, бо похибка годинника воркера лише трохи зсуває момент, коли задачу можна забрати, а не коректність переходів.

Для звірок «час запису vs зараз» поза чергою є db.server_now() — той самий серверний годинник через db.command("hello").

8. Ідемпотентність і сценарії збоїв

Сканер впав до коміту курсора → та сама сторінка змін програється повторно. Upsert-и (file_upsert_op) написані так, що replay — no-op: статуси обчислюються server-side відносно попереднього документа (md5 змінився?, був tombstone/skipped?), а не «в лоб».
Воркер впав посеред обробки → ліза протухає, задачу забирає інший (claim) або повертає сканер (reclaim). Повторна обробка безпечна: артефакти в S3 пишуться під content-addressed ключами (sha256), перезапис ідентичний.
Задачу вкрали, а «мертвий» воркер ожив → його heartbeat не пройде (worker_id-guard), а фіналізація в гіршому разі запише той самий результат того самого контенту.
Дублікат контенту → друга задача з тим самим md5 фіналізується через дедуп (_dedup_by_md5), не чіпаючи Drive; черга аналізу ключується sha256, тож аналіз одного контенту існує в одному екземплярі незалежно від кількості копій файлу.

9. Індекси та метрики

Індекси (db.ensure_indexes):

{status: 1, lease_until: 1} на обох колекціях — точно під фільтр claim/reclaim;
{md5: 1} — дедуп; {name: 1} — лукапи API; {drive_id: 1, seen_at: 1} — resync;
TTL {updated_at: 1} з partialFilterExpression={"status": "tombstone"} — автоочистка тумбстоунів за 30 днів. partialFilterExpression обов’язковий: голий TTL-індекс стер би всю колекцію.

Метрики (metrics.update_queue_gauges, пушаться сканером і воркерами): *_pending_total — розмір черги (по ній скейлиться KEDA), *_oldest_pending_age_seconds — вік найстарішої pending-задачі (головний сигнал «черга стоїть»), лічильники processed/failed/reclaimed.

10. Обмеження — свідомі

Немає пріоритетів і FIFO-гарантій: claim бере будь-яку готову задачу (порядок — natural/індексний). Для цього пайплайна порядок неважливий.
Poll, не push: воркер живе, поки черга не порожня, далі под помирає; латентність появи нового пода — на совісті KEDA, а не черги.
Ліза ≠ жорсткий mutex: після протухання лізи можлива коротка гонка «старий воркер дописує — новий уже працює». Вона безпечна через ідемпотентність фіналізації (§8), тому не закривається складнішими механізмами.