Progress notification infrastructure

crates/ark/src/lsp.rs

@@ -31,6 +31,8 @@ pub mod input_boundaries;
 pub mod inputs;
 pub mod main_loop;
 pub mod markdown;
+pub mod package_sources;
+pub mod progress;
 
 pub mod references;
 pub mod selection_range;

crates/ark/src/lsp/backend.rs

@@ -15,6 +15,7 @@ use amalthea::comm::server_comm::ServerStartMessage;
 use amalthea::comm::server_comm::ServerStartedMessage;
 use anyhow::Context;
 use crossbeam::channel::Sender;
+use oak_index::library::Library;
 use serde_json::Value;
 use stdext::result::ResultExt;
 use tokio::net::TcpListener;
@@ -45,9 +46,11 @@ use crate::lsp::help_topic::HelpTopicResponse;
 use crate::lsp::input_boundaries;
 use crate::lsp::input_boundaries::InputBoundariesParams;
 use crate::lsp::input_boundaries::InputBoundariesResponse;
+use crate::lsp::main_loop::AuxiliaryState;
 use crate::lsp::main_loop::Event;
 use crate::lsp::main_loop::GlobalState;
 use crate::lsp::main_loop::TokioUnboundedSender;
+use crate::lsp::package_sources::PackageSourcesState;
 use crate::lsp::statement_range;
 use crate::lsp::statement_range::StatementRangeParams;
 use crate::lsp::statement_range::StatementRangeResponse;
@@ -228,9 +231,9 @@ struct Backend {
     /// Channel for communication with the main loop.
     events_tx: TokioUnboundedSender<Event>,
 
-    /// Handle to main loop. Drop it to cancel the loop, all associated tasks,
-    /// and drop all owned state.
-    _main_loop: tokio::task::JoinSet<()>,
+    /// Handle to all tokio tasks (main loop, auxiliary loop, package sources loop). Drop
+    /// it to cancel the loops, all associated tasks, and drop all owned state.
+    _tokio_tasks_handle: tokio::task::JoinSet<()>,
 }
 
 impl Backend {
@@ -557,11 +560,55 @@ pub(crate) fn start_lsp(
         let (shutdown_tx, mut shutdown_rx) = tokio::sync::mpsc::channel::<()>(1);
 
         let init = |client: Client| {
-            let state = GlobalState::new(client, r_home, console_notification_tx);
-            let events_tx = state.events_tx();
+            let r = harp::command::r_executable(&r_home);
 
-            // Start main loop and hold onto the handle that keeps it alive
-            let main_loop = state.start();
+            // FIXME: We shouldn't call R code in the kernel to figure this out
+            let library_paths = r_task(|| -> anyhow::Result<Vec<String>> {
+                Ok(harp::RFunction::new("base", ".libPaths")
+                    .call()?
+                    .try_into()?)
+            });
+            let library_paths = match library_paths {
+                Ok(library_paths) => library_paths,
+                Err(err) => {
+                    log::error!("Can't evaluate `libPaths()`: {err:?}");
+                    Vec::new()
+                },
+            };
+            let library_paths: Vec<PathBuf> =
+                library_paths.into_iter().map(PathBuf::from).collect();
+
+            // If reader/writer generation fails, all features related to source references are disabled,
+            // but we don't bring the whole system down. We don't expect this to ever occur in practice.
+            let (package_cache_reader, package_cache_writer) =
+                match oak_sources::new_cache_pair(r, library_paths.clone()).log_err() {
+                    Some((reader, writer)) => (Some(reader), Some(writer)),
+                    None => (None, None),
+                };
+
+            let package_sources = package_cache_reader
+                .map(|reader| Arc::new(reader) as Arc<dyn oak_sources::PackageSources>);
+
+            // Package cache reader goes to `Library`
+            let library = Library::new(library_paths, package_sources);
+
+            // Package cache writer goes to `PackageSourcesState` event loop
+            let (package_sources_state, package_sources_event_tx) = match package_cache_writer {
+                Some(writer) => {
+                    let (state, event_tx) = PackageSourcesState::new(writer);
+                    (Some(state), Some(event_tx))
+                },
+                None => (None, None),
+            };
+
+            let auxiliary_state = AuxiliaryState::new(client.clone());
+
+            let (state, events_tx) = GlobalState::new(
+                client,
+                library,
+                console_notification_tx,
+                package_sources_event_tx,
+            );
 
             // Forward event channel along to `Console`.
             // This also updates an outdated channel after a reconnect.
@@ -576,10 +623,26 @@ pub(crate) fn start_lsp(
                 }
             });
 
+            // The handle that keeps all tokio tasks alive!
+            // Drop it to cancel everything and shut down the service.
+            let mut tokio_tasks_handle = tokio::task::JoinSet::<()>::new();
+
+            // Spawn latency-sensitive auxiliary loop. Must be first to initialise
+            // global transmission channel.
+            tokio_tasks_handle.spawn(async move { auxiliary_state.start().await });
+
+            // Spawn package sources event loop
+            if let Some(package_sources_state) = package_sources_state {
+                tokio_tasks_handle.spawn(async move { package_sources_state.start().await });
+            }
+
+            // Spawn main loop
+            tokio_tasks_handle.spawn(async move { state.start().await });
+
             Backend {
                 shutdown_tx,
                 events_tx,
-                _main_loop: main_loop,
+                _tokio_tasks_handle: tokio_tasks_handle,
             }
         };
 

crates/ark/src/lsp/capabilities.rs

@@ -17,6 +17,7 @@ pub(crate) struct Capabilities {
     dynamic_registration_for_did_change_configuration: bool,
     code_action_literal_support: bool,
     workspace_edit_document_changes: bool,
+    work_done_progress: bool,
 }
 
 impl Capabilities {
@@ -46,10 +47,17 @@ impl Capabilities {
             .and_then(|workspace_edit| workspace_edit.document_changes)
             .is_some_and(|document_changes| document_changes);
 
+        let work_done_progress = client_capabilities
+            .window
+            .as_ref()
+            .and_then(|window| window.work_done_progress)
+            .unwrap_or(false);
+
         Self {
             dynamic_registration_for_did_change_configuration,
             code_action_literal_support,
             workspace_edit_document_changes,
+            work_done_progress,
         }
     }
 
@@ -85,6 +93,10 @@ impl Capabilities {
         self
     }
 
+    pub(crate) fn work_done_progress(&self) -> bool {
+        self.work_done_progress
+    }
+
     pub(crate) fn code_action_provider_capability(&self) -> Option<CodeActionProviderCapability> {
         if !self.code_action_literal_support() {
             return None;