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..ff104c7f 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,12 @@ 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(); + // 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()); } }; @@ -279,8 +294,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 +395,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 +406,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..0aebae4d --- /dev/null +++ b/desktop/src/api.sessionOutput.test.ts @@ -0,0 +1,145 @@ +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("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("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"); + }); + 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"); + 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 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 ae31007d..9be661d1 100644 --- a/desktop/src/api.ts +++ b/desktop/src/api.ts @@ -66,6 +66,22 @@ 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; +/** 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 + * 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 +91,66 @@ 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 failures = 0; + let ended = false; + let timer: ReturnType | null = null; + const stopTimer = () => { + if (timer !== null) clearTimeout(timer); + timer = null; + }; + const acknowledge = () => { + stopTimer(); + const bytes = pending; + pending = 0; + 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 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; + stopTimer(); + timer = setTimeout(acknowledge, Math.min(OUTPUT_ACK_DELAY_MS * 2 ** failures, OUTPUT_ACK_MAX_RETRY_MS)); + }, + ); + }; channel.onmessage = (message) => { const bytes = new Uint8Array(message); - if (bytes.length === 0) onEnd(); - else onData(bytes); + if (bytes.length === 0) { + ended = true; + 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);