a73x

f9f8a3f2

refactor: DeltaTracker and socket-path identity extract with their tests

a73x   2026-08-12 19:06

Commit message
refactor: DeltaTracker and socket-path identity extract with their tests

src/delta.zig takes the row-hash tracker (Engine + protocol and nothing
else) and absorbs sendResync's seq-serviceability clauses as canServe;
the epoch clause stays at the call site, since the tracker does not know
which daemon instance it belongs to.

src/sockpath.zig takes the path claim and the dev+ino record teardown
compares against. The pair becomes PathId, so the field incident its
doc records travels with the thing it describes, and deinit's
stat-and-compare block is one call.

The six socket tests all drive a full Server, so all six stay: they pin
Server's use of the module, not the module. The two tracker tests need
only an engine, so they move.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
build.zig
Old New
@@ -173,6 +173,27 @@ pub fn build(b: *std.Build) void {
173 // file shares with the key file. xdg is a leaf, so no cycle. 173 // file shares with the key file. xdg is a leaf, so no cycle.
174 handoff_mod.addImport("xdg", xdg_mod); 174 handoff_mod.addImport("xdg", xdg_mod);
175 175
176 // Row-level change tracking behind the delta stream. Engine plus
177 // protocol and nothing else — no daemon, no clients — so the tracker's
178 // own tests drive it with an engine and no socket in sight.
179 const delta_mod = b.createModule(.{
180 .root_source_file = b.path("src/delta.zig"),
181 .target = target,
182 .optimize = optimize,
183 });
184 delta_mod.addImport("engine", engine_mod);
185 delta_mod.addImport("protocol", protocol_mod);
186
187 // 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
189 // takes a path and nothing else, and knows no Server exists.
190 const sockpath_mod = b.createModule(.{
191 .root_source_file = b.path("src/sockpath.zig"),
192 .target = target,
193 .optimize = optimize,
194 .link_libc = true,
195 });
196
176 const server_mod = b.createModule(.{ 197 const server_mod = b.createModule(.{
177 .root_source_file = b.path("src/server.zig"), 198 .root_source_file = b.path("src/server.zig"),
178 .target = target, 199 .target = target,
@@ -182,6 +203,8 @@ pub fn build(b: *std.Build) void {
182 server_mod.addImport("engine", engine_mod); 203 server_mod.addImport("engine", engine_mod);
183 server_mod.addImport("pty", pty_mod); 204 server_mod.addImport("pty", pty_mod);
184 server_mod.addImport("protocol", protocol_mod); 205 server_mod.addImport("protocol", protocol_mod);
206 server_mod.addImport("delta", delta_mod);
207 server_mod.addImport("sockpath", sockpath_mod);
185 // Both: the listener it owns, and the vocabulary it names directly 208 // Both: the listener it owns, and the vocabulary it names directly
186 // (the key it loads, the idle default it falls back to). 209 // (the key it loads, the idle default it falls back to).
187 server_mod.addImport("quic", quic_mod); 210 server_mod.addImport("quic", quic_mod);
@@ -356,12 +379,18 @@ pub fn build(b: *std.Build) void {
356 b.installArtifact(ptyclient_exe); 379 b.installArtifact(ptyclient_exe);
357 380
358 const test_step = b.step("test", "Run unit tests"); 381 const test_step = b.step("test", "Run unit tests");
382 // delta_mod and sockpath_mod sit BEFORE server_mod, deliberately: their
383 // tests are seconds-long and socket-free, while a regression in either
384 // can wedge a server test that waits on a client forever — and a wedged
385 // step prints nothing at all. Failing first is what makes the catch
386 // legible.
387 //
359 // mux_mod and exe_mod are executable roots, but they carry the argument 388 // mux_mod and exe_mod are executable roots, but they carry the argument
360 // parsers, and a test that is never built is not a test. exe_mod's 389 // parsers, and a test that is never built is not a test. exe_mod's
361 // absence here was a live hazard recorded in decisions.md — muxd's 390 // absence here was a live hazard recorded in decisions.md — muxd's
362 // entrypoint could grow tests that silently never ran, exactly as 391 // entrypoint could grow tests that silently never ran, exactly as
363 // mux_main.zig's five did before it was added. 392 // mux_main.zig's five did before it was added.
364 for ([_]*std.Build.Module{ protocol_mod, engine_mod, pty_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| { 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| {
365 const t = b.addTest(.{ .root_module = mod }); 394 const t = b.addTest(.{ .root_module = mod });
366 t.use_llvm = true; 395 t.use_llvm = true;
367 t.use_lld = true; 396 t.use_lld = true;
src/delta.zig
Old New
@@ -0,0 +1,231 @@
1 //! Row-level change tracking behind the delta stream: one content hash per
2 //! viewport row, so an engine update can be sent as just the rows that
3 //! actually changed, and a reattach can be answered with a delta instead of
4 //! a full snapshot.
5 //!
6 //! Engine and protocol are the whole of its world — no daemon, no clients,
7 //! no sockets — which is what lets the tracker be driven directly by a test
8 //! holding nothing but an engine.
9 const std = @import("std");
10 const Engine = @import("engine").Engine;
11 const proto = @import("protocol");
12
13 const Wyhash = std.hash.Wyhash;
14
15 /// Tracks a content hash per viewport row so an engine update can be sent
16 /// as just the rows that actually changed. Advances whether or not a
17 /// client is attached, so a reattach can be answered with a delta.
18 pub const DeltaTracker = struct {
19 seq: u64 = 0,
20 /// Seq at the last discontinuity (init/resize/screen switch). Clients
21 /// with have_seq older than this cannot be served a delta.
22 reset_seq: u64 = 0,
23 cols: u16 = 0,
24 rows: u16 = 0,
25 on_alt: bool = false,
26 cursor: Engine.CursorPos = .{ .x = 0, .y = 0 },
27 history_rows: u32 = 0,
28 row_hashes: []u64 = &.{},
29 row_seqs: []u64 = &.{},
30 /// Where update() writes the hashes it is computing, so a diff pass
31 /// allocates nothing. Swapped with row_hashes once per advance.
32 scratch_hashes: []u64 = &.{},
33
34 pub fn deinit(self: *DeltaTracker, alloc: std.mem.Allocator) void {
35 alloc.free(self.row_hashes);
36 alloc.free(self.row_seqs);
37 alloc.free(self.scratch_hashes);
38 }
39
40 /// Re-hash every row and mark a discontinuity: what follows can only be
41 /// carried by a full snapshot.
42 pub fn rebuild(self: *DeltaTracker, alloc: std.mem.Allocator, eng: *Engine, rows: u16, cols: u16) !void {
43 if (self.row_hashes.len != rows) {
44 // Every allocation lands before any old array is released, so a
45 // failure here leaves the tracker exactly as it was rather than
46 // half-freed. update()'s geometry check turns the resulting
47 // stale-but-consistent state into a snapshot on the next pump.
48 const hashes = try alloc.alloc(u64, rows);
49 errdefer alloc.free(hashes);
50 const seqs = try alloc.alloc(u64, rows);
51 errdefer alloc.free(seqs);
52 const scratch = try alloc.alloc(u64, rows);
53 alloc.free(self.row_hashes);
54 alloc.free(self.row_seqs);
55 alloc.free(self.scratch_hashes);
56 self.row_hashes = hashes;
57 self.row_seqs = seqs;
58 self.scratch_hashes = scratch;
59 }
60 self.cols = cols;
61 self.on_alt = eng.onAltScreen();
62 self.seq += 1;
63 self.reset_seq = self.seq;
64 self.cursor = eng.cursorPos();
65 self.history_rows = eng.historyRows();
66 // Claim no rows until every row is stamped. A dump that fails
67 // partway would otherwise leave stale seqs in the tail of
68 // row_seqs, and any of them above a client's have_seq would put
69 // that row in every delta from here on. rows == 0 sends update()
70 // down its first-run branch and sendResync to a snapshot instead.
71 self.rows = 0;
72 for (0..rows) |y| {
73 const bytes = try eng.dumpVtRow(alloc, @intCast(y));
74 defer alloc.free(bytes);
75 self.row_hashes[y] = Wyhash.hash(0, bytes);
76 self.row_seqs[y] = self.seq;
77 }
78 self.rows = rows;
79 }
80
81 pub const Update = union(enum) {
82 none,
83 discontinuity,
84 /// Something changed and the tracker now describes it; the payload
85 /// is built separately, and only if a client is there to read it.
86 advanced,
87 };
88
89 /// Diff current engine state against the tracked state. Advances seq
90 /// and tracked rows when anything changed. Allocates only the per-row
91 /// dumps it hashes.
92 pub fn update(self: *DeltaTracker, alloc: std.mem.Allocator, eng: *Engine) !Update {
93 if (self.rows == 0) return .discontinuity; // never built
94 // Geometry is load-bearing, not decorative: row_hashes is indexed
95 // by the tracker's own row count, and dumpVtRow asserts against the
96 // engine's. A tracker left stale by a failed rebuild resyncs here
97 // instead of running off the end of the grid.
98 if (self.rows != eng.term.rows or self.cols != eng.term.cols) return .discontinuity;
99 if (eng.onAltScreen() != self.on_alt) return .discontinuity;
100
101 // Safe to stamp row_seqs with the seq we may not end up taking: it
102 // is only written for rows whose hash changed, and any changed row
103 // forces the advance below.
104 const next_seq = self.seq + 1;
105 var changed: usize = 0;
106 for (0..self.rows) |y| {
107 const bytes = try eng.dumpVtRow(alloc, @intCast(y));
108 defer alloc.free(bytes);
109 const hash = Wyhash.hash(0, bytes);
110 self.scratch_hashes[y] = hash;
111 if (hash != self.row_hashes[y]) {
112 self.row_seqs[y] = next_seq;
113 changed += 1;
114 }
115 }
116
117 const cur = eng.cursorPos();
118 const hist = eng.historyRows();
119 const cursor_moved = cur.x != self.cursor.x or cur.y != self.cursor.y;
120 if (changed == 0 and !cursor_moved and hist == self.history_rows)
121 return .none;
122
123 self.seq = next_seq;
124 self.cursor = cur;
125 self.history_rows = hist;
126 std.mem.swap([]u64, &self.row_hashes, &self.scratch_hashes);
127 return .advanced;
128 }
129
130 /// Can a client holding `have_seq` be sent a delta rather than a full
131 /// snapshot? Only the tracker's half of that question: whether the seq
132 /// falls inside the span this tracker can still describe. Whether the
133 /// seq is even ours to interpret — the epoch — is the caller's, since
134 /// the tracker does not know which daemon instance it belongs to.
135 ///
136 /// 0 is never serviceable: it is what a client says when it holds
137 /// nothing.
138 pub fn canServe(self: *const DeltaTracker, have_seq: u64) bool {
139 return have_seq != 0 and
140 have_seq >= self.reset_seq and have_seq <= self.seq and
141 self.rows != 0;
142 }
143
144 /// Build a delta payload of all rows changed after `since`. The header
145 /// row_count and the appended rows MUST agree (composeDelta validates),
146 /// so both come from the same `row_seq > since` predicate over row_seqs
147 /// with nothing mutating in between.
148 ///
149 /// Changed rows get dumped twice per update — once to hash, once here
150 /// to serialize. That is the price of one row-selection routine serving
151 /// both the live stream and the attach path, and it is 1-3 rows in
152 /// steady state.
153 pub fn buildDeltaSince(self: *DeltaTracker, alloc: std.mem.Allocator, eng: *Engine, since: u64) ![]u8 {
154 var rows_changed: u16 = 0;
155 for (self.row_seqs) |s| {
156 if (s > since) rows_changed += 1;
157 }
158 var payload: std.ArrayList(u8) = .empty;
159 errdefer payload.deinit(alloc);
160 try proto.appendDeltaHeader(&payload, alloc, .{
161 .seq = self.seq,
162 .history_rows = self.history_rows,
163 .cursor_x = self.cursor.x,
164 .cursor_y = self.cursor.y,
165 .row_count = rows_changed,
166 });
167 for (self.row_seqs, 0..) |s, y| {
168 if (s <= since) continue;
169 const bytes = try eng.dumpVtRow(alloc, @intCast(y));
170 defer alloc.free(bytes);
171 try proto.appendDeltaRow(&payload, alloc, @intCast(y), bytes);
172 }
173 return payload.toOwnedSlice(alloc);
174 }
175 };
176
177 test "DeltaTracker: alt-screen flip is a discontinuity and rows follow the active screen" {
178 const alloc = std.testing.allocator;
179
180 const eng = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });
181 defer eng.deinit();
182
183 var tracker: DeltaTracker = .{};
184 defer tracker.deinit(alloc);
185 try tracker.rebuild(alloc, eng, 24, 80);
186
187 eng.feed("primary text");
188 switch (try tracker.update(alloc, eng)) {
189 .advanced => {},
190 else => return error.ExpectedAdvance,
191 }
192
193 // Switching screens replaces every row at once: the tracked hashes
194 // describe the other screen, so a delta would be a lie.
195 eng.feed("\x1b[?1049h");
196 switch (try tracker.update(alloc, eng)) {
197 .discontinuity => {},
198 else => return error.ExpectedDiscontinuity,
199 }
200
201 try tracker.rebuild(alloc, eng, 24, 80);
202 eng.feed("alt content");
203 switch (try tracker.update(alloc, eng)) {
204 .advanced => {},
205 else => return error.ExpectedAdvance,
206 }
207 const payload = try tracker.buildDeltaSince(alloc, eng, tracker.seq - 1);
208 defer alloc.free(payload);
209 const composed = try proto.composeDelta(alloc, payload);
210 defer alloc.free(composed.bytes);
211 try std.testing.expect(std.mem.indexOf(u8, composed.bytes, "alt content") != null);
212 }
213
214 test "DeltaTracker: a resize behind the tracker's back resyncs instead of over-reading" {
215 const alloc = std.testing.allocator;
216
217 const eng = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });
218 defer eng.deinit();
219
220 var tracker: DeltaTracker = .{};
221 defer tracker.deinit(alloc);
222 try tracker.rebuild(alloc, eng, 24, 80);
223
224 // Stands in for a rebuild that failed (OOM) after the engine resized:
225 // the tracker still describes 24 rows of an 80-column grid.
226 try eng.resize(80, 10);
227 switch (try tracker.update(alloc, eng)) {
228 .discontinuity => {},
229 else => return error.ExpectedDiscontinuity,
230 }
231 }
src/server.zig
Old New
@@ -9,6 +9,8 @@ const std = @import("std");
9 const Engine = @import("engine").Engine; 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;
13 const sockpath = @import("sockpath");
12 const quic = @import("quic"); 14 const quic = @import("quic");
13 const quic_server = @import("quic_server"); 15 const quic_server = @import("quic_server");
14 const xdg = @import("xdg"); 16 const xdg = @import("xdg");
@@ -75,8 +77,6 @@ fn boundUdpPort(l: *quic_server.Listener) u16 {
75 return std.net.Address.initPosix(@ptrCast(@alignCast(&actual))).getPort(); 77 return std.net.Address.initPosix(@ptrCast(@alignCast(&actual))).getPort();
76 } 78 }
77 79
78 const Wyhash = std.hash.Wyhash;
79
80 const Stats = struct { 80 const Stats = struct {
81 snapshots: u64 = 0, 81 snapshots: u64 = 0,
82 snapshot_bytes: u64 = 0, 82 snapshot_bytes: u64 = 0,
@@ -87,154 +87,6 @@ const Stats = struct {
87 snapshot_equiv_bytes: u64 = 0, 87 snapshot_equiv_bytes: u64 = 0,
88 }; 88 };
89 89
90 /// Tracks a content hash per viewport row so an engine update can be sent
91 /// as just the rows that actually changed. Advances whether or not a
92 /// client is attached, so a reattach can be answered with a delta.
93 const DeltaTracker = struct {
94 seq: u64 = 0,
95 /// Seq at the last discontinuity (init/resize/screen switch). Clients
96 /// with have_seq older than this cannot be served a delta.
97 reset_seq: u64 = 0,
98 cols: u16 = 0,
99 rows: u16 = 0,
100 on_alt: bool = false,
101 cursor: Engine.CursorPos = .{ .x = 0, .y = 0 },
102 history_rows: u32 = 0,
103 row_hashes: []u64 = &.{},
104 row_seqs: []u64 = &.{},
105 /// Where update() writes the hashes it is computing, so a diff pass
106 /// allocates nothing. Swapped with row_hashes once per advance.
107 scratch_hashes: []u64 = &.{},
108
109 fn deinit(self: *DeltaTracker, alloc: std.mem.Allocator) void {
110 alloc.free(self.row_hashes);
111 alloc.free(self.row_seqs);
112 alloc.free(self.scratch_hashes);
113 }
114
115 /// Re-hash every row and mark a discontinuity: what follows can only be
116 /// carried by a full snapshot.
117 fn rebuild(self: *DeltaTracker, alloc: std.mem.Allocator, eng: *Engine, rows: u16, cols: u16) !void {
118 if (self.row_hashes.len != rows) {
119 // Every allocation lands before any old array is released, so a
120 // failure here leaves the tracker exactly as it was rather than
121 // half-freed. update()'s geometry check turns the resulting
122 // stale-but-consistent state into a snapshot on the next pump.
123 const hashes = try alloc.alloc(u64, rows);
124 errdefer alloc.free(hashes);
125 const seqs = try alloc.alloc(u64, rows);
126 errdefer alloc.free(seqs);
127 const scratch = try alloc.alloc(u64, rows);
128 alloc.free(self.row_hashes);
129 alloc.free(self.row_seqs);
130 alloc.free(self.scratch_hashes);
131 self.row_hashes = hashes;
132 self.row_seqs = seqs;
133 self.scratch_hashes = scratch;
134 }
135 self.cols = cols;
136 self.on_alt = eng.onAltScreen();
137 self.seq += 1;
138 self.reset_seq = self.seq;
139 self.cursor = eng.cursorPos();
140 self.history_rows = eng.historyRows();
141 // Claim no rows until every row is stamped. A dump that fails
142 // partway would otherwise leave stale seqs in the tail of
143 // row_seqs, and any of them above a client's have_seq would put
144 // that row in every delta from here on. rows == 0 sends update()
145 // down its first-run branch and sendResync to a snapshot instead.
146 self.rows = 0;
147 for (0..rows) |y| {
148 const bytes = try eng.dumpVtRow(alloc, @intCast(y));
149 defer alloc.free(bytes);
150 self.row_hashes[y] = Wyhash.hash(0, bytes);
151 self.row_seqs[y] = self.seq;
152 }
153 self.rows = rows;
154 }
155
156 const Update = union(enum) {
157 none,
158 discontinuity,
159 /// Something changed and the tracker now describes it; the payload
160 /// is built separately, and only if a client is there to read it.
161 advanced,
162 };
163
164 /// Diff current engine state against the tracked state. Advances seq
165 /// and tracked rows when anything changed. Allocates only the per-row
166 /// dumps it hashes.
167 fn update(self: *DeltaTracker, alloc: std.mem.Allocator, eng: *Engine) !Update {
168 if (self.rows == 0) return .discontinuity; // never built
169 // Geometry is load-bearing, not decorative: row_hashes is indexed
170 // by the tracker's own row count, and dumpVtRow asserts against the
171 // engine's. A tracker left stale by a failed rebuild resyncs here
172 // instead of running off the end of the grid.
173 if (self.rows != eng.term.rows or self.cols != eng.term.cols) return .discontinuity;
174 if (eng.onAltScreen() != self.on_alt) return .discontinuity;
175
176 // Safe to stamp row_seqs with the seq we may not end up taking: it
177 // is only written for rows whose hash changed, and any changed row
178 // forces the advance below.
179 const next_seq = self.seq + 1;
180 var changed: usize = 0;
181 for (0..self.rows) |y| {
182 const bytes = try eng.dumpVtRow(alloc, @intCast(y));
183 defer alloc.free(bytes);
184 const hash = Wyhash.hash(0, bytes);
185 self.scratch_hashes[y] = hash;
186 if (hash != self.row_hashes[y]) {
187 self.row_seqs[y] = next_seq;
188 changed += 1;
189 }
190 }
191
192 const cur = eng.cursorPos();
193 const hist = eng.historyRows();
194 const cursor_moved = cur.x != self.cursor.x or cur.y != self.cursor.y;
195 if (changed == 0 and !cursor_moved and hist == self.history_rows)
196 return .none;
197
198 self.seq = next_seq;
199 self.cursor = cur;
200 self.history_rows = hist;
201 std.mem.swap([]u64, &self.row_hashes, &self.scratch_hashes);
202 return .advanced;
203 }
204
205 /// Build a delta payload of all rows changed after `since`. The header
206 /// row_count and the appended rows MUST agree (composeDelta validates),
207 /// so both come from the same `row_seq > since` predicate over row_seqs
208 /// with nothing mutating in between.
209 ///
210 /// Changed rows get dumped twice per update — once to hash, once here
211 /// to serialize. That is the price of one row-selection routine serving
212 /// both the live stream and the attach path, and it is 1-3 rows in
213 /// steady state.
214 fn buildDeltaSince(self: *DeltaTracker, alloc: std.mem.Allocator, eng: *Engine, since: u64) ![]u8 {
215 var rows_changed: u16 = 0;
216 for (self.row_seqs) |s| {
217 if (s > since) rows_changed += 1;
218 }
219 var payload: std.ArrayList(u8) = .empty;
220 errdefer payload.deinit(alloc);
221 try proto.appendDeltaHeader(&payload, alloc, .{
222 .seq = self.seq,
223 .history_rows = self.history_rows,
224 .cursor_x = self.cursor.x,
225 .cursor_y = self.cursor.y,
226 .row_count = rows_changed,
227 });
228 for (self.row_seqs, 0..) |s, y| {
229 if (s <= since) continue;
230 const bytes = try eng.dumpVtRow(alloc, @intCast(y));
231 defer alloc.free(bytes);
232 try proto.appendDeltaRow(&payload, alloc, @intCast(y), bytes);
233 }
234 return payload.toOwnedSlice(alloc);
235 }
236 };
237
238 var shutdown_flag = std.atomic.Value(bool).init(false); 90 var shutdown_flag = std.atomic.Value(bool).init(false);
239 91
240 fn onShutdownSignal(_: c_int) callconv(.c) void { 92 fn onShutdownSignal(_: c_int) callconv(.c) void {
@@ -365,18 +217,9 @@ pub const Server = struct {
365 pty: Pty, 217 pty: Pty,
366 listener: std.net.Server, 218 listener: std.net.Server,
367 sock_path: []const u8, 219 sock_path: []const u8,
368 /// The socket file's identity at the moment it was bound, so teardown 220 /// What the socket file was when we bound it, so teardown can tell our
369 /// can tell our socket from one that replaced it. 221 /// socket from one that replaced it. See sockpath.PathId.
370 /// 222 path_id: ?sockpath.PathId,
371 /// The PATH's dev+ino, deliberately, not the listening descriptor's: a
372 /// bound unix socket's descriptor lives in sockfs (dev 10 here) while
373 /// the path resolves to an ordinary filesystem inode (dev 38), so
374 /// comparing the two could never be equal. The guard read as careful
375 /// and was unconditionally false, which meant the daemon never unlinked
376 /// its socket on a clean exit at all — masked ever since by the stale
377 /// socket recovery in `claimSockPath` cleaning up on the next start.
378 sock_dev: u64,
379 sock_ino: u64,
380 /// The attached interactive clients. All of them see every update. 223 /// The attached interactive clients. All of them see every update.
381 clients: [max_clients]?ClientSlot = @splat(null), 224 clients: [max_clients]?ClientSlot = @splat(null),
382 /// How much unsent output one client may accumulate before the daemon 225 /// How much unsent output one client may accumulate before the daemon
@@ -430,7 +273,7 @@ pub const Server = struct {
430 pub fn init(alloc: std.mem.Allocator, opts: Options) !Server { 273 pub fn init(alloc: std.mem.Allocator, opts: Options) !Server {
431 // Before the shell is spawned, so refusing costs nobody a fork and 274 // Before the shell is spawned, so refusing costs nobody a fork and
432 // leaves no process to reap. 275 // leaves no process to reap.
433 try claimSockPath(opts.sock_path); 276 try sockpath.claim(opts.sock_path);
434 277
435 const eng = try Engine.init(alloc, .{ .cols = opts.cols, .rows = opts.rows }); 278 const eng = try Engine.init(alloc, .{ .cols = opts.cols, .rows = opts.rows });
436 errdefer eng.deinit(); 279 errdefer eng.deinit();
@@ -446,67 +289,18 @@ pub const Server = struct {
446 289
447 const addr = try std.net.Address.initUnix(opts.sock_path); 290 const addr = try std.net.Address.initUnix(opts.sock_path);
448 const listener = try addr.listen(.{}); 291 const listener = try addr.listen(.{});
449 // Stat the path we just created, not the descriptor: this is the 292 const path_id = try sockpath.PathId.of(opts.sock_path);
450 // record teardown compares against, and it has to be taken through
451 // the same lens it will be re-read through.
452 const st = try std.posix.fstatat(std.posix.AT.FDCWD, opts.sock_path, 0);
453 return .{ 293 return .{
454 .alloc = alloc, 294 .alloc = alloc,
455 .eng = eng, 295 .eng = eng,
456 .pty = pty, 296 .pty = pty,
457 .listener = listener, 297 .listener = listener,
458 .sock_path = opts.sock_path, 298 .sock_path = opts.sock_path,
459 .sock_dev = @intCast(st.dev), 299 .path_id = path_id,
460 .sock_ino = @intCast(st.ino),
461 .epoch = epoch, 300 .epoch = epoch,
462 }; 301 };
463 } 302 }
464 303
465 /// Make the socket path ours to bind, or refuse it. Field incident this
466 /// exists for: three daemons were started against one path, each
467 /// unlinking it and binding fresh. Every one of them kept running with
468 /// its sessions intact, but only the newest was reachable — the older
469 /// two were stranded, invisible, holding shells nobody could get back
470 /// to, and two terminals "in the same session" were really in two
471 /// different ones.
472 ///
473 /// So: unlink only what answers ECONNREFUSED *and* is a socket.
474 /// - something answers → a live daemon owns this path. Refuse.
475 /// - nothing there → bind, nothing to clean up.
476 /// - a dead socket file → a daemon that died without deinit's
477 /// unlink running. Ours to clear.
478 /// - anything else → propagate. A path we cannot positively
479 /// identify as a dead daemon's leftover is not
480 /// something we may delete.
481 fn claimSockPath(path: []const u8) !void {
482 if (std.net.connectUnixSocket(path)) |probe| {
483 probe.close();
484 return error.DaemonAlreadyRunning;
485 } else |err| switch (err) {
486 error.FileNotFound => return, // free path; bind straight away
487 // Nobody is listening — but this is NOT yet proof of a stale
488 // socket: Linux answers ECONNREFUSED for a regular file at the
489 // path exactly as it does for a dead socket, so connect alone
490 // cannot tell a dead daemon from `muxd run --sock notes.txt`.
491 // The stat below is what separates them.
492 error.ConnectionRefused => {},
493 else => |e| return e,
494 }
495
496 const st = std.posix.fstatat(std.posix.AT.FDCWD, path, 0) catch |err| switch (err) {
497 error.FileNotFound => return, // vanished under us; path is free
498 else => |e| return e,
499 };
500 if (!std.posix.S.ISSOCK(st.mode)) return error.SockPathNotASocket;
501
502 std.fs.cwd().deleteFile(path) catch |err| switch (err) {
503 // Someone else cleared it first. The path is free either way,
504 // which is the only thing this function was after.
505 error.FileNotFound => {},
506 else => |e| return e,
507 };
508 }
509
510 pub fn deinit(self: *Server) void { 304 pub fn deinit(self: *Server) void {
511 for (&self.clients) |*slot| { 305 for (&self.clients) |*slot| {
512 if (slot.*) |*c| { 306 if (slot.*) |*c| {
@@ -537,10 +331,7 @@ pub const Server = struct {
537 // newer daemon may have replaced the file since we bound it, and 331 // newer daemon may have replaced the file since we bound it, and
538 // deleting that one would hand its clients the same field incident 332 // deleting that one would hand its clients the same field incident
539 // the socket-steal fix exists to prevent. 333 // the socket-steal fix exists to prevent.
540 const ours: bool = ours: { 334 const ours: bool = if (self.path_id) |id| id.stillAt(self.sock_path) else false;
541 const pst = std.posix.fstatat(std.posix.AT.FDCWD, self.sock_path, 0) catch break :ours false;
542 break :ours pst.dev == self.sock_dev and pst.ino == self.sock_ino;
543 };
544 self.listener.deinit(); 335 self.listener.deinit();
545 if (ours) std.fs.cwd().deleteFile(self.sock_path) catch {}; 336 if (ours) std.fs.cwd().deleteFile(self.sock_path) catch {};
546 self.tracker.deinit(self.alloc); 337 self.tracker.deinit(self.alloc);
@@ -1628,10 +1419,7 @@ pub const Server = struct {
1628 self.resyncSnapshot(); 1419 self.resyncSnapshot();
1629 return; 1420 return;
1630 } 1421 }
1631 if (have_epoch == self.epoch and have_seq != 0 and 1422 if (have_epoch == self.epoch and self.tracker.canServe(have_seq)) {
1632 have_seq >= self.tracker.reset_seq and have_seq <= self.tracker.seq and
1633 self.tracker.rows != 0)
1634 {
1635 const payload = self.tracker.buildDeltaSince(self.alloc, self.eng, have_seq) catch { 1423 const payload = self.tracker.buildDeltaSince(self.alloc, self.eng, have_seq) catch {
1636 self.snapshotTo(i); 1424 self.snapshotTo(i);
1637 return; 1425 return;
@@ -3822,62 +3610,6 @@ fn firstStateFrame(alloc: std.mem.Allocator, fd: std.posix.fd_t, timeout_ms: u64
3822 return null; 3610 return null;
3823 } 3611 }
3824 3612
3825 test "DeltaTracker: alt-screen flip is a discontinuity and rows follow the active screen" {
3826 const alloc = std.testing.allocator;
3827
3828 const eng = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });
3829 defer eng.deinit();
3830
3831 var tracker: DeltaTracker = .{};
3832 defer tracker.deinit(alloc);
3833 try tracker.rebuild(alloc, eng, 24, 80);
3834
3835 eng.feed("primary text");
3836 switch (try tracker.update(alloc, eng)) {
3837 .advanced => {},
3838 else => return error.ExpectedAdvance,
3839 }
3840
3841 // Switching screens replaces every row at once: the tracked hashes
3842 // describe the other screen, so a delta would be a lie.
3843 eng.feed("\x1b[?1049h");
3844 switch (try tracker.update(alloc, eng)) {
3845 .discontinuity => {},
3846 else => return error.ExpectedDiscontinuity,
3847 }
3848
3849 try tracker.rebuild(alloc, eng, 24, 80);
3850 eng.feed("alt content");
3851 switch (try tracker.update(alloc, eng)) {
3852 .advanced => {},
3853 else => return error.ExpectedAdvance,
3854 }
3855 const payload = try tracker.buildDeltaSince(alloc, eng, tracker.seq - 1);
3856 defer alloc.free(payload);
3857 const composed = try proto.composeDelta(alloc, payload);
3858 defer alloc.free(composed.bytes);
3859 try std.testing.expect(std.mem.indexOf(u8, composed.bytes, "alt content") != null);
3860 }
3861
3862 test "DeltaTracker: a resize behind the tracker's back resyncs instead of over-reading" {
3863 const alloc = std.testing.allocator;
3864
3865 const eng = try Engine.init(alloc, .{ .cols = 80, .rows = 24 });
3866 defer eng.deinit();
3867
3868 var tracker: DeltaTracker = .{};
3869 defer tracker.deinit(alloc);
3870 try tracker.rebuild(alloc, eng, 24, 80);
3871
3872 // Stands in for a rebuild that failed (OOM) after the engine resized:
3873 // the tracker still describes 24 rows of an 80-column grid.
3874 try eng.resize(80, 10);
3875 switch (try tracker.update(alloc, eng)) {
3876 .discontinuity => {},
3877 else => return error.ExpectedDiscontinuity,
3878 }
3879 }
3880
3881 // Placed BEFORE the QUIC integration tests below, and that is not 3613 // Placed BEFORE the QUIC integration tests below, and that is not
3882 // cosmetic. Widening deinit's `.owned` arm to free a `.borrowed` listener 3614 // cosmetic. Widening deinit's `.owned` arm to free a `.borrowed` listener
3883 // too makes the daemon free one it does not own; every test down there hands 3615 // too makes the daemon free one it does not own; every test down there hands
src/sockpath.zig
Old New
@@ -0,0 +1,86 @@
1 //! The socket path's identity and the right to bind it: who owns a path
2 //! before a daemon starts, and whether the file sitting there at teardown
3 //! is still the one that daemon created.
4 //!
5 //! Both halves answer the same field incident from opposite ends — one
6 //! refuses to steal a live daemon's path, the other refuses to delete a
7 //! successor's socket — so they belong together and nowhere near the rest
8 //! of the daemon. Nothing here knows a Server exists; a path is all it
9 //! takes.
10 const std = @import("std");
11
12 /// A socket file's identity at the moment it was bound, so teardown can
13 /// tell our socket from one that replaced it.
14 ///
15 /// The PATH's dev+ino, deliberately, not the listening descriptor's: a
16 /// bound unix socket's descriptor lives in sockfs (dev 10 here) while
17 /// the path resolves to an ordinary filesystem inode (dev 38), so
18 /// comparing the two could never be equal. The guard read as careful
19 /// and was unconditionally false, which meant the daemon never unlinked
20 /// its socket on a clean exit at all — masked ever since by the stale
21 /// socket recovery in `claim` cleaning up on the next start.
22 pub const PathId = struct {
23 dev: u64,
24 ino: u64,
25
26 /// Stat the path we just created, not the descriptor: this is the
27 /// record teardown compares against, and it has to be taken through
28 /// the same lens it will be re-read through.
29 pub fn of(path: []const u8) !PathId {
30 const st = try std.posix.fstatat(std.posix.AT.FDCWD, path, 0);
31 return .{ .dev = @intCast(st.dev), .ino = @intCast(st.ino) };
32 }
33
34 /// Does `path` still name this exact file? A path that cannot be
35 /// stat'd is not ours — it is gone, or it is something we may not
36 /// identify — and either way the answer callers want is "leave it".
37 pub fn stillAt(self: PathId, path: []const u8) bool {
38 const pst = std.posix.fstatat(std.posix.AT.FDCWD, path, 0) catch return false;
39 return pst.dev == self.dev and pst.ino == self.ino;
40 }
41 };
42
43 /// Make the socket path ours to bind, or refuse it. Field incident this
44 /// exists for: three daemons were started against one path, each
45 /// unlinking it and binding fresh. Every one of them kept running with
46 /// its sessions intact, but only the newest was reachable — the older
47 /// two were stranded, invisible, holding shells nobody could get back
48 /// to, and two terminals "in the same session" were really in two
49 /// different ones.
50 ///
51 /// So: unlink only what answers ECONNREFUSED *and* is a socket.
52 /// - something answers → a live daemon owns this path. Refuse.
53 /// - nothing there → bind, nothing to clean up.
54 /// - a dead socket file → a daemon that died without deinit's
55 /// unlink running. Ours to clear.
56 /// - anything else → propagate. A path we cannot positively
57 /// identify as a dead daemon's leftover is not
58 /// something we may delete.
59 pub fn claim(path: []const u8) !void {
60 if (std.net.connectUnixSocket(path)) |probe| {
61 probe.close();
62 return error.DaemonAlreadyRunning;
63 } else |err| switch (err) {
64 error.FileNotFound => return, // free path; bind straight away
65 // Nobody is listening — but this is NOT yet proof of a stale
66 // socket: Linux answers ECONNREFUSED for a regular file at the
67 // path exactly as it does for a dead socket, so connect alone
68 // cannot tell a dead daemon from `muxd run --sock notes.txt`.
69 // The stat below is what separates them.
70 error.ConnectionRefused => {},
71 else => |e| return e,
72 }
73
74 const st = std.posix.fstatat(std.posix.AT.FDCWD, path, 0) catch |err| switch (err) {
75 error.FileNotFound => return, // vanished under us; path is free
76 else => |e| return e,
77 };
78 if (!std.posix.S.ISSOCK(st.mode)) return error.SockPathNotASocket;
79
80 std.fs.cwd().deleteFile(path) catch |err| switch (err) {
81 // Someone else cleared it first. The path is free either way,
82 // which is the only thing this function was after.
83 error.FileNotFound => {},
84 else => |e| return e,
85 };
86 }