1/**
2 * Coordinates access to limited resources inside transactions.
3 *
4 * A `TxSemaphore` has a fixed capacity and stores its available permit count in
5 * a `TxRef`. Acquiring or releasing permits can therefore commit atomically
6 * with other transactional state changes. This module includes operations for
7 * creating semaphores, checking capacity and availability, acquiring or
8 * releasing permits, and running effects while permits are held.
9 *
10 * @since 4.0.0
11 */
12
13import * as Effect from "./Effect.ts"
14import { dual } from "./Function.ts"
15import type { Inspectable } from "./Inspectable.ts"
16import { NodeInspectSymbol, toJson } from "./Inspectable.ts"
17import type { Pipeable } from "./Pipeable.ts"
18import { pipeArguments } from "./Pipeable.ts"
19import { hasProperty } from "./Predicate.ts"
20import type * as Scope from "./Scope.ts"
21import * as TxRef from "./TxRef.ts"
22
23const TypeId = "~effect/TxSemaphore"
24
25/**
26 * A transactional semaphore that manages permits using Software Transactional
27 * Memory (STM) semantics, providing atomic permit acquisition and release
28 * operations within Effect transactions for concurrency control over limited
29 * resources.
30 *
31 * **When to use**
32 *
33 * Use to coordinate permit accounting atomically with other transactional
34 * state changes.
35 *
36 * **Example** (Managing permits transactionally)
37 *
38 * ```ts import.meta.vitest
39 * import { Effect, TxSemaphore } from "effect"
40 *
41 * // Create a semaphore with 3 permits for managing concurrent database connections
42 * const program = Effect.gen(function*() {
43 * const dbSemaphore = yield* TxSemaphore.make(3)
44 *
45 * // Acquire a permit before accessing the database
46 * yield* TxSemaphore.acquire(dbSemaphore)
47 * const acquired = yield* TxSemaphore.available(dbSemaphore)
48 *
49 * // Perform database operations...
50 *
51 * // Release the permit when done
52 * yield* TxSemaphore.release(dbSemaphore)
53 * const released = yield* TxSemaphore.available(dbSemaphore)
54 * return [acquired, released] as const
55 * })
56 *
57 * await Effect.runPromise(program) // => [2, 3]
58 * ```
59 *
60 * @see {@link make} for creating a transactional semaphore
61 * @see {@link withPermit} for automatically acquiring and releasing one permit
62 * @see {@link acquire} for manually acquiring one permit transactionally
63 *
64 * @category models
65 * @since 4.0.0
66 */
67export interface TxSemaphore extends Inspectable, Pipeable {
68 readonly [TypeId]: typeof TypeId
69 readonly permitsRef: TxRef.TxRef<number>
70 readonly capacity: number
71}
72
73const TxSemaphoreProto: Omit<TxSemaphore, typeof TypeId | "permitsRef" | "capacity"> = {
74 [NodeInspectSymbol](this: TxSemaphore) {
75 return toJson(this)
76 },
77 toJSON(this: TxSemaphore) {
78 return {
79 _id: "TxSemaphore",
80 capacity: this.capacity
81 }
82 },
83 pipe() {
84 return pipeArguments(this, arguments)
85 }
86}
87
88const makeTxSemaphore = (permitsRef: TxRef.TxRef<number>, capacity: number): TxSemaphore => {
89 const self = Object.create(TxSemaphoreProto)
90 self[TypeId] = TypeId
91 self.permitsRef = permitsRef
92 self.capacity = capacity
93 return self
94}
95
96/**
97 * Creates a new TxSemaphore with the specified number of permits.
98 *
99 * **When to use**
100 *
101 * Use to create a transactional semaphore with a fixed permit capacity.
102 *
103 * **Example** (Creating a semaphore)
104 *
105 * ```ts import.meta.vitest
106 * import { Effect, TxSemaphore } from "effect"
107 *
108 * // Create a semaphore for managing concurrent access to a resource pool
109 * const program = Effect.gen(function*() {
110 * // Create a semaphore with 3 permits for a connection pool
111 * const connectionSemaphore = yield* TxSemaphore.make(3)
112 *
113 * // Check initial state
114 * const available = yield* TxSemaphore.available(connectionSemaphore)
115 * const capacity = yield* TxSemaphore.capacity(connectionSemaphore)
116 * return [capacity, available] as const
117 * })
118 *
119 * await Effect.runPromise(program) // => [3, 3]
120 * ```
121 *
122 * @see {@link available} for reading the current available permit count
123 * @see {@link capacity} for reading the fixed total permit count
124 *
125 * @category constructors
126 * @since 2.0.0
127 */
128export const make = (permits: number): Effect.Effect<TxSemaphore> =>
129 Effect.gen(function*() {
130 if (permits < 0) {
131 return yield* Effect.die(new Error("Permits must be non-negative"))
132 }
133
134 const permitsRef = yield* TxRef.make(permits)
135 return makeTxSemaphore(permitsRef, permits)
136 }).pipe(Effect.tx)
137
138/**
139 * Gets the current number of available permits in the semaphore.
140 *
141 * **When to use**
142 *
143 * Use to inspect how many permits are currently available.
144 *
145 * **Example** (Checking available permits)
146 *
147 * ```ts import.meta.vitest
148 * import { Effect, TxSemaphore } from "effect"
149 *
150 * const program = Effect.gen(function*() {
151 * const semaphore = yield* TxSemaphore.make(5)
152 *
153 * // Check available permits before acquiring
154 * const before = yield* TxSemaphore.available(semaphore)
155 *
156 * // Acquire some permits
157 * yield* TxSemaphore.acquire(semaphore)
158 * yield* TxSemaphore.acquire(semaphore)
159 *
160 * // Check available permits after acquiring
161 * const after = yield* TxSemaphore.available(semaphore)
162 * return [before, after] as const
163 * })
164 *
165 * await Effect.runPromise(program) // => [5, 3]
166 * ```
167 *
168 * @see {@link capacity} for reading the fixed total permit count
169 *
170 * @category combinators
171 * @since 2.0.0
172 */
173export const available = (self: TxSemaphore): Effect.Effect<number> => TxRef.get(self.permitsRef)
174
175/**
176 * Gets the maximum capacity (total permits) of the semaphore.
177 *
178 * **When to use**
179 *
180 * Use to inspect the fixed total number of permits managed by the semaphore.
181 *
182 * **Example** (Checking semaphore capacity)
183 *
184 * ```ts import.meta.vitest
185 * import { Effect, TxSemaphore } from "effect"
186 *
187 * const program = Effect.gen(function*() {
188 * const semaphore = yield* TxSemaphore.make(10)
189 *
190 * const capacity = yield* TxSemaphore.capacity(semaphore)
191 *
192 * // Capacity remains constant regardless of current permits
193 * yield* TxSemaphore.acquire(semaphore)
194 * const stillSame = yield* TxSemaphore.capacity(semaphore)
195 * return [capacity, stillSame] as const
196 * })
197 *
198 * await Effect.runPromise(program) // => [10, 10]
199 * ```
200 *
201 * @see {@link available} for reading the current available permit count
202 *
203 * @category combinators
204 * @since 4.0.0
205 */
206export const capacity = (self: TxSemaphore): Effect.Effect<number> => Effect.succeed(self.capacity)
207
208/**
209 * Acquires a single permit from the semaphore. If no permits are available,
210 * the effect will block until one becomes available.
211 *
212 * **When to use**
213 *
214 * Use to manually acquire one permit transactionally, waiting until one is
215 * available.
216 *
217 * **Example** (Acquiring a permit)
218 *
219 * ```ts import.meta.vitest
220 * import { Effect, TxSemaphore } from "effect"
221 *
222 * const program = Effect.gen(function*() {
223 * const semaphore = yield* TxSemaphore.make(2)
224 *
225 * yield* TxSemaphore.acquire(semaphore)
226 *
227 * yield* TxSemaphore.acquire(semaphore)
228 *
229 * return yield* TxSemaphore.available(semaphore)
230 * })
231 *
232 * await Effect.runPromise(program) // => 0
233 * ```
234 *
235 * @see {@link tryAcquire} for a non-blocking single-permit attempt
236 * @see {@link release} for returning one permit
237 * @see {@link withPermit} for automatic acquire and release around an effect
238 *
239 * @category combinators
240 * @since 2.0.0
241 */
242export const acquire = (self: TxSemaphore): Effect.Effect<void> =>
243 Effect.gen(function*() {
244 const permits = yield* TxRef.get(self.permitsRef)
245 if (permits <= 0) {
246 return yield* Effect.txRetry
247 }
248 yield* TxRef.set(self.permitsRef, permits - 1)
249 }).pipe(Effect.tx)
250
251/**
252 * Acquires the specified number of permits from the semaphore.
253 *
254 * **When to use**
255 *
256 * Use to manually acquire multiple permits transactionally, waiting until all
257 * requested permits are available.
258 *
259 * **Details**
260 *
261 * If fewer than `n` permits are available, the transaction retries until enough
262 * permits are released.
263 *
264 * **Gotchas**
265 *
266 * Passing a non-positive `n` dies with a defect. Passing a value greater than
267 * the semaphore capacity can wait forever because the capacity is fixed.
268 *
269 * **Example** (Acquiring multiple permits)
270 *
271 * ```ts import.meta.vitest
272 * import { Effect, TxSemaphore } from "effect"
273 *
274 * const program = Effect.gen(function*() {
275 * const semaphore = yield* TxSemaphore.make(5)
276 *
277 * yield* TxSemaphore.acquireN(semaphore, 3)
278 *
279 * return yield* TxSemaphore.available(semaphore)
280 * })
281 *
282 * await Effect.runPromise(program) // => 2
283 * ```
284 *
285 * @see {@link tryAcquireN} for a non-blocking multi-permit attempt
286 * @see {@link releaseN} for returning multiple permits
287 * @see {@link withPermits} for automatic acquire and release around an effect
288 *
289 * @category combinators
290 * @since 2.0.0
291 */
292export const acquireN: {
293 (n: number): (self: TxSemaphore) => Effect.Effect<void>
294 (self: TxSemaphore, n: number): Effect.Effect<void>
295} = dual(2, (self: TxSemaphore, n: number): Effect.Effect<void> => {
296 if (n <= 0) {
297 return Effect.die(new Error("Number of permits must be positive"))
298 }
299 return Effect.gen(function*() {
300 const permits = yield* TxRef.get(self.permitsRef)
301 if (permits < n) {
302 return yield* Effect.txRetry
303 }
304 yield* TxRef.set(self.permitsRef, permits - n)
305 }).pipe(Effect.tx)
306})
307
308/**
309 * Tries to acquire a single permit from the semaphore without blocking,
310 * returning `true` if successful or `false` if no permits are available.
311 *
312 * **When to use**
313 *
314 * Use to attempt a single-permit acquisition without retrying when no permit is
315 * available.
316 *
317 * **Example** (Trying to acquire a permit)
318 *
319 * ```ts import.meta.vitest
320 * import { Effect, TxSemaphore } from "effect"
321 *
322 * const program = Effect.gen(function*() {
323 * const semaphore = yield* TxSemaphore.make(1)
324 *
325 * // First try should succeed
326 * const first = yield* TxSemaphore.tryAcquire(semaphore)
327 *
328 * // Second try should fail (no permits left)
329 * const second = yield* TxSemaphore.tryAcquire(semaphore)
330 * return [first, second] as const
331 * })
332 *
333 * await Effect.runPromise(program) // => [true, false]
334 * ```
335 *
336 * @see {@link acquire} for waiting until one permit is available
337 * @see {@link tryAcquireN} for attempting to acquire multiple permits without blocking
338 *
339 * @category combinators
340 * @since 4.0.0
341 */
342export const tryAcquire = (self: TxSemaphore): Effect.Effect<boolean> =>
343 TxRef.modify(self.permitsRef, (permits: number) => {
344 if (permits > 0) {
345 return [true, permits - 1]
346 }
347 return [false, permits]
348 })
349
350/**
351 * Tries to acquire the specified number of permits from the semaphore without
352 * blocking, returning `true` if successful or `false` if not enough permits are
353 * available.
354 *
355 * **When to use**
356 *
357 * Use to attempt a multi-permit acquisition without retrying when not enough
358 * permits are available.
359 *
360 * **Example** (Trying to acquire multiple permits)
361 *
362 * ```ts import.meta.vitest
363 * import { Effect, TxSemaphore } from "effect"
364 *
365 * const program = Effect.gen(function*() {
366 * const semaphore = yield* TxSemaphore.make(3)
367 *
368 * // Try to acquire 2 permits (should succeed)
369 * const first = yield* TxSemaphore.tryAcquireN(semaphore, 2)
370 *
371 * // Try to acquire 2 more permits (should fail, only 1 left)
372 * const second = yield* TxSemaphore.tryAcquireN(semaphore, 2)
373 * return [first, second] as const
374 * })
375 *
376 * await Effect.runPromise(program) // => [true, false]
377 * ```
378 *
379 * @see {@link acquireN} for waiting until all requested permits are available
380 * @see {@link tryAcquire} for attempting to acquire one permit without blocking
381 *
382 * @category combinators
383 * @since 4.0.0
384 */
385export const tryAcquireN: {
386 (n: number): (self: TxSemaphore) => Effect.Effect<boolean>
387 (self: TxSemaphore, n: number): Effect.Effect<boolean>
388} = dual(2, (self: TxSemaphore, n: number): Effect.Effect<boolean> => {
389 if (n <= 0) {
390 return Effect.die(new Error("Number of permits must be positive"))
391 }
392 return TxRef.modify(self.permitsRef, (permits: number) => {
393 if (permits >= n) {
394 return [true, permits - n]
395 }
396 return [false, permits]
397 })
398})
399
400/**
401 * Releases one permit back to the semaphore, making it available for
402 * acquisition.
403 *
404 * **When to use**
405 *
406 * Use to manually return one permit after a transactional acquire.
407 *
408 * **Details**
409 *
410 * If the semaphore is already at capacity, this operation leaves the permit
411 * count unchanged.
412 *
413 * **Example** (Releasing a permit)
414 *
415 * ```ts import.meta.vitest
416 * import { Effect, TxSemaphore } from "effect"
417 *
418 * const program = Effect.gen(function*() {
419 * const semaphore = yield* TxSemaphore.make(2)
420 *
421 * // Acquire a permit
422 * yield* TxSemaphore.acquire(semaphore)
423 * const afterAcquire = yield* TxSemaphore.available(semaphore)
424 *
425 * // Release the permit
426 * yield* TxSemaphore.release(semaphore)
427 * const afterRelease = yield* TxSemaphore.available(semaphore)
428 * return [afterAcquire, afterRelease] as const
429 * })
430 *
431 * await Effect.runPromise(program) // => [1, 2]
432 * ```
433 *
434 * @see {@link acquire} for manually acquiring one permit
435 * @see {@link releaseN} for returning multiple permits
436 *
437 * @category combinators
438 * @since 2.0.0
439 */
440export const release = (self: TxSemaphore): Effect.Effect<void> =>
441 TxRef.update(self.permitsRef, (permits: number) => permits >= self.capacity ? permits : permits + 1)
442
443/**
444 * Releases the specified number of permits back to the semaphore.
445 *
446 * **When to use**
447 *
448 * Use to manually return multiple permits after a transactional acquire.
449 *
450 * **Details**
451 *
452 * The available permit count is capped at the semaphore capacity.
453 *
454 * **Gotchas**
455 *
456 * Passing a non-positive `n` dies with a defect.
457 *
458 * **Example** (Releasing multiple permits)
459 *
460 * ```ts import.meta.vitest
461 * import { Effect, TxSemaphore } from "effect"
462 *
463 * const program = Effect.gen(function*() {
464 * const semaphore = yield* TxSemaphore.make(5)
465 *
466 * // Acquire 3 permits
467 * yield* TxSemaphore.acquireN(semaphore, 3)
468 * const afterAcquire = yield* TxSemaphore.available(semaphore)
469 *
470 * // Release 2 permits
471 * yield* TxSemaphore.releaseN(semaphore, 2)
472 * const afterRelease = yield* TxSemaphore.available(semaphore)
473 * return [afterAcquire, afterRelease] as const
474 * })
475 *
476 * await Effect.runPromise(program) // => [2, 4]
477 * ```
478 *
479 * @see {@link acquireN} for manually acquiring multiple permits
480 * @see {@link release} for returning one permit
481 *
482 * @category combinators
483 * @since 2.0.0
484 */
485export const releaseN: {
486 (n: number): (self: TxSemaphore) => Effect.Effect<void>
487 (self: TxSemaphore, n: number): Effect.Effect<void>
488} = dual(2, (self: TxSemaphore, n: number): Effect.Effect<void> => {
489 if (n <= 0) {
490 return Effect.die(new Error("Number of permits must be positive"))
491 }
492 return TxRef.update(self.permitsRef, (permits: number) => {
493 const newPermits = permits + n
494 return newPermits > self.capacity ? self.capacity : newPermits
495 })
496})
497
498/**
499 * Executes an effect with a single permit from the semaphore. The permit is
500 * automatically acquired before execution and released afterwards, even if the
501 * effect fails or is interrupted.
502 *
503 * **When to use**
504 *
505 * Use to run an effect while automatically acquiring and releasing one
506 * transactional permit.
507 *
508 * **Details**
509 *
510 * The permit acquisition and release operations use atomic semantics to ensure
511 * proper resource management with Effect's scoped operations.
512 *
513 * **Example** (Running an effect with a permit)
514 *
515 * ```ts import.meta.vitest
516 * import { Effect, TxSemaphore } from "effect"
517 *
518 * const program = Effect.gen(function*() {
519 * const semaphore = yield* TxSemaphore.make(2)
520 * const events: Array<string> = []
521 *
522 * // Execute database operation with automatic permit management
523 * const result = yield* TxSemaphore.withPermit(
524 * semaphore,
525 * Effect.gen(function*() {
526 * events.push("permit acquired")
527 * yield* Effect.yieldNow
528 * events.push("operation complete")
529 * return "query result"
530 * })
531 * )
532 *
533 * // Permit is automatically released here
534 * const available = yield* TxSemaphore.available(semaphore)
535 * return [events, result, available] as const
536 * })
537 *
538 * await Effect.runPromise(program) // => [["permit acquired", "operation complete"], "query result", 2]
539 * ```
540 *
541 * @see {@link withPermits} for automatically acquiring and releasing multiple permits
542 * @see {@link withPermitScoped} for acquiring one permit for the current scope
543 * @see {@link acquire} for manual single-permit acquisition
544 *
545 * @category combinators
546 * @since 2.0.0
547 */
548export const withPermit: {
549 (self: TxSemaphore): <A, E, R>(effect: Effect.Effect<A, E, R>) => Effect.Effect<A, E, R>
550 <A, E, R>(self: TxSemaphore, effect: Effect.Effect<A, E, R>): Effect.Effect<A, E, R>
551} = ((...args: Array<any>) => {
552 if (args.length === 1) {
553 const [self] = args
554 return (effect: Effect.Effect<any, any, any>) =>
555 Effect.acquireUseRelease(
556 acquire(self),
557 () => effect,
558 () => release(self)
559 )
560 }
561 const [self, effect] = args
562 return Effect.acquireUseRelease(
563 acquire(self),
564 () => effect,
565 () => release(self)
566 )
567}) as any
568
569/**
570 * Runs an effect while holding the specified number of permits from the
571 * semaphore.
572 *
573 * **When to use**
574 *
575 * Use to run an effect while automatically acquiring and releasing multiple
576 * transactional permits.
577 *
578 * **Details**
579 *
580 * The permits are acquired before the effect starts and released after it
581 * completes, fails, or is interrupted.
582 *
583 * **Gotchas**
584 *
585 * Passing a non-positive `n` dies with a defect. Passing a value greater than
586 * the semaphore capacity can wait forever.
587 *
588 * **Example** (Running an effect with multiple permits)
589 *
590 * ```ts import.meta.vitest
591 * import { Effect, TxSemaphore } from "effect"
592 *
593 * const program = Effect.gen(function*() {
594 * const semaphore = yield* TxSemaphore.make(5)
595 * const events: Array<string> = []
596 *
597 * // Execute batch operation with 3 permits
598 * const results = yield* TxSemaphore.withPermits(
599 * semaphore,
600 * 3,
601 * Effect.gen(function*() {
602 * events.push("3 permits acquired")
603 * yield* Effect.yieldNow
604 * return ["result1", "result2", "result3"]
605 * })
606 * )
607 *
608 * // All 3 permits are automatically released here
609 * const available = yield* TxSemaphore.available(semaphore)
610 * return [events, results, available] as const
611 * })
612 *
613 * await Effect.runPromise(program) // => [["3 permits acquired"], ["result1", "result2", "result3"], 5]
614 * ```
615 *
616 * @see {@link withPermit} for automatically acquiring and releasing one permit
617 * @see {@link acquireN} for manual multi-permit acquisition
618 *
619 * @category combinators
620 * @since 2.0.0
621 */
622export const withPermits: {
623 (self: TxSemaphore, n: number): <A, E, R>(effect: Effect.Effect<A, E, R>) => Effect.Effect<A, E, R>
624 <A, E, R>(self: TxSemaphore, n: number, effect: Effect.Effect<A, E, R>): Effect.Effect<A, E, R>
625} = ((...args: Array<any>) => {
626 if (args.length === 2) {
627 const [self, n] = args
628 return (effect: Effect.Effect<any, any, any>) =>
629 Effect.acquireUseRelease(
630 acquireN(self, n),
631 () => effect,
632 () => releaseN(self, n)
633 )
634 }
635 const [self, n, effect] = args
636 return Effect.acquireUseRelease(
637 acquireN(self, n),
638 () => effect,
639 () => releaseN(self, n)
640 )
641}) as any
642
643/**
644 * Acquires a single permit from the semaphore in a scoped manner. The permit
645 * will be automatically released when the scope is closed, even if effects
646 * within the scope fail or are interrupted.
647 *
648 * **When to use**
649 *
650 * Use to acquire one transactional permit for the lifetime of the current
651 * scope.
652 *
653 * **Details**
654 *
655 * The permit acquisition and release operations use atomic semantics to ensure
656 * proper resource management with Effect's scoped operations.
657 *
658 * **Example** (Acquiring a scoped permit)
659 *
660 * ```ts import.meta.vitest
661 * import { Effect, TxSemaphore } from "effect"
662 *
663 * const program = Effect.gen(function*() {
664 * const semaphore = yield* TxSemaphore.make(3)
665 * const events: Array<string> = []
666 *
667 * yield* Effect.scoped(
668 * Effect.gen(function*() {
669 * // Acquire permit for the duration of this scope
670 * yield* TxSemaphore.withPermitScoped(semaphore)
671 * events.push("permit acquired for scope")
672 *
673 * // Do work within the scope
674 * yield* Effect.yieldNow
675 * events.push("work completed")
676 *
677 * // Permit will be automatically released when scope closes
678 * })
679 * )
680 *
681 * const available = yield* TxSemaphore.available(semaphore)
682 * return [events, available] as const
683 * })
684 *
685 * await Effect.runPromise(program) // => [["permit acquired for scope", "work completed"], 3]
686 * ```
687 *
688 * @see {@link withPermit} for acquiring one permit around a single effect
689 * @see {@link acquire} for manual single-permit acquisition
690 *
691 * @category combinators
692 * @since 2.0.0
693 */
694export const withPermitScoped = (self: TxSemaphore): Effect.Effect<void, never, Scope.Scope> =>
695 Effect.acquireRelease(
696 acquire(self),
697 () => release(self)
698 )
699
700/**
701 * Determines if the provided value is a TxSemaphore.
702 *
703 * **When to use**
704 *
705 * Use to narrow an unknown value before treating it as a `TxSemaphore`.
706 *
707 * **Example** (Checking semaphore values)
708 *
709 * ```ts import.meta.vitest
710 * import { Effect, TxSemaphore } from "effect"
711 *
712 * const program = Effect.gen(function*() {
713 * const semaphore = yield* TxSemaphore.make(5)
714 * const notSemaphore = { some: "object" }
715 *
716 * const semaphoreResult = TxSemaphore.isTxSemaphore(semaphore)
717 * const objectResult = TxSemaphore.isTxSemaphore(notSemaphore)
718 *
719 * // Useful for runtime type checking in generic functions
720 * if (TxSemaphore.isTxSemaphore(semaphore)) {
721 * const available = yield* TxSemaphore.available(semaphore)
722 * return [semaphoreResult, objectResult, available] as const
723 * }
724 * return [semaphoreResult, objectResult, 0] as const
725 * })
726 *
727 * await Effect.runPromise(program) // => [true, false, 5]
728 * ```
729 *
730 * @see {@link make} for creating a `TxSemaphore`
731 *
732 * @category guards
733 * @since 4.0.0
734 */
735export const isTxSemaphore = (u: unknown): u is TxSemaphore => hasProperty(u, TypeId)