EFF-746 Fix Schedule.fixed immediate catch-up for long-running iterations (#1831)

Author: tim-smart

.changeset/eff-746-fixed-iteration-catchup.md

@@ -0,0 +1,5 @@
+---
+"effect": patch
+---
+
+Fix `Schedule.fixed` to run the next iteration immediately when the previous action takes longer than the configured interval.

packages/effect/src/Schedule.ts

@@ -2194,14 +2194,26 @@ export const fibonacci = (one: Duration.Input): Schedule<Duration.Duration> => {
  */
 export const fixed = (interval: Duration.Input): Schedule<number> => {
   const window = Duration.toMillis(Duration.fromInputUnsafe(interval))
-  return fromStepWithMetadata(effect.succeed((meta) =>
-    effect.succeed([
-      meta.attempt - 1,
-      window === 0
-        ? Duration.zero
-        : Duration.millis(window - (meta.elapsed % window))
-    ])
-  ))
+  return fromStepWithMetadata(effect.sync(() => {
+    let start = 0
+    let lastRun = 0
+    return (meta) =>
+      effect.sync(() => {
+        if (window === 0) {
+          return [meta.attempt - 1, Duration.zero] as const
+        }
+        if (meta.attempt === 1) {
+          start = meta.now
+          lastRun = meta.now + window
+          return [0, Duration.millis(window)] as const
+        }
+        const runningBehind = meta.now > (lastRun + window)
+        const boundary = window - ((meta.now - start) % window)
+        const delay = runningBehind ? 0 : boundary === 0 ? window : boundary
+        lastRun = runningBehind ? meta.now : meta.now + delay
+        return [meta.attempt - 1, Duration.millis(delay)] as const
+      })
+  }))
 }
 
 /**

packages/effect/test/Schedule.test.ts

@@ -241,6 +241,34 @@ describe("Schedule", () => {
           Duration.zero
         ])
       }))
+
+    it.effect("matches effect v3 when action duration exceeds the interval", () =>
+      Effect.gen(function*() {
+        const delays: Array<Duration.Duration> = []
+        const schedule = Schedule.fixed("1 seconds").pipe(
+          Schedule.while(({ attempt }) => Effect.succeed(attempt <= 5)),
+          Schedule.delays,
+          Schedule.map((delay) =>
+            Effect.sync(() => {
+              delays.push(delay)
+              return delays
+            })
+          )
+        )
+        yield* Effect.sleep("1.5 seconds").pipe(
+          Effect.schedule(schedule),
+          Effect.forkChild
+        )
+        yield* TestClock.setTime(Number.POSITIVE_INFINITY)
+        expect(delays).toEqual([
+          Duration.millis(1000),
+          Duration.zero,
+          Duration.zero,
+          Duration.zero,
+          Duration.zero,
+          Duration.zero
+        ])
+      }))
   })
 
   describe("windowed", () => {