1import { assert, describe, it } from "@effect/vitest"
2import * as Clock from "effect/Clock"
3import * as Deferred from "effect/Deferred"
4import * as Duration from "effect/Duration"
5import * as Effect from "effect/Effect"
6import * as Exit from "effect/Exit"
7import * as Fiber from "effect/Fiber"
8import * as RcRef from "effect/RcRef"
9import * as Ref from "effect/Ref"
10import * as Scheduler from "effect/Scheduler"
11import * as Scope from "effect/Scope"
12import { TestClock } from "effect/testing"
13
14describe("RcRef", () => {
15 it.effect("deallocation", () =>
16 Effect.gen(function*() {
17 let acquired = 0
18 let released = 0
19 const refScope = yield* Scope.make()
20 const ref = yield* RcRef.make({
21 acquire: Effect.acquireRelease(
22 Effect.sync(() => {
23 acquired++
24 return "foo"
25 }),
26 () =>
27 Effect.sync(() => {
28 released++
29 })
30 )
31 }).pipe(
32 Scope.provide(refScope)
33 )
34
35 assert.strictEqual(acquired, 0)
36 assert.strictEqual(yield* Effect.scoped(RcRef.get(ref)), "foo")
37 assert.strictEqual(acquired, 1)
38 assert.strictEqual(released, 1)
39
40 const scopeA = yield* Scope.make()
41 const scopeB = yield* Scope.make()
42 yield* RcRef.get(ref).pipe(Scope.provide(scopeA))
43 yield* RcRef.get(ref).pipe(Scope.provide(scopeB))
44 assert.strictEqual(acquired, 2)
45 assert.strictEqual(released, 1)
46 yield* Scope.close(scopeB, Exit.void)
47 assert.strictEqual(acquired, 2)
48 assert.strictEqual(released, 1)
49 yield* Scope.close(scopeA, Exit.void)
50 assert.strictEqual(acquired, 2)
51 assert.strictEqual(released, 2)
52
53 const scopeC = yield* Scope.make()
54 yield* RcRef.get(ref).pipe(Scope.provide(scopeC))
55 assert.strictEqual(acquired, 3)
56 assert.strictEqual(released, 2)
57
58 yield* Scope.close(refScope, Exit.void)
59 assert.strictEqual(acquired, 3)
60 assert.strictEqual(released, 3)
61
62 const exit = yield* RcRef.get(ref).pipe(Effect.scoped, Effect.exit)
63 assert.isTrue(Exit.hasInterrupts(exit))
64 }))
65
66 it.effect("releasing an invalidated borrower preserves the replacement resource", () =>
67 Effect.gen(function*() {
68 let acquired = 0
69 const released: Array<number> = []
70 const owner = yield* Scope.make()
71 const scopeA = yield* Scope.make()
72 const scopeB = yield* Scope.make()
73 const scopeC = yield* Scope.make()
74 const ref = yield* RcRef.make({
75 acquire: Effect.acquireRelease(
76 Effect.sync(() => ++acquired),
77 (id) => Effect.sync(() => released.push(id))
78 )
79 }).pipe(Scope.provide(owner))
80
81 const first = yield* RcRef.get(ref).pipe(Scope.provide(scopeA))
82 yield* RcRef.invalidate(ref)
83 const replacement = yield* RcRef.get(ref).pipe(Scope.provide(scopeB))
84 yield* Scope.close(scopeA, Exit.void)
85 const next = yield* RcRef.get(ref).pipe(Scope.provide(scopeC))
86 yield* Scope.close(owner, Exit.void)
87 const releasedAtOwnerClose = [...released]
88 yield* Scope.close(scopeB, Exit.void)
89 yield* Scope.close(scopeC, Exit.void)
90
91 assert.deepStrictEqual(
92 { first, replacement, next, releasedAtOwnerClose, released },
93 { first: 1, replacement: 2, next: 2, releasedAtOwnerClose: [1, 2], released: [1, 2] }
94 )
95 }))
96
97 it.effect("releasing a borrower after owner shutdown does not reopen the reference", () =>
98 Effect.gen(function*() {
99 let acquired = 0
100 const owner = yield* Scope.make()
101 const borrower = yield* Scope.make()
102 const ref = yield* RcRef.make({
103 acquire: Effect.sync(() => ++acquired)
104 }).pipe(Scope.provide(owner))
105
106 yield* RcRef.get(ref).pipe(Scope.provide(borrower))
107 yield* Scope.close(owner, Exit.void)
108 yield* Scope.close(borrower, Exit.void)
109 const exit = yield* RcRef.get(ref).pipe(Effect.scoped, Effect.exit)
110
111 assert.isTrue(Exit.hasInterrupts(exit))
112 assert.strictEqual(acquired, 1)
113 }))
114
115 it.effect("acquisition completing after owner shutdown cannot reopen the reference", () =>
116 Effect.gen(function*() {
117 const owner = yield* Scope.make()
118 const borrower = yield* Scope.make()
119 const started = yield* Deferred.make<void>()
120 const finish = yield* Deferred.make<void>()
121 const released: Array<number> = []
122 const ref = yield* RcRef.make({
123 acquire: Effect.gen(function*() {
124 const value = yield* Effect.acquireRelease(
125 Effect.succeed(1),
126 (value) => Effect.sync(() => released.push(value))
127 )
128 yield* Deferred.succeed(started, undefined)
129 yield* Deferred.await(finish)
130 return value
131 })
132 }).pipe(Scope.provide(owner))
133 const pending = yield* RcRef.get(ref).pipe(
134 Scope.provide(borrower),
135 Effect.forkChild({ startImmediately: true })
136 )
137 yield* Deferred.await(started)
138 const queued = yield* RcRef.get(ref).pipe(
139 Scope.provide(borrower),
140 Effect.forkChild({ startImmediately: true })
141 )
142
143 yield* Scope.close(owner, Exit.void)
144 const before = yield* RcRef.get(ref).pipe(Effect.scoped, Effect.exit)
145 yield* Deferred.succeed(finish, undefined)
146 const pendingExit = yield* Fiber.await(pending)
147 const queuedExit = yield* Fiber.await(queued)
148 const after = yield* RcRef.get(ref).pipe(Effect.scoped, Effect.exit)
149 const releasedBeforeBorrowerClose = [...released]
150 yield* Scope.close(borrower, Exit.void)
151
152 assert.isTrue(Exit.hasInterrupts(before))
153 assert.isTrue(Exit.hasInterrupts(pendingExit))
154 assert.isTrue(Exit.hasInterrupts(queuedExit))
155 assert.isTrue(Exit.hasInterrupts(after))
156 assert.deepStrictEqual(releasedBeforeBorrowerClose, [1])
157 assert.deepStrictEqual(released, [1])
158 }))
159
160 it.effect("releases resources acquired before acquisition failure", () =>
161 Effect.gen(function*() {
162 const acquired = yield* Ref.make(0)
163 const released = yield* Ref.make(0)
164 const refScope = yield* Scope.make()
165 const borrowerScope = yield* Scope.make()
166 const ref = yield* RcRef.make({
167 acquire: Effect.acquireRelease(
168 Ref.updateAndGet(acquired, (n) => n + 1),
169 () => Ref.update(released, (n) => n + 1)
170 ).pipe(Effect.andThen(Effect.fail("boom")))
171 }).pipe(Scope.provide(refScope))
172
173 const getExit = yield* RcRef.get(ref).pipe(Scope.provide(borrowerScope), Effect.exit)
174 const borrowerCloseExit = yield* Scope.close(borrowerScope, Exit.void).pipe(Effect.exit)
175 const refCloseExit = yield* Scope.close(refScope, Exit.void).pipe(Effect.exit)
176
177 assert.deepStrictEqual(getExit, Exit.fail("boom"))
178 assert.deepStrictEqual(borrowerCloseExit, Exit.void)
179 assert.deepStrictEqual(refCloseExit, Exit.void)
180 assert.strictEqual(yield* Ref.get(acquired), 1)
181 assert.strictEqual(yield* Ref.get(released), 1)
182 }))
183
184 it.effect("shares one generation between concurrent first borrowers", () =>
185 Effect.gen(function*() {
186 const acquired = yield* Ref.make(0)
187 const released = yield* Ref.make(0)
188 const firstStarted = yield* Deferred.make<void>()
189 const releaseFirst = yield* Deferred.make<void>()
190 const refScope = yield* Scope.make()
191 const scopeA = yield* Scope.make()
192 const scopeB = yield* Scope.make()
193 const ref = yield* RcRef.make({
194 acquire: Effect.acquireRelease(
195 Effect.gen(function*() {
196 const id = yield* Ref.updateAndGet(acquired, (n) => n + 1)
197 if (id === 1) {
198 yield* Deferred.succeed(firstStarted, undefined)
199 yield* Deferred.await(releaseFirst)
200 }
201 return { id }
202 }),
203 () => Ref.update(released, (n) => n + 1)
204 )
205 }).pipe(Scope.provide(refScope))
206
207 const fiberA = yield* RcRef.get(ref).pipe(
208 Scope.provide(scopeA),
209 Effect.forkChild({ startImmediately: true })
210 )
211 yield* Deferred.await(firstStarted)
212 const fiberB = yield* RcRef.get(ref).pipe(
213 Scope.provide(scopeB),
214 Effect.forkChild({ startImmediately: true })
215 )
216 yield* Deferred.succeed(releaseFirst, undefined)
217
218 const exitA = yield* Fiber.await(fiberA)
219 const exitB = yield* Fiber.await(fiberB)
220 const generationA = yield* Fiber.join(fiberA)
221 const generationB = yield* Fiber.join(fiberB)
222 const closeAExit = yield* Scope.close(scopeA, Exit.void).pipe(Effect.exit)
223 const closeBExit = yield* Scope.close(scopeB, Exit.void).pipe(Effect.exit)
224 const refCloseExit = yield* Scope.close(refScope, Exit.void).pipe(Effect.exit)
225
226 assert.deepStrictEqual(
227 {
228 exitA,
229 exitB,
230 closeAExit,
231 closeBExit,
232 refCloseExit,
233 sameGeneration: generationA === generationB,
234 acquired: yield* Ref.get(acquired),
235 released: yield* Ref.get(released)
236 },
237 {
238 exitA: Exit.succeed(generationA),
239 exitB: Exit.succeed(generationA),
240 closeAExit: Exit.void,
241 closeBExit: Exit.void,
242 refCloseExit: Exit.void,
243 sameGeneration: true,
244 acquired: 1,
245 released: 1
246 }
247 )
248 }))
249
250 it.effect("idleTimeToLive reuses and releases resources", () =>
251 Effect.gen(function*() {
252 let acquired = 0
253 let released = 0
254 const ref = yield* RcRef.make({
255 acquire: Effect.acquireRelease(
256 Effect.sync(() => {
257 acquired++
258 return "foo"
259 }),
260 () =>
261 Effect.sync(() => {
262 released++
263 })
264 ),
265 idleTimeToLive: "10 millis"
266 })
267
268 assert.strictEqual(acquired, 0)
269 assert.strictEqual(yield* Effect.scoped(RcRef.get(ref)), "foo")
270 assert.strictEqual(acquired, 1)
271 assert.strictEqual(released, 0)
272
273 yield* TestClock.adjust("5 millis")
274 assert.strictEqual(yield* Effect.scoped(RcRef.get(ref)), "foo")
275 assert.strictEqual(acquired, 1)
276 assert.strictEqual(released, 0)
277
278 yield* TestClock.adjust("9 millis")
279 assert.strictEqual(released, 0)
280 yield* TestClock.adjust("1 millis")
281 assert.strictEqual(released, 1)
282
283 assert.strictEqual(yield* Effect.scoped(RcRef.get(ref)), "foo")
284 assert.strictEqual(acquired, 2)
285 assert.strictEqual(released, 1)
286
287 yield* TestClock.adjust("10 millis")
288 assert.strictEqual(released, 2)
289 }))
290
291 it.effect("idleTimeToLive 0 releases resources after a zero-duration sleep", () =>
292 Effect.gen(function*() {
293 const clock = yield* Clock.Clock
294 const sleeps: Array<number> = []
295 let released = 0
296 const ref = yield* RcRef.make({
297 acquire: Effect.acquireRelease(
298 Effect.succeed("foo"),
299 () =>
300 Effect.sync(() => {
301 released++
302 })
303 ),
304 idleTimeToLive: 0
305 }).pipe(Effect.provideService(Clock.Clock, {
306 ...clock,
307 sleep: (duration) => {
308 sleeps.push(Duration.toMillis(duration))
309 return clock.sleep(duration)
310 }
311 }))
312
313 yield* Effect.scoped(RcRef.get(ref))
314 assert.deepStrictEqual(sleeps, [0])
315 assert.strictEqual(released, 1)
316 }))
317
318 it.effect("an interrupted first get does not pin the resource", () =>
319 Effect.gen(function*() {
320 let acquired = 0
321 let released = 0
322 const ref = yield* RcRef.make({
323 acquire: Effect.acquireRelease(Effect.sync(() => ++acquired), () => Effect.sync(() => released++))
324 })
325 const fiber = yield* Effect.forkChild(
326 Effect.scoped(RcRef.get(ref)).pipe(Effect.provideService(Scheduler.MaxOpsBeforeYield, 29)),
327 { startImmediately: true }
328 )
329 yield* Fiber.interrupt(fiber)
330 yield* Effect.scoped(RcRef.get(ref))
331 assert.strictEqual(acquired, released)
332 }))
333
334 it.effect("an interrupted get of an acquired resource does not pin it", () =>
335 Effect.gen(function*() {
336 let released = 0
337 const ref = yield* RcRef.make({
338 acquire: Effect.acquireRelease(Effect.void, () => Effect.sync(() => released++))
339 })
340 const holder = yield* Effect.forkChild(Effect.scoped(Effect.andThen(RcRef.get(ref), Effect.never)), {
341 startImmediately: true
342 })
343 const fiber = yield* Effect.forkChild(
344 Effect.scoped(RcRef.get(ref)).pipe(Effect.provideService(Scheduler.MaxOpsBeforeYield, 8)),
345 { startImmediately: true }
346 )
347 yield* Fiber.interrupt(fiber)
348 yield* Fiber.interrupt(holder)
349 assert.strictEqual(released, 1)
350 }))
351
352 it.effect("invalidating a resource while a get revives it releases the resource", () =>
353 Effect.gen(function*() {
354 let released = 0
355 const ref = yield* RcRef.make({
356 acquire: Effect.acquireRelease(Effect.void, () => Effect.sync(() => released++)),
357 idleTimeToLive: "1 minute"
358 }).pipe(Effect.provideService(Scheduler.MaxOpsBeforeYield, 4))
359 yield* Effect.scoped(RcRef.get(ref))
360 const getter = yield* Effect.forkChild(Effect.scoped(RcRef.get(ref)), { startImmediately: true })
361 yield* RcRef.invalidate(ref)
362 yield* Fiber.join(getter)
363 assert.strictEqual(released, 1)
364 }))
365
366 it.effect("closing the ref while an idle resource is being released runs every finalizer", () =>
367 Effect.gen(function*() {
368 const gate = yield* Deferred.make<void>()
369 let released = 0
370 const scope = yield* Scope.make()
371 const ref = yield* RcRef.make({
372 acquire: Effect.gen(function*() {
373 yield* Effect.addFinalizer(() => Effect.sync(() => released++))
374 yield* Effect.addFinalizer(() => Effect.andThen(Deferred.await(gate), Effect.sync(() => released++)))
375 }),
376 idleTimeToLive: "1 second"
377 }).pipe(Scope.provide(scope))
378 yield* Effect.scoped(RcRef.get(ref))
379 yield* TestClock.adjust("1 second")
380 const close = yield* Effect.forkChild(Scope.close(scope, Exit.void), { startImmediately: true })
381 yield* Deferred.succeed(gate, void 0)
382 yield* Fiber.join(close)
383 assert.strictEqual(released, 2)
384 }))
385
386 describe("interruption", () => {
387 it.effect("shares an acquisition until every get is interrupted, then starts fresh", () =>
388 Effect.gen(function*() {
389 let acquired = 0
390 const started = yield* Deferred.make<void>()
391 const finalizing = yield* Deferred.make<void>()
392 const finishFinalizer = yield* Deferred.make<void>()
393 const ref = yield* RcRef.make({
394 acquire: Effect.suspend(() => {
395 if (++acquired > 1) return Effect.succeed(acquired)
396 return Deferred.succeed(started, void 0).pipe(
397 Effect.andThen(Effect.never),
398 Effect.onInterrupt(() =>
399 Deferred.succeed(finalizing, void 0).pipe(Effect.andThen(Deferred.await(finishFinalizer)))
400 )
401 )
402 })
403 })
404
405 const first = yield* Effect.forkChild(Effect.scoped(RcRef.get(ref)), { startImmediately: true })
406 yield* Deferred.await(started)
407 const second = yield* Effect.forkChild(Effect.scoped(RcRef.get(ref)), { startImmediately: true })
408 yield* Fiber.interrupt(first)
409 assert.isFalse(yield* Deferred.isDone(finalizing))
410
411 // A get made while the abandoned acquisition finalizes starts a fresh one.
412 const interruptSecond = yield* Effect.forkChild(Fiber.interrupt(second), { startImmediately: true })
413 yield* Deferred.await(finalizing)
414 const fresh = yield* Effect.forkChild(Effect.scoped(RcRef.get(ref)), { startImmediately: true })
415 yield* Deferred.succeed(finishFinalizer, void 0)
416 yield* Fiber.join(interruptSecond)
417 assert.deepStrictEqual(yield* Fiber.await(fresh), Exit.succeed(2))
418 }))
419
420 it.effect.each([false, true])(
421 "releases references of waiters interrupted after the acquisition publishes them (interrupt all: %s)",
422 (interruptAll) =>
423 Effect.gen(function*() {
424 // Sweep the op budget so the acquisition yields at every step between
425 // publishing its references and its fiber completing.
426 for (let ops = 3; ops <= 64; ops++) {
427 let acquired = 0
428 let released = 0
429 const gate = yield* Deferred.make<void>()
430 const ref = yield* RcRef.make({
431 acquire: Effect.acquireRelease(
432 Deferred.await(gate).pipe(Effect.map(() => ++acquired)),
433 () =>
434 Effect.sync(() => {
435 released++
436 })
437 )
438 })
439 const withBudget = Effect.provideService(Scheduler.MaxOpsBeforeYield, ops)
440
441 const holderScope = yield* Scope.make()
442 const holder = yield* Effect.forkChild(withBudget(RcRef.get(ref).pipe(Scope.provide(holderScope))), {
443 startImmediately: true
444 })
445 const waiter = yield* Effect.forkChild(withBudget(Effect.scoped(RcRef.get(ref))), {
446 startImmediately: true
447 })
448 for (let i = 0; i < 200; i++) yield* Effect.yieldNow
449 yield* Deferred.succeed(gate, void 0)
450 yield* Fiber.interrupt(waiter)
451 if (interruptAll) yield* Fiber.interrupt(holder)
452 yield* Fiber.await(holder)
453 yield* Scope.close(holderScope, Exit.void)
454
455 assert.deepStrictEqual([acquired, released], [1, 1], `${ops} ops`)
456 assert.strictEqual(yield* Effect.scoped(RcRef.get(ref)), 2, `${ops} ops`)
457 }
458 })
459 )
460
461 it.effect("closing the ref interrupts a pending acquisition and closes its scope", () =>
462 Effect.gen(function*() {
463 let released = 0
464 const started = yield* Deferred.make<void>()
465 const refScope = yield* Scope.make()
466 const ref = yield* RcRef.make({
467 acquire: Effect.gen(function*() {
468 yield* Effect.addFinalizer(() =>
469 Effect.sync(() => {
470 released++
471 })
472 )
473 yield* Deferred.succeed(started, void 0)
474 return yield* Effect.never
475 })
476 }).pipe(Scope.provide(refScope))
477
478 const getter = yield* Effect.forkChild(Effect.scoped(RcRef.get(ref)), { startImmediately: true })
479 yield* Deferred.await(started)
480 yield* Scope.close(refScope, Exit.void)
481
482 assert.strictEqual(released, 1)
483 assert.isTrue(Exit.hasInterrupts(yield* Fiber.await(getter)))
484 }))
485 })
486})