cordis_cli/
worker.rs

1//! Worker runtime: boot the loader, watch the config, exit on signals.
2
3use cordis::Context;
4use cordis_loader::{Loader, LoaderConfig, PluginRegistry};
5use notify::{Config as NotifyConfig, RecursiveMode, Watcher};
6use std::io::Read;
7use std::path::{Path, PathBuf};
8use std::sync::Arc;
9use std::sync::mpsc;
10use std::time::Duration;
11
12/// Exit code asking the daemon for a hot restart.
13pub const EXIT_RESTART: i32 = 51;
14/// Exit code telling the daemon to quit without restarting.
15pub const EXIT_QUIT: i32 = 52;
16/// Exit code reporting that the loader never came up (bad config,
17/// unreadable file); the daemon exits non-zero instead of masking it.
18pub const EXIT_BOOT: i32 = 53;
19/// Environment marker set by the daemon on workers it supervises. Such
20/// workers watch their stdin pipe: the daemon holds the write end, so EOF
21/// means the daemon is going away (clean shutdown or sudden death) and the
22/// worker tears itself down gracefully.
23pub const SUPERVISED_ENV: &str = "CORDIS_SUPERVISED";
24
25/// Quiet window a plugin library must stay unchanged before the worker
26/// restarts, so incremental linker output does not restart mid-write.
27const PLUGIN_CHANGE_DEBOUNCE: Duration = Duration::from_millis(250);
28
29/// Handle exposed as the `worker` service so plugins can stop or restart
30/// the process (upstream's `ctx.loader.exit` / full-reload protocol).
31pub struct WorkerHandle {
32    inner: Arc<WorkerInner>,
33}
34
35impl WorkerHandle {
36    /// Hot restart: dispose everything and ask the daemon for a new worker.
37    pub fn restart(&self) -> ! {
38        self.inner.teardown();
39        std::process::exit(EXIT_RESTART);
40    }
41
42    /// Quit: dispose everything and tell the daemon not to restart.
43    pub fn shutdown(&self) -> ! {
44        self.inner.teardown();
45        std::process::exit(EXIT_QUIT);
46    }
47}
48
49/// Everything the worker owns; shared with the signal handler.
50struct WorkerInner {
51    root: Context,
52    loader: Option<Loader>,
53}
54
55impl WorkerInner {
56    fn teardown(&self) {
57        if let Some(loader) = &self.loader {
58            let _ = loader.dispose();
59        }
60        let _ = self.root.fiber().and_then(|fiber| fiber.dispose());
61    }
62}
63
64/// Run the worker process: load dotenv, boot the loader (with dynamic
65/// plugin directories, if any), watch the entry file and the plugin
66/// directories, and block until a signal (or a `worker` service call)
67/// exits the process. Never returns.
68pub fn run(config_path: &Path, plugin_dirs: &[PathBuf]) -> ! {
69    let root = Context::new();
70    let mut registry = PluginRegistry::new();
71    if !plugin_dirs.is_empty() {
72        registry = registry.with_dynamic_dirs(plugin_dirs.iter());
73    }
74    let loader = match Loader::open(
75        &root,
76        LoaderConfig::new(config_path).with_registry(registry),
77    ) {
78        Ok(loader) => loader,
79        Err(error) => {
80            eprintln!(
81                "cordis: failed to start from {}: {error}",
82                config_path.display()
83            );
84            std::process::exit(EXIT_BOOT);
85        }
86    };
87    let inner = Arc::new(WorkerInner {
88        root: root.clone(),
89        loader: Some(loader.clone()),
90    });
91
92    let handle = Arc::new(WorkerHandle {
93        inner: Arc::clone(&inner),
94    });
95    if let Err(error) = root.provide_arc("worker", handle.clone()) {
96        eprintln!("cordis: could not expose the worker service: {error}");
97    }
98
99    let signal_inner = Arc::clone(&inner);
100    if ctrlc::set_handler(move || {
101        eprintln!("cordis: signal received, shutting down");
102        signal_inner.teardown();
103        std::process::exit(EXIT_QUIT);
104    })
105    .is_err()
106    {
107        eprintln!("cordis: could not install signal handlers");
108    }
109
110    // Under the daemon, watch the supervisor's stdin pipe. EOF (or an
111    // errored pipe) means the daemon is gone or asked us to stop — the
112    // same graceful teardown a signal triggers, and the only notification
113    // a SIGKILLed daemon can still send. Not spawned for manually-run
114    // workers, so a terminal's stdin stays untouched.
115    if std::env::var_os(SUPERVISED_ENV).is_some() {
116        let inner = Arc::clone(&inner);
117        let watched = std::thread::Builder::new()
118            .name("cordis-supervisor-watch".to_owned())
119            .spawn(move || {
120                let mut stdin = std::io::stdin();
121                let mut byte = [0_u8];
122                loop {
123                    match stdin.read(&mut byte) {
124                        Ok(0) | Err(_) => break,
125                        Ok(_) => continue,
126                    }
127                }
128                eprintln!("cordis: supervisor went away, shutting down");
129                inner.teardown();
130                std::process::exit(EXIT_QUIT);
131            });
132        if watched.is_err() {
133            eprintln!("cordis: could not watch the supervisor pipe");
134        }
135    }
136
137    match loader.watch() {
138        Ok(_watcher) => {}
139        Err(error) => eprintln!(
140            "cordis: config hot reload disabled ({error}); restart manually to apply changes"
141        ),
142    }
143
144    if !plugin_dirs.is_empty() {
145        // A changed library only takes effect through a fresh worker: the
146        // old process never unloads a mapped library, so the watcher asks
147        // the daemon for a restart instead of reloading in place.
148        match watch_plugin_dirs(plugin_dirs, handle) {
149            Ok(()) => eprintln!(
150                "cordis: dynamic plugins from {} (library changes hot-restart the worker)",
151                plugin_dirs
152                    .iter()
153                    .map(|dir| dir.display().to_string())
154                    .collect::<Vec<_>>()
155                    .join(", ")
156            ),
157            Err(error) => eprintln!(
158                "cordis: plugin library watching disabled ({error}); restart manually to apply changes"
159            ),
160        }
161    }
162
163    if let Some(error) = loader.last_error() {
164        eprintln!("cordis: startup issue: {error}");
165    }
166
167    eprintln!(
168        "cordis: worker ready ({} entries, config: {})",
169        loader.tree().entries().len(),
170        config_path.display()
171    );
172
173    // The worker's work happens on fiber threads and watcher callbacks;
174    // park until a signal or service call ends the process.
175    loop {
176        std::thread::park();
177    }
178}
179
180/// Watch `dirs` (non-recursively) for plugin library changes and hot
181/// restart the worker through `handle` once a change settles.
182///
183/// Only files with a dynamic-library extension count; matching by
184/// extension (instead of exact paths) also absorbs macOS FSEvents
185/// reporting realpaths. The thread ends the process on the first settled
186/// change, so there is nothing to return.
187fn watch_plugin_dirs(dirs: &[PathBuf], handle: Arc<WorkerHandle>) -> Result<(), String> {
188    let (tx, rx) = mpsc::channel();
189    let mut watcher = notify::RecommendedWatcher::new(tx, NotifyConfig::default())
190        .map_err(|error| format!("cannot create watcher: {error}"))?;
191    for dir in dirs {
192        watcher
193            .watch(dir, RecursiveMode::NonRecursive)
194            .map_err(|error| format!("cannot watch {}: {error}", dir.display()))?;
195    }
196
197    std::thread::Builder::new()
198        .name("cordis-plugin-watch".to_owned())
199        .spawn(move || {
200            let _watcher = watcher; // keep the watch alive for the thread's lifetime
201            let mut pending = false;
202            loop {
203                match rx.recv_timeout(PLUGIN_CHANGE_DEBOUNCE) {
204                    Ok(Ok(event)) => {
205                        if event.paths.iter().any(|path| is_plugin_library(path)) {
206                            pending = true;
207                        }
208                    }
209                    Ok(Err(_)) => {}
210                    Err(mpsc::RecvTimeoutError::Timeout) => {
211                        if pending {
212                            eprintln!("cordis: plugin library changed, restarting worker");
213                            handle.restart();
214                        }
215                    }
216                    Err(mpsc::RecvTimeoutError::Disconnected) => break,
217                }
218            }
219        })
220        .map_err(|error| format!("cannot spawn watcher thread: {error}"))?;
221    Ok(())
222}
223
224/// Whether `path` looks like a dynamic library by extension.
225fn is_plugin_library(path: &Path) -> bool {
226    path.extension().is_some_and(|extension| {
227        matches!(
228            extension.to_ascii_lowercase().to_str(),
229            Some("so" | "dylib" | "dll")
230        )
231    })
232}
233
234#[cfg(test)]
235mod tests {
236    use super::*;
237
238    #[test]
239    fn dynamic_library_extensions_match() {
240        assert!(is_plugin_library(Path::new("libgreeter.so")));
241        assert!(is_plugin_library(Path::new("libgreeter.dylib")));
242        assert!(is_plugin_library(Path::new("greeter.dll")));
243        assert!(is_plugin_library(Path::new("GREETER.SO")));
244        assert!(!is_plugin_library(Path::new("cordis.yml")));
245        assert!(!is_plugin_library(Path::new("libgreeter.so.tmp")));
246        assert!(!is_plugin_library(Path::new("plugins")));
247    }
248}
cordis_cli/worker.rs

cordis_cli/
worker.rs
