src/engine/delta.zig
Ref: Size: 20.1 KiB History
//! The daemon's half of the delta stream: this side holds the authoritative
//! grid and decides WHICH rows a client is missing, while the far side applies
//! whatever payload arrives. protocol.zig owns the bytes; this owns the rows.
//! Engine and protocol are its whole world, so a test drives the tracker with
//! nothing but an engine.
const std = @import("std");
const Engine = @import("engine.zig").Engine;
const proto = @import("term").protocol;
const Wyhash = std.hash.Wyhash;
/// A content hash per viewport row, so an update sends only the rows that
/// changed. It advances whether or not a client is attached, so a reattach can
/// be answered by delta — but with nobody attached it advances through
/// `noteBlind`, because hashing means RENDERING every row.
pub const DeltaTracker = struct {
seq: u64 = 0,
/// Seq at the last discontinuity (init/resize/screen switch). Clients
/// with have_seq older than this cannot be served a delta.
reset_seq: u64 = 0,
cols: u16 = 0,
rows: u16 = 0,
on_alt: bool = false,
cursor: Engine.CursorPos = .{ .x = 0, .y = 0 },
history_rows: u32 = 0,
row_hashes: []u64 = &.{},
row_seqs: []u64 = &.{},
/// Where update() writes the hashes it is computing, so a diff pass
/// allocates nothing. Swapped with row_hashes once per advance.
scratch_hashes: []u64 = &.{},
/// Set by noteBlind: row_hashes describe the grid from BEFORE a stretch
/// nobody watched, so they cannot be compared against. Not a tuning
/// flag — dropping it diverges the client silently, and only on a row
/// that changes and changes back. See the revert test.
hashes_stale: bool = false,
pub fn deinit(self: *DeltaTracker, alloc: std.mem.Allocator) void {
alloc.free(self.row_hashes);
alloc.free(self.row_seqs);
alloc.free(self.scratch_hashes);
}
/// Re-hash every row and mark a discontinuity: what follows can only be
/// carried by a full snapshot.
///
/// The daemon calls this through `Server.rebuildTracker` and nowhere else,
/// deliberately: moving `reset_seq` makes a recorded side-channel event
/// undeliverable, and that wrapper is where the event is dropped. A new
/// caller here leaves the user's copied text resident in a daemon that can
/// no longer give it to anyone.
pub fn rebuild(self: *DeltaTracker, alloc: std.mem.Allocator, eng: *Engine, rows: u16, cols: u16) !void {
if (self.row_hashes.len != rows) {
// Every allocation lands before any old array is released, so a
// failure here leaves the tracker exactly as it was rather than
// half-freed. update()'s geometry check turns the resulting
// stale-but-consistent state into a snapshot on the next pump.
const hashes = try alloc.alloc(u64, rows);
errdefer alloc.free(hashes);
const seqs = try alloc.alloc(u64, rows);
errdefer alloc.free(seqs);
const scratch = try alloc.alloc(u64, rows);
alloc.free(self.row_hashes);
alloc.free(self.row_seqs);
alloc.free(self.scratch_hashes);
self.row_hashes = hashes;
self.row_seqs = seqs;
self.scratch_hashes = scratch;
}
self.cols = cols;
self.on_alt = eng.onAltScreen();
self.seq += 1;
self.reset_seq = self.seq;
self.cursor = eng.cursorPos();
self.history_rows = eng.historyRows();
// Claim no rows until every row is stamped: an encode that fails partway
// leaves stale seqs in the tail of `row_seqs`, and any above a client's
// `have_seq` puts that row in every delta from here on.
self.rows = 0;
for (0..rows) |y| {
const bytes = try eng.encodeViewportRow(alloc, @intCast(y));
defer alloc.free(bytes);
self.row_hashes[y] = Wyhash.hash(0, bytes);
self.row_seqs[y] = self.seq;
}
self.rows = rows;
self.hashes_stale = false;
}
pub const Update = union(enum) {
none,
discontinuity,
/// Something changed and the tracker now describes it; the payload
/// is built separately, and only if a client is there to read it.
advanced,
};
/// Whether the tracked state can still be described as a delta against
/// `eng`, or whether the client has to be resynced from scratch.
///
/// Geometry is load-bearing, not decorative: row_hashes is indexed by the
/// tracker's own row count, and encodeViewportRow asserts against the
/// engine's. A tracker left stale by a failed rebuild resyncs here
/// instead of running
/// off the end of the grid. A tracker that was never built has no rows to
/// diff at all, and an alt-screen flip replaces the whole grid, so no row
/// seq from before it means anything.
///
/// Both `update` and `noteBlind` advance the tracker, so both must ask
/// this before they touch a row seq.
fn continuous(self: *const DeltaTracker, eng: *Engine) bool {
if (self.rows == 0) return false;
if (self.rows != eng.term.rows or self.cols != eng.term.cols) return false;
return eng.onAltScreen() == self.on_alt;
}
/// Diff current engine state against the tracked state. Advances seq
/// and tracked rows when anything changed. Allocates only the per-row
/// encodings it hashes.
pub fn update(self: *DeltaTracker, alloc: std.mem.Allocator, eng: *Engine) !Update {
if (!self.continuous(eng)) return .discontinuity;
// Safe to stamp row_seqs with the seq we may not end up taking: it
// is only written for rows whose hash changed, and any changed row
// forces the advance below.
const next_seq = self.seq + 1;
var changed: usize = 0;
for (0..self.rows) |y| {
const bytes = try eng.encodeViewportRow(alloc, @intCast(y));
defer alloc.free(bytes);
const hash = Wyhash.hash(0, bytes);
self.scratch_hashes[y] = hash;
if (self.hashes_stale or hash != self.row_hashes[y]) {
self.row_seqs[y] = next_seq;
changed += 1;
}
}
const cur = eng.cursorPos();
const hist = eng.historyRows();
const cursor_moved = cur.x != self.cursor.x or cur.y != self.cursor.y;
if (changed == 0 and !cursor_moved and hist == self.history_rows)
return .none;
self.seq = next_seq;
self.cursor = cur;
self.history_rows = hist;
std.mem.swap([]u64, &self.row_hashes, &self.scratch_hashes);
// Every row was just rendered and hashed, so nothing pre-blind
// survives. Cannot be skipped: while the flag is set every row
// counts as changed, so an advance is the only way out of it.
self.hashes_stale = false;
return .advanced;
}
/// Records that the grid moved WITHOUT rendering a row: hashing by rendering
/// dominates the daemon on a full-width repaint, and with nobody attached
/// those bytes go nowhere. Takes no allocator — a signature that can
/// allocate is one that can render.
pub fn noteBlind(self: *DeltaTracker, eng: *Engine) Update {
if (!self.continuous(eng)) return .discontinuity;
self.seq += 1;
for (self.row_seqs) |*row_seq| row_seq.* = self.seq;
self.hashes_stale = true;
self.cursor = eng.cursorPos();
self.history_rows = eng.historyRows();
return .advanced;
}
/// Only the tracker's half: whether `have_seq` is in the span it can
/// still describe. The epoch is the caller's; 0 means the client holds
/// nothing.
pub fn canServe(self: *const DeltaTracker, have_seq: u64) bool {
return have_seq != 0 and
have_seq >= self.reset_seq and have_seq <= self.seq and
self.rows != 0;
}
/// Build a delta payload of all rows changed after `since`. The header's
/// `row_count` and the appended rows MUST agree, so both come from the same
/// predicate with nothing mutating in between. Changed rows are encoded
/// twice — once to hash, once to send — which is 1-3 rows in steady state.
pub fn buildDeltaSince(self: *DeltaTracker, alloc: std.mem.Allocator, eng: *Engine, since: u64) ![]u8 {
var rows_changed: u16 = 0;
for (self.row_seqs) |s| {
if (s > since) rows_changed += 1;
}
var payload: std.ArrayList(u8) = .empty;
errdefer payload.deinit(alloc);
try proto.appendDeltaHeader(&payload, alloc, .{
.seq = self.seq,
.history_rows = self.history_rows,
.cursor_x = self.cursor.x,
.cursor_y = self.cursor.y,
.row_count = rows_changed,
});
for (self.row_seqs, 0..) |s, y| {
if (s <= since) continue;
const bytes = try eng.encodeViewportRow(alloc, @intCast(y));
defer alloc.free(bytes);
try proto.appendDeltaRow(&payload, alloc, @intCast(y), bytes);
}
return payload.toOwnedSlice(alloc);
}
};
/// The snapshot payload: the prefix, the cursor, then every viewport row as
/// cells. The cursor rides in the body rather than in the prefix so the
/// prefix keeps the layout its golden pin and its readers already have.
pub fn buildSnapshot(alloc: std.mem.Allocator, eng: *Engine, prefix: proto.SnapshotPrefix) ![]u8 {
var out: std.ArrayList(u8) = .empty;
errdefer out.deinit(alloc);
var pbuf: [proto.snapshot_prefix_len]u8 = undefined;
proto.writeSnapshotPrefix(&pbuf, prefix);
try out.appendSlice(alloc, &pbuf);
var cbuf: [proto.snapshot_cursor_len]u8 = undefined;
const cur = eng.cursorPos();
proto.writeSnapshotCursor(&cbuf, cur.x, cur.y);
try out.appendSlice(alloc, &cbuf);
var y: u16 = 0;
while (y < eng.term.rows) : (y += 1) {
const row = try eng.encodeViewportRow(alloc, y);
defer alloc.free(row);
try out.appendSlice(alloc, row);
}
return out.toOwnedSlice(alloc);
}
// ---------------------------------------------------------------------------
// Tests. A delta payload is rows of cells now, so the assertions decode it the
// way a client does — through grid.decodeRow — rather than searching bytes.
const grid = @import("term").grid;
const Replica = @import("term").replica.Replica;
/// Apply a delta payload into `g` and return the plain text, so a test can
/// say what the far side would be SHOWING after the frame.
///
/// Through `Replica.apply` rather than a loop over the rows: that is the one
/// replay core, and a fixture with its own applier grades payloads by rules
/// the product does not use. A payload the product would resync on is an
/// error here, since every caller builds one it expects to land.
fn deltaText(alloc: std.mem.Allocator, g: *grid.Grid, payload: []const u8) ![]const u8 {
var r = Replica.init(alloc, g);
const applied = try r.apply(.delta, payload);
if (applied != .painted) return error.DeltaRefused;
return g.dumpPlain(alloc);
}
test "DeltaTracker: alt-screen flip is a discontinuity and rows follow the active screen" {
const alloc = std.testing.allocator;
const eng = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });
defer eng.deinit();
var tracker: DeltaTracker = .{};
defer tracker.deinit(alloc);
try tracker.rebuild(alloc, eng, 24, 80);
eng.feed("primary text");
switch (try tracker.update(alloc, eng)) {
.advanced => {},
else => return error.ExpectedAdvance,
}
// Switching screens replaces every row at once: the tracked hashes
// describe the other screen, so a delta would be a lie.
eng.feed("\x1b[?1049h");
switch (try tracker.update(alloc, eng)) {
.discontinuity => {},
else => return error.ExpectedDiscontinuity,
}
try tracker.rebuild(alloc, eng, 24, 80);
eng.feed("alt content");
switch (try tracker.update(alloc, eng)) {
.advanced => {},
else => return error.ExpectedAdvance,
}
const payload = try tracker.buildDeltaSince(alloc, eng, tracker.seq - 1);
defer alloc.free(payload);
const g = try grid.Grid.init(alloc, 80, 24);
defer g.deinit();
const text = try deltaText(alloc, g, payload);
defer alloc.free(text);
try std.testing.expect(std.mem.indexOf(u8, text, "alt content") != null);
}
test "DeltaTracker: a resize behind the tracker's back resyncs instead of over-reading" {
const alloc = std.testing.allocator;
const eng = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });
defer eng.deinit();
var tracker: DeltaTracker = .{};
defer tracker.deinit(alloc);
try tracker.rebuild(alloc, eng, 24, 80);
// Stands in for a rebuild that failed (OOM) after the engine resized:
// the tracker still describes 24 rows of an 80-column grid.
try eng.resize(80, 10);
switch (try tracker.update(alloc, eng)) {
.discontinuity => {},
else => return error.ExpectedDiscontinuity,
}
}
test "DeltaTracker: blind output is answerable on reattach without rendering a row" {
const alloc = std.testing.allocator;
const eng = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });
defer eng.deinit();
var tracker: DeltaTracker = .{};
defer tracker.deinit(alloc);
try tracker.rebuild(alloc, eng, 24, 80);
// What a client held in its hand when it detached.
const have = tracker.seq;
eng.feed("printed with nobody watching\r\n");
switch (tracker.noteBlind(eng)) {
.advanced => {},
else => return error.ExpectedAdvance,
}
// The whole point of the blind path is that it stamps rows it never
// looked at, so the reattach delta has to carry the row anyway.
try std.testing.expect(tracker.canServe(have));
const payload = try tracker.buildDeltaSince(alloc, eng, have);
defer alloc.free(payload);
const g = try grid.Grid.init(alloc, 80, 24);
defer g.deinit();
const text = try deltaText(alloc, g, payload);
defer alloc.free(text);
try std.testing.expect(
std.mem.indexOf(u8, text, "printed with nobody watching") != null,
);
// The HEADER, not only the rows: `buildDeltaSince` serialises the tracker's
// cursor and `history_rows`, so a blind path that stamped every row and
// forgot those two repaints the right text with the cursor parked where the
// gap began. Any reattach quoting a held seq wears it.
const hdr = try proto.readDeltaHeader(payload);
const cur = eng.cursorPos();
try std.testing.expectEqual(cur.x, hdr.cursor_x);
try std.testing.expectEqual(cur.y, hdr.cursor_y);
try std.testing.expectEqual(eng.historyRows(), hdr.history_rows);
}
test "DeltaTracker: a blind chunk that changed nothing still advances seq" {
// The premise two comments in server.zig rest on. Side-channel events are
// stamped at `tracker.seq` and replayed only strictly ABOVE a reattaching
// client's watermark — and `update()` answers `.none` for a chunk that moved
// no cell, so a bare BEL during a gap is stamped where the gap began and
// never replayed. `noteBlind` has no `.none`, so those events survive.
const alloc = std.testing.allocator;
const eng = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });
defer eng.deinit();
var tracker: DeltaTracker = .{};
defer tracker.deinit(alloc);
try tracker.rebuild(alloc, eng, 24, 80);
// No feed at all: the strongest form of "this chunk changed nothing".
const before = tracker.seq;
switch (tracker.noteBlind(eng)) {
.advanced => {},
else => return error.ExpectedAdvance,
}
try std.testing.expect(tracker.seq > before);
}
test "DeltaTracker: two blind stretches never share a seq" {
// last_return.seq is stamped from tracker.seq. Two commands returning
// while nobody is attached must land on different seqs or an await
// cannot tell which of them it was just told about.
const alloc = std.testing.allocator;
const eng = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });
defer eng.deinit();
var tracker: DeltaTracker = .{};
defer tracker.deinit(alloc);
try tracker.rebuild(alloc, eng, 24, 80);
eng.feed("one\r\n");
_ = tracker.noteBlind(eng);
const first = tracker.seq;
eng.feed("two\r\n");
_ = tracker.noteBlind(eng);
try std.testing.expect(tracker.seq > first);
}
test "DeltaTracker: a row that reverts after a blind stretch is still sent" {
// Why stale hashes are unsafe, silently. Blind output moves a row A -> B and
// the reattach delta carries it, so the client holds B — but the STORED hash
// still describes A. If the row later moves B -> A, the comparison says
// "unchanged" and the client shows B for the rest of the session.
const alloc = std.testing.allocator;
const eng = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });
defer eng.deinit();
var tracker: DeltaTracker = .{};
defer tracker.deinit(alloc);
eng.feed("\x1b[1;1HAAA");
try tracker.rebuild(alloc, eng, 24, 80);
eng.feed("\x1b[1;1HBBB");
_ = tracker.noteBlind(eng);
// Stands in for the reattaching client: this is what it was handed.
const reattach = try tracker.buildDeltaSince(alloc, eng, tracker.seq - 1);
defer alloc.free(reattach);
// Back to exactly the bytes the pre-blind hashes describe. The cursor
// lands in the same column both times and no history is added, so a
// hash comparison is the only thing that can catch this.
eng.feed("\x1b[1;1HAAA");
switch (try tracker.update(alloc, eng)) {
.advanced => {},
else => return error.RevertWentUnreported,
}
const payload = try tracker.buildDeltaSince(alloc, eng, tracker.seq - 1);
defer alloc.free(payload);
const g = try grid.Grid.init(alloc, 80, 24);
defer g.deinit();
const text = try deltaText(alloc, g, payload);
defer alloc.free(text);
try std.testing.expect(std.mem.indexOf(u8, text, "AAA") != null);
}
test "DeltaTracker: a screen switch with nobody watching is still a discontinuity" {
// The blind path may skip rows; it may not skip the events it cannot
// describe at all, or the rebuild that resyncSnapshot owes a detached
// session never happens.
const alloc = std.testing.allocator;
const eng = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });
defer eng.deinit();
var tracker: DeltaTracker = .{};
defer tracker.deinit(alloc);
try tracker.rebuild(alloc, eng, 24, 80);
eng.feed("\x1b[?1049h");
switch (tracker.noteBlind(eng)) {
.discontinuity => {},
else => return error.ExpectedDiscontinuity,
}
}
test "buildSnapshot: prefix, cursor and one CellRow per viewport row" {
const alloc = std.testing.allocator;
const eng = try Engine.init(alloc, .{ .cols = 4, .rows = 2 });
defer eng.deinit();
eng.feed("a\r\nb");
const payload = try buildSnapshot(alloc, eng, .{
.seq = 5,
.history_rows = 1,
.cols = 4,
.rows = 2,
.epoch = 0x99,
});
defer alloc.free(payload);
const prefix = try proto.readSnapshotPrefix(payload);
try std.testing.expectEqual(@as(u64, 5), prefix.seq);
try std.testing.expectEqual(@as(u16, 4), prefix.cols);
try std.testing.expectEqual(@as(u16, 2), prefix.rows);
// The cursor is the engine's own, not a value the caller passed in: a
// snapshot that carried the prefix's idea of it would park the caret
// wherever the last resize left it.
const cur = try proto.readSnapshotCursor(payload);
const eng_cur = eng.cursorPos();
try std.testing.expectEqual(eng_cur.x, cur.x);
try std.testing.expectEqual(eng_cur.y, cur.y);
const body = payload[proto.snapshot_prefix_len + proto.snapshot_cursor_len ..];
const rows = try grid.decodeRows(alloc, body, 2, 4);
defer grid.freeRows(alloc, rows);
try std.testing.expectEqualStrings("a", rows[0].textOf(rows[0].cells[0]));
try std.testing.expectEqualStrings("b", rows[1].textOf(rows[1].cells[0]));
}
// Forces semantic analysis of every pub decl under `zig build test`, so an
// unreferenced decl must at least compile (the silent-module-loss hazard,
// decisions.md). Pub decls only: std.meta.declarations sees nothing private.
test {
std.testing.refAllDeclsRecursive(@This());
}