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

Background Job: CallTimeoutReaperJob

Назначение​

CallTimeoutReaperJob — периодический фоновый worker сервиса messenger, закрывающий «зависшие» звонки CallSessionEntity, которые остались в активных статусах, потому что штатный сигнал завершения не пришёл (адресат не ответил; обе стороны отвалились без hangup; сеть/сигналинг оборвались). Обрабатывает две фазы над одним агрегатом:

  • ring-timeout (INV-CALL-2) — Status = Ringing при now − StartedAt > RingTimeout (default 45 с) → Missed, EndReason = Timeout; push CallMissed инициатору;
  • stale-answered — Status = Answered, чей UpdatedAt <= now − StaleAnsweredTimeout (нет прогресса сигналинга/heartbeat; обе стороны пропали) → Failed, EndReason = Error (Answered → Failed); push обеим сторонам.

Сигнал живости для фазы stale-answered — только доменный UpdatedAt звонка: messenger вне медиа-плоскости и медиа-heartbeat не хранит (INV-CALL-4), поэтому «прогрессом» считается любая фиксация по звонку (штатный end-call, повторные сигнальные операции). Штатное завершение всегда должен слать клиент; эта фаза — грубый backstop с большим окном (StaleAnsweredTimeout, default 2 ч) на случай, когда обе стороны исчезли, не отправив end, — она закрывает утечку «вечно активного» звонка, а не измеряет реальную длительность разговора.

Оба перехода терминальны (INV-CALL-3), поднимают доменное событие (CallMissedEvent / CallEndedEvent) и в той же транзакции пишут строку-намерение realtime-push в messenger.realtime_outbox (Kind = CallMissed для Ringing → Missed, Kind = CallEnded для Answered → Failed) — persist-then-push (ADR-0055): фактическую доставку в хаб выполняет воркер realtime-dispatch. Так закрывается разрыв, при котором reaper пушил CallMissed без соответствующей outbox-строки: RealtimeDispatchKind теперь включает CallMissed. SDP/ICE/токены не сохраняются (INV-CALL-4 — в хранилище только факты). Интеграционных Kafka-событий звонок не публикует (call-session.md). Цикл — polling без Redis/TTL (ADR-0056 п.3): выборка активных звонков с single-runner блокировкой, переход фиксируется атомарно, идемпотентно (по guard-условию статуса Ringing/Answered).

MetadataЗначение
Owning teamPIN Platform (Comms squad)
Статус документаdraft
Область действияworker call-timeout-reaper, таблицы messenger.call_sessions + messenger.realtime_outbox (строка push в той же транзакции); доставка в хаб /hubs/messenger — воркером realtime-dispatch; Kafka не затрагивается
Assumptions / ограниченияосновная фаза — Ringing → Missed; целевой объём звонков мал (ADR-0056); точность закрытия — RingTimeout + Interval; заявлен в README «Background jobs»
ЗависимостиPostgreSQL (messenger.call_sessions, messenger.realtime_outbox); realtime-доставка — отдельный воркер realtime-dispatch

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

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

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

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

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

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

ПолеТипОбязательностьИсточникОписание
AsOfDateTimeOffsetДачасы системыМомент цикла для сравнения с StartedAt/UpdatedAt.
RingTimeoutTimeSpanДаCallTimeoutReaperOptions (default 45 с)Окно ответа: Ringing дольше → Missed (INV-CALL-2).
StaleAnsweredTimeoutTimeSpanДаCallTimeoutReaperOptions (default 2 ч)Answered без прогресса дольше → Failed.
BatchSizeintДаCallTimeoutReaperOptions (default 200)Верхняя граница звонков за фазу за цикл.
TenantIdTenantIdНетне применяетсяCross-tenant scan: платформенный System-worker без tenant-фильтра (звонки — вне tenant, ADR-0052 п.3).

Алгоритм​

Каждый тик обрабатывает две фазы по порядку — сначала RingTimeout, затем StaleAnswered (переход в терминал делает звонок невыбираемым во второй фазе).

  1. Контекст и доступ. Источник запуска — расписание; актор — System (scope-фильтры отключены, cross-tenant разрешён явно); correlationId = jobRunId; AsOf — текущее время. Идемпотентность — не по конверту события, а по guard-условию статуса (шаг 4) и частичному индексу активных звонков.
  2. Проверка предусловий. Внешнего ввода нет. Допустимость состояния задаётся предикатами выборки: только Ringing (фаза 1) / Answered (фаза 2); терминальные Missed/Ended/Declined/Failed не выбираются (INV-CALL-3).
  3. Что читаем (claim под блокировкой). Захватить single-runner блокировку; если занята — прогон пропустить. Иначе на каждый цикл — две выборки по частичному индексу активных звонков ix_call_active (status in ('Ringing','Answered')), порядок StartedAt asc, размер BatchSize, с пропуском уже заблокированных строк:
    • фаза RingTimeout: Status == Ringing && StartedAt <= AsOf − RingTimeout;
    • фаза StaleAnswered: Status == Answered && UpdatedAt <= AsOf − StaleAnsweredTimeout. Skip-locked пропускает строки, занятые командами answer-call/end-call, без ожидания.
  4. Бизнес-правила и переходы. Для каждой сессии:
    • фаза 1: Ringing → Missed (guard INV-CALL-3): Status = Missed, EndReason = Timeout, EndedAt = AsOf, UpdatedAt = AsOf; поднимается CallMissedEvent;
    • фаза 2: Answered → Failed: Status = Failed, EndReason = Error, EndedAt = AsOf; DurationSeconds = null (не Answered → Ended); поднимается CallEndedEvent. Повторный проход над уже терминальным звонком — no-op.
  5. Что меняется. Обновляются Status/EndReason/EndedAt/UpdatedAt затронутых сессий; в той же транзакции на каждую закрытую сессию добавляется строка messenger.realtime_outbox (Kind = CallMissed для фазы 1, Kind = CallEnded для фазы 2; TargetKind = User, адресаты user:{CallerUserId}/user:{CalleeUserId}, только typed reference-id CallId/enum — без PII). Звонок — append-only лог: единственный активный звонок и гонку закрывают guard-условие статуса + skip-locked (INV-CALL-5). Переходы батча + outbox-строки фиксируются в одном согласованном изменении.
  6. Факты и события. Доменные CallMissedEvent/CallEndedEvent (in-memory) диспетчеризуются при фиксации; их обработчики пишут строку-намерение push в messenger.realtime_outbox (см. шаг 5). Фактическую доставку CallMissed/CallEnded (non-PII: CallId/Status/MediaType) в хаб выполняет отдельный воркер realtime-dispatch — persist-then-push: reaper только фиксирует намерение, никаких синхронных вызовов хаба в цикле нет. RealtimeDispatchKind включает CallMissed (ранее reaper пушил его без outbox-строки). Интеграционных Kafka-событий нет (call-session.md); SDP/ICE не сохраняются (INV-CALL-4).
  7. Согласованность. Граница — на батч: переходы батча + строки realtime-outbox фиксируются атомарно; блокировки строк держатся до фиксации. Доставка в хаб — после фиксации, воркером realtime-dispatch (persist-then-push); потеря push безопасна (клиент перезапросит REST). Poison-сессия (guard отклонил) исключается, логируется, остаётся на следующий цикл.
  8. Результат. Счётчики цикла (MissedCount, FailedCount, SkippedLockedCount, DurationMs) — для метрик/логов, не HTTP. Инфраструктурный сбой → откат батча, освобождение блокировок, ретрай следующим циклом (те же сессии ещё активны → выберутся снова, идемпотентно).

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

ЭлементПравило
Единица работыper batch — переходы батча + строки realtime-outbox фиксируются атомарно; блокировки строк живут до фиксации
Сохраняемое состояниепереходы CallSessionEntity (Status/EndReason/EndedAt/UpdatedAt) + строки messenger.realtime_outbox (Kind = CallMissed/CallEnded) в той же транзакции; Kafka-публикация не производится (событий нет)
External callsсинхронных внешних вызовов в цикле нет; hub-push выполняет воркер realtime-dispatch по outbox-строке — после фиксации (persist-then-push)
Частичная ошибкаcontinue — poison-сессия исключается и логируется, остальной батч фиксируется; инфраструктурный сбой — откат батча, ретрай следующим циклом

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

СценарийПравило
Job restart / повторный циклCheckpoint не нужен: выборка отбирает только активные (Ringing/Answered); зафиксированный терминал невыбираем → повтор безопасен
Duplicate item (параллельные инстансы)single-runner блокировка → один runner; при обходе skip-locked сессия достаётся одному, guard-условие статуса делает второй переход no-op
Гонка reaper ↔ answer-call/end-callСтрока под командой заблокирована → skip-locked в этом цикле; если команда зафиксировала терминал/Answered до тика — сессия вне выборки
Повторяемая ошибка (инфраструктура)Откат батча, блокировки сняты; следующий цикл повторяет (кандидаты ещё активны)
Poison itemИсключается из батча, ERROR-лог с CallId; не блокирует батч; при систематике — алерт по errors_total, разбор по runbook
Потеря hub-pushбезопасна — realtime не источник истины; outbox-строка переживёт рестарт (realtime-dispatch доставит позже), а клиент при reconnect перезапрашивает журнал list-calls (REST)

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

ТипПравило
Logscycle started (jobRunId, AsOf); batch selected (phase, count); call reaped (CallId, Status, EndReason, CallerUserId, CalleeUserId); cycle completed (counts, DurationMs, SkippedLocked); item failed (CallId, exception). CallId/UserId — не PII
Metricsmessenger_call_reaper_run_total (counter — циклы, liveness); messenger_call_reaper_cycle_duration_seconds (histogram); messenger_call_reaper_items_total{outcome="missed"|"failed"}; messenger_call_reaper_active_backlog (gauge — активные звонки с превышенным окном); messenger_call_reaper_skip_locked_total; messenger_call_reaper_errors_total
TracescorrelationId = jobRunId; span цикла + per-item
Alertsjob не отработал > 2 интервалов (нет прироста run_total); active_backlog > 50 в течение 2 мин; errors_total rate > 0 устойчиво

Связанный runbook​

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

Тесты​

ТестОжидание
Empty batchнет активных с превышенным окном — no-op success, мутаций/событий нет
Ring-timeoutRinging, StartedAt <= now − 45с → Missed, EndReason=Timeout, строка realtime_outbox Kind=CallMissed в той же транзакции → realtime-dispatch пушит call_missed инициатору
Ответили в окнеRinging → Answered до тика → вне выборки фазы 1, не трогается
Stale-answeredAnswered, UpdatedAt старше StaleAnsweredTimeout → Failed, EndReason=Error, DurationSeconds=null, строка realtime_outbox Kind=CallEnded → push обеим сторонам
Идемпотентностьповторный цикл над уже Missed/Failed → guard no-op, второго события нет
Гонка с answer-callanswer зафиксировал Answered до тика → сессия не в фазе 1; либо skip-locked в этом цикле
Параллельные инстансыблокировка → один runner; при обходе skip-locked каждая сессия обработана один раз
Cancellationостановка в цикле → батч докоммичен или откачен целиком, блокировки сняты, graceful stop
SLO точностиRinging-звонок закрывается не позже StartedAt + RingTimeout + Interval

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

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

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