eska_tool_runtime/
query.rs

1//! One bounded internal response, distinct from inherited user-facing command streams.
2
3use crate::{Component, Error, Operation};
4use serde_json::Value;
5use std::{
6    ffi::OsString,
7    io::{Read, Write},
8    path::Path,
9    process::Command,
10    sync::mpsc,
11    thread,
12    time::Duration,
13};
14
15const MAX_OUTPUT: u64 = 1024 * 1024;
16const TIMEOUT: Duration = Duration::from_secs(30);
17
18/// A caller selects finite limits for its concrete response, without changing other operations.
19#[derive(Clone, Copy, Debug)]
20pub struct QueryPolicy {
21    pub timeout: Duration,
22    pub output_limit: u64,
23    pub cancellation_grace: Duration,
24}
25
26impl Default for QueryPolicy {
27    /// Ordinary metadata checks retain the original thirty-second, one-MiB bounds per stream.
28    fn default() -> Self {
29        Self {
30            timeout: TIMEOUT,
31            output_limit: MAX_OUTPUT,
32            cancellation_grace: Duration::ZERO,
33        }
34    }
35}
36
37#[derive(Clone, Copy)]
38pub enum Stream {
39    Stdout,
40    Stderr,
41}
42
43/// Query a preflighted component without invoking a shell or relying on UTF-8 paths.
44///
45/// # Errors
46/// Rejects malformed schemas, nonzero exits, oversized output, timeout and cancellation.
47/// The exact direct child is reaped on every return path.
48pub fn query(
49    component: Component,
50    binary: &Path,
51    operation: &str,
52    root: &Path,
53    options: &[OsString],
54) -> Result<Value, Error> {
55    query_with_policy(
56        component,
57        binary,
58        operation,
59        root,
60        options,
61        QueryPolicy::default(),
62    )
63}
64
65/// Receive one response under explicit operation limits, with the same strict framing and cleanup.
66///
67/// # Errors
68/// Rejects timeouts, cancellation, oversized output, failed children and invalid protocol replies.
69pub fn query_with_policy(
70    component: Component,
71    binary: &Path,
72    operation: &str,
73    root: &Path,
74    options: &[OsString],
75    policy: QueryPolicy,
76) -> Result<Value, Error> {
77    #[cfg(unix)]
78    let scope = Some(Operation::new()?);
79    #[cfg(not(unix))]
80    let scope: Option<Operation> = None;
81    let mut command = Command::new(binary);
82    command
83        .args(["__bridge", &operation.replace('_', "-"), "--project-dir"])
84        .arg(root)
85        .args(options);
86    let output = crate::capture::run(component, &mut command, scope.as_ref(), policy, || {})?;
87    if !output.stderr.is_empty() {
88        std::io::stderr()
89            .lock()
90            .write_all(&output.stderr)
91            .map_err(|source| Error::Process { component, source })?;
92    }
93    decode(
94        component,
95        output.status.success(),
96        &output.stdout,
97        operation,
98    )
99}
100
101/// Both IPC transports share framing and domain-error validation.
102pub fn decode(
103    component: Component,
104    success: bool,
105    stdout: &[u8],
106    operation: &str,
107) -> Result<Value, Error> {
108    if !success {
109        if let Ok(Value::Object(document)) = serde_json::from_slice(stdout)
110            && document.len() == 4
111            && document.get("schema_version") == Some(&Value::from(1))
112            && document.get("operation").and_then(Value::as_str) == Some(operation)
113            && document.get("data") == Some(&Value::Null)
114            && let Some((code, detail)) = document.get("error").and_then(rejection)
115        {
116            return Err(Error::Rejected {
117                component,
118                code,
119                detail,
120            });
121        }
122        return Err(Error::Failed);
123    }
124    response(stdout, operation)
125}
126
127/// Rich domain failures use a closed shape; legacy string categories remain valid.
128fn rejection(value: &Value) -> Option<(String, Option<String>)> {
129    let (code, detail) = match value {
130        Value::String(code) => (code.as_str(), None),
131        Value::Object(error) if error.len() == 2 => (
132            error.get("code")?.as_str()?,
133            Some(error.get("detail")?.as_str()?.to_owned()),
134        ),
135        _ => return None,
136    };
137    if code.is_empty()
138        || code.len() > 64
139        || !code
140            .bytes()
141            .all(|byte| byte.is_ascii_lowercase() || matches!(byte, b'_' | b'-'))
142    {
143        return None;
144    }
145    Some((code.to_owned(), detail))
146}
147
148/// Drain one stream concurrently, retaining at most the protocol's fixed maximum.
149pub fn reader(
150    component: Component,
151    stream: impl Read + Send + 'static,
152    kind: Stream,
153    sender: mpsc::Sender<(Stream, Result<Vec<u8>, Error>)>,
154    limit: u64,
155) -> Result<(), Error> {
156    thread::Builder::new()
157        .name("eska-component-output".into())
158        .spawn(move || {
159            let result = bounded_with_limit(component, stream, limit);
160            let _ = sender.send((kind, result));
161        })
162        .map(|_| ())
163        .map_err(|source| Error::Process { component, source })
164}
165
166/// Reading one extra byte distinguishes a complete small response from truncated output.
167pub fn bounded(component: Component, stream: impl Read) -> Result<Vec<u8>, Error> {
168    bounded_with_limit(component, stream, MAX_OUTPUT)
169}
170
171/// One extra byte proves overflow; checked arithmetic also handles caller-supplied limits safely.
172fn bounded_with_limit(
173    component: Component,
174    stream: impl Read,
175    limit: u64,
176) -> Result<Vec<u8>, Error> {
177    let mut bytes = Vec::new();
178    stream
179        .take(limit.checked_add(1).ok_or(Error::OutputTooLarge)?)
180        .read_to_end(&mut bytes)
181        .map_err(|source| Error::Process { component, source })?;
182    if bytes.len() as u64 > limit {
183        return Err(Error::OutputTooLarge);
184    }
185    Ok(bytes)
186}
187
188/// Both framing and the domain's success flag must be valid before any payload is trusted.
189fn response(stdout: &[u8], operation: &str) -> Result<Value, Error> {
190    let Value::Object(mut document) =
191        serde_json::from_slice(stdout).map_err(|_| Error::InvalidResponse)?
192    else {
193        return Err(Error::InvalidResponse);
194    };
195    if document.len() != 4
196        || document.get("schema_version") != Some(&Value::from(1))
197        || document.get("operation").and_then(Value::as_str) != Some(operation)
198        || document.get("error") != Some(&Value::Null)
199    {
200        return Err(Error::InvalidResponse);
201    }
202    document
203        .remove("data")
204        .filter(|value| !value.is_null())
205        .ok_or(Error::InvalidResponse)
206}
207
208#[cfg(all(test, unix))]
209mod tests {
210    use super::{Component, Error, QueryPolicy, decode, query_with_policy};
211    use crate::test_support::TestDir;
212    use std::{ffi::OsString, fs, path::Path, time::Duration};
213
214    /// Valid domain failures retain details; nonzero partial or malformed replies remain failures.
215    #[test]
216    fn domain_error_details_do_not_permit_partial_success() {
217        let reply = serde_json::json!({"schema_version":1,"operation":"patch_plan","data":null,"error":{"code":"dirty","detail":"module.bsl"}});
218        let decode_reply = |value: &serde_json::Value| {
219            decode(
220                Component::Vcs,
221                false,
222                &serde_json::to_vec(value).unwrap(),
223                "patch_plan",
224            )
225        };
226        assert!(
227            matches!(decode_reply(&reply),Err(Error::Rejected{code,detail:Some(detail),..}) if code=="dirty" && detail=="module.bsl")
228        );
229        let mut partial = reply.clone();
230        partial["data"] = serde_json::json!({"prepared":true});
231        assert!(matches!(decode_reply(&partial), Err(Error::Failed)));
232        let mut unknown = reply.clone();
233        unknown["error"]["extra"] = true.into();
234        assert!(matches!(decode_reply(&unknown), Err(Error::Failed)));
235        let mut malformed = reply;
236        malformed["error"]["code"] = "dirty\nhelp".into();
237        assert!(matches!(decode_reply(&malformed), Err(Error::Failed)));
238    }
239
240    /// A large real JSON reply needs explicit bounds; normal metadata checks retain their limit.
241    #[test]
242    fn explicit_large_response_policy_does_not_relax_default_bounds() {
243        let fixture = TestDir::new();
244        let binary =
245            Path::new(env!("CARGO_MANIFEST_DIR")).join("../../tests/fixtures/component.py");
246        let options = [
247            OsString::from("--fixture-mode"),
248            OsString::from("bridge-large-plan"),
249        ];
250        assert!(matches!(
251            query_with_policy(
252                Component::Vcs,
253                &binary,
254                "provenance",
255                &fixture.0,
256                &options,
257                QueryPolicy::default()
258            ),
259            Err(Error::OutputTooLarge)
260        ));
261        let policy = QueryPolicy {
262            output_limit: 2 * 1024 * 1024,
263            ..QueryPolicy::default()
264        };
265        let response = query_with_policy(
266            Component::Vcs,
267            &binary,
268            "provenance",
269            &fixture.0,
270            &options,
271            policy,
272        )
273        .unwrap();
274        assert_eq!(response["padding"].as_str().unwrap().len(), 1024 * 1024 + 1);
275    }
276
277    /// A stalled real process cannot keep the caller or its direct child alive after timeout.
278    #[test]
279    fn timed_out_component_is_reaped() {
280        let fixture = TestDir::new();
281        let ready = fixture.0.join("ready");
282        let binary =
283            Path::new(env!("CARGO_MANIFEST_DIR")).join("../../tests/fixtures/component.py");
284        let options = [
285            OsString::from("--fixture-mode"),
286            OsString::from("bridge-timeout"),
287            OsString::from("--fixture-ready"),
288            ready.clone().into_os_string(),
289        ];
290        let result = query_with_policy(
291            Component::Vcs,
292            &binary,
293            "provenance",
294            &fixture.0,
295            &options,
296            QueryPolicy {
297                timeout: Duration::from_secs(1),
298                ..QueryPolicy::default()
299            },
300        );
301        assert!(matches!(result, Err(Error::Timeout)), "{result:?}");
302        let pid: u32 = fs::read_to_string(ready).unwrap().parse().unwrap();
303        // /proc is a Linux observation; the same timeout test still runs on other Unix hosts.
304        #[cfg(target_os = "linux")]
305        assert!(!Path::new("/proc").join(pid.to_string()).exists());
306        #[cfg(not(target_os = "linux"))]
307        assert!(pid > 0);
308    }
309
310    /// Graceful timeout signals the owner, permits cleanup and never accepts its zero exit.
311    #[test]
312    fn graceful_timeout_keeps_failure_after_successful_child_cleanup() {
313        let fixture = TestDir::new();
314        let ready = fixture.0.join("ready");
315        let binary =
316            Path::new(env!("CARGO_MANIFEST_DIR")).join("../../tests/fixtures/component.py");
317        let result = query_with_policy(
318            Component::Build,
319            &binary,
320            "doctor",
321            &fixture.0,
322            &[
323                "--fixture-mode".into(),
324                "bridge-graceful-timeout".into(),
325                "--fixture-ready".into(),
326                ready.clone().into_os_string(),
327            ],
328            QueryPolicy {
329                timeout: Duration::from_secs(1),
330                cancellation_grace: Duration::from_secs(2),
331                ..QueryPolicy::default()
332            },
333        );
334        assert!(matches!(result, Err(Error::Timeout)), "{result:?}");
335        assert_eq!(
336            fs::read_to_string(ready.with_extension("stopped")).unwrap(),
337            signal_hook::consts::signal::SIGTERM.to_string()
338        );
339        let pid: i32 = fs::read_to_string(ready).unwrap().parse().unwrap();
340        assert_eq!(
341            nix::sys::signal::kill(nix::unistd::Pid::from_raw(pid), None),
342            Err(nix::errno::Errno::ESRCH)
343        );
344    }
345}
eska_tool_runtime/query.rs

eska_tool_runtime/
query.rs
