specfold_core/plugins/
process.rs

1//! Runs a plugin process for one request, with a deadline.
2
3use super::protocol::{PROTOCOL, Request};
4use crate::config::{Config, PluginDef};
5use anyhow::{Context, Result, anyhow, bail};
6use serde::de::DeserializeOwned;
7use serde_json::Value;
8use std::io::{Read, Write};
9use std::path::PathBuf;
10use std::process::{Command, Stdio};
11use std::time::{Duration, Instant};
12
13fn program(cfg: &Config, p: &PluginDef) -> Result<(PathBuf, Vec<String>)> {
14    let (prog, args) = p.cmd.split_first().ok_or_else(|| anyhow!("plugin {} has an empty cmd", p.name))?;
15    let local = cfg.root.join(prog);
16    let resolved = if prog.contains('/') && local.exists() { local } else { PathBuf::from(prog) };
17    Ok((resolved, args.to_vec()))
18}
19
20pub fn call<T: DeserializeOwned>(cfg: &Config, p: &PluginDef, method: &str, params: Value) -> Result<T> {
21    let (prog, args) = program(cfg, p)?;
22    let mut child = Command::new(&prog)
23        .args(&args)
24        .current_dir(&cfg.root)
25        .env("SPECFOLD_PROTOCOL", PROTOCOL)
26        .stdin(Stdio::piped())
27        .stdout(Stdio::piped())
28        .stderr(Stdio::piped())
29        .spawn()
30        .with_context(|| format!("plugin {}: cannot start {}", p.name, prog.display()))?;
31    let body = serde_json::to_vec(&Request { protocol: PROTOCOL, method, params })?;
32    let mut stdin = child.stdin.take().context("no stdin")?;
33    let writer = std::thread::spawn(move || {
34        let _ = stdin.write_all(&body);
35    });
36    // C150, C151: both pipes are drained as the plugin writes (a full stderr pipe would stall
37    // it), and neither is waited on past the deadline (a background child may keep them open).
38    let pipe = |r: Option<Box<dyn Read + Send>>| {
39        let (tx, rx) = std::sync::mpsc::channel::<Vec<u8>>();
40        if let Some(mut r) = r {
41            std::thread::spawn(move || {
42                let mut buf = [0u8; 8192];
43                while let Ok(n) = r.read(&mut buf) {
44                    if n == 0 || tx.send(buf[..n].to_vec()).is_err() {
45                        break;
46                    }
47                }
48            });
49        }
50        rx
51    };
52    let out_rx = pipe(child.stdout.take().map(|r| Box::new(r) as Box<dyn Read + Send>));
53    let err_rx = pipe(child.stderr.take().map(|r| Box::new(r) as Box<dyn Read + Send>));
54    let (mut out, mut err) = (vec![], vec![]);
55    let drain = |rx: &std::sync::mpsc::Receiver<Vec<u8>>, buf: &mut Vec<u8>, keep: usize| {
56        while let Ok(chunk) = rx.try_recv() {
57            buf.extend(chunk);
58            if keep > 0 && buf.len() > keep {
59                buf.drain(..buf.len() - keep);
60            }
61        }
62    };
63    let deadline = Instant::now() + Duration::from_millis(p.timeout_ms);
64    let status = loop {
65        drain(&out_rx, &mut out, 0);
66        drain(&err_rx, &mut err, 8192);
67        if let Some(s) = child.try_wait()? {
68            break s;
69        }
70        if Instant::now() > deadline {
71            let _ = child.kill();
72            let _ = child.wait();
73            bail!("plugin {} did not answer {method} within {} ms", p.name, p.timeout_ms);
74        }
75        std::thread::sleep(Duration::from_millis(10));
76    };
77    let _ = writer.join();
78    // The plugin has exited; take what is left of its output until the pipes close or the deadline.
79    for (rx, buf, keep) in [(&out_rx, &mut out, 0), (&err_rx, &mut err, 8192)] {
80        loop {
81            let left = deadline.saturating_duration_since(Instant::now());
82            match rx.recv_timeout(left.min(Duration::from_millis(100)).max(Duration::from_millis(1))) {
83                Ok(chunk) => {
84                    buf.extend(chunk);
85                    if keep > 0 && buf.len() > keep {
86                        buf.drain(..buf.len() - keep);
87                    }
88                }
89                Err(_) => break,
90            }
91            if Instant::now() > deadline {
92                break;
93            }
94        }
95    }
96    if !status.success() {
97        let err = String::from_utf8_lossy(&err);
98        bail!("plugin {} failed on {method}: {}", p.name, err.lines().last().unwrap_or("exit status non-zero"));
99    }
100    serde_json::from_slice(&out).with_context(|| format!("plugin {} answered {method} with invalid JSON", p.name))
101}
specfold_core/plugins/process.rs

specfold_core/plugins/
process.rs
