Background Job: AccessFanOutJob — веерная рассылка UserAccessChangedKafkaEvent при смене состава прав роли

Назначение

AccessFanOutJob — фоновый воркер ролевой модели контекста identity, который выполняет массовую рассылку события UserAccessChangedKafkaEvent по всем пользователям роли, когда меняется её состав permissions (RolePermission, set-role-permissions). Одиночный grant/revoke назначения рассылает событие сразу в самой операции (grant-role-assignment, revoke-role-assignment); эта джоба покрывает именно случай «одно изменение роли → права поменялись у многих пользователей», который нельзя разослать одним действием.

Триггер реактивный: изменение состава прав роли ставит в очередь задачу веерной рассылки (атомарно со сменой роли). Воскер разбирает очередь и постранично проходит назначения (RoleAssignment) больших ролей.

Metadata	Значение
Owning team	PIN Platform Core
Статус документа	draft
Область действия	контекст identity, таблицы identity.access_fanout_tasks / identity.role_assignments
Assumptions/ограничения	порог inline-рассылки InlineFanOutThreshold (≤ порога — события рассылаются прямо в операции; > порога — задача в очередь); порог/батч — конфиг; события pin.identity.access.v1 — at-least-once, дедуп у потребителя по (UserId, Version); Version = RoleDefinition.Version
Зависимости	PostgreSQL (identity); шина событий; постановка задачи — set-role-permissions; потребители — user-access-changed (crm, catalog, chessboard, gateway)

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

Поле	Значение
Сервис	identity
Тип job	Очередь + воркер (внутренний таймер + очередь access_fanout_tasks)
Trigger	реактивный: смена состава permissions роли → задача в очередь; воркер поллит очередь
Критичность	high (задержка рассылки → потребители держат устаревшие права в кэше авторизации: риск лишнего доступа после снятия permission); не в денежном пути, но security-sensitive

Расписание и параллелизм

Аспект	Правило
Schedule	поллинг очереди каждые PollingInterval (default 5 с; UTC); задача берётся сразу после постановки
Параллелизм	несколько экземпляров воркера берут непересекающиеся задачи (взятая задача блокируется на время обработки, остальные её пропускают) — дубль-обработки одной задачи нет; отдельная распределённая координация не требуется
Batch size	BatchSize пользователей на страницу (default 500) — постраничный проход больших ролей по ключу UserId; страница повторяется, пока она полна
Max duration	MaxLeaseDuration (default 5 мин) — срок аренды задачи; при превышении задача снова видима поллингом и дообрабатывается с чекпоинта Cursor

Входная модель AccessFanOutTask (строка очереди identity.access_fanout_tasks)

Внешнего runtime-входа нет; вход — задача очереди (дом полей — здесь) + конфиг (PollingInterval, BatchSize, MaxLeaseDuration, MaxAttempts).

Поле	Тип	Обязательность	Ограничения	Источник	Описание
Id	Guid	Да	PK	app	Идентификатор задачи fan-out.
RoleId	Guid	Да	FK → identity.role_definitions(id) restrict, index ix_access_fanout_tasks_role_id	постановка	Роль, чей состав прав изменён.
CompanyId	Guid?	Нет	index	постановка	Tenant роли; null — платформенная системная роль.
RoleVersion	long	Да	>= 1	RoleDefinition.Version на момент постановки	Кладётся в event.Version (монотонный guard потребителя).
Reason	AccessChangeReason	Да	enum (хранится как text)	постановка	RolePermissionsChanged (event).
Status	FanOutTaskStatus	Да	enum (хранится как text), index ix_access_fanout_tasks_pending (partial: Pending/InProgress)	app	Pending / InProgress / Completed / Failed.
Cursor	Guid?	Нет	чекпоинт пагинации	app	Последний обработанный UserId; null — с начала.
ProcessedCount	int	Да	>= 0, default 0	app	Сколько пользователей разослано.
AttemptCount	int	Да	>= 0, default 0	app	Счётчик аренд/повторов (poison → Failed при >= MaxAttempts).
LeaseExpiresAt	DateTimeOffset?	Нет	—	app	Видимость InProgress-задачи; истёк → задача снова берётся.
CreatedAt / StartedAt? / CompletedAt? / UpdatedAt	audit	—	—	app	Тайминги задачи.

Enum FanOutTaskStatus: Pending → InProgress → Completed (успех) / Failed (poison, AttemptCount >= MaxAttempts).

AccessChangeReason — дом event: user-access-changed; эта джоба использует только RolePermissionsChanged.

Актор и область действия

Актор — система (воркер); контекст вне tenant: выборка ограничивается самим RoleId (пользователи company-роли лежат в CompanyId роли; системная роль — платформенно), tenant-фильтр не применяется — это задокументированный system-maintenance проход, публичного входа у джобы нет.

Алгоритм

Оркестратор берёт задачу из очереди и постранично рассылает событие по её пользователям. Дом полей payload и ActiveRoleAssignmentSnapshot — event: user-access-changed, здесь — только маппинг и порядок.

1. Контекст. Актор = система, correlationId = jobRunId (на прогон), контекст вне tenant. Идемпотентность — не по ключу, а через чекпоинт Cursor задачи (шаг 5) и снапшот-семантику события: повтор батча даёт тот же снапшот пользователя.
2. Взятие задачи. Взять одну задачу очереди со статусом Pending (или InProgress с истёкшей арендой), в порядке CreatedAt; параллельные экземпляры пропускают уже взятые задачи. Перевести задачу в InProgress, выставить LeaseExpiresAt = now + MaxLeaseDuration, AttemptCount += 1, StartedAt при первом взятии. Существование роли гарантировано ссылочной целостностью; архив роли между постановкой и обработкой не мешает — снапшот отражает актуальные Active-назначения (архив уже отозвал свои, INV-R5). Version взят на постановке (RoleVersion).
3. Страница пользователей. Взять следующую страницу активных назначений роли (Status = Active, UserId > Cursor, размер BatchSize, порядок по UserId) — уникальные UserId. Для снапшота прав каждого пользователя без N+1 — одним запросом загрузить все его активные назначения и коды соответствующих ролей.
4. Сбор снапшота. Доменное состояние назначений не меняется (проход только на чтение). Для каждого UserId собрать UserAccessChangedKafkaEvent: UserId из назначения; CompanyId пользователя; ChangeReason = AccessChangeReason.RolePermissionsChanged; ActiveAssignments — список ActiveRoleAssignmentSnapshot {AssignmentId, RoleId, RoleCode, ScopeType, ScopeOrganizationId?, ScopeProjectGroupId?, ScopeProjectId?, ExpiresAt?}; Version = task.RoleVersion (event: для fan-out Version = RoleDefinition.Version).
5. Продвижение чекпоинта. Агрегаты роли/назначений не меняются. Меняется только задача: Cursor = последний UserId батча, ProcessedCount += размер батча. На последней странице (страница неполна) — Status = Completed, CompletedAt = now, LeaseExpiresAt очищается.
6. Рассылка событий. На каждого пользователя батча — UserAccessChangedKafkaEvent, topic pin.identity.access.v1, ContractVersion = 1, key/AggregateId = UserId, AggregateType = "User". Payload — шаг 4 (дом — event).
7. Согласованность. Граница — на батч: все события батча и чекпоинт задачи (Cursor/ProcessedCount/Status) фиксируются в одном согласованном изменении. Либо батч разослан и чекпоинт сдвинут, либо ничего: рестарт возобновляет с Cursor, максимум один батч может повториться (снапшот дедуплицируется потребителем). События уходят в шину после фиксации.
8. Результат. Наружу — метрики прогона (задач обработано, пользователей разослано, батчей, backlog очереди). Роль без Active-назначений → задача Completed сразу, событий нет. Инфраструктурный сбой на батче → откат, аренда истекает, задача берётся снова (AttemptCount += 1); при >= MaxAttempts → Failed + алерт (poison).

Согласованность изменения

Элемент	Правило
Граница	на батч: все события батча + чекпоинт задачи (Cursor/ProcessedCount/Status) в одном согласованном изменении
Сохраняемое состояние	access_fanout_tasks (Status/Cursor/ProcessedCount/LeaseExpiresAt/AttemptCount) + исходящие user-access-changed ×N. role_definitions/role_assignments — только чтение
Публикация событий	после фиксации изменения (at-least-once); синхронных внешних вызовов внутри изменения нет
Частичная ошибка	continue: сбой батча логируется и не откатывает уже зафиксированные батчи; задача остаётся InProgress, аренда истекает, следующий поллинг возобновляет с Cursor

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

Сценарий	Правило
Job restart	Возобновление с Cursor — уже разосланные пользователи (UserId <= Cursor) не берутся повторно; чекпоинт фиксируется атомарно с событиями батча
Параллельные экземпляры	Безопасны: аренда задачи — одну задачу обрабатывает ровно один экземпляр; чужие пропускаются
Duplicate item	В пределах задачи уникальные UserId + UserId > Cursor; повтор возможен максимум для последнего незачекпойнченного батча
Повторяемая ошибка	Инфраструктурный сбой — откат, аренда истекает, задача берётся снова (AttemptCount += 1); при >= MaxAttempts → Failed + алерт, без бесконечного retry
Дубликат события	Потребитель дедуплицирует по (UserId, Version)-guard (event); снапшот self-healing (полный список активных назначений), последний по Version выигрывает; доставка at-least-once

Кросс-сервисный контракт

Направление	Гарантия
Ожидает (вход)	Задача fan-out поставлена в очередь атомарно со сменой состава RolePermission роли (set-role-permissions), когда размер роли > InlineFanOutThreshold; в задаче зафиксирован RoleVersion = RoleDefinition.Version на момент постановки. Между постановкой и обработкой роль может быть заархивирована (INV-R5 отзовёт свои назначения) — это не мешает: снапшот отражает актуальные Active-назначения.
Предоставляет (выход)	На каждого уникального пользователя роли — UserAccessChangedKafkaEvent (topic pin.identity.access.v1, key = UserId, ContractVersion = 1, ChangeReason = RolePermissionsChanged, Version = RoleVersion, ActiveAssignments — полный снапшот активных назначений). Дом полей — user-access-changed.md.

Version-ось и переплетение с точечными событиями

Version fan-out-события = RoleDefinition.Version (ось роли), тогда как точечные grant/revoke/expire несут max(RoleAssignment.Version) (ось назначения) — это разные шкалы. Поэтому потребитель НЕ должен полагаться только на строгую монотонность (UserId, Version) между разными ChangeReason: авторитетен полный снапшот ActiveAssignments (self-healing). Порядок per-user сохраняется ключом партиции, но fan-out-батч и параллельное точечное изменение того же пользователя могут прийти в любом относительном порядке — сходимость обеспечивает снапшот-семантика; при неизвестном/расходящемся Version потребитель инвалидирует кэш пользователя целиком (safe fallback — user-access-changed.md «Версионирование и совместимость»).

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

Тип	Правило
Logs	job started (jobRunId); task dequeued (taskId, roleId, attempt); batch processed (roleId, rows, cursor); task completed (roleId, processedCount, durationMs); task failed (roleId, attempt, exception). PII (телефон/email/ФИО) в логи не пишется — payload их не содержит
Metrics	identity_access_fanout_task_duration_ms (histogram); identity_access_fanout_users_total (counter — разослано); identity_access_fanout_queue_backlog (gauge — Pending/просроченные InProgress); identity_access_fanout_errors_total; identity_access_fanout_task_lag_seconds (now − CreatedAt на завершении — контроль SLO); identity_access_fanout_size (пользователей на задачу — сверено с set-role-permissions)
Traces	span на задачу с jobRunId, taskId, roleId, атрибуты batchSize, processedCount; вложенные span на батч
Alerts	backlog очереди растёт (queue_backlog > 100 дольше 5 мин — воркер не успевает/стоит); errors_total растёт; появление Failed-задач (poison, replay через runbook); backlog топика pin.identity.access.v1 > 1000; отсутствие успешного прогона > 5 мин (см. ../README.md «Alerts»)

SLO

Показатель	Цель	Обоснование
Лаг рассылки	fan-out роли из ≤ 50k пользователей завершён за ≤ 5 мин от постановки (p99)	security-чувствительно: устаревшие права у потребителей; проверяется task_lag_seconds
Пропускная способность	≥ BatchSize событий/поллинг-цикл на экземпляр; крупные роли масштабируются несколькими экземплярами	пагинация по ключу без роста стоимости страницы
Идемпотентность	0 «потерянных» пользователей и ≤ 1 повторный батч на пользователя при рестартах/N экземплярах	Cursor-чекпоинт + аренда задач
Доставка событий	at-least-once на каждого пользователя роли; дедуп у потребителя по (UserId, Version)	снапшот-семантика события
Согласованность кэшей	потребители (crm/catalog/chessboard/gateway) сбрасывают кэш авторизации < 1 мин после доставки	event: user-access-changed («Потребители»)

Связанный runbook

Сценарий	Runbook
Backlog очереди растёт (воркер не успевает)	увеличить BatchSize/число экземпляров, проверить блокировки PostgreSQL и лаг доставки pin.identity.access.v1
Воркер стоит (нет прогонов)	проверить сервис/pod, конфиг PollingInterval, просроченные InProgress-аренды
Failed-задачи (poison)	access fan-out replay: сбросить Status → Pending, AttemptCount = 0 после устранения причины; снапшот-семантика делает повтор безопасным
Массовый рост errors_total	проверить доступность PostgreSQL/схемы и откаты; задачи остаются Pending/InProgress и дообработаются после восстановления

Тесты

Тест	Ожидание
Empty role	роль без Active-назначений — задача Completed сразу, событий нет
Массовая рассылка	роль с > BatchSize пользователей — событие на каждого уникального UserId, ProcessedCount = число пользователей, проход берёт все страницы
Полный снапшот	payload пользователя содержит все его активные назначения (не только изменённую роль), RoleCode резолвится
Идемпотентность/рестарт	падение после фиксации батча → возобновление с Cursor, уже разосланные не повторяются, дублей сверх одного батча нет
Параллельные экземпляры	две копии воркера не берут одну задачу дважды (аренда)
Version guard	event.Version = RoleVersion задачи; потребитель с бо́льшим применённым Version отбрасывает событие
Poison	AttemptCount >= MaxAttempts → Failed + алерт, очередь не блокируется другими задачами
Cancellation	остановка между батчами — graceful stop, зафиксированные батчи сохранены

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

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

• API (1): Identity Company-User API — реестр методов и общие правила области
• Фоновые задачи (1): Background Job: ExpireSessionsJob

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

• Дом полей/инвариантов: ../domain/role.md (RoleDefinition/RolePermission/RoleAssignment, INV-R5, GrantPermission/RevokePermission), ../domain/permission.md (каталог).
• Постановка/синхронный путь: ../api/company-user/set-role-permissions.md (порог inline→job), ../api/company-user/grant-role-assignment.md, ../api/company-user/revoke-role-assignment.md.
• Контракт события (дом payload): ../events/user-access-changed.md.
• Обзор/эксплуатация: ../README.md («jobs … access fan-out», «Alerts»); соседний polling-воркер — ../../user-activity/background/activity-closer.md.
• Решения: ADR-0056 (PostgreSQL-only, без Redis для координации), ADR-0052 (tenant/scope).