a73x

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