1import * as Deferred from "../../Deferred.ts"
2import * as Effect from "../../Effect.ts"
3import * as Exit from "../../Exit.ts"
4import * as Fiber from "../../Fiber.ts"
5import * as Scope from "../../Scope.ts"
6import type { EntityAddress } from "../EntityAddress.ts"
7import { aroundEntity } from "./interruptors.ts"
8
9type Pending<A> = {
10 readonly _tag: "Pending"
11 readonly scope: Scope.Closeable
12 /** Predecessor scope or ongoing release to await before acquisition. */
13 readonly releases: Scope.Closeable | Fiber.Fiber<void> | undefined
14 releasing: boolean
15 /** Shared with waiters across superseded rebuilds. */
16 readonly ready: Deferred.Deferred<A>
17 value: A | undefined
18 work: Fiber.Fiber<void> | undefined
19}
20
21type State<A> =
22 | { readonly _tag: "Closed" }
23 | { readonly _tag: "Ready"; readonly scope: Scope.Closeable; readonly value: A }
24 | Pending<A>
25 | {
26 readonly _tag: "Failed"
27 readonly ready: Deferred.Deferred<A>
28 readonly releasing: Fiber.Fiber<void> | undefined
29 }
30
31/**
32 * A replaceable resource whose acquired value identifies its generation.
33 *
34 * @internal
35 */
36export class ResourceRef<A> {
37 static from = Effect.fnUntraced(function*<A>(
38 parentScope: Scope.Scope,
39 acquire: (scope: Scope.Scope) => Effect.Effect<A>,
40 teardownAddress?: EntityAddress
41 ) {
42 // Close the ref before joining releases, so waiters wake even if release stalls.
43 const releaseScope = Scope.forkUnsafe(parentScope)
44 const ref = new ResourceRef(releaseScope, acquire, teardownAddress)
45 yield* Scope.addFinalizerExit(parentScope, (exit) => ref.close(exit))
46 const pending = ref.pending(undefined, Deferred.makeUnsafe())
47 const value = yield* acquire(pending.scope)
48 if (ref.state === pending) ref.publish(pending, value)
49 return ref
50 })
51
52 private state: State<A> = { _tag: "Closed" }
53 private readonly releaseScope: Scope.Scope
54 private readonly acquire: (scope: Scope.Scope) => Effect.Effect<A>
55 private readonly teardownAddress: EntityAddress | undefined
56 private constructor(
57 releaseScope: Scope.Scope,
58 acquire: (scope: Scope.Scope) => Effect.Effect<A>,
59 teardownAddress: EntityAddress | undefined
60 ) {
61 this.releaseScope = releaseScope
62 this.acquire = acquire
63 this.teardownAddress = teardownAddress
64 }
65
66 /** Returns the value only after acquisition and preparation complete. */
67 getUnsafe(): A | undefined {
68 return this.state._tag === "Ready" ? this.state.value : undefined
69 }
70
71 /** Waits for a ready value; interrupts when the ref closes. */
72 readonly await: Effect.Effect<A> = Effect.suspend(() => {
73 const s = this.state
74 switch (s._tag) {
75 case "Closed":
76 return Effect.interrupt
77 case "Ready":
78 return Effect.succeed(s.value)
79 case "Pending":
80 case "Failed":
81 return Deferred.await(s.ready)
82 }
83 })
84
85 /**
86 * Replaces the generation, holding waiters until acquisition and `prepare` finish.
87 * Returns `undefined` if closed, if `from` differs from the ready or preparing
88 * value, or if `from` is omitted while a rebuild is pending.
89 *
90 * Acceptance immediately blocks waiters; run the returned effect to release
91 * the predecessor and build its replacement. Closing or superseding the ref
92 * interrupts the rebuild. Failure wakes waiters with its cause; the next
93 * rebuild must omit `from` and await cleanup. Interruption during predecessor
94 * release instead carries waiters forward without awaiting that release again.
95 */
96 rebuildUnsafe(options?: {
97 readonly from?: A | undefined
98 readonly prepare?: ((value: A) => Effect.Effect<void>) | undefined
99 }): Effect.Effect<void> | undefined {
100 const s = this.state
101 const from = options?.from
102 let pending: Pending<A>
103 switch (s._tag) {
104 case "Closed":
105 return undefined
106 case "Ready": {
107 if (from !== undefined && from !== s.value) return undefined
108 pending = this.pending(s.scope, Deferred.makeUnsafe())
109 break
110 }
111 case "Failed": {
112 if (from !== undefined) return undefined
113 pending = this.pending(s.releasing, s.ready)
114 break
115 }
116 case "Pending": {
117 if (from === undefined || s.value !== from) return undefined
118 s.work?.interruptUnsafe()
119 pending = this.pending(s.scope, s.ready)
120 break
121 }
122 }
123 return this.run(pending, options?.prepare)
124 }
125
126 private pending(
127 releases: Scope.Closeable | Fiber.Fiber<void> | undefined,
128 ready: Deferred.Deferred<A>
129 ): Pending<A> {
130 const pending: Pending<A> = {
131 _tag: "Pending",
132 scope: Scope.makeUnsafe(),
133 releases,
134 releasing: false,
135 ready,
136 value: undefined,
137 work: undefined
138 }
139 this.state = pending
140 return pending
141 }
142
143 private publish(pending: Pending<A>, value: A): void {
144 this.state = { _tag: "Ready", scope: pending.scope, value }
145 Deferred.doneUnsafe(pending.ready, Exit.succeed(value))
146 }
147
148 private run(pending: Pending<A>, prepare: ((value: A) => Effect.Effect<void>) | undefined): Effect.Effect<void> {
149 const work = Effect.suspend(() => {
150 if (this.state !== pending) return Effect.interrupt
151 const releases = pending.releases
152 if (!releases) return Effect.void
153 pending.releasing = true
154 return Effect.andThen(
155 Fiber.isFiber(releases) ? Fiber.join(releases) : Effect.flatMap(this.release(releases), Fiber.join),
156 Effect.sync(() => {
157 pending.releasing = false
158 })
159 )
160 }).pipe(
161 Effect.andThen(Effect.suspend(() => this.acquire(pending.scope))),
162 Effect.tap((value) => {
163 if (this.state !== pending) return Effect.interrupt
164 pending.value = value
165 return prepare ? prepare(value) : Effect.void
166 }),
167 Effect.flatMap((value) => {
168 if (this.state !== pending) return Effect.interrupt
169 this.publish(pending, value)
170 return Effect.void
171 }),
172 Effect.onExit((exit) => {
173 if (Exit.isSuccess(exit) || this.state !== pending) {
174 return Effect.void
175 }
176 if (pending.releasing) {
177 // No resource was acquired. Carry waiters forward without rejoining the stalled release.
178 this.state = { _tag: "Failed", ready: pending.ready, releasing: undefined }
179 return Effect.void
180 }
181 return Effect.map(this.release(pending.scope), (releasing) => {
182 if (this.state !== pending) return
183 this.state = { _tag: "Failed", ready: Deferred.makeUnsafe(), releasing }
184 Deferred.doneUnsafe(pending.ready, Exit.failCause(exit.cause))
185 })
186 })
187 )
188 return Effect.flatMap(Effect.forkChild(work, { startImmediately: true }), (fiber) => {
189 pending.work = fiber
190 // Replaced or closed before the fiber handle existed.
191 if (this.state !== pending) fiber.interruptUnsafe()
192 return Fiber.join(fiber)
193 })
194 }
195
196 // A stalled release must not prevent rebuild interruption.
197 private release(scope: Scope.Closeable): Effect.Effect<Fiber.Fiber<void>> {
198 const close = Scope.close(scope, Exit.void)
199 return Effect.forkIn(
200 this.teardownAddress ? aroundEntity(this.teardownAddress, close) : close,
201 this.releaseScope,
202 { startImmediately: true }
203 )
204 }
205
206 private close(exit: Exit.Exit<unknown, unknown>): Effect.Effect<void> {
207 const s = this.state
208 this.state = { _tag: "Closed" }
209 switch (s._tag) {
210 case "Closed":
211 return Effect.void
212 case "Ready":
213 return Scope.close(s.scope, exit)
214 case "Failed": {
215 Deferred.doneUnsafe(s.ready, Effect.interrupt)
216 return Effect.void
217 }
218 case "Pending": {
219 Deferred.doneUnsafe(s.ready, Effect.interrupt)
220 s.work?.interruptUnsafe()
221 const close = Scope.close(s.scope, exit)
222 return s.releases && !Fiber.isFiber(s.releases) ? Effect.andThen(Scope.close(s.releases, exit), close) : close
223 }
224 }
225 }
226}