1import { assert, describe, it } from "@effect/vitest"
2import { deepStrictEqual, strictEqual } from "@effect/vitest/utils"
3import { Clock, Deferred, Duration, Effect, Exit, Fiber, pipe, Pool, Ref, Schedule, Scheduler, Scope } from "effect"
4import { TestClock } from "effect/testing"
5import { collectGarbage } from "./utils/gc.ts"
6
7const waitUntil = (predicate: () => boolean) =>
8 Effect.repeat(Effect.andThen(Effect.yieldNow, Effect.sync(predicate)), { until: (done) => done })
9
10describe("Pool", () => {
11 it.effect("preallocates pool items", () =>
12 Effect.gen(function*() {
13 const count = yield* Ref.make(0)
14 const get = Effect.acquireRelease(
15 Ref.updateAndGet(count, (n) => n + 1),
16 () => Ref.update(count, (n) => n - 1)
17 )
18 yield* Pool.make({ acquire: get, size: 10 })
19 yield* Effect.repeat(Ref.get(count), { until: (n) => n === 10 })
20 const result = yield* Ref.get(count)
21 strictEqual(result, 10)
22 }))
23
24 // it.effect("benchmark", () =>
25 // Effect.gen(function*() {
26 // const get = Effect.succeed("resource")
27 // const pool = yield* Pool.make({ acquire: get, size: 10 })
28 // yield* Pool.get(pool).pipe(
29 // Effect.scoped,
30 // Effect.repeatN(10000),
31 // Console.withTime("Pool.get")
32 // )
33 // }))
34
35 it.effect("cleans up items when shut down", () =>
36 Effect.gen(function*() {
37 const count = yield* Ref.make(0)
38 const get = Effect.acquireRelease(
39 Ref.updateAndGet(count, (n) => n + 1),
40 () => Ref.update(count, (n) => n - 1)
41 )
42 const scope = yield* Scope.make()
43 yield* Scope.provide(Pool.make({ acquire: get, size: 10 }), scope)
44 yield* Effect.repeat(Ref.get(count), { until: (n) => n === 10 })
45 yield* Scope.close(scope, Exit.succeed(void 0))
46 const result = yield* Ref.get(count)
47 strictEqual(result, 0)
48 }))
49
50 it.effect("defects don't prevent cleanup", () =>
51 Effect.gen(function*() {
52 const count = yield* Ref.make(0)
53 const get = Effect.acquireRelease(
54 Ref.updateAndGet(count, (n) => n + 1),
55 () => Effect.andThen(Ref.update(count, (n) => n - 1), Effect.die("boom"))
56 )
57 const scope = yield* Scope.make()
58 yield* Scope.provide(Pool.make({ acquire: get, size: 10 }), scope)
59 yield* Effect.repeat(Ref.get(count), { until: (n) => n === 10 })
60 yield* Scope.close(scope, Exit.succeed(void 0))
61 const result = yield* Ref.get(count)
62 strictEqual(result, 0)
63 }))
64
65 it.effect("acquire one item", () =>
66 Effect.gen(function*() {
67 const count = yield* Ref.make(0)
68 const get = Effect.acquireRelease(
69 Ref.updateAndGet(count, (n) => n + 1),
70 () => Ref.update(count, (n) => n - 1)
71 )
72 const pool = yield* Pool.make({ acquire: get, size: 10 })
73 yield* Effect.repeat(Ref.get(count), { until: (n) => n === 10 })
74 const item = yield* Pool.get(pool)
75 strictEqual(item, 1)
76 }))
77
78 it.effect("keeps failed background capacity until a borrower reaches it", () =>
79 Effect.gen(function*() {
80 const releaseFailure = yield* Deferred.make<void>()
81 const failed = yield* Deferred.make<void>()
82 let attempts = 0
83 const pool = yield* Pool.makeWithTTL({
84 acquire: Effect.suspend(() => {
85 const attempt = ++attempts
86 return attempt === 2
87 ? Effect.andThen(
88 Deferred.await(releaseFailure),
89 Effect.andThen(
90 Deferred.succeed(failed, undefined),
91 Effect.fail("background")
92 )
93 )
94 : Effect.succeed(attempt)
95 }),
96 min: 2,
97 max: 3,
98 timeToLive: Duration.infinity
99 })
100 yield* waitUntil(() => pool.state.availableHead?.exit._tag === "Success")
101 yield* Deferred.succeed(releaseFailure, undefined)
102 yield* Deferred.await(failed)
103 yield* waitUntil(() => pool.state.items.size === 2)
104 for (let i = 0; i < 5; i++) {
105 strictEqual(yield* Pool.use(pool, Effect.succeed), 1)
106 }
107 strictEqual(attempts, 2)
108 strictEqual(pool.state.items.size, 2)
109 const owner = yield* Scope.make()
110 strictEqual(yield* Scope.provide(Pool.get(pool), owner), 1)
111 strictEqual(yield* Pool.use(pool, Effect.succeed), 3)
112 strictEqual(attempts, 3)
113 yield* Scope.close(owner, Exit.void)
114 }))
115
116 it.live("blocks when item not available", () =>
117 Effect.gen(function*() {
118 const count = yield* Ref.make(0)
119 const get = Effect.acquireRelease(
120 Ref.updateAndGet(count, (n) => n + 1),
121 () => Ref.update(count, (n) => n - 1)
122 )
123 const pool = yield* Pool.make({ acquire: get, size: 10 })
124 yield* Effect.repeat(Ref.get(count), { until: (n) => n === 10 })
125 yield* Effect.all(Effect.replicate(10)(Pool.get(pool)))
126 const result = yield* Effect.scoped(Pool.get(pool)).pipe(
127 Effect.forkChild({ startImmediately: true })
128 )
129 yield* Effect.sleep(10)
130 assert.isUndefined(result.pollUnsafe())
131 }))
132
133 it.effect("reuse released items", () =>
134 Effect.gen(function*() {
135 const count = yield* Ref.make(0)
136 const get = Effect.acquireRelease(
137 Ref.updateAndGet(count, (n) => n + 1),
138 () => Ref.update(count, (n) => n - 1)
139 )
140 const pool = yield* Pool.make({ acquire: get, size: 10 })
141 yield* Effect.replicateEffect(99)(Effect.scoped(Pool.get(pool)))
142 const result = yield* Ref.get(count)
143 strictEqual(result, 10)
144 }))
145
146 it.effect("invalidate item", () =>
147 Effect.gen(function*() {
148 const count = yield* Ref.make(0)
149 const get = Effect.acquireRelease(
150 Ref.updateAndGet(count, (n) => n + 1),
151 () => Ref.update(count, (n) => n - 1)
152 )
153 const pool = yield* Pool.make({ acquire: get, size: 10 })
154 yield* Effect.repeat(Ref.get(count), { until: (n) => n === 10 })
155 yield* Pool.invalidate(pool, 1)
156 yield* Effect.repeat(Ref.get(count), { until: (n) => n === 10 })
157 const result = yield* Effect.scoped(Pool.get(pool))
158 const value = yield* Ref.get(count)
159 strictEqual(result, 2)
160 strictEqual(value, 10)
161 }))
162
163 it.effect("invalidate all items in pool and check that pool.get doesn't hang forever", () =>
164 Effect.gen(function*() {
165 const allocated = yield* Ref.make(0)
166 const finalized = yield* Ref.make(0)
167 const get = Effect.acquireRelease(
168 Ref.updateAndGet(allocated, (n) => n + 1),
169 () => Ref.update(finalized, (n) => n + 1)
170 )
171 const pool = yield* Pool.make({ acquire: get, size: 2 })
172 yield* Effect.repeat(Ref.get(allocated), { until: (n) => n === 2 })
173 yield* Pool.invalidate(pool, 1)
174 yield* Pool.invalidate(pool, 2)
175 const result = yield* Effect.scoped(Pool.get(pool))
176 const allocatedCount = yield* Ref.get(allocated)
177 const finalizedCount = yield* Ref.get(finalized)
178 strictEqual(result, 3)
179 strictEqual(allocatedCount, 4)
180 strictEqual(finalizedCount, 2)
181 }))
182
183 it.live("retry on failed acquire should not exhaust pool", () =>
184 Effect.gen(function*() {
185 const acquire = Effect.as(Effect.fail("error"), 1)
186 const pool = yield* Pool.makeWithTTL({ acquire, min: 0, max: 1, timeToLive: Duration.infinity })
187 const result = yield* pipe(
188 Effect.scoped(Effect.retry(Pool.get(pool), { times: 5 })),
189 Effect.timeoutOrElse({
190 orElse: () => Effect.fail("timeout"),
191 duration: Duration.seconds(1)
192 }),
193 Effect.flip
194 )
195 strictEqual(result, "error")
196 }))
197
198 it.effect("compositional retry", () =>
199 Effect.gen(function*() {
200 const cond = (i: number) => (i <= 10 ? Effect.fail(i) : Effect.succeed(i))
201 const count = yield* Ref.make(0)
202 const get = Effect.acquireRelease(
203 Ref.updateAndGet(count, (n) => n + 1).pipe(
204 Effect.flatMap(cond)
205 ),
206 () => Ref.update(count, (n) => n - 1)
207 )
208 const pool = yield* Pool.make({ acquire: get, size: 10 })
209 const result = yield* Effect.retry(Effect.scoped(Pool.get(pool)), Schedule.forever)
210 strictEqual(result, 11)
211 }))
212
213 describe("usage TTL idle timeout", () => {
214 it.effect("shrinks to min in idle order rather than acquisition order", () =>
215 Effect.gen(function*() {
216 let acquired = 0
217 const finalized: Array<number> = []
218 const pool = yield* Pool.makeWithTTL({
219 acquire: Effect.acquireRelease(
220 Effect.sync(() => ++acquired),
221 (value) => Effect.sync(() => finalized.push(value))
222 ),
223 min: 1,
224 max: 3,
225 timeToLive: 1000,
226 timeToLiveStrategy: "usage"
227 })
228 const first = yield* Scope.fork(yield* Effect.scope)
229 const second = yield* Scope.fork(yield* Effect.scope)
230 const third = yield* Scope.fork(yield* Effect.scope)
231 strictEqual(yield* Pool.get(pool).pipe(Scope.provide(first)), 1)
232 strictEqual(yield* Pool.get(pool).pipe(Scope.provide(second)), 2)
233 strictEqual(yield* Pool.get(pool).pipe(Scope.provide(third)), 3)
234
235 yield* TestClock.adjust(100)
236 yield* Scope.close(second, Exit.void)
237 yield* TestClock.adjust(100)
238 yield* Scope.close(first, Exit.void)
239 yield* TestClock.adjust(100)
240 yield* Scope.close(third, Exit.void)
241 yield* TestClock.adjust(700)
242 deepStrictEqual(finalized, [])
243 yield* TestClock.adjust(1000)
244 deepStrictEqual(finalized, [2, 1])
245 strictEqual(pool.state.items.size, 1)
246 yield* TestClock.adjust(2000)
247 deepStrictEqual(finalized, [2, 1])
248 strictEqual(yield* Pool.use(pool, Effect.succeed), 3)
249 strictEqual(acquired, 3)
250 }))
251
252 it.effect("retains items used every 3 seconds across TTL sweeps", () =>
253 Effect.gen(function*() {
254 let acquired = 0
255 const pool = yield* Pool.makeWithTTL({
256 acquire: Effect.sync(() => ++acquired),
257 min: 0,
258 max: 10,
259 timeToLive: "10 seconds",
260 timeToLiveStrategy: "usage"
261 })
262 for (let i = 0; i < 20; i++) {
263 yield* Effect.scoped(Pool.get(pool))
264 yield* TestClock.adjust("3 seconds")
265 }
266 strictEqual(acquired, 1)
267 }))
268
269 it.effect("measures idle time with the pool clock when a release runs under another clock", () =>
270 Effect.gen(function*() {
271 const pool = yield* Pool.makeWithTTL({
272 acquire: Effect.succeed("resource"),
273 min: 0,
274 max: 1,
275 timeToLive: 1000,
276 timeToLiveStrategy: "usage"
277 })
278 const clock = yield* Clock.Clock
279 yield* Pool.use(pool, Effect.succeed).pipe(
280 Effect.provideService(Clock.Clock, { ...clock, currentTimeMillisUnsafe: () => 1_000_000 })
281 )
282 yield* TestClock.adjust(999)
283 strictEqual(pool.state.items.size, 1)
284 yield* TestClock.adjust(1)
285 strictEqual(pool.state.items.size, 0)
286 }))
287 })
288
289 it.effect("max pool size", () =>
290 Effect.gen(function*() {
291 const deferred = yield* Deferred.make<void>()
292 const count = yield* Ref.make(0)
293 const acquire = Effect.acquireRelease(
294 Ref.updateAndGet(count, (n) => n + 1),
295 () => Ref.update(count, (n) => n - 1)
296 )
297 const pool = yield* Pool.makeWithTTL({
298 acquire,
299 min: 10,
300 max: 15,
301 timeToLive: Duration.seconds(60)
302 })
303 yield* pipe(
304 Effect.scoped(Effect.andThen(
305 Pool.get(pool),
306 Deferred.await(deferred)
307 )),
308 Effect.forkChild,
309 Effect.repeat({ times: 14 })
310 )
311 yield* Effect.repeat(Ref.get(count), { until: (n) => n === 15 })
312 yield* Deferred.succeed(deferred, void 0)
313 const max = yield* Ref.get(count)
314 yield* TestClock.adjust(Duration.seconds(60))
315 const min = yield* Ref.get(count)
316 strictEqual(min, 10)
317 strictEqual(max, 15)
318 }))
319
320 it.effect("max pool size with concurrency: 3", () =>
321 Effect.gen(function*() {
322 const deferred = yield* Deferred.make<void>()
323 const count = yield* Ref.make(0)
324 const acquire = Effect.acquireRelease(
325 Ref.updateAndGet(count, (n) => n + 1),
326 () => Ref.update(count, (n) => n - 1)
327 )
328 const pool = yield* Pool.makeWithTTL({
329 acquire,
330 min: 10,
331 max: 15,
332 concurrency: 3,
333 timeToLive: Duration.seconds(60)
334 })
335 yield* pipe(
336 Effect.scoped(Effect.andThen(
337 Pool.get(pool),
338 Deferred.await(deferred)
339 )),
340 Effect.forkChild,
341 Effect.repeat({ times: 14 * 3 })
342 )
343 yield* Effect.repeat(Ref.get(count), { until: (n) => n === 15 })
344 yield* Deferred.succeed(deferred, void 0)
345 const max = yield* Ref.get(count)
346 yield* TestClock.adjust(Duration.seconds(60))
347 const min = yield* Ref.get(count)
348 strictEqual(min, 10)
349 strictEqual(max, 15)
350 }))
351
352 it.effect("usage TTL retains shared items with active borrowers", () =>
353 Effect.gen(function*() {
354 let acquired = 0
355 const pool = yield* Pool.makeWithTTL({
356 acquire: Effect.sync(() => ++acquired),
357 min: 0,
358 max: 2,
359 concurrency: 2,
360 timeToLive: Duration.seconds(60)
361 })
362
363 const scope1 = yield* Scope.fork(yield* Effect.scope)
364 strictEqual(yield* Scope.provide(Pool.get(pool), scope1), 1)
365 strictEqual(yield* Pool.get(pool), 1)
366 strictEqual(yield* Effect.scoped(Pool.get(pool)), 2)
367 yield* TestClock.adjust(Duration.seconds(60))
368 strictEqual(pool.state.invalidated.size, 0)
369 strictEqual(pool.state.items.size, 1)
370 yield* Scope.close(scope1, Exit.void)
371 strictEqual(yield* Pool.get(pool), 1)
372 strictEqual(yield* Pool.get(pool), 3)
373 strictEqual(yield* Pool.get(pool), 3)
374 strictEqual(acquired, 3)
375 }))
376
377 it.effect("reserve takes an item out of shared circulation", () =>
378 Effect.gen(function*() {
379 const count = yield* Ref.make(0)
380 const acquire = Effect.acquireRelease(
381 Ref.updateAndGet(count, (n) => n + 1),
382 () => Ref.update(count, (n) => n - 1)
383 )
384 const pool = yield* Pool.makeWithTTL({
385 acquire,
386 min: 0,
387 max: 2,
388 concurrency: 2,
389 timeToLive: Duration.seconds(60)
390 })
391
392 const scope1 = yield* Scope.make()
393 strictEqual(yield* Scope.provide(Pool.get(pool), scope1), 1)
394 const reservation = yield* Scope.make()
395 yield* Scope.provide(Pool.reserve(pool, 1), reservation)
396 // The reserved item counts as fully used, so the pool opens a second one
397 // and both checkouts share it.
398 strictEqual(yield* Scope.provide(Pool.get(pool), scope1), 2)
399 strictEqual(yield* Scope.provide(Pool.get(pool), scope1), 2)
400 // Everything is saturated, so this checkout has to wait for the
401 // reservation to release its capacity.
402 const fiber = yield* pipe(
403 Scope.provide(Pool.get(pool), scope1),
404 Effect.forkChild({ startImmediately: true })
405 )
406 yield* Scope.close(reservation, Exit.void)
407 strictEqual(yield* Fiber.join(fiber), 1)
408 strictEqual(yield* Ref.get(count), 2)
409 yield* Scope.close(scope1, Exit.void)
410 }))
411
412 it.effect("releasing a shared borrower preserves an active reservation", () =>
413 Effect.gen(function*() {
414 const pool = yield* Pool.make({ acquire: Effect.succeed("resource"), size: 1, concurrency: 2 })
415 const owner = yield* Scope.make()
416 const borrower = yield* Scope.make()
417 const reservation = yield* Scope.make()
418 yield* Pool.get(pool).pipe(Scope.provide(owner))
419 yield* Pool.get(pool).pipe(Scope.provide(borrower))
420 yield* Pool.reserve(pool, "resource").pipe(Scope.provide(reservation))
421
422 yield* Scope.close(borrower, Exit.void)
423 const next = yield* Pool.use(pool, Effect.succeed).pipe(Effect.forkChild({ startImmediately: true }))
424 const beforeRelease = next.pollUnsafe()
425 yield* Scope.close(reservation, Exit.void)
426 const result = yield* Fiber.join(next)
427 yield* Scope.close(owner, Exit.void)
428
429 assert.isUndefined(beforeRelease)
430 assert.strictEqual(result, "resource")
431 }))
432
433 it.effect("overlapping reservations keep an item reserved until both close", () =>
434 Effect.gen(function*() {
435 const pool = yield* Pool.make({ acquire: Effect.succeed("resource"), size: 1, concurrency: 2 })
436 const owner = yield* Scope.make()
437 const first = yield* Scope.make()
438 const second = yield* Scope.make()
439 yield* Pool.get(pool).pipe(Scope.provide(owner))
440 yield* Pool.reserve(pool, "resource").pipe(Scope.provide(first))
441 yield* Pool.reserve(pool, "resource").pipe(Scope.provide(second))
442 // One lease plus one reservation; the overlapping reservation must not
443 // count usage again.
444 assert.strictEqual(pool.state.usage, 2)
445
446 yield* Scope.close(first, Exit.void)
447 assert.strictEqual(pool.state.usage, 2)
448 const next = yield* Pool.use(pool, Effect.succeed).pipe(Effect.forkChild({ startImmediately: true }))
449 const beforeRelease = next.pollUnsafe()
450 yield* Scope.close(second, Exit.void)
451 const result = yield* Fiber.join(next)
452 yield* Scope.close(owner, Exit.void)
453
454 assert.isUndefined(beforeRelease)
455 assert.strictEqual(result, "resource")
456 assert.strictEqual(pool.state.usage, 0)
457 }))
458
459 it.effect("releasing a reservation wakes all available slots after borrowers return", () =>
460 Effect.gen(function*() {
461 const pool = yield* Pool.make({ acquire: Effect.succeed("resource"), size: 1, concurrency: 2 })
462 const owner = yield* Scope.make()
463 const reservation = yield* Scope.make()
464 const borrowers = yield* Scope.make()
465 yield* Pool.get(pool).pipe(Scope.provide(owner))
466 yield* Pool.reserve(pool, "resource").pipe(Scope.provide(reservation))
467 yield* Scope.close(owner, Exit.void)
468
469 const first = yield* Pool.get(pool).pipe(Scope.provide(borrowers), Effect.forkChild({ startImmediately: true }))
470 const second = yield* Pool.get(pool).pipe(Scope.provide(borrowers), Effect.forkChild({ startImmediately: true }))
471 const beforeRelease = [first.pollUnsafe(), second.pollUnsafe()]
472 yield* Scope.close(reservation, Exit.void)
473 yield* TestClock.adjust(0)
474 const afterRelease = [first.pollUnsafe(), second.pollUnsafe()]
475 yield* Fiber.interrupt(first)
476 yield* Fiber.interrupt(second)
477 yield* Scope.close(borrowers, Exit.void)
478
479 assert.deepStrictEqual(beforeRelease, [undefined, undefined])
480 assert.deepStrictEqual(afterRelease, [Exit.succeed("resource"), Exit.succeed("resource")])
481 assert.strictEqual(pool.state.usage, 0)
482 }))
483
484 it.effect("usage TTL preserves reservations and waits for capacity", () =>
485 Effect.gen(function*() {
486 let acquired = 0
487 const pool = yield* Pool.makeWithTTL({
488 acquire: Effect.sync(() => ++acquired),
489 min: 0,
490 max: 2,
491 concurrency: 2,
492 timeToLive: 1000
493 })
494 const owner = yield* Scope.fork(yield* Effect.scope)
495 const reservation = yield* Scope.fork(yield* Effect.scope)
496 yield* Pool.get(pool).pipe(Scope.provide(owner))
497 yield* Pool.reserve(pool, 1).pipe(Scope.provide(reservation))
498 yield* Pool.use(pool, Effect.succeed)
499 yield* TestClock.adjust(1000)
500 assert.strictEqual(pool.state.items.size, 1)
501 assert.strictEqual(pool.state.invalidated.size, 0)
502
503 const first = yield* Pool.get(pool).pipe(Scope.provide(owner))
504 const second = yield* Pool.get(pool).pipe(Scope.provide(owner))
505 const next = yield* Pool.use(pool, Effect.succeed).pipe(Effect.forkChild({ startImmediately: true }))
506 assert.isUndefined(next.pollUnsafe())
507 yield* Scope.close(reservation, Exit.void)
508 const result = yield* Fiber.join(next)
509 yield* Scope.close(owner, Exit.void)
510
511 assert.deepStrictEqual([first, second], [3, 3])
512 assert.strictEqual(result, 1)
513 assert.strictEqual(acquired, 3)
514 assert.strictEqual(pool.state.usage, 0)
515 }))
516
517 it.effect("reserve is a no-op with concurrency one", () =>
518 Effect.gen(function*() {
519 const pool = yield* Pool.make({ acquire: Effect.succeed("resource"), size: 1 })
520 const owner = yield* Scope.make()
521 const reservation = yield* Scope.make()
522 yield* Pool.get(pool).pipe(Scope.provide(owner))
523 yield* Pool.reserve(pool, "resource").pipe(Scope.provide(reservation))
524 yield* Scope.close(owner, Exit.void)
525
526 const next = yield* Pool.use(pool, Effect.succeed).pipe(Effect.forkChild({ startImmediately: true }))
527 const beforeRelease = next.pollUnsafe()
528 yield* Scope.close(reservation, Exit.void)
529 yield* Fiber.interrupt(next)
530
531 assert.deepStrictEqual(beforeRelease, Exit.succeed("resource"))
532 }))
533
534 it.effect.each([false, true])(
535 "reservation cleanup does not revive removed items (shutdown: %s)",
536 (shutdown) =>
537 Effect.gen(function*() {
538 let acquired = 0
539 const released: Array<number> = []
540 const poolScope = yield* Scope.make()
541 const owner = yield* Scope.make()
542 const reservation = yield* Scope.make()
543 const pool = yield* Pool.make({
544 acquire: Effect.acquireRelease(
545 Effect.sync(() => ++acquired),
546 (item) => Effect.sync(() => released.push(item))
547 ),
548 size: 1,
549 concurrency: 2
550 }).pipe(Scope.provide(poolScope))
551 yield* Pool.get(pool).pipe(Scope.provide(owner))
552 yield* Pool.reserve(pool, 1).pipe(Scope.provide(reservation))
553 yield* Scope.close(owner, Exit.void)
554 if (shutdown) {
555 yield* Scope.close(poolScope, Exit.void)
556 } else {
557 yield* Pool.invalidate(pool, 1)
558 }
559 yield* Scope.close(reservation, Exit.void)
560 const next = yield* Effect.exit(Pool.use(pool, Effect.succeed))
561 yield* Scope.close(poolScope, Exit.void)
562
563 if (shutdown) {
564 assert.isTrue(Exit.hasInterrupts(next))
565 assert.deepStrictEqual(released, [1])
566 } else {
567 assert.deepStrictEqual(next, Exit.succeed(2))
568 assert.deepStrictEqual(released, [1, 2])
569 }
570 assert.strictEqual(pool.state.usage, 0)
571 })
572 )
573
574 it.effect("scale to zero", () =>
575 Effect.gen(function*() {
576 const deferred = yield* Deferred.make<void>()
577 const count = yield* Ref.make(0)
578 const acquire = Effect.acquireRelease(
579 Ref.updateAndGet(count, (n) => n + 1),
580 () => Ref.update(count, (n) => n - 1)
581 )
582 const pool = yield* Pool.makeWithTTL({
583 acquire,
584 min: 0,
585 max: 10,
586 concurrency: 3,
587 timeToLive: Duration.seconds(60)
588 })
589 yield* pipe(
590 Effect.scoped(Effect.andThen(
591 Pool.get(pool),
592 Deferred.await(deferred)
593 )),
594 Effect.forkChild,
595 Effect.repeat({ times: 29 })
596 )
597 yield* Effect.repeat(Ref.get(count), { until: (n) => n === 10 })
598 yield* Deferred.succeed(deferred, void 0)
599 const max = yield* Ref.get(count)
600 yield* TestClock.adjust(Duration.seconds(60))
601 const min = yield* Ref.get(count)
602 strictEqual(min, 0)
603 strictEqual(max, 10)
604 }))
605
606 it.effect("max pool size creation strategy", () =>
607 Effect.gen(function*() {
608 const invalidated = yield* Ref.make(0)
609 const acquire = Effect.acquireRelease(
610 Effect.succeed("resource"),
611 () => Ref.update(invalidated, (n) => n + 1)
612 )
613 const pool = yield* Pool.makeWithTTL({
614 acquire,
615 min: 10,
616 max: 15,
617 timeToLive: Duration.seconds(60),
618 timeToLiveStrategy: "creation"
619 })
620 const scope = yield* Scope.make()
621 yield* Pool.get(pool).pipe(
622 Effect.repeat({ times: 14 }),
623 Scope.provide(scope)
624 )
625 const one = yield* Ref.get(invalidated)
626 yield* TestClock.adjust(Duration.seconds(60))
627 const two = yield* Ref.get(invalidated)
628 yield* Scope.close(scope, Exit.void)
629 const three = yield* Ref.get(invalidated)
630 strictEqual(one, 0)
631 strictEqual(two, 0)
632 strictEqual(three, 15)
633 }))
634
635 it.effect("shutdown robustness", () =>
636 Effect.gen(function*() {
637 const count = yield* Ref.make(0)
638 const get = Effect.acquireRelease(
639 Ref.updateAndGet(count, (n) => n + 1),
640 () => Ref.update(count, (n) => n - 1)
641 )
642 const scope = yield* Scope.make()
643 const pool = yield* Scope.provide(Pool.make({ acquire: get, size: 10 }), scope)
644 yield* pipe(
645 Effect.scoped(Pool.get(pool)),
646 Effect.forkChild,
647 Effect.repeat({ times: 99 })
648 )
649 yield* Scope.close(scope, Exit.succeed(void 0))
650 const result = yield* Effect.repeat(Ref.get(count), { until: (n) => n === 0 })
651 strictEqual(result, 0)
652 }))
653
654 it.effect("shutdown with pending takers", () =>
655 Effect.gen(function*() {
656 const count = yield* Ref.make(0)
657 const get = Effect.acquireRelease(
658 Ref.updateAndGet(count, (n) => n + 1),
659 () => Ref.update(count, (n) => n - 1)
660 )
661 const scope = yield* Scope.make()
662 const pool = yield* Scope.provide(Pool.make({ acquire: get, size: 10 }), scope)
663 yield* pipe(
664 Pool.get(pool),
665 Scope.provide(scope),
666 Effect.forkChild,
667 Effect.repeat({ times: 99 })
668 )
669 yield* Scope.close(scope, Exit.succeed(void 0))
670 const result = yield* Effect.repeat(Ref.get(count), { until: (n) => n === 0 })
671 strictEqual(result, 0)
672 }))
673
674 it.effect("get is interruptible", () =>
675 Effect.gen(function*() {
676 const count = yield* Ref.make(0)
677 const get = Effect.acquireRelease(
678 Ref.updateAndGet(count, (n) => n + 1),
679 () => Ref.update(count, (n) => n - 1)
680 )
681 const fiberId = Fiber.getCurrent()!.id
682 const pool = yield* Pool.make({ acquire: get, size: 10 })
683 yield* Effect.repeat(Pool.get(pool), { times: 9 })
684 const fiber = yield* Effect.forkChild(Pool.get(pool))
685 yield* Fiber.interrupt(fiber)
686 deepStrictEqual(fiber.pollUnsafe(), Exit.interrupt(fiberId))
687 }))
688
689 it.effect("get is interruptible with dynamic size", () =>
690 Effect.gen(function*() {
691 const get = Effect.never.pipe(Effect.forkScoped)
692 const fiberId = Fiber.getCurrent()!.id
693 const pool = yield* Pool.makeWithTTL({ acquire: get, min: 0, max: 10, timeToLive: Duration.infinity })
694 yield* Effect.repeat(Pool.get(pool), { times: 9 })
695 const fiber = yield* Effect.forkChild(Pool.get(pool))
696 yield* Fiber.interrupt(fiber)
697 deepStrictEqual(yield* Fiber.await(fiber), Exit.interrupt(fiberId))
698 }))
699
700 it.effect("interrupts pending gets without leaking usage", () =>
701 Effect.gen(function*() {
702 const pool = yield* Pool.make({ acquire: Effect.succeed("resource"), size: 1 })
703 const scope = yield* Scope.make()
704 yield* Scope.provide(Pool.get(pool), scope)
705 const fibers: Array<Fiber.Fiber<string>> = []
706 for (let i = 0; i < 10; i++) {
707 fibers.push(yield* Effect.forkChild(Pool.get(pool), { startImmediately: true }))
708 }
709 yield* waitUntil(() => pool.state.waiters.size === fibers.length)
710 yield* Effect.all(fibers.map(Fiber.interrupt), { concurrency: "unbounded", discard: true })
711 strictEqual(pool.state.waiters.size, 0)
712 strictEqual(pool.state.usage, 1)
713 yield* Scope.close(scope, Exit.void)
714 strictEqual(pool.state.usage, 0)
715 }))
716
717 it.effect("does not re-wake waiters registered during notification", () =>
718 Effect.gen(function*() {
719 const acquire = yield* Deferred.make<void>()
720 const pool = yield* Pool.makeWithTTL({
721 acquire: Effect.as(Deferred.await(acquire), "resource"),
722 min: 0,
723 max: 1,
724 timeToLive: Duration.infinity
725 })
726 const fibers: Array<Fiber.Fiber<string>> = []
727 for (let i = 0; i < 10; i++) {
728 fibers.push(yield* Effect.forkChild(Effect.scoped(Pool.get(pool)), { startImmediately: true }))
729 }
730 yield* waitUntil(() => pool.state.waiters.size === fibers.length)
731 yield* Deferred.succeed(acquire, undefined)
732 yield* Effect.all(fibers.map(Fiber.join), { concurrency: "unbounded", discard: true })
733 strictEqual(pool.state.waiters.size, 0)
734 strictEqual(pool.state.usage, 0)
735 }))
736
737 it.effect("use borrows and returns an item", () =>
738 Effect.gen(function*() {
739 const count = yield* Ref.make(0)
740 const get = Effect.acquireRelease(
741 Ref.updateAndGet(count, (n) => n + 1),
742 () => Ref.update(count, (n) => n - 1)
743 )
744 const pool = yield* Pool.make({ acquire: get, size: 2 })
745 yield* Effect.repeat(Ref.get(count), { until: (n) => n === 2 })
746 const results: Array<number> = []
747 yield* Effect.repeat(
748 Effect.tap(Pool.use(pool, (item) => Effect.succeed(item)), (item) =>
749 Effect.sync(() => {
750 results.push(item)
751 })),
752 { times: 9 }
753 )
754 // A borrow takes the item used most recently, so a sequence of them
755 // stays on one item and leaves the other free to be reclaimed. Spreading
756 // over both would leave neither warm and neither ever idle.
757 deepStrictEqual(results, [1, 1, 1, 1, 1, 1, 1, 1, 1, 1])
758 strictEqual(yield* Ref.get(count), 2)
759 }))
760
761 it.effect("use releases the item on failure", () =>
762 Effect.gen(function*() {
763 const pool = yield* Pool.make({ acquire: Effect.succeed("resource"), size: 1 })
764 const failure = yield* Effect.flip(Pool.use(pool, () => Effect.fail("boom")))
765 strictEqual(failure, "boom")
766 strictEqual(yield* Pool.use(pool, (item) => Effect.succeed(item)), "resource")
767 }))
768
769 it.effect("use releases the item on interruption", () =>
770 Effect.gen(function*() {
771 const started = yield* Deferred.make<void>()
772 const pool = yield* Pool.make({ acquire: Effect.succeed("resource"), size: 1 })
773 const fiber = yield* Effect.forkChild(
774 Pool.use(pool, () => Effect.andThen(Deferred.succeed(started, void 0), Effect.never)),
775 { startImmediately: true }
776 )
777 yield* Deferred.await(started)
778 yield* Fiber.interrupt(fiber)
779 strictEqual(yield* Pool.use(pool, (item) => Effect.succeed(item)), "resource")
780 }))
781
782 it.effect("an interrupted use returns its lease", () =>
783 Effect.gen(function*() {
784 for (let budget = 2; budget <= 10; budget++) {
785 const pool = yield* Pool.make({ acquire: Effect.succeed("resource"), size: 1 })
786 const fiber = yield* Effect.forkChild(
787 Pool.use(pool, () => Effect.never).pipe(Effect.provideService(Scheduler.MaxOpsBeforeYield, budget)),
788 { startImmediately: true }
789 )
790 yield* Fiber.interrupt(fiber)
791 strictEqual(pool.state.usage, 0, `MaxOpsBeforeYield ${budget}`)
792 const next = yield* Effect.forkChild(Pool.use(pool, Effect.succeed), { startImmediately: true })
793 assert.isDefined(next.pollUnsafe(), `MaxOpsBeforeYield ${budget}`)
794 }
795 }))
796
797 it.effect("use waits for an available item", () =>
798 Effect.gen(function*() {
799 const pool = yield* Pool.make({ acquire: Effect.succeed("resource"), size: 1 })
800 const scope = yield* Scope.make()
801 yield* Scope.provide(Pool.get(pool), scope)
802 const fiber = yield* Effect.forkChild(
803 Pool.use(pool, (item) => Effect.succeed(item)),
804 { startImmediately: true }
805 )
806 assert.isUndefined(fiber.pollUnsafe())
807 yield* Scope.close(scope, Exit.void)
808 strictEqual(yield* Fiber.join(fiber), "resource")
809 strictEqual(pool.state.usage, 0)
810 }))
811
812 it.effect("use reports acquire failures", () =>
813 Effect.gen(function*() {
814 const pool = yield* Pool.makeWithTTL({
815 acquire: Effect.fail("nope"),
816 min: 0,
817 max: 1,
818 timeToLive: Duration.infinity
819 })
820 const failure = yield* Effect.flip(Pool.use(pool, () => Effect.void))
821 strictEqual(failure, "nope")
822 strictEqual(pool.state.usage, 0)
823 }))
824
825 it.effect("finalizer is called for failed allocations", () =>
826 Effect.gen(function*() {
827 const scope = yield* Scope.make()
828 const allocations = yield* Ref.make(0)
829 const released = yield* Ref.make(0)
830 const get = Effect.acquireRelease(
831 Ref.updateAndGet(allocations, (n) => n + 1),
832 () => Ref.update(released, (n) => n + 1)
833 ).pipe(
834 Effect.flatMap((n) => n <= 10 ? Effect.fail("boom") : Effect.succeed(n))
835 )
836 const pool = yield* Pool.make({ acquire: get, size: 10 }).pipe(
837 Scope.provide(scope)
838 )
839 yield* Effect.repeat(Ref.get(released), { until: (n) => n === 10 })
840 strictEqual(yield* Effect.scoped(Pool.get(pool)), 11)
841 strictEqual(yield* Ref.get(allocations), 11)
842 strictEqual(yield* Ref.get(released), 10)
843 yield* Scope.close(scope, Exit.void)
844 }))
845 it.effect("runs strategy callbacks for a replacement before failed background cleanup", () =>
846 Effect.gen(function*() {
847 const cleanupStarted = yield* Deferred.make<void>()
848 const resumeCleanup = yield* Deferred.make<void>()
849 const replacementAcquired = yield* Deferred.make<void>()
850 let acquired = 0
851 let callbacks = 0
852 const pool = yield* Pool.makeWithStrategy({
853 acquire: Effect.gen(function*() {
854 if (++acquired > 1) return "replacement"
855 yield* Effect.addFinalizer(() =>
856 Effect.andThen(
857 Deferred.succeed(cleanupStarted, undefined),
858 Deferred.await(resumeCleanup)
859 )
860 )
861 return yield* Effect.fail("boom")
862 }),
863 min: 1,
864 max: 2,
865 strategy: {
866 run: () => Effect.void,
867 reclaim: () => Effect.undefined,
868 onAcquire: (item) =>
869 Effect.gen(function*() {
870 callbacks++
871 if (Exit.isSuccess(item.exit)) yield* Deferred.succeed(replacementAcquired, undefined)
872 })
873 }
874 })
875 yield* Effect.addFinalizer(() => Deferred.succeed(resumeCleanup, undefined))
876 yield* Deferred.await(cleanupStarted)
877 strictEqual(yield* Effect.scoped(Pool.get(pool)), "replacement")
878 yield* Effect.yieldNow
879 strictEqual(yield* Deferred.isDone(replacementAcquired), true)
880 strictEqual(acquired, 2)
881 strictEqual(callbacks, 1)
882 yield* Deferred.succeed(resumeCleanup, undefined)
883 }))
884
885 it.effect("admits one waiter per released lease", () =>
886 Effect.gen(function*() {
887 const acquired = yield* Ref.make(0)
888 const pool = yield* Pool.makeWithTTL({
889 acquire: Effect.succeed("item"),
890 min: 0,
891 max: 1,
892 concurrency: 4,
893 timeToLive: Duration.seconds(60)
894 })
895 const release = yield* Deferred.make<void>()
896 const lease = Effect.scoped(Effect.andThen(
897 Pool.get(pool),
898 Effect.andThen(Ref.update(acquired, (n) => n + 1), Deferred.await(release))
899 ))
900
901 // Four leases saturate the only item; four more have to wait.
902 const fibers = yield* Effect.all(
903 Array.from({ length: 8 }, () => Effect.forkChild(lease, { startImmediately: true }))
904 )
905 yield* Effect.repeat(Ref.get(acquired), { until: (n) => n === 4 })
906
907 // Releasing all four at once has to admit all four waiters. Reacting only
908 // to the transition out of saturation wakes one and leaves three asleep
909 // against an item that has room for them.
910 yield* Deferred.succeed(release, undefined)
911 yield* Effect.all(fibers.map(Fiber.join))
912 strictEqual(yield* Ref.get(acquired), 8)
913 strictEqual(pool.state.usage, 0)
914 }))
915
916 describe.skipIf(process.versions.bun !== undefined || process.versions.deno !== undefined)(
917 "usage TTL retention",
918 () => {
919 it.effect("releases invalidated resources while at minimum size", () =>
920 Effect.gen(function*() {
921 const references: Array<WeakRef<object>> = []
922 const control = new WeakRef({})
923 let finalized = 0
924 const pool = yield* Pool.makeWithTTL({
925 acquire: Effect.acquireRelease(
926 Effect.sync(() => ({})),
927 () =>
928 Effect.sync(() => {
929 finalized++
930 })
931 ),
932 min: 1,
933 max: 2,
934 timeToLive: "1 hour"
935 })
936 for (let i = 0; i < 2; i++) {
937 yield* Effect.scoped(Effect.gen(function*() {
938 const value = yield* Pool.get(pool)
939 references.push(new WeakRef(value))
940 yield* Pool.invalidate(pool, value)
941 }))
942 }
943 strictEqual(finalized, 2)
944 yield* collectGarbage
945 assert.isUndefined(control.deref())
946 for (const reference of references) assert.isUndefined(reference.deref())
947 strictEqual(pool.state.items.size, 1)
948 }))
949
950 it.effect("releases consumed acquisition failures", () =>
951 Effect.gen(function*() {
952 const references: Array<WeakRef<object>> = []
953 const control = new WeakRef({})
954 const pool = yield* Pool.makeWithTTL({
955 acquire: Effect.suspend(() => {
956 const error = {}
957 references.push(new WeakRef(error))
958 return Effect.fail(error)
959 }),
960 min: 0,
961 max: 2,
962 timeToLive: "1 hour"
963 })
964 for (let i = 0; i < 2; i++) {
965 assert.isTrue(Exit.isFailure(yield* Effect.exit(Effect.scoped(Pool.get(pool)))))
966 }
967 assert.lengthOf(references, 2)
968 yield* collectGarbage
969 assert.isUndefined(control.deref())
970 for (const reference of references) assert.isUndefined(reference.deref())
971 strictEqual(pool.state.items.size, 0)
972 }))
973
974 it.effect("retries background failures during asynchronous cleanup without requeueing them", () =>
975 Effect.gen(function*() {
976 const references: Array<WeakRef<object>> = []
977 const control = new WeakRef({})
978 const cleanupStarted = yield* Deferred.make<void>()
979 const resumeCleanup = yield* Deferred.make<void>()
980 let acquired = 0
981 let finalized = 0
982 const pool = yield* Pool.makeWithTTL({
983 acquire: Effect.gen(function*() {
984 if (++acquired > 1) return "replacement"
985 yield* Effect.addFinalizer(() =>
986 Effect.gen(function*() {
987 yield* Deferred.succeed(cleanupStarted, undefined)
988 yield* Deferred.await(resumeCleanup)
989 finalized++
990 })
991 )
992 const error = {}
993 references.push(new WeakRef(error))
994 return yield* Effect.fail(error)
995 }),
996 min: 1,
997 max: 2,
998 timeToLive: "1 hour"
999 })
1000 yield* Effect.addFinalizer(() => Deferred.succeed(resumeCleanup, undefined))
1001 yield* Deferred.await(cleanupStarted)
1002 strictEqual(yield* Effect.scoped(Pool.get(pool)), "replacement")
1003 strictEqual(finalized, 0)
1004 yield* Deferred.succeed(resumeCleanup, undefined)
1005 strictEqual(acquired, 2)
1006 strictEqual(finalized, 1)
1007 assert.lengthOf(references, 1)
1008 yield* collectGarbage
1009 assert.isUndefined(control.deref())
1010 assert.isUndefined(references[0].deref())
1011 strictEqual(pool.state.items.size, 1)
1012 }))
1013
1014 it.effect("releases resources after shutdown while the pool remains reachable", () =>
1015 Effect.gen(function*() {
1016 const references: Array<WeakRef<object>> = []
1017 const control = new WeakRef({})
1018 let finalized = 0
1019 const scope = yield* Scope.fork(yield* Effect.scope)
1020 const pool = yield* Pool.makeWithTTL({
1021 acquire: Effect.acquireRelease(
1022 Effect.sync(() => ({})),
1023 () =>
1024 Effect.sync(() => {
1025 finalized++
1026 })
1027 ),
1028 min: 1,
1029 max: 2,
1030 timeToLive: "1 hour"
1031 }).pipe(Scope.provide(scope))
1032 yield* Pool.use(pool, (value) =>
1033 Effect.sync(() => {
1034 references.push(new WeakRef(value))
1035 }))
1036 yield* Scope.close(scope, Exit.void)
1037 strictEqual(finalized, 1)
1038 assert.lengthOf(references, 1)
1039 yield* collectGarbage
1040 assert.isUndefined(control.deref())
1041 assert.isUndefined(references[0].deref())
1042 strictEqual(pool.state.items.size, 0)
1043 }))
1044 }
1045 )
1046
1047 it.effect("invalidating a busy item waits for its last lease and finalizes once", () =>
1048 Effect.gen(function*() {
1049 const finalized: Array<number> = []
1050 let acquired = 0
1051 yield* Effect.scoped(Effect.gen(function*() {
1052 const pool = yield* Pool.makeWithTTL({
1053 acquire: Effect.acquireRelease(
1054 Effect.sync(() => ++acquired),
1055 (value) =>
1056 Effect.sync(() => {
1057 finalized.push(value)
1058 })
1059 ),
1060 min: 1,
1061 max: 2,
1062 concurrency: 2,
1063 timeToLive: "1 hour"
1064 })
1065 const first = yield* Scope.fork(yield* Effect.scope)
1066 const second = yield* Scope.fork(yield* Effect.scope)
1067 const value = yield* Scope.provide(Pool.get(pool), first)
1068 strictEqual(yield* Scope.provide(Pool.get(pool), second), value)
1069 yield* Pool.invalidate(pool, value)
1070 deepStrictEqual(finalized, [])
1071 yield* Scope.close(first, Exit.void)
1072 deepStrictEqual(finalized, [])
1073 yield* Scope.close(second, Exit.void)
1074 deepStrictEqual(finalized, [value])
1075 strictEqual(yield* Effect.scoped(Pool.get(pool)), 2)
1076 }))
1077 deepStrictEqual(finalized, [1, 2])
1078 }))
1079
1080 for (const [removal, removed] of [["no", undefined], ["middle", 2], ["tail", 3]] as const) {
1081 it.effect(`usage TTL preserves FIFO retirement and minimum size after ${removal} removal`, () =>
1082 Effect.gen(function*() {
1083 let acquired = 0
1084 const finalized: Array<number> = []
1085 const pool = yield* Pool.makeWithTTL({
1086 acquire: Effect.acquireRelease(
1087 Effect.sync(() => ++acquired),
1088 (value) =>
1089 Effect.sync(() => {
1090 finalized.push(value)
1091 })
1092 ),
1093 min: 1,
1094 max: 3,
1095 timeToLive: "1 second"
1096 })
1097 yield* Effect.scoped(Effect.gen(function*() {
1098 for (let i = 0; i < 3; i++) yield* Pool.get(pool)
1099 strictEqual(acquired, 3)
1100 }))
1101 if (removed !== undefined) {
1102 yield* Pool.invalidate(pool, removed)
1103 deepStrictEqual(finalized, [removed])
1104 yield* Effect.scoped(Effect.gen(function*() {
1105 for (let i = 0; i < 3; i++) yield* Pool.get(pool)
1106 strictEqual(acquired, 4)
1107 }))
1108 }
1109 const expected = removed === undefined ? [1, 2] : [removed, ...[1, 2, 3].filter((n) => n !== removed)]
1110 yield* TestClock.adjust("1 second")
1111 deepStrictEqual(finalized, expected)
1112 yield* TestClock.adjust("5 seconds")
1113 deepStrictEqual(finalized, expected)
1114 strictEqual(pool.state.items.size, 1)
1115 strictEqual(yield* Effect.scoped(Pool.get(pool)), removed === undefined ? 3 : 4)
1116 if (removed !== undefined) {
1117 yield* Effect.scoped(Effect.gen(function*() {
1118 for (let i = 0; i < 3; i++) yield* Pool.get(pool)
1119 strictEqual(acquired, 6)
1120 }))
1121 yield* TestClock.adjust("1 second")
1122 deepStrictEqual(finalized, [...expected, 4, 5])
1123 strictEqual(pool.state.items.size, 1)
1124 strictEqual(yield* Effect.scoped(Pool.get(pool)), 6)
1125 }
1126 }))
1127 }
1128})