From 83ffa74e2b9057614a1f204747faa4d54a742012 Mon Sep 17 00:00:00 2001 From: Mitchell Hashimoto Date: Tue, 28 Jul 2026 06:27:45 -0700 Subject: [PATCH] terminal/snapshot: page records --- src/terminal/snapshot/main.zig | 24 +++ src/terminal/snapshot/page.zig | 275 +++++++++++++++++++++++++++---- src/terminal/snapshot/record.zig | 241 +++++++++++++++++++-------- 3 files changed, 440 insertions(+), 100 deletions(-) diff --git a/src/terminal/snapshot/main.zig b/src/terminal/snapshot/main.zig index fedcc4618..20bc2d41c 100644 --- a/src/terminal/snapshot/main.zig +++ b/src/terminal/snapshot/main.zig @@ -42,6 +42,30 @@ //! Records have a strict order: TERMINAL, the primary SCREEN and its //! PAGE records, an optional alternate SCREEN and its PAGE records, //! CONTINUATION, READY, and FINISH. +//! +//! ## Encoding +//! +//! Encode the envelope once, then append records in the required order: +//! +//! ```zig +//! var output: std.Io.Writer.Allocating = .init(alloc); +//! defer output.deinit(); +//! +//! try envelope.encode(&output.writer); +//! try page.encode(&terminal_page, &output); +//! +//! const snapshot = output.written(); +//! ``` +//! +//! Record codecs reserve and backpatch their own framing, so callers do not +//! calculate payload lengths or checksums. `written` borrows the completed +//! bytes from the allocating writer. Use `toOwnedSlice` instead when ownership +//! must be transferred to the caller. +//! +//! `page.encode` currently appends one complete framed PAGE record. Its +//! lower-level payload codec is intentionally private. `page.decode` consumes +//! and validates one complete PAGE record, including its payload boundary, +//! checksum, and restored page integrity. pub const envelope = @import("envelope.zig"); pub const grid = @import("grid.zig"); diff --git a/src/terminal/snapshot/page.zig b/src/terminal/snapshot/page.zig index 1e18ced22..b6324860d 100644 --- a/src/terminal/snapshot/page.zig +++ b/src/terminal/snapshot/page.zig @@ -94,9 +94,11 @@ const std = @import("std"); const Allocator = std.mem.Allocator; +const envelope = @import("envelope.zig"); const grid = @import("grid.zig"); const hyperlink = @import("hyperlink.zig"); const io = @import("io.zig"); +const record = @import("record.zig"); const style = @import("style.zig"); const terminal_hyperlink = @import("../hyperlink.zig"); const terminal_page = @import("../page.zig"); @@ -115,11 +117,9 @@ const TerminalStyle = terminal_style.Style; const TerminalStyleId = terminal_style.Id; const TerminalStyleSet = terminal_style.Set; -/// Errors possible while encoding a native PAGE. -pub const EncodeError = hyperlink.EncodeError || grid.EncodeError; +const PayloadEncodeError = hyperlink.EncodeError || grid.EncodeError; -/// Errors possible while decoding a native PAGE. -pub const DecodeError = style.DecodeError || +const PayloadDecodeError = style.DecodeError || grid.DecodeError || Header.CapacityError || error{ @@ -146,11 +146,59 @@ pub const DecodeError = style.DecodeError || OutOfMemory, }; -/// Encode a complete PAGE directly from a native page. +/// Errors possible while encoding a complete PAGE record. +pub const EncodeError = PayloadEncodeError || record.Writer.FinishError; + +/// Encode one complete PAGE record from a native page. +/// +/// The record is appended to `destination`. Its header is reserved before the +/// payload is encoded, then backpatched with the payload length and CRC32C. If +/// encoding fails, the partial record is removed while earlier bytes remain. pub fn encode( page: *const TerminalPage, - writer: *std.Io.Writer, + destination: *std.Io.Writer.Allocating, ) EncodeError!void { + var record_writer = try record.Writer.init(destination, .page); + errdefer record_writer.cancel(); + try encodePayload(page, record_writer.payloadWriter()); + try record_writer.finish(); +} + +/// Errors possible while decoding and validating a complete PAGE record. +pub const DecodeError = PayloadDecodeError || + record.Reader.InitError || + record.Reader.FinishError || + TerminalPage.IntegrityError || + error{ + /// The next record is valid but is not a PAGE. + UnexpectedRecordTag, + }; + +/// Decode and validate one complete PAGE record into a fresh native page. +/// +/// The record tag, payload boundary, CRC32C, payload contents, and final page +/// integrity are all validated before the page is returned. `alloc` is used +/// only for temporary decode remaps and integrity-check storage. +pub fn decode( + reader: *std.Io.Reader, + alloc: Allocator, +) DecodeError!TerminalPage { + var record_reader: record.Reader = undefined; + try record_reader.init(reader); + if (record_reader.header.tag != .page) return error.UnexpectedRecordTag; + + var page = try decodePayload(record_reader.payloadReader(), alloc); + errdefer page.deinit(); + try record_reader.finish(); + try page.verifyIntegrity(alloc); + return page; +} + +/// Encode a PAGE payload directly from a native page. +fn encodePayload( + page: *const TerminalPage, + writer: *std.Io.Writer, +) PayloadEncodeError!void { // Write header try Header.init(page).encode(writer); @@ -172,14 +220,14 @@ pub fn encode( try grid.encode(page, writer); } -/// Decode a complete PAGE directly into a fresh native page. +/// Decode a PAGE payload directly into a fresh native page. /// /// `alloc` is used only for the temporary native-ID remap tables. Styles, /// hyperlinks, strings, graphemes, rows, and cells are stored in the page. -pub fn decode( +fn decodePayload( reader: *std.Io.Reader, alloc: Allocator, -) DecodeError!TerminalPage { +) PayloadDecodeError!TerminalPage { // Decode the header, validate capacities, init page const header = try Header.decode(reader); const capacity = try header.pageCapacity(); @@ -445,6 +493,18 @@ const test_page_fixture = "\x00\x00\x00\x03\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00" ++ "\x00\x00"; +const test_empty_framed_page_fixture = + // Record header: PAGE, 37-byte payload, CRC32C. + "\x03\x00\x25\x00\x00\x00\x8c\x05\xd6\xd3" ++ + // PAGE header: one column, one row, and zero counts/capacities. + "\x01\x00\x01\x00\x00\x00\x00\x00" ++ + "\x00\x00\x00\x00\x00\x00\x00\x00" ++ + "\x00\x00\x00\x00" ++ + // One default row and one empty cell. + "\x00\x00\x00\x00\x00\x00\x00\x00" ++ + "\x00\x00\x00\x00\x00\x00\x00\x00" ++ + "\x00"; + test "golden encoding" { const header: Header = .{ .columns = 0x0102, @@ -504,7 +564,7 @@ test "reject every truncation" { } } -test "encode and decode a sparse native page" { +test "framed PAGE encode and decode a sparse native page" { const capacity: TerminalPageCapacity = .{ .cols = 3, .rows = 2, @@ -607,11 +667,11 @@ test "encode and decode a sparse native page" { try std.testing.expectEqual(header, Header.init(&page)); var counter: std.Io.Writer.Discarding = .init(&.{}); - try encode(&page, &counter.writer); + try encodePayload(&page, &counter.writer); var encoded: [512]u8 = undefined; var writer: std.Io.Writer = .fixed(&encoded); - try encode(&page, &writer); + try encodePayload(&page, &writer); const fixture = test_page_fixture; try std.testing.expectEqualStrings(fixture, writer.buffered()); @@ -620,7 +680,10 @@ test "encode and decode a sparse native page" { var source: std.Io.Reader = .fixed(writer.buffered()); var read_buf: [1]u8 = undefined; var limited = source.limited(.unlimited, &read_buf); - var decoded = try decode(&limited.interface, std.testing.allocator); + var decoded = try decodePayload( + &limited.interface, + std.testing.allocator, + ); defer decoded.deinit(); try std.testing.expectEqual(header, Header.init(&decoded)); @@ -702,7 +765,7 @@ test "encode and decode a sparse native page" { var reencoded: [512]u8 = undefined; var rewriter: std.Io.Writer = .fixed(&reencoded); - try encode(&decoded, &rewriter); + try encodePayload(&decoded, &rewriter); try std.testing.expect(!std.mem.eql( u8, writer.buffered(), @@ -710,7 +773,7 @@ test "encode and decode a sparse native page" { )); var reencoded_reader: std.Io.Reader = .fixed(rewriter.buffered()); - var decoded_again = try decode( + var decoded_again = try decodePayload( &reencoded_reader, std.testing.allocator, ); @@ -719,11 +782,161 @@ test "encode and decode a sparse native page" { var reencoded_again: [512]u8 = undefined; var rewriter_again: std.Io.Writer = .fixed(&reencoded_again); - try encode(&decoded_again, &rewriter_again); + try encodePayload(&decoded_again, &rewriter_again); try std.testing.expectEqualStrings( rewriter.buffered(), rewriter_again.buffered(), ); + + // The public codec appends one complete PAGE record after the envelope. + var snapshot: std.Io.Writer.Allocating = .init(std.testing.allocator); + defer snapshot.deinit(); + try envelope.encode(&snapshot.writer); + try encode(&page, &snapshot); + + const snapshot_bytes = snapshot.written(); + try std.testing.expectEqualStrings( + test_page_fixture, + snapshot_bytes[envelope.encoded_len + record.Header.len ..], + ); + + var snapshot_reader: std.Io.Reader = .fixed(snapshot_bytes); + try envelope.decode(&snapshot_reader); + var framed_page = try decode( + &snapshot_reader, + std.testing.allocator, + ); + defer framed_page.deinit(); + + try std.testing.expectEqual(header, Header.init(&framed_page)); + const framed_first = framed_page.getRowAndCell(0, 0); + try std.testing.expectEqual(@as(u21, 'A'), framed_first.cell.codepoint()); + try std.testing.expectEqual( + TerminalCell.SemanticContent.prompt, + framed_first.cell.semantic_content, + ); + try std.testing.expectEqualSlices( + u21, + &.{ 0x0301, 0x0302 }, + framed_page.lookupGrapheme( + framed_page.getRowAndCell(0, 1).cell, + ).?, + ); +} + +test "framed PAGE golden empty record" { + const capacity: TerminalPageCapacity = .{ + .cols = 1, + .rows = 1, + .styles = 0, + .hyperlink_bytes = 0, + .grapheme_bytes = 0, + .string_bytes = 0, + }; + var page = try TerminalPage.init(capacity); + defer page.deinit(); + + var destination: std.Io.Writer.Allocating = .init(std.testing.allocator); + defer destination.deinit(); + try encode(&page, &destination); + try std.testing.expectEqualStrings( + test_empty_framed_page_fixture, + destination.written(), + ); + + var source: std.Io.Reader = .fixed(destination.written()); + var decoded = try decode(&source, std.testing.allocator); + defer decoded.deinit(); + try std.testing.expectEqual( + Header.init(&page), + Header.init(&decoded), + ); +} + +test "framed PAGE validates tag length checksum and exhaustion" { + { + var wrong_tag = test_empty_framed_page_fixture.*; + std.mem.writeInt(u16, wrong_tag[0..2], @intFromEnum(record.Tag.screen), .little); + var reader: std.Io.Reader = .fixed(&wrong_tag); + try std.testing.expectError( + error.UnexpectedRecordTag, + decode(&reader, std.testing.allocator), + ); + } + + { + var invalid_checksum = test_empty_framed_page_fixture.*; + invalid_checksum[6] ^= 1; + var reader: std.Io.Reader = .fixed(&invalid_checksum); + try std.testing.expectError( + error.InvalidChecksum, + decode(&reader, std.testing.allocator), + ); + } + + { + var invalid_payload = test_empty_framed_page_fixture.*; + const value_offset = record.Header.len + + Header.len + + 1 + + 8; + invalid_payload[value_offset] = 'A'; + var reader: std.Io.Reader = .fixed(&invalid_payload); + try std.testing.expectError( + error.InvalidChecksum, + decode(&reader, std.testing.allocator), + ); + } + + { + var short_payload = test_empty_framed_page_fixture.*; + std.mem.writeInt(u32, short_payload[2..6], 36, .little); + var reader: std.Io.Reader = .fixed(&short_payload); + try std.testing.expectError( + error.EndOfStream, + decode(&reader, std.testing.allocator), + ); + } + + { + var trailing: [test_empty_framed_page_fixture.len + 1]u8 = undefined; + @memcpy( + trailing[0..test_empty_framed_page_fixture.len], + test_empty_framed_page_fixture, + ); + trailing[trailing.len - 1] = 0; + std.mem.writeInt(u32, trailing[2..6], 38, .little); + + var checksum: record.Checksum = .init(.page, 38); + try checksum.writer().writeAll(trailing[record.Header.len..]); + std.mem.writeInt(u32, trailing[6..10], checksum.final(), .little); + + var reader: std.Io.Reader = .fixed(&trailing); + try std.testing.expectError( + error.PayloadNotExhausted, + decode(&reader, std.testing.allocator), + ); + } +} + +test "framed PAGE rejects every truncation and preserves following bytes" { + for (0..test_empty_framed_page_fixture.len) |len| { + var reader: std.Io.Reader = .fixed( + test_empty_framed_page_fixture[0..len], + ); + if (decode(&reader, std.testing.allocator)) |decoded_value| { + var unexpected = decoded_value; + unexpected.deinit(); + return error.ExpectedDecodeFailure; + } else |_| {} + } + + var source: std.Io.Reader = .fixed( + test_empty_framed_page_fixture ++ "next", + ); + var decoded = try decode(&source, std.testing.allocator); + defer decoded.deinit(); + try std.testing.expectEqualStrings("next", try source.take(4)); } test "decode sparse page rejects every truncation" { @@ -731,7 +944,7 @@ test "decode sparse page rejects every truncation" { var reader: std.Io.Reader = .fixed(test_page_fixture[0..len]); try std.testing.expectError( error.EndOfStream, - decode(&reader, std.testing.allocator), + decodePayload(&reader, std.testing.allocator), ); } } @@ -766,7 +979,7 @@ test "decode accepts unordered sparse style IDs and rejects zero" { var descending_reader: std.Io.Reader = .fixed( descending_writer.buffered(), ); - var decoded_descending = try decode( + var decoded_descending = try decodePayload( &descending_reader, std.testing.allocator, ); @@ -793,7 +1006,7 @@ test "decode accepts unordered sparse style IDs and rejects zero" { var zero_reader: std.Io.Reader = .fixed(zero_writer.buffered()); try std.testing.expectError( error.InvalidStyleId, - decode(&zero_reader, std.testing.allocator), + decodePayload(&zero_reader, std.testing.allocator), ); } @@ -834,7 +1047,7 @@ test "decode accepts unordered sparse hyperlink IDs" { try grid.CellHeader.encode(.{}, &writer); var reader: std.Io.Reader = .fixed(writer.buffered()); - var decoded = try decode( + var decoded = try decodePayload( &reader, std.testing.allocator, ); @@ -868,7 +1081,7 @@ test "decode defaults missing sparse cell references" { try io.writeInt(&style_writer, u32, 0); var style_reader: std.Io.Reader = .fixed(style_writer.buffered()); - var style_page = try decode( + var style_page = try decodePayload( &style_reader, std.testing.allocator, ); @@ -902,7 +1115,7 @@ test "decode defaults missing sparse cell references" { var hyperlink_reader: std.Io.Reader = .fixed( hyperlink_writer.buffered(), ); - var hyperlink_page = try decode( + var hyperlink_page = try decodePayload( &hyperlink_reader, std.testing.allocator, ); @@ -935,7 +1148,7 @@ test "decode rejects undefined row and cell values" { var row_reader: std.Io.Reader = .fixed(row_writer.buffered()); try std.testing.expectError( error.InvalidRow, - decode(&row_reader, std.testing.allocator), + decodePayload(&row_reader, std.testing.allocator), ); var invalid_cell: [Header.len + 2]u8 = undefined; @@ -947,7 +1160,7 @@ test "decode rejects undefined row and cell values" { var cell_reader: std.Io.Reader = .fixed(cell_writer.buffered()); try std.testing.expectError( error.InvalidCell, - decode(&cell_reader, std.testing.allocator), + decodePayload(&cell_reader, std.testing.allocator), ); var invalid_codepoint: [Header.len + 1 + 16]u8 = undefined; @@ -965,7 +1178,7 @@ test "decode rejects undefined row and cell values" { ); try std.testing.expectError( error.InvalidCodepoint, - decode(&codepoint_reader, std.testing.allocator), + decodePayload(&codepoint_reader, std.testing.allocator), ); var invalid_color: [Header.len + 1 + 16]u8 = undefined; @@ -981,7 +1194,7 @@ test "decode rejects undefined row and cell values" { var color_reader: std.Io.Reader = .fixed(color_writer.buffered()); try std.testing.expectError( error.InvalidColor, - decode(&color_reader, std.testing.allocator), + decodePayload(&color_reader, std.testing.allocator), ); } @@ -1049,7 +1262,7 @@ test "decode validates dimensions and native table capacities" { var reader: std.Io.Reader = .fixed(writer.buffered()); try std.testing.expectError( case.expected, - decode(&reader, std.testing.allocator), + decodePayload(&reader, std.testing.allocator), ); } } @@ -1080,7 +1293,7 @@ test "decode rejects duplicate and default style entries" { var reader: std.Io.Reader = .fixed(writer.buffered()); try std.testing.expectError( error.DuplicateStyle, - decode(&reader, std.testing.allocator), + decodePayload(&reader, std.testing.allocator), ); const default_header: Header = .{ @@ -1102,7 +1315,7 @@ test "decode rejects duplicate and default style entries" { var default_reader: std.Io.Reader = .fixed(default_writer.buffered()); try std.testing.expectError( error.DefaultStyle, - decode(&default_reader, std.testing.allocator), + decodePayload(&default_reader, std.testing.allocator), ); } @@ -1133,7 +1346,7 @@ test "decode rejects duplicate hyperlinks with empty strings" { var reader: std.Io.Reader = .fixed(writer.buffered()); try std.testing.expectError( error.DuplicateHyperlink, - decode(&reader, std.testing.allocator), + decodePayload(&reader, std.testing.allocator), ); } @@ -1166,6 +1379,6 @@ test "decode reads hyperlink strings into page storage" { var reader: std.Io.Reader = .fixed(writer.buffered()); try std.testing.expectError( error.InvalidStringCapacity, - decode(&reader, std.testing.allocator), + decodePayload(&reader, std.testing.allocator), ); } diff --git a/src/terminal/snapshot/record.zig b/src/terminal/snapshot/record.zig index 7ef9e5a7a..5a99f8607 100644 --- a/src/terminal/snapshot/record.zig +++ b/src/terminal/snapshot/record.zig @@ -141,24 +141,94 @@ pub const Checksum = struct { } }; -/// A checksum-verifying reader limited to one record payload. +/// Builds one complete record. /// -/// Initialize this only after decoding a `Header`, decode the payload through -/// `reader`, and then call `finish`. The caller owns both buffers and chooses -/// their sizes. They only need to remain valid until `finish` returns. -pub const PayloadReader = struct { +/// This Writer requires an Allocating std.Io.Writer because the record +/// format requires reading the full payload and rewinding in order to +/// write the length + CRC without encoding twice. +pub const Writer = struct { + destination: *std.Io.Writer.Allocating, + tag: Tag, + record_start: usize, + + /// Reserve space for a record at the current end of `destination`. + /// Once this is called, callers MUST NOT write anything else to + /// the writer until `finish` or `cancel` is called. + pub fn init( + destination: *std.Io.Writer.Allocating, + tag: Tag, + ) std.Io.Writer.Error!Writer { + const record_start = destination.written().len; + errdefer destination.shrinkRetainingCapacity(record_start); + try destination.writer.splatByteAll(0, Header.len); + return .{ + .destination = destination, + .tag = tag, + .record_start = record_start, + }; + } + + /// Return the writer through which the payload is encoded exactly once. + pub fn payloadWriter(self: *Writer) *std.Io.Writer { + return &self.destination.writer; + } + + pub const FinishError = error{ + /// The payload cannot be represented by the record's `u32` length. + PayloadTooLarge, + }; + + /// Marked the completed record with the payload length and CRC32C. + pub fn finish(self: *Writer) FinishError!void { + const bytes = self.destination.written(); + const payload_start = self.record_start + Header.len; + const payload_len = std.math.cast( + u32, + bytes.len - payload_start, + ) orelse return error.PayloadTooLarge; + const payload = bytes[payload_start..]; + + // Calculate our CRC + var checksum: Checksum = .init(self.tag, payload_len); + checksum.writer().writeAll(payload) catch unreachable; + + // Build the header and encode it directly into the header + const header: Header = .{ + .tag = self.tag, + .payload_len = payload_len, + .crc32c = checksum.final(), + }; + var header_writer: std.Io.Writer = .fixed(bytes[self.record_start..payload_start]); + header.encode(&header_writer) catch unreachable; + } + + /// Discard this record. This makes it safe to use the underlying + /// alloating writer again as if nothing happened. + pub fn cancel(self: *Writer) void { + self.destination.shrinkRetainingCapacity(self.record_start); + } +}; + +/// Reads one complete record. +/// +/// `init` decodes the header, `payloadReader` returns a reader limited to the +/// declared payload, and `finish` verifies exact exhaustion and the CRC32C. +pub const Reader = struct { header: Header, + + // The limited reader normally streams directly into the hashing reader's + // buffer. One byte is enough for operations that require it to buffer, + // such as peek and discard. + limited_buffer: [1]u8, + + // PAGE decoding performs many small reads. 256 bytes batches several cells + // while CRC32C is calculated without making Reader large on the stack. + hashing_buffer: [256]u8, + limited: std.Io.Reader.Limited, hashing: std.Io.Reader.Hashed(Crc32c), - /// Caller-owned storage for the reader wrappers. - pub const Buffers = struct { - /// Buffer used to enforce the payload-length boundary. - limited: []u8, - - /// Buffer used while updating CRC32C over consumed payload bytes. - hashing: []u8, - }; + pub const InitError = Header.DecodeError; /// Errors detected after a payload decoder returns. pub const FinishError = error{ @@ -169,39 +239,40 @@ pub const PayloadReader = struct { PayloadNotExhausted, }; - /// Initialize a reader for the payload described by `header`. + /// Decode a record header and initialize its payload reader. /// - /// `self` and both buffers must remain at stable addresses until `finish` - /// returns. Decode only through the reader returned by `reader`. + /// `self` must remain at a stable address until `finish` returns. Decode + /// the payload only through the reader returned by `payloadReader`. pub fn init( - self: *PayloadReader, + self: *Reader, source: *std.Io.Reader, - header: Header, - buffers: Buffers, - ) void { + ) InitError!void { self.* = undefined; - self.header = header; + self.header = try Header.decode(source); self.limited = .init( source, - .limited(header.payload_len), - buffers.limited, + .limited(self.header.payload_len), + &self.limited_buffer, ); - const checksum: Checksum = .init(header.tag, header.payload_len); + const checksum: Checksum = .init( + self.header.tag, + self.header.payload_len, + ); self.hashing = .init( &self.limited.interface, checksum.hashing.hasher, - buffers.hashing, + &self.hashing_buffer, ); } /// Return the length-limited, checksum-updating payload reader. - pub fn reader(self: *PayloadReader) *std.Io.Reader { + pub fn payloadReader(self: *Reader) *std.Io.Reader { return &self.hashing.reader; } /// Require exact payload exhaustion and validate its CRC32C. - pub fn finish(self: *PayloadReader) FinishError!void { + pub fn finish(self: *Reader) FinishError!void { if (self.hashing.reader.bufferedLen() != 0 or self.limited.remaining != .nothing) { @@ -282,7 +353,7 @@ test "reject every header truncation" { } } -test "payload reader verifies exhaustion and checksum" { +test "record reader verifies exhaustion and checksum" { const payload = "snapshot payload"; var checksum: Checksum = .init(.page, payload.len); try checksum.writer().writeAll(payload); @@ -292,23 +363,22 @@ test "payload reader verifies exhaustion and checksum" { .crc32c = checksum.final(), }; - var source: std.Io.Reader = .fixed(payload ++ "next"); - var payload_reader: PayloadReader = undefined; - var limited_buf: [1]u8 = undefined; - var hashing_buf: [1]u8 = undefined; - payload_reader.init(&source, header, .{ - .limited = &limited_buf, - .hashing = &hashing_buf, - }); + var encoded: [Header.len + payload.len + 4]u8 = undefined; + var writer: std.Io.Writer = .fixed(&encoded); + try header.encode(&writer); + try writer.writeAll(payload ++ "next"); + var source: std.Io.Reader = .fixed(writer.buffered()); + var record_reader: Reader = undefined; + try record_reader.init(&source); var decoded: [payload.len]u8 = undefined; - try payload_reader.reader().readSliceAll(&decoded); - try payload_reader.finish(); + try record_reader.payloadReader().readSliceAll(&decoded); + try record_reader.finish(); try std.testing.expectEqualStrings(payload, &decoded); try std.testing.expectEqualStrings("next", try source.take(4)); } -test "payload reader rejects remaining bytes and invalid checksum" { +test "record reader rejects remaining bytes and invalid checksum" { const payload = "payload"; const header: Header = .{ .tag = .page, @@ -316,35 +386,30 @@ test "payload reader rejects remaining bytes and invalid checksum" { .crc32c = 0, }; + var encoded: [Header.len + payload.len]u8 = undefined; + var writer: std.Io.Writer = .fixed(&encoded); + try header.encode(&writer); + try writer.writeAll(payload); + { - var source: std.Io.Reader = .fixed(payload); - var payload_reader: PayloadReader = undefined; - var limited_buf: [1]u8 = undefined; - var hashing_buf: [1]u8 = undefined; - payload_reader.init(&source, header, .{ - .limited = &limited_buf, - .hashing = &hashing_buf, - }); - _ = try payload_reader.reader().takeByte(); + var source: std.Io.Reader = .fixed(writer.buffered()); + var record_reader: Reader = undefined; + try record_reader.init(&source); + _ = try record_reader.payloadReader().takeByte(); try std.testing.expectError( error.PayloadNotExhausted, - payload_reader.finish(), + record_reader.finish(), ); } { - var source: std.Io.Reader = .fixed(payload); - var payload_reader: PayloadReader = undefined; - var limited_buf: [1]u8 = undefined; - var hashing_buf: [1]u8 = undefined; - payload_reader.init(&source, header, .{ - .limited = &limited_buf, - .hashing = &hashing_buf, - }); - try payload_reader.reader().discardAll(payload.len); + var source: std.Io.Reader = .fixed(writer.buffered()); + var record_reader: Reader = undefined; + try record_reader.init(&source); + try record_reader.payloadReader().discardAll(payload.len); try std.testing.expectError( error.InvalidChecksum, - payload_reader.finish(), + record_reader.finish(), ); } } @@ -359,20 +424,58 @@ test "payload limit does not consume the next record" { .crc32c = checksum.final(), }; - var source: std.Io.Reader = .fixed(payload ++ "next"); - var payload_reader: PayloadReader = undefined; - var limited_buf: [1]u8 = undefined; - var hashing_buf: [1]u8 = undefined; - payload_reader.init(&source, header, .{ - .limited = &limited_buf, - .hashing = &hashing_buf, - }); + var encoded: [Header.len + payload.len + 4]u8 = undefined; + var writer: std.Io.Writer = .fixed(&encoded); + try header.encode(&writer); + try writer.writeAll(payload ++ "next"); + var source: std.Io.Reader = .fixed(writer.buffered()); + var record_reader: Reader = undefined; + try record_reader.init(&source); var too_long: [3]u8 = undefined; try std.testing.expectError( error.EndOfStream, - payload_reader.reader().readSliceAll(&too_long), + record_reader.payloadReader().readSliceAll(&too_long), ); - try payload_reader.finish(); + try record_reader.finish(); try std.testing.expectEqualStrings("next", try source.take(4)); } + +test "record writer appends and backpatches framing" { + var destination: std.Io.Writer.Allocating = .init(std.testing.allocator); + defer destination.deinit(); + try destination.writer.writeAll("prefix"); + + var record_writer = try Writer.init(&destination, .page); + try record_writer.payloadWriter().writeAll("payload"); + try record_writer.finish(); + + const encoded = destination.written(); + try std.testing.expectEqualStrings("prefix", encoded[0..6]); + + var source: std.Io.Reader = .fixed(encoded[6..]); + var record_reader: Reader = undefined; + try record_reader.init(&source); + try std.testing.expectEqual(Tag.page, record_reader.header.tag); + try std.testing.expectEqual( + @as(u32, 7), + record_reader.header.payload_len, + ); + try std.testing.expectEqualStrings( + "payload", + try record_reader.payloadReader().take(7), + ); + try record_reader.finish(); +} + +test "record writer cancel preserves preceding bytes" { + var destination: std.Io.Writer.Allocating = .init(std.testing.allocator); + defer destination.deinit(); + try destination.writer.writeAll("prefix"); + + var record_writer = try Writer.init(&destination, .page); + try record_writer.payloadWriter().writeAll("partial"); + record_writer.cancel(); + + try std.testing.expectEqualStrings("prefix", destination.written()); +}