1import * as Cloudflare from "@/Cloudflare/index.ts";
2import * as RpcWebSocketClient from "@/Cloudflare/Workers/RpcWebSocketClient.ts";
3import * as Alchemy from "@/index.ts";
4import * as Test from "@/Test/Alchemy.ts";
5import { describe, expect } from "alchemy-test";
6import type { Done, TimeoutError } from "effect/Cause";
7import * as Context from "effect/Context";
8import * as Data from "effect/Data";
9import * as Deferred from "effect/Deferred";
10import * as Effect from "effect/Effect";
11import * as Exit from "effect/Exit";
12import * as Fiber from "effect/Fiber";
13import * as Layer from "effect/Layer";
14import * as ManagedRuntime from "effect/ManagedRuntime";
15import * as Queue from "effect/Queue";
16import * as Schedule from "effect/Schedule";
17import * as Schema from "effect/Schema";
18import * as Scope from "effect/Scope";
19import * as Stream from "effect/Stream";
20import * as HttpClientRequest from "effect/http/HttpClientRequest";
21import { RpcClient, RpcSerialization } from "effect/rpc";
22import type { RpcClientError } from "effect/rpc/RpcClientError";
23import * as Socket from "effect/socket/Socket";
24import { requestWorker } from "../Utils/WorkerRequest.ts";
25import {
26 Greeting,
27 Rejected,
28 SocketRpcs,
29 SocketStats,
30} from "./fixtures/rpc-websocket/rpcs.ts";
31import SocketWorker from "./fixtures/rpc-websocket/worker.ts";
32
33class SocketClient extends Context.Service<SocketClient>()("SocketClient", {
34 make: RpcClient.make(SocketRpcs),
35}) {}
36
37const observeSockets = Effect.sync(() => {
38 const sockets: {
39 socket: WebSocket;
40 closeCodes: (number | undefined)[];
41 closed: Deferred.Deferred<{ code: number; reason: string }>;
42 }[] = [];
43 const constructor = (url: string) =>
44 Layer.effect(
45 Socket.WebSocketConstructor,
46 Effect.gen(function* () {
47 // Retry only the native handshake, before constructing the RPC client.
48 const { socket: ready } = yield* rawConnect(url);
49 let consumed = false;
50 return (url: string) => {
51 const socket = consumed ? new WebSocket(url) : ready;
52 consumed = true;
53 const closeCodes: (number | undefined)[] = [];
54 const close = socket.close.bind(socket);
55 socket.close = (code, reason) => {
56 closeCodes.push(code);
57 close(code, reason);
58 };
59 const closed = Deferred.makeUnsafe<{
60 code: number;
61 reason: string;
62 }>();
63 socket.addEventListener("close", (event) =>
64 Deferred.doneUnsafe(
65 closed,
66 Exit.succeed({ code: event.code, reason: event.reason }),
67 ),
68 );
69 sockets.push({ socket, closeCodes, closed });
70 return socket;
71 };
72 }),
73 );
74 return {
75 sockets,
76 layer: (url: string) =>
77 RpcWebSocketClient.layer(
78 SocketClient,
79 SocketRpcs,
80 url.replace(/^http/, "ws"),
81 { protocol: { retryTransientErrors: false } },
82 ).pipe(Layer.provide(constructor(url))),
83 assertClosed: Effect.gen(function* () {
84 expect(sockets).toHaveLength(1);
85 const { socket, closeCodes, closed } = sockets[0]!;
86 // The RPC Layer must close before rawConnect's fallback finalizer.
87 expect(closeCodes[0]).toBe(1000);
88 expect(
89 (yield* Deferred.await(closed).pipe(Effect.timeout("10 seconds"))).code,
90 ).toBe(1000);
91 expect(socket.readyState).toBe(WebSocket.CLOSED);
92 expect(sockets).toHaveLength(1);
93 }),
94 };
95});
96
97const connect = Effect.fn(function* (url: string) {
98 const { socket: raw } = yield* rawConnect(url);
99 const protocol = RpcClient.layerProtocolSocket().pipe(
100 Layer.provide(
101 Layer.mergeAll(
102 Layer.effect(Socket.Socket, Socket.fromWebSocket(Effect.succeed(raw))),
103 RpcSerialization.layerJson,
104 ),
105 ),
106 );
107 const context = yield* Layer.build(protocol);
108 const client = yield* RpcClient.make(SocketRpcs).pipe(
109 Effect.provide(context),
110 );
111 return { client, raw };
112});
113
114class WebSocketHandshakeFailed extends Data.TaggedError(
115 "WebSocketHandshakeFailed",
116)<{
117 readonly message: string;
118}> {}
119
120type ClientLayerError =
121 | RpcClientError
122 | WebSocketHandshakeFailed
123 | TimeoutError;
124
125const rawConnect = Effect.fn(
126 function* (url: string) {
127 const opened = yield* Deferred.make<void, WebSocketHandshakeFailed>();
128 const closed = yield* Deferred.make<{ code: number; reason: string }>();
129 const messages = yield* Queue.unbounded<string>();
130 const socket = yield* Effect.acquireRelease(
131 Effect.sync(() => {
132 const socket = new WebSocket(url.replace(/^http/, "ws"));
133 socket.addEventListener("open", () =>
134 Deferred.doneUnsafe(opened, Exit.void),
135 );
136 socket.addEventListener("error", (event) =>
137 Deferred.doneUnsafe(
138 opened,
139 Exit.fail(
140 new WebSocketHandshakeFailed({
141 message: `WebSocket handshake failed: ${"message" in event ? event.message : event.type}`,
142 }),
143 ),
144 ),
145 );
146 socket.addEventListener("message", (event) =>
147 Queue.offerUnsafe(messages, String(event.data)),
148 );
149 socket.addEventListener("close", (event) =>
150 Deferred.doneUnsafe(
151 closed,
152 Exit.succeed({ code: event.code, reason: event.reason }),
153 ),
154 );
155 return socket;
156 }),
157 (socket) => Effect.sync(() => socket.close()),
158 );
159 yield* Deferred.await(opened).pipe(
160 Effect.timeout("10 seconds"),
161 Effect.onError(() => Effect.sync(() => socket.close())),
162 );
163 const send = (value: unknown) =>
164 Effect.sync(() => socket.send(JSON.stringify(value)));
165 const receive = Queue.take(messages).pipe(
166 Effect.flatMap((value) => Effect.try(() => JSON.parse(value))),
167 Effect.timeout("10 seconds"),
168 );
169 return {
170 socket,
171 send,
172 receive,
173 buffered: Queue.size(messages),
174 closed: Deferred.await(closed).pipe(Effect.timeout("10 seconds")),
175 };
176 },
177 Effect.retry({
178 // Bun hides rejected upgrade responses; retry only before any RPC is sent.
179 while: (error) => error._tag === "WebSocketHandshakeFailed",
180 schedule: Schedule.spaced("1 second"),
181 times: 8,
182 }),
183);
184
185const readStats = Effect.fn(function* (url: string) {
186 const response = yield* requestWorker(HttpClientRequest.get(url));
187 if (response.status !== 200) {
188 return yield* Effect.fail(
189 new Error(`GET ${url}: ${response.status}: ${yield* response.text}`),
190 );
191 }
192 return yield* response.json.pipe(
193 Effect.flatMap(Schema.decodeUnknownEffect(SocketStats)),
194 );
195}, Effect.timeout("15 seconds"));
196
197describe.concurrent.each([
198 { dev: true, stage: "rpc-websocket-local" },
199 { dev: false, stage: "rpc-websocket-live" },
200])(
201 "WebSocket RPC (dev: $dev)",
202 ({ dev, stage }) => {
203 const state = dev ? Alchemy.inMemoryState() : Cloudflare.state();
204 const Stack = Alchemy.Stack(
205 "RpcWebSocketStack",
206 {
207 providers: Cloudflare.providers(),
208 state,
209 },
210 Effect.gen(function* () {
211 const worker = yield* SocketWorker;
212 return { url: worker.url.as<string>() };
213 }),
214 );
215 const { test, beforeAll, afterAll, deploy, destroy } = Test.make({
216 providers: Cloudflare.providers(),
217 state,
218 dev,
219 stage,
220 });
221 const stack = beforeAll(
222 Effect.gen(function* () {
223 yield* destroy(Stack);
224 const output = yield* deploy(Stack);
225 yield* requestWorker(HttpClientRequest.get(`${output.url}/ready`)).pipe(
226 Effect.flatMap((response) =>
227 response.text.pipe(
228 Effect.flatMap((body) =>
229 response.status === 200 && body === "ready"
230 ? Effect.void
231 : Effect.fail(
232 new Test.WorkerNotReady({ status: response.status }),
233 ),
234 ),
235 ),
236 ),
237 );
238 yield* rawConnect(`${output.url}/rpc/ready`).pipe(Effect.scoped);
239 return output;
240 }),
241 { timeout: 120_000 },
242 );
243 afterAll.skipIf(!!process.env.NO_DESTROY)(destroy(Stack), {
244 timeout: 30_000,
245 });
246
247 test(
248 "readiness handling preserves application HTTP errors",
249 Effect.gen(function* () {
250 const { url } = yield* stack;
251 for (const [path, status] of [
252 ["not-found", 404],
253 ["server-error", 500],
254 ] as const) {
255 const response = yield* requestWorker(
256 HttpClientRequest.get(`${url}/${path}`),
257 );
258 expect(response.status).toBe(status);
259 expect(yield* response.text).toBe("application error");
260 }
261 }).pipe(Effect.scoped),
262 { tags: [...(dev ? ["local"] : ["live"])], timeout: 30_000 },
263 );
264
265 test(
266 "typed unary, failure, streaming, and HTTP clients share one object",
267 Effect.gen(function* () {
268 const { url } = yield* stack;
269 const { client } = yield* connect(`${url}/rpc/roundtrip`);
270 const greeting = yield* client.greet({ name: "Sam" });
271 expect(greeting).toBeInstanceOf(Greeting);
272 expect(greeting.message).toBe("Hello, Sam!");
273 expect(yield* client.increment()).toBe(1);
274 expect(yield* client.reject().pipe(Effect.flip)).toBeInstanceOf(
275 Rejected,
276 );
277 expect(
278 yield* client.numbers({ count: 40 }).pipe(Stream.runCollect),
279 ).toEqual(Array.from({ length: 40 }, (_, i) => i + 1));
280 const response = yield* requestWorker(
281 HttpClientRequest.get(`${url}/stats/roundtrip`),
282 );
283 expect(response.status).toBe(200);
284 expect(yield* response.json).toMatchObject({ count: 1 });
285 }).pipe(Effect.scoped),
286 { tags: [...(dev ? ["local"] : ["live"])], timeout: 30_000 },
287 );
288
289 test(
290 "a Layer shares one actual socket for concurrent callers without a caller scope and closes normally",
291 Effect.gen(function* () {
292 const { url } = yield* stack;
293 const endpoint = `${url}/rpc/layer-client`;
294 const observed = yield* observeSockets;
295 const program: Effect.Effect<readonly number[], ClientLayerError> =
296 Effect.gen(function* () {
297 const results = yield* Effect.all(
298 Array.from({ length: 40 }, () =>
299 Effect.gen(function* () {
300 const client = yield* SocketClient;
301 return yield* client.increment();
302 }),
303 ),
304 { concurrency: "unbounded" },
305 );
306 const client = yield* SocketClient;
307 expect(
308 yield* client.numbers({ count: 3 }).pipe(Stream.runCollect),
309 ).toEqual([1, 2, 3]);
310 expect(observed.sockets).toHaveLength(1);
311 expect(observed.sockets[0]!.socket.readyState).toBe(WebSocket.OPEN);
312 return results;
313 }).pipe(Effect.provide(observed.layer(endpoint)));
314 expect((yield* program).toSorted((a, b) => a - b)).toEqual(
315 Array.from({ length: 40 }, (_, i) => i + 1),
316 );
317 yield* observed.assertClosed;
318 expect((yield* readStats(`${url}/stats/layer-client`)).count).toBe(40);
319 }),
320 { tags: [...(dev ? ["local"] : ["live"])], timeout: 30_000 },
321 );
322
323 test(
324 "a Layer closes its actual socket with code 1000 on application failure",
325 Effect.gen(function* () {
326 const { url } = yield* stack;
327 const endpoint = `${url}/rpc/layer-failure`;
328 const observed = yield* observeSockets;
329 const program: Effect.Effect<never, Rejected | ClientLayerError> =
330 Effect.gen(function* () {
331 const client = yield* SocketClient;
332 yield* client.increment();
333 return yield* client.reject();
334 }).pipe(Effect.provide(observed.layer(endpoint)));
335 expect(yield* program.pipe(Effect.flip)).toBeInstanceOf(Rejected);
336 yield* observed.assertClosed;
337 expect((yield* readStats(`${url}/stats/layer-failure`)).count).toBe(1);
338 }),
339 { tags: [...(dev ? ["local"] : ["live"])], timeout: 30_000 },
340 );
341
342 test(
343 "interrupting a Layer closes its actual socket with code 1000 and releases the handler",
344 Effect.gen(function* () {
345 const { url } = yield* stack;
346 const endpoint = `${url}/rpc/layer-interruption`;
347 const observed = yield* observeSockets;
348 const started = yield* Deferred.make<void>();
349 const program: Effect.Effect<void, ClientLayerError> = Effect.gen(
350 function* () {
351 const client = yield* SocketClient;
352 yield* client
353 .watch({ key: "layer-interruption" })
354 .pipe(
355 Stream.runForEach(() => Deferred.succeed(started, undefined)),
356 );
357 },
358 ).pipe(Effect.provide(observed.layer(endpoint)));
359 const fiber = yield* program.pipe(Effect.forkScoped);
360 yield* Deferred.await(started).pipe(Effect.timeout("10 seconds"));
361 expect(observed.sockets).toHaveLength(1);
362 yield* Fiber.interrupt(fiber);
363 expect(Exit.hasInterrupts(yield* Fiber.await(fiber))).toBe(true);
364 yield* observed.assertClosed;
365 const status = yield* readStats(`${url}/stats/layer-interruption`).pipe(
366 Effect.repeat({
367 schedule: Schedule.spaced("100 millis"),
368 times: 10,
369 until: (state) => state.closed.includes("layer-interruption"),
370 }),
371 );
372 expect(status.closed).toEqual(["layer-interruption"]);
373 expect(status.invocations).toEqual({ "layer-interruption": 1 });
374 }).pipe(Effect.scoped),
375 { tags: [...(dev ? ["local"] : ["live"])], timeout: 30_000 },
376 );
377
378 test(
379 "ManagedRuntime.dispose closes the shared actual socket with code 1000",
380 Effect.gen(function* () {
381 const { url } = yield* stack;
382 const endpoint = `${url}/rpc/layer-managed`;
383 const observed = yield* observeSockets;
384 const runtime = yield* Effect.acquireRelease(
385 Effect.sync(() => ManagedRuntime.make(observed.layer(endpoint))),
386 (runtime) => Effect.promise(() => runtime.dispose()),
387 );
388 const call: Effect.Effect<number, RpcClientError, SocketClient> =
389 Effect.gen(function* () {
390 const client = yield* SocketClient;
391 return yield* client.increment();
392 });
393 const results = yield* Effect.all(
394 Array.from({ length: 8 }, () =>
395 Effect.sync(() => runtime.runFork(call)).pipe(
396 Effect.flatMap(Fiber.join),
397 ),
398 ),
399 { concurrency: "unbounded" },
400 );
401 expect(results.toSorted((a, b) => a - b)).toEqual([
402 1, 2, 3, 4, 5, 6, 7, 8,
403 ]);
404 expect(observed.sockets).toHaveLength(1);
405 expect(observed.sockets[0]!.socket.readyState).toBe(WebSocket.OPEN);
406 yield* Effect.promise(() => runtime.dispose());
407 yield* observed.assertClosed;
408 expect((yield* readStats(`${url}/stats/layer-managed`)).count).toBe(8);
409 }).pipe(Effect.scoped),
410 { tags: [...(dev ? ["local"] : ["live"])], timeout: 30_000 },
411 );
412
413 test(
414 "a queue subscription owns a separate scope without closing the Layer socket",
415 Effect.gen(function* () {
416 const { url } = yield* stack;
417 const endpoint = `${url}/rpc/layer-queue`;
418 const observed = yield* observeSockets;
419 const program: Effect.Effect<void, ClientLayerError | Done> =
420 Effect.gen(function* () {
421 const client = yield* SocketClient;
422 yield* Effect.gen(function* () {
423 const queue = yield* client.watch(
424 { key: "queue" },
425 { asQueue: true },
426 );
427 expect(yield* Queue.take(queue)).toBe(1);
428 }).pipe(Effect.scoped);
429 expect(observed.sockets).toHaveLength(1);
430 expect(observed.sockets[0]!.socket.readyState).toBe(WebSocket.OPEN);
431 expect((yield* client.greet({ name: "still open" })).message).toBe(
432 "Hello, still open!",
433 );
434 }).pipe(Effect.provide(observed.layer(endpoint)));
435 yield* program;
436 yield* observed.assertClosed;
437 const status = yield* readStats(`${url}/stats/layer-queue`).pipe(
438 Effect.repeat({
439 schedule: Schedule.spaced("100 millis"),
440 times: 10,
441 until: (state) => state.closed.includes("queue"),
442 }),
443 );
444 expect(status.closed).toEqual(["queue"]);
445 expect(status.invocations).toEqual({ queue: 1 });
446 }),
447 { tags: [...(dev ? ["local"] : ["live"])], timeout: 30_000 },
448 );
449
450 test(
451 "multiplexes concurrent calls without crossing instance state",
452 Effect.gen(function* () {
453 const { url } = yield* stack;
454 const { client } = yield* connect(`${url}/rpc/concurrent`);
455 const results = yield* Effect.all(
456 Array.from({ length: 40 }, () => client.increment()),
457 { concurrency: "unbounded" },
458 );
459 expect(results.sort((a, b) => a - b)).toEqual(
460 Array.from({ length: 40 }, (_, i) => i + 1),
461 );
462 const other = yield* connect(`${url}/rpc/isolated`);
463 expect((yield* other.client.stats()).count).toBe(0);
464 }).pipe(Effect.scoped),
465 { tags: [...(dev ? ["local"] : ["live"])], timeout: 30_000 },
466 );
467
468 test(
469 "stream interruption closes its scope and keeps the socket usable",
470 Effect.gen(function* () {
471 const { url } = yield* stack;
472 const { client, raw } = yield* connect(`${url}/rpc/interruption`);
473 const scope = yield* Scope.make();
474 const values = yield* client
475 .watch({ key: "cancel" }, { asQueue: true })
476 .pipe(Effect.provideService(Scope.Scope, scope));
477 expect(
478 yield* Queue.take(values).pipe(Effect.timeout("10 seconds")),
479 ).toBe(1);
480 yield* Scope.close(scope, Exit.void);
481 const status = yield* client.stats().pipe(
482 Effect.repeat({
483 schedule: Schedule.spaced("100 millis"),
484 times: 10,
485 until: (state) => state.closed.includes("cancel"),
486 }),
487 );
488 expect(status.closed).toContain("cancel");
489 expect(raw.readyState).toBe(WebSocket.OPEN);
490 expect((yield* client.greet({ name: "again" })).message).toBe(
491 "Hello, again!",
492 );
493 }).pipe(Effect.scoped),
494 { tags: [...(dev ? ["local"] : ["live"])], timeout: 30_000 },
495 );
496
497 test(
498 "socket disconnect interrupts active handler resources",
499 Effect.gen(function* () {
500 const { url } = yield* stack;
501 const connection = yield* rawConnect(`${url}/rpc/disconnect`);
502 yield* connection.send({
503 _tag: "Request",
504 id: "1",
505 tag: "watch",
506 payload: { key: "disconnect" },
507 headers: [],
508 });
509 expect(yield* connection.receive).toMatchObject({
510 _tag: "Chunk",
511 values: [1],
512 });
513 yield* Effect.sync(() => connection.socket.close());
514 yield* connection.closed;
515 const status = yield* requestWorker(
516 HttpClientRequest.get(`${url}/stats/disconnect`),
517 ).pipe(
518 Effect.flatMap((response) => response.json),
519 Effect.map((state) => state as { closed: string[] }),
520 Effect.repeat({
521 schedule: Schedule.spaced("100 millis"),
522 times: 10,
523 until: (state) => state.closed.includes("disconnect"),
524 }),
525 );
526 expect(status).toMatchObject({
527 opened: ["disconnect"],
528 closed: ["disconnect"],
529 });
530 }).pipe(Effect.scoped),
531 { tags: [...(dev ? ["local"] : ["live"])], timeout: 30_000 },
532 );
533
534 for (const mode of ["unary", "disconnect"] as const) {
535 test(
536 mode === "unary"
537 ? "unary result does not release the activation before deferred cleanup finishes"
538 : "disconnect before any output retains the activation until deferred cleanup finishes",
539 Effect.gen(function* () {
540 const { url } = yield* stack;
541 const name = `cleanup-${mode}`;
542 const connection = yield* rawConnect(`${url}/rpc/${name}`);
543 const before = yield* readStats(`${url}/stats/${name}`);
544 yield* Effect.gen(function* () {
545 yield* connection.send({
546 _tag: "Request",
547 id: "1",
548 tag: "cleanup",
549 payload: { key: mode, waitForDisconnect: mode === "disconnect" },
550 headers: [],
551 });
552 if (mode === "unary") {
553 expect(yield* connection.receive).toMatchObject({
554 _tag: "Exit",
555 requestId: "1",
556 exit: { _tag: "Success", value: before.boots },
557 });
558 } else {
559 const entered = yield* readStats(`${url}/stats/${name}`).pipe(
560 Effect.repeat({
561 schedule: Schedule.spaced("100 millis"),
562 times: 10,
563 until: (state) => state.opened.includes(mode),
564 }),
565 );
566 expect(entered.boots).toBe(before.boots);
567 expect(entered.opened).toContain(mode);
568 expect(entered.cleanupStarted).toEqual([]);
569 expect(yield* connection.buffered).toBe(0);
570 yield* Effect.sync(() => connection.socket.close());
571 yield* connection.closed;
572 expect(yield* connection.buffered).toBe(0);
573 }
574 const started = yield* readStats(`${url}/stats/${name}`).pipe(
575 Effect.repeat({
576 schedule: Schedule.spaced("100 millis"),
577 times: 10,
578 until: (state) => state.cleanupStarted.includes(mode),
579 }),
580 );
581 expect(started.boots).toBe(before.boots);
582 expect(started.cleanupStarted).toEqual([mode]);
583 expect(started.cleanupCompleted).toEqual({});
584
585 // Raw sockets send no heartbeats; only the test process has a timer.
586 yield* Effect.sleep("20 seconds");
587 const held = yield* readStats(`${url}/stats/${name}`);
588 expect(held.boots).toBe(before.boots);
589 expect(held.cleanupStarted).toEqual([mode]);
590 expect(held.cleanupCompleted).toEqual({});
591
592 if (mode === "unary") {
593 expect(connection.socket.readyState).toBe(WebSocket.OPEN);
594 const control = yield* rawConnect(`${url}/rpc/${name}`);
595 yield* control.send({
596 _tag: "Request",
597 id: "1",
598 tag: "releaseCleanup",
599 payload: { key: mode },
600 headers: [],
601 });
602 expect(yield* control.receive).toMatchObject({
603 _tag: "Exit",
604 requestId: "1",
605 exit: { _tag: "Success", value: true },
606 });
607 } else {
608 const response = yield* requestWorker(
609 HttpClientRequest.post(`${url}/release/${name}/${mode}`),
610 );
611 expect(response.status).toBe(200);
612 expect(yield* response.json).toEqual({ released: true });
613 }
614 const completed = yield* readStats(`${url}/stats/${name}`).pipe(
615 Effect.repeat({
616 schedule: Schedule.spaced("100 millis"),
617 times: 10,
618 until: (state) => state.cleanupCompleted[mode] !== undefined,
619 }),
620 );
621 expect(completed.boots).toBe(before.boots);
622 expect(completed.cleanupCompleted).toEqual({
623 [mode]: before.boots,
624 });
625 }).pipe(
626 Effect.ensuring(
627 requestWorker(
628 HttpClientRequest.post(`${url}/release/${name}/${mode}`),
629 ).pipe(Effect.timeout("15 seconds"), Effect.ignoreCause),
630 ),
631 );
632 }).pipe(Effect.scoped),
633 { tags: [...(dev ? ["local"] : ["live"])], timeout: 90_000 },
634 );
635 }
636
637 test(
638 "the same Layer-owned socket survives real hibernation and serves a new RPC",
639 Effect.gen(function* () {
640 const { url } = yield* stack;
641 const endpoint = `${url}/rpc/idle`;
642 const observed = yield* observeSockets;
643 yield* Effect.gen(function* () {
644 const client = yield* SocketClient;
645 const before = yield* client.stats();
646 expect(observed.sockets).toHaveLength(1);
647 const socket = observed.sockets[0]!.socket;
648 const after = yield* Effect.gen(function* () {
649 yield* Effect.sleep("15 seconds");
650 return yield* client.stats().pipe(Effect.timeout("5 seconds"));
651 }).pipe(
652 Effect.repeat({
653 times: 2,
654 until: (state) => state.boots > before.boots,
655 }),
656 );
657 expect(after.boots).toBeGreaterThan(before.boots);
658 expect(socket.readyState).toBe(WebSocket.OPEN);
659 expect((yield* client.greet({ name: "awake" })).message).toBe(
660 "Hello, awake!",
661 );
662 expect((yield* client.stats()).boots).toBe(after.boots);
663 expect(observed.sockets).toHaveLength(1);
664 expect(observed.sockets[0]!.socket).toBe(socket);
665 }).pipe(Effect.provide(observed.layer(endpoint)));
666 yield* observed.assertClosed;
667 }),
668 { tags: [...(dev ? ["local"] : ["live"])], timeout: 110_000 },
669 );
670
671 test(
672 "hibernation resets an actual socket with an incompatible serializer attachment",
673 Effect.gen(function* () {
674 const { url } = yield* stack;
675 const connection = yield* rawConnect(`${url}/rpc/serialization-reset`);
676 const before = yield* readStats(`${url}/stats/serialization-reset`);
677 const response = yield* requestWorker(
678 HttpClientRequest.post(`${url}/serialization/serialization-reset`),
679 );
680 expect(response.status).toBe(200);
681 expect(yield* response.json).toEqual({ changed: 1 });
682 expect(connection.socket.readyState).toBe(WebSocket.OPEN);
683 const after = yield* Effect.gen(function* () {
684 yield* Effect.sleep("15 seconds");
685 return yield* readStats(`${url}/stats/serialization-reset`);
686 }).pipe(
687 Effect.repeat({
688 times: 2,
689 until: (state) => state.boots > before.boots,
690 }),
691 );
692 expect(after.boots).toBeGreaterThan(before.boots);
693 expect(yield* connection.closed).toEqual({
694 code: 1012,
695 reason: "Durable Object RPC activation reset",
696 });
697 expect(connection.socket.readyState).toBe(WebSocket.CLOSED);
698 expect(yield* connection.buffered).toBe(0);
699 const replacement = yield* connect(`${url}/rpc/serialization-reset`);
700 expect(
701 (yield* replacement.client.greet({ name: "compatible" })).message,
702 ).toBe("Hello, compatible!");
703 expect((yield* replacement.client.stats()).boots).toBe(after.boots);
704 }).pipe(Effect.scoped),
705 { tags: [...(dev ? ["local"] : ["live"])], timeout: 110_000 },
706 );
707
708 test(
709 "native abort fails an active stream without replaying its request",
710 Effect.gen(function* () {
711 const { url } = yield* stack;
712 const connection = yield* rawConnect(`${url}/rpc/lost-stream`);
713 yield* connection.send({
714 _tag: "Request",
715 id: "1",
716 tag: "watch",
717 payload: { key: "lost" },
718 headers: [],
719 });
720 expect(yield* connection.receive).toMatchObject({
721 _tag: "Chunk",
722 values: [1],
723 });
724 yield* connection.send({ _tag: "Ack", requestId: "1" });
725 const before = yield* readStats(`${url}/stats/lost-stream`);
726 expect(before.invocations).toEqual({ lost: 1 });
727 expect(before.opened).toEqual(["lost"]);
728 const response = yield* requestWorker(
729 HttpClientRequest.post(`${url}/abort/lost-stream`),
730 ).pipe(Effect.timeout("15 seconds"));
731 expect(response.status).toBe(200);
732 expect(yield* response.json).toEqual({ aborted: true });
733 yield* connection.closed;
734 expect(connection.socket.readyState).toBe(WebSocket.CLOSED);
735 expect(yield* connection.buffered).toBe(0);
736
737 const replacement = yield* connect(`${url}/rpc/lost-stream`);
738 const status = yield* replacement.client.stats();
739 expect(status.boots).toBeGreaterThan(before.boots);
740 expect(status.invocations).toEqual({ lost: 1 });
741 expect(status.opened).toEqual([]);
742 expect(
743 (yield* replacement.client.greet({ name: "reconnected" })).message,
744 ).toBe("Hello, reconnected!");
745 expect((yield* replacement.client.stats()).invocations).toEqual({
746 lost: 1,
747 });
748 }).pipe(Effect.scoped),
749 { tags: [...(dev ? ["local"] : ["live"])], timeout: 45_000 },
750 );
751
752 test(
753 "schema and unknown-method failures leave the connection usable",
754 Effect.gen(function* () {
755 const { url } = yield* stack;
756 const connection = yield* rawConnect(`${url}/rpc/invalid-requests`);
757 for (const [id, tag, payload] of [
758 ["1", "greet", { name: 42 }],
759 ["2", "missing", {}],
760 ] as const) {
761 yield* connection.send({
762 _tag: "Request",
763 id,
764 tag,
765 payload,
766 headers: [],
767 });
768 expect(yield* connection.receive).toMatchObject({
769 _tag: "Exit",
770 requestId: id,
771 exit: { _tag: "Failure" },
772 });
773 }
774 yield* connection.send({
775 _tag: "Request",
776 id: "3",
777 tag: "greet",
778 payload: { name: "valid" },
779 headers: [],
780 });
781 expect(yield* connection.receive).toMatchObject({
782 _tag: "Exit",
783 requestId: "3",
784 exit: { _tag: "Success", value: { message: "Hello, valid!" } },
785 });
786 }).pipe(Effect.scoped),
787 { tags: [...(dev ? ["local"] : ["live"])], timeout: 30_000 },
788 );
789
790 test(
791 "malformed frames close the offending connection",
792 Effect.gen(function* () {
793 const { url } = yield* stack;
794 const connection = yield* rawConnect(`${url}/rpc/malformed`);
795 yield* Effect.sync(() => connection.socket.send("not json"));
796 expect((yield* connection.closed).code).toBeGreaterThanOrEqual(1002);
797 const healthy = yield* connect(`${url}/rpc/malformed`);
798 expect((yield* healthy.client.greet({ name: "healthy" })).message).toBe(
799 "Hello, healthy!",
800 );
801 }).pipe(Effect.scoped),
802 { tags: [...(dev ? ["local"] : ["live"])], timeout: 30_000 },
803 );
804 },
805 { tags: ["provider:cloudflare", "provider:cloudflare:worker"] },
806);