Background Job: RealtimeDispatchJob

Назначение

realtime-dispatch — фоновый воркер сервиса messenger, реализующий persist-then-push (ADR-0055): доставку уже зафиксированных в PostgreSQL событий (новое/правленое/удалённое сообщение message.md, read-receipt и member-события participant.md, системная лента chat-event.md, заявки/подтверждения контактов contact.md, сигналы 1:1-звонка call-session.md, а также media-события конференций conference-session.md/ростера call-participant.md и записи call-recording.md) подключённым клиентам через SignalR-hub /hubs/messenger — идемпотентно и at-least-once.

Надёжность обеспечивает realtime-outbox messenger.realtime_outbox: строка-намерение push-а пишется в той же транзакции, что и доменная мутация (persist-then-push — сначала фиксация ленты/состояния, затем доставка). Строка несёт только typed reference-id (ChatId/MessageId/Seq/enum-Kind), не тело; диспетчер строит типизированный контракт RealtimeMessage на выдаче и пушит в hub-группу. Realtime не source of truth: при reconnect клиент перезапрашивает REST; потеря/дубль push безопасны, клиент дедупит по (ChatId, Seq) / MessageId. Механизм — polling без Redis-очереди для outbox (ADR-0056 п.3): claim строк под блокировкой (FOR UPDATE SKIP LOCKED), backoff, poison → карантин.

Metadata	Значение
Owning team	PIN Platform (Comms squad)
Статус документа	draft
Область действия	воркер realtime-dispatch, таблица messenger.realtime_outbox, SignalR-hub /hubs/messenger (группы chat:{ChatId} / user:{UserId})
Assumptions / ограничения	доставка чата — в приватную группу chat:{ChatId} (только активные Participant, record-level); тело сообщения (PII) — только участникам чата, не в Kafka-события (message.md); offline-доставка (push/email) — отдельный путь через pin.messenger.message-posted.v1 → notifications
Зависимости	PostgreSQL (messenger.realtime_outbox), SignalR-hub (Redis backplane при multi-replica — опция, REALTIME_BACKPLANE_ENABLED, ADR-0043/0055)
Источник требований	заявлен в README «Background jobs / Runtime»

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

Поле	Значение
Сервис	messenger
Тип job	Recurring (near-real-time — короткий интервал + LISTEN/NOTIFY-wake)
Trigger	таймер каждые PollingInterval (default 1 с) + опц. PostgreSQL LISTEN realtime_outbox для немедленного пробуждения
Критичность	tier-1 (realtime UX чата; сбой → задержка in-app доставки, история не теряется — persisted, клиент перезапросит REST)

Расписание И Locking

Аспект	Правило
Schedule	PollingInterval (default 1 с, UTC) + wake по LISTEN/NOTIFY при вставке строки outbox (низкая латентность без busy-loop)
Distributed lock	primary — single-runner: единый runner через distributed lock (ключ job:scheduled:messenger-realtime-dispatch, TTL LockTtl default 30 с, PostgreSQL advisory-backend); занятый lock → прогон пропускается
Row-level	claim строк под блокировкой FOR UPDATE SKIP LOCKED (defense-in-depth) — при failover два runner-а берут непересекающиеся строки; двойной push идемпотентен (клиент дедупит)
Batch size	BatchSize строк/проход (default 200); проход повторяет выборку, пока батч полон и не истёк MaxRunDuration
Max duration	MaxRunDuration (default 5 с); остаток — следующим тиком/wake
Cancellation	graceful stop между строками при остановке хоста; закоммиченный прогресс сохраняется

Входная Модель

Внешнего runtime-входа нет: параметры — RealtimeDispatchOptions (bind + валидация на старте: PollingInterval, BatchSize > 0, MaxAttempts (default 8), backoff BackoffBase/BackoffFactor/BackoffCap (default 2 с / 2.0 / 2 мин + jitter), RetentionDone (default 24 ч — уборка Done)); рабочий вход — строки messenger.realtime_outbox.

Reliability-таблица RealtimeOutboxEntry (messenger.realtime_outbox)

Операционная структура job-а (не доменная сущность). Только typed reference-id и enum — без тела/PII.

Поле	Тип	Обязательность	Ограничения	Описание
Id	Guid	Да	PK (v7 — порядок вставки)	Идентификатор push-намерения.
Kind	RealtimeDispatchKind	Да	.HasConversion<string>(), varchar(32)	Тип push-а (см. enum).
TargetKind	RealtimeTargetKind	Да	.HasConversion<string>(), varchar(16)	ChatGroup (chat:{ChatId}) / User (user:{UserId}).
ChatId	Guid?	Нет	index ix_rto_chat	Чат-группа (для ChatGroup-push).
TargetUserId	Guid?	Нет	—	Адресат (для User-push: контакт/звонок/блокировка).
MessageId	Guid?	Нет	логическая ссылка messenger.messages(id)	Сообщение (для MessageReceived/Edited/Deleted).
RefId	Guid?	Нет	—	Id связанной сущности (contact/call/chat-event/conference/recording/participant) для non-message push.
SessionKind	SessionKind?	Нет	.HasConversion<string>(), varchar(12)	Тип медиа-сессии для media-push (Direct — 1:1-звонок / Conference — конференция); задаёт, какой payload строит диспетчер (личный vs групповой). null для не-медиа push.
Seq	long?	Нет	—	Message.Seq — ключ дедупа клиента и порядка.
Status	RealtimeOutboxStatus	Да	.HasConversion<string>(), varchar(16), default Pending	Pending/Done/Quarantined.
Attempts	int	Да	>= 0, default 0	Число попыток (инкремент при claim).
NextAttemptAt	DateTimeOffset	Да	index ix_rto_due (status,next_attempt_at)	Расписание попытки (backoff).
LastError	string?	Нет	varchar(512)	Последняя ошибка (класс+код, без PII).
CreatedAt/UpdatedAt	DateTimeOffset	Да	—	Штампы; CreatedAt — база SLO-лага.
RowVersion	uint	Да	concurrency	Гонка claim/outcome.

Enum RealtimeOutboxStatus { Pending, Done, Quarantined }. Enum RealtimeDispatchKind { MessageReceived, MessageEdited, MessageDeleted, ReadReceiptUpdated, ChatUpdated, ContactRequestReceived, ContactRequestAccepted, ContactRemoved, CallInvited, CallAnswered, CallDeclined, CallMissed, CallCancelled, CallEnded, ConferenceInvited, ConferenceStarted, ConferenceEnded, ConferenceParticipantJoined, ConferenceParticipantLeft, RecordingStarted, RecordingStopped, RecordingAvailable, RecordingFailed, ScreenShareStarted, ScreenShareStopped, BlacklistChanged }. Media-значения (звонки/конференции/запись/ демонстрация экрана) добавлены под media-стек (ADR-0062) поверх существующих значений чата; CallMissed закрывает разрыв, при котором call-timeout-reaper пушил call_missed без outbox-строки. Actor — системный (вне tenant — доступ несёт членство чата/сессии); публичного входа нет.

Алгоритм

Push-вызов — между claim и фиксацией исхода (не держим блокировку строки на сетевом I/O к hub/backplane).

1. Контекст прогона. Прогон запускает таймер (или NOTIFY); актор — системный, вне tenant. Отдельный idempotency-lookup не нужен: идемпотентность обеспечивают статус-фильтр + NextAttemptAt и клиентский дедуп по (ChatId, Seq)/MessageId.
2. Проверка конфигурации. Внешнего runtime-входа нет; конфигурация проверена на старте (BatchSize > 0, MaxAttempts > 0, backoff > 0 — иначе fail-fast хоста).
3. Захват работы (claim под блокировкой). Захватить single-runner-lock; занят → пропустить прогон. Иначе циклом батчей (до BatchSize строк, до MaxRunDuration) выбирать под блокировкой FOR UPDATE SKIP LOCKED строки outbox со статусом Pending, у которых NextAttemptAt <= now, в порядке вставки (индекс ix_rto_due, порядок по Id v7 = порядок фиксации). SKIP LOCKED разводит строки параллельного runner-а при failover.
4. Claim-фаза. Для каждой claimed-строки: Attempts += 1, NextAttemptAt = now + backoff(Attempts); фиксация claim-фазы снимает блокировку строк. Пред-инкремент безопасен: краш до доставки даёт повтор после backoff.
5. Сборка, доставка и исход. Для каждой строки построить типизированный контракт RealtimeMessage по Kind из reference-id (для MessageReceived — проекция сообщения chat:{ChatId}: { ChatId, MessageId, Seq, SenderUserId?, Kind, SentAt, Body/FileMeta/CardText } — тело только участникам приватной группы; для non-message — non-PII id/статус). Для media-push SessionKind строки выбирает форму payload: Direct — контракт 1:1-звонка (user:{TargetUserId}), Conference — конференции (chat:{ChatId}). Для приглашающих media-kind (CallInvited/ConferenceInvited/ConferenceStarted) диспетчер минтит MediaJoinModel (AccessToken, TURN/ICE-креды) на выдаче, при доставке — короткоживущие секреты никогда не пишутся в realtime_outbox (строка несёт только RoomName/reference-id); утечка строки не раскрывает доступ к комнате. Запушить: ChatGroup → в группу chat:{ChatId}; User → в user:{TargetUserId} (eventName snake_case). Затем зафиксировать исход: успех (2xx / enqueued в backplane) → Status = Done, LastError = null; transient (hub/backplane недоступен) → строка остаётся Pending (backoff из ш.4); poison (Attempts >= MaxAttempts) → Status = Quarantined + внутренний алерт RealtimeDispatchQuarantinedEvent.
6. События. Интеграционных Kafka-событий job не публикует (доставка — hub, не Kafka). Событие RealtimeDispatchQuarantinedEvent ({ OutboxEntryId, Kind, ChatId?, Attempts }) — только внутренний алерт/метрика, вне claim-транзакции.
7. Согласованность. Две короткие транзакции на строку: claim (Attempts/NextAttemptAt) и исход (Status/LastError); между ними — push вне открытой транзакции. Строка-намерение пишется в той же транзакции, что и доменная мутация владельца (send-message/edit/…/consumer — persist-then-push).
8. Результат. Наружу — метрики прогона (dispatched/retried/quarantined); сбой на строке логируется и не рвёт проход (частичный прогресс закоммичен).

Граница Транзакции

Элемент	Правило
Транзакции	Две короткие транзакции на строку: claim (Attempts/NextAttemptAt) и исход (Status/LastError); строка-намерение — в транзакции владельца (send-message/edit/consumer)
Сохраняемое состояние	messenger.realtime_outbox (Status/Attempts/NextAttemptAt/LastError); доменное состояние не мутируется (read-only проекция на build)
External calls	push к hub/backplane — между claim-tx и outcome-tx, вне открытой транзакции (нет row-lock/соединения на сетевом I/O)
Частичная ошибка	continue: сбой на строке логируется, не откатывает соседние; строка остаётся Pending (backoff) или Quarantined; следующий проход повторяет

Идемпотентность И Retry

Сценарий	Правило
Job restart	Checkpoint не нужен — прогресс = Status; рестарт заново фильтрует Pending по NextAttemptAt
Параллельные инстансы	distributed lock → один runner; при failover-лаге FOR UPDATE SKIP LOCKED разводит строки, двойной push безопасен (клиент дедупит)
Дубль push (at-least-once)	клиент дедупит по (ChatId, Seq) / MessageId; realtime не SoT — повтор не создаёт эффекта
Порядок доставки при retry	Доставка не гарантирует межстрочный порядок: строка на backoff может уйти позже вставленной после неё. Порядок для пользователя определяется Seq — клиент сортирует историю чата по (ChatId, Seq), а не по времени push. Пропущенный из-за backoff Seq клиент дозапрашивает по REST (list-messages, SoT). Для media-push (звонок/конференция) порядок не критичен — важен факт события (invite/ended), клиент сверяет актуальное состояние REST-запросом.
Повторяемая ошибка	hub/backplane недоступен → Pending + экспоненциальный backoff (cap, jitter); pre-increment Attempts при claim
Poison item	Attempts >= MaxAttempts → Quarantined + алерт; исключён из выборки; при инциденте hub — массовый разбор по runbook
Потеря push (crash до outcome)	безопасна: клиент при reconnect перезапрашивает list-messages/get-chat (REST, SoT); строка ещё Pending → повтор

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

Тип	Правило
Logs	batch claimed (rows); dispatched (Kind, ChatId/TargetUserId, Seq, Attempts); dispatch retry (EntryId, Attempts, errorCode); quarantined (EntryId, Kind); job completed (dispatched, retried, quarantined, DurationMs). Тело/PII в логи не пишется
Metrics	messenger_realtime_pending (gauge — backlog Pending); messenger_realtime_lag_seconds (histogram — now − CreatedAt на Done, SLO persist→push); messenger_realtime_dispatched_total; messenger_realtime_retried_total; messenger_realtime_quarantined_total; hub disconnect rate (README)
Traces	span на проход (jobRunId), вложенный span на диспетчеризацию строки (Kind, ChatId); корреляция с owner-tx через CreatedAt/RefId
Alerts	всплеск ошибок realtime-dispatch / рост pending (README «Alerts»); lag_seconds p95 > 1 с (нарушение SLO); quarantined_total > 0; отсутствие прогонов > 1 мин

SLO

Показатель	Цель	Обоснование
Лаг persist → push	p95 < 1 с (README SLO)	PollingInterval 1 с + NOTIFY-wake; доступный hub
Steady-state backlog	messenger_realtime_pending ≈ 0 вне инцидента hub	диспетчер успевает за фиксацией
Идемпотентность	0 видимых дублей у клиента при рестартах/N инстансах	клиентский дедуп (ChatId, Seq) + фильтр

Связанный Runbook

Сценарий	Runbook
pending/lag растёт	проверить доступность hub/backplane (REALTIME_BACKPLANE_ENABLED), disconnect rate; после восстановления backlog разгребётся — ../runbooks/realtime-dispatch.md (будущий путь)
quarantined_total > 0	разобрать LastError; ручной re-enqueue (Quarantined → Pending, Attempts = 0) после устранения причины
Worker стоит (нет прогонов)	проверить hosted-service/pod, PollingInterval, застрявший distributed lock

Тесты

Тест	Ожидание
Успешная доставка	Pending (MessageReceived) → push в chat:{ChatId} → Status=Done, dispatched_total++
Пустой проход	нет Pending с NextAttemptAt <= now — no-op success, вызовов нет
Недоступность hub → повтор	fail → строка Pending, Attempts++, NextAttemptAt сдвинут по backoff; следующий проход повторяет
Дедуп клиента	двойная доставка одного (ChatId, Seq) → клиент показывает одно сообщение
Poison → карантин	Attempts >= MaxAttempts → Quarantined + RealtimeDispatchQuarantinedEvent/алерт
Media-token at delivery	ConferenceInvited (SessionKind=Conference) → MediaJoinModel.AccessToken/TURN минтятся при доставке; в realtime_outbox секретов нет (только RoomName/reference-id)
SessionKind payload	CallInvited c SessionKind=Direct → 1:1-контракт в user:{Target}; Conference → групповой контракт в chat:{ChatId}
Персист-then-push	crash между owner-commit и push → строка Pending переживает рестарт, доставляется позже; клиент видит историю по REST
Параллельные инстансы	lock → один runner; при обходе FOR UPDATE SKIP LOCKED не даёт двойного claim одной строки
Cancellation	отмена между строками — graceful stop, claim-транзакция не рвётся, закоммиченное сохранено

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

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

• Use Cases (6): UC-MSG-001. Открыть личный чат по поиску пользователя по телефону, UC-MSG-003. Правка сообщения в окне ≤ 60 секунд, UC-MSG-004. Заблокировать пользователя (чёрный список), UC-MSG-005. Сигналинг аудио/видео-звонка, UC-MSG-006. Групповой звонок-конференция в чате сделки, UC-MSG-008. Добавленный участник появляется в чате сделки/лида
• Фоновые задачи (3): Background Job: CallTimeoutReaperJob, Background Job: ConferenceRoomReaperJob, Background Job: RecordingFinalizeRetryJob
• calls (2): Backend-реализация медиа-звонков: спека → код (messenger), Jitsi-звонки: план верификации и приёмки

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

• Источники push (доменные события → строка outbox): message.md (MessagePosted/MessageEdited/MessageDeleted), participant.md (ReadStateChanged/member-события), chat-event.md (ChatUpdated), contact.md, call-session.md, blacklist.md; media-источники (ADR-0062) — conference-session.md, call-participant.md, call-recording.md, а также воркер call-timeout-reaper.md (CallMissed/CallEnded); карта/Contracts — README.
• Прецедент паттерна (reliability-outbox, single-runner, backoff, poison): ../../news/background/snapshot-detach-retry.md.
• ADR: 0055 мини-мессенджер (persist-then-push, SignalR), 0043 SignalR backplane, 0056 simplicity-first, 0062 media-звонки и запись (media-kind, MediaJoinModel at delivery).