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

The input source returned UnixStream EOF as an error, which calloop passes
out of run() into the expect at the end of main. At box shutdown init
signals every process at once, so neshub can go before nescope reads its
SIGTERM: nescope panicked and skipped its cleanup. Deliver what was read,
remove the source and warn instead.

apps/nescope/src/input_ipc.rs

@@ -24,6 +24,22 @@ impl InputIpcSource {
     }
 }
 
+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) => return Err(e),
+            }
+        }
+    }
+}
+
 impl AsFd for InputIpcSource {
     fn as_fd(&self) -> BorrowedFd<'_> {
         self.stream.as_fd()
@@ -45,22 +61,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 +80,10 @@ impl EventSource for InputIpcSource {
             callback(payload, &mut ());
         }
 
+        if closed {
+            tracing::warn!("input IPC socket closed by the hub; no more input from it");
+            return Ok(PostAction::Remove);
+        }
         Ok(PostAction::Continue)
     }
 
@@ -88,3 +99,43 @@ 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(),
+            },
+            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]);
+    }
+}