1import { assert, it } from "@effect/vitest"
2import { Deferred, Effect, Exit, Fiber, Scope } from "effect"
3import { EntityAddress, EntityId, EntityType, ShardId } from "effect/cluster"
4import { isActive } from "effect/cluster/internal/interruptors"
5import { ResourceRef } from "effect/cluster/internal/resourceRef"
6
7// A gate signals `reached` and then waits for `open`.
8const makeGate = Effect.fnUntraced(function*() {
9 const gate = { reached: yield* Deferred.make<void>(), open: yield* Deferred.make<void>() }
10 return gate
11})
12type Gate = Effect.Success<ReturnType<typeof makeGate>>
13const pass = (gate: Gate) => Effect.andThen(Deferred.succeed(gate.reached, void 0), Deferred.await(gate.open))
14
15// Acquires 1, 2, 3, ... An acquisition listed in `gated` waits at its gate
16// before completing; the release of one listed in `gatedRelease` waits at its
17// gate too.
18const makeAcquire = Effect.fnUntraced(function*(options?: {
19 readonly gated?: ReadonlyArray<number>
20 readonly gatedRelease?: ReadonlyArray<number>
21}) {
22 let acquisitions = 0
23 let completed = 0
24 let releases = 0
25 const gates = new Map<number, Gate>()
26 const releaseGates = new Map<number, Gate>()
27 for (const acquisition of options?.gated ?? []) gates.set(acquisition, yield* makeGate())
28 for (const acquisition of options?.gatedRelease ?? []) releaseGates.set(acquisition, yield* makeGate())
29 const acquire = (scope: Scope.Scope) =>
30 Effect.gen(function*() {
31 const acquisition = ++acquisitions
32 const releaseGate = releaseGates.get(acquisition)
33 yield* Scope.addFinalizer(
34 scope,
35 Effect.andThen(releaseGate ? pass(releaseGate) : Effect.void, Effect.sync(() => releases++))
36 )
37 const gate = gates.get(acquisition)
38 if (gate) yield* pass(gate)
39 completed++
40 return acquisition
41 })
42 return {
43 acquire,
44 gate: (acquisition: number) => gates.get(acquisition)!,
45 releaseGate: (acquisition: number) => releaseGates.get(acquisition)!,
46 acquisitions: () => acquisitions,
47 /** Acquisitions that made it past their gate. */
48 completed: () => completed,
49 releases: () => releases
50 } as const
51})
52
53// Lets forked fibers run until they block.
54const settle = Effect.gen(function*() {
55 for (let i = 0; i < 10; i++) yield* Effect.yieldNow
56})
57
58it.effect("rebuilds only the current generation", () =>
59 Effect.gen(function*() {
60 const parentScope = yield* Scope.make()
61 const { acquire } = yield* makeAcquire()
62 const ref = yield* ResourceRef.from(parentScope, acquire)
63 assert.strictEqual(ref.getUnsafe(), 1)
64
65 yield* ref.rebuildUnsafe({ from: 1 })!
66 assert.strictEqual(ref.getUnsafe(), 2)
67 // A replaced value can never replace its successor.
68 assert.isUndefined(ref.rebuildUnsafe({ from: 1 }))
69
70 // Without `from`, an idle ref is rebuilt.
71 yield* ref.rebuildUnsafe()!
72 assert.strictEqual(ref.getUnsafe(), 3)
73 assert.isUndefined(ref.rebuildUnsafe({ from: 2 }))
74
75 yield* Scope.close(parentScope, Exit.void)
76 assert.isUndefined(ref.getUnsafe())
77 assert.isUndefined(ref.rebuildUnsafe({ from: 3 }))
78 assert.isUndefined(ref.rebuildUnsafe())
79 assert.isTrue(Exit.hasInterrupts(yield* Effect.exit(ref.await)))
80 }))
81
82it.effect("refuses a second rebuild of a generation without disturbing the accepted one", () =>
83 Effect.scoped(Effect.gen(function*() {
84 const parentScope = yield* Effect.scope
85 const { acquire, gate } = yield* makeAcquire({ gated: [2] })
86 const ref = yield* ResourceRef.from(parentScope, acquire)
87
88 const accepted = ref.rebuildUnsafe({ from: 1 })
89 assert.isDefined(accepted)
90 // Admission closes as soon as the rebuild is accepted, before it runs.
91 assert.isUndefined(ref.getUnsafe())
92 assert.isUndefined(ref.rebuildUnsafe({ from: 1 }))
93 assert.isUndefined(ref.rebuildUnsafe())
94
95 const rebuild = yield* Effect.forkChild(accepted!)
96 yield* Deferred.await(gate(2).reached)
97 assert.isUndefined(ref.rebuildUnsafe({ from: 1 }))
98 assert.isUndefined(ref.rebuildUnsafe())
99
100 yield* Deferred.succeed(gate(2).open, void 0)
101 assert.deepStrictEqual(yield* Fiber.await(rebuild), Exit.void)
102 assert.strictEqual(ref.getUnsafe(), 2)
103 assert.strictEqual(yield* ref.await, 2)
104 })))
105
106it.effect("interrupts prepare when its generation is replaced", () =>
107 Effect.scoped(Effect.gen(function*() {
108 const parentScope = yield* Effect.scope
109 const { acquire } = yield* makeAcquire()
110 const ref = yield* ResourceRef.from(parentScope, acquire)
111 let newer: Effect.Effect<void> | undefined
112 let finishedPreparing = false
113
114 const superseded = yield* Effect.forkChild(
115 ref.rebuildUnsafe({
116 from: 1,
117 // Replaces the value being prepared, as a replayed request that defects does.
118 prepare: (value) =>
119 Effect.gen(function*() {
120 newer = ref.rebuildUnsafe({ from: value })
121 yield* Effect.yieldNow
122 finishedPreparing = true
123 })
124 })!
125 )
126 assert.isTrue(Exit.hasInterrupts(yield* Fiber.await(superseded)))
127 assert.isFalse(finishedPreparing, "prepare must be interrupted")
128 assert.isDefined(newer)
129
130 // The superseded value is never admitted.
131 const waiter = yield* Effect.forkChild(ref.await)
132 yield* Effect.yieldNow
133 assert.isUndefined(ref.getUnsafe())
134 assert.isUndefined(waiter.pollUnsafe())
135
136 yield* newer!
137 assert.strictEqual(yield* Fiber.join(waiter), 3)
138 assert.strictEqual(ref.getUnsafe(), 3)
139 })))
140
141it.effect("closes a replaced generation whose rebuild never ran when the ref closes", () =>
142 Effect.gen(function*() {
143 const parentScope = yield* Scope.make()
144 const { acquire, acquisitions, releases } = yield* makeAcquire()
145 const ref = yield* ResourceRef.from(parentScope, acquire)
146
147 const rebuild = ref.rebuildUnsafe({ from: 1 })
148 assert.isDefined(rebuild)
149 const waiter = yield* Effect.forkChild(ref.await)
150 yield* Effect.yieldNow
151
152 yield* Scope.close(parentScope, Exit.void)
153 assert.strictEqual(releases(), 1, "the replaced generation must be closed")
154 assert.isTrue(Exit.hasInterrupts(yield* Fiber.await(waiter)))
155
156 // Running the abandoned rebuild afterwards acquires nothing.
157 assert.isTrue(Exit.hasInterrupts(yield* Effect.exit(rebuild!)))
158 assert.strictEqual(acquisitions(), 1)
159 }))
160
161it.effect("does not leak teardown membership when a rebuild is discarded", () =>
162 Effect.gen(function*() {
163 const parentScope = yield* Scope.make()
164 const address = EntityAddress.make({
165 shardId: ShardId.make("resource-ref-discard", 1),
166 entityType: EntityType.make("ResourceRefDiscard"),
167 entityId: EntityId.make("1")
168 })
169 const { acquire } = yield* makeAcquire()
170 const ref = yield* ResourceRef.from(parentScope, acquire, address)
171 assert.isDefined(ref.rebuildUnsafe({ from: 1 }))
172 assert.isFalse(isActive(address))
173 yield* Scope.close(parentScope, Exit.void)
174 assert.isFalse(isActive(address))
175 }))
176
177it.effect("does not leak teardown membership when a rebuild is interrupted", () =>
178 Effect.scoped(Effect.gen(function*() {
179 const parentScope = yield* Effect.scope
180 const address = EntityAddress.make({
181 shardId: ShardId.make("resource-ref-interrupt", 1),
182 entityType: EntityType.make("ResourceRefInterrupt"),
183 entityId: EntityId.make("1")
184 })
185 const { acquire, releaseGate } = yield* makeAcquire({ gatedRelease: [1] })
186 const ref = yield* ResourceRef.from(parentScope, acquire, address)
187 const fiber = yield* Effect.forkChild(ref.rebuildUnsafe({ from: 1 })!)
188 yield* Fiber.interrupt(fiber)
189 yield* Deferred.succeed(releaseGate(1).open, void 0)
190 assert.isFalse(isActive(address))
191 })))
192
193it.effect("stays rebuildable when a failed replacement cannot be released", () =>
194 Effect.scoped(Effect.gen(function*() {
195 const parentScope = yield* Effect.scope
196 // Acquisition 2 holds something whose release hangs, then never completes.
197 const { acquire, gate, releaseGate } = yield* makeAcquire({ gated: [2], gatedRelease: [2] })
198 const ref = yield* ResourceRef.from(parentScope, acquire)
199
200 yield* Effect.gen(function*() {
201 const rebuild = yield* Effect.forkChild(ref.rebuildUnsafe({ from: 1 })!)
202 yield* Deferred.await(gate(2).reached)
203 // A deadline abandons the rebuild, whose replacement then cannot be released.
204 rebuild.interruptUnsafe()
205 yield* settle
206 assert.isDefined(ref.rebuildUnsafe(), "a hung release must not block the next rebuild")
207 }).pipe(Effect.ensuring(Deferred.succeed(releaseGate(2).open, void 0)))
208 })))
209
210it.effect("interrupts a rebuild that is acquiring when the ref closes", () =>
211 Effect.gen(function*() {
212 const parentScope = yield* Scope.make()
213 const { acquire, completed, gate } = yield* makeAcquire({ gated: [2] })
214 const ref = yield* ResourceRef.from(parentScope, acquire)
215
216 const rebuild = yield* Effect.forkChild(ref.rebuildUnsafe({ from: 1 })!)
217 yield* Deferred.await(gate(2).reached)
218 const waiter = yield* Effect.forkChild(ref.await)
219 yield* Scope.close(parentScope, Exit.void)
220 yield* settle
221 const exit = rebuild.pollUnsafe()
222 assert.isDefined(exit, "closing the ref must interrupt acquisition")
223 assert.isTrue(Exit.hasInterrupts(exit))
224 assert.isTrue(Exit.hasInterrupts(yield* Fiber.await(waiter)))
225
226 yield* Deferred.succeed(gate(2).open, void 0)
227 yield* settle
228 assert.strictEqual(completed(), 1, "acquisition must not continue after the close")
229 }))