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

Background Job: project-aggregates-recalc (пересчёт денормализованных агрегатов)

Пересчёт денормализованных счётчиков и ценовых границ иерархии chessboard из единственного source of truth — chessboard.property-objects (и chessboard.layouts для уровня группы). Поля-агрегаты — «дома» в project.md, building.md, layout.md, layout-group.md; здесь описан только job их обновления (дельта, не переописание полей).

Назначение​

Пересчёт денормализованных агрегатов проекта после изменений цен/статусов объектов недвижимости и структуры планировок. Держит согласованными: Project.Price (границы цен свободных объектов недвижимости), Project.PropertyObjectsTotalCount/PropertyObjectsFreeCount; Building.PropertyObjectsTotalCount/PropertyObjectsFreeCount; Layout.PropertyObjectsCount; LayoutGroup.LayoutsCount/Area. Поля питают ленту/карточку catalog (фильтр «ценовой диапазон», «кол-во объектов») и измерения dev-stats; пересчёт идемпотентен и восстановим — значение есть чистая функция состояния propertyObjects/layouts, повторный прогон даёт тот же результат. Триггер — доменное событие ProjectAggregatesRecalcRequestedEvent c debounce-коалесингом 30 с по ProjectId; конкурентность — построчная блокировка PostgreSQL без Redis (ADR-0056). После пересчёта snapshot-события переопубликуются для потребителей.

MetadataЗначение
Owning teamPIN Platform Core
Статус документаdraft
Область действиясервис chessboard, проект Pin.Chessboard.Background (jobs), таблица chessboard.project_recalc_requests
ЗависимостиPostgreSQL (chessboard.property-objects/layouts, целевые агрегаты, очередь пересчёта, очередь исходящих событий), Kafka (публикация событий); upstream-раисеры — мутации объектов недвижимости/планировок, consumer crm-property-object-status, talent-chessboard-sync, job property-object-hold-expiration
Assumptions / ограниченияdebounce/очередь — на PostgreSQL (ADR-0056, без Redis; claim-lease при кратном росте нагрузки — вне scope); Building.PropertyObjects*Count — вне payload building-события (дом building-upserted.md): job обновляет колонки, но событие на чистый пересчёт счётчиков не шлёт; пороги (debounce 30 с, cap 5 мин, тик 5 с, batch 50, poison 5) — конфигурируемые
Change contextgreenfield (объявлен в ../README.md «Background», отдельного файла job не было)

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

ПолеЗначение
Сервисchessboard
Тип jobQueue worker (слив debounce-очереди пересчёта), event-triggered
Triggerдоменное событие ProjectAggregatesRecalcRequestedEvent (debounce 30 с, коалесинг по ProjectId) + внутренний тик воркера 5 с на слив созревших строк
Критичностьmedium (отказ → устаревшие счётчики/цены в ленте; продажи, статусы и цены объектов недвижимости не блокируются)
Actor / scopesystem, cross-tenant по построению (обходит все проекты, scope-фильтры отключены); TenantId каждого проекта берётся из строки очереди для partition key событий

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

АспектПравило
Scheduleпериодический воркер, тик 5 с (UTC); обрабатываются строки project_recalc_requests c DueAt <= now(). Сам пересчёт — событийный (debounce 30 с)
Distributed lockБез Redis (ADR-0056): конкурентность — построчная блокировка с пропуском занятых (FOR UPDATE SKIP LOCKED) по строке очереди (одна строка = один проект = единица коалесинга). N инстансов воркера безопасно делят очередь; занятые строки пропускаются без ожидания
Batch size50 строк за тик, сортировка по DueAt asc; каждый проект — своя транзакция
Max durationпересчёт одного проекта p99 < 2 с (индексные агрегации); тик < 30 с
Cancellationпри остановке воркера текущий проект коммитится или откатывается целиком, недообработанные строки остаются в очереди

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

Job-owned операционная строка очереди (не доменный агрегат: доменных событий/инвариантного поведения нет, tenant-фильтры не применяются — system-scope). Таблица chessboard.project_recalc_requests.

ПолеТипОбязательностьОграничения (PK/index)ИсточникОписание
ProjectIdGuidДаPK (естественный ключ = коалесинг: одна строка на проект)enqueue-handlerПроект для пересчёта; логическая ссылка на chessboard.projects.id.
TenantIdGuidДа—enqueue-handler= DeveloperId проекта; partition key re-публикуемых событий.
DueAtDateTimeOffsetДаindex ix_project_recalc_due btree(due_at) — выборка «созревших»enqueue-handlerДедлайн debounce: строка созревает в DueAt.
FirstRequestedAtDateTimeOffsetДа—enqueue-handlerЯкорь starvation-cap (страховка от вечного откладывания при непрерывных правках).
RequestedCountintДа>= 1, default 1enqueue-handlerСколько событий сколлапсировано с создания строки (метрика коалесинга).
LastReasonRecalcTriggerReasonДа.HasConversion<string>(), varchar(32)enqueue-handlerПричина последнего запроса (enum-string): PropertyObjectStatusChanged / PropertyObjectPriceChanged / PropertyObjectUpserted / LayoutChanged / TalentSync / Crud.
AttemptCountintДа>= 0, default 0drain-handlerЧисло неуспешных прогонов (poison-контроль).
LastErrorstring?Нетvarchar(1024)drain-handlerУсечённый текст последней инфраструктурной ошибки.
CreatedAt / UpdatedAtDateTimeOffsetДа—appТайминги строки.

Постановка в очередь (debounce, trailing). Событие ProjectAggregatesRecalcRequestedEvent(ProjectId, TenantId, Reason) поднимается агрегатами (дельта — «Раисеры» ниже) при их изменении и в том же согласованном изменении добавляет/обновляет строку очереди по ProjectId (триггер не теряется): новая строка → DueAt = now + 30s, FirstRequestedAt = now; существующая → DueAt = least(now + 30s, FirstRequestedAt + 5 min) (30 с тишины, но ≤ 5 мин от первого запроса), RequestedCount += 1.

Раисер (source of truth события)ДомКогда
PropertyObjectEntityproperty-object.mdсмена Status (PropertyObjectStatusChangedEvent), Price, паспортных полей, LayoutId, create/soft-delete
LayoutEntitylayout.mdcreate/archive, смена LayoutGroupId, TotalArea (влияет на LayoutGroup.Area*/LayoutsCount)
BuildingEntitybuilding.mdcreate/archive (инициализация счётчиков нового ряда)
синк / CRMconsumerstalent-chessboard-sync и crm-property-object-status при изменении статусов/цен объектов недвижимости

Предусловия И Постусловия​

Предусловия (до слива строки).

  • Раисер поднял ProjectAggregatesRecalcRequestedEvent и в том же согласованном изменении, что и мутация объекта недвижимости/планировки, добавил/обновил строку project_recalc_requests (триггер не теряется).
  • chessboard.property-objects / chessboard.layouts — актуальный source of truth: мутации, породившие запрос, уже закоммичены (значения-агрегаты суть чистая функция их текущего состояния).
  • Строка очереди созрела (DueAt <= now()), либо сработал starvation-cap (FirstRequestedAt + 5 min).

Постусловия (после успешного слива строки проекта).

  • Денорм-агрегаты проекта, зданий, планировок и групп планировок равны count(*)/min/max по текущему состоянию propertyObjects/layouts (идемпотентность by construction: повтор даёт те же значения).
  • Строка project_recalc_requests удалена — debounce-окно закрыто; запрос, пришедший во время пересчёта, вставит новую строку.
  • Переопубликованы snapshot-события только изменившихся рядов: Project/Layout/LayoutGroup UpsertedKafkaEvent (key = Id, монотонный Version); Building — не публикуется (счётчики вне payload).
  • Version затронутых рядов инкрементирован, UpdatedAt = now().
  • При архивном/удалённом проекте — строка удалена, агрегаты и события не трогаются (no-op success).

Алгоритм​

Описан цикл обработки одной строки очереди (drain); постановка в очередь (debounce-upsert) — выше.

  1. Контекст и доступ. Выполняет system-воркер (scope-фильтры отключены, cross-tenant). Воркер выбирает батч созревших строк очереди (DueAt <= now(), сортировка DueAt asc, до 50 строк) под построчной блокировкой с пропуском занятых: параллельный воркер пропустит занятую строку (без двойного пересчёта), а параллельная постановка в очередь дождётся коммита и после удаления строки вставит новую (не теряется).

  2. Проверка. По id строки очереди читается снапшот проекта: если проекта нет либо он архивирован (Status = Archived, DeletedAt != null) — строка очереди удаляется, обработка завершается no-op (info-лог): архивный проект агрегатов не несёт. Прочие статусы допустимы (Draft/Published/Hidden); tenant не фильтруется (system-scope).

  3. Что читаем. Агрегаты не загружаются — пересчёт выполняется одним набором агрегирующих запросов. Для решения о переопубликации читаются лёгкие проекции текущих (до пересчёта) денорм-счётчиков проекта, планировок и групп планировок — для сравнения на шаге 5.

  4. Пересчёт (source of truth). Каждое целевое поле — чистая функция состояния propertyObjects/layouts:

    Целевое полеФормула (по ProjectId = @p)Источник
    Project.Price (Min/Max)min/max(price) по freePropertyObjectchessboard.property-objects
    Project.PropertyObjectsFreeCountcount(*) по freePropertyObjectchessboard.property-objects
    Project.PropertyObjectsTotalCountcount(*) по activePropertyObjectchessboard.property-objects
    Building.PropertyObjectsFreeCount / PropertyObjectsTotalCountте же предикаты, group by building_idchessboard.property-objects
    Layout.PropertyObjectsCountcount(*) по activePropertyObject, group by layout_idchessboard.property-objects
    LayoutGroup.LayoutsCountcount(*) живых планировок группыchessboard.layouts
    LayoutGroup.Area (Min/Max)min/max(total_area) живых планировок группыchessboard.layouts

    где freePropertyObject = (deleted_at is null AND status = 'Free'), activePropertyObject = (deleted_at is null AND status <> 'Hidden') (Hidden и soft-deleted не учитываются); живая планировка — deleted_at is null. Пустой набор: цены/площади → NULL, счётчики → 0; границы (price_max >= price_min, area_max >= area_min) — по построению (min <= max). Version += 1, UpdatedAt = now().

  5. Что меняется. Счётчики и ценовые/площадные границы проекта, зданий, планировок и групп планировок обновляются одним набором изменений (без загрузки агрегатов): projects (цены, счётчики, версия), buildings (счётчики по своим объектам), layouts (счётчик объектов), layout_groups (число планировок, границы площадей). Меняются только скалярные счётчики существующих строк. Гонка с HTTP-правкой того же ряда сериализуется на построчной блокировке; job пишет иные колонки, чем API, — конфликтов оптимистической блокировки нет.

  6. Какие события фиксируются. После пересчёта переопубликуются snapshot-события изменённых рядов — только там, где денорм-колонки изменились (сравнение с шагом 3), топики pin.chessboard.<entity>-upserted.v1 (ContractVersion = 1, key = Id):

    • ProjectUpsertedKafkaEvent — при изменении Price/PropertyObjectsTotalCount/PropertyObjectsFreeCount (дом project-upserted.md);
    • LayoutUpsertedKafkaEvent — при изменении PropertyObjectsCount (дом layout-upserted.md);
    • LayoutGroupUpsertedKafkaEvent — при изменении LayoutsCount/Area (дом layout-group-upserted.md).

    BuildingUpsertedKafkaEvent не публикуется при чистом пересчёте счётчиков: Building.PropertyObjects*Count — вне payload building-события (дом building-upserted.md, «Когда публикуется»); building-счётчики потребители (catalog/dev-stats) выводят из property-object-потока (PropertyObjectUpsertedKafkaEvent/PropertyObjectStatusChangedKafkaEvent). Колонки здания в БД (шаг 5) всё равно обновляются — для API-чтений/шахматки. Ключ Id + монотонный Version делают повторную публикацию безопасной (потребители — upsert-by-id).

  7. Согласованность. Обновления счётчиков, записи исходящих событий и удаление строки очереди применяются в одном согласованном изменении. Удаление строки завершает debounce-окно: пришедший во время пересчёта запрос вставит новую строку. Внешних вызовов нет; события уходят потребителям после коммита (at-least-once).

  8. Результат. Успех → строка удалена, метрики записаны. Инфраструктурная ошибка → откат всей транзакции (строка остаётся, DueAt не сдвинут), затем AttemptCount += 1/LastError; повтор на следующем тике.

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

ЭлементПравило
Unit of workPer project (одна строка очереди = одна транзакция)
Сохраняемое состояниеобновление счётчиков projects/buildings/layouts/layout_groups + исходящие события (project/layout/layout-group) + удаление строки очереди
External callsНет (никаких HTTP/провайдеров); события — только транзакционно, после коммита (at-least-once)
Частичная ошибкаContinue-по-батчу: сбой одного проекта откатывает только его транзакцию; строка остаётся, остальные проекты обрабатываются

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

СценарийПравило
Job restartСтроки project_recalc_requests переживают рестарт; воркер продолжает слив по DueAt
Duplicate itemНевозможен: PK = ProjectId (коалесинг). Параллельный воркер отсекается FOR UPDATE SKIP LOCKED
Повторный прогонПересчёт — чистая функция propertyObjects/layouts → те же значения (идемпотентно by construction); повторная публикация снапшота безопасна (key=Id, монотонный Version, потребители upsert-by-id)
Повторяемая ошибкаОткат транзакции, AttemptCount += 1, LastError; retry на следующем тике (5 с)
Poison itemAttemptCount >= 5 → алерт CHESSBOARD_RECALC_POISON, строка остаётся для ручного разбора (runbook); слив прочих проектов не блокируется
StarvationНепрерывные правки не откладывают пересчёт бесконечно: DueAt <= FirstRequestedAt + 5 min (cap) → форс-прогон

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

ТипПравило
Logsrecalc_batch_started (size), project_recalculated (ProjectId, длительность, republished counts), project_skipped_archived, project_recalc_failed (ProjectId, AttemptCount, error)
Metricschessboard_recalc_queue_depth (gauge); chessboard_recalc_due_backlog (строк c DueAt < now); chessboard_recalc_process_duration_seconds (histogram); chessboard_recalc_runs_total{result}; chessboard_recalc_coalesced_events_total (Σ RequestedCount); chessboard_recalc_republished_total{entity}; лаг доставки событий pin.chessboard.*-upserted.v1
TracescorrelationId тика + per-project span; атрибуты ProjectId, TenantId, LastReason
DashboardGrafana chessboard-overview (панель recalc: depth, backlog, p99, republished by entity)
AlertsCHESSBOARD_RECALC_BACKLOG (due_backlog > порога > 5 мин — воркер завис/отстаёт); process p99 > SLO; рост runs_total{result=error}; CHESSBOARD_RECALC_POISON; воркер не тикал > 2 интервалов (нет прироста runs_total — liveness)

SLO Цикла​

ПоказательЦельОбоснование
E2E-лаг «событие → обновлённые счётчики/цены»p99 < ~40 с (debounce 30 с + обработка < 2 с + тик 5 с + запас)eventually consistent денормализация; ленте/карточке catalog секундная свежесть не требуется
Слив backlogdue_backlog → 0 в пределах одного-двух тиков после DueAtTake=50/тик × N инстансов покрывает объём проектов (тысячи, ADR-0056)
Идемпотентность0 расхождений денормализации со count(*)/min/max по propertyObjects/layouts; повторный прогон не меняет значенийчистая функция состояния + построчная блокировка с пропуском занятых
Доставка событийat-least-once на каждый изменённый ряд; дедуп у потребителей по Id+Versionтранзакционная доставка + upsert-by-id у consumer-ов

Связанный Runbook​

СценарийRunbook
Растёт due_backlog / завис воркер../runbooks/project-aggregates-recalc.md (будущий путь): lock-contention project_recalc_requests, число инстансов воркера, при необходимости рестарт
Poison-проект (AttemptCount >= 5)разобрать LastError; проверить целостность propertyObjects/layouts проекта; после фикса — сбросить AttemptCount, DueAt = now()
Ручной форс-пересчётadmin-операция: Upsert project_recalc_requests(ProjectId, DueAt=now()) — пересинк агрегатов проекта
Расхождение денормализациисверка счётчиков с count(*)/min/max по propertyObjects/layouts; форс-пересчёт устраняет дрейф
Лаг доставки событий pin.chessboard.*-upserted.v1../runbooks/outbox-backlog.md (будущий путь)

Тесты​

ТестОжидание
Empty batchнет созревших строк → тик no-op success
Проект без свободных объектов недвижимостиPrice = NULL (обе границы), PropertyObjectsFreeCount = 0, PropertyObjectsTotalCount по activePropertyObject; ProjectUpsertedKafkaEvent опубликован
Проект без объектов недвижимости/планировоквсе счётчики 0, цены/площади NULL
Coalescingburst из N событий одного проекта → 1 строка, 1 прогон, RequestedCount = N
Starvation capнепрерывные правки > 5 мин → форс-прогон на границе cap
Duplicate runповторный пересчёт того же состояния → идентичные значения; снапшот с новым Version только при diff
Building counters — без событияпересчёт Building.PropertyObjects*Count обновляет колонки, но BuildingUpsertedKafkaEvent не публикуется (вне payload)
Partial failureсбой проекта откатывает его транзакцию, строка остаётся, AttemptCount += 1; прочие проекты обработаны
Concurrencyдва воркера на одной строке → SKIP LOCKED, ровно один пересчёт
Deleted/archived projectстрока удалена, событий нет, no-op success
Cancellationотмена между проектами — graceful; текущий проект коммитится/откатывается целиком

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

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

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