Add MinimizedCheckpointStrategy and refactor checkpointing logic in event sourcing

helightdev · helightdev · commit 83e31cbad4bd · 2025-11-06T05:33:25.000+01:00
diff --git a/krescent-core/src/main/kotlin/dev/helight/krescent/checkpoint/CheckpointStrategy.kt b/krescent-core/src/main/kotlin/dev/helight/krescent/checkpoint/CheckpointStrategy.kt
@@ -1,35 +1,61 @@
 package dev.helight.krescent.checkpoint
 
 import dev.helight.krescent.event.EventMessage
+import dev.helight.krescent.source.StreamingToken
 import kotlinx.datetime.Clock
 import kotlin.concurrent.atomics.AtomicLong
 import kotlin.concurrent.atomics.ExperimentalAtomicApi
 import kotlin.concurrent.atomics.incrementAndFetch
 import kotlin.time.Duration
 
 interface CheckpointStrategy {
-    suspend fun afterMessage(eventMessage: EventMessage, lastCheckpoint: StoredCheckpoint?): Boolean
-    suspend fun afterTermination(): Boolean = false
+    suspend fun afterMessage(eventMessage: EventMessage, context: CheckpointContext): Boolean = false
+    suspend fun afterTermination(context: CheckpointContext): Boolean = false
+    suspend fun afterCallback(context: CheckpointContext): Boolean = false
 }
 
+data class CheckpointContext(
+    val lastCheckpoint: StoredCheckpoint?,
+    val lastPosition: StreamingToken<*>?,
+)
+
 @Suppress("unused")
-object NoCheckpointStrategy : CheckpointStrategy {
-    override suspend fun afterMessage(eventMessage: EventMessage, lastCheckpoint: StoredCheckpoint?): Boolean {
-        return false
-    }
-}
+object NoCheckpointStrategy : CheckpointStrategy
 
 @Suppress("unused")
 object TerminationCheckpointStrategy : CheckpointStrategy {
-    override suspend fun afterMessage(eventMessage: EventMessage, lastCheckpoint: StoredCheckpoint?): Boolean {
+    override suspend fun afterMessage(eventMessage: EventMessage, context: CheckpointContext): Boolean {
         return false
     }
 
-    override suspend fun afterTermination(): Boolean {
+    override suspend fun afterTermination(context: CheckpointContext): Boolean {
         return true
     }
 }
 
+@Suppress("unused")
+class MinimizedCheckpointStrategy : CheckpointStrategy {
+
+    private var lastPosition: StreamingToken<*>? = null
+
+    override suspend fun afterTermination(context: CheckpointContext): Boolean {
+        return checkpointIfChanged(context)
+    }
+
+    override suspend fun afterCallback(context: CheckpointContext): Boolean {
+        return checkpointIfChanged(context)
+    }
+
+    private fun checkpointIfChanged(context: CheckpointContext): Boolean {
+        if (context.lastPosition == null) return false
+        if (lastPosition == null) {
+            lastPosition = context.lastPosition
+            return true
+        }
+        return lastPosition!!.compareUnsafe(context.lastPosition) != 0
+    }
+}
+
 
 @Suppress("unused")
 @OptIn(ExperimentalAtomicApi::class)
@@ -40,11 +66,11 @@ class FixedEventRateCheckpointStrategy(
 
     private val counter = AtomicLong(0)
 
-    override suspend fun afterTermination(): Boolean {
+    override suspend fun afterTermination(context: CheckpointContext): Boolean {
         return checkpointOnTermination
     }
 
-    override suspend fun afterMessage(eventMessage: EventMessage, lastCheckpoint: StoredCheckpoint?): Boolean {
+    override suspend fun afterMessage(eventMessage: EventMessage, context: CheckpointContext): Boolean {
         return counter.incrementAndFetch() % checkpoint == 0L
     }
 }
@@ -58,7 +84,7 @@ class ManualCheckpointStrategy : CheckpointStrategy {
         shouldCheckpoint = true
     }
 
-    override suspend fun afterMessage(eventMessage: EventMessage, lastCheckpoint: StoredCheckpoint?): Boolean {
+    override suspend fun afterMessage(eventMessage: EventMessage, context: CheckpointContext): Boolean {
         if (shouldCheckpoint) {
             shouldCheckpoint = false
             return true
@@ -74,22 +100,22 @@ class FixedTimeRateCheckpointStrategy(
     val checkpointOnTermination: Boolean = true,
 ) : CheckpointStrategy {
 
-    override suspend fun afterTermination(): Boolean {
+    override suspend fun afterTermination(context: CheckpointContext): Boolean {
         return checkpointOnTermination
     }
 
-    override suspend fun afterMessage(eventMessage: EventMessage, lastCheckpoint: StoredCheckpoint?): Boolean {
-        if (lastCheckpoint == null) return true
+    override suspend fun afterMessage(eventMessage: EventMessage, context: CheckpointContext): Boolean {
+        if (context.lastCheckpoint == null) return true
         val currentTime = Clock.System.now()
-        val duration = currentTime - (lastCheckpoint.timestamp)
+        val duration = currentTime - (context.lastCheckpoint.timestamp)
         return duration >= rate
     }
 }
 
 @Suppress("unused")
 object AlwaysCheckpointStrategy : CheckpointStrategy {
 
-    override suspend fun afterMessage(eventMessage: EventMessage, lastCheckpoint: StoredCheckpoint?): Boolean {
+    override suspend fun afterMessage(eventMessage: EventMessage, context: CheckpointContext): Boolean {
         return true
     }
 }
diff --git a/krescent-core/src/main/kotlin/dev/helight/krescent/checkpoint/CheckpointingEventSourceConsumer.kt b/krescent-core/src/main/kotlin/dev/helight/krescent/checkpoint/CheckpointingEventSourceConsumer.kt
@@ -1,10 +1,7 @@
 package dev.helight.krescent.checkpoint
 
 import dev.helight.krescent.event.*
-import dev.helight.krescent.source.EventSourceConsumer
-import dev.helight.krescent.source.EventSourcingStrategy
-import dev.helight.krescent.source.StreamingEventSource
-import dev.helight.krescent.source.StreamingToken
+import dev.helight.krescent.source.*
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.NonCancellable
 import kotlinx.coroutines.withContext
@@ -23,47 +20,51 @@ class CheckpointingEventSourceConsumer(
 ) : EventSourceConsumer {
 
     private val logger = LoggerFactory.getLogger(CheckpointingEventSourceConsumer::class.java)
+    private var lastCheckpoint: StoredCheckpoint? = null
+    private var lastPosition: StreamingToken<*>? = null
+    private val context: CheckpointContext
+        get() = CheckpointContext(
+            lastCheckpoint = lastCheckpoint,
+            lastPosition = lastPosition
+        )
 
     @Suppress("UNCHECKED_CAST")
     override suspend fun strategy(strategy: EventSourcingStrategy) {
-        var lastCheckpoint = loadLastCheckpoint()
-        var lastPosition: StreamingToken<*>? = lastCheckpoint?.position?.let {
-            source.deserializeToken(it)
+        loadLastCheckpoint()
+        if (strategy is CallbackEventSourcingStrategy) {
+            strategy.addThenChain {
+                if (checkpointStrategy.afterCallback(context)) tryCheckpoint()
+            }
         }
+
         try {
             strategy.source(source, lastPosition, object : EventMessageStreamProcessor {
                 override suspend fun process(
                     message: EventMessage,
                     position: StreamingToken<*>,
                 ) {
                     consumer.process(message, position)
-                    val tickerResult = checkpointStrategy.afterMessage(message, lastCheckpoint)
-                    if (tickerResult) {
-                        val checkpoint = checkpoint(position)
-                        checkpointStorage.storeCheckpoint(checkpoint)
-                        lastCheckpoint = checkpoint // TODO: Why is not used per lint? This isn't closed and should work
-                    }
                     lastPosition = position
+                    if (checkpointStrategy.afterMessage(message, context)) tryCheckpoint()
                 }
 
                 override suspend fun forwardSystemEvent(event: Event) {
                     consumer.forwardSystemEvent(event)
                 }
             })
         } catch (e: CancellationException) {
-            withContext(NonCancellable) {
-                if (checkpointStrategy.afterTermination() && lastPosition != null) {
-                    val checkpoint = checkpoint(lastPosition)
-                    checkpointStorage.storeCheckpoint(checkpoint)
-                }
-            }
+            withContext(NonCancellable) { if (checkpointStrategy.afterTermination(context)) tryCheckpoint() }
             throw e
         }
 
-        if (checkpointStrategy.afterTermination() && lastPosition != null) {
-            val checkpoint = checkpoint(lastPosition)
-            checkpointStorage.storeCheckpoint(checkpoint)
-        }
+        if (checkpointStrategy.afterTermination(context)) tryCheckpoint()
+    }
+
+    private suspend fun tryCheckpoint() {
+        if (lastPosition == null) return
+        val checkpoint = checkpoint(lastPosition!!)
+        checkpointStorage.storeCheckpoint(checkpoint)
+        lastCheckpoint = checkpoint
     }
 
     private suspend fun loadLastCheckpoint(): StoredCheckpoint? {
@@ -90,6 +91,8 @@ class CheckpointingEventSourceConsumer(
             logger.info("No valid checkpoint found for namespace '$namespace', starting from scratch")
             consumer.forwardSystemEvent(SystemStreamHeadEvent)
         }
+        this.lastCheckpoint = lastCheckpoint
+        this.lastPosition = lastCheckpoint?.position?.let { source.deserializeToken(it) }
         return lastCheckpoint
     }
 
diff --git a/krescent-core/src/main/kotlin/dev/helight/krescent/model/EventModelBuilder.kt b/krescent-core/src/main/kotlin/dev/helight/krescent/model/EventModelBuilder.kt
@@ -1,9 +1,6 @@
 package dev.helight.krescent.model
 
-import dev.helight.krescent.checkpoint.CheckpointStorage
-import dev.helight.krescent.checkpoint.CheckpointStrategy
-import dev.helight.krescent.checkpoint.CheckpointSupport
-import dev.helight.krescent.checkpoint.CheckpointingEventSourceConsumer
+import dev.helight.krescent.checkpoint.*
 import dev.helight.krescent.event.EventCatalog
 import dev.helight.krescent.event.EventStreamProcessor
 import dev.helight.krescent.event.VirtualEventIngest
@@ -105,7 +102,7 @@ class EventModelBuilder(
      */
     fun useCheckpoints(
         checkpointStorage: CheckpointStorage,
-        strategy: CheckpointStrategy,
+        strategy: CheckpointStrategy = MinimizedCheckpointStrategy(),
     ) {
         checkpointConfig = CheckpointConfiguration(checkpointStorage, strategy)
     }
diff --git a/krescent-core/src/main/kotlin/dev/helight/krescent/source/EventSourcingStrategy.kt b/krescent-core/src/main/kotlin/dev/helight/krescent/source/EventSourcingStrategy.kt
@@ -15,6 +15,16 @@ interface EventSourcingStrategy {
     ) = consumer.process(this.first, this.second)
 }
 
-interface WriteCompatibleEventSourcingStrategy {
+interface CallbackEventSourcingStrategy {
     var then: suspend () -> Unit
-}
+
+    fun addThenChain(callback: suspend () -> Unit) {
+        val previous = then
+        then = {
+            previous()
+            callback()
+        }
+    }
+}
+
+interface WriteCompatibleEventSourcingStrategy : CallbackEventSourcingStrategy
diff --git a/krescent-core/src/main/kotlin/dev/helight/krescent/source/StreamingToken.kt b/krescent-core/src/main/kotlin/dev/helight/krescent/source/StreamingToken.kt
@@ -11,4 +11,7 @@ interface StreamingToken<SELF : StreamingToken<SELF>> : Comparable<SELF> {
      * @return A string representing the serialized token.
      */
     fun serialize(): String
+
+    @Suppress("UNCHECKED_CAST")
+    fun compareUnsafe(other: StreamingToken<*>) = this.compareTo(other as SELF)
 }
diff --git a/krescent-core/src/main/kotlin/dev/helight/krescent/source/strategy/StreamingSourcingStrategy.kt b/krescent-core/src/main/kotlin/dev/helight/krescent/source/strategy/StreamingSourcingStrategy.kt
@@ -1,6 +1,7 @@
 package dev.helight.krescent.source.strategy
 
 import dev.helight.krescent.event.*
+import dev.helight.krescent.source.CallbackEventSourcingStrategy
 import dev.helight.krescent.source.EventSourcingStrategy
 import dev.helight.krescent.source.StreamingEventSource
 import dev.helight.krescent.source.StreamingToken
@@ -21,8 +22,8 @@ import dev.helight.krescent.source.StreamingToken
  * - [SystemHintEndTransactionEvent] at the end of the catch-up phase, after the live stream has started.
  */
 class StreamingSourcingStrategy(
-    val afterCatchup: suspend () -> Unit = {},
-) : EventSourcingStrategy {
+    override var then: suspend () -> Unit = {},
+) : EventSourcingStrategy, CallbackEventSourcingStrategy {
     override suspend fun source(
         source: StreamingEventSource,
         startToken: StreamingToken<*>?,
@@ -39,8 +40,8 @@ class StreamingSourcingStrategy(
             consumer.forwardSystemEvent(SystemStreamCaughtUpEvent)
         } finally {
             consumer.forwardSystemEvent(SystemHintEndTransactionEvent)
-            afterCatchup()
         }
+        then()
 
         // Begin streaming events until interrupted
         source.streamEvents(lastToken).collect {
diff --git a/krescent-core/src/test/kotlin/dev/helight/krescent/bookstore/BookCountReadModelTest.kt b/krescent-core/src/test/kotlin/dev/helight/krescent/bookstore/BookCountReadModelTest.kt
@@ -1,13 +1,18 @@
 package dev.helight.krescent.bookstore
 
 import dev.helight.krescent.checkpoint.AlwaysCheckpointStrategy
+import dev.helight.krescent.checkpoint.MinimizedCheckpointStrategy
 import dev.helight.krescent.checkpoint.impl.InMemoryCheckpointStorage
 import dev.helight.krescent.event.Event
 import dev.helight.krescent.model.EventModelBase.Extension.withConfiguration
 import dev.helight.krescent.model.ReadModelBase.Extension.catchup
 import dev.helight.krescent.model.ReadModelBase.Extension.restoreOnly
+import dev.helight.krescent.model.ReadModelBase.Extension.stream
 import dev.helight.krescent.model.ReducingReadModel
 import dev.helight.krescent.source.impl.InMemoryEventStore
+import kotlinx.coroutines.async
+import kotlinx.coroutines.cancelAndJoin
+import kotlinx.coroutines.delay
 import kotlinx.coroutines.runBlocking
 import kotlinx.serialization.Serializable
 import java.util.concurrent.atomic.AtomicInteger
@@ -47,6 +52,38 @@ class BookCountReadModelTest {
         assertNotNull(late)
         assertEquals(9, late.target["1"])
     }
+
+    @Test
+    fun `Test minimized checkpointing`(): Unit = runBlocking {
+        val source = InMemoryEventStore()
+        val checkpointStorage = InMemoryCheckpointStorage()
+        source.publishAll(bookstoreSimulatedEventStream)
+
+        val initial = BooksAvailableReadModel().withConfiguration {
+            useCheckpoints(checkpointStorage, MinimizedCheckpointStrategy())
+        }.restoreOnly()
+        assertNull(initial)
+        assertNull(checkpointStorage.getLatestCheckpoint("books.counts"))
+
+        // Check if it happens at termination / catchup
+        val caughtUp = BooksAvailableReadModel().withConfiguration {
+            useCheckpoints(checkpointStorage, MinimizedCheckpointStrategy())
+        }.catchup(source)
+        assertEquals(9, caughtUp.target["1"])
+        assertNotNull(checkpointStorage.getLatestCheckpoint("books.counts"))
+
+        // Check if it happens at the callback as well
+        checkpointStorage.clearCheckpoints()
+        assertNull(checkpointStorage.getLatestCheckpoint("books.counts"))
+        val job = async {
+            BooksAvailableReadModel().withConfiguration {
+                useCheckpoints(checkpointStorage, MinimizedCheckpointStrategy())
+            }.stream(source)
+        }
+        delay(200)
+        assertNotNull(checkpointStorage.getLatestCheckpoint("books.counts"))
+        job.cancelAndJoin()
+    }
 }
 
 class BooksAvailableReadModel(

Original file line number	Diff line number	Diff line change
`@@ -11,4 +11,7 @@ interface StreamingToken<SELF : StreamingToken<SELF>> : Comparable<SELF> {`
`11`	`11`	`* @return A string representing the serialized token.`
`12`	`12`	`*/`
`13`	`13`	`fun serialize(): String`
	`14`	`+`
	`15`	`+ @Suppress("UNCHECKED_CAST")`
	`16`	`+ fun compareUnsafe(other: StreamingToken<*>) = this.compareTo(other as SELF)`
`14`	`17`	`}`