lib/http/src/client/stream/storage.zig
daab053ee43316e1809a84551d573ddd1e5bf3d2
1 const std = @import("std");
2 const alloc_phase = @import("alloc_phase");
3 const model = @import("root.zig");
4
5 const Header = model.Header;
6 const Scratch = model.Scratch;
7
8 pub const StorageExhaustion = error{ClientStreamCapacityExceeded};
9
10 pub const Storage = struct {
11 phase: alloc_phase.capacity.Phase,
12 capacity: model.Capacity,
13 bytes: []align(@alignOf(Header)) u8,
14
15 pub const Limits: type = model.Limits;
16 pub const Capacity: type = model.Capacity;
17 pub const Exhaustion: type = StorageExhaustion;
18 pub const InitError = std.mem.Allocator.Error || error{CapacityOverflow};
19
20 pub const claim: alloc_phase.capacity.Declaration = .{
21 .source = .{
22 .id = "http.client_stream_storage",
23 .kind = .phase_static,
24 .limit_source = .caller,
25 .storage = .{
26 .covered = &.{
27 .{
28 .id = "fixed_parsed_streaming_response_header_entries_for_e9a56e6dd10f",
29 .lifetime = .steady,
30 .detail = "fixed parsed streaming-response header entries for every stream slot",
31 },
32 .{
33 .id = "fixed_response_head_bytes_retaining_borrowed_callba_581ad8ca32dc",
34 .lifetime = .steady,
35 .detail = "fixed response-head bytes retaining borrowed callback header names and values",
36 },
37 },
38 .excluded = &.{
39 "request serialization, URL targets, TLS state, certificate bundles, sockets, and kernel queues",
40 "fixed scalar chunk-fragmentation state and handler-owned streamed output",
41 },
42 },
43 .capacity = .{
44 .inputs = &.{
45 alloc_phase.capacity.bindInput(Limits, "stream_count", "stream_count"),
46 alloc_phase.capacity.bindInput(Limits, "header_count_per_stream", "header_count_per_stream"),
47 alloc_phase.capacity.bindInput(Limits, "head_bytes_per_stream", "head_bytes_per_stream"),
48 },
49 .type_selectors = &.{
50 alloc_phase.capacity.bindType(Header, "header"),
51 },
52 .nodes = &.{
53 .{ .input = 0 },
54 .{ .input = 1 },
55 .{ .product = .{ .left = 0, .right = 1 } },
56 .{ .constant = 1 },
57 .{ .scale = .{ .node = 3, .coefficient = .{ .size_of_concrete_type = 0 } } },
58 .{ .product = .{ .left = 2, .right = 4 } },
59 .{ .input = 2 },
60 .{ .product = .{ .left = 0, .right = 6 } },
61 .{ .add = .{ .left = 5, .right = 7 } },
62 },
63 .assertions = &.{.{
64 .scope = .closure_total,
65 .measure = .retained,
66 .relation = .exact,
67 .expression = 8,
68 }},
69 },
70 .overload = .{
71 .kind = .terminal,
72 .detail = "an over-capacity response terminates that one-shot stream before publishing the head callback",
73 },
74 .risks = .{
75 .transitive = .{
76 .status = .witnessed,
77 .detail = "stream head parsing and fragmented chunk decoding allocate no storage after activation",
78 },
79 .foreign = .{
80 .status = .excluded,
81 .detail = "handler callbacks own any retained headers or body output and preserve their own error identity",
82 },
83 },
84 .obligations = &.{
85 .{ .key = "http_client_stream_capacity", .role = .capacity_model },
86 .{ .key = "http_client_stream_oom_retry", .role = .custom },
87 .{ .key = "http_client_stream_partition", .role = .custom },
88 .{ .key = "http_client_stream_sealed", .role = .transitive_risk },
89 .{ .key = "http_client_stream_atomic", .role = .overload },
90 .{ .key = "http_client_stream_boundary", .role = .custom },
91 .{ .key = "http_client_stream_callback", .role = .foreign_risk },
92 .{ .key = "http_client_stream_network", .role = .overload },
93 },
94 },
95 .bindings = .{
96 .owner = @This(),
97 .seal = .{
98 .family = alloc_phase.capacity.selector(@This().activate),
99 .premise = .{
100 .class = .checked_semantic_fact,
101 .authority = .checker,
102 },
103 },
104 .teardown = .{
105 .family = alloc_phase.capacity.selector(@This().deinit),
106 .premise = .{
107 .class = .checked_semantic_fact,
108 .authority = .checker,
109 },
110 },
111 },
112 };
113
114 pub fn init(allocator: std.mem.Allocator, limits: Limits) InitError!Storage {
115 const capacity = try Capacity.derive(limits);
116 const bytes = if (capacity.storage_bytes == 0)
117 @as([]align(@alignOf(Header)) u8, &.{})
118 else
119 try allocator.alignedAlloc(u8, .of(Header), capacity.storage_bytes);
120 return .{
121 .phase = .initialization,
122 .capacity = capacity,
123 .bytes = bytes,
124 };
125 }
126
127 pub fn activate(self: *Storage) void {
128 std.debug.assert(self.phase == .initialization);
129 std.debug.assert(self.bytes.len == self.capacity.storage_bytes);
130 self.phase = .steady;
131 }
132
133 pub fn stream(self: *Storage, index: usize) Exhaustion!Scratch {
134 std.debug.assert(self.phase == .steady);
135 if (index >= self.capacity.stream_count) {
136 return error.ClientStreamCapacityExceeded;
137 }
138 const all_headers = std.mem.bytesAsSlice(
139 Header,
140 self.bytes[0..self.capacity.header_bytes],
141 );
142 const header_start = index * self.capacity.header_count_per_stream;
143 const head_start = self.capacity.header_bytes +
144 index * self.capacity.head_bytes_per_stream;
145 return .{
146 .headers = all_headers[header_start..][0..self.capacity.header_count_per_stream],
147 .head = self.bytes[head_start..][0..self.capacity.head_bytes_per_stream],
148 };
149 }
150
151 pub fn deinit(self: *Storage, allocator: std.mem.Allocator) void {
152 std.debug.assert(self.phase != .teardown);
153 std.debug.assert(self.bytes.len == self.capacity.storage_bytes);
154 self.phase = .teardown;
155 if (self.bytes.len != 0) allocator.free(self.bytes);
156 self.bytes = &.{};
157 }
158 };
159
160 comptime {
161 alloc_phase.capacity.requireAllocatorRejectingOwnerShape(Storage);
162 }
163
164 fn checkStorageInitFailures(allocator: std.mem.Allocator) !void {
165 var storage = try Storage.init(allocator, .{
166 .stream_count = 3,
167 .header_count_per_stream = 7,
168 .head_bytes_per_stream = 127,
169 });
170 storage.deinit(allocator);
171 }
172
173 test "Client stream storage retries after every allocation failure" {
174 comptime {
175 @stardustClaim(
176 @import("alloc_phase").capacity.witness(Storage, "http_client_stream_oom_retry"),
177 null,
178 null,
179 null,
180 null,
181 null,
182 null,
183 );
184 }
185
186 try std.testing.checkAllAllocationFailures(
187 std.testing.allocator,
188 checkStorageInitFailures,
189 .{},
190 );
191 }
192
193 test "Client stream storage partitions reusable slots" {
194 comptime {
195 @stardustClaim(
196 @import("alloc_phase").capacity.witness(Storage, "http_client_stream_partition"),
197 null,
198 null,
199 null,
200 null,
201 null,
202 null,
203 );
204 }
205
206 var storage = try Storage.init(std.testing.allocator, .{
207 .stream_count = 2,
208 .header_count_per_stream = 2,
209 .head_bytes_per_stream = 3,
210 });
211 defer storage.deinit(std.testing.allocator);
212 storage.activate();
213
214 const first = try storage.stream(0);
215 const second = try storage.stream(1);
216 try std.testing.expect(first.headers.ptr + first.headers.len == second.headers.ptr);
217 try std.testing.expect(first.head.ptr + first.head.len == second.head.ptr);
218 try std.testing.expect(
219 @intFromPtr(first.head.ptr) ==
220 @intFromPtr(storage.bytes.ptr) + storage.capacity.header_bytes,
221 );
222 try std.testing.expectError(
223 error.ClientStreamCapacityExceeded,
224 storage.stream(2),
225 );
226 }
227
228 const SealedHandler = struct {
229 checksum: usize = 0,
230
231 fn handler(self: *SealedHandler) model.Handler {
232 return .{ .ctx = self, .vtable = &vtable };
233 }
234
235 fn onHead(context: *anyopaque, status: u16, headers: []const Header) anyerror!void {
236 const self: *SealedHandler = @ptrCast(@alignCast(context));
237 self.checksum +%= status;
238 self.checksum +%= headers.len;
239 }
240
241 fn onChunk(context: *anyopaque, chunk: []const u8) anyerror!void {
242 const self: *SealedHandler = @ptrCast(@alignCast(context));
243 self.checksum +%= chunk.len;
244 }
245
246 const vtable = model.HandlerVTable{
247 .onHead = &onHead,
248 .onChunk = &onChunk,
249 };
250 };
251
252 test "Client streaming remains allocation-free after storage seals" {
253 comptime {
254 @stardustClaim(
255 @import("alloc_phase").capacity.witness(Storage, "http_client_stream_sealed"),
256 null,
257 null,
258 null,
259 null,
260 null,
261 null,
262 );
263 }
264
265 var phase_allocator = try alloc_phase.SealedPhaseAllocator.init(std.testing.allocator);
266 var storage = Storage.init(phase_allocator.initializationAllocator(), .{
267 .stream_count = 1,
268 .header_count_per_stream = 4,
269 .head_bytes_per_stream = 128,
270 }) catch |err| {
271 phase_allocator.abortInitialization();
272 phase_allocator.deinit();
273 return err;
274 };
275 errdefer {
276 if (phase_allocator.phase() == .initialization) phase_allocator.abortInitialization();
277 if (phase_allocator.phase() == .steady) phase_allocator.beginTeardown();
278 if (storage.phase != .teardown) {
279 storage.deinit(phase_allocator.teardownAllocator());
280 }
281 phase_allocator.deinit();
282 }
283
284 const pointer = storage.bytes.ptr;
285 const capacity = storage.capacity;
286 phase_allocator.seal();
287 storage.activate();
288 const scratch = try storage.stream(0);
289 var handler = SealedHandler{};
290
291 var fixed_reader = std.Io.Reader.fixed(
292 "HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\nHello",
293 );
294 try std.testing.expectEqual(
295 @as(u16, 200),
296 try model.ClientStream.read(scratch, &fixed_reader, handler.handler()),
297 );
298 var chunked_reader = std.Io.Reader.fixed(
299 "HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n2\r\nHe\r\n3\r\nllo\r\n0\r\nTrace: value\r\n\r\n",
300 );
301 try std.testing.expectEqual(
302 @as(u16, 200),
303 try model.ClientStream.read(scratch, &chunked_reader, handler.handler()),
304 );
305 try std.testing.expect(handler.checksum != 0);
306 try std.testing.expect(storage.bytes.ptr == pointer);
307 try std.testing.expectEqual(capacity, storage.capacity);
308 try std.testing.expectEqual(alloc_phase.PhaseViolations{}, phase_allocator.violations());
309
310 phase_allocator.beginTeardown();
311 storage.deinit(phase_allocator.teardownAllocator());
312 phase_allocator.deinit();
313 }