terminal/snapshot: page records

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");

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),
     );
 }

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());
+}