diff --git a/src-tauri/src/modules/pty/mod.rs b/src-tauri/src/modules/pty/mod.rs index f38504e31..09726d688 100644 --- a/src-tauri/src/modules/pty/mod.rs +++ b/src-tauri/src/modules/pty/mod.rs @@ -4,7 +4,6 @@ mod session; pub(crate) mod shell_init; use std::collections::HashMap; -use std::io::Write; use std::sync::atomic::{AtomicU32, Ordering}; use std::sync::{Arc, RwLock}; use std::thread; @@ -93,11 +92,15 @@ pub async fn pty_open( } // Input is the latency-critical path: raw body + id header skips JSON -// serialization of every keystroke on both sides of the IPC boundary. +// serialization of every keystroke on both sides of the IPC boundary. Bytes +// are handed to the session's writer thread via a channel, so arrival order +// is preserved across a burst of keystrokes (the writer thread serializes the +// actual write_all calls). The queue is byte-bounded: past WRITE_QUEUE_CAP we +// return a backpressure error instead of letting a stuck PTY grow memory. #[tauri::command] -pub fn pty_write( - state: tauri::State, - request: tauri::ipc::Request, +pub async fn pty_write( + state: tauri::State<'_, PtyState>, + request: tauri::ipc::Request<'_>, ) -> Result<(), String> { let id: u32 = request .headers() @@ -108,6 +111,8 @@ pub fn pty_write( let tauri::ipc::InvokeBody::Raw(bytes) = request.body() else { return Err("pty_write: expected raw body".to_string()); }; + let bytes = bytes.clone(); + let len = bytes.len(); let session = state .sessions .read() @@ -118,19 +123,22 @@ pub fn pty_write( log::warn!("pty_write: unknown id={id}"); "no session".to_string() })?; - // Bind to a local so the MutexGuard temporary drops before `session` — - // see rustc note on tail-expression temporary drop order. - let result = session - .writer - .lock() - .unwrap() - .write_all(bytes) - .map_err(|e| { - // EPIPE is expected if the child already exited. - log::debug!("pty_write id={id} failed: {e}"); - e.to_string() - }); - result + // Reserve queue capacity first so two racing writers can't both sneak past + // the cap. Rolled back on send failure (writer thread died -> channel + // closed) so a dead session reports "no session" rather than backpressure. + let prev = session.queued_bytes.fetch_add(len, Ordering::Relaxed); + if prev + len > session::WRITE_QUEUE_CAP { + session.queued_bytes.fetch_sub(len, Ordering::Relaxed); + return Err("pty_write: input backlog full".to_string()); + } + match session.write_tx.send(bytes) { + Ok(()) => Ok(()), + Err(_) => { + session.queued_bytes.fetch_sub(len, Ordering::Relaxed); + log::debug!("pty_write id={id}: writer channel closed"); + Err("no session".to_string()) + } + } } #[tauri::command] diff --git a/src-tauri/src/modules/pty/session.rs b/src-tauri/src/modules/pty/session.rs index 792a98844..41ef1a5a8 100644 --- a/src-tauri/src/modules/pty/session.rs +++ b/src-tauri/src/modules/pty/session.rs @@ -1,5 +1,6 @@ use std::io::{Read, Write}; -use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use std::sync::mpsc::{self, Sender}; use std::sync::{Arc, Condvar, Mutex}; use std::thread; use std::time::{Duration, Instant}; @@ -29,25 +30,40 @@ const MAX_PENDING: usize = 4 * 1024 * 1024; // we're forced to discard backlog. const OVERFLOW_NOTICE: &[u8] = b"\x1bc\x1b[2m[terax: dropped output due to backpressure]\x1b[0m\r\n"; +// Cap on bytes queued for the PTY input writer. A stuck PTY (child hung, pipe +// full) would otherwise let pty_write accumulate unbounded Vecs in memory; +// once exceeded, pty_write returns a backpressure error instead of queuing. +// 16 MiB is far above any realistic keystroke burst or paste. +pub(super) const WRITE_QUEUE_CAP: usize = 16 * 1024 * 1024; pub struct Session { // Field drop order is intentional. Rust drops fields top-to-bottom: - // 1. `_job` — on Windows, closing the Job HANDLE fires + // 1. `_job`: on Windows, closing the Job HANDLE fires // KILL_ON_JOB_CLOSE, terminating the pwsh tree before the master // pipe drops. Without this, ClosePseudoConsole in `master`'s Drop // can block waiting for conhost to drain pending output, freezing // the Tauri worker thread that triggered the close. - // 2. `killer` — best-effort kill (redundant on Windows once Job + // 2. `killer`: best-effort kill (redundant on Windows once Job // closed, but harmless and required on Unix where there is no Job). - // 3. `writer` — closes the input side of the master pipe. - // 4. `master` — last; ClosePseudoConsole on Windows. By now the child + // 3. `write_tx`: drops the last channel sender, so the dedicated + // writer thread drains and exits, releasing the input-side writer + // clone it holds. (The reader thread's writer clone, used for DA + // replies, is released when the killed child's EOF unwinds it.) + // 4. `master`: last; ClosePseudoConsole on Windows. By now the child // is dead and conhost has nothing left to drain. #[cfg(windows)] _job: Option, /// PID of the shell process. 0 means unknown; callers must skip checks when 0. pub shell_pid: u32, pub killer: Mutex>, - pub writer: Arc>>, + // Input bytes are queued here and drained by a single dedicated writer + // thread, so a burst of pty_write calls stays in arrival order rather than + // racing for the writer lock from independent spawn_blocking jobs. + pub write_tx: Sender>, + // Bytes currently queued in write_tx but not yet written. Bounded by + // WRITE_QUEUE_CAP: pty_write refuses to enqueue past the cap so a stuck + // PTY can't grow memory unbounded. + pub queued_bytes: Arc, pub master: Mutex>, // Set by the waiter once the child exits, so pty_open can reap a shell // that died before it was registered. @@ -137,6 +153,36 @@ pub fn spawn( )); guard.disarm(); + // Dedicated writer thread: serializes all input writes in arrival order so + // concurrent pty_write calls can't reach the writer lock out of order and + // scramble a fast burst of keystrokes. The channel is byte-bounded (see + // queued_bytes / WRITE_QUEUE_CAP); pty_write applies backpressure instead + // of letting a stuck PTY accumulate unbounded input. The thread exits when + // the last sender drops (session close) or a write/lock fails: breaking + // drops the receiver so later sends return a channel-closed error rather + // than silently discarding input. + let (write_tx, write_rx) = mpsc::channel::>(); + let writer_for_writer = writer.clone(); + let queued_bytes = Arc::new(AtomicUsize::new(0)); + let queued_bytes_writer = queued_bytes.clone(); + thread::Builder::new() + .name("terax-pty-writer".into()) + .spawn(move || { + for bytes in write_rx { + let len = bytes.len(); + let ok = match writer_for_writer.lock() { + Ok(mut w) => w.write_all(&bytes).is_ok(), + Err(_) => false, + }; + queued_bytes_writer.fetch_sub(len, Ordering::Relaxed); + if !ok { + log::debug!("pty writer thread stopping after write/lock failure"); + break; + } + } + }) + .expect("spawn pty writer thread"); + let shell_pid = child.process_id().unwrap_or(0); #[cfg(windows)] @@ -158,7 +204,8 @@ pub fn spawn( _job: job, shell_pid, killer: Mutex::new(killer), - writer: writer.clone(), + write_tx, + queued_bytes, master: Mutex::new(pair.master), exited: exited.clone(), }); @@ -337,13 +384,15 @@ mod tests { drop(pair.slave); let killer = child.clone_killer(); - let writer: Arc>> = - Arc::new(Mutex::new(pair.master.take_writer().expect("writer"))); + // take_writer so the writer side is owned and dropped with the test, + // but the Session no longer stores it directly. + drop(pair.master.take_writer().expect("writer")); let session = Arc::new(Session { shell_pid: child.process_id().unwrap_or(0), killer: Mutex::new(killer), - writer, + write_tx: mpsc::channel().0, + queued_bytes: Arc::new(AtomicUsize::new(0)), master: Mutex::new(pair.master), exited: Arc::new(AtomicBool::new(false)), }); @@ -386,13 +435,13 @@ mod tests { let _ = child.wait(); let killer = child.clone_killer(); - let writer: Arc>> = - Arc::new(Mutex::new(pair.master.take_writer().expect("writer"))); + drop(pair.master.take_writer().expect("writer")); let session = Arc::new(Session { shell_pid: 0, killer: Mutex::new(killer), - writer, + write_tx: mpsc::channel().0, + queued_bytes: Arc::new(AtomicUsize::new(0)), master: Mutex::new(pair.master), exited: Arc::new(AtomicBool::new(false)), }); diff --git a/src/modules/terminal/lib/rendererPool.ts b/src/modules/terminal/lib/rendererPool.ts index 81c7d098a..99adc6d29 100644 --- a/src/modules/terminal/lib/rendererPool.ts +++ b/src/modules/terminal/lib/rendererPool.ts @@ -24,6 +24,19 @@ const FIT_DEBOUNCE_MS = 8; const PTY_RESIZE_DEBOUNCE_MS = 256; const SNAPSHOT_SCROLLBACK_CAP = 5_000; +// Dedup window for self-sent keyCode-229 chars vs xterm's own onData. macOS +// WKWebView fires both our self-send and a textarea-input onData for the same +// keystroke; without coordination each keystroke doubles. +const KEY_229_DEDUP_MS = 200; +// Backstop cap on a pending queue. The 200ms window already bounds real +// growth; this just guards a runaway (e.g. a misbehaving IME flood). +const DEDUP_QUEUE_MAX = 64; +type PendingKey = { char: string; at: number }; + +function prunePending(q: PendingKey[], now: number): void { + while (q.length > 0 && now - q[0].at >= KEY_229_DEDUP_MS) q.shift(); +} + export type SlotAdapter = { resolveLeaf(leafId: number): LeafBridge | null; evictLeaf(leafId: number): void; @@ -70,6 +83,14 @@ export type Slot = { lastW: number; lastH: number; lastUsedAt: number; + // FIFOs of pending keyCode-229 keystrokes awaiting their counterpart, in + // both orderings: selfSentQueue holds chars we forwarded directly that xterm + // may still echo via onData; onDataQueue holds chars xterm already forwarded + // that a trailing keydown(kc=229) must not re-send. A single slot gets + // overwritten under fast typing (cd -> ccd), so we keep an ordered queue per + // direction and consume the oldest matching entry. + selfSentQueue: PendingKey[]; + onDataQueue: PendingKey[]; }; const slots: Slot[] = []; @@ -237,17 +258,55 @@ function createSlot(): Slot { lastW: 0, lastH: 0, lastUsedAt: 0, + selfSentQueue: [], + onDataQueue: [], }; term.attachCustomKeyEventHandler((event) => { // During IME composition the browser is assembling a multi-keystroke - // character (Chinese pinyin → hanzi, Korean jamo → syllable, etc.). - // Raw keydown events — including the Enter that commits a candidate — - // must NOT be forwarded to the PTY; xterm will receive the final - // composed string through its own compositionend handler instead. - // keyCode 229 ("Process") is what Chromium reports for every key - // pressed inside an active IME session when isComposing is not yet set. - if (event.isComposing || event.keyCode === 229) return false; + // character (Chinese pinyin -> hanzi, Korean jamo -> syllable, etc.). + // Raw keydown events -- including the Enter that commits a candidate -- + // must NOT be forwarded to the PTY; xterm receives the final composed + // string through its own compositionend handler instead. + if (event.isComposing) return false; + + // macOS WKWebView tags ordinary fast typing as keyCode 229; xterm then + // reads chars from textarea input events, which coalesce under fast typing + // and drop keys. Forward kc=229 chars ourselves and dedup with onData. + if ( + event.type === "keydown" && + event.keyCode === 229 && + event.key.length === 1 + ) { + const leafId229 = slot.currentLeafId; + if (leafId229 !== null) { + const bridge229 = adapter?.resolveLeaf(leafId229); + if (bridge229) { + const now = Date.now(); + // xterm's textarea-input onData may have already fired for this + // keystroke just before keydown; if so, the char is already on its + // way and self-sending would double it. Consume the oldest matching + // pending onData entry instead of self-sending. + prunePending(slot.onDataQueue, now); + const odIdx = slot.onDataQueue.findIndex( + (e) => e.char === event.key, + ); + if (odIdx >= 0) { + slot.onDataQueue.splice(odIdx, 1); + event.preventDefault(); + return false; + } + // Otherwise self-send and enqueue so a trailing onData is dropped. + slot.selfSentQueue.push({ char: event.key, at: now }); + if (slot.selfSentQueue.length > DEDUP_QUEUE_MAX) { + slot.selfSentQueue.shift(); + } + bridge229.writeToPty(event.key); + event.preventDefault(); + return false; + } + } + } const leafId = slot.currentLeafId; if (leafId === null) return false; @@ -301,6 +360,25 @@ function createSlot(): Slot { term.onData((data) => { const leafId = slot.currentLeafId; + const now = Date.now(); + if (data.length === 1) { + // Drop xterm's own onData when it duplicates a char we already self-sent + // for a keyCode-229 keystroke within the dedup window. Consume the oldest + // matching pending entry so a fast burst (cd) dedups each char, not just + // the most recent one. + prunePending(slot.selfSentQueue, now); + const ssIdx = slot.selfSentQueue.findIndex((e) => e.char === data); + if (ssIdx >= 0) { + slot.selfSentQueue.splice(ssIdx, 1); + return; + } + // Record this onData so a trailing keydown(kc=229) for the same char + // knows not to self-send (onData-arrived-first ordering). + slot.onDataQueue.push({ char: data, at: now }); + if (slot.onDataQueue.length > DEDUP_QUEUE_MAX) { + slot.onDataQueue.shift(); + } + } if (leafId === null) return; adapter?.resolveLeaf(leafId)?.writeToPty(data); });