Background Job: RecordingFinalizeRetryJob
Назначение
RecordingFinalizeRetryJob — периодический фоновый worker сервиса messenger, гарантирующий доведение
записи звонка/конференции CallRecordingEntity до терминала
(Available либо Failed), когда штатная финализация не завершилась: рекордер media-стека выгрузил бинарь в
file storage, но ingest-callback (internal ingest) не пришёл; процесс упал между
переходом → Available и публикацией события; file storage временно недоступен. Без этой сети безопасности
запись «зависает» в Recording/Uploading, а внешние потребители (crm/notifications) не получают
call-recording-available.
Job — сеть безопасности поверх happy-path (ingest-эндпоинт финализации). Работает над
CallRecordingEntity в неконечных статусах (Pending/Recording/Uploading) и над Available, у которых
событие ещё не опубликовано. Три фазы:
- finalize-reconcile —
Recording/Uploading, чья сессия завершилась дольше FinalizeSlaTimeout
(default 30 мин) назад, а callback так и не пришёл → сверка с file storage: если бинарь на месте
(выгрузка прошла, callback потерян) — усыновить FileId/метаданные → Uploading → Available + событие
call-recording-available + realtime RecordingAvailable; иначе → Failed, FailureReason = Timeout +
алерт;
- republish — запись уже
Available, но интеграционное событие не опубликовано (краш между фиксацией и
публикацией) → повторно поставить событие в очередь (идемпотентно по RecordingId);
- transient-backoff — временные сбои file storage при сверке/выгрузке → ограниченный экспоненциальный
backoff (
Attempts/NextAttemptAt); исчерпание попыток (poison) → Quarantined + алерт.
Идемпотентность — по RecordingId и version-guard по RecordingStatus (worker правит только неконечные
статусы; зафиксированный Available/Failed повторно не перезаписывается). Бинарь в messenger не хранится —
только FileId + метаданные (INV-REC-6). Цикл — polling без Redis/планировщика (ADR-0056 п.3): выборка
кандидатов с single-runner блокировкой; переход и публикация фиксируются атомарно.
Кросс-сервисные ожидания и гарантии. Сверка finalize-reconcile ожидает от file storage стабильную
ссылку на бинарь (FileId/объектный ключ в категории CALL_RECORDINGS), пригодную для проверки наличия;
отсутствие категории/типа контента — конфигурационный сбой (предпосылки — filestorage-prerequisites),
не покрывается ретраем и требует алерта. Worker предоставляет консюмерам crm/notifications
интеграционное call-recording-available (ключ SessionId) — тот же контракт, что и happy-path ingest, но
испущенный сетью безопасности; для потребителя источник (ingest vs retry) неразличим и обрабатывается
одинаково идемпотентно по RecordingId. Окно согласованности: событие готовности приходит потребителю через
минуты после завершения сессии (SLA финализации + один PollingInterval), пока запись доступной не станет —
crm держит карточку встречи в состоянии «запись обрабатывается».
| Metadata | Значение |
|---|
| Owning team | PIN Platform (Comms squad) |
| Статус документа | draft |
| Область действия | worker recording-finalize-retry, таблицы messenger.call_recordings + операционная messenger.recording_finalize_attempts, топик pin.messenger.call-recording-available.v1, messenger.realtime_outbox (Kind = RecordingAvailable/RecordingFailed) |
| Assumptions / ограничения | happy-path финализации несёт ingest-эндпоинт, job покрывает только «застрявшие»/непубликованные записи (steady-state backlog ≈ 0); сверка идемпотентна (FileId усыновляется по стабильной ссылке file storage); групповые/записываемые звонки редки (ADR-0056); записи за флагом RECORDINGS_ENABLED |
| Зависимости | PostgreSQL (messenger), file storage (S3 — сверка наличия бинаря по FileId/категории), Kafka (публикация pin.messenger.call-recording-available.v1) |
Классификация
| Поле | Значение |
|---|
| Сервис | messenger |
| Тип job | периодический (PeriodicTimer) |
| Trigger | таймер каждые PollingInterval (default 60 с, без внешнего планировщика) |
| Критичность | tier-2 (доступность записи для сделки; сбой → запись не доедет до crm/участников, но бинарь в file storage не теряется — сверка восстановит) |
Расписание и locking
| Аспект | Правило |
|---|
| Schedule | фиксированный PollingInterval (default 60 с, UTC); следующий тик не стартует, пока идёт текущий цикл |
| Distributed lock | single-runner: попытка захватить блокировку прогона без ожидания, TTL LockTtl (default 5 мин > MaxRunDuration); backend — PostgreSQL advisory. Блокировка занята → прогон пропускается |
| Row-level | defense-in-depth: claim-выборка с пропуском уже заблокированных строк (skip-locked) — при failover два runner-а берут непересекающиеся строки; сверка/переход идемпотентны (version-guard RecordingStatus) |
| Batch size | BatchSize строк/цикл (default 100); backlog добирается следующими циклами (порядок по NextAttemptAt asc) |
| Max duration | MaxRunDuration (default 3 мин; сверка file storage — сетевой I/O, лимитируется по строке) |
| Cancellation | остановка хоста учитывается на каждом обращении к хранилищу; текущая строка докоммичивается или откатывается целиком, блокировки освобождаются |
Входная модель
Внешнего runtime-входа нет: поведение задаёт RecordingFinalizeRetryOptions (валидация на старте:
PollingInterval, BatchSize > 0, MaxAttempts > 0 (default 8), FinalizeSlaTimeout (default 30 мин),
backoff BackoffBase/BackoffFactor/BackoffCap (default 1 мин / 2.0 / 15 мин + jitter)); рабочий вход —
строки операционной таблицы messenger.recording_finalize_attempts + связанные CallRecordingEntity.
Reliability-таблица RecordingFinalizeEntry (messenger.recording_finalize_attempts)
Операционная структура job-а (не доменная сущность): планирование ретраев и карантин для финализации записи.
Домен (CallRecordingEntity) остаётся в своих статусах; Attempts/NextAttemptAt/Quarantined живут здесь,
чтобы не расширять доменный агрегат машинерией ретраев. Строка заводится при входе записи в асинхронную
финализацию (→ Uploading / завершение сессии с активной записью) и закрывается при достижении терминала.
| Поле | Тип | Обязательность | Ограничения | Описание |
|---|
Id | Guid | Да | PK | Идентификатор строки-намерения. |
RecordingId | Guid | Да | unique ux_rfa_recording; логическая ссылка messenger.call_recordings(id) | Запись; dedupe-ключ (финализация однократна на запись). |
SessionKind | SessionKind | Да | .HasConversion<string>(), varchar(12) | Тип сессии-владельца (Direct/Conference) для маршрутизации события/realtime. |
SessionId | Guid | Да | — | Сессия-владелец (CallSessionId/ConferenceSessionId); Kafka-ключ события. |
Status | RecordingFinalizeStatus | Да | хранится как varchar(16), default Pending | Pending/Done/Quarantined — состояние машинерии финализации. |
Attempts | int | Да | >= 0, default 0 | Число попыток (инкремент при claim). |
NextAttemptAt | DateTimeOffset | Да | index ix_rfa_due (status,next_attempt_at) | Расписание следующей попытки (backoff). |
LastError | string? | Нет | varchar(512) | Последняя ошибка (класс + код, без PII). |
CreatedAt/UpdatedAt | DateTimeOffset | Да | — | Штампы; CreatedAt — база SLO-лага. |
Enum RecordingFinalizeStatus { Pending, Done, Quarantined }. Актор — System (cross-tenant scan разрешён явно —
ADR-0052 п.3; доступ к записи — по участию в сессии, не строковый tenant); публичного входа нет.
Алгоритм
Claim и попытка финализации — на строку; сверка file storage — сетевой I/O между claim и фиксацией исхода
(не держим блокировку строки на внешнем вызове).
- Контекст и доступ. Источник запуска — расписание; актор — System (авто-scope-фильтры отключены);
correlationId = jobRunId; AsOf — текущее время. Идемпотентность — по RecordingId (ux_rfa_recording)
и version-guard RecordingStatus доменной записи.
- Проверка предусловий. Внешнего ввода нет. Допустимость задаётся предикатом: строки
Pending
(recording_finalize_attempts) с NextAttemptAt <= AsOf, чья запись — в неконечном статусе
(Recording/Uploading) или Available-но-непубликованная. Cross-tenant разрешён явно (System-worker).
- Что читаем (claim под блокировкой). Захватить single-runner блокировку; занята → прогон пропустить.
Иначе циклами до пустоты/
MaxRunDuration выбирать Pending-строки по ix_rfa_due (NextAttemptAt <= AsOf,
порядок NextAttemptAt asc, размер BatchSize, skip-locked) вместе со связанной CallRecordingEntity.
Skip-locked разводит строки параллельного runner-а и in-flight ingest-callback.
- Claim-фаза. Для каждой строки:
Attempts += 1, NextAttemptAt = AsOf + backoff(Attempts). Определить
фазу по статусу записи:
- republish — запись уже
Available, событие не публиковалось → повторно поставить
call-recording-available (idempotent by RecordingId) + realtime RecordingAvailable; строка → Done;
- finalize-reconcile —
Recording/Uploading, сессия завершилась дольше FinalizeSlaTimeout назад →
сверка с file storage: бинарь найден (усыновить FileId/SizeBytes/DurationSeconds/Format,
INV-REC-5) → Uploading → Available (version-guard) + событие + realtime RecordingAvailable, строка →
Done; бинаря нет и срок вышел → → Failed, FailureReason = Timeout + realtime RecordingFailed +
алерт, строка → Done (терминал);
- transient — сверка/переход упали на временной ошибке file storage → строка остаётся
Pending с
backoff (ш.4); Attempts >= MaxAttempts → Quarantined + алерт.
- Что меняется. Переход
CallRecordingEntity (Status/FileId/SizeBytes/DurationSeconds/Format/
AvailableAt или Status/FailureReason) под version-guard RecordingStatus + строка
recording_finalize_attempts (Status/Attempts/NextAttemptAt/LastError) + (для → Available)
транзакционный outbox pin.messenger.call-recording-available.v1 + messenger.realtime_outbox
(Kind = RecordingAvailable/RecordingFailed, SessionKind, reference-id — без URL/секретов). Фиксируется
в одном согласованном изменении.
- Факты и события. Доменный
RecordingAvailableEvent/RecordingFailedEvent. Интеграционное
pin.messenger.call-recording-available.v1 (ключ/AggregateId = SessionId, payload { RecordingId, SessionKind, SessionId, ChatId?, DealId?, TenantId?, MediaType, FileId, Format, DurationSeconds, SizeBytes, Trigger, StartedAt, EndedAt, AvailableAt }, без URL/бинаря) — через транзакционный outbox; контракт
call-recording-available.md, консюмеры crm/notifications.
Hub-push RecordingState(Available/Failed) (без ссылки) — по realtime-строке воркером
realtime-dispatch.
- Согласованность. Граница — на строку: переход записи + строка финализации + публикация фиксируются
атомарно; блокировка строки держится до фиксации; сверка file storage — вне открытой транзакции. Kafka —
после фиксации (at-least-once). Poison-строка (
Quarantined) исключается до ручного разбора.
- Результат. Итог по строке (
RecordingId, Status, Attempts, outcome) — для лог/метрик, не HTTP.
Инфраструктурный сбой → откат строки, освобождение блокировки, ретрай следующим циклом (строка ещё
Pending → выберется снова, идемпотентно).
Граница транзакции
| Элемент | Правило |
|---|
| Единица работы | per row — переход записи + строка финализации + публикация фиксируются атомарно; блокировка строки живёт до фиксации |
| Сохраняемое состояние | messenger.call_recordings (Status/FileId/метаданные/FailureReason/AvailableAt) + recording_finalize_attempts (Status/Attempts/NextAttemptAt) + outbox call-recording-available + messenger.realtime_outbox |
| External calls | сверка file storage — между claim и фиксацией исхода (вне открытой транзакции; нет row-lock на сетевом I/O); Kafka-публикация — после фиксации (at-least-once) |
| Частичная ошибка | continue — poison-строка (Quarantined) исключается и логируется, остальной батч фиксируется построчно; инфраструктурный сбой — откат строки, ретрай следующим циклом |
Идемпотентность и retry
| Сценарий | Правило |
|---|
| Job restart / повторный цикл | Checkpoint не нужен: прогресс = Status строки + RecordingStatus записи; зафиксированный терминал невыбираем → повтор безопасен |
| Дубликат финализации | ux_rfa_recording + version-guard RecordingStatus — второй переход → Available/Failed — no-op |
| Callback пришёл во время цикла | Строка/запись под ingest-callback заблокированы → skip-locked; если callback зафиксировал Available до тика — version-guard/фаза republish (событие уже опубликовано → строка Done) |
| Параллельные инстансы | single-runner блокировка → один runner; при failover-лаге skip-locked разводит строки, версия-guard делает второй переход no-op |
| Повторяемая ошибка (file storage) | Pending + экспоненциальный backoff (BackoffBase·Factor^n, cap, jitter); pre-increment Attempts при claim |
| Poison item | Attempts >= MaxAttempts → Quarantined + алерт; исключён из выборки до ручного re-enqueue (runbook) |
| Re-emit события | call-recording-available — key = SessionId; консюмеры crm/notifications идемпотентны по RecordingId (first-seen), повтор безопасен |
Наблюдаемость
| Тип | Правило |
|---|
| Logs | cycle started (jobRunId, AsOf); batch claimed (rows); recording reconciled (RecordingId, SessionId, FileId, Attempts); recording republished (RecordingId); recording failed (RecordingId, FailureReason=Timeout); recording quarantined (RecordingId, poison); cycle completed (reconciled, republished, failed, quarantined, DurationMs). RecordingId/SessionId/FileId — не PII; URL/бинарь в логи не пишутся |
| Metrics | messenger_recording_finalize_backlog (gauge — Pending-строки финализации); messenger_recording_finalize_reconciled_total; messenger_recording_finalize_failed_total; messenger_recording_finalize_republished_total; messenger_recording_finalize_quarantined_total; messenger_recording_finalize_oldest_age_seconds (gauge — now − min(CreatedAt)) |
| Traces | span на цикл (jobRunId), вложенный span на попытку финализации (RecordingId, phase, Attempts); DealTenantId per item |
| Alerts | failed_total rate ↑ (проблема рекордера/выгрузки); quarantined_total > 0; рост oldest_age_seconds; job не отработал > 2 интервалов; backlog публикации > 1000 |
SLO
| Показатель | Цель | Обоснование |
|---|
| Ни одна запись не «застревает» | нет записи в Recording/Uploading дольше FinalizeSlaTimeout + PollingInterval без терминализации | доступность записи для сделки |
| Republish-лаг | Available-но-непубликованная запись публикуется в пределах одного PollingInterval | восстановление после краша между commit и publish |
| Steady-state backlog | messenger_recording_finalize_backlog ≈ 0 вне инцидента file storage/рекордера | happy-path закрывает записи ingest-эндпоинтом |
Связанный runbook
| Сценарий | Runbook |
|---|
failed_total/backlog растёт | проверить доступность file storage (S3) и рекордера media-стека; после восстановления сверка усыновит бинари — ../runbooks/recording-finalize-retry.md (будущий путь) |
Poison-строка (Quarantined) | разобрать LastError; при найденном бинаре — ручной re-enqueue (Status = Pending, Attempts = 0); сверка RecordingId ↔ FileId |
| Worker стоит (нет прогонов) | проверить hosted-service/pod, PollingInterval, застрявшую блокировку прогона |
Тесты
| Тест | Ожидание |
|---|
| Empty batch | нет Pending finalize-строк с NextAttemptAt <= now — no-op success, мутаций/событий нет |
| Callback потерян, бинарь есть | Uploading, сессия завершена > FinalizeSlaTimeout, бинарь в file storage → усыновить FileId → Available + call-recording-available опубликовано |
| Callback потерян, бинаря нет | тот же случай без бинаря → Failed, FailureReason=Timeout, RecordingFailed + алерт |
| Available-но-непубликованная | краш между → Available и publish → republish call-recording-available (idempotent RecordingId), дубля у консюмера нет |
| Version-guard | callback зафиксировал Available до тика → worker-переход no-op (запись уже терминальна) |
| Transient → backoff | file storage временно недоступен → строка Pending, Attempts++, NextAttemptAt сдвинут; следующий цикл повторяет |
| Poison → карантин | Attempts >= MaxAttempts → Quarantined + алерт, исключена из выборки |
| Параллельные инстансы | блокировка → один runner; при обходе skip-locked + version-guard двойной финализации нет |
| Cancellation | остановка в цикле → текущая строка докоммичена или откачена, блокировка снята, graceful stop |
| SLO застревания | запись не остаётся в Uploading дольше FinalizeSlaTimeout + Interval без терминала |
Обратные ссылки
Автоссылки: где используется этот документ.
Связанные документы
- Дом записи (переходы INV-REC-1…7,
RecordingStatus/RecordingFailureReason/RecordingFormat,
ix_recording_active, FileId/метаданные, бинарь вне БД): call-recording.md;
сессии-владельцы — call-session.md, conference-session.md.
- Событие готовности: ../events/call-recording-available.md; happy-path
финализации (ingest-callback рекордера): ../api/internal/README.md.
- Доставка push по outbox-строке (persist-then-push,
RealtimeDispatchKind): realtime-dispatch.md.
- Прецедент паттерна (reliability-таблица, skip-locked, single-runner, backoff, poison):
chat-provision-retry.md,
../../news/background/snapshot-detach-retry.md.
- ADR: 0062 media-звонки и запись,
0055 мини-мессенджер,
0056 simplicity-first (polling без Redis),
0032 retention,
0052 tenancy (System cross-tenant).
- Шаблон — background-job-template.