farhand_core/
jobs.rs

1//! Long-running processes on the remote: a dev server, a watcher, anything
2//! that must outlive one `remote_shell` call.
3//!
4//! A command lives as long as its SSH channel. On Windows OpenSSH every
5//! process the session started is killed when the channel closes, even one
6//! detached with `Start-Process`; on POSIX a background child that keeps
7//! the output open holds the command until its timeout, which then kills
8//! the whole process group. So a job is started outside the session — in
9//! its own session with `setsid` on POSIX, by WMI (`Win32_Process.Create`)
10//! on Windows — with its output in a log file, and is recorded under
11//! `~/.farhand/jobs/` on the remote so a later FarHand session can still
12//! list and stop it.
13
14use base64::Engine;
15use serde::{Deserialize, Serialize};
16
17use crate::error::{Error, Result};
18use crate::remote::{encode_utf16_b64, ps_quote, separator, Platform, Remote};
19use crate::text::shell_quote;
20
21/// Seconds a launcher, status check or stop may take.
22const CONTROL_TIMEOUT: u64 = 30;
23/// Exit status a job reports when it cannot enter its directory.
24const CD_FAILED_EXIT: i32 = 97;
25
26/// What is recorded about a job on the remote, as `<id>.json`.
27#[derive(Debug, Clone, Serialize, Deserialize)]
28pub struct Job {
29    pub id: String,
30    pub pid: u32,
31    pub command: String,
32    pub cwd: String,
33    pub started: String,
34    /// Native path of the file the job's stdout and stderr go to.
35    pub log: String,
36}
37
38#[derive(Debug, Clone, Copy, PartialEq, Eq)]
39pub enum JobState {
40    Running,
41    /// Ended by itself with this status.
42    Exited(i32),
43    /// Not running and left no status: stopped by `remote_kill`, killed
44    /// otherwise, or the remote rebooted.
45    Gone,
46}
47
48impl Remote {
49    /// `~/.farhand/jobs` on the remote, native form.
50    async fn jobs_dir(&self) -> Result<String> {
51        let home = self.home().await?;
52        let sep = separator(self.platform().await?);
53        Ok(format!("{home}{sep}.farhand{sep}jobs"))
54    }
55
56    /// Start `command` in the background, in `cwd` (default: the workdir),
57    /// detached from the SSH session. Returns the job as recorded.
58    pub async fn spawn(&self, command: &str, cwd: Option<&str>) -> Result<Job> {
59        let platform = self.platform().await?;
60        let cwd = match cwd {
61            Some(c) => self.resolve(c).await?,
62            None => self.effective_workdir().await?,
63        };
64        let dir = self.jobs_dir().await?;
65        self.ensure_dir(&dir).await?;
66        let id = format!("{:08x}", crate::remote::random_u64() as u32);
67        let sep = separator(platform);
68        let log = format!("{dir}{sep}{id}.log");
69        let status = format!("{dir}{sep}{id}.exit");
70        let launcher = match platform {
71            Platform::Posix => posix_launcher(self.shell(), command, &cwd, &log, &status),
72            Platform::Windows => {
73                let (exe, extra) = self.powershell();
74                windows_launcher(exe, &extra, command, &cwd, &log, &status)
75            }
76        };
77        let out = self.exec(&launcher, None, Some(CONTROL_TIMEOUT)).await?;
78        let stdout = String::from_utf8_lossy(&out.stdout);
79        let pid = stdout
80            .lines()
81            .filter_map(|l| l.trim().strip_prefix("pid="))
82            .find_map(|p| p.trim().parse::<u32>().ok())
83            .ok_or_else(|| {
84                Error::Ssh(format!(
85                    "the job did not start (exit code {}): {}{}",
86                    out.exit_code,
87                    stdout.trim(),
88                    String::from_utf8_lossy(&out.stderr).trim()
89                ))
90            })?;
91        let job = Job {
92            id,
93            pid,
94            command: command.to_string(),
95            cwd,
96            started: chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true),
97            log,
98        };
99        let record = serde_json::to_vec_pretty(&job).map_err(std::io::Error::other)?;
100        self.write_file(&format!("{dir}{sep}{}.json", job.id), &record)
101            .await?;
102        Ok(job)
103    }
104
105    /// Every recorded job, newest first, with its state.
106    pub async fn jobs(&self) -> Result<Vec<(Job, JobState)>> {
107        let dir = self.jobs_dir().await?;
108        let sep = separator(self.platform().await?);
109        let mut jobs = Vec::new();
110        for name in self.list_files(&dir).await? {
111            if !name.ends_with(".json") {
112                continue;
113            }
114            let Ok((bytes, _)) = self
115                .read_file(&format!("{dir}{sep}{name}"), 64 * 1024)
116                .await
117            else {
118                continue;
119            };
120            if let Ok(job) = serde_json::from_slice::<Job>(&bytes) {
121                jobs.push(job);
122            }
123        }
124        jobs.sort_by(|a, b| b.started.cmp(&a.started));
125        let running = self.running(&jobs).await?;
126        let mut out = Vec::with_capacity(jobs.len());
127        for job in jobs {
128            let state = if running.contains(&job.pid) {
129                JobState::Running
130            } else {
131                match self
132                    .read_file(&format!("{dir}{sep}{}.exit", job.id), 64)
133                    .await
134                {
135                    Ok((b, _)) => String::from_utf8_lossy(&b)
136                        .trim()
137                        .parse()
138                        .map(JobState::Exited)
139                        .unwrap_or(JobState::Gone),
140                    Err(_) => JobState::Gone,
141                }
142            };
143            out.push((job, state));
144        }
145        Ok(out)
146    }
147
148    /// Stop the job `id` and the processes it started. Returns the job and
149    /// whether it was running.
150    pub async fn kill_job(&self, id: &str) -> Result<(Job, bool)> {
151        if id.len() != 8 || !id.bytes().all(|b| b.is_ascii_hexdigit()) {
152            return Err(Error::Invalid(format!(
153                "`{id}` is not a job id (see remote_jobs)"
154            )));
155        }
156        let (job, state) = self
157            .jobs()
158            .await?
159            .into_iter()
160            .find(|(j, _)| j.id == id)
161            .ok_or_else(|| Error::Invalid(format!("no job `{id}` (see remote_jobs)")))?;
162        if state != JobState::Running {
163            return Ok((job, false));
164        }
165        let script = match self.platform().await? {
166            // The job leads its own process group when `setsid` or `perl`
167            // was there to start it; signal the group, then the process.
168            Platform::Posix => format!(
169                "kill -TERM -- -{p} 2>/dev/null || kill -TERM {p} 2>/dev/null\n\
170                 for _ in 1 2 3 4 5; do kill -0 {p} 2>/dev/null || exit 0; sleep 1; done\n\
171                 kill -KILL -- -{p} 2>/dev/null || kill -KILL {p} 2>/dev/null\nexit 0\n",
172                p = job.pid
173            ),
174            Platform::Windows => format!("taskkill /T /F /PID {} 2>&1 | Out-Null", job.pid),
175        };
176        self.exec(&script, None, Some(CONTROL_TIMEOUT)).await?;
177        Ok((job, true))
178    }
179
180    /// Which of the jobs' processes are alive, in one round trip.
181    async fn running(&self, jobs: &[Job]) -> Result<Vec<u32>> {
182        if jobs.is_empty() {
183            return Ok(Vec::new());
184        }
185        let pids = jobs.iter().map(|j| j.pid.to_string()).collect::<Vec<_>>();
186        let script = match self.platform().await? {
187            Platform::Posix => format!(
188                "for p in {}; do kill -0 $p 2>/dev/null && echo $p; done; exit 0",
189                pids.join(" ")
190            ),
191            Platform::Windows => format!(
192                "foreach ($p in @({})) {{ if (Get-Process -Id $p -ErrorAction \
193                 SilentlyContinue) {{ $p }} }}",
194                pids.join(",")
195            ),
196        };
197        let out = self.exec(&script, None, Some(CONTROL_TIMEOUT)).await?;
198        Ok(String::from_utf8_lossy(&out.stdout)
199            .lines()
200            .filter_map(|l| l.trim().parse().ok())
201            .collect())
202    }
203}
204
205/// The POSIX launcher: runs `command` with the configured shell in a new
206/// session (`setsid`, or `perl`'s `setpgrp` where there is none, as on
207/// macOS), output to `log`, status to `status`, and prints `pid=<pid>`.
208fn posix_launcher(shell: &[String], command: &str, cwd: &str, log: &str, status: &str) -> String {
209    let b64 = |s: &str| base64::engine::general_purpose::STANDARD.encode(s.as_bytes());
210    let shell = shell
211        .iter()
212        .map(|a| shell_quote(a))
213        .collect::<Vec<_>>()
214        .join(" ");
215    let job = format!(
216        "cd {cwd} || {{ echo {CD_FAILED_EXIT} > {status}; exit {CD_FAILED_EXIT}; }}\n\
217         export NO_COLOR=1\n\
218         {shell} \"$(printf %s {cmd} | base64 -d)\"\n\
219         echo $? > {status}\n",
220        cwd = shell_quote(cwd),
221        status = shell_quote(status),
222        cmd = b64(command),
223    );
224    format!(
225        "detach() {{\n\
226         \x20 if command -v setsid >/dev/null 2>&1; then exec setsid nohup \"$@\"\n\
227         \x20 elif command -v perl >/dev/null 2>&1; then \
228         exec perl -e 'setpgrp(0, 0); exec @ARGV or die \"exec: $!\"' nohup \"$@\"\n\
229         \x20 else exec nohup \"$@\"; fi\n\
230         }}\n\
231         ( detach sh -c \"$(printf %s {job} | base64 -d)\" ) > {log} 2>&1 < /dev/null &\n\
232         echo pid=$!\n",
233        job = b64(&job),
234        log = shell_quote(log),
235    )
236}
237
238/// The Windows launcher: has WMI start `cmd.exe`, which runs the command
239/// in PowerShell with its output in `log`; WMI's process is not in the
240/// SSH session's job object, so it survives the session. Prints
241/// `pid=<pid>` of that `cmd.exe`, the root of the job's process tree.
242fn windows_launcher(
243    exe: &str,
244    extra: &[String],
245    command: &str,
246    cwd: &str,
247    log: &str,
248    status: &str,
249) -> String {
250    let job = format!(
251        "$ProgressPreference='SilentlyContinue'; $ErrorActionPreference='Continue'\n\
252         [Console]::OutputEncoding=[Text.Encoding]::UTF8; $OutputEncoding=[Text.Encoding]::UTF8\n\
253         $env:NO_COLOR='1'; $env:PYTHONUTF8='1'; $env:PYTHONIOENCODING='utf-8'\n\
254         try {{ Set-Location -LiteralPath {cwd} -ErrorAction Stop }} \
255         catch {{ Set-Content -LiteralPath {status} -Value {CD_FAILED_EXIT}; exit {CD_FAILED_EXIT} }}\n\
256         $global:LASTEXITCODE=0\n\
257         {command}\n\
258         $__fh_ok=$?\n\
259         $__fh_code=$(if ($__fh_ok) {{0}} elseif ($LASTEXITCODE) {{$LASTEXITCODE}} \
260         elseif ($Error.Count -and $Error[0].FullyQualifiedErrorId -like 'NativeCommandError*') {{0}} \
261         else {{1}})\n\
262         Set-Content -LiteralPath {status} -Value $__fh_code\n\
263         exit $__fh_code\n",
264        cwd = ps_quote(cwd),
265        status = ps_quote(status),
266    );
267    let extra = extra
268        .iter()
269        .map(|a| format!("\"{a}\""))
270        .collect::<Vec<_>>()
271        .join(" ");
272    // `/s` makes cmd strip only the outermost quotes and keep the rest.
273    let line = format!(
274        "cmd.exe /d /s /c \"\"{exe}\" {extra} -NoProfile -NonInteractive -ExecutionPolicy Bypass \
275         -EncodedCommand {enc} > \"{log}\" 2>&1\"",
276        enc = encode_utf16_b64(&job),
277    );
278    format!(
279        "$r = Invoke-CimMethod -ClassName Win32_Process -MethodName Create -Arguments \
280         @{{ CommandLine = {line}; CurrentDirectory = {cwd} }}\n\
281         if ($r.ReturnValue -ne 0) {{ [Console]::Error.WriteLine(\"Win32_Process.Create \
282         returned $($r.ReturnValue)\"); exit 1 }}\n\
283         \"pid=$($r.ProcessId)\"\n",
284        line = ps_quote(&line),
285        cwd = ps_quote(cwd),
286    )
287}
288
289#[cfg(test)]
290mod tests {
291    use super::*;
292
293    /// The launcher run locally, with a real `sh`: the job outlives the
294    /// launcher, writes its log and status, and leads its own group.
295    #[cfg(unix)]
296    #[test]
297    fn a_posix_job_outlives_its_launcher() {
298        let d = tempfile::tempdir().unwrap();
299        let p = |n: &str| d.path().join(n).to_string_lossy().into_owned();
300        let script = posix_launcher(
301            &["sh".into(), "-c".into()],
302            "sleep 1; echo \"out $NO_COLOR\"; exit 3",
303            &d.path().to_string_lossy(),
304            &p("job.log"),
305            &p("job.exit"),
306        );
307        let started = std::time::Instant::now();
308        let out = std::process::Command::new("sh")
309            .arg("-c")
310            .arg(&script)
311            .stdin(std::process::Stdio::null())
312            .output()
313            .unwrap();
314        // The launcher returns at once: the job does not hold its output.
315        assert!(started.elapsed() < std::time::Duration::from_secs(1));
316        let text = String::from_utf8_lossy(&out.stdout);
317        let pid: i32 = text.trim().strip_prefix("pid=").unwrap().parse().unwrap();
318        // A process group of its own (once it has got that far), so
319        // remote_kill can stop its children.
320        let mut pgid = 0;
321        for _ in 0..50 {
322            pgid = unsafe { libc::getpgid(pid) };
323            if pgid == pid {
324                break;
325            }
326            std::thread::sleep(std::time::Duration::from_millis(20));
327        }
328        assert_eq!(pgid, pid, "{text}");
329        for _ in 0..50 {
330            if std::path::Path::new(&p("job.exit")).exists() {
331                break;
332            }
333            std::thread::sleep(std::time::Duration::from_millis(100));
334        }
335        assert_eq!(std::fs::read_to_string(p("job.exit")).unwrap().trim(), "3");
336        assert_eq!(std::fs::read_to_string(p("job.log")).unwrap(), "out 1\n");
337    }
338
339    #[test]
340    fn the_windows_launcher_goes_through_wmi() {
341        let s = windows_launcher(
342            "powershell.exe",
343            &[],
344            "npm run dev",
345            "C:\\Users\\me\\proj",
346            "C:\\Users\\me\\.farhand\\jobs\\0badc0de.log",
347            "C:\\Users\\me\\.farhand\\jobs\\0badc0de.exit",
348        );
349        assert!(s.contains("Win32_Process -MethodName Create"), "{s}");
350        assert!(
351            s.contains("CurrentDirectory = 'C:\\Users\\me\\proj'"),
352            "{s}"
353        );
354        assert!(
355            s.contains("> \"C:\\Users\\me\\.farhand\\jobs\\0badc0de.log\" 2>&1"),
356            "{s}"
357        );
358        // The command itself travels encoded, never on the command line.
359        assert!(!s.contains("npm run dev"), "{s}");
360    }
361}
farhand_core/jobs.rs

farhand_core/
jobs.rs
