fix(core): address review of after-commit listener events (#137)

- TransactionRunner documents its contract: runInTransaction has committed when it returns, and may
  run the block more than once (Exposed retries).
- publishEvents reads the batch end time before the callbacks and is one-shot per batch;
  ProcessedBatch exposes a read-only entry list and a toString.
- Fix stale processNext references.
- Tests: the end-to-end test runs with SpringTransactionRunner and ExposedTransactionRunner as well
  as plain JDBC and fails fast if the row is missing; unit tests for a retrying runner, a failing
  processBatch and double publishing.
- Docs: counters count committed status changes; batch timer wording; how a listener that relied on
  the processing transaction now fails.

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

Author: rawo

CHANGELOG.md

@@ -113,7 +113,10 @@ Until `1.0.0`, breaking changes may appear in any release and are flagged with *
   listener runs after the commit, outside the transaction, and not at all for a batch whose commit
   failed. A listener that relied on running inside the processing transaction (e.g. to write to the
   database atomically with the status update) no longer can. `onBatchProcessed`'s duration, and with
-  it `okapi.batch.duration`, now includes the commit; empty polls are still reported.
+  it `okapi.batch.duration`, now includes the commit (and excludes the listener callbacks); empty
+  polls are still reported. Counters now count committed status changes rather than delivery
+  attempts. `TransactionRunner` now documents that `runInTransaction` must have committed when it
+  returns, and `publishEvents` throws if a batch is published twice.
   `processNext(limit)` keeps notifying at once, for callers that drive the processor themselves.
   ([#137](https://github.com/softwaremill/okapi/issues/137))
 - **`OutboxProcessorProperties` gained a `transportDispatch` constructor parameter** (see the

README.md

@@ -273,7 +273,7 @@ Metrics are then available at `/actuator/prometheus`.
 | `okapi.entries.delivered` | Counter | Successfully delivered entries |
 | `okapi.entries.retry.scheduled` | Counter | Failed attempts rescheduled for retry |
 | `okapi.entries.failed` | Counter | Permanently failed entries |
-| `okapi.batch.duration` | Timer | Processing time per batch, from the claim until the transaction has committed |
+| `okapi.batch.duration` | Timer | Processing time per batch; under the scheduler from the claim until the transaction has committed (not including listener callbacks) |
 | `okapi.entries.count` | Gauge | Current entry count (tag: `status`) |
 | `okapi.entries.lag.seconds` | Gauge | Age of the oldest entry (tag: `status`) |
 
@@ -291,7 +291,7 @@ max by (job, status) (okapi_entries_count)
 
 Without Spring Boot, construct `MicrometerOutboxListener` and `MicrometerOutboxMetrics` with your `MeterRegistry`. Refresh gauges with `OutboxMetricsRefresher` or your own scheduler.
 
-For custom reactions to delivery events, implement `OutboxProcessorListener`. `OutboxProcessor` takes a single listener; combine several with a composite of your own. The scheduler calls the listener after the processing transaction has committed, outside of it, and not for a batch whose commit failed, so every event describes persisted state and counters do not over-count when a commit fails. If you drive `OutboxProcessor` yourself, call `processBatch` inside your transaction and `publishEvents` after it commits; `processNext` does both at once, so inside a transaction its events precede the commit.
+For custom reactions to delivery events, implement `OutboxProcessorListener`. `OutboxProcessor` takes a single listener; combine several with a composite of your own. The scheduler calls the listener after the processing transaction has committed, outside of it, and not for a batch whose commit failed, so every event describes persisted state and counters do not over-count when a commit fails. Counters therefore count committed status changes, not delivery attempts: a send whose transaction then rolled back (a failed commit, or a block Exposed retried) is not counted, though the message may have reached the consumer. If you drive `OutboxProcessor` yourself, call `processBatch` inside your transaction and `publishEvents` after it commits; `processNext` does both at once, so inside a transaction its events precede the commit.
 
 ## Modules
 

UPGRADING.md

@@ -42,7 +42,13 @@ the processing transaction has committed, outside of it, and are skipped for a b
 failed.
 
 - A custom listener that wrote to the database expecting to join the processing transaction now
-  runs without one: open your own transaction in the callback, or make the write idempotent.
+  runs without one: open your own transaction in the callback, or make the write idempotent. This
+  fails quietly: an Exposed listener gets "No transaction in context", a Spring call with
+  `MANDATORY` propagation throws, and okapi only logs each such exception as a WARN.
+- Counters (`okapi.entries.delivered`, `okapi.entries.failed`, ...) now count committed status
+  changes. A send whose transaction rolled back is no longer counted.
+- A custom `TransactionRunner` must have committed when `runInTransaction` returns (now documented on
+  the interface); one that joins an outer transaction brings back pre-commit events.
 - `okapi.batch.duration` now includes the commit, so it reads slightly higher than before.
 - If you call `OutboxProcessor.processNext` inside your own transaction, events still fire before
   your commit. Switch to `val batch = runInTransaction { processor.processBatch(limit) }` followed by

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

@@ -59,7 +59,7 @@ class CompositeMessageDeliverer @JvmOverloads constructor(
      * satisfy that opt out via [TransportDispatch.SEQUENTIAL].
      *
      * The outbox transaction is open across delivery, and stays that way: the scheduler wraps the
-     * whole [OutboxProcessor.processNext] cycle — claim, deliver, update — in one transaction, so
+     * whole [OutboxProcessor.processBatch] cycle — claim, deliver, update — in one transaction, so
      * that `FOR UPDATE SKIP LOCKED` holds its row locks until the results are written. What this
      * dispatch changes is which thread the transport runs on, not that scope. The store round-trips
      * still run on the calling thread inside the transaction, but a group handed to a virtual

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

@@ -91,11 +91,14 @@ class OutboxProcessor @JvmOverloads constructor(
     /**
      * Notifies the listener of [batch]: [OutboxProcessorListener.onEntryProcessed] for each entry,
      * then [OutboxProcessorListener.onBatchProcessed], whose duration runs from the start of the batch
-     * until now, so it includes the commit when called after one. Call it at most once per batch.
+     * until this call, so it includes the commit when called after one, but not the listener
+     * callbacks. A batch's events can be published only once; a second call throws.
      */
     fun publishEvents(batch: ProcessedBatch) {
+        check(batch.markPublished()) { "Events of this batch were already published" }
+        val batchDuration = Duration.between(batch.startedAt, clock.instant())
         batch.entries.forEach { updated -> notifyEntry(updated, batch.deliveryDuration) }
-        notifyBatch(batch.count, Duration.between(batch.startedAt, clock.instant()))
+        notifyBatch(batch.count, batchDuration)
     }
 
     private fun notifyEntry(updated: OutboxEntry, duration: Duration) {

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

@@ -2,16 +2,29 @@ package com.softwaremill.okapi.core
 
 import java.time.Duration
 import java.time.Instant
+import java.util.Collections
+import java.util.concurrent.atomic.AtomicBoolean
 
 /**
  * The outcome of [OutboxProcessor.processBatch]: [entries] in the state just persisted, whose
- * listener events are held back until [OutboxProcessor.publishEvents].
+ * listener events are held back until [OutboxProcessor.publishEvents]. Only [OutboxProcessor]
+ * creates one; its constructor and timing fields are internal to okapi-core.
  */
 class ProcessedBatch internal constructor(
-    val entries: List<OutboxEntry>,
+    entries: List<OutboxEntry>,
     internal val startedAt: Instant,
     internal val deliveryDuration: Duration,
 ) {
+    /** The processed entries, in the state persisted for them. Read-only. */
+    val entries: List<OutboxEntry> = Collections.unmodifiableList(entries.toList())
+
     /** Number of entries processed (0 if the store had nothing to claim). */
     val count: Int get() = entries.size
+
+    private val published = AtomicBoolean(false)
+
+    /** Returns `true` the first time only. */
+    internal fun markPublished(): Boolean = published.compareAndSet(false, true)
+
+    override fun toString(): String = "ProcessedBatch(count=$count, startedAt=$startedAt, deliveryDuration=$deliveryDuration)"
 }

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

@@ -7,6 +7,13 @@ package com.softwaremill.okapi.core
  * - `okapi-spring-boot`: wraps Spring's `TransactionTemplate` via `SpringTransactionRunner`
  * - `okapi-exposed`: wraps Exposed's `transaction {}` via `ExposedTransactionRunner`
  * - Standalone: user-provided lambda
+ *
+ * Contract: when [runInTransaction] returns normally, the transaction it ran [block] in has
+ * committed. [OutboxScheduler] relies on this to emit listener events only for committed state
+ * (issue #137). A runner that joins an outer transaction instead (e.g. Spring's `REQUIRED` called
+ * inside `@Transactional`, or a nested Exposed `transaction {}` without nested transactions) breaks
+ * that. A runner may run [block] more than once, as Exposed does when a statement or the commit
+ * throws an `SQLException`; only the result of the attempt that committed is returned.
  */
 interface TransactionRunner {
     fun <T> runInTransaction(block: () -> T): T

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

@@ -1,5 +1,6 @@
 package com.softwaremill.okapi.core
 
+import io.kotest.assertions.throwables.shouldThrow
 import io.kotest.core.spec.style.FunSpec
 import io.kotest.matchers.collections.shouldBeEmpty
 import io.kotest.matchers.collections.shouldContainExactly
@@ -39,8 +40,8 @@ private class RecordingListener(private val log: MutableList<String>, private va
     }
 }
 
-private fun processor(listener: OutboxProcessorListener) = OutboxProcessor(
-    store = OneEntryPerClaimStore(),
+private fun processor(listener: OutboxProcessorListener, store: RouteAwareOutboxStore = OneEntryPerClaimStore()) = OutboxProcessor(
+    store = store,
     entryProcessor = OutboxEntryProcessor(
         object : MessageDeliverer {
             override val type = "stub"
@@ -113,4 +114,61 @@ class ListenerAfterCommitTest : FunSpec({
 
         log.shouldBeEmpty()
     }
+
+    test("a runner that retries the block, as Exposed does on SQLException, publishes one set of events") {
+        val log = Collections.synchronizedList(mutableListOf<String>())
+        val batchSeen = CountDownLatch(1)
+        val retryingRunner = object : TransactionRunner {
+            override fun <T> runInTransaction(block: () -> T): T {
+                block() // first attempt: its commit fails and it is rolled back
+                log += "rollback"
+                return block().also { log += "commit" }
+            }
+        }
+        val scheduler = OutboxScheduler(
+            processor(RecordingListener(log) { batchSeen.countDown() }),
+            retryingRunner,
+            OutboxSchedulerConfig(interval = Duration.ofMinutes(1)),
+        )
+
+        scheduler.start()
+        batchSeen.await(5, TimeUnit.SECONDS) shouldBe true
+        scheduler.stop()
+
+        log.take(4) shouldContainExactly listOf("rollback", "commit", "entry:Delivered", "batch:1")
+    }
+
+    test("a processBatch that throws publishes nothing") {
+        val log = Collections.synchronizedList(mutableListOf<String>())
+        val ticks = CountDownLatch(2)
+        val failingStore = object : RouteAwareOutboxStore by OneEntryPerClaimStore() {
+            override fun updateAfterProcessingBatch(entries: List<OutboxEntry>): List<OutboxEntry> {
+                ticks.countDown()
+                throw IllegalStateException("update failed")
+            }
+        }
+        val scheduler = OutboxScheduler(
+            processor(RecordingListener(log), failingStore),
+            object : TransactionRunner {
+                override fun <T> runInTransaction(block: () -> T): T = block()
+            },
+            OutboxSchedulerConfig(interval = Duration.ofMillis(20)),
+        )
+
+        scheduler.start()
+        ticks.await(5, TimeUnit.SECONDS) shouldBe true
+        scheduler.stop()
+
+        log.shouldBeEmpty()
+    }
+
+    test("a batch's events can be published only once") {
+        val log = mutableListOf<String>()
+        val processor = processor(RecordingListener(log))
+        val batch = processor.processBatch(10)
+
+        processor.publishEvents(batch)
+        shouldThrow<IllegalStateException> { processor.publishEvents(batch) }
+        log shouldContainExactly listOf("entry:Delivered", "batch:1")
+    }
 })

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

@@ -23,7 +23,7 @@ import java.util.concurrent.atomic.AtomicReference
 
 class OutboxSchedulerTest : FunSpec({
 
-    test("tick calls processNext with configured batchSize") {
+    test("tick calls processBatch with configured batchSize") {
         var capturedLimit: Int? = null
         val latch = CountDownLatch(1)
         val processor = stubProcessor { limit ->

okapi-integration-tests/src/test/kotlin/com/softwaremill/okapi/test/transaction/ListenerAfterCommitEndToEndTest.kt

@@ -11,12 +11,20 @@ import com.softwaremill.okapi.core.OutboxProcessor
 import com.softwaremill.okapi.core.OutboxProcessorListener
 import com.softwaremill.okapi.core.OutboxScheduler
 import com.softwaremill.okapi.core.OutboxSchedulerConfig
+import com.softwaremill.okapi.core.OutboxStore
 import com.softwaremill.okapi.core.RetryPolicy
 import com.softwaremill.okapi.core.TransactionRunner
+import com.softwaremill.okapi.exposed.ExposedConnectionProvider
+import com.softwaremill.okapi.exposed.ExposedTransactionRunner
 import com.softwaremill.okapi.postgres.PostgresOutboxStore
+import com.softwaremill.okapi.springboot.SpringConnectionProvider
+import com.softwaremill.okapi.springboot.SpringTransactionRunner
 import com.softwaremill.okapi.test.support.PostgresTestSupport
 import io.kotest.core.spec.style.FunSpec
 import io.kotest.matchers.shouldBe
+import org.jetbrains.exposed.v1.jdbc.Database
+import org.springframework.jdbc.datasource.DataSourceTransactionManager
+import org.springframework.transaction.support.TransactionTemplate
 import java.time.Clock
 import java.time.Duration
 import java.util.concurrent.CompletableFuture
@@ -37,23 +45,28 @@ class ListenerAfterCommitEndToEndTest : FunSpec({
     beforeSpec { db.start() }
     afterSpec { db.stop() }
 
-    test("the scheduler's listener sees the delivered status already committed") {
-        val store = PostgresOutboxStore(db.jdbc)
+    /** Runs one scheduler tick with [runner] and [store]; returns the status the listener saw from another connection. */
+    fun statusSeenByListener(runner: TransactionRunner, store: OutboxStore): String {
+        db.truncate()
         val entry = OutboxEntry.createPending(OutboxMessage("e", "{}"), AfterCommitDeliveryInfo, Clock.systemUTC().instant())
-        db.jdbc.withTransaction { store.persist(entry) }
+        db.jdbc.withTransaction { PostgresOutboxStore(db.jdbc).persist(entry) }
 
-        val statusSeenByListener = CompletableFuture<String>()
+        val seen = CompletableFuture<String>()
         val listener = object : OutboxProcessorListener {
             override fun onEntryProcessed(event: OutboxProcessingEvent) {
-                // A connection of its own: it only sees committed data.
-                db.dataSource.connection.use { conn ->
-                    conn.prepareStatement("SELECT status FROM okapi_outbox WHERE id = ?::uuid").use { stmt ->
-                        stmt.setString(1, entry.outboxId.raw.toString())
-                        stmt.executeQuery().use { rs ->
-                            rs.next()
-                            statusSeenByListener.complete(rs.getString("status"))
+                try {
+                    // A connection of its own: it only sees committed data.
+                    db.dataSource.connection.use { conn ->
+                        conn.prepareStatement("SELECT status FROM okapi_outbox WHERE id = ?::uuid").use { stmt ->
+                            stmt.setString(1, entry.outboxId.raw.toString())
+                            stmt.executeQuery().use { rs ->
+                                check(rs.next()) { "row ${entry.outboxId} not found" }
+                                seen.complete(rs.getString("status"))
+                            }
                         }
                     }
+                } catch (e: Exception) {
+                    seen.completeExceptionally(e)
                 }
             }
         }
@@ -69,16 +82,30 @@ class ListenerAfterCommitEndToEndTest : FunSpec({
             ),
             listener = listener,
         )
-        val runner = object : TransactionRunner {
-            override fun <T> runInTransaction(block: () -> T): T = db.jdbc.withTransaction(block)
-        }
         val scheduler = OutboxScheduler(processor, runner, OutboxSchedulerConfig(interval = Duration.ofMinutes(1)))
-
         scheduler.start()
         try {
-            statusSeenByListener.get(10, TimeUnit.SECONDS) shouldBe "DELIVERED"
+            return seen.get(10, TimeUnit.SECONDS)
         } finally {
             scheduler.stop()
         }
     }
+
+    test("plain JDBC runner: the listener sees the delivered status already committed") {
+        val runner = object : TransactionRunner {
+            override fun <T> runInTransaction(block: () -> T): T = db.jdbc.withTransaction(block)
+        }
+        statusSeenByListener(runner, PostgresOutboxStore(db.jdbc)) shouldBe "DELIVERED"
+    }
+
+    test("SpringTransactionRunner: the listener sees the delivered status already committed") {
+        val runner = SpringTransactionRunner(TransactionTemplate(DataSourceTransactionManager(db.dataSource)))
+        statusSeenByListener(runner, PostgresOutboxStore(SpringConnectionProvider(db.dataSource))) shouldBe "DELIVERED"
+    }
+
+    test("ExposedTransactionRunner: the listener sees the delivered status already committed") {
+        val database = Database.connect(db.dataSource)
+        statusSeenByListener(ExposedTransactionRunner(database), PostgresOutboxStore(ExposedConnectionProvider(database))) shouldBe
+            "DELIVERED"
+    }
 })