lib/http/src/connection.zig
daab053ee43316e1809a84551d573ddd1e5bf3d2
1 const std = @import("std");
2 const alloc_phase = @import("alloc_phase");
3 const message = @import("message.zig");
4 const response = @import("response.zig");
5 const sys_root = @import("sys");
6 const sys = sys_root.net;
7 const time = sys_root.time;
8
9 pub const default_input_bytes_per_connection: usize = 64 * 1024;
10
11 pub const InputLimits = struct {
12 connection_count: usize,
13 bytes_per_connection: usize,
14 };
15
16 pub const InputCapacity = struct {
17 connection_count: usize,
18 bytes_per_connection: usize,
19 storage_bytes: usize,
20
21 pub fn derive(limits: InputLimits) error{CapacityOverflow}!InputCapacity {
22 const storage_bytes = try alloc_phase.capacity.mul(
23 usize,
24 limits.connection_count,
25 limits.bytes_per_connection,
26 );
27 return .{
28 .connection_count = limits.connection_count,
29 .bytes_per_connection = limits.bytes_per_connection,
30 .storage_bytes = storage_bytes,
31 };
32 }
33 };
34
35 pub const InputExhaustion = error{
36 ConnectionCapacityExceeded,
37 InputCapacityExceeded,
38 };
39
40 const StorageLimits = InputLimits;
41 const StorageCapacity = InputCapacity;
42
43 pub const InputStorage = struct {
44 phase: alloc_phase.capacity.Phase,
45 capacity: StorageCapacity,
46 bytes: []u8,
47
48 pub const Limits: type = StorageLimits;
49 pub const Capacity: type = StorageCapacity;
50 pub const Exhaustion: type = InputExhaustion;
51 pub const InitError = std.mem.Allocator.Error || error{CapacityOverflow};
52
53 pub const claim: alloc_phase.capacity.Declaration = .{
54 .source = .{
55 .id = "http.connection_input_storage",
56 .kind = .phase_static,
57 .limit_source = .caller,
58 .storage = .{
59 .covered = &.{
60 .{
61 .id = "fixed_retained_input_byte_region_for_every_connection_slot",
62 .lifetime = .steady,
63 .detail = "fixed retained-input byte region for every connection slot",
64 },
65 },
66 .excluded = &.{
67 "request header-map buckets and decoded chunked-body output",
68 "WebSocket storage, protocol sessions, handlers, and response output",
69 "socket and kernel queues",
70 },
71 },
72 .capacity = .{
73 .inputs = &.{
74 alloc_phase.capacity.bindInput(Limits, "connection_count", "connection_count"),
75 alloc_phase.capacity.bindInput(Limits, "bytes_per_connection", "bytes_per_connection"),
76 },
77 .type_selectors = &.{},
78 .nodes = &.{
79 .{ .input = 0 },
80 .{ .input = 1 },
81 .{ .product = .{ .left = 0, .right = 1 } },
82 },
83 .assertions = &.{.{
84 .scope = .closure_total,
85 .measure = .retained,
86 .relation = .exact,
87 .expression = 2,
88 }},
89 },
90 .overload = .{
91 .kind = .reject_before_mutation,
92 .detail = "full input preserves retained bytes and leaves foreign input unread",
93 },
94 .risks = .{
95 .transitive = .{
96 .status = .open,
97 .detail = "request parsing and installed protocol callbacks own separate storage",
98 },
99 .foreign = .{
100 .status = .open,
101 .detail = "input admission reads from socket and kernel storage",
102 },
103 },
104 .obligations = &.{
105 .{ .key = "http_connection_input_capacity", .role = .capacity_model },
106 .{ .key = "http_connection_input_oom_retry", .role = .custom },
107 .{ .key = "http_connection_input_partition", .role = .custom },
108 .{ .key = "http_connection_input_sealed", .role = .overload },
109 .{ .key = "http_connection_input_partial", .role = .custom },
110 .{ .key = "http_connection_input_pipeline", .role = .custom },
111 .{ .key = "http_connection_input_overload", .role = .custom },
112 .{ .key = "http_connection_input_accept", .role = .custom },
113 },
114 },
115 .bindings = .{
116 .owner = @This(),
117 .seal = .{
118 .family = alloc_phase.capacity.selector(@This().activate),
119 .premise = .{
120 .class = .checked_semantic_fact,
121 .authority = .checker,
122 },
123 },
124 .teardown = .{
125 .family = alloc_phase.capacity.selector(@This().deinit),
126 .premise = .{
127 .class = .checked_semantic_fact,
128 .authority = .checker,
129 },
130 },
131 },
132 };
133
134 pub fn init(allocator: std.mem.Allocator, limits: StorageLimits) InitError!InputStorage {
135 const capacity = try StorageCapacity.derive(limits);
136 const bytes = if (capacity.storage_bytes == 0)
137 @as([]u8, &.{})
138 else
139 try allocator.alloc(u8, capacity.storage_bytes);
140 return .{
141 .phase = .initialization,
142 .capacity = capacity,
143 .bytes = bytes,
144 };
145 }
146
147 pub fn activate(self: *InputStorage) void {
148 std.debug.assert(self.phase == .initialization);
149 std.debug.assert(self.bytes.len == self.capacity.storage_bytes);
150 self.phase = .steady;
151 }
152
153 pub fn connection(self: *InputStorage, index: usize) Exhaustion![]u8 {
154 std.debug.assert(self.phase == .steady);
155 if (index >= self.capacity.connection_count) {
156 return error.ConnectionCapacityExceeded;
157 }
158 const start = index * self.capacity.bytes_per_connection;
159 return self.bytes[start..][0..self.capacity.bytes_per_connection];
160 }
161
162 pub fn deinit(self: *InputStorage, allocator: std.mem.Allocator) void {
163 std.debug.assert(self.phase != .teardown);
164 std.debug.assert(self.bytes.len == self.capacity.storage_bytes);
165 self.phase = .teardown;
166 if (self.bytes.len != 0) allocator.free(self.bytes);
167 self.bytes = &.{};
168 }
169 };
170
171 comptime {
172 alloc_phase.capacity.requireAllocatorRejectingOwnerShape(InputStorage);
173 }
174
175 const ConnectionInputStorage = InputStorage;
176
177 pub const InputStatus = struct {
178 capacity_rejections: u64 = 0,
179 };
180
181 pub const Connection = struct {
182 id: usize,
183 socket: sys.Socket,
184 state: std.atomic.Value(State),
185 transport: std.atomic.Value(TransportState),
186 boot_clock: time.BootClock,
187 last_activity_ns: std.atomic.Value(u64),
188 input_storage: []u8,
189 input_length: usize,
190 request_scratch: message.RequestScratch,
191 response_scratch: response.Scratch,
192 input_capacity_rejections: std.atomic.Value(u64),
193 write_timeout_ms: ?u32,
194 protocol: ?Protocol,
195 turn_disposition: TurnDisposition,
196
197 pub const State = enum(u8) {
198 http,
199 websocket,
200 closing,
201 closed,
202 };
203
204 const TurnDisposition = enum {
205 close,
206 wait_for_read,
207 consume_buffered_input,
208 };
209
210 const TransportState = enum(u8) {
211 open,
212 interrupting,
213 interrupted,
214 closed,
215 };
216
217 const Protocol = struct {
218 context: *anyopaque,
219 drive: *const fn (*anyopaque) anyerror!void,
220 deinit: *const fn (*anyopaque) void,
221 };
222
223 pub const InputLimits: type = StorageLimits;
224 pub const InputCapacity: type = StorageCapacity;
225 pub const InputStorage: type = ConnectionInputStorage;
226
227 pub fn init(
228 id: usize,
229 socket: sys.Socket,
230 input_storage: []u8,
231 request_scratch: message.RequestScratch,
232 response_scratch: response.Scratch,
233 boot_clock: time.BootClock,
234 started_at: time.BootInstant,
235 ) Connection {
236 return .{
237 .id = id,
238 .socket = socket,
239 .state = std.atomic.Value(State).init(.http),
240 .transport = std.atomic.Value(TransportState).init(.open),
241 .boot_clock = boot_clock,
242 .last_activity_ns = std.atomic.Value(u64).init(started_at.asNanoseconds()),
243 .input_storage = input_storage,
244 .input_length = 0,
245 .request_scratch = request_scratch,
246 .response_scratch = response_scratch,
247 .input_capacity_rejections = std.atomic.Value(u64).init(0),
248 .write_timeout_ms = null,
249 .protocol = null,
250 .turn_disposition = .close,
251 };
252 }
253
254 pub fn deinit(self: *Connection) void {
255 self.close();
256 if (self.protocol) |protocol| protocol.deinit(protocol.context);
257 self.protocol = null;
258 self.input_length = 0;
259 self.input_storage = &.{};
260 self.request_scratch = .{
261 .headers = &.{},
262 .body = &.{},
263 .header_line_bytes = 0,
264 };
265 self.response_scratch = .{ .headers = &.{}, .head = &.{} };
266 }
267
268 pub fn close(self: *Connection) void {
269 while (true) {
270 const current = self.transport.load(.acquire);
271 switch (current) {
272 .closed => return,
273 .interrupting => {
274 std.atomic.spinLoopHint();
275 continue;
276 },
277 .open, .interrupted => {},
278 }
279 if (self.transport.cmpxchgWeak(current, .closed, .acq_rel, .acquire) == null) break;
280 }
281 sys.close(self.socket);
282 self.state.store(.closed, .release);
283 }
284
285 pub fn takeSocket(self: *Connection) sys.Socket {
286 std.debug.assert(self.protocol == null);
287 std.debug.assert(self.transport.cmpxchgStrong(
288 .open,
289 .closed,
290 .acq_rel,
291 .acquire,
292 ) == null);
293 self.state.store(.closed, .release);
294 return self.socket;
295 }
296
297 pub fn interrupt(self: *Connection) void {
298 if (self.transport.cmpxchgStrong(.open, .interrupting, .acq_rel, .acquire) != null) return;
299 self.markClosing();
300 sys.shutdownReadWrite(self.socket);
301 self.transport.store(.interrupted, .release);
302 }
303
304 pub fn transportClosed(self: *Connection) void {
305 self.transport.store(.closed, .release);
306 self.state.store(.closed, .release);
307 }
308
309 pub fn currentState(self: *const Connection) State {
310 return self.state.load(.acquire);
311 }
312
313 pub fn markClosing(self: *Connection) void {
314 if (self.transport.load(.acquire) != .closed) self.state.store(.closing, .release);
315 }
316
317 pub fn markWebSocket(self: *Connection) void {
318 std.debug.assert(self.currentState() == .http);
319 self.state.store(.websocket, .release);
320 self.waitForInput();
321 }
322
323 pub fn beginTurn(self: *Connection) void {
324 self.turn_disposition = .close;
325 }
326
327 pub fn waitForRead(self: *Connection) void {
328 self.turn_disposition = .wait_for_read;
329 }
330
331 pub fn waitForInput(self: *Connection) void {
332 self.turn_disposition = .consume_buffered_input;
333 }
334
335 pub fn shouldWaitForRead(self: *const Connection) bool {
336 return self.turn_disposition != .close;
337 }
338
339 pub fn shouldConsumeBufferedInput(self: *const Connection) bool {
340 return self.turn_disposition == .consume_buffered_input;
341 }
342
343 pub fn lastActivity(self: *const Connection) time.BootInstant {
344 return .fromNanoseconds(self.last_activity_ns.load(.acquire));
345 }
346
347 pub fn hasBufferedInput(self: *const Connection) bool {
348 return self.input_length != 0;
349 }
350
351 pub fn bufferedInput(self: *const Connection) []const u8 {
352 return self.input_storage[0..self.input_length];
353 }
354
355 pub fn inputCapacity(self: *const Connection) usize {
356 return self.input_storage.len;
357 }
358
359 pub fn requestScratch(self: *Connection) message.RequestScratch {
360 return self.request_scratch;
361 }
362
363 pub fn responseScratch(self: *Connection) response.Scratch {
364 return self.response_scratch;
365 }
366
367 pub fn inputStatus(self: *const Connection) InputStatus {
368 return .{
369 .capacity_rejections = self.input_capacity_rejections.load(.acquire),
370 };
371 }
372
373 pub fn installProtocol(
374 self: *Connection,
375 context: anytype,
376 comptime drive: anytype,
377 comptime deinit_protocol: anytype,
378 ) void {
379 const Context = @TypeOf(context);
380 const info = @typeInfo(Context);
381 if (info != .pointer or info.pointer.size != .one) {
382 @compileError("connection protocol context must be a single-item pointer");
383 }
384 const Callbacks = struct {
385 fn run(erased: *anyopaque) anyerror!void {
386 const typed: Context = @ptrCast(@alignCast(erased));
387 return drive(typed);
388 }
389
390 fn destroy(erased: *anyopaque) void {
391 const typed: Context = @ptrCast(@alignCast(erased));
392 deinit_protocol(typed);
393 }
394 };
395 std.debug.assert(self.protocol == null);
396 self.protocol = .{
397 .context = @ptrCast(@constCast(context)),
398 .drive = Callbacks.run,
399 .deinit = Callbacks.destroy,
400 };
401 self.state.store(.websocket, .release);
402 self.waitForInput();
403 }
404
405 pub fn driveProtocol(self: *Connection) !void {
406 const protocol = self.protocol orelse return error.ProtocolNotInstalled;
407 try protocol.drive(protocol.context);
408 }
409
410 pub fn read(self: *Connection, buf: []u8) !usize {
411 if (self.transport.load(.acquire) == .closed) {
412 return error.ConnectionClosed;
413 }
414 if (buf.len == 0) return 0;
415 if (self.input_length > 0) {
416 const n = @min(buf.len, self.input_length);
417 @memcpy(buf[0..n], self.input_storage[0..n]);
418 self.consumeBufferedInput(n);
419 try self.noteActivity();
420 return n;
421 }
422 return self.readSocket(buf);
423 }
424
425 pub fn bufferInput(self: *Connection) !usize {
426 if (self.input_length == self.input_storage.len) {
427 self.recordInputCapacityRejection();
428 return error.InputCapacityExceeded;
429 }
430 const n = try self.readSocket(self.input_storage[self.input_length..]);
431 self.input_length += n;
432 return n;
433 }
434
435 pub fn retainInput(self: *Connection, data: []const u8) InputExhaustion!void {
436 if (data.len > self.input_storage.len - self.input_length) {
437 self.recordInputCapacityRejection();
438 return error.InputCapacityExceeded;
439 }
440 @memcpy(self.input_storage[self.input_length..][0..data.len], data);
441 self.input_length += data.len;
442 }
443
444 pub fn consumeBufferedInput(self: *Connection, count: usize) void {
445 std.debug.assert(count <= self.input_length);
446 const remaining = self.input_length - count;
447 std.mem.copyForwards(
448 u8,
449 self.input_storage[0..remaining],
450 self.input_storage[count..self.input_length],
451 );
452 self.input_length = remaining;
453 }
454
455 fn recordInputCapacityRejection(self: *Connection) void {
456 var current = self.input_capacity_rejections.load(.acquire);
457 while (current != std.math.maxInt(u64)) {
458 if (self.input_capacity_rejections.cmpxchgWeak(
459 current,
460 current + 1,
461 .acq_rel,
462 .acquire,
463 )) |observed| {
464 current = observed;
465 } else {
466 return;
467 }
468 }
469 }
470
471 fn readSocket(self: *Connection, buf: []u8) !usize {
472 std.debug.assert(buf.len != 0);
473 if (self.transport.load(.acquire) == .closed) {
474 return error.ConnectionClosed;
475 }
476 const n = sys.recv(self.socket, buf, 0) catch |err| switch (err) {
477 error.BadFileDescriptor => {
478 self.markClosing();
479 return error.ConnectionClosed;
480 },
481 error.ConnectionResetByPeer,
482 error.ConnectionTimedOut,
483 => {
484 self.markClosing();
485 return error.ConnectionClosed;
486 },
487 error.WouldBlock => return error.WouldBlock,
488 else => return err,
489 };
490 if (n == 0) {
491 self.markClosing();
492 return error.ConnectionClosed;
493 }
494 try self.noteActivity();
495 return n;
496 }
497
498 pub fn write(self: *Connection, data: []const u8) !void {
499 if (self.transport.load(.acquire) == .closed) {
500 return error.ConnectionClosed;
501 }
502 var sent: usize = 0;
503 while (sent < data.len) {
504 const n = sys.sendNoSignal(self.socket, data[sent..]) catch |err| switch (err) {
505 error.ConnectionResetByPeer,
506 error.BrokenPipe,
507 => {
508 self.markClosing();
509 return error.ConnectionClosed;
510 },
511 error.WouldBlock => {
512 try self.waitWritable();
513 continue;
514 },
515 else => return err,
516 };
517 if (n == 0) {
518 self.markClosing();
519 return error.ConnectionClosed;
520 }
521 sent += n;
522 try self.noteActivity();
523 }
524 }
525
526 fn waitWritable(self: *Connection) !void {
527 const ready = try sys.pollWritable(self.socket, self.writePollTimeoutMs());
528 if (!ready) {
529 self.markClosing();
530 return error.ConnectionTimedOut;
531 }
532 }
533
534 fn noteActivity(self: *Connection) time.ClockError!void {
535 const now = try self.boot_clock.now();
536 self.last_activity_ns.store(now.asNanoseconds(), .release);
537 }
538
539 fn writePollTimeoutMs(self: *const Connection) i32 {
540 const timeout = self.write_timeout_ms orelse return -1;
541 const max_timeout: u32 = @intCast(std.math.maxInt(i32));
542 return @intCast(@min(timeout, max_timeout));
543 }
544
545 pub fn setReadTimeout(self: *Connection, timeout_ms: u32) !void {
546 try sys.setReadTimeout(self.socket, timeout_ms);
547 }
548
549 pub fn setWriteTimeout(self: *Connection, timeout_ms: u32) !void {
550 try sys.setWriteTimeout(self.socket, timeout_ms);
551 self.write_timeout_ms = if (timeout_ms == 0) null else timeout_ms;
552 }
553 };
554
555 fn testSocketPair() ![2]sys.Socket {
556 return sys.socketPairUnixStream();
557 }
558
559 fn emptyRequestScratch() message.RequestScratch {
560 return .{ .headers = &.{}, .body = &.{}, .header_line_bytes = 0 };
561 }
562
563 fn emptyResponseScratch() response.Scratch {
564 return .{ .headers = &.{}, .head = &.{} };
565 }
566
567 fn fillSocketWriteBuffer(socket: sys.Socket) !void {
568 var chunk: [4096]u8 = undefined;
569 @memset(&chunk, 0xAA);
570
571 while (true) {
572 _ = sys.sendNoSignal(socket, &chunk) catch |err| switch (err) {
573 error.WouldBlock => return,
574 else => return err,
575 };
576 }
577 }
578
579 fn independentInputCapacity(limits: InputLimits) error{CapacityOverflow}!InputCapacity {
580 const storage_bytes = @as(u128, limits.connection_count) * limits.bytes_per_connection;
581 if (storage_bytes > std.math.maxInt(usize)) return error.CapacityOverflow;
582 return .{
583 .connection_count = limits.connection_count,
584 .bytes_per_connection = limits.bytes_per_connection,
585 .storage_bytes = @intCast(storage_bytes),
586 };
587 }
588
589 test "Connection input capacity matches independent arithmetic" {
590 comptime {
591 @stardustClaim(
592 @import("alloc_phase").capacity.witness(InputStorage, "http_connection_input_capacity"),
593 null,
594 null,
595 null,
596 null,
597 null,
598 null,
599 );
600 }
601
602 for (0..65) |connection_count| {
603 for (0..65) |bytes_per_connection| {
604 const limits = InputLimits{
605 .connection_count = connection_count,
606 .bytes_per_connection = bytes_per_connection,
607 };
608 try std.testing.expectEqual(
609 try independentInputCapacity(limits),
610 try InputCapacity.derive(limits),
611 );
612 }
613 }
614 try std.testing.expectError(
615 error.CapacityOverflow,
616 InputCapacity.derive(.{
617 .connection_count = 2,
618 .bytes_per_connection = std.math.maxInt(usize),
619 }),
620 );
621 }
622
623 fn checkInputStorageInitFailures(allocator: std.mem.Allocator) !void {
624 var storage = try InputStorage.init(allocator, .{
625 .connection_count = 3,
626 .bytes_per_connection = 17,
627 });
628 storage.deinit(allocator);
629 }
630
631 test "Connection input storage retries after every allocation failure" {
632 comptime {
633 @stardustClaim(
634 @import("alloc_phase").capacity.witness(InputStorage, "http_connection_input_oom_retry"),
635 null,
636 null,
637 null,
638 null,
639 null,
640 null,
641 );
642 }
643
644 try std.testing.checkAllAllocationFailures(
645 std.testing.allocator,
646 checkInputStorageInitFailures,
647 .{},
648 );
649 }
650
651 test "Connection input storage partitions reusable slots" {
652 comptime {
653 @stardustClaim(
654 @import("alloc_phase").capacity.witness(InputStorage, "http_connection_input_partition"),
655 null,
656 null,
657 null,
658 null,
659 null,
660 null,
661 );
662 }
663
664 var storage = try InputStorage.init(std.testing.allocator, .{
665 .connection_count = 2,
666 .bytes_per_connection = 4,
667 });
668 defer storage.deinit(std.testing.allocator);
669 storage.activate();
670
671 const first = try storage.connection(0);
672 const second = try storage.connection(1);
673 try std.testing.expect(first.ptr + first.len == second.ptr);
674 @memset(first, 0x11);
675 @memset(second, 0x22);
676 try std.testing.expectEqualSlices(u8, &.{ 0x11, 0x11, 0x11, 0x11 }, first);
677 try std.testing.expectEqualSlices(u8, &.{ 0x22, 0x22, 0x22, 0x22 }, second);
678 try std.testing.expect((try storage.connection(0)).ptr == first.ptr);
679 }
680
681 test "Connection input storage seals before admission" {
682 comptime {
683 @stardustClaim(
684 @import("alloc_phase").capacity.witness(InputStorage, "http_connection_input_sealed"),
685 null,
686 null,
687 null,
688 null,
689 null,
690 null,
691 );
692 }
693
694 var phase_allocator = try alloc_phase.SealedPhaseAllocator.init(std.testing.allocator);
695 var storage = InputStorage.init(
696 phase_allocator.initializationAllocator(),
697 .{ .connection_count = 1, .bytes_per_connection = 4 },
698 ) catch |err| {
699 phase_allocator.abortInitialization();
700 phase_allocator.deinit();
701 return err;
702 };
703 errdefer {
704 if (phase_allocator.phase() == .initialization) phase_allocator.abortInitialization();
705 if (phase_allocator.phase() == .steady) phase_allocator.beginTeardown();
706 if (storage.phase != .teardown) storage.deinit(phase_allocator.teardownAllocator());
707 phase_allocator.deinit();
708 }
709
710 const pointer = storage.bytes.ptr;
711 const capacity = storage.capacity;
712 phase_allocator.seal();
713 storage.activate();
714 try std.testing.expectError(
715 error.ConnectionCapacityExceeded,
716 storage.connection(1),
717 );
718
719 const sockets = try testSocketPair();
720 defer sys.close(sockets[1]);
721 var connection = Connection.init(
722 1,
723 sockets[0],
724 try storage.connection(0),
725 emptyRequestScratch(),
726 emptyResponseScratch(),
727 .system(),
728 try time.bootNow(),
729 );
730 errdefer connection.deinit();
731
732 try std.testing.expectEqual(@as(usize, 5), try sys.send(sockets[1], "abcde", 0));
733 try std.testing.expectEqual(@as(usize, 4), try connection.bufferInput());
734 try std.testing.expectEqualStrings("abcd", connection.bufferedInput());
735 try std.testing.expectError(error.InputCapacityExceeded, connection.bufferInput());
736 try std.testing.expectError(error.InputCapacityExceeded, connection.retainInput("x"));
737 try std.testing.expectEqualStrings("abcd", connection.bufferedInput());
738 try std.testing.expect(try sys.pollReadable(sockets[0], 0));
739 try std.testing.expectEqual(@as(u64, 2), connection.inputStatus().capacity_rejections);
740 try std.testing.expect(storage.bytes.ptr == pointer);
741 try std.testing.expectEqual(capacity, storage.capacity);
742 try std.testing.expectEqual(alloc_phase.PhaseViolations{}, phase_allocator.violations());
743
744 connection.consumeBufferedInput(4);
745 var final: [1]u8 = undefined;
746 try std.testing.expectEqual(@as(usize, 1), try connection.read(&final));
747 try std.testing.expectEqualStrings("e", &final);
748 connection.deinit();
749
750 phase_allocator.beginTeardown();
751 storage.deinit(phase_allocator.teardownAllocator());
752 phase_allocator.deinit();
753 }
754
755 const TestConnection = struct {
756 storage: InputStorage,
757 connection: Connection,
758
759 fn init(socket: sys.Socket) !TestConnection {
760 var storage = try InputStorage.init(std.testing.allocator, .{
761 .connection_count = 1,
762 .bytes_per_connection = 64,
763 });
764 errdefer storage.deinit(std.testing.allocator);
765 storage.activate();
766 return .{
767 .storage = storage,
768 .connection = Connection.init(
769 1,
770 socket,
771 try storage.connection(0),
772 emptyRequestScratch(),
773 emptyResponseScratch(),
774 .system(),
775 try time.bootNow(),
776 ),
777 };
778 }
779
780 fn deinit(self: *TestConnection) void {
781 self.connection.deinit();
782 self.storage.deinit(std.testing.allocator);
783 }
784 };
785
786 test "Connection init and deinit" {
787 const sockets = try testSocketPair();
788 defer sys.close(sockets[1]);
789
790 var owned = try TestConnection.init(sockets[0]);
791 owned.deinit();
792 }
793
794 test "Connection transfers an open socket without closing it" {
795 const sockets = try testSocketPair();
796 defer sys.close(sockets[1]);
797
798 var owned = try TestConnection.init(sockets[0]);
799 const detached = owned.connection.takeSocket();
800 owned.deinit();
801 defer sys.close(detached);
802
803 try std.testing.expectEqual(@as(usize, 1), try sys.send(sockets[1], "x", 0));
804 var byte: [1]u8 = undefined;
805 try std.testing.expectEqual(@as(usize, 1), try sys.recv(detached, &byte, 0));
806 try std.testing.expectEqualStrings("x", &byte);
807 }
808
809 test "Connection read and write" {
810 const sockets = try testSocketPair();
811 defer sys.close(sockets[1]);
812
813 var owned = try TestConnection.init(sockets[0]);
814 defer owned.deinit();
815 const conn = &owned.connection;
816
817 const msg = "hello http";
818 _ = try sys.send(sockets[1], msg, 0);
819
820 var buf: [64]u8 = undefined;
821 const n = try conn.read(&buf);
822 try std.testing.expectEqualStrings(msg, buf[0..n]);
823 }
824
825 test "Connection retained input is consumed before socket data" {
826 const sockets = try testSocketPair();
827 defer sys.close(sockets[1]);
828
829 var owned = try TestConnection.init(sockets[0]);
830 defer owned.deinit();
831 const conn = &owned.connection;
832
833 try conn.retainInput("first");
834 _ = try sys.send(sockets[1], "second", 0);
835
836 var buf: [8]u8 = undefined;
837 const first = try conn.read(&buf);
838 try std.testing.expectEqualStrings("first", buf[0..first]);
839 const second = try conn.read(&buf);
840 try std.testing.expectEqualStrings("second", buf[0..second]);
841 }
842
843 test "Connection retained input preserves byte order across partial reads" {
844 const sockets = try testSocketPair();
845 defer sys.close(sockets[1]);
846
847 var owned = try TestConnection.init(sockets[0]);
848 defer owned.deinit();
849 const conn = &owned.connection;
850
851 try conn.retainInput("abcdef");
852
853 var buf: [4]u8 = undefined;
854 const first = try conn.read(&buf);
855 try std.testing.expectEqualStrings("abcd", buf[0..first]);
856 const second = try conn.read(&buf);
857 try std.testing.expectEqualStrings("ef", buf[0..second]);
858 }
859
860 test "Connection write and read" {
861 const sockets = try testSocketPair();
862 defer sys.close(sockets[1]);
863
864 var owned = try TestConnection.init(sockets[0]);
865 defer owned.deinit();
866 const conn = &owned.connection;
867
868 const msg = "response from server";
869 try conn.write(msg);
870
871 var buf: [64]u8 = undefined;
872 const n = sys.recv(sockets[1], &buf, 0) catch unreachable;
873 try std.testing.expectEqualStrings(msg, buf[0..n]);
874 }
875
876 test "Connection write times out instead of spinning on backpressure" {
877 const sockets = try testSocketPair();
878 defer sys.close(sockets[1]);
879
880 try sys.setNonBlocking(sockets[0]);
881
882 var owned = try TestConnection.init(sockets[0]);
883 defer owned.deinit();
884 const conn = &owned.connection;
885
886 try conn.setWriteTimeout(1);
887 try fillSocketWriteBuffer(sockets[0]);
888
889 try std.testing.expectError(error.ConnectionTimedOut, conn.write("x"));
890 try std.testing.expectEqual(Connection.State.closing, conn.currentState());
891 }
892
893 test "Connection close prevents further read/write" {
894 const sockets = try testSocketPair();
895 defer sys.close(sockets[1]);
896
897 var owned = try TestConnection.init(sockets[0]);
898 defer owned.deinit();
899 const conn = &owned.connection;
900
901 conn.close();
902 try std.testing.expectEqual(Connection.State.closed, conn.currentState());
903
904 var buf: [64]u8 = undefined;
905 const read_err = conn.read(&buf);
906 try std.testing.expectError(error.ConnectionClosed, read_err);
907
908 const write_err = conn.write("test");
909 try std.testing.expectError(error.ConnectionClosed, write_err);
910 }
911
912 test "Connection detects peer close" {
913 const sockets = try testSocketPair();
914
915 var owned = try TestConnection.init(sockets[0]);
916 defer owned.deinit();
917 const conn = &owned.connection;
918
919 sys.close(sockets[1]);
920
921 var buf: [64]u8 = undefined;
922 const read_err = conn.read(&buf);
923 try std.testing.expectError(error.ConnectionClosed, read_err);
924 try std.testing.expectEqual(Connection.State.closing, conn.currentState());
925 }
926
927 test "Connection double close is safe" {
928 const sockets = try testSocketPair();
929 defer sys.close(sockets[1]);
930
931 var owned = try TestConnection.init(sockets[0]);
932 defer owned.deinit();
933 const conn = &owned.connection;
934
935 conn.close();
936 conn.close();
937 try std.testing.expectEqual(Connection.State.closed, conn.currentState());
938 }
939
940 test "Connection interrupt preserves descriptor ownership until close" {
941 const sockets = try testSocketPair();
942 defer sys.close(sockets[1]);
943
944 var owned = try TestConnection.init(sockets[0]);
945 defer owned.deinit();
946 const conn = &owned.connection;
947
948 conn.interrupt();
949 conn.interrupt();
950 try std.testing.expectEqual(Connection.State.closing, conn.currentState());
951
952 var byte: [1]u8 = undefined;
953 try std.testing.expectError(error.ConnectionClosed, conn.read(&byte));
954
955 conn.close();
956 try std.testing.expectEqual(Connection.State.closed, conn.currentState());
957 }
958
959 test "Connection timeout can be set" {
960 const sockets = try testSocketPair();
961 defer sys.close(sockets[1]);
962
963 var owned = try TestConnection.init(sockets[0]);
964 defer owned.deinit();
965 const conn = &owned.connection;
966
967 try conn.setReadTimeout(5000);
968 try conn.setWriteTimeout(5000);
969 try conn.setReadTimeout(0);
970 try conn.setWriteTimeout(0);
971 }