| 1 | import { assert, describe, it } from "@effect/vitest" |
| 2 | import { Cause, Deferred, Effect, Exit, Fiber, Layer, Queue, Schedule, Schema, Stream } from "effect" |
| 3 | import * as HttpClient from "effect/http/HttpClient" |
| 4 | import * as HttpClientResponse from "effect/http/HttpClientResponse" |
| 5 | import { Rpc, RpcClient, RpcGroup, RpcMessage, RpcSchema, RpcSerialization } from "effect/rpc" |
| 6 | import { RpcClientDefect, RpcClientError } from "effect/rpc/RpcClientError" |
| 7 | import * as Socket from "effect/socket/Socket" |
| 8 | import { TestClock } from "effect/testing" |
| 9 | import * as Worker from "effect/workers/Worker" |
| 10 | import { WorkerError, WorkerReceiveError } from "effect/workers/WorkerError" |
| 11 | import { vi } from "vitest" |
| 12 | import type * as RpcClientErrorModule from "../../src/rpc/RpcClientError.ts" |
| 13 | |
| 14 | const TestGroup = RpcGroup.make( |
| 15 | Rpc.make("Ping", { success: Schema.String }), |
| 16 | Rpc.make("Events", { success: RpcSchema.Stream(Schema.String, Schema.Never) }) |
| 17 | ) |
| 18 | |
| 19 | const ChunkGroup = RpcGroup.make( |
| 20 | Rpc.make("Bad", { success: RpcSchema.Stream(Schema.Number, Schema.Never) }), |
| 21 | Rpc.make("Good", { success: RpcSchema.Stream(Schema.String, Schema.Never) }), |
| 22 | Rpc.make("Unary", { success: Schema.String }) |
| 23 | ) |
| 24 | |
| 25 | const makeChunkProtocol = (beforeInterrupt: Effect.Effect<void> = Effect.void) => |
| 26 | Effect.gen(function*() { |
| 27 | const received = yield* Deferred.make<Parameters<RpcClient.Protocol["Service"]["run"]>[1]>() |
| 28 | const sent = yield* Queue.unbounded<Parameters<RpcClient.Protocol["Service"]["send"]>[1]>() |
| 29 | const protocol = RpcClient.Protocol.of({ |
| 30 | codecFor: RpcSerialization.json.codecFor, |
| 31 | supportsAck: false, |
| 32 | supportsTransferables: false, |
| 33 | run: (_clientId, handle) => Deferred.succeed(received, handle).pipe(Effect.andThen(Effect.never)), |
| 34 | send: (_clientId, message) => |
| 35 | Effect.andThen( |
| 36 | message._tag === "Interrupt" ? beforeInterrupt : Effect.void, |
| 37 | Effect.asVoid(Queue.offer(sent, message)) |
| 38 | ) |
| 39 | }) |
| 40 | const client = yield* RpcClient.make(ChunkGroup).pipe(Effect.provideService(RpcClient.Protocol, protocol)) |
| 41 | const handle = yield* Deferred.await(received) |
| 42 | return { client, handle, sent } |
| 43 | }) |
| 44 | |
| 45 | const takeRequestId = (message: Parameters<RpcClient.Protocol["Service"]["send"]>[1]) => { |
| 46 | assert(message._tag === "Request") |
| 47 | return message.id |
| 48 | } |
| 49 | |
| 50 | const makeHttpClient = (body: string): HttpClient.HttpClient => |
| 51 | HttpClient.make((request) => |
| 52 | Effect.succeed( |
| 53 | HttpClientResponse.fromWeb( |
| 54 | request, |
| 55 | new Response(body, { status: 200 }) |
| 56 | ) |
| 57 | ) |
| 58 | ) |
| 59 | |
| 60 | const makeProtocolLayerWithClient = ( |
| 61 | serializationLayer: Layer.Layer<RpcSerialization.RpcSerialization>, |
| 62 | client: HttpClient.HttpClient |
| 63 | ) => |
| 64 | RpcClient.layerProtocolHttp({ url: "http://localhost/rpc" }).pipe( |
| 65 | Layer.provideMerge(serializationLayer), |
| 66 | Layer.provideMerge(Layer.succeed(HttpClient.HttpClient, client)) |
| 67 | ) |
| 68 | |
| 69 | const makeProtocolLayer = ( |
| 70 | serializationLayer: Layer.Layer<RpcSerialization.RpcSerialization>, |
| 71 | body: string |
| 72 | ) => makeProtocolLayerWithClient(serializationLayer, makeHttpClient(body)) |
| 73 | |
| 74 | const assertEmptyResponseFailsRequest = ( |
| 75 | serializationLayer: Layer.Layer<RpcSerialization.RpcSerialization>, |
| 76 | body: string |
| 77 | ) => |
| 78 | Effect.gen(function*() { |
| 79 | const client = yield* RpcClient.make(TestGroup).pipe( |
| 80 | Effect.provide(makeProtocolLayer(serializationLayer, body)) |
| 81 | ) |
| 82 | |
| 83 | const cause = yield* client.Ping().pipe( |
| 84 | Effect.timeout("1 second"), |
| 85 | Effect.sandbox, |
| 86 | Effect.flip |
| 87 | ) |
| 88 | |
| 89 | const error = Cause.squash(cause) |
| 90 | assert.instanceOf(error, RpcClientError) |
| 91 | assert.strictEqual(error.reason._tag, "RpcClientDefect") |
| 92 | assert.strictEqual(error.reason.message, "Received empty HTTP response from RPC server") |
| 93 | }) |
| 94 | |
| 95 | describe("RpcClient", () => { |
| 96 | it.effect("does not fail a new request started synchronously during protocol error delivery", () => |
| 97 | Effect.gen(function*() { |
| 98 | const { client, handle, sent } = yield* makeChunkProtocol() |
| 99 | const error = new RpcClientError({ |
| 100 | reason: new RpcClientDefect({ message: "connection dropped", cause: undefined }) |
| 101 | }) |
| 102 | const caller = yield* client.Unary().pipe( |
| 103 | Effect.catch((received) => { |
| 104 | assert.strictEqual(received, error) |
| 105 | return client.Unary() |
| 106 | }), |
| 107 | Effect.exit, |
| 108 | Effect.forkChild |
| 109 | ) |
| 110 | takeRequestId(yield* Queue.take(sent)) |
| 111 | // Let the first call suspend before delivering the error. |
| 112 | yield* Effect.yieldNow |
| 113 | yield* handle({ _tag: "ClientProtocolError", error }) |
| 114 | |
| 115 | const retryRequestId = takeRequestId(yield* Queue.take(sent)) |
| 116 | yield* handle({ _tag: "Exit", requestId: retryRequestId, exit: { _tag: "Success", value: "ok" } }) |
| 117 | assert.deepStrictEqual(yield* Fiber.join(caller), Exit.succeed("ok")) |
| 118 | })) |
| 119 | |
| 120 | it.effect("isolates a malformed stream chunk and interrupts its server request", () => |
| 121 | Effect.gen(function*() { |
| 122 | const { client, handle, sent } = yield* makeChunkProtocol() |
| 123 | const bad = yield* client.Bad().pipe(Stream.runDrain, Effect.exit, Effect.forkChild) |
| 124 | const badRequestId = takeRequestId(yield* Queue.take(sent)) |
| 125 | |
| 126 | const good = yield* client.Good().pipe(Stream.runCollect, Effect.forkChild) |
| 127 | const goodRequestId = takeRequestId(yield* Queue.take(sent)) |
| 128 | |
| 129 | const unary = yield* client.Unary().pipe(Effect.forkChild) |
| 130 | const unaryRequestId = takeRequestId(yield* Queue.take(sent)) |
| 131 | |
| 132 | // A decode defect must not escape into the shared protocol receive loop. |
| 133 | yield* handle({ _tag: "Chunk", requestId: badRequestId, values: ["not a number"] }) |
| 134 | const badExit = yield* Fiber.join(bad) |
| 135 | assert(Exit.isFailure(badExit)) |
| 136 | assert.isFalse(Cause.hasInterruptsOnly(badExit.cause)) |
| 137 | assert.deepStrictEqual(yield* Queue.take(sent), { _tag: "Interrupt", requestId: badRequestId }) |
| 138 | |
| 139 | yield* handle({ _tag: "Chunk", requestId: goodRequestId, values: ["alive"] }) |
| 140 | yield* handle({ _tag: "Exit", requestId: goodRequestId, exit: { _tag: "Success", value: null } }) |
| 141 | yield* handle({ _tag: "Exit", requestId: unaryRequestId, exit: { _tag: "Success", value: "ok" } }) |
| 142 | assert.deepStrictEqual(Array.from(yield* Fiber.join(good)), ["alive"]) |
| 143 | assert.strictEqual(yield* Fiber.join(unary), "ok") |
| 144 | })) |
| 145 | |
| 146 | it.effect("preserves cancellation and the server Interrupt during chunk failure cleanup", () => |
| 147 | Effect.gen(function*() { |
| 148 | const sendingInterrupt = yield* Deferred.make<void>() |
| 149 | const releaseInterrupt = yield* Deferred.make<void>() |
| 150 | const { client, handle, sent } = yield* makeChunkProtocol( |
| 151 | Deferred.succeed(sendingInterrupt, void 0).pipe(Effect.andThen(Deferred.await(releaseInterrupt))) |
| 152 | ) |
| 153 | const reader = yield* client.Bad().pipe(Stream.runDrain, Effect.forkChild) |
| 154 | const requestId = takeRequestId(yield* Queue.take(sent)) |
| 155 | let observedInterrupt = false |
| 156 | const receiver = yield* handle({ _tag: "Chunk", requestId, values: ["not a number"] }).pipe( |
| 157 | Effect.onInterrupt(() => Effect.sync(() => void (observedInterrupt = true))), |
| 158 | Effect.forkChild |
| 159 | ) |
| 160 | yield* Deferred.await(sendingInterrupt) |
| 161 | assert(Exit.isFailure(yield* Fiber.await(reader))) |
| 162 | |
| 163 | // Request cancellation while the Interrupt send is suspended, without waiting for it. |
| 164 | yield* Effect.withFiber((fiber) => Effect.sync(() => receiver.interruptUnsafe(fiber.id))) |
| 165 | yield* Deferred.succeed(releaseInterrupt, void 0) |
| 166 | const receiveExit = yield* Fiber.await(receiver) |
| 167 | assert.deepStrictEqual(yield* Queue.clear(sent), [{ _tag: "Interrupt", requestId }]) |
| 168 | // The cancelled handler must end interrupted, not with the decode defect it already handled. |
| 169 | assert(Exit.isFailure(receiveExit) && Cause.hasInterruptsOnly(receiveExit.cause)) |
| 170 | assert.isTrue(observedInterrupt) |
| 171 | })) |
| 172 | |
| 173 | it.effect("releases a chunk blocked on a full buffer when its stream consumer is interrupted", () => |
| 174 | Effect.gen(function*() { |
| 175 | const consuming = yield* Deferred.make<void>() |
| 176 | const { client, handle, sent } = yield* makeChunkProtocol() |
| 177 | const reader = yield* client.Good(undefined, { streamBufferSize: 1 }).pipe( |
| 178 | Stream.runForEach(() => Deferred.succeed(consuming, void 0).pipe(Effect.andThen(Effect.never))), |
| 179 | Effect.forkChild |
| 180 | ) |
| 181 | const requestId = takeRequestId(yield* Queue.take(sent)) |
| 182 | const receiver = yield* handle({ _tag: "Chunk", requestId, values: ["a", "b", "c"] }).pipe(Effect.forkChild) |
| 183 | yield* Deferred.await(consuming) |
| 184 | assert.isUndefined(receiver.pollUnsafe()) |
| 185 | yield* Fiber.interrupt(reader) |
| 186 | |
| 187 | assert(Exit.isSuccess(yield* Fiber.await(receiver))) |
| 188 | assert.deepStrictEqual(yield* Queue.take(sent), { _tag: "Interrupt", requestId }) |
| 189 | })) |
| 190 | |
| 191 | for (const consumer of ["queue", "stream"] as const) { |
| 192 | it.effect(`releases the ${consumer} consumer when the request write is interrupted`, () => |
| 193 | Effect.gen(function*() { |
| 194 | const writing = yield* Deferred.make<Fiber.Fiber<unknown, unknown>>() |
| 195 | const { client } = yield* RpcClient.makeNoSerialization(TestGroup, { |
| 196 | onFromClient: ({ message }) => |
| 197 | message._tag === "Request" |
| 198 | ? Effect.withFiber((fiber) => Deferred.succeed(writing, fiber).pipe(Effect.andThen(Effect.never))) |
| 199 | : Effect.void |
| 200 | }) |
| 201 | const reader = yield* (consumer === "queue" |
| 202 | ? client.Events(undefined, { asQueue: true }).pipe(Effect.flatMap(Queue.take), Effect.asVoid) |
| 203 | : Stream.runDrain(client.Events())).pipe(Effect.forkChild) |
| 204 | const writer = yield* Deferred.await(writing) |
| 205 | yield* Fiber.interrupt(writer) |
| 206 | const readExit = yield* Fiber.await(reader) |
| 207 | assert(Exit.isFailure(readExit) && Cause.hasInterruptsOnly(readExit.cause)) |
| 208 | })) |
| 209 | } |
| 210 | |
| 211 | it.effect("releases a worker pool slot when the worker run fails", () => |
| 212 | Effect.gen(function*() { |
| 213 | const runFailure = yield* Deferred.make<never, WorkerError>() |
| 214 | const firstRequestSent = yield* Deferred.make<void>() |
| 215 | const secondRequestSent = yield* Deferred.make<void>() |
| 216 | const protocolErrorReceived = yield* Deferred.make<void>() |
| 217 | const sentRequestIds: Array<string | number> = [] |
| 218 | let runCount = 0 |
| 219 | const backing: Worker.Worker<any, any> = { |
| 220 | send(message) { |
| 221 | return Effect.sync(() => { |
| 222 | if (message._tag !== "Request") return |
| 223 | sentRequestIds.push(message.id) |
| 224 | }).pipe( |
| 225 | Effect.andThen( |
| 226 | message._tag === "Request" && message.id === 1 |
| 227 | ? Deferred.succeed(firstRequestSent, void 0) |
| 228 | : message._tag === "Request" && message.id === 2 |
| 229 | ? Deferred.succeed(secondRequestSent, void 0) |
| 230 | : Effect.void |
| 231 | ) |
| 232 | ) |
| 233 | }, |
| 234 | run() { |
| 235 | return runCount++ === 0 ? Deferred.await(runFailure) : Effect.never |
| 236 | } |
| 237 | } |
| 238 | const workerPlatform = Worker.WorkerPlatform.of({ |
| 239 | spawn: () => Effect.succeed(backing) |
| 240 | }) |
| 241 | const protocol = yield* RpcClient.makeProtocolWorker({ size: 1, concurrency: 1 }).pipe( |
| 242 | Effect.provideService(Worker.WorkerPlatform, workerPlatform), |
| 243 | Effect.provideService(Worker.Spawner, (() => undefined) as Worker.SpawnerFn) |
| 244 | ) |
| 245 | yield* protocol.run(0, (response) => |
| 246 | response._tag === "ClientProtocolError" |
| 247 | ? Deferred.succeed(protocolErrorReceived, void 0) |
| 248 | : Effect.void).pipe(Effect.forkScoped) |
| 249 | |
| 250 | const request = (id: number) => ({ |
| 251 | _tag: "Request" as const, |
| 252 | id, |
| 253 | tag: "Test", |
| 254 | payload: null, |
| 255 | headers: [] |
| 256 | }) |
| 257 | const first = yield* protocol.send(0, request(1)).pipe(Effect.forkChild) |
| 258 | yield* Deferred.await(firstRequestSent) |
| 259 | yield* Deferred.fail( |
| 260 | runFailure, |
| 261 | new WorkerError({ reason: new WorkerReceiveError({ message: "worker exited" }) }) |
| 262 | ) |
| 263 | yield* Deferred.await(protocolErrorReceived) |
| 264 | |
| 265 | const firstCompleted = yield* Fiber.join(first).pipe( |
| 266 | Effect.timeout("1 second"), |
| 267 | Effect.forkChild |
| 268 | ) |
| 269 | yield* TestClock.adjust("1 second") |
| 270 | yield* Fiber.join(firstCompleted) |
| 271 | |
| 272 | const second = yield* protocol.send(0, request(2)).pipe(Effect.forkChild) |
| 273 | const secondSent = yield* Deferred.await(secondRequestSent).pipe( |
| 274 | Effect.timeout("1 second"), |
| 275 | Effect.forkChild |
| 276 | ) |
| 277 | yield* TestClock.adjust("1 second") |
| 278 | yield* Fiber.join(secondSent) |
| 279 | yield* Fiber.interrupt(second) |
| 280 | assert.deepStrictEqual(sentRequestIds, [1, 2]) |
| 281 | })) |
| 282 | |
| 283 | it("preserves RpcClientError failures from a reloaded module copy", async () => { |
| 284 | vi.resetModules() |
| 285 | const ForeignRpcClientError = await vi.importActual<typeof RpcClientErrorModule>( |
| 286 | "../../src/rpc/RpcClientError.ts" |
| 287 | ) |
| 288 | const rpcClientError = new ForeignRpcClientError.RpcClientError({ |
| 289 | reason: new ForeignRpcClientError.RpcClientDefect({ message: "boom", cause: undefined }) |
| 290 | }) |
| 291 | assert.isFalse(rpcClientError instanceof RpcClientError) |
| 292 | |
| 293 | const httpClient = HttpClient.make((request) => { |
| 294 | const response = HttpClientResponse.fromWeb(request, new Response("", { status: 200 })) |
| 295 | Object.defineProperty(response, "stream", { value: Stream.fail(rpcClientError) }) |
| 296 | return Effect.succeed(response) |
| 297 | }) |
| 298 | const error = await Effect.gen(function*() { |
| 299 | const client = yield* RpcClient.make(TestGroup).pipe( |
| 300 | Effect.provide(makeProtocolLayerWithClient(RpcSerialization.layerNdjson, httpClient)) |
| 301 | ) |
| 302 | return yield* client.Ping().pipe(Effect.flip) |
| 303 | }).pipe(Effect.scoped, Effect.runPromise) |
| 304 | |
| 305 | assert.strictEqual(error, rpcClientError) |
| 306 | }) |
| 307 | |
| 308 | it.effect("fails request on empty HTTP response for unframed serialization", () => |
| 309 | assertEmptyResponseFailsRequest(RpcSerialization.layerJson, "[]")) |
| 310 | |
| 311 | it.effect("fails request on empty HTTP response for framed serialization", () => |
| 312 | assertEmptyResponseFailsRequest(RpcSerialization.layerNdjson, "")) |
| 313 | |
| 314 | it.effect("defects request when framed HTTP response closes before request completes", () => |
| 315 | Effect.gen(function*() { |
| 316 | const client = yield* RpcClient.make(TestGroup, { |
| 317 | generateRequestId: () => RpcMessage.RequestId("0") |
| 318 | }).pipe( |
| 319 | Effect.provide(makeProtocolLayer( |
| 320 | RpcSerialization.layerNdjson, |
| 321 | JSON.stringify({ _tag: "Chunk", requestId: "0", values: ["event"] }) + "\n" |
| 322 | )) |
| 323 | ) |
| 324 | |
| 325 | const cause = yield* client.Events().pipe( |
| 326 | Stream.runDrain, |
| 327 | Effect.timeout("1 second"), |
| 328 | Effect.sandbox, |
| 329 | Effect.flip |
| 330 | ) |
| 331 | |
| 332 | const error = Cause.squash(cause) |
| 333 | assert.instanceOf(error, RpcClientError) |
| 334 | assert.strictEqual(error.reason._tag, "RpcClientDefect") |
| 335 | assert.strictEqual(error.reason.message, "HTTP response ended before RPC request completed") |
| 336 | })) |
| 337 | |
| 338 | it.effect("reports transient socket open errors without failing in-flight streams", () => |
| 339 | Effect.gen(function*() { |
| 340 | const requestSent = yield* Deferred.make<void>() |
| 341 | const threeErrors = yield* Deferred.make<void>() |
| 342 | const errors: Array<RpcClientError> = [] |
| 343 | const socketError = new Socket.SocketError({ |
| 344 | reason: new Socket.SocketOpenError({ |
| 345 | kind: "Unknown", |
| 346 | cause: new Error("connection refused") |
| 347 | }) |
| 348 | }) |
| 349 | const socket = Socket.make({ |
| 350 | reader: Effect.succeed({ |
| 351 | pull: Deferred.await(requestSent).pipe(Effect.andThen(Effect.fail(socketError))), |
| 352 | upgrade: Socket.SocketUpgradeError.unsupported |
| 353 | }), |
| 354 | writer: Effect.succeed({ |
| 355 | write: () => Effect.asVoid(Deferred.succeed(requestSent, void 0)), |
| 356 | writeAll: () => Effect.asVoid(Deferred.succeed(requestSent, void 0)) |
| 357 | }) |
| 358 | }) |
| 359 | const protocol = yield* RpcClient.makeProtocolSocket({ |
| 360 | retryTransientErrors: true, |
| 361 | retryPolicy: Schedule.spaced("1 millis"), |
| 362 | onTransientError: (error) => |
| 363 | Effect.suspend(() => { |
| 364 | errors.push(error) |
| 365 | return errors.length === 3 ? Deferred.succeed(threeErrors, void 0) : Effect.void |
| 366 | }) |
| 367 | }).pipe( |
| 368 | Effect.provideService(Socket.Socket, socket), |
| 369 | Effect.provide(RpcSerialization.layerNdjson) |
| 370 | ) |
| 371 | const client = yield* RpcClient.make(TestGroup).pipe( |
| 372 | Effect.provideService(RpcClient.Protocol, protocol) |
| 373 | ) |
| 374 | const streamFiber = yield* client.Events().pipe(Stream.runDrain, Effect.forkChild) |
| 375 | |
| 376 | yield* TestClock.adjust("2 millis") |
| 377 | yield* Deferred.await(threeErrors).pipe(Effect.timeout("1 second")) |
| 378 | |
| 379 | assert.lengthOf(errors, 3) |
| 380 | for (const error of errors) { |
| 381 | assert.strictEqual(error.reason._tag, "SocketOpenError") |
| 382 | } |
| 383 | assert.isUndefined(streamFiber.pollUnsafe()) |
| 384 | })) |
| 385 | |
| 386 | it.effect("fails in-flight streams on a missed pong while retrying socket errors", () => |
| 387 | Effect.gen(function*() { |
| 388 | const requestSent = yield* Deferred.make<void>() |
| 389 | let writes = 0 |
| 390 | const write = () => |
| 391 | Effect.sync(() => writes++).pipe(Effect.andThen(Deferred.succeed(requestSent, void 0)), Effect.asVoid) |
| 392 | const socket = Socket.make({ |
| 393 | reader: Effect.succeed({ |
| 394 | pull: Effect.never, |
| 395 | upgrade: Socket.SocketUpgradeError.unsupported |
| 396 | }), |
| 397 | writer: Effect.succeed({ |
| 398 | write, |
| 399 | writeAll: write |
| 400 | }) |
| 401 | }) |
| 402 | const protocol = yield* RpcClient.makeProtocolSocket({ |
| 403 | retryTransientErrors: true, |
| 404 | retryPolicy: Schedule.spaced("1 hour") |
| 405 | }).pipe( |
| 406 | Effect.provideService(Socket.Socket, socket), |
| 407 | Effect.provide(RpcSerialization.layerNdjson) |
| 408 | ) |
| 409 | const client = yield* RpcClient.make(TestGroup).pipe( |
| 410 | Effect.provideService(RpcClient.Protocol, protocol) |
| 411 | ) |
| 412 | const streamFiber = yield* client.Events().pipe( |
| 413 | Stream.runDrain, |
| 414 | Effect.timeout("11 seconds"), |
| 415 | Effect.flip, |
| 416 | Effect.forkChild |
| 417 | ) |
| 418 | |
| 419 | yield* Deferred.await(requestSent) |
| 420 | yield* TestClock.adjust("11 seconds") |
| 421 | const error = yield* Fiber.join(streamFiber) |
| 422 | |
| 423 | assert.isAtLeast(writes, 2) // request and unanswered ping |
| 424 | assert.instanceOf(error, RpcClientError) |
| 425 | assert.strictEqual(error.reason._tag, "SocketReadError") |
| 426 | })) |
| 427 | |
| 428 | it.effect("fails in-flight streams when transient retries are exhausted", () => |
| 429 | Effect.gen(function*() { |
| 430 | const requestSent = yield* Deferred.make<void>() |
| 431 | const socketError = new Socket.SocketError({ |
| 432 | reason: new Socket.SocketOpenError({ |
| 433 | kind: "Unknown", |
| 434 | cause: new Error("connection refused") |
| 435 | }) |
| 436 | }) |
| 437 | const socket = Socket.make({ |
| 438 | reader: Effect.succeed({ |
| 439 | pull: Deferred.await(requestSent).pipe(Effect.andThen(Effect.fail(socketError))), |
| 440 | upgrade: Socket.SocketUpgradeError.unsupported |
| 441 | }), |
| 442 | writer: Effect.succeed({ |
| 443 | write: () => Effect.asVoid(Deferred.succeed(requestSent, void 0)), |
| 444 | writeAll: () => Effect.asVoid(Deferred.succeed(requestSent, void 0)) |
| 445 | }) |
| 446 | }) |
| 447 | const protocol = yield* RpcClient.makeProtocolSocket({ |
| 448 | retryTransientErrors: true, |
| 449 | retryPolicy: Schedule.recurs(2) |
| 450 | }).pipe( |
| 451 | Effect.provideService(Socket.Socket, socket), |
| 452 | Effect.provide(RpcSerialization.layerNdjson) |
| 453 | ) |
| 454 | const client = yield* RpcClient.make(TestGroup).pipe( |
| 455 | Effect.provideService(RpcClient.Protocol, protocol) |
| 456 | ) |
| 457 | const streamFiber = yield* client.Events().pipe( |
| 458 | Stream.runDrain, |
| 459 | Effect.timeout("1 second"), |
| 460 | Effect.flip, |
| 461 | Effect.forkChild |
| 462 | ) |
| 463 | |
| 464 | yield* TestClock.adjust("1 second") |
| 465 | const error = yield* Fiber.join(streamFiber) |
| 466 | |
| 467 | assert.instanceOf(error, RpcClientError) |
| 468 | assert.strictEqual(error.reason._tag, "SocketOpenError") |
| 469 | })) |
| 470 | |
| 471 | it.effect("continues retrying when the transient error hook defects", () => |
| 472 | Effect.gen(function*() { |
| 473 | const requestSent = yield* Deferred.make<void>() |
| 474 | let attempts = 0 |
| 475 | const socketError = new Socket.SocketError({ |
| 476 | reason: new Socket.SocketOpenError({ |
| 477 | kind: "Unknown", |
| 478 | cause: new Error("connection refused") |
| 479 | }) |
| 480 | }) |
| 481 | const socket = Socket.make({ |
| 482 | reader: Effect.succeed({ |
| 483 | pull: Deferred.await(requestSent).pipe( |
| 484 | Effect.tap(() => Effect.sync(() => attempts++)), |
| 485 | Effect.andThen(Effect.fail(socketError)) |
| 486 | ), |
| 487 | upgrade: Socket.SocketUpgradeError.unsupported |
| 488 | }), |
| 489 | writer: Effect.succeed({ |
| 490 | write: () => Effect.asVoid(Deferred.succeed(requestSent, void 0)), |
| 491 | writeAll: () => Effect.asVoid(Deferred.succeed(requestSent, void 0)) |
| 492 | }) |
| 493 | }) |
| 494 | const protocol = yield* RpcClient.makeProtocolSocket({ |
| 495 | retryTransientErrors: true, |
| 496 | retryPolicy: Schedule.recurs(2), |
| 497 | onTransientError: () => Effect.die("hook defect") |
| 498 | }).pipe( |
| 499 | Effect.provideService(Socket.Socket, socket), |
| 500 | Effect.provide(RpcSerialization.layerNdjson) |
| 501 | ) |
| 502 | const client = yield* RpcClient.make(TestGroup).pipe( |
| 503 | Effect.provideService(RpcClient.Protocol, protocol) |
| 504 | ) |
| 505 | const streamFiber = yield* client.Events().pipe( |
| 506 | Stream.runDrain, |
| 507 | Effect.timeout("1 second"), |
| 508 | Effect.flip, |
| 509 | Effect.forkChild |
| 510 | ) |
| 511 | |
| 512 | yield* TestClock.adjust("1 second") |
| 513 | const error = yield* Fiber.join(streamFiber) |
| 514 | |
| 515 | assert.strictEqual(attempts, 3) |
| 516 | assert.instanceOf(error, RpcClientError) |
| 517 | assert.strictEqual(error.reason._tag, "SocketOpenError") |
| 518 | })) |
| 519 | }) |