1import { NodeClusterSocket, NodeSocketServer } from "@effect/platform-node"
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"
16import * as SocketServer from "effect/socket/SocketServer"
17
18const HOST = "127.0.0.1"
19
20class TestPayload extends Schema.Class<TestPayload>("TestPayload")({
21 id: Schema.String,
22 amount: Schema.BigDecimal
23}) {
24 [PrimaryKey.symbol]() {
25 return this.id
26 }
27}
28
29const TestEntity = Entity
30 .make("TestEntity", [
31 Rpc.make("Process", {
32 payload: TestPayload,
33 success: Schema.Void
34 }).annotate(ClusterSchema.Persisted, true),
35 Rpc.make("ProcessVolatile", {
36 payload: TestPayload,
37 success: Schema.Void
38 }).annotate(ClusterSchema.Persisted, false)
39 ])
40 .annotateRpcs(ClusterSchema.Uninterruptible, true)
41
42// Build shared storage instances once, so runner and client see the same state.
43// MessageStorage.layerMemory requires ShardingConfig, so we provide a minimal one.
44const SharedStorage = Layer.mergeAll(
45 RunnerStorage.layerMemory,
46 MessageStorage.layerMemory
47).pipe(
48 Layer.provide(ShardingConfig.layerDefaults)
49)
50
51const startRunner = Effect.fnUntraced(function*(
52 entities: Layer.Layer<never, never, Sharding.Sharding>,
53 serialization: Layer.Layer<RpcSerialization.RpcSerialization> = RpcSerialization.layerSchemaBinary()
54) {
55 const socketServer = yield* NodeSocketServer.make({ host: HOST, port: 0 })
56 if (socketServer.address._tag !== "InetAddressV4") {
57 return yield* Effect.die("Expected a TCP socket server")
58 }
59 const port = socketServer.address.port
60 yield* entities.pipe(
61 Layer.provideMerge(SocketRunner.layer),
62 Layer.provide(RunnerHealth.layerNoop),
63 Layer.provide(Layer.succeed(SocketServer.SocketServer, socketServer)),
64 Layer.provide(NodeClusterSocket.layerClientProtocol),
65 Layer.provide(ShardingConfig.layer({
66 runnerAddress: Option.some(RunnerAddress.make(HOST, port)),
67 entityTerminationTimeout: 0,
68 entityMessagePollInterval: 5000,
69 sendRetryInterval: 100
70 })),
71 Layer.provide(serialization),
72 Layer.build
73 )
74 return port
75})
76
77const makeClientLayer = (port: number) =>
78 SocketRunner.layerClientOnly.pipe(
79 Layer.provide(NodeClusterSocket.layerClientProtocol),
80 Layer.provide(ShardingConfig.layer({
81 runnerAddress: Option.some(RunnerAddress.make(HOST, port)),
82 runnerListenAddress: Option.some(RunnerAddress.make(HOST, port)),
83 entityTerminationTimeout: 0,
84 entityMessagePollInterval: 5000,
85 sendRetryInterval: 100
86 })),
87 Layer.provide(RpcSerialization.layerSchemaBinary())
88 )
89
90const makeConfiguredClientLayer = (
91 port: number,
92 serialization?: "binary" | "ndjson",
93 serializationMaxBufferSize?: number | "unbounded"
94) =>
95 NodeClusterSocket.layer({
96 clientOnly: true,
97 storage: "byo",
98 ...(serialization === undefined ? undefined : { serialization }),
99 ...(serializationMaxBufferSize === undefined ? undefined : { serializationMaxBufferSize }),
100 shardingConfig: {
101 runnerAddress: Option.some(RunnerAddress.make(HOST, port)),
102 runnerListenAddress: Option.some(RunnerAddress.make(HOST, port)),
103 entityTerminationTimeout: 0,
104 entityMessagePollInterval: 5000,
105 sendRetryInterval: 100
106 }
107 })
108
109const SerializationEntity = Entity
110 .make("SerializationEntity", [
111 Rpc.make("Ping", {
112 success: Schema.String
113 })
114 ])
115 .annotateRpcs(ClusterSchema.Persisted, false)
116
117const SerializationEntityLayer = SerializationEntity.toLayer(
118 Effect.succeed({ Ping: () => Effect.succeed("pong") })
119)
120
121const assertSerialization = (
122 serverSerialization: Layer.Layer<RpcSerialization.RpcSerialization>,
123 clientSerialization?: "binary" | "ndjson"
124) =>
125 Effect.gen(function*() {
126 const port = yield* startRunner(SerializationEntityLayer, serverSerialization)
127
128 const result = yield* Effect.gen(function*() {
129 const makeClient = yield* SerializationEntity.client
130 return yield* makeClient("serialization-entity").Ping()
131 }).pipe(
132 Effect.provide(makeConfiguredClientLayer(port, clientSerialization)),
133 Effect.scoped
134 )
135
136 assert.strictEqual(result, "pong")
137 }).pipe(Effect.provide(SharedStorage))
138
139// An entity whose reply cannot be serialized: the handler returns a
140// non-integer for a `Schema.Int` success schema, so `Reply.serialize` fails
141// on the host runner when encoding the reply for the wire.
142const IsolationEntity = Entity
143 .make("IsolationEntity", [
144 Rpc.make("BadReply", {
145 payload: { id: Schema.Number },
146 success: Schema.Int
147 }),
148 Rpc.make("Slow", {
149 success: Schema.String
150 })
151 ])
152 .annotateRpcs(ClusterSchema.Persisted, false)
153
154// BigDecimal.normalize creates a circular `normalized` self-reference.
155// When a persisted message is sent with discard: true, the notify path in Runners.makeRpc
156// passes the raw envelope (with circular BigDecimal payload) to the runner,
157// causing RangeError: Maximum call stack size exceeded.
158//
159// Volatile discard should complete after the request is sent, without waiting for the
160// host runner to finish handling it.
161describe("SocketRunner", () => {
162 it.live(
163 "uses SchemaBinary by default for TCP connections",
164 () => assertSerialization(RpcSerialization.layerSchemaBinary()),
165 15_000
166 )
167
168 it.live(
169 "preserves an explicit binary selection",
170 () => assertSerialization(RpcSerialization.layerSchemaBinary(), "binary"),
171 15_000
172 )
173
174 it.live(
175 "applies serializationMaxBufferSize to the default SchemaBinary parser",
176 () =>
177 Effect.gen(function*() {
178 const port = yield* startRunner(SerializationEntityLayer)
179
180 yield* Effect.gen(function*() {
181 const makeClient = yield* SerializationEntity.client
182 assert.strictEqual(yield* makeClient("serialization-limit-entity").Ping(), "pong")
183 }).pipe(
184 Effect.provide(makeConfiguredClientLayer(port)),
185 Effect.scoped
186 )
187
188 const exit = yield* Effect.gen(function*() {
189 const makeClient = yield* SerializationEntity.client
190 return yield* makeClient("serialization-limit-entity").Ping().pipe(Effect.timeout("1 second"))
191 }).pipe(
192 Effect.provide(makeConfiguredClientLayer(port, undefined, 1)),
193 Effect.scoped,
194 Effect.exit
195 )
196
197 assert.isTrue(Exit.isFailure(exit), "the configured frame limit must reject the response")
198 }).pipe(Effect.provide(SharedStorage)),
199 15_000
200 )
201
202 it.live(
203 "discarded persisted requests serialize circular values and volatile requests do not wait for replies",
204 () =>
205 Effect.gen(function*() {
206 const volatileStarted = yield* Deferred.make<void>()
207 const releaseVolatile = yield* Deferred.make<void>()
208 const TestEntityLayer = TestEntity.toLayer(
209 Effect.succeed({
210 Process: () => Effect.void,
211 ProcessVolatile: () =>
212 Deferred.succeed(volatileStarted, void 0).pipe(
213 Effect.andThen(Deferred.await(releaseVolatile))
214 )
215 })
216 )
217
218 // Start the runner (with socket server and entity handler)
219 const port = yield* startRunner(TestEntityLayer)
220 yield* Effect.log("Before starting the client")
221
222 // Send a message from the client with discard: true.
223 // The BigDecimal is normalized to trigger the circular `normalized` self-reference.
224 yield* Effect.gen(function*() {
225 yield* Effect.log("Starting the client")
226 const makeClient = yield* TestEntity.client
227 const client = makeClient("entity-1")
228
229 const amount = BigDecimal.fromStringUnsafe("123.45")
230
231 yield* client.Process(
232 TestPayload.make({ id: "req-1", amount }),
233 { discard: true }
234 )
235
236 const volatileFiber = yield* client.ProcessVolatile(
237 TestPayload.make({ id: "req-2", amount }),
238 { discard: true }
239 ).pipe(Effect.forkChild)
240
241 yield* Deferred.await(volatileStarted).pipe(
242 Effect.timeoutOrElse({
243 duration: "5 seconds",
244 orElse: () => Effect.die("the volatile request never reached its handler")
245 })
246 )
247 yield* Fiber.join(volatileFiber).pipe(
248 Effect.timeoutOrElse({
249 duration: "1 second",
250 orElse: () => Effect.die("volatile discard waited for the entity reply")
251 })
252 )
253 yield* Deferred.succeed(releaseVolatile, void 0)
254 }).pipe(
255 Effect.provide(makeClientLayer(port)),
256 Effect.scoped
257 )
258 }).pipe(Effect.provide(
259 SharedStorage
260 )),
261 30_000
262 )
263
264 it.live(
265 "a reply serialization failure fails only its own request",
266 () =>
267 Effect.gen(function*() {
268 const slowStarted = yield* Deferred.make<void>()
269 const releaseSlow = yield* Deferred.make<void>()
270 const IsolationEntityLayer = IsolationEntity.toLayer(
271 Effect.succeed({
272 BadReply: () => Effect.succeed(1.5),
273 Slow: () =>
274 Deferred.succeed(slowStarted, void 0).pipe(
275 Effect.andThen(Deferred.await(releaseSlow)),
276 Effect.as("done")
277 )
278 })
279 )
280 const port = yield* startRunner(IsolationEntityLayer)
281
282 yield* Effect.gen(function*() {
283 const makeClient = yield* IsolationEntity.client
284
285 // a sibling request in flight on the same runner-to-runner connection
286 const slowFiber = yield* makeClient("slow-entity").Slow().pipe(Effect.forkChild)
287 yield* Deferred.await(slowStarted).pipe(
288 Effect.timeoutOrElse({
289 duration: "5 seconds",
290 orElse: () => Effect.die("the Slow request never reached its handler")
291 })
292 )
293
294 const badExit = yield* makeClient("bad-entity").BadReply({ id: 1 }).pipe(
295 Effect.exit,
296 Effect.ensuring(Deferred.succeed(releaseSlow, void 0))
297 )
298 assert.isTrue(Exit.isFailure(badExit), "the unencodable reply must fail the request")
299 const failure = Exit.isFailure(badExit) ? String(Cause.squash(badExit.cause)) : ""
300 assert.include(failure, "MalformedMessage", "the caller must receive the real encode error")
301 assert.notInclude(
302 failure,
303 "AlreadyProcessingMessage",
304 "the request must not be re-sent into the entity's dedup guard"
305 )
306
307 const slowExit = yield* Fiber.await(slowFiber)
308 assert.isTrue(
309 Exit.isSuccess(slowExit),
310 "a sibling in-flight request on the same connection must be unaffected"
311 )
312 if (Exit.isSuccess(slowExit)) {
313 assert.strictEqual(slowExit.value, "done")
314 }
315 }).pipe(
316 Effect.provide(makeClientLayer(port)),
317 Effect.scoped
318 )
319 }).pipe(Effect.provide(
320 SharedStorage
321 )),
322 30_000
323 )
324})