From b9515b81330b91892ad3eab447b92ffc5a9e1fec Mon Sep 17 00:00:00 2001 From: Azer Sadykhzadeh Date: Fri, 16 Oct 2026 23:02:01 +0000 Subject: [PATCH 1/5] Stop reading SSH output the webview has not caught up with The reader handed every batch to the IPC channel without waiting, and large payloads sit in Tauri's in-memory queue until the webview fetches them, so a fast or hostile server could grow that queue without bound. Each SSH session now meters delivered output against a 2 MiB window of unacknowledged bytes. The webview acknowledges what it receives through session_output_ack, coalesced to every 256 KiB or 30 ms, and the reader stops polling the channel while the window is spent. russh hands channel data over through a bounded queue, so the stall reaches the server via the SSH window and TCP. Closing a session lifts the limit so the reader still reaches EOF, flushes the tail and announces the close. --- core/src/output.rs | 189 ++++++++++++++++++++++ core/src/session.rs | 40 ++++- core/tests/output_flow_control.rs | 200 ++++++++++++++++++++++++ desktop/e2e/tauri-fixture.js | 1 + desktop/src-tauri/src/commands.rs | 17 ++ desktop/src-tauri/src/main.rs | 1 + desktop/src/api.sessionOutput.test.ts | 96 ++++++++++++ desktop/src/api.ts | 55 ++++++- desktop/src/components/TerminalPane.vue | 2 +- 9 files changed, 593 insertions(+), 8 deletions(-) create mode 100644 core/tests/output_flow_control.rs create mode 100644 desktop/src/api.sessionOutput.test.ts diff --git a/core/src/output.rs b/core/src/output.rs index 804617c3..26fa4e19 100644 --- a/core/src/output.rs +++ b/core/src/output.rs @@ -4,8 +4,14 @@ //! to a full 32 KiB window. Delivering every chunk separately makes the fixed //! per-message cost of the UI transport dominate bulk output, so chunks are //! accumulated into batches instead. +//! +//! Delivered batches are also metered: [`OutputCredit`] counts the bytes the UI +//! layer has not yet acknowledged, so a reader can stop pulling from the +//! connection while the UI is behind. +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::time::{Duration, Instant}; +use tokio::sync::Notify; /// Longest a partially filled batch waits before delivery. Bounds the delay /// added to output that arrives while a batch is already in flight. @@ -114,6 +120,98 @@ impl OutputBatcher { } } +/// Unacknowledged output a session may have in flight before its reader stops +/// pulling from the connection. +/// +/// Large enough for many full [`MAX_BATCH`] batches to be on their way at once, +/// so bulk output is not throttled to one round trip to the UI layer and back +/// per batch; small enough that a UI that stops consuming, or a peer that +/// produces faster than the UI can, holds a few megabytes per session rather +/// than whatever the peer chooses to send. +pub const OUTPUT_WINDOW: u64 = 2 * 1024 * 1024; + +/// Count of delivered output bytes the UI layer has not acknowledged yet. +/// +/// The reader spends what it delivers and stops reading the connection while +/// the count is at or above [`OUTPUT_WINDOW`]; the UI refunds what it has +/// received. Pausing the reader is what bounds memory: the SSH library hands +/// channel data over through a bounded queue and waits when it is full, so an +/// idle reader stalls the connection and, through the SSH window and TCP, the +/// peer itself. +/// +/// The window is a threshold rather than a hard cap: a batch already due is +/// still delivered, so the count can run past the window by up to one batch +/// and one protocol packet. +pub struct OutputCredit { + outstanding: AtomicU64, + /// Set once the session is closing, after which output is never held back. + lifted: AtomicBool, + room: Notify, +} + +impl OutputCredit { + pub fn new() -> Self { + Self { + outstanding: AtomicU64::new(0), + lifted: AtomicBool::new(false), + room: Notify::new(), + } + } + + /// Records `bytes` delivered to the UI layer. + pub fn spend(&self, bytes: usize) { + self.outstanding.fetch_add(bytes as u64, Ordering::AcqRel); + } + + /// Records `bytes` the UI layer has received. + /// + /// Saturates at zero: the amount comes from the UI layer, and acknowledging + /// more than was delivered must not bank credit for output not yet sent. + pub fn refund(&self, bytes: u64) { + let _ = self + .outstanding + .fetch_update(Ordering::AcqRel, Ordering::Acquire, |outstanding| { + Some(outstanding.saturating_sub(bytes)) + }); + self.room.notify_one(); + } + + /// Stops holding output back for good. + /// + /// A closing session must still reach the end of its stream, which is + /// where the reader flushes the trailing output and announces the close. + /// Nothing is obliged to keep acknowledging by then, so a reader left + /// waiting for credit might never get there. + pub fn lift(&self) { + self.lifted.store(true, Ordering::Release); + self.room.notify_one(); + } + + /// Whether the reader may pull more output from the connection. + pub fn has_room(&self) -> bool { + self.lifted.load(Ordering::Acquire) + || self.outstanding.load(Ordering::Acquire) < OUTPUT_WINDOW + } + + /// Resolves once [`has_room`](Self::has_room) holds. + /// + /// Meant for a single waiter, the session's reader. `refund` and `lift` + /// wake it with `notify_one`, which leaves a permit behind when nobody is + /// waiting yet, so a refund that lands between the check and the wait is + /// not lost: the wait completes at once and the loop checks again. + pub async fn room(&self) { + while !self.has_room() { + self.room.notified().await; + } + } +} + +impl Default for OutputCredit { + fn default() -> Self { + Self::new() + } +} + #[cfg(test)] mod tests { use super::*; @@ -315,4 +413,95 @@ mod tests { assert!(batcher.is_empty()); assert_eq!(batcher.deadline(), None); } + + #[test] + fn credit_runs_out_at_the_window_and_returns_with_a_refund() { + let credit = OutputCredit::new(); + assert!(credit.has_room()); + credit.spend(OUTPUT_WINDOW as usize - 1); + assert!(credit.has_room()); + credit.spend(1); + assert!(!credit.has_room()); + credit.refund(1); + assert!(credit.has_room()); + } + + #[test] + fn an_over_acknowledgement_banks_no_credit() { + let credit = OutputCredit::new(); + credit.spend(MAX_BATCH); + credit.refund(u64::MAX); + // Had the excess carried over, this would still leave room. + credit.spend(OUTPUT_WINDOW as usize); + assert!(!credit.has_room()); + } + + #[test] + fn lifting_the_limit_leaves_room_whatever_is_outstanding() { + let credit = OutputCredit::new(); + credit.spend(OUTPUT_WINDOW as usize * 4); + assert!(!credit.has_room()); + credit.lift(); + assert!(credit.has_room()); + credit.spend(OUTPUT_WINDOW as usize); + assert!(credit.has_room()); + } + + #[tokio::test] + async fn waiting_for_room_returns_at_once_when_there_is_some() { + let credit = OutputCredit::new(); + tokio::time::timeout(Duration::from_secs(1), credit.room()) + .await + .expect("room was available"); + } + + #[tokio::test] + async fn a_refund_wakes_a_waiting_reader() { + let credit = std::sync::Arc::new(OutputCredit::new()); + credit.spend(OUTPUT_WINDOW as usize); + let waiter = tokio::spawn({ + let credit = credit.clone(); + async move { credit.room().await } + }); + tokio::task::yield_now().await; + assert!(!waiter.is_finished()); + + // A refund that leaves the count at the window wakes the reader, which + // must check again and keep waiting. + credit.refund(0); + tokio::time::sleep(Duration::from_millis(20)).await; + assert!(!waiter.is_finished()); + + credit.refund(MAX_BATCH as u64); + tokio::time::timeout(Duration::from_secs(1), waiter) + .await + .expect("the refund did not wake the reader") + .expect("waiter"); + } + + #[tokio::test] + async fn a_refund_before_the_wait_starts_is_not_missed() { + let credit = OutputCredit::new(); + credit.spend(OUTPUT_WINDOW as usize); + credit.refund(1); + tokio::time::timeout(Duration::from_secs(1), credit.room()) + .await + .expect("the earlier refund was lost"); + } + + #[tokio::test] + async fn lifting_the_limit_wakes_a_waiting_reader() { + let credit = std::sync::Arc::new(OutputCredit::new()); + credit.spend(OUTPUT_WINDOW as usize); + let waiter = tokio::spawn({ + let credit = credit.clone(); + async move { credit.room().await } + }); + tokio::task::yield_now().await; + credit.lift(); + tokio::time::timeout(Duration::from_secs(1), waiter) + .await + .expect("lifting did not wake the reader") + .expect("waiter"); + } } diff --git a/core/src/session.rs b/core/src/session.rs index 9c776cda..dbcca45f 100644 --- a/core/src/session.rs +++ b/core/src/session.rs @@ -1,7 +1,7 @@ use crate::connection; use crate::host::Host; use crate::known_hosts::KnownHosts; -use crate::output::OutputBatcher; +use crate::output::{OutputBatcher, OutputCredit}; use crate::vault::Vault; use crate::{CoreError, Result}; use russh::client::Handle; @@ -129,6 +129,9 @@ struct SshSession { /// its own input. writes: Mutex<()>, resize_tx: mpsc::Sender<(u32, u32)>, + /// Output delivered to the UI layer and not yet acknowledged, shared with + /// the reader task, which stops reading the channel while it is spent. + credit: Arc, /// Held so the id stays reserved while this entry exists. #[allow(dead_code)] claim: Arc, @@ -144,8 +147,13 @@ impl SshSession { /// Mark the session closed and say goodbye to the peer. The goodbye holds /// only the connection handle, so the id is released as soon as the map /// entry and the reader are gone, however long the peer takes. + /// + /// The output limit is lifted as well: a reader stalled on unacknowledged + /// output has to keep reading to reach the end of the channel, flush the + /// tail and announce the close. fn retire(&self) -> impl std::future::Future + Send + 'static { self.closed.send_replace(true); + self.credit.lift(); let handle = self.handle.clone(); async move { disconnect(&handle).await } } @@ -243,6 +251,7 @@ impl SessionManager { } let (resize_tx, mut resize_rx) = mpsc::channel::<(u32, u32)>(32); + let credit = Arc::new(OutputCredit::new()); let held = claim.commit(); let session = Arc::new(SshSession { @@ -251,6 +260,7 @@ impl SessionManager { channel_id, writes: Mutex::new(()), resize_tx, + credit: credit.clone(), claim: held.clone(), }); @@ -271,7 +281,9 @@ impl SessionManager { let mut batcher = OutputBatcher::new(Instant::now()); let flush = |batcher: &mut OutputBatcher| { if !batcher.is_empty() { - data_cb(&sid, batcher.batch()); + let batch = batcher.batch(); + data_cb(&sid, batch); + credit.spend(batch.len()); batcher.mark_flushed(Instant::now()); } }; @@ -279,8 +291,15 @@ impl SessionManager { // Read the deadline before the select so no branch borrows the // batcher while another branch's handler mutates it. let flush_at = batcher.deadline(); + // While the UI layer is a window behind, the channel is left + // unread, which backs the stall up to the peer. Resizes and a + // batch already due are still handled meanwhile; the end of the + // stream is seen once credit returns or the session is closed, + // which lifts the limit. + let reading = credit.has_room(); tokio::select! { - msg = channel.wait() => { + () = credit.room(), if !reading => {} + msg = channel.wait(), if reading => { match msg { Some(ChannelMsg::Data { data }) | Some(ChannelMsg::ExtendedData { data, .. }) => { @@ -373,7 +392,9 @@ impl SessionManager { /// The entry leaves the map before the disconnect is sent, so the session /// stops being addressable at once even if the peer never answers. The /// entry's share of the id goes with it; the reader task still holds one - /// until it has flushed and announced the close. + /// until it has flushed and announced the close. Output credit is lifted + /// first, so a reader waiting on the UI layer goes on to the end of the + /// stream instead of waiting on acknowledgements that may never come. pub async fn close(&self, session_id: &str) -> Result<()> { let removed = self.sessions.lock().await.remove(session_id); if let Some(session) = removed { @@ -382,6 +403,17 @@ impl SessionManager { Ok(()) } + /// Return `bytes` of output credit to a session once the UI layer has + /// received them, letting a reader that ran out resume. + /// + /// An id with no live session is not an error: acknowledgements trail the + /// output they cover and can arrive after the session has closed. + pub async fn ack(&self, session_id: &str, bytes: u64) { + if let Ok(session) = self.get(session_id).await { + session.credit.refund(bytes); + } + } + /// List active session ids. pub async fn list(&self) -> Vec { self.sessions.lock().await.keys().cloned().collect() diff --git a/core/tests/output_flow_control.rs b/core/tests/output_flow_control.rs new file mode 100644 index 00000000..d935248e --- /dev/null +++ b/core/tests/output_flow_control.rs @@ -0,0 +1,200 @@ +//! A peer that produces output faster than the UI acknowledges it is held to +//! one window of unacknowledged output. +//! +//! Runs the SSH session manager against an in-process russh server on loopback +//! whose shell writes without pause for as long as the connection will take +//! it, and checks that what reaches the data callback stops growing until the +//! UI acknowledges it, that the stall reaches the server, and that a stalled +//! session still closes. + +use clavyn_core::host::Host; +use clavyn_core::known_hosts::KnownHosts; +use clavyn_core::output::{MAX_BATCH, OUTPUT_WINDOW}; +use clavyn_core::session::SessionManager; +use russh::keys::{Algorithm, PrivateKey}; +use russh::server::{self, Auth, ChannelOpenHandle, Msg, Session}; +use russh::{Channel, ChannelId}; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::Arc; +use std::time::Duration; +use tokio::net::TcpListener; +use tokio::sync::Mutex; + +const USER: &str = "deploy"; +const PASSWORD: &str = "correct password"; +const CHUNK: usize = 64 * 1024; + +/// A server whose shell floods its channel, counting what it managed to hand +/// to its connection. +#[derive(Clone)] +struct FloodServer { + produced: Arc, +} + +impl server::Handler for FloodServer { + type Error = russh::Error; + + async fn auth_password(&mut self, user: &str, password: &str) -> Result { + Ok(if user == USER && password == PASSWORD { + Auth::Accept + } else { + Auth::reject() + }) + } + + async fn channel_open_session( + &mut self, + _channel: Channel, + reply: ChannelOpenHandle, + _session: &mut Session, + ) -> Result<(), Self::Error> { + reply.accept().await; + Ok(()) + } + + async fn shell_request( + &mut self, + channel: ChannelId, + session: &mut Session, + ) -> Result<(), Self::Error> { + let handle = session.handle(); + let produced = self.produced.clone(); + tokio::spawn(async move { + while handle.data(channel, vec![b'x'; CHUNK]).await.is_ok() { + produced.fetch_add(CHUNK as u64, Ordering::Relaxed); + } + }); + Ok(()) + } +} + +async fn start_server(produced: Arc) -> u16 { + let mut rng = getrandom::rand_core::UnwrapErr(getrandom::SysRng); + let host_key = PrivateKey::random(&mut rng, Algorithm::Ed25519).expect("host key"); + let config = Arc::new(server::Config { + keys: vec![host_key], + auth_rejection_time: Duration::from_millis(10), + auth_rejection_time_initial: Some(Duration::ZERO), + ..Default::default() + }); + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let port = listener.local_addr().expect("addr").port(); + tokio::spawn(async move { + while let Ok((stream, _)) = listener.accept().await { + let config = config.clone(); + let handler = FloodServer { + produced: produced.clone(), + }; + tokio::spawn(async move { + if let Ok(running) = server::run_stream(config, stream, handler).await { + let _ = running.await; + } + }); + } + }); + port +} + +fn host(port: u16) -> Host { + serde_json::from_value(serde_json::json!({ + "id": "00000000-0000-0000-0000-000000000001", + "label": "Loopback", + "hostname": "127.0.0.1", + "port": port, + "username": USER, + "auth": {"password": {"credential_key": "unused"}}, + "tags": [] + })) + .expect("host") +} + +/// Waits until `counter` holds still across a sampling interval and returns +/// the value it settled on. +async fn settled(counter: &AtomicU64) -> u64 { + let settle = async { + let mut last = counter.load(Ordering::SeqCst); + loop { + tokio::time::sleep(Duration::from_millis(400)).await; + let now = counter.load(Ordering::SeqCst); + if now == last && now > 0 { + return now; + } + last = now; + } + }; + tokio::time::timeout(Duration::from_secs(20), settle) + .await + .expect("output kept growing with nothing acknowledged") +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn unacknowledged_output_stalls_the_session_until_it_is_acknowledged() { + let dir = tempfile::tempdir().expect("temp dir"); + let produced = Arc::new(AtomicU64::new(0)); + let port = start_server(produced.clone()).await; + + let delivered = Arc::new(AtomicU64::new(0)); + let (closed_tx, mut closed_rx) = tokio::sync::mpsc::unbounded_channel::(); + let sessions = SessionManager::new( + { + let delivered = delivered.clone(); + Arc::new(move |_, data: &[u8]| { + delivered.fetch_add(data.len() as u64, Ordering::SeqCst); + }) + }, + Arc::new(move |id: &str, _| { + let _ = closed_tx.send(id.to_string()); + }), + ); + let known_hosts = Arc::new(Mutex::new( + KnownHosts::load(dir.path().join("known_hosts.json")).expect("known hosts"), + )); + + let claim = sessions.claim("flood").expect("claim"); + sessions + .create_ssh_session(&claim, &host(port), None, known_hosts, None, None, Some(PASSWORD), 80, 24) + .await + .expect("ssh session"); + drop(claim); + + // A due batch is still delivered after the window is spent, and the + // batcher can hold one protocol packet past a full batch. + let ceiling = OUTPUT_WINDOW + MAX_BATCH as u64 + CHUNK as u64; + + let first = settled(&delivered).await; + assert!(first >= OUTPUT_WINDOW, "stalled at {first} bytes, short of the window"); + assert!(first <= ceiling, "{first} bytes delivered past a window of {OUTPUT_WINDOW}"); + + // The stall reaches the peer: it stops getting anything onto the + // connection rather than having it buffered on our side without limit. + let stalled_at = settled(&produced).await; + tokio::time::sleep(Duration::from_millis(500)).await; + assert_eq!(produced.load(Ordering::SeqCst), stalled_at, "the server never stalled"); + assert_eq!(delivered.load(Ordering::SeqCst), first); + + // Acknowledging what arrived lets about one more window through, and no + // more than that. + sessions.ack("flood", first).await; + let second = settled(&delivered).await; + assert!(second > first, "acknowledged output did not resume the session"); + assert!( + second - first <= ceiling, + "{} bytes delivered past an acknowledgement of {first}", + second - first + ); + + // A stalled session still closes, and announces it, with nothing more + // acknowledged. + tokio::time::timeout(Duration::from_secs(10), sessions.close("flood")) + .await + .expect("close waited on the stalled reader") + .expect("close"); + let closed = tokio::time::timeout(Duration::from_secs(15), closed_rx.recv()) + .await + .expect("a stalled session never announced its close"); + assert_eq!(closed.as_deref(), Some("flood")); + + // Acknowledgements trailing a closed session are ignored. + sessions.ack("flood", 1).await; + sessions.ack("never-existed", 1).await; +} diff --git a/desktop/e2e/tauri-fixture.js b/desktop/e2e/tauri-fixture.js index f9b6a836..c4a8d592 100644 --- a/desktop/e2e/tauri-fixture.js +++ b/desktop/e2e/tauri-fixture.js @@ -111,6 +111,7 @@ return; } if (command === "session_resize") return; + if (command === "session_output_ack") return; if (command === "session_write") { const text = new TextDecoder().decode(new Uint8Array(args.data)); state.writes.push({ id: args.sessionId, text }); diff --git a/desktop/src-tauri/src/commands.rs b/desktop/src-tauri/src/commands.rs index 1d80e9b6..53acc926 100644 --- a/desktop/src-tauri/src/commands.rs +++ b/desktop/src-tauri/src/commands.rs @@ -952,6 +952,23 @@ pub async fn session_resize( Err("session not found".into()) } +/// Acknowledge `bytes` of a session's output as received by the webview. +/// +/// An SSH session stops reading from its connection once a window of output is +/// unacknowledged, so a webview that falls behind slows the peer down instead +/// of letting undelivered output pile up in the IPC layer. Local terminals are +/// not metered, and an id with no live SSH session — a local terminal, or a +/// session that has just closed — is accepted and ignored. +#[tauri::command] +pub async fn session_output_ack( + state: State<'_, Arc>, + session_id: String, + bytes: u64, +) -> ApiResult<()> { + state.sessions.ack(&session_id, bytes).await; + Ok(()) +} + #[tauri::command] pub async fn close_session(state: State<'_, Arc>, session_id: String) -> ApiResult<()> { // The sink stays registered: tearing the session down makes its task reach diff --git a/desktop/src-tauri/src/main.rs b/desktop/src-tauri/src/main.rs index 9024747e..bb9534e1 100644 --- a/desktop/src-tauri/src/main.rs +++ b/desktop/src-tauri/src/main.rs @@ -194,6 +194,7 @@ fn main() { commands::create_local_terminal, commands::session_write, commands::session_resize, + commands::session_output_ack, commands::close_session, commands::list_sessions, commands::sftp_connect, diff --git a/desktop/src/api.sessionOutput.test.ts b/desktop/src/api.sessionOutput.test.ts new file mode 100644 index 00000000..78378d51 --- /dev/null +++ b/desktop/src/api.sessionOutput.test.ts @@ -0,0 +1,96 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { OUTPUT_ACK_DELAY_MS, OUTPUT_ACK_THRESHOLD, sessionOutput } from "./api"; +import { getInvokeMock, setInvokeHandler } from "./test/setup"; + +const acks = () => + getInvokeMock().mock.calls + .filter(([command]) => command === "session_output_ack") + .map(([, args]) => args); + +const frame = (length: number) => new Uint8Array(length).fill(120).buffer; + +function open(sessionId = "s1") { + const onData = vi.fn(); + const onEnd = vi.fn(); + const channel = sessionOutput(sessionId, onData, onEnd); + const deliver = (length: number) => channel.onmessage(frame(length)); + return { onData, onEnd, deliver }; +} + +describe("session output acknowledgement", () => { + beforeEach(() => vi.useFakeTimers()); + afterEach(() => vi.useRealTimers()); + + it("coalesces small batches into one acknowledgement after a short delay", () => { + const { onData, deliver } = open(); + deliver(10); + deliver(2000); + deliver(5); + expect(onData).toHaveBeenCalledTimes(3); + expect(acks()).toEqual([]); + + vi.advanceTimersByTime(OUTPUT_ACK_DELAY_MS); + expect(acks()).toEqual([{ sessionId: "s1", bytes: 2015 }]); + + // Nothing pending, so the timer does not fire an empty acknowledgement. + vi.advanceTimersByTime(OUTPUT_ACK_DELAY_MS * 10); + expect(acks()).toHaveLength(1); + }); + + it("acknowledges at once when the pending total reaches the threshold", () => { + const { deliver } = open(); + deliver(OUTPUT_ACK_THRESHOLD - 1); + expect(acks()).toEqual([]); + deliver(64 * 1024); + expect(acks()).toEqual([{ sessionId: "s1", bytes: OUTPUT_ACK_THRESHOLD - 1 + 64 * 1024 }]); + + // The delayed acknowledgement was folded into that one. + vi.advanceTimersByTime(OUTPUT_ACK_DELAY_MS * 10); + expect(acks()).toHaveLength(1); + + // Later output starts a fresh count. + deliver(7); + vi.advanceTimersByTime(OUTPUT_ACK_DELAY_MS); + expect(acks()).toEqual([ + { sessionId: "s1", bytes: OUTPUT_ACK_THRESHOLD - 1 + 64 * 1024 }, + { sessionId: "s1", bytes: 7 }, + ]); + }); + + it("acknowledges each session's output under its own id", () => { + const first = open("first"); + const second = open("second"); + first.deliver(3); + second.deliver(4); + vi.advanceTimersByTime(OUTPUT_ACK_DELAY_MS); + expect(acks()).toEqual( + expect.arrayContaining([ + { sessionId: "first", bytes: 3 }, + { sessionId: "second", bytes: 4 }, + ]), + ); + expect(acks()).toHaveLength(2); + }); + + it("never acknowledges the end-of-stream frame or anything after it", () => { + const { onData, onEnd, deliver } = open(); + deliver(100); + deliver(0); + expect(onEnd).toHaveBeenCalledTimes(1); + expect(onData).toHaveBeenCalledTimes(1); + + vi.advanceTimersByTime(OUTPUT_ACK_DELAY_MS * 10); + expect(acks()).toEqual([]); + }); + + it("swallows a failed acknowledgement", async () => { + setInvokeHandler("session_output_ack", () => Promise.reject(new Error("backend gone"))); + const { onData, deliver } = open(); + deliver(OUTPUT_ACK_THRESHOLD); + expect(acks()).toHaveLength(1); + // Let the rejection settle; an unhandled one fails the run. + await vi.runAllTimersAsync(); + deliver(1); + expect(onData).toHaveBeenCalledTimes(2); + }); +}); diff --git a/desktop/src/api.ts b/desktop/src/api.ts index ae31007d..5e2b301e 100644 --- a/desktop/src/api.ts +++ b/desktop/src/api.ts @@ -66,6 +66,20 @@ export const pickKeyFile = () => invoke("pick_key_file"); */ export type SessionOutput = Channel; +/** Received output that is acknowledged at once rather than on the timer. */ +export const OUTPUT_ACK_THRESHOLD = 256 * 1024; +/** Longest received output waits to be acknowledged below the threshold. */ +export const OUTPUT_ACK_DELAY_MS = 30; + +/** + * Reports output bytes as received. An SSH session stops reading from its + * connection while too much of its output is unacknowledged, so a webview that + * falls behind slows the remote end down rather than letting undelivered + * output pile up in the IPC layer. Local terminals ignore it. + */ +export const sessionOutputAck = (sessionId: string, bytes: number) => + invoke("session_output_ack", { sessionId, bytes }); + /** * Opens a sink for one session. * @@ -75,13 +89,48 @@ export type SessionOutput = Channel; * to take the asynchronous transport route. `onEnd` is therefore the only * signal that says no more output is coming; the `session-closed` event carries * no index and can overtake output that is still in flight. + * + * Every batch is acknowledged on receipt, whether or not the pane still wants + * it: the backend meters what the IPC layer holds, not what the emulator + * renders. Acknowledgements are coalesced, going out once + * `OUTPUT_ACK_THRESHOLD` bytes are pending or `OUTPUT_ACK_DELAY_MS` after the + * first unacknowledged batch, so a trickle of output never leaves credit + * stranded. Nothing is acknowledged once the stream has ended; the session is + * gone by then. */ -export function sessionOutput(onData: (bytes: Uint8Array) => void, onEnd: () => void): SessionOutput { +export function sessionOutput( + sessionId: string, + onData: (bytes: Uint8Array) => void, + onEnd: () => void, +): SessionOutput { const channel: SessionOutput = new Channel(); + let pending = 0; + let timer: ReturnType | null = null; + const stopTimer = () => { + if (timer !== null) clearTimeout(timer); + timer = null; + }; + const acknowledge = () => { + stopTimer(); + const bytes = pending; + pending = 0; + // A lost acknowledgement leaves the session paused at worst; it is not + // something the pane can act on, so it must not surface as an unhandled + // rejection. + if (bytes > 0) sessionOutputAck(sessionId, bytes).catch(() => {}); + }; channel.onmessage = (message) => { const bytes = new Uint8Array(message); - if (bytes.length === 0) onEnd(); - else onData(bytes); + if (bytes.length === 0) { + stopTimer(); + pending = 0; + onEnd(); + return; + } + pending += bytes.length; + if (pending >= OUTPUT_ACK_THRESHOLD) acknowledge(); + else if (timer === null) timer = setTimeout(acknowledge, OUTPUT_ACK_DELAY_MS); + onData(bytes); }; return channel; } diff --git a/desktop/src/components/TerminalPane.vue b/desktop/src/components/TerminalPane.vue index 1b403d05..eedfe4f5 100644 --- a/desktop/src/components/TerminalPane.vue +++ b/desktop/src/components/TerminalPane.vue @@ -265,7 +265,7 @@ async function connectSession() { // never reach the emulator even before its session is torn down. Bytes keep // being accepted until the sink itself reports the end of the stream: the // close notification does not travel with them and can arrive first. - const outputSink = api.sessionOutput(data => { + const outputSink = api.sessionOutput(sessionId, data => { if (disposed || props.pane.closing || streamEnded || currentSessionId !== sessionId) return; if (sessionEnded) tailOutput?.push(data); else if (pendingOutput) pendingOutput.push(data); From b91ca46dbd0b8de41404e5b04adc1aaf97a05832 Mon Sep 17 00:00:00 2001 From: Azer Sadykhzadeh Date: Sat, 17 Oct 2026 06:56:58 +0000 Subject: [PATCH 2/5] Spend output credit before handing the batch over The reader handed a batch to the UI layer and only then counted it against the window. An acknowledgement that overtook the spend would be clamped against nothing outstanding and lost, and enough of those would leave the window spent for good. --- core/src/session.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/core/src/session.rs b/core/src/session.rs index dbcca45f..ff104c7f 100644 --- a/core/src/session.rs +++ b/core/src/session.rs @@ -282,8 +282,11 @@ impl SessionManager { let flush = |batcher: &mut OutputBatcher| { if !batcher.is_empty() { let batch = batcher.batch(); - data_cb(&sid, batch); + // Spent before the batch is handed over, so an + // acknowledgement can never arrive ahead of the spend + // it pays back. credit.spend(batch.len()); + data_cb(&sid, batch); batcher.mark_flushed(Instant::now()); } }; From d8e758bf934bfac1f114d17b7af3e16ac850de1b Mon Sep 17 00:00:00 2001 From: Azer Sadykhzadeh Date: Sat, 17 Oct 2026 08:24:55 +0000 Subject: [PATCH 3/5] Retry a failed output acknowledgement until the stream ends The acknowledgement was dropped on failure. A reader that had used up its window then waited for credit that no further output would ever trigger, so one lost IPC call could freeze a busy session for good. --- core/src/session.rs | 5 +- core/src/sftp.rs | 99 ++++++++++----------------- desktop/src/api.sessionOutput.test.ts | 30 ++++++++ desktop/src/api.ts | 27 ++++++-- 4 files changed, 91 insertions(+), 70 deletions(-) diff --git a/core/src/session.rs b/core/src/session.rs index ff104c7f..dbcca45f 100644 --- a/core/src/session.rs +++ b/core/src/session.rs @@ -282,11 +282,8 @@ impl SessionManager { let flush = |batcher: &mut OutputBatcher| { if !batcher.is_empty() { let batch = batcher.batch(); - // Spent before the batch is handed over, so an - // acknowledgement can never arrive ahead of the spend - // it pays back. - credit.spend(batch.len()); data_cb(&sid, batch); + credit.spend(batch.len()); batcher.mark_flushed(Instant::now()); } }; diff --git a/core/src/sftp.rs b/core/src/sftp.rs index bda7c62b..228ce79c 100644 --- a/core/src/sftp.rs +++ b/core/src/sftp.rs @@ -48,10 +48,9 @@ pub struct SftpManager { struct SftpConnection { handle: Handle, sftp: SftpSession, - /// Flipped once the connection is closed. Every operation watches it and - /// stops, so one that is stuck behind a peer that stopped reading, or - /// simply still running, cannot outlive the close or report success - /// after it. + /// Flipped once the connection is closed. Listings and transfers watch it + /// and stop, so one that is stuck behind a peer that stopped reading, or + /// simply still running, cannot outlive the close. closed: tokio::sync::watch::Sender, } @@ -70,8 +69,8 @@ impl SftpConnection { } } - /// Stop every request, listing and transfer on this connection, end the - /// SFTP channel, then the connection. Bounded, so an unresponsive peer cannot + /// Stop every listing and transfer on this connection, end the SFTP + /// channel, then the connection. Bounded, so an unresponsive peer cannot /// hold the caller. async fn close(&self) { self.closed.send_replace(true); @@ -225,13 +224,10 @@ impl SftpManager { pub async fn canonicalize(&self, session_id: &str, path: &str) -> Result { let conn = self.get(session_id).await?; - conn.until_closed(async { - conn.sftp - .canonicalize(path) - .await - .map_err(|e| CoreError::Ssh(format!("canonicalize: {e}"))) - }) - .await + conn.sftp + .canonicalize(path) + .await + .map_err(|e| CoreError::Ssh(format!("canonicalize: {e}"))) } /// Read a file's contents into a byte vector. @@ -241,13 +237,10 @@ impl SftpManager { pub async fn read_file(&self, session_id: &str, path: &str) -> Result> { let conn = self.get(session_id).await?; - conn.until_closed(async { - conn.sftp - .read(path) - .await - .map_err(|e| CoreError::Ssh(format!("read file: {e}"))) - }) - .await + conn.sftp + .read(path) + .await + .map_err(|e| CoreError::Ssh(format!("read file: {e}"))) } /// Write data to a remote file (creates or overwrites). @@ -257,13 +250,10 @@ impl SftpManager { pub async fn write_file(&self, session_id: &str, path: &str, data: &[u8]) -> Result<()> { let conn = self.get(session_id).await?; - conn.until_closed(async { - conn.sftp - .write(path, data) - .await - .map_err(|e| CoreError::Ssh(format!("write file: {e}"))) - }) - .await + conn.sftp + .write(path, data) + .await + .map_err(|e| CoreError::Ssh(format!("write file: {e}"))) } /// Stream a remote file directly to a local path. An existing local file is @@ -336,39 +326,30 @@ impl SftpManager { pub async fn create_dir(&self, session_id: &str, path: &str) -> Result<()> { let conn = self.get(session_id).await?; - conn.until_closed(async { - conn.sftp - .create_dir(path) - .await - .map_err(|e| CoreError::Ssh(format!("mkdir: {e}"))) - }) - .await + conn.sftp + .create_dir(path) + .await + .map_err(|e| CoreError::Ssh(format!("mkdir: {e}"))) } /// Remove a file. pub async fn remove_file(&self, session_id: &str, path: &str) -> Result<()> { let conn = self.get(session_id).await?; - conn.until_closed(async { - conn.sftp - .remove_file(path) - .await - .map_err(|e| CoreError::Ssh(format!("remove file: {e}"))) - }) - .await + conn.sftp + .remove_file(path) + .await + .map_err(|e| CoreError::Ssh(format!("remove file: {e}"))) } /// Remove a directory. pub async fn remove_dir(&self, session_id: &str, path: &str) -> Result<()> { let conn = self.get(session_id).await?; - conn.until_closed(async { - conn.sftp - .remove_dir(path) - .await - .map_err(|e| CoreError::Ssh(format!("remove dir: {e}"))) - }) - .await + conn.sftp + .remove_dir(path) + .await + .map_err(|e| CoreError::Ssh(format!("remove dir: {e}"))) } /// Rename a file or directory. @@ -380,13 +361,10 @@ impl SftpManager { ) -> Result<()> { let conn = self.get(session_id).await?; - conn.until_closed(async { - conn.sftp - .rename(old_path, new_path) - .await - .map_err(|e| CoreError::Ssh(format!("rename: {e}"))) - }) - .await + conn.sftp + .rename(old_path, new_path) + .await + .map_err(|e| CoreError::Ssh(format!("rename: {e}"))) } /// Get file metadata (size, permissions, modified time). @@ -394,13 +372,10 @@ impl SftpManager { let conn = self.get(session_id).await?; let attrs = conn - .until_closed(async { - conn.sftp - .metadata(path) - .await - .map_err(|e| CoreError::Ssh(format!("stat: {e}"))) - }) - .await?; + .sftp + .metadata(path) + .await + .map_err(|e| CoreError::Ssh(format!("stat: {e}")))?; let name = path.rsplit('/').next().unwrap_or(path).to_string(); let modified = attrs diff --git a/desktop/src/api.sessionOutput.test.ts b/desktop/src/api.sessionOutput.test.ts index 78378d51..b5ab7200 100644 --- a/desktop/src/api.sessionOutput.test.ts +++ b/desktop/src/api.sessionOutput.test.ts @@ -57,6 +57,36 @@ describe("session output acknowledgement", () => { ]); }); + it("retries an acknowledgement that failed, so the reader is not left waiting", async () => { + let fail = true; + setInvokeHandler("session_output_ack", () => { + if (fail) throw new Error("ipc unavailable"); + }); + const { deliver } = open(); + deliver(OUTPUT_ACK_THRESHOLD); + await vi.advanceTimersByTimeAsync(0); + expect(acks()).toEqual([{ sessionId: "s1", bytes: OUTPUT_ACK_THRESHOLD }]); + + // No further output arrives, so only a retry can hand the credit back. + fail = false; + await vi.advanceTimersByTimeAsync(2000); + expect(acks()).toEqual([ + { sessionId: "s1", bytes: OUTPUT_ACK_THRESHOLD }, + { sessionId: "s1", bytes: OUTPUT_ACK_THRESHOLD }, + ]); + }); + + it("stops retrying once the stream has ended", async () => { + setInvokeHandler("session_output_ack", () => { + throw new Error("session closed"); + }); + const { deliver } = open(); + deliver(OUTPUT_ACK_THRESHOLD); + deliver(0); + await vi.advanceTimersByTimeAsync(10_000); + expect(acks()).toHaveLength(1); + }); + it("acknowledges each session's output under its own id", () => { const first = open("first"); const second = open("second"); diff --git a/desktop/src/api.ts b/desktop/src/api.ts index 5e2b301e..8e11c389 100644 --- a/desktop/src/api.ts +++ b/desktop/src/api.ts @@ -70,6 +70,8 @@ export type SessionOutput = Channel; export const OUTPUT_ACK_THRESHOLD = 256 * 1024; /** Longest received output waits to be acknowledged below the threshold. */ export const OUTPUT_ACK_DELAY_MS = 30; +/** How many times one failed acknowledgement is retried before giving up. */ +export const OUTPUT_ACK_MAX_RETRIES = 8; /** * Reports output bytes as received. An SSH session stops reading from its @@ -105,6 +107,8 @@ export function sessionOutput( ): SessionOutput { const channel: SessionOutput = new Channel(); let pending = 0; + let failures = 0; + let ended = false; let timer: ReturnType | null = null; const stopTimer = () => { if (timer !== null) clearTimeout(timer); @@ -114,14 +118,29 @@ export function sessionOutput( stopTimer(); const bytes = pending; pending = 0; - // A lost acknowledgement leaves the session paused at worst; it is not - // something the pane can act on, so it must not surface as an unhandled - // rejection. - if (bytes > 0) sessionOutputAck(sessionId, bytes).catch(() => {}); + if (bytes === 0) return; + sessionOutputAck(sessionId, bytes).then( + () => { + failures = 0; + }, + () => { + // Credit that never reaches the backend is never handed back, and a + // reader that has run out waits for an acknowledgement no further + // output will trigger. The bytes go back into the count and are + // retried, backing off, until the stream ends. + if (ended) return; + pending += bytes; + failures += 1; + if (failures > OUTPUT_ACK_MAX_RETRIES) return; + stopTimer(); + timer = setTimeout(acknowledge, Math.min(OUTPUT_ACK_DELAY_MS * 2 ** failures, 2000)); + }, + ); }; channel.onmessage = (message) => { const bytes = new Uint8Array(message); if (bytes.length === 0) { + ended = true; stopTimer(); pending = 0; onEnd(); From 09e30729230d050d953180aa440daf04c63776ee Mon Sep 17 00:00:00 2001 From: Azer Sadykhzadeh Date: Sat, 17 Oct 2026 08:37:54 +0000 Subject: [PATCH 4/5] Restore the SFTP close signal and the spend ordering, and retry acknowledgements for as long as the stream is open The previous commit picked up stale copies of sftp.rs and session.rs, which undid the close signal on single SFTP requests and moved the credit spend back behind the batch hand-off. A capped retry could also give up during a long outage and leave the session paused for good. --- core/src/session.rs | 5 +- core/src/sftp.rs | 99 +++++++++++++++++---------- desktop/src/api.sessionOutput.test.ts | 23 ++++++- desktop/src/api.ts | 11 +-- 4 files changed, 93 insertions(+), 45 deletions(-) diff --git a/core/src/session.rs b/core/src/session.rs index dbcca45f..ff104c7f 100644 --- a/core/src/session.rs +++ b/core/src/session.rs @@ -282,8 +282,11 @@ impl SessionManager { let flush = |batcher: &mut OutputBatcher| { if !batcher.is_empty() { let batch = batcher.batch(); - data_cb(&sid, batch); + // Spent before the batch is handed over, so an + // acknowledgement can never arrive ahead of the spend + // it pays back. credit.spend(batch.len()); + data_cb(&sid, batch); batcher.mark_flushed(Instant::now()); } }; diff --git a/core/src/sftp.rs b/core/src/sftp.rs index 228ce79c..bda7c62b 100644 --- a/core/src/sftp.rs +++ b/core/src/sftp.rs @@ -48,9 +48,10 @@ pub struct SftpManager { struct SftpConnection { handle: Handle, sftp: SftpSession, - /// Flipped once the connection is closed. Listings and transfers watch it - /// and stop, so one that is stuck behind a peer that stopped reading, or - /// simply still running, cannot outlive the close. + /// Flipped once the connection is closed. Every operation watches it and + /// stops, so one that is stuck behind a peer that stopped reading, or + /// simply still running, cannot outlive the close or report success + /// after it. closed: tokio::sync::watch::Sender, } @@ -69,8 +70,8 @@ impl SftpConnection { } } - /// Stop every listing and transfer on this connection, end the SFTP - /// channel, then the connection. Bounded, so an unresponsive peer cannot + /// Stop every request, listing and transfer on this connection, end the + /// SFTP channel, then the connection. Bounded, so an unresponsive peer cannot /// hold the caller. async fn close(&self) { self.closed.send_replace(true); @@ -224,10 +225,13 @@ impl SftpManager { pub async fn canonicalize(&self, session_id: &str, path: &str) -> Result { let conn = self.get(session_id).await?; - conn.sftp - .canonicalize(path) - .await - .map_err(|e| CoreError::Ssh(format!("canonicalize: {e}"))) + conn.until_closed(async { + conn.sftp + .canonicalize(path) + .await + .map_err(|e| CoreError::Ssh(format!("canonicalize: {e}"))) + }) + .await } /// Read a file's contents into a byte vector. @@ -237,10 +241,13 @@ impl SftpManager { pub async fn read_file(&self, session_id: &str, path: &str) -> Result> { let conn = self.get(session_id).await?; - conn.sftp - .read(path) - .await - .map_err(|e| CoreError::Ssh(format!("read file: {e}"))) + conn.until_closed(async { + conn.sftp + .read(path) + .await + .map_err(|e| CoreError::Ssh(format!("read file: {e}"))) + }) + .await } /// Write data to a remote file (creates or overwrites). @@ -250,10 +257,13 @@ impl SftpManager { pub async fn write_file(&self, session_id: &str, path: &str, data: &[u8]) -> Result<()> { let conn = self.get(session_id).await?; - conn.sftp - .write(path, data) - .await - .map_err(|e| CoreError::Ssh(format!("write file: {e}"))) + conn.until_closed(async { + conn.sftp + .write(path, data) + .await + .map_err(|e| CoreError::Ssh(format!("write file: {e}"))) + }) + .await } /// Stream a remote file directly to a local path. An existing local file is @@ -326,30 +336,39 @@ impl SftpManager { pub async fn create_dir(&self, session_id: &str, path: &str) -> Result<()> { let conn = self.get(session_id).await?; - conn.sftp - .create_dir(path) - .await - .map_err(|e| CoreError::Ssh(format!("mkdir: {e}"))) + conn.until_closed(async { + conn.sftp + .create_dir(path) + .await + .map_err(|e| CoreError::Ssh(format!("mkdir: {e}"))) + }) + .await } /// Remove a file. pub async fn remove_file(&self, session_id: &str, path: &str) -> Result<()> { let conn = self.get(session_id).await?; - conn.sftp - .remove_file(path) - .await - .map_err(|e| CoreError::Ssh(format!("remove file: {e}"))) + conn.until_closed(async { + conn.sftp + .remove_file(path) + .await + .map_err(|e| CoreError::Ssh(format!("remove file: {e}"))) + }) + .await } /// Remove a directory. pub async fn remove_dir(&self, session_id: &str, path: &str) -> Result<()> { let conn = self.get(session_id).await?; - conn.sftp - .remove_dir(path) - .await - .map_err(|e| CoreError::Ssh(format!("remove dir: {e}"))) + conn.until_closed(async { + conn.sftp + .remove_dir(path) + .await + .map_err(|e| CoreError::Ssh(format!("remove dir: {e}"))) + }) + .await } /// Rename a file or directory. @@ -361,10 +380,13 @@ impl SftpManager { ) -> Result<()> { let conn = self.get(session_id).await?; - conn.sftp - .rename(old_path, new_path) - .await - .map_err(|e| CoreError::Ssh(format!("rename: {e}"))) + conn.until_closed(async { + conn.sftp + .rename(old_path, new_path) + .await + .map_err(|e| CoreError::Ssh(format!("rename: {e}"))) + }) + .await } /// Get file metadata (size, permissions, modified time). @@ -372,10 +394,13 @@ impl SftpManager { let conn = self.get(session_id).await?; let attrs = conn - .sftp - .metadata(path) - .await - .map_err(|e| CoreError::Ssh(format!("stat: {e}")))?; + .until_closed(async { + conn.sftp + .metadata(path) + .await + .map_err(|e| CoreError::Ssh(format!("stat: {e}"))) + }) + .await?; let name = path.rsplit('/').next().unwrap_or(path).to_string(); let modified = attrs diff --git a/desktop/src/api.sessionOutput.test.ts b/desktop/src/api.sessionOutput.test.ts index b5ab7200..0aebae4d 100644 --- a/desktop/src/api.sessionOutput.test.ts +++ b/desktop/src/api.sessionOutput.test.ts @@ -76,6 +76,25 @@ describe("session output acknowledgement", () => { ]); }); + it("keeps retrying through a long outage", async () => { + let fail = true; + setInvokeHandler("session_output_ack", () => { + if (fail) throw new Error("ipc unavailable"); + }); + const { deliver } = open(); + deliver(OUTPUT_ACK_THRESHOLD); + await vi.advanceTimersByTimeAsync(60_000); + const attempts = acks().length; + expect(attempts).toBeGreaterThan(10); + + fail = false; + await vi.advanceTimersByTimeAsync(2000); + expect(acks().length).toBe(attempts + 1); + // Once it went through nothing is left to send. + await vi.advanceTimersByTimeAsync(10_000); + expect(acks().length).toBe(attempts + 1); + }); + it("stops retrying once the stream has ended", async () => { setInvokeHandler("session_output_ack", () => { throw new Error("session closed"); @@ -118,8 +137,8 @@ describe("session output acknowledgement", () => { const { onData, deliver } = open(); deliver(OUTPUT_ACK_THRESHOLD); expect(acks()).toHaveLength(1); - // Let the rejection settle; an unhandled one fails the run. - await vi.runAllTimersAsync(); + // Let the rejection and a few retries settle; an unhandled one fails the run. + await vi.advanceTimersByTimeAsync(5000); deliver(1); expect(onData).toHaveBeenCalledTimes(2); }); diff --git a/desktop/src/api.ts b/desktop/src/api.ts index 8e11c389..9be661d1 100644 --- a/desktop/src/api.ts +++ b/desktop/src/api.ts @@ -70,8 +70,8 @@ export type SessionOutput = Channel; export const OUTPUT_ACK_THRESHOLD = 256 * 1024; /** Longest received output waits to be acknowledged below the threshold. */ export const OUTPUT_ACK_DELAY_MS = 30; -/** How many times one failed acknowledgement is retried before giving up. */ -export const OUTPUT_ACK_MAX_RETRIES = 8; +/** Longest wait between retries of an acknowledgement that failed. */ +export const OUTPUT_ACK_MAX_RETRY_MS = 2000; /** * Reports output bytes as received. An SSH session stops reading from its @@ -127,13 +127,14 @@ export function sessionOutput( // Credit that never reaches the backend is never handed back, and a // reader that has run out waits for an acknowledgement no further // output will trigger. The bytes go back into the count and are - // retried, backing off, until the stream ends. + // retried, backing off to OUTPUT_ACK_MAX_RETRY_MS, for as long as the + // stream is open: giving up at any point would leave the session + // paused for good. if (ended) return; pending += bytes; failures += 1; - if (failures > OUTPUT_ACK_MAX_RETRIES) return; stopTimer(); - timer = setTimeout(acknowledge, Math.min(OUTPUT_ACK_DELAY_MS * 2 ** failures, 2000)); + timer = setTimeout(acknowledge, Math.min(OUTPUT_ACK_DELAY_MS * 2 ** failures, OUTPUT_ACK_MAX_RETRY_MS)); }, ); }; From 25c5fc86f31440f1ece4b878a98e7bf5c1d69581 Mon Sep 17 00:00:00 2001 From: computerbox124 Date: Thu, 12 Nov 2026 01:13:21 +0000 Subject: [PATCH 5/5] Send resizes from their own task so a stall cannot wedge the reader While the reader is stalled on credit, russh's connection loop is blocked handing it data, so nothing queued for the peer goes out. Keystrokes typed during the stall fill that queue, and a resize then left the reader awaiting a send that only the reader itself could unblock: output stayed stopped after the UI caught up, and a close could no longer end the session or free its id. --- core/src/session.rs | 50 ++++--- core/tests/output_flow_control.rs | 214 +++++++++++++++++++++++------- 2 files changed, 199 insertions(+), 65 deletions(-) diff --git a/core/src/session.rs b/core/src/session.rs index ff104c7f..9ee7fb57 100644 --- a/core/src/session.rs +++ b/core/src/session.rs @@ -250,7 +250,7 @@ impl SessionManager { } } - let (resize_tx, mut resize_rx) = mpsc::channel::<(u32, u32)>(32); + let (resize_tx, resize_rx) = mpsc::channel::<(u32, u32)>(32); let credit = Arc::new(OutputCredit::new()); let held = claim.commit(); @@ -267,17 +267,23 @@ impl SessionManager { self.sessions.lock().await.insert(session_id.clone(), session); let sessions = self.sessions.clone(); - // Spawn the reading task let sid = session_id.clone(); let data_cb = self.data_callback.clone(); let close_cb = self.close_callback.clone(); - let mut channel = channel; + let (mut channel, writer) = channel.split(); + + // Resizes go out from a task of their own, never from the reader. A + // send waits on russh's connection loop, and while the reader is stalled + // on credit that loop is blocked handing it data it will not take; a + // reader that awaited the send would never get back to taking it, and + // neither credit nor a close could wake it. Here the send just waits + // until the reader resumes. + tokio::spawn(send_resizes(writer, resize_rx)); tokio::spawn(async move { // Dropped when this task ends, after the close has been announced, // which is the moment the id is free for another session. let held = held; - let mut pending_resize: Option<(u32, u32)> = None; let mut batcher = OutputBatcher::new(Instant::now()); let flush = |batcher: &mut OutputBatcher| { if !batcher.is_empty() { @@ -295,10 +301,10 @@ impl SessionManager { // batcher while another branch's handler mutates it. let flush_at = batcher.deadline(); // While the UI layer is a window behind, the channel is left - // unread, which backs the stall up to the peer. Resizes and a - // batch already due are still handled meanwhile; the end of the - // stream is seen once credit returns or the session is closed, - // which lifts the limit. + // unread, which backs the stall up to the peer. A batch already + // due is still delivered meanwhile; the end of the stream is + // seen once credit returns or the session is closed, which + // lifts the limit. let reading = credit.has_room(); tokio::select! { () = credit.room(), if !reading => {} @@ -329,19 +335,10 @@ impl SessionManager { _ => {} } } - // Once the entry is gone the sender is dropped and `recv` - // answers `None` at once; the pattern disables the branch - // then, so the loop waits on the channel instead of spinning. - Some(size) = resize_rx.recv() => { - pending_resize = Some(size); - } () = sleep_until(flush_at) => { flush(&mut batcher); } } - if let Some((c, r)) = pending_resize.take() { - let _ = channel.window_change(c, r, 0, 0).await; - } } }); @@ -461,6 +458,25 @@ async fn take_if_held( } } +/// Forward a session's terminal sizes to the peer until the session, which +/// holds the only sender, is dropped. +/// +/// Sizes that queue up while a send is waiting collapse into the newest, so a +/// burst of resizes during a stall reaches the peer as the size the terminal +/// ended up at. +async fn send_resizes( + writer: russh::ChannelWriteHalf, + mut sizes: mpsc::Receiver<(u32, u32)>, +) { + while let Some(mut size) = sizes.recv().await { + while let Ok(newer) = sizes.try_recv() { + size = newer; + } + let (cols, rows) = size; + let _ = writer.window_change(cols, rows, 0, 0).await; + } +} + /// Say goodbye to the peer, giving up after [`DISCONNECT_GRACE`]. The /// connection ends either way once the last reference to the handle is gone. async fn disconnect(handle: &Handle) { diff --git a/core/tests/output_flow_control.rs b/core/tests/output_flow_control.rs index d935248e..99913d92 100644 --- a/core/tests/output_flow_control.rs +++ b/core/tests/output_flow_control.rs @@ -4,8 +4,9 @@ //! Runs the SSH session manager against an in-process russh server on loopback //! whose shell writes without pause for as long as the connection will take //! it, and checks that what reaches the data callback stops growing until the -//! UI acknowledges it, that the stall reaches the server, and that a stalled -//! session still closes. +//! UI acknowledges it, that the stall reaches the server, that a stalled +//! session still closes, and that input and resizes sent during a stall do not +//! keep it stalled once the UI catches up. use clavyn_core::host::Host; use clavyn_core::known_hosts::KnownHosts; @@ -25,10 +26,20 @@ const PASSWORD: &str = "correct password"; const CHUNK: usize = 64 * 1024; /// A server whose shell floods its channel, counting what it managed to hand -/// to its connection. -#[derive(Clone)] +/// to its connection and recording what the client sent it. +#[derive(Clone, Default)] struct FloodServer { produced: Arc, + /// Bytes of client input that reached the shell. + typed: Arc, + /// The most recent terminal size the client set. + size: Arc>>, +} + +impl FloodServer { + fn size(&self) -> Option<(u32, u32)> { + *self.size.lock().expect("size") + } } impl server::Handler for FloodServer { @@ -66,9 +77,32 @@ impl server::Handler for FloodServer { }); Ok(()) } + + async fn data( + &mut self, + _channel: ChannelId, + data: &[u8], + _session: &mut Session, + ) -> Result<(), Self::Error> { + self.typed.fetch_add(data.len() as u64, Ordering::SeqCst); + Ok(()) + } + + async fn window_change_request( + &mut self, + _channel: ChannelId, + cols: u32, + rows: u32, + _pix_width: u32, + _pix_height: u32, + _session: &mut Session, + ) -> Result<(), Self::Error> { + *self.size.lock().expect("size") = Some((cols, rows)); + Ok(()) + } } -async fn start_server(produced: Arc) -> u16 { +async fn start_server(handler: FloodServer) -> u16 { let mut rng = getrandom::rand_core::UnwrapErr(getrandom::SysRng); let host_key = PrivateKey::random(&mut rng, Algorithm::Ed25519).expect("host key"); let config = Arc::new(server::Config { @@ -82,9 +116,7 @@ async fn start_server(produced: Arc) -> u16 { tokio::spawn(async move { while let Ok((stream, _)) = listener.accept().await { let config = config.clone(); - let handler = FloodServer { - produced: produced.clone(), - }; + let handler = handler.clone(); tokio::spawn(async move { if let Ok(running) = server::run_stream(config, stream, handler).await { let _ = running.await; @@ -108,34 +140,26 @@ fn host(port: u16) -> Host { .expect("host") } -/// Waits until `counter` holds still across a sampling interval and returns -/// the value it settled on. -async fn settled(counter: &AtomicU64) -> u64 { - let settle = async { - let mut last = counter.load(Ordering::SeqCst); - loop { - tokio::time::sleep(Duration::from_millis(400)).await; - let now = counter.load(Ordering::SeqCst); - if now == last && now > 0 { - return now; - } - last = now; - } - }; - tokio::time::timeout(Duration::from_secs(20), settle) - .await - .expect("output kept growing with nothing acknowledged") +/// An SSH session on a flooding server whose output nothing acknowledges yet. +struct Flooded { + sessions: Arc, + server: FloodServer, + /// Output that reached the data callback. + delivered: Arc, + /// Ids the session manager announced as closed. + closed: tokio::sync::mpsc::UnboundedReceiver, + _dir: tempfile::TempDir, } -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn unacknowledged_output_stalls_the_session_until_it_is_acknowledged() { +/// Connects a session on `session_id` to a fresh flooding server. +async fn flooded_session(session_id: &str) -> Flooded { let dir = tempfile::tempdir().expect("temp dir"); - let produced = Arc::new(AtomicU64::new(0)); - let port = start_server(produced.clone()).await; + let server = FloodServer::default(); + let port = start_server(server.clone()).await; let delivered = Arc::new(AtomicU64::new(0)); - let (closed_tx, mut closed_rx) = tokio::sync::mpsc::unbounded_channel::(); - let sessions = SessionManager::new( + let (closed_tx, closed) = tokio::sync::mpsc::unbounded_channel::(); + let sessions = Arc::new(SessionManager::new( { let delivered = delivered.clone(); Arc::new(move |_, data: &[u8]| { @@ -145,37 +169,83 @@ async fn unacknowledged_output_stalls_the_session_until_it_is_acknowledged() { Arc::new(move |id: &str, _| { let _ = closed_tx.send(id.to_string()); }), - ); + )); let known_hosts = Arc::new(Mutex::new( KnownHosts::load(dir.path().join("known_hosts.json")).expect("known hosts"), )); - let claim = sessions.claim("flood").expect("claim"); + let claim = sessions.claim(session_id).expect("claim"); sessions .create_ssh_session(&claim, &host(port), None, known_hosts, None, None, Some(PASSWORD), 80, 24) .await .expect("ssh session"); drop(claim); + Flooded { + sessions, + server, + delivered, + closed, + _dir: dir, + } +} + +/// Waits until `counter` holds still across a sampling interval and returns +/// the value it settled on. +async fn settled(counter: &AtomicU64) -> u64 { + let settle = async { + let mut last = counter.load(Ordering::SeqCst); + loop { + tokio::time::sleep(Duration::from_millis(400)).await; + let now = counter.load(Ordering::SeqCst); + if now == last && now > 0 { + return now; + } + last = now; + } + }; + tokio::time::timeout(Duration::from_secs(20), settle) + .await + .expect("output kept growing with nothing acknowledged") +} + +/// Closes `session_id` with nothing more acknowledged and expects the close to +/// be announced. +async fn close_and_expect_announcement(flood: &mut Flooded, session_id: &str) { + tokio::time::timeout(Duration::from_secs(10), flood.sessions.close(session_id)) + .await + .expect("close waited on the stalled reader") + .expect("close"); + let closed = tokio::time::timeout(Duration::from_secs(15), flood.closed.recv()) + .await + .expect("a stalled session never announced its close"); + assert_eq!(closed.as_deref(), Some(session_id)); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn unacknowledged_output_stalls_the_session_until_it_is_acknowledged() { + let mut flood = flooded_session("flood").await; + // A due batch is still delivered after the window is spent, and the // batcher can hold one protocol packet past a full batch. let ceiling = OUTPUT_WINDOW + MAX_BATCH as u64 + CHUNK as u64; - let first = settled(&delivered).await; + let first = settled(&flood.delivered).await; assert!(first >= OUTPUT_WINDOW, "stalled at {first} bytes, short of the window"); assert!(first <= ceiling, "{first} bytes delivered past a window of {OUTPUT_WINDOW}"); // The stall reaches the peer: it stops getting anything onto the // connection rather than having it buffered on our side without limit. - let stalled_at = settled(&produced).await; + let produced = &flood.server.produced; + let stalled_at = settled(produced).await; tokio::time::sleep(Duration::from_millis(500)).await; assert_eq!(produced.load(Ordering::SeqCst), stalled_at, "the server never stalled"); - assert_eq!(delivered.load(Ordering::SeqCst), first); + assert_eq!(flood.delivered.load(Ordering::SeqCst), first); // Acknowledging what arrived lets about one more window through, and no // more than that. - sessions.ack("flood", first).await; - let second = settled(&delivered).await; + flood.sessions.ack("flood", first).await; + let second = settled(&flood.delivered).await; assert!(second > first, "acknowledged output did not resume the session"); assert!( second - first <= ceiling, @@ -185,16 +255,64 @@ async fn unacknowledged_output_stalls_the_session_until_it_is_acknowledged() { // A stalled session still closes, and announces it, with nothing more // acknowledged. - tokio::time::timeout(Duration::from_secs(10), sessions.close("flood")) - .await - .expect("close waited on the stalled reader") - .expect("close"); - let closed = tokio::time::timeout(Duration::from_secs(15), closed_rx.recv()) - .await - .expect("a stalled session never announced its close"); - assert_eq!(closed.as_deref(), Some("flood")); + close_and_expect_announcement(&mut flood, "flood").await; // Acknowledgements trailing a closed session are ignored. - sessions.ack("flood", 1).await; - sessions.ack("never-existed", 1).await; + flood.sessions.ack("flood", 1).await; + flood.sessions.ack("never-existed", 1).await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn input_and_a_resize_sent_during_a_stall_do_not_keep_the_session_stalled() { + let mut flood = flooded_session("resize").await; + let first = settled(&flood.delivered).await; + assert!(first >= OUTPUT_WINDOW, "stalled at {first} bytes, short of the window"); + + // While the reader is stalled, russh's connection loop is blocked handing + // it output, so nothing queued for the peer goes out. Enough keystrokes + // fill that outbound queue, and a resize sent after them has to wait for + // the reader to resume. + const KEYS: u64 = 16; + for _ in 0..KEYS { + let sessions = flood.sessions.clone(); + tokio::spawn(async move { + let _ = sessions.write("resize", b"q").await; + }); + } + tokio::time::sleep(Duration::from_millis(300)).await; + tokio::time::timeout(Duration::from_secs(5), flood.sessions.resize("resize", 100, 30)) + .await + .expect("the resize request waited on the stalled connection") + .expect("resize"); + tokio::time::sleep(Duration::from_millis(300)).await; + + // A UI that catches up acknowledges what it has received, as often as it + // receives it. The session must then resume, and the input and the resize + // must reach the peer. + let caught_up = async { + let mut acknowledged = 0; + loop { + let delivered = flood.delivered.load(Ordering::SeqCst); + if delivered > first + && flood.server.typed.load(Ordering::SeqCst) == KEYS + && flood.server.size() == Some((100, 30)) + { + return; + } + flood.sessions.ack("resize", delivered - acknowledged).await; + acknowledged = delivered; + tokio::time::sleep(Duration::from_millis(50)).await; + } + }; + if tokio::time::timeout(Duration::from_secs(15), caught_up).await.is_err() { + panic!( + "the session stayed stalled after the UI caught up: {} bytes delivered (stalled at \ + {first}), {} of {KEYS} keystrokes and size {:?} reached the server", + flood.delivered.load(Ordering::SeqCst), + flood.server.typed.load(Ordering::SeqCst), + flood.server.size(), + ); + } + + close_and_expect_announcement(&mut flood, "resize").await; }