diff --git a/crates/tty7-core/src/daemon/install/ssh_ops.rs b/crates/tty7-core/src/daemon/install/ssh_ops.rs index bf252a2d..df4f1b2c 100644 --- a/crates/tty7-core/src/daemon/install/ssh_ops.rs +++ b/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" ); } } diff --git a/crates/tty7-core/src/daemon/ssh/forward.rs b/crates/tty7-core/src/daemon/ssh/forward.rs index e6d055ef..00b2fb00 100644 --- a/crates/tty7-core/src/daemon/ssh/forward.rs +++ b/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( diff --git a/crates/tty7-core/src/daemon/ssh/mod.rs b/crates/tty7-core/src/daemon/ssh/mod.rs index 937c06ef..9bbc9f90 100644 --- a/crates/tty7-core/src/daemon/ssh/mod.rs +++ b/crates/tty7-core/src/daemon/ssh/mod.rs @@ -66,7 +66,52 @@ impl ConnectionKey { } } -type ConnSlot = Arc>>; +/// 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>, +} + +impl ConnPool { + fn live(&self) -> impl Iterator> + '_ { + 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> { + self.live().find(|conn| !conn.is_saturated()) + } + + fn any_live(&self) -> bool { + self.live().next().is_some() + } + + fn add(&mut self, conn: &Arc) { + 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>; + +/// 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)> { - 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, + setup: &RouteSetup, + server_command: Option<&str>, + ) -> anyhow::Result { 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 ` 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(); diff --git a/crates/tty7-core/src/daemon/ssh/session.rs b/crates/tty7-core/src/daemon/ssh/session.rs index fe38f8fe..e6386691 100644 --- a/crates/tty7-core/src/daemon/ssh/session.rs +++ b/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>>; 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>, + /// When the server last refused a session here for being at its limit. + /// See [`SshConnection::is_saturated`]. + saturated_at: Mutex>, } 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, 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]