1import { assert, describe, it } from "@effect/vitest"
2import { Deferred, Effect, Exit, Fiber, Layer, Queue, Ref, Schema, Scope, Sink, Stdio, Stream } from "effect"
3import { Headers, HttpRouter, HttpServerRequest, HttpServerResponse } from "effect/http"
4import * as NetAddress from "effect/net/NetAddress"
5import { Rpc, RpcGroup, RpcSerialization, RpcServer } from "effect/rpc"
6import * as RpcMessage from "effect/rpc/RpcMessage"
7import { Socket, SocketServer } from "effect/socket"
8
9const producedWithoutReadingFramedBody = Effect.fnUntraced(function*(
10 streamBufferSize?: number | "unbounded"
11) {
12 let produced = 0
13 const Rpcs = RpcGroup.make(Rpc.make("events", {
14 payload: Schema.Struct({}),
15 success: Schema.Number,
16 stream: true
17 }))
18 const Handlers = Rpcs.toLayerHandler(
19 "events",
20 () => Stream.fromEffectRepeat(Effect.sync(() => ++produced)).pipe(Stream.take(10_000))
21 )
22 const Server = RpcServer.layerHttp({
23 group: Rpcs,
24 path: "/rpc",
25 protocol: "http",
26 ...(streamBufferSize === undefined ? {} : { streamBufferSize })
27 }).pipe(
28 Layer.provide(Handlers),
29 Layer.provide(RpcSerialization.layerNdjson)
30 )
31 const { dispose, handler } = HttpRouter.toWebHandler(Server)
32 yield* Effect.addFinalizer(() => Effect.promise(dispose))
33
34 const response = yield* Effect.promise(() =>
35 handler(
36 new Request("http://test/rpc", {
37 method: "POST",
38 body: `{"_tag":"Request","id":1,"tag":"events","payload":{},"headers":[]}\n`
39 })
40 )
41 )
42 yield* Effect.addFinalizer(() => response.body === null ? Effect.void : Effect.promise(() => response.body!.cancel()))
43 for (let i = 0; i < 100; i++) {
44 yield* Effect.yieldNow
45 }
46
47 return produced
48})
49
50describe("RpcServer", () => {
51 it.effect("should drain the response when stdin ends during request startup", () =>
52 Effect.gen(function*() {
53 const started = yield* Deferred.make<void>()
54 const finishRequest = yield* Deferred.make<void>()
55 const output: Array<string> = []
56 const group = RpcGroup.make(Rpc.make("wait", { payload: Schema.Struct({}), success: Schema.String }))
57 const server = yield* Layer.launch(
58 RpcServer.layer(group).pipe(
59 Layer.provide(group.toLayerHandler("wait", () =>
60 Deferred.succeed(started, undefined).pipe(
61 Effect.andThen(Deferred.await(finishRequest)),
62 Effect.as("finished")
63 ))),
64 Layer.provide(RpcServer.layerProtocolStdio),
65 Layer.provide(RpcSerialization.layerNdjson),
66 Layer.provide(Stdio.layerTest({
67 stdin: Stream.make(
68 new TextEncoder().encode(
69 `{"_tag":"Request","id":1,"tag":"wait","payload":{},"headers":[]}\n`
70 )
71 ).pipe(Stream.concat(Stream.fromEffect(Deferred.await(started)).pipe(Stream.drain))),
72 stdout: () => Sink.forEach((data) => Effect.sync(() => output.push(String(data))))
73 }))
74 )
75 ).pipe(Effect.forkScoped)
76 yield* Deferred.await(started)
77 yield* Deferred.succeed(finishRequest, undefined)
78 yield* Fiber.await(server)
79 assert.deepStrictEqual(output.join("").trim().split("\n").map((line) => JSON.parse(line)), [
80 { _tag: "Exit", requestId: 1, exit: { _tag: "Success", value: "finished" } }
81 ])
82 }))
83
84 it.effect("should shut down when stdin ends after the STDIO response is written", () =>
85 Effect.gen(function*() {
86 const responseWritten = yield* Deferred.make<void>()
87 const closeStdin = yield* Deferred.make<void>()
88 const output: Array<string> = []
89 const group = RpcGroup.make(Rpc.make("hello", { payload: Schema.Struct({}), success: Schema.String }))
90 const server = yield* Layer.launch(
91 RpcServer.layer(group).pipe(
92 Layer.provide(group.toLayerHandler("hello", () => Effect.succeed("world"))),
93 Layer.provide(RpcServer.layerProtocolStdio),
94 Layer.provide(RpcSerialization.layerNdjson),
95 Layer.provide(Stdio.layerTest({
96 stdin: Stream.make(
97 new TextEncoder().encode(`{"_tag":"Request","id":1,"tag":"hello","payload":{},"headers":[]}\n`)
98 ).pipe(Stream.concat(Stream.fromEffect(Deferred.await(closeStdin)).pipe(Stream.drain))),
99 stdout: () =>
100 Sink.forEach((data) =>
101 Effect.sync(() => output.push(String(data))).pipe(
102 Effect.andThen(Deferred.succeed(responseWritten, undefined))
103 )
104 )
105 }))
106 )
107 ).pipe(Effect.forkScoped)
108 yield* Deferred.await(responseWritten)
109 assert.deepStrictEqual(output.join("").trim().split("\n").map((line) => JSON.parse(line)), [
110 { _tag: "Exit", requestId: 1, exit: { _tag: "Success", value: "world" } }
111 ])
112 yield* Deferred.succeed(closeStdin, undefined)
113 yield* Fiber.await(server)
114 }))
115
116 it.effect("should wait for a blocked STDIO write before shutting down", () =>
117 Effect.gen(function*() {
118 const writeStarted = yield* Deferred.make<void>()
119 const finishWrite = yield* Deferred.make<void>()
120 const output: Array<string> = []
121 const server = yield* Layer.launch(
122 RpcServer.layer(RpcGroup.make()).pipe(
123 Layer.provide(RpcServer.layerProtocolStdio),
124 Layer.provide(RpcSerialization.layerNdjson),
125 Layer.provide(Stdio.layerTest({
126 stdin: Stream.make(new TextEncoder().encode(`{"_tag":"Ping"}\n`)),
127 stdout: () =>
128 Sink.forEach((data) =>
129 Deferred.succeed(writeStarted, undefined).pipe(
130 Effect.andThen(Deferred.await(finishWrite)),
131 Effect.andThen(Effect.sync(() => output.push(String(data))))
132 )
133 )
134 }))
135 )
136 ).pipe(Effect.forkScoped)
137 yield* Deferred.await(writeStarted)
138 yield* Effect.yieldNow
139 assert.isUndefined(server.pollUnsafe())
140 yield* Deferred.succeed(finishWrite, undefined)
141 yield* Fiber.await(server)
142 assert.deepStrictEqual(output.join("").trim().split("\n").filter(Boolean).map((line) => JSON.parse(line)), [{
143 _tag: "Pong"
144 }])
145 }))
146
147 it.effect("should accept only cancellation of an active request when client input has ended", () =>
148 Effect.gen(function*() {
149 const entered = yield* Deferred.make<void>()
150 const interrupted = yield* Deferred.make<void>()
151 const releaseOther = yield* Deferred.make<void>()
152 const group = RpcGroup.make(Rpc.make("wait", {
153 payload: { block: Schema.Boolean },
154 success: Schema.String
155 }))
156 const messages = yield* Queue.unbounded<RpcMessage.FromServer<RpcGroup.Rpcs<typeof group>>>()
157 const server = yield* RpcServer.makeNoSerialization(group, {
158 onFromServer: (message) => Queue.offer(messages, message).pipe(Effect.asVoid)
159 }).pipe(Effect.provide(group.toLayerHandler("wait", ({ block }) =>
160 block
161 ? Deferred.succeed(entered, void 0).pipe(
162 Effect.andThen(Effect.never),
163 Effect.onInterrupt(() => Deferred.succeed(interrupted, void 0))
164 )
165 : Deferred.await(releaseOther).pipe(Effect.as("kept")))))
166 const request = {
167 _tag: "Request" as const,
168 id: RpcMessage.RequestId("same"),
169 tag: "wait" as const,
170 payload: { block: true },
171 headers: Headers.empty
172 }
173 yield* server.write(1, request)
174 yield* server.write(2, { ...request, payload: { block: false } })
175 yield* Deferred.await(entered)
176 yield* server.write(1, RpcMessage.constEof)
177 yield* server.write(2, RpcMessage.constEof)
178 for (
179 const message of [
180 { ...request, id: RpcMessage.RequestId("new") },
181 { _tag: "Ack" as const, requestId: request.id },
182 RpcMessage.constEof,
183 { _tag: "Interrupt" as const, requestId: RpcMessage.RequestId("unknown"), interruptors: [] }
184 ]
185 ) {
186 assert(Exit.isFailure(yield* Effect.exit(server.write(1, message))))
187 }
188 assert(Exit.isSuccess(
189 yield* Effect.exit(server.write(1, {
190 _tag: "Interrupt",
191 requestId: request.id,
192 interruptors: []
193 }))
194 ))
195 yield* Deferred.await(interrupted)
196 const cancelled = yield* Queue.take(messages)
197 assert.strictEqual(cancelled._tag, "Exit")
198 assert.strictEqual(cancelled.clientId, 1)
199 if (cancelled._tag === "Exit") assert(Exit.isFailure(cancelled.exit))
200 assert.deepStrictEqual(yield* Queue.take(messages), { _tag: "ClientEnd", clientId: 1 })
201 yield* Deferred.succeed(releaseOther, void 0)
202 assert.deepStrictEqual(yield* Queue.take(messages), {
203 _tag: "Exit",
204 clientId: 2,
205 requestId: request.id,
206 exit: Exit.succeed("kept")
207 })
208 assert.deepStrictEqual(yield* Queue.take(messages), { _tag: "ClientEnd", clientId: 2 })
209 }))
210
211 it.effect("should backpressure STDIO sends when the output buffer is full", () =>
212 Effect.gen(function*() {
213 const protocolReady = yield* Deferred.make<RpcServer.Protocol["Service"]>()
214 const firstWriteStarted = yield* Deferred.make<void>()
215 const releaseFirstWrite = yield* Deferred.make<void>()
216 const overflowSendCompleted = yield* Deferred.make<void>()
217 const writes = yield* Ref.make(0)
218 const stdio = Stdio.layerTest({
219 stdin: Stream.never,
220 stdout: () =>
221 Sink.forEach(() =>
222 Ref.getAndUpdate(writes, (count) => count + 1).pipe(
223 Effect.flatMap((index) =>
224 index === 0
225 ? Deferred.succeed(firstWriteStarted, undefined).pipe(
226 Effect.andThen(Deferred.await(releaseFirstWrite))
227 )
228 : Effect.void
229 )
230 )
231 )
232 })
233 yield* Effect.gen(function*() {
234 const protocol = yield* RpcServer.makeProtocolStdio
235 yield* Deferred.succeed(protocolReady, protocol)
236 return yield* protocol.run(() => Effect.void)
237 }).pipe(
238 Effect.provide(stdio),
239 Effect.provide(RpcSerialization.layerJsonRpc()),
240 Effect.forkScoped
241 )
242 const protocol = yield* Deferred.await(protocolReady)
243
244 const first = yield* protocol.send(0, { _tag: "Pong" }).pipe(Effect.forkScoped)
245 yield* Deferred.await(firstWriteStarted)
246 for (let i = 0; i < 8; i++) {
247 yield* protocol.send(0, { _tag: "Pong" })
248 }
249 const overflow = yield* protocol.send(0, { _tag: "Pong" }).pipe(
250 Effect.andThen(Deferred.succeed(overflowSendCompleted, undefined)),
251 Effect.forkScoped
252 )
253 yield* Effect.yieldNow
254 assert.strictEqual(yield* Ref.get(writes), 1)
255 assert.isFalse(yield* Deferred.isDone(overflowSendCompleted))
256
257 yield* Deferred.succeed(releaseFirstWrite, undefined)
258 yield* Fiber.join(first)
259 yield* Fiber.join(overflow)
260 assert.strictEqual(yield* Ref.get(writes), 10)
261 }))
262
263 it.effect("should continue sending to other clients when a backpressured HTTP client disconnects", () =>
264 Effect.gen(function*() {
265 const { httpEffect, protocol } = yield* RpcServer.makeProtocolWithHttpEffect({ streamBufferSize: 1 }).pipe(
266 Effect.provide(RpcSerialization.layerNdjson)
267 )
268 const connected = yield* Queue.unbounded<number>()
269 const interrupted = yield* Queue.unbounded<RpcMessage.FromClientEncoded>()
270 yield* protocol.run((clientId, message) => {
271 if (message._tag === "Request") {
272 return Queue.offer(connected, clientId).pipe(
273 Effect.andThen(protocol.send(clientId, { _tag: "Pong" }))
274 )
275 }
276 return message._tag === "Interrupt" ? Queue.offer(interrupted, message) : Effect.void
277 }).pipe(Effect.forkScoped)
278
279 const scope = yield* Scope.Scope
280 const firstScope = yield* Scope.fork(scope)
281 const request = HttpServerRequest.fromWeb(
282 new Request("http://test/rpc", {
283 method: "POST",
284 body: `{"_tag":"Request","id":1,"tag":"events","payload":{},"headers":[]}\n`
285 })
286 )
287 yield* httpEffect.pipe(
288 Effect.provideService(HttpServerRequest.HttpServerRequest, request),
289 Effect.provideService(Scope.Scope, firstScope)
290 )
291 const firstId = yield* Queue.take(connected)
292 const response = yield* httpEffect.pipe(
293 Effect.provideService(
294 HttpServerRequest.HttpServerRequest,
295 HttpServerRequest.fromWeb(
296 new Request(
297 "http://test/rpc",
298 { method: "POST", body: `{"_tag":"Request","id":2,"tag":"events","payload":{},"headers":[]}\n` }
299 )
300 )
301 )
302 )
303 const secondId = yield* Queue.take(connected)
304 yield* protocol.send(firstId, { _tag: "Pong" })
305 const delivered = yield* Deferred.make<void>()
306 const sender = yield* protocol.send(firstId, { _tag: "Pong" }).pipe(
307 Effect.andThen(protocol.send(secondId, { _tag: "Pong" })),
308 Effect.andThen(Deferred.succeed(delivered, undefined)),
309 Effect.forkScoped
310 )
311 yield* Effect.yieldNow
312 assert.isFalse(yield* Deferred.isDone(delivered))
313
314 yield* Scope.close(firstScope, Exit.void)
315 yield* Effect.yieldNow
316 assert.isTrue(yield* Deferred.isDone(delivered))
317 yield* Fiber.join(sender)
318 assert.deepStrictEqual(yield* Queue.take(interrupted), {
319 _tag: "Interrupt",
320 requestId: RpcMessage.RequestId(1)
321 })
322 yield* protocol.end(secondId)
323 const body = yield* Effect.promise(() => HttpServerResponse.toWeb(response).text())
324 assert.deepStrictEqual(body.trim().split("\n").map((line) => JSON.parse(line)), [
325 { _tag: "Pong" },
326 { _tag: "Pong" }
327 ])
328 }))
329
330 it.effect("applies backpressure to framed HTTP responses by default", () =>
331 Effect.gen(function*() {
332 const produced = yield* producedWithoutReadingFramedBody()
333
334 assert.isAbove(produced, 0)
335 assert.isBelow(produced, 10_000)
336 }))
337
338 it.effect("uses the configured framed HTTP response buffer size", () =>
339 Effect.gen(function*() {
340 const produced = yield* producedWithoutReadingFramedBody(1)
341
342 assert.isAbove(produced, 0)
343 assert.isBelow(produced, 16)
344 }))
345
346 it.effect("allows unbounded framed HTTP response buffering", () =>
347 Effect.gen(function*() {
348 const produced = yield* producedWithoutReadingFramedBody("unbounded")
349
350 assert.strictEqual(produced, 10_000)
351 }))
352
353 it.effect("keeps non-framed HTTP responses unbounded", () =>
354 Effect.gen(function*() {
355 let produced = 0
356 const Rpcs = RpcGroup.make(Rpc.make("events", {
357 payload: Schema.Struct({}),
358 success: Schema.Number,
359 stream: true
360 }))
361 const Handlers = Rpcs.toLayerHandler(
362 "events",
363 () => Stream.fromEffectRepeat(Effect.sync(() => ++produced)).pipe(Stream.take(100))
364 )
365 const Server = RpcServer.layerHttp({
366 group: Rpcs,
367 path: "/rpc",
368 protocol: "http",
369 streamBufferSize: 1
370 }).pipe(
371 Layer.provide(Handlers),
372 Layer.provide(RpcSerialization.layerJson)
373 )
374 const { dispose, handler } = HttpRouter.toWebHandler(Server)
375 yield* Effect.addFinalizer(() => Effect.promise(dispose))
376
377 yield* Effect.promise(() =>
378 handler(
379 new Request("http://test/rpc", {
380 method: "POST",
381 body: `{"_tag":"Request","id":1,"tag":"events","payload":{},"headers":[]}`
382 })
383 )
384 )
385
386 assert.strictEqual(produced, 100)
387 }))
388
389 it.effect("closes a socket when the serialization buffer limit is exceeded", () =>
390 Effect.gen(function*() {
391 const handledChunks: Array<string> = []
392 const writes: Array<Uint8Array | string | Socket.CloseEvent> = []
393 const completed = yield* Deferred.make<void>()
394 let closed = false
395
396 const write = (chunk: Uint8Array | string | Socket.CloseEvent) =>
397 Effect.sync(() => {
398 writes.push(chunk)
399 if (Socket.isCloseEvent(chunk)) {
400 closed = true
401 }
402 })
403 const socket = Socket.make({
404 reader: Effect.sync(() => {
405 const chunks = ["12", "34", "5", "{\"_tag\":\"Ping\"}\n"]
406 let index = 0
407 return {
408 pull: Effect.suspend(() => {
409 if (closed || index >= chunks.length) {
410 return Deferred.succeed(completed, void 0).pipe(
411 Effect.andThen(Effect.fail(
412 new Socket.SocketError({
413 reason: new Socket.SocketCloseError({ code: 1000 })
414 })
415 ))
416 )
417 }
418 const chunk = chunks[index++]
419 handledChunks.push(chunk)
420 return Effect.succeed([chunk] as const)
421 }),
422 upgrade: () =>
423 Effect.fail(
424 new Socket.SocketError({
425 reason: new Socket.SocketUpgradeError({})
426 })
427 )
428 }
429 }),
430 writer: Effect.succeed({
431 write,
432 writeAll: (chunks) => Effect.forEach(chunks, write, { discard: true })
433 })
434 })
435 const socketServer = SocketServer.SocketServer.of({
436 address: NetAddress.inetAddressFromStringUnsafe("127.0.0.1:0"),
437 run: (handler) => handler(socket).pipe(Effect.orDie, Effect.andThen(Effect.never))
438 })
439
440 const protocol = yield* RpcServer.makeProtocolSocketServer.pipe(
441 Effect.provide(Layer.succeed(SocketServer.SocketServer, socketServer)),
442 Effect.provide(Layer.succeed(
443 RpcSerialization.RpcSerialization,
444 RpcSerialization.makeNdjson({ maxBufferSize: 4 })
445 ))
446 )
447 yield* Effect.forkScoped(protocol.run(() => Effect.void))
448 yield* Deferred.await(completed)
449
450 assert.deepStrictEqual(handledChunks, ["12", "34", "5"])
451 assert.strictEqual(writes.length, 1)
452 const closeEvent = writes[0]
453 assert(Socket.isCloseEvent(closeEvent))
454 assert.strictEqual(closeEvent.code, 1009)
455 }))
456})