1import { assert, describe, it } from "@effect/vitest"
2import { Cause, Context, Effect, Exit, Fiber, Latch, References } from "effect"
3
4describe("Fiber", () => {
5 it("is a fiber", async () => {
6 const result = Effect.runFork(Effect.succeed(1))
7 assert.isTrue(Fiber.isFiber(result))
8 })
9
10 it("notifies all observers when an observer cancels during exit", () => {
11 const fiber = Effect.runFork(Effect.never)
12 const observed: Array<number> = []
13 let cancel = () => {}
14 cancel = fiber.addObserver(() => {
15 observed.push(1)
16 cancel()
17 })
18 fiber.addObserver(() => {
19 observed.push(2)
20 })
21
22 fiber.interruptUnsafe()
23
24 assert.deepStrictEqual(observed, [1, 2])
25 })
26
27 it("stops notifying an observer once it is removed", () => {
28 const fiber = Effect.runFork(Effect.never)
29 const observed: Array<number> = []
30 const cancels = [0, 1, 2].map((index) => fiber.addObserver(() => observed.push(index)))
31 cancels[1]()
32 cancels[1]()
33
34 fiber.interruptUnsafe()
35
36 assert.deepStrictEqual(observed, [0, 2])
37 })
38
39 describe("joinAll", () => {
40 it.effect("leaves other joins running when one is interrupted", () =>
41 Effect.gen(function*() {
42 const latch = Latch.makeUnsafe()
43 const fibers = yield* Effect.forEach(
44 [1, 2],
45 (n) => Effect.forkChild(Effect.as(latch.await, n), { startImmediately: true })
46 )
47 const interrupted = yield* Fiber.joinAll(fibers).pipe(Effect.forkChild({ startImmediately: true }))
48 const kept = yield* Fiber.joinAll(fibers).pipe(Effect.forkChild({ startImmediately: true }))
49 yield* Fiber.interrupt(interrupted)
50 yield* latch.open
51 assert.deepStrictEqual(yield* Fiber.join(kept), [1, 2])
52 }))
53
54 it.effect("cleans up observers on interruption", () =>
55 Effect.gen(function*() {
56 const fibers = yield* Effect.forEach([0, 1], () => Effect.forkChild(Effect.never))
57 const observerCounts = () =>
58 fibers.map((fiber) =>
59 (fiber as unknown as { readonly _observers: ReadonlyArray<unknown> | undefined })._observers?.length ?? 0
60 )
61 const before = observerCounts()
62 const joinFiber = yield* Fiber.joinAll(fibers).pipe(Effect.forkChild({ startImmediately: true }))
63 assert.deepStrictEqual(observerCounts(), before.map((count) => count + 1))
64 yield* Fiber.interrupt(joinFiber)
65 assert.deepStrictEqual(observerCounts(), before)
66 }))
67 })
68
69 describe("interruptAll", () => {
70 it.effect("awaits fibers passed as a one-shot iterable", () =>
71 Effect.gen(function*() {
72 let cleaned = false
73 const latch = Latch.makeUnsafe()
74 const fiber = yield* Effect.never.pipe(
75 Effect.onInterrupt(() =>
76 latch.whenOpen(Effect.sync(() => {
77 cleaned = true
78 }))
79 ),
80 Effect.forkChild({ startImmediately: true })
81 )
82 yield* Effect.forkChild(latch.open)
83 yield* Fiber.interruptAll(
84 (function*() {
85 yield fiber
86 })()
87 )
88 assert.isTrue(cleaned)
89 }))
90 })
91
92 describe("interruptAllAs", () => {
93 it.effect("awaits fibers passed as a one-shot iterable", () =>
94 Effect.gen(function*() {
95 const latch = Latch.makeUnsafe()
96 let cleaned = false
97 const fiber = yield* Effect.never.pipe(
98 Effect.onInterrupt(() =>
99 latch.whenOpen(Effect.sync(() => {
100 cleaned = true
101 }))
102 ),
103 Effect.forkChild({ startImmediately: true })
104 )
105 yield* Effect.forkChild(latch.open)
106 yield* Fiber.interruptAllAs(
107 (function*() {
108 yield fiber
109 })(),
110 0
111 )
112 assert.isTrue(cleaned)
113 }))
114 })
115
116 it.effect(
117 "delivers a synchronous self-interrupt instead of completing to success",
118 () =>
119 Effect.gen(function*() {
120 const child = yield* Effect.gen(function*() {
121 const self = Fiber.getCurrent()!
122 self.interruptUnsafe()
123 return 42
124 }).pipe(Effect.forkChild({ startImmediately: true }))
125
126 const exit = yield* Fiber.await(child)
127 assert.isTrue(Exit.hasInterrupts(exit))
128 })
129 )
130
131 it.effect("retains distinct target and interruptor stack frames", () =>
132 Effect.gen(function*() {
133 const targetFrame: References.StackFrame = {
134 name: "target-frame",
135 stack: () => "at target-call-site.ts:1:1",
136 parent: undefined
137 }
138 const interruptorFrame: References.StackFrame = {
139 name: "interruptor-frame",
140 stack: () => "at interruptor-call-site.ts:2:2",
141 parent: undefined
142 }
143 const target = yield* Effect.never.pipe(
144 Effect.provideService(References.CurrentStackFrame, targetFrame),
145 Effect.forkChild({ startImmediately: true })
146 )
147
148 yield* Fiber.interrupt(target).pipe(
149 Effect.provideService(References.CurrentStackFrame, interruptorFrame)
150 )
151
152 const exit = yield* Fiber.await(target)
153 if (exit._tag !== "Failure") {
154 return assert.fail("expected interrupted fiber to exit with failure")
155 }
156 const annotations = Cause.reasonAnnotations(exit.cause.reasons[0])
157 assert.strictEqual(Context.getUnsafe(annotations, Cause.StackTrace), targetFrame)
158 assert.strictEqual(Context.getUnsafe(annotations, Cause.InterruptorStackTrace), interruptorFrame)
159 assert.isTrue(Cause.pretty(exit.cause).includes("interruptor-call-site.ts:2:2"))
160 }))
161
162 it.effect("delivers a pending interrupt when interruptibleMask restores interruptibility", () =>
163 Effect.gen(function*() {
164 const masked = yield* Latch.make()
165 const resume = yield* Latch.make()
166 const events: Array<string> = []
167
168 const child = yield* Effect.uninterruptible(
169 Effect.gen(function*() {
170 yield* masked.open
171 yield* resume.await
172 return yield* Effect.interruptibleMask(() => {
173 events.push("interruptibleMask")
174 return Effect.never
175 })
176 })
177 ).pipe(Effect.forkChild({ startImmediately: true }))
178
179 yield* masked.await
180 events.push("masked")
181
182 yield* Effect.sync(() => {
183 child.interruptUnsafe(123)
184 events.push("interrupted")
185 })
186 assert.isUndefined(child.pollUnsafe())
187
188 yield* resume.open
189 events.push("resumed")
190 yield* Effect.yieldNow
191 yield* Effect.yieldNow
192
193 const exit = child.pollUnsafe()
194 if (exit === undefined) {
195 assert.fail("fiber did not exit after interruptibleMask restored interruptibility")
196 }
197 assert.isTrue(Exit.hasInterrupts(exit))
198 if (exit._tag !== "Failure") {
199 assert.fail("expected interrupted fiber to exit with failure")
200 }
201 assert.deepStrictEqual(Cause.interruptors(exit.cause), new Set([123]))
202 assert.deepStrictEqual(events, ["masked", "interrupted", "resumed", "interruptibleMask"])
203 }))
204
205 it.effect("runs an async interrupt finalizer exactly once, in order", () =>
206 Effect.gen(function*() {
207 const events: Array<string> = []
208
209 const child = yield* Effect.gen(function*() {
210 const self = Fiber.getCurrent()!
211 yield* Effect.suspend(() => {
212 self.interruptUnsafe()
213 events.push("acquired")
214 return Effect.void
215 }).pipe(
216 Effect.onInterrupt(() =>
217 Effect.sync(() => {
218 events.push("finalizer-start")
219 }).pipe(
220 Effect.tap(Effect.yieldNow),
221 Effect.tap(Effect.sync(() => {
222 events.push("finalizer-end")
223 }))
224 )
225 )
226 )
227 events.push("unreachable")
228 }).pipe(Effect.forkChild({ startImmediately: true }))
229
230 const exit = yield* Fiber.await(child)
231 events.push("awaited")
232 assert.isTrue(Exit.hasInterrupts(exit))
233 assert.deepStrictEqual(events, ["acquired", "finalizer-start", "finalizer-end", "awaited"])
234 }))
235})