termio, flatpak: implement process watcher with xev

This allows `termio.Exec` to track processes spawned via
`FlatpakHostCommand`, finally allowing Ghostty to function as a
Flatpak.

Alongside this is a few bug fixes:

* Don't add ghostty to PATH when running in flatpak mode since it's
  unreachable.
* Correctly handle exit status returned by Flatpak. Previously this was
  not processed and contains extra status bits.
* Use correct type for PID returned by Flatpak.

Author: alaviss

src/os/flatpak.zig

@@ -3,6 +3,7 @@ const assert = std.debug.assert;
 const Allocator = std.mem.Allocator;
 const builtin = @import("builtin");
 const posix = std.posix;
+const xev = @import("../global.zig").xev;
 
 const log = std.log.scoped(.flatpak);
 
@@ -71,18 +72,28 @@ pub const FlatpakHostCommand = struct {
 
         /// Process started with the given pid on the host.
         started: struct {
-            pid: c_int,
+            pid: u32,
+            loop_xev: ?*xev.Loop,
+            completion: ?*Completion,
             subscription: c.guint,
             loop: *c.GMainLoop,
         },
 
         /// Process exited
         exited: struct {
-            pid: c_int,
+            pid: u32,
             status: u8,
         },
     };
 
+    pub const Completion = struct {
+        callback: *const fn (ud: ?*anyopaque, l: *xev.Loop, c: *Completion, r: WaitError!u8) void = noopCallback,
+        c_xev: xev.Completion = .{},
+        userdata: ?*anyopaque = null,
+        timer: ?xev.Timer = null,
+        result: ?WaitError!u8 = null,
+    };
+
     /// Errors that are possible from us.
     pub const Error = error{
         FlatpakMustBeStarted,
@@ -91,12 +102,14 @@ pub const FlatpakHostCommand = struct {
         FlatpakRPCFail,
     };
 
+    pub const WaitError = xev.Timer.RunError || Error;
+
     /// Spawn the command. This will start the host command. On return,
     /// the pid will be available. This must only be called with the
     /// state in "init".
     ///
     /// Precondition: The self pointer MUST be stable.
-    pub fn spawn(self: *FlatpakHostCommand, alloc: Allocator) !c_int {
+    pub fn spawn(self: *FlatpakHostCommand, alloc: Allocator) !u32 {
         const thread = try std.Thread.spawn(.{}, threadMain, .{ self, alloc });
         thread.setName("flatpak-host-command") catch {};
 
@@ -135,6 +148,77 @@ pub const FlatpakHostCommand = struct {
         }
     }
 
+    /// Wait for the process to end asynchronously via libxev. This
+    /// can only be called ONCE.
+    pub fn waitXev(
+        self: *FlatpakHostCommand,
+        loop: *xev.Loop,
+        completion: *Completion,
+        comptime Userdata: type,
+        userdata: ?*Userdata,
+        comptime cb: *const fn (
+            ud: ?*Userdata,
+            l: *xev.Loop,
+            c: *Completion,
+            r: WaitError!u8,
+        ) void,
+    ) void {
+        self.state_mutex.lock();
+        defer self.state_mutex.unlock();
+
+        completion.* = .{
+            .callback = (struct {
+                fn callback(
+                    ud_: ?*anyopaque,
+                    l_inner: *xev.Loop,
+                    c_inner: *Completion,
+                    r: WaitError!u8,
+                ) void {
+                    const ud = @as(?*Userdata, if (Userdata == void) null else @ptrCast(@alignCast(ud_)));
+                    @call(.always_inline, cb, .{ ud, l_inner, c_inner, r });
+                }
+            }).callback,
+            .userdata = userdata,
+            .timer = xev.Timer.init() catch unreachable, // not great, but xev timer can't fail atm
+        };
+
+        switch (self.state) {
+            .init => completion.result = Error.FlatpakMustBeStarted,
+            .err => completion.result = Error.FlatpakSpawnFail,
+            .started => |*v| {
+                v.loop_xev = loop;
+                v.completion = completion;
+                return;
+            },
+            .exited => |v| {
+                completion.result = v.status;
+            },
+        }
+
+        completion.timer.?.run(
+            loop,
+            &completion.c_xev,
+            0,
+            anyopaque,
+            completion.userdata,
+            (struct {
+                fn callback(
+                    ud: ?*anyopaque,
+                    l_inner: *xev.Loop,
+                    c_inner: *xev.Completion,
+                    r: xev.Timer.RunError!void,
+                ) xev.CallbackAction {
+                    const c_outer: *Completion = @fieldParentPtr("c_xev", c_inner);
+                    defer if (c_outer.timer) |*t| t.deinit();
+
+                    const result = if (r) |_| c_outer.result.? else |err| err;
+                    c_outer.callback(ud, l_inner, c_outer, result);
+                    return .disarm;
+                }
+            }).callback,
+        );
+    }
+
     /// Send a signal to the started command. This does nothing if the
     /// command is not in the started state.
     pub fn signal(self: *FlatpakHostCommand, sig: u8, pg: bool) !void {
@@ -326,7 +410,7 @@ pub const FlatpakHostCommand = struct {
         };
         defer c.g_variant_unref(reply);
 
-        var pid: c_int = 0;
+        var pid: u32 = 0;
         c.g_variant_get(reply, "(u)", &pid);
         log.debug("HostCommand started pid={} subscription={}", .{
             pid,
@@ -338,6 +422,8 @@ pub const FlatpakHostCommand = struct {
                 .pid = pid,
                 .subscription = subscription_id,
                 .loop = loop,
+                .completion = null,
+                .loop_xev = null,
             },
         });
     }
@@ -366,18 +452,44 @@ pub const FlatpakHostCommand = struct {
             break :state self.state.started;
         };
 
-        var pid: c_int = 0;
-        var exit_status: c_int = 0;
-        c.g_variant_get(params.?, "(uu)", &pid, &exit_status);
+        var pid: u32 = 0;
+        var exit_status_raw: u32 = 0;
+        c.g_variant_get(params.?, "(uu)", &pid, &exit_status_raw);
         if (state.pid != pid) return;
 
+        const exit_status = posix.W.EXITSTATUS(exit_status_raw);
         // Update our state
         self.updateState(.{
             .exited = .{
                 .pid = pid,
-                .status = std.math.cast(u8, exit_status) orelse 255,
+                .status = exit_status,
             },
         });
+        if (state.completion) |completion| {
+            completion.result = exit_status;
+            completion.timer.?.run(
+                state.loop_xev.?,
+                &completion.c_xev,
+                0,
+                anyopaque,
+                completion.userdata,
+                (struct {
+                    fn callback(
+                        ud_inner: ?*anyopaque,
+                        l_inner: *xev.Loop,
+                        c_inner: *xev.Completion,
+                        r: xev.Timer.RunError!void,
+                    ) xev.CallbackAction {
+                        const c_outer: *Completion = @fieldParentPtr("c_xev", c_inner);
+                        defer if (c_outer.timer) |*t| t.deinit();
+
+                        const result = if (r) |_| c_outer.result.? else |err| err;
+                        c_outer.callback(ud_inner, l_inner, c_outer, result);
+                        return .disarm;
+                    }
+                }).callback,
+            );
+        }
         log.debug("HostCommand exited pid={} status={}", .{ pid, exit_status });
 
         // We're done now, so we can unsubscribe
@@ -386,4 +498,6 @@ pub const FlatpakHostCommand = struct {
         // We are also done with our loop so we can exit.
         c.g_main_loop_quit(state.loop);
     }
+
+    fn noopCallback(_: ?*anyopaque, _: *xev.Loop, _: *Completion, _: WaitError!u8) void {}
 };

src/termio/Exec.zig

@@ -95,11 +95,20 @@ pub fn threadEnter(
     };
     errdefer self.subprocess.stop();
 
-    // Get the pid from the subprocess
-    const pid = pid: {
-        const command = self.subprocess.command orelse return error.ProcessNotStarted;
-        break :pid command.pid orelse return error.ProcessNoPid;
+    // Watcher to detect subprocess exit
+    var process = process: {
+        // Get the pid from the subprocess
+        const pid = pid: {
+            if (self.subprocess.flatpak_command) |_| {
+                break :process null;
+            }
+            const command = self.subprocess.command orelse return error.ProcessNotStarted;
+            break :pid command.pid orelse return error.ProcessNoPid;
+        };
+
+        break :process try xev.Process.init(pid);
     };
+    errdefer if (process) |*p| p.deinit();
 
     // Track our process start time for abnormal exits
     const process_start = try std.time.Instant.now();
@@ -114,10 +123,6 @@ pub fn threadEnter(
     var stream = xev.Stream.initFd(pty_fds.write);
     errdefer stream.deinit();
 
-    // Watcher to detect subprocess exit
-    var process = try xev.Process.init(pid);
-    errdefer process.deinit();
-
     // Start our timer to read termios state changes. This is used
     // to detect things such as when password input is being done
     // so we can render the terminal in a different way.
@@ -146,13 +151,23 @@ pub fn threadEnter(
     } };
 
     // Start our process watcher
-    process.wait(
-        td.loop,
-        &td.backend.exec.process_wait_c,
-        termio.Termio.ThreadData,
-        td,
-        processExit,
-    );
+    if (self.subprocess.flatpak_command) |*c| {
+        c.waitXev(
+            td.loop,
+            &td.backend.exec.flatpak_wait_c,
+            termio.Termio.ThreadData,
+            td,
+            flatpakExit,
+        );
+    } else {
+        process.?.wait(
+            td.loop,
+            &td.backend.exec.process_wait_c,
+            termio.Termio.ThreadData,
+            td,
+            processExit,
+        );
+    }
 
     // Start our termios timer. We don't support this on Windows.
     // Fundamentally, we could support this on Windows so we're just
@@ -339,15 +354,7 @@ fn execFailedInChild() !void {
     _ = try reader.read(&buf);
 }
 
-fn processExit(
-    td_: ?*termio.Termio.ThreadData,
-    _: *xev.Loop,
-    _: *xev.Completion,
-    r: xev.Process.WaitError!u32,
-) xev.CallbackAction {
-    const exit_code = r catch unreachable;
-
-    const td = td_.?;
+fn processExitCommon(td: *termio.Termio.ThreadData, exit_code: u32) void {
     assert(td.backend == .exec);
     const execdata = &td.backend.exec;
     execdata.exited = true;
@@ -393,7 +400,7 @@ fn processExit(
         }, null);
         td.mailbox.notify();
 
-        return .disarm;
+        return;
     }
 
     // If we're purposely waiting then we just return since the process
@@ -413,17 +420,36 @@ fn processExit(
             t.modes.set(.cursor_visible, false);
         }
 
-        return .disarm;
+        return;
     }
 
     // Notify our surface we want to close
     _ = td.surface_mailbox.push(.{
         .child_exited = {},
     }, .{ .forever = {} });
+}
 
+fn processExit(
+    td_: ?*termio.Termio.ThreadData,
+    _: *xev.Loop,
+    _: *xev.Completion,
+    r: xev.Process.WaitError!u32,
+) xev.CallbackAction {
+    const exit_code = r catch unreachable;
+    processExitCommon(td_.?, exit_code);
     return .disarm;
 }
 
+fn flatpakExit(
+    td_: ?*termio.Termio.ThreadData,
+    _: *xev.Loop,
+    _: *internal_os.FlatpakHostCommand.Completion,
+    r: internal_os.FlatpakHostCommand.WaitError!u8,
+) void {
+    const exit_code = r catch unreachable;
+    processExitCommon(td_.?, exit_code);
+}
+
 fn termiosTimer(
     td_: ?*termio.Termio.ThreadData,
     _: *xev.Loop,
@@ -610,6 +636,7 @@ pub const ThreadData = struct {
     // The preallocation size for the write request pool. This should be big
     // enough to satisfy most write requests. It must be a power of 2.
     const WRITE_REQ_PREALLOC = std.math.pow(usize, 2, 5);
+    const FlatpakCompletion = if (Subprocess.FlatpakHostCommand != void) Subprocess.FlatpakHostCommand.Completion else void;
 
     /// Process start time and boolean of whether its already exited.
     start: std.time.Instant,
@@ -630,7 +657,7 @@ pub const ThreadData = struct {
     write_stream: xev.Stream,
 
     /// The process watcher
-    process: xev.Process,
+    process: ?xev.Process,
 
     /// This is the pool of available (unused) write requests. If you grab
     /// one from the pool, you must put it back when you're done!
@@ -646,6 +673,8 @@ pub const ThreadData = struct {
     /// subsequently to wait for the data_stream to close.
     process_wait_c: xev.Completion = .{},
 
+    flatpak_wait_c: FlatpakCompletion = .{},
+
     /// Reader thread state
     read_thread: std.Thread,
     read_thread_pipe: posix.fd_t,
@@ -670,7 +699,7 @@ pub const ThreadData = struct {
         self.write_buf_pool.deinit(alloc);
 
         // Stop our process watcher
-        self.process.deinit();
+        if (self.process) |*p| p.deinit();
 
         // Stop our write stream
         self.write_stream.deinit();
@@ -763,6 +792,9 @@ const Subprocess = struct {
 
         // Add our binary to the path if we can find it.
         ghostty_path: {
+            // Skip this for flatpak since host cannot reach them
+            if (internal_os.isFlatpak() and FlatpakHostCommand != void) break :ghostty_path;
+
             var exe_buf: [std.fs.max_path_bytes]u8 = undefined;
             const exe_bin_path = std.fs.selfExePath(&exe_buf) catch |err| {
                 log.warn("failed to get ghostty exe path err={}", .{err});