a73x

65f76187

feat: session epoch fences have_seq across daemon restarts

a73x   2026-08-08 14:08

Commit message
feat: session epoch fences have_seq across daemon restarts

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
docs/decisions.md
Old New
@@ -144,7 +144,8 @@
144 with overlapping counters would accept a delta belonging to a different 144 with overlapping counters would accept a delta belonging to a different
145 session. Unreachable today — the shipped client always attaches with 145 session. Unreachable today — the shipped client always attaches with
146 have_seq=0 — but a generation/epoch field is due before any real 146 have_seq=0 — but a generation/epoch field is due before any real
147 network transport. 147 network transport. **Superseded in M6:** attach and the snapshot prefix
148 now carry an epoch; the M6 section records the design.
148 - **Stats counterfactual costs a dumpState per delta send** while a client 149 - **Stats counterfactual costs a dumpState per delta send** while a client
149 is attached — deliberate prototype instrumentation; gate behind an 150 is attached — deliberate prototype instrumentation; gate behind an
150 option if it ever matters for performance comparisons. 151 option if it ever matters for performance comparisons.
src/client.zig
Old New
@@ -74,8 +74,9 @@ pub fn attach(alloc: std.mem.Allocator, sock_path: []const u8) !u8 {
74 std.posix.tcsetattr(stdin_fd, .FLUSH, t) catch {}; 74 std.posix.tcsetattr(stdin_fd, .FLUSH, t) catch {};
75 }; 75 };
76 76
77 // A fresh client process holds no state, so it asks for a full snapshot. 77 // A fresh client process holds no state, so it asks for a full snapshot:
78 try proto.writeFrame(sock, .attach, &proto.encodeAttach(size.cols, size.rows, 0)); 78 // no seq, and no epoch to interpret one in.
79 try proto.writeFrame(sock, .attach, &proto.encodeAttach(size.cols, size.rows, 0, 0));
79 80
80 var stdin_open = true; 81 var stdin_open = true;
81 // Scroll mode: 0 = live; N = viewing the page N screenfuls above live. 82 // Scroll mode: 0 = live; N = viewing the page N screenfuls above live.
@@ -90,6 +91,11 @@ pub fn attach(alloc: std.mem.Allocator, sock_path: []const u8) !u8 {
90 // arrives as exit_status before anything else; without this we could 91 // arrives as exit_status before anything else; without this we could
91 // not tell it apart from the shell exiting 1. 92 // not tell it apart from the shell exiting 1.
92 var got_state = false; 93 var got_state = false;
94 // The daemon instance we are talking to, learned from its snapshots.
95 // Nothing reads it yet: this client always attaches fresh. Reconnect
96 // logic (M6+) will send it back with a have_seq so the daemon can tell
97 // whether that seq is one of its own.
98 var session_epoch: u64 = 0;
93 var buf: [16 * 1024]u8 = undefined; 99 var buf: [16 * 1024]u8 = undefined;
94 while (true) { 100 while (true) {
95 if (winch_flag.swap(false, .acq_rel)) { 101 if (winch_flag.swap(false, .acq_rel)) {
@@ -123,6 +129,7 @@ pub fn attach(alloc: std.mem.Allocator, sock_path: []const u8) !u8 {
123 // catch in the delta arm: a short snapshot proves nothing, 129 // catch in the delta arm: a short snapshot proves nothing,
124 // while a delta's arrival alone proves we were admitted. 130 // while a delta's arrival alone proves we were admitted.
125 got_state = true; 131 got_state = true;
132 session_epoch = prefix.epoch;
126 history_rows = prefix.history_rows; 133 history_rows = prefix.history_rows;
127 if (prefix.cols != grid.cols or prefix.rows != grid.rows) { 134 if (prefix.cols != grid.cols or prefix.rows != grid.rows) {
128 try replica.resize(prefix.cols, prefix.rows); 135 try replica.resize(prefix.cols, prefix.rows);
@@ -142,7 +149,7 @@ pub fn attach(alloc: std.mem.Allocator, sock_path: []const u8) !u8 {
142 // silently skipping it and desyncing for good. By 149 // silently skipping it and desyncing for good. By
143 // latest-wins this re-attach also re-asserts our size 150 // latest-wins this re-attach also re-asserts our size
144 // onto the shared session — accepted. 151 // onto the shared session — accepted.
145 proto.writeFrame(sock, .attach, &proto.encodeAttach(size.cols, size.rows, 0)) catch {}; 152 proto.writeFrame(sock, .attach, &proto.encodeAttach(size.cols, size.rows, 0, 0)) catch {};
146 continue; 153 continue;
147 }; 154 };
148 defer alloc.free(composed.bytes); 155 defer alloc.free(composed.bytes);
src/protocol.zig
Old New
@@ -10,7 +10,7 @@ const std = @import("std");
10 10
11 pub const MsgType = enum(u8) { 11 pub const MsgType = enum(u8) {
12 // client -> daemon 12 // client -> daemon
13 attach = 0x01, // payload: u16 LE cols, u16 LE rows, u64 LE have_seq 13 attach = 0x01, // payload: u16 LE cols, u16 LE rows, u64 LE have_seq, u64 LE have_epoch
14 input = 0x02, // payload: raw bytes for the PTY 14 input = 0x02, // payload: raw bytes for the PTY
15 resize = 0x03, // payload: u16 LE cols, u16 LE rows 15 resize = 0x03, // payload: u16 LE cols, u16 LE rows
16 detach = 0x04, // payload: empty 16 detach = 0x04, // payload: empty
@@ -138,42 +138,58 @@ pub fn decodeScrollbackReq(payload: []const u8) !ScrollbackReq {
138 }; 138 };
139 } 139 }
140 140
141 pub const AttachReq = struct { cols: u16, rows: u16, have_seq: u64 }; 141 /// What a (re)attaching client already holds. `have_seq` is a sequence
142 /// number in ONE daemon instance's stream: after a restart the new daemon
143 /// counts from zero again, so the same number denotes different state.
144 /// `have_epoch` names the instance those seqs came from — a mismatch (or 0,
145 /// "I hold nothing") makes have_seq unusable and forces a full snapshot.
146 pub const AttachReq = struct { cols: u16, rows: u16, have_seq: u64, have_epoch: u64 };
142 147
143 pub fn encodeAttach(cols: u16, rows: u16, have_seq: u64) [12]u8 { 148 pub const attach_len = 20;
144 var buf: [12]u8 = undefined; 149
150 pub fn encodeAttach(cols: u16, rows: u16, have_seq: u64, have_epoch: u64) [attach_len]u8 {
151 var buf: [attach_len]u8 = undefined;
145 std.mem.writeInt(u16, buf[0..2], cols, .little); 152 std.mem.writeInt(u16, buf[0..2], cols, .little);
146 std.mem.writeInt(u16, buf[2..4], rows, .little); 153 std.mem.writeInt(u16, buf[2..4], rows, .little);
147 std.mem.writeInt(u64, buf[4..12], have_seq, .little); 154 std.mem.writeInt(u64, buf[4..12], have_seq, .little);
155 std.mem.writeInt(u64, buf[12..20], have_epoch, .little);
148 return buf; 156 return buf;
149 } 157 }
150 158
151 pub fn decodeAttach(payload: []const u8) !AttachReq { 159 pub fn decodeAttach(payload: []const u8) !AttachReq {
152 if (payload.len != 12) return error.BadPayload; 160 if (payload.len != attach_len) return error.BadPayload;
153 return .{ 161 return .{
154 .cols = std.mem.readInt(u16, payload[0..2], .little), 162 .cols = std.mem.readInt(u16, payload[0..2], .little),
155 .rows = std.mem.readInt(u16, payload[2..4], .little), 163 .rows = std.mem.readInt(u16, payload[2..4], .little),
156 .have_seq = std.mem.readInt(u64, payload[4..12], .little), 164 .have_seq = std.mem.readInt(u64, payload[4..12], .little),
165 .have_epoch = std.mem.readInt(u64, payload[12..20], .little),
157 }; 166 };
158 } 167 }
159 168
160 /// Fixed prefix of every snapshot payload. Carries the grid size because 169 /// Fixed prefix of every snapshot payload. Carries the grid size because
161 /// under the latest-wins resize policy (M5) a client's tty may not match 170 /// under the latest-wins resize policy (M5) a client's tty may not match
162 /// the authoritative grid; the replica must follow the grid, not the tty. 171 /// the authoritative grid; the replica must follow the grid, not the tty.
172 /// `epoch` identifies the daemon instance that produced `seq`: a client
173 /// echoes it back on reattach so the daemon can tell "you are current" from
174 /// "you are current in a session that no longer exists". Deltas carry no
175 /// epoch — they only ever arrive on a live connection to the instance whose
176 /// snapshot opened it, and that connection dies with the daemon.
163 pub const SnapshotPrefix = struct { 177 pub const SnapshotPrefix = struct {
164 seq: u64, 178 seq: u64,
165 history_rows: u32, 179 history_rows: u32,
166 cols: u16, 180 cols: u16,
167 rows: u16, 181 rows: u16,
182 epoch: u64,
168 }; 183 };
169 184
170 pub const snapshot_prefix_len = 16; 185 pub const snapshot_prefix_len = 24;
171 186
172 pub fn writeSnapshotPrefix(buf: *[snapshot_prefix_len]u8, p: SnapshotPrefix) void { 187 pub fn writeSnapshotPrefix(buf: *[snapshot_prefix_len]u8, p: SnapshotPrefix) void {
173 std.mem.writeInt(u64, buf[0..8], p.seq, .little); 188 std.mem.writeInt(u64, buf[0..8], p.seq, .little);
174 std.mem.writeInt(u32, buf[8..12], p.history_rows, .little); 189 std.mem.writeInt(u32, buf[8..12], p.history_rows, .little);
175 std.mem.writeInt(u16, buf[12..14], p.cols, .little); 190 std.mem.writeInt(u16, buf[12..14], p.cols, .little);
176 std.mem.writeInt(u16, buf[14..16], p.rows, .little); 191 std.mem.writeInt(u16, buf[14..16], p.rows, .little);
192 std.mem.writeInt(u64, buf[16..24], p.epoch, .little);
177 } 193 }
178 194
179 pub fn readSnapshotPrefix(payload: []const u8) !SnapshotPrefix { 195 pub fn readSnapshotPrefix(payload: []const u8) !SnapshotPrefix {
@@ -183,6 +199,7 @@ pub fn readSnapshotPrefix(payload: []const u8) !SnapshotPrefix {
183 .history_rows = std.mem.readInt(u32, payload[8..12], .little), 199 .history_rows = std.mem.readInt(u32, payload[8..12], .little),
184 .cols = std.mem.readInt(u16, payload[12..14], .little), 200 .cols = std.mem.readInt(u16, payload[12..14], .little),
185 .rows = std.mem.readInt(u16, payload[14..16], .little), 201 .rows = std.mem.readInt(u16, payload[14..16], .little),
202 .epoch = std.mem.readInt(u64, payload[16..24], .little),
186 }; 203 };
187 } 204 }
188 205
@@ -373,11 +390,12 @@ test "size encode/decode round trip" {
373 try std.testing.expectEqual(@as(u16, 58), sz.rows); 390 try std.testing.expectEqual(@as(u16, 58), sz.rows);
374 } 391 }
375 392
376 test "attach v2 encode/decode round trip" { 393 test "attach v3 encode/decode round trip" {
377 const a = try decodeAttach(&encodeAttach(120, 40, 987654321)); 394 const a = try decodeAttach(&encodeAttach(120, 40, 987654321, 0xA1B2C3D4E5F60718));
378 try std.testing.expectEqual(@as(u16, 120), a.cols); 395 try std.testing.expectEqual(@as(u16, 120), a.cols);
379 try std.testing.expectEqual(@as(u16, 40), a.rows); 396 try std.testing.expectEqual(@as(u16, 40), a.rows);
380 try std.testing.expectEqual(@as(u64, 987654321), a.have_seq); 397 try std.testing.expectEqual(@as(u64, 987654321), a.have_seq);
398 try std.testing.expectEqual(@as(u64, 0xA1B2C3D4E5F60718), a.have_epoch);
381 } 399 }
382 400
383 test "delta build/iterate round trip" { 401 test "delta build/iterate round trip" {
@@ -497,7 +515,8 @@ test "encodeAttach golden bytes" {
497 0x78, 0x00, // cols 120 515 0x78, 0x00, // cols 120
498 0x28, 0x00, // rows 40 516 0x28, 0x00, // rows 40
499 0xB1, 0x68, 0xDE, 0x3A, 0x00, 0x00, 0x00, 0x00, // have_seq 987654321 517 0xB1, 0x68, 0xDE, 0x3A, 0x00, 0x00, 0x00, 0x00, // have_seq 987654321
500 }, &encodeAttach(120, 40, 987654321)); 518 0x18, 0x07, 0xF6, 0xE5, 0xD4, 0xC3, 0xB2, 0xA1, // have_epoch 0xA1B2C3D4E5F60718
519 }, &encodeAttach(120, 40, 987654321, 0xA1B2C3D4E5F60718));
501 } 520 }
502 521
503 test "delta header golden bytes" { 522 test "delta header golden bytes" {
@@ -522,7 +541,12 @@ test "delta header golden bytes" {
522 541
523 test "decodeAttach rejects a wrong-length payload" { 542 test "decodeAttach rejects a wrong-length payload" {
524 try std.testing.expectError(error.BadPayload, decodeAttach(&encodeSize(80, 24))); 543 try std.testing.expectError(error.BadPayload, decodeAttach(&encodeSize(80, 24)));
525 try std.testing.expectError(error.BadPayload, decodeAttach(&[_]u8{0} ** 13)); 544 try std.testing.expectError(error.BadPayload, decodeAttach(&[_]u8{0} ** 19));
545 try std.testing.expectError(error.BadPayload, decodeAttach(&[_]u8{0} ** 21));
546 // A v2 attach (12 bytes, no epoch) carries no epoch to check, so it is
547 // rejected outright rather than read as epoch 0 — an old client must
548 // fail loudly, not be handed a session it cannot reason about.
549 try std.testing.expectError(error.BadPayload, decodeAttach(&[_]u8{0} ** 12));
526 } 550 }
527 551
528 test "readDeltaHeader rejects a payload one byte short" { 552 test "readDeltaHeader rejects a payload one byte short" {
@@ -530,14 +554,21 @@ test "readDeltaHeader rejects a payload one byte short" {
530 } 554 }
531 555
532 test "snapshot prefix round trip and golden bytes" { 556 test "snapshot prefix round trip and golden bytes" {
533 const p = SnapshotPrefix{ .seq = 258, .history_rows = 7, .cols = 120, .rows = 40 }; 557 const p = SnapshotPrefix{
558 .seq = 258,
559 .history_rows = 7,
560 .cols = 120,
561 .rows = 40,
562 .epoch = 0xDEADBEEFCAFEF00D,
563 };
534 var buf: [snapshot_prefix_len]u8 = undefined; 564 var buf: [snapshot_prefix_len]u8 = undefined;
535 writeSnapshotPrefix(&buf, p); 565 writeSnapshotPrefix(&buf, p);
536 try std.testing.expectEqualSlices(u8, &[_]u8{ 566 try std.testing.expectEqualSlices(u8, &[_]u8{
537 0x02, 0x01, 0, 0, 0, 0, 0, 0, // seq u64 LE 567 0x02, 0x01, 0, 0, 0, 0, 0, 0, // seq u64 LE
538 0x07, 0, 0, 0, // history_rows u32 LE 568 0x07, 0, 0, 0, // history_rows u32 LE
539 0x78, 0, // cols u16 LE 569 0x78, 0, // cols u16 LE
540 0x28, 0, // rows u16 LE 570 0x28, 0, // rows u16 LE
571 0x0D, 0xF0, 0xFE, 0xCA, 0xEF, 0xBE, 0xAD, 0xDE, // epoch u64 LE
541 }, &buf); 572 }, &buf);
542 const q = try readSnapshotPrefix(&buf); 573 const q = try readSnapshotPrefix(&buf);
543 try std.testing.expectEqual(p, q); 574 try std.testing.expectEqual(p, q);
@@ -545,6 +576,10 @@ test "snapshot prefix round trip and golden bytes" {
545 576
546 test "snapshot prefix rejects short payloads" { 577 test "snapshot prefix rejects short payloads" {
547 try std.testing.expectError(error.BadPayload, readSnapshotPrefix(&[_]u8{0} ** 15)); 578 try std.testing.expectError(error.BadPayload, readSnapshotPrefix(&[_]u8{0} ** 15));
579 // A v1 prefix (16 bytes, no epoch) is short now: honouring it would mean
580 // inventing an epoch, and an invented epoch is exactly the thing the
581 // field exists to prevent.
582 try std.testing.expectError(error.BadPayload, readSnapshotPrefix(&[_]u8{0} ** 23));
548 } 583 }
549 584
550 test "truncated row header is rejected" { 585 test "truncated row header is rejected" {
src/server.zig
Old New
@@ -225,6 +225,12 @@ pub const Server = struct {
225 /// Connections that haven't attached (muxd dump, or a client waiting 225 /// Connections that haven't attached (muxd dump, or a client waiting
226 /// to attach). May send debug_dump; attach promotes into a client slot. 226 /// to attach). May send debug_dump; attach promotes into a client slot.
227 observers: [max_observers]?std.posix.fd_t = @splat(null), 227 observers: [max_observers]?std.posix.fd_t = @splat(null),
228 /// Identifies this daemon instance in every snapshot it sends. Seqs are
229 /// only meaningful within one instance: a restarted daemon counts from
230 /// zero over different content, so a reattach quoting a pre-restart
231 /// have_seq must be snapshotted, not deltaed. Never 0 — that value is
232 /// reserved for a client saying "I hold nothing".
233 epoch: u64,
228 /// Row-level change tracking behind the delta stream. 234 /// Row-level change tracking behind the delta stream.
229 tracker: DeltaTracker = .{}, 235 tracker: DeltaTracker = .{},
230 stats: Stats = .{}, 236 stats: Stats = .{},
@@ -243,6 +249,12 @@ pub const Server = struct {
243 var pty = try Pty.spawn(.{ .cols = opts.cols, .rows = opts.rows, .shell = opts.shell }); 249 var pty = try Pty.spawn(.{ .cols = opts.cols, .rows = opts.rows, .shell = opts.shell });
244 errdefer pty.deinit(); 250 errdefer pty.deinit();
245 251
252 // Random rather than a counter or a timestamp: nothing on disk
253 // survives a daemon, and two daemons started in the same
254 // millisecond (tests do exactly this) must still differ.
255 var epoch: u64 = 0;
256 while (epoch == 0) epoch = std.crypto.random.int(u64);
257
246 // systemd socket activation: LISTEN_FDS=1 hands us the listener as fd 3. 258 // systemd socket activation: LISTEN_FDS=1 hands us the listener as fd 3.
247 if (listenFdFromSystemd()) |fd| { 259 if (listenFdFromSystemd()) |fd| {
248 return .{ 260 return .{
@@ -255,6 +267,7 @@ pub const Server = struct {
255 }, 267 },
256 .sock_path = opts.sock_path, 268 .sock_path = opts.sock_path,
257 .owns_sock_file = false, 269 .owns_sock_file = false,
270 .epoch = epoch,
258 }; 271 };
259 } 272 }
260 273
@@ -267,6 +280,7 @@ pub const Server = struct {
267 .listener = try addr.listen(.{}), 280 .listener = try addr.listen(.{}),
268 .sock_path = opts.sock_path, 281 .sock_path = opts.sock_path,
269 .owns_sock_file = true, 282 .owns_sock_file = true,
283 .epoch = epoch,
270 }; 284 };
271 } 285 }
272 286
@@ -650,7 +664,7 @@ pub const Server = struct {
650 // Latest wins: a size change broadcasts, repainting every 664 // Latest wins: a size change broadcasts, repainting every
651 // client at the new attacher's size. A same-size join is 665 // client at the new attacher's size. A same-size join is
652 // the joiner's business alone — see sendResync. 666 // the joiner's business alone — see sendResync.
653 self.sendResync(slot, sz.have_seq, size_changed); 667 self.sendResync(slot, sz.have_seq, sz.have_epoch, size_changed);
654 }, 668 },
655 .debug_dump => self.replyDumpObserver(fd, frame.payload) catch self.dropObserver(i), 669 .debug_dump => self.replyDumpObserver(fd, frame.payload) catch self.dropObserver(i),
656 .stats_req => self.replyStatsObserver(fd) catch self.dropObserver(i), 670 .stats_req => self.replyStatsObserver(fd) catch self.dropObserver(i),
@@ -756,7 +770,7 @@ pub const Server = struct {
756 } 770 }
757 } 771 }
758 772
759 /// The snapshot payload for the grid as it stands: 16-byte prefix ++ 773 /// The snapshot payload for the grid as it stands: the fixed prefix ++
760 /// full-state dump. Caller owns the result. Reads tracker.seq, so it 774 /// full-state dump. Caller owns the result. Reads tracker.seq, so it
761 /// must be called after the rebuild that stamps it. 775 /// must be called after the rebuild that stamps it.
762 fn buildSnapshotPayload(self: *Server) ![]u8 { 776 fn buildSnapshotPayload(self: *Server) ![]u8 {
@@ -769,6 +783,7 @@ pub const Server = struct {
769 .history_rows = self.eng.historyRows(), 783 .history_rows = self.eng.historyRows(),
770 .cols = self.colsNow(), 784 .cols = self.colsNow(),
771 .rows = self.rowsNow(), 785 .rows = self.rowsNow(),
786 .epoch = self.epoch,
772 }); 787 });
773 @memcpy(payload[proto.snapshot_prefix_len..], state); 788 @memcpy(payload[proto.snapshot_prefix_len..], state);
774 return payload; 789 return payload;
@@ -824,12 +839,17 @@ pub const Server = struct {
824 /// this client only. This is the universal case in practice — the 839 /// this client only. This is the universal case in practice — the
825 /// shipped client attaches with have_seq=0 — which is exactly why 840 /// shipped client attaches with have_seq=0 — which is exactly why
826 /// it must not broadcast. 841 /// it must not broadcast.
827 fn sendResync(self: *Server, i: usize, have_seq: u64, size_changed: bool) void { 842 ///
843 /// Serviceable means the seq is ours to interpret: `have_epoch` must
844 /// name THIS daemon instance. Without that check a client holding
845 /// seq 900 from a daemon that has since been restarted would be told
846 /// "you are current" against a session it has never seen a byte of.
847 fn sendResync(self: *Server, i: usize, have_seq: u64, have_epoch: u64, size_changed: bool) void {
828 if (size_changed) { 848 if (size_changed) {
829 self.resyncSnapshot(); 849 self.resyncSnapshot();
830 return; 850 return;
831 } 851 }
832 if (have_seq != 0 and 852 if (have_epoch == self.epoch and have_seq != 0 and
833 have_seq >= self.tracker.reset_seq and have_seq <= self.tracker.seq and 853 have_seq >= self.tracker.reset_seq and have_seq <= self.tracker.seq and
834 self.tracker.rows != 0) 854 self.tracker.rows != 0)
835 { 855 {
@@ -929,7 +949,7 @@ test "Server: survives a client that dies without detaching; next attach works"
929 949
930 // Client 1 attaches, provokes output, then vanishes without detach. 950 // Client 1 attaches, provokes output, then vanishes without detach.
931 const a = try std.net.connectUnixSocket(sock_path); 951 const a = try std.net.connectUnixSocket(sock_path);
932 try proto.writeFrame(a.handle, .attach, &proto.encodeAttach(80, 24, 0)); 952 try proto.writeFrame(a.handle, .attach, &proto.encodeAttach(80, 24, 0, 0));
933 try proto.writeFrame(a.handle, .input, "echo pre-kill\n"); 953 try proto.writeFrame(a.handle, .input, "echo pre-kill\n");
934 std.Thread.sleep(300 * std.time.ns_per_ms); 954 std.Thread.sleep(300 * std.time.ns_per_ms);
935 a.close(); // abrupt: no detach frame 955 a.close(); // abrupt: no detach frame
@@ -940,7 +960,7 @@ test "Server: survives a client that dies without detaching; next attach works"
940 // Daemon must still be serving: a fresh attach gets a snapshot. 960 // Daemon must still be serving: a fresh attach gets a snapshot.
941 const b = try std.net.connectUnixSocket(sock_path); 961 const b = try std.net.connectUnixSocket(sock_path);
942 defer b.close(); 962 defer b.close();
943 try proto.writeFrame(b.handle, .attach, &proto.encodeAttach(80, 24, 0)); 963 try proto.writeFrame(b.handle, .attach, &proto.encodeAttach(80, 24, 0, 0));
944 var got_snapshot = false; 964 var got_snapshot = false;
945 var deadline_ms: u64 = 5000; 965 var deadline_ms: u64 = 5000;
946 while (deadline_ms > 0 and !got_snapshot) { 966 while (deadline_ms > 0 and !got_snapshot) {
@@ -977,7 +997,7 @@ test "Server: serves scrollback chunks on request" {
977 997
978 const c = try std.net.connectUnixSocket(sock_path); 998 const c = try std.net.connectUnixSocket(sock_path);
979 defer c.close(); 999 defer c.close();
980 try proto.writeFrame(c.handle, .attach, &proto.encodeAttach(80, 24, 0)); 1000 try proto.writeFrame(c.handle, .attach, &proto.encodeAttach(80, 24, 0, 0));
981 try proto.writeFrame(c.handle, .input, "seq 1 100\n"); 1001 try proto.writeFrame(c.handle, .input, "seq 1 100\n");
982 1002
983 // Wait until an update reports enough history, then fetch the oldest 1003 // Wait until an update reports enough history, then fetch the oldest
@@ -1053,7 +1073,7 @@ test "Server: a full session refuses the next attach instead of displacing anyon
1053 while (opened < max_clients) : (opened += 1) { 1073 while (opened < max_clients) : (opened += 1) {
1054 streams[opened] = try std.net.connectUnixSocket(sock_path); 1074 streams[opened] = try std.net.connectUnixSocket(sock_path);
1055 const fd = streams[opened].handle; 1075 const fd = streams[opened].handle;
1056 try proto.writeFrame(fd, .attach, &proto.encodeAttach(80, 24, 0)); 1076 try proto.writeFrame(fd, .attach, &proto.encodeAttach(80, 24, 0, 0));
1057 const first = try firstStateFrame(alloc, fd, 10_000); 1077 const first = try firstStateFrame(alloc, fd, 10_000);
1058 try std.testing.expect(first != null); 1078 try std.testing.expect(first != null);
1059 } 1079 }
@@ -1061,7 +1081,7 @@ test "Server: a full session refuses the next attach instead of displacing anyon
1061 // The ninth attach is refused; nobody already attached is evicted. 1081 // The ninth attach is refused; nobody already attached is evicted.
1062 const extra = try std.net.connectUnixSocket(sock_path); 1082 const extra = try std.net.connectUnixSocket(sock_path);
1063 defer extra.close(); 1083 defer extra.close();
1064 try proto.writeFrame(extra.handle, .attach, &proto.encodeAttach(80, 24, 0)); 1084 try proto.writeFrame(extra.handle, .attach, &proto.encodeAttach(80, 24, 0, 0));
1065 var refused = false; 1085 var refused = false;
1066 var deadline_ms: u64 = 5000; 1086 var deadline_ms: u64 = 5000;
1067 while (deadline_ms > 0 and !refused) { 1087 while (deadline_ms > 0 and !refused) {
@@ -1124,10 +1144,10 @@ test "Server: two clients converge on one session" {
1124 1144
1125 const a = try std.net.connectUnixSocket(sock_path); 1145 const a = try std.net.connectUnixSocket(sock_path);
1126 defer a.close(); 1146 defer a.close();
1127 try proto.writeFrame(a.handle, .attach, &proto.encodeAttach(80, 24, 0)); 1147 try proto.writeFrame(a.handle, .attach, &proto.encodeAttach(80, 24, 0, 0));
1128 const b = try std.net.connectUnixSocket(sock_path); 1148 const b = try std.net.connectUnixSocket(sock_path);
1129 defer b.close(); 1149 defer b.close();
1130 try proto.writeFrame(b.handle, .attach, &proto.encodeAttach(80, 24, 0)); 1150 try proto.writeFrame(b.handle, .attach, &proto.encodeAttach(80, 24, 0, 0));
1131 1151
1132 // Input through A must reach both replicas. 1152 // Input through A must reach both replicas.
1133 try proto.writeFrame(a.handle, .input, "echo both-see-this\n"); 1153 try proto.writeFrame(a.handle, .input, "echo both-see-this\n");
@@ -1231,7 +1251,7 @@ test "Server: a same-size join snapshots the joiner only" {
1231 1251
1232 const a = try std.net.connectUnixSocket(sock_path); 1252 const a = try std.net.connectUnixSocket(sock_path);
1233 defer a.close(); 1253 defer a.close();
1234 try proto.writeFrame(a.handle, .attach, &proto.encodeAttach(80, 24, 0)); 1254 try proto.writeFrame(a.handle, .attach, &proto.encodeAttach(80, 24, 0, 0));
1235 1255
1236 var replica_a = try Engine.init(alloc, .{ .cols = 80, .rows = 24 }); 1256 var replica_a = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });
1237 defer replica_a.deinit(); 1257 defer replica_a.deinit();
@@ -1263,7 +1283,7 @@ test "Server: a same-size join snapshots the joiner only" {
1263 // not be repainted. B, which has nothing, must be. 1283 // not be repainted. B, which has nothing, must be.
1264 const b = try std.net.connectUnixSocket(sock_path); 1284 const b = try std.net.connectUnixSocket(sock_path);
1265 defer b.close(); 1285 defer b.close();
1266 try proto.writeFrame(b.handle, .attach, &proto.encodeAttach(80, 24, 0)); 1286 try proto.writeFrame(b.handle, .attach, &proto.encodeAttach(80, 24, 0, 0));
1267 1287
1268 var b_snapshot = false; 1288 var b_snapshot = false;
1269 deadline_ms = 5000; 1289 deadline_ms = 5000;
@@ -1617,7 +1637,7 @@ test "Server: the shell's exit status reaches an attached client" {
1617 1637
1618 const c = try std.net.connectUnixSocket(sock_path); 1638 const c = try std.net.connectUnixSocket(sock_path);
1619 defer c.close(); 1639 defer c.close();
1620 try proto.writeFrame(c.handle, .attach, &proto.encodeAttach(80, 24, 0)); 1640 try proto.writeFrame(c.handle, .attach, &proto.encodeAttach(80, 24, 0, 0));
1621 1641
1622 // Settle first, so the shell is ready for input and the exit below is 1642 // Settle first, so the shell is ready for input and the exit below is
1623 // the next thing that happens. 1643 // the next thing that happens.
@@ -1856,7 +1876,7 @@ test "Server: latest attacher's size wins; earlier client is resnapshotted at th
1856 1876
1857 const a = try std.net.connectUnixSocket(sock_path); 1877 const a = try std.net.connectUnixSocket(sock_path);
1858 defer a.close(); 1878 defer a.close();
1859 try proto.writeFrame(a.handle, .attach, &proto.encodeAttach(80, 24, 0)); 1879 try proto.writeFrame(a.handle, .attach, &proto.encodeAttach(80, 24, 0, 0));
1860 1880
1861 // A is attached once it has been answered; only then can B's attach be 1881 // A is attached once it has been answered; only then can B's attach be
1862 // the *later* event this test is about. 1882 // the *later* event this test is about.
@@ -1881,7 +1901,7 @@ test "Server: latest attacher's size wins; earlier client is resnapshotted at th
1881 1901
1882 const b = try std.net.connectUnixSocket(sock_path); 1902 const b = try std.net.connectUnixSocket(sock_path);
1883 defer b.close(); 1903 defer b.close();
1884 try proto.writeFrame(b.handle, .attach, &proto.encodeAttach(100, 30, 0)); 1904 try proto.writeFrame(b.handle, .attach, &proto.encodeAttach(100, 30, 0, 0));
1885 1905
1886 // The grid follows the newest attacher, and A is told about it. 1906 // The grid follows the newest attacher, and A is told about it.
1887 var a_resized = false; 1907 var a_resized = false;
@@ -1928,10 +1948,10 @@ test "Server: scrollback fetch is per-client and independent" {
1928 1948
1929 const a = try std.net.connectUnixSocket(sock_path); 1949 const a = try std.net.connectUnixSocket(sock_path);
1930 defer a.close(); 1950 defer a.close();
1931 try proto.writeFrame(a.handle, .attach, &proto.encodeAttach(80, 24, 0)); 1951 try proto.writeFrame(a.handle, .attach, &proto.encodeAttach(80, 24, 0, 0));
1932 const b = try std.net.connectUnixSocket(sock_path); 1952 const b = try std.net.connectUnixSocket(sock_path);
1933 defer b.close(); 1953 defer b.close();
1934 try proto.writeFrame(b.handle, .attach, &proto.encodeAttach(80, 24, 0)); 1954 try proto.writeFrame(b.handle, .attach, &proto.encodeAttach(80, 24, 0, 0));
1935 1955
1936 // Typed by B; both clients see the history it produces. 1956 // Typed by B; both clients see the history it produces.
1937 try proto.writeFrame(b.handle, .input, "seq 1 100\n"); 1957 try proto.writeFrame(b.handle, .input, "seq 1 100\n");
@@ -2030,7 +2050,7 @@ test "Server: replica rebuilt from snapshots matches the authoritative grid" {
2030 var replica = try Engine.init(alloc, .{ .cols = 100, .rows = 30 }); 2050 var replica = try Engine.init(alloc, .{ .cols = 100, .rows = 30 });
2031 defer replica.deinit(); 2051 defer replica.deinit();
2032 2052
2033 try proto.writeFrame(fd, .attach, &proto.encodeAttach(100, 30, 0)); 2053 try proto.writeFrame(fd, .attach, &proto.encodeAttach(100, 30, 0, 0));
2034 try proto.writeFrame(fd, .input, "printf 'fidelity-%s\\n' ok\n"); 2054 try proto.writeFrame(fd, .input, "printf 'fidelity-%s\\n' ok\n");
2035 2055
2036 // Consume the attach snapshot and the deltas that follow it until the 2056 // Consume the attach snapshot and the deltas that follow it until the
@@ -2102,7 +2122,7 @@ test "Server: typing produces deltas, not snapshots; stats track both" {
2102 2122
2103 const c = try std.net.connectUnixSocket(sock_path); 2123 const c = try std.net.connectUnixSocket(sock_path);
2104 defer c.close(); 2124 defer c.close();
2105 try proto.writeFrame(c.handle, .attach, &proto.encodeAttach(80, 24, 0)); 2125 try proto.writeFrame(c.handle, .attach, &proto.encodeAttach(80, 24, 0, 0));
2106 2126
2107 // The attach is answered with a full snapshot carrying the tracker seq. 2127 // The attach is answered with a full snapshot carrying the tracker seq.
2108 var attach_seq: u64 = 0; 2128 var attach_seq: u64 = 0;
@@ -2178,7 +2198,7 @@ test "Server: typing produces deltas, not snapshots; stats track both" {
2178 try std.testing.expect(std.mem.indexOf(u8, stats_text.?, "snapshots=") != null); 2198 try std.testing.expect(std.mem.indexOf(u8, stats_text.?, "snapshots=") != null);
2179 } 2199 }
2180 2200
2181 test "Server: reattach with a recent have_seq gets a delta, stale gets snapshot" { 2201 test "Server: reattach needs a recent have_seq AND this daemon's epoch to get a delta" {
2182 const alloc = std.testing.allocator; 2202 const alloc = std.testing.allocator;
2183 2203
2184 var tmp = std.testing.tmpDir(.{}); 2204 var tmp = std.testing.tmpDir(.{});
@@ -2196,71 +2216,213 @@ test "Server: reattach with a recent have_seq gets a delta, stale gets snapshot"
2196 defer th.join(); 2216 defer th.join();
2197 defer stop.store(true, .release); 2217 defer stop.store(true, .release);
2198 2218
2199 // Session 1: attach cold, produce output, note the newest seq we hold. 2219 // Session 1: attach cold, produce output, note the newest seq we hold
2220 // and the session epoch the daemon stamps into its snapshots.
2200 var last_seq: u64 = 0; 2221 var last_seq: u64 = 0;
2222 var epoch: u64 = 0;
2201 { 2223 {
2202 const c1 = try std.net.connectUnixSocket(sock_path); 2224 const c1 = try std.net.connectUnixSocket(sock_path);
2203 defer c1.close(); 2225 defer c1.close();
2204 try proto.writeFrame(c1.handle, .attach, &proto.encodeAttach(80, 24, 0)); 2226 try proto.writeFrame(c1.handle, .attach, &proto.encodeAttach(80, 24, 0, 0));
2205 // A resize is a discontinuity, so this pins reset_seq above 1 no 2227 // A resize is a discontinuity, so this pins reset_seq above 1 no
2206 // matter how the shell's first output raced the attach — session 3 2228 // matter how the shell's first output raced the attach — session 3
2207 // below needs have_seq=1 to be reliably stale. 2229 // below needs have_seq=1 to be reliably stale.
2208 try proto.writeFrame(c1.handle, .resize, &proto.encodeSize(80, 24)); 2230 try proto.writeFrame(c1.handle, .resize, &proto.encodeSize(80, 24));
2209 try proto.writeFrame(c1.handle, .input, "echo before-detach\n"); 2231 try proto.writeFrame(c1.handle, .input, "echo before-detach\n");
2210 2232
2211 var quiet_ms: u64 = 0; 2233 const held = try drainHeld(alloc, c1.handle, 10_000);
2212 var deadline_ms: u64 = 10_000; 2234 last_seq = held.seq;
2213 while (deadline_ms > 0 and quiet_ms < 500) { 2235 epoch = held.epoch;
2214 var pfd = [_]std.posix.pollfd{
2215 .{ .fd = c1.handle, .events = std.posix.POLL.IN, .revents = 0 },
2216 };
2217 const ready = try std.posix.poll(&pfd, 100);
2218 deadline_ms -|= 100;
2219 if (ready == 0) {
2220 quiet_ms += 100;
2221 continue;
2222 }
2223 quiet_ms = 0;
2224 const frame = (try proto.readFrame(alloc, c1.handle)) orelse break;
2225 defer frame.deinit(alloc);
2226 switch (frame.type) {
2227 .snapshot => {
2228 last_seq = (try proto.readSnapshotPrefix(frame.payload)).seq;
2229 },
2230 .delta => last_seq = (try proto.readDeltaHeader(frame.payload)).seq,
2231 else => {},
2232 }
2233 }
2234 try std.testing.expect(last_seq > 0); 2236 try std.testing.expect(last_seq > 0);
2237 // 0 is reserved for "I hold no epoch", so a live daemon never has it.
2238 try std.testing.expect(epoch != 0);
2235 try proto.writeFrame(c1.handle, .detach, ""); 2239 try proto.writeFrame(c1.handle, .detach, "");
2236 } 2240 }
2237 2241
2238 // Session 2: we are up to date as of last_seq, so the daemon owes us a 2242 // Session 2: the seq is real but the epoch belongs to some other daemon
2239 // delta (possibly empty), never a full repaint. 2243 // instance. Those seqs describe a different history; honouring one would
2244 // paint this session's screen with another's rows. Snapshot.
2240 { 2245 {
2241 const c2 = try std.net.connectUnixSocket(sock_path); 2246 const c2 = try std.net.connectUnixSocket(sock_path);
2242 defer c2.close(); 2247 defer c2.close();
2243 try proto.writeFrame(c2.handle, .attach, &proto.encodeAttach(80, 24, last_seq)); 2248 try proto.writeFrame(c2.handle, .attach, &proto.encodeAttach(80, 24, last_seq, epoch ^ 1));
2244 const first = try firstStateFrame(alloc, c2.handle, 10_000); 2249 const first = try firstStateFrame(alloc, c2.handle, 10_000);
2245 try std.testing.expect(first != null); 2250 try std.testing.expect(first != null);
2246 try std.testing.expectEqual(proto.MsgType.delta, first.?); 2251 try std.testing.expectEqual(proto.MsgType.snapshot, first.?.type);
2252 // That snapshot is itself a discontinuity (rebuild bumps seq and
2253 // reset_seq), so the sessions below must resync off it, not off the
2254 // seq session 1 learned.
2255 last_seq = first.?.seq;
2256 try std.testing.expectEqual(epoch, first.?.epoch);
2247 try proto.writeFrame(c2.handle, .detach, ""); 2257 try proto.writeFrame(c2.handle, .detach, "");
2248 } 2258 }
2249 2259
2250 // Session 3: have_seq predates the last discontinuity, so no delta can 2260 // Session 3: a real seq with have_epoch = 0 — a pre-epoch client, or one
2251 // reconstruct our state — full snapshot. 2261 // hoping 0 means "any". It means "none", and none is never serviceable.
2252 { 2262 {
2253 const c3 = try std.net.connectUnixSocket(sock_path); 2263 const c3 = try std.net.connectUnixSocket(sock_path);
2254 defer c3.close(); 2264 defer c3.close();
2255 try proto.writeFrame(c3.handle, .attach, &proto.encodeAttach(80, 24, 1)); 2265 try proto.writeFrame(c3.handle, .attach, &proto.encodeAttach(80, 24, last_seq, 0));
2256 const first = try firstStateFrame(alloc, c3.handle, 10_000); 2266 const first = try firstStateFrame(alloc, c3.handle, 10_000);
2257 try std.testing.expect(first != null); 2267 try std.testing.expect(first != null);
2258 try std.testing.expectEqual(proto.MsgType.snapshot, first.?); 2268 try std.testing.expectEqual(proto.MsgType.snapshot, first.?.type);
2269 last_seq = first.?.seq;
2270 try proto.writeFrame(c3.handle, .detach, "");
2271 }
2272
2273 // Session 4: right seq, right epoch — we really are up to date, so the
2274 // daemon owes us a delta (possibly empty), never a full repaint.
2275 {
2276 const c4 = try std.net.connectUnixSocket(sock_path);
2277 defer c4.close();
2278 try proto.writeFrame(c4.handle, .attach, &proto.encodeAttach(80, 24, last_seq, epoch));
2279 const first = try firstStateFrame(alloc, c4.handle, 10_000);
2280 try std.testing.expect(first != null);
2281 try std.testing.expectEqual(proto.MsgType.delta, first.?.type);
2282 try proto.writeFrame(c4.handle, .detach, "");
2283 }
2284
2285 // Session 5: right epoch, but have_seq predates the last discontinuity,
2286 // so no delta can reconstruct our state — full snapshot.
2287 {
2288 const c5 = try std.net.connectUnixSocket(sock_path);
2289 defer c5.close();
2290 try proto.writeFrame(c5.handle, .attach, &proto.encodeAttach(80, 24, 1, epoch));
2291 const first = try firstStateFrame(alloc, c5.handle, 10_000);
2292 try std.testing.expect(first != null);
2293 try std.testing.expectEqual(proto.MsgType.snapshot, first.?.type);
2294 }
2295 }
2296
2297 test "Server: a daemon restart invalidates have_seq even with the old epoch presented" {
2298 const alloc = std.testing.allocator;
2299
2300 var tmp = std.testing.tmpDir(.{});
2301 defer tmp.cleanup();
2302 var path_buf: [256]u8 = undefined;
2303 const dir_path = try tmp.dir.realpath(".", &path_buf);
2304 const sock_path = try std.fmt.allocPrint(alloc, "{s}/restart.sock", .{dir_path});
2305 defer alloc.free(sock_path);
2306
2307 // Daemon A: what a client is holding at the moment the daemon dies
2308 // under it — a seq, and the epoch that seq is counted in.
2309 var held_a: Held = undefined;
2310 {
2311 var srv_a = try Server.init(alloc, .{ .sock_path = sock_path, .shell = "/bin/sh" });
2312 defer srv_a.deinit();
2313 var stop_a = std.atomic.Value(bool).init(false);
2314 const th_a = try std.Thread.spawn(.{}, serverThread, .{ &srv_a, &stop_a });
2315 defer th_a.join();
2316 defer stop_a.store(true, .release);
2317
2318 const c = try std.net.connectUnixSocket(sock_path);
2319 defer c.close();
2320 try proto.writeFrame(c.handle, .attach, &proto.encodeAttach(80, 24, 0, 0));
2321 try proto.writeFrame(c.handle, .input, "echo before-restart\n");
2322 held_a = try drainHeld(alloc, c.handle, 10_000);
2323 try std.testing.expect(held_a.seq > 0);
2324 try std.testing.expect(held_a.epoch != 0);
2325 try proto.writeFrame(c.handle, .detach, "");
2326 }
2327 // A is gone: thread joined, deinit ran, socket unlinked.
2328
2329 // Daemon B on the same path — the restart. Its seqs start over, so A's
2330 // numbers now name rows B has never produced.
2331 var srv_b = try Server.init(alloc, .{ .sock_path = sock_path, .shell = "/bin/sh" });
2332 defer srv_b.deinit();
2333 var stop_b = std.atomic.Value(bool).init(false);
2334 const th_b = try std.Thread.spawn(.{}, serverThread, .{ &srv_b, &stop_b });
2335 defer th_b.join();
2336 defer stop_b.store(true, .release);
2337
2338 // Learn what B itself is up to, so the check below can vary the epoch
2339 // alone. Two sequential inits are observed to differ here: that is the
2340 // distinctness this whole mechanism rests on, and without pinning it a
2341 // constant epoch would satisfy every other test in the file.
2342 var held_b: Held = undefined;
2343 {
2344 const c = try std.net.connectUnixSocket(sock_path);
2345 defer c.close();
2346 try proto.writeFrame(c.handle, .attach, &proto.encodeAttach(80, 24, 0, 0));
2347 held_b = try drainHeld(alloc, c.handle, 10_000);
2348 try std.testing.expect(held_b.seq > 0);
2349 try std.testing.expect(held_b.epoch != 0);
2350 try std.testing.expect(held_b.epoch != held_a.epoch);
2351 try proto.writeFrame(c.handle, .detach, "");
2352 }
2353
2354 // A seq B issued moments ago, quoted back with A's epoch. Everything
2355 // about this attach is serviceable except the epoch, so a delta here
2356 // would mean the epoch is not being checked at all.
2357 {
2358 const c = try std.net.connectUnixSocket(sock_path);
2359 defer c.close();
2360 try proto.writeFrame(c.handle, .attach, &proto.encodeAttach(80, 24, held_b.seq, held_a.epoch));
2361 const first = try firstStateFrame(alloc, c.handle, 10_000);
2362 try std.testing.expect(first != null);
2363 try std.testing.expectEqual(proto.MsgType.snapshot, first.?.type);
2364 // Read off the wire, not off srv_b: the server thread is live.
2365 try std.testing.expectEqual(held_b.epoch, first.?.epoch);
2366 try proto.writeFrame(c.handle, .detach, "");
2367 }
2368
2369 // And the literal restart case: the exact pair the pre-restart client
2370 // held. B has no way to reconstruct that state, so: snapshot, stamped
2371 // with B's own epoch, which the client adopts in place of A's.
2372 {
2373 const c = try std.net.connectUnixSocket(sock_path);
2374 defer c.close();
2375 try proto.writeFrame(c.handle, .attach, &proto.encodeAttach(80, 24, held_a.seq, held_a.epoch));
2376 const first = try firstStateFrame(alloc, c.handle, 10_000);
2377 try std.testing.expect(first != null);
2378 try std.testing.expectEqual(proto.MsgType.snapshot, first.?.type);
2379 try std.testing.expect(first.?.epoch != held_a.epoch);
2259 } 2380 }
2260 } 2381 }
2261 2382
2262 /// Test helper: the type of the first snapshot-or-delta frame to arrive. 2383 /// Test helper: the first snapshot-or-delta frame to arrive, reduced to what
2263 fn firstStateFrame(alloc: std.mem.Allocator, fd: std.posix.fd_t, timeout_ms: u64) !?proto.MsgType { 2384 /// the resync tests assert on. `epoch` is 0 for a delta — only snapshots
2385 /// carry the session epoch.
2386 const StateFrame = struct { type: proto.MsgType, seq: u64, epoch: u64 };
2387
2388 /// What a client would be holding if it detached now: the newest seq it has
2389 /// seen and the epoch stamped on the last snapshot (0 if only deltas came).
2390 const Held = struct { seq: u64, epoch: u64 };
2391
2392 /// Test helper: read an attached connection until it goes quiet, then report
2393 /// what it holds. "Quiet" rather than a frame count because a shell's
2394 /// startup output arrives as an unpredictable number of frames.
2395 fn drainHeld(alloc: std.mem.Allocator, fd: std.posix.fd_t, timeout_ms: u64) !Held {
2396 var held: Held = .{ .seq = 0, .epoch = 0 };
2397 var quiet_ms: u64 = 0;
2398 var deadline_ms = timeout_ms;
2399 while (deadline_ms > 0 and quiet_ms < 500) {
2400 var pfd = [_]std.posix.pollfd{
2401 .{ .fd = fd, .events = std.posix.POLL.IN, .revents = 0 },
2402 };
2403 const ready = try std.posix.poll(&pfd, 100);
2404 deadline_ms -|= 100;
2405 if (ready == 0) {
2406 quiet_ms += 100;
2407 continue;
2408 }
2409 quiet_ms = 0;
2410 const frame = (try proto.readFrame(alloc, fd)) orelse break;
2411 defer frame.deinit(alloc);
2412 switch (frame.type) {
2413 .snapshot => {
2414 const p = try proto.readSnapshotPrefix(frame.payload);
2415 held.seq = p.seq;
2416 held.epoch = p.epoch;
2417 },
2418 .delta => held.seq = (try proto.readDeltaHeader(frame.payload)).seq,
2419 else => {},
2420 }
2421 }
2422 return held;
2423 }
2424
2425 fn firstStateFrame(alloc: std.mem.Allocator, fd: std.posix.fd_t, timeout_ms: u64) !?StateFrame {
2264 var deadline_ms = timeout_ms; 2426 var deadline_ms = timeout_ms;
2265 while (deadline_ms > 0) { 2427 while (deadline_ms > 0) {
2266 var pfd = [_]std.posix.pollfd{ 2428 var pfd = [_]std.posix.pollfd{
@@ -2271,7 +2433,17 @@ fn firstStateFrame(alloc: std.mem.Allocator, fd: std.posix.fd_t, timeout_ms: u64
2271 if (ready == 0) continue; 2433 if (ready == 0) continue;
2272 const frame = (try proto.readFrame(alloc, fd)) orelse return null; 2434 const frame = (try proto.readFrame(alloc, fd)) orelse return null;
2273 defer frame.deinit(alloc); 2435 defer frame.deinit(alloc);
2274 if (frame.type == .snapshot or frame.type == .delta) return frame.type; 2436 switch (frame.type) {
2437 .snapshot => {
2438 const p = try proto.readSnapshotPrefix(frame.payload);
2439 return .{ .type = .snapshot, .seq = p.seq, .epoch = p.epoch };
2440 },
2441 .delta => {
2442 const h = try proto.readDeltaHeader(frame.payload);
2443 return .{ .type = .delta, .seq = h.seq, .epoch = 0 };
2444 },
2445 else => {},
2446 }
2275 } 2447 }
2276 return null; 2448 return null;
2277 } 2449 }