eska_tool_runtime/
exchange.rs

1//! Interactive composition keeps stderr/stdin attached while receiving one bounded machine reply.
2
3use crate::{
4    Component, Error,
5    process::{OwnedChild, SignalState},
6    query::{bounded, decode},
7};
8use serde_json::Value;
9use std::{
10    ffi::OsString,
11    io,
12    path::Path,
13    process::{Command, Stdio},
14    sync::{Arc, atomic::Ordering, mpsc},
15    thread,
16    time::{Duration, Instant},
17};
18
19/// One sticky cancellation scope spans selection, preparation, confirmation and publication.
20#[derive(Clone)]
21pub struct Operation(Arc<SignalState>);
22
23impl std::fmt::Debug for Operation {
24    /// Report cancellation state without exposing OS subscription internals.
25    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
26        formatter
27            .debug_struct("Operation")
28            .field("cancelled", &self.is_cancelled())
29            .finish_non_exhaustive()
30    }
31}
32
33impl Operation {
34    /// Install only this operation's signal subscriptions before invoking component work.
35    ///
36    /// # Errors
37    /// Returns OS handler installation failures before creating a child or changing files.
38    pub fn new() -> Result<Self, Error> {
39        SignalState::new()
40            .map(|state| Self(Arc::new(state)))
41            .map_err(Error::Signal)
42    }
43
44    /// Internal process controllers preserve the exact originally received Unix signal.
45    pub(crate) fn received_signal(&self) -> Option<i32> {
46        i32::try_from(self.0.received.load(Ordering::Acquire))
47            .ok()
48            .filter(|value| *value != 0)
49    }
50
51    /// A received cancellation remains set until the entire surrounding operation is dropped.
52    #[must_use]
53    pub fn is_cancelled(&self) -> bool {
54        self.0.received.load(Ordering::Acquire) != 0
55    }
56}
57
58/// Receive a machine result while the component owns interactive input and live diagnostics.
59///
60/// Native work retains its owner's duration policy. Cancellation forwards the exact Unix signal
61/// to the direct child and gives its rollback thirty seconds before the final kill/reap fallback.
62///
63/// # Errors
64/// Rejects cancellation, invalid replies, nonzero exits, oversized stdout and process failures.
65pub fn exchange(
66    component: Component,
67    binary: &Path,
68    args: &[OsString],
69    operation: &str,
70    scope: &Operation,
71) -> Result<Value, Error> {
72    if scope.is_cancelled() {
73        return Err(Error::Cancelled);
74    }
75    let child = Command::new(binary)
76        .args(args)
77        .stdin(Stdio::inherit())
78        .stdout(Stdio::piped())
79        .stderr(Stdio::inherit())
80        .spawn()
81        .map_err(|source| Error::Process { component, source })?;
82    let mut child = OwnedChild(child);
83    let stdout = child.0.stdout.take().ok_or(Error::InvalidResponse)?;
84    let (sender, receiver) = mpsc::channel();
85    thread::Builder::new()
86        .name("eska-component-reply".into())
87        .spawn(move || {
88            let _ = sender.send(bounded(component, stdout));
89        })
90        .map_err(|source| Error::Process { component, source })?;
91    let mut output = None;
92    let mut status = None;
93    let mut cancellation = None;
94    loop {
95        match receiver.try_recv() {
96            Ok(result) => output = Some(result?),
97            Err(mpsc::TryRecvError::Disconnected) if output.is_none() => {
98                return Err(Error::InvalidResponse);
99            }
100            Err(_) => {}
101        }
102        if status.is_none() {
103            status = child
104                .0
105                .try_wait()
106                .map_err(|source| Error::Process { component, source })?;
107        }
108        if let Some(status) = status
109            && let Some(output) = &output
110        {
111            return if scope.is_cancelled() {
112                Err(Error::Cancelled)
113            } else {
114                decode(component, status.success(), output, operation)
115            };
116        }
117        if scope.is_cancelled() && cancellation.is_none() {
118            forward_cancel(&child, scope).map_err(Error::Signal)?;
119            cancellation = Some(Instant::now());
120        }
121        if cancellation.is_some_and(|started| started.elapsed() >= Duration::from_secs(30)) {
122            return Err(Error::Cancelled);
123        }
124        thread::sleep(Duration::from_millis(10));
125    }
126}
127
128/// Never broadcast to the caller's group or enumerate unrelated native processes.
129#[cfg(unix)]
130fn forward_cancel(child: &OwnedChild, scope: &Operation) -> io::Result<()> {
131    use nix::{
132        errno::Errno,
133        sys::signal::{Signal, kill},
134        unistd::Pid,
135    };
136    let pid = i32::try_from(child.0.id()).map_err(io::Error::other)?;
137    let signal =
138        i32::try_from(scope.0.received.load(Ordering::Acquire)).map_err(io::Error::other)?;
139    match kill(
140        Pid::from_raw(pid),
141        Signal::try_from(signal).map_err(io::Error::other)?,
142    ) {
143        Ok(()) | Err(Errno::ESRCH) => Ok(()),
144        Err(error) => Err(io::Error::from(error)),
145    }
146}
147
148/// Console cancellation reaches both attached processes on Windows.
149#[cfg(windows)]
150const fn forward_cancel(_: &OwnedChild, _: &Operation) -> io::Result<()> {
151    Ok(())
152}
eska_tool_runtime/exchange.rs

eska_tool_runtime/
exchange.rs
