lib/http/src/server/test.zig
daab053ee43316e1809a84551d573ddd1e5bf3d2
1 const std = @import("std");
2 const http = @import("../root.zig");
3 const sys_root = @import("sys");
4 const sys = sys_root.net;
5 const thread = sys_root.thread;
6 const time = sys_root.time;
7 const root = @import("root.zig");
8
9 const Server = root.Server;
10 const Connection = http.Connection;
11
12 test "Server init and deinit" {
13 const server = try Server.init(std.testing.allocator, .{
14 .port = 0,
15 .num_workers = 2,
16 });
17 server.deinit();
18 }
19
20 test "Server init with custom config" {
21 const server = try Server.init(std.testing.allocator, .{
22 .address = "127.0.0.1",
23 .port = 0,
24 .num_workers = 4,
25 .max_connections = 100,
26 .read_timeout_ms = 5000,
27 .write_timeout_ms = 5000,
28 .tcp_no_delay = true,
29 .connection_input_bytes_per_connection = 321,
30 .request_header_count_per_connection = 3,
31 .request_header_line_bytes = 29,
32 .request_body_bytes_per_connection = 17,
33 .response_header_count_per_connection = 5,
34 .response_head_bytes_per_connection = 257,
35 });
36 defer server.deinit();
37
38 try std.testing.expectEqual(@as(usize, 4), server.pool.capacity.workers);
39 try std.testing.expectEqual(@as(usize, 100), server.pool.capacity.tasks);
40 try std.testing.expectEqual(@as(usize, 100), server.connection_slots.len);
41 try std.testing.expect(server.config.tcp_no_delay);
42 try std.testing.expectEqual(
43 http.ConnectionInputCapacity{
44 .connection_count = 100,
45 .bytes_per_connection = 321,
46 .storage_bytes = 32_100,
47 },
48 server.connectionInputCapacity(),
49 );
50 try std.testing.expectEqual(
51 http.RequestCapacity{
52 .request_count = 100,
53 .header_count_per_request = 3,
54 .header_line_bytes = 29,
55 .body_bytes_per_request = 17,
56 .header_count = 300,
57 .header_bytes = 300 * @sizeOf(http.RequestHeader),
58 .body_bytes = 1700,
59 .storage_bytes = 300 * @sizeOf(http.RequestHeader) + 1700,
60 },
61 server.requestCapacity(),
62 );
63 try std.testing.expectEqual(
64 http.ResponseCapacity{
65 .response_count = 100,
66 .header_count_per_response = 5,
67 .head_bytes_per_response = 257,
68 .header_count = 500,
69 .header_bytes = 500 * @sizeOf(http.ResponseHeader),
70 .head_bytes = 25_700,
71 .storage_bytes = 500 * @sizeOf(http.ResponseHeader) + 25_700,
72 },
73 server.responseCapacity(),
74 );
75 try std.testing.expect(server.listener != null);
76 }
77
78 fn testMarkConnection(flag: *std.atomic.Value(bool), _: *Connection) void {
79 flag.store(true, .release);
80 }
81
82 fn testListen(server: *Server) void {
83 server.listen() catch |err| std.debug.panic("server listen failed: {s}", .{@errorName(err)});
84 }
85
86 test "Server stop unblocks listen" {
87 const server = try Server.init(std.testing.allocator, .{
88 .port = 0,
89 .num_workers = 1,
90 });
91
92 const port = try sys.socketPort(server.listener.?);
93 var connection_handled = std.atomic.Value(bool).init(false);
94 server.setHandler(&connection_handled, testMarkConnection);
95 const listen_thread = try thread.spawn(testListen, .{server});
96
97 const client = try sys.tcpStreamSocket(.{});
98 defer sys.close(client);
99 const server_addr = try sys.ip4AddressForHost("127.0.0.1", port);
100 try sys.connectIp4(client, server_addr);
101 _ = try sys.send(client, "x", 0);
102
103 var waited: usize = 0;
104 while (!connection_handled.load(.acquire) and waited < 5000) : (waited += 1) {
105 sleepMillis(1);
106 }
107 try std.testing.expect(connection_handled.load(.acquire));
108
109 server.stop();
110 listen_thread.join();
111 server.deinit();
112 }
113
114 test "Server accepts connection" {
115 comptime {
116 @stardustClaim(
117 @import("alloc_phase").capacity.witness(@import("../root.zig").ConnectionInputStorage, "http_connection_input_accept"),
118 null,
119 null,
120 null,
121 null,
122 null,
123 null,
124 );
125 }
126
127 var failing = std.testing.FailingAllocator.init(std.testing.allocator, .{});
128 const server = try Server.init(failing.allocator(), .{
129 .port = 0,
130 .num_workers = 2,
131 });
132 failing.fail_index = failing.alloc_index;
133
134 const port = try sys.socketPort(server.listener.?);
135 var connection_handled = std.atomic.Value(bool).init(false);
136 server.setHandler(&connection_handled, testMarkConnection);
137 const listen_thread = try thread.spawn(testListen, .{server});
138
139 const client = try sys.tcpStreamSocket(.{});
140 defer sys.close(client);
141 const server_addr = try sys.ip4AddressForHost("127.0.0.1", port);
142 try sys.connectIp4(client, server_addr);
143 _ = try sys.send(client, "x", 0);
144
145 var waited: usize = 0;
146 while (!connection_handled.load(.acquire) and waited < 5000) : (waited += 1) {
147 sleepMillis(1);
148 }
149
150 server.stop();
151 listen_thread.join();
152 server.deinit();
153 try std.testing.expect(connection_handled.load(.acquire));
154 try std.testing.expect(!failing.has_induced_failure);
155 }
156
157 const CapacityGate = struct {
158 release: std.atomic.Value(bool) = std.atomic.Value(bool).init(false),
159 handled: std.atomic.Value(usize) = std.atomic.Value(usize).init(0),
160 };
161
162 fn holdConnection(gate: *CapacityGate, conn: *Connection) void {
163 _ = gate.handled.fetchAdd(1, .acq_rel);
164 var rounds: usize = 0;
165 while (!gate.release.load(.acquire) and rounds < 5000) : (rounds += 1) {
166 sleepMillis(1);
167 }
168 conn.markClosing();
169 }
170
171 fn waitForCount(value: *const std.atomic.Value(usize), target: usize) bool {
172 var rounds: usize = 0;
173 while (value.load(.acquire) < target and rounds < 5000) : (rounds += 1) {
174 sleepMillis(1);
175 }
176 return value.load(.acquire) >= target;
177 }
178
179 fn waitForNoConnections(server: *Server) bool {
180 var rounds: usize = 0;
181 while (server.connectionCount() != 0 and rounds < 5000) : (rounds += 1) {
182 sleepMillis(1);
183 }
184 return server.connectionCount() == 0;
185 }
186
187 fn peerIsClosed(socket: sys.Socket) !bool {
188 if (!try sys.pollReadable(socket, 2000)) return false;
189 var byte: [1]u8 = undefined;
190 const received = sys.recv(socket, &byte, 0) catch |err| switch (err) {
191 error.ConnectionResetByPeer => return true,
192 else => return err,
193 };
194 return received == 0;
195 }
196
197 test "Server closes overload and reuses bounded connection slots" {
198 const server = try Server.init(std.testing.allocator, .{
199 .port = 0,
200 .num_workers = 1,
201 .max_connections = 1,
202 });
203 var gate = CapacityGate{};
204 server.setHandler(&gate, holdConnection);
205 const port = try sys.socketPort(server.listener.?);
206 const address = try sys.ip4AddressForHost("127.0.0.1", port);
207 const listen_thread = try thread.spawn(testListen, .{server});
208 defer {
209 gate.release.store(true, .release);
210 server.stop();
211 listen_thread.join();
212 server.deinit();
213 }
214
215 const first = try sys.tcpStreamSocket(.{});
216 defer sys.close(first);
217 try sys.connectIp4(first, address);
218 _ = try sys.send(first, "x", 0);
219 try std.testing.expect(waitForCount(&gate.handled, 1));
220 try std.testing.expectEqual(@as(usize, 1), server.connectionCount());
221
222 const overload = try sys.tcpStreamSocket(.{});
223 defer sys.close(overload);
224 try sys.connectIp4(overload, address);
225 try std.testing.expect(try peerIsClosed(overload));
226 try std.testing.expectEqual(@as(usize, 1), server.connectionCount());
227
228 gate.release.store(true, .release);
229 try std.testing.expect(waitForNoConnections(server));
230
231 const reused = try sys.tcpStreamSocket(.{});
232 defer sys.close(reused);
233 try sys.connectIp4(reused, address);
234 _ = try sys.send(reused, "x", 0);
235 try std.testing.expect(waitForCount(&gate.handled, 2));
236 try std.testing.expect(waitForNoConnections(server));
237 }
238
239 const SchedulingState = struct {
240 requests: std.atomic.Value(usize) = std.atomic.Value(usize).init(0),
241 websocket_messages: std.atomic.Value(usize) = std.atomic.Value(usize).init(0),
242 };
243
244 const SchedulingRouter = http.Router(*SchedulingState);
245 const Request = http.Request;
246 const Response = http.Response;
247 const WebSocket = http.WebSocket;
248 const websocket_limits: http.WebSocketLimits = .{
249 .frame_payload_bytes = 4096,
250 .message_payload_bytes = 4096,
251 };
252
253 fn schedulingHttp(state: *SchedulingState, _: *Request, response: *Response) !void {
254 _ = state.requests.fetchAdd(1, .acq_rel);
255 response.status = 200;
256 response.status_text = "OK";
257 response.body = "ready";
258 }
259
260 fn schedulingUpgrade(_: *SchedulingState, _: *Request, response: *Response) !void {
261 response.status = 200;
262 response.status_text = "OK";
263 }
264
265 fn schedulingWebSocket(state: *SchedulingState, websocket: *WebSocket) !void {
266 while (try websocket.receive()) |received| {
267 _ = received;
268 _ = state.websocket_messages.fetchAdd(1, .acq_rel);
269 }
270 }
271
272 fn schedulingConnection(router: *SchedulingRouter, connection: *Connection) void {
273 while (router.handleRequest(connection) catch false) {}
274 }
275
276 fn connectClient(address: sys.Ip4Address) !sys.Socket {
277 const socket = try sys.tcpStreamSocket(.{});
278 errdefer sys.close(socket);
279 try sys.connectIp4(socket, address);
280 return socket;
281 }
282
283 fn expectSocketText(socket: sys.Socket, expected: []const u8) !void {
284 var bytes: [1024]u8 = undefined;
285 var length: usize = 0;
286 while (length < bytes.len) {
287 if (std.mem.indexOf(u8, bytes[0..length], expected) != null) return;
288 if (!try sys.pollReadable(socket, 2000)) return error.ResponseTimeout;
289 const read = try sys.recv(socket, bytes[length..], 0);
290 if (read == 0) break;
291 length += read;
292 }
293 try std.testing.expect(std.mem.indexOf(u8, bytes[0..length], expected) != null);
294 }
295
296 fn waitForValue(value: *const std.atomic.Value(usize), expected: usize) bool {
297 var rounds: usize = 0;
298 while (value.load(.acquire) < expected and rounds < 2000) : (rounds += 1) sleepMillis(1);
299 return value.load(.acquire) >= expected;
300 }
301
302 test "Server schedules fresh requests beyond idle keepalive count" {
303 const idle_count = 32;
304 var state = SchedulingState{};
305 var router = SchedulingRouter.init(std.testing.allocator, &state);
306 defer router.deinit();
307 try router.get("/work", schedulingHttp);
308
309 const server = try Server.init(std.testing.allocator, .{
310 .port = 0,
311 .num_workers = 4,
312 .max_connections = idle_count + 2,
313 });
314 server.setHandler(&router, schedulingConnection);
315 const port = try sys.socketPort(server.listener.?);
316 const address = try sys.ip4AddressForHost("127.0.0.1", port);
317 const listen_thread = try thread.spawn(testListen, .{server});
318 var idle: [idle_count]sys.Socket = undefined;
319 var idle_open: usize = 0;
320 defer {
321 for (idle[0..idle_open]) |socket| sys.close(socket);
322 server.stop();
323 listen_thread.join();
324 server.deinit();
325 }
326
327 for (&idle) |*socket| {
328 socket.* = try connectClient(address);
329 idle_open += 1;
330 _ = try sys.send(socket.*, "GET /work HTTP/1.1\r\nHost: example.test\r\n\r\n", 0);
331 try expectSocketText(socket.*, "\r\n\r\nready");
332 }
333
334 const active = try connectClient(address);
335 defer sys.close(active);
336 _ = try sys.send(active, "GET /work HTTP/1.1\r\nHost: example.test\r\nConnection: close\r\n\r\n", 0);
337 try expectSocketText(active, "\r\n\r\nready");
338 try std.testing.expect(waitForValue(&state.requests, idle_count + 1));
339 }
340
341 test "Server schedules around a partial HTTP request" {
342 var state = SchedulingState{};
343 var router = SchedulingRouter.init(std.testing.allocator, &state);
344 defer router.deinit();
345 try router.get("/work", schedulingHttp);
346
347 const server = try Server.init(std.testing.allocator, .{
348 .port = 0,
349 .num_workers = 1,
350 .max_connections = 3,
351 });
352 server.setHandler(&router, schedulingConnection);
353 const port = try sys.socketPort(server.listener.?);
354 const address = try sys.ip4AddressForHost("127.0.0.1", port);
355 const listen_thread = try thread.spawn(testListen, .{server});
356 defer {
357 server.stop();
358 listen_thread.join();
359 server.deinit();
360 }
361
362 const partial = try connectClient(address);
363 defer sys.close(partial);
364 _ = try sys.send(partial, "GET /work HTTP/1.1\r\nHost: exam", 0);
365
366 const fresh = try connectClient(address);
367 defer sys.close(fresh);
368 _ = try sys.send(fresh, "GET /work HTTP/1.1\r\nHost: example.test\r\nConnection: close\r\n\r\n", 0);
369 try expectSocketText(fresh, "\r\n\r\nready");
370
371 _ = try sys.send(partial, "ple.test\r\nConnection: close\r\n\r\n", 0);
372 try expectSocketText(partial, "\r\n\r\nready");
373 try std.testing.expect(waitForValue(&state.requests, 2));
374 }
375
376 test "Server expires idle connections without occupying a worker" {
377 const server = try Server.init(std.testing.allocator, .{
378 .port = 0,
379 .num_workers = 1,
380 .max_connections = 2,
381 .read_timeout_ms = 20,
382 });
383 var handled = std.atomic.Value(bool).init(false);
384 server.setHandler(&handled, testMarkConnection);
385 const port = try sys.socketPort(server.listener.?);
386 const address = try sys.ip4AddressForHost("127.0.0.1", port);
387 const listen_thread = try thread.spawn(testListen, .{server});
388 defer {
389 server.stop();
390 listen_thread.join();
391 server.deinit();
392 }
393
394 const idle = try connectClient(address);
395 defer sys.close(idle);
396 try std.testing.expect(try peerIsClosed(idle));
397 try std.testing.expect(!handled.load(.acquire));
398 try std.testing.expect(waitForNoConnections(server));
399 }
400
401 test "Server schedules HTTP and WebSocket turns beyond idle upgrades" {
402 const idle_count = 16;
403 var state = SchedulingState{};
404 var router = SchedulingRouter.init(std.testing.allocator, &state);
405 defer router.deinit();
406 try router.get("/work", schedulingHttp);
407 try router.websocket("/ws", websocket_limits, schedulingUpgrade, schedulingWebSocket);
408
409 const server = try Server.init(std.testing.allocator, .{
410 .port = 0,
411 .num_workers = 4,
412 .max_connections = idle_count + 2,
413 });
414 server.setHandler(&router, schedulingConnection);
415 const port = try sys.socketPort(server.listener.?);
416 const address = try sys.ip4AddressForHost("127.0.0.1", port);
417 const listen_thread = try thread.spawn(testListen, .{server});
418 var idle: [idle_count]sys.Socket = undefined;
419 var idle_open: usize = 0;
420 defer {
421 for (idle[0..idle_open]) |socket| sys.close(socket);
422 server.stop();
423 listen_thread.join();
424 server.deinit();
425 }
426
427 const handshake =
428 "GET /ws HTTP/1.1\r\n" ++
429 "Host: example.test\r\n" ++
430 "Upgrade: websocket\r\n" ++
431 "Connection: Upgrade\r\n" ++
432 "Sec-WebSocket-Version: 13\r\n" ++
433 "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n\r\n";
434 for (&idle) |*socket| {
435 socket.* = try connectClient(address);
436 idle_open += 1;
437 _ = try sys.send(socket.*, handshake, 0);
438 try expectSocketText(socket.*, "101 Switching Protocols");
439 }
440
441 const active = try connectClient(address);
442 defer sys.close(active);
443 _ = try sys.send(active, "GET /work HTTP/1.1\r\nHost: example.test\r\nConnection: close\r\n\r\n", 0);
444 try expectSocketText(active, "\r\n\r\nready");
445
446 const masked_text = [_]u8{ 0x81, 0x81, 1, 2, 3, 4, 'x' ^ 1 };
447 _ = try sys.send(idle[0], &masked_text, 0);
448 try std.testing.expect(waitForValue(&state.websocket_messages, 1));
449 }
450
451 fn sleepMillis(ms: u64) void {
452 time.sleepMilliseconds(ms);
453 }