Eric Bower
·
2026-08-11
1const std = @import("std");
2const cross = @import("cross.zig");
3const socket = @import("socket.zig");
4const lib_posix = @import("posix.zig");
5
6pub const Tag = enum(u8) {
7 Input = 0,
8 Output = 1,
9 Resize = 2,
10 Detach = 3,
11 DetachAll = 4,
12 Kill = 5,
13 Info = 6,
14 Init = 7,
15 History = 8,
16 Run = 9,
17 Ack = 10,
18 Switch = 11,
19 Write = 12,
20 TaskComplete = 13,
21 LabelGet = 14,
22 LabelSet = 15,
23 LabelClear = 16,
24 LabelData = 17,
25 Send = 18,
26 // Non-exhaustive: this enum comes off the wire via bytesToValue and
27 // @enumFromInt, so out-of-range values are representable
28 // rather than UB. Switches must handle `_` (unknown tag).
29 _,
30};
31
32comptime {
33 if (@typeInfo(Tag).@"enum".is_exhaustive) @compileError(
34 "ipc.Tag must stay non-exhaustive -- old daemons rely on `_` to ignore unknown tags",
35 );
36}
37
38pub const Header = packed struct {
39 tag: Tag,
40 len: u32,
41};
42
43pub const Resize = packed struct {
44 rows: u16,
45 cols: u16,
46 xpixel: u16 = 0,
47 ypixel: u16 = 0,
48};
49
50pub fn getTerminalSize(fd: i32) Resize {
51 var ws: cross.c.struct_winsize = undefined;
52 if (cross.c.ioctl(fd, cross.c.TIOCGWINSZ, &ws) == 0 and ws.ws_row > 0 and ws.ws_col > 0) {
53 return .{ .rows = ws.ws_row, .cols = ws.ws_col, .xpixel = ws.ws_xpixel, .ypixel = ws.ws_ypixel };
54 }
55 inline for (.{ lib_posix.STDOUT_FILENO, lib_posix.STDIN_FILENO, lib_posix.STDERR_FILENO }) |fallback_fd| {
56 if (fallback_fd != fd) {
57 if (cross.c.ioctl(fallback_fd, cross.c.TIOCGWINSZ, &ws) == 0 and ws.ws_row > 0 and ws.ws_col > 0) {
58 return .{ .rows = ws.ws_row, .cols = ws.ws_col, .xpixel = ws.ws_xpixel, .ypixel = ws.ws_ypixel };
59 }
60 }
61 }
62 if (lib_posix.open("/dev/tty", .{ .ACCMODE = .RDWR }, 0)) |tty_fd| {
63 defer lib_posix.close(tty_fd);
64 if (cross.c.ioctl(tty_fd, cross.c.TIOCGWINSZ, &ws) == 0 and ws.ws_row > 0 and ws.ws_col > 0) {
65 return .{ .rows = ws.ws_row, .cols = ws.ws_col, .xpixel = ws.ws_xpixel, .ypixel = ws.ws_ypixel };
66 }
67 } else |_| {}
68 return .{ .rows = 24, .cols = 120 };
69}
70
71pub const MAX_CMD_LEN = 256;
72pub const MAX_CWD_LEN = 256;
73
74/// Frozen wire shape. Do NOT add fields! New stats go in new `Tag` values
75/// so old daemons (whose `_` arm ignores unknown tags) stay reachable.
76/// Changing `@sizeOf(Info)` breaks `zmx list` against running daemons.
77pub const Info = extern struct {
78 clients_len: u64,
79 pid: i32,
80 cmd_len: u16,
81 cwd_len: u16,
82 cmd: [MAX_CMD_LEN]u8,
83 cwd: [MAX_CWD_LEN]u8,
84 created_at: u64,
85 task_ended_at: u64,
86 task_exit_code: u8,
87};
88
89pub fn expectedLength(data: []const u8) ?usize {
90 if (data.len < @sizeOf(Header)) return null;
91 const header = std.mem.bytesToValue(Header, data[0..@sizeOf(Header)]);
92 // header.len comes off the wire; widen to usize before adding so a
93 // near-u32-max value can't wrap (panic in safe mode, UB in release).
94 return @as(usize, @sizeOf(Header)) + @as(usize, header.len);
95}
96
97pub fn send(fd: i32, tag: Tag, data: []const u8) !void {
98 const header = Header{
99 .tag = tag,
100 .len = @intCast(data.len),
101 };
102 const header_bytes = std.mem.asBytes(&header);
103 try writeAll(fd, header_bytes);
104 if (data.len > 0) {
105 try writeAll(fd, data);
106 }
107}
108
109pub fn appendMessage(
110 gpa: std.mem.Allocator,
111 list: *std.ArrayList(u8),
112 tag: Tag,
113 data: []const u8,
114) !void {
115 const header = Header{
116 .tag = tag,
117 .len = @intCast(data.len),
118 };
119 // Guarantee capacity for header + payload in one check to avoid
120 // intermediate realloc between the two appends on the hot path.
121 try list.ensureTotalCapacity(gpa, list.items.len + @sizeOf(Header) + data.len);
122 list.appendSliceAssumeCapacity(std.mem.asBytes(&header));
123 if (data.len > 0) {
124 list.appendSliceAssumeCapacity(data);
125 }
126}
127
128fn writeAll(fd: i32, data: []const u8) !void {
129 var index: usize = 0;
130 while (index < data.len) {
131 const n = try lib_posix.write(fd, data[index..]);
132 if (n == 0) return error.DiskQuota;
133 index += n;
134 }
135}
136
137pub const Message = struct {
138 tag: Tag,
139 data: []u8,
140
141 pub fn deinit(self: Message, alloc: std.mem.Allocator) void {
142 if (self.data.len > 0) {
143 alloc.free(self.data);
144 }
145 }
146};
147
148pub const SocketMsg = struct {
149 header: Header,
150 payload: []const u8,
151};
152
153pub const SocketBuffer = struct {
154 buf: std.ArrayList(u8),
155 alloc: std.mem.Allocator,
156 head: usize,
157
158 pub fn init(alloc: std.mem.Allocator) !SocketBuffer {
159 return .{
160 .buf = try std.ArrayList(u8).initCapacity(alloc, 4096),
161 .alloc = alloc,
162 .head = 0,
163 };
164 }
165
166 pub fn deinit(self: *SocketBuffer) void {
167 self.buf.deinit(self.alloc);
168 }
169
170 /// Reads from fd into buffer.
171 /// Returns number of bytes read.
172 /// Propagates error.WouldBlock and other errors to caller.
173 /// Returns 0 on EOF.
174 pub fn read(self: *SocketBuffer, fd: i32) !usize {
175 if (self.head > 0) {
176 const remaining = self.buf.items.len - self.head;
177 if (remaining > 0) {
178 std.mem.copyForwards(u8, self.buf.items[0..remaining], self.buf.items[self.head..]);
179 self.buf.items.len = remaining;
180 } else {
181 self.buf.clearRetainingCapacity();
182 }
183 self.head = 0;
184 }
185
186 var tmp: [4096]u8 = undefined;
187 const n = try lib_posix.read(fd, &tmp);
188 if (n > 0) {
189 try self.buf.appendSlice(self.alloc, tmp[0..n]);
190 }
191 return n;
192 }
193
194 /// Returns the next complete message or `null` when none available.
195 /// `buf` is advanced automatically; caller keeps the returned slices
196 /// valid until the following `next()` (or `deinit`).
197 pub fn next(self: *SocketBuffer) ?SocketMsg {
198 const available = self.buf.items[self.head..];
199 const total = expectedLength(available) orelse return null;
200 if (available.len < total) return null;
201
202 const hdr = std.mem.bytesToValue(Header, available[0..@sizeOf(Header)]);
203 const pay = available[@sizeOf(Header)..total];
204
205 self.head += total;
206 return .{ .header = hdr, .payload = pay };
207 }
208};
209
210const ConnectError = error{
211 ConnectionRefused,
212 Unexpected,
213};
214
215/// Connect-only liveness check. Callers that don't read `Info` should use
216/// this (not `probeSession`) so they survive `Info` shape changes.
217pub fn connectSession(socket_path: []const u8) ConnectError!i32 {
218 return socket.sessionConnect(socket_path) catch |err| switch (err) {
219 error.ConnectionRefused => return error.ConnectionRefused,
220 else => return error.Unexpected,
221 };
222}
223
224const SessionProbeError = error{
225 Timeout,
226 ConnectionRefused,
227 Unexpected,
228 InfoSizeMismatch,
229};
230
231const SessionProbeResult = struct {
232 fd: i32,
233 info: Info,
234 labels: ?[]const u8,
235 alloc: std.mem.Allocator,
236
237 pub fn deinit(self: *const SessionProbeResult) void {
238 if (self.labels) |lbl| self.alloc.free(lbl);
239 lib_posix.close(self.fd);
240 }
241};
242
243pub fn probeSession(
244 alloc: std.mem.Allocator,
245 socket_path: []const u8,
246) SessionProbeError!SessionProbeResult {
247 const timeout_ms = 1000;
248 const fd = try connectSession(socket_path);
249 errdefer lib_posix.close(fd);
250
251 send(fd, .Info, "") catch return error.Unexpected;
252 send(fd, .LabelGet, "") catch {};
253
254 var poll_fds = [_]lib_posix.pollfd{.{ .fd = fd, .events = lib_posix.POLL.IN, .revents = 0 }};
255 const poll_result = lib_posix.poll(&poll_fds, timeout_ms) catch return error.Unexpected;
256 if (poll_result == 0) {
257 return error.Timeout;
258 }
259
260 var sb = SocketBuffer.init(alloc) catch return error.Unexpected;
261 defer sb.deinit();
262
263 const n = sb.read(fd) catch return error.Unexpected;
264 if (n == 0) return error.Unexpected;
265
266 var info_result: ?Info = null;
267 var labels: ?[]const u8 = null;
268 errdefer if (labels) |lbl| alloc.free(lbl);
269
270 while (true) {
271 if (sb.next()) |msg| {
272 if (msg.header.tag == .Info) {
273 if (msg.payload.len != @sizeOf(Info)) return error.InfoSizeMismatch;
274 info_result = std.mem.bytesToValue(Info, msg.payload[0..@sizeOf(Info)]);
275 }
276 if (msg.header.tag == .LabelData) {
277 labels = alloc.dupe(u8, msg.payload) catch null;
278 }
279
280 if (info_result != null and labels != null) break;
281 continue;
282 }
283
284 // No complete message available, wait for more data
285 const more = lib_posix.poll(&poll_fds, 50) catch break;
286 if (more == 0) break;
287 const n_read = sb.read(fd) catch break;
288 if (n_read == 0) break;
289 }
290
291 if (info_result) |info| {
292 return .{
293 .fd = fd,
294 .info = info,
295 .labels = labels,
296 .alloc = alloc,
297 };
298 }
299 return error.Unexpected;
300}
301
302// WIRE PROTOCOL FREEZE: read before "fixing" any test below.
303//
304// Changing these constants does not fix the test; it breaks every
305// running daemon for every user until they `pkill -f zmx`.
306//
307// Need a new field? → add a new `Tag` value (next free integer).
308// Need to remove one? → don't. Reserve the integer, stop sending it.
309test "Info wire size is frozen" {
310 try std.testing.expectEqual(@as(usize, 552), @sizeOf(Info));
311 // packed struct{u8,u32} backs to u40 → @sizeOf rounds to 8, not 5.
312 try std.testing.expectEqual(@as(usize, 8), @sizeOf(Header));
313}
314
315test "Tag wire values are frozen" {
316 inline for (.{
317 .{ Tag.Input, 0 }, .{ Tag.Output, 1 }, .{ Tag.Resize, 2 },
318 .{ Tag.Detach, 3 }, .{ Tag.DetachAll, 4 }, .{ Tag.Kill, 5 },
319 .{ Tag.Info, 6 }, .{ Tag.Init, 7 }, .{ Tag.History, 8 },
320 .{ Tag.Run, 9 }, .{ Tag.Ack, 10 }, .{ Tag.Switch, 11 },
321 .{ Tag.Write, 12 }, .{ Tag.TaskComplete, 13 }, .{ Tag.LabelGet, 14 },
322 .{ Tag.LabelSet, 15 }, .{ Tag.LabelClear, 16 }, .{ Tag.LabelData, 17 },
323 .{ Tag.Send, 18 },
324 }) |p| try std.testing.expectEqual(@as(u8, p[1]), @intFromEnum(p[0]));
325}
326
327pub fn roundTripForTag(
328 alloc: std.mem.Allocator,
329 socket_path: []const u8,
330 request_tag: Tag,
331 payload: []const u8,
332 expected_tag: Tag,
333) SessionProbeError![]u8 {
334 const timeout_ms = 1000;
335 const fd = try connectSession(socket_path);
336 defer lib_posix.close(fd);
337
338 send(fd, request_tag, payload) catch return error.Unexpected;
339
340 var poll_fds = [_]lib_posix.pollfd{.{ .fd = fd, .events = lib_posix.POLL.IN, .revents = 0 }};
341 const poll_result = lib_posix.poll(&poll_fds, timeout_ms) catch return error.Unexpected;
342 if (poll_result == 0) return error.Timeout;
343
344 var sb = SocketBuffer.init(alloc) catch return error.Unexpected;
345 defer sb.deinit();
346
347 const n = sb.read(fd) catch return error.Unexpected;
348 if (n == 0) return error.Unexpected;
349
350 while (sb.next()) |msg| {
351 if (msg.header.tag == expected_tag) {
352 return alloc.dupe(u8, msg.payload) catch return error.Unexpected;
353 }
354 }
355 return error.Unexpected;
356}
357
358test "zeroed Info has no stack garbage in wire bytes" {
359 var info = std.mem.zeroes(Info);
360 info.clients_len = 3;
361 info.pid = 999;
362 info.task_exit_code = 7;
363 const bytes = std.mem.asBytes(&info);
364 // Tail padding after task_exit_code must be zero (asBytes ships it).
365 const last_field_end = @offsetOf(Info, "task_exit_code") + @sizeOf(u8);
366 for (bytes[last_field_end..]) |b| try std.testing.expectEqual(@as(u8, 0), b);
367}