harmos_sidecar/
attach.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10/* === Modules === */
11
12#[cfg(unix)]
13pub(crate) mod control;
14#[cfg(windows)]
15#[path = "attach/windows.rs"]
16pub(crate) mod control;
17
18/* === Re-Exports === */
19
20/* === Imports === */
21
22// Standard Library Imports
23use std::ffi::OsString;
24use std::future::Future;
25use std::io::Error;
26use std::path::{Path, PathBuf};
27use std::pin::Pin;
28use std::process::Stdio;
29use std::sync::Arc;
30// Third Party Imports
31use sha2::{Digest, Sha256};
32use tokio::io::{AsyncRead, AsyncWrite};
33use tokio::process::{Child, Command};
34// Local Imports
35use crate::StreamDeclaration;
36use crate::inventory::{Declaration, DeclarationError, Facility};
37
38/* === Constants === */
39
40#[cfg(windows)]
41const CREATE_NO_WINDOW: u32 = 0x0800_0000;
42
43/* === Macros === */
44
45/* === Type Aliases === */
46
47type EndWork = Pin<Box<dyn Future<Output = std::io::Result<()>> + Send>>;
48type EndStart = Box<dyn FnOnce() -> EndWork + Send>;
49
50/* === Structs === */
51
52#[must_use = "retain the ending until physical settlement succeeds"]
53pub struct Ending {
54    receipt: Box<dyn End>,
55}
56
57struct Terminal {
58    start: Option<EndStart>,
59    work: Option<EndWork>,
60    result: Option<Result<(), Arc<Error>>>,
61}
62
63struct ProcessEnd {
64    child: Child,
65    signaled: bool,
66    completed: bool,
67}
68
69pub struct Attached {
70    pub(super) control: Option<crate::transport::Io>,
71    pub(super) write: Box<dyn AsyncWrite + Send + Unpin>,
72    pub(super) read: Box<dyn AsyncRead + Send + Unpin>,
73    pub(super) diagnostics: Option<Box<dyn AsyncRead + Send + Unpin>>,
74    pub(super) end: Option<Ending>,
75}
76
77#[derive(Clone, Debug)]
78pub struct Executable {
79    program: PathBuf,
80    declaration: Declaration,
81    sha256: [u8; 32],
82    arguments: Vec<OsString>,
83    directory: Option<PathBuf>,
84    environment: Vec<(OsString, OsString)>,
85}
86
87/* === Enums === */
88
89#[derive(Debug, thiserror::Error)]
90pub enum ExecutableError {
91    #[error("the sidecar executable `{path}` is not an absolute path")]
92    Relative { path: PathBuf },
93
94    #[error("the sidecar executable `{path}` cannot be read: {source}")]
95    Unreadable { path: PathBuf, source: std::io::Error },
96
97    #[error("the sidecar executable `{path}` is not a file")]
98    NotFile { path: PathBuf },
99
100    #[error("the sidecar executable `{path}` has no valid declaration: {source}")]
101    Declaration {
102        path: PathBuf,
103
104        #[source]
105        source: Box<DeclarationError>,
106    },
107}
108
109/* === Traits === */
110
111pub trait End: Send + 'static {
112    fn settle(&mut self) -> Pin<Box<dyn Future<Output = std::io::Result<()>> + Send + '_>>;
113}
114
115pub trait Attach: Clone + Send + Sync + 'static {
116    type Error: std::fmt::Display + Send + 'static;
117
118    fn owner(&self) -> &str;
119
120    fn publications(&self) -> Vec<StreamDeclaration> {
121        Vec::new()
122    }
123
124    fn observers(&self) -> Vec<crate::inventory::observers::Declaration> {
125        Vec::new()
126    }
127
128    fn operations(&self) -> harmos_domain::Result<Vec<crate::operations::Definition>> {
129        Ok(Vec::new())
130    }
131
132    fn operation_contracts(&self) -> harmos_domain::Result<Option<Vec<crate::operations::Contract>>> {
133        Ok(None)
134    }
135
136    fn required_contracts(&self) -> harmos_domain::Result<Option<Vec<crate::inventory::contracts::Required>>> {
137        Ok(None)
138    }
139
140    fn sdk_lifecycle(&self) -> harmos_domain::Result<Option<[crate::operations::lifecycle::Descriptor; 2]>> {
141        Ok(None)
142    }
143
144    fn observe_contract(&self) -> harmos_domain::Result<Option<crate::operations::Contract>> {
145        Ok(None)
146    }
147
148    fn listener_declarations(&self) -> harmos_domain::Result<Option<Vec<harmos_domain::listeners::Definition>>> {
149        Ok(None)
150    }
151
152    fn attach(&self) -> impl Future<Output = Result<Attached, Self::Error>> + Send;
153}
154
155/* === Implementations === */
156
157impl Ending {
158    pub fn new(receipt: impl End) -> Self {
159        Self { receipt: Box::new(receipt) }
160    }
161
162    pub fn terminal<F, U>(end: F) -> Self
163    where
164        F: FnOnce() -> U + Send + 'static,
165        U: Future<Output = std::io::Result<()>> + Send + 'static,
166    {
167        Self::new(Terminal { start: Some(Box::new(move || Box::pin(end()))), work: None, result: None })
168    }
169
170    pub async fn settle(&mut self) -> std::io::Result<()> {
171        self.receipt.settle().await
172    }
173}
174
175impl End for Terminal {
176    fn settle(&mut self) -> Pin<Box<dyn Future<Output = std::io::Result<()>> + Send + '_>> {
177        Box::pin(async move {
178            if let Some(result) = &self.result {
179                return result.clone().map_err(Error::other);
180            }
181            if self.work.is_none() {
182                let start = self.start.take().ok_or_else(|| Error::other("terminal ending construction did not return"))?;
183                self.work = Some(start());
184            }
185            let result = self.work.as_mut().expect("original terminal work is retained").await;
186            self.work = None;
187            self.result = Some(result.map_err(Arc::new));
188            self.result.as_ref().expect("terminal outcome is retained").clone().map_err(Error::other)
189        })
190    }
191}
192
193impl End for ProcessEnd {
194    fn settle(&mut self) -> Pin<Box<dyn Future<Output = std::io::Result<()>> + Send + '_>> {
195        Box::pin(async move {
196            if self.completed {
197                return Ok(());
198            }
199            if !self.signaled {
200                self.child.start_kill()?;
201                self.signaled = true;
202            }
203            self.child.wait().await?;
204            self.completed = true;
205            Ok(())
206        })
207    }
208}
209
210impl Attached {
211    pub fn new(write: impl AsyncWrite + Send + Unpin + 'static, read: impl AsyncRead + Send + Unpin + 'static) -> Self {
212        Self { control: None, write: Box::new(write), read: Box::new(read), diagnostics: None, end: None }
213    }
214
215    pub fn control(mut self, channel: crate::transport::Io) -> Self {
216        self.control = Some(channel);
217        self
218    }
219
220    pub fn diagnostics(mut self, read: impl AsyncRead + Send + Unpin + 'static) -> Self {
221        self.diagnostics = Some(Box::new(read));
222        self
223    }
224
225    pub fn ended_by<F, U>(self, end: F) -> Self
226    where
227        F: FnOnce() -> U + Send + 'static,
228        U: Future<Output = ()> + Send + 'static,
229    {
230        self.ending(Ending::terminal(move || async move {
231            end().await;
232            Ok(())
233        }))
234    }
235
236    pub fn ending(mut self, ending: Ending) -> Self {
237        self.end = Some(ending);
238        self
239    }
240
241    #[allow(clippy::type_complexity)]
242    pub fn into_parts(
243        self,
244    ) -> (
245        Box<dyn AsyncWrite + Send + Unpin>,
246        Box<dyn AsyncRead + Send + Unpin>,
247        Option<crate::transport::Io>,
248        Option<Box<dyn AsyncRead + Send + Unpin>>,
249        Option<Ending>,
250    ) {
251        (self.write, self.read, self.control, self.diagnostics, self.end)
252    }
253}
254
255impl Executable {
256    pub fn at(path: impl AsRef<Path>) -> Result<Self, ExecutableError> {
257        let program = path.as_ref().to_path_buf();
258        if !program.is_absolute() {
259            return Err(ExecutableError::Relative { path: program });
260        }
261        let held = std::fs::metadata(&program).map_err(|source| ExecutableError::Unreadable { path: program.clone(), source })?;
262        if !held.is_file() {
263            return Err(ExecutableError::NotFile { path: program });
264        }
265        let bytes = std::fs::read(&program).map_err(|source| ExecutableError::Unreadable { path: program.clone(), source })?;
266        let declaration = Declaration::inspect(&bytes, Facility::Sidecar)
267            .map_err(|source| ExecutableError::Declaration { path: program.clone(), source: Box::new(source) })?;
268        Ok(Self {
269            program,
270            declaration,
271            sha256: Sha256::digest(&bytes).into(),
272            arguments: Vec::new(),
273            directory: None,
274            environment: Vec::new(),
275        })
276    }
277
278    pub fn sha256(&self) -> [u8; 32] {
279        self.sha256
280    }
281
282    pub fn declaration(&self) -> &Declaration {
283        &self.declaration
284    }
285
286    pub fn path(&self) -> &Path {
287        &self.program
288    }
289
290    pub fn arguments(&self) -> &[OsString] {
291        &self.arguments
292    }
293
294    pub fn working_directory(&self) -> Option<&Path> {
295        self.directory.as_deref()
296    }
297
298    pub fn environment(&self) -> &[(OsString, OsString)] {
299        &self.environment
300    }
301
302    pub fn argument(mut self, argument: impl Into<OsString>) -> Self {
303        self.arguments.push(argument.into());
304        self
305    }
306
307    pub fn directory(mut self, path: impl Into<PathBuf>) -> Self {
308        self.directory = Some(path.into());
309        self
310    }
311
312    pub fn variable(mut self, name: impl Into<OsString>, value: impl Into<OsString>) -> Self {
313        self.environment.push((name.into(), value.into()));
314        self
315    }
316}
317
318impl Attach for Executable {
319    type Error = Error;
320
321    fn owner(&self) -> &str {
322        &self.declaration.id
323    }
324
325    fn publications(&self) -> Vec<StreamDeclaration> {
326        self.declaration.streams.clone()
327    }
328
329    fn observers(&self) -> Vec<crate::inventory::observers::Declaration> {
330        self.declaration.observers.clone()
331    }
332
333    fn operations(&self) -> harmos_domain::Result<Vec<crate::operations::Definition>> {
334        Ok(self.declaration.operations.clone())
335    }
336
337    fn operation_contracts(&self) -> harmos_domain::Result<Option<Vec<crate::operations::Contract>>> {
338        Ok(Some(self.declaration.contracts.provides.clone()))
339    }
340
341    fn required_contracts(&self) -> harmos_domain::Result<Option<Vec<crate::inventory::contracts::Required>>> {
342        Ok(Some(self.declaration.contracts.requires.clone()))
343    }
344
345    fn sdk_lifecycle(&self) -> harmos_domain::Result<Option<[crate::operations::lifecycle::Descriptor; 2]>> {
346        Ok(Some(self.declaration.contracts.lifecycle.clone()))
347    }
348
349    fn observe_contract(&self) -> harmos_domain::Result<Option<crate::operations::Contract>> {
350        Ok(self.declaration.observe_contract())
351    }
352
353    fn listener_declarations(&self) -> harmos_domain::Result<Option<Vec<harmos_domain::listeners::Definition>>> {
354        Ok(Some(self.declaration.listeners.clone()))
355    }
356
357    async fn attach(&self) -> Result<Attached, Error> {
358        let mut command = command(&self.program, &self.arguments, self.directory.as_deref(), &self.environment)?;
359
360        #[cfg(any(unix, windows))]
361        let control = control::prepare(&mut command)?;
362        let mut child = command.spawn()?;
363        drop(command);
364        #[cfg(windows)]
365        let control = control.io(child.id().expect("a spawned child has its process id"));
366        let write = child.stdin.take().ok_or_else(|| Error::other("the sidecar was started without a standard input"))?;
367        let read = child.stdout.take().ok_or_else(|| Error::other("the sidecar was started without a standard output"))?;
368        let diagnostics = child.stderr.take();
369
370        let ending = Ending::new(ProcessEnd { child, signaled: false, completed: false });
371        let mut attached = Attached::new(write, read).ending(ending);
372        #[cfg(any(unix, windows))]
373        {
374            attached = attached.control(control);
375        }
376        if let Some(diagnostics) = diagnostics {
377            attached = attached.diagnostics(diagnostics);
378        }
379        Ok(attached)
380    }
381}
382
383/* === Public Functions === */
384
385/* === Private Functions === */
386
387fn command(
388    program: &Path, arguments: &[OsString], directory: Option<&Path>, environment: &[(OsString, OsString)],
389) -> Result<Command, Error> {
390    let mut command = Command::new(program);
391    command.args(arguments).env_clear().stdin(Stdio::piped()).stdout(Stdio::piped()).stderr(Stdio::piped()).kill_on_drop(true);
392    #[cfg(windows)]
393    {
394        command.creation_flags(CREATE_NO_WINDOW);
395        command.env("SystemRoot", system_root()?);
396        let temporary = std::env::temp_dir();
397        command.env("TMP", &temporary).env("TEMP", &temporary);
398    }
399    if !cfg!(windows)
400        && let Some(tmpdir) = std::env::var_os("TMPDIR").filter(|value| !value.is_empty())
401    {
402        command.env("TMPDIR", tmpdir);
403    }
404    command.envs(environment.iter().cloned());
405    if let Some(directory) = directory {
406        command.current_dir(directory);
407    }
408
409    Ok(command)
410}
411
412#[cfg(windows)]
413fn system_root() -> Result<OsString, Error> {
414    use std::os::windows::ffi::OsStringExt;
415
416    use windows_sys::Win32::System::SystemInformation::GetSystemWindowsDirectoryW;
417
418    if let Some(root) = std::env::var_os("SystemRoot").filter(|value| !value.is_empty()) {
419        return Ok(root);
420    }
421
422    let mut buffer = vec![0_u16; 260];
423    loop {
424        // SAFETY: the buffer is writable for the supplied number of UTF-16
425        // units, and the API retains no pointer after this synchronous call.
426        #[allow(unsafe_code)]
427        let length = unsafe { GetSystemWindowsDirectoryW(buffer.as_mut_ptr(), buffer.len() as u32) };
428        if length == 0 {
429            return Err(Error::last_os_error());
430        }
431        if (length as usize) < buffer.len() {
432            return Ok(OsString::from_wide(&buffer[..length as usize]));
433        }
434        buffer.resize(length as usize, 0);
435    }
436}
437
438/* === Tests === */
439
440#[cfg(test)]
441mod tests {
442    use super::*;
443
444    #[test]
445    fn command_configuration_preserved() {
446        let arguments = [OsString::from("--plugin")];
447        let directory = Path::new("work");
448        let environment = [
449            (OsString::from("HARMOS_EXPLICIT"), OsString::from("kept")),
450            (OsString::from("TMPDIR"), OsString::from("chosen-unix-temp")),
451            (OsString::from("TMP"), OsString::from("chosen-windows-temp")),
452            (OsString::from("TEMP"), OsString::from("chosen-windows-temp")),
453        ];
454        let configured = command(Path::new("sidecar"), &arguments, Some(directory), &environment).unwrap();
455        let configured = configured.as_std();
456        assert_eq!(configured.get_program(), "sidecar");
457        assert_eq!(configured.get_args().collect::<Vec<_>>(), [arguments[0].as_os_str()]);
458        assert_eq!(configured.get_current_dir(), Some(directory));
459        let variables = configured.get_envs().collect::<std::collections::BTreeMap<_, _>>();
460        for (name, value) in &environment {
461            assert_eq!(variables.get(name.as_os_str()), Some(&Some(value.as_os_str())));
462        }
463        assert!(!variables.contains_key(std::ffi::OsStr::new("PATH")));
464        #[cfg(windows)]
465        assert!(variables.contains_key(std::ffi::OsStr::new("SystemRoot")));
466    }
467}
harmos_sidecar/attach.rs

harmos_sidecar/
attach.rs
