deps: Bump smol to 2.0 (#22956)

The collateral of this is that code size is increased by ~300kB, but I
think we can stomach it.

Release Notes:

- N/A

Cargo.toml

@@ -342,7 +342,7 @@ ashpd = { version = "0.10", default-features = false, features = ["async-std"]}
 async-compat = "0.2.1"
 async-compression = { version = "0.4", features = ["gzip", "futures-io"] }
 async-dispatcher = "0.1"
-async-fs = "1.6"
+async-fs = "2.1"
 async-pipe = { git = "https://github.com/zed-industries/async-pipe-rs", rev = "82d00a04211cf4e1236029aa03e6b6ce2a74c553" }
 async-recursion = "1.0.0"
 async-tar = "0.5.0"
@@ -461,7 +461,7 @@ signal-hook = "0.3.17"
 similar = "1.3"
 simplelog = "0.12.2"
 smallvec = { version = "1.6", features = ["union"] }
-smol = "1.2"
+smol = "2.0"
 sqlformat = "0.2"
 strsim = "0.11"
 strum = { version = "0.26.0", features = ["derive"] }

crates/client/src/client.rs

@@ -1958,8 +1958,8 @@ mod tests {
         });
         let server = FakeServer::for_client(user_id, &client, cx).await;
 
-        let (done_tx1, mut done_rx1) = smol::channel::unbounded();
-        let (done_tx2, mut done_rx2) = smol::channel::unbounded();
+        let (done_tx1, done_rx1) = smol::channel::unbounded();
+        let (done_tx2, done_rx2) = smol::channel::unbounded();
         AnyProtoClient::from(client.clone()).add_model_message_handler(
             move |model: Model<TestModel>, _: TypedEnvelope<proto::JoinProject>, mut cx| {
                 match model.update(&mut cx, |model, _| model.id).unwrap() {
@@ -2001,8 +2001,8 @@ mod tests {
 
         server.send(proto::JoinProject { project_id: 1 });
         server.send(proto::JoinProject { project_id: 2 });
-        done_rx1.next().await.unwrap();
-        done_rx2.next().await.unwrap();
+        done_rx1.recv().await.unwrap();
+        done_rx2.recv().await.unwrap();
     }
 
     #[gpui::test]
@@ -2020,7 +2020,7 @@ mod tests {
 
         let model = cx.new_model(|_| TestModel::default());
         let (done_tx1, _done_rx1) = smol::channel::unbounded();
-        let (done_tx2, mut done_rx2) = smol::channel::unbounded();
+        let (done_tx2, done_rx2) = smol::channel::unbounded();
         let subscription1 = client.add_message_handler(
             model.downgrade(),
             move |_, _: TypedEnvelope<proto::Ping>, _| {
@@ -2037,7 +2037,7 @@ mod tests {
             },
         );
         server.send(proto::Ping {});
-        done_rx2.next().await.unwrap();
+        done_rx2.recv().await.unwrap();
     }
 
     #[gpui::test]
@@ -2054,7 +2054,7 @@ mod tests {
         let server = FakeServer::for_client(user_id, &client, cx).await;
 
         let model = cx.new_model(|_| TestModel::default());
-        let (done_tx, mut done_rx) = smol::channel::unbounded();
+        let (done_tx, done_rx) = smol::channel::unbounded();
         let subscription = client.add_message_handler(
             model.clone().downgrade(),
             move |model: Model<TestModel>, _: TypedEnvelope<proto::Ping>, mut cx| {
@@ -2069,7 +2069,7 @@ mod tests {
             model.subscription = Some(subscription);
         });
         server.send(proto::Ping {});
-        done_rx.next().await.unwrap();
+        done_rx.recv().await.unwrap();
     }
 
     #[derive(Default)]

crates/collab/src/tests/integration_tests.rs

@@ -4936,7 +4936,7 @@ async fn test_project_search(
 
     // Perform a search as the guest.
     let mut results = HashMap::default();
-    let mut search_rx = project_b.update(cx_b, |project, cx| {
+    let search_rx = project_b.update(cx_b, |project, cx| {
         project.search(
             SearchQuery::text(
                 "world",
@@ -4951,7 +4951,7 @@ async fn test_project_search(
             cx,
         )
     });
-    while let Some(result) = search_rx.next().await {
+    while let Ok(result) = search_rx.recv().await {
         match result {
             SearchResult::Buffer { buffer, ranges } => {
                 results.entry(buffer).or_insert(ranges);

crates/collab/src/tests/random_project_collaboration_tests.rs

@@ -6,7 +6,6 @@ use call::ActiveCall;
 use collections::{BTreeMap, HashMap};
 use editor::Bias;
 use fs::{FakeFs, Fs as _};
-use futures::StreamExt;
 use git::repository::GitFileStatus;
 use gpui::{BackgroundExecutor, Model, TestAppContext};
 use language::{
@@ -873,7 +872,7 @@ impl RandomizedTest for ProjectCollaborationTest {
                     if detach { "detaching" } else { "awaiting" }
                 );
 
-                let mut search = project.update(cx, |project, cx| {
+                let search = project.update(cx, |project, cx| {
                     project.search(
                         SearchQuery::text(
                             query,
@@ -891,7 +890,7 @@ impl RandomizedTest for ProjectCollaborationTest {
                 drop(project);
                 let search = cx.executor().spawn(async move {
                     let mut results = HashMap::default();
-                    while let Some(result) = search.next().await {
+                    while let Ok(result) = search.recv().await {
                         if let SearchResult::Buffer { buffer, ranges } = result {
                             results.entry(buffer).or_insert(ranges);
                         }

crates/fs/src/fs.rs

@@ -1007,7 +1007,7 @@ impl FakeFs {
     const SYSTEMTIME_INTERVAL: Duration = Duration::from_nanos(100);
 
     pub fn new(executor: gpui::BackgroundExecutor) -> Arc<Self> {
-        let (tx, mut rx) = smol::channel::bounded::<PathBuf>(10);
+        let (tx, rx) = smol::channel::bounded::<PathBuf>(10);
 
         let this = Arc::new_cyclic(|this| Self {
             this: this.clone(),
@@ -1035,7 +1035,7 @@ impl FakeFs {
         executor.spawn({
             let this = this.clone();
             async move {
-                while let Some(git_event) = rx.next().await {
+                while let Ok(git_event) = rx.recv().await {
                     if let Some(mut state) = this.state.try_lock() {
                         state.emit_event([(git_event, None)]);
                     } else {

crates/gpui/Cargo.toml

@@ -118,7 +118,7 @@ util = { workspace = true, features = ["test-support"] }
 http_client = { workspace = true, features = ["test-support"] }
 unicode-segmentation.workspace = true
 
-[build-dependencies]
+[target.'cfg(target_os = "windows")'.build-dependencies]
 embed-resource = "3.0"
 
 [target.'cfg(target_os = "macos")'.build-dependencies]

crates/gpui/build.rs

@@ -13,6 +13,7 @@ fn main() {
             #[cfg(target_os = "macos")]
             macos::build();
         }
+        #[cfg(target_os = "windows")]
         Ok("windows") => {
             let manifest = std::path::Path::new("resources/windows/gpui.manifest.xml");
             let rc_file = std::path::Path::new("resources/windows/gpui.rc");

crates/gpui/src/test.rs

@@ -32,6 +32,7 @@ use smol::channel;
 use std::{
     env,
     panic::{self, RefUnwindSafe},
+    pin::Pin,
 };
 
 /// Run the given test function with the configured parameters.
@@ -85,7 +86,7 @@ pub fn run_test(
 
 /// A test struct for converting an observation callback into a stream.
 pub struct Observation<T> {
-    rx: channel::Receiver<T>,
+    rx: Pin<Box<channel::Receiver<T>>>,
     _subscription: Subscription,
 }
 
@@ -108,6 +109,7 @@ pub fn observe<T: 'static>(entity: &impl Entity<T>, cx: &mut TestAppContext) ->
             let _ = smol::block_on(tx.send(()));
         })
     });
+    let rx = Box::pin(rx);
 
     Observation { rx, _subscription }
 }

crates/lsp/src/lsp.rs

@@ -1396,10 +1396,8 @@ impl FakeLanguageServer {
     pub async fn try_receive_notification<T: notification::Notification>(
         &mut self,
     ) -> Option<T::Params> {
-        use futures::StreamExt as _;
-
         loop {
-            let (method, params) = self.notifications_rx.next().await?;
+            let (method, params) = self.notifications_rx.recv().await.ok()?;
             if method == T::METHOD {
                 return Some(serde_json::from_str::<T::Params>(&params).unwrap());
             } else {

crates/project/src/buffer_store.rs

@@ -30,6 +30,7 @@ use std::{
     io,
     ops::Range,
     path::{Path, PathBuf},
+    pin::pin,
     str::FromStr as _,
     sync::Arc,
     time::Instant,
@@ -1483,7 +1484,7 @@ impl BufferStore {
         }
 
         const MAX_CONCURRENT_BUFFER_OPENS: usize = 64;
-        let mut project_paths_rx = self
+        let project_paths_rx = self
             .worktree_store
             .update(cx, |worktree_store, cx| {
                 worktree_store.find_search_candidates(query.clone(), limit, open_buffers, fs, cx)
@@ -1495,6 +1496,7 @@ impl BufferStore {
                 tx.send(buffer).await.ok();
             }
 
+            let mut project_paths_rx = pin!(project_paths_rx);
             while let Some(project_paths) = project_paths_rx.next().await {
                 let buffers = this.update(&mut cx, |this, cx| {
                     project_paths

crates/project/src/lsp_store.rs

@@ -818,7 +818,7 @@ impl LocalLspStore {
                     let name = name.to_string();
                     async move {
                         let actions = params.actions.unwrap_or_default();
-                        let (tx, mut rx) = smol::channel::bounded(1);
+                        let (tx, rx) = smol::channel::bounded(1);
                         let request = LanguageServerPromptRequest {
                             level: match params.typ {
                                 lsp::MessageType::ERROR => PromptLevel::Critical,
@@ -837,9 +837,9 @@ impl LocalLspStore {
                             })
                             .is_ok();
                         if did_update {
-                            let response = rx.next().await;
+                            let response = rx.recv().await?;
 
-                            Ok(response)
+                            Ok(Some(response))
                         } else {
                             Ok(None)
                         }

crates/project/src/project.rs

@@ -79,6 +79,7 @@ use std::{
     borrow::Cow,
     ops::Range,
     path::{Component, Path, PathBuf},
+    pin::pin,
     str,
     sync::Arc,
     time::Duration,
@@ -3019,6 +3020,7 @@ impl Project {
             // 64 buffers at a time to avoid overwhelming the main thread. For each
             // opened buffer, we will spawn a background task that retrieves all the
             // ranges in the buffer matched by the query.
+            let mut chunks = pin!(chunks);
             'outer: while let Some(matching_buffer_chunk) = chunks.next().await {
                 let mut chunk_results = Vec::new();
                 for buffer in matching_buffer_chunk {
@@ -3748,6 +3750,7 @@ impl Project {
         // next `flush_effects()` call.
         drop(this);
 
+        let mut rx = pin!(rx);
         let answer = rx.next().await;
 
         Ok(LanguageServerPromptResponse {
@@ -3889,15 +3892,15 @@ impl Project {
                 .query
                 .ok_or_else(|| anyhow!("missing query field"))?,
         )?;
-        let mut results = this.update(&mut cx, |this, cx| {
+        let results = this.update(&mut cx, |this, cx| {
             this.find_search_candidate_buffers(&query, message.limit as _, cx)
         })?;
 
         let mut response = proto::FindSearchCandidatesResponse {
             buffer_ids: Vec::new(),
         };
 
-        while let Some(buffer) = results.next().await {
+        while let Ok(buffer) = results.recv().await {
             this.update(&mut cx, |this, cx| {
                 let buffer_id = this.create_buffer_for_peer(&buffer, peer_id, cx);
                 response.buffer_ids.push(buffer_id.to_proto());

crates/project/src/project_tests.rs

@@ -5657,9 +5657,9 @@ async fn search(
     query: SearchQuery,
     cx: &mut gpui::TestAppContext,
 ) -> Result<HashMap<String, Vec<Range<usize>>>> {
-    let mut search_rx = project.update(cx, |project, cx| project.search(query, cx));
+    let search_rx = project.update(cx, |project, cx| project.search(query, cx));
     let mut results = HashMap::default();
-    while let Some(search_result) = search_rx.next().await {
+    while let Ok(search_result) = search_rx.recv().await {
         match search_result {
             SearchResult::Buffer { buffer, ranges } => {
                 results.entry(buffer).or_insert(ranges);

crates/project/src/worktree_store.rs

@@ -1,5 +1,6 @@
 use std::{
     path::{Path, PathBuf},
+    pin::pin,
     sync::{atomic::AtomicUsize, Arc},
 };
 
@@ -648,7 +649,7 @@ impl WorktreeStore {
         // We spawn a number of workers that take items from the filter channel and check the query
         // against the version of the file on disk.
         let (filter_tx, filter_rx) = smol::channel::bounded(64);
-        let (output_tx, mut output_rx) = smol::channel::bounded(64);
+        let (output_tx, output_rx) = smol::channel::bounded(64);
         let (matching_paths_tx, matching_paths_rx) = smol::channel::unbounded();
 
         let input = cx.background_executor().spawn({
@@ -685,7 +686,7 @@ impl WorktreeStore {
         cx.background_executor()
             .spawn(async move {
                 let mut matched = 0;
-                while let Some(mut receiver) = output_rx.next().await {
+                while let Ok(mut receiver) = output_rx.recv().await {
                     let Some(path) = receiver.next().await else {
                         continue;
                     };
@@ -990,6 +991,7 @@ impl WorktreeStore {
         mut input: Receiver<MatchingEntry>,
         query: &SearchQuery,
     ) -> Result<()> {
+        let mut input = pin!(input);
         while let Some(mut entry) = input.next().await {
             let abs_path = entry.worktree_path.join(&entry.path.path);
             let Some(file) = fs.open_sync(&abs_path).await.log_err() else {

crates/remote_server/src/headless_project.rs

@@ -515,7 +515,7 @@ impl HeadlessProject {
                 .query
                 .ok_or_else(|| anyhow!("missing query field"))?,
         )?;
-        let mut results = this.update(&mut cx, |this, cx| {
+        let results = this.update(&mut cx, |this, cx| {
             this.buffer_store.update(cx, |buffer_store, cx| {
                 buffer_store.find_search_candidates(&query, message.limit as _, this.fs.clone(), cx)
             })
@@ -527,7 +527,7 @@ impl HeadlessProject {
 
         let buffer_store = this.read_with(&cx, |this, _| this.buffer_store.clone())?;
 
-        while let Some(buffer) = results.next().await {
+        while let Ok(buffer) = results.recv().await {
             let buffer_id = buffer.update(&mut cx, |this, _| this.remote_id())?;
             response.buffer_ids.push(buffer_id.to_proto());
             buffer_store

crates/remote_server/src/remote_editing_tests.rs

@@ -187,7 +187,7 @@ async fn test_remote_project_search(cx: &mut TestAppContext, server_cx: &mut Tes
     cx.run_until_parked();
 
     async fn do_search(project: &Model<Project>, mut cx: TestAppContext) -> Model<Buffer> {
-        let mut receiver = project.update(&mut cx, |project, cx| {
+        let receiver = project.update(&mut cx, |project, cx| {
             project.search(
                 SearchQuery::text(
                     "project",
@@ -203,7 +203,7 @@ async fn test_remote_project_search(cx: &mut TestAppContext, server_cx: &mut Tes
             )
         });
 
-        let first_response = receiver.next().await.unwrap();
+        let first_response = receiver.recv().await.unwrap();
         let SearchResult::Buffer { buffer, .. } = first_response else {
             panic!("incorrect result");
         };
@@ -214,7 +214,7 @@ async fn test_remote_project_search(cx: &mut TestAppContext, server_cx: &mut Tes
             )
         });
 
-        assert!(receiver.next().await.is_none());
+        assert!(receiver.recv().await.is_err());
         buffer
     }
 

crates/remote_server/src/unix.rs

@@ -249,7 +249,7 @@ impl ServerListeners {
 
 fn start_server(
     listeners: ServerListeners,
-    mut log_rx: Receiver<Vec<u8>>,
+    log_rx: Receiver<Vec<u8>>,
     cx: &mut AppContext,
 ) -> Arc<ChannelClient> {
     // This is the server idle timeout. If no connection comes in in this timeout, the server will shut down.
@@ -351,8 +351,8 @@ fn start_server(
                         }
                     }
 
-                    log_message = log_rx.next().fuse() => {
-                        if let Some(log_message) = log_message {
+                    log_message = log_rx.recv().fuse() => {
+                        if let Ok(log_message) = log_message {
                             if let Err(error) = stderr_stream.write_all(&log_message).await {
                                 log::error!("failed to write log message to stderr: {:?}", error);
                                 break;