1import { assert, describe, it } from "@effect/vitest"
2import { Cause, Context, Deferred, Effect, Exit, Layer, Option, Queue, Schema, Stream } from "effect"
3import {
4 ClusterError,
5 ClusterSchema,
6 Entity,
7 EntityAddress,
8 EntityId,
9 EntityType,
10 Envelope,
11 HttpRunner,
12 Message,
13 MessageStorage,
14 type Reply,
15 RunnerAddress,
16 RunnerHealth,
17 Runners,
18 RunnerServer,
19 RunnerStorage,
20 ShardId,
21 Sharding,
22 ShardingConfig,
23 Snowflake
24} from "effect/cluster"
25import { Headers, HttpClient, HttpClientResponse } from "effect/http"
26import { Rpc, RpcClient, RpcGroup, RpcMessage, RpcSerialization, RpcServer, RpcTest } from "effect/rpc"
27import { RpcClientError } from "effect/rpc/RpcClientError"
28import type { FromClientEncoded, FromServerEncoded } from "effect/rpc/RpcMessage"
29import { Socket } from "effect/socket"
30import { TestClock } from "effect/testing"
31
32// An entity whose replies cannot be serialized: the handlers return
33// non-integers for a `Schema.Int` success schema, so `Reply.serialize` fails
34// on encode.
35const BadReplyEntity = Entity.make("BadReplyEntity", [
36 Rpc.make("BadReply", {
37 success: Schema.Int,
38 payload: { id: Schema.Number }
39 }),
40 Rpc.make("BadStream", {
41 success: Schema.Int,
42 payload: { id: Schema.Number },
43 stream: true
44 })
45]).annotateRpcs(ClusterSchema.Persisted, false)
46
47const BadReplyEntityLayer = BadReplyEntity.toLayer({
48 BadReply: () => Effect.succeed(1.5),
49 BadStream: () => Stream.make(2.5)
50})
51
52const TestShardingConfig = ShardingConfig.layer({
53 entityMailboxCapacity: 10,
54 entityTerminationTimeout: 0,
55 entityMessagePollInterval: 5000,
56 sendRetryInterval: 100,
57 refreshAssignmentsInterval: 0
58})
59
60const layerServerProtocol = (codecFor: RpcSerialization.CodecFor) =>
61 Layer.effect(RpcServer.Protocol)(
62 Effect.map(Queue.unbounded<number>(), (disconnects) =>
63 RpcServer.Protocol.of({
64 run: () => Effect.never,
65 disconnects,
66 send: () => Effect.void,
67 end: () => Effect.void,
68 clientIds: Effect.succeed(new Set()),
69 initialMessage: Effect.succeedNone,
70 supportsAck: false,
71 supportsTransferables: false,
72 supportsSpanPropagation: false,
73 supportsNotifications: false,
74 codecFor
75 }))
76 )
77
78const RunnerServerHandlers = RunnerServer.layerHandlers.pipe(
79 Layer.provide(layerServerProtocol(Schema.toCodecJson as RpcSerialization.CodecFor)),
80 Layer.provideMerge(BadReplyEntityLayer),
81 Layer.provideMerge(Sharding.layer),
82 Layer.provideMerge(Snowflake.layerGenerator),
83 Layer.provide(RunnerStorage.layerMemory),
84 Layer.provide(RunnerHealth.layerNoop),
85 Layer.provide(Runners.layerNoop),
86 Layer.provideMerge(MessageStorage.layerMemory),
87 Layer.provide(TestShardingConfig)
88)
89
90describe.concurrent("HttpRunner", () => {
91 const address = RunnerAddress.make("localhost", 8080)
92 const Rpcs = RpcGroup.make(Rpc.make("Ping", { success: Schema.String }))
93
94 it.effect("preserves a leading slash in HTTP client paths", () => {
95 let requestUrl: string | undefined
96 const httpClient = HttpClient.make((request) => {
97 requestUrl = request.url
98 return Effect.succeed(HttpClientResponse.fromWeb(
99 request,
100 new Response("[{\"_tag\":\"Exit\",\"requestId\":\"0\",\"exit\":{\"_tag\":\"Success\",\"value\":\"pong\"}}]")
101 ))
102 })
103 return Effect.gen(function*() {
104 const runnerProtocol = yield* Runners.RpcClientProtocol
105 const protocol = yield* runnerProtocol.make(address)
106 const client = yield* RpcClient.make(Rpcs, {
107 generateRequestId: () => RpcMessage.RequestId("0")
108 }).pipe(Effect.provideService(RpcClient.Protocol, protocol))
109
110 yield* client.Ping()
111
112 assert.strictEqual(requestUrl, "http://localhost:8080/runner/")
113 }).pipe(
114 Effect.provide(HttpRunner.layerClientProtocolHttp({ path: "/runner" })),
115 Effect.provide(RpcSerialization.layerJson),
116 Effect.provideService(HttpClient.HttpClient, httpClient)
117 )
118 })
119
120 it.effect("preserves a leading slash in WebSocket client paths", () =>
121 Effect.gen(function*() {
122 const connected = yield* Deferred.make<void>()
123 let socketUrl: string | undefined
124 const constructor: Socket.WebSocketConstructor["Service"] = (url) => {
125 socketUrl = url
126 return {
127 readyState: 1,
128 addEventListener() {},
129 removeEventListener() {},
130 close() {},
131 send() {}
132 }
133 }
134
135 yield* Effect.gen(function*() {
136 const runnerProtocol = yield* Runners.RpcClientProtocol
137 const protocol = yield* runnerProtocol.make(address)
138 yield* protocol.run(0, () => Effect.void).pipe(Effect.forkScoped)
139 yield* Deferred.await(connected)
140
141 assert.strictEqual(socketUrl, "ws://localhost:8080/runner")
142 }).pipe(
143 Effect.provide(HttpRunner.layerClientProtocolWebsocket({ path: "/runner" })),
144 Effect.provide(RpcSerialization.layerJson),
145 Effect.provideService(Socket.WebSocketConstructor, constructor),
146 Effect.provideService(RpcClient.ConnectionHooks, {
147 onConnect: Deferred.succeed(connected, undefined).pipe(Effect.asVoid),
148 onDisconnect: Effect.void
149 }),
150 Effect.scoped
151 )
152 }))
153})
154
155describe.concurrent("RunnerServer", () => {
156 const makeRequest = (options: {
157 readonly entityId: string
158 readonly tag: string
159 }) =>
160 Effect.gen(function*() {
161 const sharding = yield* Sharding.Sharding
162 const snowflakeGen = yield* Snowflake.Generator
163 const entityId = EntityId.make(options.entityId)
164 const address = EntityAddress.make({
165 shardId: sharding.getShardId(entityId, BadReplyEntity.getShardGroup(entityId)),
166 entityType: EntityType.make("BadReplyEntity"),
167 entityId
168 })
169 return {
170 _tag: "Request",
171 requestId: snowflakeGen.nextUnsafe(),
172 address,
173 tag: options.tag,
174 payload: { id: 1 },
175 headers: Headers.empty
176 } as Envelope.PartialRequest
177 })
178
179 it.effect("a reply that fails to serialize fails only its own request", () =>
180 Effect.gen(function*() {
181 yield* TestClock.adjust(1)
182 const client = yield* RpcTest.makeClient(Runners.Rpcs)
183 const request = yield* makeRequest({ entityId: "bad-1", tag: "BadReply" })
184
185 const exit = yield* Effect.exit(client.Effect({ request, persisted: false }))
186 if (!Exit.isSuccess(exit)) {
187 return assert.fail("Effect rpc must not defect on a reply serialization failure")
188 }
189 const reply = exit.value
190 if (reply._tag !== "WithExit" || reply.exit._tag !== "Failure") {
191 return assert.fail("expected a WithExit reply with a failure exit")
192 }
193 assert.strictEqual(reply.requestId, String(request.requestId))
194 const die = reply.exit.cause.find((entry) => entry._tag === "Die")
195 assert.isDefined(die, "the reply exit must carry the encode failure as a defect")
196 assert.include(
197 JSON.stringify(die),
198 "MalformedMessage",
199 "the defect must identify the encode failure"
200 )
201 }).pipe(Effect.provide(RunnerServerHandlers)))
202
203 it.effect("a stream reply that fails to serialize ends the stream with the defect", () =>
204 Effect.gen(function*() {
205 yield* TestClock.adjust(1)
206 const client = yield* RpcTest.makeClient(Runners.Rpcs)
207 const request = yield* makeRequest({ entityId: "bad-2", tag: "BadStream" })
208
209 const queue = yield* client.Stream({ request, persisted: false }, { asQueue: true })
210 const replies: Array<Reply.Encoded> = []
211 yield* Queue.take(queue).pipe(
212 Effect.flatMap((reply) =>
213 Effect.sync(() => {
214 replies.push(reply)
215 })
216 ),
217 Effect.forever,
218 Effect.catchTag("Done", () => Effect.void)
219 )
220
221 assert.strictEqual(replies.length, 1)
222 const last = replies[0]
223 if (last._tag !== "WithExit" || last.exit._tag !== "Failure") {
224 return assert.fail("expected a terminal WithExit reply with a failure exit")
225 }
226 assert.strictEqual(last.requestId, String(request.requestId))
227 assert.include(
228 JSON.stringify(last.exit.cause),
229 "MalformedMessage",
230 "the defect must identify the encode failure"
231 )
232 }).pipe(Effect.provide(RunnerServerHandlers)))
233})
234
235describe.concurrent("Runners.makeRpc", () => {
236 const runnerAddress = RunnerAddress.make("localhost", 42_000)
237
238 const TestRpc = Rpc.make("TestRpc", {
239 success: Schema.Number,
240 payload: { id: Schema.Number }
241 }).annotate(ClusterSchema.Persisted, false)
242
243 const TestRpcPersisted = Rpc.make("TestRpcPersisted", {
244 success: Schema.Number,
245 payload: { id: Schema.Number }
246 }).annotate(ClusterSchema.Persisted, true)
247
248 type SendRpc = typeof TestRpc | typeof TestRpcPersisted
249
250 const makeOutgoingRequest = (
251 rpc: SendRpc,
252 requestId: Snowflake.Snowflake,
253 respond: (reply: Reply.Reply<SendRpc>) => Effect.Effect<void>
254 ): Message.OutgoingRequest<SendRpc> =>
255 new Message.OutgoingRequest({
256 envelope: Envelope.makeRequest<SendRpc>({
257 requestId,
258 address: EntityAddress.make({
259 shardId: ShardId.make("default", 1),
260 entityType: EntityType.make("TestRpcEntity"),
261 entityId: EntityId.make("1")
262 }),
263 tag: rpc._tag,
264 payload: { id: 1 },
265 headers: Headers.empty
266 }),
267 rpc,
268 context: Context.empty(),
269 lastReceivedReply: Option.none(),
270 respond,
271 annotations: Context.empty()
272 })
273
274 // A hole codec that is observably different from `Schema.toCodecJson`: the
275 // entity payload becomes a JSON string inside the runner envelope.
276 const codecForJsonString =
277 (<S extends Schema.Top>(schema: S) =>
278 Schema.fromJsonString(Schema.toCodecJson(schema as any))) as RpcSerialization.CodecFor
279
280 const layerFakeProtocol = (
281 onRequest: (
282 request: FromClientEncoded,
283 write: (data: FromServerEncoded) => Effect.Effect<void>
284 ) => Effect.Effect<void, RpcClientError>,
285 codecFor: RpcSerialization.CodecFor = Schema.toCodecJson as RpcSerialization.CodecFor
286 ) =>
287 Layer.succeed(Runners.RpcClientProtocol)({
288 codecFor,
289 make: () =>
290 Effect.sync(() => {
291 let write!: (data: FromServerEncoded) => Effect.Effect<void>
292 return RpcClient.Protocol.of({
293 run(_clientId, f) {
294 write = f
295 return Effect.never
296 },
297 send(_clientId, request) {
298 return onRequest(request, write)
299 },
300 supportsAck: true,
301 supportsTransferables: false,
302 codecFor
303 })
304 })
305 })
306
307 const layerRunners = (protocol: Layer.Layer<Runners.RpcClientProtocol>) =>
308 Runners.layerRpc.pipe(
309 Layer.provideMerge(Snowflake.layerGenerator),
310 Layer.provide(protocol),
311 Layer.provideMerge(MessageStorage.layerNoop),
312 Layer.provide(TestShardingConfig)
313 )
314
315 it.effect("uses the client protocol codec for simulated remote serialization", () => {
316 let codecCalls = 0
317 const codecFor = (<S extends Schema.Top>(schema: S) => {
318 codecCalls++
319 return Schema.fromJsonString(Schema.toCodecJson(schema as any))
320 }) as RpcSerialization.CodecFor
321 const codecForJson = Schema.toCodecJson as RpcSerialization.CodecFor
322 return Effect.gen(function*() {
323 const runners = yield* Runners.Runners
324 const snowflake = yield* Snowflake.Generator
325 const message = makeOutgoingRequest(TestRpc, snowflake.nextUnsafe(), () => Effect.void)
326
327 yield* runners.sendLocal({
328 message,
329 simulateRemoteSerialization: true,
330 send: (incoming) =>
331 Effect.sync(() => {
332 assert.strictEqual(incoming._tag, "IncomingRequestLocal")
333 if (incoming._tag === "IncomingRequestLocal") {
334 assert.strictEqual((incoming.envelope.payload as { readonly id: number }).id, 1)
335 }
336 })
337 })
338
339 assert.strictEqual(codecCalls, 2)
340 assert.strictEqual(message.encodedCache?.codecFor, codecFor)
341 const encodedJson = yield* Message.serialize(message, codecForJson)
342 assert.strictEqual(encodedJson._tag, "Request")
343 if (encodedJson._tag === "Request") {
344 assert.deepStrictEqual(encodedJson.payload, { id: 1 })
345 }
346 assert.strictEqual(message.encodedCache?.codecFor, codecForJson)
347 }).pipe(
348 Effect.provide(layerRunners(layerFakeProtocol(() => Effect.void, codecFor)))
349 )
350 })
351
352 const respondWithDefect = (request: FromClientEncoded, write: (data: FromServerEncoded) => Effect.Effect<void>) =>
353 request._tag === "Request" ? write({ _tag: "Defect", defect: "boom" }) : Effect.void
354
355 const failTransport = () =>
356 Effect.fail(
357 new RpcClientError({
358 reason: new Socket.SocketCloseError({ code: 1006 })
359 })
360 )
361
362 it.effect("a server-delivered defect resolves the request instead of RunnerUnavailable", () =>
363 Effect.gen(function*() {
364 const runners = yield* Runners.Runners
365 const snowflakeGen = yield* Snowflake.Generator
366 const replies: Array<Reply.Reply<SendRpc>> = []
367 const message = makeOutgoingRequest(TestRpc, snowflakeGen.nextUnsafe(), (reply) =>
368 Effect.sync(() => {
369 replies.push(reply)
370 }))
371
372 const exit = yield* Effect.exit(runners.send({ address: runnerAddress, message }))
373 assert.isTrue(Exit.isSuccess(exit), "send must not fail with RunnerUnavailable for a delivered defect")
374 assert.strictEqual(replies.length, 1)
375 const reply = replies[0]
376 if (reply._tag !== "WithExit" || !Exit.isFailure(reply.exit)) {
377 return assert.fail("expected a WithExit reply with a failure exit")
378 }
379 assert.include(
380 String(Cause.squash(reply.exit.cause)),
381 "boom",
382 "the reply must carry the server defect"
383 )
384 }).pipe(Effect.provide(layerRunners(layerFakeProtocol(respondWithDefect)))))
385
386 it.effect("transport failures still map to RunnerUnavailable", () =>
387 Effect.gen(function*() {
388 const runners = yield* Runners.Runners
389 const snowflakeGen = yield* Snowflake.Generator
390 const message = makeOutgoingRequest(TestRpc, snowflakeGen.nextUnsafe(), () => Effect.void)
391
392 const exit = yield* Effect.exit(runners.send({ address: runnerAddress, message }))
393 if (!Exit.isFailure(exit)) {
394 return assert.fail("send must fail for a transport failure")
395 }
396 assert.instanceOf(Cause.squash(exit.cause), ClusterError.RunnerUnavailable)
397 }).pipe(Effect.provide(layerRunners(layerFakeProtocol(failTransport)))))
398
399 it.effect("volatile notification transport failures map to RunnerUnavailable", () =>
400 Effect.gen(function*() {
401 const runners = yield* Runners.Runners
402 const snowflakeGen = yield* Snowflake.Generator
403 const message = makeOutgoingRequest(TestRpc, snowflakeGen.nextUnsafe(), () => Effect.void)
404
405 const exit = yield* Effect.exit(runners.notify({
406 address: Option.some(runnerAddress),
407 message,
408 discard: true
409 }))
410 if (!Exit.isFailure(exit)) {
411 return assert.fail("volatile notification must fail when delivery fails")
412 }
413 assert.instanceOf(Cause.squash(exit.cause), ClusterError.RunnerUnavailable)
414 }).pipe(Effect.provide(layerRunners(layerFakeProtocol(failTransport)))))
415
416 it.effect("persisted notification transport failures are ignored", () =>
417 Effect.gen(function*() {
418 const runners = yield* Runners.Runners
419 const snowflakeGen = yield* Snowflake.Generator
420 const message = makeOutgoingRequest(TestRpcPersisted, snowflakeGen.nextUnsafe(), () => Effect.void)
421
422 yield* runners.notify({
423 address: Option.some(runnerAddress),
424 message,
425 discard: true
426 })
427 }).pipe(Effect.provide(layerRunners(layerFakeProtocol(failTransport)))))
428
429 it.effect("a delivered defect for a persisted request maps to RunnerUnavailable for storage recovery", () =>
430 Effect.gen(function*() {
431 const runners = yield* Runners.Runners
432 const snowflakeGen = yield* Snowflake.Generator
433 const message = makeOutgoingRequest(TestRpcPersisted, snowflakeGen.nextUnsafe(), () => Effect.void)
434
435 const exit = yield* Effect.exit(runners.send({ address: runnerAddress, message }))
436 if (!Exit.isFailure(exit)) {
437 return assert.fail("send must fail for a delivered defect on a persisted request")
438 }
439 assert.instanceOf(Cause.squash(exit.cause), ClusterError.RunnerUnavailable)
440 }).pipe(Effect.provide(layerRunners(layerFakeProtocol(respondWithDefect)))))
441
442 it.effect("encodes the entity payload with the protocol codec and leaves the hole untouched", () => {
443 const sent: Array<FromClientEncoded> = []
444 const protocol = layerFakeProtocol(
445 (request, write) => {
446 sent.push(request)
447 // the defect hole is filled by the same codec
448 return request._tag === "Request"
449 ? write({ _tag: "Defect", defect: JSON.stringify("boom") })
450 : Effect.void
451 },
452 codecForJsonString
453 )
454 return Effect.gen(function*() {
455 const runners = yield* Runners.Runners
456 const snowflakeGen = yield* Snowflake.Generator
457 const message = makeOutgoingRequest(TestRpc, snowflakeGen.nextUnsafe(), () => Effect.void)
458
459 yield* runners.send({ address: runnerAddress, message })
460
461 assert.strictEqual(sent.length, 1)
462 const request = sent[0]
463 if (request._tag !== "Request") {
464 return assert.fail("expected a runner Request")
465 }
466 // the outer runner payload is filled by the protocol codec
467 assert.strictEqual(typeof request.payload, "string")
468 const outer = JSON.parse(request.payload as string)
469 // the entity payload was encoded with the same codec and carried through
470 // the opaque hole without being re-encoded
471 assert.strictEqual(outer.request.payload, JSON.stringify({ id: 1 }))
472 }).pipe(Effect.provide(layerRunners(protocol)))
473 })
474})