Remove unnecessary waiting when handling save RPC requests

Add saving to the randomized integration test

Co-Authored-By: Nathan Sobo <nathan@zed.dev>

crates/language/src/buffer.rs

@@ -1283,10 +1283,6 @@ impl Buffer {
         self.text.wait_for_edits(edit_ids)
     }
 
-    pub fn wait_for_version(&mut self, version: clock::Global) -> impl Future<Output = ()> {
-        self.text.wait_for_version(version)
-    }
-
     pub fn set_active_selections(
         &mut self,
         selections: Arc<[Selection<Anchor>]>,

crates/project/src/project.rs

@@ -2289,11 +2289,12 @@ impl Project {
             Ok::<_, anyhow::Error>((project_id, buffer))
         })?;
 
-        buffer
+        if !buffer
-            .update(&mut cx, |buffer, _| {
+            .read_with(&cx, |buffer, _| buffer.version())
-                buffer.wait_for_version(requested_version)
+            .observed_all(&requested_version)
-            })
+        {
-            .await;
+            Err(anyhow!("save request depends on unreceived edits"))?;
+        }
 
         let (saved_version, mtime) = buffer.update(&mut cx, |buffer, cx| buffer.save(cx)).await?;
         Ok(proto::BufferSaved {

crates/server/src/rpc.rs

@@ -4215,6 +4215,21 @@ mod tests {
                             .await
                             .expect("completion request failed");
                     }
+                    20..=29 if buffer.read_with(&cx, |buffer, _| buffer.is_dirty()) => {
+                        let (requested_version, save) = buffer.update(&mut cx, |buffer, cx| {
+                            log::info!(
+                                "Guest {}: saving buffer {:?}",
+                                guest_id,
+                                buffer.file().unwrap().full_path(cx)
+                            );
+                            (buffer.version(), buffer.save(cx))
+                        });
+                        let (saved_version, _) = save.await.expect("completion request failed");
+                        buffer.read_with(&cx, |buffer, _| {
+                            assert!(buffer.version().observed_all(&saved_version));
+                            assert!(saved_version.observed_all(&requested_version));
+                        });
+                    }
                     _ => {
                         buffer.update(&mut cx, |buffer, cx| {
                             log::info!(

crates/text/src/text.rs

@@ -21,7 +21,7 @@ use operation_queue::OperationQueue;
 pub use patch::Patch;
 pub use point::*;
 pub use point_utf16::*;
-use postage::{barrier, oneshot, prelude::*};
+use postage::{oneshot, prelude::*};
 #[cfg(any(test, feature = "test-support"))]
 pub use random_char_iter::*;
 use rope::TextDimension;
@@ -53,7 +53,6 @@ pub struct Buffer {
     pub lamport_clock: clock::Lamport,
     subscriptions: Topic,
     edit_id_resolvers: HashMap<clock::Local, Vec<oneshot::Sender<()>>>,
-    version_barriers: Vec<(clock::Global, barrier::Sender)>,
 }
 
 #[derive(Clone, Debug)]
@@ -575,7 +574,6 @@ impl Buffer {
             lamport_clock,
             subscriptions: Default::default(),
             edit_id_resolvers: Default::default(),
-            version_barriers: Default::default(),
         }
     }
 
@@ -837,8 +835,6 @@ impl Buffer {
                 }
             }
         }
-        self.version_barriers
-            .retain(|(version, _)| !self.snapshot.version().observed_all(version));
         Ok(())
     }
 
@@ -1309,16 +1305,6 @@ impl Buffer {
         }
     }
 
-    pub fn wait_for_version(&mut self, version: clock::Global) -> impl Future<Output = ()> {
-        let (tx, mut rx) = barrier::channel();
-        if !self.snapshot.version.observed_all(&version) {
-            self.version_barriers.push((version, tx));
-        }
-        async move {
-            rx.recv().await;
-        }
-    }
-
     fn resolve_edit(&mut self, edit_id: clock::Local) {
         for mut tx in self
             .edit_id_resolvers