lib/http/src/client/websocket/storage.zig

daab053ee43316e1809a84551d573ddd1e5bf3d2

  1 const std = @import("std");
  2 const alloc_phase = @import("alloc_phase");
  3 const model = @import("root.zig");
  4 
  5 const Scratch = model.Scratch;
  6 
  7 pub const StorageExhaustion = error{
  8     ClientWebSocketCapacityExceeded,
  9 };
 10 
 11 pub const Storage = struct {
 12     phase: alloc_phase.capacity.Phase,
 13     capacity: model.Capacity,
 14     bytes: []u8,
 15 
 16     pub const Limits: type = model.Limits;
 17     pub const Capacity: type = model.Capacity;
 18     pub const Exhaustion: type = StorageExhaustion;
 19     pub const InitError =
 20         std.mem.Allocator.Error || error{CapacityOverflow};
 21 
 22     pub const claim: alloc_phase.capacity.Declaration = .{
 23         .source = .{
 24             .id = "http.client_websocket_storage",
 25             .kind = .phase_static,
 26             .limit_source = .caller,
 27             .storage = .{
 28                 .covered = &.{
 29                     .{
 30                         .id = "one_fixed_inbound_wire_frame_region_for_every_client_websocket",
 31                         .lifetime = .steady,
 32                         .detail = "one fixed inbound wire-frame region for every client WebSocket",
 33                     },
 34                     .{
 35                         .id = "one_fixed_masked_outbound_wire_frame_region_for_eve_72613139f624",
 36                         .lifetime = .steady,
 37                         .detail = "one fixed masked outbound wire-frame region for every client WebSocket",
 38                     },
 39                 },
 40                 .excluded = &.{
 41                     "URL preparation, HTTP upgrade headers, TLS state, certificate bundles, sockets, and kernel queues",
 42                     "application payload owners and protocol effects",
 43                 },
 44             },
 45             .capacity = .{
 46                 .inputs = &.{
 47                     alloc_phase.capacity.bindInput(Limits, "websocket_count", "websocket_count"),
 48                     alloc_phase.capacity.bindInput(Limits, "frame_payload_bytes_per_websocket", "frame_payload_bytes_per_websocket"),
 49                 },
 50                 .type_selectors = &.{},
 51                 .nodes = &.{
 52                     .{ .input = 0 },
 53                     .{ .scale = .{ .node = 0, .coefficient = .{ .literal = 2 } } },
 54                     .{ .constant = 14 },
 55                     .{ .input = 1 },
 56                     .{ .add = .{ .left = 2, .right = 3 } },
 57                     .{ .product = .{ .left = 1, .right = 4 } },
 58                 },
 59                 .assertions = &.{.{
 60                     .scope = .closure_total,
 61                     .measure = .retained,
 62                     .relation = .exact,
 63                     .expression = 5,
 64                 }},
 65             },
 66             .overload = .{
 67                 .kind = .terminal,
 68                 .detail = "capacity exhaustion rejects before opening a WebSocket or publishing a frame",
 69             },
 70             .risks = .{
 71                 .transitive = .{
 72                     .status = .witnessed,
 73                     .detail = "WebSocket frame send and receive allocate no storage after activation",
 74                 },
 75                 .foreign = .{
 76                     .status = .excluded,
 77                     .detail = "TLS and operating-system network owners remain independent",
 78                 },
 79             },
 80             .obligations = &.{
 81                 .{ .key = "http_client_websocket_capacity", .role = .capacity_model },
 82                 .{ .key = "http_client_websocket_oom_retry", .role = .custom },
 83                 .{ .key = "http_client_websocket_partition", .role = .overload },
 84                 .{ .key = "http_client_websocket_sealed", .role = .transitive_risk },
 85                 .{ .key = "http_client_websocket_network_overload", .role = .overload },
 86                 .{ .key = "http_client_websocket_network_transitive_risk", .role = .transitive_risk },
 87                 .{ .key = "http_client_websocket_network_foreign_risk", .role = .foreign_risk },
 88             },
 89         },
 90         .bindings = .{
 91             .owner = @This(),
 92             .seal = .{
 93                 .family = alloc_phase.capacity.selector(@This().activate),
 94                 .premise = .{
 95                     .class = .checked_semantic_fact,
 96                     .authority = .checker,
 97                 },
 98             },
 99             .teardown = .{
100                 .family = alloc_phase.capacity.selector(@This().deinit),
101                 .premise = .{
102                     .class = .checked_semantic_fact,
103                     .authority = .checker,
104                 },
105             },
106         },
107     };
108 
109     pub fn init(
110         allocator: std.mem.Allocator,
111         limits: Limits,
112     ) InitError!Storage {
113         const capacity = try Capacity.derive(limits);
114         const bytes = if (capacity.storage_bytes == 0)
115             @as([]u8, &.{})
116         else
117             try allocator.alloc(u8, capacity.storage_bytes);
118         return .{
119             .phase = .initialization,
120             .capacity = capacity,
121             .bytes = bytes,
122         };
123     }
124 
125     pub fn activate(self: *Storage) void {
126         std.debug.assert(self.phase == .initialization);
127         std.debug.assert(
128             self.bytes.len == self.capacity.storage_bytes,
129         );
130         self.phase = .steady;
131     }
132 
133     pub fn websocket(
134         self: *Storage,
135         index: usize,
136     ) StorageExhaustion!Scratch {
137         std.debug.assert(self.phase == .steady);
138         if (index >= self.capacity.websocket_count) {
139             return error.ClientWebSocketCapacityExceeded;
140         }
141         const frame_bytes =
142             self.capacity.frame_bytes_per_websocket;
143         const read_start = index * frame_bytes;
144         const write_region_start =
145             self.capacity.websocket_count * frame_bytes;
146         const write_start =
147             write_region_start + index * frame_bytes;
148         return .{
149             .read = self.bytes[read_start..][0..frame_bytes],
150             .write = self.bytes[write_start..][0..frame_bytes],
151             .frame_payload_bytes = self.capacity.frame_payload_bytes_per_websocket,
152         };
153     }
154 
155     pub fn deinit(
156         self: *Storage,
157         allocator: std.mem.Allocator,
158     ) void {
159         std.debug.assert(self.phase != .teardown);
160         std.debug.assert(
161             self.bytes.len == self.capacity.storage_bytes,
162         );
163         self.phase = .teardown;
164         if (self.bytes.len != 0) allocator.free(self.bytes);
165         self.bytes = &.{};
166     }
167 };
168 
169 comptime {
170     alloc_phase.capacity.requireAllocatorRejectingOwnerShape(
171         Storage,
172     );
173 }
174 
175 fn checkStorageInitFailures(
176     allocator: std.mem.Allocator,
177 ) !void {
178     var storage = try Storage.init(allocator, .{
179         .websocket_count = 3,
180         .frame_payload_bytes_per_websocket = 257,
181     });
182     storage.deinit(allocator);
183 }
184 
185 test "Client WebSocket storage retries after every allocation failure" {
186     comptime {
187         @stardustClaim(
188             @import("alloc_phase").capacity.witness(Storage, "http_client_websocket_oom_retry"),
189             null,
190             null,
191             null,
192             null,
193             null,
194             null,
195         );
196     }
197 
198     try std.testing.checkAllAllocationFailures(
199         std.testing.allocator,
200         checkStorageInitFailures,
201         .{},
202     );
203 }
204 
205 test "Client WebSocket storage partitions reusable slots" {
206     comptime {
207         @stardustClaim(
208             @import("alloc_phase").capacity.witness(Storage, "http_client_websocket_partition"),
209             null,
210             null,
211             null,
212             null,
213             null,
214             null,
215         );
216     }
217 
218     var storage = try Storage.init(std.testing.allocator, .{
219         .websocket_count = 2,
220         .frame_payload_bytes_per_websocket = 3,
221     });
222     defer storage.deinit(std.testing.allocator);
223     storage.activate();
224 
225     const first = try storage.websocket(0);
226     const second = try storage.websocket(1);
227     try std.testing.expect(
228         first.read.ptr + first.read.len == second.read.ptr,
229     );
230     try std.testing.expect(
231         first.write.ptr + first.write.len == second.write.ptr,
232     );
233     try std.testing.expect(
234         @intFromPtr(first.write.ptr) ==
235             @intFromPtr(storage.bytes.ptr) +
236                 storage.capacity.websocket_count *
237                     storage.capacity.frame_bytes_per_websocket,
238     );
239     try std.testing.expectError(
240         error.ClientWebSocketCapacityExceeded,
241         storage.websocket(2),
242     );
243 }
244 
245 test "Client WebSocket storage remains fixed after activation" {
246     comptime {
247         @stardustClaim(
248             @import("alloc_phase").capacity.witness(Storage, "http_client_websocket_sealed"),
249             null,
250             null,
251             null,
252             null,
253             null,
254             null,
255         );
256     }
257 
258     var phase_allocator =
259         try alloc_phase.SealedPhaseAllocator.init(
260             std.testing.allocator,
261         );
262     var storage = Storage.init(
263         phase_allocator.initializationAllocator(),
264         .{
265             .websocket_count = 1,
266             .frame_payload_bytes_per_websocket = 64,
267         },
268     ) catch |err| {
269         phase_allocator.abortInitialization();
270         phase_allocator.deinit();
271         return err;
272     };
273     errdefer {
274         if (phase_allocator.phase() == .initialization) {
275             phase_allocator.abortInitialization();
276         }
277         if (phase_allocator.phase() == .steady) {
278             phase_allocator.beginTeardown();
279         }
280         if (storage.phase != .teardown) {
281             storage.deinit(
282                 phase_allocator.teardownAllocator(),
283             );
284         }
285         phase_allocator.deinit();
286     }
287 
288     const pointer = storage.bytes.ptr;
289     const capacity = storage.capacity;
290     phase_allocator.seal();
291     storage.activate();
292     const scratch = try storage.websocket(0);
293     try std.testing.expect(scratch.read.ptr == pointer);
294     try std.testing.expectEqual(capacity, storage.capacity);
295     try std.testing.expectEqual(
296         alloc_phase.PhaseViolations{},
297         phase_allocator.violations(),
298     );
299 
300     phase_allocator.beginTeardown();
301     storage.deinit(phase_allocator.teardownAllocator());
302     phase_allocator.deinit();
303 }