a73x

bdc6eeb5

refactor: replica.zig owns replay — extracted from session(), finally unit-tested

a73x   2026-08-13 10:13

Commit message
refactor: replica.zig owns replay — extracted from session(), finally unit-tested

The replay core (snapshot/delta application, resume coordinates,
authoritative-grid tracking, scroll-page math) moves out of session()'s
local variables into a Replica struct that BORROWS its engine — the
caller stays the one owner. client.zig delegates; server.zig's
applyFrame test helper now replays through the production code instead
of a hand-rolled twin; the module is platform-free by construction so it
can compile for wasm32 (M-web Task 4).

Seven new unit tests cover what was e2e-only before: prefix adoption,
grid-following resize, short-snapshot BadPayload, delta advance, the
resync-on-garbled-delta path (including the arrival-proves-admission
subtlety), attach quoting, and scrollStart saturation.

make test green; make e2e green (20 scenarios, 33 convergence points).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
build.zig
Old New
@@ -184,6 +184,18 @@ pub fn build(b: *std.Build) void {
184 delta_mod.addImport("engine", engine_mod); 184 delta_mod.addImport("engine", engine_mod);
185 delta_mod.addImport("protocol", protocol_mod); 185 delta_mod.addImport("protocol", protocol_mod);
186 186
187 // The replay core: snapshot/delta application and the resume
188 // coordinates, shared by the CLI client, the wasm core, and the
189 // server's test fixtures. Engine plus protocol and nothing else, and
190 // deliberately platform-free — it must compile for wasm32.
191 const replica_mod = b.createModule(.{
192 .root_source_file = b.path("src/replica.zig"),
193 .target = target,
194 .optimize = optimize,
195 });
196 replica_mod.addImport("engine", engine_mod);
197 replica_mod.addImport("protocol", protocol_mod);
198
187 // The socket path's identity and the right to bind it: the stale-socket 199 // The socket path's identity and the right to bind it: the stale-socket
188 // claim and the dev+ino record teardown compares against. A leaf — it 200 // claim and the dev+ino record teardown compares against. A leaf — it
189 // takes a path and nothing else, and knows no Server exists. 201 // takes a path and nothing else, and knows no Server exists.
@@ -204,6 +216,7 @@ pub fn build(b: *std.Build) void {
204 server_mod.addImport("pty", pty_mod); 216 server_mod.addImport("pty", pty_mod);
205 server_mod.addImport("protocol", protocol_mod); 217 server_mod.addImport("protocol", protocol_mod);
206 server_mod.addImport("delta", delta_mod); 218 server_mod.addImport("delta", delta_mod);
219 server_mod.addImport("replica", replica_mod);
207 server_mod.addImport("sockpath", sockpath_mod); 220 server_mod.addImport("sockpath", sockpath_mod);
208 // Both: the listener it owns, and the vocabulary it names directly 221 // Both: the listener it owns, and the vocabulary it names directly
209 // (the key it loads, the idle default it falls back to). 222 // (the key it loads, the idle default it falls back to).
@@ -258,6 +271,7 @@ pub fn build(b: *std.Build) void {
258 }); 271 });
259 client_mod.addImport("engine", engine_mod); 272 client_mod.addImport("engine", engine_mod);
260 client_mod.addImport("protocol", protocol_mod); 273 client_mod.addImport("protocol", protocol_mod);
274 client_mod.addImport("replica", replica_mod);
261 client_mod.addImport("testtmp", testtmp_mod); 275 client_mod.addImport("testtmp", testtmp_mod);
262 client_mod.addImport("quic_client", quic_client_mod); 276 client_mod.addImport("quic_client", quic_client_mod);
263 // The client is the only thing that predicts: the overlay is a local 277 // The client is the only thing that predicts: the overlay is a local
@@ -390,7 +404,7 @@ pub fn build(b: *std.Build) void {
390 // absence here was a live hazard recorded in decisions.md — muxd's 404 // absence here was a live hazard recorded in decisions.md — muxd's
391 // entrypoint could grow tests that silently never ran, exactly as 405 // entrypoint could grow tests that silently never ran, exactly as
392 // mux_main.zig's five did before it was added. 406 // mux_main.zig's five did before it was added.
393 for ([_]*std.Build.Module{ protocol_mod, engine_mod, pty_mod, delta_mod, sockpath_mod, server_mod, client_mod, proxy_mod, mux_mod, quic_mod, quic_server_mod, exe_mod, testtmp_mod, quic_client_mod, predict_mod, rawmode_mod, delaypipe_mod, xdg_mod, spawn_mod, handoff_mod, paint_mod, render_mod, ptyclient_mod }) |mod| { 407 for ([_]*std.Build.Module{ protocol_mod, engine_mod, pty_mod, delta_mod, replica_mod, sockpath_mod, server_mod, client_mod, proxy_mod, mux_mod, quic_mod, quic_server_mod, exe_mod, testtmp_mod, quic_client_mod, predict_mod, rawmode_mod, delaypipe_mod, xdg_mod, spawn_mod, handoff_mod, paint_mod, render_mod, ptyclient_mod }) |mod| {
394 const t = b.addTest(.{ .root_module = mod }); 408 const t = b.addTest(.{ .root_module = mod });
395 t.use_llvm = true; 409 t.use_llvm = true;
396 t.use_lld = true; 410 t.use_lld = true;
src/client.zig
Old New
@@ -7,6 +7,7 @@
7 //! Detach chord: Ctrl-\ (0x1c). No keybinding layer in this prototype. 7 //! Detach chord: Ctrl-\ (0x1c). No keybinding layer in this prototype.
8 const std = @import("std"); 8 const std = @import("std");
9 const Engine = @import("engine").Engine; 9 const Engine = @import("engine").Engine;
10 const Replica = @import("replica").Replica;
10 const proto = @import("protocol"); 11 const proto = @import("protocol");
11 const TmpDir = @import("testtmp").TmpDir; 12 const TmpDir = @import("testtmp").TmpDir;
12 const quic_client = @import("quic_client"); 13 const quic_client = @import("quic_client");
@@ -899,8 +900,11 @@ fn session(
899 900
900 var size = ttySize(stdout_fd) orelse proto.Size{ .cols = 80, .rows = 24 }; 901 var size = ttySize(stdout_fd) orelse proto.Size{ .cols = 80, .rows = 24 };
901 902
902 var replica = try Engine.init(alloc, .{ .cols = size.cols, .rows = size.rows }); 903 var eng = try Engine.init(alloc, .{ .cols = size.cols, .rows = size.rows });
903 defer replica.deinit(); 904 defer eng.deinit();
905 // The replay core, extracted to replica.zig; the engine stays owned
906 // here (the Replica borrows it), so the deinit above is the one owner.
907 var rep = Replica.init(alloc, eng);
904 908
905 // Registered before the terminal-restore defer so it runs after it: 909 // Registered before the terminal-restore defer so it runs after it:
906 // messages land on the normal screen, not the wiped alternate one. 910 // messages land on the normal screen, not the wiped alternate one.
@@ -973,28 +977,9 @@ fn session(
973 977
974 var stdin_open = true; 978 var stdin_open = true;
975 // Scroll mode: 0 = live; N = viewing the page N screenfuls above live. 979 // Scroll mode: 0 = live; N = viewing the page N screenfuls above live.
980 // View state, not replay state, so it stays here rather than in the
981 // Replica (which holds grid/seq/epoch/history — see replica.zig).
976 var scroll_pages: u32 = 0; 982 var scroll_pages: u32 = 0;
977 var history_rows: u32 = 0;
978 // The authoritative grid size, learned from snapshot prefixes. Under
979 // latest-wins another client's attach or resize can make it differ
980 // from our tty (`size`); the replica follows the grid, the paint is
981 // clipped to the tty.
982 var grid = size;
983 // Whether state has arrived *on the current attach*. A refusal (session
984 // full) arrives as exit_status before anything else; without this we
985 // could not tell it apart from the shell exiting 1. Per attach rather
986 // than per session because a reconnect produces the identical shape:
987 // what we still hold from before the disconnect says nothing about
988 // whether this attach was let in.
989 var state_since_attach = false;
990 // The daemon instance we are talking to, learned from its snapshots,
991 // and quoted back on reconnect so the daemon can tell whether the seq
992 // we hold is one of its own (a restarted daemon counts from zero over
993 // different content).
994 var session_epoch: u64 = 0;
995 // The newest seq we hold. Sent with a reconnect attach; the daemon
996 // answers with a delta when it can still reach us from there.
997 var last_seq: u64 = 0;
998 // Set by any site that finds the transport dead; serviced at the top of 983 // Set by any site that finds the transport dead; serviced at the top of
999 // the loop so the bookkeeping around a reconnect lives in one place. 984 // the loop so the bookkeeping around a reconnect lives in one place.
1000 var needs_reconnect = false; 985 var needs_reconnect = false;
@@ -1020,8 +1005,8 @@ fn session(
1020 // `state_since_attach` cannot serve here: it is cleared on 1005 // `state_since_attach` cannot serve here: it is cleared on
1021 // every re-attach, so mid-session it would send us down this 1006 // every re-attach, so mid-session it would send us down this
1022 // exit path exactly when resuming is what we want. 1007 // exit path exactly when resuming is what we want.
1023 if (session_epoch == 0) { 1008 if (rep.session_epoch == 0) {
1024 exit_msg = lostMsg(target, session_epoch); 1009 exit_msg = lostMsg(target, rep.session_epoch);
1025 return 1; 1010 return 1;
1026 } 1011 }
1027 // A resync repaints live state, so a history page would be 1012 // A resync repaints live state, so a history page would be
@@ -1046,8 +1031,8 @@ fn session(
1046 transport, 1031 transport,
1047 target, 1032 target,
1048 size, 1033 size,
1049 last_seq, 1034 rep.last_seq,
1050 session_epoch, 1035 rep.session_epoch,
1051 stdin_fd, 1036 stdin_fd,
1052 stdout_fd, 1037 stdout_fd,
1053 is_tty, 1038 is_tty,
@@ -1057,7 +1042,9 @@ fn session(
1057 exit_msg = "mux: detached while reconnecting (session still running; run mux to reattach)"; 1042 exit_msg = "mux: detached while reconnecting (session still running; run mux to reattach)";
1058 return 0; 1043 return 0;
1059 } 1044 }
1060 state_since_attach = false; 1045 // Only the caller knows a re-attach happened; the Replica's
1046 // contract says this clear is ours to do.
1047 rep.state_since_attach = false;
1061 if (reconnect_grace_until == null) { 1048 if (reconnect_grace_until == null) {
1062 reconnect_grace_until = std.time.milliTimestamp() + reconnect_grace_ms; 1049 reconnect_grace_until = std.time.milliTimestamp() + reconnect_grace_ms;
1063 } 1050 }
@@ -1103,7 +1090,7 @@ fn session(
1103 // reconcile will never run again and the glyph would otherwise stay 1090 // reconcile will never run again and the glyph would otherwise stay
1104 // on screen for the rest of the session. 1091 // on screen for the rest of the session.
1105 if (overlay.expire(std.time.milliTimestamp()) == .contradicted and scroll_pages == 0) { 1092 if (overlay.expire(std.time.milliTimestamp()) == .contradicted and scroll_pages == 0) {
1106 try paint_mod.renderClipped(alloc, replica, size, stdout_fd); 1093 try paint_mod.renderClipped(alloc, rep.eng, size, stdout_fd);
1107 } 1094 }
1108 1095
1109 // Labelled, because "no whole frame yet" must leave the REST of this 1096 // Labelled, because "no whole frame yet" must leave the REST of this
@@ -1151,41 +1138,34 @@ fn session(
1151 } 1138 }
1152 switch (frame.type) { 1139 switch (frame.type) {
1153 .snapshot => { 1140 .snapshot => {
1154 const prefix = proto.readSnapshotPrefix(frame.payload) catch continue; 1141 // The Replica adopts seq/epoch/history, resizes to the
1155 // Set after the prefix parse here but before the compose 1142 // prefix's grid, and rebuilds — even while scrolled, so
1156 // catch in the delta arm: a short snapshot proves nothing, 1143 // leaving scroll mode paints current state. A short
1157 // while a delta's arrival alone proves we were admitted. 1144 // snapshot proves nothing (BadPayload leaves everything
1158 state_since_attach = true; 1145 // untouched, state_since_attach included); anything
1146 // else out of the resize stays loud.
1147 _ = rep.apply(.snapshot, frame.payload) catch |err| switch (err) {
1148 error.BadPayload => continue,
1149 else => |e| return e,
1150 };
1159 reconnect_grace_until = null; 1151 reconnect_grace_until = null;
1160 session_epoch = prefix.epoch;
1161 last_seq = prefix.seq;
1162 history_rows = prefix.history_rows;
1163 if (prefix.cols != grid.cols or prefix.rows != grid.rows) {
1164 try replica.resize(prefix.cols, prefix.rows);
1165 grid = .{ .cols = prefix.cols, .rows = prefix.rows };
1166 }
1167 // Authoritative resync: rebuild the replica even while
1168 // scrolled, so leaving scroll mode paints current state.
1169 replica.reset();
1170 replica.feed(frame.payload[proto.snapshot_prefix_len..]);
1171 // A snapshot answers a resize, ends a reconnect, and 1152 // A snapshot answers a resize, ends a reconnect, and
1172 // rebuilds the screen under anything outstanding. None 1153 // rebuilds the screen under anything outstanding. None
1173 // of that says a prediction was wrong — it says we can 1154 // of that says a prediction was wrong — it says we can
1174 // no longer find out, so the queue goes and the counters 1155 // no longer find out, so the queue goes and the counters
1175 // do not move. 1156 // do not move.
1176 overlay.setGrid(grid.cols, grid.rows); 1157 overlay.setGrid(rep.grid.cols, rep.grid.rows);
1177 overlay.setResizePending(false); 1158 overlay.setResizePending(false);
1178 overlay.flush(); 1159 overlay.flush();
1179 overlay.noteSeq(prefix.seq); 1160 overlay.noteSeq(rep.last_seq);
1180 if (scroll_pages == 0) { 1161 if (scroll_pages == 0) {
1181 try paint_mod.renderClipped(alloc, replica, size, stdout_fd); 1162 try paint_mod.renderClipped(alloc, rep.eng, size, stdout_fd);
1182 repaint_after_resync = false; // banner painted over 1163 repaint_after_resync = false; // banner painted over
1183 } 1164 }
1184 }, 1165 },
1185 .delta => { 1166 .delta => {
1186 state_since_attach = true;
1187 reconnect_grace_until = null; 1167 reconnect_grace_until = null;
1188 const composed = proto.composeDelta(alloc, frame.payload) catch { 1168 if (try rep.apply(.delta, frame.payload) == .resync) {
1189 // A rejected delta means the replica can no longer be 1169 // A rejected delta means the replica can no longer be
1190 // trusted; ask for a fresh snapshot rather than 1170 // trusted; ask for a fresh snapshot rather than
1191 // silently skipping it and desyncing for good. By 1171 // silently skipping it and desyncing for good. By
@@ -1203,18 +1183,14 @@ fn session(
1203 &proto.encodeAttach(size.cols, size.rows, 0, 0), 1183 &proto.encodeAttach(size.cols, size.rows, 0, 0),
1204 ) catch {}; 1184 ) catch {};
1205 continue; 1185 continue;
1206 }; 1186 }
1207 defer alloc.free(composed.bytes);
1208 history_rows = composed.header.history_rows;
1209 last_seq = composed.header.seq;
1210 replica.feed(composed.bytes);
1211 // Judged against the replica the frame has just been fed 1187 // Judged against the replica the frame has just been fed
1212 // into, which is the only authority there is. 1188 // into, which is the only authority there is.
1213 const verdict = reconcileOverlay( 1189 const verdict = reconcileOverlay(
1214 alloc, 1190 alloc,
1215 &overlay, 1191 &overlay,
1216 replica, 1192 rep.eng,
1217 last_seq, 1193 rep.last_seq,
1218 std.time.milliTimestamp(), 1194 std.time.milliTimestamp(),
1219 ); 1195 );
1220 // While scrolled the replica still tracks live output; the 1196 // While scrolled the replica still tracks live output; the
@@ -1233,7 +1209,7 @@ fn session(
1233 // It is affordable precisely because reconcile v2 1209 // It is affordable precisely because reconcile v2
1234 // made contradictions rare — a burst outrunning 1210 // made contradictions rare — a burst outrunning
1235 // the round trip is no longer one. 1211 // the round trip is no longer one.
1236 try paint_mod.renderClipped(alloc, replica, size, stdout_fd); 1212 try paint_mod.renderClipped(alloc, rep.eng, size, stdout_fd);
1237 repaint_after_resync = false; 1213 repaint_after_resync = false;
1238 } else { 1214 } else {
1239 try paint_mod.paintDeltaClipped(alloc, frame.payload, size, stdout_fd); 1215 try paint_mod.paintDeltaClipped(alloc, frame.payload, size, stdout_fd);
@@ -1241,7 +1217,7 @@ fn session(
1241 // Last, and after either paint: the rows the daemon 1217 // Last, and after either paint: the rows the daemon
1242 // just sent have overwritten anything drawn on them, 1218 // just sent have overwritten anything drawn on them,
1243 // including predictions that are still outstanding. 1219 // including predictions that are still outstanding.
1244 paintOverlay(alloc, &overlay, replica.cursorPos(), size, stdout_fd); 1220 paintOverlay(alloc, &overlay, rep.eng.cursorPos(), size, stdout_fd);
1245 } 1221 }
1246 }, 1222 },
1247 .pty_mode => { 1223 .pty_mode => {
@@ -1253,7 +1229,7 @@ fn session(
1253 const had_pending = overlay.pendingCount() > 0; 1229 const had_pending = overlay.pendingCount() > 0;
1254 overlay.setMode(flags); 1230 overlay.setMode(flags);
1255 if (had_pending and overlay.pendingCount() == 0 and scroll_pages == 0) { 1231 if (had_pending and overlay.pendingCount() == 0 and scroll_pages == 0) {
1256 try paint_mod.renderClipped(alloc, replica, size, stdout_fd); 1232 try paint_mod.renderClipped(alloc, rep.eng, size, stdout_fd);
1257 } 1233 }
1258 }, 1234 },
1259 .scrollback_chunk => { 1235 .scrollback_chunk => {
@@ -1264,7 +1240,7 @@ fn session(
1264 // Before any session state, exit_status is almost always 1240 // Before any session state, exit_status is almost always
1265 // the daemon refusing the attach — say so, or it looks 1241 // the daemon refusing the attach — say so, or it looks
1266 // exactly like the shell itself exiting non-zero. 1242 // exactly like the shell itself exiting non-zero.
1267 if (!state_since_attach) { 1243 if (!rep.state_since_attach) {
1268 // Right after a reconnect the usual cause is the 1244 // Right after a reconnect the usual cause is the
1269 // daemon not having reaped our dead predecessor's 1245 // daemon not having reaped our dead predecessor's
1270 // slot yet, so the session is full of *us*. Worth a 1246 // slot yet, so the session is full of *us*. Worth a
@@ -1305,14 +1281,14 @@ fn session(
1305 const scroll_up = "\x1b[5;2~"; // Shift+PageUp 1281 const scroll_up = "\x1b[5;2~"; // Shift+PageUp
1306 const scroll_dn = "\x1b[6;2~"; // Shift+PageDown 1282 const scroll_dn = "\x1b[6;2~"; // Shift+PageDown
1307 if (std.mem.eql(u8, buf[0..n], scroll_up)) { 1283 if (std.mem.eql(u8, buf[0..n], scroll_up)) {
1308 if (history_rows > 0) { 1284 if (rep.history_rows > 0) {
1309 const max_pages: u32 = (history_rows + size.rows - 1) / size.rows; 1285 const max_pages: u32 = (rep.history_rows + size.rows - 1) / size.rows;
1310 if (scroll_pages < max_pages) scroll_pages += 1; 1286 if (scroll_pages < max_pages) scroll_pages += 1;
1311 // The cursor is no longer where the user is looking, 1287 // The cursor is no longer where the user is looking,
1312 // so a prediction painted at it would land in the 1288 // so a prediction painted at it would land in the
1313 // middle of history. 1289 // middle of history.
1314 overlay.setScrollMode(true); 1290 overlay.setScrollMode(true);
1315 requestScrollPage(transport, scroll_pages, history_rows, size) catch { 1291 requestScrollPage(transport, &rep, scroll_pages, size) catch {
1316 needs_reconnect = true; 1292 needs_reconnect = true;
1317 continue; 1293 continue;
1318 }; 1294 };
@@ -1321,9 +1297,9 @@ fn session(
1321 if (scroll_pages > 0) scroll_pages -= 1; 1297 if (scroll_pages > 0) scroll_pages -= 1;
1322 if (scroll_pages == 0) { 1298 if (scroll_pages == 0) {
1323 overlay.setScrollMode(false); 1299 overlay.setScrollMode(false);
1324 try paint_mod.renderClipped(alloc, replica, size, stdout_fd); 1300 try paint_mod.renderClipped(alloc, rep.eng, size, stdout_fd);
1325 } else { 1301 } else {
1326 requestScrollPage(transport, scroll_pages, history_rows, size) catch { 1302 requestScrollPage(transport, &rep, scroll_pages, size) catch {
1327 needs_reconnect = true; 1303 needs_reconnect = true;
1328 continue; 1304 continue;
1329 }; 1305 };
@@ -1332,12 +1308,12 @@ fn session(
1332 // Any other key exits scroll mode (swallowed, not forwarded). 1308 // Any other key exits scroll mode (swallowed, not forwarded).
1333 scroll_pages = 0; 1309 scroll_pages = 0;
1334 overlay.setScrollMode(false); 1310 overlay.setScrollMode(false);
1335 try paint_mod.renderClipped(alloc, replica, size, stdout_fd); 1311 try paint_mod.renderClipped(alloc, rep.eng, size, stdout_fd);
1336 } else { 1312 } else {
1337 // Speculate before sending, so the glyph is on screen 1313 // Speculate before sending, so the glyph is on screen
1338 // while the keystroke is still in flight. The bytes that 1314 // while the keystroke is still in flight. The bytes that
1339 // go out are unchanged either way. 1315 // go out are unchanged either way.
1340 offerKeystroke(alloc, &overlay, replica, buf[0..n], size, stdout_fd); 1316 offerKeystroke(alloc, &overlay, rep.eng, buf[0..n], size, stdout_fd);
1341 transport.writeFrame(.input, buf[0..n]) catch { 1317 transport.writeFrame(.input, buf[0..n]) catch {
1342 // These keystrokes are lost with the transport, by 1318 // These keystrokes are lost with the transport, by
1343 // the same policy that drops what is typed while 1319 // the same policy that drops what is typed while
@@ -1517,14 +1493,13 @@ fn dumpPredictStats(c: predict.Counters) void {
1517 1493
1518 fn requestScrollPage( 1494 fn requestScrollPage(
1519 transport: *Transport, 1495 transport: *Transport,
1496 rep: *const Replica,
1520 pages_up: u32, 1497 pages_up: u32,
1521 history_rows: u32,
1522 size: proto.Size, 1498 size: proto.Size,
1523 ) !void { 1499 ) !void {
1524 // Page N shows `size.rows` rows ending N*rows above the live viewport 1500 // Page N shows `size.rows` rows ending N*rows above the live viewport
1525 // top (screen-space row index history_rows). 1501 // top; the row math lives with the replica's history_rows (replica.zig).
1526 const rows: u32 = size.rows; 1502 const start = rep.scrollStart(pages_up, size.rows);
1527 const start = history_rows -| (pages_up * rows);
1528 try transport.writeFrame(.fetch_scrollback, &proto.encodeScrollbackReq(start, size.rows)); 1503 try transport.writeFrame(.fetch_scrollback, &proto.encodeScrollbackReq(start, size.rows));
1529 } 1504 }
1530 1505
src/replica.zig
Old New
@@ -0,0 +1,299 @@
1 //! The replay core: what a mux client does to mirror a daemon's session.
2 //! Applies snapshot and delta frames to a local engine, tracks the resume
3 //! coordinates (`last_seq`/`session_epoch`), and follows the authoritative
4 //! grid. Extracted from client.zig's session() (M-web Task 1) so the CLI
5 //! client, the wasm core, and the server's test fixtures replay through one
6 //! implementation instead of three.
7 //!
8 //! Deliberately platform-free: no posix, no fds, no clocks — this module
9 //! must compile for wasm32-freestanding. Anything that needs a terminal,
10 //! a timestamp, or a transport belongs to the caller.
11 //!
12 //! The Replica BORROWS its engine. The caller owns the Engine's lifetime
13 //! (the CLI's session() and the server's tests already do); a Replica is
14 //! bookkeeping around it, not a second owner.
15
16 const std = @import("std");
17 const Engine = @import("engine").Engine;
18 const proto = @import("protocol");
19
20 pub const Replica = struct {
21 alloc: std.mem.Allocator,
22 /// Borrowed. The engine the frames are replayed into.
23 eng: *Engine,
24 /// The authoritative grid size, learned from snapshot prefixes. Under
25 /// latest-wins another client's attach or resize can make it differ
26 /// from any local tty; the replica follows the grid, not the tty.
27 grid: proto.Size,
28 /// The daemon instance we are talking to, learned from its snapshots,
29 /// and quoted back on reconnect so the daemon can tell whether the seq
30 /// we hold is one of its own (a restarted daemon counts from zero over
31 /// different content). 0 until the first snapshot: "I hold nothing".
32 session_epoch: u64 = 0,
33 /// The newest seq we hold. Quoted on reconnect; the daemon answers
34 /// with a delta when it can still reach us from there.
35 last_seq: u64 = 0,
36 history_rows: u32 = 0,
37 /// Whether state has arrived *on the current attach*. A refusal
38 /// (session full) arrives as exit_status before anything else; without
39 /// this the caller could not tell it apart from the shell exiting 1.
40 /// Set here by any replay frame's arrival — a delta's arrival alone
41 /// proves we were admitted, even one that fails to decode — and
42 /// cleared by the CALLER on every re-attach, because only the caller
43 /// knows when one happened.
44 state_since_attach: bool = false,
45
46 pub const Applied = enum {
47 /// The frame landed; the engine reflects it.
48 painted,
49 /// A delta that could not be trusted (rejected by composeDelta).
50 /// The engine was not touched. The caller must re-attach with
51 /// have_seq=0 — quoting a seq would invite the delta that cannot
52 /// fix us — and this is deliberately NOT the reconnect path: the
53 /// transport is alive, it is the replica that is suspect.
54 resync,
55 };
56
57 pub fn init(alloc: std.mem.Allocator, eng: *Engine) Replica {
58 return .{
59 .alloc = alloc,
60 .eng = eng,
61 .grid = .{ .cols = eng.term.cols, .rows = eng.term.rows },
62 };
63 }
64
65 /// Consume one replay frame. Only .snapshot and .delta are replay
66 /// frames; callers dispatch on type before calling.
67 ///
68 /// .snapshot: a short/garbled prefix is error.BadPayload with nothing
69 /// consumed and `state_since_attach` untouched — a short snapshot
70 /// proves nothing. A good prefix adopts seq/epoch/history, resizes the
71 /// engine when the grid moved, then reset+feed: an authoritative
72 /// rebuild. Resize failure (OOM) propagates.
73 ///
74 /// .delta: arrival alone sets `state_since_attach`. A payload
75 /// composeDelta rejects returns .resync (see Applied); a good one
76 /// advances history_rows/last_seq and feeds the composed paint.
77 pub fn apply(self: *Replica, t: proto.MsgType, payload: []const u8) !Applied {
78 switch (t) {
79 .snapshot => {
80 const prefix = try proto.readSnapshotPrefix(payload);
81 self.state_since_attach = true;
82 self.session_epoch = prefix.epoch;
83 self.last_seq = prefix.seq;
84 self.history_rows = prefix.history_rows;
85 if (prefix.cols != self.grid.cols or prefix.rows != self.grid.rows) {
86 try self.eng.resize(prefix.cols, prefix.rows);
87 self.grid = .{ .cols = prefix.cols, .rows = prefix.rows };
88 }
89 self.eng.reset();
90 self.eng.feed(payload[proto.snapshot_prefix_len..]);
91 return .painted;
92 },
93 .delta => {
94 self.state_since_attach = true;
95 const composed = proto.composeDelta(self.alloc, payload) catch
96 return .resync;
97 defer self.alloc.free(composed.bytes);
98 self.history_rows = composed.header.history_rows;
99 self.last_seq = composed.header.seq;
100 self.eng.feed(composed.bytes);
101 return .painted;
102 },
103 else => unreachable, // not a replay frame; callers dispatch
104 }
105 }
106
107 pub const AttachArgs = struct { have_seq: u64, have_epoch: u64 };
108
109 /// What an attach frame quotes: (0,0) until the first snapshot — a
110 /// fresh replica holds nothing — and (last_seq, session_epoch) after.
111 /// The delta-resync re-attach quotes (0,0) EXPLICITLY at its call site
112 /// instead of using this: there the point is that what we hold is
113 /// untrusted.
114 pub fn attachArgs(self: *const Replica) AttachArgs {
115 return .{ .have_seq = self.last_seq, .have_epoch = self.session_epoch };
116 }
117
118 /// Screen-space start row for scroll page N: page N shows `view_rows`
119 /// rows ending N*view_rows above the live viewport top (row index
120 /// history_rows). Saturates at the oldest retained row.
121 pub fn scrollStart(self: *const Replica, pages_up: u32, view_rows: u16) u32 {
122 return self.history_rows -| (pages_up * @as(u32, view_rows));
123 }
124 };
125
126 // ---------------------------------------------------------------------------
127 // Tests. The wire layouts these build are golden-pinned in protocol.zig, so
128 // constructing real payloads here is mechanical, not speculative.
129
130 fn testSnapshot(
131 alloc: std.mem.Allocator,
132 p: proto.SnapshotPrefix,
133 state: []const u8,
134 ) ![]u8 {
135 var payload = try alloc.alloc(u8, proto.snapshot_prefix_len + state.len);
136 proto.writeSnapshotPrefix(payload[0..proto.snapshot_prefix_len], p);
137 @memcpy(payload[proto.snapshot_prefix_len..], state);
138 return payload;
139 }
140
141 test "snapshot replay: prefix consumed, state fed, epoch and seq adopted" {
142 const alloc = std.testing.allocator;
143 var eng = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });
144 defer eng.deinit();
145 var r = Replica.init(alloc, eng);
146
147 const payload = try testSnapshot(alloc, .{
148 .seq = 7,
149 .history_rows = 3,
150 .cols = 80,
151 .rows = 24,
152 .epoch = 0xABCD,
153 }, "hi");
154 defer alloc.free(payload);
155
156 try std.testing.expectEqual(Replica.Applied.painted, try r.apply(.snapshot, payload));
157 try std.testing.expectEqual(@as(u64, 7), r.last_seq);
158 try std.testing.expectEqual(@as(u64, 0xABCD), r.session_epoch);
159 try std.testing.expectEqual(@as(u32, 3), r.history_rows);
160 try std.testing.expect(r.state_since_attach);
161
162 const dump = try eng.dumpPlain(alloc);
163 defer alloc.free(dump);
164 try std.testing.expect(std.mem.startsWith(u8, dump, "hi"));
165 }
166
167 test "snapshot at a new grid size resizes the replica engine first" {
168 const alloc = std.testing.allocator;
169 var eng = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });
170 defer eng.deinit();
171 var r = Replica.init(alloc, eng);
172
173 const payload = try testSnapshot(alloc, .{
174 .seq = 1,
175 .history_rows = 0,
176 .cols = 100,
177 .rows = 30,
178 .epoch = 1,
179 }, "wide");
180 defer alloc.free(payload);
181
182 try std.testing.expectEqual(Replica.Applied.painted, try r.apply(.snapshot, payload));
183 try std.testing.expectEqual(@as(u16, 100), r.grid.cols);
184 try std.testing.expectEqual(@as(u16, 30), r.grid.rows);
185 try std.testing.expectEqual(@as(u16, 100), eng.term.cols);
186 try std.testing.expectEqual(@as(u16, 30), eng.term.rows);
187 }
188
189 test "short snapshot proves nothing: BadPayload, state_since_attach untouched" {
190 const alloc = std.testing.allocator;
191 var eng = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });
192 defer eng.deinit();
193 var r = Replica.init(alloc, eng);
194
195 const short = [_]u8{0} ** (proto.snapshot_prefix_len - 1);
196 try std.testing.expectError(error.BadPayload, r.apply(.snapshot, &short));
197 try std.testing.expect(!r.state_since_attach);
198 try std.testing.expectEqual(@as(u64, 0), r.last_seq);
199 }
200
201 test "delta replay: composed rows land, last_seq advances, history follows" {
202 const alloc = std.testing.allocator;
203 var eng = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });
204 defer eng.deinit();
205 var r = Replica.init(alloc, eng);
206
207 const snap = try testSnapshot(alloc, .{
208 .seq = 7,
209 .history_rows = 0,
210 .cols = 80,
211 .rows = 24,
212 .epoch = 0xABCD,
213 }, "hi");
214 defer alloc.free(snap);
215 _ = try r.apply(.snapshot, snap);
216
217 var delta: std.ArrayList(u8) = .empty;
218 defer delta.deinit(alloc);
219 try proto.appendDeltaHeader(&delta, alloc, .{
220 .seq = 8,
221 .history_rows = 2,
222 .cursor_x = 2,
223 .cursor_y = 0,
224 .row_count = 1,
225 });
226 try proto.appendDeltaRow(&delta, alloc, 0, "yo");
227
228 try std.testing.expectEqual(Replica.Applied.painted, try r.apply(.delta, delta.items));
229 try std.testing.expectEqual(@as(u64, 8), r.last_seq);
230 try std.testing.expectEqual(@as(u32, 2), r.history_rows);
231
232 const dump = try eng.dumpPlain(alloc);
233 defer alloc.free(dump);
234 try std.testing.expect(std.mem.startsWith(u8, dump, "yo"));
235 }
236
237 test "delta decode failure reports .resync — and its arrival still proves admission" {
238 const alloc = std.testing.allocator;
239 var eng = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });
240 defer eng.deinit();
241 var r = Replica.init(alloc, eng);
242
243 // Header claims two rows, payload carries one: composeDelta rejects it
244 // (the row_count check protocol.zig pins).
245 var delta: std.ArrayList(u8) = .empty;
246 defer delta.deinit(alloc);
247 try proto.appendDeltaHeader(&delta, alloc, .{
248 .seq = 9,
249 .history_rows = 0,
250 .cursor_x = 0,
251 .cursor_y = 0,
252 .row_count = 2,
253 });
254 try proto.appendDeltaRow(&delta, alloc, 0, "x");
255
256 try std.testing.expectEqual(Replica.Applied.resync, try r.apply(.delta, delta.items));
257 // The subtlety the CLI relies on: a delta's ARRIVAL alone proves the
258 // attach was admitted, decodable or not...
259 try std.testing.expect(r.state_since_attach);
260 // ...while the resume coordinates stay exactly where they were.
261 try std.testing.expectEqual(@as(u64, 0), r.last_seq);
262 }
263
264 test "attach args: first attach quotes (0,0); after a snapshot, (last_seq, epoch)" {
265 const alloc = std.testing.allocator;
266 var eng = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });
267 defer eng.deinit();
268 var r = Replica.init(alloc, eng);
269
270 const fresh = r.attachArgs();
271 try std.testing.expectEqual(@as(u64, 0), fresh.have_seq);
272 try std.testing.expectEqual(@as(u64, 0), fresh.have_epoch);
273
274 const snap = try testSnapshot(alloc, .{
275 .seq = 42,
276 .history_rows = 0,
277 .cols = 80,
278 .rows = 24,
279 .epoch = 0xFEED,
280 }, "");
281 defer alloc.free(snap);
282 _ = try r.apply(.snapshot, snap);
283
284 const held = r.attachArgs();
285 try std.testing.expectEqual(@as(u64, 42), held.have_seq);
286 try std.testing.expectEqual(@as(u64, 0xFEED), held.have_epoch);
287 }
288
289 test "scrollStart: pages count up from the live viewport top, saturating at row 0" {
290 const alloc = std.testing.allocator;
291 var eng = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });
292 defer eng.deinit();
293 var r = Replica.init(alloc, eng);
294 r.history_rows = 100;
295
296 try std.testing.expectEqual(@as(u32, 76), r.scrollStart(1, 24));
297 try std.testing.expectEqual(@as(u32, 52), r.scrollStart(2, 24));
298 try std.testing.expectEqual(@as(u32, 0), r.scrollStart(5, 24)); // 100 -| 120
299 }
src/server.zig
Old New
@@ -10,6 +10,9 @@ const Engine = @import("engine").Engine;
10 const Pty = @import("pty").Pty; 10 const Pty = @import("pty").Pty;
11 const proto = @import("protocol"); 11 const proto = @import("protocol");
12 const DeltaTracker = @import("delta").DeltaTracker; 12 const DeltaTracker = @import("delta").DeltaTracker;
13 // Test-only consumer (applyFrame): the tests replay daemon frames through
14 // the production client's replay core rather than a hand-rolled twin.
15 const replica_mod = @import("replica");
13 const sockpath = @import("sockpath"); 16 const sockpath = @import("sockpath");
14 const quic = @import("quic"); 17 const quic = @import("quic");
15 const quic_server = @import("quic_server"); 18 const quic_server = @import("quic_server");
@@ -1495,24 +1498,16 @@ fn serverThread(srv: *Server, stop: *std.atomic.Value(bool)) void {
1495 } 1498 }
1496 1499
1497 /// Test helper: bring a replica engine up to date with one daemon frame. 1500 /// Test helper: bring a replica engine up to date with one daemon frame.
1498 /// Snapshots are a full reset + state feed (skipping the fixed prefix); 1501 /// The replay itself lives in replica.zig — the production client's — so
1499 /// deltas are composed into row paints. 1502 /// these tests replay through the same code the client ships, not a
1503 /// hand-rolled twin. The Replica here is per-call scaffolding around the
1504 /// borrowed engine; resume state is not what these tests assert.
1500 fn applyFrame(alloc: std.mem.Allocator, replica: *Engine, frame: proto.Frame) !void { 1505 fn applyFrame(alloc: std.mem.Allocator, replica: *Engine, frame: proto.Frame) !void {
1501 switch (frame.type) { 1506 if (frame.type != .snapshot and frame.type != .delta) return;
1502 .snapshot => { 1507 var r = replica_mod.Replica.init(alloc, replica);
1503 // Not a discarded read: this is the bounds check that makes the 1508 // The old helper propagated a garbled delta as BadPayload; .resync is
1504 // fixed-width slice below safe on a short/garbled payload. 1509 // that same event with the production name on it.
1505 _ = try proto.readSnapshotPrefix(frame.payload); 1510 if (try r.apply(frame.type, frame.payload) == .resync) return error.BadPayload;
1506 replica.reset();
1507 replica.feed(frame.payload[proto.snapshot_prefix_len..]);
1508 },
1509 .delta => {
1510 const composed = try proto.composeDelta(alloc, frame.payload);
1511 defer alloc.free(composed.bytes);
1512 replica.feed(composed.bytes);
1513 },
1514 else => {},
1515 }
1516 } 1511 }
1517 1512
1518 test "Server: survives a client that dies without detaching; next attach works" { 1513 test "Server: survives a client that dies without detaching; next attach works" {