Files
IdeA/crates/app-tauri/tests/pty_bridge.rs
Blomios b9fd2fb925 fix(terminals): stop PTY output duplication on re-attach
Each open/reattach spawned a pump thread feeding pty_bridge[session]. The
broadcast hub added a subscriber per attach without dropping the previous one,
and the stale pump thread kept delivering (its send_output still succeeded
against the new channel) — so every byte was delivered twice (more after
further re-attaches). Accumulated on each tab/layout/agent switch; window
resize made it glaring.

- Broadcast::subscribe() is now single-consumer: a new subscription supersedes
  the previous one, ending the old pump thread's stream.
- PtyBridge gains a per-session generation; register() returns it and
  unregister_if(session, gen) only removes when still current, so a dying
  superseded thread can't tear down the channel that replaced it.
- All three attach sites (open/reattach/agent launch) use unregister_if.

Tests: 6 added (generation guard + single-consumer broadcast). Workspace green.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-07 11:52:24 +02:00

135 lines
4.8 KiB
Rust

//! L1 tests for [`PtyBridge`] — the PTY↔Channel registry — exercised with a
//! real [`tauri::ipc::Channel`] built from a capturing closure (no Tauri runtime
//! and no real PTY needed).
use std::sync::{Arc, Mutex};
use tauri::ipc::{Channel, InvokeResponseBody};
use app_tauri_lib::pty::PtyBridge;
use domain::ids::SessionId;
use uuid::Uuid;
/// Builds a `Channel<Vec<u8>>` whose sent chunks are recorded into `sink`.
///
/// `Vec<u8>` is `Serialize`, so chunks arrive as a JSON array string in an
/// `InvokeResponseBody::Json`; we parse them back to bytes for assertions.
fn capturing_channel(sink: Arc<Mutex<Vec<Vec<u8>>>>) -> Channel<Vec<u8>> {
Channel::new(move |body: InvokeResponseBody| {
let bytes: Vec<u8> = match body {
InvokeResponseBody::Json(s) => serde_json::from_str(&s).unwrap(),
InvokeResponseBody::Raw(b) => b,
};
sink.lock().unwrap().push(bytes);
Ok(())
})
}
fn sid() -> SessionId {
SessionId::from_uuid(Uuid::new_v4())
}
#[test]
fn register_increases_active_sessions() {
let bridge = PtyBridge::new();
assert_eq!(bridge.active_sessions(), 0);
let sink = Arc::new(Mutex::new(Vec::new()));
bridge.register(sid(), capturing_channel(sink));
assert_eq!(bridge.active_sessions(), 1);
}
#[test]
fn send_output_delivers_bytes_to_registered_channel() {
let bridge = PtyBridge::new();
let session = sid();
let sink = Arc::new(Mutex::new(Vec::new()));
bridge.register(session, capturing_channel(Arc::clone(&sink)));
let delivered = bridge.send_output(&session, vec![104, 105]);
assert!(delivered, "send_output should return true for a live session");
let captured = sink.lock().unwrap();
assert_eq!(captured.as_slice(), &[vec![104, 105]]);
}
#[test]
fn send_output_to_unknown_session_returns_false() {
let bridge = PtyBridge::new();
assert!(!bridge.send_output(&sid(), vec![0]));
}
#[test]
fn unregister_removes_session_and_stops_delivery() {
let bridge = PtyBridge::new();
let session = sid();
let sink = Arc::new(Mutex::new(Vec::new()));
bridge.register(session, capturing_channel(Arc::clone(&sink)));
assert_eq!(bridge.active_sessions(), 1);
bridge.unregister(&session);
assert_eq!(bridge.active_sessions(), 0);
assert!(!bridge.send_output(&session, vec![1]));
assert!(sink.lock().unwrap().is_empty());
}
#[test]
fn register_same_session_twice_replaces_channel() {
let bridge = PtyBridge::new();
let session = sid();
let first = Arc::new(Mutex::new(Vec::new()));
let second = Arc::new(Mutex::new(Vec::new()));
bridge.register(session, capturing_channel(Arc::clone(&first)));
bridge.register(session, capturing_channel(Arc::clone(&second)));
assert_eq!(bridge.active_sessions(), 1, "same id is replaced, not added");
bridge.send_output(&session, vec![9]);
assert!(first.lock().unwrap().is_empty(), "old channel no longer used");
assert_eq!(second.lock().unwrap().as_slice(), &[vec![9]]);
}
#[test]
fn register_returns_monotonic_generation_per_session() {
let bridge = PtyBridge::new();
let session = sid();
let g0 = bridge.register(session, capturing_channel(Arc::new(Mutex::new(Vec::new()))));
let g1 = bridge.register(session, capturing_channel(Arc::new(Mutex::new(Vec::new()))));
assert_eq!(g0, 0);
assert_eq!(g1, 1, "re-attaching the same session bumps the generation");
}
/// The regression guard for on-resize output duplication: a superseded attach's
/// pump thread (older generation) must NOT tear down the channel of the
/// re-attach that replaced it.
#[test]
fn unregister_if_is_a_noop_for_a_superseded_generation() {
let bridge = PtyBridge::new();
let session = sid();
let old = Arc::new(Mutex::new(Vec::new()));
let new = Arc::new(Mutex::new(Vec::new()));
let old_gen = bridge.register(session, capturing_channel(Arc::clone(&old)));
let _new_gen = bridge.register(session, capturing_channel(Arc::clone(&new)));
// The stale (old) pump thread ends and tries to clean up with its own gen.
bridge.unregister_if(&session, old_gen);
// The current channel survives and still delivers — no duplication, no drop.
assert_eq!(bridge.active_sessions(), 1, "live re-attach must not be removed");
assert!(bridge.send_output(&session, vec![7]));
assert_eq!(new.lock().unwrap().as_slice(), &[vec![7]]);
}
#[test]
fn unregister_if_removes_when_generation_is_current() {
let bridge = PtyBridge::new();
let session = sid();
let gen = bridge.register(session, capturing_channel(Arc::new(Mutex::new(Vec::new()))));
// The current attach's pump thread ends (PTY EOF): it owns the live channel.
bridge.unregister_if(&session, gen);
assert_eq!(bridge.active_sessions(), 0);
assert!(!bridge.send_output(&session, vec![1]));
}