terminal/snapshot: add incremental decoder

This adds a new `terminal.snapshot.Decoder` that allows for incremental
decoding of a snapshot stream. There are two methods: `ready` builds up
the entire terminal up to READY. Then `next` acts like a Zig iterator
and applies incremental history as it becomes available. In between
calls to `ready` and `next` the caller can do whatever.
This commit is contained in:
Mitchell Hashimoto
2026-08-02 15:29:18 -07:00
parent d351d9ce07
commit e37865bedc
5 changed files with 1097 additions and 172 deletions

View File

@@ -19,6 +19,10 @@
//! * `decode` restores one complete snapshot per loop from bytes prepared
//! during setup. The restored terminal is destroyed inside the step, so
//! this measures the complete restore lifecycle.
//! * `decode-ready` restores only the renderable prefix per loop using the
//! incremental decoder and stops at READY, skipping history. Compared
//! against `decode`, this is the time-to-interactive win of applying
//! history off the critical path.
//! * `report` encodes once and prints total and per-record-tag sizes. It
//! is for inspecting the wire shape, not timing comparisons.
//!
@@ -92,6 +96,9 @@ pub const Mode = enum {
/// Restore one complete snapshot per loop, including its teardown.
decode,
/// Restore one renderable READY prefix per loop, including teardown.
@"decode-ready",
/// Print encoded sizes by record tag. Not a timing benchmark.
report,
};
@@ -122,6 +129,7 @@ pub fn benchmark(self: *TerminalSnapshot) Benchmark {
.noop => stepNoop,
.encode => stepEncode,
.decode => stepDecode,
.@"decode-ready" => stepDecodeReady,
.report => stepReport,
},
.setupFn = setup,
@@ -231,6 +239,28 @@ fn stepDecode(ptr: *anyopaque) Benchmark.Error!void {
}
}
/// The renderable-prefix half of decode: everything through READY, leaving
/// all history unread. The gap between this and `stepDecode` is the latency
/// class the incremental decoder removes from time-to-interactive.
fn stepDecodeReady(ptr: *anyopaque) Benchmark.Error!void {
const self: *TerminalSnapshot = @ptrCast(@alignCast(ptr));
const bytes = self.encoded.written();
for (0..self.opts.loops) |_| {
var reader: std.Io.Reader = .fixed(bytes);
var decoder: snapshot.Decoder = .init(&reader);
var decoded = decoder.ready(
self.alloc,
global.io(),
.{ .max_continuation_bytes = 1024 * 1024 },
) catch |err| {
log.warn("snapshot READY decoding failed err={}", .{err});
return error.BenchmarkFailed;
};
std.mem.doNotOptimizeAway(&decoded);
decoded.deinit(self.alloc);
}
}
/// Print the encoded size grouped by record tag. This shares the encoder
/// with encode mode but deliberately makes no timing claims.
fn stepReport(ptr: *anyopaque) Benchmark.Error!void {
@@ -305,3 +335,16 @@ test "TerminalSnapshot decode round trip" {
const bench = impl.benchmark();
_ = try bench.run(.once);
}
test "TerminalSnapshot decode READY prefix" {
const testing = std.testing;
const impl: *TerminalSnapshot = try .create(testing.allocator, .{
.mode = .@"decode-ready",
.@"terminal-rows" = 4,
.@"terminal-cols" = 8,
});
defer impl.destroy(testing.allocator);
const bench = impl.benchmark();
_ = try bench.run(.once);
}

View File

@@ -193,6 +193,40 @@ pub const DecodeError = Decoder.InitError ||
UnexpectedScreenKey,
};
/// Errors possible while restoring one history PAGE into a native Screen.
pub const DecodePageError = Allocator.Error ||
page.DecodeError ||
TerminalPageList.PageAllocation.FinalizeError;
/// Restore the next history PAGE record and prepend it to the native Screen.
///
/// Returns the number of rows added above the screen's existing content.
/// The page is decoded directly into a detached PageList-pooled allocation
/// and prepended only after its record validates, so a failure leaves the
/// screen unchanged. A limit failure from `finalize` occurs after the record
/// bytes were fully consumed, leaving `source` aligned on the next record.
pub fn decodePage(
source: *std.Io.Reader,
alloc: Allocator,
terminal_screen: *TerminalScreen,
) DecodePageError!usize {
// PAGE exposes its exact capacity before decoding the payload, allowing
// the destination PageList to allocate the final backing memory once.
var decoder: page.Decoder = undefined;
try decoder.init(source);
var allocation = try terminal_screen.pages.allocatePage(
decoder.capacity(),
);
defer allocation.deinit();
try decoder.decode(allocation.page(), alloc);
const rows = allocation.page().size.rows;
const contains_prompt = hasSemanticPrompt(allocation.page());
try allocation.finalize(.prepend);
if (contains_prompt) terminal_screen.semantic_prompt.seen = true;
return rows;
}
/// A decoded HISTORY manifest ready to restore its following PAGE records.
///
/// Keeping manifest decoding separate lets the full snapshot wrapper route a
@@ -209,13 +243,10 @@ pub const Decoder = struct {
UnexpectedRecordTag,
};
pub const RestoreError = Allocator.Error ||
page.DecodeError ||
TerminalPageList.PageAllocation.FinalizeError ||
error{
/// The Screen already contains complete pages before its active page.
ExistingHistory,
};
pub const RestoreError = DecodePageError || error{
/// The Screen already contains complete pages before its active page.
ExistingHistory,
};
/// Decode and finish the self-contained HISTORY manifest.
pub fn init(self: *Decoder, source: *std.Io.Reader) InitError!void {
@@ -245,20 +276,7 @@ pub const Decoder = struct {
// Native row totals remain derived from the actual PAGE dimensions.
for (0..self.header.page_count) |_| {
// PAGE exposes its exact capacity before decoding the payload,
// allowing the destination PageList to allocate the final backing
// memory once.
var decoder: page.Decoder = undefined;
try decoder.init(self.source);
var allocation = try terminal_screen.pages.allocatePage(
decoder.capacity(),
);
defer allocation.deinit();
try decoder.decode(allocation.page(), alloc);
const contains_prompt = hasSemanticPrompt(allocation.page());
try allocation.finalize(.prepend);
if (contains_prompt) terminal_screen.semantic_prompt.seen = true;
_ = try decodePage(self.source, alloc, terminal_screen);
}
terminal_screen.pages.assertIntegrity();

View File

@@ -10,13 +10,11 @@
//! To do that, it sends the active terminal state followed by a READY record,
//! then complete history.
//!
//! READY denotes that enough authenticated state is present to render the
//! terminal and reconstruct its unfinished standard Stream state. The current
//! synchronous decoder still returns only after FINISH. A caller moves the
//! Terminal into final storage, replays CONTINUATION once, and only then applies
//! PTY bytes belonging after the snapshot cut.
//! READY denotes that enough state has been sent down to render the
//! terminal and reconstruct its unfinished VT Stream state.
//!
//! After READY, we send history pages (scrollback).
//! After READY, we send history pages (scrollback). Finally, the
//! snapshot ends with a FINISH payload.
//!
//! ## Snapshot Format
//!
@@ -132,6 +130,27 @@
//! Use `snapshot.decodeExact` for a bounded file or buffer that must contain
//! only one snapshot. It preserves the stricter end-of-file check, which may
//! block when used with a live stream.
//!
//! `snapshot.Decoder` decodes the same stream incrementally so the terminal
//! becomes usable at READY, before history has arrived. Each `next` call
//! applies one history page to the by-then live terminal and returning null
//! validates FINISH:
//!
//! ```zig
//! var decoder: snapshot.Decoder = .init(&reader);
//! var decoded = try decoder.ready(alloc, io, .{
//! .max_continuation_bytes = 1024 * 1024,
//! });
//! defer decoded.deinit(alloc);
//!
//! var terminal = decoded.toOwned();
//! defer terminal.deinit(alloc);
//!
//! // Render, replay the continuation, process input...
//! while (try decoder.next(alloc, &terminal)) |progress| {
//! _ = progress; // Scrollback grew.
//! }
//! ```
pub const checkpoint = @import("checkpoint.zig");
pub const continuation = @import("continuation.zig");
@@ -153,6 +172,7 @@ pub const Continuation = codec.Continuation;
pub const EncodeOptions = codec.EncodeOptions;
pub const DecodeOptions = codec.DecodeOptions;
pub const Decoded = codec.Decoded;
pub const Decoder = codec.Decoder;
pub const encode = codec.encode;
pub const decode = codec.decode;
pub const decodeExact = codec.decodeExact;

View File

@@ -277,6 +277,34 @@ pub const Decoder = struct {
}
};
/// Errors possible while discarding one PAGE record without decoding it.
pub const DiscardError = record.Reader.InitError ||
record.Reader.FinishError ||
std.Io.Reader.Error ||
error{
/// The next record is valid but is not a PAGE.
UnexpectedRecordTag,
};
/// Consume exactly one complete PAGE record, validating its framing and
/// CRC32C while discarding the payload without structural validation.
///
/// The payload bytes still stream through `source`, so an enclosing running
/// digest (such as the FINISH checkpoint) continues to cover them. This lets
/// a decoder stay aligned with the record sequence, and keep authenticating
/// it, while dropping page content it can no longer apply.
pub fn discard(source: *std.Io.Reader) DiscardError!void {
var record_reader: record.Reader = undefined;
try record_reader.init(source);
if (record_reader.header.tag != .page) {
return error.UnexpectedRecordTag;
}
try record_reader.payloadReader().discardAll(
record_reader.header.payload_len,
);
try record_reader.finish();
}
/// Encode a PAGE payload directly from a native page.
fn encodePayload(
page: *const TerminalPage,
@@ -1421,6 +1449,76 @@ test "decode reuses duplicate hyperlinks" {
);
}
test "discard consumes exactly one PAGE record and keeps digest coverage" {
const testing = std.testing;
// Discard validates framing only, so an arbitrary payload keeps this test
// focused on record consumption rather than page structure.
var encoded: std.Io.Writer.Allocating = .init(testing.allocator);
defer encoded.deinit();
var stream: record.Writer = .init(testing.allocator, &encoded.writer);
defer stream.deinit();
{
const payload = stream.begin(.page);
errdefer stream.cancel();
try payload.writeAll("undecodable page payload");
try stream.finish();
}
const record_len = encoded.written().len;
try encoded.writer.writeAll("next");
// Discarded payload bytes must still update an enclosing running digest,
// which is what lets a snapshot FINISH checkpoint authenticate records
// whose content was dropped.
var source: std.Io.Reader = .fixed(encoded.written());
var hashed: record.StreamReader = .init(&source);
try discard(hashed.reader());
var expected_digest: record.PrefixDigest = undefined;
std.crypto.hash.Blake3.hash(
encoded.written()[0..record_len],
&expected_digest,
.{},
);
try testing.expectEqual(expected_digest, hashed.prefixDigest());
try testing.expectEqualStrings("next", try source.take(4));
// Only PAGE records may be discarded; the tag remains strict.
var wrong_tag: std.Io.Writer.Allocating = .init(testing.allocator);
defer wrong_tag.deinit();
var wrong_tag_stream: record.Writer = .init(
testing.allocator,
&wrong_tag.writer,
);
defer wrong_tag_stream.deinit();
{
const payload = wrong_tag_stream.begin(.history);
errdefer wrong_tag_stream.cancel();
try payload.writeAll("payload");
try wrong_tag_stream.finish();
}
var wrong_tag_source: std.Io.Reader = .fixed(wrong_tag.written());
try testing.expectError(
error.UnexpectedRecordTag,
discard(&wrong_tag_source),
);
// Discarded bytes are still covered by the record CRC.
const corrupted = try testing.allocator.dupe(
u8,
encoded.written()[0..record_len],
);
defer testing.allocator.free(corrupted);
corrupted[corrupted.len - 1] ^= 1;
var corrupted_source: std.Io.Reader = .fixed(corrupted);
try testing.expectError(error.InvalidChecksum, discard(&corrupted_source));
// A truncated payload is detected before `finish`.
var truncated_source: std.Io.Reader = .fixed(
encoded.written()[0 .. record_len - 1],
);
try testing.expectError(error.EndOfStream, discard(&truncated_source));
}
test "decode ignores empty hyperlink strings" {
const header: Header = .{
.columns = 1,

File diff suppressed because it is too large Load Diff