Перейти к основному содержимому

0040. Явная concurrency воркеров sync-очередей WB

Копия ADR из репозитория: backend/docs/adr/0040-sync-queue-concurrency.md. Канонический текст — в репозитории; портал обновляется командой npm run sync:adr.

  • Статус: accepted
  • Дата: 2026-07-27
  • Связанные задачи: OP-216 (F5 · Очереди); связано: OP-209, OP-212, ADR-0007

Контекст и проблема​

Все per-account sync-процессоры (digitization-sync, rnp-sync, rnp-finance-sync, rnp-ads-sync, inventory-sync, catalog-sync, autobidder-sync) объявлялись @Processor(QUEUE) без опций → BullMQ concurrency = 1: кабинеты обрабатывались строго по одному. При этом и наш rate-лимитер (ключ wb:rl:{accountId}:{scope}:{path}), и лимиты WB — per кабинет, т.е. параллель между кабинетами легальна. Блокирующий acquire (OP-209) спит прямо в воркере и держит единственный слот. Полный круг rnp-sync ≈ 2,5 × M минут для M кабинетов — главный multi-tenant-затык.

Требование: синхронизировать кабинеты параллельно, не нарушая лимиты WB и anti-overlap, с настраиваемой степенью параллелизма.

Рассмотренные варианты​

  1. { concurrency: N } в декораторе @Processor, N из env. BullMQ создаёт воркер с N слотами; N кабинетов обрабатываются одновременно. Плюсы: минимальные правки (только декоратор), опция применяется на самом раннем этапе (создание воркера), не трогает конструкторы/DI. Минус: значение читается из process.env при загрузке класса (до готовности ConfigService) — в локальном .env берётся дефолт; в prod/stage/CI env приходит из окружения процесса и читается корректно.
  2. Установка worker.concurrency в onApplicationBootstrap через AppConfigService. Плюсы: значение валидировано и работает в dev/.env. Минусы: AppConfigService + lifecycle-хук в каждый из 7 процессоров → ломает существующие unit-конструкторы и их тесты; больше кода при равном поведении в проде (где затык и существует).
  3. Отдельная очередь user-импорта. Решает соседство user/плановых джоб, но не параллелит кабинеты и умножает инфраструктуру очередей.

Решение​

Вариант 1. @Processor(QUEUE, { concurrency: resolveSyncQueueConcurrency() }) для всех 7 sync-очередей; resolveSyncQueueConcurrency читает SYNC_QUEUE_CONCURRENCY (дефолт 5, core/queue/sync-concurrency.ts). Значение дополнительно валидируется zod-схемой env на старте (опечатка в проде уронит bootstrap, а не применится молча). Multi-tenant-параллель нужна именно в проде, где env задаётся окружением, — ограничение локального .env не критично (single-tenant dev в параллели не нуждается).

Anti-overlap и приоритеты не затронуты. Детерминированные jobId (composeJobId(job, account, slot)) отсекают дубль постановки независимо от числа слотов; джобы-оркестраторы (*-all) остаются без priority (в BullMQ v5 берутся раньше приоритизированных — не голодают при N > 1); приоритеты user/плановых джоб (OP-214) сохраняются в пределах слота.

Давление на зависимости при N параллельных кабинетах (AC3).

  • Postgres. Импорт пишет chunked-upsert по 500 строк (лимит bind-параметров). N = 5 параллельных апсертов держат ≤ 5 соединений из пула — в пределах дефолтного пула TypeORM/pg; всплеск строк ограничен размером чанка, не числом слотов. При росте N следить за max пула и pg_stat_activity.
  • Redis-лимитер. Lua-скрипт acquire атомарен; ключи изолированы по кабинету (wb:rl:{accountId}:...), поэтому параллельные кабинеты не делят бакет и не провоцируют 429. Соседство с BullMQ-трафиком безопасно — блокирующие команды BullMQ живут на выделенном соединении (ADR-0007).
  • WB API. Лимиты per-кабинет соблюдает WB-клиент (OP-34); параллель между кабинетами их не нарушает.

Подтверждение эффекта (AC4) — существующая observability (OP-43/46): optima_queue_jobs{queue,state="active"} показывает число одновременно обрабатываемых кабинетов, optima_queue_job_duration_ms{queue,quantile} — p50/p95 длительности джобы кабинета. Вместе они показывают, что круг ≈ длительность самого «тяжёлого» кабинета, а не сумма. Отдельная метрика не вводится.

Последствия​

  • Положительные: полный круг синка сокращается с ≈ суммы длительностей кабинетов до ≈ длительности самого «тяжёлого»; degree параллелизма крутится через env без релиза; blast-radius минимален (декоратор), существующие тесты не затронуты.
  • Отрицательные / компромиссы: в локальном .env SYNC_QUEUE_CONCURRENCY не подхватывается (берётся дефолт) — осознанный компромисс ради статической опции воркера; при значительном росте N нужно ревизовать пул Postgres.
  • Дальше: неблокирующее ожидание лимитера (reschedule вместо блокирующего acquire) снимет риск, что N слотов одновременно «уснут» на 1/мин-бакетах (синергия с OP-218 · token-bucket); консолидация планировщиков фидов — OP-217.