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::OperationDefinition>> {
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<[harmos_domain::model::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 {
168            start: Some(Box::new(move || Box::pin(end()))),
169            work: None,
170            result: None,
171        })
172    }
173
174    pub async fn settle(&mut self) -> std::io::Result<()> {
175        self.receipt.settle().await
176    }
177}
178
179impl End for Terminal {
180    fn settle(&mut self) -> Pin<Box<dyn Future<Output = std::io::Result<()>> + Send + '_>> {
181        Box::pin(async move {
182            if let Some(result) = &self.result {
183                return result.clone().map_err(Error::other);
184            }
185            if self.work.is_none() {
186                let start = self.start.take().ok_or_else(|| Error::other("terminal ending construction did not return"))?;
187                self.work = Some(start());
188            }
189            let result = self.work.as_mut().expect("original terminal work is retained").await;
190            self.work = None;
191            self.result = Some(result.map_err(Arc::new));
192            self.result.as_ref().expect("terminal outcome is retained").clone().map_err(Error::other)
193        })
194    }
195}
196
197impl End for ProcessEnd {
198    fn settle(&mut self) -> Pin<Box<dyn Future<Output = std::io::Result<()>> + Send + '_>> {
199        Box::pin(async move {
200            if self.completed {
201                return Ok(());
202            }
203            if !self.signaled {
204                self.child.start_kill()?;
205                self.signaled = true;
206            }
207            self.child.wait().await?;
208            self.completed = true;
209            Ok(())
210        })
211    }
212}
213
214impl Attached {
215    pub fn new(write: impl AsyncWrite + Send + Unpin + 'static, read: impl AsyncRead + Send + Unpin + 'static) -> Self {
216        Self {
217            control: None,
218            write: Box::new(write),
219            read: Box::new(read),
220            diagnostics: None,
221            end: None,
222        }
223    }
224
225    pub fn control(mut self, channel: crate::transport::Io) -> Self {
226        self.control = Some(channel);
227        self
228    }
229
230    pub fn diagnostics(mut self, read: impl AsyncRead + Send + Unpin + 'static) -> Self {
231        self.diagnostics = Some(Box::new(read));
232        self
233    }
234
235    pub fn ended_by<F, U>(self, end: F) -> Self
236    where
237        F: FnOnce() -> U + Send + 'static,
238        U: Future<Output = ()> + Send + 'static,
239    {
240        self.ending(Ending::terminal(move || async move {
241            end().await;
242            Ok(())
243        }))
244    }
245
246    pub fn ending(mut self, ending: Ending) -> Self {
247        self.end = Some(ending);
248        self
249    }
250
251    #[allow(clippy::type_complexity)]
252    pub fn into_parts(
253        self,
254    ) -> (
255        Box<dyn AsyncWrite + Send + Unpin>,
256        Box<dyn AsyncRead + Send + Unpin>,
257        Option<crate::transport::Io>,
258        Option<Box<dyn AsyncRead + Send + Unpin>>,
259        Option<Ending>,
260    ) {
261        (self.write, self.read, self.control, self.diagnostics, self.end)
262    }
263}
264
265impl Executable {
266    pub fn at(path: impl AsRef<Path>) -> Result<Self, ExecutableError> {
267        let program = path.as_ref().to_path_buf();
268        if !program.is_absolute() {
269            return Err(ExecutableError::Relative { path: program });
270        }
271        let held = std::fs::metadata(&program).map_err(|source| ExecutableError::Unreadable { path: program.clone(), source })?;
272        if !held.is_file() {
273            return Err(ExecutableError::NotFile { path: program });
274        }
275        let bytes = std::fs::read(&program).map_err(|source| ExecutableError::Unreadable { path: program.clone(), source })?;
276        let declaration = Declaration::inspect(&bytes, Facility::Sidecar).map_err(|source| ExecutableError::Declaration {
277            path: program.clone(),
278            source: Box::new(source),
279        })?;
280        Ok(Self {
281            program,
282            declaration,
283            sha256: Sha256::digest(&bytes).into(),
284            arguments: Vec::new(),
285            directory: None,
286            environment: Vec::new(),
287        })
288    }
289
290    pub fn sha256(&self) -> [u8; 32] {
291        self.sha256
292    }
293
294    pub fn declaration(&self) -> &Declaration {
295        &self.declaration
296    }
297
298    pub fn path(&self) -> &Path {
299        &self.program
300    }
301
302    pub fn arguments(&self) -> &[OsString] {
303        &self.arguments
304    }
305
306    pub fn working_directory(&self) -> Option<&Path> {
307        self.directory.as_deref()
308    }
309
310    pub fn environment(&self) -> &[(OsString, OsString)] {
311        &self.environment
312    }
313
314    pub fn argument(mut self, argument: impl Into<OsString>) -> Self {
315        self.arguments.push(argument.into());
316        self
317    }
318
319    pub fn directory(mut self, path: impl Into<PathBuf>) -> Self {
320        self.directory = Some(path.into());
321        self
322    }
323
324    pub fn variable(mut self, name: impl Into<OsString>, value: impl Into<OsString>) -> Self {
325        self.environment.push((name.into(), value.into()));
326        self
327    }
328}
329
330impl Attach for Executable {
331    type Error = Error;
332
333    fn owner(&self) -> &str {
334        &self.declaration.id
335    }
336
337    fn publications(&self) -> Vec<StreamDeclaration> {
338        self.declaration.streams.clone()
339    }
340
341    fn observers(&self) -> Vec<crate::inventory::observers::Declaration> {
342        self.declaration.observers.clone()
343    }
344
345    fn operations(&self) -> harmos_domain::Result<Vec<crate::operations::OperationDefinition>> {
346        Ok(self.declaration.operations.clone())
347    }
348
349    fn operation_contracts(&self) -> harmos_domain::Result<Option<Vec<crate::operations::Contract>>> {
350        Ok(Some(self.declaration.contracts.provides.clone()))
351    }
352
353    fn required_contracts(&self) -> harmos_domain::Result<Option<Vec<crate::inventory::contracts::Required>>> {
354        Ok(Some(self.declaration.contracts.requires.clone()))
355    }
356
357    fn sdk_lifecycle(&self) -> harmos_domain::Result<Option<[harmos_domain::model::operations::lifecycle::Descriptor; 2]>> {
358        Ok(Some(self.declaration.contracts.lifecycle.clone()))
359    }
360
361    fn observe_contract(&self) -> harmos_domain::Result<Option<crate::operations::Contract>> {
362        Ok(self.declaration.observe_contract())
363    }
364
365    fn listener_declarations(&self) -> harmos_domain::Result<Option<Vec<harmos_domain::listeners::Definition>>> {
366        Ok(Some(self.declaration.listeners.clone()))
367    }
368
369    async fn attach(&self) -> Result<Attached, Error> {
370        let mut command = command(&self.program, &self.arguments, self.directory.as_deref(), &self.environment)?;
371
372        #[cfg(any(unix, windows))]
373        let control = control::prepare(&mut command)?;
374        let mut child = command.spawn()?;
375        tracing::debug!(target: "harmos::carrier", provider = %self.declaration.id, remote_pid = child.id().expect("a spawned child has its process id"), "sidecar attached");
376        drop(command);
377        #[cfg(windows)]
378        let control = control.io(child.id().expect("a spawned child has its process id"));
379        let write = child.stdin.take().ok_or_else(|| Error::other("the sidecar was started without a standard input"))?;
380        let read = child.stdout.take().ok_or_else(|| Error::other("the sidecar was started without a standard output"))?;
381        let diagnostics = child.stderr.take();
382
383        let ending = Ending::new(ProcessEnd { child, signaled: false, completed: false });
384        let mut attached = Attached::new(write, read).ending(ending);
385        #[cfg(any(unix, windows))]
386        {
387            attached = attached.control(control);
388        }
389        if let Some(diagnostics) = diagnostics {
390            attached = attached.diagnostics(diagnostics);
391        }
392        Ok(attached)
393    }
394}
395
396/* === Public Functions === */
397
398/* === Private Functions === */
399
400fn command(
401    program: &Path, arguments: &[OsString], directory: Option<&Path>, environment: &[(OsString, OsString)],
402) -> Result<Command, Error> {
403    let mut command = Command::new(program);
404    command.args(arguments).env_clear().stdin(Stdio::piped()).stdout(Stdio::piped()).stderr(Stdio::piped()).kill_on_drop(true);
405    #[cfg(windows)]
406    {
407        command.creation_flags(CREATE_NO_WINDOW);
408        command.env("SystemRoot", system_root()?);
409        let temporary = std::env::temp_dir();
410        command.env("TMP", &temporary).env("TEMP", &temporary);
411    }
412    if !cfg!(windows)
413        && let Some(tmpdir) = std::env::var_os("TMPDIR").filter(|value| !value.is_empty())
414    {
415        command.env("TMPDIR", tmpdir);
416    }
417    command.envs(environment.iter().cloned());
418    if let Some(directory) = directory {
419        command.current_dir(directory);
420    }
421
422    Ok(command)
423}
424
425#[cfg(windows)]
426fn system_root() -> Result<OsString, Error> {
427    use std::os::windows::ffi::OsStringExt;
428
429    use windows_sys::Win32::System::SystemInformation::GetSystemWindowsDirectoryW;
430
431    if let Some(root) = std::env::var_os("SystemRoot").filter(|value| !value.is_empty()) {
432        return Ok(root);
433    }
434
435    let mut buffer = vec![0_u16; 260];
436    loop {
437        // SAFETY: the buffer is writable for the supplied number of UTF-16
438        // units, and the API retains no pointer after this synchronous call.
439        #[allow(unsafe_code)]
440        let length = unsafe { GetSystemWindowsDirectoryW(buffer.as_mut_ptr(), buffer.len() as u32) };
441        if length == 0 {
442            return Err(Error::last_os_error());
443        }
444        if (length as usize) < buffer.len() {
445            return Ok(OsString::from_wide(&buffer[..length as usize]));
446        }
447        buffer.resize(length as usize, 0);
448    }
449}
450
451/* === Tests === */
452
453#[cfg(test)]
454mod tests {
455    use super::*;
456
457    #[test]
458    fn command_configuration_preserved() {
459        let arguments = [OsString::from("--plugin")];
460        let directory = Path::new("work");
461        let environment = [
462            (OsString::from("HARMOS_EXPLICIT"), OsString::from("kept")),
463            (OsString::from("TMPDIR"), OsString::from("chosen-unix-temp")),
464            (OsString::from("TMP"), OsString::from("chosen-windows-temp")),
465            (OsString::from("TEMP"), OsString::from("chosen-windows-temp")),
466        ];
467        let configured = command(Path::new("sidecar"), &arguments, Some(directory), &environment).unwrap();
468        let configured = configured.as_std();
469        assert_eq!(configured.get_program(), "sidecar");
470        assert_eq!(configured.get_args().collect::<Vec<_>>(), [arguments[0].as_os_str()]);
471        assert_eq!(configured.get_current_dir(), Some(directory));
472        let variables = configured.get_envs().collect::<std::collections::BTreeMap<_, _>>();
473        for (name, value) in &environment {
474            assert_eq!(variables.get(name.as_os_str()), Some(&Some(value.as_os_str())));
475        }
476        assert!(!variables.contains_key(std::ffi::OsStr::new("PATH")));
477        #[cfg(windows)]
478        assert!(variables.contains_key(std::ffi::OsStr::new("SystemRoot")));
479    }
480}
harmos_sidecar/attach.rs

harmos_sidecar/
attach.rs
