isothrum_runner/
runner.rs

1//! The engine: one per embedder. It holds what the launcher kept in process-wide statics (the
2//! operations in flight, the per-run locks, the ports being pinned), so two runners in one
3//! process never share them.
4
5use std::collections::BTreeMap;
6use std::path::{Path, PathBuf};
7use std::sync::{Arc, Mutex, PoisonError};
8
9use isoloom_manifest::{HostOs, Protocol};
10
11use crate::Result;
12use crate::clock::{Clock, SystemClock};
13use crate::events::Event;
14use crate::hooks::Hooks;
15use crate::manifest::load_manifest;
16use crate::name::RunName;
17use crate::net::{HostPorts, LoopbackPorts};
18use crate::process::{ProcessRunner, SystemProcesses};
19use crate::run::Running;
20use crate::state::Op;
21use crate::store::{RunLocks, RunRecord, Store};
22
23/// How a runner is built: its home, and the ports it reaches the world through.
24#[derive(Clone)]
25#[non_exhaustive]
26pub struct RunnerConfig {
27    /// The runner's home (`runs/` under it). The launcher passes its app data folder; the
28    /// `isothrum` command line uses `~/.isothrum`.
29    pub home: PathBuf,
30    /// Starts the tools.
31    pub processes: Arc<dyn ProcessRunner>,
32    /// The host's ports.
33    pub ports: Arc<dyn HostPorts>,
34    /// The time.
35    pub clock: Arc<dyn Clock>,
36    /// The host's operating system (it picks per-host argv and filters steps).
37    pub host: HostOs,
38    /// The tools' `PATH`, where the macOS relays look for the real `ovftool` and put their
39    /// wrapper first (default: the runner's own).
40    pub path: Option<String>,
41}
42
43impl RunnerConfig {
44    /// A runner at `home` on this host: real processes, the loopback, the system's clock.
45    pub fn new(home: impl Into<PathBuf>) -> RunnerConfig {
46        RunnerConfig {
47            home: home.into(),
48            processes: Arc::new(SystemProcesses::default()),
49            ports: Arc::new(LoopbackPorts),
50            clock: Arc::new(SystemClock),
51            host: this_host(),
52            path: std::env::var("PATH").ok(),
53        }
54    }
55}
56
57impl RunnerConfig {
58    /// Starts the tools through `processes` (a fake in tests).
59    #[must_use]
60    pub fn with_processes(mut self, processes: Arc<dyn ProcessRunner>) -> RunnerConfig {
61        self.processes = processes;
62        self
63    }
64
65    /// Reads the host's ports through `ports`.
66    #[must_use]
67    pub fn with_ports(mut self, ports: Arc<dyn HostPorts>) -> RunnerConfig {
68        self.ports = ports;
69        self
70    }
71
72    /// Reads the time from `clock`.
73    #[must_use]
74    pub fn with_clock(mut self, clock: Arc<dyn Clock>) -> RunnerConfig {
75        self.clock = clock;
76        self
77    }
78}
79
80impl std::fmt::Debug for RunnerConfig {
81    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
82        f.debug_struct("RunnerConfig")
83            .field("home", &self.home)
84            .field("host", &self.host)
85            .finish_non_exhaustive()
86    }
87}
88
89/// This build's host operating system.
90#[must_use]
91pub fn this_host() -> HostOs {
92    if cfg!(windows) {
93        HostOs::Windows
94    } else if cfg!(target_os = "macos") {
95        HostOs::Macos
96    } else {
97        HostOs::Linux
98    }
99}
100
101/// The runner.
102pub struct Runner {
103    pub(crate) config: RunnerConfig,
104    pub(crate) store: Store,
105    pub(crate) hooks: Vec<Arc<dyn Hooks>>,
106    pub(crate) locks: RunLocks,
107    flight: Mutex<BTreeMap<RunName, Op>>,
108    reserved: Mutex<BTreeMap<(u16, Protocol), RunName>>,
109}
110
111impl Runner {
112    /// A runner with `config`; its home's `runs/` folder is created owner-only.
113    ///
114    /// # Errors
115    ///
116    /// Fails with `io` when the home's `runs/` folder can't be created.
117    pub fn new(config: RunnerConfig) -> Result<Runner> {
118        let store = Store::open(&config.home)?;
119        Ok(Runner {
120            config,
121            store,
122            hooks: Vec::new(),
123            locks: RunLocks::default(),
124            flight: Mutex::default(),
125            reserved: Mutex::default(),
126        })
127    }
128
129    /// Adds the embedder's hooks around lifecycle edges.
130    pub fn add_hooks(&mut self, hooks: Arc<dyn Hooks>) {
131        self.hooks.push(hooks);
132    }
133
134    /// Reads and checks the manifest at `manifest`, as run `name` (default: the run's name).
135    ///
136    /// # Errors
137    ///
138    /// Fails as [`load_manifest`](crate::load_manifest) does; with `run_name` when the run's name
139    /// can't name a run; with `unsupported` when this runner can't run the manifest (its driver, a
140    /// delivery, this host).
141    pub fn open(&self, manifest: &Path, name: Option<RunName>) -> Result<Running<'_>> {
142        let loaded = load_manifest(manifest)?;
143        let name = match name {
144            Some(name) => name,
145            None => RunName::new(loaded.manifest.environment.name.as_str())?,
146        };
147        Running::new(self, loaded, name)
148    }
149
150    /// Every recorded run, with the state its last operation left it in. Reads files only.
151    ///
152    /// # Errors
153    ///
154    /// Fails with `io` when a record can't be read.
155    pub fn runs(&self) -> Result<Vec<RunRecord>> {
156        self.store.records()
157    }
158
159    /// Run `name`'s logged events, oldest first.
160    ///
161    /// # Errors
162    ///
163    /// Fails with `io` when the run's event log can't be read.
164    pub fn events(&self, name: &RunName) -> Result<Vec<Event>> {
165        self.store.events(name)
166    }
167
168    /// The public half of the runner's SSH key, one line: what an embedder authorizes where a
169    /// run's SSH goes (a Proxmox node it jumps through). The key is made when the runner has none.
170    ///
171    /// # Errors
172    ///
173    /// Fails with `tool_missing` when `ssh-keygen` isn't installed, `tool_failed` or `timeout`
174    /// when it can't make the key, and `io` when the runner's `ssh/` folder can't be created.
175    pub async fn ssh_public_key(&self) -> Result<String> {
176        crate::ssh::ensure_key(self.config.processes.as_ref(), &self.config.home).await
177    }
178
179    /// The operations in flight in this runner (a quit guard, a UI rehydrating).
180    pub fn in_flight(&self) -> Vec<(RunName, Op)> {
181        self.flight
182            .lock()
183            .unwrap_or_else(PoisonError::into_inner)
184            .iter()
185            .map(|(n, o)| (n.clone(), *o))
186            .collect()
187    }
188
189    /// Marks `op` on `name` in flight until the guard drops.
190    pub(crate) fn enter(&self, name: &RunName, op: Op) -> Flight<'_> {
191        self.flight.lock().unwrap_or_else(PoisonError::into_inner).insert(name.clone(), op);
192        Flight {
193            runner: self,
194            name: name.clone(),
195        }
196    }
197
198    /// The operation in flight on `name`, if any.
199    pub(crate) fn op_of(&self, name: &RunName) -> Option<Op> {
200        self.flight.lock().unwrap_or_else(PoisonError::into_inner).get(name).copied()
201    }
202
203    /// Reserves `port` for `name` while it starts; false when another run holds it.
204    pub(crate) fn reserve(&self, name: &RunName, port: u16, protocol: Protocol) -> bool {
205        let mut reserved = self.reserved.lock().unwrap_or_else(PoisonError::into_inner);
206        let holder = reserved.entry((port, protocol)).or_insert_with(|| name.clone());
207        holder == name
208    }
209
210    /// Whether another run than `name` holds `port`.
211    pub(crate) fn reserved_elsewhere(&self, name: &RunName, port: u16, protocol: Protocol) -> bool {
212        self.reserved
213            .lock()
214            .unwrap_or_else(PoisonError::into_inner)
215            .get(&(port, protocol))
216            .is_some_and(|h| h != name)
217    }
218
219    /// Frees every port `name` reserved.
220    pub(crate) fn release(&self, name: &RunName) {
221        self.reserved.lock().unwrap_or_else(PoisonError::into_inner).retain(|_, holder| holder != name);
222    }
223}
224
225impl std::fmt::Debug for Runner {
226    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
227        f.debug_struct("Runner")
228            .field("config", &self.config)
229            .field("hooks", &self.hooks.len())
230            .field("in_flight", &self.in_flight())
231            .finish_non_exhaustive()
232    }
233}
234
235/// An operation in flight; cleared when dropped, whatever the outcome (an error, a cancellation).
236pub(crate) struct Flight<'r> {
237    runner: &'r Runner,
238    name: RunName,
239}
240
241impl Drop for Flight<'_> {
242    fn drop(&mut self) {
243        self.runner.flight.lock().unwrap_or_else(PoisonError::into_inner).remove(&self.name);
244        self.runner.release(&self.name);
245    }
246}
247
248#[cfg(test)]
249mod tests {
250    use super::*;
251
252    fn runner(home: &Path) -> Runner {
253        Runner::new(RunnerConfig::new(home)).unwrap()
254    }
255
256    #[test]
257    fn two_runners_share_nothing() {
258        // Regression (R5): the launcher's in-flight map, run locks and probe cache were statics,
259        // so two embedders (or two tests) in one process saw each other's operations.
260        let (a, b) = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap());
261        let (a, b) = (runner(a.path()), runner(b.path()));
262        let name = RunName::new("lab").unwrap();
263        let flight = a.enter(&name, Op::Up);
264        assert!(a.reserve(&name, 8080, Protocol::Tcp));
265        assert_eq!(a.in_flight(), [(name.clone(), Op::Up)]);
266        assert_eq!(b.in_flight(), []);
267        assert!(b.reserve(&RunName::new("other").unwrap(), 8080, Protocol::Tcp), "b's ports aren't a's");
268        drop(flight);
269        assert_eq!(a.in_flight(), []);
270    }
271
272    #[test]
273    fn a_port_reserved_by_one_run_is_refused_to_another_until_released() {
274        let home = tempfile::tempdir().unwrap();
275        let r = runner(home.path());
276        let (one, two) = (RunName::new("one").unwrap(), RunName::new("two").unwrap());
277        assert!(r.reserve(&one, 41000, Protocol::Tcp));
278        assert!(r.reserve(&one, 41000, Protocol::Tcp), "its own reservation again");
279        assert!(!r.reserve(&two, 41000, Protocol::Tcp));
280        assert!(r.reserved_elsewhere(&two, 41000, Protocol::Tcp) && !r.reserved_elsewhere(&one, 41000, Protocol::Tcp));
281        assert!(r.reserve(&two, 41000, Protocol::Udp), "another protocol");
282        r.release(&one);
283        assert!(r.reserve(&two, 41000, Protocol::Tcp));
284    }
285}
isothrum_runner/runner.rs

isothrum_runner/
runner.rs
