fix(core): address review of retry backoff (#131)

- Count a batch retry's backoff from when delivery ended, as process() already did, so slow
  failures (timeouts) do not eat into it.
- Postgres 004: backfill in its own changeset so ADD COLUMN holds ACCESS EXCLUSIVE only briefly.
- MySQL 004: one changeset per DDL step with preconditions, so a migration that stopped half-way
  (MySQL commits each DDL) resumes; backfill FAILED rows too so a manual requeue is claimable.
- RetryPolicy: compute in nanoseconds, cap maxBackoff at 365 days, require a finite multiplier;
  same checks on the Spring properties. OutboxEntry constructor is @jvmoverloads for Java callers.
- Docs: all migrations up to 004 before updated workers, clock-skew note, manual requeue recipe.
- Tests: migration tests claim through the route-aware claim (backfill, 1.x insert default, FAILED
  requeue), batch backoff timing, sub-ms backoff, Spring property binding and validation.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

Author: rawo

CHANGELOG.md

@@ -115,7 +115,8 @@ Until `1.0.0`, breaking changes may appear in any release and are flagged with *
   `backoffMultiplier`, `maxBackoff` and `jitter` (all defaulted, `@JvmOverloads`) and `backoff(retry)`;
   `OutboxEntry` gained `nextAttemptAt` (defaults to `createdAt`) and `retry` an optional
   `nextAttemptAt`; `RouteAwareOutboxStore.claimPending` takes the current time (`now`); and
-  `OutboxProcessorProperties` gained four parameters. As with every data-class change, Kotlin code
+  `OutboxProcessorProperties` gained four parameters. `OutboxEntry`'s constructor is `@JvmOverloads`,
+  so Java callers can still omit `nextAttemptAt`. As with every data-class change, Kotlin code
   calling `copy(...)` or default-argument constructors of these classes must be recompiled; custom
   `RouteAwareOutboxStore`s must filter on and write `next_attempt_at`. The legacy
   `OutboxStore.claimPending(limit)` ignores the backoff. Tests or callers that relied on immediate

README.md

@@ -222,7 +222,7 @@ If you use another migration tool, apply every SQL migration for your database,
 - **PostgreSQL:** [001](okapi-postgres/src/main/resources/com/softwaremill/okapi/db/postgres/001__create_okapi_outbox_table.sql) table and initial indexes, [002](okapi-postgres/src/main/resources/com/softwaremill/okapi/db/postgres/002__route_aware_claim_index.sql) route-aware claim index, [003](okapi-postgres/src/main/resources/com/softwaremill/okapi/db/postgres/003__timestamptz_columns.sql) `TIMESTAMPTZ` timestamp columns, [004](okapi-postgres/src/main/resources/com/softwaremill/okapi/db/postgres/004__retry_backoff.sql) retry backoff column and claim index (its second changeset must run outside a transaction)
 - **MySQL:** [001](okapi-mysql/src/main/resources/com/softwaremill/okapi/db/mysql/001__create_okapi_outbox_table.sql) table and initial indexes, [002](okapi-mysql/src/main/resources/com/softwaremill/okapi/db/mysql/002__route_aware_claim_index.sql) route-aware claim index, [003](okapi-mysql/src/main/resources/com/softwaremill/okapi/db/mysql/003__datetime_columns.sql) `DATETIME(6)` timestamp columns, [004](okapi-mysql/src/main/resources/com/softwaremill/okapi/db/mysql/004__retry_backoff.sql) retry backoff column and claim index
 
-For an existing installation, apply `002` before starting updated workers. MySQL claims explicitly select the new index and fail if it is missing. PostgreSQL's `002` uses `CREATE INDEX CONCURRENTLY`, so run it outside a transaction when using another migration tool. The bundled Liquibase changelog handles this automatically. `CREATE INDEX CONCURRENTLY` waits for every transaction in the database that started before it, not only those on `okapi_outbox`, so long-running transactions (including old workers delivering a claimed batch) delay it, and with Liquibase they delay application startup while the changelog lock is held. Run it when no long transactions are open, or apply it ahead of the rollout.
+For an existing installation, apply every new migration (up to `004`) before starting updated workers. MySQL claims explicitly select the claim index from `004` and fail if it is missing. PostgreSQL's `002` and the index changeset of `004` use `CREATE INDEX CONCURRENTLY`, so run them outside a transaction when using another migration tool. The bundled Liquibase changelog handles this automatically. `CREATE INDEX CONCURRENTLY` waits for every transaction in the database that started before it, not only those on `okapi_outbox`, so long-running transactions (including old workers delivering a claimed batch) delay it, and with Liquibase they delay application startup while the changelog lock is held. Run it when no long transactions are open, or apply it ahead of the rollout.
 
 PostgreSQL's `003` converts the timestamp columns from `TIMESTAMP` (the okapi JVM's local wall-clock time) to `TIMESTAMPTZ`. It reads existing values in the session `TimeZone`; pgjdbc sets that to the JVM default zone, so the bundled Liquibase changelog converts them correctly when it runs inside the application. When applying it by hand, run it inside a transaction (`psql -1`) after `SET TIME ZONE '<zone your okapi JVMs run in>'` in the same session. It takes an `ACCESS EXCLUSIVE` lock: it first waits (up to 30 s, then fails and retries on the next start) for open transactions on `okapi_outbox`, queuing every other statement on the table meanwhile, and then, unless the session zone is UTC (PostgreSQL 12+), rewrites the table while reads and writes wait. 1.x instances keep working after it, apart from one failed poll per pooled connection. See UPGRADING.md for when the in-app conversion uses the wrong zone. See [UPGRADING.md](UPGRADING.md#from-1x-to-200).
 

UPGRADING.md

@@ -79,19 +79,30 @@ rows and replaces 002's claim index with one on `(delivery_type, next_attempt_at
 
 - **PostgreSQL:** the column is `NOT NULL DEFAULT CURRENT_TIMESTAMP` (added without a table rewrite
   on PostgreSQL 11+), so rows that 1.x instances insert during a rolling deploy are claimable at once.
-  The index is built in a second changeset with `CREATE INDEX CONCURRENTLY`, which must run outside a
+  The `ADD COLUMN` holds an `ACCESS EXCLUSIVE` lock only for itself (waiting at most 30 s for it); the
+  backfill of `PENDING` rows runs in a separate changeset with row locks only. The index is built in a second changeset with `CREATE INDEX CONCURRENTLY`, which must run outside a
   transaction when you apply the script by hand, and which waits for every transaction that started
   before it.
-- **MySQL:** the column is nullable (an instant change on 8.0.12+) and only `PENDING` rows are
-  backfilled. 1.x instances are already stopped for 003, so nothing writes rows without it.
+- **MySQL:** the column is nullable (an instant change on 8.0.12+); `PENDING` and `FAILED` rows are
+  backfilled, `DELIVERED` rows stay `NULL`. 1.x instances are already stopped for 003, so nothing
+  writes rows without it. Each step is a changeset of its own with a precondition, so a migration that
+  stopped half-way (e.g. on its 30 s `lock_wait_timeout`) resumes where it stopped.
 - **Behaviour:** retriable failures now wait 10s, 20s, 40s, ... (capped at 10m, 20% jitter) before
   the next attempt, so with `max-retries` 5 an entry becomes `FAILED` after about five minutes rather
   than six seconds. Set `okapi.processor.initial-backoff`, `backoff-multiplier`, `max-backoff`,
   `backoff-jitter` or `max-retries` to change that. 1.x workers still running during a rolling deploy
   ignore the backoff for the rows they claim.
+- **Clocks:** a new entry is due at its `createdAt`, taken from the publishing instance's clock, and
+  claimed once the processing instance's clock reaches it, so a publisher whose clock runs ahead of
+  the processors' delays its messages by the difference. Keep instance clocks NTP-synchronised.
+- **Requeuing a `FAILED` row by hand** now needs a due time as well:
+  `UPDATE okapi_outbox SET status = 'PENDING', retries = 0, next_attempt_at = <now in UTC> WHERE ...`
+  (`CURRENT_TIMESTAMP` on PostgreSQL, `UTC_TIMESTAMP(6)` on MySQL). A row whose `next_attempt_at`
+  lies in the future is not claimed until then.
 - **Code:** a custom `RouteAwareOutboxStore` must implement `claimPending(deliveryTypes, limit, now)`,
   return only rows with `next_attempt_at <= now`, and write `nextAttemptAt` in `persist` and the update
-  methods. Tests that drive retries tick by tick need a zero backoff:
+  methods. `OutboxEntry` keeps a Java constructor without `nextAttemptAt` (`@JvmOverloads`). Tests
+  that drive retries tick by tick need a zero backoff:
   `RetryPolicy(maxRetries, initialBackoff = Duration.ZERO, maxBackoff = Duration.ZERO)`.
 
 ## From 0.2.x to 1.0.0

okapi-core/src/main/kotlin/com/softwaremill/okapi/core/OutboxEntry.kt

@@ -8,7 +8,7 @@ import java.time.Instant
  * Created via [createPending] and progressed through [retry], [toDelivered],
  * or [toFailed] — each returning a new immutable copy.
  */
-data class OutboxEntry(
+data class OutboxEntry @JvmOverloads constructor(
     @get:JvmName("getOutboxId")
     val outboxId: OutboxId,
     val messageType: String,

okapi-core/src/main/kotlin/com/softwaremill/okapi/core/OutboxEntryProcessor.kt

@@ -31,8 +31,11 @@ class OutboxEntryProcessor(
         check(entries.all { it.deliveryType in supportedDeliveryTypes }) {
             "Batch contains an entry with no registered deliverer"
         }
+        val outcomes = deliverer.deliverBatch(entries)
+        // Read after delivery, like process(): a retry's backoff counts from when the attempt ended, so
+        // a slow failure (e.g. a 30 s HTTP timeout) does not eat into it (issue #131).
         val now = clock.instant()
-        return deliverer.deliverBatch(entries).map { (entry, result) -> applyResult(entry, result, now) }
+        return outcomes.map { (entry, result) -> applyResult(entry, result, now) }
     }
 
     private fun applyResult(entry: OutboxEntry, result: DeliveryResult, now: java.time.Instant): OutboxEntry = when (result) {

okapi-core/src/main/kotlin/com/softwaremill/okapi/core/RetryPolicy.kt

@@ -23,8 +23,11 @@ data class RetryPolicy @JvmOverloads constructor(
     init {
         require(maxRetries >= 0) { "maxRetries must be >= 0, got: $maxRetries" }
         require(!initialBackoff.isNegative) { "initialBackoff must be >= 0, got: $initialBackoff" }
-        require(backoffMultiplier >= 1.0) { "backoffMultiplier must be >= 1, got: $backoffMultiplier" }
+        require(
+            backoffMultiplier >= 1.0 && backoffMultiplier.isFinite(),
+        ) { "backoffMultiplier must be finite and >= 1, got: $backoffMultiplier" }
         require(maxBackoff >= initialBackoff) { "maxBackoff ($maxBackoff) must be >= initialBackoff ($initialBackoff)" }
+        require(maxBackoff <= MAX_ALLOWED_BACKOFF) { "maxBackoff must be at most $MAX_ALLOWED_BACKOFF, got: $maxBackoff" }
         require(jitter in 0.0..1.0) { "jitter must be between 0 and 1, got: $jitter" }
     }
 
@@ -35,17 +38,21 @@ data class RetryPolicy @JvmOverloads constructor(
     @JvmOverloads
     fun backoff(retry: Int, random: RandomGenerator = ThreadLocalRandom.current()): Duration {
         require(retry >= 1) { "retry must be >= 1, got: $retry" }
-        val maxMillis = maxBackoff.toMillis().toDouble()
-        val exponential = initialBackoff.toMillis() * Math.pow(backoffMultiplier, (retry - 1).toDouble())
-        val capped = minOf(exponential, maxMillis)
+        // In nanoseconds as doubles: no sub-millisecond truncation, and pow() overflowing to infinity
+        // is simply capped. MAX_ALLOWED_BACKOFF keeps the result inside Duration.ofNanos's range.
+        val exponential = initialBackoff.toNanos() * Math.pow(backoffMultiplier, (retry - 1).toDouble())
+        val capped = minOf(exponential, maxBackoff.toNanos().toDouble())
         val jittered = capped * (1.0 - jitter * random.nextDouble())
-        return Duration.ofMillis(jittered.toLong())
+        return Duration.ofNanos(jittered.toLong())
     }
 
     companion object {
         val DEFAULT_INITIAL_BACKOFF: Duration = Duration.ofSeconds(10)
         const val DEFAULT_BACKOFF_MULTIPLIER: Double = 2.0
         val DEFAULT_MAX_BACKOFF: Duration = Duration.ofMinutes(10)
         const val DEFAULT_JITTER: Double = 0.2
+
+        /** Upper bound on [maxBackoff]. */
+        val MAX_ALLOWED_BACKOFF: Duration = Duration.ofDays(365)
     }
 }

okapi-core/src/test/kotlin/com/softwaremill/okapi/core/RetryPolicyTest.kt

@@ -63,6 +63,13 @@ class RetryPolicyTest : FunSpec({
         shouldThrow<IllegalArgumentException> { RetryPolicy(1, initialBackoff = Duration.ofMinutes(2), maxBackoff = Duration.ofMinutes(1)) }
         shouldThrow<IllegalArgumentException> { RetryPolicy(1, jitter = 1.5) }
         shouldThrow<IllegalArgumentException> { noJitter.backoff(0) }
+        shouldThrow<IllegalArgumentException> { RetryPolicy(1, maxBackoff = Duration.ofDays(366)) }
+        shouldThrow<IllegalArgumentException> { RetryPolicy(1, backoffMultiplier = Double.POSITIVE_INFINITY) }
+    }
+
+    test("sub-millisecond backoffs are kept") {
+        RetryPolicy(1, initialBackoff = Duration.ofNanos(500_000), maxBackoff = Duration.ofMillis(1), jitter = 0.0).backoff(1) shouldBe
+            Duration.ofNanos(500_000)
     }
 
     test("a retriable failure schedules the next attempt after the backoff") {
@@ -83,6 +90,31 @@ class RetryPolicyTest : FunSpec({
         processor.process(first).nextAttemptAt shouldBe now.plusSeconds(20)
     }
 
+    test("in a batch the backoff counts from when the slow attempt ended, not when it started") {
+        var now = Instant.parse("2024-01-01T00:00:00Z")
+        val clock = object : Clock() {
+            override fun getZone() = ZoneOffset.UTC
+            override fun withZone(zone: java.time.ZoneId) = this
+            override fun instant() = now
+        }
+        val timingOut = object : MessageDeliverer {
+            override val type = "stub"
+            override fun deliver(entry: OutboxEntry): DeliveryResult {
+                now = now.plusSeconds(30) // e.g. an HTTP request timeout
+                return DeliveryResult.RetriableFailure("timeout")
+            }
+        }
+        val info = object : DeliveryInfo {
+            override val type = "stub"
+            override fun serialize(): String = "{}"
+        }
+        val entry = OutboxEntry.createPending(OutboxMessage("e", "{}"), info, Instant.EPOCH)
+
+        val retried = OutboxEntryProcessor(timingOut, noJitter, clock).processBatch(listOf(entry)).single()
+
+        retried.nextAttemptAt shouldBe Instant.parse("2024-01-01T00:00:40Z")
+    }
+
     test("a new entry is due when it is created") {
         val info = object : DeliveryInfo {
             override val type = "stub"

okapi-integration-tests/src/test/kotlin/com/softwaremill/okapi/test/store/MysqlDatetimeMigrationTest.kt

@@ -61,7 +61,9 @@ class MysqlDatetimeMigrationTest : FunSpec({
                     """
                     INSERT INTO okapi_outbox (id, message_type, payload, delivery_type, status, created_at, updated_at, retries, last_attempt, delivery_metadata)
                     VALUES ('00000000-0000-0000-0000-000000000001', 'm', '{}', 'http', 'PENDING',
-                            '2024-06-01 17:00:00.250000', '2024-06-01 17:00:01', 1, '2024-06-01 17:00:02', '{}')
+                            '2024-06-01 17:00:00.250000', '2024-06-01 17:00:01', 1, '2024-06-01 17:00:02', '{}'),
+                           ('00000000-0000-0000-0000-000000000002', 'm', '{}', 'http', 'FAILED',
+                            '2024-06-01 16:00:00', '2024-06-01 16:00:01', 5, '2024-06-01 16:00:02', '{}')
                     """.trimIndent(),
                 )
             }
@@ -113,10 +115,28 @@ class MysqlDatetimeMigrationTest : FunSpec({
                 setPassword(container.password)
             },
         )
-        val entry = jdbc.withTransaction { MysqlOutboxStore(jdbc).claimPending(10) }.single()
+        val store = MysqlOutboxStore(jdbc)
+        // The route-aware claim, which only returns rows whose next_attempt_at (backfilled by 004) is due.
+        val entry = jdbc.withTransaction { store.claimPending(setOf("http"), 10, Instant.now()) }.single()
         entry.createdAt shouldBe Instant.parse("2024-06-01T12:00:00.25Z")
         entry.updatedAt shouldBe Instant.parse("2024-06-01T12:00:01Z")
         entry.lastAttempt shouldBe Instant.parse("2024-06-01T12:00:02Z")
+        entry.nextAttemptAt shouldBe entry.createdAt
         entry.status shouldBe OutboxStatus.PENDING
+
+        // 004 also backfilled the FAILED row, so requeuing it by hand makes it claimable.
+        jdbc.withTransaction {
+            jdbc.withConnection { conn ->
+                conn.createStatement().use {
+                    it.executeUpdate(
+                        "UPDATE okapi_outbox SET status = 'PENDING', retries = 0 WHERE id = '00000000-0000-0000-0000-000000000002'",
+                    )
+                }
+            }
+        }
+        jdbc.withTransaction { store.claimPending(setOf("http"), 10, Instant.now()) }.map { it.outboxId.raw.toString() } shouldBe listOf(
+            "00000000-0000-0000-0000-000000000002",
+            "00000000-0000-0000-0000-000000000001",
+        )
     }
 })

okapi-integration-tests/src/test/kotlin/com/softwaremill/okapi/test/store/PostgresTimestamptzMigrationTest.kt

@@ -119,7 +119,9 @@ class PostgresTimestamptzMigrationTest : FunSpec({
             }
         }
         val (jdbc, store) = store(url)
-        val entry = jdbc.withTransaction { store.claimPending(10) }.single()
+        // The route-aware claim, which only returns rows whose next_attempt_at (backfilled by 004) is due.
+        val entry = jdbc.withTransaction { store.claimPending(setOf("http"), 10, Instant.now()) }.single()
+        entry.nextAttemptAt shouldBe createdAt
         entry.createdAt shouldBe createdAt
         entry.updatedAt shouldBe createdAt
         entry.lastAttempt shouldBe lastAttempt
@@ -158,6 +160,8 @@ class PostgresTimestamptzMigrationTest : FunSpec({
         }
 
         val (jdbc, store) = store(url)
-        jdbc.withTransaction { store.claimPending(10) }.map { it.createdAt } shouldBe listOf(before, after)
+        // `after` was inserted by 1.x after 004, so only the column DEFAULT made it claimable.
+        jdbc.withTransaction { store.claimPending(setOf("http"), 10, Instant.now().plusSeconds(5)) }
+            .map { it.createdAt }.sorted() shouldBe listOf(before, after)
     }
 })