Send diagnostic summaries synchronously

crates/collab/src/rpc.rs

@@ -201,7 +201,7 @@ impl Server {
             .add_request_handler(update_worktree)
             .add_message_handler(start_language_server)
             .add_message_handler(update_language_server)
-            .add_request_handler(update_diagnostic_summary)
+            .add_message_handler(update_diagnostic_summary)
             .add_request_handler(forward_project_request::<proto::GetHover>)
             .add_request_handler(forward_project_request::<proto::GetDefinition>)
             .add_request_handler(forward_project_request::<proto::GetTypeDefinition>)
@@ -1187,14 +1187,13 @@ async fn update_worktree(
 }
 
 async fn update_diagnostic_summary(
-    request: proto::UpdateDiagnosticSummary,
+    message: proto::UpdateDiagnosticSummary,
-    response: Response<proto::UpdateDiagnosticSummary>,
     session: Session,
 ) -> Result<()> {
     let guest_connection_ids = session
         .db()
         .await
-        .update_diagnostic_summary(&request, session.connection_id)
+        .update_diagnostic_summary(&message, session.connection_id)
         .await?;
 
     broadcast(
@@ -1203,11 +1202,10 @@ async fn update_diagnostic_summary(
         |connection_id| {
             session
                 .peer
-                .forward_send(session.connection_id, connection_id, request.clone())
+                .forward_send(session.connection_id, connection_id, message.clone())
         },
     );
 
-    response.send(proto::Ack {})?;
     Ok(())
 }
 

crates/project/src/worktree.rs

@@ -168,9 +168,7 @@ enum ScanState {
 struct ShareState {
     project_id: u64,
     snapshots_tx: watch::Sender<LocalSnapshot>,
-    diagnostic_summaries_tx: mpsc::UnboundedSender<(Arc<Path>, DiagnosticSummary)>,
     _maintain_remote_snapshot: Task<Option<()>>,
-    _maintain_remote_diagnostic_summaries: Task<()>,
 }
 
 pub enum Event {
@@ -532,9 +530,18 @@ impl LocalWorktree {
         let updated = !old_summary.is_empty() || !new_summary.is_empty();
         if updated {
             if let Some(share) = self.share.as_ref() {
-                let _ = share
+                self.client
-                    .diagnostic_summaries_tx
+                    .send(proto::UpdateDiagnosticSummary {
-                    .unbounded_send((worktree_path.clone(), new_summary));
+                        project_id: share.project_id,
+                        worktree_id: self.id().to_proto(),
+                        summary: Some(proto::DiagnosticSummary {
+                            path: worktree_path.to_string_lossy().to_string(),
+                            language_server_id: language_server_id as u64,
+                            error_count: new_summary.error_count as u32,
+                            warning_count: new_summary.warning_count as u32,
+                        }),
+                    })
+                    .log_err();
             }
         }
 
@@ -968,6 +975,16 @@ impl LocalWorktree {
             let (snapshots_tx, mut snapshots_rx) = watch::channel_with(self.snapshot());
             let worktree_id = cx.model_id() as u64;
 
+            for (path, summary) in self.diagnostic_summaries.iter() {
+                if let Err(e) = self.client.send(proto::UpdateDiagnosticSummary {
+                    project_id,
+                    worktree_id,
+                    summary: Some(summary.to_proto(&path.0)),
+                }) {
+                    return Task::ready(Err(e));
+                }
+            }
+
             let maintain_remote_snapshot = cx.background().spawn({
                 let rpc = self.client.clone();
                 async move {
@@ -1017,31 +1034,10 @@ impl LocalWorktree {
                 .log_err()
             });
 
-            let (diagnostic_summaries_tx, mut diagnostic_summaries_rx) = mpsc::unbounded();
-            for (path, summary) in self.diagnostic_summaries.iter() {
-                let _ = diagnostic_summaries_tx.unbounded_send((path.0.clone(), summary.clone()));
-            }
-            let maintain_remote_diagnostic_summaries = cx.background().spawn({
-                let rpc = self.client.clone();
-                async move {
-                    while let Some((path, summary)) = diagnostic_summaries_rx.next().await {
-                        rpc.request(proto::UpdateDiagnosticSummary {
-                            project_id,
-                            worktree_id,
-                            summary: Some(summary.to_proto(&path)),
-                        })
-                        .await
-                        .log_err();
-                    }
-                }
-            });
-
             self.share = Some(ShareState {
                 project_id,
                 snapshots_tx,
-                diagnostic_summaries_tx,
                 _maintain_remote_snapshot: maintain_remote_snapshot,
-                _maintain_remote_diagnostic_summaries: maintain_remote_diagnostic_summaries,
             });
         }
 

crates/rpc/src/proto.rs

@@ -228,7 +228,6 @@ request_messages!(
     (ShareProject, ShareProjectResponse),
     (Test, Test),
     (UpdateBuffer, Ack),
-    (UpdateDiagnosticSummary, Ack),
     (UpdateParticipantLocation, Ack),
     (UpdateProject, Ack),
     (UpdateWorktree, Ack),