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

cordis_cli/
worker.rs
