--- title: Черга задач на MongoDB description: 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`: ```python { "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. **Відпустити задачу можна трьома способами:** 1. `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 розсинхронізовує повтори). 2. `release_not_an_attempt(id)` — revision race: файл змінився **під час** завантаження (md5 не збігся). Це не помилка нікого: `attempts` відкочується на −1, задача повертається в `pending` з короткою «not before»-лізою, щоб не крутитись у гарячому циклі, поки сканер не запише новий md5. 3. `fail(id, error)` — фатальна помилка (бите відео, GeminiFatalError) або вичерпані спроби: `status=failed`, ліза знімається. `failed` — термінальний до ручного втручання оператора. Успішне завершення — не метод черги: воркер сам фіналізує документ (`worker._finalize`: `status=done`, sha256, probe, зняття лізи). ## 6. Reclaim: прибирання за впалими воркерами Раз на цикл сканер викликає `reclaim_expired()` для обох черг, два `update_many` у строгому порядку: 1. `processing` + протухла ліза + `attempts >= max_attempts` → `failed` (задача згоріла всі спроби — селити її назад у чергу немає сенсу); 2. решта `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), тому не закривається складнішими механізмами.