lsp: Reference-count log streams across views

LSP log streams can be consumed by split views and downstream clients.
Treating enablement as a single value lets one consumer disable a
stream that another consumer is still using.

Track the streams enabled by each view and reference-count each exact
server and log kind. Keep in-process view ownership separate from
downstream requests, preserve RPC capture while either has an active
consumer, and notify remote hosts only on the first enable and final
disable.

Represent log kinds independently so split views can consume logs,
traces, and RPC messages without overwriting one another.

Add regression coverage for both local-view and downstream release
orderings.

Release Notes:

- Fixed closing one LSP Logs view stopping streams used by another view
  or downstream client

Author: aviatesk

crates/language_tools/src/lsp_log_view.rs

@@ -1,4 +1,4 @@
-use collections::VecDeque;
+use collections::{HashMap, HashSet, VecDeque};
 use edit_prediction::EditPredictionStore;
 use editor::{Editor, EditorEvent, MultiBufferOffset, actions::MoveToEnd, scroll::Autoscroll};
 use gpui::{
@@ -102,6 +102,7 @@ pub struct LspLogView {
     log_store: Entity<LogStore>,
     current_server_key: Option<LanguageServerLogKey>,
     active_entry_kind: LogKind,
+    enabled_streams: HashMap<LanguageServerLogKey, HashSet<LogKind>>,
     project: Entity<Project>,
     focus_handle: FocusHandle,
     _log_store_subscriptions: Vec<Subscription>,
@@ -261,18 +262,19 @@ impl LspLogView {
         });
 
         cx.on_release(|log_view, cx| {
-            let project = log_view.project.downgrade();
-            let lsp_store = log_view.project.read(cx).lsp_store().downgrade();
-            log_view.log_store.update(cx, |log_store, cx| {
-                for (key, state) in &log_store.language_servers {
-                    if key.is_for_project(&project, &lsp_store)
-                        && let Some(log_kind) = state.toggled_log_kind
-                        && let Some(log_type) = log_type(log_kind)
-                    {
-                        send_toggle_log_message(key, false, log_type, cx);
-                    }
-                }
-            });
+            let enabled_streams = log_view
+                .enabled_streams
+                .iter()
+                .flat_map(|(key, log_kinds)| {
+                    log_kinds
+                        .iter()
+                        .map(|log_kind| (key.clone(), *log_kind))
+                        .collect::<Vec<_>>()
+                })
+                .collect::<Vec<_>>();
+            for (key, log_kind) in enabled_streams {
+                log_view.set_stream_enabled(&key, log_kind, false, cx);
+            }
         })
         .detach();
 
@@ -284,6 +286,7 @@ impl LspLogView {
             log_store,
             current_server_key: None,
             active_entry_kind: LogKind::Logs,
+            enabled_streams: HashMap::default(),
             _log_store_subscriptions: vec![
                 model_changes_subscription,
                 events_subscriptions,
@@ -463,6 +466,62 @@ impl LspLogView {
         Some(rows)
     }
 
+    fn set_stream_enabled(
+        &mut self,
+        key: &LanguageServerLogKey,
+        log_kind: LogKind,
+        enabled: bool,
+        cx: &mut App,
+    ) {
+        if enabled {
+            let already_enabled = self
+                .enabled_streams
+                .get(key)
+                .is_some_and(|log_kinds| log_kinds.contains(&log_kind));
+            if already_enabled {
+                return;
+            }
+            let Some(is_first) = self.log_store.update(cx, |log_store, _| {
+                log_store.retain_view_log_stream(key, log_kind)
+            }) else {
+                return;
+            };
+            self.enabled_streams
+                .entry(key.clone())
+                .or_default()
+                .insert(log_kind);
+            if is_first && let Some(log_type) = log_type(log_kind) {
+                send_toggle_log_message(key, true, log_type, cx);
+            }
+        } else {
+            let Some(log_kinds) = self.enabled_streams.get_mut(key) else {
+                return;
+            };
+            if !log_kinds.remove(&log_kind) {
+                return;
+            }
+            if log_kinds.is_empty() {
+                self.enabled_streams.remove(key);
+            }
+            let is_last = self
+                .log_store
+                .update(cx, |log_store, _| {
+                    log_store.release_view_log_stream(key, log_kind)
+                })
+                .unwrap_or(true);
+            if is_last && let Some(log_type) = log_type(log_kind) {
+                send_toggle_log_message(key, false, log_type, cx);
+            }
+        }
+    }
+
+    fn disable_streams_for_server(&mut self, key: &LanguageServerLogKey, cx: &mut App) {
+        let log_kinds = self.enabled_streams.get(key).cloned().unwrap_or_default();
+        for log_kind in log_kinds {
+            self.set_stream_enabled(key, log_kind, false, cx);
+        }
+    }
+
     fn show_logs_for_server(
         &mut self,
         key: LanguageServerLogKey,
@@ -490,12 +549,7 @@ impl LspLogView {
             cx.notify();
         }
         self.editor.read(cx).focus_handle(cx).focus(window, cx);
-        self.log_store.update(cx, |log_store, cx| {
-            let state = log_store.get_language_server_state(&key)?;
-            state.toggled_log_kind = Some(LogKind::Logs);
-            send_toggle_log_message(&key, true, LogType::Log, cx);
-            Some(())
-        });
+        self.set_stream_enabled(&key, LogKind::Logs, true, cx);
     }
 
     fn update_log_level(
@@ -547,12 +601,7 @@ impl LspLogView {
             let (editor, editor_subscriptions) = Self::editor_for_logs(log_contents, window, cx);
             self.editor = editor;
             self.editor_subscriptions = editor_subscriptions;
-            self.log_store.update(cx, |log_store, cx| {
-                let state = log_store.get_language_server_state(&key)?;
-                state.toggled_log_kind = Some(LogKind::Trace);
-                send_toggle_log_message(&key, true, LogType::Trace, cx);
-                Some(())
-            });
+            self.set_stream_enabled(&key, LogKind::Trace, true, cx);
             cx.notify();
         }
         self.editor.read(cx).focus_handle(cx).focus(window, cx);
@@ -567,7 +616,9 @@ impl LspLogView {
         self.toggle_rpc_trace_for_server(key.clone(), true, window, cx);
         let rpc_log = self.log_store.update(cx, |log_store, _| {
             log_store
-                .enable_rpc_trace_for_language_server(&key)
+                .get_language_server_state(&key)?
+                .rpc_state
+                .as_ref()
                 .map(|state| log_contents(&state.rpc_messages, ()))
         });
         if let Some(rpc_log) = rpc_log {
@@ -609,17 +660,7 @@ impl LspLogView {
         window: &mut Window,
         cx: &mut Context<Self>,
     ) {
-        self.log_store.update(cx, |log_store, cx| {
-            if enabled {
-                log_store.enable_rpc_trace_for_language_server(&key);
-            } else {
-                log_store.disable_rpc_trace_for_language_server(&key);
-            }
-
-            if log_store.language_servers.contains_key(&key) {
-                send_toggle_log_message(&key, enabled, LogType::Rpc, cx);
-            }
-        });
+        self.set_stream_enabled(&key, LogKind::Rpc, enabled, cx);
         if !enabled && self.current_server_key.as_ref() == Some(&key) {
             self.show_logs_for_server(key, window, cx);
             cx.notify();
@@ -704,15 +745,7 @@ impl LspLogView {
         self.editor_subscriptions = editor_subscriptions;
         cx.notify();
         self.editor.read(cx).focus_handle(cx).focus(window, cx);
-        self.log_store.update(cx, |log_store, cx| {
-            let state = log_store.get_language_server_state(&key)?;
-            if let Some(log_kind) = state.toggled_log_kind.take()
-                && let Some(log_type) = log_type(log_kind)
-            {
-                send_toggle_log_message(&key, false, log_type, cx);
-            }
-            Some(())
-        });
+        self.disable_streams_for_server(&key, cx);
     }
 }
 

crates/language_tools/src/lsp_log_view_tests.rs

@@ -11,7 +11,7 @@ use language::{
 use lsp::LanguageServerName;
 use project::{
     FakeFs, Project,
-    lsp_store::log_store::{LanguageServerKind, LogKind, LogStore},
+    lsp_store::log_store::{LanguageServerKind, LanguageServerLogKey, LogKind, LogStore},
 };
 use serde_json::json;
 use settings::SettingsStore;
@@ -127,6 +127,91 @@ async fn test_lsp_log_view_filters_servers_from_other_projects(cx: &mut TestAppC
     });
 }
 
+#[gpui::test]
+async fn test_local_views_and_downstream_requests_own_rpc_streams_independently(
+    cx: &mut TestAppContext,
+) {
+    init_test(cx);
+
+    let fs = FakeFs::new(cx.background_executor.clone());
+    fs.insert_tree(path!("/stream-ownership"), json!({ "test.rs": "" }))
+        .await;
+    let project = Project::test(fs, [path!("/stream-ownership").as_ref()], cx).await;
+    let server_id = LanguageServerId(100);
+    let server_key = LanguageServerLogKey::new(
+        LanguageServerKind::Local {
+            project: project.downgrade(),
+        },
+        server_id,
+    );
+    let log_store = cx.new(|cx| LogStore::new(false, cx));
+
+    log_store.update(cx, |store, cx| {
+        store.add_language_server(
+            server_key.kind.clone(),
+            server_id,
+            Some(LanguageServerName::new_static("test-server")),
+            None,
+            None,
+            cx,
+        );
+
+        store.toggle_lsp_logs(&server_key, true, LogKind::Rpc);
+        assert!(
+            store
+                .language_servers
+                .get(&server_key)
+                .is_some_and(|state| state.rpc_state.is_some())
+        );
+        assert_eq!(
+            store.retain_view_log_stream(&server_key, LogKind::Rpc),
+            Some(true)
+        );
+        assert_eq!(
+            store.release_view_log_stream(&server_key, LogKind::Rpc),
+            Some(true)
+        );
+        assert!(
+            store
+                .language_servers
+                .get(&server_key)
+                .is_some_and(|state| state.rpc_state.is_some()),
+            "releasing the final local view must preserve a downstream RPC request"
+        );
+        store.toggle_lsp_logs(&server_key, false, LogKind::Rpc);
+        assert!(
+            store
+                .language_servers
+                .get(&server_key)
+                .is_some_and(|state| state.rpc_state.is_none())
+        );
+
+        assert_eq!(
+            store.retain_view_log_stream(&server_key, LogKind::Rpc),
+            Some(true)
+        );
+        store.toggle_lsp_logs(&server_key, true, LogKind::Rpc);
+        store.toggle_lsp_logs(&server_key, false, LogKind::Rpc);
+        assert!(
+            store
+                .language_servers
+                .get(&server_key)
+                .is_some_and(|state| state.rpc_state.is_some()),
+            "disabling a downstream RPC request must preserve a local view's stream"
+        );
+        assert_eq!(
+            store.release_view_log_stream(&server_key, LogKind::Rpc),
+            Some(true)
+        );
+        assert!(
+            store
+                .language_servers
+                .get(&server_key)
+                .is_some_and(|state| state.rpc_state.is_none())
+        );
+    });
+}
+
 #[gpui::test]
 async fn test_lsp_log_view(cx: &mut TestAppContext) {
     zlog::init_test();