1// Exercise the raw Node HTTP boundary before Effect attaches serving handlers.
2// @effect-diagnostics nodeBuiltinImport:off
3import * as NodeHttp from "node:http";
4import * as NodeNet from "node:net";
5import { afterEach, describe, expect, it } from "vite-plus/test";
6
7import * as HttpResponseErrorGuard from "./httpResponseErrorGuard.ts";
8
9const { guardHttpResponseWriteErrors } = HttpResponseErrorGuard;
10
11const servers: NodeHttp.Server[] = [];
12const requests: NodeHttp.ClientRequest[] = [];
13const sockets: NodeNet.Socket[] = [];
14
15function listen(server: NodeHttp.Server): Promise<number> {
16 servers.push(server);
17 server.on("connection", (socket) => sockets.push(socket));
18 return new Promise((resolve, reject) => {
19 server.once("error", reject);
20 server.listen(0, "127.0.0.1", () => {
21 resolve((server.address() as NodeNet.AddressInfo).port);
22 });
23 });
24}
25
26function fetchResponse(
27 port: number,
28 path: string,
29 upgrade = false,
30): Promise<NodeHttp.IncomingMessage> {
31 return new Promise((resolve, reject) => {
32 const request = NodeHttp.get(
33 {
34 host: "127.0.0.1",
35 port,
36 path,
37 agent: false,
38 headers: upgrade
39 ? {
40 connection: "Upgrade",
41 upgrade: "websocket",
42 "sec-websocket-key": "dGhlIHNhbXBsZSBub25jZQ==",
43 "sec-websocket-version": "13",
44 }
45 : {},
46 },
47 (response) => {
48 response.on("error", reject);
49 response.on("end", () => resolve(response));
50 response.resume();
51 },
52 );
53 requests.push(request);
54 request.on("error", reject);
55 request.on("upgrade", (response, socket) => {
56 socket.destroy();
57 resolve(response);
58 });
59 });
60}
61
62function fetchStatus(port: number, path: string): Promise<number> {
63 return fetchResponse(port, path).then((response) => response.statusCode ?? 0);
64}
65
66afterEach(async () => {
67 for (const request of requests.splice(0)) {
68 request.destroy();
69 }
70 for (const socket of sockets.splice(0)) {
71 socket.destroy();
72 }
73 await Promise.all(
74 servers
75 .splice(0)
76 .map((server) => new Promise<void>((resolve) => server.close(() => resolve()))),
77 );
78});
79
80describe("guardHttpResponseWriteErrors", () => {
81 it("contains an upgrade socket write failure instead of crashing the process", async () => {
82 const writeErrors: unknown[] = [];
83 const failureObserved = Promise.withResolvers<void>();
84 const server = guardHttpResponseWriteErrors(NodeHttp.createServer(), (error) => {
85 writeErrors.push(error);
86 failureObserved.resolve();
87 });
88
89 server.on("upgrade", (_request, socket) => {
90 // Node's HTTP server no longer listens for errors on an upgraded socket.
91 socket.destroy(Object.assign(new Error("write EPIPE"), { code: "EPIPE" }));
92 });
93
94 const port = await listen(server);
95
96 const client = NodeNet.connect(port, "127.0.0.1", () => {
97 client.write(
98 [
99 "GET /rpc HTTP/1.1",
100 "Host: 127.0.0.1",
101 "Connection: Upgrade",
102 "Upgrade: websocket",
103 "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==",
104 "Sec-WebSocket-Version: 13",
105 "",
106 "",
107 ].join("\r\n"),
108 );
109 });
110 sockets.push(client);
111 client.on("error", () => {});
112
113 await failureObserved.promise;
114 client.destroy();
115
116 expect(writeErrors).toHaveLength(1);
117 expect(writeErrors[0]).toBeInstanceOf(Error);
118 expect((writeErrors[0] as NodeJS.ErrnoException).code).toBe("EPIPE");
119
120 server.on("request", (_request, response) => {
121 response.writeHead(200, { "content-type": "text/plain" });
122 response.end("ok");
123 });
124 await expect(fetchStatus(port, "/")).resolves.toBe(200);
125 });
126});
127
128describe("keepIdleConnections", () => {
129 it("closes a connection whose response finishes after the server stopped listening", async () => {
130 const server = HttpResponseErrorGuard.keepIdleConnections(NodeHttp.createServer());
131 const body = "x".repeat(1_000_000);
132 const closed = Promise.withResolvers<void>();
133 server.on("request", (_request, response) => {
134 // The server shuts down while this request is in flight.
135 server.close(() => closed.resolve());
136 response.end(body);
137 });
138 const port = await listen(server);
139 const agent = new NodeHttp.Agent({ keepAlive: true });
140
141 const received = await new Promise<number>((resolve, reject) => {
142 const request = NodeHttp.get({ host: "127.0.0.1", port, agent }, (response) => {
143 let bytes = 0;
144 response.on("data", (chunk: Buffer) => {
145 bytes += chunk.length;
146 });
147 response.on("end", () => resolve(bytes));
148 response.on("error", reject);
149 });
150 requests.push(request);
151 request.on("error", reject);
152 });
153
154 expect(received).toBe(body.length);
155 // Left alone, the kept-alive connection holds the server open for the keep-alive time.
156 await closed.promise;
157 agent.destroy();
158 });
159
160 it("keeps a connection until every response on it has finished", async () => {
161 const server = HttpResponseErrorGuard.keepIdleConnections(NodeHttp.createServer());
162 const closed = Promise.withResolvers<void>();
163 // Larger than a socket buffer, so it still has output to send after `end`.
164 const thirdBody = "x".repeat(8_000_000);
165 const held: NodeHttp.ServerResponse[] = [];
166 server.on("request", (request, response) => {
167 if (request.url !== "/third") {
168 held.push(response);
169 return;
170 }
171 const [first, second] = held;
172 // All three pipelined requests are in flight when the server shuts down. The third
173 // response is ended before the first two, and the second ends after the first.
174 response.end(thirdBody);
175 server.close(() => closed.resolve());
176 first?.once("finish", () => setImmediate(() => second?.end("second-body")));
177 first?.end("first-body");
178 });
179 const port = await listen(server);
180
181 const received = await new Promise<string>((resolve, reject) => {
182 let data = "";
183 const client = NodeNet.connect(port, "127.0.0.1", () => {
184 client.write(
185 ["/first", "/second", "/third"]
186 .map((path) => `GET ${path} HTTP/1.1\r\nHost: 127.0.0.1\r\n\r\n`)
187 .join(""),
188 );
189 });
190 sockets.push(client);
191 client.on("data", (chunk: Buffer) => {
192 data += chunk.toString();
193 });
194 client.on("end", () => resolve(data));
195 client.on("error", reject);
196 });
197
198 expect(received.slice(0, 1_000)).toContain("first-body");
199 expect(received.slice(0, 1_000)).toContain("second-body");
200 expect(received.endsWith(thirdBody)).toBe(true);
201 await closed.promise;
202 });
203});
204
205describe("rejectRequestsUntilServing", () => {
206 it("rejects early HTTP and upgrades, then hands each off to its serving handler", async () => {
207 const writeErrors: unknown[] = [];
208 const server = HttpResponseErrorGuard.rejectRequestsUntilServing(
209 guardHttpResponseWriteErrors(NodeHttp.createServer(), (error) => writeErrors.push(error)),
210 );
211 const port = await listen(server);
212
213 const earlyHttp = await fetchResponse(port, "/");
214 expect(earlyHttp.statusCode).toBe(503);
215 expect(earlyHttp.headers["retry-after"]).toBe("1");
216
217 const earlyUpgrade = await fetchResponse(port, "/rpc", true);
218 expect(earlyUpgrade.statusCode).toBe(503);
219 expect(earlyUpgrade.headers["retry-after"]).toBe("1");
220
221 server.on("request", (_request, response) => response.writeHead(204).end());
222
223 const servingHttp = await fetchResponse(port, "/");
224 expect(servingHttp.statusCode).toBe(204);
225 expect(servingHttp.headers["retry-after"]).toBeUndefined();
226
227 const pendingUpgrade = await fetchResponse(port, "/rpc", true);
228 expect(pendingUpgrade.statusCode).toBe(503);
229 expect(pendingUpgrade.headers["retry-after"]).toBe("1");
230
231 server.on("upgrade", (_request, socket) => {
232 socket.end(
233 "HTTP/1.1 101 Switching Protocols\r\nConnection: Upgrade\r\nUpgrade: websocket\r\n\r\n",
234 );
235 });
236
237 const servingUpgrade = await fetchResponse(port, "/rpc", true);
238 expect(servingUpgrade.statusCode).toBe(101);
239 expect(servingUpgrade.headers["retry-after"]).toBeUndefined();
240 expect(writeErrors).toEqual([]);
241 });
242});