lib/trace/src/store/format/block.zig
daab053ee43316e1809a84551d573ddd1e5bf3d2
1 const std = @import("std");
2 const event = @import("../../root.zig").event;
3 const codec = @import("codec.zig");
4
5 pub const target_bytes: usize = 256 * 1024;
6 pub const header_bytes: usize = 16;
7 pub const record_header_bytes: usize = 4;
8
9 const magic = [_]u8{ 't', 'r', 'c', 'b' };
10 const version: u8 = 1;
11 const version_offset: usize = 4;
12 const reserved_offset: usize = 5;
13 const event_count_offset: usize = 8;
14 const payload_bytes_offset: usize = 12;
15
16 pub fn capacity(max_event_bytes: usize) error{CapacityOverflow}!usize {
17 const event_record_bytes = std.math.add(usize, record_header_bytes, max_event_bytes) catch
18 return error.CapacityOverflow;
19 const minimum = std.math.add(usize, header_bytes, event_record_bytes) catch
20 return error.CapacityOverflow;
21 return @max(target_bytes, minimum);
22 }
23
24 pub const Builder = struct {
25 bytes: []u8,
26 len: usize = header_bytes,
27 event_count: u32 = 0,
28
29 pub fn init(bytes: []u8) Builder {
30 std.debug.assert(bytes.len >= header_bytes + record_header_bytes);
31 return .{ .bytes = bytes };
32 }
33
34 pub fn empty(self: *const Builder) bool {
35 return self.event_count == 0;
36 }
37
38 pub fn canAppend(self: *const Builder, event_bytes: usize) bool {
39 const record_bytes = std.math.add(usize, record_header_bytes, event_bytes) catch return false;
40 return record_bytes <= self.bytes.len - self.len;
41 }
42
43 pub fn append(self: *Builder, item: event.Event, event_bytes: usize) codec.Error![]u8 {
44 std.debug.assert(event_bytes <= std.math.maxInt(u32));
45 std.debug.assert(self.canAppend(event_bytes));
46 if (self.event_count == std.math.maxInt(u32)) return error.CapacityOverflow;
47 const prefix = self.bytes[self.len..][0..record_header_bytes];
48 std.mem.writeInt(u32, prefix, @intCast(event_bytes), .big);
49 const start = self.len + record_header_bytes;
50 const encoded = try codec.encode(self.bytes[start..][0..event_bytes], item);
51 std.debug.assert(encoded.len == event_bytes);
52 self.len = start + encoded.len;
53 self.event_count += 1;
54 return encoded;
55 }
56
57 pub fn finish(self: *Builder) []const u8 {
58 std.debug.assert(!self.empty());
59 @memset(self.bytes[0..header_bytes], 0);
60 @memcpy(self.bytes[0..magic.len], &magic);
61 self.bytes[version_offset] = version;
62 std.mem.writeInt(u32, self.bytes[event_count_offset..][0..4], self.event_count, .big);
63 std.mem.writeInt(u32, self.bytes[payload_bytes_offset..][0..4], @intCast(self.len - header_bytes), .big);
64 return self.bytes[0..self.len];
65 }
66
67 pub fn reset(self: *Builder) void {
68 self.len = header_bytes;
69 self.event_count = 0;
70 }
71 };
72
73 pub const View = struct {
74 bytes: []u8,
75 offset: usize = header_bytes,
76 remaining: u32,
77
78 pub fn init(bytes: []u8) error{InvalidBlock}!View {
79 if (bytes.len < header_bytes) return error.InvalidBlock;
80 if (!std.mem.eql(u8, bytes[0..magic.len], &magic)) return error.InvalidBlock;
81 if (bytes[version_offset] != version) return error.InvalidBlock;
82 for (bytes[reserved_offset..event_count_offset]) |byte| {
83 if (byte != 0) return error.InvalidBlock;
84 }
85 const event_count = std.mem.readInt(u32, bytes[event_count_offset..][0..4], .big);
86 const payload_bytes: usize = std.mem.readInt(u32, bytes[payload_bytes_offset..][0..4], .big);
87 if (event_count == 0 or payload_bytes != bytes.len - header_bytes) return error.InvalidBlock;
88 var offset = header_bytes;
89 for (0..event_count) |_| {
90 if (record_header_bytes > bytes.len - offset) return error.InvalidBlock;
91 const event_bytes: usize = std.mem.readInt(u32, bytes[offset..][0..record_header_bytes], .big);
92 offset += record_header_bytes;
93 if (event_bytes == 0 or event_bytes > bytes.len - offset) return error.InvalidBlock;
94 offset += event_bytes;
95 }
96 if (offset != bytes.len) return error.InvalidBlock;
97 return .{ .bytes = bytes, .remaining = event_count };
98 }
99
100 pub fn peek(self: *const View) ?[]u8 {
101 if (self.remaining == 0) return null;
102 const event_bytes: usize = std.mem.readInt(
103 u32,
104 self.bytes[self.offset..][0..record_header_bytes],
105 .big,
106 );
107 return self.bytes[self.offset + record_header_bytes ..][0..event_bytes];
108 }
109
110 pub fn advance(self: *View) void {
111 std.debug.assert(self.remaining != 0);
112 const encoded = self.peek().?;
113 self.offset += record_header_bytes + encoded.len;
114 self.remaining -= 1;
115 }
116 };
117
118 test "binary event block round trips bounded records" {
119 var bytes: [512]u8 = undefined;
120 var builder = Builder.init(&bytes);
121 const first = event.Event.user(.{ .seq = 1 }, "first", "payload");
122 const second = event.Event.user(.{ .seq = 2 }, "second", "");
123 _ = try builder.append(first, try codec.encodedSize(first));
124 _ = try builder.append(second, try codec.encodedSize(second));
125 const encoded = builder.finish();
126 try std.testing.expectEqualSlices(u8, "trcb", encoded[0..magic.len]);
127 var view = try View.init(@constCast(encoded));
128 try std.testing.expect(first.eqlForReplay(try codec.decode(view.peek().?)));
129 view.advance();
130 try std.testing.expect(second.eqlForReplay(try codec.decode(view.peek().?)));
131 view.advance();
132 try std.testing.expectEqual(@as(?[]u8, null), view.peek());
133 }
134
135 test "binary event block rejects truncation and trailing bytes" {
136 var bytes: [256]u8 = undefined;
137 var builder = Builder.init(&bytes);
138 const item = event.Event.user(.{}, "event", "payload");
139 _ = try builder.append(item, try codec.encodedSize(item));
140 const encoded = builder.finish();
141 try std.testing.expectError(error.InvalidBlock, View.init(@constCast(encoded[0 .. encoded.len - 1])));
142 bytes[encoded.len] = 0;
143 try std.testing.expectError(error.InvalidBlock, View.init(bytes[0 .. encoded.len + 1]));
144 }