feat: add window_title to shpool list --json

libshpool/src/daemon/server.rs

@@ -839,41 +839,80 @@ impl Server {
 
     #[instrument(skip_all)]
     fn handle_list(&self, mut stream: UnixStream) -> anyhow::Result<()> {
-        let _s = span!(Level::INFO, "lock(shells)").entered();
-        let shells = self.shells.lock();
-        let sessions: anyhow::Result<Vec<Session>> = shells
-            .iter()
-            .map(|(session_name, session)| {
-                let status = match session.inner.try_lock() {
-                    Some(_) => SessionStatus::Disconnected,
-                    None => SessionStatus::Attached,
-                };
+        // None if the session's shell->client thread has exited.
+        type Reply = Option<crossbeam_channel::Receiver<shell::SessionMetadata>>;
 
-                let lifecycle_state = session.lifecycle.snapshot();
-                let last_connected_at_unix_ms = lifecycle_state
-                    .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 = lifecycle_state
-                    .last_disconnected_at
-                    .map(|t| t.duration_since(time::UNIX_EPOCH).map(|d| d.as_millis() as i64))
-                    .transpose()?;
-
-                Ok(Session {
-                    name: session_name.to_string(),
-                    started_at_unix_ms: session
-                        .started_at
-                        .duration_since(time::UNIX_EPOCH)?
-                        .as_millis() as i64,
-                    last_connected_at_unix_ms,
-                    last_disconnected_at_unix_ms,
-                    status,
-                    shell_pid: session.child_pid,
-                    attachments: lifecycle_state.attachment.into_iter().collect(),
+        // Ask every session for its metadata up front, so the shell->client threads
+        // work on their answers in parallel. Asking never blocks, so it is safe under
+        // the shells lock.
+        let sessions: anyhow::Result<Vec<(Session, Reply)>> = {
+            let _s = span!(Level::INFO, "lock(shells)").entered();
+            let shells = self.shells.lock();
+            shells
+                .iter()
+                .map(|(session_name, session)| {
+                    let status = match session.inner.try_lock() {
+                        Some(_) => SessionStatus::Disconnected,
+                        None => SessionStatus::Attached,
+                    };
+
+                    let lifecycle_state = session.lifecycle.snapshot();
+                    let last_connected_at_unix_ms = lifecycle_state
+                        .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 = lifecycle_state
+                        .last_disconnected_at
+                        .map(|t| t.duration_since(time::UNIX_EPOCH).map(|d| d.as_millis() as i64))
+                        .transpose()?;
+
+                    let (reply_tx, reply_rx) = crossbeam_channel::bounded(1);
+                    let reply = session.fetch_metadata.send(reply_tx).ok().map(|()| reply_rx);
+
+                    Ok((
+                        Session {
+                            name: session_name.to_string(),
+                            started_at_unix_ms: session
+                                .started_at
+                                .duration_since(time::UNIX_EPOCH)?
+                                .as_millis() as i64,
+                            last_connected_at_unix_ms,
+                            last_disconnected_at_unix_ms,
+                            status,
+                            shell_pid: session.child_pid,
+                            attachments: lifecycle_state.attachment.into_iter().collect(),
+                            window_title: None,
+                        },
+                        reply,
+                    ))
                 })
-            })
-            .collect();
-        let sessions = sessions.context("collecting running session metadata")?;
+                .collect()
+        };
+        let (mut sessions, replies): (Vec<Session>, Vec<Reply>) =
+            sessions.context("collecting running session metadata")?.into_iter().unzip();
+
+        // The lock is released by now, so we can wait. A busy shell->client thread
+        // (e.g. one building a large restore buffer for a reattach) may miss the
+        // deadline, in which case its session is listed without its metadata.
+        let deadline = Instant::now() + SESSION_MSG_TIMEOUT;
+        for (session, reply) in sessions.iter_mut().zip(replies) {
+            match reply.map(|reply| reply.recv_deadline(deadline)) {
+                Some(Ok(shell::SessionMetadata { window_title })) => {
+                    session.window_title = window_title
+                }
+                Some(Err(crossbeam_channel::RecvTimeoutError::Timeout)) => {
+                    info!(
+                        "listing {} without metadata, which it did not report in time",
+                        session.name
+                    )
+                }
+                None | Some(Err(crossbeam_channel::RecvTimeoutError::Disconnected)) => info!(
+                    "listing {} without metadata, since its shell->client thread has exited",
+                    session.name
+                ),
+            }
+        }
+
         write_reply(&mut stream, ListReply { sessions })?;
         Ok(())
     }
@@ -1196,6 +1235,9 @@ impl Server {
         // no ack chan so we can afford to buffer a bit.
         let (maybe_switch_tx, maybe_switch_rx) = crossbeam_channel::bounded(10);
 
+        // Unbounded so that sending a metadata request never blocks.
+        let (fetch_metadata_tx, fetch_metadata_rx) = crossbeam_channel::unbounded();
+
         let shell_to_client_ctl = Arc::new(Mutex::new(shell::ShellToClientCtl {
             next_heartbeat_id: std::sync::atomic::AtomicU64::new(0),
             client_connection: client_connection_tx,
@@ -1240,6 +1282,7 @@ impl Server {
                 heartbeat_ack: heartbeat_ack_tx,
                 maybe_switch: maybe_switch_rx,
                 child_exit_notifier: shell_to_client_child_exit_notifier,
+                fetch_metadata: fetch_metadata_rx,
             })?);
 
         if let Some(ttl_secs) = header.ttl_secs {
@@ -1256,6 +1299,7 @@ impl Server {
             child_exit_notifier,
             started_at: time::SystemTime::now(),
             lifecycle: shell::SessionLifecycle::default(),
+            fetch_metadata: fetch_metadata_tx,
             inner: Arc::new(Mutex::new(session_inner)),
         })
     }

libshpool/src/daemon/shell.rs

@@ -15,7 +15,7 @@
 use std::{
     io,
     io::{Read, Write},
-    net,
+    iter, net,
     ops::Add,
     os::unix::net::UnixStream,
     sync::{
@@ -114,6 +114,10 @@ pub struct Session {
     pub child_pid: libc::pid_t,
     pub child_exit_notifier: Arc<ExitNotifier>,
     pub shell_to_client_ctl: Arc<Mutex<ShellToClientCtl>>,
+    /// Asks the shell->client thread for the session's [`SessionMetadata`]. It
+    /// is kept here rather than in [`ShellToClientCtl`] so that requesters
+    /// don't wait for that lock.
+    pub fetch_metadata: crossbeam_channel::Sender<MetadataRequest>,
     pub pager_ctl: Arc<Mutex<Option<PagerCtl>>>,
     /// Mutable state with the lock held by the servicing handle_attach thread
     /// while a tty is attached to the session. Probing the mutex can be used
@@ -245,6 +249,46 @@ pub enum ClientConnectionMsg {
     Disconnect,
 }
 
+/// What the shell->client thread returns in response to a
+/// [`Session::fetch_metadata`] request.
+#[derive(Debug, Clone)]
+pub struct SessionMetadata {
+    /// The window title. Any non-UTF-8 bytes are replaced with U+FFFD.
+    pub window_title: Option<String>,
+}
+
+/// A request on [`Session::fetch_metadata`]. It is its own reply channel, so
+/// each requester gets a fresh private channel.
+pub type MetadataRequest = crossbeam_channel::Sender<SessionMetadata>;
+
+/// Handle `request`, received from `requests`, the shell->client thread's
+/// [`Session::fetch_metadata`] channel. Replies to it and to every request
+/// already queued behind it, so concurrent requests get answered as a single
+/// batch. If the channel has closed because the [`Session`] is gone, `requests`
+/// is replaced with [`crossbeam_channel::never`].
+fn handle_fetch_metadata(
+    spool: &dyn session_restore::SessionSpool,
+    request: Result<MetadataRequest, crossbeam_channel::RecvError>,
+    requests: &mut crossbeam_channel::Receiver<MetadataRequest>,
+) {
+    let Ok(request) = request else {
+        // A closed channel is always ready, so leaving it in place would make the
+        // select that produced `request` pick it on every pass. The thread runs on
+        // until the shell exits or its control channels close.
+        *requests = crossbeam_channel::never();
+        return;
+    };
+    let metadata = SessionMetadata {
+        window_title: spool
+            .window_title()
+            .map(|window_title| String::from_utf8_lossy(window_title).into_owned()),
+    };
+    for reply in iter::once(request).chain(requests.try_iter()) {
+        // Nobody needs the reply if the requester has stopped waiting.
+        let _ = reply.send(metadata.clone());
+    }
+}
+
 pub struct ShellToClientArgs {
     pub conn_id: usize,
     pub tty_size: TtySize,
@@ -260,6 +304,9 @@ pub struct ShellToClientArgs {
     // true if the client is still live, false if it has hung up on us.
     pub heartbeat_ack: crossbeam_channel::Sender<(u64, bool)>,
     pub child_exit_notifier: Arc<ExitNotifier>,
+    // The only sending end is the Session's fetch_metadata field, so this closes when the Session
+    // is dropped.
+    pub fetch_metadata: crossbeam_channel::Receiver<MetadataRequest>,
 }
 
 impl SessionInner {
@@ -291,6 +338,7 @@ impl SessionInner {
 
             let mut output_spool =
                 session_restore::new(config, &args.tty_size, args.scrollback_lines, &name);
+            let mut fetch_metadata = args.fetch_metadata;
             let mut buf: Vec<u8> = vec![0; consts::BUF_SIZE];
             let mut poll_fds = [poll::PollFd::new(
                 watchable_master.borrow_fd(),
@@ -300,8 +348,18 @@ impl SessionInner {
             // block until we get the first connection attached so that we don't
             // drop the initial prompt on the floor
             info!("waiting for initial client connection");
-            let mut client_conn: ClientConnectionMsg =
-                args.client_connection.recv().context("waiting for initial client connection")?;
+            let mut client_conn: ClientConnectionMsg = loop {
+                crossbeam_channel::select! {
+                    recv(args.client_connection) -> client_conn => {
+                        break client_conn.context("waiting for initial client connection")?;
+                    }
+                    // The session is already in `shells`, so metadata requests can arrive before
+                    // its first client.
+                    recv(fetch_metadata) -> request => {
+                        handle_fetch_metadata(&*output_spool, request, &mut fetch_metadata);
+                    }
+                }
+            };
             args.client_connection_ack
                 .send(ClientConnectionStatus::New)
                 .context("sending initial client connection ack")?;
@@ -482,6 +540,9 @@ impl SessionInner {
                             }
                         }
                     }
+                    recv(fetch_metadata) -> request => {
+                        handle_fetch_metadata(&*output_spool, request, &mut fetch_metadata);
+                    }
 
                     // make this select non-blocking so we spend most of our time parked
                     // in poll

libshpool/src/session_restore.rs

@@ -56,6 +56,10 @@ pub trait SessionSpool {
 
     /// Process bytes from pty master.
     fn process(&mut self, bytes: &[u8]);
+
+    /// The current window title, or None if it is unset or empty. The title is
+    /// not checked to be UTF-8.
+    fn window_title(&self) -> Option<&[u8]>;
 }
 
 /// A spool that doesn't do anything.
@@ -69,6 +73,10 @@ impl SessionSpool for NullSpool {
     }
 
     fn process(&mut self, _: &[u8]) {}
+
+    fn window_title(&self) -> Option<&[u8]> {
+        None
+    }
 }
 
 /// A spool that restores the last screenful of content using shpool_vt100.
@@ -100,6 +108,11 @@ impl SessionSpool for Vt100Screen {
     fn process(&mut self, bytes: &[u8]) {
         self.parser.process(bytes)
     }
+
+    fn window_title(&self) -> Option<&[u8]> {
+        Some(self.parser.screen().title().as_bytes())
+            .filter(|window_title| !window_title.is_empty())
+    }
 }
 
 /// A spool that restores the last n lines of content using shpool_vt100.
@@ -130,6 +143,11 @@ impl SessionSpool for Vt100Lines {
     fn process(&mut self, bytes: &[u8]) {
         self.parser.process(bytes)
     }
+
+    fn window_title(&self) -> Option<&[u8]> {
+        Some(self.parser.screen().title().as_bytes())
+            .filter(|window_title| !window_title.is_empty())
+    }
 }
 
 /// A spool that restores the last screenful of content using shpool-vterm.
@@ -157,6 +175,10 @@ impl SessionSpool for Vterm {
     fn process(&mut self, bytes: &[u8]) {
         self.term.process(bytes);
     }
+
+    fn window_title(&self) -> Option<&[u8]> {
+        self.term.title()
+    }
 }
 
 /// Creates a spool given a `mode`.

shpool-protocol/src/lib.rs

@@ -294,6 +294,12 @@ pub struct Session {
     /// independently of `status`, so they can briefly disagree.
     #[serde(default)]
     pub attachments: Vec<Attachment>,
+    /// This session's window title. None if it is unset or empty, or
+    /// `session_restore_mode` is `simple`, or the session's shell has exited,
+    /// or the session was too busy to report it in time. Non-UTF-8 bytes are
+    /// replaced with U+FFFD.
+    #[serde(default)]
+    pub window_title: Option<String>,
 }
 
 /// A session's attachment.