Fix settings schema missing extension-provided LSP settings on remote

crates/extension_host/Cargo.toml

@@ -65,6 +65,7 @@ language = { workspace = true, features = ["test-support"] }
 language_extension.workspace = true
 parking_lot.workspace = true
 project = { workspace = true, features = ["test-support"] }
+remote = { workspace = true, features = ["test-support"] }
 reqwest_client.workspace = true
 theme = { workspace = true, features = ["test-support"] }
 theme_settings.workspace = true

crates/extension_host/src/extension_host.rs

@@ -42,11 +42,12 @@ use language::{
 use node_runtime::NodeRuntime;
 use project::ContextProviderWithTasks;
 use release_channel::ReleaseChannel;
-use remote::RemoteClient;
+use remote::{OnRemoteClientCreated, RemoteClient, RemoteClientEvent};
 use semver::Version;
 use serde::{Deserialize, Serialize};
 use settings::{SemanticTokenRules, Settings, SettingsStore};
 use std::ops::RangeInclusive;
+use std::rc::Rc;
 use std::str::FromStr;
 use std::sync::LazyLock;
 use std::{
@@ -145,7 +146,7 @@ pub struct ExtensionStore {
     pub wasm_extensions: Vec<(Arc<ExtensionManifest>, WasmExtension)>,
     pub tasks: Vec<Task<()>>,
     pub remote_clients: Vec<WeakEntity<RemoteClient>>,
-    pub ssh_registered_tx: UnboundedSender<()>,
+    pub ssh_registered_tx: UnboundedSender<Option<WeakEntity<RemoteClient>>>,
 }
 
 #[derive(Clone, Copy)]
@@ -292,6 +293,12 @@ pub fn init(
     });
 
     cx.set_global(GlobalExtensionStore(store));
+
+    cx.set_global(OnRemoteClientCreated(Rc::new(|client, cx| {
+        if let Some(store) = ExtensionStore::try_global(cx) {
+            store.update(cx, |store, cx| store.register_remote_client(client, cx));
+        }
+    })));
 }
 
 impl ExtensionStore {
@@ -425,8 +432,17 @@ impl ExtensionStore {
 
                             Self::update_remote_clients(&this, cx).await?;
                         }
-                        _ = connection_registered_rx.next() => {
-                            debounce_timer = cx.background_executor().timer(RELOAD_DEBOUNCE_DURATION).fuse()
+                        client = connection_registered_rx.next() => {
+                            match client {
+                                Some(Some(client)) => {
+                                    Self::sync_extensions_to_remotes(&this, client, cx)
+                                        .await
+                                        .log_err();
+                                }
+                                _ => {
+                                    debounce_timer = cx.background_executor().timer(RELOAD_DEBOUNCE_DURATION).fuse()
+                                }
+                            }
                         }
                         extension_id = reload_rx.next() => {
                             let Some(extension_id) = extension_id else { break; };
@@ -1959,13 +1975,18 @@ impl ExtensionStore {
         anyhow::Ok(())
     }
 
-    pub fn register_remote_client(
-        &mut self,
-        client: Entity<RemoteClient>,
-        _cx: &mut Context<Self>,
-    ) {
+    pub fn register_remote_client(&mut self, client: Entity<RemoteClient>, cx: &mut Context<Self>) {
         self.remote_clients.push(client.downgrade());
-        self.ssh_registered_tx.unbounded_send(()).ok();
+        self.ssh_registered_tx
+            .unbounded_send(Some(client.downgrade()))
+            .ok();
+
+        cx.subscribe(&client, |store, _client, event, _cx| {
+            if matches!(event, RemoteClientEvent::Reconnected) {
+                store.ssh_registered_tx.unbounded_send(None).ok();
+            }
+        })
+        .detach();
     }
 }
 

crates/extension_host/src/extension_store_test.rs

@@ -4,11 +4,12 @@ use crate::{
     RELOAD_DEBOUNCE_DURATION, SchemaVersion, load_plugin_queries,
 };
 use async_compression::futures::bufread::GzipEncoder;
+use client::{AnyProtoClient, TypedEnvelope, proto};
 use collections::{BTreeMap, HashSet};
 use extension::ExtensionHostProxy;
 use fs::{FakeFs, Fs, RealFs};
 use futures::{AsyncReadExt, FutureExt, StreamExt, io::BufReader};
-use gpui::{AppContext as _, BackgroundExecutor, TaskExt, TestAppContext};
+use gpui::{AppContext as _, BackgroundExecutor, Entity, TaskExt, TestAppContext};
 use http_client::{FakeHttpClient, Response};
 use language::{BinaryStatus, LanguageMatcher, LanguageName, LanguageRegistry, QueryFiles};
 use language_extension::LspAccess;
@@ -17,13 +18,17 @@ use node_runtime::NodeRuntime;
 use parking_lot::Mutex;
 use project::{DEFAULT_COMPLETION_CONTEXT, Project};
 use release_channel::AppVersion;
+use remote::{RemoteClient, RemoteClientEvent, RemoteConnectionOptions};
 use reqwest_client::ReqwestClient;
 use serde_json::json;
 use settings::SettingsStore;
 use std::{
     ffi::OsString,
     path::{Path, PathBuf},
-    sync::Arc,
+    sync::{
+        Arc,
+        atomic::{AtomicUsize, Ordering},
+    },
 };
 use theme::ThemeRegistry;
 use util::{rel_path::rel_path_buf, test::TempTree};
@@ -1275,3 +1280,170 @@ fn init_test(cx: &mut TestAppContext) {
         gpui_tokio::init(cx);
     });
 }
+
+struct SyncRequestCounter(Arc<AtomicUsize>);
+
+/// Creates a mock remote connection whose server responds to connection
+/// handshake requests and counts every `SyncExtensions` request it receives.
+/// The returned `counter` entity must be kept alive for the handler to work.
+async fn setup_mock_remote(
+    cx: &mut TestAppContext,
+    server_cx: &mut TestAppContext,
+) -> (
+    RemoteConnectionOptions,
+    Entity<SyncRequestCounter>,
+    Arc<AtomicUsize>,
+) {
+    let sync_count = Arc::new(AtomicUsize::new(0));
+    let counter = server_cx.new(|_| SyncRequestCounter(sync_count.clone()));
+    let (opts, server_client, _) = RemoteClient::fake_server(cx, server_cx);
+    register_server_handlers(&server_client, counter.clone());
+    (opts, counter, sync_count)
+}
+
+fn register_server_handlers(server_client: &AnyProtoClient, counter: Entity<SyncRequestCounter>) {
+    // The remote client pings the server as part of the connection handshake.
+    server_client.add_request_handler::<proto::Ping, SyncRequestCounter, _, _>(
+        counter.downgrade(),
+        |_counter, _envelope: TypedEnvelope<proto::Ping>, _cx| async move { Ok(proto::Ack {}) },
+    );
+    server_client.add_request_handler::<proto::SyncExtensions, SyncRequestCounter, _, _>(
+        counter.downgrade(),
+        |counter, _envelope: TypedEnvelope<proto::SyncExtensions>, mut cx| async move {
+            counter.update(&mut cx, |counter, _cx| {
+                counter.0.fetch_add(1, Ordering::SeqCst);
+            });
+            Ok(proto::SyncExtensionsResponse {
+                missing_extensions: Vec::new(),
+                tmp_dir: String::new(),
+            })
+        },
+    );
+}
+
+fn create_extension_store(cx: &mut TestAppContext) -> Entity<ExtensionStore> {
+    let fs = FakeFs::new(cx.executor());
+    let http_client = FakeHttpClient::with_200_response();
+    let proxy = Arc::new(ExtensionHostProxy::new());
+    let node_runtime = NodeRuntime::unavailable();
+
+    let store = cx.new(|cx| {
+        ExtensionStore::new(
+            PathBuf::from("/extensions"),
+            None,
+            proxy,
+            fs.clone(),
+            http_client.clone(),
+            http_client.clone(),
+            None,
+            node_runtime.clone(),
+            cx,
+        )
+    });
+    cx.run_until_parked();
+
+    store
+}
+
+#[gpui::test]
+async fn test_register_remote_client_syncs_only_the_new_client(
+    cx: &mut TestAppContext,
+    server_cx: &mut TestAppContext,
+) {
+    init_test(cx);
+    let store = create_extension_store(cx);
+
+    store.update(cx, |store, _cx| {
+        store.extension_index.extensions.insert(
+            "foo-lsp".into(),
+            remote_sync_entry(
+                "foo-lsp",
+                r#"
+                [language_servers.foo]
+                language = "Foo"
+                "#,
+            ),
+        );
+    });
+
+    let (opts_a, _counter_a, sync_count_a) = setup_mock_remote(cx, server_cx).await;
+    let (opts_b, _counter_b, sync_count_b) = setup_mock_remote(cx, server_cx).await;
+
+    let client_a = RemoteClient::connect_mock(opts_a, cx).await;
+    let client_b = RemoteClient::connect_mock(opts_b, cx).await;
+
+    store.update(cx, |store, cx| {
+        store.register_remote_client(client_a.clone(), cx)
+    });
+    cx.run_until_parked();
+    assert_eq!(
+        sync_count_a.load(Ordering::SeqCst),
+        1,
+        "registering a client should sync extensions to it once"
+    );
+    assert_eq!(sync_count_b.load(Ordering::SeqCst), 0);
+
+    store.update(cx, |store, cx| {
+        store.register_remote_client(client_b.clone(), cx)
+    });
+    cx.run_until_parked();
+    assert_eq!(
+        sync_count_a.load(Ordering::SeqCst),
+        1,
+        "registering a new client should not re-sync already-registered clients"
+    );
+    assert_eq!(
+        sync_count_b.load(Ordering::SeqCst),
+        1,
+        "registering a client should sync extensions to it once"
+    );
+}
+
+#[gpui::test]
+async fn test_register_remote_client_resyncs_extensions_on_reconnect(
+    cx: &mut TestAppContext,
+    server_cx: &mut TestAppContext,
+) {
+    init_test(cx);
+    let store = create_extension_store(cx);
+
+    store.update(cx, |store, _cx| {
+        store.extension_index.extensions.insert(
+            "foo-lsp".into(),
+            remote_sync_entry(
+                "foo-lsp",
+                r#"
+                [language_servers.foo]
+                language = "Foo"
+                "#,
+            ),
+        );
+    });
+
+    let (opts, _counter, sync_count) = setup_mock_remote(cx, server_cx).await;
+    let client = RemoteClient::connect_mock(opts, cx).await;
+    store.update(cx, |store, cx| {
+        store.register_remote_client(client.clone(), cx)
+    });
+    cx.run_until_parked();
+    assert_eq!(
+        sync_count.load(Ordering::SeqCst),
+        1,
+        "registering a remote client should sync extensions to it once"
+    );
+
+    // Simulate the remote client reconnecting (e.g. after an SSH reconnect
+    // restarted the remote server): the reconnect emits `Reconnected`, which
+    // should re-sync extensions to the client.
+    client.update(cx, |_client, cx| {
+        cx.emit(RemoteClientEvent::Reconnected);
+    });
+    cx.executor().advance_clock(RELOAD_DEBOUNCE_DURATION);
+    cx.run_until_parked();
+
+    assert_eq!(
+        sync_count.load(Ordering::SeqCst),
+        2,
+        "reconnecting should re-sync extensions to the remote client"
+    );
+}

crates/extension_host/src/headless_host.rs

@@ -4,8 +4,8 @@ use anyhow::{Context as _, Result};
 use client::{TypedEnvelope, proto};
 use collections::{HashMap, HashSet};
 use extension::{
-    Extension, ExtensionDebugAdapterProviderProxy, ExtensionHostProxy, ExtensionLanguageProxy,
-    ExtensionLanguageServerProxy, ExtensionManifest,
+    Event, Extension, ExtensionDebugAdapterProviderProxy, ExtensionEvents, ExtensionHostProxy,
+    ExtensionLanguageProxy, ExtensionLanguageServerProxy, ExtensionManifest,
 };
 use fs::{Fs, RemoveOptions, RenameOptions};
 use futures::future::{FutureExt as _, join_all};
@@ -85,24 +85,37 @@ impl HeadlessExtensionStore {
             })
             .collect();
 
-        cx.spawn(async move |this, cx| {
+        cx.spawn(async move |store, cx| {
             let mut missing = Vec::new();
+            let mut extensions_changed = false;
 
             for extension_id in to_remove {
                 log::info!("removing extension: {}", extension_id);
-                this.update(cx, |this, cx| this.uninstall_extension(&extension_id, cx))?
+                store
+                    .update(cx, |store, cx| store.uninstall_extension(&extension_id, cx))?
                     .await?;
+                extensions_changed = true;
             }
 
             for extension in to_load {
-                if let Err(e) = Self::load_extension(this.clone(), extension.clone(), cx).await {
-                    log::info!("failed to load extension: {}, {:#}", extension.id, e);
-                    missing.push(extension)
-                } else if extension.dev {
-                    missing.push(extension)
+                match Self::load_extension(store.clone(), extension.clone(), cx).await {
+                    Ok(()) => {
+                        extensions_changed = true;
+                        if extension.dev {
+                            missing.push(extension)
+                        }
+                    }
+                    Err(e) => {
+                        log::info!("failed to load extension: {}, {:#}", extension.id, e);
+                        missing.push(extension)
+                    }
                 }
             }
 
+            if extensions_changed {
+                store.update(cx, |_, cx| notify_extensions_changed(cx)).ok();
+            }
+
             Ok(missing)
         })
     }
@@ -275,19 +288,24 @@ impl HeadlessExtensionStore {
         let path = self.extension_dir.join(&extension.id);
         let fs = self.fs.clone();
 
-        cx.spawn(async move |this, cx| {
+        cx.spawn(async move |store, cx| {
             if fs.is_dir(&path).await {
-                this.update(cx, |this, cx| {
-                    this.uninstall_extension(&extension.id.clone().into(), cx)
-                })?
-                .await?;
+                store
+                    .update(cx, |store, cx| {
+                        store.uninstall_extension(&extension.id.clone().into(), cx)
+                    })?
+                    .await?;
             }
 
             fs.rename(&tmp_path, &path, RenameOptions::default())
                 .await
                 .with_context(|| format!("Failed to rename {tmp_path:?} to {path:?}"))?;
 
-            Self::load_extension(this, extension, cx).await
+            Self::load_extension(store.clone(), extension, cx).await?;
+
+            store.update(cx, |_, cx| notify_extensions_changed(cx)).ok();
+
+            Ok(())
         })
     }
 
@@ -354,3 +372,11 @@ impl HeadlessExtensionStore {
         Ok(proto::Ack {})
     }
 }
+
+fn notify_extensions_changed(cx: &mut App) {
+    if let Some(events) = ExtensionEvents::try_global(cx) {
+        events.update(cx, |this, cx| {
+            this.emit(Event::ExtensionsInstalledChanged, cx)
+        });
+    }
+}