xianyun_cli/commands/
editmodeprocesssupervisor.rs

1//! `caproni edit-mode-process-supervisor`: hidden internal command started
2//! by `caproni run` for each reloader.
3//!
4//! Port of Go's `cmd/editmodeprocesssupervisor` package.
5
6use std::collections::HashMap;
7
8use anyhow::Context as _;
9use clap::Args;
10use thiserror::Error;
11use xianyun_core::config::global::Context;
12use xianyun_core::config::latest::{PROCESS_CONTEXT_MIRRORD, ProcessConfig};
13
14pub const COMMAND_NAME: &str = "edit-mode-process-supervisor";
15
16/// Default process context, mirroring Go's `latest.ProcessContextMirrord`.
17pub const DEFAULT_PROCESS_CONTEXT: &str = PROCESS_CONTEXT_MIRRORD;
18
19#[derive(Args, Debug)]
20pub struct EditModeProcessSupervisorArgs {
21    /// Name of the reloader to run processes for.
22    #[arg(long = "reloader", required = true)]
23    pub reloader: String,
24
25    /// Directory to write process log files.
26    #[arg(long = "logs-dir")]
27    pub logs_dir: Option<String>,
28
29    /// Supervise edit mode processes in this context (mirrord or local).
30    #[arg(long = "context", default_value = DEFAULT_PROCESS_CONTEXT)]
31    pub context: String,
32}
33
34/// Ported from the package-level `var` sentinel errors.
35#[derive(Debug, Error, PartialEq, Eq, Clone)]
36pub enum SupervisorError {
37    #[error("reloader not found: {0:?}")]
38    ReloaderNotFound(String),
39    #[error("reloader has no edit mode processes configured: {0:?}")]
40    NoEditModeProcesses(String),
41    #[error("repository not found: {0:?}")]
42    RepositoryNotFound(String),
43    #[error("process exited abnormally")]
44    ProcessFailed,
45}
46
47/// Ported from `escapeFilename`-adjacent `shellwords.Parse` error path.
48/// This crate has no `mattn/go-shellwords` equivalent dependency, so a
49/// minimal POSIX-lite splitter (single/double quotes, backslash escapes
50/// outside quotes) is implemented here directly -- just enough to detect
51/// the unbalanced-quote case Go's library rejects, without adding a new
52/// external dependency for one narrow use.
53#[derive(Debug, Error, PartialEq, Eq, Clone, Copy)]
54pub enum ShellwordsError {
55    #[error("unmatched quote")]
56    UnmatchedQuote,
57}
58
59/// Splits a command string into words, honoring single/double quotes and
60/// backslash escapes outside of quotes. Ported (narrowly) from
61/// `mattn/go-shellwords`'s `Parse`.
62pub fn split_command(input: &str) -> Result<Vec<String>, ShellwordsError> {
63    #[derive(PartialEq)]
64    enum State {
65        Normal,
66        Word,
67        Single,
68        Double,
69    }
70
71    let mut state = State::Normal;
72    let mut words = Vec::new();
73    let mut current = String::new();
74    let mut chars = input.chars();
75
76    while let Some(c) = chars.next() {
77        match state {
78            State::Normal | State::Word => match c {
79                _ if c.is_whitespace() && state == State::Word => {
80                    words.push(std::mem::take(&mut current));
81                    state = State::Normal;
82                }
83                _ if c.is_whitespace() => {}
84                '\'' => state = State::Single,
85                '"' => state = State::Double,
86                '\\' => {
87                    if let Some(next) = chars.next() {
88                        current.push(next);
89                    }
90                    state = State::Word;
91                }
92                _ => {
93                    current.push(c);
94                    state = State::Word;
95                }
96            },
97            State::Single => {
98                if c == '\'' {
99                    state = State::Word;
100                } else {
101                    current.push(c);
102                }
103            }
104            State::Double => {
105                if c == '"' {
106                    state = State::Word;
107                } else {
108                    current.push(c);
109                }
110            }
111        }
112    }
113
114    match state {
115        State::Normal => {}
116        State::Word => words.push(current),
117        State::Single | State::Double => return Err(ShellwordsError::UnmatchedQuote),
118    }
119
120    Ok(words)
121}
122
123/// Options for a supervised child process.
124///
125/// Ported from `overseer.Options` as returned by Go's `processOptions`.
126/// Go sets `LineBufferSize: 1 << 20` because `ShinyTrinkets/overseer`
127/// scans child output with a `bufio.Scanner` whose *default* token limit
128/// is 64 KiB -- a line longer than that stops the stream for good, and
129/// every later write by the process fails with `EPIPE`. `RetryTimes`,
130/// `Buffered`, and `Streaming` configure overseer's restart/buffering
131/// behaviour, which belongs to `supervise_all`'s not-yet-ported restart
132/// orchestration rather than to a single child's streaming, so they have
133/// no field here.
134///
135/// [`new_cmd`]'s line reader has no equivalent limit to configure: it
136/// reads with [`std::io::BufRead::read_until`], which grows its buffer to
137/// fit whatever is on the line instead of capping it, so arbitrarily long
138/// lines stream through whole by construction.
139#[derive(Debug, Clone)]
140pub struct ProcessOptions {
141    pub working_dir: String,
142}
143
144/// Port of `processOptions`.
145#[must_use]
146pub fn process_options(working_dir: &str) -> ProcessOptions {
147    ProcessOptions {
148        working_dir: working_dir.to_string(),
149    }
150}
151
152/// The outcome of a [`SupervisedCmd`] run.
153///
154/// Ported from the subset of `overseer.Status` that `cmd.Start()`'s
155/// result channel carries in the Go test this is a port of.
156#[derive(Debug, Clone, Default)]
157pub struct CmdStatus {
158    pub error: Option<String>,
159}
160
161/// A single child process, configured but not yet started.
162///
163/// Ported from `overseer.NewCmd`'s return value -- narrowly, for the one
164/// capability `editmodeprocesssupervisor`'s tests actually exercise on
165/// it: streaming stdout line-by-line to a channel while the process runs,
166/// then reporting whether it exited cleanly. `overseer.NewCmd`'s restart
167/// bookkeeping, PID file, and `Cmd.Stop`/signal handling belong to
168/// `supervise_all`'s orchestration layer (still not ported, see that
169/// function's doc comment) rather than to a single child's lifecycle, so
170/// they are not part of this struct.
171pub struct SupervisedCmd {
172    name: String,
173    args: Vec<String>,
174    working_dir: String,
175}
176
177/// Port of `overseer.NewCmd(name, args, opts)`.
178#[must_use]
179pub fn new_cmd(name: &str, args: &[String], opts: ProcessOptions) -> SupervisedCmd {
180    SupervisedCmd {
181        name: name.to_string(),
182        args: args.to_vec(),
183        working_dir: opts.working_dir,
184    }
185}
186
187impl SupervisedCmd {
188    /// Port of `overseer.Cmd.Start()`.
189    ///
190    /// Spawns the child and returns immediately with a receiver that
191    /// yields each line of its combined stdout as it is produced, and a
192    /// second receiver that yields exactly one [`CmdStatus`] once the
193    /// child exits. Both channels are closed from a background thread
194    /// (mirroring overseer's own internal goroutine) once the child has
195    /// finished and its output has been fully drained.
196    ///
197    /// Unlike Go's `overseer.Cmd.Start`, which returns a channel even on
198    /// spawn failure (closing `Stdout` immediately and sending a status
199    /// carrying the spawn error), this returns `Err` synchronously on
200    /// spawn failure, since Rust has no equivalent need to defer that
201    /// error onto a channel the caller must still drain.
202    ///
203    /// # Errors
204    ///
205    /// Returns an error if the child process fails to spawn.
206    ///
207    /// # Panics
208    ///
209    /// Panics if spawning succeeds but the child's stdout was not piped,
210    /// which cannot happen since `start` always configures `Stdio::piped()`.
211    pub fn start(
212        self,
213    ) -> std::io::Result<(
214        std::sync::mpsc::Receiver<String>,
215        std::sync::mpsc::Receiver<CmdStatus>,
216    )> {
217        use std::io::BufRead as _;
218        use std::process::{Command, Stdio};
219        use std::sync::mpsc;
220
221        let mut command = Command::new(&self.name);
222        command.args(&self.args);
223        if !self.working_dir.is_empty() {
224            command.current_dir(&self.working_dir);
225        }
226        command.stdin(Stdio::null());
227        command.stdout(Stdio::piped());
228        command.stderr(Stdio::null());
229
230        let mut child = command.spawn()?;
231        let stdout = child.stdout.take().expect("stdout was piped");
232
233        let (line_tx, line_rx) = mpsc::channel::<String>();
234        let (status_tx, status_rx) = mpsc::channel::<CmdStatus>();
235
236        std::thread::spawn(move || {
237            let mut reader = std::io::BufReader::new(stdout);
238            let mut buf = Vec::new();
239
240            loop {
241                buf.clear();
242                match reader.read_until(b'\n', &mut buf) {
243                    Ok(0) | Err(_) => break,
244                    Ok(_) => {
245                        if buf.last() == Some(&b'\n') {
246                            buf.pop();
247                        }
248                        let line = String::from_utf8_lossy(&buf).into_owned();
249                        if line_tx.send(line).is_err() {
250                            break;
251                        }
252                    }
253                }
254            }
255            drop(line_tx);
256
257            let error = match child.wait() {
258                Ok(exit_status) if exit_status.success() => None,
259                Ok(exit_status) => Some(format!("exit status {exit_status}")),
260                Err(error) => Some(error.to_string()),
261            };
262
263            let _ = status_tx.send(CmdStatus { error });
264        });
265
266        Ok((line_rx, status_rx))
267    }
268}
269
270/// A named edit-mode process selected to run in the requested context.
271struct NamedProcess {
272    name: String,
273    config: ProcessConfig,
274}
275
276pub fn run(args: &EditModeProcessSupervisorArgs, ctx: &Context) -> anyhow::Result<()> {
277    run_supervisor(
278        ctx,
279        &args.reloader,
280        args.logs_dir.as_deref().unwrap_or(""),
281        &args.context,
282    )
283}
284
285/// Port of `editmodeprocesssupervisor.run`: loads the configuration,
286/// resolves the edit-mode processes configured for `reloader_name` that
287/// run in `proc_context`, and supervises them.
288///
289/// Deviation from Go: actual process supervision (restart-on-failure via
290/// `ShinyTrinkets/overseer`) has no Rust equivalent in this worktree yet,
291/// so this port covers every step up to (and including) the port
292/// conflict/invalid-command checks `superviseAll` performs before handing
293/// off to `overseer.NewOverseer()` -- exactly the paths every non-
294/// integration Go test in this package exercises. If processes are
295/// actually selected to run, this returns an explicit error rather than
296/// silently doing nothing or pretending to supervise.
297pub fn run_supervisor(
298    ctx: &Context,
299    reloader_name: &str,
300    logs_dir: &str,
301    proc_context: &str,
302) -> anyhow::Result<()> {
303    let _ = logs_dir; // threaded through for parity with Go's signature; log setup isn't ported (see module docs).
304
305    xianyun_executor::tempfile::ensure_temp_dir_exists().context("creating temporary directory")?;
306
307    let container =
308        xianyun_core::config::caproni::load_from_context(ctx).context("loading configuration")?;
309
310    let reloader_config = container
311        .caproni
312        .reloaders
313        .as_ref()
314        .and_then(|reloaders| reloaders.get(&reloader_name.to_string()))
315        .ok_or_else(|| SupervisorError::ReloaderNotFound(reloader_name.to_string()))?;
316
317    let edit_mode_processes = reloader_config
318        .mirror
319        .as_ref()
320        .and_then(|mirror| mirror.edit_mode_processes.as_ref())
321        .ok_or_else(|| SupervisorError::NoEditModeProcesses(reloader_name.to_string()))?;
322
323    let repo_config = container
324        .caproni
325        .repositories
326        .as_ref()
327        .and_then(|repos| repos.get(&reloader_config.repository))
328        .ok_or_else(|| SupervisorError::RepositoryNotFound(reloader_config.repository.clone()))?;
329
330    let working_dir = container.caproni.resolve_path(&repo_config.directory);
331
332    let processes: Vec<NamedProcess> = edit_mode_processes
333        .iter()
334        .filter_map(|(name, proc_config)| {
335            let run_in = if proc_config.context.is_empty() {
336                PROCESS_CONTEXT_MIRRORD
337            } else {
338                proc_config.context.as_str()
339            };
340
341            if run_in == proc_context {
342                Some(NamedProcess {
343                    name: name.clone(),
344                    config: proc_config.clone(),
345                })
346            } else {
347                None
348            }
349        })
350        .collect();
351
352    if processes.is_empty() {
353        return Ok(());
354    }
355
356    supervise_all(&processes, &working_dir)
357}
358
359/// Port of `superviseAll`'s pre-flight checks (port conflicts, command
360/// parsing). See [`run_supervisor`]'s doc comment for why the actual
361/// `overseer.SuperviseAll()` call has no port here.
362fn supervise_all(processes: &[NamedProcess], _working_dir: &str) -> anyhow::Result<()> {
363    for process in processes {
364        if process.config.port > 0
365            && xianyun_executor::portprobe::in_use(&format!("localhost:{}", process.config.port))
366        {
367            anyhow::bail!(
368                "cannot start {}: port {} already in use",
369                process.name,
370                process.config.port
371            );
372        }
373    }
374
375    for process in processes {
376        split_command(&process.config.command).map_err(|error| {
377            anyhow::anyhow!("invalid command for process {:?}: {error}", process.name)
378        })?;
379    }
380
381    anyhow::bail!(
382        "supervising edit mode processes requires an overseer-equivalent process supervisor, \
383        which isn't implemented in this worktree yet"
384    );
385}
386
387/// Ported from `overseer.ProcessJSON`'s subset `collectProcessErrors` reads.
388#[derive(Debug, Clone, Default)]
389pub struct ProcessStatus {
390    pub state: String,
391    pub exit_code: i32,
392    pub error: Option<String>,
393}
394
395/// Port of `collectProcessErrors`: checks the final state of each named
396/// process and returns a combined error for any that exited abnormally.
397/// Processes in the "interrupted" state are not treated as errors since
398/// they were stopped intentionally (e.g. via `SIGTERM` from the user).
399pub fn collect_process_errors(
400    statuses: &HashMap<String, ProcessStatus>,
401    names: &[String],
402) -> anyhow::Result<()> {
403    let mut messages = Vec::new();
404
405    for name in names {
406        let Some(status) = statuses.get(name) else {
407            continue;
408        };
409
410        match status.state.as_str() {
411            "finished" if status.exit_code != 0 => {
412                messages.push(format!(
413                    "process {name:?} exit code {}: {}",
414                    status.exit_code,
415                    SupervisorError::ProcessFailed
416                ));
417            }
418            "fatal" => {
419                if let Some(error) = &status.error {
420                    messages.push(format!(
421                        "process {name:?}: {error}: {}",
422                        SupervisorError::ProcessFailed
423                    ));
424                } else {
425                    messages.push(format!(
426                        "process {name:?}: {}",
427                        SupervisorError::ProcessFailed
428                    ));
429                }
430            }
431            _ => {}
432        }
433    }
434
435    if messages.is_empty() {
436        return Ok(());
437    }
438
439    anyhow::bail!(messages.join("\n"));
440}
xianyun_cli/commands/editmodeprocesssupervisor.rs

xianyun_cli/commands/
editmodeprocesssupervisor.rs
