lib/sql/src/history/segment/format.zig
daab053ee43316e1809a84551d573ddd1e5bf3d2
1 const std = @import("std");
2 const alloc_phase = @import("alloc_phase");
3 const sql = @import("../../root.zig");
4
5 const Hash = sql.Hash;
6 const Sha256 = std.crypto.hash.sha2.Sha256;
7
8 pub const header_bytes: usize = 256;
9 pub const block_header_bytes: usize = 64;
10 pub const closure_record_bytes: usize = 40;
11 pub const trailer_bytes: usize = Sha256.digest_length;
12 pub const blocks_max: usize = 64;
13 pub const closure_digests_max: usize = 64 * 1024;
14 pub const block_bytes_max: usize = 64 * 1024 * 1024;
15 pub const encoded_bytes_max: usize = header_bytes +
16 blocks_max * block_header_bytes +
17 closure_digests_max * closure_record_bytes +
18 block_bytes_max + trailer_bytes;
19
20 const RangeLifetime = struct {
21 const maximum_seconds_per_year: u64 = 366 * 24 * 60 * 60;
22 const service_lifetime_years: u64 = 1_000;
23 const service_lifetime_seconds: u64 =
24 maximum_seconds_per_year * service_lifetime_years;
25 };
26
27 fn totalRange(
28 unit_precision: u64,
29 units_per_second: u64,
30 lifetime_seconds: u64,
31 ) ?u64 {
32 const units = std.math.mul(
33 u64,
34 unit_precision,
35 units_per_second,
36 ) catch return null;
37 return std.math.mul(u64, units, lifetime_seconds) catch null;
38 }
39
40 pub const SegmentRange = struct {
41 pub const Count = u64;
42 pub const unit_precision_segments: Count = 1;
43 pub const maximum_units_per_second: Count = 1_000_000;
44 pub const maximum_seconds_per_year: Count =
45 RangeLifetime.maximum_seconds_per_year;
46 pub const service_lifetime_years: Count =
47 RangeLifetime.service_lifetime_years;
48 pub const service_lifetime_seconds: Count =
49 RangeLifetime.service_lifetime_seconds;
50 pub const maximum_increment_segments: Count = 1;
51 pub const budget_segments: Count = total(
52 maximum_units_per_second,
53 service_lifetime_seconds,
54 ).?;
55
56 pub fn total(units_per_second: Count, lifetime_seconds: Count) ?Count {
57 return totalRange(
58 unit_precision_segments,
59 units_per_second,
60 lifetime_seconds,
61 );
62 }
63
64 pub fn advance(current: Count) ?Count {
65 if (current >= budget_segments) return null;
66 return std.math.add(Count, current, maximum_increment_segments) catch null;
67 }
68
69 pub fn containsOrdinal(ordinal: Count) bool {
70 return advance(ordinal) != null;
71 }
72
73 pub fn containsCount(count: Count) bool {
74 return count > 0 and count <= budget_segments;
75 }
76 };
77
78 pub const EventRange = struct {
79 pub const Count = u64;
80 pub const unit_precision_events: Count = 1;
81 pub const maximum_units_per_second: Count = 512 * 1024 * 1024;
82 pub const maximum_seconds_per_year: Count =
83 RangeLifetime.maximum_seconds_per_year;
84 pub const service_lifetime_years: Count =
85 RangeLifetime.service_lifetime_years;
86 pub const service_lifetime_seconds: Count =
87 RangeLifetime.service_lifetime_seconds;
88 pub const maximum_increment_events: Count =
89 @as(Count, blocks_max) * @as(Count, std.math.maxInt(u32));
90 pub const budget_events: Count = total(
91 maximum_units_per_second,
92 service_lifetime_seconds,
93 ).?;
94
95 pub fn total(units_per_second: Count, lifetime_seconds: Count) ?Count {
96 return totalRange(
97 unit_precision_events,
98 units_per_second,
99 lifetime_seconds,
100 );
101 }
102
103 pub fn advance(current: Count, increment: Count) ?Count {
104 if (current > budget_events or increment == 0 or
105 increment > maximum_increment_events)
106 {
107 return null;
108 }
109 const next = std.math.add(Count, current, increment) catch return null;
110 return if (next <= budget_events) next else null;
111 }
112 };
113
114 comptime {
115 std.debug.assert(SegmentRange.budget_segments > 0);
116 std.debug.assert(
117 @as(u128, SegmentRange.budget_segments) +
118 @as(u128, SegmentRange.maximum_increment_segments) <=
119 std.math.maxInt(SegmentRange.Count),
120 );
121 std.debug.assert(
122 EventRange.budget_events >= EventRange.maximum_increment_events,
123 );
124 std.debug.assert(
125 @as(u128, EventRange.budget_events) +
126 @as(u128, EventRange.maximum_increment_events) <=
127 std.math.maxInt(EventRange.Count),
128 );
129 }
130
131 const magic = "TINYSEGM".*;
132 const format_version: u32 = 1;
133 const block_kind: u8 = 1;
134 const closure_kind: u8 = 2;
135
136 pub const Error = error{
137 CapacityOverflow,
138 CapacityExceeded,
139 EmptySegment,
140 InvalidSegment,
141 SegmentBytesExceeded,
142 StorageTooShort,
143 TooManyBlocks,
144 TooManyClosureDigests,
145 };
146
147 pub const Spec = struct {
148 epoch: u64,
149 ordinal: SegmentRange.Count,
150 first_event: EventRange.Count,
151 event_count: EventRange.Count,
152 previous_segment: Hash,
153 event_chain: Hash,
154 checkpoint_root: Hash,
155 schema_identity: Hash,
156 stream_identity: Hash,
157 };
158
159 pub const Block = struct {
160 first_event: EventRange.Count,
161 event_count: u32,
162 raw_bytes: u32,
163 bytes: []const u8,
164 };
165
166 pub const Sealed = struct {
167 bytes: []const u8,
168 digest: Hash,
169 spec: Spec,
170 };
171
172 pub const Sealer = struct {
173 pub const storage_alignment: usize = 8;
174 pub const Storage = []align(storage_alignment) u8;
175 pub const Limits = struct {
176 blocks: usize,
177 block_bytes: usize,
178 closure_digests: usize,
179 };
180
181 pub const Capacity = struct {
182 blocks: usize,
183 block_bytes: usize,
184 closure_digests: usize,
185 storage_bytes: usize,
186
187 pub const DeriveError = error{
188 CapacityOverflow,
189 CapacityExceeded,
190 EmptySegment,
191 };
192
193 pub fn derive(limits: Limits) DeriveError!Capacity {
194 if (limits.blocks == 0 or limits.block_bytes == 0) {
195 return error.EmptySegment;
196 }
197 if (limits.blocks > blocks_max or
198 limits.closure_digests > closure_digests_max or
199 limits.block_bytes > block_bytes_max)
200 {
201 return error.CapacityExceeded;
202 }
203 const block_headers = try multiply(limits.blocks, block_header_bytes);
204 const closures = try multiply(
205 limits.closure_digests,
206 closure_record_bytes,
207 );
208 var storage_bytes = try add(header_bytes, block_headers);
209 storage_bytes = try add(storage_bytes, closures);
210 storage_bytes = try add(storage_bytes, limits.block_bytes);
211 storage_bytes = try add(storage_bytes, trailer_bytes);
212 return .{
213 .blocks = limits.blocks,
214 .block_bytes = limits.block_bytes,
215 .closure_digests = limits.closure_digests,
216 .storage_bytes = storage_bytes,
217 };
218 }
219 };
220
221 pub const Exhaustion = error{
222 InvalidSegment,
223 SegmentBytesExceeded,
224 TooManyBlocks,
225 TooManyClosureDigests,
226 };
227 pub const InitError = Capacity.DeriveError || error{StorageTooShort};
228 pub const work_limits: alloc_phase.capacity.WorkLimits = .{
229 .transition_steps_max = blocks_max + closure_digests_max,
230 .cleanup_steps_per_call_max = 0,
231 .cleanup_calls_at_capacity_max = 0,
232 };
233
234 pub const claim: alloc_phase.capacity.Declaration = .{
235 .source = .{
236 .id = "sql.history_segment_sealer",
237 .kind = .phase_static,
238 .limit_source = .caller,
239 .storage = .{
240 .covered = &.{.{
241 .id = "reserved_immutable_segment_output",
242 .lifetime = .transferred,
243 .detail = "exact header record payload and trailer output",
244 }},
245 .excluded = &.{
246 "caller-owned event blocks and content closure digests",
247 "filesystem writes manifest pages and archive transport",
248 },
249 },
250 .capacity = .{
251 .inputs = &.{
252 alloc_phase.capacity.bindInput(Limits, "blocks", "blocks"),
253 alloc_phase.capacity.bindInput(
254 Limits,
255 "block_bytes",
256 "block_bytes",
257 ),
258 alloc_phase.capacity.bindInput(
259 Limits,
260 "closure_digests",
261 "closure_digests",
262 ),
263 },
264 .type_selectors = &.{},
265 .nodes = &.{
266 .{ .input = 0 },
267 .{ .input = 1 },
268 .{ .input = 2 },
269 .{ .constant = 64 },
270 .{ .product = .{ .left = 0, .right = 3 } },
271 .{ .constant = 40 },
272 .{ .product = .{ .left = 2, .right = 5 } },
273 .{ .constant = 256 },
274 .{ .add = .{ .left = 7, .right = 4 } },
275 .{ .add = .{ .left = 8, .right = 6 } },
276 .{ .add = .{ .left = 9, .right = 1 } },
277 .{ .constant = 32 },
278 .{ .add = .{ .left = 10, .right = 11 } },
279 },
280 .assertions = &.{.{
281 .scope = .closure_total,
282 .measure = .retained,
283 .relation = .exact,
284 .expression = 12,
285 }},
286 },
287 .overload = .{
288 .kind = .reject_before_mutation,
289 .detail = "record count and byte exhaustion reject before output changes",
290 },
291 .risks = .{
292 .transitive = .{
293 .status = .witnessed,
294 .detail = "sealing hashes and writes only the caller reservation",
295 },
296 .foreign = .{
297 .status = .excluded,
298 .detail = "durable file and archive publication are outside this owner",
299 },
300 },
301 .work = .{
302 .equation = "record writes <= blocks + closure_digests",
303 },
304 .obligations = &.{
305 .{
306 .key = "sql_history_segment_capacity",
307 .role = .capacity_model,
308 },
309 .{
310 .key = "sql_history_segment_overload",
311 .role = .overload,
312 },
313 .{
314 .key = "sql_history_segment_verification",
315 .role = .transitive_risk,
316 },
317 .{
318 .key = "sql_history_segment_work_bound",
319 .role = .work_bound,
320 },
321 },
322 },
323 .bindings = .{
324 .owner = @This(),
325 .seal = .{
326 .family = alloc_phase.capacity.selector(@This().activate),
327 .premise = .{
328 .class = .checked_semantic_fact,
329 .authority = .checker,
330 },
331 },
332 .teardown = .{
333 .family = alloc_phase.capacity.selector(@This().deinit),
334 .premise = .{
335 .class = .checked_semantic_fact,
336 .authority = .checker,
337 },
338 },
339 },
340 };
341
342 phase: alloc_phase.capacity.Phase,
343 capacity: Capacity,
344 storage: Storage,
345 cursor: usize = header_bytes,
346 blocks: usize = 0,
347 block_bytes: usize = 0,
348 closure_digests: usize = 0,
349 first_event: ?EventRange.Count = null,
350 next_event: EventRange.Count = 0,
351 blocks_started: bool = false,
352
353 pub fn init(storage: Storage, limits: Limits) InitError!Sealer {
354 const capacity = try Capacity.derive(limits);
355 if (storage.len < capacity.storage_bytes) return error.StorageTooShort;
356 return .{
357 .phase = .initialization,
358 .capacity = capacity,
359 .storage = storage[0..capacity.storage_bytes],
360 };
361 }
362
363 pub fn activate(self: *Sealer) void {
364 std.debug.assert(self.phase == .initialization);
365 std.debug.assert(self.storage.len == self.capacity.storage_bytes);
366 self.phase = .steady;
367 }
368
369 pub fn addClosure(self: *Sealer, value: Hash) Exhaustion!void {
370 self.assertSteady();
371 if (self.blocks_started) return error.InvalidSegment;
372 if (self.closure_digests == self.capacity.closure_digests) {
373 return error.TooManyClosureDigests;
374 }
375 const end = self.cursor + closure_record_bytes;
376 writeClosure(self.storage[self.cursor..end], value);
377 self.cursor = end;
378 self.closure_digests += 1;
379 }
380
381 pub fn addBlock(self: *Sealer, block: Block) Exhaustion!void {
382 self.assertSteady();
383 if (block.event_count == 0 or block.raw_bytes == 0 or block.bytes.len == 0) {
384 return error.InvalidSegment;
385 }
386 if (self.blocks == self.capacity.blocks) return error.TooManyBlocks;
387 if (block.bytes.len > self.capacity.block_bytes - self.block_bytes) {
388 return error.SegmentBytesExceeded;
389 }
390 const next_event = EventRange.advance(
391 block.first_event,
392 block.event_count,
393 ) orelse return error.InvalidSegment;
394 if (self.first_event) |_| {
395 if (block.first_event != self.next_event) return error.InvalidSegment;
396 } else self.first_event = block.first_event;
397 const end = self.cursor + block_header_bytes + block.bytes.len;
398 writeBlock(self.storage[self.cursor..end], block);
399 self.cursor = end;
400 self.blocks += 1;
401 self.block_bytes += block.bytes.len;
402 self.next_event = next_event;
403 self.blocks_started = true;
404 }
405
406 pub fn finish(self: *Sealer, spec: Spec) Exhaustion!Sealed {
407 self.assertSteady();
408 if (self.blocks != self.capacity.blocks or
409 self.block_bytes != self.capacity.block_bytes or
410 self.closure_digests != self.capacity.closure_digests)
411 {
412 return error.InvalidSegment;
413 }
414 if (!SegmentRange.containsOrdinal(spec.ordinal)) {
415 return error.InvalidSegment;
416 }
417 const first_event = self.first_event orelse return error.InvalidSegment;
418 if (spec.first_event != first_event) return error.InvalidSegment;
419 const next_event = EventRange.advance(
420 spec.first_event,
421 spec.event_count,
422 ) orelse return error.InvalidSegment;
423 if (next_event != self.next_event) {
424 return error.InvalidSegment;
425 }
426 std.debug.assert(self.cursor + trailer_bytes == self.storage.len);
427 var records_digest: Hash = undefined;
428 Sha256.hash(self.storage[header_bytes..self.cursor], &records_digest, .{});
429 writeHeader(self.storage[0..header_bytes], spec, self.capacity, records_digest);
430 const trailer_digest = digest(self.storage[0..self.cursor]);
431 @memcpy(self.storage[self.cursor..], &trailer_digest);
432 const segment_digest = digest(self.storage);
433 const view = verify(self.storage) catch return error.InvalidSegment;
434 std.debug.assert(std.mem.eql(u8, &view.digest, &segment_digest));
435 return .{ .bytes = self.storage, .digest = segment_digest, .spec = spec };
436 }
437
438 pub fn deinit(self: *Sealer) Storage {
439 std.debug.assert(self.phase == .steady);
440 const storage = self.storage;
441 self.phase = .teardown;
442 self.* = undefined;
443 return storage;
444 }
445
446 fn assertSteady(self: *const Sealer) void {
447 std.debug.assert(self.phase == .steady);
448 std.debug.assert(self.storage.len == self.capacity.storage_bytes);
449 std.debug.assert(self.blocks <= self.capacity.blocks);
450 std.debug.assert(self.block_bytes <= self.capacity.block_bytes);
451 std.debug.assert(self.closure_digests <= self.capacity.closure_digests);
452 }
453 };
454
455 comptime {
456 alloc_phase.capacity.requireProvisionedRejectingOwnerShape(Sealer);
457 }
458
459 pub const View = struct {
460 bytes: []const u8,
461 digest: Hash,
462 spec: Spec,
463 block_count: u32,
464 closure_count: u32,
465 block_bytes: u64,
466
467 pub fn iterator(self: *const View) Iterator {
468 return .{
469 .view = self,
470 .cursor = header_bytes,
471 .closures_remaining = self.closure_count,
472 .blocks_remaining = self.block_count,
473 .next_event = self.spec.first_event,
474 };
475 }
476
477 pub fn demand(self: *const View) Error!ReplayDemand {
478 var demand_value = ReplayDemand{
479 .blocks = self.block_count,
480 .events = self.spec.event_count,
481 .encoded_block_bytes = self.block_bytes,
482 .closure_digests = self.closure_count,
483 .checkpoint_roots = @intFromBool(!allZero(&self.spec.checkpoint_root)),
484 .segment_bytes = @intCast(self.bytes.len),
485 };
486 var iterator_value = self.iterator();
487 while (try iterator_value.next()) |record| switch (record) {
488 .closure => {},
489 .block => |block| {
490 demand_value.raw_event_bytes = std.math.add(
491 u64,
492 demand_value.raw_event_bytes,
493 block.raw_bytes,
494 ) catch return error.InvalidSegment;
495 demand_value.max_record_bytes = @max(
496 demand_value.max_record_bytes,
497 @as(u64, block_header_bytes) +
498 @as(u64, @intCast(block.bytes.len)),
499 );
500 },
501 };
502 return demand_value;
503 }
504 };
505
506 pub const ReplayDemand = struct {
507 segments: u64 = 1,
508 blocks: u64,
509 events: u64,
510 raw_event_bytes: u64 = 0,
511 encoded_block_bytes: u64,
512 max_record_bytes: u64 = closure_record_bytes,
513 checkpoint_roots: u64,
514 closure_digests: u64,
515 segment_bytes: u64,
516 io_operations: u64 = 1,
517 };
518
519 pub const Record = union(enum) {
520 closure: Hash,
521 block: Block,
522 };
523
524 pub const Iterator = struct {
525 view: *const View,
526 cursor: usize,
527 closures_remaining: u32,
528 blocks_remaining: u32,
529 next_event: EventRange.Count,
530
531 pub fn next(self: *Iterator) Error!?Record {
532 if (self.closures_remaining != 0) {
533 const end = self.cursor + closure_record_bytes;
534 if (end > self.view.bytes.len - trailer_bytes) {
535 return error.InvalidSegment;
536 }
537 const value = try readClosure(self.view.bytes[self.cursor..end]);
538 self.cursor = end;
539 self.closures_remaining -= 1;
540 return .{ .closure = value };
541 }
542 if (self.blocks_remaining == 0) {
543 if (self.cursor != self.view.bytes.len - trailer_bytes) {
544 return error.InvalidSegment;
545 }
546 return null;
547 }
548 const block = try readBlock(self.view.bytes, &self.cursor);
549 if (block.first_event != self.next_event) return error.InvalidSegment;
550 self.next_event = EventRange.advance(
551 block.first_event,
552 block.event_count,
553 ) orelse return error.InvalidSegment;
554 self.blocks_remaining -= 1;
555 return .{ .block = block };
556 }
557 };
558
559 pub fn verify(bytes: []const u8) Error!View {
560 if (bytes.len < header_bytes + block_header_bytes + trailer_bytes) {
561 return error.InvalidSegment;
562 }
563 const decoded = try readHeader(bytes[0..header_bytes]);
564 const expected_len = try encodedBytes(
565 decoded.block_count,
566 decoded.block_bytes,
567 decoded.closure_count,
568 );
569 if (bytes.len != expected_len) return error.InvalidSegment;
570 const records_end = bytes.len - trailer_bytes;
571 if (!std.mem.eql(
572 u8,
573 &decoded.records_digest,
574 &digest(bytes[header_bytes..records_end]),
575 )) return error.InvalidSegment;
576 const trailer_digest = digest(bytes[0..records_end]);
577 if (!std.mem.eql(u8, &trailer_digest, bytes[records_end..])) {
578 return error.InvalidSegment;
579 }
580 const segment_digest = digest(bytes);
581 var view = View{
582 .bytes = bytes,
583 .digest = segment_digest,
584 .spec = decoded.spec,
585 .block_count = decoded.block_count,
586 .closure_count = decoded.closure_count,
587 .block_bytes = decoded.block_bytes,
588 };
589 try verifyRecords(&view);
590 return view;
591 }
592
593 const DecodedHeader = struct {
594 spec: Spec,
595 block_count: u32,
596 closure_count: u32,
597 block_bytes: u64,
598 records_digest: Hash,
599 };
600
601 fn verifyRecords(view: *const View) Error!void {
602 var iterator = view.iterator();
603 var blocks: u32 = 0;
604 var closures: u32 = 0;
605 var block_bytes: u64 = 0;
606 while (try iterator.next()) |record| switch (record) {
607 .closure => closures += 1,
608 .block => |block| {
609 blocks += 1;
610 block_bytes = std.math.add(u64, block_bytes, block.bytes.len) catch
611 return error.InvalidSegment;
612 },
613 };
614 if (blocks != view.block_count or closures != view.closure_count or
615 block_bytes != view.block_bytes or
616 iterator.next_event - view.spec.first_event != view.spec.event_count)
617 {
618 return error.InvalidSegment;
619 }
620 }
621
622 fn writeHeader(
623 out: []u8,
624 spec: Spec,
625 capacity: Sealer.Capacity,
626 records_digest: Hash,
627 ) void {
628 std.debug.assert(out.len == header_bytes);
629 std.debug.assert(SegmentRange.containsOrdinal(spec.ordinal));
630 std.debug.assert(EventRange.advance(spec.first_event, spec.event_count) != null);
631 @memset(out, 0);
632 @memcpy(out[0..8], &magic);
633 std.mem.writeInt(u32, out[8..12], format_version, .little);
634 std.mem.writeInt(u32, out[12..16], header_bytes, .little);
635 std.mem.writeInt(u64, out[16..24], spec.epoch, .little);
636 std.mem.writeInt(u64, out[24..32], spec.ordinal, .little);
637 std.mem.writeInt(u64, out[32..40], spec.first_event, .little);
638 std.mem.writeInt(u64, out[40..48], spec.event_count, .little);
639 std.mem.writeInt(u32, out[48..52], @intCast(capacity.blocks), .little);
640 std.mem.writeInt(u32, out[52..56], @intCast(capacity.closure_digests), .little);
641 std.mem.writeInt(u64, out[56..64], capacity.block_bytes, .little);
642 @memcpy(out[64..96], &spec.previous_segment);
643 @memcpy(out[96..128], &spec.event_chain);
644 @memcpy(out[128..160], &spec.checkpoint_root);
645 @memcpy(out[160..192], &spec.schema_identity);
646 @memcpy(out[192..224], &spec.stream_identity);
647 @memcpy(out[224..256], &records_digest);
648 }
649
650 fn readHeader(bytes: []const u8) Error!DecodedHeader {
651 if (bytes.len != header_bytes or !std.mem.eql(u8, bytes[0..8], &magic)) {
652 return error.InvalidSegment;
653 }
654 if (std.mem.readInt(u32, bytes[8..12], .little) != format_version or
655 std.mem.readInt(u32, bytes[12..16], .little) != header_bytes)
656 {
657 return error.InvalidSegment;
658 }
659 const block_count = std.mem.readInt(u32, bytes[48..52], .little);
660 const closure_count = std.mem.readInt(u32, bytes[52..56], .little);
661 const block_bytes = std.mem.readInt(u64, bytes[56..64], .little);
662 const first_event = std.mem.readInt(u64, bytes[32..40], .little);
663 const event_count = std.mem.readInt(u64, bytes[40..48], .little);
664 if (block_count == 0 or block_count > blocks_max or
665 closure_count > closure_digests_max or block_bytes > block_bytes_max or
666 !SegmentRange.containsOrdinal(std.mem.readInt(u64, bytes[24..32], .little)) or
667 EventRange.advance(first_event, event_count) == null)
668 {
669 return error.InvalidSegment;
670 }
671 return .{
672 .spec = .{
673 .epoch = std.mem.readInt(u64, bytes[16..24], .little),
674 .ordinal = std.mem.readInt(u64, bytes[24..32], .little),
675 .first_event = first_event,
676 .event_count = event_count,
677 .previous_segment = bytes[64..96].*,
678 .event_chain = bytes[96..128].*,
679 .checkpoint_root = bytes[128..160].*,
680 .schema_identity = bytes[160..192].*,
681 .stream_identity = bytes[192..224].*,
682 },
683 .block_count = block_count,
684 .closure_count = closure_count,
685 .block_bytes = block_bytes,
686 .records_digest = bytes[224..256].*,
687 };
688 }
689
690 fn writeClosure(out: []u8, value: Hash) void {
691 std.debug.assert(out.len == closure_record_bytes);
692 @memset(out, 0);
693 out[0] = closure_kind;
694 @memcpy(out[8..40], &value);
695 }
696
697 fn readClosure(bytes: []const u8) Error!Hash {
698 if (bytes.len != closure_record_bytes or bytes[0] != closure_kind) {
699 return error.InvalidSegment;
700 }
701 if (!allZero(bytes[1..8])) return error.InvalidSegment;
702 return bytes[8..40].*;
703 }
704
705 fn writeBlock(out: []u8, block: Block) void {
706 std.debug.assert(out.len == block_header_bytes + block.bytes.len);
707 @memset(out[0..block_header_bytes], 0);
708 out[0] = block_kind;
709 std.mem.writeInt(u32, out[4..8], @intCast(block.bytes.len), .little);
710 std.mem.writeInt(u64, out[8..16], block.first_event, .little);
711 std.mem.writeInt(u32, out[16..20], block.event_count, .little);
712 std.mem.writeInt(u32, out[20..24], block.raw_bytes, .little);
713 const block_digest = digest(block.bytes);
714 @memcpy(out[24..56], &block_digest);
715 @memcpy(out[block_header_bytes..], block.bytes);
716 }
717
718 fn readBlock(bytes: []const u8, cursor: *usize) Error!Block {
719 if (cursor.* > bytes.len - trailer_bytes or
720 bytes.len - trailer_bytes - cursor.* < block_header_bytes)
721 {
722 return error.InvalidSegment;
723 }
724 const header = bytes[cursor.*..][0..block_header_bytes];
725 if (header[0] != block_kind or !allZero(header[1..4]) or
726 !allZero(header[56..64])) return error.InvalidSegment;
727 const encoded_len = std.mem.readInt(u32, header[4..8], .little);
728 const event_count = std.mem.readInt(u32, header[16..20], .little);
729 const raw_bytes = std.mem.readInt(u32, header[20..24], .little);
730 const first_event = std.mem.readInt(u64, header[8..16], .little);
731 if (encoded_len == 0 or event_count == 0 or raw_bytes == 0) {
732 return error.InvalidSegment;
733 }
734 if (EventRange.advance(first_event, event_count) == null) {
735 return error.InvalidSegment;
736 }
737 const end = std.math.add(
738 usize,
739 cursor.* + block_header_bytes,
740 encoded_len,
741 ) catch return error.InvalidSegment;
742 if (end > bytes.len - trailer_bytes) return error.InvalidSegment;
743 const encoded = bytes[cursor.* + block_header_bytes .. end];
744 if (!std.mem.eql(u8, header[24..56], &digest(encoded))) {
745 return error.InvalidSegment;
746 }
747 cursor.* = end;
748 return .{
749 .first_event = first_event,
750 .event_count = event_count,
751 .raw_bytes = raw_bytes,
752 .bytes = encoded,
753 };
754 }
755
756 fn encodedBytes(
757 block_count: anytype,
758 block_bytes: anytype,
759 closure_count: anytype,
760 ) Error!usize {
761 const blocks = std.math.cast(usize, block_count) orelse
762 return error.CapacityOverflow;
763 const payload = std.math.cast(usize, block_bytes) orelse
764 return error.CapacityOverflow;
765 const closures = std.math.cast(usize, closure_count) orelse
766 return error.CapacityOverflow;
767 return (try Sealer.Capacity.derive(.{
768 .blocks = blocks,
769 .block_bytes = payload,
770 .closure_digests = closures,
771 })).storage_bytes;
772 }
773
774 fn digest(bytes: []const u8) Hash {
775 var value: Hash = undefined;
776 Sha256.hash(bytes, &value, .{});
777 return value;
778 }
779
780 fn allZero(bytes: []const u8) bool {
781 for (bytes) |byte| if (byte != 0) return false;
782 return true;
783 }
784
785 fn multiply(left: usize, right: usize) error{CapacityOverflow}!usize {
786 return std.math.mul(usize, left, right) catch error.CapacityOverflow;
787 }
788
789 fn add(left: usize, right: usize) error{CapacityOverflow}!usize {
790 return std.math.add(usize, left, right) catch error.CapacityOverflow;
791 }