Preserve LSP message order (#65856)

Closes https://github.com/zed-industries/zed/issues/65667

Release Notes:

- Fixed LSP formatting requests bringing in incorrect changes sometimes
This commit is contained in:
Kirill Bulatov authored and GitHub committed 2026-10-21 15:35:05 +00:00
1 parent cc23f35a7d
commit e80af52575
3 files changed
+133 -65

No files matched your search

+20 -20
View File
@@ -22175,28 +22175,28 @@ async fn test_strip_whitespace_and_format_via_lsp(cx: &mut TestAppContext) {
cx.lsp
.set_request_handler::<lsp::request::Formatting, _, _>({
move |_, _| {
assert_eq!(
buffer_changes.lock().get(1..),
Some(
[
(
lsp::Range::new(lsp::Position::new(0, 3), lsp::Position::new(0, 4)),
String::new(),
),
(
lsp::Range::new(lsp::Position::new(2, 5), lsp::Position::new(2, 6)),
String::new(),
),
(
lsp::Range::new(lsp::Position::new(3, 4), lsp::Position::new(3, 4)),
language::LineEnding::default().as_str().to_owned(),
),
]
.as_slice(),
),
);
// Insert blank lines between each line of the buffer.
async move {
// TODO: this assertion is not reliably true. Currently nothing guarantees that we deliver
// DidChangedTextDocument to the LSP before sending the formatting request.
// assert_eq!(
// &buffer_changes.lock()[1..],
// &[
// (
// lsp::Range::new(lsp::Position::new(0, 3), lsp::Position::new(0, 4)),
// "".into()
// ),
// (
// lsp::Range::new(lsp::Position::new(2, 5), lsp::Position::new(2, 6)),
// "".into()
// ),
// (
// lsp::Range::new(lsp::Position::new(3, 4), lsp::Position::new(3, 4)),
// "\n".into()
// ),
// ]
// );
Ok(Some(vec![
lsp::TextEdit {
range: lsp::Range::new(
+10 -5
View File
@@ -7292,13 +7292,18 @@ let c = 3;"#
})
.await
.unwrap();
let editor =
cx.add_window(|window, cx| Editor::for_buffer(buffer, Some(project), window, cx));
let mut fake_server_a = fake_servers_a.next().await.unwrap();
let mut fake_server_b = fake_servers_b.next().await.unwrap();
fake_server_a
.receive_notification::<lsp::notification::DidOpenTextDocument>()
.await;
fake_server_b
.receive_notification::<lsp::notification::DidOpenTextDocument>()
.await;
cx.executor().run_until_parked();
let _fake_server_a = fake_servers_a.next().await.unwrap();
let fake_server_b = fake_servers_b.next().await.unwrap();
let editor =
cx.add_window(|window, cx| Editor::for_buffer(buffer, Some(project), window, cx));
editor
.update(cx, |editor, window, cx| {
editor.set_visible_line_count(50.0, window, cx);
+103 -40
View File
@@ -109,14 +109,16 @@ pub struct LanguageServerBinaryOptions {
pub pre_release: bool,
}
struct NotificationSerializer(Box<dyn FnOnce() -> String + Send + Sync>);
enum OutboundMessage {
Serialized(String),
PendingSerialization(Box<dyn FnOnce() -> String + Send + Sync>),
}
/// A running language server process.
pub struct LanguageServer {
server_id: LanguageServerId,
next_id: AtomicI32,
outbound_tx: async_channel::Sender<String>,
notification_tx: async_channel::Sender<NotificationSerializer>,
outbound_tx: async_channel::Sender<OutboundMessage>,
name: LanguageServerName,
version: Option<SharedString>,
process_name: Arc<str>,
@@ -600,7 +602,8 @@ impl LanguageServer {
Stderr: AsyncRead + Unpin + Send + 'static,
F: Fn(&NotificationOrRequest) -> bool + 'static + Send + Sync + Clone,
{
let (outbound_tx, outbound_rx) = async_channel::unbounded::<String>();
let (outbound_tx, outbound_rx) = async_channel::unbounded::<OutboundMessage>();
let (serialized_tx, serialized_rx) = async_channel::unbounded::<String>();
let (output_done_tx, output_done_rx) = barrier::channel();
let notification_handlers =
Arc::new(Mutex::new(HashMap::<_, NotificationHandler>::default()));
@@ -622,7 +625,10 @@ impl LanguageServer {
result: None,
};
if let Ok(response) = serde_json::to_string(&response) {
response_channel.send(response).await.ok();
response_channel
.send(OutboundMessage::Serialized(response))
.await
.ok();
}
}
}
@@ -663,7 +669,7 @@ impl LanguageServer {
let output_task = cx.background_spawn({
Self::handle_outgoing_messages(
stdin,
outbound_rx,
serialized_rx,
output_done_tx,
response_handlers.clone(),
io_handlers.clone(),
@@ -676,25 +682,22 @@ impl LanguageServer {
}
.into();
let (notification_tx, notification_rx) =
async_channel::unbounded::<NotificationSerializer>();
cx.background_spawn({
let outbound_tx = outbound_tx.clone();
async move {
while let Ok(serializer) = notification_rx.recv().await {
let serialized = (serializer.0)();
let Ok(_) = outbound_tx.send(serialized).await else {
return;
};
}
outbound_tx.close();
cx.background_spawn(async move {
while let Ok(message) = outbound_rx.recv().await {
let serialized = match message {
OutboundMessage::Serialized(message) => message,
OutboundMessage::PendingSerialization(serialize) => serialize(),
};
let Ok(_) = serialized_tx.send(serialized).await else {
return;
};
}
serialized_tx.close();
})
.detach();
Self {
server_id,
notification_handlers,
notification_tx,
response_handlers,
pending_respond_tasks,
io_handlers,
@@ -1203,13 +1206,11 @@ impl LanguageServer {
let next_id = AtomicI32::new(self.next_id.load(SeqCst));
let outbound_tx = self.outbound_tx.clone();
let executor = self.executor.clone();
let notification_serializers = self.notification_tx.clone();
let mut output_done = self.output_done_rx.lock().take().unwrap();
let shutdown_request = Self::request_internal::<request::Shutdown>(
&next_id,
&response_handlers,
&outbound_tx,
&notification_serializers,
&executor,
SERVER_SHUTDOWN_TIMEOUT,
(),
@@ -1246,8 +1247,8 @@ impl LanguageServer {
};
response_handlers.lock().take();
Self::notify_internal::<notification::Exit>(&notification_serializers, ()).ok();
notification_serializers.close();
Self::notify_internal::<notification::Exit>(&outbound_tx, ()).ok();
outbound_tx.close();
if !shutdown_timed_out {
select! {
_ = output_done.recv().fuse() => {},
@@ -1391,7 +1392,9 @@ impl LanguageServer {
if let Some(response) =
serde_json::to_string(&response).log_err()
{
outbound_tx.try_send(response).ok();
outbound_tx
.try_send(OutboundMessage::Serialized(response))
.ok();
}
pending_respond_tasks.lock().remove(&id);
}
@@ -1411,7 +1414,9 @@ impl LanguageServer {
)),
};
if let Some(response) = serde_json::to_string(&response).log_err() {
outbound_tx.try_send(response).ok();
outbound_tx
.try_send(OutboundMessage::Serialized(response))
.ok();
}
}
}
@@ -1512,7 +1517,6 @@ impl LanguageServer {
&self.next_id,
&self.response_handlers,
&self.outbound_tx,
&self.notification_tx,
&self.executor,
request_timeout,
params,
@@ -1522,8 +1526,7 @@ impl LanguageServer {
fn request_internal_with_timer<T, U>(
next_id: &AtomicI32,
response_handlers: &Arc<Mutex<Option<HashMap<RequestId, ResponseHandler>>>>,
outbound_tx: &async_channel::Sender<String>,
notification_serializers: &async_channel::Sender<NotificationSerializer>,
outbound_tx: &async_channel::Sender<OutboundMessage>,
executor: &BackgroundExecutor,
timer: U,
params: T::Params,
@@ -1571,11 +1574,11 @@ impl LanguageServer {
});
let send = outbound_tx
.try_send(message)
.try_send(OutboundMessage::Serialized(message))
.context("failed to write to language server's stdin");
let response_handlers = Arc::clone(response_handlers);
let notification_serializers = notification_serializers.downgrade();
let outbound_tx = outbound_tx.downgrade();
let started = Instant::now();
LspRequest::new(id, async move {
if let Err(e) = handle_response {
@@ -1586,9 +1589,9 @@ impl LanguageServer {
}
let cancel_on_drop = gpui_util::defer(move || {
if let Some(notification_serializers) = notification_serializers.upgrade() {
if let Some(outbound_tx) = outbound_tx.upgrade() {
Self::notify_internal::<notification::Cancel>(
&notification_serializers,
&outbound_tx,
CancelParams {
id: NumberOrString::Number(id),
},
@@ -1632,8 +1635,7 @@ impl LanguageServer {
fn request_internal<T>(
next_id: &AtomicI32,
response_handlers: &Arc<Mutex<Option<HashMap<RequestId, ResponseHandler>>>>,
outbound_tx: &async_channel::Sender<String>,
notification_serializers: &async_channel::Sender<NotificationSerializer>,
outbound_tx: &async_channel::Sender<OutboundMessage>,
executor: &BackgroundExecutor,
request_timeout: Duration,
params: T::Params,
@@ -1646,7 +1648,6 @@ impl LanguageServer {
next_id,
response_handlers,
outbound_tx,
notification_serializers,
executor,
Self::request_timeout_future(executor.clone(), request_timeout),
params,
@@ -1679,15 +1680,14 @@ impl LanguageServer {
///
/// [LSP Specification](https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#notificationMessage)
pub fn notify<T: notification::Notification>(&self, params: T::Params) -> Result<()> {
let outbound = self.notification_tx.clone();
Self::notify_internal::<T>(&outbound, params)
Self::notify_internal::<T>(&self.outbound_tx, params)
}
fn notify_internal<T: notification::Notification>(
outbound_tx: &async_channel::Sender<NotificationSerializer>,
outbound_tx: &async_channel::Sender<OutboundMessage>,
params: T::Params,
) -> Result<()> {
let serializer = NotificationSerializer(Box::new(move || {
let message = OutboundMessage::PendingSerialization(Box::new(move || {
serde_json::to_string(&Notification {
jsonrpc: JSON_RPC_VERSION,
method: T::METHOD,
@@ -1696,7 +1696,7 @@ impl LanguageServer {
.unwrap()
}));
outbound_tx.send_blocking(serializer)?;
outbound_tx.send_blocking(message)?;
Ok(())
}
@@ -2260,6 +2260,69 @@ mod tests {
fake.receive_notification::<notification::Exit>().await;
}
#[gpui::test]
async fn test_notifications_and_requests_preserve_send_order(cx: &mut TestAppContext) {
let (server, mut stdout, output) = shutdown_test_server(None, cx);
let uri = Uri::from_str("file:///test.py").expect("valid document URI");
let notify_change = |version, text: &str| {
server
.notify::<notification::DidChangeTextDocument>(DidChangeTextDocumentParams {
text_document: VersionedTextDocumentIdentifier::new(uri.clone(), version),
content_changes: vec![TextDocumentContentChangeEvent {
range: None,
range_length: None,
text: text.to_owned(),
}],
})
.expect("document change must be queued");
};
notify_change(1, "α = 1\n");
let formatting = server.request::<request::Formatting>(
DocumentFormattingParams {
text_document: TextDocumentIdentifier::new(uri.clone()),
options: FormattingOptions {
tab_size: 4,
insert_spaces: true,
..FormattingOptions::default()
},
work_done_progress_params: WorkDoneProgressParams::default(),
},
DEFAULT_LSP_REQUEST_TIMEOUT,
);
notify_change(2, "α = 2\n");
cx.run_until_parked();
let expected = [
r#"{"jsonrpc":"2.0","method":"textDocument/didChange","params":{"textDocument":{"uri":"file:///test.py","version":1},"contentChanges":[{"text":"α = 1\n"}]}}"#,
r#"{"jsonrpc":"2.0","id":0,"method":"textDocument/formatting","params":{"textDocument":{"uri":"file:///test.py"},"options":{"tabSize":4,"insertSpaces":true}}}"#,
r#"{"jsonrpc":"2.0","method":"textDocument/didChange","params":{"textDocument":{"uri":"file:///test.py","version":2},"contentChanges":[{"text":"α = 2\n"}]}}"#,
]
.into_iter()
.flat_map(framed_test_message)
.collect::<Vec<_>>();
assert_eq!(
String::from_utf8(output.lock().bytes.clone()).expect("UTF-8 output"),
String::from_utf8(expected.clone()).expect("UTF-8 expected output"),
);
stdout
.write_all(&framed_test_message(
r#"{"jsonrpc":"2.0","id":0,"result":[]}"#,
))
.await
.expect("formatting response must reach stdout");
assert_eq!(
formatting
.await
.into_response()
.expect("formatting response"),
Some(Vec::new()),
);
cx.run_until_parked();
assert_eq!(output.lock().bytes, expected);
}
#[gpui::test]
async fn test_shutdown_bounds_output_drain(cx: &mut TestAppContext) {
for remaining_flushes in [Some(0), Some(1), None] {