diff --git a/Cargo.lock b/Cargo.lock index c437f149d..67bfe6551 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -532,6 +532,16 @@ dependencies = [ "memchr", ] +[[package]] +name = "command-group" +version = "5.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a68fa787550392a9d58f44c21a3022cfb3ea3e2458b7f85d3b399d0ceeccf409" +dependencies = [ + "nix", + "winapi", +] + [[package]] name = "compact_str" version = "0.9.1" @@ -1753,6 +1763,8 @@ name = "iris-build" version = "0.1.0" dependencies = [ "building", + "command-group", + "configuration", "diagnostics", "dunce", "files", @@ -1862,6 +1874,25 @@ dependencies = [ "tracing", ] +[[package]] +name = "iris-workspace" +version = "0.1.0" +dependencies = [ + "analyzer", + "building", + "configuration", + "files", + "iris-build", + "iris-workspace", + "lsp-types", + "parking_lot", + "serde_json", + "tempfile", + "thiserror", + "tokio", + "url", +] + [[package]] name = "is_terminal_polyfill" version = "1.70.2" @@ -2362,6 +2393,17 @@ dependencies = [ "smallvec", ] +[[package]] +name = "nix" +version = "0.27.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2eb04e9c688eff1c89d72b407f168cf79bb9e867a9d3323ed6c01519eb9cc053" +dependencies = [ + "bitflags 2.13.1", + "cfg-if", + "libc", +] + [[package]] name = "nohash-hasher" version = "0.2.0" diff --git a/compiler-bin/iris-build/Cargo.toml b/compiler-bin/iris-build/Cargo.toml index 655340b75..5bd45528d 100644 --- a/compiler-bin/iris-build/Cargo.toml +++ b/compiler-bin/iris-build/Cargo.toml @@ -9,6 +9,8 @@ repository = "https://github.com/purefunctor/purescript-iris" [dependencies] building = { version = "0.1.0", path = "../../compiler-core/building" } +command-group = "5.0.1" +configuration = { version = "0.1.0", path = "../../compiler-lsp/configuration" } iris-progress = { version = "0.1.0", path = "../iris-progress" } diagnostics = { version = "0.1.0", path = "../../compiler-frontend/diagnostics" } dunce = "1.0.5" diff --git a/compiler-bin/iris-build/src/analysis.rs b/compiler-bin/iris-build/src/analysis.rs new file mode 100644 index 000000000..bc99376bc --- /dev/null +++ b/compiler-bin/iris-build/src/analysis.rs @@ -0,0 +1,353 @@ +//! Editor compilation preparation without semantic query warming or transport state. +//! +//! A prepared compilation has discovered disk inputs and authoritative overlays installed. +//! It has not been checked: source diagnostics do not prevent readiness. Consumers may reconcile +//! newer overlays through the compilation lifecycle before issuing their first query. + +use std::collections::{BTreeMap, BTreeSet}; +use std::io::{Read, Seek, SeekFrom}; +use std::path::{Path, PathBuf}; +use std::process::{Command, ExitStatus, Stdio}; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::{Duration, Instant}; +use std::{io, thread}; + +use building::{ForeignEvent, LifecycleEvent, SourceEvent, SourceUnitKey}; +use command_group::{CommandGroup, GroupChild}; +pub use configuration::SourceDiscovery; +use files::ForeignSourceKind; +use path_absolutize::Absolutize; +use thiserror::Error; +use url::Url; + +use crate::compilation::{CompilationState, MaterializedPrim}; +use crate::compile::{self, CompileError}; +use crate::events::{BuildEvent, BuildEventSink, BuildOutcome}; +use crate::plan::{BuildPlan, BuildPlanError, PackageInput, SelectedSource}; +use crate::walk; + +#[derive(Clone, Debug)] +pub struct AnalysisConfig { + pub root: PathBuf, + pub sources: SourceDiscovery, +} + +#[derive(Clone, Debug)] +pub struct AnalysisOverlay { + pub uri: Url, + pub text: Arc, + pub version: i32, +} + +#[derive(Clone, Debug, Default)] +pub struct CancellationToken { + cancelled: Arc, +} + +impl CancellationToken { + pub fn new() -> CancellationToken { + CancellationToken::default() + } + + pub fn cancel(&self) { + self.cancelled.store(true, Ordering::Relaxed); + } + + pub fn is_cancelled(&self) -> bool { + self.cancelled.load(Ordering::Relaxed) + } + + fn check(&self) -> Result<(), AnalysisError> { + if self.is_cancelled() { Err(AnalysisError::Cancelled) } else { Ok(()) } + } +} + +#[derive(Debug, Error)] +pub enum AnalysisError { + #[error("analysis preparation cancelled")] + Cancelled, + #[error(transparent)] + Io(#[from] io::Error), + #[error(transparent)] + Compile(#[from] CompileError), + #[error(transparent)] + Plan(#[from] BuildPlanError), + #[error(transparent)] + Spago(#[from] spago::LockfileGlobSetError), + #[error("source discovery command exited with {status}: {stderr}")] + CommandFailed { status: ExitStatus, stderr: String }, + #[error("source discovery output is not UTF-8: {0}")] + CommandOutput(#[from] std::string::FromUtf8Error), + #[error("failed to convert path to a file URL: {}", .0.display())] + InvalidPath(PathBuf), + #[error("unsupported analysis document URI: {0}")] + UnsupportedDocument(Url), +} + +/// File locators must round-trip through a native path without changing their identity. +/// This does not canonicalize the filesystem or require the file to exist. +pub fn file_path(uri: &Url) -> Option { + if uri.query().is_some() || uri.fragment().is_some() { + return None; + } + let path = uri.to_file_path().ok()?; + (Url::from_file_path(&path).ok().as_ref() == Some(uri)).then_some(path) +} + +/// Supported source and foreign documents share the same locator rule as disk loading. +pub fn document_path(uri: &Url) -> Option { + let path = file_path(uri)?; + match path.extension().and_then(|extension| extension.to_str()) { + Some("purs" | "js" | "jsx") => Some(path), + _ => None, + } +} + +#[derive(Clone, Debug)] +pub struct AnalysisSourceRoot { + pub path: PathBuf, + pub editable: bool, +} + +#[derive(Clone, Debug)] +pub struct AnalysisSelection { + /// Disk selection only; opening a buffer does not add it to this set. + pub selected_sources: BTreeSet>, + /// Most specific package roots precede their ancestors, including canonical aliases. + pub source_roots: Vec, + root: PathBuf, + metadata: BTreeMap, +} + +impl AnalysisSelection { + /// Editable metadata for a supported file document, including buffer-only sources. + /// Like the editor's open-document policy, files outside known roots participate read-only. + /// Foreign documents inherit the metadata of their associated PureScript source. + pub fn metadata(&self, uri: &Url) -> Option { + let path = document_path(uri)?; + let source = if uri.path().ends_with(".purs") { path } else { path.with_extension("purs") }; + if let Some(editable) = self.metadata.get(&source) { + return Some(*editable); + } + let package = self.source_roots.iter().find(|root| source.starts_with(&root.path)); + Some(package.map_or_else(|| source.starts_with(&self.root), |root| root.editable)) + } +} + +pub struct PreparedAnalysis { + pub compilation: CompilationState, + pub selection: AnalysisSelection, +} + +pub fn prepare( + config: &AnalysisConfig, + overlays: &[AnalysisOverlay], + prim: Arc, + cancellation: &CancellationToken, + events: &impl BuildEventSink, +) -> Result { + cancellation.check()?; + for overlay in overlays { + if document_path(&overlay.uri).is_none() { + return Err(AnalysisError::UnsupportedDocument(Url::clone(&overlay.uri))); + } + } + let started = Instant::now(); + events.send(BuildEvent::Preparing); + let root = config.root.absolutize()?.into_owned(); + let (selection, packages) = discover(&root, &config.sources, cancellation)?; + cancellation.check()?; + let selected = selection.metadata.keys().map(|path| { + let identity = dunce::canonicalize(path)?; + Ok(SelectedSource { path: path.clone(), identity }) + }); + let selected = selected.collect::, io::Error>>()?; + let packages = packages.into_iter().map(|package| { + let identities = package.source_identities.iter().map(dunce::canonicalize); + let source_identities = identities.collect::, _>>()?; + Ok(PackageInput { source_identities, ..package }) + }); + let packages = packages.collect::, io::Error>>()?; + let plan = BuildPlan::new(selected, packages)?; + events.send(BuildEvent::PlanReady { package_count: plan.package_count() }); + + let mut compilation = CompilationState::new(prim, false); + let source_paths = plan.packages().flat_map(|package| package.source_paths.iter().cloned()); + let mut source_paths = source_paths.collect::>(); + for overlay in overlays { + cancellation.check()?; + let Some(metadata) = selection.metadata(&overlay.uri) else { + continue; + }; + let is_source = overlay.uri.path().ends_with(".purs"); + let source_uri = + if is_source { overlay.uri.clone() } else { sibling_uri(&overlay.uri, "purs") }; + let source = source_uri.to_file_path().expect("supported document has a file path"); + let javascript_uri = sibling_uri(&source_uri, "js"); + let jsx_uri = sibling_uri(&source_uri, "jsx"); + let unit = SourceUnitKey::with_foreign_sources( + source_uri.as_str(), + javascript_uri.as_str(), + jsx_uri.as_str(), + ); + let text = Arc::clone(&overlay.text); + let version = overlay.version; + let event = if is_source { + source_paths.insert(source); + LifecycleEvent::Source { unit, event: SourceEvent::Opened { text, version, metadata } } + } else { + let kind = if overlay.uri.path().ends_with(".js") { + ForeignSourceKind::JavaScript + } else { + ForeignSourceKind::Jsx + }; + LifecycleEvent::Foreign { unit, kind, event: ForeignEvent::Opened { text, version } } + }; + compilation.apply(event); + } + for path in source_paths { + cancellation.check()?; + let uri = file_uri(&path)?; + let metadata = selection.metadata(&uri).expect("source selection must have metadata"); + compile::load_source(&mut compilation, &path, metadata)?; + } + cancellation.check()?; + let outcome = if compilation.source_ids().next().is_some() { + BuildOutcome::Succeeded + } else { + BuildOutcome::NoInputs + }; + events.send(BuildEvent::Finished { duration: started.elapsed(), outcome }); + Ok(PreparedAnalysis { compilation, selection }) +} + +fn discover( + root: &Path, + sources: &SourceDiscovery, + cancellation: &CancellationToken, +) -> Result<(AnalysisSelection, Vec), AnalysisError> { + let mut metadata = BTreeMap::new(); + let mut source_roots = vec![]; + let mut packages = vec![]; + match sources { + SourceDiscovery::Spago {} => { + for (name, package) in spago::source_files_by_package(root)? { + cancellation.check()?; + let editable = matches!( + package.reference, + spago::PackageReference::Workspace | spago::PackageReference::Local + ); + for path in &package.sources { + metadata.insert(path.clone(), editable); + } + for path in package.roots { + let path = root.join(path).absolutize()?.into_owned(); + source_roots.push(AnalysisSourceRoot { path: path.clone(), editable }); + if let Ok(canonical) = dunce::canonicalize(&path) + && canonical != path + { + source_roots.push(AnalysisSourceRoot { path: canonical, editable }); + } + } + packages.push(PackageInput { + name, + source_identities: package.sources, + dependencies: package.dependencies.into_iter().collect(), + }); + } + } + SourceDiscovery::Command { program, arguments } => { + let output = source_command(root, program, arguments, cancellation)?; + let walked = walk::walk_filtered(root, output.lines(), std::iter::empty::<&Path>()) + .map_err(CompileError::from)?; + let files = walked + .files + .into_iter() + .filter(|path| path.extension().is_some_and(|extension| extension == "purs")); + let files = files.collect::>(); + for path in &files { + metadata.insert(path.clone(), path.starts_with(root)); + } + source_roots.push(AnalysisSourceRoot { path: root.to_path_buf(), editable: true }); + packages.push(PackageInput { + name: "unmanaged".into(), + source_identities: files, + dependencies: vec![], + }); + } + } + source_roots.sort_by_key(|root| std::cmp::Reverse(root.path.components().count())); + let locators = metadata.keys().map(|path| file_uri(path).map(|uri| Arc::from(uri.as_str()))); + let selected_sources = locators.collect::>()?; + let selection = + AnalysisSelection { selected_sources, source_roots, root: root.to_path_buf(), metadata }; + Ok((selection, packages)) +} + +fn file_uri(path: &Path) -> Result { + Url::from_file_path(path).map_err(|_| AnalysisError::InvalidPath(path.to_path_buf())) +} + +fn sibling_uri(uri: &Url, extension: &str) -> Url { + let path = uri.path(); + let file_name_start = path.rfind('/').map_or(0, |index| index + 1); + let extension_start = path[file_name_start..] + .rfind('.') + .filter(|index| *index > 0) + .map_or(path.len(), |index| file_name_start + index); + let mut sibling = uri.clone(); + sibling.set_path(&format!("{}.{extension}", &path[..extension_start])); + sibling +} + +struct SourceCommand(GroupChild); + +impl Drop for SourceCommand { + fn drop(&mut self) { + // The leader may exit while descendants still run. Keep group ownership until cleanup. + let _ = self.0.kill(); + let _ = self.0.wait(); + } +} + +fn source_command( + root: &Path, + program: &str, + arguments: &[String], + cancellation: &CancellationToken, +) -> Result { + cancellation.check()?; + let mut stdout = tempfile::tempfile()?; + let mut stderr = tempfile::tempfile()?; + let child = Command::new(program) + .args(arguments) + .current_dir(root) + .stdin(Stdio::null()) + .stdout(stdout.try_clone()?) + .stderr(stderr.try_clone()?) + .group_spawn()?; + let mut child = SourceCommand(child); + let status = loop { + cancellation.check()?; + if let Some(status) = child.0.try_wait()? { + break status; + } + thread::sleep(Duration::from_millis(10)); + }; + drop(child); + cancellation.check()?; + if !status.success() { + stderr.seek(SeekFrom::Start(0))?; + let mut output = vec![]; + stderr.read_to_end(&mut output)?; + return Err(AnalysisError::CommandFailed { + status, + stderr: String::from_utf8_lossy(&output).into_owned(), + }); + } + stdout.seek(SeekFrom::Start(0))?; + let mut output = vec![]; + stdout.read_to_end(&mut output)?; + Ok(String::from_utf8(output)?) +} diff --git a/compiler-bin/iris-build/src/compilation.rs b/compiler-bin/iris-build/src/compilation.rs index 2eeb089ee..d55b68b2c 100644 --- a/compiler-bin/iris-build/src/compilation.rs +++ b/compiler-bin/iris-build/src/compilation.rs @@ -29,22 +29,23 @@ impl MaterializedPrim { pub struct CompilationState { engine: QueryEngine, - files: FileLifecycle, + pub(crate) files: FileLifecycle, sources: BTreeSet, - prim: MaterializedPrim, + prim: Arc, } pub struct CompilationParts { pub engine: QueryEngine, pub files: FileLifecycle, - pub prim: MaterializedPrim, + pub prim: Arc, } impl CompilationState { pub fn new( - prim: MaterializedPrim, + prim: impl Into>, prim_metadata: Metadata, ) -> CompilationState { + let prim = prim.into(); let engine = QueryEngine::default(); let mut files = FileLifecycle::default(); @@ -74,6 +75,15 @@ impl CompilationState CompilationState { engine, files, sources: BTreeSet::new(), prim } } + pub fn apply(&mut self, event: LifecycleEvent) -> LifecycleChange { + let change = self.files.apply(&self.engine, event); + self.sources.extend(change.changed_sources()); + for removed in change.removed_sources() { + self.sources.remove(&removed.file_id); + } + change + } + pub fn observe_source( &mut self, unit: SourceUnitKey, @@ -82,12 +92,7 @@ impl CompilationState ) -> LifecycleChange { let event = LifecycleEvent::Source { unit, event: SourceEvent::DiskObserved { disk, metadata } }; - let change = self.files.apply(&self.engine, event); - self.sources.extend(change.changed_sources()); - for removed in change.removed_sources() { - self.sources.remove(&removed.file_id); - } - change + self.apply(event) } pub fn observe_foreign( @@ -98,7 +103,7 @@ impl CompilationState ) -> LifecycleChange { let event = LifecycleEvent::Foreign { unit, kind, event: ForeignEvent::DiskObserved { disk } }; - self.files.apply(&self.engine, event) + self.apply(event) } pub fn source_content(&self, locator: &str) -> Result>, QueryError> { @@ -117,7 +122,7 @@ impl CompilationState self.engine.snapshot() } - pub(crate) fn query_engine(&self) -> &QueryEngine { + pub fn query_engine(&self) -> &QueryEngine { &self.engine } diff --git a/compiler-bin/iris-build/src/compile.rs b/compiler-bin/iris-build/src/compile.rs index 813ee0374..263691667 100644 --- a/compiler-bin/iris-build/src/compile.rs +++ b/compiler-bin/iris-build/src/compile.rs @@ -4,7 +4,7 @@ use std::sync::Arc; use std::time::Instant; use std::{fs, io}; -use building::{DiskObservation, QueryError, SourceUnitKey}; +use building::{DiskObservation, DocumentKey, QueryError, SourceUnitKey}; use diagnostics::Severity; use files::{FileId, ForeignSourceKind}; use itertools::Itertools; @@ -290,7 +290,7 @@ pub(crate) fn finish_initial( Ok(RebuildResult { outcome, outputs }) } -fn load_source( +pub(crate) fn load_source( compilation: &mut CompilationState, path: &Path, metadata: Metadata, @@ -305,17 +305,30 @@ where let foreign_url = Url::from_file_path(&foreign_path) .map_err(|_| CompileError::InvalidPath(PathBuf::clone(&foreign_path)))?; let unit = SourceUnitKey::new(source_url.as_str(), foreign_url.as_str()); - let content = fs::read_to_string(path)?; - let change = compilation.observe_source( - SourceUnitKey::clone(&unit), - DiskObservation::Found(content.into()), - metadata, - ); - let file_id = change - .changed_sources() - .next() - .expect("invariant violated: newly loaded source did not change its lifecycle"); + let file_id = if compilation.files.is_open(&DocumentKey::Source(unit.clone())) { + compilation.files.source_id(unit.source()).expect("open source must have an identity") + } else { + let content = fs::read_to_string(path)?; + let change = compilation.observe_source( + SourceUnitKey::clone(&unit), + DiskObservation::Found(content.into()), + metadata, + ); + change.changed_sources().next().expect("newly loaded source must change its lifecycle") + }; + load_foreign(compilation, path, &unit)?; + Ok(file_id) +} + +pub(crate) fn load_foreign( + compilation: &mut CompilationState, + path: &Path, + unit: &SourceUnitKey, +) -> Result<(), CompileError> { for kind in ForeignSourceKind::ALL { + if compilation.files.is_open(&DocumentKey::Foreign(unit.clone(), kind)) { + continue; + } let foreign_path = path.with_extension(kind.extension()); let disk = match fs::read_to_string(&foreign_path) { Ok(content) => DiskObservation::Found(content.into()), @@ -324,7 +337,7 @@ where }; compilation.observe_foreign(SourceUnitKey::clone(&unit), kind, disk); } - Ok(file_id) + Ok(()) } fn query_package(engine: &building::QueryEngine, sources: &[FileId]) -> Result<(), CompileError> { diff --git a/compiler-bin/iris-build/src/lib.rs b/compiler-bin/iris-build/src/lib.rs index 40228895d..d07a0c093 100644 --- a/compiler-bin/iris-build/src/lib.rs +++ b/compiler-bin/iris-build/src/lib.rs @@ -1,3 +1,4 @@ +pub mod analysis; pub mod events; pub mod executor; pub mod plan; diff --git a/compiler-bin/iris-lsp/src/server/workspace.rs b/compiler-bin/iris-lsp/src/server/workspace.rs index 9195e00ed..1b00e3189 100644 --- a/compiler-bin/iris-lsp/src/server/workspace.rs +++ b/compiler-bin/iris-lsp/src/server/workspace.rs @@ -69,7 +69,7 @@ pub(super) struct ReadyWorkspace { pub(super) workspace_symbols_cache: Arc>, pub(super) suggestions_cache: Arc>, pub(super) diagnostics: DiagnosticScheduler, - _prim: MaterializedPrim, + _prim: Arc, } pub(super) struct PreparedInitialWorkspace { diff --git a/compiler-core/building/src/engine.rs b/compiler-core/building/src/engine.rs index be08c3b9f..5e33e4074 100644 --- a/compiler-core/building/src/engine.rs +++ b/compiler-core/building/src/engine.rs @@ -200,6 +200,36 @@ fn state_references_removed_file( #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] struct SnapshotId(u32); +/// Cooperative cancellation for one operation and its descendant snapshots. +/// +/// Queries observe cancellation at query entry and before computation. Cancelling +/// does not interrupt computation between these checks or wake queries waiting +/// for another snapshot's result. Those waits end when the producer completes or +/// drops its promise, and may still return a successful result after cancellation. +/// A cancelled producer drops its promise, so its waiters can receive +/// [`QueryError::Cancelled`] even when their own operation tokens are not cancelled. +#[derive(Debug, Clone, Default)] +pub struct QueryCancellation { + cancelled: Arc, +} + +impl QueryCancellation { + /// Creates an independent, initially uncancelled operation token. + pub fn new() -> QueryCancellation { + QueryCancellation::default() + } + + /// Permanently cancels this token and its clones without waiting for queries. + pub fn cancel(&self) { + self.cancelled.store(true, Ordering::Relaxed); + } + + /// Returns whether cancellation has been requested for this operation. + pub fn is_cancelled(&self) -> bool { + self.cancelled.load(Ordering::Relaxed) + } +} + #[derive(Default)] struct GlobalState { /// An atomic token that determines if query execution had been cancelled. @@ -339,6 +369,7 @@ struct QueryControl { id: SnapshotId, local: Arc, global: Arc, + cancellation: Option, } impl QueryControl { @@ -347,7 +378,13 @@ impl QueryControl { let local = Arc::new(LocalState::default()); let global = Arc::clone(&self.global); let id = global.next_snapshot(); - QueryControl { _guard, id, local, global } + let cancellation = self.cancellation.clone(); + QueryControl { _guard, id, local, global, cancellation } + } + + fn is_cancelled(&self) -> bool { + self.global.cancelled.load(Ordering::Relaxed) + || self.cancellation.as_ref().is_some_and(QueryCancellation::is_cancelled) } } @@ -357,7 +394,8 @@ impl Default for QueryControl { let local = Arc::new(LocalState::default()); let global = Arc::new(GlobalState::default()); let id = global.next_snapshot(); - QueryControl { _guard, id, local, global } + let cancellation = None; + QueryControl { _guard, id, local, global, cancellation } } } @@ -375,6 +413,9 @@ impl QueryEngine { /// Snapshots are read locks over the [`QueryEngine`] that must /// be sent across threads to perform query execution. /// + /// Descendants inherit this snapshot's operation cancellation + /// token, if any. + /// /// As with read locks, keeping snapshots alive indefinitely is /// a logic error and will cause a deadlock on mutation or on a /// [cancellation request]. @@ -388,6 +429,18 @@ impl QueryEngine { QueryEngine { input, derived, interned, control } } + /// Creates a snapshot using the given operation token instead of inheriting + /// this engine's token. Descendants created with [`Self::snapshot`] inherit it. + /// + /// Cancelling the token does not cancel the engine or independent snapshots. + /// The snapshot still holds a read lock until dropped; see [`Self::snapshot`] + /// and [`QueryCancellation`] for locking and cooperative cancellation limits. + pub fn snapshot_with_cancellation(&self, token: QueryCancellation) -> QueryEngine { + let mut snapshot = self.snapshot(); + snapshot.control.cancellation = Some(token); + snapshot + } + /// Creates a cancellation request for queries. /// /// Query cancellation is cooperative. A cancellation flag is read @@ -495,7 +548,7 @@ impl QueryEngine { ComputeFn: Fn(&QueryEngine) -> QueryResult, V: Eq + Clone, { - if self.control.global.cancelled.load(Ordering::Relaxed) { + if self.control.is_cancelled() { return Err(QueryError::Cancelled); } @@ -613,7 +666,7 @@ impl QueryEngine { ComputeFn: Fn(&QueryEngine) -> QueryResult, V: Eq + Clone, { - if self.control.global.cancelled.load(Ordering::Relaxed) { + if self.control.is_cancelled() { return Err(QueryError::Cancelled); } diff --git a/compiler-lsp/iris-workspace/Cargo.toml b/compiler-lsp/iris-workspace/Cargo.toml new file mode 100644 index 000000000..5e9d1a4b8 --- /dev/null +++ b/compiler-lsp/iris-workspace/Cargo.toml @@ -0,0 +1,27 @@ +[package] +name = "iris-workspace" +version = "0.1.0" +edition = "2024" +description = "Stateful, transport-independent Iris workspace analysis." +license = "BSD-3-Clause" + +[features] +test-support = [] + +[dependencies] +analyzer = { path = "../analyzer" } +building = { path = "../../compiler-core/building" } +configuration = { path = "../configuration" } +files = { path = "../../compiler-core/files" } +iris-build = { path = "../../compiler-bin/iris-build" } +lsp-types.workspace = true +parking_lot = "0.12.5" +serde_json = "1.0.151" +thiserror = "2.0.20" +tokio = { version = "1", features = ["sync"] } +url = "2.5.8" + +[dev-dependencies] +iris-workspace = { path = ".", features = ["test-support"] } +tempfile = "3.27.0" +tokio = { version = "1", features = ["macros", "rt", "time", "net", "io-util"] } diff --git a/compiler-lsp/iris-workspace/README.md b/compiler-lsp/iris-workspace/README.md new file mode 100644 index 000000000..eb761ec2a --- /dev/null +++ b/compiler-lsp/iris-workspace/README.md @@ -0,0 +1,58 @@ +# Iris workspace service + +`iris-workspace` manages editor workspace analysis independently of `iris-lsp`. It uses +`iris-build` to find and load project files, then runs the analyzer to answer requests such as +hover and completion. Communication with the editor, matching responses to requests, agreeing +on supported features, and displaying progress are handled outside this crate. + +Text from open editor documents takes precedence over files on disk. The service accepts and +orders input changes without waiting for compiler work to finish. A separate worker runs +compilation and analysis. Cancelled work may still be running while a rebuild waits to start. + +Documents use local file URLs without query strings or fragments. Converting a URL to a local +path and back must preserve it, so editor buffers and disk reads identify the same document. +Files need not exist on disk, and filenames containing spaces remain supported. + +Rebuilds keep open documents but make analysis unavailable until the latest inputs have been +loaded. If preparation fails, analysis remains unavailable rather than falling back to the +previous compiler state. Type errors in the source do not prevent the service from answering +analysis requests once those inputs are loaded. + +Edits cancel unfinished analysis requests and discard queued requests for the old inputs. The +caller decides whether to request analysis again. Once a response has been placed in its channel, +later edits do not revoke it, even if the caller has not received it yet. This applies to both +successful responses and errors. Identical configuration and diagnostic-policy changes leave +analysis requests running. + +Diagnostics are checked separately when the language server handles them, immediately before +passing them to the transport. `Delivery::release` discards diagnostics made obsolete by newer +inputs. No further check is required while the transport queues or sends the notification. +Cancelling diagnostics does not itself schedule another run; document events and rebuilds +determine when to collect them. + +The service keeps the latest queued status, progress, and diagnostics rather than accumulating +every update. Callers should finish handling each event before requesting the next, and allow +event handling to stop when the connection closes. + +## Runnable walkthrough + +The [language-server example](examples/language_server.rs) uses one temporary project and workspace, +with separate functions demonstrating: + +- Lifecycle: configuration, open buffers, hover, completed responses surviving edits, + rebuilds, cancellation, and shutdown. +- Diagnostics: an unsaved error, an incremental edit, saving and file + watcher notifications, and clearing diagnostics when a buffer-only document closes. +- Completion and rename: editor-owned request IDs, completion + resolution, conflicting rename confirmation, versioned edits, and outdated completion tokens. + +It explains the caller's responsibilities alongside executable requests and assertions, prints +labelled results, and removes its temporary project on exit. Node.js is required. + +```sh +cargo run -p iris-workspace --example language_server +``` + +Tests in `tests/` run sequences of workspace commands against the real compiler. They pause work +at predefined points to control the order of operations. +Run them with `cargo nextest run -p iris-workspace`. diff --git a/compiler-lsp/iris-workspace/examples/language_server.rs b/compiler-lsp/iris-workspace/examples/language_server.rs new file mode 100644 index 000000000..05a069beb --- /dev/null +++ b/compiler-lsp/iris-workspace/examples/language_server.rs @@ -0,0 +1,424 @@ +//! Editor lifecycle, diagnostics, completion, and rename using one workspace. +//! Run with `cargo run -p iris-workspace --example language_server` (requires Node.js). + +use std::time::Duration; + +use iris_workspace::{ + AnalyzerCapabilities, Command, Configuration, ConfigurationInput, Document, Event, + EventReceiver, InputSequence, LanguageServer, Options, Reply, RequestFailure, SourceDiscovery, + Status, Workspace, WorkspaceSession, +}; +use lsp_types::{ + CompletionItem, CompletionResponse, DiagnosticSeverity, DocumentChanges, HoverContents, OneOf, + Position, PublishDiagnosticsParams, Range, TextDocumentContentChangeEvent, Url, +}; + +const SOURCE: &str = "module Main where\n\nvalue :: Int\nvalue = 1\n\nuse = value\n"; +const RENAMED: &str = "module Main where\n\ncount :: Int\ncount = 1\n\nuse = count\n"; +const BUFFER: &str = "module Main where\nvalue = \"buffer\"\n"; +const EDITED: &str = "module Main where\nvalue = true\n"; +const VALID: &str = "module Main where\nvalue :: Int\nvalue = 42\n"; +const BROKEN: &str = "module Main where\nvalue :: Int\nvalue = \"wrong\"\n"; + +fn main() { + let directory = tempfile::tempdir().expect("create the example project"); + let path = directory.path().join("Main.purs"); + std::fs::write(&path, SOURCE).expect("write Main.purs"); + let uri = Url::from_file_path(path).expect("convert Main.purs to a file URI"); + + let sources = SourceDiscovery::Command { + program: "node".into(), + arguments: vec!["-e".into(), "console.log('Main.purs')".into()], + }; + let mut settings = Configuration { sources, ..Configuration::default() }; + settings.diagnostics.on_change = true; + let configuration = ConfigurationInput { root: directory.path().to_path_buf(), settings }; + + // The language server translates initialize capabilities into analyzer options. This client + // can display change annotations and ask for confirmation before applying conflicting edits. + let options = Options { + capabilities: AnalyzerCapabilities::default().with_change_annotations(), + ..Options::default() + }; + let WorkspaceSession { workspace, mut events, join } = + Workspace::start(options).expect("start the workspace service"); + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("create the async runtime"); + let result = runtime.block_on(async { + let scenarios = async { + println!("Iris workspace: language-server walkthrough"); + lifecycle(&workspace, &mut events, configuration, &uri).await; + diagnostics(&workspace, &mut events, &uri).await; + completion_and_rename(&workspace, &mut events, &uri).await; + shutdown(&workspace, &mut events).await; + }; + tokio::time::timeout(Duration::from_secs(30), scenarios).await + }); + + // Joining waits for compiler snapshots and discovery descendants to retire. Do it outside + // the async protocol loop, and before deleting the temporary project's files. + join.join().expect("join the workspace controller without a panic"); + result.expect("language-server walkthrough timed out"); + println!("\nAll scenarios passed."); +} + +async fn lifecycle( + workspace: &Workspace, + events: &mut EventReceiver, + configuration: ConfigurationInput, + uri: &Url, +) { + println!("\n1. Open buffers, rebuilds, and cancellation"); + + // Open buffers belong to the controller, not an engine incarnation. They can arrive even + // before configuration and override disk when preparation eventually loads the project. + let document = Document::Open { uri: Url::clone(uri), text: BUFFER.into(), version: 1 }; + workspace.send(Command::Document(document)).expect("admit the buffer before configuration"); + let sequence = + workspace.send(Command::Configure(configuration)).expect("configure the project"); + // Keep the sequence beside the protocol request ID. Configuration outcomes are retained even + // if later input replaces this configuration; status updates alone cannot complete requests. + loop { + let delivery = events.recv().await.expect("receive the configuration outcome"); + if let Ok(Event::ConfigurationFinished { sequence: completed, outcome }) = + delivery.release() + { + assert_eq!(completed, sequence); + assert_eq!(outcome, iris_workspace::ConfigurationOutcome::Rebuilt); + println!(" Configuration {}: {outcome:?}", sequence.value); + break; + } + } + ready(workspace, events, sequence).await; + + let (reply, request) = Reply::channel(); + let command = + LanguageServer::Hover { uri: Url::clone(uri), position: Position::new(1, 1), reply }; + workspace.send(Command::LanguageServer(command)).expect("admit hover on the ready workspace"); + let held = request.await.expect("compute hover before the document changes"); + + // This reply has settled. Later edits cancel unfinished reads, but do not recall replies + // already placed in their channels. The editor may receive this response after its edit. + let change = + TextDocumentContentChangeEvent { range: None, range_length: None, text: EDITED.into() }; + let document = Document::Change { uri: Url::clone(uri), version: 2, changes: vec![change] }; + let sequence = workspace.send(Command::Document(document)).expect("admit the version 2 edit"); + assert!(held.is_some()); + println!(" Completed hover retained after didChange."); + ready(workspace, events, sequence).await; + + let before = workspace.status(); + let sequence = workspace.send(Command::Reload).expect("admit the full rebuild"); + ready(workspace, events, sequence).await; + assert_ne!(before, workspace.status()); + + // Each variant fixes its reply type. Interactive replies need no publication-time check. + let (reply, request) = Reply::channel(); + let command = + LanguageServer::Hover { uri: Url::clone(uri), position: Position::new(1, 1), reply }; + workspace.send(Command::LanguageServer(command)).expect("admit hover after rebuilding"); + let hover = request + .await + .expect("compute hover from the preserved buffer") + .expect("the value declaration must have hover information"); + print_message("Hover after rebuild", serde_json::json!({"result": hover})); + + let expected = lsp_types::MarkedString::LanguageString(lsp_types::LanguageString { + language: "purescript".into(), + value: "value :: Boolean".into(), + }); + assert_eq!(hover.contents, HoverContents::Array(vec![expected])); + + // Request cancellation is monotonic and nonblocking. It does not mean the worker has freed + // its slot; future work still waits for the cancelled computation to actually finish. + let (reply, request) = Reply::channel(); + request.cancellation().cancel(); + let command = LanguageServer::DocumentSymbols { uri: Url::clone(uri), reply }; + workspace.send(Command::LanguageServer(command)).expect("admit the already-cancelled request"); + assert!(matches!(request.await, Err(RequestFailure::Cancelled))); + println!(" Cancelled request returned Cancelled."); + + let sequence = workspace + .send(Command::Document(Document::Close(Url::clone(uri)))) + .expect("close the lifecycle buffer before the next scenario"); + ready(workspace, events, sequence).await; +} + +async fn diagnostics(workspace: &Workspace, events: &mut EventReceiver, uri: &Url) { + println!("\n2. Diagnostics through edit, save, and close"); + + // textDocument/didOpen supplies the editor's text and version. The valid disk file must not + // replace it: diagnostics should describe the unsaved error. + let open = Document::Open { uri: Url::clone(uri), text: BROKEN.into(), version: 7 }; + workspace.send(Command::Document(open)).expect("accept didOpen"); + + let initial = publish_next(events, uri, Some(7)).await; + let mismatch = initial.diagnostics.iter().any(|diagnostic| { + diagnostic.severity == Some(DiagnosticSeverity::ERROR) + && diagnostic.message.contains("Int") + && diagnostic.message.contains("String") + }); + assert!(mismatch, "the unsaved buffer must report its Int/String mismatch"); + + // textDocument/didChange uses the position encoding negotiated during initialization. + // Options::default selects UTF-16. This range replaces the quoted literal, including quotes. + let change = TextDocumentContentChangeEvent { + range: Some(Range::new(Position::new(2, 8), Position::new(2, 15))), + range_length: None, + text: "42".into(), + }; + let document = Document::Change { uri: Url::clone(uri), version: 8, changes: vec![change] }; + workspace.send(Command::Document(document)).expect("accept the incremental didChange"); + + let fixed = publish_next(events, uri, Some(8)).await; + assert!(fixed.diagnostics.is_empty(), "the corrected buffer must clear the error"); + + // didSave is a notification, not a command to write a file. The editor performs the write. + // A watcher can then report that same write; the open document remains authoritative. + let path = uri.to_file_path().expect("recover the example's source path"); + std::fs::write(path, VALID).expect("simulate the editor saving the corrected buffer"); + workspace.send(Command::Document(Document::Save(Url::clone(uri)))).expect("accept didSave"); + workspace.send(Command::FilesChanged(vec![Url::clone(uri)])).expect("accept the watcher event"); + + let rebuilt = publish_next(events, uri, Some(8)).await; + assert!(rebuilt.diagnostics.is_empty(), "the saved buffer must remain valid after rediscovery"); + + // A buffer-only document has no disk source to restore on close. Publishing an empty list + // is still necessary: silence would leave its last error visible in the editor. + let scratch = uri.join("Scratch.purs").expect("construct a buffer-only file URI"); + let open = Document::Open { + uri: Url::clone(&scratch), + text: "module Scratch where\nvalue :: Int\nvalue = \"wrong\"\n".into(), + version: 1, + }; + workspace.send(Command::Document(open)).expect("accept didOpen for the buffer-only source"); + + let broken = publish_next(events, &scratch, Some(1)).await; + assert!(!broken.diagnostics.is_empty(), "the scratch buffer must have an error to clear"); + + workspace + .send(Command::Document(Document::Close(Url::clone(&scratch)))) + .expect("accept didClose"); + let closed = publish_next(events, &scratch, None).await; + assert!(closed.diagnostics.is_empty(), "closing the scratch buffer must clear its diagnostics"); + + let sequence = workspace + .send(Command::Document(Document::Close(Url::clone(uri)))) + .expect("close the diagnostics buffer before the next scenario"); + ready(workspace, events, sequence).await; +} + +async fn completion_and_rename(workspace: &Workspace, events: &mut EventReceiver, uri: &Url) { + println!("\n3. Completion resolution and rename"); + + let open = Document::Open { uri: Url::clone(uri), text: SOURCE.into(), version: 6 }; + let sequence = workspace.send(Command::Document(open)).expect("accept didOpen at version 6"); + ready(workspace, events, sequence).await; + + // The incoming textDocument/completion ID stays in the handler. The workspace receives only + // semantic inputs and a reply channel whose result type is fixed by the command variant. + let (reply, request) = Reply::channel(); + let command = + LanguageServer::Completion { uri: Url::clone(uri), position: Position::new(5, 9), reply }; + workspace.send(Command::LanguageServer(command)).expect("admit the completion request"); + let response = request.await.expect("compute completion suggestions"); + print_message( + "Completion", + serde_json::json!({"jsonrpc": "2.0", "id": 101, "result": response}), + ); + + let items = match response.expect("the value prefix must have completions") { + CompletionResponse::Array(items) => items, + CompletionResponse::List(list) => list.items, + }; + + // Keep the original item to resolve its outdated token again after the rename below. + let item = items + .into_iter() + .find(|item| item.label == "value") + .expect("completion must suggest the declared value"); + assert!(item.data.as_ref().is_some_and(serde_json::Value::is_string)); + + // completionItem/resolve echoes the selected item's data unchanged. The string is an opaque + // workspace token, not a file ID or something the adapter should decode or manufacture. + let (reply, request) = Reply::channel(); + let command = LanguageServer::ResolveCompletion { item: CompletionItem::clone(&item), reply }; + workspace.send(Command::LanguageServer(command)).expect("admit completionItem/resolve"); + let resolved = request.await.expect("resolve the completion's type information"); + print_message( + "Resolved completion", + serde_json::json!({"jsonrpc": "2.0", "id": 102, "result": resolved}), + ); + assert!(resolved.detail.as_ref().is_some_and(|detail| detail.contains("Int"))); + assert!(resolved.data.is_none(), "resolved items must not expose compiler identities"); + + // Renaming value to use would change name resolution. With annotation support the result + // asks the editor for confirmation, and records the open document version for its edits. + let (reply, request) = Reply::channel(); + let command = LanguageServer::Rename { + uri: Url::clone(uri), + position: Position::new(5, 7), + new_name: "use".into(), + reply, + }; + workspace.send(Command::LanguageServer(command)).expect("admit the conflicting rename"); + let edit = request + .await + .expect("compute annotated rename edits") + .expect("the selected value must be renameable"); + print_message( + "Rename requiring confirmation", + serde_json::json!({"jsonrpc": "2.0", "id": 103, "result": edit}), + ); + + let annotations = edit.change_annotations.expect("conflicting edits need annotations"); + assert!(annotations.values().any(|annotation| annotation.needs_confirmation == Some(true))); + let DocumentChanges::Edits(changes) = + edit.document_changes.expect("rename must edit documents") + else { + panic!("this rename must contain text edits, not file operations"); + }; + assert_eq!(changes.len(), 1); + assert_eq!(&changes[0].text_document.uri, uri); + assert_eq!(changes[0].text_document.version, Some(6)); + assert_eq!(changes[0].edits.len(), 3); + assert!(changes[0].edits.iter().any(|edit| matches!(edit, OneOf::Right(_)))); + + // The editor declines that rename and requests a non-conflicting name instead. Returning + // WorkspaceEdit does not modify the workspace: only subsequent didChange updates its text. + let (reply, request) = Reply::channel(); + let command = LanguageServer::Rename { + uri: Url::clone(uri), + position: Position::new(5, 7), + new_name: "count".into(), + reply, + }; + workspace.send(Command::LanguageServer(command)).expect("admit the non-conflicting rename"); + let edit = + request.await.expect("compute the count rename").expect("value must have rename edits"); + print_message( + "Rename to count", + serde_json::json!({"jsonrpc": "2.0", "id": 104, "result": edit}), + ); + + let changes = edit.changes.expect("non-conflicting rename must return ordinary edits"); + let edits = changes.get(uri).expect("rename must edit Main.purs"); + for edit in edits { + assert_eq!(edit.new_text, "count"); + } + + let mut ranges = edits.iter().map(|edit| edit.range).collect::>(); + ranges.sort_by_key(|range| (range.start.line, range.start.character)); + assert_eq!( + ranges, + vec![ + Range::new(Position::new(2, 0), Position::new(2, 5)), + Range::new(Position::new(3, 0), Position::new(3, 5)), + Range::new(Position::new(5, 6), Position::new(5, 11)), + ] + ); + + let change = + TextDocumentContentChangeEvent { range: None, range_length: None, text: RENAMED.into() }; + let document = Document::Change { uri: Url::clone(uri), version: 7, changes: vec![change] }; + let sequence = + workspace.send(Command::Document(document)).expect("accept the editor's applied rename"); + ready(workspace, events, sequence).await; + + // An editor may still hold a completion from before the rename. Resolving its old token + // returns the item without stale type information, rather than looking up recycled IDs. + let (reply, request) = Reply::channel(); + let command = LanguageServer::ResolveCompletion { item, reply }; + workspace + .send(Command::LanguageServer(command)) + .expect("admit resolution of an outdated completion"); + let outdated = request.await.expect("handle the outdated completion token"); + assert!(outdated.data.is_none()); + assert!(outdated.detail.is_none()); + println!(" Outdated completion token discarded after didChange."); +} + +async fn shutdown(workspace: &Workspace, events: &mut EventReceiver) { + println!("\n4. Shutdown"); + workspace.send(Command::Shutdown).expect("admit workspace shutdown"); + while let Some(delivery) = events.recv().await { + if matches!(delivery.release(), Ok(Event::StatusChanged(Status::Stopped))) { + break; + } + } + assert_eq!(workspace.status(), Status::Stopped); + println!(" Workspace stopped."); +} + +async fn ready(workspace: &Workspace, events: &mut EventReceiver, sequence: InputSequence) { + loop { + match workspace.status() { + Status::Ready { sequence: applied, .. } if applied >= sequence => return, + Status::Failed { message, .. } => panic!("preparation failed: {message}"), + _ => {} + } + + // These scripted scenarios drain events at milestones. A language server continuously + // consumes the stream, including diagnostics and progress, in its output loop. + let delivery = events.recv().await.expect("workspace stopped before becoming ready"); + if let Ok(Event::InputRejected { failure, .. }) = delivery.release() { + panic!("invalid example input: {failure}"); + } + } +} + +async fn publish_next( + events: &mut EventReceiver, + expected_uri: &Url, + expected_version: Option, +) -> PublishDiagnosticsParams { + loop { + let delivery = events.recv().await.expect("receive a workspace event before shutdown"); + + // Check in the service loop immediately before handing diagnostics to the transport. + // Forward the guarded delivery to that loop, awaiting acknowledgement before receiving + // another event. Later transport queueing does not require another freshness check. + let Ok(event) = delivery.release() else { + continue; + }; + + match event { + Event::Diagnostics { uri, version, diagnostics } => { + let publication = PublishDiagnosticsParams { uri, version, diagnostics }; + let name = publication + .uri + .path_segments() + .and_then(|mut segments| segments.next_back()) + .expect("diagnostics must refer to a source file"); + let version = publication + .version + .map_or("unversioned".into(), |version| format!("version {version}")); + println!(" {name} ({version}) — diagnostics: {}", publication.diagnostics.len()); + for diagnostic in &publication.diagnostics { + println!(" {}", diagnostic.message); + } + + // The output above summarizes textDocument/publishDiagnostics. Only the scripted + // editor waits for a particular version; every valid publication is displayed. + if publication.uri == *expected_uri && publication.version == expected_version { + return publication; + } + } + Event::InputRejected { failure, .. } => panic!("editor sent invalid input: {failure}"), + Event::StatusChanged(Status::Failed { message, .. }) => { + panic!("preparation failed: {message}") + } + _ => {} + } + } +} + +fn print_message(label: &str, message: serde_json::Value) { + println!("\n {label}"); + let message = serde_json::to_string_pretty(&message).expect("format the protocol message"); + for line in message.lines() { + println!(" {line}"); + } +} diff --git a/compiler-lsp/iris-workspace/src/controller.rs b/compiler-lsp/iris-workspace/src/controller.rs new file mode 100644 index 000000000..56be28532 --- /dev/null +++ b/compiler-lsp/iris-workspace/src/controller.rs @@ -0,0 +1,524 @@ +use std::collections::{BTreeSet, VecDeque}; +use std::sync::{Arc, mpsc}; +use std::thread; + +use iris_build::events::BuildEvent; +use lsp_types::Url; + +use crate::documents::Documents; +use crate::events::EventSender; +use crate::transport::{Fence, Shared, WorkerState}; +use crate::worker::{self, Completed, Work}; +use crate::{ + AnalysisStamp, Cancellation, Command, ConfigurationInput, ConfigurationOutcome, Delivery, + Document, Event, Generation, Incarnation, InputSequence, LanguageServer, Options, Outcome, + Phase, RequestFailure, Status, +}; + +const MESSAGE_BATCH_SIZE: usize = 64; + +pub(crate) enum Message { + Command { + sequence: InputSequence, + revision: InputSequence, + generation: Generation, + command: Command, + }, + Completed(Completed), + Progress { + generation: Generation, + event: BuildEvent, + }, +} + +// Desired lifecycle is independent of outstanding work: a superseded query may still be running. +#[derive(Clone, Copy)] +enum Lifecycle { + AwaitingConfiguration, + PreparationPending, + Preparing, + CatchingUp { stamp: AnalysisStamp }, + Active { stamp: AnalysisStamp }, + Failed, + Stopping, +} + +impl Lifecycle { + fn has_attempt(self) -> bool { + matches!(self, Lifecycle::Preparing | Lifecycle::CatchingUp { .. }) + } + + fn accepts_completion(self) -> bool { + matches!( + self, + Lifecycle::Preparing | Lifecycle::CatchingUp { .. } | Lifecycle::Active { .. } + ) + } +} + +pub(crate) struct Controller { + options: Options, + shared: Shared, + receiver: mpsc::Receiver, + events: EventSender, + worker: mpsc::Sender, + worker_thread: Option>, + hooks: crate::testing::Hooks, + documents: Documents, + configuration: Option, + pending_configuration: Option, + generation: Generation, + sequence: InputSequence, + revision: InputSequence, + incarnation: Incarnation, + lifecycle: Lifecycle, + dirty: BTreeSet, + requests: VecDeque, + diagnostics: BTreeSet, + sources: BTreeSet, + request_diagnostics: bool, +} + +impl Controller { + pub(crate) fn new( + options: Options, + shared: Shared, + sender: mpsc::Sender, + receiver: mpsc::Receiver, + events: EventSender, + hooks: crate::testing::Hooks, + ) -> std::io::Result { + let (worker, work) = mpsc::channel(); + let worker_hooks = crate::testing::Hooks::clone(&hooks); + let worker_thread = thread::Builder::new() + .name("iris-compilation".into()) + .spawn(move || worker::run(work, sender, options, worker_hooks))?; + Ok(Controller { + options, + shared, + receiver, + events, + worker, + worker_thread: Some(worker_thread), + hooks, + documents: Documents::default(), + configuration: None, + pending_configuration: None, + generation: Generation::default(), + sequence: InputSequence::default(), + revision: InputSequence::default(), + incarnation: Incarnation::default(), + lifecycle: Lifecycle::AwaitingConfiguration, + dirty: BTreeSet::new(), + requests: VecDeque::new(), + diagnostics: BTreeSet::new(), + sources: BTreeSet::new(), + request_diagnostics: false, + }) + } + + pub(crate) fn run(mut self) { + self.emit(Event::StatusChanged(Status::AwaitingConfiguration), Fence::Status(0)); + while let Ok(message) = self.receiver.recv() { + if !self.handle(message) { + break; + } + // Coalesce queued inputs without indefinitely postponing scheduling. The first + // message already counts toward the batch; its size is only a throughput choice. + for _ in 1..MESSAGE_BATCH_SIZE { + if let Ok(message) = self.receiver.try_recv() { + if !self.handle(message) { + return; + } + } else { + break; + } + } + self.schedule(); + } + if let Some(worker) = self.worker_thread.take() { + let _ = worker.join(); + } + } + + fn emit(&self, event: Event, fence: Fence) { + self.events.send(Delivery::new(event, Arc::clone(&self.shared), fence)); + } + + fn status(&self, status: Status) { + let mut admission = self.shared.lock(); + if admission.generation != self.generation && !matches!(status, Status::Stopped) { + return; + } + if matches!(admission.status, Status::Stopping | Status::Stopped) + && !matches!(status, Status::Stopping | Status::Stopped) + { + return; + } + admission.status = Status::clone(&status); + admission.status_revision += 1; + let revision = admission.status_revision; + drop(admission); + self.emit(Event::StatusChanged(status), Fence::Status(revision)); + } + + fn finish_attempt(&mut self, outcome: Outcome) { + if self.lifecycle.has_attempt() { + let generation = self.generation; + self.emit(Event::Finished { generation, outcome }, Fence::Unconditional); + } + } + + fn finish_configuration(&mut self, outcome: ConfigurationOutcome) { + if let Some(sequence) = self.pending_configuration.take() { + self.emit(Event::ConfigurationFinished { sequence, outcome }, Fence::Unconditional); + } + } + + fn clear_diagnostics(&mut self) { + self.diagnostics.clear(); + let mut admission = self.shared.lock(); + let clears = admission.publications.iter_mut().map(|(uri, revision)| { + *revision += 1; + let event = + Event::Diagnostics { uri: Url::clone(uri), version: None, diagnostics: vec![] }; + let fence = + Fence::Publication { uri: Url::clone(uri), revision: *revision, analysis: None }; + Delivery::new(event, Arc::clone(&self.shared), fence) + }); + + let clears = clears.collect::>(); + drop(admission); + + for clear in clears { + self.events.send(clear); + } + } + + fn reject_requests(&mut self, failure: RequestFailure) { + for mut request in self.requests.drain(..) { + request.reject(RequestFailure::clone(&failure)); + } + } + + fn clear_source_diagnostics(&self, uri: Url) { + let mut admission = self.shared.lock(); + let revision = admission.publications.entry(Url::clone(&uri)).or_default(); + *revision += 1; + let fence = + Fence::Publication { uri: Url::clone(&uri), revision: *revision, analysis: None }; + drop(admission); + self.emit(Event::Diagnostics { uri, version: None, diagnostics: vec![] }, fence); + } + + fn handle(&mut self, message: Message) -> bool { + match message { + Message::Command { sequence, revision, generation, command } => { + if revision != self.revision { + self.revision = revision; + self.diagnostics.clear(); + self.request_diagnostics = false; + self.reject_requests(RequestFailure::Stale); + } + match command { + Command::LanguageServer(mut request) => match self.lifecycle { + Lifecycle::Active { .. } => self.requests.push_back(request), + Lifecycle::Stopping => request.reject(RequestFailure::Cancelled), + _ => request.reject(RequestFailure::Unavailable), + }, + Command::Document(document) => { + self.sequence = sequence; + let triggers = self + .configuration + .as_ref() + .map(|configuration| &configuration.settings.diagnostics); + let collect = match (&document, triggers) { + (Document::Open { .. }, Some(triggers)) => triggers.on_open, + (Document::Change { .. }, Some(triggers)) => triggers.on_change, + (Document::Save(_), Some(triggers)) => triggers.on_save, + (Document::Close(_), _) => true, + _ => false, + }; + match self.documents.apply(document, self.options.position_encoding) { + Ok(uri) => { + self.dirty.insert(uri); + self.request_diagnostics |= collect; + } + Err(failure) => self.emit( + Event::InputRejected { sequence, failure }, + Fence::Unconditional, + ), + } + } + Command::Configure(configuration) => { + let unchanged = self.configuration.as_ref().is_some_and(|previous| { + previous.settings == configuration.settings + && previous.root == configuration.root + }); + self.configuration = Some(configuration); + if generation != self.generation { + self.rebuild(sequence, generation); + self.pending_configuration = Some(sequence); + } else { + self.sequence = sequence; + let outcome = match self.shared.lock().status { + Status::Failed { ref message, .. } => { + ConfigurationOutcome::Failed { message: Arc::clone(message) } + } + _ if unchanged => ConfigurationOutcome::Unchanged, + _ => ConfigurationOutcome::PolicyUpdated, + }; + self.emit( + Event::ConfigurationFinished { sequence, outcome }, + Fence::Unconditional, + ); + } + } + Command::FilesChanged(uris) if generation == self.generation => { + self.sequence = sequence; + self.dirty.extend(uris); + self.request_diagnostics = true; + } + Command::Reload | Command::FilesChanged(_) => { + self.rebuild(sequence, generation) + } + Command::Shutdown => { + self.sequence = sequence; + self.reject_requests(RequestFailure::Cancelled); + self.finish_attempt(Outcome::Cancelled); + self.finish_configuration(ConfigurationOutcome::Cancelled); + self.lifecycle = Lifecycle::Stopping; + self.clear_diagnostics(); + self.status(Status::Stopping); + } + } + } + Message::Progress { generation, event } => { + if generation == self.generation { + let phase = match event { + BuildEvent::Preparing => Phase::Discovering, + BuildEvent::PlanReady { .. } | BuildEvent::PackageCompleted { .. } => { + Phase::Building + } + BuildEvent::Finalizing { .. } | BuildEvent::Finished { .. } => { + Phase::Reconciling + } + }; + self.status(Status::Rebuilding { generation, phase }); + self.emit(Event::Progress { generation, phase }, Fence::Generation(generation)); + } + } + Message::Completed(completed) => { + { + let mut admission = self.shared.lock(); + admission.worker = WorkerState::Idle; + } + match completed { + Completed::Reconciled { generation, stamp, sources } => { + if generation == self.generation && self.lifecycle.accepts_completion() { + let rebuilding = self.lifecycle.has_attempt(); + self.lifecycle = if rebuilding { + Lifecycle::CatchingUp { stamp } + } else { + Lifecycle::Active { stamp } + }; + let sources = sources.into_iter().collect::>(); + let removed = + self.sources.difference(&sources).cloned().collect::>(); + for uri in removed { + self.clear_source_diagnostics(uri); + } + self.sources = sources; + if stamp.revision == self.revision + && (self.request_diagnostics || rebuilding) + { + self.diagnostics.extend(self.sources.iter().cloned()); + self.request_diagnostics = false; + } + } + } + Completed::Analyzed => {} + Completed::Diagnostics { uri, stamp, version, result } => { + if let Err(failure) = &result { + if !matches!(failure, RequestFailure::Cancelled | RequestFailure::Stale) + { + self.emit( + Event::DiagnosticsFailed { + uri: Url::clone(&uri), + message: failure.to_string().into(), + }, + Fence::Unconditional, + ); + } + } + let mut admission = self.shared.lock(); + let current = admission.revision == stamp.revision + && matches!(admission.status, Status::Ready { stamp: current, .. } if current == stamp); + if current { + if let Ok(diagnostics) = result { + let revision = + admission.publications.entry(Url::clone(&uri)).or_default(); + *revision += 1; + let revision = *revision; + drop(admission); + self.emit( + Event::Diagnostics { + uri: Url::clone(&uri), + version, + diagnostics, + }, + Fence::Publication { uri, revision, analysis: Some(stamp) }, + ); + } + } + } + Completed::Failed { generation, failure } => { + self.hooks.reach(crate::testing::Point::BeforeFailure); + if generation == self.generation && self.lifecycle.accepts_completion() { + self.clear_diagnostics(); + self.reject_requests(RequestFailure::Unavailable); + self.finish_attempt(Outcome::Failed); + self.finish_configuration(ConfigurationOutcome::Failed { + message: failure.to_string().into(), + }); + self.lifecycle = Lifecycle::Failed; + self.status(Status::Failed { + generation, + message: failure.to_string().into(), + }); + } + } + Completed::Stopped => { + self.status(Status::Stopped); + if let Some(worker) = self.worker_thread.take() { + let _ = worker.join(); + } + return false; + } + } + } + } + true + } + + fn rebuild(&mut self, sequence: InputSequence, generation: Generation) { + self.finish_attempt(Outcome::Superseded); + self.finish_configuration(ConfigurationOutcome::Superseded); + self.generation = generation; + self.sequence = sequence; + self.lifecycle = Lifecycle::PreparationPending; + self.reject_requests(RequestFailure::Unavailable); + self.clear_diagnostics(); + self.status(Status::Rebuilding { generation, phase: Phase::Retiring }); + } + + fn schedule(&mut self) { + let mut admission = self.shared.lock(); + if !matches!(admission.worker, WorkerState::Idle) { + return; + } + if matches!(self.lifecycle, Lifecycle::Stopping) { + admission.worker = WorkerState::Stopping; + let _ = self.worker.send(Work::Stop); + return; + } + if admission.sequence != self.sequence || admission.generation != self.generation { + return; + } + if matches!(self.lifecycle, Lifecycle::PreparationPending) { + let Some(configuration) = self.configuration.clone() else { + self.lifecycle = Lifecycle::AwaitingConfiguration; + drop(admission); + self.status(Status::AwaitingConfiguration); + return; + }; + self.incarnation.advance(); + let stamp = AnalysisStamp { incarnation: self.incarnation, revision: self.revision }; + let cancellation = Cancellation::default(); + admission.worker = + WorkerState::Preparing { cancellation: Cancellation::clone(&cancellation) }; + self.lifecycle = Lifecycle::Preparing; + self.dirty.clear(); + let work = Work::Prepare { + configuration, + documents: self.documents.open.clone(), + generation: self.generation, + stamp, + cancellation, + }; + let _ = self.worker.send(work); + drop(admission); + self.status(Status::Rebuilding { + generation: self.generation, + phase: Phase::Discovering, + }); + return; + } + let (Lifecycle::CatchingUp { stamp } | Lifecycle::Active { stamp }) = self.lifecycle else { + return; + }; + if stamp.revision != self.revision { + let stamp = AnalysisStamp { revision: self.revision, ..stamp }; + let work = Work::Reconcile { + documents: self.documents.open.clone(), + dirty: std::mem::take(&mut self.dirty), + stamp, + }; + admission.worker = WorkerState::Reconciling; + let _ = self.worker.send(work); + drop(admission); + if self.lifecycle.has_attempt() { + self.status(Status::Rebuilding { + generation: self.generation, + phase: Phase::Reconciling, + }); + } + return; + } + let status = Status::Ready { generation: self.generation, sequence: self.sequence, stamp }; + if admission.status != status { + admission.status = Status::clone(&status); + admission.status_revision += 1; + let revision = admission.status_revision; + self.emit(Event::StatusChanged(status), Fence::Status(revision)); + } + drop(admission); + self.finish_attempt(Outcome::Ready); + self.finish_configuration(ConfigurationOutcome::Rebuilt); + self.lifecycle = Lifecycle::Active { stamp }; + while let Some(mut request) = self.requests.pop_front() { + if request.cancellation().is_cancelled() { + request.reject(RequestFailure::Cancelled); + continue; + } + if request.stamp() != stamp { + request.reject(RequestFailure::Stale); + continue; + } + let mut admission = self.shared.lock(); + if admission.revision != stamp.revision || admission.generation != self.generation { + drop(admission); + request.reject(RequestFailure::Stale); + continue; + } + admission.worker = WorkerState::Querying { cancellation: request.cancellation() }; + let result = self.worker.send(Work::Analyze(request)); + drop(admission); + drop(result); + return; + } + if let Some(uri) = self.diagnostics.pop_first() { + let mut admission = self.shared.lock(); + if admission.revision != stamp.revision || admission.generation != self.generation { + self.diagnostics.insert(uri); + return; + } + let cancellation = Cancellation::default(); + admission.worker = + WorkerState::Querying { cancellation: Cancellation::clone(&cancellation) }; + let _ = self.worker.send(Work::Diagnostics { uri, stamp, cancellation }); + return; + } + self.hooks.reach(crate::testing::Point::Idle); + } +} diff --git a/compiler-lsp/iris-workspace/src/documents.rs b/compiler-lsp/iris-workspace/src/documents.rs new file mode 100644 index 000000000..fa82de490 --- /dev/null +++ b/compiler-lsp/iris-workspace/src/documents.rs @@ -0,0 +1,92 @@ +use std::collections::BTreeMap; +use std::sync::Arc; + +use analyzer::position::{PositionConverter, PositionEncoding}; +use lsp_types::Url; + +use crate::{Document, InputFailure}; + +#[derive(Clone, Debug)] +pub(crate) struct OpenDocument { + pub(crate) text: Arc, + pub(crate) version: i32, + pub(crate) lifetime: u64, +} + +#[derive(Default)] +pub(crate) struct Documents { + pub(crate) open: BTreeMap, + next_lifetime: u64, +} + +pub(crate) fn document_path(uri: &Url) -> Result { + iris_build::analysis::document_path(uri) + .ok_or_else(|| InputFailure::UnsupportedDocument(Url::clone(uri))) +} + +impl Documents { + pub(crate) fn apply( + &mut self, + command: Document, + encoding: PositionEncoding, + ) -> Result { + match command { + Document::Open { uri, text, version } => { + document_path(&uri)?; + if self.open.contains_key(&uri) { + return Err(InputFailure::AlreadyOpen(uri)); + } + self.next_lifetime = + self.next_lifetime.checked_add(1).expect("document lifetime overflow"); + self.open.insert( + Url::clone(&uri), + OpenDocument { text, version, lifetime: self.next_lifetime }, + ); + Ok(uri) + } + Document::Change { uri, version, changes } => { + document_path(&uri)?; + let document = self + .open + .get_mut(&uri) + .ok_or_else(|| InputFailure::NotOpen(Url::clone(&uri)))?; + if version <= document.version { + return Err(InputFailure::StaleVersion(uri)); + } + let mut text = document.text.to_string(); + for change in changes { + match change.range { + None => text = change.text, + Some(range) => { + let positions = PositionConverter::new(&text, encoding); + let offset = |position| { + let position = positions.protocol_position_to_utf8(position)?; + positions.utf8_position_to_offset(position).map(usize::from) + }; + let start = offset(range.start) + .ok_or_else(|| InputFailure::InvalidRange(Url::clone(&uri)))?; + let end = offset(range.end) + .ok_or_else(|| InputFailure::InvalidRange(Url::clone(&uri)))?; + if start > end { + return Err(InputFailure::InvalidRange(uri)); + } + text.replace_range(start..end, &change.text); + } + } + } + document.text = text.into(); + document.version = version; + Ok(uri) + } + Document::Close(uri) => { + document_path(&uri)?; + self.open.remove(&uri).ok_or_else(|| InputFailure::NotOpen(Url::clone(&uri)))?; + Ok(uri) + } + Document::Save(uri) => { + document_path(&uri)?; + Ok(uri) + } + } + } +} diff --git a/compiler-lsp/iris-workspace/src/events.rs b/compiler-lsp/iris-workspace/src/events.rs new file mode 100644 index 000000000..372eaf626 --- /dev/null +++ b/compiler-lsp/iris-workspace/src/events.rs @@ -0,0 +1,69 @@ +use std::collections::VecDeque; +use std::sync::Arc; + +use parking_lot::Mutex; +use tokio::sync::mpsc; + +use crate::{Delivery, Event}; + +/// One consumer of workspace publications. Status, progress and per-URI diagnostics coalesce; +/// terminal outcomes and rejected inputs remain ordered and must be drained by the consumer. +/// An adapter may hold one delivery until its service loop acknowledges handoff or discard, +/// then receive the next. Do not eagerly release into an unbounded forwarding queue. Connection +/// teardown must drop or acknowledge the held item so the pump can stop. +pub struct EventReceiver { + pending: Arc>>>, + wake: mpsc::Receiver<()>, +} + +pub(crate) struct EventSender { + pending: Arc>>>, + wake: mpsc::Sender<()>, +} + +impl EventSender { + pub(crate) fn channel() -> (EventSender, EventReceiver) { + let pending = Arc::new(Mutex::new(VecDeque::new())); + let (sender, receiver) = mpsc::channel(1); + ( + EventSender { pending: Arc::clone(&pending), wake: sender }, + EventReceiver { pending, wake: receiver }, + ) + } + + pub(crate) fn send(&self, delivery: Delivery) { + if self.wake.is_closed() { + return; + } + let mut pending = self.pending.lock(); + pending.retain(|previous| !replaces(&delivery.value, &previous.value)); + pending.push_back(delivery); + drop(pending); + let _ = self.wake.try_send(()); + } +} + +impl EventReceiver { + pub async fn recv(&mut self) -> Option> { + loop { + if let Some(event) = self.pending.lock().pop_front() { + return Some(event); + } + self.wake.recv().await?; + } + } +} + +fn replaces(next: &Event, previous: &Event) -> bool { + match (next, previous) { + (Event::StatusChanged(_), Event::StatusChanged(_)) => true, + (Event::Diagnostics { uri: next, .. }, Event::Diagnostics { uri: previous, .. }) => { + next == previous + } + ( + Event::Progress { generation: next, .. }, + Event::Progress { generation: previous, .. }, + ) => next == previous, + _ => false, + } +} diff --git a/compiler-lsp/iris-workspace/src/language_server.rs b/compiler-lsp/iris-workspace/src/language_server.rs new file mode 100644 index 000000000..fde9a68c1 --- /dev/null +++ b/compiler-lsp/iris-workspace/src/language_server.rs @@ -0,0 +1,287 @@ +use std::collections::BTreeMap; + +use analyzer::completion::SuggestionsCache; +use analyzer::symbols::WorkspaceSymbolsCache; +use analyzer::{AnalyzerContext, AnalyzerError, AnalyzerHost}; +use building::{FileLifecycle, QueryCancellation, QueryEngine, QueryError}; +use files::FileId; +use lsp_types::*; + +use crate::transport::Shared; +use crate::{AnalysisStamp, Cancellation, Options, Reply, RequestFailure}; + +#[derive(Clone, Debug, thiserror::Error)] +pub enum LanguageServerFailure { + #[error("rename rejected: {0}")] + RenameRejected(String), + #[error("analysis failed: {0}")] + Analysis(String), +} + +macro_rules! language_requests { + ($($name:ident { $($field:ident: $input:ty),* } => $output:ty),* $(,)?) => { + /// Reference queries return the analyzer's reference set without declaration-inclusion filtering. + pub enum LanguageServer { + $($name { $($field: $input,)* reply: Reply<$output> }),* + } + + impl LanguageServer { + pub(crate) fn set_hooks(&mut self, hooks: crate::testing::Hooks) { + match self { $(LanguageServer::$name { reply, .. } => reply.hooks = hooks),* } + } + + pub(crate) fn admit(&mut self, shared: Shared, stamp: AnalysisStamp) { + match self { $(LanguageServer::$name { reply, .. } => reply.admit(shared, stamp)),* } + } + + pub(crate) fn reject(&mut self, failure: RequestFailure) { + match self { $(LanguageServer::$name { reply, .. } => reply.reject(failure)),* } + } + + pub(crate) fn cancellation(&self) -> Cancellation { + match self { $(LanguageServer::$name { reply, .. } => Cancellation::clone(&reply.cancellation)),* } + } + + pub(crate) fn stamp(&self) -> AnalysisStamp { + match self { $(LanguageServer::$name { reply, .. } => reply.stamp()),* } + } + } + }; +} + +language_requests! { + Hover { uri: Url, position: Position } => Option, + Definition { uri: Url, position: Position } => Option, + References { uri: Url, position: Position } => Option>, + Completion { uri: Url, position: Position } => Option, + ResolveCompletion { item: CompletionItem } => CompletionItem, + Rename { uri: Url, position: Position, new_name: String } => Option, + PrepareRename { uri: Url, position: Position } => Option, + DocumentHighlight { uri: Url, position: Position } => Option>, + DocumentSymbols { uri: Url } => Option, + WorkspaceSymbols { query: String } => Option, + SemanticTokens { uri: Url } => Option, + CodeAction { uri: Url, range: Range, context: CodeActionContext } => Option, +} + +impl LanguageServer { + pub(crate) fn validate(&self) -> Result<(), crate::InputFailure> { + let uri = match self { + LanguageServer::Hover { uri, .. } + | LanguageServer::Definition { uri, .. } + | LanguageServer::References { uri, .. } + | LanguageServer::Completion { uri, .. } + | LanguageServer::Rename { uri, .. } + | LanguageServer::PrepareRename { uri, .. } + | LanguageServer::DocumentHighlight { uri, .. } + | LanguageServer::DocumentSymbols { uri, .. } + | LanguageServer::SemanticTokens { uri, .. } + | LanguageServer::CodeAction { uri, .. } => uri, + LanguageServer::ResolveCompletion { .. } | LanguageServer::WorkspaceSymbols { .. } => { + return Ok(()); + } + }; + crate::documents::document_path(uri).map(|_| ()) + } +} + +pub(crate) struct Host<'a> { + pub(crate) engine: &'a QueryEngine, + pub(crate) files: &'a FileLifecycle, +} + +impl AnalyzerHost for Host<'_> { + type Queries = QueryEngine; + + fn queries(&self) -> &QueryEngine { + self.engine + } + + fn file_id(&self, uri: &str) -> Option { + self.files.source_id(uri) + } + + fn file_uri(&self, file_id: FileId) -> Result, url::ParseError> { + self.files.source_path(file_id).map(|uri| Url::parse(&uri)).transpose() + } + + fn active_files(&self) -> impl Iterator { + self.files.source_ids() + } + + fn is_editable(&self, file_id: FileId) -> bool { + self.files.source_metadata(file_id).copied().unwrap_or(false) + } +} + +pub(crate) fn failure(error: AnalyzerError) -> RequestFailure { + match error { + AnalyzerError::QueryError(QueryError::Cancelled) => RequestFailure::Cancelled, + AnalyzerError::RenameRejected(message) => { + LanguageServerFailure::RenameRejected(message).into() + } + error => LanguageServerFailure::Analysis(error.to_string()).into(), + } +} + +fn optional(result: Result, AnalyzerError>) -> Result, RequestFailure> { + match result { + Err(AnalyzerError::NonFatal) => Ok(None), + result => result.map_err(failure), + } +} + +#[derive(Default)] +pub(crate) struct Analysis { + suggestions: SuggestionsCache, + symbols: WorkspaceSymbolsCache, + resolve: BTreeMap, + next_resolve: u64, + session: String, +} + +impl Analysis { + pub(crate) fn new(session: String) -> Analysis { + Analysis { session, ..Analysis::default() } + } + + pub(crate) fn invalidate(&mut self) { + self.suggestions = SuggestionsCache::default(); + self.symbols = WorkspaceSymbolsCache::default(); + self.resolve.clear(); + } + + fn protect_completions(&mut self, response: &mut CompletionResponse) { + let items = match response { + CompletionResponse::Array(items) => items, + CompletionResponse::List(list) => &mut list.items, + }; + for item in items { + if let Some(data) = item.data.take() { + self.next_resolve = + self.next_resolve.checked_add(1).expect("completion token overflow"); + let token = format!("{}:{}", self.session, self.next_resolve); + if self.resolve.len() >= 4096 { + self.resolve.pop_first(); + } + self.resolve.insert(String::clone(&token), data); + item.data = Some(serde_json::Value::String(token)); + } + } + } + + pub(crate) fn execute( + &mut self, + mut command: LanguageServer, + engine: &QueryEngine, + files: &FileLifecycle, + options: Options, + hooks: &crate::testing::Hooks, + ) { + command.set_hooks(crate::testing::Hooks::clone(hooks)); + let cancellation = command.cancellation(); + let snapshot = + engine.snapshot_with_cancellation(QueryCancellation::clone(&cancellation.query)); + hooks.reach(crate::testing::Point::SnapshotActive); + let host = Host { engine: &snapshot, files }; + let context = AnalyzerContext::new(&host, options.position_encoding, options.capabilities); + match command { + LanguageServer::Hover { uri, position, reply } => { + reply.finish(optional(analyzer::hover::implementation(&context, uri, position))); + } + LanguageServer::Definition { uri, position, reply } => { + reply.finish(optional(analyzer::definition::implementation( + &context, uri, position, + ))); + } + LanguageServer::References { uri, position, reply } => { + reply.finish(optional(analyzer::references::implementation( + &context, uri, position, + ))); + } + LanguageServer::Completion { uri, position, reply } => { + let result = optional(analyzer::completion::implementation( + &context, + &mut self.suggestions, + uri, + position, + )); + reply.finish(result.map(|response| { + response.map(|mut response| { + self.protect_completions(&mut response); + response + }) + })); + } + LanguageServer::ResolveCompletion { mut item, reply } => { + let target = item.data.take().and_then(|data| { + data.as_str().and_then(|token| self.resolve.get(token)).cloned() + }); + item.data = target; + let result = analyzer::completion::resolve::implementation(&snapshot, item); + reply.finish(match result { + Ok(item) | Err((AnalyzerError::NonFatal, item)) => Ok(item), + Err((error, _)) => Err(failure(error)), + }); + } + LanguageServer::Rename { uri, position, new_name, reply } => { + let result = + optional(analyzer::rename::implementation(&context, uri, position, new_name)); + reply.finish(result.map(|edit| { + edit.map(|mut edit| { + version_edits(&mut edit, files); + edit + }) + })); + } + LanguageServer::PrepareRename { uri, position, reply } => { + reply.finish(optional(analyzer::rename::prepare(&context, uri, position))); + } + LanguageServer::DocumentHighlight { uri, position, reply } => { + reply.finish(optional(analyzer::document_highlight::implementation( + &context, uri, position, + ))); + } + LanguageServer::DocumentSymbols { uri, reply } => { + reply.finish(optional(analyzer::symbols::document(&context, uri))); + } + LanguageServer::WorkspaceSymbols { query, reply } => { + reply.finish(optional(analyzer::symbols::workspace( + &context, + &mut self.symbols, + &query, + ))); + } + LanguageServer::SemanticTokens { uri, reply } => { + reply.finish(optional(analyzer::semantic_tokens::implementation(&context, uri))); + } + LanguageServer::CodeAction { uri, range, context: action_context, reply } => { + reply.finish(optional(analyzer::code_action::implementation( + &context, + uri, + range, + action_context, + ))); + } + } + } +} + +fn version_edits(edit: &mut WorkspaceEdit, files: &FileLifecycle) { + let update = |edit: &mut TextDocumentEdit| { + edit.text_document.version = files + .source_id(edit.text_document.uri.as_str()) + .and_then(|file_id| files.source_version(file_id)); + }; + match &mut edit.document_changes { + Some(DocumentChanges::Edits(edits)) => edits.iter_mut().for_each(update), + Some(DocumentChanges::Operations(operations)) => { + for operation in operations { + if let DocumentChangeOperation::Edit(edit) = operation { + update(edit); + } + } + } + None => {} + } +} diff --git a/compiler-lsp/iris-workspace/src/lib.rs b/compiler-lsp/iris-workspace/src/lib.rs new file mode 100644 index 000000000..45b5ce892 --- /dev/null +++ b/compiler-lsp/iris-workspace/src/lib.rs @@ -0,0 +1,253 @@ +//! A single-consumer workspace service, independent of protocol transports. +//! +//! Inputs are admitted synchronously through [`Workspace::send`]. Compiler work runs on a +//! separate worker; neither input admission nor cancellation waits for compiler snapshots. +//! Interactive replies settle once; background publications are checked at service-loop handoff. +//! Configuration replacement discards compilation state but preserves open documents. + +mod controller; +mod documents; +mod events; +mod language_server; +mod transport; +mod worker; + +#[cfg(feature = "test-support")] +pub mod testing; +#[cfg(not(feature = "test-support"))] +mod testing; + +pub use analyzer::AnalyzerCapabilities; +pub use analyzer::position::PositionEncoding; +pub use configuration::{Configuration, SourceDiscovery}; +pub use events::EventReceiver; +pub use language_server::{LanguageServer, LanguageServerFailure}; +pub use transport::{ + Cancellation, Delivery, Reply, Request, Workspace, WorkspaceJoin, WorkspaceSession, +}; + +use std::path::PathBuf; +use std::sync::Arc; + +use lsp_types::{Diagnostic, SemanticTokensLegend, TextDocumentContentChangeEvent, Url}; + +/// Token indices in analysis responses refer to these analyzer-owned tables. +pub fn semantic_tokens_legend() -> SemanticTokensLegend { + SemanticTokensLegend { + token_types: analyzer::semantic_tokens::TOKEN_TYPES.to_vec(), + token_modifiers: analyzer::semantic_tokens::TOKEN_MODIFIERS.to_vec(), + } +} + +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub struct Generation { + pub value: u64, +} + +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub struct Incarnation { + pub value: u64, +} + +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Ord, PartialOrd)] +pub struct InputSequence { + pub value: u64, +} + +impl Generation { + pub(crate) fn advance(&mut self) { + self.value = self.value.checked_add(1).expect("configuration generation overflow"); + } +} + +impl Incarnation { + pub(crate) fn advance(&mut self) { + self.value = self.value.checked_add(1).expect("engine incarnation overflow"); + } +} + +impl InputSequence { + pub(crate) fn advance(&mut self) { + self.value = self.value.checked_add(1).expect("input sequence overflow"); + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct AnalysisStamp { + pub incarnation: Incarnation, + /// Sequence of the last potential analysis write, excluding configuration policy updates. + pub revision: InputSequence, +} + +#[derive(Clone, Debug)] +pub struct ConfigurationInput { + /// Discovery commands run in this directory, with arguments passed without shell evaluation. + pub root: PathBuf, + pub settings: Configuration, +} + +pub enum Command { + /// Ready workspaces retain their compiler when root and source discovery are unchanged. + /// Diagnostic triggers affect future inputs; pending diagnostics finish against current inputs. + /// Other states start a new preparation attempt, including retries of a failed configuration. + /// Every accepted command produces a retained `Event::ConfigurationFinished`. + Configure(ConfigurationInput), + /// Rediscover and reload all sources, then schedule diagnostics for editable sources. + Reload, + Document(Document), + /// Foreign-only changes reconcile locally. Other changes conservatively rediscover all sources. + FilesChanged(Vec), + LanguageServer(LanguageServer), + Shutdown, +} + +impl Command { + pub(crate) fn rebuilds(&self) -> bool { + match self { + Command::Configure(_) | Command::Reload => true, + Command::FilesChanged(uris) => { + uris.iter().any(|uri| !uri.path().ends_with(".js") && !uri.path().ends_with(".jsx")) + } + _ => false, + } + } +} + +/// Documents use file URLs that round-trip through their local path unchanged, without a query +/// or fragment. This does not resolve filesystem aliases or require a file to exist. +pub enum Document { + Open { uri: Url, text: Arc, version: i32 }, + Change { uri: Url, version: i32, changes: Vec }, + Save(Url), + Close(Url), +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum Status { + AwaitingConfiguration, + Rebuilding { + generation: Generation, + phase: Phase, + }, + Ready { + generation: Generation, + /// Last reconciled input, including inputs that do not invalidate analysis. + sequence: InputSequence, + stamp: AnalysisStamp, + }, + Failed { + generation: Generation, + message: Arc, + }, + Stopping, + /// The worker has stopped. The controller is joined separately through `WorkspaceJoin`. + Stopped, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum Phase { + Retiring, + Discovering, + Building, + Reconciling, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum Outcome { + Ready, + Failed, + Superseded, + Cancelled, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum ConfigurationOutcome { + Unchanged, + PolicyUpdated, + Rebuilt, + Failed { message: Arc }, + Superseded, + Cancelled, +} + +#[derive(Debug)] +pub enum Event { + StatusChanged(Status), + Diagnostics { + uri: Url, + version: Option, + diagnostics: Vec, + }, + /// Indeterminate progress: the current phase, not a delta or percentage. + Progress { + generation: Generation, + phase: Phase, + }, + DiagnosticsFailed { + uri: Url, + message: Arc, + }, + Finished { + generation: Generation, + outcome: Outcome, + }, + InputRejected { + sequence: InputSequence, + failure: InputFailure, + }, + /// Exactly one retained terminal outcome for each admitted Configure, keyed by its sequence. + ConfigurationFinished { + sequence: InputSequence, + outcome: ConfigurationOutcome, + }, +} + +#[derive(Clone, Debug, thiserror::Error, Eq, PartialEq)] +pub enum InputFailure { + #[error("unsupported document URI: {0}")] + UnsupportedDocument(Url), + #[error("document is not open: {0}")] + NotOpen(Url), + #[error("document is already open: {0}")] + AlreadyOpen(Url), + #[error("document version is not newer: {0}")] + StaleVersion(Url), + #[error("invalid document edit range: {0}")] + InvalidRange(Url), +} + +#[derive(Clone, Debug, thiserror::Error)] +pub enum RequestFailure { + #[error("workspace is unavailable")] + Unavailable, + #[error("workspace request capacity is exhausted")] + Busy, + #[error("analysis has been invalidated")] + Stale, + #[error("request was cancelled")] + Cancelled, + #[error(transparent)] + InvalidInput(#[from] InputFailure), + #[error("workspace failed: {0}")] + Workspace(Arc), + #[error(transparent)] + LanguageServer(#[from] LanguageServerFailure), +} + +#[derive(Clone, Copy, Debug)] +pub struct Options { + pub position_encoding: PositionEncoding, + pub capabilities: AnalyzerCapabilities, + /// Includes executing requests; diagnostics do not consume interactive capacity. + pub request_capacity: usize, +} + +impl Default for Options { + fn default() -> Options { + Options { + position_encoding: PositionEncoding::Utf16, + capabilities: AnalyzerCapabilities::default(), + request_capacity: 32, + } + } +} diff --git a/compiler-lsp/iris-workspace/src/testing.rs b/compiler-lsp/iris-workspace/src/testing.rs new file mode 100644 index 000000000..e61652f8f --- /dev/null +++ b/compiler-lsp/iris-workspace/src/testing.rs @@ -0,0 +1,77 @@ +//! Deterministic execution gates for the command-sequence suite. Not enabled in production. + +#[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd)] +pub enum Point { + BeforePreparation, + BeforeAcknowledgement, + BeforeAnalysis, + SnapshotActive, + BeforeReplySettlement, + AfterReplySettlement, + BeforeDiagnostics, + AfterDiagnostics, + Idle, + BeforeFailure, +} + +#[derive(Clone, Default)] +pub struct Hooks { + #[cfg(feature = "test-support")] + gates: std::sync::Arc>>, +} + +#[cfg(feature = "test-support")] +struct Gate { + entered: tokio::sync::oneshot::Sender<()>, + release: std::sync::mpsc::Receiver<()>, +} + +#[cfg(feature = "test-support")] +pub struct Pause { + entered: Option>, + release: std::sync::mpsc::Sender<()>, +} + +impl Hooks { + #[cfg(feature = "test-support")] + pub fn pause_next(&self, point: Point) -> Pause { + let (entered, receiver) = tokio::sync::oneshot::channel(); + let (release, wait) = std::sync::mpsc::channel(); + assert!( + self.gates.lock().insert(point, Gate { entered, release: wait }).is_none(), + "gate already installed" + ); + Pause { entered: Some(receiver), release } + } + + pub(crate) fn reach(&self, point: Point) { + #[cfg(feature = "test-support")] + { + let gate = self.gates.lock().remove(&point); + if let Some(gate) = gate { + let _ = gate.entered.send(()); + let _ = gate.release.recv(); + } + } + #[cfg(not(feature = "test-support"))] + let _ = point; + } +} + +#[cfg(feature = "test-support")] +impl Pause { + pub async fn entered(&mut self) { + self.entered + .take() + .expect("gate already awaited") + .await + .expect("worker exited before gate"); + } +} + +#[cfg(feature = "test-support")] +impl Drop for Pause { + fn drop(&mut self) { + let _ = self.release.send(()); + } +} diff --git a/compiler-lsp/iris-workspace/src/transport.rs b/compiler-lsp/iris-workspace/src/transport.rs new file mode 100644 index 000000000..c8020b902 --- /dev/null +++ b/compiler-lsp/iris-workspace/src/transport.rs @@ -0,0 +1,446 @@ +use std::collections::BTreeMap; +use std::future::Future; +use std::pin::Pin; +use std::sync::{Arc, mpsc}; +use std::task::{Context, Poll, Waker}; +use std::thread; + +use building::QueryCancellation; +use iris_build::analysis::CancellationToken; +use lsp_types::Url; +use parking_lot::Mutex; +use tokio::sync::oneshot; + +use crate::controller::{Controller, Message}; +use crate::events::EventSender; +use crate::{ + AnalysisStamp, Command, ConfigurationInput, EventReceiver, Generation, InputSequence, Options, + Phase, RequestFailure, Status, +}; + +#[derive(Clone, Default)] +pub struct Cancellation { + pub(crate) query: QueryCancellation, + pub(crate) build: CancellationToken, + terminal: Arc>, +} + +enum Terminal { + Pending { waker: Option }, + Cancelled, + Settled, +} + +impl Default for Terminal { + fn default() -> Terminal { + Terminal::Pending { waker: None } + } +} + +impl Cancellation { + /// Cancel an unfinished request without waiting for its worker to retire. A settled reply + /// is unchanged. Query interruption is cooperative; cancellation does not release snapshots. + pub fn cancel(&self) { + let mut terminal = self.terminal.lock(); + let Terminal::Pending { waker } = &mut *terminal else { return }; + let waker = waker.take(); + *terminal = Terminal::Cancelled; + self.query.cancel(); + self.build.cancel(); + drop(terminal); + if let Some(waker) = waker { + waker.wake(); + } + } + + pub fn is_cancelled(&self) -> bool { + self.query.is_cancelled() + } +} + +// Cancellation does not free the worker slot; only its completion does. +pub(crate) enum WorkerState { + Idle, + Preparing { cancellation: Cancellation }, + Reconciling, + Querying { cancellation: Cancellation }, + Stopping, +} + +pub(crate) struct Admission { + pub(crate) sequence: InputSequence, + pub(crate) revision: InputSequence, + pub(crate) generation: Generation, + pub(crate) status: Status, + pub(crate) requests: usize, + pub(crate) worker: WorkerState, + pub(crate) publications: BTreeMap, + pub(crate) status_revision: u64, + configuration: Option, +} + +impl Default for Admission { + fn default() -> Admission { + Admission { + sequence: InputSequence::default(), + revision: InputSequence::default(), + generation: Generation::default(), + status: Status::AwaitingConfiguration, + requests: 0, + worker: WorkerState::Idle, + publications: BTreeMap::new(), + status_revision: 0, + configuration: None, + } + } +} + +pub(crate) type Shared = Arc>; + +#[derive(Clone)] +pub(crate) enum Fence { + Analysis(AnalysisStamp), + Publication { uri: Url, revision: u64, analysis: Option }, + Status(u64), + Generation(Generation), + Unconditional, +} + +impl Fence { + fn valid(&self, admission: &Admission) -> bool { + match self { + Fence::Analysis(stamp) => { + matches!(admission.status, Status::Ready { stamp: current, .. } if current.incarnation == stamp.incarnation) + && admission.revision == stamp.revision + } + Fence::Publication { uri, revision, analysis } => { + admission.publications.get(uri) == Some(revision) + && analysis.is_none_or(|stamp| Fence::Analysis(stamp).valid(admission)) + } + Fence::Status(revision) => admission.status_revision == *revision, + Fence::Generation(generation) => { + admission.generation == *generation + && !matches!(admission.status, Status::Stopping | Status::Stopped) + } + Fence::Unconditional => true, + } + } +} + +pub struct Delivery { + pub(crate) value: T, + shared: Shared, + fence: Fence, +} + +impl std::fmt::Debug for Delivery { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.debug_struct("Delivery").finish_non_exhaustive() + } +} + +impl Delivery { + pub(crate) fn new(value: T, shared: Shared, fence: Fence) -> Delivery { + Delivery { value, shared, fence } + } + + /// Validate a publication in the service loop immediately before transport handoff. + /// Keep the delivery guarded while forwarding it to that loop. Once released, later inputs + /// do not recall the output; transport queueing and socket flush need no further check. + pub fn release(self) -> Result { + let admission = self.shared.lock(); + if !self.fence.valid(&admission) { + return Err(RequestFailure::Stale); + } + Ok(self.value) + } +} + +struct ReplyAdmission { + shared: Shared, + stamp: AnalysisStamp, +} + +impl Drop for ReplyAdmission { + fn drop(&mut self) { + self.shared.lock().requests -= 1; + } +} + +pub struct Reply { + sender: Option>>, + pub(crate) cancellation: Cancellation, + pub(crate) hooks: crate::testing::Hooks, + admission: Option, +} + +/// A reply settles once. Edits and cancellation do not revoke a settled result, even if unread. +pub struct Request { + receiver: Option>>, + cancellation: Cancellation, + completed: bool, +} + +impl Reply { + pub fn channel() -> (Reply, Request) { + let (sender, receiver) = oneshot::channel(); + let cancellation = Cancellation::default(); + ( + Reply { + sender: Some(sender), + cancellation: Cancellation::clone(&cancellation), + admission: None, + hooks: crate::testing::Hooks::default(), + }, + Request { receiver: Some(receiver), cancellation, completed: false }, + ) + } + + pub(crate) fn admit(&mut self, shared: Shared, stamp: AnalysisStamp) { + self.admission = Some(ReplyAdmission { shared, stamp }); + } + + pub(crate) fn stamp(&self) -> AnalysisStamp { + self.admission.as_ref().expect("analysis reply must be admitted").stamp + } + + pub(crate) fn reject(&mut self, failure: RequestFailure) { + self.settle(Err(failure)); + self.admission = None; + } + + pub(crate) fn finish(mut self, result: Result) { + self.hooks.reach(crate::testing::Point::BeforeReplySettlement); + let shared = + Arc::clone(&self.admission.as_ref().expect("analysis reply must be admitted").shared); + // Input admission takes this same lock, so an edit and reply settlement have one order. + let admission = shared.lock(); + self.settle(result); + drop(admission); + self.hooks.reach(crate::testing::Point::AfterReplySettlement); + } + + fn settle(&mut self, result: Result) { + let mut terminal = self.cancellation.terminal.lock(); + let result = match *terminal { + Terminal::Cancelled => Err(RequestFailure::Cancelled), + Terminal::Pending { .. } => { + *terminal = Terminal::Settled; + result + } + Terminal::Settled => return, + }; + if let Some(sender) = self.sender.take() { + let _ = sender.send(result); + } + } +} + +impl Request { + pub fn cancellation(&self) -> Cancellation { + Cancellation::clone(&self.cancellation) + } +} + +impl Future for Request { + type Output = Result; + + fn poll(mut self: Pin<&mut Request>, context: &mut Context<'_>) -> Poll { + let cancelled = { + let mut terminal = self.cancellation.terminal.lock(); + match &mut *terminal { + Terminal::Pending { waker } => { + *waker = Some(Waker::clone(context.waker())); + false + } + Terminal::Cancelled => true, + Terminal::Settled => false, + } + }; + if cancelled { + self.completed = true; + return Poll::Ready(Err(RequestFailure::Cancelled)); + } + let receiver = self.receiver.as_mut().expect("request polled after completion"); + match Pin::new(receiver).poll(context) { + Poll::Pending => Poll::Pending, + Poll::Ready(result) => { + self.completed = true; + self.receiver = None; + Poll::Ready(result.unwrap_or(Err(RequestFailure::Unavailable))) + } + } + } +} + +impl Drop for Request { + fn drop(&mut self) { + if !self.completed { + self.cancellation.cancel(); + } + } +} + +#[derive(Clone)] +pub struct Workspace { + sender: mpsc::Sender, + shared: Shared, + capacity: usize, +} + +pub struct WorkspaceSession { + pub workspace: Workspace, + pub events: EventReceiver, + pub join: WorkspaceJoin, +} + +/// Owns controller-thread cleanup independently of cloneable command handles. +/// Dropping this owner requests shutdown but does not wait; call `join` to await cleanup. +pub struct WorkspaceJoin { + workspace: Workspace, + controller: Option>, +} + +impl Workspace { + pub fn start(options: Options) -> std::io::Result { + Workspace::start_inner(options, crate::testing::Hooks::default()) + } + + #[cfg(feature = "test-support")] + pub fn start_with_hooks( + options: Options, + hooks: crate::testing::Hooks, + ) -> std::io::Result { + Workspace::start_inner(options, hooks) + } + + fn start_inner( + options: Options, + hooks: crate::testing::Hooks, + ) -> std::io::Result { + let shared = Arc::new(Mutex::new(Admission::default())); + let (sender, receiver) = mpsc::channel(); + let (events, event_receiver) = EventSender::channel(); + let controller = + Controller::new(options, Arc::clone(&shared), sender.clone(), receiver, events, hooks)?; + let controller = + thread::Builder::new().name("iris-workspace".into()).spawn(move || controller.run())?; + let workspace = Workspace { sender, shared, capacity: options.request_capacity }; + let join = + WorkspaceJoin { workspace: Workspace::clone(&workspace), controller: Some(controller) }; + Ok(WorkspaceSession { workspace, events: event_receiver, join }) + } + + pub fn status(&self) -> Status { + Status::clone(&self.shared.lock().status) + } + + /// Admit input without waiting for analysis. Potential writes cancel executing reads and + /// discard queued reads; requests are never replayed automatically. Document saves and + /// inputs later rejected by document validation conservatively count as potential writes. + /// + /// Handle `Busy` as a request failure rather than pausing the protocol input loop: edits + /// must remain admissible while the request capacity is exhausted. + pub fn send(&self, mut command: Command) -> Result { + if let Command::FilesChanged(uris) = &command { + for uri in uris { + if iris_build::analysis::file_path(uri).is_none() { + return Err(crate::InputFailure::UnsupportedDocument(Url::clone(uri)).into()); + } + } + } + if let Command::LanguageServer(request) = &mut command { + if let Err(failure) = request.validate() { + request.reject(RequestFailure::InvalidInput(crate::InputFailure::clone(&failure))); + return Err(failure.into()); + } + } + let mut admission = self.shared.lock(); + if matches!(admission.status, Status::Stopping | Status::Stopped) { + return Err(RequestFailure::Unavailable); + } + if let Command::LanguageServer(request) = &mut command { + let Status::Ready { mut stamp, .. } = admission.status else { + drop(admission); + request.reject(RequestFailure::Unavailable); + return Err(RequestFailure::Unavailable); + }; + if admission.requests >= self.capacity { + drop(admission); + request.reject(RequestFailure::Busy); + return Err(RequestFailure::Busy); + } + stamp.revision = admission.revision; + admission.requests += 1; + request.admit(Arc::clone(&self.shared), stamp); + } else { + admission.sequence.advance(); + let rebuilds = match &command { + Command::Configure(configuration) => { + let unchanged_sources = + admission.configuration.as_ref().is_some_and(|previous| { + previous.root == configuration.root + && previous.settings.sources == configuration.settings.sources + }); + let rebuilds = + !unchanged_sources || !matches!(admission.status, Status::Ready { .. }); + admission.configuration = Some(ConfigurationInput::clone(configuration)); + rebuilds + } + _ => command.rebuilds(), + }; + let invalidates = !matches!(&command, Command::Configure(_)) || rebuilds; + if invalidates { + admission.revision = admission.sequence; + if let WorkerState::Querying { cancellation } = &admission.worker { + cancellation.cancel(); + } + } + match &command { + _ if rebuilds => { + admission.generation.advance(); + if let WorkerState::Preparing { cancellation } = &admission.worker { + cancellation.cancel(); + } + admission.status = Status::Rebuilding { + generation: admission.generation, + phase: Phase::Retiring, + }; + admission.status_revision += 1; + } + Command::Shutdown => { + if let WorkerState::Preparing { cancellation } = &admission.worker { + cancellation.cancel(); + } + admission.status = Status::Stopping; + admission.status_revision += 1; + } + _ => {} + } + } + let sequence = admission.sequence; + let revision = admission.revision; + let generation = admission.generation; + let message = Message::Command { sequence, revision, generation, command }; + // Keep admission locked through enqueue, so readiness cannot overtake this input. + let result = self.sender.send(message); + drop(admission); + result.map_err(|_| RequestFailure::Unavailable)?; + Ok(sequence) + } +} + +impl WorkspaceJoin { + /// Request shutdown and wait for the controller, worker, and process cleanup. Call on a + /// blocking thread, not a protocol loop. `Status::Stopped` alone does not join the controller. + pub fn join(mut self) -> thread::Result<()> { + let _ = self.workspace.send(Command::Shutdown); + self.controller.take().expect("workspace controller missing").join() + } +} + +impl Drop for WorkspaceJoin { + fn drop(&mut self) { + let _ = self.workspace.send(Command::Shutdown); + } +} diff --git a/compiler-lsp/iris-workspace/src/worker.rs b/compiler-lsp/iris-workspace/src/worker.rs new file mode 100644 index 000000000..25ad7269f --- /dev/null +++ b/compiler-lsp/iris-workspace/src/worker.rs @@ -0,0 +1,394 @@ +use std::collections::{BTreeMap, BTreeSet}; +use std::path::Path; +use std::sync::{Arc, mpsc}; +use std::{fs, io, panic}; + +use analyzer::AnalyzerContext; +use building::{ + DiskObservation, FileLifecycle, ForeignEvent, LifecycleEvent, QueryEngine, SourceEvent, + SourceUnitKey, +}; +use files::ForeignSourceKind; +use iris_build::analysis::{self, AnalysisConfig, AnalysisOverlay, AnalysisSelection}; +use iris_build::compilation::{CompilationParts, MaterializedPrim}; +use iris_build::events::{BuildEvent, BuildEventSink}; +use lsp_types::{Diagnostic, Url}; + +use crate::controller::Message; +use crate::documents::{OpenDocument, document_path}; +use crate::language_server::{Analysis, Host, failure}; +use crate::{ + AnalysisStamp, Cancellation, ConfigurationInput, Generation, LanguageServer, Options, + RequestFailure, +}; + +pub(crate) enum Work { + Prepare { + configuration: ConfigurationInput, + documents: BTreeMap, + generation: Generation, + stamp: AnalysisStamp, + cancellation: Cancellation, + }, + Reconcile { + documents: BTreeMap, + dirty: BTreeSet, + stamp: AnalysisStamp, + }, + Analyze(LanguageServer), + Diagnostics { + uri: Url, + stamp: AnalysisStamp, + cancellation: Cancellation, + }, + Stop, +} + +pub(crate) enum Completed { + Reconciled { + generation: Generation, + stamp: AnalysisStamp, + sources: Vec, + }, + Analyzed, + Diagnostics { + uri: Url, + stamp: AnalysisStamp, + version: Option, + result: Result, RequestFailure>, + }, + Failed { + generation: Generation, + failure: RequestFailure, + }, + Stopped, +} + +struct Events { + sender: mpsc::Sender, + generation: Generation, +} + +impl BuildEventSink for Events { + fn send(&self, event: BuildEvent) { + let _ = self.sender.send(Message::Progress { generation: self.generation, event }); + } +} + +struct Compilation { + engine: QueryEngine, + files: FileLifecycle, + selection: AnalysisSelection, + documents: BTreeMap, + analysis: Analysis, + generation: Generation, + stamp: AnalysisStamp, +} + +impl Compilation { + fn source_uris(&self) -> Vec { + self.files + .source_ids() + .filter_map(|file_id| { + if self.files.source_metadata(file_id) != Some(&true) { + return None; + } + self.files.source_path(file_id).and_then(|uri| Url::parse(&uri).ok()) + }) + .collect() + } + + fn reconcile( + &mut self, + documents: BTreeMap, + dirty: BTreeSet, + stamp: AnalysisStamp, + ) -> Result<(), RequestFailure> { + self.analysis.invalidate(); + for uri in dirty { + let path = document_path(&uri)?; + let unit = unit(&path)?; + let current = documents.get(&uri); + let previous = self.documents.get(&uri); + let metadata = self.selection.metadata(&uri).unwrap_or(false); + let source = path.extension().is_some_and(|extension| extension == "purs"); + let event = if source { + let event = match (previous, current) { + (_, Some(document)) + if previous + .is_none_or(|previous| previous.lifetime != document.lifetime) => + { + SourceEvent::Opened { + text: Arc::clone(&document.text), + version: document.version, + metadata, + } + } + (Some(previous), Some(document)) if previous.version != document.version => { + SourceEvent::Changed { + text: Arc::clone(&document.text), + version: document.version, + } + } + (_, Some(_)) => continue, + (Some(_), None) if !self.selection.selected_sources.contains(uri.as_str()) => { + SourceEvent::Closed { disk: DiskObservation::NotFound } + } + (Some(_), None) => SourceEvent::Closed { disk: observe(&path)? }, + (None, None) if !self.selection.selected_sources.contains(uri.as_str()) => { + continue; + } + (None, None) => SourceEvent::DiskObserved { disk: observe(&path)?, metadata }, + }; + LifecycleEvent::Source { unit: SourceUnitKey::clone(&unit), event } + } else { + let kind = if path.extension().is_some_and(|extension| extension == "jsx") { + ForeignSourceKind::Jsx + } else { + ForeignSourceKind::JavaScript + }; + let event = match (previous, current) { + (_, Some(document)) + if previous + .is_none_or(|previous| previous.lifetime != document.lifetime) => + { + ForeignEvent::Opened { + text: Arc::clone(&document.text), + version: document.version, + } + } + (Some(previous), Some(document)) if previous.version != document.version => { + ForeignEvent::Changed { + text: Arc::clone(&document.text), + version: document.version, + } + } + (_, Some(_)) => continue, + (Some(_), None) => ForeignEvent::Closed { disk: observe(&path)? }, + (None, None) => ForeignEvent::DiskObserved { disk: observe(&path)? }, + }; + LifecycleEvent::Foreign { unit: SourceUnitKey::clone(&unit), kind, event } + }; + self.files.apply(&self.engine, event); + if source && (previous.is_none() || current.is_none()) { + for kind in ForeignSourceKind::ALL { + let foreign_path = path.with_extension(kind.extension()); + let foreign_uri = Url::from_file_path(&foreign_path) + .map_err(|()| RequestFailure::Workspace("invalid foreign path".into()))?; + if let Some(document) = documents.get(&foreign_uri) { + self.files.apply( + &self.engine, + LifecycleEvent::Foreign { + unit: SourceUnitKey::clone(&unit), + kind, + event: ForeignEvent::Opened { + text: Arc::clone(&document.text), + version: document.version, + }, + }, + ); + } else { + self.files.apply( + &self.engine, + LifecycleEvent::Foreign { + unit: SourceUnitKey::clone(&unit), + kind, + event: ForeignEvent::DiskObserved { disk: observe(&foreign_path)? }, + }, + ); + } + } + } + } + self.documents = documents; + self.stamp = stamp; + Ok(()) + } +} + +fn observe(path: &Path) -> Result { + match fs::read_to_string(path) { + Ok(text) => Ok(DiskObservation::Found(text.into())), + Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(DiskObservation::NotFound), + Err(error) => Err(RequestFailure::Workspace(format!("{}: {error}", path.display()).into())), + } +} + +fn unit(path: &Path) -> Result { + let source = Url::from_file_path(path.with_extension("purs")) + .map_err(|()| RequestFailure::Workspace("invalid source path".into()))?; + let foreign = Url::from_file_path(path.with_extension("js")) + .map_err(|()| RequestFailure::Workspace("invalid foreign path".into()))?; + Ok(SourceUnitKey::new(source.as_str(), foreign.as_str())) +} + +pub(crate) fn run( + receiver: mpsc::Receiver, + sender: mpsc::Sender, + options: Options, + hooks: crate::testing::Hooks, +) { + let mut compilation: Option = None; + let mut prim: Option> = None; + let mut generation = Generation::default(); + while let Ok(work) = receiver.recv() { + let result = panic::catch_unwind(panic::AssertUnwindSafe( + || -> Result { + match work { + Work::Prepare { + configuration, + documents, + generation: next_generation, + stamp, + cancellation, + } => { + generation = next_generation; + compilation = None; + hooks.reach(crate::testing::Point::BeforePreparation); + if cancellation.is_cancelled() { + return Err(RequestFailure::Cancelled); + } + let materialized = match &prim { + Some(prim) => Arc::clone(prim), + None => { + let materialized = + Arc::new(MaterializedPrim::new().map_err(|error| { + RequestFailure::Workspace(error.to_string().into()) + })?); + prim = Some(Arc::clone(&materialized)); + materialized + } + }; + let configuration = AnalysisConfig { + root: configuration.root, + sources: configuration.settings.sources, + }; + let overlays = documents + .iter() + .map(|(uri, document)| AnalysisOverlay { + uri: Url::clone(uri), + text: Arc::clone(&document.text), + version: document.version, + }) + .collect::>(); + let events = Events { sender: sender.clone(), generation }; + let prepared = analysis::prepare( + &configuration, + &overlays, + materialized, + &cancellation.build, + &events, + ) + .map_err(|error| { + if matches!(error, analysis::AnalysisError::Cancelled) { + RequestFailure::Cancelled + } else { + RequestFailure::Workspace(error.to_string().into()) + } + })?; + let CompilationParts { engine, files, .. } = + prepared.compilation.into_parts(); + let prim_id = engine.module_file("Prim").expect("Prim must be registered"); + let prim_uri = + files.source_path(prim_id).expect("Prim must have a locator"); + let analysis = + Analysis::new(format!("{prim_uri}:{}", stamp.incarnation.value)); + compilation = Some(Compilation { + engine, + files, + selection: prepared.selection, + documents, + analysis, + generation, + stamp, + }); + let sources = compilation.as_ref().unwrap().source_uris(); + Ok(Completed::Reconciled { generation, stamp, sources }) + } + Work::Reconcile { documents, dirty, stamp } => { + let compilation = + compilation.as_mut().ok_or(RequestFailure::Unavailable)?; + compilation.reconcile(documents, dirty, stamp)?; + Ok(Completed::Reconciled { + generation: compilation.generation, + stamp, + sources: compilation.source_uris(), + }) + } + Work::Analyze(mut command) => { + hooks.reach(crate::testing::Point::BeforeAnalysis); + match compilation.as_mut() { + Some(compilation) if compilation.stamp == command.stamp() => { + if command.cancellation().is_cancelled() { + command.reject(RequestFailure::Cancelled); + } else { + compilation.analysis.execute( + command, + &compilation.engine, + &compilation.files, + options, + &hooks, + ); + } + } + _ => command.reject(RequestFailure::Stale), + } + Ok(Completed::Analyzed) + } + Work::Diagnostics { uri, stamp, cancellation } => { + hooks.reach(crate::testing::Point::BeforeDiagnostics); + let compilation = + compilation.as_ref().ok_or(RequestFailure::Unavailable)?; + let version = + compilation.documents.get(&uri).map(|document| document.version); + let result = if compilation.stamp != stamp || cancellation.is_cancelled() { + Err(RequestFailure::Cancelled) + } else if let Some(file_id) = compilation.files.source_id(uri.as_str()) { + let snapshot = + compilation.engine.snapshot_with_cancellation(cancellation.query); + let host = Host { engine: &snapshot, files: &compilation.files }; + let context = AnalyzerContext::new( + &host, + options.position_encoding, + options.capabilities, + ); + analyzer::diagnostics::implementation(&context, file_id) + .map(|collected| collected.diagnostics) + .map_err(failure) + } else { + Ok(vec![]) + }; + hooks.reach(crate::testing::Point::AfterDiagnostics); + Ok(Completed::Diagnostics { uri, stamp, version, result }) + } + Work::Stop => { + compilation = None; + prim = None; + Ok(Completed::Stopped) + } + } + }, + )); + let completed = match result { + Ok(Ok(completed)) => completed, + Ok(Err(failure)) => { + compilation = None; + Completed::Failed { generation, failure } + } + Err(_) => { + compilation = None; + Completed::Failed { + generation, + failure: RequestFailure::Workspace("compilation worker panicked".into()), + } + } + }; + if matches!(completed, Completed::Reconciled { .. }) { + hooks.reach(crate::testing::Point::BeforeAcknowledgement); + } + let stopped = matches!(completed, Completed::Stopped); + if sender.send(Message::Completed(completed)).is_err() || stopped { + break; + } + } +} diff --git a/compiler-lsp/iris-workspace/tests/sequences.rs b/compiler-lsp/iris-workspace/tests/sequences.rs new file mode 100644 index 000000000..ae15291b9 --- /dev/null +++ b/compiler-lsp/iris-workspace/tests/sequences.rs @@ -0,0 +1,1524 @@ +use std::fs; +use std::future::Future; +use std::ops::Deref; +use std::task::Poll; +use std::time::Duration; + +use configuration::{Configuration, SourceDiscovery}; +use iris_workspace::testing::{Hooks, Point}; +use iris_workspace::{ + AnalysisStamp, Command, ConfigurationInput, Document, Event, EventReceiver, InputFailure, + InputSequence, LanguageServer, Options, Outcome, Reply, Request, RequestFailure, Status, + Workspace, WorkspaceJoin, WorkspaceSession, +}; +use lsp_types::{ + CompletionItem, CompletionResponse, DocumentSymbolResponse, Hover, HoverContents, Position, + Range, TextDocumentContentChangeEvent, Url, +}; +use tempfile::TempDir; + +const ORIGINAL: &str = "module Main where\n\nvalue :: Int\nvalue = 1\n\nuse = value\n"; +const CHANGED: &str = + "module Main where\n\nchanged :: String\nchanged = \"hello\"\n\nuse = changed\n"; + +struct Harness { + workspace: Workspace, + join: Option, + events: EventReceiver, + hooks: Hooks, + directory: TempDir, + uri: Url, +} + +impl Deref for Harness { + type Target = Workspace; + + fn deref(&self) -> &Workspace { + &self.workspace + } +} + +impl Drop for Harness { + fn drop(&mut self) { + if let Some(join) = self.join.take() { + join.join().unwrap(); + } + } +} + +async fn bounded(future: impl std::future::Future) -> T { + tokio::time::timeout(Duration::from_secs(15), future) + .await + .expect("workspace sequence timed out") +} + +// Call while an Idle gate holds the controller, so an empty queue means it has been drained. +async fn queued_events(harness: &mut Harness) -> Vec> { + let mut events = vec![]; + loop { + let mut receive = std::pin::pin!(harness.events.recv()); + let event = std::future::poll_fn(|context| match receive.as_mut().poll(context) { + Poll::Ready(event) => Poll::Ready(event), + Poll::Pending => Poll::Ready(None), + }) + .await; + let Some(event) = event else { return events }; + events.push(event); + } +} + +impl Harness { + fn new(options: Options) -> Harness { + let directory = tempfile::tempdir().unwrap(); + fs::write(directory.path().join("Main.purs"), ORIGINAL).unwrap(); + let uri = Url::from_file_path(directory.path().join("Main.purs")).unwrap(); + + let hooks = Hooks::default(); + let WorkspaceSession { workspace, events, join } = + Workspace::start_with_hooks(options, Hooks::clone(&hooks)).unwrap(); + + Harness { workspace, join: Some(join), events, hooks, directory, uri } + } + + fn configuration(&self) -> ConfigurationInput { + let sources = SourceDiscovery::Command { + program: "node".into(), + arguments: vec!["-e".into(), "console.log('*.purs')".into()], + }; + + ConfigurationInput { + root: self.directory.path().to_path_buf(), + settings: Configuration { sources, ..Configuration::default() }, + } + } + + fn configure(&self) -> InputSequence { + self.send(Command::Configure(self.configuration())).unwrap() + } + + async fn next(&mut self) -> Event { + bounded(async { + loop { + let delivery = self.events.recv().await.expect("workspace event stream closed"); + if let Ok(event) = delivery.release() { + return event; + } + } + }) + .await + } + + async fn ready(&mut self, sequence: InputSequence) -> AnalysisStamp { + loop { + match self.status() { + Status::Ready { stamp, sequence: applied, .. } if applied >= sequence => { + return stamp; + } + Status::Failed { message, .. } => panic!("workspace failed: {message}"), + _ => { + self.next().await; + } + } + } + } + + fn open(&self, text: &str, version: i32) -> InputSequence { + self.send(Command::Document(Document::Open { + uri: Url::clone(&self.uri), + text: text.into(), + version, + })) + .unwrap() + } + + fn change(&self, text: &str, version: i32) -> InputSequence { + let change = + TextDocumentContentChangeEvent { range: None, range_length: None, text: text.into() }; + + self.send(Command::Document(Document::Change { + uri: Url::clone(&self.uri), + version, + changes: vec![change], + })) + .unwrap() + } + + fn hover_request(&self) -> Request> { + let (reply, request) = Reply::channel(); + self.send(Command::LanguageServer(LanguageServer::Hover { + uri: Url::clone(&self.uri), + position: Position::new(5, 7), + reply, + })) + .unwrap(); + + request + } + + async fn hover(&self) -> String { + let hover = bounded(self.hover_request()).await.unwrap().unwrap(); + + match hover.contents { + HoverContents::Markup(markup) => markup.value, + other => format!("{other:?}"), + } + } + + async fn symbols(&self) -> Vec { + let (reply, request) = Reply::channel(); + self.send(Command::LanguageServer(LanguageServer::DocumentSymbols { + uri: Url::clone(&self.uri), + reply, + })) + .unwrap(); + + match bounded(request).await.unwrap().unwrap() { + DocumentSymbolResponse::Flat(symbols) => { + symbols.into_iter().map(|symbol| symbol.name).collect() + } + DocumentSymbolResponse::Nested(symbols) => { + symbols.into_iter().map(|symbol| symbol.name).collect() + } + } + } + + async fn completion(&self) -> Vec { + let (reply, request) = Reply::channel(); + self.send(Command::LanguageServer(LanguageServer::Completion { + uri: Url::clone(&self.uri), + position: Position::new(5, 9), + reply, + })) + .unwrap(); + + match bounded(request).await.unwrap().unwrap() { + CompletionResponse::Array(items) => items, + CompletionResponse::List(list) => list.items, + } + } + + async fn resolve(&self, item: CompletionItem) -> CompletionItem { + let (reply, request) = Reply::channel(); + self.send(Command::LanguageServer(LanguageServer::ResolveCompletion { item, reply })) + .unwrap(); + + bounded(request).await.unwrap() + } +} + +#[tokio::test] +async fn startup_reconciles_edits_admitted_before_ready() { + let mut harness = Harness::new(Options::default()); + let mut acknowledgement = harness.hooks.pause_next(Point::BeforeAcknowledgement); + + harness.configure(); + bounded(acknowledgement.entered()).await; + + harness.open(ORIGINAL, 1); + let sequence = harness.change(CHANGED, 2); + + let (reply, request) = Reply::channel(); + let result = harness.send(Command::LanguageServer(LanguageServer::Hover { + uri: Url::clone(&harness.uri), + position: Position::new(5, 7), + reply, + })); + + assert!(matches!(result, Err(RequestFailure::Unavailable))); + assert!(matches!(bounded(request).await, Err(RequestFailure::Unavailable))); + + drop(acknowledgement); + assert_eq!(harness.ready(sequence).await.revision, sequence); + assert!(harness.hover().await.contains("String")); + + let symbols = harness.symbols().await; + assert!(symbols.contains(&"changed".into())); + assert!(!symbols.contains(&"value".into())); +} + +#[tokio::test] +async fn supersession_and_failed_rebuild_preserve_buffers_without_rollback() { + let mut harness = Harness::new(Options::default()); + let initial = harness.configure(); + let first = harness.ready(initial).await; + + harness.open(CHANGED, 11); + let mut preparation = harness.hooks.pause_next(Point::BeforePreparation); + harness.send(Command::Reload).unwrap(); + bounded(preparation.entered()).await; + + let mut invalid = harness.configuration(); + invalid.settings.sources = SourceDiscovery::Command { + program: "node".into(), + arguments: vec![ + "-e".into(), + "process.stderr.write('expected failure'); process.exit(7)".into(), + ], + }; + harness.send(Command::Configure(invalid)).unwrap(); + drop(preparation); + + loop { + if let Event::StatusChanged(Status::Failed { message, .. }) = harness.next().await { + assert!(message.contains("7"), "{message}"); + break; + } + } + assert!(matches!(harness.status(), Status::Failed { .. })); + + harness.change(ORIGINAL, 12); + let sequence = harness.configure(); + let recovered = harness.ready(sequence).await; + + assert_ne!(first.incarnation, recovered.incarnation); + assert!(harness.hover().await.contains("Int")); +} + +#[tokio::test] +async fn settled_replies_survive_edits_but_completion_tokens_expire() { + let mut harness = Harness::new(Options::default()); + let sequence = harness.configure(); + harness.ready(sequence).await; + + let held = bounded(harness.hover_request()).await.unwrap(); + let completions = harness.completion().await; + let item = completions + .into_iter() + .find(|item| item.data.is_some()) + .expect("completion needs an opaque resolve token"); + assert!(item.data.as_ref().unwrap().is_string()); + + let sequence = harness.open(CHANGED, 1); + assert!(held.is_some()); + harness.ready(sequence).await; + + let resolved = harness.resolve(CompletionItem::clone(&item)).await; + assert!(resolved.data.is_none()); + assert_eq!(resolved.documentation, item.documentation); + + let sequence = harness.send(Command::Reload).unwrap(); + harness.ready(sequence).await; + + let resolved = harness.resolve(item).await; + assert!(resolved.data.is_none()); + + let forged = CompletionItem { + label: "forged".into(), + data: Some(serde_json::json!({"TermItem": [4294967295u32, 4294967295u32]})), + ..CompletionItem::default() + }; + assert!(harness.resolve(forged).await.data.is_none()); +} + +#[tokio::test] +async fn overload_and_request_cancellation_do_not_block_inputs() { + let mut harness = Harness::new(Options { request_capacity: 1, ..Options::default() }); + let sequence = harness.configure(); + harness.ready(sequence).await; + + let mut analysis = harness.hooks.pause_next(Point::BeforeAnalysis); + let request = harness.hover_request(); + bounded(analysis.entered()).await; + + let (reply, rejected) = Reply::channel(); + let result = harness.send(Command::LanguageServer(LanguageServer::Hover { + uri: Url::clone(&harness.uri), + position: Position::new(5, 7), + reply, + })); + + assert!(matches!(result, Err(RequestFailure::Busy))); + assert!(matches!(bounded(rejected).await, Err(RequestFailure::Busy))); + + request.cancellation().cancel(); + assert!(matches!(bounded(request).await, Err(RequestFailure::Cancelled))); + + let (reply, rejected) = Reply::channel(); + let command = LanguageServer::DocumentSymbols { uri: Url::clone(&harness.uri), reply }; + assert!(matches!(harness.send(Command::LanguageServer(command)), Err(RequestFailure::Busy))); + assert!(matches!(bounded(rejected).await, Err(RequestFailure::Busy))); + + let sequence = harness.open(CHANGED, 1); + drop(analysis); + harness.ready(sequence).await; + assert!(harness.hover().await.contains("String")); +} + +#[tokio::test] +async fn computed_rename_rejections_survive_edits_after_settlement() { + let mut harness = Harness::new(Options::default()); + let sequence = harness.configure(); + harness.ready(sequence).await; + + for replacement in [ + None, + Some(Command::Document(Document::Open { + uri: Url::clone(&harness.uri), + text: ORIGINAL.into(), + version: 1, + })), + Some(Command::Configure(harness.configuration())), + ] { + let (reply, request) = Reply::channel(); + let command = LanguageServer::Rename { + uri: Url::clone(&harness.uri), + position: Position::new(5, 7), + new_name: "use".into(), + reply, + }; + let mut settlement = harness.hooks.pause_next(Point::AfterReplySettlement); + harness.send(Command::LanguageServer(command)).unwrap(); + bounded(settlement.entered()).await; + + if let Some(command) = replacement { + harness.send(command).unwrap(); + } + request.cancellation().cancel(); + assert!(matches!( + bounded(request).await, + Err(RequestFailure::LanguageServer( + iris_workspace::LanguageServerFailure::RenameRejected(_) + )) + )); + drop(settlement); + let sequence = harness.configure(); + harness.ready(sequence).await; + } +} + +#[tokio::test] +async fn edits_and_explicit_cancellation_win_before_reply_settlement() { + for new_name in ["count", "use"] { + for edit in [false, true] { + let mut harness = Harness::new(Options::default()); + let sequence = harness.configure(); + harness.ready(sequence).await; + + let mut settlement = harness.hooks.pause_next(Point::BeforeReplySettlement); + let (reply, request) = Reply::channel(); + let command = LanguageServer::Rename { + uri: Url::clone(&harness.uri), + position: Position::new(5, 7), + new_name: new_name.into(), + reply, + }; + harness.send(Command::LanguageServer(command)).unwrap(); + bounded(settlement.entered()).await; + + if edit { + harness.open(CHANGED, 1); + } else { + request.cancellation().cancel(); + } + assert!(matches!(bounded(request).await, Err(RequestFailure::Cancelled))); + drop(settlement); + } + } +} + +#[tokio::test] +async fn unread_success_survives_reload_but_keeps_the_worker_occupied() { + let mut harness = Harness::new(Options { request_capacity: 1, ..Options::default() }); + let sequence = harness.configure(); + harness.ready(sequence).await; + + let mut settlement = harness.hooks.pause_next(Point::AfterReplySettlement); + let request = harness.hover_request(); + bounded(settlement.entered()).await; + request.cancellation().cancel(); + + let (reply, rejected) = Reply::channel(); + let command = LanguageServer::DocumentSymbols { uri: Url::clone(&harness.uri), reply }; + assert!(matches!(harness.send(Command::LanguageServer(command)), Err(RequestFailure::Busy))); + assert!(matches!(bounded(rejected).await, Err(RequestFailure::Busy))); + + let sequence = harness.send(Command::Reload).unwrap(); + let hover = bounded(request).await.unwrap().unwrap(); + assert!(format!("{:?}", hover.contents).contains("Int")); + assert!(matches!(harness.status(), Status::Rebuilding { .. })); + + drop(settlement); + harness.ready(sequence).await; + assert!(harness.hover().await.contains("Int")); +} + +#[tokio::test] +async fn cross_file_edits_cancel_active_reads_and_discard_queued_reads() { + let mut harness = Harness::new(Options::default()); + let sequence = harness.configure(); + harness.ready(sequence).await; + + let mut snapshot = harness.hooks.pause_next(Point::SnapshotActive); + let (reply, request) = Reply::channel(); + let command = LanguageServer::References { + uri: Url::clone(&harness.uri), + position: Position::new(5, 7), + reply, + }; + harness.send(Command::LanguageServer(command)).unwrap(); + bounded(snapshot.entered()).await; + let queued = harness.hover_request(); + + let uri = harness.uri.join("Other.purs").unwrap(); + let document = Document::Open { + uri, + text: "module Other where\nimport Main\nother = value\n".into(), + version: 1, + }; + let sequence = harness.send(Command::Document(document)).unwrap(); + assert!(matches!(bounded(request).await, Err(RequestFailure::Cancelled))); + assert!(matches!(bounded(queued).await, Err(RequestFailure::Stale))); + + drop(snapshot); + harness.ready(sequence).await; + assert!(harness.hover().await.contains("Int")); +} + +#[tokio::test] +async fn configuration_policy_preserves_running_reads_and_completion_identity() { + let mut harness = Harness::new(Options::default()); + let sequence = harness.configure(); + configuration_outcome(&mut harness, sequence).await; + let stamp = harness.ready(sequence).await; + let item = harness.completion().await.into_iter().find(|item| item.label == "value").unwrap(); + + let mut snapshot = harness.hooks.pause_next(Point::SnapshotActive); + let request = harness.hover_request(); + bounded(snapshot.entered()).await; + + let sequence = harness.configure(); + assert_eq!( + configuration_outcome(&mut harness, sequence).await, + iris_workspace::ConfigurationOutcome::Unchanged + ); + + let mut policy = harness.configuration(); + policy.settings.diagnostics.on_change = true; + let sequence = harness.send(Command::Configure(policy)).unwrap(); + assert_eq!( + configuration_outcome(&mut harness, sequence).await, + iris_workspace::ConfigurationOutcome::PolicyUpdated + ); + assert!(!request.cancellation().is_cancelled()); + + drop(snapshot); + assert!(bounded(request).await.unwrap().is_some()); + assert_eq!(harness.ready(sequence).await, stamp); + let resolved = harness.resolve(item).await; + assert!(resolved.detail.unwrap().contains("Int")); +} + +#[tokio::test] +async fn cancelled_diagnostics_do_not_retry_without_a_trigger() { + for point in [Point::BeforeDiagnostics, Point::AfterDiagnostics] { + let mut harness = Harness::new(Options::default()); + let mut diagnostics = harness.hooks.pause_next(point); + harness.open(ORIGINAL, 1); + harness.configure(); + bounded(diagnostics.entered()).await; + + let mut idle = harness.hooks.pause_next(Point::Idle); + let sequence = harness.change(CHANGED, 2); + drop(diagnostics); + bounded(idle.entered()).await; + assert!( + matches!(harness.status(), Status::Ready { sequence: applied, .. } if applied == sequence) + ); + + let events = queued_events(&mut harness).await; + for event in events { + assert!(!matches!(event.release(), Ok(Event::Diagnostics { .. }))); + } + drop(idle); + + harness.send(Command::Document(Document::Save(Url::clone(&harness.uri)))).unwrap(); + let uri = Url::clone(&harness.uri); + assert!(diagnostics_for(&mut harness, &uri).await.is_empty()); + } +} + +#[tokio::test] +async fn held_diagnostics_survive_policy_updates_and_removal_clears_survive_other_edits() { + let mut harness = Harness::new(Options::default()); + fs::remove_file(harness.directory.path().join("Main.purs")).unwrap(); + let mut idle = harness.hooks.pause_next(Point::Idle); + harness.open("module Main where\nvalue :: Int\nvalue = \"wrong\"\n", 1); + harness.configure(); + bounded(idle.entered()).await; + let held = queued_events(&mut harness).await; + + let mut policy = harness.configuration(); + policy.settings.diagnostics.on_change = true; + harness.send(Command::Configure(policy)).unwrap(); + let published = held.into_iter().any(|event| { + matches!(event.release(), Ok(Event::Diagnostics { diagnostics, .. }) if !diagnostics.is_empty()) + }); + assert!(published); + + let mut closed = harness.hooks.pause_next(Point::Idle); + harness.send(Command::Document(Document::Close(Url::clone(&harness.uri)))).unwrap(); + drop(idle); + bounded(closed.entered()).await; + let held = queued_events(&mut harness).await; + + let uri = harness.uri.join("Other.purs").unwrap(); + harness + .send(Command::Document(Document::Open { + uri, + text: "module Other where\nother = 1\n".into(), + version: 1, + })) + .unwrap(); + let cleared = held.into_iter().any(|event| { + matches!(event.release(), Ok(Event::Diagnostics { uri, diagnostics, .. }) if uri == harness.uri && diagnostics.is_empty()) + }); + assert!(cleared); + drop(closed); +} + +#[tokio::test] +async fn sequential_unicode_edits_are_atomic_and_versions_reset_only_on_reopen() { + let mut harness = Harness::new(Options::default()); + harness.configure(); + let text = "module Main where\nvalue = \"a😀b\"\n"; + let sequence = harness.open(text, 7); + harness.ready(sequence).await; + + let edit = |start, end, text: &str| TextDocumentContentChangeEvent { + range: Some(Range::new(Position::new(1, start), Position::new(1, end))), + range_length: None, + text: text.into(), + }; + let command = Document::Change { + uri: Url::clone(&harness.uri), + version: 8, + changes: vec![edit(10, 12, "xyz"), edit(8, 15, "42")], + }; + let sequence = harness.send(Command::Document(command)).unwrap(); + harness.ready(sequence).await; + + let (reply, request) = Reply::channel(); + let command = LanguageServer::Hover { + uri: Url::clone(&harness.uri), + position: Position::new(1, 1), + reply, + }; + harness.send(Command::LanguageServer(command)).unwrap(); + + let hover = bounded(request).await.unwrap().unwrap(); + assert!(format!("{:?}", hover.contents).contains("Int")); + + let command = Document::Change { + uri: Url::clone(&harness.uri), + version: 9, + changes: vec![edit(0, 5, "broken"), edit(8, 1, "invalid")], + }; + let sequence = harness.send(Command::Document(command)).unwrap(); + + loop { + if matches!( + harness.next().await, + Event::InputRejected { failure: InputFailure::InvalidRange(_), .. } + ) { + break; + } + } + harness.ready(sequence).await; + assert!(harness.symbols().await.contains(&"value".into())); + + let sequence = harness.change(CHANGED, 8); + loop { + if matches!( + harness.next().await, + Event::InputRejected { failure: InputFailure::StaleVersion(_), .. } + ) { + break; + } + } + harness.ready(sequence).await; + assert!(!harness.symbols().await.contains(&"changed".into())); + + harness.send(Command::Document(Document::Close(Url::clone(&harness.uri)))).unwrap(); + let sequence = harness.open(CHANGED, 1); + harness.ready(sequence).await; + assert!(harness.hover().await.contains("String")); +} + +#[tokio::test] +async fn disk_changes_do_not_replace_open_authority_and_close_restores_disk() { + let mut harness = Harness::new(Options::default()); + harness.configure(); + let sequence = harness.open(CHANGED, 1); + harness.ready(sequence).await; + + fs::write(harness.directory.path().join("Main.purs"), ORIGINAL).unwrap(); + let sequence = harness.send(Command::FilesChanged(vec![Url::clone(&harness.uri)])).unwrap(); + harness.ready(sequence).await; + assert!(harness.hover().await.contains("String")); + + let sequence = + harness.send(Command::Document(Document::Close(Url::clone(&harness.uri)))).unwrap(); + harness.ready(sequence).await; + assert!(harness.hover().await.contains("Int")); +} + +#[tokio::test] +async fn diagnostics_are_cleared_on_rebuild_and_old_publications_cannot_escape() { + let mut harness = Harness::new(Options::default()); + fs::write( + harness.directory.path().join("Main.purs"), + "module Main where\nvalue :: Int\nvalue = \"wrong\"\n", + ) + .unwrap(); + + let mut diagnostics = harness.hooks.pause_next(Point::BeforeDiagnostics); + harness.configure(); + loop { + if matches!( + harness.next().await, + Event::ConfigurationFinished { + outcome: iris_workspace::ConfigurationOutcome::Rebuilt, + .. + } + ) { + break; + } + } + bounded(diagnostics.entered()).await; + drop(diagnostics); + let old = bounded(harness.events.recv()).await.unwrap(); + + let mut preparation = harness.hooks.pause_next(Point::BeforePreparation); + let sequence = harness.send(Command::Reload).unwrap(); + bounded(preparation.entered()).await; + assert!(matches!(old.release(), Err(RequestFailure::Stale))); + + loop { + if let Event::Diagnostics { diagnostics, .. } = harness.next().await { + assert!(diagnostics.is_empty()); + break; + } + } + + drop(preparation); + harness.ready(sequence).await; + loop { + if let Event::Diagnostics { diagnostics, .. } = harness.next().await { + assert!(!diagnostics.is_empty()); + let mismatch = diagnostics.iter().any(|diagnostic| { + diagnostic.message.contains("Int") && diagnostic.message.contains("String") + }); + assert!(mismatch); + break; + } + } +} + +async fn diagnostics_for(harness: &mut Harness, uri: &Url) -> Vec { + loop { + let Event::Diagnostics { uri: published, diagnostics, .. } = harness.next().await else { + continue; + }; + if published == *uri { + return diagnostics; + } + } +} + +#[tokio::test] +async fn foreign_buffers_survive_rebuild_and_reopen_with_reset_versions() { + let mut harness = Harness::new(Options::default()); + let source = "module Main where\nforeign import value :: Int\n"; + fs::write(harness.directory.path().join("Main.purs"), source).unwrap(); + let javascript = Url::from_file_path(harness.directory.path().join("Main.js")).unwrap(); + let jsx = Url::from_file_path(harness.directory.path().join("Main.jsx")).unwrap(); + + let command = Document::Open { + uri: Url::clone(&javascript), + text: "export const value = 1;".into(), + version: 8, + }; + harness.send(Command::Document(command)).unwrap(); + harness.open(source, 1); + let sequence = harness.configure(); + harness.ready(sequence).await; + + let uri = Url::clone(&harness.uri); + assert!(diagnostics_for(&mut harness, &uri).await.is_empty()); + + let command = Document::Open { + uri: Url::clone(&jsx), + text: "export const value = 2;".into(), + version: 9, + }; + let sequence = harness.send(Command::Document(command)).unwrap(); + harness.ready(sequence).await; + + let ambiguous = diagnostics_for(&mut harness, &uri).await; + assert!(!ambiguous.is_empty()); + + let sequence = harness.send(Command::Reload).unwrap(); + harness.ready(sequence).await; + + let rebuilt = loop { + let diagnostics = diagnostics_for(&mut harness, &uri).await; + if !diagnostics.is_empty() { + break diagnostics; + } + }; + assert_eq!(ambiguous, rebuilt); + + harness.send(Command::Document(Document::Close(javascript))).unwrap(); + harness.send(Command::Document(Document::Close(Url::clone(&jsx)))).unwrap(); + + let command = Document::Open { uri: jsx, text: "export const value = 3;".into(), version: 1 }; + let sequence = harness.send(Command::Document(command)).unwrap(); + harness.ready(sequence).await; + assert!(diagnostics_for(&mut harness, &uri).await.is_empty()); +} + +#[tokio::test] +async fn closing_a_buffer_only_source_clears_its_diagnostics() { + let mut harness = Harness::new(Options::default()); + fs::remove_file(harness.directory.path().join("Main.purs")).unwrap(); + + harness.open("module Main where\nvalue :: Int\nvalue = \"wrong\"\n", 1); + let sequence = harness.configure(); + harness.ready(sequence).await; + + let uri = Url::clone(&harness.uri); + assert!(!diagnostics_for(&mut harness, &uri).await.is_empty()); + + let sequence = harness.send(Command::Document(Document::Close(Url::clone(&uri)))).unwrap(); + assert!(diagnostics_for(&mut harness, &uri).await.is_empty()); + harness.ready(sequence).await; + + let (reply, request) = Reply::channel(); + harness.send(Command::LanguageServer(LanguageServer::DocumentSymbols { uri, reply })).unwrap(); + assert!(bounded(request).await.unwrap().is_none()); +} + +#[tokio::test] +async fn analysis_commands_return_locations_edits_and_stable_prim_uris() { + let mut harness = Harness::new(Options { + capabilities: iris_workspace::AnalyzerCapabilities::default().with_change_annotations(), + ..Options::default() + }); + harness.open(ORIGINAL, 6); + let sequence = harness.configure(); + harness.ready(sequence).await; + + let (reply, request) = Reply::channel(); + let command = LanguageServer::References { + uri: Url::clone(&harness.uri), + position: Position::new(5, 7), + reply, + }; + harness.send(Command::LanguageServer(command)).unwrap(); + + let locations = bounded(request).await.unwrap().unwrap(); + let reference = locations + .iter() + .any(|location| location.uri == harness.uri && location.range.start.line == 5); + assert!(reference); + + let (reply, request) = Reply::channel(); + let command = LanguageServer::Rename { + uri: Url::clone(&harness.uri), + position: Position::new(5, 7), + new_name: "use".into(), + reply, + }; + harness.send(Command::LanguageServer(command)).unwrap(); + + let edit = bounded(request).await.unwrap().unwrap(); + let changes = match edit.document_changes.unwrap() { + lsp_types::DocumentChanges::Edits(edits) => edits, + lsp_types::DocumentChanges::Operations(operations) => { + let edits = operations.into_iter().filter_map(|operation| match operation { + lsp_types::DocumentChangeOperation::Edit(edit) => Some(edit), + _ => None, + }); + + edits.collect() + } + }; + + assert_eq!(changes.len(), 1); + assert_eq!(changes[0].text_document.version, Some(6)); + assert_eq!(changes[0].edits.len(), 3); + for edit in &changes[0].edits { + let edit = match edit { + lsp_types::OneOf::Left(edit) => edit, + lsp_types::OneOf::Right(edit) => &edit.text_edit, + }; + assert_eq!(edit.new_text, "use"); + } + + let (reply, request) = Reply::channel(); + let command = LanguageServer::SemanticTokens { uri: Url::clone(&harness.uri), reply }; + harness.send(Command::LanguageServer(command)).unwrap(); + + let tokens = bounded(request).await.unwrap().unwrap(); + assert!(!tokens.data.is_empty()); + let legend = iris_workspace::semantic_tokens_legend(); + let keyword = &tokens.data[0]; + assert_eq!(keyword.length, 6); + assert_eq!( + legend.token_types[keyword.token_type as usize], + lsp_types::SemanticTokenType::KEYWORD + ); + assert_eq!(legend.token_modifiers, vec![lsp_types::SemanticTokenModifier::DECLARATION]); + + async fn prim(harness: &Harness) -> Url { + let (reply, request) = Reply::channel(); + let command = LanguageServer::Definition { + uri: Url::clone(&harness.uri), + position: Position::new(2, 10), + reply, + }; + harness.send(Command::LanguageServer(command)).unwrap(); + + match bounded(request).await.unwrap().unwrap() { + lsp_types::GotoDefinitionResponse::Scalar(location) => location.uri, + lsp_types::GotoDefinitionResponse::Array(locations) => Url::clone(&locations[0].uri), + lsp_types::GotoDefinitionResponse::Link(locations) => { + Url::clone(&locations[0].target_uri) + } + } + } + + let first = prim(&harness).await; + assert!(first.to_file_path().unwrap().is_file()); + + let sequence = harness.send(Command::Reload).unwrap(); + harness.ready(sequence).await; + assert_eq!(first, prim(&harness).await); + assert!(first.to_file_path().unwrap().is_file()); +} + +#[tokio::test] +async fn discovery_preserves_root_and_literal_arguments_and_rejects_invalid_output() { + let mut harness = Harness::new(Options::default()); + fs::write(harness.directory.path().join("Ignored.purs"), "module Ignored where\nignored = 0\n") + .unwrap(); + fs::write( + harness.directory.path().join("sources.cjs"), + "require('fs').writeFileSync('arguments', process.argv[2]); console.log('Main.purs');", + ) + .unwrap(); + + let argument = "argument with spaces; $NOT_SHELL"; + let mut configuration = harness.configuration(); + configuration.settings.sources = SourceDiscovery::Command { + program: "node".into(), + arguments: vec!["sources.cjs".into(), argument.into()], + }; + let sequence = harness.send(Command::Configure(configuration)).unwrap(); + harness.ready(sequence).await; + + assert_eq!(fs::read_to_string(harness.directory.path().join("arguments")).unwrap(), argument); + + let (reply, request) = Reply::channel(); + let command = LanguageServer::WorkspaceSymbols { query: "ignored".into(), reply }; + harness.send(Command::LanguageServer(command)).unwrap(); + + let result = bounded(request).await.unwrap(); + assert!(match result { + None => true, + Some(lsp_types::WorkspaceSymbolResponse::Flat(symbols)) => symbols.is_empty(), + Some(lsp_types::WorkspaceSymbolResponse::Nested(symbols)) => symbols.is_empty(), + }); + + let mut configuration = harness.configuration(); + configuration.settings.sources = SourceDiscovery::Command { + program: "node".into(), + arguments: vec!["-e".into(), "process.stdout.write(Buffer.from([255]))".into()], + }; + harness.send(Command::Configure(configuration)).unwrap(); + + loop { + if let Event::StatusChanged(Status::Failed { message, .. }) = harness.next().await { + assert!(message.contains("UTF-8")); + break; + } + } +} + +async fn process_tree(leader_exits: bool) { + use tokio::io::{AsyncBufReadExt, AsyncReadExt, BufReader}; + + let mut harness = Harness::new(Options::default()); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + + let child = r#" +const address = { port: Number(process.argv[2]), host: '127.0.0.1' }; +const socket = require('net').connect(address, () => { + socket.write('ready\n'); + process.send('ready'); +}); + +setInterval(() => {}, 1000); +setTimeout(() => process.exit(0), 30000); +"#; + fs::write(harness.directory.path().join("child.cjs"), child).unwrap(); + + let completion = if leader_exits { + "console.log('Main.purs'); process.exit(0);" + } else { + "setInterval(() => {}, 1000);" + }; + let parent = format!( + r#" +const child = require('child_process').spawn( + process.execPath, + ['child.cjs', process.argv[2]], + {{ stdio: ['ignore', 'inherit', 'inherit', 'ipc'] }}, +); + +child.on('message', () => {{ {completion} }}); +setTimeout(() => process.exit(0), 30000); +"# + ); + fs::write(harness.directory.path().join("sources.cjs"), parent).unwrap(); + + let mut configuration = harness.configuration(); + configuration.settings.sources = SourceDiscovery::Command { + program: "node".into(), + arguments: vec!["sources.cjs".into(), port.to_string()], + }; + let sequence = harness.send(Command::Configure(configuration)).unwrap(); + + let (socket, _) = bounded(listener.accept()).await.unwrap(); + let mut reader = BufReader::new(socket); + let mut line = String::new(); + bounded(reader.read_line(&mut line)).await.unwrap(); + assert_eq!(line, "ready\n"); + + if leader_exits { + harness.ready(sequence).await; + assert!(harness.hover().await.contains("Int")); + } else { + harness.open(CHANGED, 1); + harness.send(Command::Shutdown).unwrap(); + assert!(matches!(harness.status(), Status::Stopping | Status::Stopped)); + loop { + if matches!(harness.next().await, Event::StatusChanged(Status::Stopped)) { + break; + } + } + } + + let mut remainder = vec![]; + assert_eq!(bounded(reader.read_to_end(&mut remainder)).await.unwrap(), 0); +} + +#[tokio::test] +async fn shutdown_reaps_source_command_descendants() { + process_tree(false).await; +} + +#[tokio::test] +async fn successful_discovery_reaps_descendants_after_leader_exit() { + process_tree(true).await; +} + +#[tokio::test] +async fn authoritative_source_and_foreign_buffers_bypass_unreadable_disk_content() { + let mut harness = Harness::new(Options::default()); + let source = "module Main where\nforeign import value :: Int\n"; + let foreign_path = harness.directory.path().join("Main.js"); + let foreign_uri = Url::from_file_path(&foreign_path).unwrap(); + fs::write(harness.directory.path().join("Main.purs"), [255]).unwrap(); + fs::write(&foreign_path, [255]).unwrap(); + + harness.open(source, 3); + let command = + Document::Open { uri: foreign_uri, text: "export const value = 1;".into(), version: 5 }; + harness.send(Command::Document(command)).unwrap(); + let sequence = harness.configure(); + harness.ready(sequence).await; + + let uri = Url::clone(&harness.uri); + assert!(diagnostics_for(&mut harness, &uri).await.is_empty()); + assert!(harness.symbols().await.contains(&"value".into())); + + let sequence = harness.send(Command::Reload).unwrap(); + harness.ready(sequence).await; + assert!(harness.symbols().await.contains(&"value".into())); +} + +#[tokio::test] +async fn closing_an_excluded_source_does_not_restore_it_from_disk() { + let mut harness = Harness::new(Options::default()); + harness.open(CHANGED, 1); + let sequence = harness.configure(); + harness.ready(sequence).await; + + let mut configuration = harness.configuration(); + configuration.settings.sources = SourceDiscovery::Command { + program: "node".into(), + arguments: vec!["-e".into(), "process.exit(0)".into()], + }; + let sequence = harness.send(Command::Configure(configuration)).unwrap(); + harness.ready(sequence).await; + assert!(harness.hover().await.contains("String")); + + harness.send(Command::Document(Document::Close(Url::clone(&harness.uri)))).unwrap(); + let sequence = + harness.send(Command::Document(Document::Save(Url::clone(&harness.uri)))).unwrap(); + harness.ready(sequence).await; + + let (reply, request) = Reply::channel(); + let command = LanguageServer::DocumentSymbols { uri: Url::clone(&harness.uri), reply }; + harness.send(Command::LanguageServer(command)).unwrap(); + assert!(bounded(request).await.unwrap().is_none()); + assert!(harness.uri.to_file_path().unwrap().is_file()); +} + +#[tokio::test] +async fn failed_disk_reconciliation_clears_previously_published_diagnostics() { + let mut harness = Harness::new(Options::default()); + harness.open("module Main where\nvalue :: Int\nvalue = \"wrong\"\n", 1); + let sequence = harness.configure(); + harness.ready(sequence).await; + + let uri = Url::clone(&harness.uri); + assert!(!diagnostics_for(&mut harness, &uri).await.is_empty()); + + fs::write(harness.directory.path().join("Main.purs"), [255]).unwrap(); + harness.send(Command::Document(Document::Close(Url::clone(&uri)))).unwrap(); + assert!(diagnostics_for(&mut harness, &uri).await.is_empty()); + + loop { + if matches!(harness.status(), Status::Failed { .. }) { + break; + } + harness.next().await; + } + + fs::write(harness.directory.path().join("Main.purs"), ORIGINAL).unwrap(); + let sequence = harness.send(Command::Reload).unwrap(); + harness.ready(sequence).await; + assert!(harness.hover().await.contains("Int")); +} + +#[tokio::test] +async fn request_admitted_behind_failure_is_rejected_without_another_input() { + let mut harness = Harness::new(Options::default()); + harness.open(ORIGINAL, 1); + let sequence = harness.configure(); + harness.ready(sequence).await; + + let mut failure = harness.hooks.pause_next(Point::BeforeFailure); + fs::write(harness.directory.path().join("Main.purs"), [255]).unwrap(); + harness.send(Command::Document(Document::Close(Url::clone(&harness.uri)))).unwrap(); + bounded(failure.entered()).await; + + let request = harness.hover_request(); + drop(failure); + + assert!(matches!(bounded(request).await, Err(RequestFailure::Unavailable))); + assert!(matches!(harness.status(), Status::Failed { .. })); +} + +#[tokio::test] +async fn startup_waits_for_multiple_acknowledgements_and_finishes_once() { + let mut harness = Harness::new(Options::default()); + let mut preparation = harness.hooks.pause_next(Point::BeforeAcknowledgement); + harness.configure(); + bounded(preparation.entered()).await; + + harness.open(ORIGINAL, 1); + let mut reconciliation = harness.hooks.pause_next(Point::BeforeAcknowledgement); + drop(preparation); + bounded(reconciliation.entered()).await; + assert!(matches!(harness.status(), Status::Rebuilding { .. })); + + let sequence = harness.change(CHANGED, 2); + drop(reconciliation); + let mut finished = Vec::new(); + loop { + match harness.next().await { + Event::Finished { outcome, .. } => finished.push(outcome), + Event::StatusChanged(Status::Ready { stamp, .. }) => { + assert_eq!(stamp.revision, sequence); + break; + } + _ => {} + } + } + assert!(harness.hover().await.contains("String")); + + harness.send(Command::Shutdown).unwrap(); + while let Some(delivery) = bounded(harness.events.recv()).await { + if let Ok(Event::Finished { outcome, .. }) = delivery.release() { + finished.push(outcome); + } + } + assert_eq!(finished, vec![Outcome::Ready]); +} + +#[tokio::test] +async fn shutdown_waits_for_cancelled_worker_and_ignores_its_late_acknowledgement() { + let mut harness = Harness::new(Options::default()); + let mut acknowledgement = harness.hooks.pause_next(Point::BeforeAcknowledgement); + harness.configure(); + bounded(acknowledgement.entered()).await; + + harness.send(Command::Shutdown).unwrap(); + assert_eq!(harness.status(), Status::Stopping); + assert!(matches!(harness.send(Command::Reload), Err(RequestFailure::Unavailable))); + drop(acknowledgement); + + let mut finished = Vec::new(); + while let Some(delivery) = bounded(harness.events.recv()).await { + let Ok(event) = delivery.release() else { + continue; + }; + assert!(!matches!(event, Event::StatusChanged(Status::Ready { .. }))); + if let Event::Finished { outcome, .. } = event { + finished.push(outcome); + } + } + assert_eq!(finished, vec![Outcome::Cancelled]); + assert_eq!(harness.status(), Status::Stopped); +} + +#[tokio::test] +async fn command_handles_and_cleanup_have_independent_ownership() { + let WorkspaceSession { workspace, mut events, join } = + Workspace::start(Options::default()).unwrap(); + let handle = Workspace::clone(&workspace); + drop(workspace); + + handle.send(Command::Reload).unwrap(); + let held = bounded(events.recv()).await.unwrap(); + handle.send(Command::Shutdown).unwrap(); + drop(held); + drop(events); + + bounded(tokio::task::spawn_blocking(move || join.join())).await.unwrap().unwrap(); + assert_eq!(handle.status(), Status::Stopped); + assert!(matches!(handle.send(Command::Reload), Err(RequestFailure::Unavailable))); +} + +#[tokio::test] +async fn progress_reports_a_phase_without_requiring_prior_events() { + let mut harness = Harness::new(Options::default()); + let mut acknowledgement = harness.hooks.pause_next(Point::BeforeAcknowledgement); + harness.configure(); + bounded(acknowledgement.entered()).await; + + loop { + if let Event::Progress { phase: iris_workspace::Phase::Reconciling, .. } = + harness.next().await + { + break; + } + } + drop(acknowledgement); +} + +async fn configuration_outcome( + harness: &mut Harness, + expected: InputSequence, +) -> iris_workspace::ConfigurationOutcome { + loop { + if let Event::ConfigurationFinished { sequence, outcome } = harness.next().await { + assert_eq!(sequence, expected); + return outcome; + } + } +} + +#[tokio::test] +async fn configurations_classify_changes_and_retry_the_same_failed_settings() { + use iris_workspace::ConfigurationOutcome; + + let mut harness = Harness::new(Options::default()); + let sequence = harness.configure(); + assert_eq!(configuration_outcome(&mut harness, sequence).await, ConfigurationOutcome::Rebuilt); + let initial = harness.ready(sequence).await; + + let sequence = harness.configure(); + assert_eq!( + configuration_outcome(&mut harness, sequence).await, + ConfigurationOutcome::Unchanged + ); + assert_eq!(harness.ready(sequence).await.incarnation, initial.incarnation); + + let mut policy = harness.configuration(); + policy.settings.diagnostics.on_change = true; + let sequence = harness.send(Command::Configure(policy)).unwrap(); + assert_eq!( + configuration_outcome(&mut harness, sequence).await, + ConfigurationOutcome::PolicyUpdated + ); + assert_eq!(harness.ready(sequence).await.incarnation, initial.incarnation); + + let source = "module Main where\nvalue :: Int\nvalue = \"wrong\"\n"; + let sequence = harness.open(source, 1); + harness.ready(sequence).await; + let uri = Url::clone(&harness.uri); + assert!(!diagnostics_for(&mut harness, &uri).await.is_empty()); + + let script = harness.directory.path().join("sources.cjs"); + fs::write(&script, "process.stderr.write('cannot discover'); process.exit(7);").unwrap(); + let mut configuration = harness.configuration(); + configuration.settings.sources = + SourceDiscovery::Command { program: "node".into(), arguments: vec!["sources.cjs".into()] }; + let sequence = + harness.send(Command::Configure(ConfigurationInput::clone(&configuration))).unwrap(); + let ConfigurationOutcome::Failed { message } = + configuration_outcome(&mut harness, sequence).await + else { + panic!("discovery must fail"); + }; + assert!(message.contains("cannot discover")); + + fs::write(script, "console.log('Main.purs');").unwrap(); + let sequence = harness.send(Command::Configure(configuration)).unwrap(); + assert_eq!(configuration_outcome(&mut harness, sequence).await, ConfigurationOutcome::Rebuilt); + assert_ne!(harness.ready(sequence).await.incarnation, initial.incarnation); + assert!(!diagnostics_for(&mut harness, &uri).await.is_empty()); + + harness.send(Command::Shutdown).unwrap(); + while let Some(delivery) = bounded(harness.events.recv()).await { + assert!(!matches!(delivery.release(), Ok(Event::ConfigurationFinished { .. }))); + } +} + +#[tokio::test] +async fn pending_configurations_each_finish_even_without_starting_preparation() { + use iris_workspace::ConfigurationOutcome; + + let mut harness = Harness::new(Options::default()); + let initial = harness.configure(); + assert_eq!(configuration_outcome(&mut harness, initial).await, ConfigurationOutcome::Rebuilt); + harness.ready(initial).await; + + let mut analysis = harness.hooks.pause_next(Point::BeforeAnalysis); + let request = harness.hover_request(); + bounded(analysis.entered()).await; + + let mut configuration = harness.configuration(); + configuration.settings.sources = SourceDiscovery::Command { + program: "node".into(), + arguments: vec!["-e".into(), "console.log('Main.*')".into()], + }; + let first = harness.send(Command::Configure(configuration)).unwrap(); + let second = harness.configure(); + harness.send(Command::Shutdown).unwrap(); + + assert_eq!(configuration_outcome(&mut harness, first).await, ConfigurationOutcome::Superseded); + assert_eq!(configuration_outcome(&mut harness, second).await, ConfigurationOutcome::Cancelled); + drop(analysis); + assert!(matches!(bounded(request).await, Err(RequestFailure::Cancelled))); + + while let Some(delivery) = bounded(harness.events.recv()).await { + assert!(!matches!(delivery.release(), Ok(Event::ConfigurationFinished { .. }))); + } +} + +#[tokio::test] +async fn running_preparation_reports_supersession_once() { + use iris_workspace::ConfigurationOutcome; + + let mut harness = Harness::new(Options::default()); + let mut acknowledgement = harness.hooks.pause_next(Point::BeforeAcknowledgement); + let first = harness.configure(); + bounded(acknowledgement.entered()).await; + + let second = harness.configure(); + assert_eq!(configuration_outcome(&mut harness, first).await, ConfigurationOutcome::Superseded); + drop(acknowledgement); + assert_eq!(configuration_outcome(&mut harness, second).await, ConfigurationOutcome::Rebuilt); + + harness.send(Command::Shutdown).unwrap(); + while let Some(delivery) = bounded(harness.events.recv()).await { + assert!(!matches!(delivery.release(), Ok(Event::ConfigurationFinished { .. }))); + } +} + +#[tokio::test] +async fn policy_updates_preserve_pending_diagnostics_and_control_future_edits() { + let mut harness = Harness::new(Options::default()); + let mut diagnostics = harness.hooks.pause_next(Point::BeforeDiagnostics); + let initial = harness.configure(); + configuration_outcome(&mut harness, initial).await; + bounded(diagnostics.entered()).await; + + let mut policy = harness.configuration(); + policy.settings.diagnostics.on_open = false; + policy.settings.diagnostics.on_change = true; + let sequence = harness.send(Command::Configure(policy)).unwrap(); + assert_eq!( + configuration_outcome(&mut harness, sequence).await, + iris_workspace::ConfigurationOutcome::PolicyUpdated + ); + drop(diagnostics); + + let uri = Url::clone(&harness.uri); + assert!(diagnostics_for(&mut harness, &uri).await.is_empty()); + harness.open(ORIGINAL, 1); + harness.change("module Main where\nvalue :: Int\nvalue = \"wrong\"\n", 2); + assert!(!diagnostics_for(&mut harness, &uri).await.is_empty()); +} + +#[tokio::test] +async fn unsupported_uri_aliases_are_rejected_without_disturbing_open_authority() { + use iris_build::analysis::{AnalysisConfig, AnalysisError, AnalysisOverlay, CancellationToken}; + use iris_build::compilation::MaterializedPrim; + use iris_build::events::SilentBuildEvents; + use std::sync::Arc; + + let mut harness = Harness::new(Options::default()); + harness.open(CHANGED, 1); + let sequence = harness.configure(); + harness.ready(sequence).await; + + let prim = Arc::new(MaterializedPrim::new().unwrap()); + let configuration = AnalysisConfig { + root: harness.directory.path().to_path_buf(), + sources: SourceDiscovery::default(), + }; + let mut aliases = ["?view=1", "#selection", "?view=1#selection", "?", "#"] + .map(|suffix| Url::parse(&format!("{}{suffix}", harness.uri)).unwrap()) + .to_vec(); + aliases.push(Url::parse(&harness.uri.as_str().replace("Main.purs", "%4dain.purs")).unwrap()); + + for uri in aliases { + let document = Document::Open { uri: Url::clone(&uri), text: ORIGINAL.into(), version: 2 }; + let sequence = harness.send(Command::Document(document)).unwrap(); + loop { + if let Event::InputRejected { sequence: rejected, failure } = harness.next().await { + assert_eq!(sequence, rejected); + assert_eq!(failure, InputFailure::UnsupportedDocument(Url::clone(&uri))); + break; + } + } + harness.ready(sequence).await; + assert!(harness.hover().await.contains("String")); + + assert!(matches!( + harness.send(Command::FilesChanged(vec![Url::clone(&uri)])), + Err(RequestFailure::InvalidInput(_)) + )); + let (reply, request) = Reply::channel(); + let command = LanguageServer::DocumentSymbols { uri: Url::clone(&uri), reply }; + assert!(matches!( + harness.send(Command::LanguageServer(command)), + Err(RequestFailure::InvalidInput(_)) + )); + assert!(matches!(bounded(request).await, Err(RequestFailure::InvalidInput(_)))); + + let overlay = AnalysisOverlay { uri: Url::clone(&uri), text: ORIGINAL.into(), version: 1 }; + let result = iris_build::analysis::prepare( + &configuration, + &[overlay], + Arc::clone(&prim), + &CancellationToken::new(), + &SilentBuildEvents, + ); + assert!( + matches!(result, Err(AnalysisError::UnsupportedDocument(rejected)) if rejected == uri) + ); + } +} + +#[tokio::test] +async fn encoded_file_uris_keep_importers_and_rename_on_the_open_buffer() { + let mut harness = Harness::new(Options::default()); + let importer = "module Main where\nimport Library\nuse :: String\nuse = value\n"; + fs::write(harness.directory.path().join("Main.purs"), importer).unwrap(); + let path = harness.directory.path().join("Library Source.purs"); + fs::write(&path, "module Library where\nvalue :: Int\nvalue = 1\n").unwrap(); + let uri = Url::from_file_path(&path).unwrap(); + assert!(uri.as_str().contains("%20")); + let localhost = Url::parse(&uri.as_str().replacen("file:///", "file://localhost/", 1)).unwrap(); + assert_eq!(iris_build::analysis::document_path(&localhost), Some(path)); + + let source = "module Library where\nvalue :: String\nvalue = \"buffer\"\n"; + harness + .send(Command::Document(Document::Open { + uri: Url::clone(&localhost), + text: source.into(), + version: 4, + })) + .unwrap(); + let sequence = harness.configure(); + configuration_outcome(&mut harness, sequence).await; + let main = Url::clone(&harness.uri); + assert!(diagnostics_for(&mut harness, &main).await.is_empty()); + + for command in [Command::Reload, Command::FilesChanged(vec![Url::clone(&uri)])] { + let sequence = harness.send(command).unwrap(); + harness.ready(sequence).await; + assert!(diagnostics_for(&mut harness, &main).await.is_empty()); + + let (reply, request) = Reply::channel(); + let command = LanguageServer::Rename { + uri: Url::clone(&uri), + position: Position::new(2, 1), + new_name: "renamed".into(), + reply, + }; + harness.send(Command::LanguageServer(command)).unwrap(); + let edit = bounded(request).await.unwrap().unwrap(); + let edits = edit.changes.unwrap(); + assert_eq!(edits.len(), 2); + assert!(edits.contains_key(&uri)); + assert!(edits.contains_key(&main)); + } + + let sequence = harness.send(Command::Document(Document::Close(uri))).unwrap(); + harness.ready(sequence).await; + assert!(!diagnostics_for(&mut harness, &main).await.is_empty()); +} + +#[tokio::test] +async fn closing_sources_reobserves_deleted_and_changed_foreign_siblings() { + for extension in ["js", "jsx"] { + let mut harness = Harness::new(Options::default()); + let source = "module Main where\nforeign import value :: Int\n"; + fs::write(harness.directory.path().join("Main.purs"), source).unwrap(); + let sibling = harness.directory.path().join(format!("Main.{extension}")); + fs::write(&sibling, "export const value = 1;").unwrap(); + harness.open(source, 1); + let sequence = harness.configure(); + configuration_outcome(&mut harness, sequence).await; + let uri = Url::clone(&harness.uri); + assert!(diagnostics_for(&mut harness, &uri).await.is_empty()); + + fs::remove_file(&sibling).unwrap(); + let sequence = harness.send(Command::Document(Document::Close(Url::clone(&uri)))).unwrap(); + harness.ready(sequence).await; + assert!(!diagnostics_for(&mut harness, &uri).await.is_empty()); + + fs::write(&sibling, "export const value = 1;").unwrap(); + let sequence = harness.open(source, 2); + harness.ready(sequence).await; + assert!(diagnostics_for(&mut harness, &uri).await.is_empty()); + + fs::write(&sibling, "export const other = 1;").unwrap(); + let sequence = harness.send(Command::Document(Document::Close(Url::clone(&uri)))).unwrap(); + harness.ready(sequence).await; + assert!(!diagnostics_for(&mut harness, &uri).await.is_empty()); + + let sequence = harness.open(source, 3); + harness.ready(sequence).await; + assert!(!diagnostics_for(&mut harness, &uri).await.is_empty()); + + fs::write(&sibling, [255]).unwrap(); + harness.send(Command::Document(Document::Close(Url::clone(&uri)))).unwrap(); + loop { + if let Event::StatusChanged(Status::Failed { message, .. }) = harness.next().await { + assert!(message.contains(&format!("Main.{extension}"))); + break; + } + } + assert!(matches!(harness.status(), Status::Failed { .. })); + } +}