@@ -1,7 +1,10 @@
11use std::collections::{HashMap, HashSet};
22use std::path::{Path, PathBuf};
3−use std::time::SystemTime;
3+use std::sync::{Arc, Mutex};
4+use std::thread;
5+use std::time::{Duration, Instant, SystemTime};
46
7+use crossbeam_channel::{select, Receiver, RecvTimeoutError, Sender};
58use lsp_server::{Connection, Message, Notification, Request, RequestId, Response};
69use lsp_types::notification::Notification as _;
710use lsp_types::*;
@@ -43,10 +46,37 @@ struct ServerState {
4346 documents: HashMap<Uri, DocumentState>,
4447 workspace_root: Option<PathBuf>,
4548 /// Cached parsed imports: canonical path → (mtime, module, source text).
46− import_cache: HashMap<PathBuf, (SystemTime, Module, String)>,
49+ /// Shared with the analysis worker thread.
50+ import_cache: Arc<Mutex<HashMap<PathBuf, (SystemTime, Module, String)>>>,
4751 /// Cached LSP diagnostics for unopened workspace files, keyed by mtime.
4852 /// Refreshed lazily; stale entries are dropped when mtime advances.
4953 workspace_diag_cache: HashMap<PathBuf, (SystemTime, Vec<Diagnostic>)>,
54+ /// Edited but not-yet-analyzed sources. When present, this is the source
55+ /// the next analysis run will see. Subsequent didChange edits stack on top
56+ /// of this rather than the (stale) analyzed source.
57+ pending_sources: HashMap<Uri, PendingSource>,
58+ /// Sender side of the analysis-task channel. Cloned per outgoing task.
59+ analysis_tx: Sender<AnalysisTask>,
60+}
61+
62+struct PendingSource {
63+ source: String,
64+ /// Latest LSP version observed for this URI (`None` for didOpen).
65+ version: Option<i32>,
66+}
67+
68+/// Work item handed to the analysis worker.
69+struct AnalysisTask {
70+ uri: Uri,
71+ source: String,
72+ version: Option<i32>,
73+}
74+
75+/// Output from the analysis worker.
76+struct AnalysisResult {
77+ uri: Uri,
78+ version: Option<i32>,
79+ doc: DocumentState,
5080}
5181
5282// ── Semantic token legend ───────────────────────────────────────────
@@ -208,11 +238,25 @@ fn main() {
208238 })
209239 .and_then(|u| uri_to_path(&u));
210240
241+ // Spawn the analysis worker. It owns no per-request state of its own —
242+ // the import cache is shared (Arc<Mutex>) so the main thread can read it
243+ // for auto-import completion suggestions without a round trip.
244+ let import_cache = Arc::new(Mutex::new(HashMap::new()));
245+ let (analysis_tx, analysis_rx) = crossbeam_channel::unbounded::<AnalysisTask>();
246+ let (results_tx, results_rx) = crossbeam_channel::unbounded::<AnalysisResult>();
247+ let worker_cache = Arc::clone(&import_cache);
248+ let worker = thread::Builder::new()
249+ .name("knot-lsp-analysis".into())
250+ .spawn(move || analysis_worker(analysis_rx, results_tx, worker_cache))
251+ .expect("failed to spawn analysis worker");
252+
211253 let mut state = ServerState {
212254 documents: HashMap::new(),
213255 workspace_root,
214− import_cache: HashMap::new(),
256+ import_cache,
215257 workspace_diag_cache: HashMap::new(),
258+ pending_sources: HashMap::new(),
259+ analysis_tx,
216260 };
217261
218262 // Register for file watcher notifications (.knot files)
@@ -238,24 +282,124 @@ fn main() {
238282 .unwrap(),
239283 )));
240284
241− for msg in &connection.receiver {
242− match msg {
243− Message::Request(req) => {
244− if connection.handle_shutdown(&req).unwrap_or(false) {
245− return;
285+ 'outer: loop {
286+ select! {
287+ recv(connection.receiver) -> msg => {
288+ let msg = match msg {
289+ Ok(m) => m,
290+ Err(_) => break 'outer,
291+ };
292+ match msg {
293+ Message::Request(req) => {
294+ if connection.handle_shutdown(&req).unwrap_or(false) {
295+ break 'outer;
296+ }
297+ handle_request(&mut state, &connection, req);
298+ }
299+ Message::Notification(not) => {
300+ handle_notification(&mut state, &connection, not);
301+ }
302+ Message::Response(_) => {}
246303 }
247− handle_request(&mut state, &connection, req);
248304 }
249− Message::Notification(not) => {
250− handle_notification(&mut state, &connection, not);
305+ recv(results_rx) -> result => {
306+ let result = match result {
307+ Ok(r) => r,
308+ Err(_) => break 'outer,
309+ };
310+ apply_analysis_result(&mut state, &connection, result);
251311 }
252− Message::Response(_) => {}
253312 }
254313 }
255314
315+ // Dropping `state` (and thus `analysis_tx`) closes the channel, prompting
316+ // the worker to exit on its next blocking `recv`.
317+ drop(state);
318+ let _ = worker.join();
256319 io_threads.join().unwrap();
257320}
258321
322+// ── Analysis worker ─────────────────────────────────────────────────
323+
324+/// Quiet period after the most recent task before processing begins.
325+const ANALYSIS_DEBOUNCE: Duration = Duration::from_millis(150);
326+/// Hard upper bound on how long a task can sit in the debounce queue. Prevents
327+/// continuous typing from indefinitely starving the analysis pass.
328+const ANALYSIS_MAX_WAIT: Duration = Duration::from_millis(500);
329+
330+fn analysis_worker(
331+ rx: Receiver<AnalysisTask>,
332+ tx: Sender<AnalysisResult>,
333+ import_cache: Arc<Mutex<HashMap<PathBuf, (SystemTime, Module, String)>>>,
334+) {
335+ loop {
336+ // Block until something arrives.
337+ let first = match rx.recv() {
338+ Ok(t) => t,
339+ Err(_) => return,
340+ };
341+
342+ // Coalesce: keep the latest task per URI within a debounce window.
343+ let mut pending: HashMap<Uri, AnalysisTask> = HashMap::new();
344+ pending.insert(first.uri.clone(), first);
345+
346+ let batch_start = Instant::now();
347+ loop {
348+ let elapsed = batch_start.elapsed();
349+ if elapsed >= ANALYSIS_MAX_WAIT {
350+ break;
351+ }
352+ let timeout = ANALYSIS_DEBOUNCE.min(ANALYSIS_MAX_WAIT - elapsed);
353+ match rx.recv_timeout(timeout) {
354+ Ok(task) => {
355+ pending.insert(task.uri.clone(), task);
356+ }
357+ Err(RecvTimeoutError::Timeout) => break,
358+ Err(RecvTimeoutError::Disconnected) => return,
359+ }
360+ }
361+
362+ for (_, task) in pending {
363+ let mut cache = import_cache.lock().unwrap();
364+ let doc = analyze_document(&task.uri, &task.source, &mut cache);
365+ drop(cache);
366+ if tx
367+ .send(AnalysisResult {
368+ uri: task.uri,
369+ version: task.version,
370+ doc,
371+ })
372+ .is_err()
373+ {
374+ return;
375+ }
376+ }
377+ }
378+}
379+
380+/// Apply a finished analysis: replace the document state and publish diagnostics.
381+fn apply_analysis_result(state: &mut ServerState, conn: &Connection, result: AnalysisResult) {
382+ // Document closed while analysis was in flight: drop the result rather
383+ // than resurrecting a removed document.
384+ let tracked = state.documents.contains_key(&result.uri)
385+ || state.pending_sources.contains_key(&result.uri);
386+ if !tracked {
387+ return;
388+ }
389+
390+ // If a newer edit was applied while analysis was running, drop the result.
391+ // The newer edit will already have queued a fresh task.
392+ if let Some(pending) = state.pending_sources.get(&result.uri) {
393+ if pending.source != result.doc.source {
394+ return;
395+ }
396+ state.pending_sources.remove(&result.uri);
397+ }
398+
399+ publish_diagnostics(conn, &result.uri, &result.doc, result.version);
400+ state.documents.insert(result.uri, result.doc);
401+}
402+
259403// ── Request dispatch ────────────────────────────────────────────────
260404
261405fn handle_request(state: &mut ServerState, conn: &Connection, req: Request) {
@@ -370,67 +514,82 @@ fn handle_notification(state: &mut ServerState, conn: &Connection, not: Notifica
370514 if not.method == notification::DidOpenTextDocument::METHOD {
371515 let params: DidOpenTextDocumentParams = serde_json::from_value(not.params).unwrap();
372516 let uri = params.text_document.uri.clone();
373− let doc = analyze_document(&uri, ¶ms.text_document.text, &mut state.import_cache);
374− publish_diagnostics(conn, &uri, &doc);
375− state.documents.insert(uri, doc);
517+ let version = Some(params.text_document.version);
518+ // Park the source as pending until analysis catches up. This lets
519+ // subsequent didChange edits stack on top of the freshest text even
520+ // before the worker has produced its first AST.
521+ state.pending_sources.insert(
522+ uri.clone(),
523+ PendingSource {
524+ source: params.text_document.text.clone(),
525+ version,
526+ },
527+ );
528+ queue_analysis(state, uri, params.text_document.text, version);
376529 } else if not.method == notification::DidChangeTextDocument::METHOD {
377530 let params: DidChangeTextDocumentParams = serde_json::from_value(not.params).unwrap();
378531 let uri = params.text_document.uri.clone();
379− // Get current source or start empty
532+ let version = Some(params.text_document.version);
533+ // Apply edits to the freshest source we know about: the pending one
534+ // (if mid-debounce) or the last analyzed one. Falling back to the
535+ // analyzed source could lose interleaved edits, so prefer pending.
380536 let mut source = state
381− .documents
537+ .pending_sources
382538 .get(&uri)
383− .map(|d| d.source.clone())
539+ .map(|p| p.source.clone())
540+ .or_else(|| state.documents.get(&uri).map(|d| d.source.clone()))
384541 .unwrap_or_default();
385− // Apply incremental edits (or full replacement if range is None)
386542 for change in params.content_changes {
387543 if let Some(range) = change.range {
388544 let start = position_to_offset(&source, range.start);
389545 let end = position_to_offset(&source, range.end);
390546 source.replace_range(start..end, &change.text);
391547 } else {
392− // Full document replacement (fallback)
393548 source = change.text;
394549 }
395550 }
396− // Skip re-analysis if the source matches what we already analyzed.
397− // Triggered when a no-op edit fires (e.g. format-on-save with no changes).
551+
552+ // No-op edits (e.g. format-on-save with no changes) shouldn't queue
553+ // redundant work. The pending check covers an edit that was already
554+ // queued; the analyzed check covers a steady state.
555+ let already_pending = state
556+ .pending_sources
557+ .get(&uri)
558+ .map(|p| p.source == source)
559+ .unwrap_or(false);
398560 let unchanged = state
399561 .documents
400562 .get(&uri)
401563 .map(|d| d.source == source)
402564 .unwrap_or(false);
403− if !unchanged {
404− let doc = analyze_document(&uri, &source, &mut state.import_cache);
405− publish_diagnostics(conn, &uri, &doc);
406− state.documents.insert(uri.clone(), doc);
565+ if already_pending {
566+ // Just refresh the version so the result-routing check in
567+ // `apply_analysis_result` keeps working.
568+ if let Some(p) = state.pending_sources.get_mut(&uri) {
569+ p.version = version;
570+ }
571+ return;
572+ }
573+ if unchanged {
574+ state.pending_sources.remove(&uri);
575+ return;
407576 }
408577
409− // Re-analyze open documents that import from the changed file
578+ state.pending_sources.insert(
579+ uri.clone(),
580+ PendingSource {
581+ source: source.clone(),
582+ version,
583+ },
584+ );
585+ queue_analysis(state, uri.clone(), source, version);
586+
587+ // When a file changes, dependents may need a fresh pass too.
410588 if let Some(changed_path) = uri_to_path(&uri).and_then(|p| p.canonicalize().ok()) {
411− let uris_to_refresh: Vec<(Uri, String)> = state
412− .documents
413− .iter()
414− .filter(|(other_uri, other_doc)| {
415− **other_uri != uri
416− && other_doc
417− .imported_files
418− .keys()
419− .any(|p| *p == changed_path)
420− })
421− .map(|(u, d)| (u.clone(), d.source.clone()))
422− .collect();
423− for (refresh_uri, refresh_source) in uris_to_refresh {
424− let doc =
425− analyze_document(&refresh_uri, &refresh_source, &mut state.import_cache);
426− publish_diagnostics(conn, &refresh_uri, &doc);
427− state.documents.insert(refresh_uri, doc);
428− }
589+ queue_dependent_analyses(state, &uri, &changed_path);
429590 }
430591 } else if not.method == notification::DidChangeWatchedFiles::METHOD {
431592 let params: DidChangeWatchedFilesParams = serde_json::from_value(not.params).unwrap();
432− // When .knot files change on disk, re-analyze open documents that might
433− // import them (their cross-file navigation data may be stale)
434593 let changed_paths: HashSet<PathBuf> = params
435594 .changes
436595 .iter()
@@ -439,8 +598,7 @@ fn handle_notification(state: &mut ServerState, conn: &Connection, not: Notifica
439598 .collect();
440599
441600 if !changed_paths.is_empty() {
442− // Re-analyze all open documents whose imports overlap with changed files
443− let uris_to_refresh: Vec<(Uri, String)> = state
601+ let dependents: Vec<(Uri, String)> = state
444602 .documents
445603 .iter()
446604 .filter(|(_, doc)| {
@@ -448,18 +606,24 @@ fn handle_notification(state: &mut ServerState, conn: &Connection, not: Notifica
448606 .keys()
449607 .any(|p| changed_paths.contains(p))
450608 })
451− .map(|(uri, doc)| (uri.clone(), doc.source.clone()))
609+ .map(|(uri, doc)| {
610+ let src = state
611+ .pending_sources
612+ .get(uri)
613+ .map(|p| p.source.clone())
614+ .unwrap_or_else(|| doc.source.clone());
615+ (uri.clone(), src)
616+ })
452617 .collect();
453618
454− for (uri, source) in uris_to_refresh {
455− let doc = analyze_document(&uri, &source, &mut state.import_cache);
456− publish_diagnostics(conn, &uri, &doc);
457− state.documents.insert(uri, doc);
619+ for (dep_uri, dep_source) in dependents {
620+ queue_analysis(state, dep_uri, dep_source, None);
458621 }
459622 }
460623 } else if not.method == notification::DidCloseTextDocument::METHOD {
461624 let params: DidCloseTextDocumentParams = serde_json::from_value(not.params).unwrap();
462625 state.documents.remove(¶ms.text_document.uri);
626+ state.pending_sources.remove(¶ms.text_document.uri);
463627 // Clear diagnostics
464628 let diags = PublishDiagnosticsParams::new(params.text_document.uri, vec![], None);
465629 let not = Notification::new(notification::PublishDiagnostics::METHOD.into(), diags);
@@ -467,6 +631,38 @@ fn handle_notification(state: &mut ServerState, conn: &Connection, not: Notifica
467631 }
468632}
469633
634+/// Send an analysis task to the worker. Errors here are unrecoverable — the
635+/// worker has died — so we eprintln and continue (other features still work
636+/// against the last good analysis).
637+fn queue_analysis(state: &ServerState, uri: Uri, source: String, version: Option<i32>) {
638+ if let Err(e) = state.analysis_tx.send(AnalysisTask { uri, source, version }) {
639+ eprintln!("knot-lsp: analysis worker channel closed: {e}");
640+ }
641+}
642+
643+/// Re-queue analysis for any open document whose imports point at `changed_path`.
644+fn queue_dependent_analyses(state: &mut ServerState, source_uri: &Uri, changed_path: &Path) {
645+ let dependents: Vec<(Uri, String)> = state
646+ .documents
647+ .iter()
648+ .filter(|(other_uri, other_doc)| {
649+ *other_uri != source_uri
650+ && other_doc.imported_files.keys().any(|p| p == changed_path)
651+ })
652+ .map(|(u, d)| {
653+ let src = state
654+ .pending_sources
655+ .get(u)
656+ .map(|p| p.source.clone())
657+ .unwrap_or_else(|| d.source.clone());
658+ (u.clone(), src)
659+ })
660+ .collect();
661+ for (dep_uri, dep_source) in dependents {
662+ queue_analysis(state, dep_uri, dep_source, None);
663+ }
664+}
665+
470666// ── Document analysis ───────────────────────────────────────────────
471667
472668fn analyze_document(
@@ -683,14 +879,14 @@ fn resolve_import_navigation(
683879 (imported_files, import_defs, import_origins)
684880}
685881
686−fn publish_diagnostics(conn: &Connection, uri: &Uri, doc: &DocumentState) {
882+fn publish_diagnostics(conn: &Connection, uri: &Uri, doc: &DocumentState, version: Option<i32>) {
687883 let lsp_diags: Vec<Diagnostic> = doc
688884 .knot_diagnostics
689885 .iter()
690886 .filter_map(|d| to_lsp_diagnostic(d, &doc.source, uri))
691887 .collect();
692888
693− let params = PublishDiagnosticsParams::new(uri.clone(), lsp_diags, None);
889+ let params = PublishDiagnosticsParams::new(uri.clone(), lsp_diags, version);
694890 let not = Notification::new(
695891 notification::PublishDiagnostics::METHOD.into(),
696892 params,
@@ -1796,8 +1992,14 @@ fn handle_completion(
17961992
17971993 // Reuse the cached parsed module if available (populated by
17981994 // resolve_import_navigation when other files have imported it).
1799− let module = if let Some((_, cached_mod, _)) = state.import_cache.get(&canonical) {
1800− cached_mod.clone()
1995+ let cached_module = state
1996+ .import_cache
1997+ .lock()
1998+ .unwrap()
1999+ .get(&canonical)
2000+ .map(|(_, m, _)| m.clone());
2001+ let module = if let Some(m) = cached_module {
2002+ m
18012003 } else {
18022004 let source_text = match std::fs::read_to_string(&canonical) {
18032005 Ok(s) => s,
@@ -7321,3 +7523,107 @@ fn error_code_for_diagnostic(message: &str) -> Option<String> {
73217523 None
73227524 }
73237525}
7526+
7527+// ── Tests ───────────────────────────────────────────────────────────
7528+
7529+#[cfg(test)]
7530+mod tests {
7531+ use super::*;
7532+
7533+ fn fake_uri(s: &str) -> Uri {
7534+ s.parse().unwrap()
7535+ }
7536+
7537+ /// Spin up the analysis worker and feed it tasks. Used to verify the
7538+ /// debounce/coalesce behavior without going through stdio.
7539+ fn spawn_worker() -> (Sender<AnalysisTask>, Receiver<AnalysisResult>, thread::JoinHandle<()>) {
7540+ let (tx, rx) = crossbeam_channel::unbounded::<AnalysisTask>();
7541+ let (rtx, rrx) = crossbeam_channel::unbounded::<AnalysisResult>();
7542+ let cache = Arc::new(Mutex::new(HashMap::new()));
7543+ let handle = thread::spawn(move || analysis_worker(rx, rtx, cache));
7544+ (tx, rrx, handle)
7545+ }
7546+
7547+ #[test]
7548+ fn worker_returns_result_for_single_task() {
7549+ let (tx, rx, handle) = spawn_worker();
7550+ let uri = fake_uri("file:///tmp/a.knot");
7551+ tx.send(AnalysisTask {
7552+ uri: uri.clone(),
7553+ source: "x = 1".into(),
7554+ version: Some(1),
7555+ })
7556+ .unwrap();
7557+
7558+ let result = rx
7559+ .recv_timeout(Duration::from_secs(5))
7560+ .expect("worker should produce a result");
7561+ assert_eq!(result.uri, uri);
7562+ assert_eq!(result.version, Some(1));
7563+ assert_eq!(result.doc.source, "x = 1");
7564+
7565+ drop(tx);
7566+ handle.join().unwrap();
7567+ }
7568+
7569+ #[test]
7570+ fn worker_coalesces_rapid_edits_to_latest() {
7571+ let (tx, rx, handle) = spawn_worker();
7572+ let uri = fake_uri("file:///tmp/b.knot");
7573+ // Three rapid edits within the debounce window — only the latest
7574+ // should turn into a result.
7575+ for (i, src) in ["x = 1", "x = 2", "x = 3"].iter().enumerate() {
7576+ tx.send(AnalysisTask {
7577+ uri: uri.clone(),
7578+ source: (*src).into(),
7579+ version: Some(i as i32 + 1),
7580+ })
7581+ .unwrap();
7582+ }
7583+
7584+ let result = rx
7585+ .recv_timeout(Duration::from_secs(5))
7586+ .expect("worker should produce a result");
7587+ assert_eq!(result.doc.source, "x = 3");
7588+ assert_eq!(result.version, Some(3));
7589+
7590+ // No follow-up result should arrive — the earlier two were dropped.
7591+ assert!(rx.recv_timeout(Duration::from_millis(300)).is_err());
7592+
7593+ drop(tx);
7594+ handle.join().unwrap();
7595+ }
7596+
7597+ #[test]
7598+ fn worker_keeps_distinct_uris_separate() {
7599+ let (tx, rx, handle) = spawn_worker();
7600+ let a = fake_uri("file:///tmp/a.knot");
7601+ let b = fake_uri("file:///tmp/b.knot");
7602+ tx.send(AnalysisTask {
7603+ uri: a.clone(),
7604+ source: "x = 1".into(),
7605+ version: Some(1),
7606+ })
7607+ .unwrap();
7608+ tx.send(AnalysisTask {
7609+ uri: b.clone(),
7610+ source: "y = 2".into(),
7611+ version: Some(1),
7612+ })
7613+ .unwrap();
7614+
7615+ let mut got = Vec::new();
7616+ for _ in 0..2 {
7617+ let r = rx
7618+ .recv_timeout(Duration::from_secs(5))
7619+ .expect("worker should produce two results");
7620+ got.push((r.uri, r.doc.source));
7621+ }
7622+ got.sort_by(|x, y| x.0.as_str().cmp(y.0.as_str()));
7623+ assert_eq!(got[0], (a, "x = 1".into()));
7624+ assert_eq!(got[1], (b, "y = 2".into()));
7625+
7626+ drop(tx);
7627+ handle.join().unwrap();
7628+ }
7629+}