From c1259c310561273a11bf80baca2d1a0b8097fd02 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=94=D0=BC=D0=B8=D1=82=D1=80=D0=B8=D0=B9=20=D0=A1=D0=BE?= =?UTF-8?q?=D0=BB=D0=BE=D0=B2=D1=8C=D0=B5=D0=B2?= Date: Sat, 6 Jun 2026 23:47:19 +0300 Subject: [PATCH] =?UTF-8?q?pcp-request-service:=20=D0=B0=D1=83=D0=B4=D0=B8?= =?UTF-8?q?=D1=82-=D1=84=D0=B8=D0=BA=D1=81=D1=8B=20RS1/RS3/RS6/RS7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit RS1 (#29): планировщик истечения заявок — RequestExpiryService (@Scheduled + FOR UPDATE SKIP LOCKED, мультиинстанс-safe), refreshStatus вызывается и при создании заявки. RS7 (#35): единый топик pcp.request.status.v1 → geoportal (COMPLETED/EXPIRED в payload), замена orphan pcp.request.completed.v1; эмиссия из matching и expiry через тот же outbox; backoff/next_attempt_at/cap для outbox FAILED. RS3 (#31): DefaultErrorHandler + DeadLetterPublishingRecoverer (DLQ pcp.route.georeference-dlq.v1), bounded retry, контрактные ошибки в DLQ. RS6 (#34): проброс coverageRequiredPercent из API, разведение терминов покрытие/важность. Доки: бэклог, аудит сервиса заданий, карты взаимодействий. RS2 (#30) отложено до #28, RS5 (#33) — до роста требований к точности. --- config-repo/pcp-request-service.yaml | 21 +- docs/architect/PCP_ARCHITECTURE.md | 8 +- docs/architect/PCP_SERVICE_INTERACTIONS.md | 4 +- docs/аудит-сервис-заданий.md | 310 ++++++++++++++++++ docs/бэклог-pcp-tgu-ops-ui.md | 42 ++- .../config/KafkaConsumerConfig.kt | 131 ++++++++ .../requestservice/config/OutboxProperties.kt | 9 +- .../config/OutboxTransactionConfig.kt | 11 + .../config/RequestExpiryProperties.kt | 19 ++ .../config/RouteConsumerProperties.kt | 28 ++ .../requestservice/dto/RequestApiDtos.kt | 9 + .../entity/OutboxEventEntity.kt | 8 + .../repository/OutboxEventRepository.kt | 15 +- .../repository/RequestRepository.kt | 23 ++ .../service/OutboxPublisherService.kt | 54 ++- .../service/RequestExpiryService.kt | 118 +++++++ .../requestservice/service/RequestService.kt | 12 +- ...rvice.kt => RequestStatusOutboxService.kt} | 34 +- .../service/RouteMatchingService.kt | 4 +- .../src/main/resources/application.yaml | 5 +- .../db/migration/V1__create_schema.sql | 3 + .../requestservice/MigrationSchemaTest.kt | 3 + .../config/KafkaConsumerConfigTest.kt | 175 ++++++++++ .../RequestControllerOpenApiSkeletonTest.kt | 7 +- .../dto/RequestApiDtoContractTest.kt | 25 ++ .../OutboxEventRepositoryJpaTest.kt | 66 +++- .../repository/RequestRepositoryJpaTest.kt | 104 ++++++ .../service/OutboxPublisherServiceTest.kt | 86 +++-- .../RequestCompletedOutboxServiceTest.kt | 65 ---- .../service/RequestExpiryServiceTest.kt | 203 ++++++++++++ .../service/RequestServiceTest.kt | 20 +- .../service/RequestStatusOutboxServiceTest.kt | 107 ++++++ .../service/RouteMatchingServiceTest.kt | 10 +- 33 files changed, 1600 insertions(+), 139 deletions(-) create mode 100644 docs/аудит-сервис-заданий.md create mode 100644 services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/config/KafkaConsumerConfig.kt create mode 100644 services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/config/RequestExpiryProperties.kt create mode 100644 services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/config/RouteConsumerProperties.kt create mode 100644 services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/service/RequestExpiryService.kt rename services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/service/{RequestCompletedOutboxService.kt => RequestStatusOutboxService.kt} (55%) create mode 100644 services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/config/KafkaConsumerConfigTest.kt create mode 100644 services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/repository/RequestRepositoryJpaTest.kt delete mode 100644 services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/service/RequestCompletedOutboxServiceTest.kt create mode 100644 services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/service/RequestExpiryServiceTest.kt create mode 100644 services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/service/RequestStatusOutboxServiceTest.kt diff --git a/config-repo/pcp-request-service.yaml b/config-repo/pcp-request-service.yaml index ef094a1..239cacd 100644 --- a/config-repo/pcp-request-service.yaml +++ b/config-repo/pcp-request-service.yaml @@ -3,9 +3,18 @@ server: pcp: outbox: - request-completed-topic: pcp.request.completed.v1 + request-status-topic: pcp.request.status.v1 publish-batch-size: 50 publish-fixed-delay-ms: 5000 + max-attempts: 10 + initial-backoff-ms: 1000 + max-backoff-ms: 60000 + request: + expiry: + enabled: true + fixed-delay-ms: 60000 + batch-size: 200 + max-batches-per-run: 50 spring: boot: @@ -60,10 +69,12 @@ logging: app: kafka: topics: - request: pcp.request.in.v1 route: pcp.route.georeference.v1 - request: - cleanup-retention-days: 30 - cleanup-cron: "0 0 3 * * *" + consumer: + route: + dlq-topic: pcp.route.georeference-dlq.v1 + max-attempts: 5 + backoff-ms: 1000 + max-backoff-ms: 30000 grid: enabled: true diff --git a/docs/architect/PCP_ARCHITECTURE.md b/docs/architect/PCP_ARCHITECTURE.md index 3f96ce4..51a7197 100644 --- a/docs/architect/PCP_ARCHITECTURE.md +++ b/docs/architect/PCP_ARCHITECTURE.md @@ -14,7 +14,7 @@ | Сервис | Зона ответственности | REST base | Хранит | Kafka (in / out) | Зрелость* | |---|---|---|---|---|---| -| **pcp-request-service** | Приём и жизненный цикл заявок на съёмку; сетка Земли (earth-grid), геопривязка ячеек, сопоставление маршрутов с заявками | `/v1/requests`, `/v1/cells`, `/v1/grid` | заявки, ячейки сетки, outbox | in: `pcp.route.georeference.v1` · out: `pcp.request.completed.v1` (outbox) | 🟢 высокая | +| **pcp-request-service** | Приём и жизненный цикл заявок на съёмку; сетка Земли (earth-grid), геопривязка ячеек, сопоставление маршрутов с заявками | `/v1/requests`, `/v1/cells`, `/v1/grid` | заявки, ячейки сетки, outbox | in: `pcp.route.georeference.v1` · out: `pcp.request.status.v1` (outbox → geoportal) | 🟢 высокая | | **pcp-route-processing-service** | Приём «паспортов маршрутов», нормализация/геопривязка, ретрансляция дальше; собственный outbox + DLQ | `/api/route-processing` (test) | outbox маршрутов | in: `pcp.request.survey-georeference.v1` · out: `pcp.route.georeference.v1`, `pcp.route.in.v1`, DLQ | 🟢 высокая | | **pcp-satellite-catalog-service** | Каталог КА (источник истины): спутники, группы, observation/slot-профили | `/api/satellites`, `/api/satellite-groups` | каталог КА, профили | out: `pcp.satellites` (`SatelliteDeletedEvent`) | 🟡 средняя (мало тестов) | | **pcp-ballistics-service** | Баллистика: орбиты, ЗРВ/RVA, узлы, трассы, time-extract по TLE | `/api/satellites`, `/api/obj-view`, `/api/tle`, `/api/rva` | орбитальные данные | in: `pcp.tle` (IC/ICRV/TLE фильтры), `pcp.satellites` | 🟡 средняя | @@ -89,7 +89,7 @@ flowchart LR EXT[Внешний источник / UI ⚠] -->|pcp.request.survey-georeference.v1| RP[route-processing] RP -->|pcp.route.georeference.v1| REQ[request-service] RP -->|pcp.route.in.v1| MP[mission-planing] - REQ -->|pcp.request.completed.v1 ⚠ потребитель| Q1{{?}} + REQ -->|pcp.request.status.v1 · COMPLETED/EXPIRED| GEO[geoportal · внешний] CAT[satellite-catalog] -->|pcp.satellites / SatelliteDeletedEvent| BAL[ballistics] CAT --> TLEM[tle-monitoring] @@ -175,7 +175,7 @@ sequenceDiagram | `pcp.route.georeference.v1` | route-processing | request-service | `RoutePassportDto` (геопривязанный) | ⚠ | | `pcp.route.in.v1` | route-processing | mission-planing (`ModeStatusChangedEvent`) | processed route / mode-status | ⚠ | | `pcp.request.in.v1` | request-service / UI ⚠ | ⚠ требует уточнения | `RequestPlacedEvent` | ⚠ | -| `pcp.request.completed.v1` | request-service (outbox) | ⚠ требует уточнения | RequestCompleted | `aggregateId` (UUID заявки) | +| `pcp.request.status.v1` | request-service (outbox) | **geoportal** (внешний) | `REQUEST_TERMINAL_STATUS` (COMPLETED/EXPIRED) | `aggregateId` (UUID заявки) | | `pcp.slots.status.v1` | slots-service | mission-planing, slots-service (self) | `BookedSlotsStatusChangedEvent` | ⚠ | | `pcp.tgu.satellite-plan-decision.v1` | Mission Ops / КА ⚠ | tgu-service | `PlanDecisionMessage` (ACCEPTED/REJECTED, `attemptId`) | ⚠ | | `pcp.tgu.visibility-windows-changed.v1` | tgu-service | tgu-service | `VisibilityWindowsChangedMessage` | ⚠ | @@ -255,7 +255,7 @@ tgu : GET /api/plans/{spacecraftId} ; /api/platforms ; Эти пункты помечены `⚠` выше — их стоит зафиксировать до изменений: 1. Producer топиков `pcp.request.survey-georeference.v1` и `pcp.request.in.v1` (внешняя система или UI?). -2. Consumer `pcp.request.completed.v1` — кто слушает завершённые заявки (предположительно complex-mission/планирование). +2. ~~Consumer `pcp.request.completed.v1`~~ — уточнено: потребитель терминальных статусов заявки = внешний **geoportal**, топик переименован в `pcp.request.status.v1` (COMPLETED/EXPIRED, #35). 3. Producer `pcp.tgu.satellite-plan-decision.v1` — реальный Mission Ops/КА vs текущая имитация в `FlowService`. 4. Точные DTO payload для `pcp.tle`, `pcp.route.in.v1`, `pcp.slots.status.v1` — зафиксировать схемы. 5. Возможная циклическая зависимость `complex-mission ⇄ coverage-scheme` в рантайме. diff --git a/docs/architect/PCP_SERVICE_INTERACTIONS.md b/docs/architect/PCP_SERVICE_INTERACTIONS.md index 9e1dd06..bd0c5ac 100644 --- a/docs/architect/PCP_SERVICE_INTERACTIONS.md +++ b/docs/architect/PCP_SERVICE_INTERACTIONS.md @@ -55,7 +55,7 @@ | `pcp.route.georeference.v1` | `pcp-route-processing-service` (`route-georeference`) | `pcp-request-service` (`RequestKafkaListener.consumeRoute`) | геопривязка маршрута к заявке | | `pcp.request.survey-georeference.v1` | (инжект `RouteProcessingTestController` / внешний источник) | `pcp-route-processing-service` (`RoutePassportMessageListener`, вход) | входной маршрутный паспорт на обработку | | `pcp.request.survey-georeference-dlq.v1` | `pcp-route-processing-service` (DLQ) | — | dead-letter для входных паспортов | -| `pcp.request.completed.v1` | `pcp-request-service` (`OutboxPublisherService`, outbox) | *(в репозитории нет консьюмера)* | заявка завершена (REQUEST_COMPLETED) | +| `pcp.request.status.v1` | `pcp-request-service` (`OutboxPublisherService`, outbox) | **geoportal** (внешний, в репо нет) | терминальный статус заявки `COMPLETED`/`EXPIRED` (`REQUEST_TERMINAL_STATUS`) | | `pcp.request.in.v1` | — | *(объявлен в конфиге `pcp-request-service`, активного слушателя нет)* | входящие заявки (legacy/зарезервировано) | ### Поток данных (упрощённо) @@ -76,7 +76,7 @@ route-processing ──pcp.route.georeference.v1──> request-service (вход) pcp.request.survey-georeference.v1 ──> route-processing mission-planing ──pcp.slots.status.v1(BookedSlots)──────> slots-service -request-service ──pcp.request.completed.v1 (outbox)─────> (нет консьюмера в репо) +request-service ──pcp.request.status.v1 (outbox)────────> geoportal (внешний) ``` ### ТГУ-топики (`pcp.tgu.*`) diff --git a/docs/аудит-сервис-заданий.md b/docs/аудит-сервис-заданий.md new file mode 100644 index 0000000..dc353b0 --- /dev/null +++ b/docs/аудит-сервис-заданий.md @@ -0,0 +1,310 @@ +# Аудит сервиса заданий — pcp-request-service + +Дата: 2026-06-06. Автор: Claude (Opus 4.8). +Охват: сервис `pcp-request-service` («сервис заданий») — задача [#26](бэклог-pcp-tgu-ops-ui.md). +Метод: статический разбор кода (контроллеры, Kafka-listener, сервисы сопоставления и жизненного +цикла, outbox, earth-grid), схема БД и Flyway-миграция, конфиг в `config-repo`, обзор тестов. + +Парные документы: общий [аудит-проекта-2026-06-06.md](аудит-проекта-2026-06-06.md), +карты [PCP_ARCHITECTURE.md](architect/PCP_ARCHITECTURE.md) и +[PCP_SERVICE_INTERACTIONS.md](architect/PCP_SERVICE_INTERACTIONS.md). +Стык с сервисом паспорта маршрута (`pcp-route-processing-service`) — отдельная задача +[#28](бэклог-pcp-tgu-ops-ui.md), здесь только обозначен. + +## Сводка по состоянию + +Базовое здоровье — хорошее. Сервис держит жизненный цикл заявок на съёмку, ведёт учёт отснятой +территории геометрическим вычитанием и сопоставляет приходящие паспорта маршрутов с заявками. +Конкурентность и работа в нескольких инстансах продуманы: per-row пессимистичные блокировки, +`SKIP LOCKED` в outbox, advisory-lock на пересборку сетки, дедуп по уникальным ключам, +transactional outbox. 19 тестовых файлов с покрытием ключевых сервисов и репозиториев. + +Главные риски — не в инфраструктуре, а в **бизнес-логике закрытия и сопоставления**: +закрытие «по времени» само по себе не срабатывает (нет планировщика истечения), а матчинг +засчитывает покрытие без проверки совместимости аппарата. Приоритеты: **RS1**, **RS2** (высокий), +далее RS3/RS4 (средний), затем долг RS5–RS7. + +Порт 7005. БД `pcp_requests` (PostgreSQL + PostGIS), Flyway `V1`, `ddl-auto: validate`. +Пакет `org.nstart.dep265.requestservice` (несогласован с остальными — см. B3 общего аудита). + +## Назначение и контур + +Заявки создаются **только по HTTP**; паспорта маршрутов приходят **из Kafka**; завершение +заявки уходит наружу через **transactional outbox**. + +``` + HTTP (оператор / pcp-tgu-ops-ui через Vite-прокси) + POST/GET/DELETE /v1/requests · GET /v1/cells · /v1/grid + │ + ▼ + ┌───────────────────────────────────┐ + Kafka │ pcp-request-service │ Kafka + IN ───▶│ заявки · покрытие · lifecycle · │── OUT ──▶ pcp.request.status.v1 → geoportal + │ earth-grid · transactional outbox │ (outbox; COMPLETED/EXPIRED, #35) + └───────────────────────────────────┘ + ▲ │ + │ ▼ +pcp.route.georeference.v1 DB: pcp_requests (PostgreSQL + PostGIS) + │ + pcp-route-processing-service ◀── связь для задачи #28 + («сервис паспорта маршрута») +``` + +### Карта взаимодействий + +| Направление | Канал | Контрагент | Примечание | +|---|---|---|---| +| IN (HTTP) | `/v1/requests` (POST/GET/`/map`/`/{id}`/`/{id}/cells`/DELETE) | UI / оператор | заявки заводятся только здесь | +| IN (HTTP) | `/v1/cells` (+`/priority-map`), `/v1/grid` (settings, `/rebuild`) | UI / оператор | сетка и карта важности | +| IN (Kafka) | `pcp.route.georeference.v1` | **pcp-route-processing-service** | `KafkaMessage`, контракт из `pcp-types-lib` | +| OUT (Kafka) | `pcp.request.status.v1` | **geoportal** (внешний) | outbox `REQUEST_TERMINAL_STATUS`; терминальный статус `COMPLETED`/`EXPIRED` в payload (#35) | +| — | `pcp.request.in.v1` (`app.kafka.topics.request`) | — | в конфиге есть, **слушателя нет** — мёртвый конфиг (RS4) | +| DB | `pcp_requests` (Postgres + PostGIS) | — | Flyway `V1`, `ddl-auto: validate` | + +HTTP-поверхность: [RequestController](../services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/controller/RequestController.kt), +[CellsController](../services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/controller/CellsController.kt), +[GridManagementController](../services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/controller/GridManagementController.kt). + +## Доменная модель и БД + +Источник: [V1__create_schema.sql](../services/pcp-request-service/src/main/resources/db/migration/V1__create_schema.sql). + +- **`requests`** — заявка/задание: `geometry` (исходная) + `remaining_geometry` (что осталось снять), + `geometry_area` / `remaining_area`, `coverage_required_percent`, окно `begin/end_date_time`, + `status`, `match_count`, `last_matched_at`, `completed_at`, мягкое удаление `deleted_at`, + `@Version version` (оптимистичная блокировка). Статусы CHECK-ограничением: + `ACCEPTED · ACTIVE · COMPLETED · EXPIRED · DELETED`. +- **`request_optics_params` / `request_rsa_params`** — параметры съёмки (разрешение, угол Солнца, + облачность / поляризация, интерферометрия). Хранятся, но в матчинге **не используются** (RS2). +- **`request_route_matches`** — аудит и идемпотентность сопоставления: `UNIQUE(request_id, route_id)`, + `original_intersection_*`, `applied_intersection_*`, `coverage_delta_percent`, `contributes_to_coverage`. +- **`outbox_events`** — transactional outbox: `UNIQUE(event_type, aggregate_id)`, статусы `NEW/PUBLISHED/FAILED`. +- **`earth_cells` / `request_cells` / `earth_grid_settings`** — сетка Земли и проекция заявок на неё; + PostGIS-триггеры заполняют `contour_geom` и пересчитывают `earth_cells.importance` + (= Σ `request_cell.importance × coverage_percent/100`). Это **карта важности**, не прогресс съёмки (RS6). + +Геометрия хранится как WKT (`TEXT`); вычисления — JTS (`jts-core`), без hibernate-spatial. + +## Учёт отснятой территории + +Ядро — [RouteMatchingService](../services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/service/RouteMatchingService.kt), +геометрия — [GeometryService](../services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/service/GeometryService.kt). + +На каждый паспорт маршрута: +1. Кандидаты выбираются по **окну времени + статусу** (`ACCEPTED`/`ACTIVE`, не удалённые) — + [findRequestIdsForRoute](../services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/repository/RequestRepository.kt). +2. Идемпотентность: `existsByRequestIdAndRouteId` + `request_route_matches UNIQUE` + обработка гонки вставки. +3. На каждую заявку — своя транзакция с `findByIdForUpdate` (row-lock): маршрут пересекается с + **`remaining_geometry`** (не с исходной), остаток уменьшается `difference`, пересчитывается `remaining_area`. +4. `coveredPercent = (geometry_area − remaining_area) / geometry_area × 100`. + +Поскольку учёт идёт по остатку, **перекрывающиеся маршруты не задваиваются** — второй маршрут +засчитывает только новую площадь. Это корректная кумулятивная модель покрытия. + +Ограничения учёта вынесены в находки **RS2** (игнор совместимости аппарата), **RS5** (планарная +площадь в градусах), **RS6** (earth-grid — это важность, а не покрытие). + +## Жизненный цикл и закрытие заданий + +Вся логика — пересчёт статуса «с нуля» в +[RequestLifecycleService.refreshStatus](../services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/service/RequestLifecycleService.kt), +по приоритету сверху вниз: + +| Условие | Статус | +|---|---| +| `coveredPercent ≥ coverage_required_percent` | **COMPLETED** (ставится `completed_at`) — приоритет над сроком | +| иначе `now < begin_date_time` | ACCEPTED | +| иначе `now > end_date_time` | **EXPIRED** | +| иначе | ACTIVE | + +При переходе в терминальный статус (`COMPLETED` или `EXPIRED`) — транзакционный outbox-эвент +`REQUEST_TERMINAL_STATUS` (идемпотентно по `(event_type, aggregate_id)`, +[RequestStatusOutboxService](../services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/service/RequestStatusOutboxService.kt)), +который затем публикуется +[OutboxPublisherService](../services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/service/OutboxPublisherService.kt) +в `pcp.request.status.v1` (geoportal). См. #35. + +Ключевая особенность (бывший источник RS1): `refreshStatus` исторически вызывался **только из матчинга +маршрута**. После #29 его дополнительно дёргают планировщик истечения (`RequestExpiryService`) и +`createRequest`, поэтому закрытие «по времени» теперь срабатывает (см. RS1). + +## Горизонтальное масштабирование (несколько инстансов) + +Сервис **спроектирован под несколько одинаковых инстансов** — это сильная сторона. + +Безопасно при N инстансов: +- **Матчинг** — `findByIdForUpdate` (`SELECT … FOR UPDATE`, пессимистичная блокировка строки заявки) + в отдельной транзакции на заявку (`PROPAGATION_REQUIRED` из не-транзакционного listener → + одна независимая транзакция на заявку), плюс `@Version` и дедуп `(request_id, route_id)`. + Инстансы делят партиции топика (consumer-group `pcp-request-service-route`). +- **Outbox** — `SELECT … FOR UPDATE SKIP LOCKED` + `UNIQUE(event_type, aggregate_id)` → + параллельные инстансы не публикуют эвент дважды + ([OutboxEventRepository](../services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/repository/OutboxEventRepository.kt)). +- **Пересборка сетки** — `pg_try_advisory_xact_lock` + ([GridRebuildLockService](../services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/service/GridRebuildLockService.kt)) → + ребилдит только один инстанс. +- **Создание заявки** — `existsById` + PK + обработка `23505` → дубликат не пройдёт при гонке. + +Семантика доставки — **at-least-once**; повторная обработка маршрута безопасна за счёт дедупа. +После #31 «ядовитое» сообщение уходит в DLQ (`DefaultErrorHandler` + bounded retry), а не блокирует +партицию. Outbox FAILED после #35 ретраится с backoff и пределом попыток. Параллелизм потребления +маршрутов ограничен числом партиций топика. + +--- + +## Находки + +### RS1 (высокий). Нет планировщика истечения — закрытие «по времени» не срабатывает — ✅ устранено (#29) +Файлы: [RequestLifecycleService.kt](../services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/service/RequestLifecycleService.kt), +[RouteMatchingService.kt](../services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/service/RouteMatchingService.kt) + +`refreshStatus` — единственный путь смены статуса — вызывается **только при сопоставлении маршрута**. +Единственный `@Scheduled` в сервисе — публикация outbox. Поэтому если по заявке после `end_date_time` +**не пришло ни одного маршрута**, она навсегда останется `ACTIVE`/`ACCEPTED` и **никогда не станет +`EXPIRED`**. В `EXPIRED` заявка попадает лишь побочно — если какой-то маршрут случайно пересечётся +с ней уже после истечения срока. Дополнительно: `createRequest` всегда ставит `ACCEPTED`, даже если +окно уже в прошлом (`refreshStatus` при создании не вызывается). + +Рекомендация: периодический job (`@Scheduled` + advisory-lock или `SKIP LOCKED`-выборка), который +переводит просроченные заявки в `EXPIRED` без участия маршрутов; вызывать `refreshStatus` и при создании. + +**Готово (#29):** добавлен [RequestExpiryService](../services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/service/RequestExpiryService.kt) — +`@Scheduled`-воркер переводит просроченные `ACCEPTED`/`ACTIVE` в `EXPIRED`, забирая заявки батчами через +`findExpirableForUpdateSkipLocked` (`FOR UPDATE SKIP LOCKED` + верхний лимит батчей — безопасно при нескольких +инстансах). `refreshStatus` теперь вызывается и в `createRequest`, поэтому статус приёма отражает окно +(`ACCEPTED`/`ACTIVE`/`EXPIRED`). Конфиг — `pcp.request.expiry.*`. Тесты: `RequestExpiryServiceTest`, +`RequestRepositoryJpaTest`. + +### RS2 (высокий). Матчинг игнорирует совместимость аппарата +Файлы: [RouteMatchingService.kt](../services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/service/RouteMatchingService.kt), +[RequestValidationService.kt](../services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/service/RequestValidationService.kt) + +Кандидаты подбираются только по окну времени и статусу, а покрытие засчитывается по любому +геометрическому пересечению. `survey_type` (оптика/РСА), разрешение, угол Солнца, облачность и +**крен** в фильтрации **не участвуют**: крен парсится и валидируется (0–90°), но как фильтр не +используется; параметры съёмки лежат в БД мёртвым грузом. Следствие: РСА-маршрут «закроет» +оптическую заявку, маршрут с негодным разрешением/освещённостью засчитается как покрытие. +Перекликается с UI-задачей #4. + +Рекомендация: добавить проверку совместимости маршрут↔заявка перед зачётом покрытия (как минимум +survey_type и крен; в идеале — разрешение и угол Солнца). Требует, чтобы паспорт маршрута нёс тип +сенсора — согласовать в рамках #28. + +**Решение (2026-06-06): отложено до #28.** `RoutePassportDto` сейчас не несёт явный `surveyType` +(есть `polarisation`, `resolutionRange`/`Azimuth`, `rollAngle`, `visirAngle`). Фиксируем контракт +паспорта в рамках аудита связи сервисов #28 и тогда же добавляем фильтр совместимости. До этого +matching не трогаем. + +### RS3 (средний). Нет DLQ / обработчика ошибок Kafka — poison-сообщение блокирует партицию — ✅ устранено (#31) +Файл: [RequestKafkaListener.kt](../services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/kafka/RequestKafkaListener.kt) + +Listener ловит исключение, логирует и **перебрасывает** его. DLQ и кастомный `CommonErrorHandler` +не настроены — работают дефолты Spring Kafka. Невалидное («ядовитое») сообщение маршрута будет +ретраиться бесконечно и **заблокирует партицию**. У парного `pcp-route-processing-service` DLQ есть, +здесь — нет. + +Рекомендация: `DefaultErrorHandler` с ограниченными ретраями и `DeadLetterPublishingRecoverer` +(топик вида `pcp.route.georeference-dlq.v1`); невалидные по контракту сообщения отправлять в DLQ, +а не зацикливать. + +**Готово (#31):** [KafkaConsumerConfig](../services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/config/KafkaConsumerConfig.kt) — +`DefaultErrorHandler` + `DeadLetterPublishingRecoverer` (DLQ `pcp.route.georeference-dlq.v1`), +экспоненциальный bounded retry; контрактные ошибки (`JsonProcessingException`/`ConstraintViolationException`) +не ретраятся, а сразу уходят в DLQ с диагностическими заголовками (`pcp-exception-class`, +`pcp-original-topic`/`-partition`/`-offset`). Настройки — `app.kafka.consumer.route.*`. По образцу +`pcp-route-processing-service`. Тест — `KafkaConsumerConfigTest`. + +### RS4 (средний). Мёртвый конфиг: cleanup и входной топик заявок — ✅ устранено (#32) +Файл: [config-repo/pcp-request-service.yaml](../config-repo/pcp-request-service.yaml) + +`app.request.cleanup-cron` / `cleanup-retention-days` были заданы, но **кода под них нет** — архивации +или удаления старых заявок не происходило. Аналогично `app.kafka.topics.request: pcp.request.in.v1` +был сконфигурирован, но **без слушателя** (заявки заводятся только по HTTP). Конфиг вводил в заблуждение. + +Решение (#32): ключи удалены из `config-repo` (привязок `@Value` не было). Если cleanup/приём заявок +из Kafka понадобятся — заводить конфиг вместе с реализацией. + +### RS5 (низкий). Планарная площадь в градусах вместо геодезической +Файл: [GeometryService.kt](../services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/service/GeometryService.kt) + +Площади считаются JTS `geometry.area` в координатах EPSG:4326 (градусы). Для компактных заявок +процент покрытия как отношение почти корректен, но для крупных и высокоширотных областей даёт +систематическое искажение (градус долготы у полюса короче). + +Рекомендация: при росте требований к точности — считать площадь геодезически (например, PostGIS +`ST_Area(geography)` или проекция в равновеликую СК), а не планарно в градусах. + +**Решение (2026-06-06): отложено.** Покрытие считается отношением площадей, поэтому для обычных AOI +планарная оценка близка к геодезической. Переключение `area()` на геодезию ripple-эффектом затронет +учёт `coverageDeltaPercent` и точечные ассерты площадей в `RouteMatchingServiceTest`; делаем при +реальном росте требований к точности (крупные/высокоширотные заявки). + +### RS6 (низкий). `coverage_required_percent` захардкожен в 100%; earth-grid ≠ покрытие — ✅ устранено (#34) +Файлы: [RequestService.kt](../services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/service/RequestService.kt), +[RequestGridProjectionService.kt](../services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/service/RequestGridProjectionService.kt) + +`createRequest` всегда ставит `DEFAULT_COVERAGE_REQUIRED_PERCENT = 100.0`, игнорируя значение из API, +хотя колонка и поле его поддерживают. Отдельно стоит зафиксировать: `request_cells.coverage_percent` +и `/priority-map` — это **доля футпринта заявки в ячейке × важность** (карта приоритетов), считается +один раз при создании и **маршрутами не обновляется**. Реальный прогресс съёмки живёт только в +`remaining_geometry` / `remaining_area`. Это легко перепутать по названию поля. + +Рекомендация: пробрасывать требуемый процент из API; в документации/именовании развести «важность +ячейки» и «прогресс покрытия». + +**Готово (#34):** в `CreateRequestDto` добавлено опциональное поле `coverageRequiredPercent` +(0 < x ≤ 100, по умолчанию 100); оно пробрасывается в сущность и служит порогом `COMPLETED`, +константа `DEFAULT_COVERAGE_REQUIRED_PERCENT` убрана. В KDoc разведены «прогресс покрытия» +(`coverageRequiredPercent`/`remaining_area`) и «важность заявки» (`importance` → карта приоритетов +ячеек). Тесты — `RequestServiceTest`, `RequestApiDtoContractTest`. + +### RS7 (средний). Уведомление geoportal о финале заявки; outbox FAILED без backoff — ✅ устранено (#35) +Файлы: [OutboxPublisherService.kt](../services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/service/OutboxPublisherService.kt), +[OutboxEventRepository.kt](../services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/repository/OutboxEventRepository.kt), +[RequestStatusOutboxService.kt](../services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/service/RequestStatusOutboxService.kt) + +Потребитель завершения заявок — внешний сервис **geoportal** (в этом монорепо его нет, поэтому в +[PCP_SERVICE_INTERACTIONS.md](architect/PCP_SERVICE_INTERACTIONS.md) консьюмер не виден). Ему нужно +сообщать про **оба** терминальных исхода: «выполнено» (`COMPLETED`) и «срок вышел» (`EXPIRED`). Сейчас +эмитится только `COMPLETED` (в `pcp.request.completed.v1`), а `EXPIRED`-эвента нет вовсе. Плюс +публикуемая выборка берёт `NEW` + `FAILED`, поэтому FAILED **ретраится**, но без backoff и без cap: +при устойчивом сбое Kafka каждые 5 с все инстансы перевыбирают FAILED (нагрузка, но не потеря). + +Решение: свести уведомления в единый топик **`pcp.request.status.v1`** (по аналогии с +`pcp.slots.status.v1`) с терминальным статусом в payload, заменив orphan `pcp.request.completed.v1`; +эмитить `EXPIRED`-эвент через тот же transactional outbox (зависит от планировщика истечения **RS1**, +который и создаёт переход в `EXPIRED`). Outbox дополнить `next_attempt_at`/экспоненциальным backoff и +пределом попыток — по образцу `RoutePassportOutboxWorker` из `pcp-route-processing-service`. + +**Готово (#35):** единый топик `pcp.request.status.v1` с терминальным статусом в payload +(`COMPLETED`/`EXPIRED`, плюс `occurredAt`/`coveragePercent`) — заменил orphan `pcp.request.completed.v1`. +Единое событие `REQUEST_TERMINAL_STATUS` пишется в той же транзакции из двух мест: +[RouteMatchingService](../services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/service/RouteMatchingService.kt) +(по достижению покрытия) и [RequestExpiryService](../services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/service/RequestExpiryService.kt) +(по истечению срока). Outbox получил `attempts`/`next_attempt_at`: при сбое — удвоение backoff от +`initial-backoff-ms` до `max-backoff-ms`, после `max-attempts` событие становится терминальным `FAILED` +и из выборки выпадает. Конфиг — `pcp.outbox.*`. Мелкий зазор: заявка, **созданная** уже просроченной, +терминальное событие не порождает (вырожденный кейс; обычные пути ACTIVE→COMPLETED/EXPIRED покрыты). + +## Что в порядке (намеренно не отмечаем как проблемы) + +- Конкурентность и мультиинстанс: row-lock на заявку, `SKIP LOCKED` в outbox, advisory-lock на сетку, + `@Version`, дедуп по уникальным ключам — сделано грамотно. +- Transactional outbox: терминальный эвент `REQUEST_TERMINAL_STATUS` пишется в той же транзакции, что и завершение/истечение заявки. +- Идемпотентность сопоставления: повтор маршрута безопасен (`request_route_matches UNIQUE` + проверки). +- Кумулятивный учёт покрытия по остатку (без задвоения перекрытий). +- Тесты: `RouteMatchingServiceTest`, `OutboxPublisherServiceTest`, + `RequestStatusOutboxServiceTest`, `RequestExpiryServiceTest`, `RequestRepositoryJpaTest`, `MigrationSchemaTest`. + +## Статус работ + +Сделано: +- ✅ **RS1** — устранено (#29): планировщик истечения `RequestExpiryService` + `refreshStatus` при создании. +- ✅ **RS3** — устранено (#31): `DefaultErrorHandler` + DLQ `pcp.route.georeference-dlq.v1`. +- ✅ **RS4** — устранено (#32): мёртвый конфиг удалён из `config-repo`. +- ✅ **RS6** — устранено (#34): проброс `coverageRequiredPercent` из API + разведение терминов. +- ✅ **RS7** — устранено (#35): единый топик `pcp.request.status.v1` (COMPLETED/EXPIRED) + backoff/cap outbox. + +Отложено (осознанно): +- ⏸ **RS2** (#30) — фильтр совместимости аппарата. Отложено **до #28**: паспорт маршрута не несёт явный `surveyType`, контракт фиксируется в аудите связи сервисов. +- ⏸ **RS5** (#33) — геодезическая площадь. Отложено до роста требований к точности (переключение `area()` затрагивает matching и тесты; покрытие — отношение площадей). diff --git a/docs/бэклог-pcp-tgu-ops-ui.md b/docs/бэклог-pcp-tgu-ops-ui.md index b406ffa..8ede38b 100644 --- a/docs/бэклог-pcp-tgu-ops-ui.md +++ b/docs/бэклог-pcp-tgu-ops-ui.md @@ -10,7 +10,10 @@ У каждой задачи есть стабильный номер `#N` — он не меняется при правках списка, на него удобно ссылаться. Новые задачи получают следующий свободный номер. -_Обновлено: 2026-06-05_ +> Задачи #12–#25 заведены по результатам аудита. Подробности, обоснование и рекомендации +> по каждой — в [аудит-проекта-2026-06-06.md](аудит-проекта-2026-06-06.md) (ссылки на коды находок B/F/R/C). + +_Обновлено: 2026-06-06_ --- @@ -31,12 +34,49 @@ _Обновлено: 2026-06-05_ ### Бэкенд / инфраструктура - [ ] **#7** — Перевести сервис станций на ordinis (по примеру получения аппаратов в сервисе tgu). +### Проверка сервисов +- [ ] **#27** — Проверить сервис получения паспорта маршрута по отдельности. +- [ ] **#28** — Проверить связь сервиса заданий и сервиса получения паспорта маршрута между собой. + ### Баги - [ ] **#8** — Ошибка при попытке принять план на вкладке «Планы». +### Аудит — надёжность и бэкенд +_Источник и рекомендации: [аудит-проекта-2026-06-06.md](аудит-проекта-2026-06-06.md)._ +- [ ] **#12** — (B1, высокий) Camunda job-воркеры в `FlowService` молча глотают исключения (`catch { logger.warn(ex.message) }`) при `autoComplete=false`: задача не завершается и не падает, BPMN-процесс зависает без инцидента. Завершать через `newFailCommand`/проброс наверх, логировать `error(msg, ex)`. +- [ ] **#13** — (B2, средний) Снять симуляционную заглушку `Thread.sleep(5000)` («имитация подтверждения плана») в `FlowService.sendPlan` при появлении реального канала подтверждений. +- [ ] **#14** — (B3, средний) Привести в порядок несогласованные пакеты-копипасты (ballistics→`pcp_request_service`, complex-mission→`pcp_satellites_service`, ui-service и slots-service делят `slots_service`, request→`org.nstart.dep265`, tgu→без `.pcp.`). Отдельной задачей из-за широкого охвата. +- [ ] **#15** — (B4, низкий) Опечатка «srpring» в `pcp-srpring-boot-admin-server` растиражирована по каталогу, пакету, CI, docker-compose, gradle. Чинить вместе с #14. +- [ ] **#16** — (B5, низкий) ~9 мест `logger.warn/error(ex.message)` теряют stack trace — заменить на `logger.error("...", ex)`. + +### Аудит — фронтенд +_Источник и рекомендации: [аудит-проекта-2026-06-06.md](аудит-проекта-2026-06-06.md)._ +- [ ] **#17** — (F1, средний) Вынести общий `src/api/http.ts` (`fetchJson` + `extractErrorText` + маппинг ошибок): `extractErrorText` продублирован в 5 клиентах, fetch-обвязка повторяется ~22 раза. Заодно устранить расхождение приоритета полей ошибки (F5). +- [ ] **#18** — (F2, низкий) Перевести нетипизированный островок `features/constellation/*` (`.jsx`/`.js`, `allowJs`) на TS или добавить декларации типов на границе. +- [ ] **#19** — (F3, низкий) Удалить мёртвый barrel `features/constellation/index.js` (нигде не импортируется). +- [ ] **#20** — (F4, низкий) Декомпозировать God-компонент `TguEditorTab.tsx` (1727 строк): вынести хуки (`useDraftHistory`, `useRouteBuilder`), разнести панели. + +### Аудит — гигиена репозитория и конфиги +_Источник и рекомендации: [аудит-проекта-2026-06-06.md](аудит-проекта-2026-06-06.md)._ +- [ ] **#21** — (R1, средний) Убрать из git крупные дампы в корне (`pcp_requests_converted_create_requests.json` ~228 КБ, `response_заданияКубернетс.json` ~157 КБ, `response_заданияЕмельянов.json` ~66 КБ) — перенести в `src/test/resources`/`docs` или удалить с добавлением в `.gitignore`. +- [ ] **#22** — (R2, низкий) Удалить посторонние артефакты в корне: `node_modules/` + пустой `package-lock.json`, `obj_terra.zip`; при желании добавить `/package-lock.json` в `.gitignore`. +- [ ] **#23** — (R3, низкий) Прогнать `./gradlew clean` / удалить 26 устаревших каталогов `bin/` (задваивают исходники, шумят при grep). +- [ ] **#24** — (C1, средний) Унифицировать креды БД в config-repo: привести все сервисы к переопределяемому `password: ${DB_PASSWORD:...}` (как в tgu-service), убрать зашитые `password: password`. +- [ ] **#25** — (C2, низкий) Сузить CORS для прода в `pcp-ui-service` (`allowedOriginPatterns("*")` + `allowCredentials`) до конкретных origin; держать в курсе отсутствие CORS у напрямую вызываемых сервисов (работает только через прокси/nginx). + +### Аудит — сервис заданий (pcp-request-service) +_Источник и рекомендации: [аудит-сервис-заданий.md](аудит-сервис-заданий.md) (коды находок RS1–RS7)._ +- [ ] **#30** — (RS2, высокий) Матчинг засчитывает покрытие по любому геометрическому пересечению, игнорируя совместимость аппарата (`survey_type`, крен, разрешение, угол Солнца). Добавить фильтр совместимости маршрут↔заявка перед зачётом покрытия; тип сенсора в паспорте маршрута согласовать в рамках #28. _Отложено до #28 (решение, 2026-06-06): `RoutePassportDto` не несёт явный `surveyType` — контракт фиксируется в рамках аудита связи сервисов #28, тогда же делаем фильтр._ +- [ ] **#33** — (RS5, низкий) Площади считаются планарно в градусах (JTS `area` на EPSG:4326) — искажение для крупных/высокоширотных заявок. Перейти на геодезию (`ST_Area(geography)` или равновеликая проекция), когда вырастут требования к точности. _Отложено по решению аудита: покрытие считается отношением площадей (планар ≈ геодезия для обычных AOI), а переключение `area()` затрагивает matching и точечные ассерты тестов — вводим при росте требований к точности._ --- ## Выполнено +- [x] **#34** — (RS6) Проброс требуемого процента покрытия из API + разведение терминов. _Готово: в `CreateRequestDto` добавлено опциональное поле `coverageRequiredPercent` (0 < x ≤ 100, по умолчанию 100), оно пробрасывается в сущность и используется как порог COMPLETED; захардкоженная константа убрана. В KDoc/коде разведены «прогресс покрытия» (`coverageRequiredPercent`) и «важность заявки» (`importance`, карта приоритетов ячеек). Тесты: `RequestServiceTest`, `RequestApiDtoContractTest`._ +- [x] **#31** — (RS3) DLQ и error-handler для Kafka-листенера маршрутов. _Готово: `KafkaConsumerConfig` — `DefaultErrorHandler` + `DeadLetterPublishingRecoverer` в топик `pcp.route.georeference-dlq.v1`, экспоненциальный bounded retry, контрактные ошибки (`JsonProcessingException`/`ConstraintViolationException`) → DLQ без ретраев, диагностические заголовки. Настройки `app.kafka.consumer.route.*`, активируется при заданном `spring.kafka.bootstrap-servers`. Тест `KafkaConsumerConfigTest`._ +- [x] **#35** — (RS7) Уведомление **geoportal** о финале заявки. _Готово: единый топик `pcp.request.status.v1` (заменил orphan `pcp.request.completed.v1`) с терминальным статусом в payload (`COMPLETED`/`EXPIRED`). Эмиссия из двух мест через тот же transactional outbox: `RouteMatchingService` (по покрытию) и `RequestExpiryService` (по сроку, #29). Сервис `RequestCompletedOutboxService`→`RequestStatusOutboxService`, eventType `REQUEST_TERMINAL_STATUS`. Outbox получил backoff: колонки `attempts`/`next_attempt_at`, удвоение `initial→max`, предел `max-attempts` (терминальный FAILED). Конфиг `pcp.outbox.*`. Тесты: `OutboxPublisherServiceTest`, `OutboxEventRepositoryJpaTest`, `RequestStatusOutboxServiceTest`, `RequestExpiryServiceTest`, `RouteMatchingServiceTest`. Док: интеракции обновлены. Известный мелкий зазор: заявка, созданная уже просроченной, события не шлёт (вырожденный кейс)._ +- [x] **#29** — (RS1) Планировщик истечения заявок. _Готово: `RequestExpiryService` (`@Scheduled`, `FOR UPDATE SKIP LOCKED`, батчи с верхним лимитом) переводит просроченные `ACCEPTED`/`ACTIVE` в `EXPIRED`; несколько инстансов безопасны. `refreshStatus` теперь вызывается и при создании заявки — статус приёма отражает окно (`ACCEPTED`/`ACTIVE`/`EXPIRED`), а не всегда `ACCEPTED`. Конфиг `pcp.request.expiry.*` в config-repo. Тесты: `RequestExpiryServiceTest` (юнит) + `RequestRepositoryJpaTest` (H2). Разблокирует EXPIRED-эмиссию в #35._ +- [x] **#32** — (RS4) Удалить мёртвый конфиг сервиса заданий. _Готово: из `config-repo/pcp-request-service.yaml` убраны `app.request.cleanup-cron`/`cleanup-retention-days` и неиспользуемый топик `app.kafka.topics.request` (`pcp.request.in.v1`); кода под ними не было, `@Value`-привязок нет._ +- [x] **#26** — Проверить сервис заданий (`pcp-request-service`) по отдельности. _Готово: аудит в [аудит-сервис-заданий.md](аудит-сервис-заданий.md) — карта взаимодействий, учёт покрытия, жизненный цикл, мультиинстанс; находки RS1–RS7 (RS1/RS2 — высокий приоритет)._ - [x] **#9** — Расширить вправо левую панель информации на вкладке «Редактор» (цифры у границы смотрелись плохо). _Готово: commit 69dfd80, рейл 15.5rem → 17rem._ - [x] **#10** — Проставлять статусы слотов только после того, как план заложен на КА. _Готово: commit 69dfd80, слоты занимаются по `LAID_IN`, а не по `ACCEPT`._ - [x] **#11** — Запретить нажимать «Подтвердить закладку», пока начало закладки в будущем. _Готово: commit 69dfd80._ diff --git a/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/config/KafkaConsumerConfig.kt b/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/config/KafkaConsumerConfig.kt new file mode 100644 index 0000000..8a998c8 --- /dev/null +++ b/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/config/KafkaConsumerConfig.kt @@ -0,0 +1,131 @@ +package org.nstart.dep265.requestservice.config + +import com.fasterxml.jackson.core.JsonProcessingException +import jakarta.validation.ConstraintViolationException +import org.apache.kafka.clients.consumer.ConsumerConfig +import org.apache.kafka.clients.consumer.ConsumerRecord +import org.apache.kafka.common.TopicPartition +import org.apache.kafka.common.header.Headers +import org.apache.kafka.common.header.internals.RecordHeader +import org.apache.kafka.common.header.internals.RecordHeaders +import org.apache.kafka.common.serialization.StringDeserializer +import org.springframework.beans.factory.annotation.Value +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty +import org.springframework.context.annotation.Bean +import org.springframework.context.annotation.Configuration +import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory +import org.springframework.kafka.core.ConsumerFactory +import org.springframework.kafka.core.DefaultKafkaConsumerFactory +import org.springframework.kafka.core.KafkaTemplate +import org.springframework.kafka.listener.DeadLetterPublishingRecoverer +import org.springframework.kafka.listener.DefaultErrorHandler +import org.springframework.util.backoff.ExponentialBackOff +import java.nio.charset.StandardCharsets + +/** + * Обработка ошибок Kafka-консьюмера маршрутов: bounded retry + DLQ. + * + * Без этого «ядовитое» сообщение бесконечно ретраится и блокирует партицию. Невалидные по контракту + * (ошибка десериализации/валидации) уходят в DLQ сразу, прочие — после ограниченного числа попыток. + * + * Активируется только при заданном `spring.kafka.bootstrap-servers`, чтобы не требовать Kafka в тестах + * без брокера. + */ +@Configuration +@ConditionalOnProperty(name = ["spring.kafka.bootstrap-servers"], matchIfMissing = false) +class KafkaConsumerConfig( + @param:Value("\${spring.kafka.bootstrap-servers}") + private val bootstrapServers: String, + @param:Value("\${spring.kafka.consumer.group-id}") + private val groupId: String, + @param:Value("\${spring.kafka.consumer.auto-offset-reset:latest}") + private val autoOffsetReset: String, +) { + + @Bean + fun routeConsumerFactory(): ConsumerFactory = + DefaultKafkaConsumerFactory( + mutableMapOf( + ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG to bootstrapServers, + ConsumerConfig.GROUP_ID_CONFIG to groupId, + ConsumerConfig.AUTO_OFFSET_RESET_CONFIG to autoOffsetReset, + ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG to false, + ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG to StringDeserializer::class.java, + ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG to StringDeserializer::class.java, + ), + ) + + @Bean + fun kafkaListenerContainerFactory( + routeKafkaErrorHandler: DefaultErrorHandler, + ): ConcurrentKafkaListenerContainerFactory = + ConcurrentKafkaListenerContainerFactory().apply { + setConsumerFactory(routeConsumerFactory()) + setCommonErrorHandler(routeKafkaErrorHandler) + } + + @Bean + fun routeDeadLetterPublishingRecoverer( + kafkaTemplate: KafkaTemplate, + routeConsumerProperties: RouteConsumerProperties, + ): DeadLetterPublishingRecoverer = + DeadLetterPublishingRecoverer(kafkaTemplate) { record, _ -> + TopicPartition(routeConsumerProperties.dlqTopic, record.partition()) + }.apply { + addHeadersFunction { record, exception -> routeDlqHeaders(record, exception) } + } + + @Bean + fun routeKafkaErrorHandler( + routeDeadLetterPublishingRecoverer: DeadLetterPublishingRecoverer, + routeConsumerProperties: RouteConsumerProperties, + ): DefaultErrorHandler = + DefaultErrorHandler( + routeDeadLetterPublishingRecoverer, + routeConsumerBackOff(routeConsumerProperties), + ).apply { + // Контрактные ошибки (битый JSON, провал валидации) бесполезно ретраить — сразу в DLQ. + addNotRetryableExceptions( + JsonProcessingException::class.java, + ConstraintViolationException::class.java, + ) + setCommitRecovered(true) + } + + private fun routeConsumerBackOff(properties: RouteConsumerProperties): ExponentialBackOff = + ExponentialBackOff().apply { + initialInterval = properties.backoffMs + multiplier = RETRY_BACKOFF_MULTIPLIER + maxInterval = properties.maxBackoffMs + maxAttempts = (properties.maxAttempts - 1).coerceAtLeast(0) + } + + private fun routeDlqHeaders( + record: ConsumerRecord<*, *>, + exception: Exception, + ): Headers = + RecordHeaders().apply { + addStringHeader(HEADER_SERVICE_NAME, SERVICE_NAME) + addStringHeader(HEADER_EXCEPTION_CLASS, exception::class.java.name) + addStringHeader(HEADER_EXCEPTION_MESSAGE, exception.message.orEmpty()) + addStringHeader(HEADER_ORIGINAL_TOPIC, record.topic()) + addStringHeader(HEADER_ORIGINAL_PARTITION, record.partition().toString()) + addStringHeader(HEADER_ORIGINAL_OFFSET, record.offset().toString()) + } + + private fun Headers.addStringHeader(name: String, value: String) { + add(RecordHeader(name, value.toByteArray(StandardCharsets.UTF_8))) + } + + private companion object { + const val SERVICE_NAME = "pcp-request-service" + const val RETRY_BACKOFF_MULTIPLIER = 2.0 + + const val HEADER_SERVICE_NAME = "pcp-service-name" + const val HEADER_EXCEPTION_CLASS = "pcp-exception-class" + const val HEADER_EXCEPTION_MESSAGE = "pcp-exception-message" + const val HEADER_ORIGINAL_TOPIC = "pcp-original-topic" + const val HEADER_ORIGINAL_PARTITION = "pcp-original-partition" + const val HEADER_ORIGINAL_OFFSET = "pcp-original-offset" + } +} diff --git a/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/config/OutboxProperties.kt b/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/config/OutboxProperties.kt index 2bcd7d9..96721b0 100644 --- a/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/config/OutboxProperties.kt +++ b/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/config/OutboxProperties.kt @@ -4,7 +4,14 @@ import org.springframework.boot.context.properties.ConfigurationProperties @ConfigurationProperties(prefix = "pcp.outbox") data class OutboxProperties( - val requestCompletedTopic: String = "pcp.request.completed.v1", + /** Топик терминальных статусов заявки (COMPLETED/EXPIRED) для geoportal. */ + val requestStatusTopic: String = "pcp.request.status.v1", val publishBatchSize: Int = 50, val publishFixedDelayMs: Long = 5_000, + /** Сколько раз пытаемся опубликовать событие до перевода в терминальный FAILED. */ + val maxAttempts: Int = 10, + /** Backoff после первой неудачной публикации. */ + val initialBackoffMs: Long = 1_000, + /** Верхняя граница backoff между попытками. */ + val maxBackoffMs: Long = 60_000, ) diff --git a/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/config/OutboxTransactionConfig.kt b/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/config/OutboxTransactionConfig.kt index f4d6de2..db18f26 100644 --- a/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/config/OutboxTransactionConfig.kt +++ b/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/config/OutboxTransactionConfig.kt @@ -22,4 +22,15 @@ class OutboxTransactionConfig { propagationBehavior = TransactionDefinition.PROPAGATION_REQUIRES_NEW } } + + /** + * Каждый батч планировщика истечения коммитится отдельно, поэтому между батчами + * row-level блокировки (FOR UPDATE SKIP LOCKED) освобождаются. + */ + @Bean("requestExpiryTransactionOperations") + fun requestExpiryTransactionOperations(transactionManager: PlatformTransactionManager): TransactionOperations { + return TransactionTemplate(transactionManager).apply { + propagationBehavior = TransactionDefinition.PROPAGATION_REQUIRES_NEW + } + } } diff --git a/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/config/RequestExpiryProperties.kt b/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/config/RequestExpiryProperties.kt new file mode 100644 index 0000000..967e151 --- /dev/null +++ b/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/config/RequestExpiryProperties.kt @@ -0,0 +1,19 @@ +package org.nstart.dep265.requestservice.config + +import org.springframework.boot.context.properties.ConfigurationProperties + +/** + * Настройки планировщика истечения заявок: периодически переводит просроченные заявки + * (status ACCEPTED/ACTIVE, окно которых уже закончилось) в EXPIRED. + */ +@ConfigurationProperties(prefix = "pcp.request.expiry") +data class RequestExpiryProperties( + /** Включён ли планировщик. */ + val enabled: Boolean = true, + /** Период запуска (мс), читается также в @Scheduled. */ + val fixedDelayMs: Long = 60_000, + /** Сколько заявок захватывать за одну транзакцию (FOR UPDATE SKIP LOCKED). */ + val batchSize: Int = 200, + /** Верхняя граница числа батчей за один прогон, чтобы не удерживать поток планировщика. */ + val maxBatchesPerRun: Int = 50, +) diff --git a/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/config/RouteConsumerProperties.kt b/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/config/RouteConsumerProperties.kt new file mode 100644 index 0000000..dd92386 --- /dev/null +++ b/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/config/RouteConsumerProperties.kt @@ -0,0 +1,28 @@ +package org.nstart.dep265.requestservice.config + +import jakarta.validation.constraints.Min +import org.springframework.boot.context.properties.ConfigurationProperties +import org.springframework.validation.annotation.Validated + +/** + * Настройки обработки ошибок Kafka-консьюмера маршрутов: DLQ-топик и bounded retry. + * «Ядовитое» (невалидное по контракту) сообщение уходит в DLQ, а не блокирует партицию. + */ +@Validated +@ConfigurationProperties(prefix = "app.kafka.consumer.route") +data class RouteConsumerProperties( + /** Топик dead-letter для маршрутов, не прошедших обработку. */ + val dlqTopic: String = "pcp.route.georeference-dlq.v1", + + /** Максимум попыток обработки, включая первичную доставку. */ + @field:Min(1) + val maxAttempts: Long = 5, + + /** Начальная задержка retry, мс. */ + @field:Min(0) + val backoffMs: Long = 1_000, + + /** Верхняя граница задержки retry, мс. */ + @field:Min(0) + val maxBackoffMs: Long = 30_000, +) diff --git a/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/dto/RequestApiDtos.kt b/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/dto/RequestApiDtos.kt index a67df9a..efdddf1 100644 --- a/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/dto/RequestApiDtos.kt +++ b/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/dto/RequestApiDtos.kt @@ -60,6 +60,15 @@ data class CreateRequestDto( /** Признак высокоприоритетной передачи. */ val highPriorityTransmit: Boolean = false, + /** + * Требуемый процент покрытия заявки (0 < x ≤ 100), при достижении которого заявка + * считается выполненной (COMPLETED). Это **прогресс съёмки**, не путать с [importance] + * (важность заявки для карты приоритетов ячеек). По умолчанию — 100%. + */ + @field:DecimalMin(value = "0", inclusive = false) + @field:DecimalMax("100") + val coverageRequiredPercent: Double = 100.0, + /** Параметры оптической съемки. */ @field:Valid val optics: OpticsParamsDto? = null, diff --git a/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/entity/OutboxEventEntity.kt b/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/entity/OutboxEventEntity.kt index 77c8805..03fdee2 100644 --- a/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/entity/OutboxEventEntity.kt +++ b/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/entity/OutboxEventEntity.kt @@ -48,6 +48,14 @@ class OutboxEventEntity( @Column(name = "created_at", nullable = false) var createdAt: LocalDateTime, + /** Число выполненных попыток публикации. */ + @Column(name = "attempts", nullable = false) + var attempts: Int = 0, + + /** Не раньше этого момента событие снова попадёт в выборку публикации (backoff). */ + @Column(name = "next_attempt_at", nullable = false) + var nextAttemptAt: LocalDateTime = createdAt, + @Column(name = "published_at") var publishedAt: LocalDateTime? = null, diff --git a/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/repository/OutboxEventRepository.kt b/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/repository/OutboxEventRepository.kt index 02459c1..890ac1c 100644 --- a/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/repository/OutboxEventRepository.kt +++ b/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/repository/OutboxEventRepository.kt @@ -16,8 +16,9 @@ interface OutboxEventRepository : JpaRepository { fun findByStatusOrderByCreatedAtAsc(status: OutboxEventStatus, pageable: Pageable): List /** - * Locks publishable REQUEST_COMPLETED events so parallel service instances skip rows already - * being published by another transaction. + * Locks publishable events so parallel service instances skip rows already being published by + * another transaction. Учитывает backoff (`next_attempt_at`) и предел попыток (`attempts`): + * терминальные FAILED (`attempts >= maxAttempts`) и ещё не «дозревшие» строки исключаются. */ @Query( value = """ @@ -25,6 +26,8 @@ interface OutboxEventRepository : JpaRepository { FROM outbox_events WHERE event_type = :eventType AND status IN ('NEW', 'FAILED') + AND attempts < :maxAttempts + AND next_attempt_at <= :now ORDER BY CASE WHEN status = 'NEW' THEN 0 ELSE 1 END, created_at ASC @@ -33,8 +36,10 @@ interface OutboxEventRepository : JpaRepository { """, nativeQuery = true, ) - fun findRequestCompletedPublishableForUpdateSkipLocked( + fun findPublishableForUpdateSkipLocked( @Param("eventType") eventType: String, + @Param("now") now: LocalDateTime, + @Param("maxAttempts") maxAttempts: Int, @Param("batchSize") batchSize: Int, ): List @@ -52,6 +57,8 @@ interface OutboxEventRepository : JpaRepository { aggregate_id, payload, status, + attempts, + next_attempt_at, created_at, published_at, error_message @@ -63,6 +70,8 @@ interface OutboxEventRepository : JpaRepository { :aggregateId, CAST(:payload AS jsonb), :status, + 0, + :createdAt, :createdAt, NULL, NULL diff --git a/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/repository/RequestRepository.kt b/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/repository/RequestRepository.kt index 8b05de9..895660d 100644 --- a/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/repository/RequestRepository.kt +++ b/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/repository/RequestRepository.kt @@ -106,4 +106,27 @@ interface RequestRepository : JpaRepository { ): List fun findByStatusIn(statuses: Collection): List + + /** + * Захватывает порцию просроченных заявок (окно уже закончилось, статус ещё незавершённый) + * под row-level lock. FOR UPDATE SKIP LOCKED позволяет нескольким инстансам разбирать + * непересекающиеся строки, не конкурируя за одни и те же заявки. + */ + @Query( + value = """ + SELECT * + FROM requests + WHERE status IN ('ACCEPTED', 'ACTIVE') + AND end_date_time < :now + AND deleted_at IS NULL + ORDER BY end_date_time ASC + LIMIT :batchSize + FOR UPDATE SKIP LOCKED + """, + nativeQuery = true, + ) + fun findExpirableForUpdateSkipLocked( + @Param("now") now: LocalDateTime, + @Param("batchSize") batchSize: Int, + ): List } diff --git a/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/service/OutboxPublisherService.kt b/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/service/OutboxPublisherService.kt index 31627eb..7077639 100644 --- a/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/service/OutboxPublisherService.kt +++ b/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/service/OutboxPublisherService.kt @@ -11,6 +11,7 @@ import org.springframework.scheduling.annotation.Scheduled import org.springframework.stereotype.Service import org.springframework.transaction.support.TransactionCallback import org.springframework.transaction.support.TransactionOperations +import java.time.Duration import java.time.LocalDateTime import java.util.concurrent.ExecutionException @@ -41,60 +42,91 @@ class OutboxPublisherService( return try { transactionOperations.execute( TransactionCallback { - val events = outboxEventRepository.findRequestCompletedPublishableForUpdateSkipLocked( - eventType = RequestCompletedOutboxService.REQUEST_COMPLETED_EVENT_TYPE, + val now = LocalDateTime.now() + val events = outboxEventRepository.findPublishableForUpdateSkipLocked( + eventType = RequestStatusOutboxService.REQUEST_STATUS_EVENT_TYPE, + now = now, + maxAttempts = outboxProperties.maxAttempts, batchSize = batchSize, ).distinctBy { it.id } events.forEach { event -> - publishAndUpdateStatus(event) + publishAndUpdateStatus(event, now) } events.size }, ) ?: 0 } catch (exception: Exception) { - log.error("Failed to process REQUEST_COMPLETED outbox event batch", exception) + log.error("Failed to process REQUEST_TERMINAL_STATUS outbox event batch", exception) 0 } } - private fun publishAndUpdateStatus(event: OutboxEventEntity) { + private fun publishAndUpdateStatus(event: OutboxEventEntity, now: LocalDateTime) { try { kafkaTemplate .send( - outboxProperties.requestCompletedTopic, + outboxProperties.requestStatusTopic, event.aggregateId.toString(), event.payload.toString(), ) .get() event.status = OutboxEventStatus.PUBLISHED - event.publishedAt = LocalDateTime.now() + event.publishedAt = now event.errorMessage = null outboxEventRepository.save(event) log.info( - "Published REQUEST_COMPLETED outbox event: eventId={}, requestId={}, topic={}", + "Published REQUEST_TERMINAL_STATUS outbox event: eventId={}, requestId={}, topic={}", event.id, event.aggregateId, - outboxProperties.requestCompletedTopic, + outboxProperties.requestStatusTopic, ) } catch (exception: Exception) { val publishException = exception.unwrapExecutionException() + + event.attempts += 1 event.status = OutboxEventStatus.FAILED event.errorMessage = publishException.shortMessage() + // attempts >= maxAttempts → терминальный FAILED (выборка публикации его уже не вернёт); + // иначе планируем повтор с backoff. + if (event.attempts < outboxProperties.maxAttempts) { + event.nextAttemptAt = now.plus(Duration.ofMillis(calculateBackoffMs(event.attempts))) + } outboxEventRepository.save(event) + val exhausted = event.attempts >= outboxProperties.maxAttempts log.warn( - "Failed to publish REQUEST_COMPLETED outbox event: eventId={}, requestId={}, topic={}", + "Failed to publish REQUEST_TERMINAL_STATUS outbox event: eventId={}, requestId={}, " + + "topic={}, attempts={}, maxAttempts={}, exhausted={}, nextAttemptAt={}", event.id, event.aggregateId, - outboxProperties.requestCompletedTopic, + outboxProperties.requestStatusTopic, + event.attempts, + outboxProperties.maxAttempts, + exhausted, + event.nextAttemptAt, publishException, ) } } + /** + * Удваивает backoff от [OutboxProperties.initialBackoffMs] до [OutboxProperties.maxBackoffMs]. + */ + private fun calculateBackoffMs(attempts: Int): Long { + var backoffMs = outboxProperties.initialBackoffMs.coerceAtMost(outboxProperties.maxBackoffMs) + repeat((attempts - 1).coerceAtLeast(0)) { + backoffMs = if (backoffMs >= outboxProperties.maxBackoffMs / 2) { + outboxProperties.maxBackoffMs + } else { + (backoffMs * 2).coerceAtMost(outboxProperties.maxBackoffMs) + } + } + return backoffMs + } + private fun Exception.unwrapExecutionException(): Throwable { return if (this is ExecutionException && cause != null) { cause!! diff --git a/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/service/RequestExpiryService.kt b/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/service/RequestExpiryService.kt new file mode 100644 index 0000000..fd73d08 --- /dev/null +++ b/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/service/RequestExpiryService.kt @@ -0,0 +1,118 @@ +package org.nstart.dep265.requestservice.service + +import org.nstart.dep265.requestservice.config.RequestExpiryProperties +import org.nstart.dep265.requestservice.entity.RequestStatus +import org.nstart.dep265.requestservice.repository.RequestRepository +import org.slf4j.LoggerFactory +import org.springframework.beans.factory.annotation.Qualifier +import org.springframework.scheduling.annotation.Scheduled +import org.springframework.stereotype.Service +import org.springframework.transaction.support.TransactionCallback +import org.springframework.transaction.support.TransactionOperations +import java.time.LocalDateTime + +/** + * Планировщик истечения заявок. + * + * [RequestLifecycleService.refreshStatus] вызывается только при матчинге маршрута, поэтому заявка, + * к которой не пришло ни одного маршрута, никогда не уходила в EXPIRED после окончания окна. + * Этот воркер периодически переводит такие заявки в EXPIRED. + * + * Несколько инстансов безопасны: батч строк захватывается через FOR UPDATE SKIP LOCKED, так что + * параллельные инстансы разбирают непересекающиеся заявки. + */ +@Service +class RequestExpiryService( + private val requestRepository: RequestRepository, + + private val requestLifecycleService: RequestLifecycleService, + + private val requestStatusOutboxService: RequestStatusOutboxService, + + private val requestExpiryProperties: RequestExpiryProperties, + + @param:Qualifier("requestExpiryTransactionOperations") + private val transactionOperations: TransactionOperations, +) { + private val log = LoggerFactory.getLogger(this::class.java) + + @Scheduled(fixedDelayString = "\${pcp.request.expiry.fixed-delay-ms:60000}") + fun expireOverdueScheduledBatch() { + if (!requestExpiryProperties.enabled) { + return + } + expireOverdue() + } + + /** + * Переводит просроченные заявки (ACCEPTED/ACTIVE с истёкшим окном) в EXPIRED. + * + * @return число заявок, перешедших в EXPIRED за прогон. + */ + fun expireOverdue(): Int { + val batchSize = requestExpiryProperties.batchSize.coerceAtLeast(0) + if (batchSize == 0) { + return 0 + } + val maxBatches = requestExpiryProperties.maxBatchesPerRun.coerceAtLeast(1) + val now = LocalDateTime.now() + + var expiredTotal = 0 + var batchIndex = 0 + while (batchIndex < maxBatches) { + val result = try { + transactionOperations.execute( + TransactionCallback { expireBatch(now, batchSize) }, + ) ?: ExpireBatchResult.EMPTY + } catch (exception: Exception) { + log.error("Failed to process request expiry batch", exception) + break + } + + expiredTotal += result.expiredCount + batchIndex += 1 + if (result.claimedCount < batchSize) { + break + } + } + + if (expiredTotal > 0) { + log.info("Request expiry sweep finished: expired={}", expiredTotal) + } + return expiredTotal + } + + private fun expireBatch(now: LocalDateTime, batchSize: Int): ExpireBatchResult { + val candidates = requestRepository.findExpirableForUpdateSkipLocked(now, batchSize) + var expiredCount = 0 + + candidates.forEach { request -> + val refresh = requestLifecycleService.refreshStatus(request, now) + if (refresh.statusChanged) { + request.updatedAt = now + requestRepository.save(request) + if (request.status == RequestStatus.EXPIRED) { + expiredCount += 1 + // Терминальный статус → уведомляем geoportal через тот же transactional outbox. + requestStatusOutboxService.createTerminalIfAbsent(request, now) + log.info( + "Request expired: requestId={}, endDateTime={}", + request.id, + request.endDateTime, + ) + } + } + } + + return ExpireBatchResult(claimedCount = candidates.size, expiredCount = expiredCount) + } + + private data class ExpireBatchResult( + val claimedCount: Int, + val expiredCount: Int, + ) { + companion object { + val EMPTY = ExpireBatchResult(claimedCount = 0, expiredCount = 0) + } + } +} diff --git a/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/service/RequestService.kt b/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/service/RequestService.kt index 13d366e..e420f55 100644 --- a/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/service/RequestService.kt +++ b/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/service/RequestService.kt @@ -51,6 +51,7 @@ class RequestService( private val requestCellRepository: RequestCellRepository, private val geometryService: GeometryService, private val requestGridProjectionService: RequestGridProjectionService, + private val requestLifecycleService: RequestLifecycleService, private val entityManager: EntityManager, ) { @Transactional @@ -78,7 +79,9 @@ class RequestService( remainingGeometry = normalizedGeometryWkt, geometryArea = geometryArea, remainingArea = geometryArea, - coverageRequiredPercent = DEFAULT_COVERAGE_REQUIRED_PERCENT, + // Порог завершения по покрытию приходит из API (по умолчанию 100%); это прогресс съёмки, + // в отличие от importance — важности заявки для карты приоритетов ячеек. + coverageRequiredPercent = request.coverageRequiredPercent, importance = importance, beginDateTime = request.beginDateTime, endDateTime = request.endDateTime, @@ -90,6 +93,10 @@ class RequestService( updatedAt = now, ) + // Окно заявки может уже идти или закончиться к моменту приёма — сразу выставляем + // корректный статус (ACCEPTED/ACTIVE/EXPIRED), а не всегда ACCEPTED. + requestLifecycleService.refreshStatus(requestEntity, now) + persistNewRequest(requestEntity) replaceSurveyParams(requestEntity.id, request.optics, request.rsa) requestGridProjectionService.rebuildProjection(requestEntity) @@ -97,7 +104,7 @@ class RequestService( return CreateRequestResponseDto( id = requestEntity.id, name = requestEntity.name, - status = RequestStatusDto.ACCEPTED, + status = requestEntity.status.toDto(), surveyType = requestEntity.surveyType.toDto(), importance = requestEntity.importance, message = "Заявка успешно создана", @@ -369,7 +376,6 @@ class RequestService( } private companion object { - const val DEFAULT_COVERAGE_REQUIRED_PERCENT = 100.0 const val POSTGRES_UNIQUE_VIOLATION = "23505" const val REQUESTS_PRIMARY_KEY = "requests_pkey" const val DEFAULT_SORT_FIELD = "createdAt" diff --git a/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/service/RequestCompletedOutboxService.kt b/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/service/RequestStatusOutboxService.kt similarity index 55% rename from services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/service/RequestCompletedOutboxService.kt rename to services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/service/RequestStatusOutboxService.kt index 044c9af..088dfa1 100644 --- a/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/service/RequestCompletedOutboxService.kt +++ b/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/service/RequestStatusOutboxService.kt @@ -12,8 +12,15 @@ import org.springframework.transaction.annotation.Transactional import java.time.LocalDateTime import java.util.UUID +/** + * Складывает в transactional outbox единое событие о **терминальном** статусе заявки + * (`COMPLETED` или `EXPIRED`) для geoportal. Идемпотентно по `(event_type, aggregate_id)`: + * на заявку приходится не более одного терминального события. + * + * Публикуется [OutboxPublisherService] в топик `pcp.request.status.v1` со статусом в payload. + */ @Service -class RequestCompletedOutboxService( +class RequestStatusOutboxService( private val outboxEventRepository: OutboxEventRepository, private val objectMapper: ObjectMapper, @@ -23,12 +30,17 @@ class RequestCompletedOutboxService( private val log = LoggerFactory.getLogger(this::class.java) @Transactional(propagation = Propagation.MANDATORY) - fun createRequestCompletedIfAbsent(request: RequestEntity, eventCreatedAt: LocalDateTime) { - val completedAt = request.completedAt ?: return - val payload = RequestCompletedPayload( + fun createTerminalIfAbsent(request: RequestEntity, eventCreatedAt: LocalDateTime) { + val status = request.status + require(status == RequestStatus.COMPLETED || status == RequestStatus.EXPIRED) { + "Терминальное outbox-событие требует COMPLETED/EXPIRED, получено: $status" + } + + val payload = RequestStatusPayload( requestId = request.id, - status = RequestStatus.COMPLETED.name, - completedAt = completedAt, + status = status.name, + // Для COMPLETED — момент достижения покрытия, для EXPIRED — момент фиксации истечения. + occurredAt = request.completedAt ?: eventCreatedAt, coveragePercent = geometryService.coveredPercent(request.geometryArea, request.remainingArea), matchedAt = request.lastMatchedAt, eventCreatedAt = eventCreatedAt, @@ -36,7 +48,7 @@ class RequestCompletedOutboxService( val inserted = outboxEventRepository.insertIfAbsent( id = UUID.randomUUID(), - eventType = REQUEST_COMPLETED_EVENT_TYPE, + eventType = REQUEST_STATUS_EVENT_TYPE, aggregateType = REQUEST_AGGREGATE_TYPE, aggregateId = request.id, payload = objectMapper.writeValueAsString(payload), @@ -46,23 +58,23 @@ class RequestCompletedOutboxService( if (inserted == 0) { log.debug( - "REQUEST_COMPLETED outbox event already exists for requestId={}", + "Терминальное outbox-событие уже существует для requestId={}", request.id, ) } } - private data class RequestCompletedPayload( + private data class RequestStatusPayload( val requestId: UUID, val status: String, - val completedAt: LocalDateTime, + val occurredAt: LocalDateTime, val coveragePercent: Double, val matchedAt: LocalDateTime?, val eventCreatedAt: LocalDateTime, ) companion object { - const val REQUEST_COMPLETED_EVENT_TYPE = "REQUEST_COMPLETED" + const val REQUEST_STATUS_EVENT_TYPE = "REQUEST_TERMINAL_STATUS" const val REQUEST_AGGREGATE_TYPE = "REQUEST" } } diff --git a/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/service/RouteMatchingService.kt b/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/service/RouteMatchingService.kt index 0fb6ed7..5e5934c 100644 --- a/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/service/RouteMatchingService.kt +++ b/services/pcp-request-service/src/main/kotlin/org/nstart/dep265/requestservice/service/RouteMatchingService.kt @@ -31,7 +31,7 @@ class RouteMatchingService( private val requestLifecycleService: RequestLifecycleService, - private val requestCompletedOutboxService: RequestCompletedOutboxService, + private val requestStatusOutboxService: RequestStatusOutboxService, @param:Qualifier("routeMatchingTransactionOperations") private val routeMatchingTransactionOperations: TransactionOperations, @@ -168,7 +168,7 @@ class RouteMatchingService( val statusRefresh = requestLifecycleService.refreshStatus(request, matchedAt) requestRepository.save(request) if (statusRefresh.completedNow) { - requestCompletedOutboxService.createRequestCompletedIfAbsent(request, matchedAt) + requestStatusOutboxService.createTerminalIfAbsent(request, matchedAt) } return true } diff --git a/services/pcp-request-service/src/main/resources/application.yaml b/services/pcp-request-service/src/main/resources/application.yaml index c3632e3..962dede 100644 --- a/services/pcp-request-service/src/main/resources/application.yaml +++ b/services/pcp-request-service/src/main/resources/application.yaml @@ -14,6 +14,9 @@ spring: pcp: outbox: - request-completed-topic: pcp.request.completed.v1 + request-status-topic: pcp.request.status.v1 publish-batch-size: 50 publish-fixed-delay-ms: 5000 + max-attempts: 10 + initial-backoff-ms: 1000 + max-backoff-ms: 60000 diff --git a/services/pcp-request-service/src/main/resources/db/migration/V1__create_schema.sql b/services/pcp-request-service/src/main/resources/db/migration/V1__create_schema.sql index 96df62e..6ba242d 100644 --- a/services/pcp-request-service/src/main/resources/db/migration/V1__create_schema.sql +++ b/services/pcp-request-service/src/main/resources/db/migration/V1__create_schema.sql @@ -100,6 +100,8 @@ CREATE TABLE IF NOT EXISTS outbox_events ( aggregate_id UUID NOT NULL, payload JSONB NOT NULL, status VARCHAR(32) NOT NULL, + attempts INTEGER NOT NULL DEFAULT 0, + next_attempt_at TIMESTAMP NOT NULL DEFAULT now(), created_at TIMESTAMP NOT NULL, published_at TIMESTAMP NULL, error_message TEXT NULL, @@ -203,6 +205,7 @@ CREATE INDEX IF NOT EXISTS idx_request_kpp_request_id ON request_kpp(request_id) CREATE INDEX IF NOT EXISTS idx_request_kpp_kpp ON request_kpp(kpp); CREATE INDEX IF NOT EXISTS idx_request_route_matches_route_id ON request_route_matches(route_id); CREATE INDEX IF NOT EXISTS idx_outbox_events_status_created_at ON outbox_events(status, created_at); +CREATE INDEX IF NOT EXISTS idx_outbox_events_due ON outbox_events(event_type, status, next_attempt_at, created_at); CREATE INDEX IF NOT EXISTS idx_earth_cells_cell_num ON earth_cells(cell_num); CREATE INDEX IF NOT EXISTS idx_earth_cells_contour_geom ON earth_cells USING GIST (contour_geom); CREATE INDEX IF NOT EXISTS idx_request_cells_request_id ON request_cells(request_id); diff --git a/services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/MigrationSchemaTest.kt b/services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/MigrationSchemaTest.kt index 7fb6226..42a202a 100644 --- a/services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/MigrationSchemaTest.kt +++ b/services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/MigrationSchemaTest.kt @@ -49,8 +49,11 @@ class MigrationSchemaTest { assertTrue(outboxEvents.contains("payload jsonb not null")) assertTrue(outboxEvents.contains("status varchar(32) not null")) + assertTrue(outboxEvents.contains("attempts integer not null default 0")) + assertTrue(outboxEvents.contains("next_attempt_at timestamp not null")) assertTrue(outboxEvents.contains("constraint uk_outbox_events_event_type_aggregate_id unique (event_type, aggregate_id)")) assertTrue(schema.contains("on outbox_events(status, created_at)")) + assertTrue(schema.contains("on outbox_events(event_type, status, next_attempt_at, created_at)")) } @Test diff --git a/services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/config/KafkaConsumerConfigTest.kt b/services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/config/KafkaConsumerConfigTest.kt new file mode 100644 index 0000000..b5c74b7 --- /dev/null +++ b/services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/config/KafkaConsumerConfigTest.kt @@ -0,0 +1,175 @@ +package org.nstart.dep265.requestservice.config + +import jakarta.validation.ConstraintViolation +import jakarta.validation.ConstraintViolationException +import org.apache.kafka.clients.consumer.Consumer +import org.apache.kafka.clients.consumer.ConsumerRecord +import org.apache.kafka.clients.producer.ProducerRecord +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertFalse +import org.junit.jupiter.api.Assertions.assertNotNull +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test +import org.mockito.ArgumentCaptor +import org.mockito.ArgumentMatchers.any +import org.mockito.Mockito.mock +import org.mockito.Mockito.never +import org.mockito.Mockito.times +import org.mockito.Mockito.verify +import org.mockito.Mockito.`when` +import org.springframework.boot.autoconfigure.AutoConfigurations +import org.springframework.boot.autoconfigure.context.ConfigurationPropertiesAutoConfiguration +import org.springframework.boot.context.properties.EnableConfigurationProperties +import org.springframework.boot.test.context.runner.ApplicationContextRunner +import org.springframework.context.annotation.Bean +import org.springframework.context.annotation.Configuration +import org.springframework.kafka.core.KafkaTemplate +import org.springframework.kafka.listener.DeadLetterPublishingRecoverer +import org.springframework.kafka.listener.DefaultErrorHandler +import org.springframework.kafka.listener.MessageListenerContainer +import org.springframework.kafka.support.SendResult +import java.nio.charset.StandardCharsets +import java.util.concurrent.CompletableFuture + +class KafkaConsumerConfigTest { + + private val contextRunner = ApplicationContextRunner() + .withConfiguration(AutoConfigurations.of(ConfigurationPropertiesAutoConfiguration::class.java)) + .withUserConfiguration(TestConfiguration::class.java, KafkaConsumerConfig::class.java) + .withPropertyValues( + "spring.kafka.bootstrap-servers=localhost:9092", + "spring.kafka.consumer.group-id=pcp-request-service", + "spring.kafka.consumer.auto-offset-reset=earliest", + "app.kafka.consumer.route.dlq-topic=$DLQ_TOPIC", + "app.kafka.consumer.route.max-attempts=3", + "app.kafka.consumer.route.backoff-ms=0", + "app.kafka.consumer.route.max-backoff-ms=0", + ) + + @Test + fun `creates error handler and recoverer beans`() { + contextRunner.run { context -> + assertNotNull(context.getBean(DefaultErrorHandler::class.java)) + assertNotNull(context.getBean(DeadLetterPublishingRecoverer::class.java)) + + val properties = context.getBean(RouteConsumerProperties::class.java) + assertEquals(DLQ_TOPIC, properties.dlqTopic) + assertEquals(3L, properties.maxAttempts) + } + } + + @Test + fun `recoverer publishes original record to configured dlq topic with diagnostics headers`() { + contextRunner.run { context -> + val kafkaTemplate = context.kafkaTemplate() + val recoverer = context.getBean(DeadLetterPublishingRecoverer::class.java) + val record = consumerRecord() + + recoverer.accept(record, ConstraintViolationException("route payload invalid", emptySet>())) + + val published = capturePublishedRecord(kafkaTemplate) + assertEquals(DLQ_TOPIC, published.topic()) + assertEquals(record.partition(), published.partition()) + assertEquals(record.key(), published.key()) + assertEquals(record.value(), published.value()) + assertEquals("pcp-request-service", published.headers().stringHeader("pcp-service-name")) + assertEquals( + ConstraintViolationException::class.java.name, + published.headers().stringHeader("pcp-exception-class"), + ) + assertEquals("route payload invalid", published.headers().stringHeader("pcp-exception-message")) + assertEquals(ROUTE_TOPIC, published.headers().stringHeader("pcp-original-topic")) + assertEquals("3", published.headers().stringHeader("pcp-original-partition")) + assertEquals("42", published.headers().stringHeader("pcp-original-offset")) + } + } + + @Test + fun `contract violation is recovered to dlq without retry`() { + contextRunner.run { context -> + val kafkaTemplate = context.kafkaTemplate() + val errorHandler = context.getBean(DefaultErrorHandler::class.java) + + val handled = errorHandler.handleOne( + ConstraintViolationException("route payload invalid", emptySet>()), + consumerRecord(), + mockConsumer(), + mock(MessageListenerContainer::class.java), + ) + + assertTrue(handled) + verify(kafkaTemplate, times(1)).send(anyProducerRecord()) + } + } + + @Test + fun `transient exception is retried with bounded policy before dlq recovery`() { + contextRunner.run { context -> + val kafkaTemplate = context.kafkaTemplate() + val errorHandler = context.getBean(DefaultErrorHandler::class.java) + val record = consumerRecord() + val consumer = mockConsumer() + val container = mock(MessageListenerContainer::class.java) + val exception = IllegalStateException("database is temporarily unavailable") + + // maxAttempts=3 → две первые доставки ретраятся (seek), на третьей — DLQ. + assertFalse(errorHandler.handleOne(exception, record, consumer, container)) + assertFalse(errorHandler.handleOne(exception, record, consumer, container)) + verify(kafkaTemplate, never()).send(anyProducerRecord()) + + assertTrue(errorHandler.handleOne(exception, record, consumer, container)) + verify(kafkaTemplate, times(1)).send(anyProducerRecord()) + } + } + + private fun consumerRecord(): ConsumerRecord = + ConsumerRecord(ROUTE_TOPIC, 3, 42L, "route-key", """{"routeId":"route-1"}""") + + private fun capturePublishedRecord( + kafkaTemplate: KafkaTemplate, + ): ProducerRecord { + val captor = producerRecordCaptor() + verify(kafkaTemplate, times(1)).send(captor.capture()) + return captor.value + } + + @Suppress("UNCHECKED_CAST") + private fun producerRecordCaptor(): ArgumentCaptor> = + ArgumentCaptor.forClass(ProducerRecord::class.java) as ArgumentCaptor> + + @Suppress("UNCHECKED_CAST") + private fun mockConsumer(): Consumer = + mock(Consumer::class.java) as Consumer + + private fun org.apache.kafka.common.header.Headers.stringHeader(name: String): String? = + lastHeader(name)?.value()?.toString(StandardCharsets.UTF_8) + + @Configuration(proxyBeanMethods = false) + @EnableConfigurationProperties(RouteConsumerProperties::class) + private class TestConfiguration { + + @Bean + @Suppress("UNCHECKED_CAST") + fun kafkaTemplate(): KafkaTemplate { + val kafkaTemplate = mock(KafkaTemplate::class.java) as KafkaTemplate + val sendResult = mock(SendResult::class.java) as SendResult + `when`(kafkaTemplate.send(anyProducerRecord())) + .thenReturn(CompletableFuture.completedFuture(sendResult)) + return kafkaTemplate + } + } + + private companion object { + const val ROUTE_TOPIC = "pcp.route.georeference.v1" + const val DLQ_TOPIC = "pcp.route.georeference-dlq.v1" + + @Suppress("UNCHECKED_CAST") + fun anyProducerRecord(): ProducerRecord = + (any(ProducerRecord::class.java) as ProducerRecord?) + ?: ProducerRecord("stub-topic", "stub-key", "stub-payload") + + @Suppress("UNCHECKED_CAST") + fun org.springframework.context.ApplicationContext.kafkaTemplate(): KafkaTemplate = + getBean(KafkaTemplate::class.java) as KafkaTemplate + } +} diff --git a/services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/controller/RequestControllerOpenApiSkeletonTest.kt b/services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/controller/RequestControllerOpenApiSkeletonTest.kt index c91a450..13dbf15 100644 --- a/services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/controller/RequestControllerOpenApiSkeletonTest.kt +++ b/services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/controller/RequestControllerOpenApiSkeletonTest.kt @@ -25,6 +25,7 @@ import org.nstart.dep265.requestservice.repository.RequestRepository import org.nstart.dep265.requestservice.repository.RequestRsaParamsRepository import org.nstart.dep265.requestservice.service.GeometryService import org.nstart.dep265.requestservice.service.RequestGridProjectionService +import org.nstart.dep265.requestservice.service.RequestLifecycleService import org.nstart.dep265.requestservice.service.RequestService import org.springframework.data.domain.PageImpl import org.springframework.data.domain.PageRequest @@ -98,6 +99,7 @@ class RequestControllerOpenApiSkeletonTest { requestCellRepository = requestCellRepository, geometryService = GeometryService(), requestGridProjectionService = mock(RequestGridProjectionService::class.java), + requestLifecycleService = RequestLifecycleService(GeometryService()), entityManager = entityManager, ) ) @@ -114,6 +116,7 @@ class RequestControllerOpenApiSkeletonTest { @Test fun `post v1 requests returns create response`() { + // Окно в будущем → приём детерминированно даёт ACCEPTED (createRequest зовёт refreshStatus). mockMvc.perform( post("/v1/requests") .contentType(MediaType.APPLICATION_JSON) @@ -125,8 +128,8 @@ class RequestControllerOpenApiSkeletonTest { "name": "Тест1", "geometry": "POLYGON ((37.5 55.5, 37.7 55.5, 37.7 55.7, 37.5 55.7, 37.5 55.5))", "importance": 10.0, - "beginDateTime": "2026-01-01T00:00:00Z", - "endDateTime": "2026-01-31T23:59:59Z", + "beginDateTime": "2090-01-01T00:00:00Z", + "endDateTime": "2090-01-31T23:59:59Z", "kpp": [1, 2, 3], "highPriorityTransmit": false, "optics": { diff --git a/services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/dto/RequestApiDtoContractTest.kt b/services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/dto/RequestApiDtoContractTest.kt index 3996a45..d9cabac 100644 --- a/services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/dto/RequestApiDtoContractTest.kt +++ b/services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/dto/RequestApiDtoContractTest.kt @@ -45,6 +45,31 @@ class RequestApiDtoContractTest { assertEquals(10.0, request.importance) assertEquals(LocalDateTime.of(2026, 1, 1, 0, 0, 0), request.beginDateTime) assertEquals(OpticsResultTypeDto.PANCHROMATIC, request.optics?.resultType) + // Поле опционально: при отсутствии в payload — порог покрытия по умолчанию 100%. + assertEquals(100.0, request.coverageRequiredPercent) + } + + @Test + fun `create request dto binds explicit coverage required percent`() { + val request = objectMapper.readValue( + """ + { + "id": "3fa85f64-5717-4562-b3fc-2c963f66afa6", + "name": "Тест1", + "geometry": "POLYGON ((37.5 55.5, 37.7 55.5, 37.7 55.7, 37.5 55.7, 37.5 55.5))", + "importance": 10.0, + "beginDateTime": "2026-01-01T00:00:00", + "endDateTime": "2026-01-31T23:59:59", + "coverageRequiredPercent": 80.0, + "optics": { + "resultType": "PANCHROMATIC", + "resolution": 1.0 + } + } + """.trimIndent(), + ) + + assertEquals(80.0, request.coverageRequiredPercent) } @Test diff --git a/services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/repository/OutboxEventRepositoryJpaTest.kt b/services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/repository/OutboxEventRepositoryJpaTest.kt index 7cf9b2c..38fe235 100644 --- a/services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/repository/OutboxEventRepositoryJpaTest.kt +++ b/services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/repository/OutboxEventRepositoryJpaTest.kt @@ -82,19 +82,77 @@ class OutboxEventRepositoryJpaTest @Autowired constructor( ) outboxEventRepository.saveAllAndFlush(listOf(oldFailedEvent, newEvent, newerFailedEvent, publishedEvent, otherEventType)) - val events = outboxEventRepository.findRequestCompletedPublishableForUpdateSkipLocked( - eventType = "REQUEST_COMPLETED", + val events = outboxEventRepository.findPublishableForUpdateSkipLocked( + eventType = "REQUEST_TERMINAL_STATUS", + now = LocalDateTime.parse("2026-06-06T00:00:00"), + maxAttempts = 10, batchSize = 10, ) assertEquals(listOf(newEvent.id, oldFailedEvent.id, newerFailedEvent.id), events.map { it.id }) } + @Test + @Transactional + fun `publishable query excludes events whose backoff has not elapsed`() { + val now = LocalDateTime.parse("2026-06-06T00:00:00") + val dueFailed = outboxEvent( + requestId = UUID.fromString("aaaaaaaa-0000-4000-8000-000000000001"), + status = OutboxEventStatus.FAILED, + attempts = 1, + nextAttemptAt = now.minusMinutes(1), + ) + val backoffPendingFailed = outboxEvent( + requestId = UUID.fromString("bbbbbbbb-0000-4000-8000-000000000002"), + status = OutboxEventStatus.FAILED, + attempts = 1, + nextAttemptAt = now.plusMinutes(5), + ) + outboxEventRepository.saveAllAndFlush(listOf(dueFailed, backoffPendingFailed)) + + val events = outboxEventRepository.findPublishableForUpdateSkipLocked( + eventType = "REQUEST_TERMINAL_STATUS", + now = now, + maxAttempts = 10, + batchSize = 10, + ) + + assertEquals(listOf(dueFailed.id), events.map { it.id }) + } + + @Test + @Transactional + fun `publishable query excludes events that exhausted max attempts`() { + val now = LocalDateTime.parse("2026-06-06T00:00:00") + val retriable = outboxEvent( + requestId = UUID.fromString("cccccccc-0000-4000-8000-000000000003"), + status = OutboxEventStatus.FAILED, + attempts = 9, + ) + val exhausted = outboxEvent( + requestId = UUID.fromString("dddddddd-0000-4000-8000-000000000004"), + status = OutboxEventStatus.FAILED, + attempts = 10, + ) + outboxEventRepository.saveAllAndFlush(listOf(retriable, exhausted)) + + val events = outboxEventRepository.findPublishableForUpdateSkipLocked( + eventType = "REQUEST_TERMINAL_STATUS", + now = now, + maxAttempts = 10, + batchSize = 10, + ) + + assertEquals(listOf(retriable.id), events.map { it.id }) + } + private fun outboxEvent( requestId: UUID, - eventType: String = "REQUEST_COMPLETED", + eventType: String = "REQUEST_TERMINAL_STATUS", status: OutboxEventStatus = OutboxEventStatus.NEW, createdAt: LocalDateTime = LocalDateTime.parse("2026-01-01T10:00:00"), + attempts: Int = 0, + nextAttemptAt: LocalDateTime = createdAt, ): OutboxEventEntity { return OutboxEventEntity( id = UUID.randomUUID(), @@ -104,6 +162,8 @@ class OutboxEventRepositoryJpaTest @Autowired constructor( payload = objectMapper.readTree("""{"requestId":"$requestId","status":"COMPLETED"}"""), status = status, createdAt = createdAt, + attempts = attempts, + nextAttemptAt = nextAttemptAt, ) } } diff --git a/services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/repository/RequestRepositoryJpaTest.kt b/services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/repository/RequestRepositoryJpaTest.kt new file mode 100644 index 0000000..aa4179f --- /dev/null +++ b/services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/repository/RequestRepositoryJpaTest.kt @@ -0,0 +1,104 @@ +package org.nstart.dep265.requestservice.repository + +import org.junit.jupiter.api.BeforeEach +import org.junit.jupiter.api.Test +import org.nstart.dep265.requestservice.PcpRequestServiceApplication +import org.nstart.dep265.requestservice.entity.RequestEntity +import org.nstart.dep265.requestservice.entity.RequestStatus +import org.nstart.dep265.requestservice.entity.RequestSurveyType +import org.springframework.beans.factory.annotation.Autowired +import org.springframework.boot.test.context.SpringBootTest +import org.springframework.transaction.annotation.Transactional +import java.time.LocalDateTime +import java.util.UUID +import kotlin.test.assertEquals + +@SpringBootTest( + classes = [PcpRequestServiceApplication::class], + webEnvironment = SpringBootTest.WebEnvironment.NONE, + properties = [ + "spring.config.import=", + "spring.cloud.config.enabled=false", + "spring.flyway.enabled=false", + "spring.jpa.hibernate.ddl-auto=create-drop", + "spring.datasource.url=jdbc:h2:mem:request-expiry;MODE=PostgreSQL;DB_CLOSE_DELAY=-1;DATABASE_TO_UPPER=false", + "spring.datasource.driver-class-name=org.h2.Driver", + "spring.datasource.username=sa", + "spring.datasource.password=", + "spring.jpa.database-platform=org.hibernate.dialect.H2Dialect", + "spring.kafka.listener.auto-startup=false", + "spring.kafka.consumer.group-id=request-service-test", + "app.kafka.topics.route=request-service-test-route", + ], +) +class RequestRepositoryJpaTest @Autowired constructor( + private val requestRepository: RequestRepository, +) { + private val now: LocalDateTime = LocalDateTime.parse("2026-06-06T12:00:00") + + @BeforeEach + fun setUp() { + requestRepository.deleteAll() + } + + @Test + @Transactional + fun `expirable query returns only overdue accepted and active requests`() { + val overdueAccepted = request(status = RequestStatus.ACCEPTED, end = now.minusHours(1)) + val overdueActive = request(status = RequestStatus.ACTIVE, end = now.minusHours(2)) + val activeInWindow = request(status = RequestStatus.ACTIVE, end = now.plusHours(1)) + val alreadyCompleted = request(status = RequestStatus.COMPLETED, end = now.minusHours(1)) + val alreadyExpired = request(status = RequestStatus.EXPIRED, end = now.minusHours(1)) + val deletedOverdue = request( + status = RequestStatus.DELETED, + end = now.minusHours(1), + deletedAt = now.minusMinutes(30), + ) + requestRepository.saveAllAndFlush( + listOf(overdueAccepted, overdueActive, activeInWindow, alreadyCompleted, alreadyExpired, deletedOverdue), + ) + + val expirable = requestRepository.findExpirableForUpdateSkipLocked(now = now, batchSize = 100) + + // Только незавершённые (ACCEPTED/ACTIVE) с истёкшим окном и без soft-delete, по возрастанию end. + assertEquals(listOf(overdueActive.id, overdueAccepted.id), expirable.map { it.id }) + } + + @Test + @Transactional + fun `expirable query honours the batch size limit`() { + val oldest = request(status = RequestStatus.ACTIVE, end = now.minusHours(3)) + val middle = request(status = RequestStatus.ACCEPTED, end = now.minusHours(2)) + val newest = request(status = RequestStatus.ACTIVE, end = now.minusHours(1)) + requestRepository.saveAllAndFlush(listOf(newest, oldest, middle)) + + val expirable = requestRepository.findExpirableForUpdateSkipLocked(now = now, batchSize = 2) + + assertEquals(listOf(oldest.id, middle.id), expirable.map { it.id }) + } + + private fun request( + status: RequestStatus, + end: LocalDateTime, + begin: LocalDateTime = end.minusDays(1), + deletedAt: LocalDateTime? = null, + ): RequestEntity { + val geometry = "POLYGON ((37.5 55.5, 37.7 55.5, 37.7 55.7, 37.5 55.7, 37.5 55.5))" + return RequestEntity( + id = UUID.randomUUID(), + name = "Тест1", + geometry = geometry, + remainingGeometry = geometry, + geometryArea = 1.0, + remainingArea = 1.0, + importance = 10.0, + beginDateTime = begin, + endDateTime = end, + surveyType = RequestSurveyType.OPTICS, + status = status, + createdAt = begin.minusDays(1), + updatedAt = begin.minusDays(1), + deletedAt = deletedAt, + ) + } +} diff --git a/services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/service/OutboxPublisherServiceTest.kt b/services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/service/OutboxPublisherServiceTest.kt index 83b327c..fac212c 100644 --- a/services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/service/OutboxPublisherServiceTest.kt +++ b/services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/service/OutboxPublisherServiceTest.kt @@ -6,6 +6,8 @@ import org.junit.jupiter.api.Assertions.assertNotNull import org.junit.jupiter.api.Assertions.assertNull import org.junit.jupiter.api.Assertions.assertTrue import org.junit.jupiter.api.Test +import org.mockito.ArgumentMatchers.any +import org.mockito.ArgumentMatchers.anyInt import org.mockito.ArgumentMatchers.anyString import org.mockito.Mockito.doReturn import org.mockito.Mockito.doAnswer @@ -30,13 +32,16 @@ import java.util.concurrent.CompletableFuture class OutboxPublisherServiceTest { private val objectMapper = jacksonObjectMapper() private val properties = OutboxProperties( - requestCompletedTopic = "pcp.request.completed.v1", + requestStatusTopic = "pcp.request.status.v1", publishBatchSize = 50, publishFixedDelayMs = 5_000, + maxAttempts = 10, + initialBackoffMs = 1_000, + maxBackoffMs = 60_000, ) @Test - fun `new request completed event is published and marked as published`() { + fun `new terminal event is published and marked as published`() { val requestId = UUID.fromString("6cc8ed4b-0db1-4cfb-bef0-d56eed2edce4") val event = outboxEvent( requestId = requestId, @@ -50,7 +55,7 @@ class OutboxPublisherServiceTest { assertEquals(1, processedCount) verify(kafkaTemplate).send( - properties.requestCompletedTopic, + properties.requestStatusTopic, requestId.toString(), event.payload.toString(), ) @@ -61,11 +66,12 @@ class OutboxPublisherServiceTest { } @Test - fun `failed request completed event is picked for retry`() { + fun `failed event is picked for retry`() { val requestId = UUID.fromString("9986e096-886a-429e-af7f-3a6178c29b4f") val event = outboxEvent( requestId = requestId, status = OutboxEventStatus.FAILED, + attempts = 1, errorMessage = "previous failure", ) val outboxEventRepository = repositoryReturning(listOf(event)) @@ -75,7 +81,7 @@ class OutboxPublisherServiceTest { service.publishPending() verify(kafkaTemplate).send( - properties.requestCompletedTopic, + properties.requestStatusTopic, requestId.toString(), event.payload.toString(), ) @@ -95,14 +101,14 @@ class OutboxPublisherServiceTest { service.publishPending() verify(kafkaTemplate).send( - properties.requestCompletedTopic, + properties.requestStatusTopic, requestId.toString(), objectMapper.readTree(savedPayload).toString(), ) } @Test - fun `kafka failure marks event as failed and stores short error message`() { + fun `kafka failure increments attempts, schedules backoff and stores short error message`() { val requestId = UUID.fromString("116eb4c7-cd6d-4837-8525-21c1d16ecbbb") val event = outboxEvent(requestId = requestId) val outboxEventRepository = repositoryReturning(listOf(event)) @@ -113,17 +119,42 @@ class OutboxPublisherServiceTest { assertEquals(1, processedCount) assertEquals(OutboxEventStatus.FAILED, event.status) + assertEquals(1, event.attempts) assertTrue(event.errorMessage!!.contains("broker unavailable")) assertNull(event.publishedAt) + // backoff: следующая попытка отложена в будущее относительно создания события. + assertTrue(event.nextAttemptAt.isAfter(event.createdAt)) verify(outboxEventRepository).save(event) } + @Test + fun `event exhausting max attempts becomes terminal and is not rescheduled`() { + val requestId = UUID.fromString("c0ffee00-0000-4000-8000-000000000001") + val event = outboxEvent( + requestId = requestId, + status = OutboxEventStatus.FAILED, + attempts = properties.maxAttempts - 1, + errorMessage = "previous failure", + ) + val outboxEventRepository = repositoryReturning(listOf(event)) + val kafkaTemplate = kafkaTemplateReturning(failedSend(IllegalStateException("still broken"))) + val service = publisherService(outboxEventRepository, kafkaTemplate) + + service.publishPending() + + assertEquals(OutboxEventStatus.FAILED, event.status) + assertEquals(properties.maxAttempts, event.attempts) + // Исчерпав попытки, событие терминально: backoff не назначается (next_attempt_at не двигаем). + assertEquals(event.createdAt, event.nextAttemptAt) + } + @Test fun `failed event is not retried repeatedly in same publish pending call`() { val requestId = UUID.fromString("f31992a5-efae-4d9a-8262-78381da835f7") val event = outboxEvent( requestId = requestId, status = OutboxEventStatus.FAILED, + attempts = 1, errorMessage = "previous failure", ) val outboxEventRepository = repositoryReturning(listOf(event)) @@ -134,15 +165,18 @@ class OutboxPublisherServiceTest { assertEquals(1, processedCount) verify(kafkaTemplate, times(1)).send( - properties.requestCompletedTopic, + properties.requestStatusTopic, requestId.toString(), event.payload.toString(), ) - verify(outboxEventRepository, times(1)).findRequestCompletedPublishableForUpdateSkipLocked( - RequestCompletedOutboxService.REQUEST_COMPLETED_EVENT_TYPE, - properties.publishBatchSize, + verify(outboxEventRepository, times(1)).findPublishableForUpdateSkipLocked( + anyString(), + anyLocalDateTime(), + anyInt(), + anyInt(), ) assertEquals(OutboxEventStatus.FAILED, event.status) + assertEquals(2, event.attempts) assertTrue(event.errorMessage!!.contains("still broken")) } @@ -153,6 +187,7 @@ class OutboxPublisherServiceTest { val failedEvent = outboxEvent( requestId = failedRequestId, status = OutboxEventStatus.FAILED, + attempts = 1, errorMessage = "previous failure", ) val newEvent = outboxEvent(requestId = newRequestId) @@ -164,12 +199,12 @@ class OutboxPublisherServiceTest { assertEquals(2, processedCount) verify(kafkaTemplate).send( - properties.requestCompletedTopic, + properties.requestStatusTopic, failedRequestId.toString(), failedEvent.payload.toString(), ) verify(kafkaTemplate).send( - properties.requestCompletedTopic, + properties.requestStatusTopic, newRequestId.toString(), newEvent.payload.toString(), ) @@ -206,9 +241,11 @@ class OutboxPublisherServiceTest { val outboxEventRepository = mock(OutboxEventRepository::class.java) doReturn(events) .`when`(outboxEventRepository) - .findRequestCompletedPublishableForUpdateSkipLocked( - RequestCompletedOutboxService.REQUEST_COMPLETED_EVENT_TYPE, - properties.publishBatchSize, + .findPublishableForUpdateSkipLocked( + anyString(), + anyLocalDateTime(), + anyInt(), + anyInt(), ) return outboxEventRepository } @@ -256,20 +293,31 @@ class OutboxPublisherServiceTest { requestId: UUID, payload: String = """{"requestId":"$requestId","status":"COMPLETED"}""", status: OutboxEventStatus = OutboxEventStatus.NEW, + attempts: Int = 0, errorMessage: String? = null, ): OutboxEventEntity { + val createdAt = LocalDateTime.parse("2026-01-01T10:00:00") return OutboxEventEntity( id = UUID.randomUUID(), - eventType = RequestCompletedOutboxService.REQUEST_COMPLETED_EVENT_TYPE, - aggregateType = RequestCompletedOutboxService.REQUEST_AGGREGATE_TYPE, + eventType = RequestStatusOutboxService.REQUEST_STATUS_EVENT_TYPE, + aggregateType = RequestStatusOutboxService.REQUEST_AGGREGATE_TYPE, aggregateId = requestId, payload = objectMapper.readTree(payload), status = status, - createdAt = LocalDateTime.parse("2026-01-01T10:00:00"), + createdAt = createdAt, + attempts = attempts, + nextAttemptAt = createdAt, errorMessage = errorMessage, ) } + // Регистрирует matcher, но возвращает non-null заглушку (any() на non-null LocalDateTime в Kotlin + // вернул бы null и сломал стек матчеров Mockito). + private fun anyLocalDateTime(): LocalDateTime { + any(LocalDateTime::class.java) + return LocalDateTime.now() + } + private class ImmediateTransactionOperations : TransactionOperations { override fun execute(action: TransactionCallback): T { return action.doInTransaction(SimpleTransactionStatus()) diff --git a/services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/service/RequestCompletedOutboxServiceTest.kt b/services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/service/RequestCompletedOutboxServiceTest.kt deleted file mode 100644 index 944eea5..0000000 --- a/services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/service/RequestCompletedOutboxServiceTest.kt +++ /dev/null @@ -1,65 +0,0 @@ -package org.nstart.dep265.requestservice.service - -import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule -import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper -import org.junit.jupiter.api.Assertions.assertEquals -import org.junit.jupiter.api.Assertions.assertNotNull -import org.junit.jupiter.api.Test -import org.mockito.Mockito.mock -import org.mockito.Mockito.mockingDetails -import org.nstart.dep265.requestservice.entity.RequestEntity -import org.nstart.dep265.requestservice.entity.RequestStatus -import org.nstart.dep265.requestservice.entity.RequestSurveyType -import org.nstart.dep265.requestservice.repository.OutboxEventRepository -import java.time.LocalDateTime -import java.util.UUID - -class RequestCompletedOutboxServiceTest { - private val objectMapper = jacksonObjectMapper() - .registerModule(JavaTimeModule()) - .findAndRegisterModules() - - @Test - fun `request completed payload contains required fields`() { - val outboxEventRepository = mock(OutboxEventRepository::class.java) - val service = RequestCompletedOutboxService( - outboxEventRepository = outboxEventRepository, - objectMapper = objectMapper, - geometryService = GeometryService(), - ) - val requestId = UUID.fromString("5f57cd22-44a7-4fe8-956f-7fd0c52df846") - val completedAt = LocalDateTime.parse("2026-01-01T10:15:30") - val matchedAt = LocalDateTime.parse("2026-01-01T10:14:00") - val eventCreatedAt = LocalDateTime.parse("2026-01-01T10:16:00") - val request = RequestEntity( - id = requestId, - name = "Тест1", - geometry = "POLYGON((0 0, 1 0, 1 1, 0 1, 0 0))", - remainingGeometry = "POLYGON EMPTY", - geometryArea = 1.0, - remainingArea = 0.0, - importance = 10.0, - beginDateTime = LocalDateTime.parse("2026-01-01T10:00:00"), - endDateTime = LocalDateTime.parse("2026-01-01T11:00:00"), - surveyType = RequestSurveyType.OPTICS, - status = RequestStatus.COMPLETED, - lastMatchedAt = matchedAt, - completedAt = completedAt, - createdAt = LocalDateTime.parse("2026-01-01T09:00:00"), - updatedAt = completedAt, - ) - - service.createRequestCompletedIfAbsent(request, eventCreatedAt) - - val insertInvocation = mockingDetails(outboxEventRepository).invocations.single { - invocation -> invocation.method.name == "insertIfAbsent" - } - val payload = objectMapper.readTree(insertInvocation.arguments[4] as String) - assertEquals(requestId.toString(), payload["requestId"].asText()) - assertEquals("COMPLETED", payload["status"].asText()) - assertNotNull(payload["completedAt"]) - assertEquals(100.0, payload["coveragePercent"].asDouble()) - assertNotNull(payload["matchedAt"]) - assertNotNull(payload["eventCreatedAt"]) - } -} diff --git a/services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/service/RequestExpiryServiceTest.kt b/services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/service/RequestExpiryServiceTest.kt new file mode 100644 index 0000000..31c0cb8 --- /dev/null +++ b/services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/service/RequestExpiryServiceTest.kt @@ -0,0 +1,203 @@ +package org.nstart.dep265.requestservice.service + +import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule +import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test +import org.mockito.ArgumentMatchers.any +import org.mockito.ArgumentMatchers.anyInt +import org.mockito.Mockito.doReturn +import org.mockito.Mockito.mock +import org.mockito.Mockito.mockingDetails +import org.mockito.Mockito.never +import org.mockito.Mockito.times +import org.mockito.Mockito.verify +import org.mockito.Mockito.verifyNoInteractions +import org.nstart.dep265.requestservice.config.RequestExpiryProperties +import org.nstart.dep265.requestservice.entity.RequestEntity +import org.nstart.dep265.requestservice.entity.RequestStatus +import org.nstart.dep265.requestservice.entity.RequestSurveyType +import org.nstart.dep265.requestservice.repository.OutboxEventRepository +import org.nstart.dep265.requestservice.repository.RequestRepository +import org.springframework.transaction.TransactionStatus +import org.springframework.transaction.support.SimpleTransactionStatus +import org.springframework.transaction.support.TransactionCallback +import org.springframework.transaction.support.TransactionOperations +import java.time.LocalDateTime +import java.util.UUID + +class RequestExpiryServiceTest { + + @Test + fun `expires overdue accepted and active requests and emits terminal events`() { + val requestRepository = mock(RequestRepository::class.java) + val outboxEventRepository = mock(OutboxEventRepository::class.java) + val service = expiryService(requestRepository, outboxEventRepository, RequestExpiryProperties(batchSize = 200)) + + val expiredAccepted = overdueRequest(status = RequestStatus.ACCEPTED) + val expiredActive = overdueRequest(status = RequestStatus.ACTIVE) + doReturn(listOf(expiredAccepted, expiredActive)) + .`when`(requestRepository) + .findExpirableForUpdateSkipLocked(anyLocalDateTime(), anyInt()) + + val expiredCount = service.expireOverdue() + + assertEquals(2, expiredCount) + assertEquals(RequestStatus.EXPIRED, expiredAccepted.status) + assertEquals(RequestStatus.EXPIRED, expiredActive.status) + verify(requestRepository).save(expiredAccepted) + verify(requestRepository).save(expiredActive) + + // По каждой истёкшей заявке — терминальное событие в outbox для geoportal. + val terminalInserts = mockingDetails(outboxEventRepository).invocations + .filter { invocation -> invocation.method.name == "insertIfAbsent" } + assertEquals(2, terminalInserts.size) + assertTrue(terminalInserts.all { invocation -> invocation.arguments[1] == "REQUEST_TERMINAL_STATUS" }) + } + + @Test + fun `keeps fetching while a full batch is returned`() { + val requestRepository = mock(RequestRepository::class.java) + val service = expiryService( + requestRepository, + mock(OutboxEventRepository::class.java), + RequestExpiryProperties(batchSize = 1), + ) + + val first = overdueRequest(status = RequestStatus.ACTIVE) + val second = overdueRequest(status = RequestStatus.ACCEPTED) + doReturn(listOf(first), listOf(second), emptyList()) + .`when`(requestRepository) + .findExpirableForUpdateSkipLocked(anyLocalDateTime(), anyInt()) + + val expiredCount = service.expireOverdue() + + assertEquals(2, expiredCount) + verify(requestRepository, times(3)) + .findExpirableForUpdateSkipLocked(anyLocalDateTime(), anyInt()) + } + + @Test + fun `empty sweep returns zero and emits nothing`() { + val requestRepository = mock(RequestRepository::class.java) + val outboxEventRepository = mock(OutboxEventRepository::class.java) + val service = expiryService(requestRepository, outboxEventRepository, RequestExpiryProperties(batchSize = 200)) + + doReturn(emptyList()) + .`when`(requestRepository) + .findExpirableForUpdateSkipLocked(anyLocalDateTime(), anyInt()) + + val expiredCount = service.expireOverdue() + + assertEquals(0, expiredCount) + verify(requestRepository, never()).save(any(RequestEntity::class.java)) + verifyNoInteractions(outboxEventRepository) + } + + @Test + fun `request still inside its window is not expired`() { + val requestRepository = mock(RequestRepository::class.java) + val outboxEventRepository = mock(OutboxEventRepository::class.java) + val service = expiryService(requestRepository, outboxEventRepository, RequestExpiryProperties(batchSize = 200)) + + val now = LocalDateTime.now() + val inWindow = request( + status = RequestStatus.ACTIVE, + beginDateTime = now.minusDays(1), + endDateTime = now.plusDays(1), + ) + doReturn(listOf(inWindow)) + .`when`(requestRepository) + .findExpirableForUpdateSkipLocked(anyLocalDateTime(), anyInt()) + + val expiredCount = service.expireOverdue() + + assertEquals(0, expiredCount) + assertEquals(RequestStatus.ACTIVE, inWindow.status) + verify(requestRepository, never()).save(any(RequestEntity::class.java)) + verifyNoInteractions(outboxEventRepository) + } + + @Test + fun `disabled scheduler does not sweep`() { + val requestRepository = mock(RequestRepository::class.java) + val service = expiryService( + requestRepository, + mock(OutboxEventRepository::class.java), + RequestExpiryProperties(enabled = false), + ) + + service.expireOverdueScheduledBatch() + + verify(requestRepository, never()) + .findExpirableForUpdateSkipLocked(anyLocalDateTime(), anyInt()) + } + + private fun expiryService( + requestRepository: RequestRepository, + outboxEventRepository: OutboxEventRepository, + properties: RequestExpiryProperties, + ): RequestExpiryService { + val geometryService = GeometryService() + return RequestExpiryService( + requestRepository = requestRepository, + requestLifecycleService = RequestLifecycleService(geometryService), + requestStatusOutboxService = RequestStatusOutboxService( + outboxEventRepository = outboxEventRepository, + objectMapper = jacksonObjectMapper() + .registerModule(JavaTimeModule()) + .findAndRegisterModules(), + geometryService = geometryService, + ), + requestExpiryProperties = properties, + transactionOperations = ImmediateTransactionOperations(), + ) + } + + private fun overdueRequest(status: RequestStatus): RequestEntity { + val now = LocalDateTime.now() + return request( + status = status, + beginDateTime = now.minusDays(10), + endDateTime = now.minusDays(1), + ) + } + + private fun request( + status: RequestStatus, + beginDateTime: LocalDateTime, + endDateTime: LocalDateTime, + ): RequestEntity { + val geometry = "POLYGON ((0 0, 1 0, 1 1, 0 1, 0 0))" + return RequestEntity( + id = UUID.randomUUID(), + name = "Тест1", + geometry = geometry, + remainingGeometry = geometry, + geometryArea = 1.0, + remainingArea = 1.0, + importance = 10.0, + beginDateTime = beginDateTime, + endDateTime = endDateTime, + surveyType = RequestSurveyType.OPTICS, + status = status, + createdAt = beginDateTime.minusDays(1), + updatedAt = beginDateTime.minusDays(1), + ) + } + + // Регистрирует matcher, но возвращает non-null заглушку — иначе any() на non-null LocalDateTime + // в Kotlin отдаёт null и ломает стек матчеров Mockito. + private fun anyLocalDateTime(): LocalDateTime { + any(LocalDateTime::class.java) + return LocalDateTime.now() + } + + private class ImmediateTransactionOperations : TransactionOperations { + override fun execute(action: TransactionCallback): T { + val status: TransactionStatus = SimpleTransactionStatus() + return action.doInTransaction(status) + } + } +} diff --git a/services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/service/RequestServiceTest.kt b/services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/service/RequestServiceTest.kt index 3f02619..4b95e86 100644 --- a/services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/service/RequestServiceTest.kt +++ b/services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/service/RequestServiceTest.kt @@ -50,6 +50,7 @@ class RequestServiceTest { requestCellRepository = requestCellRepository, geometryService = GeometryService(), requestGridProjectionService = requestGridProjectionService, + requestLifecycleService = RequestLifecycleService(GeometryService()), entityManager = entityManager, ) @@ -95,6 +96,22 @@ class RequestServiceTest { verify(requestGridProjectionService).rebuildProjection(persistedRequests.single()) } + @Test + fun `create request passes through explicit coverage required percent from api`() { + val persistedRequests = mutableListOf() + doReturn(false) + .`when`(requestRepository) + .existsById(REQUEST_ID) + doAnswer { invocation -> + persistedRequests += invocation.getArgument(0) + null + }.`when`(entityManager).persist(any(RequestEntity::class.java)) + + service.createRequest(createRequestDto(coverageRequiredPercent = 80.0)) + + assertEquals(80.0, persistedRequests.single().coverageRequiredPercent) + } + @Test fun `get request by id returns name and importance`() { val request = existingRequest() @@ -279,7 +296,7 @@ class RequestServiceTest { verify(requestCellRepository).deleteByRequestId(REQUEST_ID) } - private fun createRequestDto(): CreateRequestDto { + private fun createRequestDto(coverageRequiredPercent: Double = 100.0): CreateRequestDto { return CreateRequestDto( id = REQUEST_ID, name = REQUEST_NAME, @@ -287,6 +304,7 @@ class RequestServiceTest { importance = REQUEST_IMPORTANCE, beginDateTime = LocalDateTime.of(2026, 1, 1, 0, 0, 0), endDateTime = LocalDateTime.of(2026, 1, 31, 23, 59, 59), + coverageRequiredPercent = coverageRequiredPercent, optics = OpticsParamsDto( resultType = OpticsResultTypeDto.PANCHROMATIC, resolution = 1.0, diff --git a/services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/service/RequestStatusOutboxServiceTest.kt b/services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/service/RequestStatusOutboxServiceTest.kt new file mode 100644 index 0000000..cc8f322 --- /dev/null +++ b/services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/service/RequestStatusOutboxServiceTest.kt @@ -0,0 +1,107 @@ +package org.nstart.dep265.requestservice.service + +import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule +import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertNotNull +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test +import org.mockito.Mockito.mock +import org.mockito.Mockito.mockingDetails +import org.nstart.dep265.requestservice.entity.RequestEntity +import org.nstart.dep265.requestservice.entity.RequestStatus +import org.nstart.dep265.requestservice.entity.RequestSurveyType +import org.nstart.dep265.requestservice.repository.OutboxEventRepository +import java.time.LocalDateTime +import java.util.UUID + +class RequestStatusOutboxServiceTest { + private val objectMapper = jacksonObjectMapper() + .registerModule(JavaTimeModule()) + .findAndRegisterModules() + + @Test + fun `completed terminal payload contains required fields`() { + val outboxEventRepository = mock(OutboxEventRepository::class.java) + val service = service(outboxEventRepository) + val requestId = UUID.fromString("5f57cd22-44a7-4fe8-956f-7fd0c52df846") + val completedAt = LocalDateTime.parse("2026-01-01T10:15:30") + val matchedAt = LocalDateTime.parse("2026-01-01T10:14:00") + val eventCreatedAt = LocalDateTime.parse("2026-01-01T10:16:00") + val request = RequestEntity( + id = requestId, + name = "Тест1", + geometry = "POLYGON((0 0, 1 0, 1 1, 0 1, 0 0))", + remainingGeometry = "POLYGON EMPTY", + geometryArea = 1.0, + remainingArea = 0.0, + importance = 10.0, + beginDateTime = LocalDateTime.parse("2026-01-01T10:00:00"), + endDateTime = LocalDateTime.parse("2026-01-01T11:00:00"), + surveyType = RequestSurveyType.OPTICS, + status = RequestStatus.COMPLETED, + lastMatchedAt = matchedAt, + completedAt = completedAt, + createdAt = LocalDateTime.parse("2026-01-01T09:00:00"), + updatedAt = completedAt, + ) + + service.createTerminalIfAbsent(request, eventCreatedAt) + + val payload = objectMapper.readTree(insertedPayload(outboxEventRepository)) + assertEquals(requestId.toString(), payload["requestId"].asText()) + assertEquals("COMPLETED", payload["status"].asText()) + assertEquals(completedAt, objectMapper.treeToValue(payload["occurredAt"], LocalDateTime::class.java)) + assertEquals(100.0, payload["coveragePercent"].asDouble()) + assertNotNull(payload["matchedAt"]) + assertNotNull(payload["eventCreatedAt"]) + } + + @Test + fun `expired terminal payload carries expired status and partial coverage`() { + val outboxEventRepository = mock(OutboxEventRepository::class.java) + val service = service(outboxEventRepository) + val requestId = UUID.fromString("9c2f0c4b-3a51-4e2d-9d8a-1f4b2c6e7a01") + val eventCreatedAt = LocalDateTime.parse("2026-02-01T00:00:00") + val request = RequestEntity( + id = requestId, + name = "Тест1", + geometry = "POLYGON((0 0, 1 0, 1 1, 0 1, 0 0))", + remainingGeometry = "POLYGON((0 0, 0.5 0, 0.5 1, 0 1, 0 0))", + geometryArea = 1.0, + remainingArea = 0.5, + importance = 10.0, + beginDateTime = LocalDateTime.parse("2026-01-01T00:00:00"), + endDateTime = LocalDateTime.parse("2026-01-31T23:59:59"), + surveyType = RequestSurveyType.OPTICS, + status = RequestStatus.EXPIRED, + completedAt = null, + createdAt = LocalDateTime.parse("2025-12-31T00:00:00"), + updatedAt = eventCreatedAt, + ) + + service.createTerminalIfAbsent(request, eventCreatedAt) + + val payload = objectMapper.readTree(insertedPayload(outboxEventRepository)) + assertEquals("EXPIRED", payload["status"].asText()) + // completedAt отсутствует у EXPIRED → occurredAt = момент фиксации истечения. + assertEquals(eventCreatedAt, objectMapper.treeToValue(payload["occurredAt"], LocalDateTime::class.java)) + assertEquals(50.0, payload["coveragePercent"].asDouble()) + assertTrue(payload["matchedAt"].isNull) + } + + private fun service(outboxEventRepository: OutboxEventRepository): RequestStatusOutboxService { + return RequestStatusOutboxService( + outboxEventRepository = outboxEventRepository, + objectMapper = objectMapper, + geometryService = GeometryService(), + ) + } + + private fun insertedPayload(outboxEventRepository: OutboxEventRepository): String { + val insertInvocation = mockingDetails(outboxEventRepository).invocations.single { invocation -> + invocation.method.name == "insertIfAbsent" + } + return insertInvocation.arguments[4] as String + } +} diff --git a/services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/service/RouteMatchingServiceTest.kt b/services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/service/RouteMatchingServiceTest.kt index 6af3400..d4fb832 100644 --- a/services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/service/RouteMatchingServiceTest.kt +++ b/services/pcp-request-service/src/test/kotlin/org/nstart/dep265/requestservice/service/RouteMatchingServiceTest.kt @@ -288,7 +288,7 @@ class RouteMatchingServiceTest { invocation -> invocation.method.name == "insertIfAbsent" } assertEquals(1, insertInvocations.size) - assertEquals("REQUEST_COMPLETED", insertInvocations.single().arguments[1]) + assertEquals("REQUEST_TERMINAL_STATUS", insertInvocations.single().arguments[1]) assertEquals(request.id, insertInvocations.single().arguments[3]) assertEquals(RequestStatus.COMPLETED, request.status) } @@ -305,16 +305,16 @@ class RouteMatchingServiceTest { requestValidationService = RequestValidationService(geometryService), geometryService = geometryService, requestLifecycleService = RequestLifecycleService(geometryService), - requestCompletedOutboxService = requestCompletedOutboxService(outboxEventRepository, geometryService), + requestStatusOutboxService = requestStatusOutboxService(outboxEventRepository, geometryService), routeMatchingTransactionOperations = ImmediateTransactionOperations(), ) } - private fun requestCompletedOutboxService( + private fun requestStatusOutboxService( outboxEventRepository: OutboxEventRepository, geometryService: GeometryService, - ): RequestCompletedOutboxService { - return RequestCompletedOutboxService( + ): RequestStatusOutboxService { + return RequestStatusOutboxService( outboxEventRepository = outboxEventRepository, objectMapper = jacksonObjectMapper() .registerModule(JavaTimeModule())