Merge pull request #2267 from dead10ck/fix-write-fail

Write path fixes

2 files changed, 222 insertions, 61 deletions

diff --git a/helix-view/src/document.rs b/helix-view/src/document.rs
index 0daa983f..78c6d032 100644
--- a/helix-view/src/document.rs
+++ b/helix-view/src/document.rs
@@ -83,6 +83,18 @@ impl Serialize for Mode {
     }
 }
 
+/// A snapshot of the text of a document that we want to write out to disk
+#[derive(Debug, Clone)]
+pub struct DocumentSavedEvent {
+    pub revision: usize,
+    pub doc_id: DocumentId,
+    pub path: PathBuf,
+    pub text: Rope,
+}
+
+pub type DocumentSavedEventResult = Result<DocumentSavedEvent, anyhow::Error>;
+pub type DocumentSavedEventFuture = BoxFuture<'static, DocumentSavedEventResult>;
+
 pub struct Document {
     pub(crate) id: DocumentId,
     text: Rope,
@@ -492,45 +504,61 @@ impl Document {
         Some(fut.boxed())
     }
 
-    pub fn save(&mut self, force: bool) -> impl Future<Output = Result<(), anyhow::Error>> {
-        self.save_impl::<futures_util::future::Ready<_>>(None, force)
-    }
-
-    pub fn format_and_save(
+    pub fn save<P: Into<PathBuf>>(
         &mut self,
-        formatting: Option<impl Future<Output = Result<Transaction, FormatterError>>>,
+        path: Option<P>,
         force: bool,
-    ) -> impl Future<Output = anyhow::Result<()>> {
-        self.save_impl(formatting, force)
+    ) -> Result<
+        impl Future<Output = Result<DocumentSavedEvent, anyhow::Error>> + 'static + Send,
+        anyhow::Error,
+    > {
+        let path = path.map(|path| path.into());
+        self.save_impl(path, force)
+
+        // futures_util::future::Ready<_>,
     }
 
-    // TODO: do we need some way of ensuring two save operations on the same doc can't run at once?
-    // or is that handled by the OS/async layer
     /// The `Document`'s text is encoded according to its encoding and written to the file located
     /// at its `path()`.
-    ///
-    /// If `formatting` is present, it supplies some changes that we apply to the text before saving.
-    fn save_impl<F: Future<Output = Result<Transaction, FormatterError>>>(
+    fn save_impl(
         &mut self,
-        formatting: Option<F>,
+        path: Option<PathBuf>,
         force: bool,
-    ) -> impl Future<Output = Result<(), anyhow::Error>> {
+    ) -> Result<
+        impl Future<Output = Result<DocumentSavedEvent, anyhow::Error>> + 'static + Send,
+        anyhow::Error,
+    > {
+        log::debug!(
+            "submitting save of doc '{:?}'",
+            self.path().map(|path| path.to_string_lossy())
+        );
+
         // we clone and move text + path into the future so that we asynchronously save the current
         // state without blocking any further edits.
+        let text = self.text().clone();
 
-        let mut text = self.text().clone();
-        let path = self.path.clone().expect("Can't save with no path set!");
-        let identifier = self.identifier();
+        let path = match path {
+            Some(path) => helix_core::path::get_canonicalized_path(&path)?,
+            None => {
+                if self.path.is_none() {
+                    bail!("Can't save with no path set!");
+                }
 
+                self.path.as_ref().unwrap().clone()
+            }
+        };
+
+        let identifier = self.path().map(|_| self.identifier());
         let language_server = self.language_server.clone();
 
         // mark changes up to now as saved
-        self.reset_modified();
+        let current_rev = self.get_current_revision();
+        let doc_id = self.id();
 
         let encoding = self.encoding;
 
         // We encode the file according to the `Document`'s encoding.
-        async move {
+        let future = async move {
             use tokio::fs::File;
             if let Some(parent) = path.parent() {
                 // TODO: display a prompt asking the user if the directories should be created
@@ -543,39 +571,34 @@ impl Document {
                 }
             }
 
-            if let Some(fmt) = formatting {
-                match fmt.await {
-                    Ok(transaction) => {
-                        let success = transaction.changes().apply(&mut text);
-                        if !success {
-                            // This shouldn't happen, because the transaction changes were generated
-                            // from the same text we're saving.
-                            log::error!("failed to apply format changes before saving");
-                        }
-                    }
-                    Err(err) => {
-                        // formatting failed: report error, and save file without modifications
-                        log::error!("{}", err);
-                    }
-                }
-            }
-
-            let mut file = File::create(path).await?;
+            let mut file = File::create(&path).await?;
             to_writer(&mut file, encoding, &text).await?;
 
+            let event = DocumentSavedEvent {
+                revision: current_rev,
+                doc_id,
+                path,
+                text: text.clone(),
+            };
+
             if let Some(language_server) = language_server {
                 if !language_server.is_initialized() {
-                    return Ok(());
+                    return Ok(event);
                 }
-                if let Some(notification) =
-                    language_server.text_document_did_save(identifier, &text)
-                {
-                    notification.await?;
+
+                if let Some(identifier) = identifier {
+                    if let Some(notification) =
+                        language_server.text_document_did_save(identifier, &text)
+                    {
+                        notification.await?;
+                    }
                 }
             }
 
-            Ok(())
-        }
+            Ok(event)
+        };
+
+        Ok(future)
     }
 
     /// Detect the programming language based on the file type.
@@ -930,6 +953,12 @@ impl Document {
         let history = self.history.take();
         let current_revision = history.current_revision();
         self.history.set(history);
+        log::debug!(
+            "id {} modified - last saved: {}, current: {}",
+            self.id,
+            self.last_saved_revision,
+            current_revision
+        );
         current_revision != self.last_saved_revision || !self.changes.is_empty()
     }
 
@@ -941,6 +970,30 @@ impl Document {
         self.last_saved_revision = current_revision;
     }
 
+    /// Set the document's latest saved revision to the given one.
+    pub fn set_last_saved_revision(&mut self, rev: usize) {
+        log::debug!(
+            "doc {} revision updated {} -> {}",
+            self.id,
+            self.last_saved_revision,
+            rev
+        );
+        self.last_saved_revision = rev;
+    }
+
+    /// Get the document's latest saved revision.
+    pub fn get_last_saved_revision(&mut self) -> usize {
+        self.last_saved_revision
+    }
+
+    /// Get the current revision number
+    pub fn get_current_revision(&mut self) -> usize {
+        let history = self.history.take();
+        let current_revision = history.current_revision();
+        self.history.set(history);
+        current_revision
+    }
+
     /// Corresponding language scope name. Usually `source.<lang>`.
     pub fn language_scope(&self) -> Option<&str> {
         self.language
diff --git a/helix-view/src/editor.rs b/helix-view/src/editor.rs
index e9a3c639..cd2b1ad4 100644
--- a/helix-view/src/editor.rs
+++ b/helix-view/src/editor.rs
@@ -1,6 +1,6 @@
 use crate::{
     clipboard::{get_clipboard_provider, ClipboardProvider},
-    document::Mode,
+    document::{DocumentSavedEventFuture, DocumentSavedEventResult, Mode},
     graphics::{CursorKind, Rect},
     info::Info,
     input::KeyEvent,
@@ -9,8 +9,9 @@ use crate::{
     Document, DocumentId, View, ViewId,
 };
 
-use futures_util::future;
 use futures_util::stream::select_all::SelectAll;
+use futures_util::{future, StreamExt};
+use helix_lsp::Call;
 use tokio_stream::wrappers::UnboundedReceiverStream;
 
 use std::{
@@ -28,7 +29,7 @@ use tokio::{
     time::{sleep, Duration, Instant, Sleep},
 };
 
-use anyhow::Error;
+use anyhow::{anyhow, bail, Error};
 
 pub use helix_core::diagnostic::Severity;
 pub use helix_core::register::Registers;
@@ -65,7 +66,7 @@ where
     )
 }
 
-#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "kebab-case", default, deny_unknown_fields)]
 pub struct FilePickerConfig {
     /// IgnoreOptions
@@ -172,7 +173,7 @@ pub struct Config {
     pub color_modes: bool,
 }
 
-#[derive(Debug, Default, Clone, PartialEq, Serialize, Deserialize)]
+#[derive(Debug, Default, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(default, rename_all = "kebab-case", deny_unknown_fields)]
 pub struct TerminalConfig {
     pub command: String,
@@ -225,7 +226,7 @@ pub fn get_terminal_provider() -> Option<TerminalConfig> {
     None
 }
 
-#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(default, rename_all = "kebab-case", deny_unknown_fields)]
 pub struct LspConfig {
     /// Display LSP progress messages below statusline
@@ -246,7 +247,7 @@ impl Default for LspConfig {
     }
 }
 
-#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "kebab-case", default, deny_unknown_fields)]
 pub struct SearchConfig {
     /// Smart case: Case insensitive searching unless pattern contains upper case characters. Defaults to true.
@@ -255,7 +256,7 @@ pub struct SearchConfig {
     pub wrap_around: bool,
 }
 
-#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "kebab-case", default, deny_unknown_fields)]
 pub struct StatusLineConfig {
     pub left: Vec<StatusLineElement>,
@@ -279,7 +280,7 @@ impl Default for StatusLineConfig {
     }
 }
 
-#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "kebab-case", default, deny_unknown_fields)]
 pub struct ModeConfig {
     pub normal: String,
@@ -458,7 +459,7 @@ impl std::str::FromStr for GutterType {
     }
 }
 
-#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(default)]
 pub struct WhitespaceConfig {
     pub render: WhitespaceRender,
@@ -643,12 +644,21 @@ pub struct Breakpoint {
     pub log_message: Option<String>,
 }
 
+use futures_util::stream::{Flatten, Once};
+
 pub struct Editor {
     /// Current editing mode.
     pub mode: Mode,
     pub tree: Tree,
     pub next_document_id: DocumentId,
     pub documents: BTreeMap<DocumentId, Document>,
+
+    // We Flatten<> to resolve the inner DocumentSavedEventFuture. For that we need a stream of streams, hence the Once<>.
+    // https://stackoverflow.com/a/66875668
+    pub saves: HashMap<DocumentId, UnboundedSender<Once<DocumentSavedEventFuture>>>,
+    pub save_queue: SelectAll<Flatten<UnboundedReceiverStream<Once<DocumentSavedEventFuture>>>>,
+    pub write_count: usize,
+
     pub count: Option<std::num::NonZeroUsize>,
     pub selected_register: Option<char>,
     pub registers: Registers,
@@ -688,6 +698,15 @@ pub struct Editor {
     pub config_events: (UnboundedSender<ConfigEvent>, UnboundedReceiver<ConfigEvent>),
 }
 
+#[derive(Debug)]
+pub enum EditorEvent {
+    DocumentSaved(DocumentSavedEventResult),
+    ConfigEvent(ConfigEvent),
+    LanguageServerMessage((usize, Call)),
+    DebuggerEvent(dap::Payload),
+    IdleTimer,
+}
+
 #[derive(Debug, Clone)]
 pub enum ConfigEvent {
     Refresh,
@@ -719,6 +738,8 @@ pub enum CloseError {
     DoesNotExist,
     /// Buffer is modified
     BufferModified(String),
+    /// Document failed to save
+    SaveError(anyhow::Error),
 }
 
 impl Editor {
@@ -739,6 +760,9 @@ impl Editor {
             tree: Tree::new(area),
             next_document_id: DocumentId::default(),
             documents: BTreeMap::new(),
+            saves: HashMap::new(),
+            save_queue: SelectAll::new(),
+            write_count: 0,
             count: None,
             selected_register: None,
             macro_recording: None,
@@ -804,12 +828,16 @@ impl Editor {
 
     #[inline]
     pub fn set_status<T: Into<Cow<'static, str>>>(&mut self, status: T) {
-        self.status_msg = Some((status.into(), Severity::Info));
+        let status = status.into();
+        log::debug!("editor status: {}", status);
+        self.status_msg = Some((status, Severity::Info));
     }
 
     #[inline]
     pub fn set_error<T: Into<Cow<'static, str>>>(&mut self, error: T) {
-        self.status_msg = Some((error.into(), Severity::Error));
+        let error = error.into();
+        log::error!("editor error: {}", error);
+        self.status_msg = Some((error, Severity::Error));
     }
 
     #[inline]
@@ -1034,6 +1062,13 @@ impl Editor {
             DocumentId(unsafe { NonZeroUsize::new_unchecked(self.next_document_id.0.get() + 1) });
         doc.id = id;
         self.documents.insert(id, doc);
+
+        let (save_sender, save_receiver) = tokio::sync::mpsc::unbounded_channel();
+        self.saves.insert(id, save_sender);
+
+        let stream = UnboundedReceiverStream::new(save_receiver).flatten();
+        self.save_queue.push(stream);
+
         id
     }
 
@@ -1080,16 +1115,19 @@ impl Editor {
     }
 
     pub fn close_document(&mut self, doc_id: DocumentId, force: bool) -> Result<(), CloseError> {
-        let doc = match self.documents.get(&doc_id) {
+        let doc = match self.documents.get_mut(&doc_id) {
             Some(doc) => doc,
             None => return Err(CloseError::DoesNotExist),
         };
-
         if !force && doc.is_modified() {
             return Err(CloseError::BufferModified(doc.display_name().into_owned()));
         }
 
+        // This will also disallow any follow-up writes
+        self.saves.remove(&doc_id);
+
         if let Some(language_server) = doc.language_server() {
+            // TODO: track error
             tokio::spawn(language_server.text_document_did_close(doc.identifier()));
         }
 
@@ -1152,6 +1190,32 @@ impl Editor {
         Ok(())
     }
 
+    pub fn save<P: Into<PathBuf>>(
+        &mut self,
+        doc_id: DocumentId,
+        path: Option<P>,
+        force: bool,
+    ) -> anyhow::Result<()> {
+        // convert a channel of futures to pipe into main queue one by one
+        // via stream.then() ? then push into main future
+
+        let path = path.map(|path| path.into());
+        let doc = doc_mut!(self, &doc_id);
+        let future = doc.save(path, force)?;
+
+        use futures_util::stream;
+
+        self.saves
+            .get(&doc_id)
+            .ok_or_else(|| anyhow::format_err!("saves are closed for this document!"))?
+            .send(stream::once(Box::pin(future)))
+            .map_err(|err| anyhow!("failed to send save event: {}", err))?;
+
+        self.write_count += 1;
+
+        Ok(())
+    }
+
     pub fn resize(&mut self, area: Rect) {
         if self.tree.resize(area) {
             self._refresh();
@@ -1252,14 +1316,14 @@ impl Editor {
         }
     }
 
-    /// Closes language servers with timeout. The default timeout is 500 ms, use
+    /// Closes language servers with timeout. The default timeout is 10000 ms, use
     /// `timeout` parameter to override this.
     pub async fn close_language_servers(
         &self,
         timeout: Option<u64>,
     ) -> Result<(), tokio::time::error::Elapsed> {
         tokio::time::timeout(
-            Duration::from_millis(timeout.unwrap_or(500)),
+            Duration::from_millis(timeout.unwrap_or(3000)),
             future::join_all(
                 self.language_servers
                     .iter_clients()
@@ -1269,4 +1333,48 @@ impl Editor {
         .await
         .map(|_| ())
     }
+
+    pub async fn wait_event(&mut self) -> EditorEvent {
+        tokio::select! {
+            biased;
+
+            Some(event) = self.save_queue.next() => {
+                self.write_count -= 1;
+                EditorEvent::DocumentSaved(event)
+            }
+            Some(config_event) = self.config_events.1.recv() => {
+                EditorEvent::ConfigEvent(config_event)
+            }
+            Some(message) = self.language_servers.incoming.next() => {
+                EditorEvent::LanguageServerMessage(message)
+            }
+            Some(event) = self.debugger_events.next() => {
+                EditorEvent::DebuggerEvent(event)
+            }
+            _ = &mut self.idle_timer => {
+                EditorEvent::IdleTimer
+            }
+        }
+    }
+
+    pub async fn flush_writes(&mut self) -> anyhow::Result<()> {
+        while self.write_count > 0 {
+            if let Some(save_event) = self.save_queue.next().await {
+                self.write_count -= 1;
+
+                let save_event = match save_event {
+                    Ok(event) => event,
+                    Err(err) => {
+                        self.set_error(err.to_string());
+                        bail!(err);
+                    }
+                };
+
+                let doc = doc_mut!(self, &save_event.doc_id);
+                doc.set_last_saved_revision(save_event.revision);
+            }
+        }
+
+        Ok(())
+    }
 }