tiny.trace.TraceReader

1pub const Reader = struct {2    storage: ?*Storage = null,3    root_path: []u8 = &.{},4    header: format.Header = undefined,5    block_buffer: []u8 = &.{},6    block_view: ?block.View = null,7    sql_allocator: std.heap.FixedBufferAllocator = undefined,8    database_workspace: sql.FileDatabase.Workspace = undefined,9    database_workspace_open: bool = false,10    file: sql.FileDatabase = undefined,11    file_open: bool = false,12    events: sql.Tree = undefined,13    chunk_event_count: u64 = 0,14    chunk_byte_count: u64 = 0,15    event_hasher: std.hash.Wyhash = format.newHasher(),16    loaded_chunk_count: u64 = 0,17    loaded_event_count: u64 = 0,18    loaded_event_bytes: u64 = 0,19    loaded_block_count: u64 = 0,20    loaded_block_bytes: u64 = 0,21    current: ?event.Event = null,22    initialized: bool = false,23    verified: bool = false,2425    pub fn open(self: *Reader, storage: *Storage, root_path: []const u8) !void {26        if (self.initialized) return error.TraceReaderAlreadyOpen;27        const regions = try storage.acquire(.{ .root_path_bytes = root_path.len });28        var acquired = true;29        errdefer if (acquired) storage.release();30        std.mem.copyForwards(u8, regions.root_path, root_path);31        const header_value = try readHeader(32            regions.root_path,33            regions.input,34            regions.path,35            regions.json_stack,36            storage.capacity.json_nesting,37        );38        try storage.admit(header_value.limits);39        if (!header_value.status_closed) return error.TraceNotClosed;40        const block_bytes = block.capacity(header_value.limits.max_event_bytes) catch41            return error.InvalidTraceManifest;42        std.debug.assert(block_bytes <= regions.input.len);43        self.* = .{44            .storage = storage,45            .root_path = regions.root_path,46            .header = header_value,47            .block_buffer = regions.input[0..block_bytes],48            .sql_allocator = std.heap.FixedBufferAllocator.init(regions.sql),49            .initialized = true,50        };51        acquired = false;52        errdefer self.deinit();53        self.database_workspace = try sql.FileDatabase.Workspace.allocate(54            self.sql_allocator.allocator(),55            database.workspaceLimits(storage.capacity.database),56        );57        self.database_workspace_open = true;58        self.file = try database.openReader(59            self.sql_allocator.allocator(),60            &self.database_workspace,61            self.root_path,62            storage.capacity.database,63        );64        self.file_open = true;65        self.events = try database.events(&self.file);66        const identity = try self.events.identity();67        const expected_key_bytes = std.math.mul(u64, self.header.block_count, database.key_bytes) catch68            return error.InvalidTraceManifest;69        if (identity.entries != self.header.block_count or70            identity.key_bytes != expected_key_bytes or71            identity.value_bytes != self.header.block_bytes)72        {73            return error.InvalidTraceManifest;74        }75    }7677    pub fn deinit(self: *Reader) void {78        if (!self.initialized) return;79        if (self.file_open) {80            self.file.deinit();81            self.file_open = false;82        }83        if (self.database_workspace_open) {84            self.database_workspace.deallocate(self.sql_allocator.allocator());85            self.database_workspace_open = false;86        }87        const storage = self.storage.?;88        self.* = .{};89        storage.release();90    }9192    pub fn source(self: *Reader) event.Source {93        return .{94            .context = self,95            .peekFn = peekFromSource,96            .advanceFn = advanceFromSource,97            .countFn = countFromSource,98        };99    }100101    pub fn peek(self: *Reader) !?*const event.Event {102        if (!self.initialized) return error.TraceReaderNotOpen;103        if (self.current) |*item| return item;104        if (self.verified) return null;105        const maybe_encoded = try self.peekEncoded();106        if (self.loaded_event_count == self.header.event_count) {107            if (maybe_encoded != null) return error.InvalidTraceManifest;108            try self.finishTrace();109            return null;110        }111        const encoded = maybe_encoded orelse return error.InvalidTraceManifest;112        if (encoded.len == 0 or encoded.len > self.header.limits.max_event_bytes) {113            return error.InvalidTraceEvent;114        }115        const prospective_chunk_bytes = std.math.add(u64, self.chunk_byte_count, encoded.len) catch116            return error.InvalidTraceManifest;117        var prior_chunk_event_count = self.chunk_event_count;118        var prior_chunk_byte_count = self.chunk_byte_count;119        var next_loaded_chunk_count = self.loaded_chunk_count;120        if (self.chunk_event_count != 0 and prospective_chunk_bytes > self.header.limits.max_chunk_bytes) {121            next_loaded_chunk_count = std.math.add(u64, self.loaded_chunk_count, 1) catch122                return error.InvalidTraceManifest;123            prior_chunk_event_count = 0;124            prior_chunk_byte_count = 0;125        }126        const next_chunk_event_count = std.math.add(u64, prior_chunk_event_count, 1) catch127            return error.InvalidTraceManifest;128        const next_chunk_byte_count = std.math.add(u64, prior_chunk_byte_count, encoded.len) catch129            return error.InvalidTraceManifest;130        const next_event_count = std.math.add(u64, self.loaded_event_count, 1) catch131            return error.InvalidTraceManifest;132        const next_event_bytes = std.math.add(u64, self.loaded_event_bytes, encoded.len) catch133            return error.InvalidTraceManifest;134        if (next_chunk_byte_count > self.header.limits.max_chunk_bytes or135            next_event_count > self.header.event_count or136            next_event_bytes > self.header.event_bytes)137        {138            return error.InvalidTraceManifest;139        }140        const parsed = codec.decode(encoded) catch return error.InvalidTraceEvent;141        var key_buffer: [database.key_bytes]u8 = undefined;142        const key = database.sequenceKey(&key_buffer, self.loaded_event_count);143        self.advanceEncoded();144        database.updateChecksum(&self.event_hasher, key, encoded);145        self.loaded_chunk_count = next_loaded_chunk_count;146        self.chunk_event_count = next_chunk_event_count;147        self.chunk_byte_count = next_chunk_byte_count;148        self.loaded_event_count = next_event_count;149        self.loaded_event_bytes = next_event_bytes;150        self.current = parsed;151        return &self.current.?;152    }153154    fn peekEncoded(self: *Reader) !?[]u8 {155        while (true) {156            if (self.block_view) |*view| {157                if (view.peek()) |encoded| return encoded;158                self.block_view = null;159            }160            if (self.loaded_block_count == self.header.block_count) return null;161            try self.loadBlock();162        }163    }164165    fn advanceEncoded(self: *Reader) void {166        self.block_view.?.advance();167    }168169    fn loadBlock(self: *Reader) !void {170        var key_buffer: [database.key_bytes]u8 = undefined;171        const key = database.sequenceKey(&key_buffer, self.loaded_block_count);172        const encoded = self.events.getInto(key, self.block_buffer) catch |err| switch (err) {173            error.OutputTooSmall => return error.InvalidTraceEvent,174            else => |read_err| return read_err,175        } orelse return error.InvalidTraceManifest;176        const view = block.View.init(encoded) catch return error.InvalidTraceEvent;177        const next_block_count = std.math.add(u64, self.loaded_block_count, 1) catch178            return error.InvalidTraceManifest;179        const next_block_bytes = std.math.add(u64, self.loaded_block_bytes, encoded.len) catch180            return error.InvalidTraceManifest;181        if (next_block_count > self.header.block_count or next_block_bytes > self.header.block_bytes) {182            return error.InvalidTraceManifest;183        }184        self.block_view = view;185        self.loaded_block_count = next_block_count;186        self.loaded_block_bytes = next_block_bytes;187    }188189    pub fn advance(self: *Reader) void {190        std.debug.assert(self.current != null);191        self.current = null;192    }193194    pub fn eventCount(self: *const Reader) u64 {195        return self.header.event_count;196    }197198    pub fn verify(self: *Reader) !VerifyResult {199        while (try self.peek()) |_| self.advance();200        return .{201            .chunk_count = self.loaded_chunk_count,202            .event_count = self.loaded_event_count,203            .event_bytes = self.loaded_event_bytes,204            .checksum = self.event_hasher.final(),205        };206    }207208    fn finishTrace(self: *Reader) !void {209        if (self.chunk_event_count != 0) {210            self.loaded_chunk_count = std.math.add(u64, self.loaded_chunk_count, 1) catch211                return error.InvalidTraceManifest;212            self.chunk_event_count = 0;213            self.chunk_byte_count = 0;214        }215        if (self.loaded_chunk_count != self.header.chunk_count or216            self.loaded_event_count != self.header.event_count or217            self.loaded_event_bytes != self.header.event_bytes or218            self.loaded_block_count != self.header.block_count or219            self.loaded_block_bytes != self.header.block_bytes)220        {221            return error.InvalidTraceManifest;222        }223        if (self.event_hasher.final() != self.header.event_checksum) return error.TraceChecksumMismatch;224        self.verified = true;225    }226227    fn peekFromSource(context: *anyopaque) !?*const event.Event {228        const self: *Reader = @ptrCast(@alignCast(context));229        return try self.peek();230    }231232    fn advanceFromSource(context: *anyopaque) void {233        const self: *Reader = @ptrCast(@alignCast(context));234        self.advance();235    }236237    fn countFromSource(context: *anyopaque) u64 {238        const self: *Reader = @ptrCast(@alignCast(context));239        return self.eventCount();240    }241};

1pub const TraceReader = store.Reader;