Repository: kevincianfarini/cardiologist (kotlin)

Issue:
Account for time drift in long running operations. 
Clock drift occurs when a machine's internal clock does not tick at the same rate as a reference clock does. This is solved by the network time protocol, which synchronizes a machine's clock with the reference clock. During synchronization, a system's clock can experience a discontinuous jump, and we should handle that jump. 

The cron utility accounts for time drift by waking up and checking for a difference between "virtual time" and wall time every minute. https://github.com/cronie-crond/cronie/blob/adbd221f2ee1f12cad4924fb5a53f88346de885c/src/cron.c#L349-L480

We should do the same for the following operations: 

* [`Clock.delayFor`](https://kevincianfarini.github.io/cardiologist/docs/0.x/cardiologist/io.github.kevincianfarini.cardiologist/delay-for.html) when the amount of time delayed is >= 1 minute
* [`Clock.delayUntil`](https://kevincianfarini.github.io/cardiologist/docs/0.x/cardiologist/io.github.kevincianfarini.cardiologist/delay-until.html) when the instant in time is >= 1 minute in the future. 
* [`Clock.intervalPulse`](https://kevincianfarini.github.io/cardiologist/docs/0.x/cardiologist/io.github.kevincianfarini.cardiologist/interval-pulse.html) when the interval is >= 1 minute. 
* [`Clock.schedulePulse`](https://kevincianfarini.github.io/cardiologist/docs/0.x/cardiologist/io.github.kevincianfarini.cardiologist/schedule-pulse.html) when the next Pulse is scheduled to occur more than 1 minute in the future.

Pull request:
Closes #32

Patch:
diff --git a/src/commonMain/kotlin/io/github/kevincianfarini/cardiologist/Pulse.kt b/src/commonMain/kotlin/io/github/kevincianfarini/cardiologist/Pulse.kt
index 79bad21..36340e0 100644
--- a/src/commonMain/kotlin/io/github/kevincianfarini/cardiologist/Pulse.kt
+++ b/src/commonMain/kotlin/io/github/kevincianfarini/cardiologist/Pulse.kt
@@ -44,6 +44,5 @@ public value class Pulse internal constructor(private val flow: Flow<Instant>) {
                 }
             }
         }
-        RecurringJobMode.DelayBetweenSequential -> flow.collect(action)
     }
 }
\ No newline at end of file
diff --git a/src/commonMain/kotlin/io/github/kevincianfarini/cardiologist/RecurringJobMode.kt b/src/commonMain/kotlin/io/github/kevincianfarini/cardiologist/RecurringJobMode.kt
index f0bc31f..8d4f094 100644
--- a/src/commonMain/kotlin/io/github/kevincianfarini/cardiologist/RecurringJobMode.kt
+++ b/src/commonMain/kotlin/io/github/kevincianfarini/cardiologist/RecurringJobMode.kt
@@ -1,22 +1,14 @@
 package io.github.kevincianfarini.cardiologist
 
 public enum class RecurringJobMode {
-
     /**
      * Schedules recurring jobs concurrently. That is if job `n` is still active when job
      * `n + 1` should begin, then job `n` continues to run concurrently alongside job `n + 1`.
      */
     Concurrent,
-
     /**
      * Schedules recurring jobs sequentially by cancelling an unfinished job if a new one
      * should begin executing.
      */
     CancellingSequential,
-
-    /**
-     * Schedules recurring jobs sequentially by ensuring delay periods are interleaved between
-     * jobs.
-     */
-    DelayBetweenSequential,
 }
diff --git a/src/commonMain/kotlin/io/github/kevincianfarini/cardiologist/schedule.kt b/src/commonMain/kotlin/io/github/kevincianfarini/cardiologist/schedule.kt
index f3c70f5..666cad4 100644
--- a/src/commonMain/kotlin/io/github/kevincianfarini/cardiologist/schedule.kt
+++ b/src/commonMain/kotlin/io/github/kevincianfarini/cardiologist/schedule.kt
@@ -1,6 +1,5 @@
 package io.github.kevincianfarini.cardiologist
 
-import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.flow
 import kotlinx.datetime.*
 import io.github.kevincianfarini.cardiologist.impl.nextMatch
@@ -11,9 +10,11 @@ import kotlin.time.Duration
  */
 public fun Clock.intervalPulse(interval: Duration): Pulse {
     val flow = flow {
+        var nextPulse: Instant = now() + interval
         while (true) {
             emit(now())
-            delay(interval)
+            delayUntil(nextPulse)
+            nextPulse += interval
         }
     }
     return Pulse(flow)
@@ -24,9 +25,11 @@ public fun Clock.intervalPulse(interval: Duration): Pulse {
  */
 public fun Clock.intervalPulse(period: DateTimePeriod, timeZone: TimeZone): Pulse {
     val flow = flow {
+        var nextPulse: Instant = now().plus(period, timeZone)
         while (true) {
             emit(now())
-            delayFor(period, timeZone)
+            delayUntil(nextPulse)
+            nextPulse = nextPulse.plus(period, timeZone)
         }
     }
     return Pulse(flow)
@@ -36,7 +39,7 @@ public fun Clock.intervalPulse(period: DateTimePeriod, timeZone: TimeZone): Puls
  * Schedule a [Pulse] whose beats occur [atSecond], [atMinute], [atHour], [onDayOfMonth], and [inMonth] for a specific
  * [timeZone].
  *
- * Null values for any parameter indiciates that a beat can occur on _any_ second, minute, hour, etc. For example,
+ * Null values for any parameter indicate that a beat can occur on _any_ second, minute, hour, etc. For example,
  * scheduling a pulse on the fifth second of every minute would look like the following:
  *
  * ```kt
@@ -69,9 +72,9 @@ public fun Clock.schedulePulse(
     inMonth: Month? = null,
 ): Pulse {
     val flow = flow {
+        var lastPulse: LocalDateTime = now().toLocalDateTime(timeZone)
         while (true) {
-            val nowLocal = now().toLocalDateTime(timeZone)
-            val nextPulse = nowLocal.nextMatch(
+            val nextPulse = lastPulse.nextMatch(
                 atSeconds = atSecond?.let { it..it } ?: 0..59,
                 atMinutes = atMinute?.let { it..it } ?: 0..59,
                 atHours = atHour?.let { it..it } ?: 0..23,
@@ -80,6 +83,7 @@ public fun Clock.schedulePulse(
             )
             delayUntil(nextPulse, timeZone)
             emit(now())
+            lastPulse = nextPulse
         }
     }
     return Pulse(flow)
diff --git a/src/commonMain/kotlin/io/github/kevincianfarini/cardiologist/suspend.kt b/src/commonMain/kotlin/io/github/kevincianfarini/cardiologist/suspend.kt
index 72d3896..2588f87 100644
--- a/src/commonMain/kotlin/io/github/kevincianfarini/cardiologist/suspend.kt
+++ b/src/commonMain/kotlin/io/github/kevincianfarini/cardiologist/suspend.kt
@@ -1,24 +1,64 @@
 package io.github.kevincianfarini.cardiologist
 
+import kotlin.time.Duration.Companion.minutes
 import kotlinx.coroutines.delay
-import kotlinx.datetime.*
+import kotlinx.datetime.Clock
+import kotlinx.datetime.DateTimePeriod
+import kotlinx.datetime.Instant
+import kotlinx.datetime.LocalDateTime
+import kotlinx.datetime.LocalTime
+import kotlinx.datetime.TimeZone
+import kotlinx.datetime.atTime
+import kotlinx.datetime.plus
+import kotlinx.datetime.toInstant
+import kotlinx.datetime.toLocalDateTime
 
+/**
+ * Suspend the coroutine until [instant] without blocking the thread. If [Clock.now] is greater than [instant], this
+ * function returns immediately.
+ *
+ * This function checks for time drift in one minute increments. Coroutines delayed for more than one minute will be
+ * periodically resumed, allowing them to account for drift.
+ */
 public suspend fun Clock.delayUntil(instant: Instant) {
-    val duration = instant - now()
-    delay(duration)
+    var now = now()
+    while (now < instant) {
+        val durationUntilInstant = instant - now
+        val delayIncrement = minOf(durationUntilInstant, 1.minutes)
+        delay(delayIncrement)
+        now = now()
+    }
 }
 
+/**
+ * Suspend the coroutine until [dateTime] in [timeZone] without blocking the thread. If [Clock.now] is greater than
+ * the [Instant] associated with [dateTime] and [timeZone], this function returns immediately.
+ *
+ * This function checks for time drift in one minute increments. Coroutines delayed for more than one minute will be
+ * periodically resumed, allowing them to account for drift.
+ */
 public suspend fun Clock.delayUntil(dateTime: LocalDateTime, timeZone: TimeZone) {
     delayUntil(instant = dateTime.toInstant(timeZone))
 }
 
+/**
+ * Suspend the coroutine for [period] in [timeZone] without blocking the thread.
+ *
+ * This function checks for time drift in one minute increments. Coroutines delayed for more than one minute will be
+ * periodically resumed, allowing them to account for drift.
+ */
 public suspend fun Clock.delayFor(period: DateTimePeriod, timeZone: TimeZone) {
     val now = now()
     val futureInstant = now.plus(period, timeZone)
-    val duration = futureInstant - now
-    delay(duration)
+    delayUntil(futureInstant)
 }
 
+/**
+ * Suspend the coroutine until the next occurance of [time] in [timeZone] without blocking the thread.
+ *
+ * This function checks for time drift in one minute increments. Coroutines delayed for more than one minute will be
+ * periodically resumed, allowing them to account for drift.
+ */
 public suspend fun Clock.delayUntilNext(time: LocalTime, timeZone: TimeZone) {
     val now = now()
     val nowLocal = now.toLocalDateTime(timeZone)
@@ -27,7 +67,5 @@ public suspend fun Clock.delayUntilNext(time: LocalTime, timeZone: TimeZone) {
         desiredLocal < nowLocal -> desiredLocal.toInstant(timeZone).plus(DateTimePeriod(days = 1), timeZone)
         else -> desiredLocal.toInstant(timeZone)
     }
-
-    val duration = futureInstant - now
-    delay(duration)
+    delayUntil(futureInstant)
 }
\ No newline at end of file


Test patch:
diff --git a/src/commonTest/kotlin/io/github/kevincianfarini/cardiologist/FakeClock.kt b/src/commonTest/kotlin/io/github/kevincianfarini/cardiologist/FakeClock.kt
index 29f2598..369e13a 100644
--- a/src/commonTest/kotlin/io/github/kevincianfarini/cardiologist/FakeClock.kt
+++ b/src/commonTest/kotlin/io/github/kevincianfarini/cardiologist/FakeClock.kt
@@ -3,7 +3,10 @@ package io.github.kevincianfarini.cardiologist
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.test.TestCoroutineScheduler
 import kotlinx.coroutines.test.TestScope
+import kotlinx.coroutines.test.advanceTimeBy
+import kotlinx.coroutines.test.currentTime
 import kotlinx.datetime.*
+import kotlin.time.Duration
 
 @OptIn(ExperimentalCoroutinesApi::class)
 private class TestClock(private val scheduler: TestCoroutineScheduler) : Clock {
@@ -12,10 +15,20 @@ private class TestClock(private val scheduler: TestCoroutineScheduler) : Clock {
 
 val TestScope.testClock: Clock get() = TestClock(testScheduler)
 
-private class StaticClock(private val now: Instant) : Clock {
-    override fun now() = now
+@OptIn(ExperimentalCoroutinesApi::class)
+fun TestScope.testClockAt(instant: Instant): Clock {
+    val currentTime = Instant.fromEpochMilliseconds(currentTime)
+    advanceTimeBy(instant - currentTime)
+    return TestClock(testScheduler)
+}
+
+fun TestScope.testClockAt(localDateTime: LocalDateTime, timeZone: TimeZone): Clock {
+    return testClockAt(localDateTime.toInstant(timeZone))
 }
 
-fun Instant.asClock(): Clock = StaticClock(this)
+private class QueueClock(private val instants: List<Instant>) : Clock {
+    private var index = 0
+    override fun now(): Instant = instants[index++]
+}
 
-fun LocalDateTime.asClock(timeZone: TimeZone): Clock = StaticClock(toInstant(timeZone))
+fun List<Instant>.asClock(): Clock = QueueClock(this)
\ No newline at end of file
diff --git a/src/commonTest/kotlin/io/github/kevincianfarini/cardiologist/PulseTests.kt b/src/commonTest/kotlin/io/github/kevincianfarini/cardiologist/PulseTests.kt
index 699da74..b54db1d 100644
--- a/src/commonTest/kotlin/io/github/kevincianfarini/cardiologist/PulseTests.kt
+++ b/src/commonTest/kotlin/io/github/kevincianfarini/cardiologist/PulseTests.kt
@@ -1,6 +1,8 @@
 package io.github.kevincianfarini.cardiologist
 
 import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.runBlocking
 import kotlinx.coroutines.test.runTest
 import kotlinx.coroutines.test.testTimeSource
 import kotlinx.datetime.*
@@ -10,7 +12,7 @@ import kotlin.time.Duration.Companion.seconds
 import kotlin.time.ExperimentalTime
 import kotlin.time.measureTime
 
-@OptIn(ExperimentalTime::class, ExperimentalCoroutinesApi::class)
+@OptIn(ExperimentalCoroutinesApi::class)
 class PulseTests {
 
     @Test fun durationPulse_emits_initial_instant_before_delaying() = runTest {
diff --git a/src/commonTest/kotlin/io/github/kevincianfarini/cardiologist/SuspendingTests.kt b/src/commonTest/kotlin/io/github/kevincianfarini/cardiologist/SuspendingTests.kt
index 519e725..b53ac04 100644
--- a/src/commonTest/kotlin/io/github/kevincianfarini/cardiologist/SuspendingTests.kt
+++ b/src/commonTest/kotlin/io/github/kevincianfarini/cardiologist/SuspendingTests.kt
@@ -3,10 +3,7 @@ package io.github.kevincianfarini.cardiologist
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.test.runTest
 import kotlinx.coroutines.test.testTimeSource
-import kotlinx.datetime.DateTimePeriod
-import kotlinx.datetime.LocalDateTime
-import kotlinx.datetime.LocalTime
-import kotlinx.datetime.TimeZone
+import kotlinx.datetime.*
 import kotlin.test.Test
 import kotlin.test.assertEquals
 import kotlin.time.Duration.Companion.days
@@ -14,10 +11,9 @@ import kotlin.time.Duration.Companion.hours
 import kotlin.time.Duration.Companion.milliseconds
 import kotlin.time.Duration.Companion.minutes
 import kotlin.time.Duration.Companion.seconds
-import kotlin.time.ExperimentalTime
 import kotlin.time.measureTime
 
-@OptIn(ExperimentalTime::class, ExperimentalCoroutinesApi::class)
+@OptIn(ExperimentalCoroutinesApi::class)
 class SuspendingTests {
 
     @Test fun delays_for_proper_amount_of_time_until_future_instant() = runTest {
@@ -26,16 +22,48 @@ class SuspendingTests {
         val elapsed = testTimeSource.measureTime {
             testClock.delayUntil(future)
         }
-
         assertEquals(expected = 10.minutes, actual = elapsed)
     }
 
+    @Test fun delayUntil_considers_positive_time_drift() = runTest {
+        val instants = listOf(
+            Instant.fromEpochMilliseconds(0),
+            Instant.fromEpochSeconds(65),
+            Instant.fromEpochSeconds(120),
+        )
+        val clock = instants.asClock()
+        val elapsed = testTimeSource.measureTime {
+            clock.delayUntil(Instant.fromEpochSeconds(120))
+        }
+        // Below we assert that 115 seconds elapsed because the test time source is monotonic. It doesn't care about
+        // time drift on the Clock. The above setup implies that between the first and the second invocation of
+        // Clock.now, we experienced 5 seconds of positive time drift. This happens in scenarios where the device's
+        // clock runs slowly and NTP adjusts it forwards five seconds.
+        assertEquals(expected = 115.seconds, actual = elapsed)
+    }
+
+    @Test fun considers_negative_time_drift() = runTest {
+        val instants = listOf(
+            Instant.fromEpochMilliseconds(0),
+            Instant.fromEpochSeconds(55),
+            Instant.fromEpochSeconds(60),
+        )
+        val clock = instants.asClock()
+        val elapsed = testTimeSource.measureTime {
+            clock.delayUntil(Instant.fromEpochSeconds(60))
+        }
+        // Below we assert that 65 seconds elapsed because the test time source is monotonic. It doesn't care about
+        // time drift on the Clock. The above setup implies that between the first and the second invocation of
+        // Clock.now, we experienced 5 seconds of negative time drift. This happens in scenarios where the device's
+        // clock runs quickly and NTP adjusts it backwards five seconds.
+        assertEquals(expected = 65.seconds, actual = elapsed)
+    }
+
     @Test fun resumes_immediately_for_0_duration() = runTest {
         val now = testClock.now()
         val elapsed = testTimeSource.measureTime {
             testClock.delayUntil(now)
         }
-
         assertEquals(expected = 0.minutes, actual = elapsed)
     }
 
@@ -45,7 +73,6 @@ class SuspendingTests {
         val elapsed = testTimeSource.measureTime {
             testClock.delayUntil(past)
         }
-
         assertEquals(expected = 0.minutes, actual = elapsed)
     }
 
@@ -54,7 +81,6 @@ class SuspendingTests {
         val elapsed = testTimeSource.measureTime {
             testClock.delayFor(timePeriod, TimeZone.UTC)
         }
-
         assertEquals(
             expected = 10.minutes + 10.seconds + 10.milliseconds,
             actual = elapsed,
@@ -66,50 +92,46 @@ class SuspendingTests {
         val elapsed = testTimeSource.measureTime {
             testClock.delayFor(datePeriod, TimeZone.UTC)
         }
-
         assertEquals(expected = 1.days, actual = elapsed)
     }
 
     @Test fun delayUntilNext_simple_in_future_delays_for_proper_duration() = runTest {
         val tz = TimeZone.UTC
-        val clock = LocalDateTime(year = 2023, monthNumber = 10, dayOfMonth = 4, hour = 16, minute = 0).asClock(tz)
+        val localDt = LocalDateTime(year = 2023, monthNumber = 10, dayOfMonth = 4, hour = 16, minute = 0)
+        val clock = testClockAt(localDt, tz)
         val elapsed = testTimeSource.measureTime {
             clock.delayUntilNext(LocalTime(hour = 17, minute = 0), tz)
         }
-
-        assertEquals(
-            expected = 1.hours,
-            actual = elapsed
-        )
+        assertEquals(expected = 1.hours, actual = elapsed)
     }
 
     @Test fun delayUntilNext_simple_in_past_delays_until_tomorrow_duration() = runTest {
         val tz = TimeZone.UTC
-        val clock = LocalDateTime(year = 2023, monthNumber = 10, dayOfMonth = 4, hour = 16, minute = 0).asClock(tz)
+        val localDt = LocalDateTime(year = 2023, monthNumber = 10, dayOfMonth = 4, hour = 16, minute = 0)
+        val clock = testClockAt(localDt, tz)
         val elapsed = testTimeSource.measureTime {
             clock.delayUntilNext(LocalTime(hour = 14, minute = 0), tz)
         }
-
         assertEquals(expected = 22.hours, actual = elapsed)
     }
 
     @Test fun delayUntilNext_time_zone_switch_forwards_delays_proper_duration() = runTest {
         val tz = TimeZone.of("America/New_York")
-        val clock = LocalDateTime(year = 2023, monthNumber = 3, dayOfMonth = 12, hour = 1, minute = 0).asClock(tz)
+        val localDt = LocalDateTime(year = 2023, monthNumber = 3, dayOfMonth = 12, hour = 1, minute = 0)
+        val clock = testClockAt(localDt, tz)
         val elapsed = testTimeSource.measureTime {
             clock.delayUntilNext(LocalTime(hour = 2, minute = 30), tz)
         }
-
         assertEquals(expected = 1.hours + 30.minutes, actual = elapsed)
     }
 
     @Test fun delayUntilNext_time_zone_switch_backwards_delays_proper_duration() = runTest {
         val tz = TimeZone.of("America/New_York")
-        val clock = LocalDateTime(year = 2023, monthNumber = 11, dayOfMonth = 5, hour = 1, minute = 0).asClock(tz)
+        val localDt = LocalDateTime(year = 2023, monthNumber = 11, dayOfMonth = 5, hour = 1, minute = 0)
+        val clock = testClockAt(localDt, tz)
         val elapsed = testTimeSource.measureTime {
             clock.delayUntilNext(LocalTime(hour = 2, minute = 0), tz)
         }
-
         assertEquals(expected = 2.hours, actual = elapsed)
     }
 }
\ No newline at end of file