Feat/events socket

EVENTS.md

@@ -0,0 +1,66 @@
+# Events
+
+`shpool` exposes an event stream so that external programs can react to changes
+without polling. This way, a program (e.g. a TUI) can call `shpool list` (or the
+equivalent `ConnectHeader::List` request over the main socket; see the
+[`shpool-protocol`](./shpool-protocol) crate) after each event so that its model
+is always consistent with shpool's state.
+
+## The events socket
+
+The daemon binds a sibling Unix socket next to the main shpool socket:
+
+```bash
+<runtime_dir>/shpool/shpool.socket   # main socket
+<runtime_dir>/shpool/events.socket   # events socket (this protocol)
+```
+
+A subscriber connects to `events.socket` and reads events. The daemon ignores anything written to the events socket, so for subscribers it's effectively read-only.
+
+## Event types
+
+| `type`             | Meaning                                                  |
+| ------------------ | -------------------------------------------------------- |
+| `session.created`  | A new session was added to the table.                    |
+| `session.attached` | A client attached or reattached to a session.            |
+| `session.detached` | A client disconnected from a still-running session.      |
+| `session.removed`  | A session was removed (shell exited, killed, or reaped). |
+
+Subscribers should ignore unknown `type` values so that future event types do
+not break older consumers.
+
+## Wire format
+
+The daemon writes one JSON object per line (JSONL). Each event looks like:
+
+```json
+{"type":"<event-type>"}
+```
+
+There are no other fields. To learn what the event refers to (which session,
+when, etc.), call `shpool list` (or use `ConnectHeader::List`).
+
+The format is robust: literal newline characters only appear as delimiters
+between events. Any newlines within JSON string values are automatically
+escaped (as `\n`) by the daemon.
+
+## Subscribing
+
+For ad-hoc use, `shpool events` connects to the events socket and prints each
+event line to stdout, flushing after each line:
+
+```bash
+shpool events | while read -r ev; do
+  echo "got: $ev"
+  shpool list
+done
+
+shpool events | jq .
+```
+
+## Slow subscribers
+
+Each subscriber has a bounded outbound queue. A subscriber that falls too far
+behind is dropped by the daemon (in which case the subscriber can always reconnect).
+There is no replay, so events that fired while a subscriber was disconnected are
+lost.

libshpool/src/daemon/mod.rs

@@ -17,7 +17,7 @@ use std::{env, os::unix::net::UnixListener, path::PathBuf};
 use anyhow::Context;
 use tracing::{info, instrument};
 
-use crate::{config, consts, hooks};
+use crate::{config, consts, events, hooks};
 
 mod etc_environment;
 mod exit_notify;
@@ -81,8 +81,17 @@ pub fn run(
             (Some(socket.clone()), UnixListener::bind(&socket).context("binding to socket")?)
         }
     };
-    // spawn the signal handler thread in the background
-    signals::Handler::new(cleanup_socket.clone()).spawn()?;
+    let events_socket = events::socket_path(&socket);
+
+    // spawn the signal handler thread in the background. Both sockets need
+    // explicit cleanup on signal exit because the process exits before any
+    // RAII guard can run.
+    let mut socks_to_clean: Vec<PathBuf> = cleanup_socket.iter().cloned().collect();
+    socks_to_clean.push(events_socket.clone());
+    signals::Handler::new(socks_to_clean).spawn().context("spawning signal handler")?;
+
+    let _events_guard =
+        server.start_events_listener(events_socket).context("starting events listener")?;
 
     server::Server::serve(server, listener)?;
 

libshpool/src/daemon/server.rs

@@ -49,7 +49,7 @@ use crate::{
         etc_environment, exit_notify::ExitNotifier, hooks, pager, pager::PagerError, prompt, shell,
         show_motd, ttl_reaper,
     },
-    protocol, test_hooks, tty, user,
+    events, protocol, test_hooks, tty, user,
 };
 
 const DEFAULT_INITIAL_SHELL_PATH: &str = "/usr/bin:/bin:/usr/sbin:/sbin";
@@ -74,6 +74,7 @@ pub struct Server {
     runtime_dir: PathBuf,
     register_new_reapable_session: crossbeam_channel::Sender<(String, Instant)>,
     hooks: Box<dyn hooks::Hooks + Send + Sync>,
+    events_bus: Arc<events::EventBus>,
     daily_messenger: Arc<show_motd::DailyMessenger>,
     log_level_handle: tracing_subscriber::reload::Handle<
         tracing_subscriber::filter::LevelFilter,
@@ -93,13 +94,17 @@ impl Server {
         >,
     ) -> anyhow::Result<Arc<Self>> {
         let shells = Arc::new(Mutex::new(HashMap::new()));
+        let events_bus = events::EventBus::new();
         // buffered so that we are unlikely to block when setting up a
         // new session
         let (new_sess_tx, new_sess_rx) = crossbeam_channel::bounded(10);
-        let shells_tab = Arc::clone(&shells);
-        thread::spawn(move || {
-            if let Err(e) = ttl_reaper::run(new_sess_rx, shells_tab) {
-                warn!("ttl reaper exited with error: {:?}", e);
+        thread::spawn({
+            let shells = Arc::clone(&shells);
+            let events_bus = Arc::clone(&events_bus);
+            move || {
+                if let Err(e) = ttl_reaper::run(new_sess_rx, shells, events_bus) {
+                    warn!("ttl reaper exited with error: {:?}", e);
+                }
             }
         });
 
@@ -110,11 +115,21 @@ impl Server {
             runtime_dir,
             register_new_reapable_session: new_sess_tx,
             hooks,
+            events_bus,
             daily_messenger,
             log_level_handle,
         }))
     }
 
+    /// Bind the events socket and spawn the accept thread. The returned
+    /// guard unlinks the socket file on drop.
+    pub fn start_events_listener(
+        self: &Arc<Self>,
+        socket_path: PathBuf,
+    ) -> anyhow::Result<events::ListenerGuard> {
+        events::start_listener(socket_path, Arc::clone(&self.events_bus))
+    }
+
     #[instrument(skip_all)]
     pub fn serve(server: Arc<Self>, listener: UnixListener) -> anyhow::Result<()> {
         test_hooks::emit("daemon-about-to-listen");
@@ -310,7 +325,12 @@ impl Server {
                 {
                     let _s = span!(Level::INFO, "2_lock(shells)").entered();
                     let mut shells = self.shells.lock().unwrap();
-                    shells.remove(&header.name);
+                    // The publish below is gated on `is_some()` because a
+                    // concurrent kill or reaper may have already removed the
+                    // entry (and published) while we were waiting for the lock.
+                    if shells.remove(&header.name).is_some() {
+                        self.events_bus.publish(&events::Event::SessionRemoved);
+                    }
                 }
 
                 // The child shell has exited, so the shell->client thread should
@@ -331,6 +351,7 @@ impl Server {
                     if let Some(session) = shells.get(&header.name) {
                         session.lifecycle_timestamps.lock().unwrap().last_disconnected_at =
                             Some(time::SystemTime::now());
+                        self.events_bus.publish(&events::Event::SessionDetached);
                     }
                 }
                 if let Err(err) = self.hooks.on_client_disconnect(&header.name) {
@@ -405,9 +426,11 @@ impl Server {
                                     "child_exited chan unclosed, but shell->client thread has exited, clobbering with new subshell"
                                 );
                                 status = AttachStatus::Created { warnings };
+                            } else {
+                                // Reattach confirmed; the create path won't run
+                                // and clobber the entry, so it's safe to publish.
+                                self.events_bus.publish(&events::Event::SessionAttached);
                             }
-
-                            // status is already attached
                         }
                         Some(exit_status) => {
                             // the channel is closed so we know the subshell exited
@@ -472,7 +495,16 @@ impl Server {
                 // we unwrap to propagate the poison as an unwind
                 let _s = span!(Level::INFO, "select_shell_lock_2(shells)").entered();
                 let mut shells = self.shells.lock().unwrap();
+                // If we're replacing a stale entry whose shell process is
+                // gone, surface that to subscribers before announcing the
+                // replacement.
+                let clobbered = shells.contains_key(&header.name);
                 shells.insert(header.name.clone(), Box::new(session));
+                if clobbered {
+                    self.events_bus.publish(&events::Event::SessionRemoved);
+                }
+                self.events_bus.publish(&events::Event::SessionCreated);
+                self.events_bus.publish(&events::Event::SessionAttached);
             }
             // fallthrough to bidi streaming
         } else if let Err(err) = self.hooks.on_reattach(&header.name) {
@@ -572,6 +604,10 @@ impl Server {
                     if let shell::ClientConnectionStatus::DetachNone = status {
                         not_attached_sessions.push(session);
                     } else {
+                        // The bidi-loop unwind in handle_attach owns the
+                        // SessionDetached publish; we just update
+                        // last_disconnected_at eagerly so a concurrent list()
+                        // reflects the detach immediately.
                         s.lifecycle_timestamps.lock().unwrap().last_disconnected_at =
                             Some(time::SystemTime::now());
                     }
@@ -631,6 +667,7 @@ impl Server {
 
             for session in to_remove.iter() {
                 shells.remove(session);
+                self.events_bus.publish(&events::Event::SessionRemoved);
             }
             if !to_remove.is_empty() {
                 test_hooks::emit("daemon-handle-kill-removed-shells");
@@ -646,8 +683,7 @@ impl Server {
     fn handle_list(&self, mut stream: UnixStream) -> anyhow::Result<()> {
         let _s = span!(Level::INFO, "lock(shells)").entered();
         let shells = self.shells.lock().unwrap();
-
-        let sessions: anyhow::Result<Vec<Session>> = shells
+        let sessions: Vec<Session> = shells
             .iter()
             .map(|(k, v)| {
                 let status = match v.inner.try_lock() {
@@ -660,7 +696,6 @@ impl Server {
                     .last_connected_at
                     .map(|t| t.duration_since(time::UNIX_EPOCH).map(|d| d.as_millis() as i64))
                     .transpose()?;
-
                 let last_disconnected_at_unix_ms = timestamps
                     .last_disconnected_at
                     .map(|t| t.duration_since(time::UNIX_EPOCH).map(|d| d.as_millis() as i64))
@@ -675,11 +710,9 @@ impl Server {
                     status,
                 })
             })
-            .collect();
-        let sessions = sessions.context("collecting running session metadata")?;
-
+            .collect::<anyhow::Result<_>>()
+            .context("collecting running session metadata")?;
         write_reply(&mut stream, ListReply { sessions })?;
-
         Ok(())
     }
 

libshpool/src/daemon/signals.rs

@@ -23,11 +23,11 @@ use signal_hook::{consts::TERM_SIGNALS, flag, iterator::Signals};
 use tracing::{error, info};
 
 pub struct Handler {
-    sock: Option<PathBuf>,
+    socks: Vec<PathBuf>,
 }
 impl Handler {
-    pub fn new(sock: Option<PathBuf>) -> Self {
-        Handler { sock }
+    pub fn new(socks: Vec<PathBuf>) -> Self {
+        Handler { socks }
     }
 
     pub fn spawn(self) -> anyhow::Result<()> {
@@ -57,10 +57,12 @@ impl Handler {
             for signal in &mut signals {
                 assert!(TERM_SIGNALS.contains(&signal));
 
-                info!("term sig handler: cleaning up socket");
-                if let Some(sock) = self.sock {
-                    if let Err(e) = std::fs::remove_file(sock).context("cleaning up socket") {
-                        error!("error cleaning up socket file: {}", e);
+                info!("term sig handler: cleaning up sockets");
+                for sock in &self.socks {
+                    match std::fs::remove_file(sock) {
+                        Ok(()) => {}
+                        Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
+                        Err(e) => error!("error cleaning up socket {:?}: {}", sock, e),
                     }
                 }
 

libshpool/src/daemon/ttl_reaper.rs

@@ -30,12 +30,14 @@ use std::{
 use tracing::{info, span, warn, Level};
 
 use super::shell;
+use crate::events;
 
 /// Run the reaper thread loop. Should be invoked in a dedicated
 /// thread.
 pub fn run(
     new_sess: crossbeam_channel::Receiver<(String, Instant)>,
     shells: Arc<Mutex<HashMap<String, Box<shell::Session>>>>,
+    events_bus: Arc<events::EventBus>,
 ) -> anyhow::Result<()> {
     let _s = span!(Level::INFO, "ttl_reaper").entered();
 
@@ -115,6 +117,7 @@ pub fn run(
                         continue;
                     }
                     shells.remove(&reapable.session_name);
+                    events_bus.publish(&events::Event::SessionRemoved);
                 }
             }
         }