1/**
2 * Shares scoped resources across fibers.
3 *
4 * A `Pool<A, E>` acquires resource-backed values with a scoped effect, lets
5 * fibers borrow them with `get`, can invalidate broken values, and releases all
6 * acquired values when the pool scope closes. This module includes fixed-size
7 * pools, pools that resize with a time-to-live policy, custom strategy pools,
8 * per-item concurrency limits, and runtime state types used by pool strategies.
9 *
10 * @since 2.0.0
11 */
12import type * as Cause from "./Cause.ts"
13import { Clock } from "./Clock.ts"
14import * as Context from "./Context.ts"
15import * as Duration from "./Duration.ts"
16import * as Effect from "./Effect.ts"
17import type * as Exit from "./Exit.ts"
18import * as Fiber from "./Fiber.ts"
19import { constant, dual, identity } from "./Function.ts"
20import * as core from "./internal/core.ts"
21import * as internal from "./internal/effect.ts"
22import { type Pipeable, pipeArguments } from "./Pipeable.ts"
23import { hasProperty } from "./Predicate.ts"
24import * as Queue from "./Queue.ts"
25import { UnhandledLogLevel } from "./References.ts"
26import * as Scope from "./Scope.ts"
27import * as Semaphore from "./Semaphore.ts"
28
29const TypeId = "~effect/Pool"
30
31const Acquire = Symbol()
32const AcquireContext = Symbol()
33const PoolClock = Symbol()
34
35interface PoolImpl<A, E> extends Pool<A, E> {
36 readonly [Acquire]: Effect.Effect<A, E, Scope.Scope>
37 readonly [AcquireContext]: Context.Context<Scope.Scope>
38 readonly [PoolClock]: Clock
39}
40
41/**
42 * A `Pool<A, E>` is a pool of items of type `A`, each of which may be
43 * associated with the acquisition and release of resources. An attempt to get
44 * an item `A` from a pool may fail with an error of type `E`.
45 *
46 * **When to use**
47 *
48 * Use when you need to share a bounded set of scoped resources across fibers
49 * while the pool manages acquisition, reuse, and release.
50 *
51 * @see {@link make} for creating a pool with size bounds
52 * @see {@link makeWithTTL} for creating a pool with idle item expiration
53 * @see {@link makeWithStrategy} for creating a pool with a custom strategy
54 * @see {@link get} for acquiring an item from a pool
55 * @see {@link invalidate} for removing a broken item from the pool
56 *
57 * @category models
58 * @since 2.0.0
59 */
60export interface Pool<in out A, in out E = never> extends Pipeable {
61 readonly [TypeId]: typeof TypeId
62 readonly config: Config<A, E>
63 readonly state: State<A, E>
64}
65
66/**
67 * Normalized configuration used by a `Pool`.
68 *
69 * **When to use**
70 *
71 * Use as the normalized, read-only description of how a pool acquires, sizes,
72 * shares, and resizes its items after construction.
73 *
74 * **Details**
75 *
76 * The config stores the acquire effect, size bounds, per-item concurrency,
77 * target utilization, and resizing strategy used by the pool implementation.
78 *
79 * @see {@link Pool} for the value exposing this configuration
80 * @see {@link State} for mutable runtime state instead of static configuration
81 * @see {@link Strategy} for the resizing and reclamation contract stored on the config
82 *
83 * @category models
84 * @since 4.0.0
85 */
86export interface Config<A, E> {
87 readonly acquire: Effect.Effect<A, E, Scope.Scope>
88 readonly concurrency: number
89 readonly isFixed: boolean
90 readonly minSize: number
91 readonly maxSize: number
92 readonly strategy: Strategy<A, E>
93 readonly targetUtilization: number
94}
95
96/**
97 * Mutable runtime state maintained by a `Pool`.
98 *
99 * **When to use**
100 *
101 * Use when you need to inspect or support the runtime state backing a `Pool`,
102 * including its scope, item sets, semaphores, waiters, invalidation tracking,
103 * and shutdown flag.
104 *
105 * **Details**
106 *
107 * This state is exposed for inspection and implementation support. User code
108 * should prefer the high-level pool operations.
109 *
110 * @see {@link Pool} for the pool value exposing this state
111 * @see {@link PoolItem} for the entries stored in the runtime item sets
112 * @see {@link get} for acquiring items through the high-level API
113 * @see {@link invalidate} for invalidating items through the high-level API
114 *
115 * @category models
116 * @since 4.0.0
117 */
118export interface State<A, E> {
119 readonly scope: Scope.Scope
120 isShuttingDown: boolean
121 usage: number
122 readonly resizeSemaphore: Semaphore.Semaphore
123 readonly items: Set<PoolItem<A, E>>
124 availableHead: PoolItem<A, E> | undefined
125 availableTail: PoolItem<A, E> | undefined
126 readonly invalidated: Set<PoolItem<A, E>>
127 readonly waiters: Set<(failure?: Exit.Failure<A, E>) => void>
128}
129
130/**
131 * Internal record for a value managed by a `Pool`.
132 *
133 * **When to use**
134 *
135 * Use when implementing a custom pool `Strategy` that needs to inspect
136 * acquired items, track reference counts, or return reclaimable items to the
137 * pool.
138 *
139 * **Details**
140 *
141 * Each item stores the acquisition `Exit`, its finalizer, the current
142 * reference count, when it last became idle, and whether automatic reclaiming
143 * has been disabled because the item was invalidated.
144 *
145 * @see {@link Strategy} for the custom strategy callbacks that receive and return pool items
146 * @see {@link State} for the runtime sets that store active, available, and invalidated pool items
147 *
148 * @category models
149 * @since 4.0.0
150 */
151export interface PoolItem<A, E> {
152 readonly exit: Exit.Exit<A, E>
153 finalizer: Effect.Effect<void>
154 refCount: number
155 /** Clock time at which the item was acquired or its last borrower released it. */
156 idleSince: number
157 disableReclaim: boolean
158 isAvailable: boolean
159 availablePrevious: PoolItem<A, E> | undefined
160 availableNext: PoolItem<A, E> | undefined
161 release: (exit: Exit.Exit<any, any>) => Effect.Effect<void>
162}
163
164/**
165 * Strategy used by a `Pool` to manage background resizing and item
166 * reclamation.
167 *
168 * **When to use**
169 *
170 * Use when defining a custom pool lifecycle policy that needs to run background
171 * work, observe acquired items, or choose items for reclamation.
172 *
173 * **Details**
174 *
175 * `run` starts any strategy-specific background work, `onAcquire` is invoked
176 * when an item is acquired, and `reclaim` selects an item that can be removed
177 * or replaced.
178 *
179 * @see {@link makeWithStrategy} for constructing a pool from a custom `Strategy`
180 *
181 * @category models
182 * @since 4.0.0
183 */
184export interface Strategy<A, E> {
185 readonly run: (pool: Pool<A, E>) => Effect.Effect<void>
186 readonly onAcquire: (item: PoolItem<A, E>) => Effect.Effect<void>
187 readonly reclaim: (pool: Pool<A, E>) => Effect.Effect<PoolItem<A, E> | undefined>
188}
189
190/**
191 * Returns `true` if the specified value is a `Pool`, `false` otherwise.
192 *
193 * **When to use**
194 *
195 * Use to validate unknown values at runtime boundaries before treating them as
196 * `Pool` values.
197 *
198 * **Details**
199 *
200 * This predicate narrows the input to `Pool<unknown, unknown>`.
201 *
202 * @category guards
203 * @since 2.0.0
204 */
205export const isPool = (u: unknown): u is Pool<unknown, unknown> => hasProperty(u, TypeId)
206
207/**
208 * Makes a new pool of the specified fixed size.
209 *
210 * **When to use**
211 *
212 * Use when you need a fixed-size pool with no growth or shrinkage.
213 *
214 * **Details**
215 *
216 * The pool is returned in a `Scope`, which governs the lifetime of the pool.
217 * When the pool is shutdown because the `Scope` is closed, the individual
218 * items allocated by the pool will be released in some unspecified order.
219 *
220 * By setting the `concurrency` parameter, you can control the level of concurrent
221 * access per pool item. By default, the number of permits is set to `1`.
222 *
223 * `targetUtilization` determines when to create new pool items. It is a value
224 * between 0 and 1, where 1 means only create new pool items when all the existing
225 * items are fully utilized.
226 *
227 * A `targetUtilization` of 0.5 will create new pool items when the existing items are
228 * 50% utilized.
229 *
230 * @see {@link makeWithTTL} for pools with min/max sizes and a TTL-based shrinking policy
231 * @see {@link makeWithStrategy} for pools with a custom resizing and reclamation strategy
232 * @category constructors
233 * @since 2.0.0
234 */
235export const make = <A, E, R>(options: {
236 readonly acquire: Effect.Effect<A, E, R>
237 readonly size: number
238 readonly concurrency?: number | undefined
239 readonly targetUtilization?: number | undefined
240}): Effect.Effect<Pool<A, E>, never, R | Scope.Scope> =>
241 makeWithStrategy({ ...options, min: options.size, max: options.size, strategy: strategyNoop })
242
243/**
244 * Creates a scoped pool with minimum and maximum sizes and a time-to-live
245 * policy for shrinking unused excess items.
246 *
247 * **When to use**
248 *
249 * Use to create an elastic scoped pool that can grow up to a maximum size and
250 * later reclaim unused excess items.
251 *
252 * **Details**
253 *
254 * The returned pool requires `Scope`; when that scope is closed, allocated
255 * items are released in an unspecified order. `concurrency` controls how many
256 * fibers may use each pool item at once and defaults to `1`.
257 *
258 * `targetUtilization` controls when new items are created and is clamped by the
259 * pool implementation. A value of `1` waits until existing items are fully
260 * utilized before creating more items.
261 *
262 * `timeToLiveStrategy` defaults to `"usage"`, which expires idle, unreserved
263 * excess items after `timeToLive` since their last borrower released them
264 * (or acquisition if never borrowed). It checks once per `timeToLive`, so
265 * retirement can take up to twice that duration. `"creation"` measures from
266 * item creation instead.
267 *
268 * **Example** (Creating a connection pool)
269 *
270 * ```ts import.meta.vitest
271 * import { Duration, Effect, Pool } from "effect"
272 *
273 * interface Connection {
274 * readonly execute: (sql: string) => Effect.Effect<ReadonlyArray<string>>
275 * readonly close: Effect.Effect<void>
276 * }
277 *
278 * const acquireDBConnection = Effect.acquireRelease(
279 * Effect.succeed({
280 * execute: (sql) => Effect.succeed([`executed: ${sql}`]),
281 * close: Effect.void
282 * } satisfies Connection),
283 * (connection) => connection.close
284 * )
285 *
286 * const program = Effect.scoped(
287 * Effect.flatMap(
288 * Pool.makeWithTTL({
289 * acquire: acquireDBConnection,
290 * min: 10,
291 * max: 20,
292 * timeToLive: Duration.seconds(60)
293 * }),
294 * (pool) => Effect.flatMap(Pool.get(pool), (connection) => connection.execute("select 1"))
295 * )
296 * )
297 *
298 * await Effect.runPromise(program) // => ["executed: select 1"]
299 * ```
300 *
301 * @category constructors
302 * @since 2.0.0
303 */
304export const makeWithTTL = <A, E, R>(options: {
305 readonly acquire: Effect.Effect<A, E, R>
306 readonly min: number
307 readonly max: number
308 readonly concurrency?: number | undefined
309 readonly targetUtilization?: number | undefined
310 readonly timeToLive: Duration.Input
311 readonly timeToLiveStrategy?: "creation" | "usage" | undefined
312}): Effect.Effect<Pool<A, E>, never, R | Scope.Scope> =>
313 Effect.flatMap(
314 options.timeToLiveStrategy === "creation" ?
315 strategyCreationTTL<A, E>(options.timeToLive) :
316 strategyUsageTTL<A, E>(options.timeToLive),
317 (strategy) => makeWithStrategy({ ...options, strategy })
318 )
319
320/**
321 * Creates a scoped pool using a custom resizing and reclamation strategy.
322 *
323 * **When to use**
324 *
325 * Use to build a pool whose item lifecycle is controlled by an explicit
326 * `Strategy`, such as custom background resizing, replacement, or reclamation.
327 *
328 * **Details**
329 *
330 * The returned pool requires `Scope`; closing the scope shuts down the pool and
331 * releases allocated items.
332 *
333 * @see {@link make} for fixed-size pools without custom resizing or reclamation
334 * @see {@link makeWithTTL} for min/max pools that shrink excess items with a TTL policy
335 * @see {@link Strategy} for the custom strategy contract consumed by this constructor
336 *
337 * @category constructors
338 * @since 4.0.0
339 */
340export const makeWithStrategy = <A, E, R>(options: {
341 readonly acquire: Effect.Effect<A, E, R>
342 readonly min: number
343 readonly max: number
344 readonly concurrency?: number | undefined
345 readonly targetUtilization?: number | undefined
346 readonly strategy: Strategy<A, E>
347}): Effect.Effect<Pool<A, E>, never, Scope.Scope | R> =>
348 Effect.uninterruptibleMask(Effect.fnUntraced(function*(restore) {
349 const services = yield* Effect.context<R | Scope.Scope>()
350 const scope = Context.get(services, Scope.Scope)
351 const acquire = Effect.updateContext(
352 options.acquire,
353 (input) => Context.merge(services, input)
354 ) as Effect.Effect<A, E, Scope.Scope>
355 const concurrency = options.concurrency ?? 1
356
357 const config: Config<A, E> = {
358 acquire,
359 concurrency,
360 isFixed: options.min === options.max,
361 minSize: options.min,
362 maxSize: options.max,
363 strategy: options.strategy,
364 targetUtilization: Math.min(Math.max(options.targetUtilization ?? 1, 0.1), 1)
365 }
366 const state: State<A, E> = {
367 scope,
368 isShuttingDown: false,
369 usage: 0,
370 resizeSemaphore: Semaphore.makeUnsafe(1),
371 items: new Set(),
372 availableHead: undefined,
373 availableTail: undefined,
374 invalidated: new Set(),
375 waiters: new Set()
376 }
377 const self: PoolImpl<A, E> = {
378 [TypeId]: TypeId,
379 [Acquire]: options.acquire as Effect.Effect<A, E, Scope.Scope>,
380 [AcquireContext]: services as Context.Context<Scope.Scope>,
381 [PoolClock]: yield* Clock,
382 config,
383 state,
384 pipe() {
385 return pipeArguments(this, arguments)
386 }
387 }
388 yield* Scope.addFinalizer(scope, shutdown(self))
389 if (config.minSize > 0) {
390 yield* Effect.tap(
391 Effect.forkDetach(restore(resize(self)), { startImmediately: true }),
392 (fiber) => Scope.addFinalizer(scope, Fiber.interrupt(fiber))
393 )
394 }
395 if (options.strategy !== strategyNoop) {
396 yield* Effect.tap(
397 Effect.forkDetach(restore(options.strategy.run(self))),
398 (fiber) => Scope.addFinalizer(scope, Fiber.interrupt(fiber))
399 )
400 }
401 return self
402 }))
403
404const shutdown = Effect.fnUntraced(function*<A, E>(self: Pool<A, E>) {
405 if (self.state.isShuttingDown) return
406 self.state.isShuttingDown = true
407 const size = self.state.items.size
408 const semaphore = Semaphore.makeUnsafe(size)
409 for (const item of self.state.items) {
410 if (item.refCount > 0) {
411 item.finalizer = Effect.tap(item.finalizer, semaphore.release(1))
412 self.state.invalidated.add(item)
413 yield* semaphore.take(1)
414 } else {
415 self.state.items.delete(item)
416 removeAvailable(self, item)
417 self.state.invalidated.delete(item)
418 yield* item.finalizer
419 }
420 }
421 yield* semaphore.releaseAll
422 if (self.state.waiters.size > 0) {
423 const waiters = Array.from(self.state.waiters)
424 self.state.waiters.clear()
425 for (const notify of waiters) notify()
426 }
427 yield* semaphore.take(size)
428})
429
430/**
431 * Retrieves an item from the pool in a scoped effect.
432 *
433 * **When to use**
434 *
435 * Use to borrow a pooled resource for the lifetime of the current scope so it
436 * is automatically returned when that scope closes.
437 *
438 * **Details**
439 *
440 * The returned effect waits for an available item when the pool is at capacity.
441 * If acquiring a new item fails, the effect fails with the acquisition error.
442 *
443 * **Gotchas**
444 *
445 * Retrying a failed `get` can repeat the acquisition attempt.
446 *
447 * @see {@link invalidate} for removing an unhealthy item from future reuse
448 *
449 * @category getters
450 * @since 2.0.0
451 */
452export const get = <A, E>(self: Pool<A, E>): Effect.Effect<A, E, Scope.Scope> =>
453 core.withFiber((fiber) => {
454 const state = self.state
455 if (state.isShuttingDown) return internal.interrupt
456 if (state.availableHead?.exit._tag === "Success") {
457 state.usage++
458 if (self.config.isFixed || targetSize(self) <= activeSize(self)) {
459 return leaseItem(self, state.availableHead, fiber)
460 }
461 state.usage--
462 }
463 return getSlowWith(self, leaseItemWith)
464 })
465
466/**
467 * Borrows an item while an effect runs and returns it when the effect exits.
468 *
469 * **When to use**
470 *
471 * Use when an item is needed by one effect. Unlike `Effect.scoped` with
472 * {@link get}, this avoids allocating a scope and registering a finalizer.
473 *
474 * **Example** (Running a single operation with a pooled item)
475 *
476 * ```ts import.meta.vitest
477 * import { Effect, Pool } from "effect"
478 *
479 * const program = Effect.scoped(
480 * Effect.flatMap(
481 * Pool.make({ acquire: Effect.succeed("resource"), size: 2 }),
482 * (pool) => Pool.use(pool, (item) => Effect.succeed(item.length))
483 * )
484 * )
485 *
486 * await Effect.runPromise(program) // => 8
487 * ```
488 *
489 * @see {@link get} for borrowing an item for the lifetime of a scope
490 *
491 * @category combinators
492 * @since 4.0.0
493 */
494export const use: {
495 <A, B, E2, R2>(
496 f: (item: A) => Effect.Effect<B, E2, R2>
497 ): <E>(self: Pool<A, E>) => Effect.Effect<B, E | E2, R2>
498 <A, E, B, E2, R2>(
499 self: Pool<A, E>,
500 f: (item: A) => Effect.Effect<B, E2, R2>
501 ): Effect.Effect<B, E | E2, R2>
502} = dual(2, <A, E, B, E2, R2>(
503 self: Pool<A, E>,
504 f: (item: A) => Effect.Effect<B, E2, R2>
505): Effect.Effect<B, E | E2, R2> =>
506 core.withFiber((fiber) => {
507 const state = self.state
508 if (state.isShuttingDown) return internal.interrupt
509 if (state.availableHead?.exit._tag === "Success") {
510 state.usage++
511 if (self.config.isFixed || targetSize(self) <= activeSize(self)) {
512 return useItem(self, state.availableHead, f, fiber)
513 }
514 state.usage--
515 }
516 return getSlowWith(self, (self, item, fiber, restore) => useItem(self, item, f, fiber, restore))
517 }))
518
519// Count the lease and install its release in the same step.
520const useItem = <A, E, B, E2, R2>(
521 self: Pool<A, E>,
522 item: PoolItem<A, E>,
523 f: (item: A) => Effect.Effect<B, E2, R2>,
524 fiber: Fiber.Fiber<unknown, unknown>,
525 restore?: <AX, EX, RX>(effect: Effect.Effect<AX, EX, RX>) => Effect.Effect<AX, EX, RX>
526): Effect.Effect<B, E | E2, R2> => {
527 leaseItemBookkeeping(self, item)
528 internal.onExitUnsafe(fiber, item.release)
529 let body: Effect.Effect<B, E2, R2>
530 try {
531 body = f((item.exit as Exit.Success<A, E>).value)
532 } catch (defect) {
533 return core.exitDie(defect)
534 }
535 return restore !== undefined ? restore(body) : body
536}
537
538const getSlowWith = <A, E, X, R>(
539 self: Pool<A, E>,
540 lease: (
541 self: Pool<A, E>,
542 item: PoolItem<A, E>,
543 fiber: Fiber.Fiber<unknown, unknown>,
544 restore: <AX, EX, RX>(effect: Effect.Effect<AX, EX, RX>) => Effect.Effect<AX, EX, RX>
545 ) => Effect.Effect<X, any, R>
546): Effect.Effect<X, any, R> =>
547 internal.uninterruptibleMask((restore) => {
548 const state = self.state
549 state.usage++
550 const wait: Effect.Effect<X, any, R> = internal.flatMap(
551 internal.onError(
552 restore(waitForItem(self)),
553 () =>
554 internal.sync(() => {
555 state.usage--
556 })
557 ),
558 () => loop
559 )
560 const step: Effect.Effect<X, any, R> = core.withFiber((fiber) => {
561 if (state.isShuttingDown) {
562 state.usage--
563 return internal.interrupt
564 }
565 // Replace failed slots only when the pool cannot grow.
566 const item = firstHealthyAvailable(self) ??
567 (targetSize(self) <= activeSize(self) ? state.availableHead : undefined)
568 if (item === undefined) {
569 return wait
570 }
571 if (item.exit._tag === "Failure") {
572 removeFailedItem(self, item)
573 return loop
574 }
575 return lease(self, item, fiber, restore)
576 })
577 const loop: Effect.Effect<X, any, R> = internal.suspend(() => {
578 if (state.isShuttingDown) {
579 state.usage--
580 return internal.interrupt
581 }
582 return targetSize(self) > activeSize(self)
583 ? internal.flatMap(
584 state.resizeSemaphore.withPermitsIfAvailable(1)(
585 Effect.forkIn(Effect.interruptible(resize(self)), state.scope)
586 ),
587 () => step
588 )
589 : step
590 })
591 return loop
592 })
593
594const leaseItemBookkeeping = <A, E>(self: Pool<A, E>, item: PoolItem<A, E>): void => {
595 item.refCount++
596 if (item.refCount >= self.config.concurrency) {
597 removeAvailable(self, item)
598 }
599}
600
601const removeFailedItem = <A, E>(self: Pool<A, E>, item: PoolItem<A, E>): void => {
602 self.state.items.delete(item)
603 self.state.invalidated.delete(item)
604 removeAvailable(self, item)
605}
606
607const leaseItem = <A, E>(
608 self: Pool<A, E>,
609 item: PoolItem<A, E>,
610 fiber: Fiber.Fiber<unknown, unknown>
611): Effect.Effect<A, E> => {
612 leaseItemBookkeeping(self, item)
613 const scope = Context.getUnsafe(fiber.context, Scope.Scope)
614 if (scope.state._tag === "Closed") {
615 internal.onExitUnsafe(fiber, item.release)
616 } else {
617 internal.scopeAddFinalizerUnsafe(scope, {}, item.release)
618 }
619 return item.exit
620}
621
622const leaseItemWith = <A, E>(
623 self: Pool<A, E>,
624 item: PoolItem<A, E>,
625 fiber: Fiber.Fiber<unknown, unknown>
626): Effect.Effect<A, E, Scope.Scope> => leaseItem(self, item, fiber)
627
628const releaseItem = <A, E>(self: Pool<A, E>, item: PoolItem<A, E>): Effect.Effect<void> =>
629 core.withFiber((fiber) => {
630 const state = self.state
631 item.refCount--
632 state.usage--
633 if (item.refCount === 0) {
634 item.idleSince = (self as PoolImpl<A, E>)[PoolClock].currentTimeMillisUnsafe()
635 }
636 if (state.invalidated.has(item)) {
637 return invalidatePoolItem(self, item)
638 }
639 // Every release frees one slot, so it can admit one waiter. Reacting only
640 // to the saturated-to-unsaturated transition strands the rest: several
641 // leases returning at once would wake a single waiter and leave the others
642 // asleep against an item that has capacity for them. `addAvailable` is
643 // idempotent, so re-adding an available item is free.
644 if (item.refCount < self.config.concurrency) {
645 addAvailableFront(self, item)
646 wakeWaiters(self, fiber, 1)
647 }
648 return internal.void
649 })
650
651const firstHealthyAvailable = <A, E>(self: Pool<A, E>): PoolItem<A, E> | undefined => {
652 let item = self.state.availableHead
653 while (item !== undefined && item.exit._tag === "Failure") item = item.availableNext
654 return item
655}
656
657const waitForItem = <A, E>(self: Pool<A, E>): Effect.Effect<void, E> =>
658 internal.callback((resume) => {
659 const state = self.state
660 if (firstHealthyAvailable(self) !== undefined || state.isShuttingDown) {
661 return resume(internal.void)
662 }
663 const observer = (failure?: Exit.Failure<A, E>) => {
664 state.waiters.delete(observer)
665 resume(failure ?? internal.void)
666 }
667 state.waiters.add(observer)
668 return internal.sync(() => {
669 state.waiters.delete(observer)
670 })
671 })
672
673const wakeWaiters = <A, E>(self: Pool<A, E>, fiber: Fiber.Fiber<unknown, unknown>, count: number) => {
674 const waiters = self.state.waiters
675 if (waiters.size === 0) return
676 fiber.currentDispatcher.scheduleTask(() => {
677 let remaining = count
678 const toWake: Array<(failure?: Exit.Failure<A, E>) => void> = []
679 for (const notify of waiters) {
680 if (remaining-- <= 0) break
681 toWake.push(notify)
682 }
683 for (let i = 0; i < toWake.length; i++) {
684 toWake[i]()
685 }
686 }, 0)
687}
688
689const wakeAll = <A, E>(self: Pool<A, E>): Effect.Effect<void> =>
690 core.withFiber((fiber) => {
691 wakeWaiters(self, fiber, Number.POSITIVE_INFINITY)
692 return internal.void
693 })
694
695// Reservations prevent reuse without extending the lifetime of borrowed items.
696const reservations = new WeakMap<PoolItem<unknown, unknown>, number>()
697
698/** Adds a freshly acquired item, which has no use behind it, at the back. */
699const addAvailable = <A, E>(self: Pool<A, E>, item: PoolItem<A, E>): void => {
700 if (item.isAvailable || reservations.has(item)) return
701 item.isAvailable = true
702 item.availablePrevious = self.state.availableTail
703 item.availableNext = undefined
704 if (self.state.availableTail !== undefined) {
705 self.state.availableTail.availableNext = item
706 } else {
707 self.state.availableHead = item
708 }
709 self.state.availableTail = item
710}
711
712/**
713 * Returns a released item at the front, so the next borrow gets the one used
714 * most recently. Borrowers take from the front, so the list runs warmest
715 * first.
716 *
717 * Sending it to the back instead spreads a sequence of borrows evenly over
718 * every item the pool has open. For a pool of connections that means none of
719 * them is ever the hot one - each borrow lands on a peer that has been sitting
720 * idle, losing whatever warmth it had - and it means `timeToLive` never
721 * reclaims anything, because a pool that grew for one burst keeps every item
722 * equally fresh forever. Under saturation the two orders agree, since every
723 * item is checked out either way.
724 */
725const addAvailableFront = <A, E>(self: Pool<A, E>, item: PoolItem<A, E>): void => {
726 if (item.isAvailable || reservations.has(item)) return
727 item.isAvailable = true
728 item.availablePrevious = undefined
729 item.availableNext = self.state.availableHead
730 if (self.state.availableHead !== undefined) {
731 self.state.availableHead.availablePrevious = item
732 } else {
733 self.state.availableTail = item
734 }
735 self.state.availableHead = item
736}
737
738const removeAvailable = <A, E>(self: Pool<A, E>, item: PoolItem<A, E>): void => {
739 if (!item.isAvailable) return
740 item.isAvailable = false
741 if (item.availablePrevious !== undefined) {
742 item.availablePrevious.availableNext = item.availableNext
743 } else {
744 self.state.availableHead = item.availableNext
745 }
746 if (item.availableNext !== undefined) {
747 item.availableNext.availablePrevious = item.availablePrevious
748 } else {
749 self.state.availableTail = item.availablePrevious
750 }
751 item.availablePrevious = undefined
752 item.availableNext = undefined
753}
754
755/**
756 * Invalidates the specified item so the pool can remove it and reallocate the
757 * item, lazily if needed.
758 *
759 * **When to use**
760 *
761 * Use to prevent a pooled item from being reused after it becomes unsuitable,
762 * such as a stale connection or a resource that failed a health check.
763 *
764 * **Gotchas**
765 *
766 * The item is matched with strict equality. Passing an equivalent but different
767 * object instance does nothing.
768 *
769 * @see {@link get} for retrieving scoped items from the pool
770 *
771 * @category combinators
772 * @since 2.0.0
773 */
774export const invalidate: {
775 <A>(item: A): <E>(self: Pool<A, E>) => Effect.Effect<void>
776 <A, E>(self: Pool<A, E>, item: A): Effect.Effect<void>
777} = dual(2, <A, E>(self: Pool<A, E>, item: A): Effect.Effect<void> =>
778 Effect.suspend(() => {
779 if (self.state.isShuttingDown) return Effect.void
780 for (const poolItem of self.state.items) {
781 if (poolItem.exit._tag === "Success" && poolItem.exit.value === item) {
782 poolItem.disableReclaim = true
783 return Effect.uninterruptible(invalidatePoolItem(self, poolItem))
784 }
785 }
786 return Effect.void
787 }))
788
789/**
790 * Reserves a leased item for exclusive use until the scope closes. This
791 * removes the item's remaining capacity from the pool but does not wait
792 * for existing leases to finish. It has no effect when per-item concurrency is
793 * `1` or the pool does not contain the item.
794 *
795 * @see {@link get} for acquiring an item
796 *
797 * @category combinators
798 * @since 4.0.0
799 */
800export const reserve: {
801 <A>(item: A): <E>(self: Pool<A, E>) => Effect.Effect<void, never, Scope.Scope>
802 <A, E>(self: Pool<A, E>, item: A): Effect.Effect<void, never, Scope.Scope>
803} = dual(
804 2,
805 <A, E>(self: Pool<A, E>, item: A): Effect.Effect<void, never, Scope.Scope> =>
806 Effect.asVoid(Effect.acquireRelease(
807 Effect.sync(() => {
808 if (self.config.concurrency === 1) return undefined
809 for (const poolItem of self.state.items) {
810 if (poolItem.exit._tag !== "Success" || poolItem.exit.value !== item) continue
811 const existing = reservations.get(poolItem)
812 if (existing === undefined) self.state.usage += self.config.concurrency - 1
813 reservations.set(poolItem, (existing ?? 0) + 1)
814 removeAvailable(self, poolItem)
815 return poolItem
816 }
817 return undefined
818 }),
819 (poolItem) =>
820 core.withFiber((fiber) => {
821 if (poolItem === undefined) return internal.void
822 const remaining = (reservations.get(poolItem) ?? 1) - 1
823 if (remaining > 0) {
824 reservations.set(poolItem, remaining)
825 return internal.void
826 }
827 reservations.delete(poolItem)
828 self.state.usage -= self.config.concurrency - 1
829 if (
830 !self.state.isShuttingDown &&
831 self.state.items.has(poolItem) &&
832 !self.state.invalidated.has(poolItem) &&
833 poolItem.refCount < self.config.concurrency
834 ) {
835 addAvailableFront(self, poolItem)
836 wakeWaiters(self, fiber, self.config.concurrency - poolItem.refCount)
837 }
838 return internal.void
839 })
840 ))
841)
842
843const invalidatePoolItem = <A, E>(self: Pool<A, E>, poolItem: PoolItem<A, E>): Effect.Effect<void> =>
844 Effect.suspend(() => {
845 if (!self.state.items.has(poolItem)) {
846 return Effect.void
847 } else if (poolItem.refCount === 0) {
848 self.state.items.delete(poolItem)
849 removeAvailable(self, poolItem)
850 self.state.invalidated.delete(poolItem)
851 return Effect.asVoid(Effect.flatMap(
852 poolItem.finalizer,
853 () => Effect.forkIn(Effect.interruptible(resize(self)), self.state.scope, { startImmediately: true })
854 ))
855 }
856 self.state.invalidated.add(poolItem)
857 removeAvailable(self, poolItem)
858 // An invalidated item stops counting towards the pool's active size, so the
859 // pool is now below target and has to top itself back up. Waiting for the
860 // last lease to be returned would strand anybody already queued: the item
861 // they are waiting for is never coming back.
862 return Effect.asVoid(
863 Effect.forkIn(Effect.interruptible(resize(self)), self.state.scope, { startImmediately: true })
864 )
865 })
866
867const resize = <A, E>(self: Pool<A, E>): Effect.Effect<void> =>
868 self.state.resizeSemaphore.withPermits(1)(resizeLoop(self))
869
870const resizeLoop = <A, E>(self: Pool<A, E>): Effect.Effect<void> =>
871 Effect.suspend(() => {
872 const active = activeSize(self)
873 const target = targetSize(self)
874 if (active >= target) {
875 return Effect.void
876 }
877 const toAcquire = target - active
878 const acquireOne = self.config.strategy === strategyNoop
879 ? allocate(self)
880 : Effect.flatMap(
881 self.config.strategy.reclaim(self),
882 (item) => item ? Effect.succeed(item) : allocate(self)
883 )
884 if (toAcquire === 1) {
885 const acquired = Effect.tap(acquireOne, wakeAll(self))
886 return self.config.isFixed
887 ? Effect.asVoid(acquired)
888 : Effect.flatMap(acquired, (item) => item.exit._tag === "Failure" ? Effect.void : resizeLoop(self))
889 }
890 const acquired = acquireOne.pipe(
891 Effect.replicateEffect(toAcquire, { concurrency: toAcquire }),
892 Effect.tap(wakeAll(self))
893 )
894 return self.config.isFixed
895 ? Effect.asVoid(acquired)
896 : Effect.flatMap(
897 acquired,
898 (items) => items.some((_) => _.exit._tag === "Failure") ? Effect.void : resizeLoop(self)
899 )
900 })
901
902const allocate = <A, E>(self: Pool<A, E>): Effect.Effect<PoolItem<A, E>> =>
903 internal.uninterruptibleMask((restore) =>
904 core.withFiber((fiber) => {
905 const impl = self as PoolImpl<A, E>
906 const scope = internal.scopeMakeUnsafe()
907 const previousContext = fiber.context
908 fiber.setContext(Context.add(impl[AcquireContext], Scope.Scope, scope))
909 const use = Effect.flatMap(Effect.exit(impl[Acquire]), (exit) => {
910 const item: PoolItem<A, E> = {
911 exit,
912 finalizer: Effect.catchCause(Scope.close(scope, exit), reportUnhandledError),
913 refCount: 0,
914 idleSince: impl[PoolClock].currentTimeMillisUnsafe(),
915 disableReclaim: false,
916 isAvailable: false,
917 availablePrevious: undefined,
918 availableNext: undefined,
919 release: undefined as any
920 }
921 item.release = constant(releaseItem(self, item))
922 const onAcquire = Effect.suspend(() =>
923 // A borrower may have removed the item before the callback runs.
924 self.state.items.has(item) ? self.config.strategy.onAcquire(item) : Effect.void
925 )
926 if (exit._tag === "Success") {
927 self.state.items.add(item)
928 addAvailable(self, item)
929 return self.config.strategy === strategyNoop ? Effect.succeed(item) : Effect.as(onAcquire, item)
930 }
931 // Deliver a waiting borrower its own failure; otherwise retain the slot.
932 const waiter = firstHealthyAvailable(self) === undefined ? self.state.waiters.values().next().value : undefined
933 if (waiter !== undefined) {
934 waiter(exit)
935 } else {
936 self.state.items.add(item)
937 addAvailable(self, item)
938 }
939 // Do not let cleanup delay a replacement acquisition.
940 return Effect.as(
941 Effect.forkIn(Effect.andThen(item.finalizer, onAcquire), self.state.scope, { startImmediately: true }),
942 item
943 )
944 })
945 return internal.onExitPrimitive(
946 restore(use) as Effect.Effect<PoolItem<A, E>>,
947 (exit) => {
948 fiber.setContext(previousContext)
949 return exit._tag === "Failure" ? internal.scopeCloseUnsafe(scope, exit) : undefined
950 },
951 true
952 )
953 })
954 )
955
956const targetSize = <A, E>(self: Pool<A, E>) => {
957 if (self.state.isShuttingDown) return 0
958 if (self.config.isFixed) return self.config.minSize
959 const utilization = self.state.usage / self.config.targetUtilization
960 const target = Math.ceil(utilization / self.config.concurrency)
961 return Math.min(Math.max(self.config.minSize, target), self.config.maxSize)
962}
963
964const activeSize = <A, E>(self: Pool<A, E>) => {
965 return self.state.items.size - self.state.invalidated.size
966}
967
968// -----------------------------------------------------------------------------
969// Strategy
970// -----------------------------------------------------------------------------
971
972const strategyNoop: Strategy<any, any> = {
973 run: (_) => Effect.void,
974 onAcquire: (_) => Effect.void,
975 reclaim: (_) => Effect.undefined
976}
977
978const strategyCreationTTL = Effect.fnUntraced(function*<A, E>(ttl: Duration.Input) {
979 const clock = yield* Clock
980 const queue = yield* Queue.unbounded<PoolItem<A, E>>()
981 const ttlMillis = Duration.toMillis(Duration.fromInputUnsafe(ttl))
982 const creationTimes = new WeakMap<PoolItem<A, E>, number>()
983 return identity<Strategy<A, E>>({
984 run: (pool) => {
985 const process = (item: PoolItem<A, E>): Effect.Effect<void> =>
986 Effect.suspend(() => {
987 if (!pool.state.items.has(item) || pool.state.invalidated.has(item)) {
988 return Effect.void
989 }
990 const now = clock.currentTimeMillisUnsafe()
991 const created = creationTimes.get(item)!
992 const remaining = ttlMillis - (now - created)
993 return remaining > 0
994 ? Effect.delay(process(item), remaining)
995 : invalidatePoolItem(pool, item)
996 })
997 return Queue.take(queue).pipe(
998 Effect.tap(process),
999 Effect.forever({ disableYield: true })
1000 )
1001 },
1002 onAcquire: (item) =>
1003 Effect.suspend(() => {
1004 creationTimes.set(item, clock.currentTimeMillisUnsafe())
1005 return Queue.offer(queue, item)
1006 }),
1007 reclaim: (_) => Effect.undefined
1008 })
1009})
1010
1011const strategyUsageTTL = Effect.fnUntraced(function*<A, E>(ttl: Duration.Input) {
1012 const clock = yield* Clock
1013 const ttlMillis = Duration.toMillis(Duration.fromInputUnsafe(ttl))
1014 return identity<Strategy<A, E>>({
1015 run: (pool) => {
1016 const process: Effect.Effect<void> = Effect.suspend(() => {
1017 if (activeSize(pool) <= targetSize(pool)) return Effect.void
1018 const expired = clock.currentTimeMillisUnsafe() - ttlMillis
1019 let oldest: PoolItem<A, E> | undefined
1020 for (const item of pool.state.items) {
1021 if (item.refCount > 0 || item.idleSince > expired) continue
1022 if (pool.state.invalidated.has(item) || reservations.has(item)) continue
1023 if (oldest === undefined || item.idleSince < oldest.idleSince) oldest = item
1024 }
1025 return oldest === undefined ? Effect.void : Effect.flatMap(invalidatePoolItem(pool, oldest), () => process)
1026 })
1027 return process.pipe(
1028 Effect.delay(ttl),
1029 Effect.forever({ disableYield: true })
1030 )
1031 },
1032 onAcquire: (_) => Effect.void,
1033 reclaim: (_) => Effect.undefined
1034 })
1035})
1036
1037const reportUnhandledError = <E>(cause: Cause.Cause<E>) =>
1038 Effect.withFiber<void>((fiber) => {
1039 const unhandledLogLevel = fiber.getRef(UnhandledLogLevel)
1040 if (unhandledLogLevel) {
1041 return Effect.logWithLevel(unhandledLogLevel)(
1042 "Unhandled error in pool finalizer",
1043 cause
1044 )
1045 }
1046 return Effect.void
1047 })