fix(ssh): a connection at the server's session limit is full, not dead

sshd allows ten sessions per connection by default and every remote pane is
one of them. The eleventh was refused with ConnectFailed, and the connection
was marked dead for it: taken out of the cache while all ten panes on it were
still using it. Every pane after that dialled its own fresh link, paying a full
handshake and server probe, and so did every short-lived route afterwards,
because nothing held the replacement and it went away as soon as that route
closed.

The cache now keeps a pool per destination. A refusal marks that connection
saturated for a while instead of dead; the pool hands out the first live one
with room and dials another beside them only when all are full. A route or a
native SSH pane that lands on a connection that turns out to be full goes back
to the pool for another.

crates/tty7-core/src/daemon/install/ssh_ops.rs

@@ -261,7 +261,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn a_refused_command_channel_retires_the_connection() {
+    async fn a_refused_command_channel_marks_the_connection_full() {
         let sshd = FakeSshd::connect(Exec::Hangs, Some(0)).await;
         let err = exec(&sshd.conn, "true")
             .await
@@ -271,8 +271,8 @@ mod tests {
             "{err}"
         );
         assert!(
-            !sshd.conn.is_alive(),
-            "Try Again must dial afresh, not retry the spent link"
+            sshd.conn.is_saturated(),
+            "Try Again must be handed another connection, not this full one"
         );
     }
 }

crates/tty7-core/src/daemon/ssh/forward.rs

@@ -904,8 +904,9 @@ impl SshManager {
         let key = ConnectionKey::from_spec(spec);
         let slot = self.conns.lock().unwrap().get(&key).cloned()?;
         let guard = slot.try_lock().ok()?;
-        let conn = guard.upgrade()?;
-        conn.is_alive().then_some(conn)
+        // Any live one: a forward rides a tcpip channel, which the server's
+        // session limit does not count.
+        guard.live().next()
     }
 
     pub fn add_workspace_forward(

crates/tty7-core/src/daemon/ssh/mod.rs

@@ -66,7 +66,52 @@ impl ConnectionKey {
     }
 }
 
-type ConnSlot = Arc<tokio::sync::Mutex<Weak<SshConnection>>>;
+/// Every connection held open to one destination, oldest first.
+///
+/// Usually one. A second appears when the first is at the server's session
+/// limit: sshd allows ten sessions per connection by default and each remote
+/// pane is one of them, so a workspace with more panes than that spans several
+/// connections. Weak, like the single connection this used to be — a link
+/// lives as long as something is using it, not as long as the cache.
+#[derive(Default)]
+struct ConnPool {
+    conns: Vec<Weak<SshConnection>>,
+}
+
+impl ConnPool {
+    fn live(&self) -> impl Iterator<Item = Arc<SshConnection>> + '_ {
+        self.conns
+            .iter()
+            .filter_map(Weak::upgrade)
+            .filter(|conn| conn.is_alive())
+    }
+
+    /// A connection that can take another session, as far as anyone knows.
+    fn usable(&self) -> Option<Arc<SshConnection>> {
+        self.live().find(|conn| !conn.is_saturated())
+    }
+
+    fn any_live(&self) -> bool {
+        self.live().next().is_some()
+    }
+
+    fn add(&mut self, conn: &Arc<SshConnection>) {
+        self.prune();
+        self.conns.push(Arc::downgrade(conn));
+    }
+
+    fn prune(&mut self) {
+        self.conns
+            .retain(|weak| weak.upgrade().is_some_and(|conn| conn.is_alive()));
+    }
+}
+
+type ConnSlot = Arc<tokio::sync::Mutex<ConnPool>>;
+
+/// How many times one open goes back to the cache after a connection turned
+/// out to be full. Each refusal takes that connection out of rotation for a
+/// while, so this only bounds a pathological server, not ordinary use.
+const MAX_SATURATED_RETRIES: usize = 8;
 
 pub struct SshManager {
     runtime: tokio::runtime::Runtime,
@@ -250,8 +295,24 @@ impl SshManager {
 
         broker.status(SshPhase::Connected);
 
-        let channel = match conn.open_session_channel().await {
+        let mut channel = conn.open_session_channel().await;
+        let mut retries = 0;
+        while channel.is_err() && conn.is_saturated() && reused && retries < MAX_SATURATED_RETRIES {
+            retries += 1;
+            (conn, _) = self
+                .open_connection(spec, broker)
+                .await
+                .map_err(|e| format!("{e}"))?;
+            *conn_slot.lock().unwrap() = Arc::downgrade(&conn);
+            channel = conn.open_session_channel().await;
+        }
+        let channel = match channel {
             Ok(channel) => channel,
+            Err(e) if conn.is_saturated() => {
+                return Err(format!(
+                    "open shell channel failed: the server refused another session ({e})"
+                ));
+            }
             Err(e) if reused => {
                 log::info!(
                     "reused ssh connection to {}:{} was dead ({e}); reconnecting",
@@ -335,8 +396,33 @@ impl SshManager {
         setup: &RouteSetup,
         server_command: Option<&str>,
     ) -> anyhow::Result<(RemoteLink, Arc<SshConnection>)> {
-        let (conn, _reused) = self.open_connection(spec, &setup.broker).await?;
+        let mut retries = 0;
+        loop {
+            let (conn, reused) = self.open_connection(spec, &setup.broker).await?;
+            match self.open_remote_link_on(&conn, setup, server_command).await {
+                // Full, not broken: the cache passes this one over now, so
+                // asking again lands on a connection with room, or a new one.
+                // A connection dialled just for this that refuses its very
+                // first sessions will refuse the next one too.
+                Err(e) if conn.is_saturated() && reused && retries < MAX_SATURATED_RETRIES => {
+                    retries += 1;
+                    log::info!(
+                        "ssh {:?}: this connection is at the server's session limit ({e}); \
+                         trying another",
+                        conn.key()
+                    );
+                }
+                result => return result.map(|link| (link, conn)),
+            }
+        }
+    }
 
+    async fn open_remote_link_on(
+        &self,
+        conn: &Arc<SshConnection>,
+        setup: &RouteSetup,
+        server_command: Option<&str>,
+    ) -> anyhow::Result<RemoteLink> {
         let installed = {
             let install_conn = conn.clone();
             setup
@@ -359,7 +445,7 @@ impl SshManager {
             },
             RouteChannel::Control => {
                 conn.remote_entry_or_init(|| async {
-                    let env = probe_remote_env(&conn).await;
+                    let env = probe_remote_env(conn).await;
                     let socket = env.as_ref().and_then(remote_link::remote_control_socket);
                     remote_link::choose_entry(socket.as_deref(), true, &command)
                 })
@@ -369,7 +455,7 @@ impl SshManager {
 
         if let RemoteEntry::StreamLocal { socket } = &entry {
             match conn.open_direct_streamlocal(socket).await {
-                Ok(channel) => return Ok((RemoteLink::stream_local(channel), conn)),
+                Ok(channel) => return Ok(RemoteLink::stream_local(channel)),
                 Err(e) => {
                     log::info!(
                         "ssh {:?}: direct-streamlocal to {socket} refused ({e}); \
@@ -390,7 +476,7 @@ impl SshManager {
             .exec(false, command.as_bytes())
             .await
             .map_err(|e| anyhow::anyhow!("exec `{command}` on the remote failed: {e}"))?;
-        Ok((RemoteLink::session_exec(channel), conn))
+        Ok(RemoteLink::session_exec(channel))
     }
 
     pub async fn restart_remote_server(
@@ -467,7 +553,7 @@ impl SshManager {
         if let Some(slot) = slot
             && let Ok(mut held) = slot.try_lock()
         {
-            *held = Weak::new();
+            held.prune();
         }
     }
 
@@ -477,7 +563,7 @@ impl SshManager {
             .iter()
             .map(|(key, slot)| {
                 let connected = match slot.try_lock() {
-                    Ok(weak) => weak.upgrade().is_some_and(|conn| conn.is_alive()),
+                    Ok(pool) => pool.any_live(),
                     // The slot is held by whoever is opening or using this link
                     // right now. Busy is not down — and callers act on this:
                     // the CLI's `-m <machine>` refuses to route over a link it
@@ -513,7 +599,7 @@ impl SshManager {
                 // A probe the link would not carry says nothing about the
                 // shell. Remembered, its nothing would keep integration off
                 // this host for the rest of the run, fresh links included.
-                if probed.is_some() || conn.is_alive() {
+                if probed.is_some() || (conn.is_alive() && !conn.is_saturated()) {
                     self.probes.lock().unwrap().insert(key, probed.clone());
                 }
                 probed
@@ -554,16 +640,18 @@ impl SshManager {
                 true => {
                     let slot: ConnSlot = {
                         let mut map = self.conns.lock().unwrap();
-                        map.entry(key.clone())
-                            .or_insert_with(|| Arc::new(tokio::sync::Mutex::new(Weak::new())))
-                            .clone()
+                        map.entry(key.clone()).or_default().clone()
                     };
                     let guard = slot.lock_owned().await;
-                    if let Some(conn) = guard.upgrade()
-                        && conn.is_alive()
-                    {
+                    if let Some(conn) = guard.usable() {
                         return Ok((conn, true));
                     }
+                    if guard.any_live() {
+                        log::info!(
+                            "ssh {key:?}: every connection is at the server's session limit; \
+                             dialling another beside them"
+                        );
+                    }
                     Some(guard)
                 }
                 false => None,
@@ -623,7 +711,7 @@ impl SshManager {
 
             let conn = SshConnection::new(handle, key, remote_forwards);
             if let Some(guard) = guard.as_mut() {
-                **guard = Arc::downgrade(&conn);
+                guard.add(&conn);
             }
             Ok((conn, false))
         })
@@ -763,7 +851,10 @@ mod tests {
         mgr.runtime.block_on(async {
             let sshd = FakeSshd::connect(Exec::Hangs, Some(0)).await;
             assert!(mgr.remote_bootstrap(&sshd.conn).await.is_none());
-            assert!(!sshd.conn.is_alive(), "a refused session retires the link");
+            assert!(
+                sshd.conn.is_saturated(),
+                "a refused session marks the link full"
+            );
             assert!(
                 !mgr.probes.lock().unwrap().contains_key(sshd.conn.key()),
                 "the next session, on a fresh link, must ask again"
@@ -834,7 +925,7 @@ mod tests {
     fn evict_connection_empties_the_slot_without_replacing_it() {
         let mgr = bare_manager();
         let key = ConnectionKey::from_spec(&base_spec());
-        let slot: ConnSlot = Arc::new(tokio::sync::Mutex::new(Weak::new()));
+        let slot: ConnSlot = ConnSlot::default();
         mgr.conns.lock().unwrap().insert(key.clone(), slot.clone());
 
         mgr.evict_connection(&key);
@@ -850,7 +941,7 @@ mod tests {
         assert!(
             held.try_lock()
                 .expect("nobody holds it here")
-                .upgrade()
+                .usable()
                 .is_none(),
             "what the slot pointed at is gone, so the next dial does not reuse it"
         );
@@ -863,7 +954,7 @@ mod tests {
     fn evict_connection_leaves_a_slot_that_is_being_dialled_on_alone() {
         let mgr = bare_manager();
         let key = ConnectionKey::from_spec(&base_spec());
-        let slot: ConnSlot = Arc::new(tokio::sync::Mutex::new(Weak::new()));
+        let slot: ConnSlot = ConnSlot::default();
         mgr.conns.lock().unwrap().insert(key.clone(), slot.clone());
 
         let _dialling = slot.try_lock().expect("nobody else holds it in this test");
@@ -892,10 +983,10 @@ mod tests {
         let mut other = base_spec();
         other.host = "build-box".into();
         for spec in [&base_spec(), &other] {
-            mgr.conns.lock().unwrap().insert(
-                ConnectionKey::from_spec(spec),
-                Arc::new(tokio::sync::Mutex::new(Weak::new())),
-            );
+            mgr.conns
+                .lock()
+                .unwrap()
+                .insert(ConnectionKey::from_spec(spec), ConnSlot::default());
         }
 
         let routes = mgr.routes();
@@ -925,7 +1016,7 @@ mod tests {
             forwards: SshForwardRegistry::default(),
             probes: Mutex::new(HashMap::new()),
         };
-        let slot: ConnSlot = Arc::new(tokio::sync::Mutex::new(Weak::new()));
+        let slot: ConnSlot = ConnSlot::default();
         mgr.conns
             .lock()
             .unwrap()
@@ -944,6 +1035,58 @@ mod tests {
         );
     }
 
+    /// A workspace with more panes than the server's `MaxSessions` fills its
+    /// first connection. The pool has to hand the next pane the connection
+    /// with room rather than the full one — and must not drop the full one,
+    /// which is carrying every pane that fit.
+    #[tokio::test]
+    async fn a_full_connection_is_passed_over_for_one_with_room() {
+        let full = FakeSshd::connect(Exec::Hangs, Some(1)).await;
+        let roomy = FakeSshd::connect(Exec::Hangs, None).await;
+        let _pane = full
+            .conn
+            .open_session_channel()
+            .await
+            .expect("the one slot");
+        assert!(full.conn.open_session_channel().await.is_err());
+
+        let mut pool = ConnPool::default();
+        pool.add(&full.conn);
+        pool.add(&roomy.conn);
+
+        let picked = pool.usable().expect("the second connection has room");
+        assert!(Arc::ptr_eq(&picked, &roomy.conn));
+        assert_eq!(pool.live().count(), 2, "the full one stays in the pool");
+        assert!(pool.any_live(), "and the machine still reads as connected");
+    }
+
+    /// Only full connections: nothing to hand out, so the caller dials one
+    /// more — but the machine is not down.
+    #[tokio::test]
+    async fn a_pool_of_full_connections_asks_for_another_without_reading_as_down() {
+        let full = FakeSshd::connect(Exec::Hangs, Some(0)).await;
+        assert!(full.conn.open_session_channel().await.is_err());
+
+        let mut pool = ConnPool::default();
+        pool.add(&full.conn);
+        assert!(pool.usable().is_none());
+        assert!(pool.any_live());
+    }
+
+    #[tokio::test]
+    async fn a_dead_connection_leaves_the_pool() {
+        let gone = FakeSshd::connect(Exec::Hangs, None).await;
+        let kept = FakeSshd::connect(Exec::Hangs, None).await;
+        let mut pool = ConnPool::default();
+        pool.add(&gone.conn);
+        pool.add(&kept.conn);
+
+        gone.conn.mark_dead();
+        pool.prune();
+        assert_eq!(pool.conns.len(), 1);
+        assert!(Arc::ptr_eq(&pool.usable().unwrap(), &kept.conn));
+    }
+
     #[test]
     fn connection_key_includes_jump_chain() {
         let mut with_jump = base_spec();

crates/tty7-core/src/daemon/ssh/session.rs

@@ -1,6 +1,7 @@
 use std::io::{self, Read, Write};
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Mutex, Weak};
+use std::time::{Duration, Instant};
 
 use russh::client::Msg;
 use russh::{Channel, ChannelMsg};
@@ -14,6 +15,12 @@ use super::forward::RemoteForwardTable;
 
 const DATA_CHANNEL_DEPTH: usize = 16;
 
+/// How long a connection the server refused a session on is passed over when
+/// the cache hands one out. Nothing tells us when one of its sessions closes,
+/// so after this it is asked again: one refused round trip if it is still
+/// full, a free slot if a pane on it has gone since.
+const SATURATION_HOLD: Duration = Duration::from_secs(30);
+
 pub type SharedConnection = Arc<Mutex<Weak<SshConnection>>>;
 
 pub enum ChannelCmd {
@@ -216,7 +223,8 @@ impl Drop for CommandChannel {
 
 /// sshd's answer to a session open once the connection's `MaxSessions` are
 /// all taken, by channels still running or by ones never closed. It says
-/// nothing about this channel and everything about the connection.
+/// nothing about this channel and everything about the connection — which is
+/// full, not broken: every session already on it keeps working.
 fn refuses_every_session(e: &russh::Error) -> bool {
     matches!(
         e,
@@ -234,6 +242,9 @@ pub struct SshConnection {
     /// What this connection's server probe proved, once it has. See
     /// [`SshConnection::proved_server`].
     proved_server: Mutex<Option<ProvedServer>>,
+    /// When the server last refused a session here for being at its limit.
+    /// See [`SshConnection::is_saturated`].
+    saturated_at: Mutex<Option<Instant>>,
 }
 
 impl SshConnection {
@@ -249,6 +260,7 @@ impl SshConnection {
             alive: AtomicBool::new(true),
             remote_entry: tokio::sync::Mutex::new(None),
             proved_server: Mutex::new(None),
+            saturated_at: Mutex::new(None),
         })
     }
 
@@ -271,13 +283,36 @@ impl SshConnection {
         self.alive.store(false, Ordering::SeqCst);
     }
 
+    /// Whether the server recently refused a session here because every one
+    /// it allows on this connection is taken (sshd's `MaxSessions`, ten by
+    /// default — one per remote pane).
+    ///
+    /// A full connection is not a dead one. Retiring it used to be the answer,
+    /// and it took the connection out of the cache while every pane on it was
+    /// still using it: the next pane dialled a fresh link and paid a whole
+    /// handshake and server probe for it, and so did everything short-lived
+    /// after that, because nothing held the replacement. The cache passes a
+    /// full connection over instead and adds another beside it.
+    pub fn is_saturated(&self) -> bool {
+        self.saturated_at
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner())
+            .is_some_and(|at| at.elapsed() < SATURATION_HOLD)
+    }
+
+    fn set_saturated(&self, saturated: bool) {
+        *self
+            .saturated_at
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner()) = saturated.then(Instant::now);
+    }
+
     pub async fn open_session_channel(&self) -> Result<Channel<Msg>, russh::Error> {
         let opened = self.handle.lock().await.channel_open_session().await;
-        // `is_alive` is what decides whether the cache hands this connection
-        // out again, and one that refuses sessions keeps refusing them until
-        // it is dropped: every retry on it would fail exactly this way.
-        if opened.as_ref().is_err_and(refuses_every_session) {
-            self.mark_dead();
+        match &opened {
+            Ok(_) => self.set_saturated(false),
+            Err(e) if refuses_every_session(e) => self.set_saturated(true),
+            Err(_) => {}
         }
         opened
     }
@@ -500,9 +535,9 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn a_link_that_refuses_a_session_is_not_handed_out_again() {
+    async fn a_link_at_its_session_limit_is_full_not_dead() {
         let sshd = FakeSshd::connect(Exec::Hangs, Some(1)).await;
-        let _held = sshd
+        let held = sshd
             .conn
             .open_session_channel()
             .await
@@ -520,10 +555,26 @@ mod tests {
             "the fake answers like sshd at MaxSessions: {refused:?}"
         );
         assert!(
-            !sshd.conn.is_alive(),
-            "the cache must dial afresh rather than retry this link"
+            sshd.conn.is_saturated(),
+            "the cache must pass this link over for the next session"
+        );
+        assert!(
+            sshd.conn.is_alive(),
+            "the session already on it is still in use"
         );
         assert_eq!(sshd.refused(), 1);
+
+        held.close().await.expect("close the held session");
+        sshd.wait_for_closed(1).await;
+        let _again = sshd
+            .conn
+            .open_session_channel()
+            .await
+            .expect("a slot freed up");
+        assert!(
+            !sshd.conn.is_saturated(),
+            "a session that opens says the link has room again"
+        );
     }
 
     #[test]