chore(write): serialize write operations within a Document

The way that document writes are handled are by submitting them to the
async job pool, which are all executed opportunistically out of order. It
was discovered that this can lead to write inconsistencies when there
are multiple writes to the same file in quick succession.

This seeks to fix this problem by removing document writes from the
general pool of jobs and into its own specialized event. Now when a
user submits a write with one of the write commands, a request is simply
queued up in a new mpsc channel that each Document makes to handle its own
writes. This way, if multiple writes are submitted on the same document,
they are executed in order, while still allowing concurrent writes for
different documents.

4 files changed, 112 insertions, 40 deletions

diff --git a/helix-term/src/application.rs b/helix-term/src/application.rs
index 4bb36b59..a9e25d08 100644
--- a/helix-term/src/application.rs
+++ b/helix-term/src/application.rs
@@ -6,7 +6,14 @@ use helix_core::{
     pos_at_coords, syntax, Selection,
 };
 use helix_lsp::{lsp, util::lsp_pos_to_pos, LspProgressMap};
-use helix_view::{align_view, editor::ConfigEvent, theme, tree::Layout, Align, Editor};
+use helix_view::{
+    align_view,
+    document::DocumentSaveEventResult,
+    editor::{ConfigEvent, EditorEvent},
+    theme,
+    tree::Layout,
+    Align, Editor,
+};
 use serde_json::json;
 
 use crate::{
@@ -19,7 +26,7 @@ use crate::{
     ui::{self, overlay::overlayed},
 };
 
-use log::{error, warn};
+use log::{debug, error, warn};
 use std::{
     io::{stdin, stdout, Write},
     sync::Arc,
@@ -294,26 +301,6 @@ impl Application {
                 Some(signal) = self.signals.next() => {
                     self.handle_signals(signal).await;
                 }
-                Some((id, call)) = self.editor.language_servers.incoming.next() => {
-                    self.handle_language_server_message(call, id).await;
-                    // limit render calls for fast language server messages
-                    let last = self.editor.language_servers.incoming.is_empty();
-
-                    if last || self.last_render.elapsed() > LSP_DEADLINE {
-                        self.render();
-                        self.last_render = Instant::now();
-                    }
-                }
-                Some(payload) = self.editor.debugger_events.next() => {
-                    let needs_render = self.editor.handle_debugger_message(payload).await;
-                    if needs_render {
-                        self.render();
-                    }
-                }
-                Some(config_event) = self.editor.config_events.1.recv() => {
-                    self.handle_config_events(config_event);
-                    self.render();
-                }
                 Some(callback) = self.jobs.futures.next() => {
                     self.jobs.handle_callback(&mut self.editor, &mut self.compositor, callback);
                     self.render();
@@ -322,20 +309,47 @@ impl Application {
                     self.jobs.handle_callback(&mut self.editor, &mut self.compositor, callback);
                     self.render();
                 }
-                _ = &mut self.editor.idle_timer => {
-                    // idle timeout
-                    self.editor.clear_idle_timer();
-                    self.handle_idle_timeout();
+                event = self.editor.wait_event() => {
+                    match event {
+                        EditorEvent::DocumentSave(event) => {
+                            self.handle_document_write(event);
+                            self.render();
+                        }
+                        EditorEvent::ConfigEvent(event) => {
+                            self.handle_config_events(event);
+                            self.render();
+                        }
+                        EditorEvent::LanguageServerMessage((id, call)) => {
+                            self.handle_language_server_message(call, id).await;
+                            // limit render calls for fast language server messages
+                            let last = self.editor.language_servers.incoming.is_empty();
+
+                            if last || self.last_render.elapsed() > LSP_DEADLINE {
+                                self.render();
+                                self.last_render = Instant::now();
+                            }
+                        }
+                        EditorEvent::DebuggerEvent(payload) => {
+                            let needs_render = self.editor.handle_debugger_message(payload).await;
+                            if needs_render {
+                                self.render();
+                            }
+                        }
+                        EditorEvent::IdleTimer => {
+                            self.editor.clear_idle_timer();
+                            self.handle_idle_timeout();
 
-                    #[cfg(feature = "integration")]
-                    {
-                        idle_handled = true;
+                            #[cfg(feature = "integration")]
+                            {
+                                idle_handled = true;
+                            }
+                        }
                     }
                 }
             }
 
             // for integration tests only, reset the idle timer after every
-            // event to make a signal when test events are done processing
+            // event to signal when test events are done processing
             #[cfg(feature = "integration")]
             {
                 if idle_handled {
@@ -446,6 +460,46 @@ impl Application {
         }
     }
 
+    pub fn handle_document_write(&mut self, doc_save_event: DocumentSaveEventResult) {
+        if let Err(err) = doc_save_event {
+            self.editor.set_error(err.to_string());
+            return;
+        }
+
+        let doc_save_event = doc_save_event.unwrap();
+        let doc = self.editor.document_mut(doc_save_event.doc_id);
+
+        if doc.is_none() {
+            warn!(
+                "received document saved event for non-existent doc id: {}",
+                doc_save_event.doc_id
+            );
+
+            return;
+        }
+
+        let doc = doc.unwrap();
+
+        debug!(
+            "document {:?} saved with revision {}",
+            doc.path(),
+            doc_save_event.revision
+        );
+
+        doc.set_last_saved_revision(doc_save_event.revision);
+        let lines = doc.text().len_lines();
+        let bytes = doc.text().len_bytes();
+
+        let path_str = doc
+            .path()
+            .expect("document written without path")
+            .to_string_lossy()
+            .into_owned();
+
+        self.editor
+            .set_status(format!("'{}' written, {}L {}B", path_str, lines, bytes));
+    }
+
     pub fn handle_terminal_events(&mut self, event: Result<CrosstermEvent, crossterm::ErrorKind>) {
         let mut cx = crate::compositor::Context {
             editor: &mut self.editor,
@@ -866,11 +920,28 @@ impl Application {
 
         self.event_loop(input_stream).await;
 
-        let err = self.close().await.err();
+        let mut save_errs = Vec::new();
+
+        for doc in self.editor.documents_mut() {
+            if let Some(Err(err)) = doc.close().await {
+                save_errs.push((
+                    doc.path()
+                        .map(|path| path.to_string_lossy().into_owned())
+                        .unwrap_or_else(|| "".into()),
+                    err,
+                ));
+            }
+        }
 
+        let close_err = self.close().await.err();
         restore_term()?;
 
-        if let Some(err) = err {
+        for (path, err) in save_errs {
+            self.editor.exit_code = 1;
+            eprintln!("Error closing '{}': {}", path, err);
+        }
+
+        if let Some(err) = close_err {
             self.editor.exit_code = 1;
             eprintln!("Error: {}", err);
         }
diff --git a/helix-term/src/commands.rs b/helix-term/src/commands.rs
index 5073651b..f38434e2 100644
--- a/helix-term/src/commands.rs
+++ b/helix-term/src/commands.rs
@@ -51,7 +51,7 @@ use crate::{
     ui::{self, overlay::overlayed, FilePicker, Picker, Popup, Prompt, PromptEvent},
 };
 
-use crate::job::{self, Job, Jobs};
+use crate::job::{self, Jobs};
 use futures_util::{FutureExt, StreamExt};
 use std::{collections::HashMap, fmt, future::Future};
 use std::{collections::HashSet, num::NonZeroUsize};
diff --git a/helix-term/src/commands/typed.rs b/helix-term/src/commands/typed.rs
index 1bfc8153..d82dd7fe 100644
--- a/helix-term/src/commands/typed.rs
+++ b/helix-term/src/commands/typed.rs
@@ -77,7 +77,9 @@ fn buffer_close_by_ids_impl(
     let (modified_ids, modified_names): (Vec<_>, Vec<_>) = doc_ids
         .iter()
         .filter_map(|&doc_id| {
-            if let Err(CloseError::BufferModified(name)) = editor.close_document(doc_id, force) {
+            if let Err(CloseError::BufferModified(name)) =
+                helix_lsp::block_on(editor.close_document(doc_id, force))
+            {
                 Some((doc_id, name))
             } else {
                 None
@@ -269,6 +271,7 @@ fn write_impl(
         doc.set_path(Some(path.as_ref().as_ref()))
             .context("invalid filepath")?;
     }
+
     if doc.path().is_none() {
         bail!("cannot write a buffer without a filename");
     }
@@ -287,8 +290,8 @@ fn write_impl(
     } else {
         None
     };
-    let future = doc.format_and_save(fmt, force);
-    cx.jobs.add(Job::new(future).wait_before_exiting());
+
+    doc.format_and_save(fmt, force)?;
 
     if path.is_some() {
         let id = doc.id();
@@ -602,8 +605,8 @@ fn write_all_impl(
         } else {
             None
         };
-        let future = doc.format_and_save(fmt, force);
-        jobs.add(Job::new(future).wait_before_exiting());
+
+        doc.format_and_save(fmt, force)?;
     }
 
     if quit {
diff --git a/helix-term/tests/test/write.rs b/helix-term/tests/test/write.rs
index 8869d881..4ac850c1 100644
--- a/helix-term/tests/test/write.rs
+++ b/helix-term/tests/test/write.rs
@@ -62,7 +62,6 @@ async fn test_write_quit() -> anyhow::Result<()> {
 }
 
 #[tokio::test]
-#[ignore]
 async fn test_write_concurrent() -> anyhow::Result<()> {
     let mut file = tempfile::NamedTempFile::new()?;
     let mut command = String::new();
@@ -92,7 +91,6 @@ async fn test_write_concurrent() -> anyhow::Result<()> {
 }
 
 #[tokio::test]
-#[ignore]
 async fn test_write_fail_mod_flag() -> anyhow::Result<()> {
     let file = helpers::new_readonly_tempfile()?;