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

Background Job: ConferenceRoomReaperJob

Назначение​

ConferenceRoomReaperJob — периодический фоновый worker сервиса messenger, закрывающий осиротевшие и «зависшие» конференции ConferenceSessionEntity, которые остались в активных статусах (Provisioning/Active), потому что штатное завершение не наступило: последний участник отвалился без корректного leave; комната не поднялась в media-стеке; presence-heartbeat'ы перестали приходить. Обрабатывает три фазы над одним агрегатом:

  • empty-room (INV-CONF-8) — Status = Active, ActiveParticipantCount = 0 дольше EmptyRoomGrace (default 90 с) → Ended, EndReason = EmptyTimeout;
  • provision-timeout — Status = Provisioning старше ProvisionTimeout (default 2 мин; media-стек не подтвердил admission) → Failed, EndReason = Error;
  • stale-live — Status = Active без presence-активности (LastActivityAt <= now − StaleLiveTimeout, default 4 ч; heartbeat'ы пропали, но комната не опустела формально) → Ended, EndReason = Timeout.

Все переходы терминальны (INV-CONF-4). Для Active → Ended/Failed (фазы empty-room и stale-live) поднимается доменное ConferenceEndedEvent и в той же транзакции пишутся: строка транзакционного outbox интеграционного события pin.messenger.conference-ended.v1 (ключ DealId) и строка messenger.realtime_outbox (Kind = ConferenceEnded) для hub-push ConferenceSignal(Ended) (persist-then-push через realtime-dispatch). Фаза provision-timeout (Provisioning → Failed) не публикует conference-ended: конференция так и не стала Active, встречи не было — есть только внутренний алерт. Если у закрываемой сессии есть активная запись (call-recording.md), reaper запрашивает остановку рекордера media-стека; финализацию бинаря доводит собственный путь записи (сеть безопасности — recording-finalize-retry).

Медиа/SDP/ICE/токены доступа не персистятся (INV-CONF-7 — в хранилище только факты жизненного цикла и ростер). Цикл — polling без Redis/TTL (ADR-0056 п.3): выборка активных конференций по частичному индексу ix_conf_active с single-runner блокировкой; переход фиксируется атомарно, идемпотентно (guard-условие статуса Provisioning/Active).

MetadataЗначение
Owning teamPIN Platform (Comms squad)
Статус документаdraft
Область действияworker conference-room-reaper, таблица messenger.conference_sessions (+ messenger.realtime_outbox, транзакционный outbox pin.messenger.conference-ended.v1); остановка записи — сигнал рекордеру media-стека; доставка в хаб — realtime-dispatch
Assumptions / ограничениягрупповые звонки редки на целевой нагрузке (ADR-0056); happy-path завершения несёт end-conference, reaper — сеть безопасности для осиротевших/зависших комнат; точность закрытия пустой комнаты — EmptyRoomGrace + Interval; media за флагом CONFERENCE_CALLS_ENABLED
ЗависимостиPostgreSQL (messenger.conference_sessions, messenger.realtime_outbox), Kafka (публикация pin.messenger.conference-ended.v1), media-стек (сигнал остановки рекордера)

Классификация​

ПолеЗначение
Сервисmessenger
Тип jobпериодический (PeriodicTimer)
Triggerтаймер каждые PollingInterval (default 30 с — низкий, ради точности окна EmptyTimeout)
Критичностьtier-2 (UX конференций сделки + освобождение комнат media-стека; сбой → «висящая» комната и лишняя карточка в UI, не потеря данных)

Расписание и locking​

АспектПравило
Scheduleфиксированный PollingInterval (default 30 с, UTC); тик не стартует поверх идущего цикла
Distributed locksingle-runner: попытка захватить блокировку прогона без ожидания, TTL LockTtl (default 2 мин); backend — PostgreSQL advisory. Блокировка занята → прогон пропускается
Row-levelclaim-выборка с пропуском уже заблокированных строк (skip-locked, defense-in-depth); переход идемпотентен по guard-условию статуса
Batch sizeBatchSize конференций/цикл на фазу (default 100); backlog добирается циклами (порядок по LastActivityAt asc / CreatedAt asc)
Max durationMaxRunDuration (default 30 с; цель < 5 с/цикл)
Cancellationостановка хоста учитывается на каждом обращении к хранилищу; текущий batch докоммичивается или откатывается целиком, блокировки снимаются

Входная модель​

Внешнего runtime-входа нет; параметры — ConferenceRoomReaperOptions (валидация на старте).

ПолеТипОбязательностьИсточникОписание
AsOfDateTimeOffsetДачасы системыМомент цикла для сравнения с LastActivityAt/UpdatedAt/CreatedAt.
EmptyRoomGraceTimeSpanДаConferenceRoomReaperOptions (default 90 с)Grace для Active с ActiveParticipantCount = 0 → EmptyTimeout (INV-CONF-8).
ProvisionTimeoutTimeSpanДаConferenceRoomReaperOptions (default 2 мин)Provisioning дольше → Failed(Error) (admission не пришёл).
StaleLiveTimeoutTimeSpanДаConferenceRoomReaperOptions (default 4 ч)Active без presence-активности дольше → Ended(Timeout).
BatchSizeintДаConferenceRoomReaperOptions (default 100)Верхняя граница конференций за фазу за цикл.
TenantIdTenantIdНетне применяетсяCross-tenant scan: платформенный System-worker без tenant-фильтра (доступ к конференции — членство в Deal-чате, не строковый tenant; ADR-0052 п.3).

Алгоритм​

Каждый тик обрабатывает три фазы по порядку — empty-room, provision-timeout, stale-live (переход в терминал делает конференцию невыбираемой в остальных фазах).

  1. Контекст и доступ. Источник запуска — расписание; актор — System (scope-фильтры отключены, cross-tenant разрешён явно); correlationId = jobRunId; AsOf — текущее время. Идемпотентность — не по конверту события, а по guard-условию статуса (шаг 4) и частичному индексу активных конференций ix_conf_active.
  2. Проверка предусловий. Внешнего ввода нет. Допустимость состояния задаётся предикатами выборки: только Provisioning/Active; терминальные Ended/Failed не выбираются (INV-CONF-4).
  3. Что читаем (claim под блокировкой). Захватить single-runner блокировку; если занята — прогон пропустить. Иначе на каждый цикл — три выборки по частичному индексу ix_conf_active (status in ('Provisioning','Active')), размер BatchSize, с пропуском уже заблокированных строк:
    • empty-room: Status == Active && ActiveParticipantCount == 0 && LastActivityAt <= AsOf − EmptyRoomGrace;
    • provision-timeout: Status == Provisioning && CreatedAt <= AsOf − ProvisionTimeout;
    • stale-live: Status == Active && LastActivityAt <= AsOf − StaleLiveTimeout. Skip-locked пропускает строки, занятые командами join/leave/end-conference и admission-вебхуком, без ожидания.
  4. Бизнес-правила и переходы. Для каждой сессии (guard INV-CONF-4):
    • empty-room: Active → Ended, EndReason = EmptyTimeout, EndedAt = AsOf, DurationSeconds = AsOf − StartedAt; поднимается ConferenceEndedEvent;
    • stale-live: Active → Ended, EndReason = Timeout (аналогично, для «зависшей» живой комнаты);
    • provision-timeout: Provisioning → Failed, EndReason = Error, EndedAt = AsOf (комната не поднялась; StartedAt не задавался, встречи не было). Повторный проход над уже терминальной конференцией — no-op. Если у сессии есть активная запись (Pending/Recording/Uploading) — сигнал рекордеру на остановку (идемпотентно).
  5. Что меняется. Обновляются Status/EndReason/EndedAt/DurationSeconds/UpdatedAt затронутых конференций. В той же транзакции для завершений Active → Ended: строка транзакционного outbox pin.messenger.conference-ended.v1 + строка messenger.realtime_outbox (Kind = ConferenceEnded, TargetKind = ChatGroup, ChatId, SessionKind = Conference — только reference-id/enum, без PII). Для Provisioning → Failed интеграционное событие не пишется (встречи не было); realtime-строка — опционально ConferenceEnded/ConferenceSignal(Failed) для клиентов, ждавших admission. Переходы батча + outbox-строки фиксируются в одном согласованном изменении.
  6. Факты и события. Доменное ConferenceEndedEvent (in-memory) диспетчеризуется при фиксации. Интеграционное pin.messenger.conference-ended.v1 (ключ/AggregateId = DealId, payload { ConferenceId, ChatId, DealId, TenantId, MediaType, Status, EndReason, StartedByUserId, StartedAt, EndedAt, DurationSeconds, PeakParticipantCount, Participants[], WasRecorded, RecordingId? }, без ФИО/медиа) — через транзакционный outbox; контракт conference-ended.md, консюмеры crm (карточка встречи в таймлайне)/notifications/dev-stats. Hub-push ConferenceSignal(Ended) — по realtime-строке воркером realtime-dispatch. Готовность возможной записи придёт отдельным событием call-recording-available.md. Окно согласованности и промежуточное состояние: conference-ended (с флагом WasRecorded и, если запись велась, RecordingId) публикуется сразу при закрытии комнаты, а call-recording-available — только после финализации бинаря (сигнал рекордеру здесь best-effort, финализацию доводит recording-finalize-retry). Порядок для консюмера crm: сначала карточка встречи в таймлайне помечается «запись обрабатывается», затем — при отдельном событии готовности — к ней привязывается доступная запись; окно между ними — от секунд до минут (зависит от размера файла и выгрузки).
  7. Согласованность. Граница — на батч: переходы + строки outbox (Kafka и realtime) фиксируются атомарно; блокировки строк держатся до фиксации. Kafka-доставка — после фиксации (at-least-once, outbox-dispatcher); hub-push — после фиксации (realtime-dispatch). Poison-сессия (guard отклонил) исключается, логируется, остаётся на следующий цикл.
  8. Результат. Счётчики цикла (EmptyTimeoutCount, ProvisionFailedCount, StaleTimeoutCount, SkippedLockedCount, DurationMs) — для метрик/логов, не HTTP. Инфраструктурный сбой → откат батча, освобождение блокировок, ретрай следующим циклом (кандидаты ещё активны → выберутся снова, идемпотентно).

Граница транзакции​

ЭлементПравило
Единица работыper batch — переходы батча + строки outbox (Kafka + realtime) фиксируются атомарно; блокировки строк живут до фиксации
Сохраняемое состояниепереходы ConferenceSessionEntity (Status/EndReason/EndedAt/DurationSeconds/UpdatedAt) + транзакционный outbox pin.messenger.conference-ended.v1 (для Active → Ended) + messenger.realtime_outbox (Kind = ConferenceEnded)
External callsсинхронных внешних вызовов в цикле нет; Kafka-публикация и hub-push — после фиксации (outbox / realtime-dispatch); сигнал рекордеру — best-effort вне транзакции (финализацию доводит путь записи)
Частичная ошибкаcontinue — poison-сессия исключается и логируется, остальной батч фиксируется; инфраструктурный сбой — откат батча, ретрай следующим циклом

Идемпотентность и retry​

СценарийПравило
Job restart / повторный циклCheckpoint не нужен: выборка отбирает только активные (Provisioning/Active); зафиксированный терминал невыбираем → повтор безопасен
Duplicate item (параллельные инстансы)single-runner блокировка → один runner; при обходе skip-locked сессия достаётся одному, guard-условие статуса делает второй переход no-op
Гонка reaper ↔ join/leave/end-conference/admissionСтрока под командой/вебхуком заблокирована → skip-locked в этом цикле; если участник вошёл (ActiveParticipantCount > 0) или команда завершила конференцию до тика — сессия вне выборки
Повторяемая ошибка (инфраструктура)Откат батча, блокировки сняты; следующий цикл повторяет (кандидаты ещё активны)
Re-emit conference-endedключ DealId; консюмеры идемпотентны по ConferenceId (first-seen) — повтор безопасен
Poison itemИсключается из батча, ERROR-лог с ConferenceId; не блокирует батч; при систематике — алерт по errors_total, разбор по runbook
Потеря hub-pushбезопасна — realtime не источник истины; outbox-строка переживёт рестарт; клиент при reconnect перезапрашивает get-conference/list-conferences (REST)

Наблюдаемость​

ТипПравило
Logscycle started (jobRunId, AsOf); batch selected (phase, count); conference reaped (ConferenceId, ChatId, DealId, Status, EndReason, DurationSeconds); cycle completed (counts, DurationMs, SkippedLocked); item failed (ConferenceId, exception). ConferenceId/ChatId/DealId — не PII
Metricsmessenger_conf_reaper_run_total (counter — циклы, liveness); messenger_conf_reaper_cycle_duration_seconds (histogram); messenger_conf_reaper_items_total{outcome="empty_timeout"|"provision_failed"|"stale_timeout"}; messenger_conf_reaper_active_backlog (gauge — активные конференции с превышенным окном); messenger_conf_reaper_skip_locked_total; messenger_conf_reaper_errors_total
TracescorrelationId = jobRunId; span цикла + per-item (ConferenceId, phase); DealTenantId per item
Alertsjob не отработал > 2 интервалов (нет прироста run_total); active_backlog > 20 в течение 2 мин; provision_failed rate ↑ (проблема admission media-стека); errors_total rate > 0 устойчиво

SLO​

ПоказательЦельОбоснование
Закрытие пустой комнатыActive/ActiveParticipantCount=0 закрыт не позже LastActivityAt + EmptyRoomGrace + PollingIntervalосвобождение комнаты media-стека и корректная карточка встречи
Provision-timeout«неподнявшаяся» Provisioning терминализуется в пределах ProvisionTimeout + PollingIntervalклиент не ждёт admission бесконечно
Steady-state backlogmessenger_conf_reaper_active_backlog ≈ 0 вне инцидента media-стекаreaper успевает за завершениями

Связанный runbook​

СценарийRunbook
Worker стоит / backlog активных растётпроверить hosted-service/pod, PollingInterval, застрявшую блокировку прогона — ../runbooks/conference-room-reaper.md (будущий путь)
Рост provision_failedпроверить доступность media-стека/admission-вебхука (internal); после восстановления backlog разгребётся
Poison-сессия не закрываетсяразобрать LastError/статус; ручной end-conference/корректировка статуса по runbook

Тесты​

ТестОжидание
Empty batchнет активных с превышенным окном — no-op success, мутаций/событий нет
Empty-roomActive, ActiveParticipantCount=0, LastActivityAt <= now − 90с → Ended, EndReason=EmptyTimeout, conference-ended (Kafka) + ConferenceEnded (realtime)
Кто-то вошёл в graceActiveParticipantCount > 0 до тика → вне выборки empty-room, не трогается
Provision-timeoutProvisioning старше ProvisionTimeout → Failed, EndReason=Error, без conference-ended
Stale-liveActive, LastActivityAt старше StaleLiveTimeout → Ended, EndReason=Timeout, событие опубликовано
Активная записьу закрываемой конференции запись Recording → сигнал остановки рекордеру; финализацию доводит путь записи
Идемпотентностьповторный цикл над уже Ended/Failed → guard no-op, второго события нет
Гонка с end-conferenceкоманда завершила до тика → сессия вне выборки; либо skip-locked в этом цикле
Параллельные инстансыблокировка → один runner; при обходе skip-locked каждая сессия обработана один раз
Cancellationостановка в цикле → батч докоммичен или откачен целиком, блокировки сняты, graceful stop
SLO точностипустая комната закрывается не позже LastActivityAt + EmptyRoomGrace + Interval

Обратные ссылки​

Автоссылки: где используется этот документ.

Связанные документы​