1import { assert, describe, it } from "@effect/vitest"
2import { type Cause, Deferred, Effect, Exit, Fiber, Option, Queue, Schema, Scope, Stream } from "effect"
3import { Entity, ShardingConfig } from "effect/cluster"
4import { CurrentActivationScope } from "effect/cluster/internal/entityActivation"
5import { Rpc } from "effect/rpc"
6import { CallerId, ContextBleedEntity, ContextBleedLayer, TestEntity, TestEntityLayer, User } from "./TestEntity.ts"
7
8const StreamEntity = Entity.make("StreamEntity", [
9 Rpc.make("Watch", {
10 success: Schema.Number,
11 stream: true
12 })
13])
14
15const FatalDefectEntity = Entity.make("FatalDefectEntity", [
16 Rpc.make("Hold", { success: Schema.Number }),
17 Rpc.make("Bad", { error: Schema.String })
18])
19
20const snapshot = <A, E>(exit: Exit.Exit<A, E>) =>
21 Exit.isSuccess(exit)
22 ? { _tag: "Success", value: exit.value }
23 : {
24 _tag: "Failure",
25 reasons: exit.cause.reasons.map((reason) =>
26 reason._tag === "Die"
27 ? { _tag: reason._tag, defect: reason.defect }
28 : reason._tag === "Fail"
29 ? { _tag: reason._tag, error: reason.error }
30 : { _tag: reason._tag }
31 )
32 }
33
34const observeFatalDefect = (disableFatalDefects: boolean | undefined, defecting: boolean, separateIds = false) =>
35 Effect.gen(function*() {
36 const entered = yield* Deferred.make<void>()
37 const release = yield* Deferred.make<void>()
38 const layer = FatalDefectEntity.toLayer({
39 Hold: () =>
40 Effect.gen(function*() {
41 yield* Deferred.succeed(entered, undefined)
42 yield* Deferred.await(release)
43 return 42
44 }),
45 Bad: () => defecting ? Effect.die("fixture defect") : Effect.fail("typed failure")
46 }, {
47 concurrency: "unbounded",
48 ...(disableFatalDefects === undefined ? {} : { disableFatalDefects })
49 })
50 const clientFor = yield* Entity.makeTestClient(FatalDefectEntity, layer)
51 const holdClient = yield* clientFor("one")
52 const badClient = yield* clientFor(separateIds ? "two" : "one")
53 yield* Effect.addFinalizer(() => Deferred.succeed(release, undefined))
54 const good = yield* Effect.forkChild(holdClient.Hold())
55 yield* Deferred.await(entered)
56 const badExit = yield* Effect.exit(badClient.Bad())
57 yield* Deferred.succeed(release, undefined)
58 const goodExit = yield* Fiber.await(good)
59 return { badExit: snapshot(badExit), goodExit: snapshot(goodExit) }
60 }).pipe(Effect.provide(ShardingConfig.layerDefaults), Effect.timeout("5 seconds"))
61
62describe.concurrent("Entity", () => {
63 describe("makeTestClient", () => {
64 it.effect("provides an activation scope per entity and closes it with the client", () =>
65 Effect.gen(function*() {
66 const entity = Entity.make("ActivationScope", [Rpc.make("Read", { success: Schema.String })])
67 const activations: Array<Scope.Scope> = []
68 const closed: Array<string> = []
69 const layer = entity.toLayer(Effect.gen(function*() {
70 const address = yield* Entity.CurrentAddress
71 const activation = yield* Effect.serviceOption(CurrentActivationScope)
72 assert(Option.isSome(activation), "makeTestClient must provide the entity activation scope")
73 activations.push(activation.value)
74 yield* Scope.addFinalizer(activation.value, Effect.sync(() => closed.push(address.entityId)))
75 return { Read: () => Effect.succeed(address.entityId) }
76 }))
77 yield* Effect.scoped(Effect.gen(function*() {
78 const makeClient = yield* Entity.makeTestClient(entity, layer)
79 const first = yield* makeClient("one")
80 const second = yield* makeClient("two")
81 assert.strictEqual(yield* first.Read(), "one")
82 assert.strictEqual(yield* second.Read(), "two")
83 assert.strictEqual(activations.length, 2)
84 assert.notStrictEqual(activations[0], activations[1])
85 assert.deepStrictEqual(closed, [])
86 }))
87 assert.deepStrictEqual(closed.sort(), ["one", "two"])
88 }).pipe(Effect.provide(TestShardingConfig)))
89
90 it.effect("creates an in-memory client for an entity layer", () =>
91 Effect.gen(function*() {
92 const makeClient = yield* Entity.makeTestClient(TestEntity, TestEntityLayer)
93 const client = yield* makeClient("123")
94 const user = yield* client.GetUser({ id: 1 })
95 assert.deepEqual(user, new User({ id: 1, name: "User 1" }))
96 }).pipe(Effect.provide(TestShardingConfig)))
97
98 it.effect("does not freeze the acquiring fiber's context into the entity server", () =>
99 Effect.gen(function*() {
100 const makeClient = yield* Entity.makeTestClient(ContextBleedEntity, ContextBleedLayer)
101
102 const client = yield* makeClient("1").pipe(Effect.provideService(CallerId, "A"))
103
104 const observed = yield* client.ReadCaller()
105 assert.strictEqual(observed, "none")
106 }).pipe(Effect.provide(TestShardingConfig)))
107
108 for (const flag of [true, false, undefined]) {
109 const label = flag === undefined ? "omitted" : String(flag)
110 it.live(`isolates defects when disableFatalDefects is ${label}`, () =>
111 Effect.gen(function*() {
112 const actual = yield* observeFatalDefect(flag, true)
113 assert.deepEqual(actual.badExit, snapshot(Exit.die("fixture defect")), "Bad preserves the original defect")
114 assert.deepEqual(
115 actual.goodExit,
116 snapshot(flag === true ? Exit.succeed(42) : Exit.die("fixture defect")),
117 "Hold isolation follows the registered flag"
118 )
119 }))
120
121 it.live(`keeps typed failures request-local when disableFatalDefects is ${label}`, () =>
122 Effect.gen(function*() {
123 const actual = yield* observeFatalDefect(flag, false)
124 assert.deepEqual(actual.badExit, snapshot(Exit.fail("typed failure")))
125 assert.deepEqual(actual.goodExit, snapshot(Exit.succeed(42)))
126 }))
127 }
128
129 it.live("does not send fatal defects across entity IDs", () =>
130 Effect.gen(function*() {
131 const actual = yield* observeFatalDefect(false, true, true)
132 assert.deepEqual(actual.badExit, snapshot(Exit.die("fixture defect")))
133 assert.deepEqual(actual.goodExit, snapshot(Exit.succeed(42)))
134 }))
135 })
136
137 describe("toLayerQueue", () => {
138 it.effect("replies to a streaming RPC with a Stream", () =>
139 Effect.gen(function*() {
140 const layer = StreamEntity.toLayerQueue((mailbox, replier) =>
141 Effect.gen(function*() {
142 while (true) {
143 const req = yield* Queue.take(mailbox)
144 yield* replier.succeed(req, Stream.make(1, 2, 3))
145 }
146 })
147 )
148
149 const makeClient = yield* Entity.makeTestClient(StreamEntity, layer)
150 const client = yield* makeClient("entity-1")
151 const results: Array<number> = []
152 yield* client.Watch().pipe(
153 Stream.take(3),
154 Stream.runForEach((n) => Effect.sync(() => results.push(n)))
155 )
156 assert.deepEqual(results, [1, 2, 3])
157 }).pipe(Effect.provide(TestShardingConfig)))
158
159 it.effect("replies to a streaming RPC with a Dequeue", () =>
160 Effect.gen(function*() {
161 const layer = StreamEntity.toLayerQueue((mailbox, replier) =>
162 Effect.gen(function*() {
163 while (true) {
164 const req = yield* Queue.take(mailbox)
165 const q = yield* Queue.make<number, Cause.Done>()
166 yield* replier.succeed(req, q)
167 yield* Queue.offer(q, 1)
168 yield* Queue.offer(q, 2)
169 yield* Queue.offer(q, 3)
170 yield* Queue.end(q)
171 }
172 }) as Effect.Effect<never>
173 )
174
175 const makeClient = yield* Entity.makeTestClient(StreamEntity, layer)
176 const client = yield* makeClient("entity-1")
177 const results: Array<number> = []
178 yield* client.Watch().pipe(
179 Stream.take(3),
180 Stream.runForEach((n) => Effect.sync(() => results.push(n)))
181 )
182 assert.deepEqual(results, [1, 2, 3])
183 }).pipe(Effect.provide(TestShardingConfig)))
184 })
185})
186
187const TestShardingConfig = ShardingConfig.layer({
188 shardsPerGroup: 300,
189 entityMailboxCapacity: 10,
190 entityTerminationTimeout: 0,
191 entityMessagePollInterval: 5000,
192 sendRetryInterval: 100
193})