char/avril

(wip) why'd you have to go and make things so complicaaaaated?

git clone https://git.t4t.associates/char/avril

Charlotte Somexperiment: use native typescript lsp supportc7a9569

main
branchesmainall refs →
11.4 KiB313 linesraw
1use crate::{BACKEND_VERSION, Result, wire};
2use serde::de::DeserializeOwned;
3use serde_json::{Value, json};
4use std::{
5    collections::HashMap,
6    path::Path,
7    process::Stdio,
8    sync::{Arc, Mutex},
9    time::Duration,
10};
11use tokio::{
12    io::BufReader,
13    process::{Child, Command},
14    sync::{mpsc, oneshot},
15    task::{JoinHandle, JoinSet},
16};
17
18pub trait Callbacks: Send + Sync + 'static {
19    fn call(&self, method: &str, params: Value) -> Result<Value>;
20}
21
22type Reply = Result<Value>;
23#[derive(Default)]
24struct State {
25    next: u64,
26    failed: Option<String>,
27    pending: HashMap<u64, oneshot::Sender<Reply>>,
28}
29impl State {
30    fn fail(&mut self, error: String) {
31        self.failed.get_or_insert(error.clone());
32        for (_, reply) in self.pending.drain() {
33            let _ = reply.send(Err(error.clone().into()));
34        }
35    }
36}
37
38const WRITER_CAPACITY: usize = 16;
39
40type Writer = mpsc::Sender<WriteRequest>;
41struct WriteRequest {
42    value: Value,
43    result: oneshot::Sender<Result<()>>,
44}
45
46async fn send(writer: &Writer, value: Value) -> Result<()> {
47    let (result, rx) = oneshot::channel();
48    writer.send(WriteRequest { value, result }).await?;
49    rx.await?
50}
51
52async fn writer_loop<W: tokio::io::AsyncWrite + Unpin>(
53    mut output: W,
54    mut requests: mpsc::Receiver<WriteRequest>,
55    state: Arc<Mutex<State>>,
56) {
57    while let Some(request) = requests.recv().await {
58        let result = wire::write(&mut output, request.value).await;
59        let failed = result.as_ref().err().map(ToString::to_string);
60        let _ = request.result.send(result);
61        if let Some(error) = failed {
62            state.lock().unwrap().fail(error);
63            break;
64        }
65    }
66}
67
68pub struct Backend {
69    child: Child,
70    state: Arc<Mutex<State>>,
71    writer: Option<Writer>,
72    writer_task: Option<JoinHandle<()>>,
73    reader: Option<JoinHandle<()>>,
74}
75
76impl Backend {
77    pub async fn start(path: &Path, host: Arc<dyn Callbacks>) -> Result<Self> {
78        let version = tokio::time::timeout(
79            Duration::from_secs(10),
80            Command::new(path)
81                .env_clear()
82                .kill_on_drop(true)
83                .arg("--version")
84                .output(),
85        )
86        .await??;
87        if !version.status.success()
88            || String::from_utf8_lossy(&version.stdout).trim()
89                != format!("Version {BACKEND_VERSION}")
90        {
91            return Err(format!("Expected TypeScript {BACKEND_VERSION} native executable").into());
92        }
93        let mut child = Command::new(path)
94            .env_clear()
95            .kill_on_drop(true)
96            .args([
97                "--api",
98                "--async",
99                "--cwd",
100                "/avril-lsp",
101                "--callbacks",
102                concat!(
103                    "readFile,fileExists,directoryExists,getAccessibleEntries,",
104                    "realpath,stat:fakeStat,writeFile:error,removeFile:error"
105                ),
106            ])
107            .stdin(Stdio::piped())
108            .stdout(Stdio::piped())
109            .stderr(Stdio::inherit())
110            .spawn()?;
111        let stdin = child.stdin.take().ok_or("No backend stdin")?;
112        let (writer, requests) = mpsc::channel::<WriteRequest>(WRITER_CAPACITY);
113        let mut stdout = BufReader::new(child.stdout.take().ok_or("No backend stdout")?);
114        let state = Arc::new(Mutex::new(State::default()));
115        let writer_task = tokio::spawn(writer_loop(stdin, requests, state.clone()));
116        let reader = tokio::spawn({
117            let state = state.clone();
118            let writer = writer.clone();
119            async move {
120                let mut callbacks = JoinSet::new();
121                let result: Result<()> = async {
122                    loop {
123                        tokio::select! {
124                            completed = callbacks.join_next(), if !callbacks.is_empty() => { completed.ok_or("Missing callback task")???; }
125                            message = wire::read(&mut stdout) => {
126                                let message = message?.ok_or("TypeScript backend closed stdout")?;
127                                if let Some(method) = message["method"].as_str() {
128                                    if let Some(id) = message.get("id").cloned() {
129                                        let method = method.to_owned();
130                                        let host = host.clone();
131                                        let writer = writer.clone();
132                                        callbacks.spawn(async move {
133                                            let response = match host.call(&method, message["params"].clone()) {
134                                                Ok(result) => json!({"id":id,"result":result}),
135                                                Err(error) => json!({"id":id,"error":{"code":-32603,"message":error.to_string()}}),
136                                            };
137                                            send(&writer, response).await
138                                        });
139                                    }
140                                } else if let Some(id) = message["id"].as_u64() {
141                                    if let Some(reply) = state.lock().unwrap().pending.remove(&id) {
142                                        let result = if let Some(error) = message.get("error") {
143                                            Err(serde_json::from_value::<wire::RpcError>(error.clone())?.into())
144                                        } else {
145                                            Ok(message["result"].clone())
146                                        };
147                                        let _ = reply.send(result);
148                                    }
149                                } else { return Err("Invalid backend response ID".into()); }
150                            }
151                        }
152                    }
153                }.await;
154                callbacks.abort_all();
155                state.lock().unwrap().fail(
156                    result
157                        .err()
158                        .map_or("Backend reader stopped".into(), |e| e.to_string()),
159                );
160            }
161        });
162        Ok(Self {
163            child,
164            state,
165            writer: Some(writer),
166            writer_task: Some(writer_task),
167            reader: Some(reader),
168        })
169    }
170
171    pub fn failed(&self) -> bool {
172        self.state.lock().unwrap().failed.is_some()
173    }
174
175    pub async fn call<T: DeserializeOwned>(&self, method: &str, params: Value) -> Result<T> {
176        let (tx, rx) = oneshot::channel();
177        let id = {
178            let mut state = self.state.lock().unwrap();
179            if let Some(error) = &state.failed {
180                return Err(error.clone().into());
181            }
182            state.next += 1;
183            let id = state.next;
184            state.pending.insert(id, tx);
185            id
186        };
187        // Remove registrations even if the editor cancels this future.
188        struct Registration(Arc<Mutex<State>>, u64, Writer);
189        impl Drop for Registration {
190            fn drop(&mut self) {
191                if self.0.lock().unwrap().pending.remove(&self.1).is_some() {
192                    let (result, _) = oneshot::channel();
193                    let _ = self.2.try_send(WriteRequest {
194                        value: json!({"method":"$/cancelRequest","params":{"id":self.1}}),
195                        result,
196                    });
197                }
198            }
199        }
200        let writer = self.writer.as_ref().ok_or("Backend stdin closed")?.clone();
201        let _registration = Registration(self.state.clone(), id, writer.clone());
202        let operation = async {
203            send(&writer, json!({"id":id,"method":method,"params":params})).await?;
204            let value = rx.await??;
205            Ok::<T, crate::Error>(serde_json::from_value(value)?)
206        };
207        match tokio::time::timeout(Duration::from_secs(30), operation).await {
208            Ok(result) => result,
209            Err(_) => {
210                self.state
211                    .lock()
212                    .unwrap()
213                    .fail(format!("TypeScript API timed out: {method}"));
214                Err(format!("TypeScript API timed out: {method}").into())
215            }
216        }
217    }
218
219    pub async fn close(&mut self) -> Result<()> {
220        self.state
221            .lock()
222            .unwrap()
223            .fail("Backend shutting down".into());
224        if let Some(reader) = self.reader.take() {
225            reader.abort();
226            let _ = reader.await;
227        }
228        self.writer.take();
229        if let Some(mut writer_task) = self.writer_task.take() {
230            match tokio::time::timeout(Duration::from_secs(2), &mut writer_task).await {
231                Ok(Ok(())) => {}
232                Ok(Err(error)) => return Err(error.into()),
233                Err(_) => {
234                    writer_task.abort();
235                    let _ = writer_task.await;
236                    self.child.kill().await?;
237                    self.child.wait().await?;
238                    return Err("TypeScript backend writer did not shut down".into());
239                }
240            }
241        }
242        match tokio::time::timeout(Duration::from_secs(2), self.child.wait()).await {
243            Ok(status) => {
244                if status?.success() {
245                    Ok(())
246                } else {
247                    Err("TypeScript backend exited unsuccessfully".into())
248                }
249            }
250            Err(_) => {
251                self.child.kill().await?;
252                self.child.wait().await?;
253                Err("TypeScript backend did not shut down".into())
254            }
255        }
256    }
257}
258impl Drop for Backend {
259    fn drop(&mut self) {
260        if let Some(reader) = &self.reader {
261            reader.abort();
262        }
263        if let Some(writer) = &self.writer_task {
264            writer.abort();
265        }
266    }
267}
268
269#[cfg(test)]
270mod tests {
271    use super::*;
272    use tokio::io::{AsyncReadExt, BufReader};
273
274    #[tokio::test]
275    async fn canceled_frame_does_not_corrupt_next_frame() {
276        let state = Arc::new(Mutex::new(State::default()));
277        let (writer, requests) = mpsc::channel(WRITER_CAPACITY);
278        let (backend, mut output) = tokio::io::duplex(8);
279        let task = tokio::spawn(writer_loop(backend, requests, state));
280
281        let large = json!({"id":1,"method":"large","params":{"text":"x".repeat(1024 * 1024)}});
282        let first_call = tokio::spawn({
283            let writer = writer.clone();
284            async move { send(&writer, large).await }
285        });
286
287        let mut prefix = [0; 8];
288        output.read_exact(&mut prefix).await.unwrap();
289        first_call.abort();
290        let _ = first_call.await;
291
292        let (second_result, second_rx) = oneshot::channel();
293        writer
294            .send(WriteRequest {
295                value: json!({"id":2,"method":"query","params":{}}),
296                result: second_result,
297            })
298            .await
299            .unwrap();
300        drop(writer);
301        let mut bytes = prefix.to_vec();
302        let (second_result, read_result) = tokio::join!(second_rx, output.read_to_end(&mut bytes));
303        second_result.unwrap().unwrap();
304        read_result.unwrap();
305        task.await.unwrap();
306        let mut input = BufReader::new(bytes.as_slice());
307        let first = wire::read(&mut input).await.unwrap().unwrap();
308        let second = wire::read(&mut input).await.unwrap().unwrap();
309        assert_eq!(first["id"], 1);
310        assert_eq!(second["id"], 2);
311        assert_eq!(second["method"], "query");
312    }
313}