Repository: kevincianfarini/cardiologist (kotlin)

Issue:
Consider removing `occurred: Instant` lambda parameters from `Pulse.beat` and `Clock.executeAt`
These values might be misleading if someone is blocking the event loop. Blocking the event loop is obviously bad, but we won't want people using these `occurred` values in their health checks if they're possibly not accurate. Plus, they provide very little value to the end user since they can just do this instead. 

```kt
clock.fixedPeriodPulse(10.seconds) { scheduled -> 
  val occurred = clock.now()
  ...
}

clock.executeAt(...) {
  val occurred = clock.now()
  ...
}
```

This would also allow us to remove the parameter-less `Pulse.beat` function since you can implicitly omit a single lambda parameter versus needing to do `_, _ ->` everywhere.

Pull request description:
Closes #121

Patch:
diff --git a/README.md b/README.md
index 109b99f..2cd28cd 100644
--- a/README.md
+++ b/README.md
@@ -6,8 +6,8 @@ Build job schedules with [kotlinx-datetime](https://github.com/Kotlin/kotlinx-da
 
 ```kt
 val tz = TimeZone.of("America/New_York")
-Clock.System.schedulePulse(timeZone = tz, atSecond = 0).beat { scheduled, occurred ->
-    println("A pulse was scheduled in $tz for ${scheduled.toLocalDateTime(tz)} and occurred at ${occurred.toLocalDateTime(tz)}.")
+Clock.System.schedulePulse(timeZone = tz, atSecond = 0).beat { scheduled ->
+    println("A pulse was scheduled in $tz for ${scheduled.toLocalDateTime(tz)} and occurred at ${Clock.System.now().toLocalDateTime(tz)}.")
 }
 ```
 
@@ -88,12 +88,12 @@ val cronPulse: Pulse = clock.schedulePulse(cronExpression = "* * * * *", timeZon
 
 #### Executing jobs
 
-Pulses alone don't do any work. You need to beat a pulse for it to do something! The lambda exposes two `Instant` 
-parameters denoting when a pulse was scheduled and when it actually occurred. 
+Pulses alone don't do any work. You need to beat a pulse for it to do something! The lambda exposes the `Instant` 
+denoting when a pulse was scheduled to occur. 
 
 ```kt
-clock.fixedPeriodPulse(30.seconds).beat { scheduled, occurred ->
-    println("My job was scheduled for $scheduled and occurred at $occurred.")
+clock.fixedPeriodPulse(30.seconds).beat { scheduled ->
+    println("My job was scheduled for $scheduled and occurred at ${clock.now()}.")
 }
 ```
 
@@ -122,14 +122,14 @@ Cardiologist provides `Clock.executeAt` which executes an action at a specific `
 
 ```kt
 val instant = Instant.parse("2025-05-10T13:15:00Z")
-clock.executeAt(instant) { occurred: Instant ->
-    println("I executed at $occurred!")
+clock.executeAt(instant) { 
+    println("I executed at ${clock.now()}!")
 }
 
 val localDateTime = LocalDatetime(2025, 5, 10, 13, 15, 0)
 val timeZone = TimeZone.of("America/New_York")
-clock.executeAt(localDateTime, timeZone) { occurred: LocalDateTime ->
-    println("I executed at $occurred!")
+clock.executeAt(localDateTime, timeZone) { 
+    println("I executed at ${clock.now().toLocalDateTime(timeZone)}!")
 }
 ```
 
@@ -142,7 +142,7 @@ like "My job which started executing at 2025-05-07T15:24:00"
 ```kt
 val scheduleName = CoroutineName("5-second Pulse schedule")
 withContext(scheduleName) {
-    clock.fixedPeriodPulse(5.seconds).beat(strategy = PulseBackpressureStrategy.CancelPrevious) { scheduled, _ ->
+    clock.fixedPeriodPulse(5.seconds).beat(strategy = PulseBackpressureStrategy.CancelPrevious) { scheduled ->
         val jobName = CoroutineName("My job which started executing at $scheduled")
         withContext(jobName) { myJob() }
     }
diff --git a/src/commonMain/kotlin/io/github/kevincianfarini/cardiologist/Pulse.kt b/src/commonMain/kotlin/io/github/kevincianfarini/cardiologist/Pulse.kt
index 5215f57..4900797 100644
--- a/src/commonMain/kotlin/io/github/kevincianfarini/cardiologist/Pulse.kt
+++ b/src/commonMain/kotlin/io/github/kevincianfarini/cardiologist/Pulse.kt
@@ -14,7 +14,7 @@ import kotlin.jvm.JvmInline
  * A [Pulse] is a cadence which informs consumers when to execute work by calling [Pulse.beat].
  */
 @JvmInline
-public value class Pulse internal constructor(private val flow: Flow<Pair<Instant, Instant>>) {
+public value class Pulse internal constructor(private val flow: Flow<Instant>) {
 
     /**
      * Returns a pulse that beats [count] times.
@@ -26,41 +26,27 @@ public value class Pulse internal constructor(private val flow: Flow<Pair<Instan
     /**
      * Returns a pulse that beats while [predicate] is satisfied.
      */
-    public fun takeWhile(predicate: (Instant, Instant) -> Boolean): Pulse = Pulse(
-        flow.takeWhile { (scheduled, occurred) -> predicate(scheduled, occurred) }
+    public fun takeWhile(predicate: (scheduled: Instant) -> Boolean): Pulse = Pulse(
+        flow.takeWhile(predicate)
     )
 
     /**
-     * Invoke [action] every time this Pulse is set to execute. [Action][action] provides two [instants][Instant]
-     * denoting when the pulse was scheduled to occur, and when it actually occurred.
+     * Invoke [action] every time this Pulse is set to execute. [Action][action] provides an [Instant] denoting when the
+     * pulse was scheduled to occur.
      *
      * This operator will execute [action] according to which [strategy] is specified.
      */
     public suspend fun beat(
         strategy: PulseBackpressureStrategy = PulseBackpressureStrategy.ExecuteConcurrently,
-        action: suspend (scheduled: Instant, occurred: Instant) -> Unit,
+        action: suspend (scheduled: Instant) -> Unit,
     ): Unit = when (strategy) {
-        PulseBackpressureStrategy.CancelPrevious -> flow.collectLatest { (scheduled, occurred) ->
-            action(scheduled, occurred)
-        }
+        PulseBackpressureStrategy.CancelPrevious -> flow.collectLatest(action)
         PulseBackpressureStrategy.ExecuteConcurrently -> coroutineScope {
-            flow.collect { (scheduled, occurred) ->
-                launch { action(scheduled, occurred) }
+            flow.collect { scheduled ->
+                launch { action(scheduled) }
             }
         }
-        PulseBackpressureStrategy.SkipNext -> flow.collectCurrent { (scheduled, occurred) ->
-            action(scheduled, occurred)
-        }
+        PulseBackpressureStrategy.SkipNext -> flow.collectCurrent(action)
         else -> error("Impossible.")
     }
-
-    /**
-     * Invoke [action] every time this Pulse is set to execute.
-     *
-     * This operator will execute [action] according to which [strategy] is specified.
-     */
-    public suspend fun beat(
-        strategy: PulseBackpressureStrategy = PulseBackpressureStrategy.ExecuteConcurrently,
-        action: suspend () -> Unit,
-    ): Unit = beat(strategy) { _, _ -> action() }
 }
\ No newline at end of file
diff --git a/src/commonMain/kotlin/io/github/kevincianfarini/cardiologist/schedule.kt b/src/commonMain/kotlin/io/github/kevincianfarini/cardiologist/schedule.kt
index ddd4997..b399124 100644
--- a/src/commonMain/kotlin/io/github/kevincianfarini/cardiologist/schedule.kt
+++ b/src/commonMain/kotlin/io/github/kevincianfarini/cardiologist/schedule.kt
@@ -25,7 +25,7 @@ public fun Clock.fixedPeriodPulse(period: Duration): Pulse {
         var nextPulse: Instant = now() + period
         while (true) {
             delayUntil(nextPulse)
-            emit(Pair(nextPulse, now()))
+            emit(nextPulse)
             nextPulse += period
         }
     }
@@ -44,7 +44,7 @@ public fun Clock.fixedPeriodPulse(period: DateTimePeriod, timeZone: TimeZone): P
         var nextPulse: Instant = now().plus(period, timeZone)
         while (true) {
             delayUntil(nextPulse)
-            emit(Pair(nextPulse, now()))
+            emit(nextPulse)
             nextPulse = nextPulse.plus(period, timeZone)
         }
     }
@@ -98,7 +98,7 @@ public fun Clock.schedulePulse(
                 inMonths = inMonth?.let { it..it } ?: Month.JANUARY..Month.DECEMBER,
             )
             delayUntil(nextPulse, timeZone)
-            emit(Pair(nextPulse.toInstant(timeZone), now()))
+            emit(nextPulse.toInstant(timeZone))
             lastPulse = nextPulse
         }
     }
diff --git a/src/commonMain/kotlin/io/github/kevincianfarini/cardiologist/suspend.kt b/src/commonMain/kotlin/io/github/kevincianfarini/cardiologist/suspend.kt
index bdfbf29..fd49e69 100644
--- a/src/commonMain/kotlin/io/github/kevincianfarini/cardiologist/suspend.kt
+++ b/src/commonMain/kotlin/io/github/kevincianfarini/cardiologist/suspend.kt
@@ -7,22 +7,19 @@ import kotlinx.datetime.Instant
 import kotlinx.datetime.LocalDateTime
 import kotlinx.datetime.TimeZone
 import kotlinx.datetime.toInstant
-import kotlinx.datetime.toLocalDateTime
 
 /**
  * Execute the provided [action] at or after [instant].
  *
  * The coroutine will be suspended until [instant] without blocking the thread. If [Clock.now] returns an [Instant]
  * greater than the supplied [instant], this function executes [action] without suspending.
- *
- * The supplied [action] is invoked with an [Instant] parameter specifying when the action was actually executed.
  */
 public suspend fun <T> Clock.executeAt(
     instant: Instant,
-    action: suspend (occurred: Instant) -> T,
+    action: suspend () -> T,
 ): T {
     delayUntil(instant)
-    return action(now())
+    return action()
 }
 
 /**
@@ -32,17 +29,16 @@ public suspend fun <T> Clock.executeAt(
  * [Instant] greater than the [Instant] associated with [dateTime] in [timeZone], this function executes [action] without
  * suspending.
  *
- * The supplied [action] is invoked with a [LocalDateTime] parameter specifying the instant that [action] was executed
- * in [timeZone]. Local time conversion is sometimes ambiguous, and therefore it's recommended to schedule pulses in a
+ * Local time conversion is sometimes ambiguous, and therefore it's recommended to schedule execution in a
  * fixed UTC offset timezone. See [LocalDateTime.toInstant] for more details.
  */
 public suspend fun <T> Clock.executeAt(
     dateTime: LocalDateTime,
     timeZone: TimeZone,
-    action: suspend (occurred: LocalDateTime) -> T,
+    action: suspend () -> T,
 ): T {
     delayUntil(dateTime, timeZone)
-    return action(now().toLocalDateTime(timeZone))
+    return action()
 }
 
 /**


Test patch:
diff --git a/src/commonTest/kotlin/io/github/kevincianfarini/cardiologist/PulseTests.kt b/src/commonTest/kotlin/io/github/kevincianfarini/cardiologist/PulseTests.kt
index 87ee90e..6d2ef1d 100644
--- a/src/commonTest/kotlin/io/github/kevincianfarini/cardiologist/PulseTests.kt
+++ b/src/commonTest/kotlin/io/github/kevincianfarini/cardiologist/PulseTests.kt
@@ -57,9 +57,9 @@ class PulseTests {
             Instant.fromEpochSeconds(120), // Second emissions in intervalPulse.
         ).asClock()
         val pulse = clock.fixedPeriodPulse(60.seconds)
-        pulse.take(2).beat { scheduled, occurred ->
+        pulse.take(2).beat { scheduled ->
             assertEquals(expected = scheduledInstants[index], scheduled)
-            assertEquals(expected = occurredInstants[index++], occurred)
+            assertEquals(expected = occurredInstants[index++], clock.now())
         }
     }
 
@@ -86,9 +86,9 @@ class PulseTests {
             Instant.fromEpochSeconds(120), // Second emissions in intervalPulse.
         ).asClock()
         val pulse = clock.fixedPeriodPulse(60.seconds)
-        pulse.take(2).beat { scheduled, occurred ->
+        pulse.take(2).beat { scheduled ->
             assertEquals(expected = scheduledInstants[index], scheduled)
-            assertEquals(expected = occurredInstants[index++], occurred)
+            assertEquals(expected = occurredInstants[index++], clock.now())
         }
     }
 
@@ -113,9 +113,9 @@ class PulseTests {
             Instant.fromEpochSeconds(120), // Second emissions in intervalPulse.
         ).asClock()
         val pulse = clock.fixedPeriodPulse(DateTimePeriod(minutes = 1), timeZone = TimeZone.UTC)
-        pulse.take(2).beat { scheduled, occurred ->
+        pulse.take(2).beat { scheduled ->
             assertEquals(expected = scheduledInstants[index], scheduled)
-            assertEquals(expected = occurredInstants[index++], occurred)
+            assertEquals(expected = occurredInstants[index++], clock.now())
         }
     }
 
@@ -142,9 +142,9 @@ class PulseTests {
             Instant.fromEpochSeconds(120), // Second emissions in intervalPulse.
         ).asClock()
         val pulse = clock.fixedPeriodPulse(DateTimePeriod(minutes = 1), timeZone = TimeZone.UTC)
-        pulse.take(2).beat { scheduled, occurred ->
+        pulse.take(2).beat { scheduled ->
             assertEquals(expected = scheduledInstants[index], scheduled)
-            assertEquals(expected = occurredInstants[index++], occurred)
+            assertEquals(expected = occurredInstants[index++], clock.now())
         }
     }
 
@@ -157,9 +157,9 @@ class PulseTests {
             Instant.fromEpochSeconds(65), // During emission in intervalPulse
         ).asClock()
         val pulse = clock.schedulePulse(timeZone = TimeZone.UTC, atMinute = 1, atSecond = 0)
-        pulse.take(1).beat { scheduled, occurred ->
+        pulse.take(1).beat { scheduled ->
             assertEquals(expected = Instant.fromEpochSeconds(60), scheduled)
-            assertEquals(expected = Instant.fromEpochSeconds(65), occurred)
+            assertEquals(expected = Instant.fromEpochSeconds(65), clock.now())
         }
     }
 
@@ -188,9 +188,9 @@ class PulseTests {
             Instant.fromEpochSeconds(120),
         ).asClock()
         val pulse = clock.schedulePulse(timeZone = TimeZone.UTC, atSecond = 0)
-        pulse.take(2).beat { scheduled, occurred ->
+        pulse.take(2).beat { scheduled ->
             assertEquals(expected = scheduledInstants[index], scheduled)
-            assertEquals(expected = occurredInstants[index++], occurred)
+            assertEquals(expected = occurredInstants[index++], clock.now())
         }
     }
 
@@ -217,7 +217,7 @@ class PulseTests {
         val list = MutableStateFlow<List<Instant>>(emptyList())
         backgroundScope.launch {
             val pulse = testClock.fixedPeriodPulse(1.seconds).take(3)
-            pulse.beat(PulseBackpressureStrategy.CancelPrevious) { scheduled, _ ->
+            pulse.beat(PulseBackpressureStrategy.CancelPrevious) { scheduled ->
                 try {
                     list.update { l -> l + scheduled }
                     awaitCancellation()
@@ -238,7 +238,7 @@ class PulseTests {
         val list = MutableStateFlow<List<Instant>>(emptyList())
         backgroundScope.launch {
             val pulse = testClock.fixedPeriodPulse(1.seconds).take(3)
-            pulse.beat(PulseBackpressureStrategy.SkipNext) { scheduled, _ ->
+            pulse.beat(PulseBackpressureStrategy.SkipNext) { scheduled ->
                 try {
                     list.update { l -> l + scheduled }
                     awaitCancellation()
diff --git a/src/commonTest/kotlin/io/github/kevincianfarini/cardiologist/SuspendingTests.kt b/src/commonTest/kotlin/io/github/kevincianfarini/cardiologist/SuspendingTests.kt
index c8fc678..f97f2cd 100644
--- a/src/commonTest/kotlin/io/github/kevincianfarini/cardiologist/SuspendingTests.kt
+++ b/src/commonTest/kotlin/io/github/kevincianfarini/cardiologist/SuspendingTests.kt
@@ -83,8 +83,8 @@ class SuspendingTests {
     fun executeAt_invokes_with_correct_occurred_instant() = runTest {
         val tenSeconds = Instant.fromEpochSeconds(10)
         val duration = testTimeSource.measureTime {
-            testClock.executeAt(tenSeconds) { occurred ->
-                assertEquals(expected = tenSeconds, actual = occurred)
+            testClock.executeAt(tenSeconds) {
+                assertEquals(expected = tenSeconds, actual = testClock.now())
             }
         }
         assertEquals(expected = 10.seconds, actual = duration)
@@ -95,8 +95,8 @@ class SuspendingTests {
         val localDateTime = LocalDateTime(1970, 1, 2, 0, 0)
         val tz = TimeZone.UTC
         val duration = testTimeSource.measureTime {
-            testClock.executeAt(localDateTime, tz) { occurred ->
-                assertEquals(expected = localDateTime, actual = occurred)
+            testClock.executeAt(localDateTime, tz) {
+                assertEquals(expected = localDateTime, actual = testClock.now().toLocalDateTime(tz))
             }
         }
         assertEquals(expected = 24.hours, actual = duration)