coop/
run.rs

1use std::io::Write;
2
3use anyhow::{Context, Result, bail};
4
5use crate::config::Host;
6use crate::transport::Transport;
7use crate::wrapper::{Job, JobId, dispatch_script, new_id};
8
9pub fn dispatch(
10    transport: &dyn Transport,
11    host: &Host,
12    cmd: &str,
13    cwd: Option<&str>,
14    max_secs: Option<u64>,
15) -> Result<JobId> {
16    dispatch_with_warnings(transport, host, cmd, cwd, max_secs, &mut std::io::stderr())
17}
18
19#[doc(hidden)]
20pub fn dispatch_with_warnings(
21    transport: &dyn Transport,
22    host: &Host,
23    cmd: &str,
24    cwd: Option<&str>,
25    max_secs: Option<u64>,
26    warnings: &mut dyn Write,
27) -> Result<JobId> {
28    // `ssh -O check` measured at 0s and opens no session channel, so it is the
29    // one transport call deliberately outside the lock.
30    if !transport.master_alive(host) {
31        return Err(crate::errors::no_master(host));
32    }
33
34    let job = Job {
35        // Generated, so it parses by construction; the parse is what keeps a
36        // CLI-supplied id from reaching the remote shell unvalidated.
37        id: new_id().parse::<JobId>().expect("generated ids are valid"),
38        cmd: cmd.to_owned(),
39        cwd: cwd.map(str::to_owned),
40        max_secs: max_secs.unwrap_or(host.max_job_secs),
41    };
42    let script = format!(
43        "{}; {} && {{ tmux -L {} list-sessions -F '#{{session_name}}' 2>/dev/null | grep -c '^coop-' || true; }}",
44        crate::jobs::prune(host),
45        dispatch_script(host, &job),
46        host.tmux_socket
47    );
48
49    // The retrospective count shares the measured 0s dispatch round trip. A
50    // pre-flight warning would double both ssh round trips and lock cycles for
51    // advisory backpressure.
52    let output = transport.run(host, &script)?;
53    if output.code != 0 {
54        bail!(
55            "dispatch failed for {}: {}",
56            host.name,
57            output.stderr.trim()
58        );
59    }
60    let running: u32 = output
61        .text()
62        .trim()
63        .parse()
64        .context("invalid running-session count in dispatch reply")?;
65    if running > host.max_running {
66        writeln!(
67            warnings,
68            "coop: dispatched {}; {running} now running on {}, cap {}",
69            job.id, host.name, host.max_running
70        )?;
71    }
72
73    Ok(job.id)
74}
coop/run.rs

coop/
run.rs
