1import { assert, describe, it } from "@effect/vitest"
2import { assertTrue, deepStrictEqual, strictEqual } from "@effect/vitest/utils"
3import {
4 Array,
5 Cause,
6 Context,
7 Duration,
8 Effect,
9 ExecutionPlan,
10 Exit,
11 Fiber,
12 Latch,
13 Layer,
14 Scheduler,
15 Stream
16} from "effect"
17
18describe("ExecutionPlan", () => {
19 class Service extends Context.Service<Service>()("Service", {
20 make: Effect.succeed({
21 stream: Stream.fail("A") as Stream.Stream<number, string>
22 })
23 }) {
24 static A = Layer.effect(this, this.make)
25
26 static B = Layer.succeed(
27 Service,
28 Service.of({
29 stream: Stream.fail("B")
30 })
31 )
32 static C = Layer.succeed(
33 Service,
34 Service.of({
35 stream: Stream.make(1, 2, 3)
36 })
37 )
38 }
39
40 const Plan = ExecutionPlan.make({
41 provide: Service.A
42 }, {
43 provide: Service.B
44 }, {
45 provide: Service.C
46 })
47
48 const PlanPartial = ExecutionPlan.make({
49 provide: Layer.succeed(
50 Service,
51 Service.of({
52 stream: Stream.make(1, 2, 3).pipe(
53 Stream.concat(Stream.fail("Partial"))
54 )
55 })
56 )
57 }, {
58 provide: Service.C
59 })
60
61 describe("make", () => {
62 it("rejects zero attempts", () => {
63 assert.throws(
64 () =>
65 ExecutionPlan.make({
66 provide: Context.empty(),
67 attempts: 0
68 }),
69 /ExecutionPlan\.make: step\[0\]\.attempts must be greater than 0/
70 )
71 })
72 })
73
74 describe("captureRequirements", () => {
75 class Policy extends Context.Service<Policy, { allow: boolean }>()("ExecutionPlan.test/Policy") {}
76
77 const failOnce = () => {
78 let attempts = 0
79 return Effect.suspend(() => ++attempts === 1 ? Effect.fail("busy") : Effect.succeed("ok"))
80 }
81
82 const plan = ExecutionPlan.make({
83 provide: Context.empty(),
84 attempts: 2,
85 while: (_: string) => Effect.map(Policy, ({ allow }) => allow)
86 })
87
88 it.effect("captures while predicate requirements", () =>
89 Effect.gen(function*() {
90 const captured = yield* plan.captureRequirements.pipe(Effect.provideService(Policy, { allow: true }))
91 const result = yield* Effect.withExecutionPlan(failOnce(), captured)
92 strictEqual(result, "ok")
93 }))
94
95 it.effect("prefers captured while predicate requirements over the ambient context", () =>
96 Effect.gen(function*() {
97 const captured = yield* plan.captureRequirements.pipe(Effect.provideService(Policy, { allow: true }))
98 const result = yield* Effect.withExecutionPlan(failOnce(), captured).pipe(
99 Effect.provideService(Policy, { allow: false })
100 )
101 strictEqual(result, "ok")
102 }))
103 })
104
105 describe("Stream.withExecutionPlan", () => {
106 it.effect("limits attempts after partial stream failures", () =>
107 Effect.gen(function*() {
108 let runs = 0
109 const stream = Stream.fromEffect(Effect.sync(() => {
110 runs++
111 return 1
112 })).pipe(Stream.concat(Stream.fail("boom")))
113 const plan = ExecutionPlan.make({
114 provide: Context.empty(),
115 attempts: 3
116 })
117 const items = Array.empty<number>()
118
119 const result = yield* stream.pipe(
120 Stream.withExecutionPlan(plan),
121 Stream.runForEach((item) =>
122 Effect.sync(() => {
123 items.push(item)
124 })
125 ),
126 Effect.exit
127 )
128
129 deepStrictEqual(items, [1, 1, 1])
130 deepStrictEqual(runs, 3)
131 deepStrictEqual(result, Exit.fail("boom"))
132 }))
133
134 it.effect("allows interruption between retries after partial stream failures", () =>
135 Effect.gen(function*() {
136 let runs = 0
137 const retried = yield* Latch.make()
138 const stream = Stream.fromEffect(Effect.gen(function*() {
139 runs++
140 if (runs > 1) {
141 yield* retried.open
142 }
143 return 1
144 })).pipe(Stream.concat(Stream.fail("boom")))
145 const plan = ExecutionPlan.make({
146 provide: Context.empty(),
147 attempts: Number.MAX_SAFE_INTEGER
148 })
149 const fiber = yield* stream.pipe(
150 Stream.withExecutionPlan(plan),
151 Stream.runDrain,
152 Effect.provideService(Scheduler.PreventSchedulerYield, true),
153 Effect.forkChild
154 )
155
156 yield* retried.await
157 yield* Fiber.interrupt(fiber)
158 const exit = yield* Fiber.await(fiber)
159
160 assertTrue(runs > 1)
161 assertTrue(Exit.hasInterrupts(exit))
162 }))
163
164 it.effect("falls back through failing layers and records stream attempt metadata", () =>
165 Effect.gen(function*() {
166 const stream = Stream.unwrap(Effect.map(Service, (_) => _.stream))
167 const items = Array.empty<number>()
168 const metadata = Array.empty<ExecutionPlan.Metadata>()
169 const result = yield* stream.pipe(
170 Stream.onStart(ExecutionPlan.CurrentMetadata.use((meta) => {
171 metadata.push(meta)
172 return Effect.void
173 })),
174 Stream.withExecutionPlan(Plan),
175 Stream.runForEach((n) =>
176 Effect.sync(() => {
177 items.push(n)
178 })
179 ),
180 Effect.exit
181 )
182 deepStrictEqual(items, [1, 2, 3])
183 deepStrictEqual(metadata, [{
184 attempt: 1,
185 stepIndex: 0
186 }, {
187 attempt: 2,
188 stepIndex: 1
189 }, {
190 attempt: 3,
191 stepIndex: 2
192 }])
193 assertTrue(Exit.isSuccess(result))
194 }))
195
196 it.effect("falls back after a stream emits partial items by default", () =>
197 Effect.gen(function*() {
198 const stream = Stream.unwrap(Effect.map(Service, (_) => _.stream))
199 const items = Array.empty<number>()
200 const result = yield* stream.pipe(
201 Stream.withExecutionPlan(PlanPartial),
202 Stream.runForEach((n) =>
203 Effect.sync(() => {
204 items.push(n)
205 })
206 ),
207 Effect.exit
208 )
209 deepStrictEqual(items, [1, 2, 3, 1, 2, 3])
210 assertTrue(Exit.isSuccess(result))
211 }))
212
213 it.effect("fails on a partial stream when fallback is disabled", () =>
214 Effect.gen(function*() {
215 const stream = Stream.unwrap(Effect.map(Service, (_) => _.stream))
216 const items = Array.empty<number>()
217 const result = yield* stream.pipe(
218 Stream.withExecutionPlan(PlanPartial, {
219 preventFallbackOnPartialStream: true
220 }),
221 Stream.runForEach((n) =>
222 Effect.sync(() => {
223 items.push(n)
224 })
225 ),
226 Effect.exit
227 )
228 deepStrictEqual(items, [1, 2, 3])
229 deepStrictEqual(result, Exit.fail("Partial"))
230 }))
231 })
232
233 describe("onEvent", () => {
234 const makeCollector = () => {
235 const events = Array.empty<ExecutionPlan.Event<unknown>>()
236 const onEvent = (event: ExecutionPlan.Event<unknown>) =>
237 Effect.sync(() => {
238 events.push(event)
239 })
240 return { events, onEvent }
241 }
242
243 const simplify = (event: ExecutionPlan.Event<unknown>) =>
244 event._tag === "AttemptFailure"
245 ? {
246 _tag: event._tag,
247 attempt: event.attempt,
248 stepAttempt: event.stepAttempt,
249 stepIndex: event.stepIndex,
250 error: Cause.squash(event.cause)
251 }
252 : {
253 _tag: event._tag,
254 attempt: event.attempt,
255 stepAttempt: event.stepAttempt,
256 stepIndex: event.stepIndex
257 }
258
259 describe("Effect.withExecutionPlan", () => {
260 it.effect("emits start and success on first attempt", () =>
261 Effect.gen(function*() {
262 const { events, onEvent } = makeCollector()
263 const result = yield* Effect.succeed(1).pipe(
264 Effect.withExecutionPlan(ExecutionPlan.make({ provide: Context.empty() }), { onEvent })
265 )
266 strictEqual(result, 1)
267 deepStrictEqual(events.map(simplify), [
268 { _tag: "AttemptStart", attempt: 1, stepAttempt: 1, stepIndex: 0 },
269 { _tag: "AttemptSuccess", attempt: 1, stepAttempt: 1, stepIndex: 0 }
270 ])
271 assertTrue(events[1]._tag === "AttemptSuccess" && Duration.isDuration(events[1].duration))
272 }))
273
274 it.effect("observer defects do not change the outcome or leave attempts unpaired", () =>
275 Effect.gen(function*() {
276 const events = Array.empty<string>()
277 const exit = yield* Effect.succeed("source-success").pipe(
278 Effect.withExecutionPlan(ExecutionPlan.make({ provide: Context.empty() }), {
279 onEvent: (event) =>
280 Effect.sync(() => events.push(event._tag)).pipe(
281 Effect.andThen(Effect.die("observer-defect"))
282 )
283 }),
284 Effect.exit
285 )
286
287 deepStrictEqual(
288 { exit, events },
289 { exit: Exit.succeed("source-success"), events: ["AttemptStart", "AttemptSuccess"] },
290 "observer defects must not affect the source outcome"
291 )
292 }))
293
294 it.effect("emits an event pair per attempt within a step", () =>
295 Effect.gen(function*() {
296 const { events, onEvent } = makeCollector()
297 let runs = 0
298 const task = Effect.suspend(() => {
299 runs++
300 return runs < 3 ? Effect.fail(`fail-${runs}`) : Effect.succeed(runs)
301 })
302 const plan = ExecutionPlan.make({ provide: Context.empty(), attempts: 3 })
303 const result = yield* Effect.withExecutionPlan(task, plan, { onEvent })
304 strictEqual(result, 3)
305 deepStrictEqual(events.map(simplify), [
306 { _tag: "AttemptStart", attempt: 1, stepAttempt: 1, stepIndex: 0 },
307 { _tag: "AttemptFailure", attempt: 1, stepAttempt: 1, stepIndex: 0, error: "fail-1" },
308 { _tag: "AttemptStart", attempt: 2, stepAttempt: 2, stepIndex: 0 },
309 { _tag: "AttemptFailure", attempt: 2, stepAttempt: 2, stepIndex: 0, error: "fail-2" },
310 { _tag: "AttemptStart", attempt: 3, stepAttempt: 3, stepIndex: 0 },
311 { _tag: "AttemptSuccess", attempt: 3, stepAttempt: 3, stepIndex: 0 }
312 ])
313 }))
314
315 it.effect("fails over to the next step without replaying events", () =>
316 Effect.gen(function*() {
317 const { events, onEvent } = makeCollector()
318 let runs = 0
319 const task = Effect.suspend(() => {
320 runs++
321 return runs < 3 ? Effect.fail(`fail-${runs}`) : Effect.succeed(runs)
322 })
323 const plan = ExecutionPlan.make(
324 { provide: Context.empty(), attempts: 2 },
325 { provide: Context.empty(), attempts: 2 }
326 )
327 const result = yield* Effect.withExecutionPlan(task, plan, { onEvent })
328 strictEqual(result, 3)
329 deepStrictEqual(events.map(simplify), [
330 { _tag: "AttemptStart", attempt: 1, stepAttempt: 1, stepIndex: 0 },
331 { _tag: "AttemptFailure", attempt: 1, stepAttempt: 1, stepIndex: 0, error: "fail-1" },
332 { _tag: "AttemptStart", attempt: 2, stepAttempt: 2, stepIndex: 0 },
333 { _tag: "AttemptFailure", attempt: 2, stepAttempt: 2, stepIndex: 0, error: "fail-2" },
334 { _tag: "AttemptStart", attempt: 3, stepAttempt: 1, stepIndex: 1 },
335 { _tag: "AttemptSuccess", attempt: 3, stepAttempt: 1, stepIndex: 1 }
336 ])
337 }))
338
339 it.effect("event attempt matches CurrentMetadata across a failover boundary", () =>
340 Effect.gen(function*() {
341 const { events, onEvent } = makeCollector()
342 const inside = Array.empty<ExecutionPlan.Metadata>()
343 let runs = 0
344 const task = Effect.gen(function*() {
345 yield* ExecutionPlan.CurrentMetadata.use((meta) => {
346 inside.push(meta)
347 return Effect.void
348 })
349 runs++
350 return runs < 3 ? yield* Effect.fail("boom") : runs
351 })
352 const plan = ExecutionPlan.make(
353 { provide: Context.empty(), attempts: 2 },
354 { provide: Context.empty(), attempts: 2 }
355 )
356 const result = yield* Effect.withExecutionPlan(task, plan, { onEvent })
357 strictEqual(result, 3)
358 deepStrictEqual(inside, [
359 { attempt: 1, stepIndex: 0 },
360 { attempt: 2, stepIndex: 0 },
361 { attempt: 3, stepIndex: 1 }
362 ])
363 deepStrictEqual(
364 events.filter((event) => event._tag === "AttemptStart")
365 .map((event) => ({ attempt: event.attempt, stepIndex: event.stepIndex })),
366 inside
367 )
368 }))
369
370 it.effect("emits AttemptFailure when a step layer fails to build", () =>
371 Effect.gen(function*() {
372 const { events, onEvent } = makeCollector()
373 const plan = ExecutionPlan.make(
374 { provide: Layer.effect(Service, Effect.fail("nope")) },
375 { provide: Service.C }
376 )
377 const result = yield* Stream.runCollect(Stream.unwrap(Effect.map(Service, (_) => _.stream))).pipe(
378 Effect.withExecutionPlan(plan, { onEvent })
379 )
380 deepStrictEqual(result, [1, 2, 3])
381 deepStrictEqual(events.map(simplify), [
382 { _tag: "AttemptStart", attempt: 1, stepAttempt: 1, stepIndex: 0 },
383 { _tag: "AttemptFailure", attempt: 1, stepAttempt: 1, stepIndex: 0, error: "nope" },
384 { _tag: "AttemptStart", attempt: 2, stepAttempt: 1, stepIndex: 1 },
385 { _tag: "AttemptSuccess", attempt: 2, stepAttempt: 1, stepIndex: 1 }
386 ])
387 }))
388
389 it.effect("emits AttemptFailure when interrupted mid-attempt", () =>
390 Effect.gen(function*() {
391 const { events, onEvent } = makeCollector()
392 const started = yield* Latch.make()
393 const task = Effect.gen(function*() {
394 yield* started.open
395 return yield* Effect.never
396 })
397 const plan = ExecutionPlan.make({ provide: Context.empty(), attempts: 2 })
398 const fiber = yield* task.pipe(
399 Effect.withExecutionPlan(plan, { onEvent }),
400 Effect.forkChild
401 )
402 yield* started.await
403 yield* Fiber.interrupt(fiber)
404 const exit = yield* Fiber.await(fiber)
405 assertTrue(Exit.hasInterrupts(exit))
406 strictEqual(events.length, 2)
407 deepStrictEqual(simplify(events[0]), { _tag: "AttemptStart", attempt: 1, stepAttempt: 1, stepIndex: 0 })
408 const failure = events[1]
409 assertTrue(failure._tag === "AttemptFailure")
410 strictEqual(failure.attempt, 1)
411 assertTrue(Cause.hasInterrupts(failure.cause))
412 }))
413
414 it.effect("emits AttemptFailure for defects without retrying", () =>
415 Effect.gen(function*() {
416 const { events, onEvent } = makeCollector()
417 const plan = ExecutionPlan.make(
418 { provide: Context.empty(), attempts: 2 },
419 { provide: Context.empty() }
420 )
421 const exit = yield* Effect.die("boom").pipe(
422 Effect.withExecutionPlan(plan, { onEvent }),
423 Effect.exit
424 )
425 deepStrictEqual(exit, Exit.die("boom"))
426 deepStrictEqual(events.map(simplify), [
427 { _tag: "AttemptStart", attempt: 1, stepAttempt: 1, stepIndex: 0 },
428 { _tag: "AttemptFailure", attempt: 1, stepAttempt: 1, stepIndex: 0, error: "boom" }
429 ])
430 }))
431 })
432
433 describe("Stream.withExecutionPlan", () => {
434 it.effect("observer defects do not change the outcome or leave attempts unpaired", () =>
435 Effect.gen(function*() {
436 const events = Array.empty<string>()
437 const exit = yield* Stream.succeed("source-success").pipe(
438 Stream.withExecutionPlan(ExecutionPlan.make({ provide: Context.empty() }), {
439 onEvent: (event) =>
440 Effect.sync(() => events.push(event._tag)).pipe(
441 Effect.andThen(Effect.die("observer-defect"))
442 )
443 }),
444 Stream.runCollect,
445 Effect.exit
446 )
447
448 deepStrictEqual(
449 { exit, events },
450 { exit: Exit.succeed(["source-success"]), events: ["AttemptStart", "AttemptSuccess"] },
451 "observer defects must not affect the source outcome"
452 )
453 }))
454
455 it.effect("emits events for each step attempt", () =>
456 Effect.gen(function*() {
457 const { events, onEvent } = makeCollector()
458 const stream = Stream.unwrap(Effect.map(Service, (_) => _.stream))
459 const items = yield* stream.pipe(
460 Stream.withExecutionPlan(Plan, { onEvent }),
461 Stream.runCollect
462 )
463 deepStrictEqual(items, [1, 2, 3])
464 deepStrictEqual(events.map(simplify), [
465 { _tag: "AttemptStart", attempt: 1, stepAttempt: 1, stepIndex: 0 },
466 { _tag: "AttemptFailure", attempt: 1, stepAttempt: 1, stepIndex: 0, error: "A" },
467 { _tag: "AttemptStart", attempt: 2, stepAttempt: 1, stepIndex: 1 },
468 { _tag: "AttemptFailure", attempt: 2, stepAttempt: 1, stepIndex: 1, error: "B" },
469 { _tag: "AttemptStart", attempt: 3, stepAttempt: 1, stepIndex: 2 },
470 { _tag: "AttemptSuccess", attempt: 3, stepAttempt: 1, stepIndex: 2 }
471 ])
472 }))
473
474 it.effect("emits an event pair per attempt within a step", () =>
475 Effect.gen(function*() {
476 const { events, onEvent } = makeCollector()
477 const stream = Stream.make(1).pipe(Stream.concat(Stream.fail("boom")))
478 const plan = ExecutionPlan.make({ provide: Context.empty(), attempts: 3 })
479 const result = yield* stream.pipe(
480 Stream.withExecutionPlan(plan, { onEvent }),
481 Stream.runDrain,
482 Effect.exit
483 )
484 deepStrictEqual(result, Exit.fail("boom"))
485 deepStrictEqual(events.map(simplify), [
486 { _tag: "AttemptStart", attempt: 1, stepAttempt: 1, stepIndex: 0 },
487 { _tag: "AttemptFailure", attempt: 1, stepAttempt: 1, stepIndex: 0, error: "boom" },
488 { _tag: "AttemptStart", attempt: 2, stepAttempt: 2, stepIndex: 0 },
489 { _tag: "AttemptFailure", attempt: 2, stepAttempt: 2, stepIndex: 0, error: "boom" },
490 { _tag: "AttemptStart", attempt: 3, stepAttempt: 3, stepIndex: 0 },
491 { _tag: "AttemptFailure", attempt: 3, stepAttempt: 3, stepIndex: 0, error: "boom" }
492 ])
493 }))
494
495 it.effect("emits events when falling back after a partial stream", () =>
496 Effect.gen(function*() {
497 const { events, onEvent } = makeCollector()
498 const stream = Stream.unwrap(Effect.map(Service, (_) => _.stream))
499 const items = yield* stream.pipe(
500 Stream.withExecutionPlan(PlanPartial, { onEvent }),
501 Stream.runCollect
502 )
503 deepStrictEqual(items, [1, 2, 3, 1, 2, 3])
504 deepStrictEqual(events.map(simplify), [
505 { _tag: "AttemptStart", attempt: 1, stepAttempt: 1, stepIndex: 0 },
506 { _tag: "AttemptFailure", attempt: 1, stepAttempt: 1, stepIndex: 0, error: "Partial" },
507 { _tag: "AttemptStart", attempt: 2, stepAttempt: 1, stepIndex: 1 },
508 { _tag: "AttemptSuccess", attempt: 2, stepAttempt: 1, stepIndex: 1 }
509 ])
510 }))
511
512 it.effect("emits AttemptFailure when interrupted mid-attempt", () =>
513 Effect.gen(function*() {
514 const { events, onEvent } = makeCollector()
515 const started = yield* Latch.make()
516 const stream = Stream.fromEffect(Effect.gen(function*() {
517 yield* started.open
518 return yield* Effect.never
519 }))
520 const plan = ExecutionPlan.make({ provide: Context.empty(), attempts: 2 })
521 const fiber = yield* stream.pipe(
522 Stream.withExecutionPlan(plan, { onEvent }),
523 Stream.runDrain,
524 Effect.forkChild
525 )
526 yield* started.await
527 yield* Fiber.interrupt(fiber)
528 const exit = yield* Fiber.await(fiber)
529 assertTrue(Exit.hasInterrupts(exit))
530 strictEqual(events.length, 2)
531 deepStrictEqual(simplify(events[0]), { _tag: "AttemptStart", attempt: 1, stepAttempt: 1, stepIndex: 0 })
532 const failure = events[1]
533 assertTrue(failure._tag === "AttemptFailure")
534 strictEqual(failure.attempt, 1)
535 assertTrue(Cause.hasInterrupts(failure.cause))
536 }))
537
538 it.effect("emits AttemptFailure for defects without retrying", () =>
539 Effect.gen(function*() {
540 const { events, onEvent } = makeCollector()
541 const plan = ExecutionPlan.make(
542 { provide: Context.empty(), attempts: 2 },
543 { provide: Context.empty() }
544 )
545 const exit = yield* Stream.fromEffect(Effect.die("boom")).pipe(
546 Stream.withExecutionPlan(plan, { onEvent }),
547 Stream.runDrain,
548 Effect.exit
549 )
550 deepStrictEqual(exit, Exit.die("boom"))
551 deepStrictEqual(events.map(simplify), [
552 { _tag: "AttemptStart", attempt: 1, stepAttempt: 1, stepIndex: 0 },
553 { _tag: "AttemptFailure", attempt: 1, stepAttempt: 1, stepIndex: 0, error: "boom" }
554 ])
555 }))
556
557 it.effect("emits events when a partial stream fails with fallback disabled", () =>
558 Effect.gen(function*() {
559 const { events, onEvent } = makeCollector()
560 const stream = Stream.unwrap(Effect.map(Service, (_) => _.stream))
561 const result = yield* stream.pipe(
562 Stream.withExecutionPlan(PlanPartial, {
563 preventFallbackOnPartialStream: true,
564 onEvent
565 }),
566 Stream.runDrain,
567 Effect.exit
568 )
569 deepStrictEqual(result, Exit.fail("Partial"))
570 deepStrictEqual(events.map(simplify), [
571 { _tag: "AttemptStart", attempt: 1, stepAttempt: 1, stepIndex: 0 },
572 { _tag: "AttemptFailure", attempt: 1, stepAttempt: 1, stepIndex: 0, error: "Partial" }
573 ])
574 }))
575 })
576 })
577})