cordis_cli/
lib.rs

1//! Command-line runner for the [cordis-rs](https://crates.io/crates/cordis-rs)
2//! plugin framework: `cordis run <config.yml>`.
3//!
4//! The process model follows upstream Cordis' NodeLoader: `cordis run`
5//! supervises a worker subprocess running the loader. The worker exits
6//! with code `51` to request a hot restart (respawned with a doubling,
7//! capped backoff), `52` to quit, and `53` when the loader never came up.
8//! The daemon exits `0` on clean shutdown, `1` when the worker never
9//! booted or died abnormally, and otherwise propagates the worker's code —
10//! deployment pipelines always see the real outcome. `SIGINT` / `SIGTERM`
11//! dispose the root context gracefully and exit `52` (daemon `0`). `.env`
12//! and `.env.local` are loaded (without overriding existing variables)
13//! before the worker boots, and the entry file is watched for hot reload.
14//!
15//! `--plugin-dir <dir>` (repeatable) additionally resolves entries from
16//! dynamic-library plugins compiled against the same toolchain; a change
17//! to a library in those directories hot-restarts the worker so the new
18//! build is loaded by a fresh process.
19//!
20//! Plugins reach the process controls through the `worker` service
21//! ([`worker::WorkerHandle`]): `restart()` maps to a full worker reload,
22//! and `shutdown()` stops the whole application.
23
24#![deny(unsafe_code)]
25#![warn(missing_docs)]
26
27pub mod dotenv;
28pub mod worker;
29
30use std::path::PathBuf;
31use std::sync::Arc;
32use std::sync::atomic::{AtomicBool, Ordering};
33use std::time::Duration;
34
35/// Initial delay before respawning a restart-looping worker.
36const RESTART_BACKOFF_START: Duration = Duration::from_millis(100);
37/// Ceiling for the doubling restart delay.
38const RESTART_BACKOFF_MAX: Duration = Duration::from_secs(5);
39/// A worker that stayed up at least this long resets the restart backoff.
40const RESTART_BACKOFF_RESET_AFTER: Duration = Duration::from_secs(10);
41
42/// What the supervisor does after a worker exits.
43#[derive(Debug, Clone, Copy, PartialEq, Eq)]
44pub enum Action {
45    /// Spawn a fresh worker (hot restart).
46    Restart,
47    /// Stop supervising.
48    Stop,
49}
50
51/// Decide whether to restart the worker after it exited with `exit_code`.
52///
53/// Only exit code [`worker::EXIT_RESTART`] (51) restarts, and never once
54/// the daemon itself received a shutdown signal.
55pub fn supervisor_action(exit_code: Option<i32>, shutdown: bool) -> Action {
56    if shutdown || exit_code != Some(worker::EXIT_RESTART) {
57        Action::Stop
58    } else {
59        Action::Restart
60    }
61}
62
63/// The daemon's own exit code after the worker stopped.
64///
65/// A clean quit (52) and a daemon-side shutdown signal exit 0; a worker
66/// that never booted (53) exits 1 so deployment pipelines see the failure;
67/// a crashed worker's code is propagated instead of being masked as
68/// success.
69pub fn daemon_exit_code(exit_code: Option<i32>, shutdown: bool) -> i32 {
70    if shutdown {
71        return 0;
72    }
73    match exit_code {
74        Some(code) if code == worker::EXIT_QUIT => 0,
75        Some(code) if code == worker::EXIT_RESTART => 0,
76        Some(code) if code == worker::EXIT_BOOT => 1,
77        Some(code) => code,
78        None => 1,
79    }
80}
81
82/// The delay before respawning a restart-looping worker: doubling from
83/// [`RESTART_BACKOFF_START`], capped at [`RESTART_BACKOFF_MAX`], and reset
84/// whenever the previous worker stayed up at least
85/// [`RESTART_BACKOFF_RESET_AFTER`].
86fn next_backoff(previous: Option<Duration>, ran_for: Duration) -> Duration {
87    let Some(previous) = previous else {
88        return RESTART_BACKOFF_START;
89    };
90    if ran_for >= RESTART_BACKOFF_RESET_AFTER {
91        return RESTART_BACKOFF_START;
92    }
93    previous.saturating_mul(2).min(RESTART_BACKOFF_MAX)
94}
95
96/// Parsed command line.
97#[derive(Debug, PartialEq, Eq)]
98pub struct Options {
99    /// Entry config file.
100    pub config: PathBuf,
101    /// Directories searched for dynamic-library plugins (repeatable).
102    pub plugin_dirs: Vec<PathBuf>,
103    /// Run as the daemon's worker process instead of supervising.
104    pub worker: bool,
105}
106
107/// Parse `cordis` arguments (after the binary name).
108pub fn parse_args(args: &[String]) -> Result<Options, String> {
109    let Some(command) = args.first() else {
110        return Err(usage());
111    };
112    if command != "run" {
113        return Err(format!("unknown command `{command}`\n\n{}", usage()));
114    }
115    let mut config = None;
116    let mut plugin_dirs = Vec::new();
117    let mut worker = false;
118    let mut rest = args[1..].iter();
119    while let Some(arg) = rest.next() {
120        match arg.as_str() {
121            "--worker" | "-w" => worker = true,
122            "--help" | "-h" => return Err(usage()),
123            "--plugin-dir" => {
124                let Some(value) = rest.next() else {
125                    return Err(format!("--plugin-dir requires a directory\n\n{}", usage()));
126                };
127                plugin_dirs.push(PathBuf::from(value));
128            }
129            other if other.starts_with("--plugin-dir=") => {
130                let value = other.strip_prefix("--plugin-dir=").unwrap();
131                if value.is_empty() {
132                    return Err(format!("--plugin-dir requires a directory\n\n{}", usage()));
133                }
134                plugin_dirs.push(PathBuf::from(value));
135            }
136            other if other.starts_with('-') => {
137                return Err(format!("unknown flag `{other}`\n\n{}", usage()));
138            }
139            other => {
140                if config.replace(PathBuf::from(other)).is_some() {
141                    return Err(format!(
142                        "unexpected extra argument `{other}`\n\n{}",
143                        usage()
144                    ));
145                }
146            }
147        }
148    }
149    match config {
150        Some(config) => Ok(Options {
151            config,
152            plugin_dirs,
153            worker,
154        }),
155        None => Err(format!("missing config file\n\n{}", usage())),
156    }
157}
158
159fn usage() -> String {
160    "usage: cordis run <config.yml> [--plugin-dir <dir>]... [--worker]
161
162  run             start the loader from an entry config file
163  --plugin-dir    also resolve plugins from dynamic libraries in <dir>
164                  (repeatable); library changes there hot-restart the worker
165  --worker        internal: run as the daemon's worker process
166
167Worker exit codes: 51 = hot restart, 52 = quit, 53 = boot failure.
168Daemon exit codes: 0 = clean shutdown, 1 = worker never booted or died
169abnormally, otherwise the worker's own code."
170        .to_owned()
171}
172
173/// Entry point used by the `cordis` binary: parse arguments, load dotenv,
174/// then supervise or run as the worker.
175pub fn run<I, S>(args: I) -> i32
176where
177    I: IntoIterator<Item = S>,
178    S: Into<String>,
179{
180    let args: Vec<String> = args.into_iter().map(Into::into).collect();
181    let options = match parse_args(&args) {
182        Ok(options) => options,
183        Err(message) => {
184            eprintln!("{message}");
185            return 2;
186        }
187    };
188    if let Ok(dir) = std::env::current_dir() {
189        dotenv::load(&dir);
190    }
191    if options.worker {
192        worker::run(&options.config, &options.plugin_dirs);
193    }
194    supervise(&options.config, &options.plugin_dirs)
195}
196
197/// Supervise worker subprocesses until they quit or a signal arrives.
198fn supervise(config: &std::path::Path, plugin_dirs: &[PathBuf]) -> i32 {
199    let shutdown = Arc::new(AtomicBool::new(false));
200    let signal_flag = Arc::clone(&shutdown);
201    if ctrlc::set_handler(move || {
202        eprintln!("cordis: shutdown requested");
203        signal_flag.store(true, Ordering::SeqCst);
204    })
205    .is_err()
206    {
207        eprintln!("cordis: could not install signal handlers");
208    }
209
210    let exe = match std::env::current_exe() {
211        Ok(exe) => exe,
212        Err(error) => {
213            eprintln!("cordis: cannot resolve own executable: {error}");
214            return 1;
215        }
216    };
217    // Restart backoff so a restart loop cannot spin the CPU; a worker that
218    // stayed up long enough resets it.
219    let mut backoff: Option<Duration> = None;
220    let exit_code;
221    loop {
222        if shutdown.load(Ordering::SeqCst) {
223            exit_code = Some(worker::EXIT_QUIT);
224            break;
225        }
226        let started = std::time::Instant::now();
227        let mut command = std::process::Command::new(&exe);
228        command.arg("run").arg(config).arg("--worker");
229        for dir in plugin_dirs {
230            command.arg("--plugin-dir").arg(dir);
231        }
232        let mut child = match command.spawn() {
233            Ok(child) => child,
234            Err(error) => {
235                eprintln!("cordis: cannot spawn worker: {error}");
236                return 1;
237            }
238        };
239        let code = child.wait().ok().and_then(|status| status.code());
240        if supervisor_action(code, shutdown.load(Ordering::SeqCst)) == Action::Stop {
241            exit_code = code;
242            break;
243        }
244        let delay = next_backoff(backoff, started.elapsed());
245        eprintln!(
246            "cordis: worker requested restart, respawning in {}ms",
247            delay.as_millis()
248        );
249        backoff = Some(delay);
250        std::thread::sleep(delay);
251    }
252    daemon_exit_code(exit_code, shutdown.load(Ordering::SeqCst))
253}
254
255#[cfg(test)]
256mod tests {
257    use super::*;
258
259    fn args(list: &[&str]) -> Vec<String> {
260        list.iter().map(ToString::to_string).collect()
261    }
262
263    #[test]
264    fn parses_run_command_and_flags() {
265        assert_eq!(
266            parse_args(&args(&["run", "cordis.yml"])).unwrap(),
267            Options {
268                config: "cordis.yml".into(),
269                plugin_dirs: Vec::new(),
270                worker: false,
271            }
272        );
273        assert_eq!(
274            parse_args(&args(&["run", "cordis.yml", "--worker"])).unwrap(),
275            Options {
276                config: "cordis.yml".into(),
277                plugin_dirs: Vec::new(),
278                worker: true,
279            }
280        );
281    }
282
283    #[test]
284    fn parses_plugin_dirs_in_both_forms() {
285        let options = parse_args(&args(&[
286            "run",
287            "cordis.yml",
288            "--plugin-dir",
289            "a",
290            "--plugin-dir=b",
291        ]))
292        .unwrap();
293        assert_eq!(
294            options.plugin_dirs,
295            [PathBuf::from("a"), PathBuf::from("b")]
296        );
297        assert!(!options.worker);
298    }
299
300    #[test]
301    fn rejects_malformed_plugin_dirs() {
302        assert!(parse_args(&args(&["run", "cordis.yml", "--plugin-dir"])).is_err());
303        assert!(parse_args(&args(&["run", "cordis.yml", "--plugin-dir="])).is_err());
304    }
305
306    #[test]
307    fn rejects_missing_or_unknown_arguments() {
308        assert!(parse_args(&args(&[])).is_err());
309        assert!(parse_args(&args(&["start", "cordis.yml"])).is_err());
310        assert!(parse_args(&args(&["run"])).is_err());
311        assert!(parse_args(&args(&["run", "a.yml", "b.yml"])).is_err());
312        assert!(parse_args(&args(&["run", "a.yml", "--nope"])).is_err());
313    }
314
315    #[test]
316    fn only_code_51_restarts_and_never_after_shutdown() {
317        assert_eq!(supervisor_action(Some(51), false), Action::Restart);
318        assert_eq!(supervisor_action(Some(51), true), Action::Stop);
319        assert_eq!(supervisor_action(Some(52), false), Action::Stop);
320        assert_eq!(supervisor_action(Some(0), false), Action::Stop);
321        assert_eq!(supervisor_action(None, false), Action::Stop);
322    }
323
324    #[test]
325    fn daemon_exit_code_reflects_how_the_worker_ended() {
326        // Clean quit and daemon-side shutdown stay successful.
327        assert_eq!(daemon_exit_code(Some(52), false), 0);
328        assert_eq!(daemon_exit_code(Some(52), true), 0);
329        assert_eq!(daemon_exit_code(Some(51), true), 0);
330        // A worker that never booted fails the daemon.
331        assert_eq!(daemon_exit_code(Some(53), false), 1);
332        // Crashes propagate instead of being masked as success.
333        assert_eq!(daemon_exit_code(Some(101), false), 101);
334        assert_eq!(daemon_exit_code(None, false), 1);
335    }
336
337    #[test]
338    fn restart_backoff_doubles_resets_and_caps() {
339        assert_eq!(
340            next_backoff(None, Duration::from_secs(0)),
341            RESTART_BACKOFF_START
342        );
343        assert_eq!(
344            next_backoff(Some(Duration::from_millis(100)), Duration::from_secs(1)),
345            Duration::from_millis(200)
346        );
347        assert_eq!(
348            next_backoff(Some(Duration::from_secs(4)), Duration::from_secs(1)),
349            RESTART_BACKOFF_MAX
350        );
351        assert_eq!(
352            next_backoff(Some(Duration::from_secs(4)), Duration::from_secs(60)),
353            RESTART_BACKOFF_START,
354            "a worker that stayed up resets the backoff"
355        );
356    }
357}
cordis_cli/lib.rs

cordis_cli/
lib.rs
