Ensure documents are synced before calculating completions

Author: tobias-tengler

compiler/crates/relay-lsp/src/server/mod.rs

@@ -152,7 +152,7 @@ where
     );
 
     let task_processor = LSPTaskProcessor;
-    let task_queue = TaskQueue::new(Arc::new(task_processor));
+    let mut task_queue = TaskQueue::new(Arc::new(task_processor));
     let task_scheduler = task_queue.get_scheduler();
 
     config.artifact_writer = Box::new(NoopArtifactWriter);
@@ -190,6 +190,12 @@ fn next_task(
     }
 }
 
+static NOTIFCATIONS_MUTATING_LSP_STATE: [&str; 3] = [
+    "textDocument/didOpen",
+    "textDocument/didChange",
+    "textDocument/didClose",
+];
+
 struct LSPTaskProcessor;
 
 impl<TPerfLogger: PerfLogger + 'static, TSchemaDocumentation: SchemaDocumentation + 'static>
@@ -209,6 +215,15 @@ impl<TPerfLogger: PerfLogger + 'static, TSchemaDocumentation: SchemaDocumentatio
             }
         }
     }
+
+    fn is_serial_task(&self, task: &Task) -> bool {
+        match task {
+            Task::InboundMessage(Message::Notification(notification)) => {
+                NOTIFCATIONS_MUTATING_LSP_STATE.contains(&notification.method.as_str())
+            }
+            _ => false,
+        }
+    }
 }
 
 fn handle_request<TPerfLogger: PerfLogger + 'static, TSchemaDocumentation: SchemaDocumentation>(

compiler/crates/relay-lsp/src/server/task_queue.rs

@@ -7,6 +7,7 @@
 
 use std::sync::Arc;
 use std::thread;
+use std::thread::JoinHandle;
 use std::time::Instant;
 
 use crossbeam::channel::unbounded;
@@ -18,10 +19,13 @@ pub struct TaskQueue<S, T> {
     processor: Arc<dyn TaskProcessor<S, T>>,
     pub receiver: Receiver<T>,
     scheduler: Arc<TaskScheduler<T>>,
+    active_thread_handles: Vec<JoinHandle<()>>,
 }
 
 pub trait TaskProcessor<S, T>: Send + Sync + 'static {
     fn process(&self, state: Arc<S>, task: T);
+
+    fn is_serial_task(&self, task: &T) -> bool;
 }
 
 pub struct TaskScheduler<T> {
@@ -48,25 +52,62 @@ where
             processor,
             receiver,
             scheduler: Arc::new(TaskScheduler { sender }),
+            active_thread_handles: Vec::new(),
         }
     }
 
     pub fn get_scheduler(&self) -> Arc<TaskScheduler<T>> {
         Arc::clone(&self.scheduler)
     }
 
-    pub fn process(&self, state: Arc<S>, task: T) {
+    pub fn process(&mut self, state: Arc<S>, task: T) {
+        let processor = Arc::clone(&self.processor);
+        let is_serial_task = processor.is_serial_task(&task);
+
+        if is_serial_task {
+            // Before starting a serial task, we need to make sure that all
+            // previous tasks have been completed, otherwise the serial task
+            // might interfere with them.
+            self.ensure_previous_tasks_completed();
+        } else {
+            // We do this in order for the active threads to not grow
+            // indefinitely, if there hasn't been a serial task in a while.
+            self.cleanup_already_finished_tasks();
+        }
+
         let task_str = format!("{:?}", &task);
         let now = Instant::now();
         debug!("Processing task {:?}", &task_str);
-        let processor = Arc::clone(&self.processor);
-        thread::spawn(move || {
+
+        let handle = thread::spawn(move || {
             processor.process(state, task);
+
             debug!(
                 "task {} completed in {}ms",
                 task_str,
                 now.elapsed().as_millis()
             );
         });
+
+        if is_serial_task {
+            // If the task is serial, we need to wait for its thread
+            // to complete, before moving onto the next task.
+            let _ = handle.join();
+        } else {
+            self.active_thread_handles.push(handle);
+        }
+    }
+
+    fn ensure_previous_tasks_completed(&mut self) {
+        for handle in self.active_thread_handles.drain(..) {
+            // We don't actually care whether the thread has panicked or not,
+            // we just want to make sure it's finished.
+            let _ = handle.join();
+        }
+    }
+
+    fn cleanup_already_finished_tasks(&mut self) {
+        self.active_thread_handles
+            .retain(|handle| handle.is_finished() == false);
     }
 }