Less calls to store (#34)

* Add an `isEmpty` to the retry store interface and `hasNext` to the retry flow to allow skipping lots of calls to the store in the event the retry store is empty
* Fix the last retry strategy from querying overflow value as `last`

ft-coroutines-kafka-retry/src/main/kotlin/tech/figure/kafka/coroutines/retry/flow/KafkaFlowRetry.kt

@@ -28,6 +28,8 @@ open class KafkaFlowRetry<K, V>(
 ) : FlowRetry<ConsumerRecord<K, V>> {
     private val log = KotlinLogging.logger {}
 
+    override suspend fun hasNext(): Boolean = !store.isEmpty()
+
     override suspend fun produceNext(
         attemptRange: IntRange,
         olderThan: OffsetDateTime,

ft-coroutines-kafka-retry/src/main/kotlin/tech/figure/kafka/coroutines/retry/store/InMemoryConsumerRecordStore.kt

@@ -11,6 +11,9 @@ fun <K, V> inMemoryConsumerRecordStore(
 ) = object : RetryRecordStore<ConsumerRecord<K, V>> {
     val log = KotlinLogging.logger {}
 
+    override suspend fun isEmpty(): Boolean =
+        data.isEmpty()
+
     override suspend fun select(
         attemptRange: IntRange,
         lastAttempted: OffsetDateTime,

ft-coroutines-retry/src/main/kotlin/tech/figure/coroutines/retry/RetryStrategy.kt

@@ -26,7 +26,7 @@ val defaultRetryStrategies = listOf(
     RetryStrategy(1.hours, 5),
     RetryStrategy(3.hours, 2),
     RetryStrategy(1.days, 8),
-    RetryStrategy(1.days, Int.MAX_VALUE)
+    RetryStrategy(1.days, Int.MAX_VALUE - 23) // max - sum of other retry counts. otherwise it overflows to Int.MIN_VALUE
 )
 
 internal fun List<RetryStrategy>.invert(): Map<IntRange, RetryStrategy> {

ft-coroutines-retry/src/main/kotlin/tech/figure/coroutines/retry/flow/FlowRetry.kt

@@ -8,6 +8,13 @@ import kotlinx.coroutines.flow.Flow
  * Callbacks and hooks to generate a limited [Flow] of [RetryRecord] to retry.
  */
 interface FlowRetry<T> : FlowProcessor<T> {
+    /**
+     * Determine if there is anything in the hopper to process.
+     *
+     * @return True if there are pending items, false otherwise
+     */
+    suspend fun hasNext(): Boolean
+
     /**
      * Generate the next group of items to retry.
      *

ft-coroutines-retry/src/main/kotlin/tech/figure/coroutines/retry/flow/RetryFlow.kt

@@ -1,21 +1,20 @@
 package tech.figure.coroutines.retry.flow
 
-import tech.figure.coroutines.retry.RetryStrategy
-import tech.figure.coroutines.retry.defaultRetryStrategies
-import tech.figure.coroutines.retry.invert
-import tech.figure.coroutines.retry.store.RetryRecord
-import tech.figure.coroutines.tryMap
 import java.time.OffsetDateTime
 import kotlin.time.Duration
 import kotlin.time.Duration.Companion.seconds
-import kotlin.time.ExperimentalTime
 import kotlin.time.toJavaDuration
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.collect
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.onStart
 import mu.KotlinLogging
+import tech.figure.coroutines.retry.RetryStrategy
+import tech.figure.coroutines.retry.defaultRetryStrategies
+import tech.figure.coroutines.retry.invert
+import tech.figure.coroutines.retry.store.RetryRecord
+import tech.figure.coroutines.tryMap
 
 internal val DEFAULT_RETRY_INTERVAL = 10.seconds
 
@@ -28,7 +27,7 @@ internal val DEFAULT_RETRY_INTERVAL = 10.seconds
  *
  * Once a record is successfully processed, emit the data element out to the flow.
  */
-@OptIn(ExperimentalTime::class, ExperimentalCoroutinesApi::class)
+@OptIn(ExperimentalCoroutinesApi::class)
 fun <T> retryFlow(
     flowRetry: FlowRetry<T>,
     retryInterval: Duration = DEFAULT_RETRY_INTERVAL,
@@ -39,6 +38,10 @@ fun <T> retryFlow(
     val strategies = retryStrategies.invert()
 
     return pollingFlow(retryInterval) {
+        if (!flowRetry.hasNext()) {
+            return@pollingFlow
+        }
+
         for (strategy in strategies) {
             val lastAttempted = OffsetDateTime.now().minus(strategy.value.lastAttempted.toJavaDuration())
 

ft-coroutines-retry/src/main/kotlin/tech/figure/coroutines/retry/flow/SimpleQueueFlowRetry.kt

@@ -3,17 +3,21 @@ package tech.figure.coroutines.retry.flow
 import tech.figure.coroutines.channels.receiveQueued
 import tech.figure.coroutines.retry.store.RetryRecord
 import java.time.OffsetDateTime
+import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.asFlow
 
-class SimpleChannelFlowRetry<T>(
+open class SimpleChannelFlowRetry<T>(
     private val queue: Channel<RetryRecord<T>> = Channel(capacity = UNLIMITED),
-    private val onSuccess: (T) -> Unit = {},
-    private val onFailure: (T) -> Unit = {},
-    private val block: (T, Int) -> Unit,
+    private val onSuccess: (item: T) -> Unit = {},
+    private val onFailure: (item: T) -> Unit = {},
+    private val block: (item: T, attempt: Int) -> Unit,
 ) : FlowRetry<T> {
+    @OptIn(ExperimentalCoroutinesApi::class)
+    override suspend fun hasNext(): Boolean = !queue.isEmpty
+
     override suspend fun send(item: T, e: Throwable) {
         queue.send(RetryRecord(item, lastException = e.localizedMessage))
     }

ft-coroutines-retry/src/main/kotlin/tech/figure/coroutines/retry/store/RetryRecord.kt

@@ -8,5 +8,6 @@ open class RetryRecord<T>(
     var lastAttempted: OffsetDateTime = OffsetDateTime.now(),
     var lastException: String = ""
 ) {
-    override fun toString(): String = "RetryRecord(attempt:$attempt Exception:$lastException lastAttempted:$lastAttempted data:$data)"
+    override fun toString(): String =
+        "RetryRecord(attempt:$attempt Exception:$lastException lastAttempted:$lastAttempted data:$data)"
 }

ft-coroutines-retry/src/main/kotlin/tech/figure/coroutines/retry/store/RetryRecordStore.kt

@@ -4,6 +4,7 @@ import tech.figure.coroutines.retry.flow.DEFAULT_FETCH_LIMIT
 import java.time.OffsetDateTime
 
 interface RetryRecordStore<T> {
+    suspend fun isEmpty(): Boolean
     suspend fun select(attemptRange: IntRange, lastAttempted: OffsetDateTime, limit: Int = DEFAULT_FETCH_LIMIT): List<RetryRecord<T>>
     suspend fun getOne(item: T): RetryRecord<T>?
     suspend fun putOne(item: T, lastException: Throwable? = null, mutator: RetryRecord<T>.() -> Unit)

ft-coroutines-retry/src/test/kotlin/tech/figure/coroutines/retry/flow/TestRetryFlow.kt

@@ -0,0 +1,68 @@
+package tech.figure.coroutines.retry.flow
+
+import io.kotest.core.spec.style.AnnotationSpec
+import java.time.OffsetDateTime
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.flow.toList
+import kotlinx.coroutines.test.runTest
+import org.junit.jupiter.api.fail
+import tech.figure.coroutines.retry.store.RetryRecord
+
+@OptIn(ExperimentalCoroutinesApi::class)
+class TestRetryFlow : AnnotationSpec() {
+    suspend fun <E> List<E>.asChannel(): Channel<E> {
+        val ch = Channel<E>(capacity = size)
+        forEach {
+            ch.send(it)
+        }
+        return ch
+    }
+
+    suspend fun <T> testFlowRetry(
+        list: List<RetryRecord<T>>,
+        onSuccess: (item: T) -> Unit,
+        onFailure: (item: T) -> Unit,
+        block: (item: T, attempt: Int) -> Unit,
+    ): SimpleChannelFlowRetry<T> {
+        return SimpleChannelFlowRetry(
+            queue = list.asChannel(),
+            onSuccess = onSuccess,
+            onFailure = onFailure,
+            block = block
+        )
+    }
+
+    @Test
+    fun testRetryFlowEmpty() = runTest {
+        val flow = testFlowRetry(
+            emptyList<RetryRecord<Int>>(),
+            { fail("should not be called") },
+            { fail("should not be called") },
+            { _, _ -> fail("should not be called") },
+        )
+
+        assert(!flow.hasNext()) { "hasNext should be false" }
+        assert(flow.produceNext(0..Int.MAX_VALUE, OffsetDateTime.MIN).toList().isEmpty()) { "should produce nothing" }
+    }
+
+    @Test
+    fun testRetryFlowNotEmpty() = runTest {
+        val flow = testFlowRetry(
+            listOf(RetryRecord(1), RetryRecord(2), RetryRecord(3)),
+            { },
+            { fail("should not be called") },
+            { _, _ -> }
+        )
+
+        assert(flow.drain().size == 3) { "not all elements produced" }
+    }
+}
+
+private suspend fun <T> FlowRetry<T>.drain(): List<RetryRecord<T>> {
+    val list = mutableListOf<RetryRecord<T>>()
+    while (hasNext()) {
+        list.addAll(produceNext(0..Int.MAX_VALUE, OffsetDateTime.MIN, Int.MAX_VALUE).toList())
+    }
+    return list
+}