Add Schedule.CurrentIterationMetadata Reference (#4828)

Co-authored-by: Tim <hello@timsmart.co>

Author: 2 people

.changeset/calm-geese-carry.md

@@ -0,0 +1,74 @@
+---
+"effect": minor
+---
+
+`Schedule.CurrentIterationMetadata` has been added
+
+```ts
+import { Effect, Schedule } from "effect"
+
+Effect.gen(function* () {
+  const currentIterationMetadata = yield* Schedule.CurrentIterationMetadata
+  //     ^? Schedule.IterationMetadata
+
+  console.log(currentIterationMetadata)
+}).pipe(Effect.repeat(Schedule.recurs(2)))
+// {
+//   elapsed: Duration.zero,
+//   elapsedSincePrevious: Duration.zero,
+//   input: undefined,
+//   now: 0,
+//   recurrence: 0,
+//   start: 0
+// }
+// {
+//   elapsed: Duration.zero,
+//   elapsedSincePrevious: Duration.zero,
+//   input: undefined,
+//   now: 0,
+//   recurrence: 1,
+//   start: 0
+// }
+// {
+//   elapsed: Duration.zero,
+//   elapsedSincePrevious: Duration.zero,
+//   input: undefined,
+//   now: 0,
+//   recurrence: 2,
+//   start: 0
+// }
+
+Effect.gen(function* () {
+  const currentIterationMetadata = yield* Schedule.CurrentIterationMetadata
+
+  console.log(currentIterationMetadata)
+}).pipe(
+  Effect.schedule(
+    Schedule.intersect(Schedule.fibonacci("1 second"), Schedule.recurs(3))
+  )
+)
+// {
+//   elapsed: Duration.zero,
+//   elapsedSincePrevious: Duration.zero,
+//   recurrence: 1,
+//   input: undefined,
+//   now: 0,
+//   start: 0
+// },
+// {
+//   elapsed: Duration.seconds(1),
+//   elapsedSincePrevious: Duration.seconds(1),
+//   recurrence: 2,
+//   input: undefined,
+//   now: 1000,
+//   start: 0
+// },
+// {
+//   elapsed: Duration.seconds(2),
+//   elapsedSincePrevious: Duration.seconds(1),
+//   recurrence: 3,
+//   input: undefined,
+//   now: 2000,
+//   start: 0
+// }
+```

packages/effect/src/Schedule.ts

@@ -14,6 +14,7 @@ import * as internal from "./internal/schedule.js"
 import type * as Option from "./Option.js"
 import type { Pipeable } from "./Pipeable.js"
 import type { Predicate } from "./Predicate.js"
+import type * as Ref from "./Ref.js"
 import type * as ScheduleDecision from "./ScheduleDecision.js"
 import type * as Intervals from "./ScheduleIntervals.js"
 import type * as Types from "./Types.js"
@@ -140,6 +141,7 @@ export declare namespace Schedule {
  */
 export interface ScheduleDriver<out Out, in In = unknown, out R = never> extends Schedule.DriverVariance<Out, In, R> {
   readonly state: Effect.Effect<unknown>
+  readonly iterationMeta: Ref.Ref<IterationMetadata>
   readonly last: Effect.Effect<Out, Cause.NoSuchElementException>
   readonly reset: Effect.Effect<void>
   next(input: In): Effect.Effect<Out, Option.Option<never>, R>
@@ -2184,3 +2186,33 @@ export const zipWith: {
     f: (out: Out, out2: Out2) => Out3
   ): Schedule<Out3, In & In2, R | R2>
 } = internal.zipWith
+
+/**
+ * @since 3.15.0
+ * @category models
+ */
+export interface CurrentIterationMetadata {
+  readonly _: unique symbol
+}
+
+/**
+ * @since 3.15.0
+ * @category models
+ */
+export interface IterationMetadata {
+  readonly input: unknown
+  readonly recurrence: number
+  readonly start: number
+  readonly now: number
+  readonly elapsed: Duration.Duration
+  readonly elapsedSincePrevious: Duration.Duration
+}
+
+/**
+ * @since 3.15.0
+ * @category models
+ */
+export const CurrentIterationMetadata: Context.Reference<
+  CurrentIterationMetadata,
+  IterationMetadata
+> = internal.CurrentIterationMetadata

packages/effect/src/internal/schedule.ts

@@ -45,6 +45,22 @@ export const ScheduleDriverTypeId: Schedule.ScheduleDriverTypeId = Symbol.for(
   ScheduleDriverSymbolKey
 ) as Schedule.ScheduleDriverTypeId
 
+/** @internal */
+const defaultIterationMetadata: Schedule.IterationMetadata = {
+  start: 0,
+  now: 0,
+  input: undefined,
+  elapsed: Duration.zero,
+  elapsedSincePrevious: Duration.zero,
+  recurrence: 0
+}
+
+/** @internal */
+export const CurrentIterationMetadata = Context.Reference<Schedule.CurrentIterationMetadata>()(
+  "effect/Schedule/CurrentIterationMetadata",
+  { defaultValue: () => defaultIterationMetadata }
+)
+
 const scheduleVariance = {
   /* c8 ignore next */
   _Out: (_: never) => _,
@@ -80,6 +96,31 @@ class ScheduleImpl<S, Out, In, R> implements Schedule.Schedule<Out, In, R> {
   }
 }
 
+/** @internal */
+const updateInfo = (
+  iterationMetaRef: Ref.Ref<Schedule.IterationMetadata>,
+  now: number,
+  input: unknown
+) =>
+  ref.update(iterationMetaRef, (prev) =>
+    (prev.recurrence === 0) ?
+      {
+        now,
+        input,
+        recurrence: prev.recurrence + 1,
+        elapsed: Duration.zero,
+        elapsedSincePrevious: Duration.zero,
+        start: now
+      } :
+      {
+        now,
+        input,
+        recurrence: prev.recurrence + 1,
+        elapsed: Duration.millis(now - prev.start),
+        elapsedSincePrevious: Duration.millis(now - prev.now),
+        start: prev.start
+      })
+
 /** @internal */
 class ScheduleDriverImpl<Out, In, R> implements Schedule.ScheduleDriver<Out, In, R> {
   [ScheduleDriverTypeId] = scheduleDriverVariance
@@ -106,8 +147,12 @@ class ScheduleDriverImpl<Out, In, R> implements Schedule.ScheduleDriver<Out, In,
     })
   }
 
+  iterationMeta = ref.unsafeMake(defaultIterationMetadata)
+
   get reset(): Effect.Effect<void> {
-    return ref.set(this.ref, [Option.none(), this.schedule.initial])
+    return ref.set(this.ref, [Option.none(), this.schedule.initial]).pipe(
+      core.zipLeft(ref.set(this.iterationMeta, defaultIterationMetadata))
+    )
   }
 
   next(input: In): Effect.Effect<Out, Option.Option<never>, R> {
@@ -122,15 +167,22 @@ class ScheduleDriverImpl<Out, In, R> implements Schedule.ScheduleDriver<Out, In,
               core.flatMap(([state, out, decision]) => {
                 const setState = ref.set(this.ref, [Option.some(out), state] as const)
                 if (ScheduleDecision.isDone(decision)) {
-                  return core.zipRight(setState, core.fail(Option.none()))
+                  return setState.pipe(
+                    core.zipRight(core.fail(Option.none()))
+                  )
                 }
                 const millis = Intervals.start(decision.intervals) - now
                 if (millis <= 0) {
-                  return core.as(setState, out)
+                  return setState.pipe(
+                    core.zipRight(updateInfo(this.iterationMeta, now, input)),
+                    core.as(out)
+                  )
                 }
+                const duration = Duration.millis(millis)
                 return pipe(
                   setState,
-                  core.zipRight(effect.sleep(Duration.millis(millis))),
+                  core.zipRight(updateInfo(this.iterationMeta, now, input)),
+                  core.zipRight(effect.sleep(duration)),
                   core.as(out)
                 )
               })
@@ -1874,7 +1926,22 @@ export const repeatOrElse_Effect = dual<
   core.flatMap(driver(schedule), (driver) =>
     core.matchEffect(self, {
       onFailure: (error) => orElse(error, Option.none()),
-      onSuccess: (value) => repeatOrElseEffectLoop(self, driver, orElse, value)
+      onSuccess: (value) =>
+        repeatOrElseEffectLoop(
+          effect.provideServiceEffect(
+            self,
+            CurrentIterationMetadata,
+            ref.get(driver.iterationMeta)
+          ),
+          driver,
+          (error, option) =>
+            effect.provideServiceEffect(
+              orElse(error, option),
+              CurrentIterationMetadata,
+              ref.get(driver.iterationMeta)
+            ),
+          value
+        )
     })))
 
 /** @internal */
@@ -1883,16 +1950,15 @@ const repeatOrElseEffectLoop = <A, E, R, R1, B, C, E2, R2>(
   driver: Schedule.ScheduleDriver<B, A, R1>,
   orElse: (error: E, option: Option.Option<B>) => Effect.Effect<C, E2, R2>,
   value: A
-): Effect.Effect<B | C, E2, R | R1 | R2> => {
-  return core.matchEffect(driver.next(value), {
+): Effect.Effect<B | C, E2, R | R1 | R2> =>
+  core.matchEffect(driver.next(value), {
     onFailure: () => core.orDie(driver.last),
     onSuccess: (b) =>
       core.matchEffect(self, {
         onFailure: (error) => orElse(error, Option.some(b)),
         onSuccess: (value) => repeatOrElseEffectLoop(self, driver, orElse, value)
       })
   })
-}
 
 /** @internal */
 export const retry_Effect = dual<
@@ -1970,7 +2036,21 @@ export const retryOrElse_Effect = dual<
 >(3, (self, policy, orElse) =>
   core.flatMap(
     driver(policy),
-    (driver) => retryOrElse_EffectLoop(self, driver, orElse)
+    (driver) =>
+      retryOrElse_EffectLoop(
+        effect.provideServiceEffect(
+          self,
+          CurrentIterationMetadata,
+          ref.get(driver.iterationMeta)
+        ),
+        driver,
+        (e, out) =>
+          effect.provideServiceEffect(
+            orElse(e, out),
+            CurrentIterationMetadata,
+            ref.get(driver.iterationMeta)
+          )
+      )
   ))
 
 /** @internal */
@@ -2022,7 +2102,16 @@ export const scheduleFrom_Effect = dual<
 >(3, (self, initial, schedule) =>
   core.flatMap(
     driver(schedule),
-    (driver) => scheduleFrom_EffectLoop(self, initial, driver)
+    (driver) =>
+      scheduleFrom_EffectLoop(
+        effect.provideServiceEffect(
+          self,
+          CurrentIterationMetadata,
+          ref.get(driver.iterationMeta)
+        ),
+        initial,
+        driver
+      )
   ))
 
 /** @internal */
@@ -2033,7 +2122,11 @@ const scheduleFrom_EffectLoop = <In, E, R, R2, Out>(
 ): Effect.Effect<Out, E, R | R2> =>
   core.matchEffect(driver.next(initial), {
     onFailure: () => core.orDie(driver.last),
-    onSuccess: () => core.flatMap(self, (a) => scheduleFrom_EffectLoop(self, a, driver))
+    onSuccess: () =>
+      core.flatMap(
+        self,
+        (a) => scheduleFrom_EffectLoop(self, a, driver)
+      )
   })
 
 /** @internal */

packages/effect/src/internal/stream.ts

@@ -5382,8 +5382,13 @@ export const repeatWith = dual<
     return pipe(
       Schedule.driver(schedule),
       Effect.map((driver) => {
+        const provideLastIterationInfo = provideServiceEffect(
+          Schedule.CurrentIterationMetadata,
+          Ref.get(driver.iterationMeta)
+        )
+
         const scheduleOutput = pipe(driver.last, Effect.orDie, Effect.map(options.onSchedule))
-        const process = pipe(self, map(options.onElement), toChannel)
+        const process = pipe(self, provideLastIterationInfo, map(options.onElement), toChannel)
         const loop: Channel.Channel<Chunk.Chunk<C>, unknown, E, unknown, void, unknown, R | R2> = channel.unwrap(
           Effect.match(driver.next(void 0), {
             onFailure: () => core.void,
@@ -5419,15 +5424,21 @@ export const repeatEffectWithSchedule = <A, E, R, X, A0 extends A, R2>(
 ): Stream.Stream<A, E, R | R2> =>
   flatMap(
     fromEffect(Effect.zip(effect, Schedule.driver(schedule))),
-    ([a, driver]) =>
-      concat(
+    ([a, driver]) => {
+      const provideLastIterationInfo = Effect.provideServiceEffect(
+        Schedule.CurrentIterationMetadata,
+        Ref.get(driver.iterationMeta)
+      )
+      return concat(
         succeed(a),
         unfoldEffect(a, (s) =>
           Effect.matchEffect(driver.next(s as A0), {
             onFailure: Effect.succeed,
-            onSuccess: () => Effect.map(effect, (nextA) => Option.some([nextA, nextA] as const))
+            onSuccess: () =>
+              Effect.map(provideLastIterationInfo(effect), (nextA) => Option.some([nextA, nextA] as const))
           }))
       )
+    }
   )
 
 /** @internal */
@@ -5447,6 +5458,11 @@ export const retry = dual<
   ): Stream.Stream<A, E, R | R2> =>
     Schedule.driver(schedule).pipe(
       Effect.map((driver) => {
+        const provideLastIterationInfo = provideServiceEffect(
+          Schedule.CurrentIterationMetadata,
+          Ref.get(driver.iterationMeta)
+        )
+
         const loop: Channel.Channel<
           Chunk.Chunk<A>,
           unknown,
@@ -5455,7 +5471,7 @@ export const retry = dual<
           unknown,
           unknown,
           R | R2
-        > = toChannel(self).pipe(
+        > = toChannel(provideLastIterationInfo(self)).pipe(
           channel.mapOutEffect((out) => Effect.as(driver.reset, out)),
           channel.catchAll((error) =>
             driver.next(error as E0).pipe(