aria_primitives/process/
sidecar.rs

1//! A supervised child process that answers one JSON line per request.
2//!
3//! The program is handed to its interpreter on the command line rather than
4//! read from disk, so the source cannot drift from the code that talks to it.
5//! Replies arrive on their own thread, which is what lets a caller give up:
6//! killing the child ends the read, so cancelling never waits for the work.
7//!
8//! The frames carry no meaning here. What a reply says, and which failures are
9//! recoverable, belongs to whoever speaks the protocol on top.
10
11use std::io::{self, BufRead, BufReader, Read, Write};
12use std::path::{Path, PathBuf};
13use std::process::{ChildStderr, ChildStdin, ChildStdout, Command, Stdio};
14use std::sync::mpsc::{self, Receiver, RecvTimeoutError};
15use std::thread;
16
17use serde::Serialize;
18use serde::de::DeserializeOwned;
19
20use super::POLL_INTERVAL;
21use super::child::ChildProcess;
22use crate::file_inspection::{InspectedFile, InspectionError};
23use crate::user_home::{self, UserHomeError};
24
25// A reply past this is a broken child, not a long answer, and reading on
26// would be unbounded memory.
27const MAX_REPLY_BYTES: u64 = 64 * 1024;
28
29/// What to run, and the name its log lines and threads carry.
30pub struct Script<'a> {
31    pub name: &'static str,
32    pub interpreter: &'a Path,
33    pub source: &'a str,
34    pub argument: &'a str,
35    /// Variables copied from the parent. The child also receives HOME and
36    /// PYTHONSAFEPATH; the remaining environment is cleared.
37    pub inherited_variables: &'a [&'a str],
38}
39
40pub struct Sidecar {
41    name: &'static str,
42    process: ChildProcess,
43    requests: ChildStdin,
44    replies: Receiver<io::Result<String>>,
45    /// A stopped sidecar is never reused, even when stopping it failed.
46    stopped: bool,
47}
48
49impl Sidecar {
50    pub fn start(script: &Script<'_>) -> Result<Self, SidecarError> {
51        let mut command = Command::new(script.interpreter);
52        command
53            .arg("-c")
54            .arg(script.source)
55            .arg(script.argument)
56            .stdin(Stdio::piped())
57            .stdout(Stdio::piped())
58            .stderr(Stdio::piped());
59        restrict_environment(&mut command, script.inherited_variables)?;
60
61        let mut process = ChildProcess::spawn(&mut command).map_err(SidecarError::Spawn)?;
62        let (Some(requests), Some(stdout), Some(stderr)) = (
63            process.take_stdin(),
64            process.take_stdout(),
65            process.take_stderr(),
66        ) else {
67            return Err(SidecarError::Pipes);
68        };
69        log_stderr(script.name, stderr);
70
71        Ok(Self {
72            name: script.name,
73            process,
74            requests,
75            replies: read_replies(script.name, stdout),
76            stopped: false,
77        })
78    }
79
80    pub fn send(&mut self, request: &impl Serialize) -> Result<(), SidecarError> {
81        let line = serde_json::to_string(request).map_err(SidecarError::Encode)?;
82        writeln!(self.requests, "{line}")
83            .and_then(|()| self.requests.flush())
84            .map_err(SidecarError::Write)
85    }
86
87    /// Reads one reply, killing the child if the caller gives up first.
88    pub fn receive<R: DeserializeOwned>(
89        &mut self,
90        should_cancel: &(impl Fn() -> bool + Sync),
91    ) -> Result<R, SidecarError> {
92        let line = self.read_line(should_cancel)?;
93        serde_json::from_str(line.trim()).map_err(SidecarError::Decode)
94    }
95
96    /// Whether the child is still there to serve another request.
97    pub fn is_running(&mut self) -> bool {
98        !self.stopped && matches!(self.process.try_wait(), Ok(None))
99    }
100
101    fn read_line(
102        &mut self,
103        should_cancel: &(impl Fn() -> bool + Sync),
104    ) -> Result<String, SidecarError> {
105        loop {
106            match self.replies.recv_timeout(POLL_INTERVAL) {
107                Ok(reply) => return reply.map_err(SidecarError::Read),
108                Err(RecvTimeoutError::Disconnected) => return Err(SidecarError::Closed),
109                Err(RecvTimeoutError::Timeout) => {}
110            }
111            if should_cancel() {
112                self.stop();
113                return Err(SidecarError::Cancelled);
114            }
115        }
116    }
117
118    fn stop(&mut self) {
119        self.stopped = true;
120        if let Err(error) = self.process.terminate_in_place() {
121            tracing::error!(?error, sidecar = self.name, "failed to stop the sidecar");
122        }
123    }
124}
125
126#[derive(Debug, thiserror::Error)]
127pub enum SidecarError {
128    #[error("the sidecar interpreter cannot be used")]
129    Interpreter(#[source] InspectionError),
130
131    #[error("a sidecar needs HOME to start")]
132    Home(#[source] UserHomeError),
133
134    #[error("failed to start the sidecar process")]
135    Spawn(#[source] io::Error),
136
137    #[error("the sidecar was started without all three of its pipes")]
138    Pipes,
139
140    #[error("failed to encode a request")]
141    Encode(#[source] serde_json::Error),
142
143    #[error("failed to send a request")]
144    Write(#[source] io::Error),
145
146    #[error("failed to read a reply")]
147    Read(#[source] io::Error),
148
149    #[error("the sidecar sent a reply that could not be understood")]
150    Decode(#[source] serde_json::Error),
151
152    #[error("the sidecar stopped before replying")]
153    Closed,
154
155    #[error("the caller gave up waiting")]
156    Cancelled,
157}
158
159/// Checks the interpreter the path resolves to, but keeps the configured path
160/// to run: a virtual environment is only itself when entered through its own
161/// `bin/python3`, and the canonical target is the bare interpreter.
162pub fn validate_interpreter(path: PathBuf) -> Result<PathBuf, SidecarError> {
163    InspectedFile::inspect(path.clone())
164        .and_then(InspectedFile::require_executable)
165        .and_then(InspectedFile::require_trusted)
166        .map_err(SidecarError::Interpreter)?;
167    Ok(path)
168}
169
170/// The child inherits nothing but `HOME` and the variables its caller named.
171fn restrict_environment(
172    command: &mut Command,
173    inherited_variables: &[&str],
174) -> Result<(), SidecarError> {
175    let home = user_home::resolve().map_err(SidecarError::Home)?;
176    // `python -c` puts the working directory on sys.path, so a file there
177    // named like an imported package would shadow it.
178    command
179        .env_clear()
180        .env("HOME", home)
181        .env("PYTHONSAFEPATH", "1");
182    for name in inherited_variables {
183        if let Some(value) = std::env::var_os(name) {
184            command.env(name, value);
185        }
186    }
187    Ok(())
188}
189
190/// Reads replies on their own thread so a blocked read never pins the caller.
191/// The thread ends when the child closes its output, which killing it does.
192fn read_replies(name: &'static str, stdout: ChildStdout) -> Receiver<io::Result<String>> {
193    let (sender, receiver) = mpsc::channel();
194    spawn_helper(format!("{name}-replies"), move || {
195        let mut reader = BufReader::new(stdout);
196        while let Some(reply) = next_reply(&mut reader) {
197            let failed = reply.is_err();
198            if sender.send(reply).is_err() || failed {
199                return;
200            }
201        }
202    });
203    receiver
204}
205
206/// One reply, or `None` once the child closes its output. Blank lines carry no
207/// frame, so they are skipped rather than delivered.
208fn next_reply(reader: &mut BufReader<ChildStdout>) -> Option<io::Result<String>> {
209    loop {
210        let mut line = String::new();
211        match reader.by_ref().take(MAX_REPLY_BYTES).read_line(&mut line) {
212            Ok(0) => return None,
213            Ok(size) if size as u64 == MAX_REPLY_BYTES => {
214                return Some(Err(io::Error::other(
215                    "sidecar reply exceeded the maximum frame size",
216                )));
217            }
218            Ok(_size) if line.trim().is_empty() => {}
219            Ok(_size) => return Some(Ok(line)),
220            Err(error) => return Some(Err(error)),
221        }
222    }
223}
224
225/// The child's own output is diagnostics, not protocol, so it joins the daemon log.
226fn log_stderr(name: &'static str, stderr: ChildStderr) {
227    spawn_helper(format!("{name}-stderr"), move || {
228        for line in BufReader::new(stderr).lines().map_while(Result::ok) {
229            if !line.trim().is_empty() {
230                tracing::warn!(sidecar = name, "{line}");
231            }
232        }
233    });
234}
235
236/// Both helper threads are best-effort. One that cannot start is logged, and
237/// the caller meets the missing output as a sidecar that stopped replying.
238fn spawn_helper(name: String, body: impl FnOnce() + Send + 'static) {
239    thread::Builder::new()
240        .name(name.clone())
241        .spawn(body)
242        .map(|_detached| ())
243        .unwrap_or_else(|error| {
244            tracing::error!(?error, thread = name, "failed to start a sidecar thread");
245        });
246}
247
248#[cfg(test)]
249mod tests {
250    use std::time::{Duration, Instant};
251
252    use super::*;
253
254    #[derive(Serialize)]
255    struct Ping {
256        op: &'static str,
257    }
258
259    #[derive(serde::Deserialize)]
260    struct Pong {
261        text: String,
262    }
263
264    // /bin/sh takes its script from -c the same way a Python child does, so
265    // the transport can be exercised without Python or a model.
266    fn stub(source: &str) -> Sidecar {
267        Sidecar::start(&Script {
268            name: "stub",
269            interpreter: Path::new("/bin/sh"),
270            source,
271            argument: "unused",
272            inherited_variables: &[],
273        })
274        .expect("the stub should start")
275    }
276
277    fn ping(sidecar: &mut Sidecar) -> Result<Pong, SidecarError> {
278        sidecar.send(&Ping { op: "ping" })?;
279        sidecar.receive(&|| false)
280    }
281
282    // The child exits on its own schedule, so poll instead of checking once.
283    fn stops_running(sidecar: &mut Sidecar) -> bool {
284        let deadline = Instant::now() + Duration::from_secs(5);
285        while Instant::now() < deadline {
286            if !sidecar.is_running() {
287                return true;
288            }
289            thread::sleep(POLL_INTERVAL);
290        }
291        false
292    }
293
294    #[test]
295    fn a_reply_is_read_back_from_the_child() {
296        let mut sidecar = stub(r#"read line; echo '{"text":"hello there"}'"#);
297
298        let pong = ping(&mut sidecar).expect("the stub should answer");
299
300        assert_eq!(pong.text, "hello there");
301    }
302
303    #[test]
304    fn the_child_is_told_to_keep_cwd_off_the_python_path() {
305        let mut sidecar = stub(r#"read line; echo "{\"text\":\"$PYTHONSAFEPATH\"}""#);
306
307        let pong = ping(&mut sidecar).expect("the stub should answer");
308
309        assert_eq!(pong.text, "1");
310    }
311
312    #[test]
313    fn blank_lines_are_not_replies() {
314        let mut sidecar = stub(r#"read line; echo; echo; echo '{"text":"after the blanks"}'"#);
315
316        let pong = ping(&mut sidecar).expect("the stub should answer past the blank lines");
317
318        assert_eq!(pong.text, "after the blanks");
319    }
320
321    #[test]
322    fn a_child_that_keeps_serving_is_still_running() {
323        let mut sidecar = stub(r#"while read line; do echo '{"text":"again"}'; done"#);
324
325        ping(&mut sidecar).expect("the stub should answer");
326
327        assert!(sidecar.is_running());
328    }
329
330    #[test]
331    fn a_cancelled_wait_stops_the_child_instead_of_blocking() {
332        let mut sidecar = stub("sleep 600");
333
334        sidecar.send(&Ping { op: "ping" }).expect("the send lands");
335        let outcome = sidecar.receive::<Pong>(&|| true);
336
337        assert!(matches!(outcome, Err(SidecarError::Cancelled)));
338        assert!(
339            !sidecar.is_running(),
340            "a cancelled sidecar was killed and must not be reused"
341        );
342    }
343
344    #[test]
345    fn a_child_that_exits_is_reported_rather_than_waited_on() {
346        let mut sidecar = stub("exit 0");
347
348        let outcome = ping(&mut sidecar);
349
350        assert!(
351            matches!(
352                outcome,
353                Err(SidecarError::Closed) | Err(SidecarError::Write(_))
354            ),
355            "a dead sidecar must surface as an error"
356        );
357        assert!(
358            stops_running(&mut sidecar),
359            "a dead sidecar must stop reporting itself as running"
360        );
361        assert!(
362            !sidecar.is_running(),
363            "a dead sidecar must stay dead on a second look"
364        );
365    }
366
367    #[test]
368    fn a_reply_that_is_not_json_is_a_loud_error() {
369        let mut sidecar = stub("read line; echo 'not json'");
370
371        let outcome = ping(&mut sidecar);
372
373        assert!(matches!(outcome, Err(SidecarError::Decode(_))));
374    }
375
376    #[test]
377    fn an_oversized_reply_is_refused_rather_than_buffered() {
378        let mut sidecar = stub(&format!(
379            "read line; head -c {} /dev/zero | tr '\\0' 'a'; echo",
380            MAX_REPLY_BYTES + 1
381        ));
382
383        let outcome = ping(&mut sidecar);
384
385        assert!(matches!(outcome, Err(SidecarError::Read(_))));
386    }
387
388    #[test]
389    fn a_relative_interpreter_is_rejected() {
390        let error = validate_interpreter(PathBuf::from("relative/python3"))
391            .expect_err("a relative interpreter must be rejected");
392
393        assert!(matches!(
394            error,
395            SidecarError::Interpreter(InspectionError::NotAbsolute { .. })
396        ));
397    }
398
399    #[test]
400    fn a_missing_interpreter_is_rejected() {
401        let error = validate_interpreter(PathBuf::from("/nonexistent/python3"))
402            .expect_err("a missing interpreter must be rejected");
403
404        assert!(matches!(
405            error,
406            SidecarError::Interpreter(InspectionError::Inspect { .. })
407        ));
408    }
409
410    #[test]
411    fn a_file_that_cannot_be_run_is_not_an_interpreter() {
412        let error = validate_interpreter(PathBuf::from("/etc/hosts"))
413            .expect_err("a data file must be rejected");
414
415        assert!(matches!(
416            error,
417            SidecarError::Interpreter(InspectionError::NotExecutable { .. })
418        ));
419    }
420}
aria_primitives/process/sidecar.rs

aria_primitives/process/
sidecar.rs
