fix(nescope): a hub that closes the input socket is not a panic (#440)

Found in a passing multi-launch probe on NestriTest
(nesprobe-1792327018): at shutdown nescope logged `panicked at
apps/nescope/src/main.rs:639:10: Event loop error: ... IPC socket
closed`.

Cause: `InputIpcSource::process_events` returned EOF as `Err`; calloop
passes that out of `run()` into the `.expect("Event loop error")`. At
box shutdown nesinit does `kill(-1, SIGTERM)`, so neshub can exit before
nescope reads its own SIGTERM, and the panic skips the cleanup after the
loop (socket removal, child kill, the clean-exit log).

Change: on EOF, deliver the frames already read, warn, and return
`PostAction::Remove` so the loop carries on and the signal or auto-exit
path ends it. Read loop split into `fill()` so it is unit-tested with a
socket pair.

Tested on nestripc-1 (CARGO_TARGET_DIR=~/target-fix-nescope-ipc-exit):
`cargo test -p nescope`, 3 new tests pass.

apps/nescope/src/input_ipc.rs

@@ -1,12 +1,16 @@
 use std::io::{self, Read};
 use std::os::unix::io::{AsFd, BorrowedFd};
 use std::os::unix::net::UnixStream;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::Arc;
 
 use calloop::{EventSource, Interest, Mode, Poll, PostAction, Readiness, Token, TokenFactory};
 
 pub struct InputIpcSource {
     stream: UnixStream,
     buf: Vec<u8>,
+    /// Set once the hub has closed its end, for the main loop to act on.
+    closed: Arc<AtomicBool>,
 }
 
 impl InputIpcSource {
@@ -16,14 +20,39 @@ impl InputIpcSource {
         Ok(Self {
             stream,
             buf: Vec::new(),
+            closed: Arc::new(AtomicBool::new(false)),
         })
     }
 
+    /// True once the hub has closed the socket. The main loop gives that a
+    /// grace period and then ends nescope: a compositor with no input holds
+    /// the GPU for nothing.
+    pub fn closed_flag(&self) -> Arc<AtomicBool> {
+        self.closed.clone()
+    }
+
     pub fn try_clone(&self) -> io::Result<UnixStream> {
         self.stream.try_clone()
     }
 }
 
+impl InputIpcSource {
+    /// Read everything available into `buf`. `true` if the peer has closed.
+    fn fill(&mut self) -> io::Result<bool> {
+        let mut tmp = [0u8; 4096];
+        loop {
+            match self.stream.read(&mut tmp) {
+                Ok(0) => return Ok(true),
+                Ok(n) => self.buf.extend_from_slice(&tmp[..n]),
+                Err(e) if e.kind() == io::ErrorKind::WouldBlock => return Ok(false),
+                Err(e) if e.kind() == io::ErrorKind::Interrupted => {}
+                Err(e) if e.kind() == io::ErrorKind::ConnectionReset => return Ok(true),
+                Err(e) => return Err(e),
+            }
+        }
+    }
+}
+
 impl AsFd for InputIpcSource {
     fn as_fd(&self) -> BorrowedFd<'_> {
         self.stream.as_fd()
@@ -45,22 +74,13 @@ impl EventSource for InputIpcSource {
     where
         F: FnMut(Self::Event, &mut Self::Metadata),
     {
-        let mut tmp = [0u8; 4096];
-        loop {
-            match self.stream.read(&mut tmp) {
-                Ok(0) => {
-                    return Err(io::Error::new(
-                        io::ErrorKind::UnexpectedEof,
-                        "IPC socket closed",
-                    ));
-                }
-                Ok(n) => {
-                    self.buf.extend_from_slice(&tmp[..n]);
-                }
-                Err(e) if e.kind() == io::ErrorKind::WouldBlock => break,
-                Err(e) => return Err(e),
-            }
-        }
+        // The hub closing its end is not an error of this source: it is what
+        // the hub does when the box is going down (init signals every process
+        // at once, so the hub can be gone before our own SIGTERM is read), and
+        // an `Err` here is returned out of the event loop and panics `main`,
+        // skipping the cleanup after it. Frames already read are still
+        // delivered, then the source removes itself and the loop carries on.
+        let closed = self.fill()?;
 
         while self.buf.len() >= 2 {
             let payload_len = u16::from_le_bytes([self.buf[0], self.buf[1]]) as usize;
@@ -73,6 +93,11 @@ impl EventSource for InputIpcSource {
             callback(payload, &mut ());
         }
 
+        if closed {
+            self.closed.store(true, Ordering::Relaxed);
+            tracing::warn!("input IPC socket closed by the hub; no more input from it");
+            return Ok(PostAction::Remove);
+        }
         Ok(PostAction::Continue)
     }
 
@@ -88,3 +113,44 @@ impl EventSource for InputIpcSource {
         poll.unregister(self.as_fd())
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::io::Write;
+
+    fn pair() -> (InputIpcSource, UnixStream) {
+        let (a, b) = UnixStream::pair().unwrap();
+        a.set_nonblocking(true).unwrap();
+        (
+            InputIpcSource {
+                stream: a,
+                buf: Vec::new(),
+                closed: Arc::new(AtomicBool::new(false)),
+            },
+            b,
+        )
+    }
+
+    #[test]
+    fn an_open_peer_with_nothing_to_say_is_not_closed() {
+        let (mut source, _peer) = pair();
+        assert!(!source.fill().unwrap());
+    }
+
+    #[test]
+    fn a_closed_peer_is_closed_not_an_error() {
+        let (mut source, peer) = pair();
+        drop(peer);
+        assert!(source.fill().unwrap());
+    }
+
+    #[test]
+    fn what_the_peer_sent_before_closing_is_kept() {
+        let (mut source, mut peer) = pair();
+        peer.write_all(&[1, 0, 7]).unwrap();
+        drop(peer);
+        assert!(source.fill().unwrap());
+        assert_eq!(source.buf, vec![1, 0, 7]);
+    }
+}

apps/nescope/src/main.rs

@@ -389,11 +389,15 @@ fn main() {
 
     // ── Input IPC source ─────────────────────────────────────────────────
     // Connect to the neshub input socket and feed events into the
-    // compositor seat. Reconnection is handled in the idle callback.
+    // compositor seat. There is no reconnection: when the hub closes the socket
+    // the source removes itself, and the loop below ends nescope after a short
+    // grace (a signal that arrives first ends it sooner).
     let ipc_path = args.input_ipc.clone();
+    let mut hub_closed_flag: Option<std::sync::Arc<std::sync::atomic::AtomicBool>> = None;
     match input_ipc::InputIpcSource::connect(&ipc_path) {
         Ok(source) => {
             tracing::info!("Connected to input IPC socket: {ipc_path}");
+            hub_closed_flag = Some(source.closed_flag());
             match source.try_clone() {
                 Ok(write_stream) => {
                     state.ipc_write = Some(write_stream);
@@ -476,6 +480,7 @@ fn main() {
     // never arrives there is nothing to wait for and no game to run.
     let startup = std::time::Instant::now();
     let mut xwayland_timed_out = false;
+    let mut hub_closed_at: Option<std::time::Instant> = None;
 
     event_loop
         .run(Some(Duration::from_secs(1)), &mut data, move |data| {
@@ -578,6 +583,24 @@ fn main() {
                 }
             }
 
+            // ── Hub gone ──────────────────────────────────────────────
+            // neshub closing the input socket mid-run leaves a compositor
+            // nobody can drive. The grace lets a shutdown that is already
+            // under way (SIGTERM stops the loop itself) finish its own way;
+            // otherwise nescope ends, and nesinit ends the box on seeing it.
+            if hub_closed_flag
+                .as_ref()
+                .is_some_and(|f| f.load(std::sync::atomic::Ordering::Relaxed))
+            {
+                let since = hub_closed_at.get_or_insert_with(std::time::Instant::now);
+                if since.elapsed() > Duration::from_secs(2) {
+                    tracing::error!("input hub gone for 2 s with no shutdown — exiting");
+                    kill_all_children();
+                    data.loop_signal.stop();
+                    return;
+                }
+            }
+
             // ── Poll primary process ──────────────────────────────────
             // The launcher (e.g. Steam's shell wrapper) may exit quickly
             // while the real game client stays alive as a reparented