diff --git a/crates/server/src/lib.rs b/crates/server/src/lib.rs index 5a0eb803..83236fd4 100644 --- a/crates/server/src/lib.rs +++ b/crates/server/src/lib.rs @@ -7494,6 +7494,10 @@ struct AppStateInner { /// native YAS session. The delivery loop updates the shared cache first; /// receivers then rebuild their own catalogue without a polling delay. surface_catalogue_updates: watch::Sender, + /// Bumped once per PTY exit, after its output is drained and its exit + /// status collected. Terminal catalogue publication must not wait for the + /// next display frame or catalogue poll. + terminal_exit_updates: watch::Sender, /// Retained shutdown state for every accept loop, including fd-channel. shutdown_started: watch::Sender, /// Broadcast to everything this process hosts alongside the server, so a @@ -8394,6 +8398,9 @@ async fn cleanup_pty_internal(pty_id: u16, generation: Option, state: &AppS // timeout for output that is never coming. let end_seq = pty.driver.cursor_seq().0; pty.journal.note_pty_exit(end_seq); + state + .terminal_exit_updates + .send_modify(|revision| *revision = revision.wrapping_add(1)); } } @@ -8622,6 +8629,7 @@ pub async fn run_hosted(config: Config, hosted: Option) { pty_fds: Arc::new(std::sync::RwLock::new(FxHashMap::default())), delivery_notify: Arc::new(Notify::new()), surface_catalogue_updates: watch::channel(0).0, + terminal_exit_updates: watch::channel(0).0, shutdown_started: watch::channel(false).0, hosted_shutdown: Arc::new(Notify::new()), connections: Arc::new(ConnectionRegistry::default()), @@ -13218,6 +13226,7 @@ mod tests { pty_fds: Arc::new(std::sync::RwLock::new(FxHashMap::default())), delivery_notify: Arc::new(Notify::new()), surface_catalogue_updates: watch::channel(0).0, + terminal_exit_updates: watch::channel(0).0, shutdown_started: watch::channel(false).0, hosted_shutdown: Arc::new(Notify::new()), connections: Arc::new(ConnectionRegistry::default()), @@ -13862,6 +13871,55 @@ mod tests { assert_sync_output_drained(true).await; } + #[cfg(unix)] + #[tokio::test] + async fn terminal_exit_update_follows_drained_output_once() { + let service = process::Server::new(false, true); + let state = process_transport::test_state(service.clone()); + let mut exits = state.terminal_exit_updates.subscribe(); + let terminal = pty::spawn_pty( + "/bin/sh", + "", + 24, + 80, + 1, + "", + pty::ChildSpec { + command: Some("printf 'LAST-OUTPUT'; exit 7"), + ..Default::default() + }, + None, + 100, + state.clone(), + None, + ) + .unwrap(); + state.session.lock().await.ptys.insert(1, terminal); + tokio::time::timeout(Duration::from_secs(5), async { + while !exits.has_changed().unwrap() { + tick(&state).await; + tokio::time::sleep(Duration::from_millis(1)).await; + } + }) + .await + .expect("PTY exit was announced"); + exits.borrow_and_update(); + { + let sess = state.session.lock().await; + let terminal = &sess.ptys[&1]; + assert!(terminal.exited); + assert_eq!(terminal.exit_status, 7); + let text = terminal.driver.seq_text(0, 0, None, 10000).text; + assert!(text.contains("LAST-OUTPUT"), "{text}"); + } + cleanup_pty_internal(1, None, &state).await; + assert!( + !exits.has_changed().unwrap(), + "a second cleanup announced the exit again" + ); + service.shutdown().await; + } + fn test_client() -> ClientState { let (client, _rx) = test_client_with_capacity(0); client diff --git a/crates/server/src/yas.rs b/crates/server/src/yas.rs index 7ae436c8..9374d526 100644 --- a/crates/server/src/yas.rs +++ b/crates/server/src/yas.rs @@ -169,6 +169,7 @@ const TERMINAL_MAX_INFLIGHT_FRAMES: u8 = 3; // fifth of the floor. const TERMINAL_FRAME_CELL_BUCKET: u32 = 8 * 1024; const MAX_SURFACE_OPERATION_REPLAYS: usize = 256; +const MAX_RETIRED_TERMINAL_VIEWS: usize = 256; const MAX_SURFACE_CAPTURE_RETAINED_BYTES: u64 = SERVER_MAX_BUFFERED; const SURFACE_CODEC_SELECTION_TIMEOUT: Duration = Duration::from_secs(30); // Surface frames share the reliable byte stream with latency probes and @@ -2347,6 +2348,16 @@ async fn serve_registered( .map(|state| state.surface_catalogue_updates.subscribe()) }) .flatten(); + let mut terminal_exit_updates = negotiated + .selected + .contains(&family::TERMINAL) + .then(|| { + services + .app_state + .as_ref() + .map(|state| state.terminal_exit_updates.subscribe()) + }) + .flatten(); let (channel, mut channel_rx, mut channel_terminal_rx) = if negotiated.selected.contains(&family::CHANNEL) { match services.app_state.clone() { @@ -2561,6 +2572,19 @@ async fn serve_registered( }, Err(_) => surface_catalogue_updates = None, }, + changed = async { + match terminal_exit_updates.as_mut() { + Some(updates) => updates.changed().await, + None => std::future::pending().await, + } + }, if terminal_exit_updates.is_some() => match changed { + Ok(()) => { + if session.refresh_terminal_catalogue().await { + session.publish_terminal_catalogue(); + } + } + Err(_) => terminal_exit_updates = None, + }, _ = catalogue_tick.tick(), if poll_terminal_catalogue || poll_client_catalogue || poll_desktop_catalogue || poll_media_catalogue => { if poll_terminal_catalogue && session.refresh_terminal_catalogue().await { session.publish_terminal_catalogue(); @@ -3882,6 +3906,16 @@ struct TerminalRuntime { operations: HashMap<[u8; 16], TerminalOperationReplay>, next_view_id: u32, views: HashMap, + /// Views this connection closed or retired, oldest first. The peer may + /// already have sent feedback for one when the removal crosses it. + retired_views: VecDeque, +} + +#[derive(Clone, Copy)] +struct RetiredTerminalView { + view_id: u32, + acknowledged_sequence: u32, + last_sent: u32, } struct TerminalOperationReplay { @@ -4987,6 +5021,7 @@ impl TerminalRuntime { operations: HashMap::new(), next_view_id: 1, views: HashMap::new(), + retired_views: VecDeque::new(), } } @@ -4994,12 +5029,31 @@ impl TerminalRuntime { for _ in 0..u32::MAX { let view_id = self.next_view_id.max(1); self.next_view_id = self.next_view_id.wrapping_add(1).max(1); - if !self.views.contains_key(&view_id) { + if !self.views.contains_key(&view_id) && self.retired_view(view_id).is_none() { return Some(view_id); } } None } + + fn remove_view(&mut self, view_id: u32) -> Option { + let view = self.views.remove(&view_id)?; + if self.retired_views.len() >= MAX_RETIRED_TERMINAL_VIEWS { + self.retired_views.pop_front(); + } + self.retired_views.push_back(RetiredTerminalView { + view_id, + acknowledged_sequence: view.acknowledged_sequence, + last_sent: view.next_sequence.wrapping_sub(1), + }); + Some(view) + } + + fn retired_view(&mut self, view_id: u32) -> Option<&mut RetiredTerminalView> { + self.retired_views + .iter_mut() + .find(|retired| retired.view_id == view_id) + } } async fn terminal_catalogue_snapshot(state: &AppState) -> TerminalCatalogue { @@ -14162,12 +14216,16 @@ impl Session { } yas_wire::schema::terminal::event::INPUT => { let value = yas_terminal::Input::decode(&frame.payload).map_err(|_| ())?; - let pty_id = self.update_terminal_feedback(&value.feedback).await?; + let Some(pty_id) = self.update_terminal_feedback(&value.feedback).await? else { + return Ok(()); + }; self.write_terminal_input(pty_id, &value.data).await } yas_wire::schema::terminal::event::MOUSE => { let value = yas_terminal::Mouse::decode(&frame.payload).map_err(|_| ())?; - let pty_id = self.update_terminal_feedback(&value.feedback).await?; + let Some(pty_id) = self.update_terminal_feedback(&value.feedback).await? else { + return Ok(()); + }; let column = u16::try_from(value.column).map_err(|_| ())?; let row = u16::try_from(value.row).map_err(|_| ())?; self.write_terminal_mouse( @@ -14182,7 +14240,9 @@ impl Session { } yas_wire::schema::terminal::event::WHEEL => { let value = yas_terminal::Wheel::decode(&frame.payload).map_err(|_| ())?; - let pty_id = self.update_terminal_feedback(&value.feedback).await?; + let Some(pty_id) = self.update_terminal_feedback(&value.feedback).await? else { + return Ok(()); + }; if value.dx_32_32 != 0 || value.dy_32_32 == 0 { return Err(()); } @@ -14191,7 +14251,9 @@ impl Session { } yas_wire::schema::terminal::event::WHEEL_AT => { let value = yas_terminal::WheelAt::decode(&frame.payload).map_err(|_| ())?; - let pty_id = self.update_terminal_feedback(&value.feedback).await?; + let Some(pty_id) = self.update_terminal_feedback(&value.feedback).await? else { + return Ok(()); + }; if value.dx_32_32 != 0 || value.dy_32_32 == 0 { return Err(()); } @@ -14209,16 +14271,30 @@ impl Session { } } + /// Applies view feedback and returns the view's PTY, or `None` for a view + /// this connection already retired, whose input is dropped. async fn update_terminal_feedback( &mut self, feedback: &yas_terminal::ViewFeedback, - ) -> Result { + ) -> Result, ()> { self.settle_completed_terminal_frame(feedback.view_id, true) .await?; let (state, pty_id, backend_view, owed_final_guard) = { let terminal = self.native.as_mut().ok_or(())?; let state = terminal.state.clone(); - let view = terminal.views.get_mut(&feedback.view_id).ok_or(())?; + let Some(view) = terminal.views.get_mut(&feedback.view_id) else { + let retired = terminal.retired_view(feedback.view_id).ok_or(())?; + if terminal_feedback_advance( + retired.acknowledged_sequence, + retired.last_sent, + feedback.presented_sequence, + )? + .is_some() + { + retired.acknowledged_sequence = feedback.presented_sequence; + } + return Ok(None); + }; let credit_opened = apply_terminal_feedback(view, feedback)?; let owed_final_guard = (credit_opened && view.frame_owed).then(|| { view.frame_owed = false; @@ -14237,7 +14313,7 @@ impl Session { view.frame_owed = true; view.frame_owed_final_guard = owed_final_guard; } - Ok(pty_id) + Ok(Some(pty_id)) } async fn rearm_terminal_view( @@ -15026,7 +15102,7 @@ impl Session { .collect::>(); view_ids .into_iter() - .filter_map(|view_id| terminal.views.remove(&view_id)) + .filter_map(|view_id| terminal.remove_view(view_id)) .collect::>() }; let backend_views = removed_views @@ -15668,8 +15744,7 @@ impl Session { .native .as_mut() .ok_or(())? - .views - .remove(&request.view_id) + .remove_view(request.view_id) .expect("Terminal view checked"); // CLOSE_VIEW is Control while Terminal frames are Data. Drain this // view's already-queued frames so the scheduler cannot write one @@ -27788,7 +27863,7 @@ impl Session { .collect::>(); retired_view_ids .into_iter() - .filter_map(|view_id| terminal.views.remove(&view_id)) + .filter_map(|view_id| terminal.remove_view(view_id)) .collect() }; let retired_receipts = retired_views @@ -55241,6 +55316,178 @@ mod tests { process_service.shutdown().await; } + #[cfg(unix)] + #[tokio::test(flavor = "multi_thread")] + async fn terminal_feedback_crossing_close_is_ignored_but_unsent_frames_are_rejected() { + let process_service = super::super::process::Server::new(false, true); + let state = super::super::tests::process_transport::test_state(process_service.clone()); + let delivery_state = state.clone(); + let delivery = tokio::spawn(async move { + loop { + tokio::select! { + _ = delivery_state.delivery_notify.notified() => {} + _ = tokio::time::sleep(Duration::from_millis(10)) => {} + } + let _ = super::super::tick(&delivery_state).await; + } + }); + let (mut client, codec, _, task) = + start_registered_session(state.clone(), &[family::TERMINAL]).await; + let mut pending_frames = Vec::new(); + write_request( + &mut client, + &codec, + family::TERMINAL, + yas_wire::schema::terminal::request::CREATE, + 100, + &yas_terminal::Create { + rows: 12, + cols: 40, + operation_id: [0xb0; 16], + launch: yas_terminal::Launch { + command: yas_terminal::Command::Argv(vec![ + b"/bin/sh".to_vec(), + b"-c".to_vec(), + b"exec cat".to_vec(), + ]), + cwd: yas_terminal::Cwd::ServerDefault, + environment_base: yas_terminal::EnvironmentBase::Server, + environment: Vec::new(), + extensions: Extensions::default(), + }, + extensions: Extensions::default(), + }, + ) + .await; + let created = next_terminal_result( + &mut client, + &codec, + yas_wire::schema::terminal::request::CREATE, + 100, + &mut pending_frames, + ) + .await; + assert_eq!(created.status, Status::Ok); + let created = yas_terminal::CreateResult::decode(&created.body).unwrap(); + write_request( + &mut client, + &codec, + family::TERMINAL, + yas_wire::schema::terminal::request::OPEN_VIEW, + 101, + &yas_terminal::OpenView { + terminal_handle: created.terminal_handle, + rows: 12, + cols: 40, + max_fps: 60, + codec_versions: vec![1], + extensions: Extensions::default(), + }, + ) + .await; + let opened = next_terminal_result( + &mut client, + &codec, + yas_wire::schema::terminal::request::OPEN_VIEW, + 101, + &mut pending_frames, + ) + .await; + assert_eq!(opened.status, Status::Ok); + let opened = yas_terminal::OpenViewResult::decode(&opened.body).unwrap(); + let first = + next_terminal_frame_for(&mut client, &codec, opened.view_id, &mut pending_frames).await; + + write_request( + &mut client, + &codec, + family::TERMINAL, + yas_wire::schema::terminal::request::CLOSE, + 102, + &yas_terminal::Close { + terminal_handle: created.terminal_handle, + operation_id: [0xb2; 16], + }, + ) + .await; + assert_eq!( + next_terminal_result( + &mut client, + &codec, + yas_wire::schema::terminal::request::CLOSE, + 102, + &mut pending_frames, + ) + .await + .status, + Status::Ok, + ); + // The client presented the frame before it learned of the close. + let feedback = yas_terminal::ViewFeedback { + view_id: opened.view_id, + presented_sequence: first.frame_sequence, + decoder_queue_depth: 0, + available_frame_slots: 1, + }; + write_event( + &mut client, + &codec, + family::TERMINAL, + yas_wire::schema::terminal::event::FRAME_ACK, + &feedback, + ) + .await; + write_event( + &mut client, + &codec, + family::TERMINAL, + yas_wire::schema::terminal::event::INPUT, + &yas_terminal::Input { + feedback, + data: b"late keystroke".to_vec(), + }, + ) + .await; + write_request( + &mut client, + &codec, + family::TERMINAL, + yas_wire::schema::terminal::request::CLOSE, + 103, + &yas_terminal::Close { + terminal_handle: created.terminal_handle, + operation_id: [0xb3; 16], + }, + ) + .await; + next_terminal_result( + &mut client, + &codec, + yas_wire::schema::terminal::request::CLOSE, + 103, + &mut pending_frames, + ) + .await; + + write_event( + &mut client, + &codec, + family::TERMINAL, + yas_wire::schema::terminal::event::FRAME_ACK, + &yas_terminal::ViewFeedback { + presented_sequence: first.frame_sequence.wrapping_add(1), + ..feedback + }, + ) + .await; + timeout(TEST_TIMEOUT, task) + .await + .expect("feedback for an unsent frame ends the session") + .unwrap(); + delivery.abort(); + process_service.shutdown().await; + } + #[cfg(unix)] #[tokio::test(flavor = "multi_thread")] async fn terminal_two_views_keep_independent_configuration_scroll_and_lifetime() { diff --git a/docs/design/yas.md b/docs/design/yas.md index 711e648c..38b89703 100644 --- a/docs/design/yas.md +++ b/docs/design/yas.md @@ -1874,7 +1874,10 @@ View input and FRAME_ACK share this ten-byte feedback prefix: Feedback is cumulative. Before presenting any frame, `presented_sequence` is `first_sequence - 1` modulo 2^32. Queue depth and available slots are exact -because a view has at most 255 in-flight frames. Stale feedback is ignored. The server sends no more than +because a view has at most 255 in-flight frames. Stale feedback is ignored. +A closed or retired view's feedback can cross its removal on the wire, so the +server keeps a bounded record of recent views and ignores their feedback, and +drops input carried with it, as long as it acknowledges no unsent frame. The server sends no more than `max_inflight_frames` logical frames beyond the highest presented sequence and also obeys the most recent available-slot report. diff --git a/js/core/src/YasNativeWorkspaceConnection.ts b/js/core/src/YasNativeWorkspaceConnection.ts index 28680cf4..8e1507ed 100644 --- a/js/core/src/YasNativeWorkspaceConnection.ts +++ b/js/core/src/YasNativeWorkspaceConnection.ts @@ -999,11 +999,11 @@ export class YasNativeWorkspaceConnection { this.focusedSessionId = sessionId; if (previous) { const view = this.views.get(this.handleForSession(previous)); - if (view) void this.terminal.setFocus(view.view.result.viewId, false); + if (view) this.setViewFocus(view.view.result.viewId, false); } if (sessionId) { const view = this.views.get(this.handleForSession(sessionId)); - if (view) void this.terminal.setFocus(view.view.result.viewId, true); + if (view) this.setViewFocus(view.view.result.viewId, true); } this.refreshSnapshot(); } @@ -3862,7 +3862,19 @@ export class YasNativeWorkspaceConnection { this.configureView(handle, latestSize); } if (this.focusedSessionId === this.sessionId(handle)) - void this.terminal.setFocus(view.result.viewId, true); + this.setViewFocus(view.result.viewId, true); + } + + // Focus follows workspace state, including the fallback chosen when a + // session closes, so the view can already be gone when the request lands. + private setViewFocus(viewId: number, focused: boolean): void { + void (async () => { + try { + await this.terminal.setFocus(viewId, focused); + } catch { + // The view closed first; there is nothing left to focus. + } + })(); } private terminalGrid( diff --git a/js/core/src/__tests__/YasNativeWorkspaceConnection.test.ts b/js/core/src/__tests__/YasNativeWorkspaceConnection.test.ts index ba49259a..cdbfc41a 100644 --- a/js/core/src/__tests__/YasNativeWorkspaceConnection.test.ts +++ b/js/core/src/__tests__/YasNativeWorkspaceConnection.test.ts @@ -1161,6 +1161,39 @@ describe("YasNativeWorkspaceConnection", () => { expect(disposeFont).toHaveBeenCalledOnce(); }); + it("handles SET_FOCUS rejected because the view closed first", async () => { + // A rejected promise nobody observes terminates a Node embedder. + const observed: number[] = []; + const setFocus = vi.fn((viewId: number) => ({ + then(_resolve: unknown, reject?: (error: Error) => void) { + if (reject) observed.push(viewId); + reject?.(new Error("not found")); + }, + })); + const connection = Object.create( + YasNativeWorkspaceConnection.prototype, + ) as YasNativeWorkspaceConnection; + Object.assign(connection as object, { + id: "remote", + views: new Map([ + [1n, { view: { result: { viewId: 11 } } }], + [2n, { view: { result: { viewId: 12 } } }], + ]), + sessions: new Map([ + ["remote:terminal:1", { ptyId: 1n, state: "active" }], + ["remote:terminal:2", { ptyId: 2n, state: "active" }], + ]), + terminalClient: { setFocus }, + focusedSessionId: "remote:terminal:1", + refreshSnapshot: vi.fn(), + }); + + connection.focusSession("remote:terminal:2"); + await flush(); + + expect(observed).toEqual([11, 12]); + }); + it("deduplicates concurrent Terminal OPEN_VIEW requests per handle", async () => { const result = deferred<{ result: {