lattice_lsp/
pending.rs

1//! `Pending<T>` -- the typed handle returned by every actor-side
2//! request. Mirrors `lattice_runtime::Pending` (DESIGN.md §5.2.1)
3//! but parameterised over [`LspError`] so server-side error codes
4//! survive on the way back to the caller.
5//!
6//! Three usage patterns, identical to the runtime's `Pending`:
7//!
8//! 1. **Async caller** (the editor's input pipeline): `pending.await`.
9//! 2. **Sync caller in tokio** (`#[tokio::test]`): same.
10//! 3. **Sync caller outside tokio** (the TUI input loop):
11//!    `pending.blocking_recv()` parks the current thread.
12
13use std::fmt;
14use std::sync::atomic::{AtomicU64, Ordering};
15
16use tokio::sync::oneshot;
17
18use crate::error::{LspError, LspResult};
19
20/// Process-monotonic id for an LSP request -- distinct from the
21/// JSON-RPC `RequestId` on the wire (which is per-server). Useful
22/// for telemetry: an `InvocationId` survives a server restart
23/// while the wire id resets.
24#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
25pub struct InvocationId(pub u64);
26
27impl InvocationId {
28    /// Allocate the next id. Lock-free; never collides with the
29    /// runtime's `InvocationId` because the two crates use
30    /// independent counters.
31    pub fn next() -> Self {
32        static SEQ: AtomicU64 = AtomicU64::new(1);
33        Self(SEQ.fetch_add(1, Ordering::Relaxed))
34    }
35}
36
37impl fmt::Display for InvocationId {
38    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
39        write!(f, "lsp#{}", self.0)
40    }
41}
42
43/// One in-flight LSP request awaiting a response.
44///
45/// `Pending` is neither `Clone` nor `Copy`: the underlying
46/// oneshot is single-use, matching the "one response per request"
47/// LSP contract. Dropping a `Pending` cancels the wait but does
48/// not interrupt the actor; callers that need true cancellation
49/// pass a [`lattice_runtime::CancellationToken`] alongside the
50/// request and the actor sends `$/cancelRequest` to the server.
51#[must_use = "the LSP response is dropped if the Pending is not awaited or block_on'd"]
52pub struct Pending<T> {
53    pub id: InvocationId,
54    rx: oneshot::Receiver<LspResult<T>>,
55}
56
57impl<T> Pending<T> {
58    pub(crate) fn new(id: InvocationId, rx: oneshot::Receiver<LspResult<T>>) -> Self {
59        Self { id, rx }
60    }
61
62    /// Eagerly resolved error -- used at construction sites where
63    /// the request never reaches the actor (e.g. mailbox closed).
64    pub(crate) fn ready_err(err: LspError) -> Self {
65        let (tx, rx) = oneshot::channel();
66        // Sender will only fail if the receiver was already
67        // dropped, which it can't be since we just made it.
68        let _ = tx.send(Err(err));
69        Self {
70            id: InvocationId::next(),
71            rx,
72        }
73    }
74
75    /// Block the current thread until the actor responds. Used by
76    /// the TUI input loop and tests.
77    pub fn blocking_recv(self) -> LspResult<T> {
78        match self.rx.blocking_recv() {
79            Ok(res) => res,
80            Err(_) => Err(LspError::ResponseDropped),
81        }
82    }
83}
84
85impl<T> std::future::Future for Pending<T> {
86    type Output = LspResult<T>;
87
88    fn poll(
89        mut self: std::pin::Pin<&mut Self>,
90        cx: &mut std::task::Context<'_>,
91    ) -> std::task::Poll<Self::Output> {
92        use std::task::Poll;
93        match std::pin::Pin::new(&mut self.rx).poll(cx) {
94            Poll::Ready(Ok(res)) => Poll::Ready(res),
95            Poll::Ready(Err(_)) => Poll::Ready(Err(LspError::ResponseDropped)),
96            Poll::Pending => Poll::Pending,
97        }
98    }
99}
100
101impl<T> fmt::Debug for Pending<T> {
102    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
103        f.debug_struct("Pending").field("id", &self.id).finish()
104    }
105}
106
107#[cfg(test)]
108mod tests {
109    use super::*;
110
111    #[tokio::test]
112    async fn pending_resolves_with_value() {
113        let (tx, rx) = oneshot::channel();
114        let p: Pending<i32> = Pending::new(InvocationId::next(), rx);
115        tx.send(Ok(42)).unwrap();
116        assert_eq!(p.await.unwrap(), 42);
117    }
118
119    #[tokio::test]
120    async fn pending_yields_response_dropped_when_sender_dies() {
121        let (tx, rx) = oneshot::channel::<LspResult<i32>>();
122        let p = Pending::new(InvocationId::next(), rx);
123        drop(tx);
124        match p.await {
125            Err(LspError::ResponseDropped) => {}
126            other => panic!("expected ResponseDropped, got {other:?}"),
127        }
128    }
129
130    #[tokio::test]
131    async fn ready_err_short_circuits() {
132        let p: Pending<i32> = Pending::ready_err(LspError::NotInitialized);
133        match p.await {
134            Err(LspError::NotInitialized) => {}
135            other => panic!("expected NotInitialized, got {other:?}"),
136        }
137    }
138
139    #[test]
140    fn invocation_ids_are_monotonic() {
141        let a = InvocationId::next().0;
142        let b = InvocationId::next().0;
143        assert!(b > a);
144    }
145}
lattice_lsp/pending.rs

lattice_lsp/
pending.rs
