1import { DenoClusterSocket } from "@effect/platform-deno"
2import { assert, describe, it } from "@effect/vitest"
3import { BigDecimal, Cause, Deferred, Effect, Exit, Fiber, Layer, Option, PrimaryKey, Schema } from "effect"
4import type { Sharding } from "effect/cluster"
5import {
6 ClusterSchema,
7 Entity,
8 MessageStorage,
9 RunnerAddress,
10 RunnerHealth,
11 RunnerStorage,
12 ShardingConfig,
13 SocketRunner
14} from "effect/cluster"
15import { Rpc, RpcSerialization } from "effect/rpc"
16
17class TestPayload extends Schema.Class<TestPayload>("TestPayload")({
18 id: Schema.String,
19 amount: Schema.BigDecimal
20}) {
21 [PrimaryKey.symbol]() {
22 return this.id
23 }
24}
25
26const TestEntity = Entity
27 .make("TestEntity", [
28 Rpc.make("Process", {
29 payload: TestPayload,
30 success: Schema.Void
31 }).annotate(ClusterSchema.Persisted, true),
32 Rpc.make("ProcessVolatile", {
33 payload: TestPayload,
34 success: Schema.Void
35 }).annotate(ClusterSchema.Persisted, false)
36 ])
37 .annotateRpcs(ClusterSchema.Uninterruptible, true)
38
39const RUNNER_PORT = 50_125
40const SharedStorage = Layer.mergeAll(
41 RunnerStorage.layerMemory,
42 MessageStorage.layerMemory
43).pipe(
44 Layer.provide(ShardingConfig.layerDefaults)
45)
46
47const makeRunnerLayer = (port: number, entities: Layer.Layer<never, never, Sharding.Sharding>) =>
48 entities.pipe(
49 Layer.provideMerge(SocketRunner.layer),
50 Layer.provide(RunnerHealth.layerNoop),
51 Layer.provide(DenoClusterSocket.layerSocketServer),
52 Layer.provide(DenoClusterSocket.layerClientProtocol),
53 Layer.provide(ShardingConfig.layer({
54 runnerAddress: Option.some(RunnerAddress.make("127.0.0.1", port)),
55 entityTerminationTimeout: 0,
56 entityMessagePollInterval: 5000,
57 sendRetryInterval: 100
58 })),
59 Layer.provide(RpcSerialization.layerSchemaBinary())
60 )
61
62const makeClientLayer = (port: number) =>
63 SocketRunner.layerClientOnly.pipe(
64 Layer.provide(DenoClusterSocket.layerClientProtocol),
65 Layer.provide(ShardingConfig.layer({
66 runnerAddress: Option.some(RunnerAddress.make("127.0.0.1", port)),
67 runnerListenAddress: Option.some(RunnerAddress.make("127.0.0.1", port)),
68 entityTerminationTimeout: 0,
69 entityMessagePollInterval: 5000,
70 sendRetryInterval: 100
71 })),
72 Layer.provide(RpcSerialization.layerSchemaBinary())
73 )
74
75const IsolationEntity = Entity
76 .make("IsolationEntity", [
77 Rpc.make("BadReply", {
78 payload: { id: Schema.Number },
79 success: Schema.Int
80 }),
81 Rpc.make("Slow", {
82 success: Schema.String
83 })
84 ])
85 .annotateRpcs(ClusterSchema.Persisted, false)
86
87const ISOLATION_PORT = 50_126
88
89// BigDecimal.normalize creates a circular `normalized` self-reference.
90// When a persisted message is sent with discard: true, the notify path in Runners.makeRpc
91// passes the raw envelope (with circular BigDecimal payload) to the runner,
92// causing RangeError: Maximum call stack size exceeded.
93//
94// Volatile discard should complete after the request is sent, without waiting for the
95// host runner to finish handling it.
96describe("SocketRunner", () => {
97 it.live(
98 "discarded persisted requests serialize circular values and volatile requests do not wait for replies",
99 () =>
100 Effect.gen(function*() {
101 const volatileStarted = yield* Deferred.make<void>()
102 const releaseVolatile = yield* Deferred.make<void>()
103 const TestEntityLayer = TestEntity.toLayer(
104 Effect.succeed({
105 Process: () => Effect.void,
106 ProcessVolatile: () =>
107 Deferred.succeed(volatileStarted, void 0).pipe(
108 Effect.andThen(Deferred.await(releaseVolatile))
109 )
110 })
111 )
112
113 yield* Layer.launch(makeRunnerLayer(RUNNER_PORT, TestEntityLayer)).pipe(Effect.forkScoped)
114
115 yield* Effect.sleep("2 seconds")
116 yield* Effect.gen(function*() {
117 yield* Effect.sleep("2 seconds")
118 const makeClient = yield* TestEntity.client
119 yield* Effect.sleep("3 seconds")
120 const client = makeClient("entity-1")
121
122 const amount = BigDecimal.fromStringUnsafe("123.45")
123 yield* client.Process(
124 TestPayload.make({ id: "req-1", amount }),
125 { discard: true }
126 )
127
128 const volatileFiber = yield* client.ProcessVolatile(
129 TestPayload.make({ id: "req-2", amount }),
130 { discard: true }
131 ).pipe(Effect.forkChild)
132
133 yield* Deferred.await(volatileStarted)
134 yield* Fiber.join(volatileFiber).pipe(
135 Effect.timeoutOrElse({
136 duration: "1 second",
137 orElse: () => Effect.die("volatile discard waited for the entity reply")
138 })
139 )
140 yield* Deferred.succeed(releaseVolatile, void 0)
141 }).pipe(
142 Effect.provide(makeClientLayer(RUNNER_PORT)),
143 Effect.scoped
144 )
145 }).pipe(Effect.provide(SharedStorage)),
146 30_000
147 )
148
149 it.live(
150 "a reply serialization failure fails only its own request",
151 () =>
152 Effect.gen(function*() {
153 const slowStarted = yield* Deferred.make<void>()
154 const IsolationEntityLayer = IsolationEntity.toLayer(
155 Effect.succeed({
156 BadReply: () => Effect.succeed(1.5),
157 Slow: () =>
158 Deferred.succeed(slowStarted, void 0).pipe(
159 Effect.andThen(Effect.as(Effect.sleep("2 seconds"), "done"))
160 )
161 })
162 )
163
164 yield* Layer.launch(makeRunnerLayer(ISOLATION_PORT, IsolationEntityLayer)).pipe(Effect.forkScoped)
165 yield* Effect.sleep("2 seconds")
166
167 yield* Effect.gen(function*() {
168 const makeClient = yield* IsolationEntity.client
169 yield* Effect.sleep("3 seconds")
170
171 const slowFiber = yield* makeClient("slow-entity").Slow().pipe(Effect.forkChild)
172 yield* Deferred.await(slowStarted)
173
174 const badExit = yield* makeClient("bad-entity").BadReply({ id: 1 }).pipe(Effect.exit)
175 assert.isTrue(Exit.isFailure(badExit), "the unencodable reply must fail the request")
176 const failure = Exit.isFailure(badExit) ? String(Cause.squash(badExit.cause)) : ""
177 assert.include(failure, "MalformedMessage", "the caller must receive the real encode error")
178 assert.notInclude(
179 failure,
180 "AlreadyProcessingMessage",
181 "the request must not be re-sent into the entity's dedup guard"
182 )
183
184 const slowExit = yield* Fiber.await(slowFiber)
185 assert.isTrue(
186 Exit.isSuccess(slowExit),
187 "a sibling in-flight request on the same connection must be unaffected"
188 )
189 if (Exit.isSuccess(slowExit)) {
190 assert.strictEqual(slowExit.value, "done")
191 }
192 }).pipe(
193 Effect.provide(makeClientLayer(ISOLATION_PORT)),
194 Effect.scoped
195 )
196 }).pipe(Effect.provide(SharedStorage)),
197 30_000
198 )
199})