добавить масштабируемый outbox worker для паспортов маршрутов

- добавить properties и config values для route passport outbox worker - включить Spring scheduling в route-processing service - добавить транзакционный сервис claim и status transitions для outbox - публиковать claimed outbox rows через существующий Kafka publisher - обрабатывать publish failures по каждой row с bounded exponential backoff - обеспечить безопасность при нескольких pod через FOR UPDATE SKIP LOCKED и lockedUntil - добавить unit tests для claim, publish transitions, retry/fail behavior и batch-обработки worker
2026-05-25 15:05:04 +03:00
parent 9d6f838692
commit b5990a55ac
5 changed files with 694 additions and 0 deletions
@@ -0,0 +1,44 @@
+package space.nstart.pcp.routepassportconsumer.config
+
+import jakarta.validation.constraints.AssertTrue
+import jakarta.validation.constraints.Min
+import org.springframework.boot.context.properties.ConfigurationProperties
+import org.springframework.validation.annotation.Validated
+
+/**
+ * Operational settings for the route passport outbox worker.
+ *
+ * Values control scheduling cadence, DB claim size, lock lease time and bounded retry backoff.
+ */
+@Validated
+@ConfigurationProperties(prefix = "app.route-passport.outbox")
+data class RoutePassportOutboxProperties(
+
+    /** Enables scheduled publication of pending route passport outbox rows. */
+    val enabled: Boolean = true,
+
+    /** Maximum number of due rows claimed by one worker tick. */
+    @field:Min(1)
+    val batchSize: Int = 50,
+
+    /** Delay between scheduled worker runs. */
+    @field:Min(100)
+    val fixedDelayMs: Long = 1_000,
+
+    /** Lease duration for a claimed row; expired locks can be claimed by another instance. */
+    @field:Min(1_000)
+    val lockTimeoutMs: Long = 60_000,
+
+    /** Backoff after the first failed publication attempt. */
+    @field:Min(100)
+    val initialBackoffMs: Long = 1_000,
+
+    /** Maximum backoff between publication attempts. */
+    @field:Min(100)
+    val maxBackoffMs: Long = 60_000,
+) {
+
+    @get:AssertTrue(message = "maxBackoffMs must be greater than or equal to initialBackoffMs")
+    val isMaxBackoffGreaterThanOrEqualToInitialBackoff: Boolean
+        get() = maxBackoffMs >= initialBackoffMs
+}
@@ -0,0 +1,107 @@
+package space.nstart.pcp.routepassportconsumer.service
+
+import org.springframework.stereotype.Service
+import org.springframework.transaction.annotation.Transactional
+import space.nstart.pcp.routepassportconsumer.config.RoutePassportOutboxProperties
+import space.nstart.pcp.routepassportconsumer.entity.RoutePassportOutboxEntity
+import space.nstart.pcp.routepassportconsumer.entity.RoutePassportOutboxStatus
+import space.nstart.pcp.routepassportconsumer.repository.RoutePassportOutboxRepository
+import java.time.Duration
+import java.time.LocalDateTime
+import java.util.UUID
+
+/**
+ * Owns transactional claim and status transitions for route passport outbox rows.
+ */
+@Service
+class RoutePassportOutboxService(
+    private val routePassportOutboxRepository: RoutePassportOutboxRepository,
+    private val outboxProperties: RoutePassportOutboxProperties,
+) {
+
+    /**
+     * Claims due pending rows in one DB transaction using SELECT FOR UPDATE SKIP LOCKED.
+     */
+    @Transactional
+    fun claimDueBatch(): List<RoutePassportOutboxEntity> {
+        if (!outboxProperties.enabled) {
+            return emptyList()
+        }
+
+        val now = LocalDateTime.now()
+        val lockedUntil = now.plus(Duration.ofMillis(outboxProperties.lockTimeoutMs))
+        val claimedRows = routePassportOutboxRepository.findDueForUpdateSkipLocked(outboxProperties.batchSize)
+
+        claimedRows.forEach { outbox ->
+            outbox.lockedUntil = lockedUntil
+            outbox.updatedAt = now
+        }
+
+        return routePassportOutboxRepository.saveAll(claimedRows).toList()
+    }
+
+    /**
+     * Marks a successfully published row as terminally published.
+     */
+    @Transactional
+    fun markPublished(outboxId: UUID) {
+        val now = LocalDateTime.now()
+        val outbox = findOutbox(outboxId)
+
+        outbox.status = RoutePassportOutboxStatus.PUBLISHED
+        outbox.publishedAt = now
+        outbox.lockedUntil = null
+        outbox.lastError = null
+        outbox.updatedAt = now
+
+        routePassportOutboxRepository.save(outbox)
+    }
+
+    /**
+     * Records a publish failure and either schedules bounded retry or marks the row as failed.
+     */
+    @Transactional
+    fun markRetryOrFailed(outboxId: UUID, exception: Exception) {
+        val now = LocalDateTime.now()
+        val outbox = findOutbox(outboxId)
+
+        outbox.attempts += 1
+        outbox.lastError = compactError(exception)
+        outbox.lockedUntil = null
+        outbox.updatedAt = now
+
+        if (outbox.attempts >= outbox.maxAttempts) {
+            outbox.status = RoutePassportOutboxStatus.FAILED
+        } else {
+            outbox.status = RoutePassportOutboxStatus.PENDING
+            outbox.nextAttemptAt = now.plus(Duration.ofMillis(calculateBackoffMs(outbox.attempts)))
+        }
+
+        routePassportOutboxRepository.save(outbox)
+    }
+
+    private fun findOutbox(outboxId: UUID): RoutePassportOutboxEntity =
+        routePassportOutboxRepository.findById(outboxId)
+            .orElseThrow { IllegalStateException("Route passport outbox row not found: $outboxId") }
+
+    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 compactError(exception: Exception): String =
+        "${exception::class.java.name}: ${exception.message ?: ""}".take(MAX_ERROR_LENGTH)
+
+    private companion object {
+        const val MAX_ERROR_LENGTH = 2_000
+    }
+}
@@ -0,0 +1,100 @@
+package space.nstart.pcp.routepassportconsumer.service
+
+import org.slf4j.LoggerFactory
+import org.springframework.scheduling.annotation.Scheduled
+import org.springframework.stereotype.Component
+import space.nstart.pcp.routepassportconsumer.config.RoutePassportOutboxProperties
+import space.nstart.pcp.routepassportconsumer.entity.RoutePassportOutboxEntity
+import space.nstart.pcp.routepassportconsumer.entity.RoutePassportOutboxStatus
+
+/**
+ * Scheduled worker that publishes already persisted route passport outbox rows.
+ */
+@Component
+class RoutePassportOutboxWorker(
+    private val routePassportOutboxService: RoutePassportOutboxService,
+    private val routePassportKafkaPublisherService: RoutePassportKafkaPublisherService,
+    private val outboxProperties: RoutePassportOutboxProperties,
+) {
+    private val log = LoggerFactory.getLogger(this::class.java)
+
+    @Scheduled(fixedDelayString = "\${app.route-passport.outbox.fixed-delay-ms:1000}")
+    fun publishDueOutboxRows() {
+        if (!outboxProperties.enabled) {
+            return
+        }
+
+        val outboxRows = routePassportOutboxService.claimDueBatch()
+        outboxRows.forEach { outbox ->
+            logClaimed(outbox)
+        }
+
+        outboxRows.forEach { outbox ->
+            try {
+                publishOne(outbox)
+            } catch (exception: Exception) {
+                log.error(
+                    "Kafka outbox row processing failed: outboxId={}, routeId={}, eventType={}, topic={}, attempts={}, status={}, exceptionClass={}, exceptionMessage={}",
+                    outbox.id,
+                    outbox.routeId,
+                    outbox.eventType,
+                    outbox.topic,
+                    outbox.attempts,
+                    outbox.status,
+                    exception::class.java.name,
+                    exception.message,
+                )
+            }
+        }
+    }
+
+    private fun publishOne(outbox: RoutePassportOutboxEntity) {
+        try {
+            routePassportKafkaPublisherService.publish(outbox)
+            routePassportOutboxService.markPublished(outbox.id)
+
+            log.info(
+                "Kafka outbox row marked published: outboxId={}, routeId={}, eventType={}, topic={}, attempts={}, status={}",
+                outbox.id,
+                outbox.routeId,
+                outbox.eventType,
+                outbox.topic,
+                outbox.attempts,
+                RoutePassportOutboxStatus.PUBLISHED,
+            )
+        } catch (exception: Exception) {
+            routePassportOutboxService.markRetryOrFailed(outbox.id, exception)
+
+            val attemptsAfterFailure = outbox.attempts + 1
+            val statusAfterFailure = if (attemptsAfterFailure >= outbox.maxAttempts) {
+                RoutePassportOutboxStatus.FAILED
+            } else {
+                RoutePassportOutboxStatus.PENDING
+            }
+
+            log.warn(
+                "Kafka outbox row publish failed: outboxId={}, routeId={}, eventType={}, topic={}, attempts={}, status={}, exceptionClass={}, exceptionMessage={}",
+                outbox.id,
+                outbox.routeId,
+                outbox.eventType,
+                outbox.topic,
+                attemptsAfterFailure,
+                statusAfterFailure,
+                exception::class.java.name,
+                exception.message,
+            )
+        }
+    }
+
+    private fun logClaimed(outbox: RoutePassportOutboxEntity) {
+        log.info(
+            "Kafka outbox row claimed: outboxId={}, routeId={}, eventType={}, topic={}, attempts={}, status={}",
+            outbox.id,
+            outbox.routeId,
+            outbox.eventType,
+            outbox.topic,
+            outbox.attempts,
+            outbox.status,
+        )
+    }
+}
@@ -0,0 +1,249 @@
+package space.nstart.pcp.routepassportconsumer.service
+
+import org.junit.jupiter.api.Test
+import org.mockito.ArgumentMatchers.any
+import org.mockito.ArgumentMatchers.anyInt
+import org.mockito.Mockito.doAnswer
+import org.mockito.Mockito.doReturn
+import org.mockito.Mockito.mock
+import org.mockito.Mockito.never
+import org.mockito.Mockito.verify
+import space.nstart.pcp.pcp_types_lib.message.PcpKafkaEvent
+import space.nstart.pcp.routepassportconsumer.config.RoutePassportOutboxProperties
+import space.nstart.pcp.routepassportconsumer.entity.RoutePassportOutboxEntity
+import space.nstart.pcp.routepassportconsumer.entity.RoutePassportOutboxStatus
+import space.nstart.pcp.routepassportconsumer.repository.RoutePassportOutboxRepository
+import tools.jackson.databind.ObjectMapper
+import tools.jackson.databind.json.JsonMapper
+import tools.jackson.module.kotlin.KotlinModule
+import java.time.Duration
+import java.time.LocalDateTime
+import java.util.Optional
+import java.util.UUID
+import kotlin.test.assertEquals
+import kotlin.test.assertNotNull
+import kotlin.test.assertNull
+import kotlin.test.assertTrue
+
+class RoutePassportOutboxServiceTest {
+
+    private val objectMapper: ObjectMapper = JsonMapper.builder()
+        .addModule(KotlinModule.Builder().build())
+        .build()
+
+    @Test
+    fun `claimDueBatch returns empty when outbox is disabled`() {
+        val repository = mock(RoutePassportOutboxRepository::class.java)
+        val service = service(
+            repository = repository,
+            properties = properties(enabled = false),
+        )
+
+        val claimedRows = service.claimDueBatch()
+
+        assertTrue(claimedRows.isEmpty())
+        verify(repository, never()).findDueForUpdateSkipLocked(anyInt())
+    }
+
+    @Test
+    fun `claimDueBatch locks returned rows`() {
+        val repository = mock(RoutePassportOutboxRepository::class.java)
+        val service = service(repository)
+        val first = outbox(id = UUID.fromString("3ed5e75c-a5c0-4d63-88be-a701b5222b62"))
+        val second = outbox(id = UUID.fromString("5d75d482-e402-46d4-8e2b-4c19bff6f795"))
+        val originalUpdatedAt = first.updatedAt
+
+        doReturn(listOf(first, second))
+            .`when`(repository)
+            .findDueForUpdateSkipLocked(50)
+        stubSaveAll(repository)
+
+        val claimedRows = service.claimDueBatch()
+
+        assertEquals(listOf(first, second), claimedRows)
+        claimedRows.forEach { outbox ->
+            assertNotNull(outbox.lockedUntil)
+            assertEquals(60_000, Duration.between(outbox.updatedAt, outbox.lockedUntil).toMillis())
+            assertTrue(outbox.updatedAt != originalUpdatedAt)
+        }
+        verify(repository).saveAll(listOf(first, second))
+    }
+
+    @Test
+    fun `markPublished moves row to published state`() {
+        val repository = mock(RoutePassportOutboxRepository::class.java)
+        val service = service(repository)
+        val outbox = outbox(
+            lockedUntil = LocalDateTime.of(2026, 4, 1, 10, 20),
+            lastError = "previous failure",
+        )
+
+        stubFindById(repository, outbox)
+        stubSave(repository)
+
+        service.markPublished(outbox.id)
+
+        assertEquals(RoutePassportOutboxStatus.PUBLISHED, outbox.status)
+        assertNotNull(outbox.publishedAt)
+        assertNull(outbox.lockedUntil)
+        assertNull(outbox.lastError)
+        verify(repository).save(outbox)
+    }
+
+    @Test
+    fun `markRetryOrFailed below max attempts schedules pending retry`() {
+        val repository = mock(RoutePassportOutboxRepository::class.java)
+        val service = service(repository)
+        val outbox = outbox(
+            attempts = 0,
+            maxAttempts = 3,
+            lockedUntil = LocalDateTime.of(2026, 4, 1, 10, 20),
+        )
+        val exception = IllegalStateException("broker unavailable")
+
+        stubFindById(repository, outbox)
+        stubSave(repository)
+
+        service.markRetryOrFailed(outbox.id, exception)
+
+        assertEquals(1, outbox.attempts)
+        assertEquals(RoutePassportOutboxStatus.PENDING, outbox.status)
+        assertNull(outbox.lockedUntil)
+        assertTrue(outbox.lastError?.contains(IllegalStateException::class.java.name) == true)
+        assertTrue(outbox.lastError?.contains("broker unavailable") == true)
+        assertEquals(1_000, Duration.between(outbox.updatedAt, outbox.nextAttemptAt).toMillis())
+        verify(repository).save(outbox)
+    }
+
+    @Test
+    fun `markRetryOrFailed at max attempts moves row to failed state`() {
+        val repository = mock(RoutePassportOutboxRepository::class.java)
+        val service = service(repository)
+        val outbox = outbox(
+            attempts = 2,
+            maxAttempts = 3,
+            lockedUntil = LocalDateTime.of(2026, 4, 1, 10, 20),
+        )
+
+        stubFindById(repository, outbox)
+        stubSave(repository)
+
+        service.markRetryOrFailed(outbox.id, IllegalStateException("broker unavailable"))
+
+        assertEquals(3, outbox.attempts)
+        assertEquals(RoutePassportOutboxStatus.FAILED, outbox.status)
+        assertNull(outbox.lockedUntil)
+        assertTrue(outbox.lastError?.contains("broker unavailable") == true)
+        verify(repository).save(outbox)
+    }
+
+    @Test
+    fun `markRetryOrFailed caps exponential backoff by maxBackoffMs`() {
+        val repository = mock(RoutePassportOutboxRepository::class.java)
+        val service = service(
+            repository = repository,
+            properties = properties(
+                initialBackoffMs = 1_000,
+                maxBackoffMs = 1_500,
+            ),
+        )
+        val outbox = outbox(
+            attempts = 2,
+            maxAttempts = 10,
+            lockedUntil = LocalDateTime.of(2026, 4, 1, 10, 20),
+        )
+
+        stubFindById(repository, outbox)
+        stubSave(repository)
+
+        service.markRetryOrFailed(outbox.id, IllegalStateException("broker unavailable"))
+
+        assertEquals(3, outbox.attempts)
+        assertEquals(RoutePassportOutboxStatus.PENDING, outbox.status)
+        assertEquals(1_500, Duration.between(outbox.updatedAt, outbox.nextAttemptAt).toMillis())
+    }
+
+    private fun service(
+        repository: RoutePassportOutboxRepository,
+        properties: RoutePassportOutboxProperties = properties(),
+    ): RoutePassportOutboxService =
+        RoutePassportOutboxService(
+            routePassportOutboxRepository = repository,
+            outboxProperties = properties,
+        )
+
+    private fun properties(
+        enabled: Boolean = true,
+        initialBackoffMs: Long = 1_000,
+        maxBackoffMs: Long = 60_000,
+    ): RoutePassportOutboxProperties =
+        RoutePassportOutboxProperties(
+            enabled = enabled,
+            batchSize = 50,
+            fixedDelayMs = 1_000,
+            lockTimeoutMs = 60_000,
+            initialBackoffMs = initialBackoffMs,
+            maxBackoffMs = maxBackoffMs,
+        )
+
+    private fun stubFindById(
+        repository: RoutePassportOutboxRepository,
+        outbox: RoutePassportOutboxEntity,
+    ) {
+        doReturn(Optional.of(outbox))
+            .`when`(repository)
+            .findById(outbox.id)
+    }
+
+    private fun stubSave(repository: RoutePassportOutboxRepository) {
+        doAnswer { invocation ->
+            @Suppress("UNCHECKED_CAST")
+            invocation.arguments[0] as RoutePassportOutboxEntity
+        }
+            .`when`(repository)
+            .save(anyOutboxEntity())
+    }
+
+    private fun stubSaveAll(repository: RoutePassportOutboxRepository) {
+        doAnswer { invocation ->
+            @Suppress("UNCHECKED_CAST")
+            invocation.arguments[0] as Iterable<RoutePassportOutboxEntity>
+        }.`when`(repository).saveAll(anyOutboxIterable())
+    }
+
+    private fun outbox(
+        id: UUID = UUID.fromString("3ed5e75c-a5c0-4d63-88be-a701b5222b62"),
+        attempts: Int = 0,
+        maxAttempts: Int = 10,
+        lockedUntil: LocalDateTime? = null,
+        lastError: String? = null,
+    ): RoutePassportOutboxEntity {
+        val routeId = UUID.fromString("51b07d12-1823-4bd5-b55d-de02a5304c03")
+        val now = LocalDateTime.of(2026, 4, 1, 10, 15, 0)
+
+        return RoutePassportOutboxEntity(
+            id = id,
+            routeId = routeId,
+            eventType = PcpKafkaEvent.ModeStatusChangedEvent.name,
+            topic = "pcp.route.in.v1",
+            messageKey = routeId.toString(),
+            payload = objectMapper.readTree("""{"type":"ModeStatusChangedEvent","data":{"routeNameFull":"ROUTE-001"}}"""),
+            headers = null,
+            status = RoutePassportOutboxStatus.PENDING,
+            attempts = attempts,
+            maxAttempts = maxAttempts,
+            nextAttemptAt = now,
+            lockedUntil = lockedUntil,
+            lastError = lastError,
+            createdAt = now,
+            updatedAt = now,
+        )
+    }
+
+    private fun anyOutboxEntity(): RoutePassportOutboxEntity =
+        any(RoutePassportOutboxEntity::class.java) ?: outbox()
+
+    @Suppress("UNCHECKED_CAST")
+    private fun anyOutboxIterable(): Iterable<RoutePassportOutboxEntity> =
+        (any(Iterable::class.java) as Iterable<RoutePassportOutboxEntity>?) ?: emptyList()
+}
@@ -0,0 +1,194 @@
+package space.nstart.pcp.routepassportconsumer.service
+
+import org.junit.jupiter.api.Test
+import org.mockito.Mockito.doReturn
+import org.mockito.Mockito.doThrow
+import org.mockito.Mockito.mock
+import org.mockito.Mockito.never
+import org.mockito.Mockito.verify
+import space.nstart.pcp.pcp_types_lib.message.PcpKafkaEvent
+import space.nstart.pcp.routepassportconsumer.config.RoutePassportOutboxProperties
+import space.nstart.pcp.routepassportconsumer.entity.RoutePassportOutboxEntity
+import space.nstart.pcp.routepassportconsumer.entity.RoutePassportOutboxStatus
+import tools.jackson.databind.ObjectMapper
+import tools.jackson.databind.json.JsonMapper
+import tools.jackson.module.kotlin.KotlinModule
+import java.time.LocalDateTime
+import java.util.UUID
+
+class RoutePassportOutboxWorkerTest {
+
+    private val objectMapper: ObjectMapper = JsonMapper.builder()
+        .addModule(KotlinModule.Builder().build())
+        .build()
+
+    @Test
+    fun `when disabled no claim or publish happens`() {
+        val outboxService = mock(RoutePassportOutboxService::class.java)
+        val publisherService = mock(RoutePassportKafkaPublisherService::class.java)
+        val worker = worker(
+            outboxService = outboxService,
+            publisherService = publisherService,
+            properties = properties(enabled = false),
+        )
+
+        worker.publishDueOutboxRows()
+
+        verify(outboxService, never()).claimDueBatch()
+        verify(publisherService, never()).publish(anyOutboxEntity())
+    }
+
+    @Test
+    fun `when claim returns rows publisher is called for each row`() {
+        val outboxService = mock(RoutePassportOutboxService::class.java)
+        val publisherService = mock(RoutePassportKafkaPublisherService::class.java)
+        val first = outbox(id = UUID.fromString("3ed5e75c-a5c0-4d63-88be-a701b5222b62"))
+        val second = outbox(id = UUID.fromString("5d75d482-e402-46d4-8e2b-4c19bff6f795"))
+        val worker = worker(outboxService, publisherService)
+
+        doReturn(listOf(first, second))
+            .`when`(outboxService)
+            .claimDueBatch()
+
+        worker.publishDueOutboxRows()
+
+        verify(publisherService).publish(first)
+        verify(publisherService).publish(second)
+    }
+
+    @Test
+    fun `successful publish marks row as published`() {
+        val outboxService = mock(RoutePassportOutboxService::class.java)
+        val publisherService = mock(RoutePassportKafkaPublisherService::class.java)
+        val outbox = outbox()
+        val worker = worker(outboxService, publisherService)
+
+        doReturn(listOf(outbox))
+            .`when`(outboxService)
+            .claimDueBatch()
+
+        worker.publishDueOutboxRows()
+
+        verify(publisherService).publish(outbox)
+        verify(outboxService).markPublished(outbox.id)
+    }
+
+    @Test
+    fun `publish failure marks row retry or failed`() {
+        val outboxService = mock(RoutePassportOutboxService::class.java)
+        val publisherService = mock(RoutePassportKafkaPublisherService::class.java)
+        val outbox = outbox()
+        val failure = IllegalStateException("broker unavailable")
+        val worker = worker(outboxService, publisherService)
+
+        doReturn(listOf(outbox))
+            .`when`(outboxService)
+            .claimDueBatch()
+        doThrow(failure)
+            .`when`(publisherService)
+            .publish(outbox)
+
+        worker.publishDueOutboxRows()
+
+        verify(outboxService).markRetryOrFailed(outbox.id, failure)
+        verify(outboxService, never()).markPublished(outbox.id)
+    }
+
+    @Test
+    fun `failure of one row does not prevent processing next row`() {
+        val outboxService = mock(RoutePassportOutboxService::class.java)
+        val publisherService = mock(RoutePassportKafkaPublisherService::class.java)
+        val first = outbox(id = UUID.fromString("3ed5e75c-a5c0-4d63-88be-a701b5222b62"))
+        val second = outbox(id = UUID.fromString("5d75d482-e402-46d4-8e2b-4c19bff6f795"))
+        val failure = IllegalStateException("broker unavailable")
+        val worker = worker(outboxService, publisherService)
+
+        doReturn(listOf(first, second))
+            .`when`(outboxService)
+            .claimDueBatch()
+        doThrow(failure)
+            .`when`(publisherService)
+            .publish(first)
+
+        worker.publishDueOutboxRows()
+
+        verify(outboxService).markRetryOrFailed(first.id, failure)
+        verify(publisherService).publish(second)
+        verify(outboxService).markPublished(second.id)
+    }
+
+    @Test
+    fun `row status transition failure does not prevent processing next row`() {
+        val outboxService = mock(RoutePassportOutboxService::class.java)
+        val publisherService = mock(RoutePassportKafkaPublisherService::class.java)
+        val first = outbox(id = UUID.fromString("3ed5e75c-a5c0-4d63-88be-a701b5222b62"))
+        val second = outbox(id = UUID.fromString("5d75d482-e402-46d4-8e2b-4c19bff6f795"))
+        val publishFailure = IllegalStateException("broker unavailable")
+        val transitionFailure = IllegalStateException("database unavailable")
+        val worker = worker(outboxService, publisherService)
+
+        doReturn(listOf(first, second))
+            .`when`(outboxService)
+            .claimDueBatch()
+        doThrow(publishFailure)
+            .`when`(publisherService)
+            .publish(first)
+        doThrow(transitionFailure)
+            .`when`(outboxService)
+            .markRetryOrFailed(first.id, publishFailure)
+
+        worker.publishDueOutboxRows()
+
+        verify(publisherService).publish(second)
+        verify(outboxService).markPublished(second.id)
+    }
+
+    private fun worker(
+        outboxService: RoutePassportOutboxService,
+        publisherService: RoutePassportKafkaPublisherService,
+        properties: RoutePassportOutboxProperties = properties(),
+    ): RoutePassportOutboxWorker =
+        RoutePassportOutboxWorker(
+            routePassportOutboxService = outboxService,
+            routePassportKafkaPublisherService = publisherService,
+            outboxProperties = properties,
+        )
+
+    private fun properties(enabled: Boolean = true): RoutePassportOutboxProperties =
+        RoutePassportOutboxProperties(
+            enabled = enabled,
+            batchSize = 50,
+            fixedDelayMs = 1_000,
+            lockTimeoutMs = 60_000,
+            initialBackoffMs = 1_000,
+            maxBackoffMs = 60_000,
+        )
+
+    private fun outbox(
+        id: UUID = UUID.fromString("3ed5e75c-a5c0-4d63-88be-a701b5222b62"),
+    ): RoutePassportOutboxEntity {
+        val routeId = UUID.fromString("51b07d12-1823-4bd5-b55d-de02a5304c03")
+        val now = LocalDateTime.of(2026, 4, 1, 10, 15, 0)
+
+        return RoutePassportOutboxEntity(
+            id = id,
+            routeId = routeId,
+            eventType = PcpKafkaEvent.ModeStatusChangedEvent.name,
+            topic = "pcp.route.in.v1",
+            messageKey = routeId.toString(),
+            payload = objectMapper.readTree("""{"type":"ModeStatusChangedEvent","data":{"routeNameFull":"ROUTE-001"}}"""),
+            headers = null,
+            status = RoutePassportOutboxStatus.PENDING,
+            attempts = 0,
+            maxAttempts = 10,
+            nextAttemptAt = now,
+            lockedUntil = now.plusMinutes(1),
+            lastError = null,
+            createdAt = now,
+            updatedAt = now,
+        )
+    }
+
+    private fun anyOutboxEntity(): RoutePassportOutboxEntity =
+        org.mockito.ArgumentMatchers.any(RoutePassportOutboxEntity::class.java) ?: outbox()
+}