pcp-request-service: аудит-фиксы RS1/RS3/RS6/RS7

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) — до роста требований к точности.
2026-06-06 23:47:19 +03:00
parent 2499aebf2d
commit c1259c3105
33 changed files with 1600 additions and 139 deletions
@@ -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<String, String> =
+        DefaultKafkaConsumerFactory(
+            mutableMapOf<String, Any>(
+                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<String, String> =
+        ConcurrentKafkaListenerContainerFactory<String, String>().apply {
+            setConsumerFactory(routeConsumerFactory())
+            setCommonErrorHandler(routeKafkaErrorHandler)
+        }
+
+    @Bean
+    fun routeDeadLetterPublishingRecoverer(
+        kafkaTemplate: KafkaTemplate<String, String>,
+        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"
+    }
+}
@@ -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,
 )
@@ -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
+        }
+    }
 }
@@ -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,
+)
@@ -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,
+)
@@ -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,
@@ -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,

@@ -16,8 +16,9 @@ interface OutboxEventRepository : JpaRepository<OutboxEventEntity, UUID> {
    fun findByStatusOrderByCreatedAtAsc(status: OutboxEventStatus, pageable: Pageable): List<OutboxEventEntity>

    /**
-     * 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<OutboxEventEntity, UUID> {
            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<OutboxEventEntity, UUID> {
        """,
        nativeQuery = true,
    )
-    fun findRequestCompletedPublishableForUpdateSkipLocked(
+    fun findPublishableForUpdateSkipLocked(
        @Param("eventType") eventType: String,
+        @Param("now") now: LocalDateTime,
+        @Param("maxAttempts") maxAttempts: Int,
        @Param("batchSize") batchSize: Int,
    ): List<OutboxEventEntity>

@@ -52,6 +57,8 @@ interface OutboxEventRepository : JpaRepository<OutboxEventEntity, UUID> {
                aggregate_id,
                payload,
                status,
+                attempts,
+                next_attempt_at,
                created_at,
                published_at,
                error_message
@@ -63,6 +70,8 @@ interface OutboxEventRepository : JpaRepository<OutboxEventEntity, UUID> {
                :aggregateId,
                CAST(:payload AS jsonb),
                :status,
+                0,
+                :createdAt,
                :createdAt,
                NULL,
                NULL
@@ -106,4 +106,27 @@ interface RequestRepository : JpaRepository<RequestEntity, UUID> {
    ): List<RequestEntity>

    fun findByStatusIn(statuses: Collection<RequestStatus>): List<RequestEntity>
+
+    /**
+     * Захватывает порцию просроченных заявок (окно уже закончилось, статус ещё незавершённый)
+     * под 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<RequestEntity>
 }
@@ -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!!
@@ -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)
+        }
+    }
+}
@@ -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"
@@ -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"
    }
 }
@@ -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
    }
@@ -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
@@ -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);
@@ -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
@@ -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<ConstraintViolation<*>>()))
+
+            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<ConstraintViolation<*>>()),
+                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<String, String> =
+        ConsumerRecord(ROUTE_TOPIC, 3, 42L, "route-key", """{"routeId":"route-1"}""")
+
+    private fun capturePublishedRecord(
+        kafkaTemplate: KafkaTemplate<String, String>,
+    ): ProducerRecord<String, String> {
+        val captor = producerRecordCaptor()
+        verify(kafkaTemplate, times(1)).send(captor.capture())
+        return captor.value
+    }
+
+    @Suppress("UNCHECKED_CAST")
+    private fun producerRecordCaptor(): ArgumentCaptor<ProducerRecord<String, String>> =
+        ArgumentCaptor.forClass(ProducerRecord::class.java) as ArgumentCaptor<ProducerRecord<String, String>>
+
+    @Suppress("UNCHECKED_CAST")
+    private fun mockConsumer(): Consumer<String, String> =
+        mock(Consumer::class.java) as Consumer<String, String>
+
+    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<String, String> {
+            val kafkaTemplate = mock(KafkaTemplate::class.java) as KafkaTemplate<String, String>
+            val sendResult = mock(SendResult::class.java) as SendResult<String, String>
+            `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<String, String> =
+            (any(ProducerRecord::class.java) as ProducerRecord<String, String>?)
+                ?: ProducerRecord("stub-topic", "stub-key", "stub-payload")
+
+        @Suppress("UNCHECKED_CAST")
+        fun org.springframework.context.ApplicationContext.kafkaTemplate(): KafkaTemplate<String, String> =
+            getBean(KafkaTemplate::class.java) as KafkaTemplate<String, String>
+    }
+}
@@ -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": {
@@ -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<CreateRequestDto>(
+            """
+            {
+              "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
@@ -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,
        )
    }
 }
@@ -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,
+        )
+    }
+}
@@ -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 <T : Any?> execute(action: TransactionCallback<T>): T {
            return action.doInTransaction(SimpleTransactionStatus())
@@ -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"])
-    }
-}
@@ -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<RequestEntity>())
+            .`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<RequestEntity>())
+            .`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 <T> execute(action: TransactionCallback<T>): T {
+            val status: TransactionStatus = SimpleTransactionStatus()
+            return action.doInTransaction(status)
+        }
+    }
+}
@@ -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<RequestEntity>()
+        doReturn(false)
+            .`when`(requestRepository)
+            .existsById(REQUEST_ID)
+        doAnswer { invocation ->
+            persistedRequests += invocation.getArgument<RequestEntity>(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,
@@ -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
+    }
+}
@@ -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())