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>
This commit is contained in:
@ -163,11 +163,11 @@ pub async fn open_terminal(
|
||||
let session_id = output.session.id;
|
||||
|
||||
// (2) Register the xterm output channel for this session.
|
||||
state.pty_bridge.register(session_id, on_output);
|
||||
let gen = state.pty_bridge.register(session_id, on_output);
|
||||
|
||||
// (3) Subscribe to the PTY's byte stream and pump it to the channel. The
|
||||
// stream is a blocking iterator, so it runs on a dedicated OS thread; it
|
||||
// ends when the PTY hits EOF (process exit) or the bridge channel is gone.
|
||||
// ends when the PTY hits EOF (process exit) or this attach is superseded.
|
||||
let handle = PtyHandle { session_id };
|
||||
match state.pty_port.subscribe_output(&handle) {
|
||||
Ok(stream) => {
|
||||
@ -178,8 +178,9 @@ pub async fn open_terminal(
|
||||
break;
|
||||
}
|
||||
}
|
||||
// Stream ended (process exited): drop the channel registration.
|
||||
bridge.unregister(&session_id);
|
||||
// Stream ended: drop the channel only if still ours (a re-attach
|
||||
// may have superseded this generation — don't tear down its channel).
|
||||
bridge.unregister_if(&session_id, gen);
|
||||
});
|
||||
}
|
||||
Err(e) => {
|
||||
@ -283,10 +284,12 @@ pub fn reattach_terminal(
|
||||
.map_err(|e| ErrorDto::from(AppError::from(e)))?;
|
||||
|
||||
// (2) Register the new output channel for this session, replacing any stale
|
||||
// one from a previous attach.
|
||||
state.pty_bridge.register(sid, on_output);
|
||||
// one from a previous attach (and bumping the generation).
|
||||
let gen = state.pty_bridge.register(sid, on_output);
|
||||
|
||||
// (3) Subscribe afresh to the live byte stream and pump it to the channel.
|
||||
// The fresh subscription supersedes the previous attach's, so its pump thread
|
||||
// ends and stops double-delivering this session's bytes.
|
||||
match state.pty_port.subscribe_output(&handle) {
|
||||
Ok(stream) => {
|
||||
let bridge: std::sync::Arc<PtyBridge> = std::sync::Arc::clone(&state.pty_bridge);
|
||||
@ -296,7 +299,7 @@ pub fn reattach_terminal(
|
||||
break;
|
||||
}
|
||||
}
|
||||
bridge.unregister(&sid);
|
||||
bridge.unregister_if(&sid, gen);
|
||||
});
|
||||
}
|
||||
Err(e) => {
|
||||
@ -767,11 +770,11 @@ pub async fn launch_agent(
|
||||
let session_id = output.session.id;
|
||||
|
||||
// Register the xterm output channel for this session.
|
||||
state.pty_bridge.register(session_id, on_output);
|
||||
let gen = state.pty_bridge.register(session_id, on_output);
|
||||
|
||||
// Subscribe to the PTY's byte stream and pump it to the channel.
|
||||
// The stream is a blocking iterator; it runs on a dedicated OS thread and
|
||||
// ends when the PTY hits EOF or the bridge channel is gone.
|
||||
// ends when the PTY hits EOF or this attach is superseded.
|
||||
let handle = PtyHandle { session_id };
|
||||
match state.pty_port.subscribe_output(&handle) {
|
||||
Ok(stream) => {
|
||||
@ -782,7 +785,7 @@ pub async fn launch_agent(
|
||||
break;
|
||||
}
|
||||
}
|
||||
bridge.unregister(&session_id);
|
||||
bridge.unregister_if(&session_id, gen);
|
||||
});
|
||||
}
|
||||
Err(e) => {
|
||||
|
||||
@ -34,7 +34,11 @@ pub type PtyChunk = Vec<u8>;
|
||||
/// Thread-safe; cloned `Arc<PtyBridge>` is held in [`crate::state::AppState`].
|
||||
#[derive(Default)]
|
||||
pub struct PtyBridge {
|
||||
channels: Mutex<HashMap<SessionId, Channel<PtyChunk>>>,
|
||||
/// Per session: a monotonically-increasing **generation** plus the current
|
||||
/// output channel. The generation distinguishes successive (re-)attaches so a
|
||||
/// superseded pump thread can't tear down the channel of the attach that
|
||||
/// replaced it (see [`PtyBridge::unregister_if`]).
|
||||
channels: Mutex<HashMap<SessionId, (u64, Channel<PtyChunk>)>>,
|
||||
}
|
||||
|
||||
impl PtyBridge {
|
||||
@ -46,21 +50,39 @@ impl PtyBridge {
|
||||
}
|
||||
}
|
||||
|
||||
/// Registers the output channel for a session (called when a terminal is
|
||||
/// opened, from a `#[tauri::command]` that receives the `Channel` argument).
|
||||
pub fn register(&self, session: SessionId, channel: Channel<PtyChunk>) {
|
||||
/// Registers (or replaces) the output channel for a session and returns the
|
||||
/// **generation** of this registration. Each call for a session bumps the
|
||||
/// generation, so the caller's pump thread can later tear down *only its own*
|
||||
/// registration via [`unregister_if`](Self::unregister_if).
|
||||
pub fn register(&self, session: SessionId, channel: Channel<PtyChunk>) -> u64 {
|
||||
if let Ok(mut map) = self.channels.lock() {
|
||||
map.insert(session, channel);
|
||||
let gen = map.get(&session).map_or(0, |(g, _)| g.wrapping_add(1));
|
||||
map.insert(session, (gen, channel));
|
||||
gen
|
||||
} else {
|
||||
0
|
||||
}
|
||||
}
|
||||
|
||||
/// Removes a session's channel (terminal closed).
|
||||
/// Removes a session's channel unconditionally (terminal explicitly closed).
|
||||
pub fn unregister(&self, session: &SessionId) {
|
||||
if let Ok(mut map) = self.channels.lock() {
|
||||
map.remove(session);
|
||||
}
|
||||
}
|
||||
|
||||
/// Removes a session's channel **only if** `gen` is still the current
|
||||
/// generation. A pump thread calls this when its output stream ends: if the
|
||||
/// session has since been re-attached (newer generation), this is a no-op, so
|
||||
/// the dying thread never unregisters the live channel that superseded it.
|
||||
pub fn unregister_if(&self, session: &SessionId, gen: u64) {
|
||||
if let Ok(mut map) = self.channels.lock() {
|
||||
if matches!(map.get(session), Some((g, _)) if *g == gen) {
|
||||
map.remove(session);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Forwards a chunk of output bytes to a session's channel.
|
||||
///
|
||||
/// Returns `true` if the chunk was delivered, `false` if no channel is
|
||||
@ -71,7 +93,7 @@ impl PtyBridge {
|
||||
return false;
|
||||
};
|
||||
match map.get(session) {
|
||||
Some(channel) => channel.send(chunk).is_ok(),
|
||||
Some((_, channel)) => channel.send(chunk).is_ok(),
|
||||
None => false,
|
||||
}
|
||||
}
|
||||
|
||||
@ -88,3 +88,47 @@ fn register_same_session_twice_replaces_channel() {
|
||||
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]));
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user