Yehowshua/ZIm
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Separate timeout and connection dropped errors out (#30457)

Browse source
This commit is contained in:
Kirill Bulatov 2025-05-10 15:12:58 +03:00 • committed by GitHub
parent 39da72161f
commit 471e02d48f
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
25 changed files with 313 additions and 115 deletions
Download patch file Download diff file Expand all files Collapse all files

1
Cargo.lock generated
View file

@ -3068,6 +3068,7 @@ dependencies = [
"gpui", "gpui",
"http_client", "http_client",
"language", "language",
"log",
"menu", "menu",
"notifications", "notifications",
"picker", "picker",

21
crates/assistant_context_editor/src/context_editor.rs
View file

@ -1894,11 +1894,24 @@ impl ContextEditor {
.log_err(); .log_err();
if let Some(client) = client { if let Some(client) = client {
cx.spawn(async move |this, cx| { cx.spawn(async move |context_editor, cx| {
client.authenticate_and_connect(true, cx).await?; match client.authenticate_and_connect(true, cx).await {
this.update(cx, |_, cx| cx.notify()) util::ConnectionResult::Timeout => {
log::error!("Authentication timeout")
}
util::ConnectionResult::ConnectionReset => {
log::error!("Connection reset")
}
util::ConnectionResult::Result(r) => {
if r.log_err().is_some() {
context_editor
.update(cx, |_, cx| cx.notify())
.ok();
}
}
}
}) })
.detach_and_log_err(cx) .detach()
} }
})), })),
) )

77
crates/client/src/client.rs
View file

@ -49,7 +49,7 @@ use telemetry::Telemetry;
use thiserror::Error; use thiserror::Error;
use tokio::net::TcpStream; use tokio::net::TcpStream;
use url::Url; use url::Url;
use util::{ResultExt, TryFutureExt}; use util::{ConnectionResult, ResultExt};
pub use rpc::*; pub use rpc::*;
pub use telemetry_events::Event; pub use telemetry_events::Event;
@ -151,9 +151,19 @@ pub fn init(client: &Arc<Client>, cx: &mut App) {
let client = client.clone(); let client = client.clone();
move |_: &SignIn, cx| { move |_: &SignIn, cx| {
if let Some(client) = client.upgrade() { if let Some(client) = client.upgrade() {
cx.spawn(async move |cx| { cx.spawn(
client.authenticate_and_connect(true, &cx).log_err().await async move |cx| match client.authenticate_and_connect(true, &cx).await {
}) ConnectionResult::Timeout => {
log::error!("Initial authentication timed out");
}
ConnectionResult::ConnectionReset => {
log::error!("Initial authentication connection reset");
}
ConnectionResult::Result(r) => {
r.log_err();
}
},
)
.detach(); .detach();
} }
} }
@ -658,7 +668,7 @@ impl Client {
state._reconnect_task = None; state._reconnect_task = None;
} }
Status::ConnectionLost => { Status::ConnectionLost => {
let this = self.clone(); let client = self.clone();
state._reconnect_task = Some(cx.spawn(async move |cx| { state._reconnect_task = Some(cx.spawn(async move |cx| {
#[cfg(any(test, feature = "test-support"))] #[cfg(any(test, feature = "test-support"))]
let mut rng = StdRng::seed_from_u64(0); let mut rng = StdRng::seed_from_u64(0);
@ -666,10 +676,25 @@ impl Client {
let mut rng = StdRng::from_entropy(); let mut rng = StdRng::from_entropy();
let mut delay = INITIAL_RECONNECTION_DELAY; let mut delay = INITIAL_RECONNECTION_DELAY;
while let Err(error) = this.authenticate_and_connect(true, &cx).await { loop {
log::error!("failed to connect {}", error); match client.authenticate_and_connect(true, &cx).await {
if matches!(*this.status().borrow(), Status::ConnectionError) { ConnectionResult::Timeout => {
this.set_status( log::error!("client connect attempt timed out")
}
ConnectionResult::ConnectionReset => {
log::error!("client connect attempt reset")
}
ConnectionResult::Result(r) => {
if let Err(error) = r {
log::error!("failed to connect: {error}");
} else {
break;
}
}
}
if matches!(*client.status().borrow(), Status::ConnectionError) {
client.set_status(
Status::ReconnectionError { Status::ReconnectionError {
next_reconnection: Instant::now() + delay, next_reconnection: Instant::now() + delay,
}, },
@ -827,7 +852,7 @@ impl Client {
self: &Arc<Self>, self: &Arc<Self>,
try_provider: bool, try_provider: bool,
cx: &AsyncApp, cx: &AsyncApp,
) -> anyhow::Result<()> { ) -> ConnectionResult<()> {
let was_disconnected = match *self.status().borrow() { let was_disconnected = match *self.status().borrow() {
Status::SignedOut => true, Status::SignedOut => true,
Status::ConnectionError Status::ConnectionError
@ -836,9 +861,14 @@ impl Client {
| Status::Reauthenticating { .. } | Status::Reauthenticating { .. }
| Status::ReconnectionError { .. } => false, | Status::ReconnectionError { .. } => false,
Status::Connected { .. } | Status::Connecting { .. } | Status::Reconnecting { .. } => { Status::Connected { .. } | Status::Connecting { .. } | Status::Reconnecting { .. } => {
return Ok(()); return ConnectionResult::Result(Ok(()));
}
Status::UpgradeRequired => {
return ConnectionResult::Result(
Err(EstablishConnectionError::UpgradeRequired)
.context("client auth and connect"),
);
} }
Status::UpgradeRequired => return Err(EstablishConnectionError::UpgradeRequired)?,
}; };
if was_disconnected { if was_disconnected {
self.set_status(Status::Authenticating, cx); self.set_status(Status::Authenticating, cx);
@ -862,12 +892,12 @@ impl Client {
Ok(creds) => credentials = Some(creds), Ok(creds) => credentials = Some(creds),
Err(err) => { Err(err) => {
self.set_status(Status::ConnectionError, cx); self.set_status(Status::ConnectionError, cx);
return Err(err); return ConnectionResult::Result(Err(err));
} }
} }
} }
_ = status_rx.next().fuse() => { _ = status_rx.next().fuse() => {
return Err(anyhow!("authentication canceled")); return ConnectionResult::Result(Err(anyhow!("authentication canceled")));
} }
} }
} }
@ -892,10 +922,10 @@ impl Client {
} }
futures::select_biased! { futures::select_biased! {
result = self.set_connection(conn, cx).fuse() => result, result = self.set_connection(conn, cx).fuse() => ConnectionResult::Result(result.context("client auth and connect")),
_ = timeout => { _ = timeout => {
self.set_status(Status::ConnectionError, cx); self.set_status(Status::ConnectionError, cx);
Err(anyhow!("timed out waiting on hello message from server")) ConnectionResult::Timeout
} }
} }
} }
@ -907,22 +937,22 @@ impl Client {
self.authenticate_and_connect(false, cx).await self.authenticate_and_connect(false, cx).await
} else { } else {
self.set_status(Status::ConnectionError, cx); self.set_status(Status::ConnectionError, cx);
Err(EstablishConnectionError::Unauthorized)? ConnectionResult::Result(Err(EstablishConnectionError::Unauthorized).context("client auth and connect"))
} }
} }
Err(EstablishConnectionError::UpgradeRequired) => { Err(EstablishConnectionError::UpgradeRequired) => {
self.set_status(Status::UpgradeRequired, cx); self.set_status(Status::UpgradeRequired, cx);
Err(EstablishConnectionError::UpgradeRequired)? ConnectionResult::Result(Err(EstablishConnectionError::UpgradeRequired).context("client auth and connect"))
} }
Err(error) => { Err(error) => {
self.set_status(Status::ConnectionError, cx); self.set_status(Status::ConnectionError, cx);
Err(error)? ConnectionResult::Result(Err(error).context("client auth and connect"))
} }
} }
} }
_ = &mut timeout => { _ = &mut timeout => {
self.set_status(Status::ConnectionError, cx); self.set_status(Status::ConnectionError, cx);
Err(anyhow!("timed out trying to establish connection")) ConnectionResult::Timeout
} }
} }
} }
@ -938,10 +968,7 @@ impl Client {
let peer_id = async { let peer_id = async {
log::debug!("waiting for server hello"); log::debug!("waiting for server hello");
let message = incoming let message = incoming.next().await.context("no hello message received")?;
.next()
.await
.ok_or_else(|| anyhow!("no hello message received"))?;
log::debug!("got server hello"); log::debug!("got server hello");
let hello_message_type_name = message.payload_type_name().to_string(); let hello_message_type_name = message.payload_type_name().to_string();
let hello = message let hello = message
@ -1743,7 +1770,7 @@ mod tests {
status.next().await, status.next().await,
Some(Status::ConnectionError { .. }) Some(Status::ConnectionError { .. })
)); ));
auth_and_connect.await.unwrap_err(); auth_and_connect.await.into_response().unwrap_err();
// Allow the connection to be established. // Allow the connection to be established.
let server = FakeServer::for_client(user_id, &client, cx).await; let server = FakeServer::for_client(user_id, &client, cx).await;

1
crates/client/src/test.rs
View file

@ -107,6 +107,7 @@ impl FakeServer {
client client
.authenticate_and_connect(false, &cx.to_async()) .authenticate_and_connect(false, &cx.to_async())
.await .await
.into_response()
.unwrap(); .unwrap();
server server

2
crates/collab/src/tests/editor_tests.rs
View file

@ -1740,6 +1740,7 @@ async fn test_mutual_editor_inlay_hint_cache_update(
fake_language_server fake_language_server
.request::<lsp::request::InlayHintRefreshRequest>(()) .request::<lsp::request::InlayHintRefreshRequest>(())
.await .await
.into_response()
.expect("inlay refresh request failed"); .expect("inlay refresh request failed");
executor.run_until_parked(); executor.run_until_parked();
@ -1930,6 +1931,7 @@ async fn test_inlay_hint_refresh_is_forwarded(
fake_language_server fake_language_server
.request::<lsp::request::InlayHintRefreshRequest>(()) .request::<lsp::request::InlayHintRefreshRequest>(())
.await .await
.into_response()
.expect("inlay refresh request failed"); .expect("inlay refresh request failed");
executor.run_until_parked(); executor.run_until_parked();
editor_a.update(cx_a, |editor, _| { editor_a.update(cx_a, |editor, _| {

6
crates/collab/src/tests/integration_tests.rs
View file

@ -1253,6 +1253,7 @@ async fn test_calls_on_multiple_connections(
client_b1 client_b1
.authenticate_and_connect(false, &cx_b1.to_async()) .authenticate_and_connect(false, &cx_b1.to_async())
.await .await
.into_response()
.unwrap(); .unwrap();
// User B hangs up, and user A calls them again. // User B hangs up, and user A calls them again.
@ -1633,6 +1634,7 @@ async fn test_project_reconnect(
client_a client_a
.authenticate_and_connect(false, &cx_a.to_async()) .authenticate_and_connect(false, &cx_a.to_async())
.await .await
.into_response()
.unwrap(); .unwrap();
executor.run_until_parked(); executor.run_until_parked();
@ -1761,6 +1763,7 @@ async fn test_project_reconnect(
client_b client_b
.authenticate_and_connect(false, &cx_b.to_async()) .authenticate_and_connect(false, &cx_b.to_async())
.await .await
.into_response()
.unwrap(); .unwrap();
executor.run_until_parked(); executor.run_until_parked();
@ -4317,6 +4320,7 @@ async fn test_collaborating_with_lsp_progress_updates_and_diagnostics_ordering(
token: lsp::NumberOrString::String("the-disk-based-token".to_string()), token: lsp::NumberOrString::String("the-disk-based-token".to_string()),
}) })
.await .await
.into_response()
.unwrap(); .unwrap();
fake_language_server.notify::<lsp::notification::Progress>(&lsp::ProgressParams { fake_language_server.notify::<lsp::notification::Progress>(&lsp::ProgressParams {
token: lsp::NumberOrString::String("the-disk-based-token".to_string()), token: lsp::NumberOrString::String("the-disk-based-token".to_string()),
@ -5699,6 +5703,7 @@ async fn test_contacts(
client_c client_c
.authenticate_and_connect(false, &cx_c.to_async()) .authenticate_and_connect(false, &cx_c.to_async())
.await .await
.into_response()
.unwrap(); .unwrap();
executor.run_until_parked(); executor.run_until_parked();
@ -6229,6 +6234,7 @@ async fn test_contact_requests(
client client
.authenticate_and_connect(false, &cx.to_async()) .authenticate_and_connect(false, &cx.to_async())
.await .await
.into_response()
.unwrap(); .unwrap();
} }
} }

1
crates/collab/src/tests/test_server.rs
View file

@ -313,6 +313,7 @@ impl TestServer {
client client
.authenticate_and_connect(false, &cx.to_async()) .authenticate_and_connect(false, &cx.to_async())
.await .await
.into_response()
.unwrap(); .unwrap();
let client = TestClient { let client = TestClient {

1
crates/collab_ui/Cargo.toml
View file

@ -42,6 +42,7 @@ futures.workspace = true
fuzzy.workspace = true fuzzy.workspace = true
gpui.workspace = true gpui.workspace = true
language.workspace = true language.workspace = true
log.workspace = true
menu.workspace = true menu.workspace = true
notifications.workspace = true notifications.workspace = true
picker.workspace = true picker.workspace = true

1
crates/collab_ui/src/collab_panel.rs
View file

@ -2227,6 +2227,7 @@ impl CollabPanel {
client client
.authenticate_and_connect(true, &cx) .authenticate_and_connect(true, &cx)
.await .await
.into_response()
.notify_async_err(cx); .notify_async_err(cx);
}) })
.detach() .detach()

16
crates/collab_ui/src/notification_panel.rs
View file

@ -646,10 +646,20 @@ impl Render for NotificationPanel {
let client = client.clone(); let client = client.clone();
window window
.spawn(cx, async move |cx| { .spawn(cx, async move |cx| {
client match client
.authenticate_and_connect(true, &cx) .authenticate_and_connect(true, &cx)
.log_err() .await
.await; {
util::ConnectionResult::Timeout => {
log::error!("Connection timeout");
}
util::ConnectionResult::ConnectionReset => {
log::error!("Connection reset");
}
util::ConnectionResult::Result(r) => {
r.log_err();
}
}
}) })
.detach() .detach()
} }

36
crates/copilot/src/copilot.rs
View file

@ -5,7 +5,7 @@ mod sign_in;
use crate::sign_in::initiate_sign_in_within_workspace; use crate::sign_in::initiate_sign_in_within_workspace;
use ::fs::Fs; use ::fs::Fs;
use anyhow::{Result, anyhow}; use anyhow::{Context as _, Result, anyhow};
use collections::{HashMap, HashSet}; use collections::{HashMap, HashSet};
use command_palette_hooks::CommandPaletteFilter; use command_palette_hooks::CommandPaletteFilter;
use futures::{Future, FutureExt, TryFutureExt, channel::oneshot, future::Shared}; use futures::{Future, FutureExt, TryFutureExt, channel::oneshot, future::Shared};
@ -531,11 +531,15 @@ impl Copilot {
.request::<request::CheckStatus>(request::CheckStatusParams { .request::<request::CheckStatus>(request::CheckStatusParams {
local_checks_only: false, local_checks_only: false,
}) })
.await?; .await
.into_response()
.context("copilot: check status")?;
server server
.request::<request::SetEditorInfo>(editor_info) .request::<request::SetEditorInfo>(editor_info)
.await?; .await
.into_response()
.context("copilot: set editor info")?;
anyhow::Ok((server, status)) anyhow::Ok((server, status))
}; };
@ -581,7 +585,9 @@ impl Copilot {
.request::<request::SignInInitiate>( .request::<request::SignInInitiate>(
request::SignInInitiateParams {}, request::SignInInitiateParams {},
) )
.await?; .await
.into_response()
.context("copilot sign-in")?;
match sign_in { match sign_in {
request::SignInInitiateResult::AlreadySignedIn { user } => { request::SignInInitiateResult::AlreadySignedIn { user } => {
Ok(request::SignInStatus::Ok { user: Some(user) }) Ok(request::SignInStatus::Ok { user: Some(user) })
@ -609,7 +615,9 @@ impl Copilot {
user_code: flow.user_code, user_code: flow.user_code,
}, },
) )
.await?; .await
.into_response()
.context("copilot: sign in confirm")?;
Ok(response) Ok(response)
} }
} }
@ -656,7 +664,9 @@ impl Copilot {
cx.background_spawn(async move { cx.background_spawn(async move {
server server
.request::<request::SignOut>(request::SignOutParams {}) .request::<request::SignOut>(request::SignOutParams {})
.await?; .await
.into_response()
.context("copilot: sign in confirm")?;
anyhow::Ok(()) anyhow::Ok(())
}) })
} }
@ -873,7 +883,10 @@ impl Copilot {
uuid: completion.uuid.clone(), uuid: completion.uuid.clone(),
}); });
cx.background_spawn(async move { cx.background_spawn(async move {
request.await?; request
.await
.into_response()
.context("copilot: notify accepted")?;
Ok(()) Ok(())
}) })
} }
@ -897,7 +910,10 @@ impl Copilot {
.collect(), .collect(),
}); });
cx.background_spawn(async move { cx.background_spawn(async move {
request.await?; request
.await
.into_response()
.context("copilot: notify rejected")?;
Ok(()) Ok(())
}) })
} }
@ -957,7 +973,9 @@ impl Copilot {
version: version.try_into().unwrap(), version: version.try_into().unwrap(),
}, },
}) })
.await?; .await
.into_response()
.context("copilot: get completions")?;
let completions = result let completions = result
.completions .completions
.into_iter() .into_iter()

2
crates/editor/src/editor_tests.rs
View file

@ -8900,6 +8900,7 @@ async fn test_multiple_formatters(cx: &mut TestAppContext) {
}, },
}) })
.await .await
.into_response()
.unwrap(); .unwrap();
Ok(Some(json!(null))) Ok(Some(json!(null)))
} }
@ -19153,6 +19154,7 @@ async fn test_apply_code_lens_actions_with_commands(cx: &mut gpui::TestAppContex
}, },
) )
.await .await
.into_response()
.unwrap(); .unwrap();
Ok(Some(json!(null))) Ok(Some(json!(null)))
} }

5
crates/editor/src/inlay_hint_cache.rs
View file

@ -1409,6 +1409,7 @@ pub mod tests {
fake_server fake_server
.request::<lsp::request::InlayHintRefreshRequest>(()) .request::<lsp::request::InlayHintRefreshRequest>(())
.await .await
.into_response()
.expect("inlay refresh request failed"); .expect("inlay refresh request failed");
cx.executor().run_until_parked(); cx.executor().run_until_parked();
editor editor
@ -1492,6 +1493,7 @@ pub mod tests {
token: lsp::ProgressToken::String(progress_token.to_string()), token: lsp::ProgressToken::String(progress_token.to_string()),
}) })
.await .await
.into_response()
.expect("work done progress create request failed"); .expect("work done progress create request failed");
cx.executor().run_until_parked(); cx.executor().run_until_parked();
fake_server.notify::<lsp::notification::Progress>(&lsp::ProgressParams { fake_server.notify::<lsp::notification::Progress>(&lsp::ProgressParams {
@ -1863,6 +1865,7 @@ pub mod tests {
fake_server fake_server
.request::<lsp::request::InlayHintRefreshRequest>(()) .request::<lsp::request::InlayHintRefreshRequest>(())
.await .await
.into_response()
.expect("inlay refresh request failed"); .expect("inlay refresh request failed");
cx.executor().run_until_parked(); cx.executor().run_until_parked();
editor editor
@ -2008,6 +2011,7 @@ pub mod tests {
fake_server fake_server
.request::<lsp::request::InlayHintRefreshRequest>(()) .request::<lsp::request::InlayHintRefreshRequest>(())
.await .await
.into_response()
.expect("inlay refresh request failed"); .expect("inlay refresh request failed");
cx.executor().run_until_parked(); cx.executor().run_until_parked();
editor editor
@ -2070,6 +2074,7 @@ pub mod tests {
fake_server fake_server
.request::<lsp::request::InlayHintRefreshRequest>(()) .request::<lsp::request::InlayHintRefreshRequest>(())
.await .await
.into_response()
.expect("inlay refresh request failed"); .expect("inlay refresh request failed");
cx.executor().run_until_parked(); cx.executor().run_until_parked();
editor editor

9
crates/language_models/src/provider/cloud.rs
View file

@ -180,9 +180,12 @@ impl State {
fn authenticate(&self, cx: &mut Context<Self>) -> Task<Result<()>> { fn authenticate(&self, cx: &mut Context<Self>) -> Task<Result<()>> {
let client = self.client.clone(); let client = self.client.clone();
cx.spawn(async move |this, cx| { cx.spawn(async move |state, cx| {
client.authenticate_and_connect(true, &cx).await?; client
this.update(cx, |_, cx| cx.notify()) .authenticate_and_connect(true, &cx)
.await
.into_response()?;
state.update(cx, |_, cx| cx.notify())
}) })
} }

68
crates/lsp/src/lsp.rs
View file

@ -5,7 +5,12 @@ pub use lsp_types::*;
use anyhow::{Context as _, Result, anyhow}; use anyhow::{Context as _, Result, anyhow};
use collections::HashMap; use collections::HashMap;
use futures::{AsyncRead, AsyncWrite, Future, FutureExt, channel::oneshot, io::BufWriter, select}; use futures::{
AsyncRead, AsyncWrite, Future, FutureExt,
channel::oneshot::{self, Canceled},
io::BufWriter,
select,
};
use gpui::{App, AppContext as _, AsyncApp, BackgroundExecutor, SharedString, Task}; use gpui::{App, AppContext as _, AsyncApp, BackgroundExecutor, SharedString, Task};
use notification::DidChangeWorkspaceFolders; use notification::DidChangeWorkspaceFolders;
use parking_lot::{Mutex, RwLock}; use parking_lot::{Mutex, RwLock};
@ -39,7 +44,7 @@ use std::{
time::{Duration, Instant}, time::{Duration, Instant},
}; };
use std::{path::Path, process::Stdio}; use std::{path::Path, process::Stdio};
use util::{ResultExt, TryFutureExt}; use util::{ConnectionResult, ResultExt, TryFutureExt};
const JSON_RPC_VERSION: &str = "2.0"; const JSON_RPC_VERSION: &str = "2.0";
const CONTENT_LEN_HEADER: &str = "Content-Length: "; const CONTENT_LEN_HEADER: &str = "Content-Length: ";
@ -259,7 +264,7 @@ struct Error {
message: String, message: String,
} }
pub trait LspRequestFuture<O>: Future<Output = O> { pub trait LspRequestFuture<O>: Future<Output = ConnectionResult<O>> {
fn id(&self) -> i32; fn id(&self) -> i32;
} }
@ -284,7 +289,10 @@ impl<F: Future> Future for LspRequest<F> {
} }
} }
impl<F: Future> LspRequestFuture<F::Output> for LspRequest<F> { impl<F, O> LspRequestFuture<O> for LspRequest<F>
where
F: Future<Output = ConnectionResult<O>>,
{
fn id(&self) -> i32 { fn id(&self) -> i32 {
self.id self.id
} }
@ -824,7 +832,17 @@ impl LanguageServer {
cx: &App, cx: &App,
) -> Task<Result<Arc<Self>>> { ) -> Task<Result<Arc<Self>>> {
cx.spawn(async move |_| { cx.spawn(async move |_| {
let response = self.request::<request::Initialize>(params).await?; let response = self
.request::<request::Initialize>(params)
.await
.into_response()
.with_context(|| {
format!(
"initializing server {}, id {}",
self.name(),
self.server_id()
)
})?;
if let Some(info) = response.server_info { if let Some(info) = response.server_info {
self.process_name = info.name.into(); self.process_name = info.name.into();
} }
@ -863,7 +881,13 @@ impl LanguageServer {
select! { select! {
request_result = shutdown_request.fuse() => { request_result = shutdown_request.fuse() => {
request_result?; match request_result {
ConnectionResult::Timeout => {
log::warn!("timeout waiting for language server {name} to shutdown");
},
ConnectionResult::ConnectionReset => {},
ConnectionResult::Result(r) => r?,
}
} }
_ = timer => { _ = timer => {
@ -1084,7 +1108,7 @@ impl LanguageServer {
pub fn request<T: request::Request>( pub fn request<T: request::Request>(
&self, &self,
params: T::Params, params: T::Params,
) -> impl LspRequestFuture<Result<T::Result>> + use<T> ) -> impl LspRequestFuture<T::Result> + use<T>
where where
T::Result: 'static + Send, T::Result: 'static + Send,
{ {
@ -1097,15 +1121,18 @@ impl LanguageServer {
) )
} }
fn request_internal<T: request::Request>( fn request_internal<T>(
next_id: &AtomicI32, next_id: &AtomicI32,
response_handlers: &Mutex<Option<HashMap<RequestId, ResponseHandler>>>, response_handlers: &Mutex<Option<HashMap<RequestId, ResponseHandler>>>,
outbound_tx: &channel::Sender<String>, outbound_tx: &channel::Sender<String>,
executor: &BackgroundExecutor, executor: &BackgroundExecutor,
params: T::Params, params: T::Params,
) -> impl LspRequestFuture<Result<T::Result>> + use<T> ) -> impl LspRequestFuture<T::Result> + use<T>
where where
T::Result: 'static + Send, T::Result: 'static + Send,
T: request::Request,
// TODO kb
// <T as lsp_types::request::Request>::Result: ConnectionResult,
{ {
let id = next_id.fetch_add(1, SeqCst); let id = next_id.fetch_add(1, SeqCst);
let message = serde_json::to_string(&Request { let message = serde_json::to_string(&Request {
@ -1120,7 +1147,7 @@ impl LanguageServer {
let handle_response = response_handlers let handle_response = response_handlers
.lock() .lock()
.as_mut() .as_mut()
.ok_or_else(|| anyhow!("server shut down")) .context("server shut down")
.map(|handlers| { .map(|handlers| {
let executor = executor.clone(); let executor = executor.clone();
handlers.insert( handlers.insert(
@ -1153,8 +1180,12 @@ impl LanguageServer {
let mut timeout = executor.timer(LSP_REQUEST_TIMEOUT).fuse(); let mut timeout = executor.timer(LSP_REQUEST_TIMEOUT).fuse();
let started = Instant::now(); let started = Instant::now();
LspRequest::new(id, async move { LspRequest::new(id, async move {
handle_response?; if let Err(e) = handle_response {
send?; return ConnectionResult::Result(Err(e));
}
if let Err(e) = send {
return ConnectionResult::Result(Err(e));
}
let cancel_on_drop = util::defer(move || { let cancel_on_drop = util::defer(move || {
if let Some(outbound_tx) = outbound_tx.upgrade() { if let Some(outbound_tx) = outbound_tx.upgrade() {
@ -1174,12 +1205,18 @@ impl LanguageServer {
let elapsed = started.elapsed(); let elapsed = started.elapsed();
log::trace!("Took {elapsed:?} to receive response to {method:?} id {id}"); log::trace!("Took {elapsed:?} to receive response to {method:?} id {id}");
cancel_on_drop.abort(); cancel_on_drop.abort();
response? match response {
Ok(response_result) => ConnectionResult::Result(response_result),
Err(Canceled) => {
log::error!("Server reset connection for a request {method:?} id {id}");
ConnectionResult::ConnectionReset
},
}
} }
_ = timeout => { _ = timeout => {
log::error!("Cancelled LSP request task for {method:?} id {id} which took over {LSP_REQUEST_TIMEOUT:?}"); log::error!("Cancelled LSP request task for {method:?} id {id} which took over {LSP_REQUEST_TIMEOUT:?}");
anyhow::bail!("LSP request timeout"); ConnectionResult::Timeout
} }
} }
}) })
@ -1506,7 +1543,7 @@ impl FakeLanguageServer {
} }
/// See [`LanguageServer::request`]. /// See [`LanguageServer::request`].
pub async fn request<T>(&self, params: T::Params) -> Result<T::Result> pub async fn request<T>(&self, params: T::Params) -> ConnectionResult<T::Result>
where where
T: request::Request, T: request::Request,
T::Result: 'static + Send, T::Result: 'static + Send,
@ -1608,6 +1645,7 @@ impl FakeLanguageServer {
token: NumberOrString::String(token.clone()), token: NumberOrString::String(token.clone()),
}) })
.await .await
.into_response()
.unwrap(); .unwrap();
self.notify::<notification::Progress>(&ProgressParams { self.notify::<notification::Progress>(&ProgressParams {
token: NumberOrString::String(token), token: NumberOrString::String(token),

8
crates/prettier/src/prettier.rs
View file

@ -452,7 +452,12 @@ impl Prettier {
})? })?
.context("prettier params calculation")?; .context("prettier params calculation")?;
let response = local.server.request::<Format>(params).await?; let response = local
.server
.request::<Format>(params)
.await
.into_response()
.context("prettier format")?;
let diff_task = buffer.update(cx, |buffer, cx| buffer.diff(response.text, cx))?; let diff_task = buffer.update(cx, |buffer, cx| buffer.diff(response.text, cx))?;
Ok(diff_task.await) Ok(diff_task.await)
} }
@ -482,6 +487,7 @@ impl Prettier {
.server .server
.request::<ClearCache>(()) .request::<ClearCache>(())
.await .await
.into_response()
.context("prettier clear cache"), .context("prettier clear cache"),
#[cfg(any(test, feature = "test-support"))] #[cfg(any(test, feature = "test-support"))]
Self::Test(_) => Ok(()), Self::Test(_) => Ok(()),

94
crates/project/src/lsp_store.rs
View file

@ -86,7 +86,7 @@ use std::{
use text::{Anchor, BufferId, LineEnding, OffsetRangeExt}; use text::{Anchor, BufferId, LineEnding, OffsetRangeExt};
use url::Url; use url::Url;
use util::{ use util::{
ResultExt, TryFutureExt as _, debug_panic, defer, maybe, merge_json_value_into, ResultExt as _, debug_panic, defer, maybe, merge_json_value_into,
paths::{PathExt, SanitizedPath}, paths::{PathExt, SanitizedPath},
post_inc, post_inc,
}; };
@ -1769,7 +1769,8 @@ impl LocalLspStore {
..Default::default() ..Default::default()
}, },
) )
.await; .await
.into_response();
if execute_command_result.is_err() { if execute_command_result.is_err() {
zlog::error!( zlog::error!(
@ -1894,7 +1895,8 @@ impl LocalLspStore {
options: lsp_command::lsp_formatting_options(settings), options: lsp_command::lsp_formatting_options(settings),
work_done_progress_params: Default::default(), work_done_progress_params: Default::default(),
}) })
.await? .await
.into_response()?
{ {
edits.get_or_insert_with(Vec::new).append(&mut edit); edits.get_or_insert_with(Vec::new).append(&mut edit);
} }
@ -1945,7 +1947,8 @@ impl LocalLspStore {
options: lsp_command::lsp_formatting_options(settings), options: lsp_command::lsp_formatting_options(settings),
work_done_progress_params: Default::default(), work_done_progress_params: Default::default(),
}) })
.await? .await
.into_response()?
} else if matches!(range_formatting_provider, Some(p) if *p != OneOf::Left(false)) { } else if matches!(range_formatting_provider, Some(p) if *p != OneOf::Left(false)) {
let _timer = zlog::time!(logger => "format-range"); let _timer = zlog::time!(logger => "format-range");
let buffer_start = lsp::Position::new(0, 0); let buffer_start = lsp::Position::new(0, 0);
@ -1957,7 +1960,8 @@ impl LocalLspStore {
options: lsp_command::lsp_formatting_options(settings), options: lsp_command::lsp_formatting_options(settings),
work_done_progress_params: Default::default(), work_done_progress_params: Default::default(),
}) })
.await? .await
.into_response()?
} else { } else {
None None
}; };
@ -2065,7 +2069,8 @@ impl LocalLspStore {
*lsp_action = Box::new( *lsp_action = Box::new(
lang_server lang_server
.request::<lsp::request::CodeActionResolveRequest>(*lsp_action.clone()) .request::<lsp::request::CodeActionResolveRequest>(*lsp_action.clone())
.await?, .await
.into_response()?,
); );
} }
} }
@ -2073,7 +2078,8 @@ impl LocalLspStore {
if !action.resolved && GetCodeLens::can_resolve_lens(&lang_server.capabilities()) { if !action.resolved && GetCodeLens::can_resolve_lens(&lang_server.capabilities()) {
*lens = lang_server *lens = lang_server
.request::<lsp::request::CodeLensResolve>(lens.clone()) .request::<lsp::request::CodeLensResolve>(lens.clone())
.await?; .await
.into_response()?;
} }
} }
LspAction::Command(_) => {} LspAction::Command(_) => {}
@ -2578,7 +2584,9 @@ impl LocalLspStore {
arguments: command.arguments.clone().unwrap_or_default(), arguments: command.arguments.clone().unwrap_or_default(),
..Default::default() ..Default::default()
}) })
.await?; .await
.into_response()
.context("execute command")?;
lsp_store.update(cx, |this, _| { lsp_store.update(cx, |this, _| {
if let LspStoreMode::Local(mode) = &mut this.mode { if let LspStoreMode::Local(mode) = &mut this.mode {
@ -4223,7 +4231,7 @@ impl LspStore {
language_server.name(), language_server.name(),
err err
); );
log::warn!("{}", message); log::warn!("{message}");
return Task::ready(Err(anyhow!(message))); return Task::ready(Err(anyhow!(message)));
} }
}; };
@ -4268,7 +4276,7 @@ impl LspStore {
None None
}; };
let result = lsp_request.await; let result = lsp_request.await.into_response();
let response = result.map_err(|err| { let response = result.map_err(|err| {
let message = format!( let message = format!(
@ -4277,7 +4285,7 @@ impl LspStore {
language_server.name(), language_server.name(),
err err
); );
log::warn!("{}", message); log::warn!("{message}");
anyhow!(message) anyhow!(message)
})?; })?;
@ -4521,15 +4529,14 @@ impl LspStore {
.remove(&lang_server.server_id()); .remove(&lang_server.server_id());
})?; })?;
let result = lang_server let _result = lang_server
.request::<lsp::request::ExecuteCommand>(lsp::ExecuteCommandParams { .request::<lsp::request::ExecuteCommand>(lsp::ExecuteCommandParams {
command: command.command.clone(), command: command.command.clone(),
arguments: command.arguments.clone().unwrap_or_default(), arguments: command.arguments.clone().unwrap_or_default(),
..Default::default() ..lsp::ExecuteCommandParams::default()
}) })
.await; .await.into_response()
.context("execute command")?;
result?;
return this.update(cx, |this, _| { return this.update(cx, |this, _| {
this.as_local_mut() this.as_local_mut()
@ -4649,6 +4656,7 @@ impl LspStore {
); );
let resolved_hint = resolve_task let resolved_hint = resolve_task
.await .await
.into_response()
.context("inlay hint resolve LSP request")?; .context("inlay hint resolve LSP request")?;
let resolved_hint = InlayHints::lsp_to_project_hint( let resolved_hint = InlayHints::lsp_to_project_hint(
resolved_hint, resolved_hint,
@ -5232,7 +5240,10 @@ impl LspStore {
} }
} }
}; };
let resolved_completion = request.await?; let resolved_completion = request
.await
.into_response()
.context("resolve completion")?;
let mut updated_insert_range = None; let mut updated_insert_range = None;
if let Some(text_edit) = resolved_completion.text_edit.as_ref() { if let Some(text_edit) = resolved_completion.text_edit.as_ref() {
@ -5847,27 +5858,30 @@ impl LspStore {
..Default::default() ..Default::default()
}, },
) )
.log_err()
.map(move |response| { .map(move |response| {
let lsp_symbols = response.flatten().map(|symbol_response| match symbol_response { let lsp_symbols = response.into_response()
lsp::WorkspaceSymbolResponse::Flat(flat_responses) => { .context("workspace symbols request")
flat_responses.into_iter().map(|lsp_symbol| { .log_err()
(lsp_symbol.name, lsp_symbol.kind, lsp_symbol.location) .flatten()
}).collect::<Vec<_>>() .map(|symbol_response| match symbol_response {
} lsp::WorkspaceSymbolResponse::Flat(flat_responses) => {
lsp::WorkspaceSymbolResponse::Nested(nested_responses) => { flat_responses.into_iter().map(|lsp_symbol| {
nested_responses.into_iter().filter_map(|lsp_symbol| { (lsp_symbol.name, lsp_symbol.kind, lsp_symbol.location)
let location = match lsp_symbol.location { }).collect::<Vec<_>>()
OneOf::Left(location) => location, }
OneOf::Right(_) => { lsp::WorkspaceSymbolResponse::Nested(nested_responses) => {
log::error!("Unexpected: client capabilities forbid symbol resolutions in workspace.symbol.resolveSupport"); nested_responses.into_iter().filter_map(|lsp_symbol| {
return None let location = match lsp_symbol.location {
} OneOf::Left(location) => location,
}; OneOf::Right(_) => {
Some((lsp_symbol.name, lsp_symbol.kind, location)) log::error!("Unexpected: client capabilities forbid symbol resolutions in workspace.symbol.resolveSupport");
}).collect::<Vec<_>>() return None
} }
}).unwrap_or_default(); };
Some((lsp_symbol.name, lsp_symbol.kind, location))
}).collect::<Vec<_>>()
}
}).unwrap_or_default();
WorkspaceSymbolsResult { WorkspaceSymbolsResult {
server_id, server_id,
@ -7517,8 +7531,10 @@ impl LspStore {
.request::<WillRenameFiles>(RenameFilesParams { .request::<WillRenameFiles>(RenameFilesParams {
files: vec![FileRename { old_uri, new_uri }], files: vec![FileRename { old_uri, new_uri }],
}) })
.log_err()
.await .await
.into_response()
.context("will rename files")
.log_err()
.flatten()?; .flatten()?;
LocalLspStore::deserialize_workspace_edit( LocalLspStore::deserialize_workspace_edit(
@ -7788,6 +7804,8 @@ impl LspStore {
server server
.request::<lsp::request::ResolveCompletionItem>(lsp_completion) .request::<lsp::request::ResolveCompletionItem>(lsp_completion)
.await .await
.into_response()
.context("resolve completion item")
} else { } else {
anyhow::Ok(lsp_completion) anyhow::Ok(lsp_completion)
} }

5
crates/project/src/project.rs
View file

@ -1224,7 +1224,10 @@ impl Project {
fs: Arc<dyn Fs>, fs: Arc<dyn Fs>,
cx: AsyncApp, cx: AsyncApp,
) -> Result<Entity<Self>> { ) -> Result<Entity<Self>> {
client.authenticate_and_connect(true, &cx).await?; client
.authenticate_and_connect(true, &cx)
.await
.into_response()?;
let subscriptions = [ let subscriptions = [
EntitySubscription::Project(client.subscribe_to_entity::<Self>(remote_id)?), EntitySubscription::Project(client.subscribe_to_entity::<Self>(remote_id)?),

2
crates/project/src/project_tests.rs
View file

@ -1089,6 +1089,7 @@ async fn test_reporting_fs_changes_to_language_servers(cx: &mut gpui::TestAppCon
}], }],
}) })
.await .await
.into_response()
.unwrap(); .unwrap();
fake_server.handle_notification::<lsp::notification::DidChangeWatchedFiles, _>({ fake_server.handle_notification::<lsp::notification::DidChangeWatchedFiles, _>({
let file_changes = file_changes.clone(); let file_changes = file_changes.clone();
@ -3431,6 +3432,7 @@ async fn test_apply_code_actions_with_commands(cx: &mut gpui::TestAppContext) {
}, },
) )
.await .await
.into_response()
.unwrap(); .unwrap();
Ok(Some(json!(null))) Ok(Some(json!(null)))
} }

1
crates/title_bar/src/title_bar.rs
View file

@ -664,6 +664,7 @@ impl TitleBar {
client client
.authenticate_and_connect(true, &cx) .authenticate_and_connect(true, &cx)
.await .await
.into_response()
.notify_async_err(cx); .notify_async_err(cx);
}) })
.detach(); .detach();

23
crates/util/src/util.rs
View file

@ -1025,6 +1025,29 @@ pub fn get_system_shell() -> String {
} }
} }
#[derive(Debug)]
pub enum ConnectionResult<O> {
Timeout,
ConnectionReset,
Result(anyhow::Result<O>),
}
impl<O> ConnectionResult<O> {
pub fn into_response(self) -> anyhow::Result<O> {
match self {
ConnectionResult::Timeout => anyhow::bail!("Request timed out"),
ConnectionResult::ConnectionReset => anyhow::bail!("Server reset the connection"),
ConnectionResult::Result(r) => r,
}
}
}
impl<O> From<anyhow::Result<O>> for ConnectionResult<O> {
fn from(result: anyhow::Result<O>) -> Self {
ConnectionResult::Result(result)
}
}
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::*; use super::*;

2
crates/workspace/src/notifications.rs
View file

@ -693,7 +693,7 @@ pub mod simple_message_notification {
) )
} else { } else {
Tooltip::for_action( Tooltip::for_action(
"Close.", "Close",
&menu::Cancel, &menu::Cancel,
window, window,
cx, cx,

6
crates/worktree/src/worktree.rs
View file

@ -4358,11 +4358,7 @@ impl BackgroundScanner {
let canonical_path = match self.fs.canonicalize(&child_abs_path).await { let canonical_path = match self.fs.canonicalize(&child_abs_path).await {
Ok(path) => path, Ok(path) => path,
Err(err) => { Err(err) => {
log::error!( log::error!("error reading target of symlink {child_abs_path:?}: {err:#}",);
"error reading target of symlink {:?}: {:?}",
child_abs_path,
err
);
continue; continue;
} }
}; };

35
crates/zed/src/main.rs
View file

@ -44,7 +44,7 @@ use theme::{
ActiveTheme, IconThemeNotFoundError, SystemAppearance, ThemeNotFoundError, ThemeRegistry, ActiveTheme, IconThemeNotFoundError, SystemAppearance, ThemeNotFoundError, ThemeRegistry,
ThemeSettings, ThemeSettings,
}; };
use util::{ResultExt, TryFutureExt, maybe}; use util::{ConnectionResult, ResultExt, TryFutureExt, maybe};
use uuid::Uuid; use uuid::Uuid;
use welcome::{BaseKeymap, FIRST_OPEN, show_welcome_view}; use welcome::{BaseKeymap, FIRST_OPEN, show_welcome_view};
use workspace::{AppState, SerializedWorkspaceLocation, WorkspaceSettings, WorkspaceStore}; use workspace::{AppState, SerializedWorkspaceLocation, WorkspaceSettings, WorkspaceStore};
@ -612,9 +612,17 @@ fn main() {
cx.spawn({ cx.spawn({
let client = app_state.client.clone(); let client = app_state.client.clone();
async move |cx| authenticate(client, &cx).await async move |cx| match authenticate(client, &cx).await {
ConnectionResult::Timeout => log::error!("Timeout during initial auth"),
ConnectionResult::ConnectionReset => {
log::error!("Connection reset during initial auth")
}
ConnectionResult::Result(r) => {
r.log_err();
}
}
}) })
.detach_and_log_err(cx); .detach();
let urls: Vec<_> = args let urls: Vec<_> = args
.paths_or_urls .paths_or_urls
@ -727,7 +735,15 @@ fn handle_open_request(request: OpenRequest, app_state: Arc<AppState>, cx: &mut
let client = app_state.client.clone(); let client = app_state.client.clone();
// we continue even if authentication fails as join_channel/ open channel notes will // we continue even if authentication fails as join_channel/ open channel notes will
// show a visible error message. // show a visible error message.
authenticate(client, &cx).await.log_err(); match authenticate(client, &cx).await {
ConnectionResult::Timeout => {
log::error!("Timeout during open request handling")
}
ConnectionResult::ConnectionReset => {
log::error!("Connection reset during open request handling")
}
ConnectionResult::Result(r) => r?,
};
if let Some(channel_id) = request.join_channel { if let Some(channel_id) = request.join_channel {
cx.update(|cx| { cx.update(|cx| {
@ -777,17 +793,18 @@ fn handle_open_request(request: OpenRequest, app_state: Arc<AppState>, cx: &mut
} }
} }
async fn authenticate(client: Arc<Client>, cx: &AsyncApp) -> Result<()> { async fn authenticate(client: Arc<Client>, cx: &AsyncApp) -> ConnectionResult<()> {
if stdout_is_a_pty() { if stdout_is_a_pty() {
if client::IMPERSONATE_LOGIN.is_some() { if client::IMPERSONATE_LOGIN.is_some() {
client.authenticate_and_connect(false, cx).await?; return client.authenticate_and_connect(false, cx).await;
} else if client.has_credentials(cx).await { } else if client.has_credentials(cx).await {
client.authenticate_and_connect(true, cx).await?; return client.authenticate_and_connect(true, cx).await;
} }
} else if client.has_credentials(cx).await { } else if client.has_credentials(cx).await {
client.authenticate_and_connect(true, cx).await?; return client.authenticate_and_connect(true, cx).await;
} }
Ok::<_, anyhow::Error>(())
ConnectionResult::Result(Ok(()))
} }
async fn system_id() -> Result<IdType> { async fn system_id() -> Result<IdType> {

5
crates/zeta/src/onboarding_modal.rs
View file

@ -139,7 +139,10 @@ impl ZedPredictModal {
self.sign_in_status = SignInStatus::Waiting; self.sign_in_status = SignInStatus::Waiting;
cx.spawn(async move |this, cx| { cx.spawn(async move |this, cx| {
let result = client.authenticate_and_connect(true, &cx).await; let result = client
.authenticate_and_connect(true, &cx)
.await
.into_response();
let status = match result { let status = match result {
Ok(_) => SignInStatus::SignedIn, Ok(_) => SignInStatus::SignedIn,