1import { assert, describe, expect, it } from "@effect/vitest"
2import { Array, Deferred, Duration, Effect, Fiber, Pull, Random, Result, Schedule } from "effect"
3import { constant, constUndefined } from "effect/Function"
4import { TestClock } from "effect/testing"
5
6describe("Schedule", () => {
7 describe("constructors", () => {
8 it.effect("during recurs while the duration has not elapsed", () =>
9 Effect.gen(function*() {
10 const step = yield* Schedule.toStep(Schedule.during("1 second"))
11 const completed = yield* Pull.catchDone(
12 step(0, undefined).pipe(Effect.as(false)),
13 () => Effect.succeed(true)
14 )
15
16 assert.isFalse(completed)
17 }))
18 })
19
20 describe("combining", () => {
21 it.effect("max - outputs the slowest schedule duration", () =>
22 Effect.gen(function*() {
23 const schedule = Schedule.max([
24 Schedule.fixed("5 seconds"),
25 Schedule.exponential("5 seconds"),
26 Schedule.spaced("10 seconds")
27 ])
28 const inputs = Array.makeBy(3, constUndefined)
29 const outputs = yield* runDelays(schedule, inputs)
30 assert.deepStrictEqual(outputs, [
31 Duration.seconds(10),
32 Duration.seconds(10),
33 Duration.seconds(20)
34 ])
35 }))
36
37 it.effect("max - stops when any schedule completes", () =>
38 Effect.gen(function*() {
39 const schedule = Schedule.max([
40 Schedule.duration("1 second"),
41 Schedule.spaced("5 seconds")
42 ])
43 const inputs = Array.makeBy(3, constUndefined)
44 const outputs = yield* runDelays(schedule, inputs)
45 assert.deepStrictEqual(outputs, [
46 Duration.seconds(5),
47 Duration.zero
48 ])
49 }))
50
51 it.effect("min - outputs the fastest schedule duration", () =>
52 Effect.gen(function*() {
53 const schedule = Schedule.min([
54 Schedule.fixed("5 seconds"),
55 Schedule.exponential("5 seconds"),
56 Schedule.spaced("10 seconds")
57 ])
58 const inputs = Array.makeBy(3, constUndefined)
59 const outputs = yield* runDelays(schedule, inputs)
60 assert.deepStrictEqual(outputs, [
61 Duration.seconds(5),
62 Duration.seconds(5),
63 Duration.seconds(5)
64 ])
65 }))
66
67 it.effect("min - continues after a schedule completes", () =>
68 Effect.gen(function*() {
69 const schedule = Schedule.min([
70 Schedule.duration("1 second"),
71 Schedule.spaced("5 seconds")
72 ])
73 const inputs = Array.makeBy(3, constUndefined)
74 const outputs = yield* runDelays(schedule, inputs)
75 assert.deepStrictEqual(outputs, [
76 Duration.seconds(1),
77 Duration.seconds(5),
78 Duration.seconds(5)
79 ])
80 }))
81 })
82
83 describe("sequencing", () => {
84 it.effect("concat - sequences self then other and merges their outputs", () =>
85 Effect.gen(function*() {
86 const left = Schedule.identity<string>().pipe(
87 Schedule.upTo({ times: 2 }),
88 Schedule.map(({ output }) => `left:${output}`)
89 )
90 const right = Schedule.identity<string>().pipe(
91 Schedule.map(({ output }) => `right:${output}`)
92 )
93 const step = yield* Schedule.toStep(Schedule.concat(left, right))
94
95 const first = yield* step(0, "a")
96 const second = yield* step(0, "b")
97 const third = yield* step(0, "c")
98
99 assert.deepStrictEqual([first, second, third], [
100 ["left:a", Duration.zero],
101 ["left:b", Duration.zero],
102 ["right:c", Duration.zero]
103 ])
104 }))
105
106 it.effect("tap - provides full metadata", () =>
107 Effect.gen(function*() {
108 const observed: Array<Schedule.Metadata<number, string>> = []
109 const schedule = Schedule.spaced(Duration.millis(250)).pipe(
110 Schedule.tap((metadata: Schedule.Metadata<number, string>) =>
111 Effect.sync(() => {
112 observed.push(metadata)
113 })
114 )
115 )
116 const step = yield* Schedule.toStep(schedule)
117 const first = yield* step(1_000, "a")
118 const second = yield* step(1_250, "b")
119
120 expect(first).toEqual([0, Duration.millis(250)])
121 expect(second).toEqual([1, Duration.millis(250)])
122 expect(observed).toEqual([
123 {
124 input: "a",
125 output: 0,
126 duration: Duration.millis(250),
127 attempt: 1,
128 start: 1_000,
129 now: 1_000,
130 elapsed: 0,
131 elapsedSincePrevious: 0
132 },
133 {
134 input: "b",
135 output: 1,
136 duration: Duration.millis(250),
137 attempt: 2,
138 start: 1_000,
139 now: 1_250,
140 elapsed: 250,
141 elapsedSincePrevious: 250
142 }
143 ])
144 }))
145
146 it.effect("modifyDelay - provides full metadata", () =>
147 Effect.gen(function*() {
148 const observed: Array<Schedule.Metadata<number, string>> = []
149 const schedule = Schedule.spaced(Duration.millis(250)).pipe(
150 Schedule.modifyDelay((metadata: Schedule.Metadata<number, string>) =>
151 Effect.sync(() => {
152 observed.push(metadata)
153 return Duration.sum(metadata.duration, Duration.millis(metadata.elapsedSincePrevious))
154 })
155 )
156 )
157 const step = yield* Schedule.toStep(schedule)
158 const first = yield* step(1_000, "a")
159 const second = yield* step(1_250, "b")
160
161 assert.deepStrictEqual(first, [0, Duration.millis(250)])
162 assert.deepStrictEqual(second, [1, Duration.millis(500)])
163 assert.deepStrictEqual(observed, [
164 {
165 input: "a",
166 output: 0,
167 duration: Duration.millis(250),
168 attempt: 1,
169 start: 1_000,
170 now: 1_000,
171 elapsed: 0,
172 elapsedSincePrevious: 0
173 },
174 {
175 input: "b",
176 output: 1,
177 duration: Duration.millis(250),
178 attempt: 2,
179 start: 1_000,
180 now: 1_250,
181 elapsed: 250,
182 elapsedSincePrevious: 250
183 }
184 ])
185 }))
186
187 it.effect("addDelay - provides full metadata", () =>
188 Effect.gen(function*() {
189 const observed: Array<Schedule.Metadata<number, string>> = []
190 const schedule = Schedule.spaced(Duration.millis(250)).pipe(
191 Schedule.addDelay((metadata: Schedule.Metadata<number, string>) =>
192 Effect.sync(() => {
193 observed.push(metadata)
194 return Duration.millis(metadata.elapsedSincePrevious)
195 })
196 )
197 )
198 const step = yield* Schedule.toStep(schedule)
199 const first = yield* step(1_000, "a")
200 const second = yield* step(1_250, "b")
201
202 assert.deepStrictEqual(first, [0, Duration.millis(250)])
203 assert.deepStrictEqual(second, [1, Duration.millis(500)])
204 assert.deepStrictEqual(observed, [
205 {
206 input: "a",
207 output: 0,
208 duration: Duration.millis(250),
209 attempt: 1,
210 start: 1_000,
211 now: 1_000,
212 elapsed: 0,
213 elapsedSincePrevious: 0
214 },
215 {
216 input: "b",
217 output: 1,
218 duration: Duration.millis(250),
219 attempt: 2,
220 start: 1_000,
221 now: 1_250,
222 elapsed: 250,
223 elapsedSincePrevious: 250
224 }
225 ])
226 }))
227
228 it.effect("concatResult - sequences self then other when collecting delays", () =>
229 Effect.gen(function*() {
230 const left = Schedule.fixed("500 millis").pipe(
231 Schedule.while(({ attempt }) => Effect.succeed(attempt <= 3))
232 )
233 const right = Schedule.fixed("1 second")
234 const schedule = Schedule.concatResult(left, right)
235 const inputs = Array.makeBy(6, constUndefined)
236 const outputs = yield* runDelays(schedule, inputs)
237 expect(outputs).toEqual([
238 Duration.millis(500),
239 Duration.millis(500),
240 Duration.millis(500),
241 Duration.seconds(1),
242 Duration.seconds(1),
243 Duration.seconds(1)
244 ])
245 }))
246
247 it.effect("concatResult - includes finite other completion when collecting delays", () =>
248 Effect.gen(function*() {
249 const left = Schedule.fixed("500 millis").pipe(
250 Schedule.while(({ attempt }) => Effect.succeed(attempt <= 2))
251 )
252 const right = Schedule.duration("1 second")
253 const schedule = Schedule.concatResult(left, right)
254 const inputs = Array.makeBy(5, constUndefined)
255 const outputs = yield* runDelays(schedule, inputs)
256 expect(outputs).toEqual([
257 Duration.millis(500),
258 Duration.millis(500),
259 Duration.seconds(1),
260 Duration.zero
261 ])
262 }))
263
264 it.effect("concatResult - wraps self outputs as Failure and other outputs as Success", () =>
265 Effect.gen(function*() {
266 const left = Schedule.identity<string>().pipe(Schedule.upTo({ times: 2 }))
267 const right = Schedule.identity<string>()
268 const step = yield* Schedule.toStep(Schedule.concatResult(left, right))
269
270 const first = yield* step(0, "left-1")
271 const second = yield* step(0, "left-2")
272 const third = yield* step(0, "right-1")
273
274 assert.deepStrictEqual([first, second, third], [
275 [Result.fail("left-1"), Duration.zero],
276 [Result.fail("left-2"), Duration.zero],
277 [Result.succeed("right-1"), Duration.zero]
278 ])
279 }))
280 })
281
282 describe("cron", () => {
283 it.effect("should recur on interval matching cron expression", () =>
284 Effect.gen(function*() {
285 const now = new Date(2024, 0, 1, 0, 0, 35).getTime()
286 // At every second minute
287 const schedule = Schedule.cron("*/2 * * * *")
288 const inputs = Array.makeBy(4, constUndefined)
289 yield* TestClock.setTime(now)
290 const [, outputs] = yield* runDelays(schedule, inputs).pipe(
291 Effect.map(Array.mapAccum(now, (next, delay) => {
292 const timestamp = next + Duration.toMillis(delay)
293 return [timestamp, format(timestamp)]
294 }))
295 )
296 expect(outputs).toEqual([
297 "Mon Jan 01 2024 00:02:00",
298 "Mon Jan 01 2024 00:04:00",
299 "Mon Jan 01 2024 00:06:00",
300 "Mon Jan 01 2024 00:08:00"
301 ])
302 }))
303
304 it.effect("should recur on interval matching cron expression (second granularity)", () =>
305 Effect.gen(function*() {
306 const now = new Date(2024, 0, 1, 0, 0, 0).getTime()
307 // At every third minute
308 const schedule = Schedule.cron("*/3 * * * * *")
309 const inputs = Array.makeBy(10, constUndefined)
310 yield* TestClock.setTime(now)
311 const [, outputs] = yield* runDelays(schedule, inputs).pipe(
312 Effect.map(Array.mapAccum(now, (next, delay) => {
313 const timestamp = next + Duration.toMillis(delay)
314 return [timestamp, format(timestamp)]
315 }))
316 )
317 expect(outputs).toEqual([
318 "Mon Jan 01 2024 00:00:03",
319 "Mon Jan 01 2024 00:00:06",
320 "Mon Jan 01 2024 00:00:09",
321 "Mon Jan 01 2024 00:00:12",
322 "Mon Jan 01 2024 00:00:15",
323 "Mon Jan 01 2024 00:00:18",
324 "Mon Jan 01 2024 00:00:21",
325 "Mon Jan 01 2024 00:00:24",
326 "Mon Jan 01 2024 00:00:27",
327 "Mon Jan 01 2024 00:00:30"
328 ])
329 }))
330
331 it.effect("should recur at time matching cron expression", () =>
332 Effect.gen(function*() {
333 const now = new Date(2024, 0, 1, 0, 0, 0).getTime()
334 // At 04:30 on day-of-month 5 and 15 and on Wednesday.
335 const schedule = Schedule.cron("30 4 5,15 * WED")
336 const inputs = Array.makeBy(6, constUndefined)
337 yield* TestClock.setTime(now)
338 const [, outputs] = yield* runDelays(schedule, inputs).pipe(
339 Effect.map(Array.mapAccum(now, (next, delay) => {
340 const timestamp = next + Duration.toMillis(delay)
341 return [timestamp, format(timestamp)]
342 }))
343 )
344 expect(outputs).toEqual([
345 "Wed Jan 03 2024 04:30:00",
346 "Fri Jan 05 2024 04:30:00",
347 "Wed Jan 10 2024 04:30:00",
348 "Mon Jan 15 2024 04:30:00",
349 "Wed Jan 17 2024 04:30:00",
350 "Wed Jan 24 2024 04:30:00"
351 ])
352 }))
353
354 it.effect("does not fail when the test clock is adjusted to infinity", () =>
355 Effect.gen(function*() {
356 const latch = yield* Deferred.make<void>()
357 const fiber = yield* Deferred.await(latch).pipe(
358 Effect.repeat(Schedule.cron("0 0 4 8-14 * *", "UTC")),
359 Effect.forkChild
360 )
361
362 yield* TestClock.adjust(Infinity)
363 yield* Deferred.succeed(latch, void 0)
364 yield* Fiber.join(fiber)
365 }))
366 })
367
368 describe("duration", () => {
369 it.effect("recurs once after the provided duration", () =>
370 Effect.gen(function*() {
371 const schedule = Schedule.duration(Duration.seconds(1))
372 const inputs = Array.makeBy(5, constUndefined)
373 const output = yield* runDelays(schedule, inputs)
374 expect(output).toEqual([Duration.seconds(1), Duration.zero])
375 }))
376 })
377
378 describe("once", () => {
379 it.effect("recurs immediately once, then completes with void", () =>
380 Effect.gen(function*() {
381 const step = yield* Schedule.toStep(Schedule.once)
382 const first = yield* step(0, "first")
383 const second = yield* Pull.matchEffect(step(1_000, "second"), {
384 onSuccess: () => Effect.succeed("unexpected success"),
385 onFailure: () => Effect.succeed("unexpected failure"),
386 onDone: (value) => Effect.succeed(value)
387 })
388
389 assert.deepStrictEqual(first, [undefined, Duration.zero])
390 assert.isUndefined(second)
391 }))
392
393 it.effect("repeats an effect exactly once after its initial execution", () =>
394 Effect.gen(function*() {
395 let executions = 0
396 yield* Effect.sync(() => {
397 executions++
398 }).pipe(Effect.repeat(Schedule.once))
399
400 assert.strictEqual(executions, 2)
401 }))
402 })
403
404 describe("upTo", () => {
405 it.effect("limits by times", () =>
406 Effect.gen(function*() {
407 const schedule = Schedule.identity<string>().pipe(Schedule.upTo({ times: 2 }))
408 const step = yield* Schedule.toStep(schedule)
409
410 const first = yield* step(0, "a")
411 const second = yield* step(0, "b")
412 const third = yield* Pull.matchEffect(step(0, "c"), {
413 onSuccess: () => Effect.succeed("unexpected success"),
414 onFailure: () => Effect.succeed("unexpected failure"),
415 onDone: (value) => Effect.succeed(value)
416 })
417
418 assert.deepStrictEqual([first, second, third], [
419 ["a", Duration.zero],
420 ["b", Duration.zero],
421 "c"
422 ])
423 }))
424
425 it.effect("limits by duration", () =>
426 Effect.gen(function*() {
427 const schedule = Schedule.identity<string>().pipe(Schedule.upTo({ duration: "1 second" }))
428 const step = yield* Schedule.toStep(schedule)
429
430 const first = yield* step(0, "a")
431 const second = yield* step(1_000, "b")
432 const third = yield* Pull.matchEffect(step(1_001, "c"), {
433 onSuccess: () => Effect.succeed("unexpected success"),
434 onFailure: () => Effect.succeed("unexpected failure"),
435 onDone: (value) => Effect.succeed(value)
436 })
437
438 assert.deepStrictEqual([first, second, third], [
439 ["a", Duration.zero],
440 ["b", Duration.zero],
441 "c"
442 ])
443 }))
444
445 it.effect("limits by the first exhausted option", () =>
446 Effect.gen(function*() {
447 const schedule = Schedule.identity<string>().pipe(Schedule.upTo({ duration: "1 hour", times: 1 }))
448 const step = yield* Schedule.toStep(schedule)
449
450 const first = yield* step(0, "a")
451 const second = yield* Pull.matchEffect(step(0, "b"), {
452 onSuccess: () => Effect.succeed("unexpected success"),
453 onFailure: () => Effect.succeed("unexpected failure"),
454 onDone: (value) => Effect.succeed(value)
455 })
456
457 assert.deepStrictEqual([first, second], [["a", Duration.zero], "b"])
458 }))
459
460 it.effect("leaves the schedule unchanged when no options are specified", () =>
461 Effect.gen(function*() {
462 const schedule = Schedule.identity<string>().pipe(Schedule.upTo({}))
463 const step = yield* Schedule.toStep(schedule)
464
465 const first = yield* step(0, "a")
466 const second = yield* step(0, "b")
467
468 assert.deepStrictEqual([first, second], [
469 ["a", Duration.zero],
470 ["b", Duration.zero]
471 ])
472 }))
473 })
474
475 describe("jittered", () => {
476 it.effect("keeps delays within 80%-120% of the original", () =>
477 Effect.gen(function*() {
478 const schedule = Schedule.jittered(Schedule.spaced(Duration.seconds(1)))
479 const inputs = Array.makeBy(20, constUndefined)
480 const output = yield* runDelays(schedule, inputs).pipe(Random.withSeed("jittered-bounds"))
481 expect(output.every((delay) => {
482 const millis = Duration.toMillis(delay)
483 return millis >= 800 && millis <= 1200
484 })).toBe(true)
485 }))
486
487 it.effect("does not change completion output", () =>
488 Effect.gen(function*() {
489 const schedule = Schedule.jittered(Schedule.duration(Duration.seconds(1)))
490 const inputs = Array.makeBy(5, constUndefined)
491 const output = yield* runDelays(schedule, inputs).pipe(Random.withSeed("jittered-completion"))
492 expect(output.length).toEqual(2)
493 expect(Duration.toMillis(output[0])).toBeGreaterThanOrEqual(800)
494 expect(Duration.toMillis(output[0])).toBeLessThanOrEqual(1200)
495 expect(output[1]).toEqual(Duration.zero)
496 }))
497 })
498
499 describe("spaced", () => {
500 it.effect("constant delays", () =>
501 Effect.gen(function*() {
502 const schedule = Schedule.spaced(Duration.seconds(1))
503 const inputs = Array.makeBy(5, constUndefined)
504 const output = yield* runDelays(schedule, inputs)
505 expect(output).toEqual(Array.makeBy(5, constant(Duration.seconds(1))))
506 }))
507 })
508
509 describe("fixed", () => {
510 it.effect("constant delays", () =>
511 Effect.gen(function*() {
512 const schedule = Schedule.fixed(Duration.seconds(1))
513 const inputs = Array.makeBy(5, constUndefined)
514 const output = yield* runDelays(schedule, inputs)
515 expect(output).toEqual(Array.makeBy(5, constant(Duration.seconds(1))))
516 }))
517
518 it.effect("delays until the nearest window boundary when action is slow", () =>
519 Effect.gen(function*() {
520 const delays: Array<Duration.Duration> = []
521 const schedule = Schedule.fixed("1 seconds").pipe(
522 Schedule.while(({ attempt }) => Effect.succeed(attempt <= 5)),
523 Schedule.tap((metadata) =>
524 Effect.sync(() => {
525 delays.push(metadata.duration)
526 })
527 )
528 )
529 yield* Effect.sleep("500 millis").pipe(
530 Effect.schedule(schedule),
531 Effect.andThen(Effect.sync(() => delays.push(Duration.zero))),
532 Effect.forkChild
533 )
534 yield* TestClock.setTime(Number.POSITIVE_INFINITY)
535 expect(delays).toEqual([
536 Duration.millis(1000),
537 Duration.millis(500),
538 Duration.millis(500),
539 Duration.millis(500),
540 Duration.millis(500),
541 Duration.zero
542 ])
543 }))
544
545 it.effect("matches effect v3 when action duration exceeds the interval", () =>
546 Effect.gen(function*() {
547 const delays: Array<Duration.Duration> = []
548 const schedule = Schedule.fixed("1 seconds").pipe(
549 Schedule.while(({ attempt }) => Effect.succeed(attempt <= 5)),
550 Schedule.tap((metadata) =>
551 Effect.sync(() => {
552 delays.push(metadata.duration)
553 })
554 )
555 )
556 yield* Effect.sleep("1.5 seconds").pipe(
557 Effect.schedule(schedule),
558 Effect.andThen(Effect.sync(() => delays.push(Duration.zero))),
559 Effect.forkChild
560 )
561 yield* TestClock.setTime(Number.POSITIVE_INFINITY)
562 expect(delays).toEqual([
563 Duration.millis(1000),
564 Duration.zero,
565 Duration.zero,
566 Duration.zero,
567 Duration.zero,
568 Duration.zero
569 ])
570 }))
571 })
572
573 describe("windowed", () => {
574 it.effect("constant delays", () =>
575 Effect.gen(function*() {
576 const schedule = Schedule.windowed(Duration.seconds(1))
577 const inputs = Array.makeBy(5, constUndefined)
578 const output = yield* runDelays(schedule, inputs)
579 expect(output).toEqual(Array.makeBy(5, constant(Duration.seconds(1))))
580 }))
581
582 it.effect("delays until the nearest window boundary", () =>
583 Effect.gen(function*() {
584 const delays: Array<Duration.Duration> = []
585 const schedule = Schedule.windowed("1 seconds").pipe(
586 Schedule.while(({ attempt }) => Effect.succeed(attempt <= 5)),
587 Schedule.tap((metadata) =>
588 Effect.sync(() => {
589 delays.push(metadata.duration)
590 })
591 )
592 )
593 yield* Effect.sleep("1.5 seconds").pipe(
594 Effect.schedule(schedule),
595 Effect.andThen(Effect.sync(() => delays.push(Duration.zero))),
596 Effect.forkChild
597 )
598 yield* TestClock.setTime(Number.POSITIVE_INFINITY)
599 expect(delays).toEqual([
600 Duration.millis(1000),
601 Duration.millis(500),
602 Duration.millis(500),
603 Duration.millis(500),
604 Duration.millis(500),
605 Duration.zero
606 ])
607 }))
608 })
609})
610
611const run = Effect.fnUntraced(function*<A, E, R>(effect: Effect.Effect<A, E, R>) {
612 const fiber = yield* Effect.forkChild(effect)
613 yield* TestClock.setTime(Number.POSITIVE_INFINITY)
614 return yield* Fiber.join(fiber)
615})
616
617const runDelays = <Output, Input, Error, Env>(
618 schedule: Schedule.Schedule<Output, Input, Error, Env>,
619 input: Iterable<Input>
620) =>
621 Effect.gen(function*() {
622 const step = yield* Schedule.toStepWithMetadata(schedule)
623 const out: Array<Duration.Duration> = []
624 yield* Effect.gen(function*() {
625 for (const value of input) {
626 out.push((yield* step(value)).duration)
627 }
628 }).pipe(Pull.catchDone(() => {
629 out.push(Duration.zero)
630 return Effect.void
631 }))
632 return out
633 }).pipe(run)
634
635const format = (timestamp: number | string | Date): string => {
636 const date = new Date(timestamp)
637 const hours = `0${date.getHours()}`.slice(-2)
638 const minutes = `0${date.getMinutes()}`.slice(-2)
639 const seconds = `0${date.getSeconds()}`.slice(-2)
640 return `${date.toDateString()} ${hours}:${minutes}:${seconds}`
641}