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

eska_tool_runtime/
exchange.rs
