//! Where a run's steps run: in a shell on the runner's host, or in one
//! container the runner's docker daemon starts for the run with the
//! checkout mounted, so each step finds what the steps before it left,
//! installed tools included, as on one machine.

use anyhow::{Context, Result, bail};
use rustix::process::{Pid, Signal};
use std::collections::{BTreeSet, HashSet};
use std::path::{Path, PathBuf};
use std::process::Stdio;
use std::sync::Mutex;
use std::time::{Duration, Instant};
use tokio::process::Command;

/// The runner's own variables a step sees, so it finds its tools and the
/// docker client its daemon. Nothing else of the runner's environment, its
/// credentials least of all, reaches a step.
const PASSED: [&str; 7] = [
    "PATH",
    "HOME",
    "LANG",
    "TMPDIR",
    "USER",
    "DOCKER_HOST",
    "DOCKER_CONFIG",
];

#[derive(Clone, Debug, PartialEq)]
pub enum Executor {
    /// `sh -c` on the host, as the runner's user. For trusted repositories.
    Shell,
    /// `docker run --rm`, in the workflow's image or this one, within
    /// `limits`.
    Docker { image: String, limits: Limits },
}

/// What one run's container may take of the host, under docker: each
/// unset, as much as it likes, as before there were limits. A runner
/// running code it does not trust sets them all, so that no run takes
/// the machine from the rest, or from what else runs there.
#[derive(Clone, Debug, Default, PartialEq)]
pub struct Limits {
    /// CPUs, as docker's `--cpus` takes them: `2`, `1.5`.
    pub cpus: Option<String>,
    /// Bytes of memory, swap none beyond it: a run past it is killed by
    /// the kernel, its step failing.
    pub memory: Option<u64>,
    /// Processes and threads at once, so a fork bomb ends at it.
    pub pids: Option<u64>,
    /// The docker network runs join, made if missing as a bridge whose
    /// containers cannot reach each other: so a run reaches neither
    /// another run nor what the host's other containers serve on theirs.
    pub network: Option<String>,
    /// The container runtime docker runs it with, as its `--runtime`
    /// takes one: `runsc`, gVisor's, puts a kernel of its own between a
    /// run and the host's.
    pub runtime: Option<String>,
    /// The runtime, in place of `runtime`, of a run [`Setup::vouched`]
    /// for: `runc`, docker's own, spares it gVisor's cost, as the code
    /// runs on the merge queue's runner with none once it is queued.
    pub trusted_runtime: Option<String>,
    /// A file on the docker host that runs see as their
    /// `/etc/resolv.conf`. Docker's own resolver on a network of its
    /// making is reached through NAT rules in the container's network
    /// namespace, which gVisor's network stack does not apply: under
    /// `runsc` a run resolves nothing without one naming resolvers of
    /// its own.
    pub resolv: Option<String>,
}

/// The option a network made for [`Limits::network`] is made with: no
/// traffic between its containers.
const ISOLATED: &str = "com.docker.network.bridge.enable_icc=false";

/// What a run's steps run in, from its first step to its last.
#[derive(Debug, PartialEq)]
pub enum Session {
    /// A shell in the checkout, with the run's variables.
    Shell {
        dir: PathBuf,
        vars: Vec<(String, String)>,
    },
    /// A running container, by ID, through this docker client, and the
    /// overlay volumes it mounts.
    Docker {
        docker: PathBuf,
        id: String,
        volumes: Vec<String>,
    },
}

/// What a step came to.
#[derive(Debug, PartialEq)]
pub struct Exit {
    /// None when it was killed: out of time, or cancelled.
    pub code: Option<i32>,
    /// Whether it was killed for being cancelled.
    pub cancelled: bool,
    pub elapsed: Duration,
}

impl Exit {
    pub fn success(&self) -> bool {
        self.code == Some(0)
    }
}

/// What a run's steps start in.
pub struct Setup<'a> {
    /// The checkout, which steps run in.
    pub dir: &'a Path,
    pub vars: &'a [(String, String)],
    /// The image, where not the executor's own.
    pub image: Option<&'a str>,
    /// Overlays of what is kept from run to run, under docker only.
    pub caches: &'a [Cache],
    /// Where steps leave what the run made to be looked at, named to them
    /// as `KNOTLESS_RUN_OUTPUTS`.
    pub outputs: &'a Path,
    /// The checkouts of the shared steps' repositories, if any: mounted
    /// read-only at [`STEPS`] under docker.
    pub steps: Option<&'a Path>,
    /// Whose the run is, as its container is labeled under docker.
    pub owner: Owner<'a>,
    /// Whether it gives way to everything else on the host for CPU, as a
    /// pull's run does: under docker, a weight of about 10 against a
    /// container's 100, so a burst of pulls leaves the site, deploys and
    /// main's runs what they need.
    pub behind: bool,
    /// Whether its code is vouched for by an owner the runner trusts, to
    /// run under [`Limits::trusted_runtime`].
    pub vouched: bool,
}

/// The `--cpu-shares` a run behind the rest is given: docker makes it a
/// cgroup v2 `cpu.weight` of 10.
const BEHIND: &str = "256";

/// The runner a run is of, by its account's DID, and the run, by its
/// check's key.
#[derive(Clone, Copy)]
pub struct Owner<'a> {
    pub runner: &'a str,
    pub run: &'a str,
}

/// The variable naming [`Setup::outputs`] to the steps.
pub const OUTPUTS: &str = "KNOTLESS_RUN_OUTPUTS";

/// Where the shared steps' repositories are, in the container: each at
/// `<STEPS>/<n>`, named to its steps as `KNOTLESS_STEP_DIR`.
pub const STEPS: &str = "/steps";

/// An overlay of a cache, mounted at `path` in a run's container: a
/// volume of docker's `local` driver over a base, `lower`, read only, with
/// the run's own `upper` layer taking what it writes.
#[derive(Debug, Clone, PartialEq)]
pub struct Cache {
    pub volume: String,
    pub path: String,
    /// The run it is of, which the volume is labelled with.
    pub run: String,
    pub lower: PathBuf,
    pub upper: PathBuf,
    pub work: PathBuf,
}

/// The label a run's overlay volumes carry, naming the run.
pub const OVERLAY_LABEL: &str = "dev.knotless.overlay";

/// The labels a run's container carries: the runner's DID, and the run's
/// check's key.
pub const RUNNER_LABEL: &str = "dev.knotless.runner";
pub const RUN_LABEL: &str = "dev.knotless.run";

impl Executor {
    pub fn parse(name: &str, image: &str) -> Option<Executor> {
        match name {
            "shell" => Some(Executor::Shell),
            "docker" => Some(Executor::Docker {
                image: image.to_string(),
                limits: Limits::default(),
            }),
            _ => None,
        }
    }

    /// Where the steps find the `i`th shared step's repository, checked
    /// out at `at` on the host: under docker, where it is mounted.
    pub fn step_dir(&self, i: usize, at: &Path) -> String {
        match self {
            Executor::Shell => at.to_string_lossy().into_owned(),
            Executor::Docker { .. } => format!("{STEPS}/{i}"),
        }
    }

    /// Starts what a run's steps run in, as `setup` says: under docker, a
    /// container whose start, an image pulled among it, is written to
    /// `log` and may take `timeout`.
    pub async fn start(
        &self,
        setup: &Setup<'_>,
        log: &std::fs::File,
        timeout: Duration,
    ) -> Result<Session> {
        self.start_with(Path::new("docker"), setup, log, timeout)
            .await
    }

    /// [`Executor::start`], with `docker` as the docker client.
    pub async fn start_with(
        &self,
        docker: &Path,
        setup: &Setup<'_>,
        log: &std::fs::File,
        timeout: Duration,
    ) -> Result<Session> {
        let Setup {
            dir,
            vars,
            image,
            caches,
            outputs,
            steps: _,
            owner: _,
            behind: _,
            vouched: _,
        } = *setup;
        let mut vars = vars.to_vec();
        let Executor::Docker {
            image: default,
            limits,
        } = self
        else {
            vars.push((OUTPUTS.into(), outputs.display().to_string()));
            return Ok(Session::Shell {
                dir: dir.to_path_buf(),
                vars,
            });
        };
        vars.push((OUTPUTS.into(), "/outputs".into()));
        let image = image.unwrap_or(default);
        let mut volumes = Vec::new();
        let mut made = Ok(());
        for cache in caches {
            made = overlay(docker, cache, log).await;
            if made.is_err() {
                break;
            }
            volumes.push(cache.volume.clone());
        }
        let started = match made {
            Ok(()) => {
                let start = Start {
                    image,
                    limits,
                    vars: &vars,
                    timeout,
                };
                container(docker, setup, start, log).await
            }
            Err(err) => Err(err),
        };
        // A container that did not start leaves no overlay mounted, for
        // its layers to be removed.
        if started.is_err() {
            remove_volumes(docker, &volumes).await;
        }
        Ok(Session::Docker {
            docker: docker.to_path_buf(),
            id: started?,
            volumes,
        })
    }
}

/// Makes the overlay volume `cache`, writing what goes wrong to `log`.
async fn overlay(docker: &Path, cache: &Cache, log: &std::fs::File) -> Result<()> {
    let layers = format!(
        "o=lowerdir={},upperdir={},workdir={},userxattr",
        cache.lower.display(),
        cache.upper.display(),
        cache.work.display()
    );
    let mut cmd = client(docker, &[]);
    cmd.args(["volume", "create", "--driver", "local", "--label"])
        .arg(format!("{OVERLAY_LABEL}={}", cache.run))
        .args(["--opt", "type=overlay", "--opt", "device=overlay", "--opt"])
        .arg(layers)
        .arg(&cache.volume)
        .stdout(Stdio::null())
        .stderr(log.try_clone()?);
    let made = cmd.status().await.context("making a cache's overlay")?;
    if !made.success() {
        bail!("the overlay at {} was not made: see the log", cache.path);
    }
    Ok(())
}

/// What a run's container starts from: its image, within its limits, with
/// its variables, in at most its timeout.
struct Start<'a> {
    image: &'a str,
    limits: &'a Limits,
    vars: &'a [(String, String)],
    timeout: Duration,
}

/// Starts the container a run's steps run in, as `start` says, mounting
/// the checkout, the outputs and the overlays `setup` names: its ID.
async fn container(
    docker: &Path,
    setup: &Setup<'_>,
    start: Start<'_>,
    log: &std::fs::File,
) -> Result<String> {
    let Start {
        image,
        limits,
        vars,
        timeout,
    } = start;
    if let Some(network) = &limits.network {
        isolated(docker, network, log).await?;
    }
    let mut cmd = client(docker, vars);
    cmd.args(["run", "-d", "--init", "--label"])
        .arg(format!("{RUNNER_LABEL}={}", setup.owner.runner))
        .arg("--label")
        .arg(format!("{RUN_LABEL}={}", setup.owner.run))
        .arg("-v")
        .arg(format!("{}:/work", setup.dir.display()))
        .arg("-v")
        .arg(format!("{}:/outputs", setup.outputs.display()));
    if let Some(steps) = setup.steps {
        cmd.arg("-v").arg(format!("{}:{STEPS}:ro", steps.display()));
    }
    cmd.args(["-w", "/work"]);
    if setup.behind {
        cmd.args(["--cpu-shares", BEHIND]);
    }
    cmd.args(limited(limits, setup.vouched));
    for cache in setup.caches {
        let mount = format!("type=volume,src={},dst={}", cache.volume, cache.path);
        cmd.arg("--mount").arg(mount);
    }
    // Named only: the values reach the container from the client's
    // environment, never its arguments.
    for (k, _) in vars {
        cmd.arg("-e").arg(k);
    }
    // Kept running until the run ends, for each step to be run in.
    cmd.args([image, "tail", "-f", "/dev/null"])
        .stdin(Stdio::null())
        .stderr(log.try_clone()?)
        .kill_on_drop(true);
    let started = tokio::time::timeout(timeout, cmd.output()).await;
    let out = match started {
        Ok(out) => out.context("starting a container")?,
        Err(_) => bail!("{image} did not start in {}s", timeout.as_secs()),
    };
    if !out.status.success() {
        bail!("{image} did not start: see the log");
    }
    let id = String::from_utf8_lossy(&out.stdout).trim().to_string();
    crate::telemetry::container("container started", &id, Some(image));
    Ok(id)
}

/// The arguments to `docker run` that hold a container to `limits`, a
/// `vouched` one's under its trusted runtime, if any.
fn limited(limits: &Limits, vouched: bool) -> Vec<String> {
    let mut args = Vec::new();
    if let Some(cpus) = &limits.cpus {
        args.extend(["--cpus".into(), cpus.clone()]);
    }
    if let Some(bytes) = limits.memory {
        // Swap as large as the memory is none beyond it.
        let bytes = bytes.to_string();
        args.extend([
            "--memory".into(),
            bytes.clone(),
            "--memory-swap".into(),
            bytes,
        ]);
    }
    if let Some(pids) = limits.pids {
        args.extend(["--pids-limit".into(), pids.to_string()]);
    }
    if let Some(network) = &limits.network {
        args.extend(["--network".into(), network.clone()]);
    }
    let trusted = limits.trusted_runtime.as_ref().filter(|_| vouched);
    if let Some(runtime) = trusted.or(limits.runtime.as_ref()) {
        args.extend(["--runtime".into(), runtime.clone()]);
    }
    if let Some(resolv) = &limits.resolv {
        args.extend(["-v".into(), format!("{resolv}:/etc/resolv.conf:ro")]);
    }
    args
}

/// Makes the network `name` if docker has none of that name, as one whose
/// containers cannot reach each other, writing what goes wrong to `log`.
/// One there already is used as it is, and so is one another run's start
/// made meanwhile.
async fn isolated(docker: &Path, name: &str, log: &std::fs::File) -> Result<()> {
    if network_there(docker, name).await? {
        return Ok(());
    }
    let mut create = client(docker, &[]);
    create
        .args([
            "network", "create", "--driver", "bridge", "--opt", ISOLATED, name,
        ])
        .stdout(Stdio::null())
        .stderr(log.try_clone()?);
    let made = create.status().await.context("making a network")?;
    if !made.success() && !network_there(docker, name).await? {
        bail!("the network {name} was not made: see the log");
    }
    Ok(())
}

/// Whether docker has a network named `name`.
async fn network_there(docker: &Path, name: &str) -> Result<bool> {
    let mut inspect = client(docker, &[]);
    inspect
        .args(["network", "inspect", name])
        .stdout(Stdio::null())
        .stderr(Stdio::null());
    let asked = inspect.status().await.context("asking for a network")?;
    Ok(asked.success())
}

impl Executor {
    /// Removes what runs of an earlier runner left in docker, under
    /// docker: their overlay volumes, and the volumes caches were kept in
    /// before they were overlays. What it returns is the runs whose
    /// overlays are still mounted, and so stay.
    pub async fn clean_caches(&self) -> Vec<String> {
        self.clean_caches_with(Path::new("docker")).await
    }

    /// [`Executor::clean_caches`], with `docker` as the docker client.
    pub async fn clean_caches_with(&self, docker: &Path) -> Vec<String> {
        if !matches!(self, Executor::Docker { .. }) {
            return Vec::new();
        }
        let overlays = format!("label={OVERLAY_LABEL}");
        for filter in [overlays.as_str(), "name=knotless-cache-"] {
            let names = listed(docker, &["-q", "--filter", filter]).await;
            remove_volumes(docker, &names).await;
        }
        let runs = format!("{{{{.Label \"{OVERLAY_LABEL}\"}}}}");
        let mut left = listed(docker, &["--filter", &overlays, "--format", &runs]).await;
        left.sort();
        left.dedup();
        left
    }

    /// Removes the containers the runner `runner` left, under docker:
    /// those labeled as its own, and with `work`, those that mount a path
    /// under it, as a run's checkout and outputs are, unless of a run in
    /// `going`, by its check's key, read once the containers are listed, so
    /// one started meanwhile is never taken for left. Called as it starts, with none going,
    /// each is left by a runner that was killed or crashed, one from before
    /// runs were labeled among them; and from time to time after. Any other
    /// container is never touched.
    pub async fn remove_left(
        &self,
        runner: &str,
        going: &Mutex<HashSet<String>>,
        work: Option<&Path>,
    ) {
        let docker = Path::new("docker");
        self.remove_left_with(docker, runner, going, work).await
    }

    /// [`Executor::remove_left`], with `docker` as the docker client.
    pub async fn remove_left_with(
        &self,
        docker: &Path,
        runner: &str,
        going: &Mutex<HashSet<String>>,
        work: Option<&Path>,
    ) {
        if !matches!(self, Executor::Docker { .. }) {
            return;
        }
        let all = words(client(docker, &[]).args(["ps", "-aq"])).await;
        if all.is_empty() {
            return;
        }
        // Each as `<id>|<runner>|<run>|<source>;<source>;...`.
        let format = format!(
            "{{{{.Id}}}}|{{{{index .Config.Labels \"{RUNNER_LABEL}\"}}}}|{{{{index .Config.Labels \"{RUN_LABEL}\"}}}}|{{{{range .Mounts}}}}{{{{.Source}}}};{{{{end}}}}"
        );
        let mut inspect = client(docker, &[]);
        inspect.args(["inspect", "--format", &format]).args(&all);
        let described = words(&mut inspect).await;
        let left: Vec<&str> = described
            .iter()
            .filter_map(|line| {
                let mut fields = line.splitn(4, '|');
                let (id, mine, run) = (fields.next()?, fields.next()?, fields.next()?);
                let mounts = fields.next().unwrap_or_default().split(';');
                let under = |m: &str| work.is_some_and(|w| Path::new(m).starts_with(w));
                let ours = mine == runner || mounts.into_iter().any(under);
                (ours && !going.lock().unwrap().contains(run)).then_some(id)
            })
            .collect();
        if left.is_empty() {
            return;
        }
        let _ = client(docker, &[])
            .args(["rm", "-f"])
            .args(left)
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .status()
            .await;
    }
}

/// The lines `cmd` prints, or none if it cannot be run.
async fn words(cmd: &mut Command) -> Vec<String> {
    let listed = cmd.stderr(Stdio::null()).output().await;
    let text = listed.map(|o| String::from_utf8_lossy(&o.stdout).into_owned());
    let text = text.unwrap_or_default();
    text.lines()
        .filter(|l| !l.is_empty())
        .map(str::to_string)
        .collect()
}

/// The words `docker volume ls` prints, given `args`.
async fn listed(docker: &Path, args: &[&str]) -> Vec<String> {
    let listed = client(docker, &[])
        .args(["volume", "ls"])
        .args(args)
        .stderr(Stdio::null())
        .output()
        .await;
    let text = listed.map(|o| String::from_utf8_lossy(&o.stdout).into_owned());
    let text = text.unwrap_or_default();
    text.split_whitespace().map(str::to_string).collect()
}

/// Removes the volumes `names`, as far as nothing still uses them.
async fn remove_volumes(docker: &Path, names: &[String]) {
    if names.is_empty() {
        return;
    }
    let _ = client(docker, &[])
        .args(["volume", "rm", "-f"])
        .args(names)
        .stdout(Stdio::null())
        .stderr(Stdio::null())
        .status()
        .await;
}

/// Lists the processes in a container by ID, but for itself, from `/proc`,
/// which every image has where `ps` may not be.
const RUNNING: &str =
    r#"for p in /proc/[0-9]*; do p=${p#/proc/}; [ "$p" = "$$" ] || echo "$p"; done"#;

/// The docker client, seeing `vars` and the variables in [`PASSED`].
fn client(docker: &Path, vars: &[(String, String)]) -> Command {
    let mut cmd = Command::new(docker);
    passed(&mut cmd, vars);
    cmd
}

/// Clears `cmd`'s environment to the variables in [`PASSED`] and `vars`.
fn passed(cmd: &mut Command, vars: &[(String, String)]) {
    cmd.env_clear();
    for name in PASSED {
        if let Some(v) = std::env::var_os(name) {
            cmd.env(name, v);
        }
    }
    cmd.envs(vars.iter().map(|(k, v)| (k, v)));
}

impl Session {
    /// The process that runs `script` as a step: `sh -e`, so a command
    /// that fails ends the step, failed, rather than only its last one
    /// counting. It sees the run's variables and `own`, this step's alone,
    /// such as the secrets it is given, which the container never holds.
    pub fn command(&self, script: &str, own: &[(String, String)]) -> Command {
        match self {
            Session::Shell { dir, vars } => {
                let mut cmd = Command::new("sh");
                cmd.arg("-ec").arg(script).current_dir(dir);
                passed(&mut cmd, vars);
                cmd.envs(own.iter().map(|(k, v)| (k, v)));
                cmd
            }
            Session::Docker { docker, id, .. } => {
                // The container has the run's variables already. Named
                // only: the values reach the step from the client's
                // environment, never its arguments.
                let mut cmd = client(docker, own);
                cmd.arg("exec");
                for (k, _) in own {
                    cmd.arg("-e").arg(k);
                }
                cmd.args([id, "sh", "-ec", script]);
                cmd
            }
        }
    }

    /// The processes running in the container, by ID, for
    /// [`Session::end_left`] to tell which a step left: none under the
    /// shell executor, whose steps share the runner's host.
    pub async fn running(&self) -> Option<BTreeSet<u32>> {
        let Session::Docker { docker, id, .. } = self else {
            return None;
        };
        let mut list = client(docker, &[]);
        list.args(["exec", id, "sh", "-c", RUNNING]);
        let ids = words(&mut list).await;
        Some(ids.iter().filter_map(|p| p.parse().ok()).collect())
    }

    /// Ends what is running in the container that was not in `before`, as
    /// [`Session::running`] had it before `step`, and says so in `log`: the
    /// processes a step given secrets left, which would hold them for the
    /// steps after it to read. Nothing, with no `before`.
    pub async fn end_left(
        &self,
        before: Option<&BTreeSet<u32>>,
        step: &str,
        mut log: &std::fs::File,
    ) -> Result<()> {
        let (Some(before), Session::Docker { docker, id, .. }) = (before, self) else {
            return Ok(());
        };
        let now = self.running().await.unwrap_or_default();
        let left: Vec<String> = now.difference(before).map(u32::to_string).collect();
        if left.is_empty() {
            return Ok(());
        }
        let mut kill = client(docker, &[]);
        kill.args(["exec", id, "kill", "-9"]).args(&left);
        let _ = kill
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .status()
            .await;
        use std::io::Write;
        let n = left.len();
        let said = format!("==> ended {n} of what {step} left running, as it was given secrets\n");
        log.write_all(said.as_bytes())?;
        Ok(())
    }

    /// Ends the run's container, and with it anything a step left
    /// running, one killed for its time among them; then removes its
    /// overlay volumes, which unmounts them, for its layers to be kept or
    /// removed.
    pub async fn stop(self) {
        if let Session::Docker {
            docker,
            id,
            volumes,
        } = self
        {
            let _ = client(&docker, &[])
                .args(["rm", "-f", &id])
                .stdout(Stdio::null())
                .stderr(Stdio::null())
                .status()
                .await;
            crate::telemetry::container("container stopped", &id, None);
            remove_volumes(&docker, &volumes).await;
        }
    }
}

/// Runs `cmd` to its end, `timeout` or `cancelled`, writing what it prints
/// to `log`. One that runs out of time or is cancelled is killed with
/// everything it started.
pub async fn run(
    mut cmd: Command,
    log: &std::fs::File,
    timeout: Duration,
    cancelled: impl Future<Output = ()>,
) -> Result<Exit> {
    let started = Instant::now();
    cmd.stdin(Stdio::null())
        .stdout(log.try_clone()?)
        .stderr(log.try_clone()?)
        .kill_on_drop(true)
        .process_group(0);
    let mut child = cmd.spawn().context("starting a step")?;
    // Its exit, or, killed, whether for being cancelled.
    let ended = tokio::select! {
        status = tokio::time::timeout(timeout, child.wait()) => status.map_err(|_| false),
        _ = cancelled => Err(true),
    };
    let (code, cancelled) = match ended {
        Ok(status) => (Some(status?.code().unwrap_or(-1)), false),
        Err(cancelled) => {
            // The group the step leads: a shell's children with it. Killed
            // by the call itself, not a `kill` command, which the runner's
            // image does not have: without one, a step cancelled or out of
            // time ran on to its end, and one that hung held its job, and
            // a stop, for good.
            let group = child.id().and_then(|pid| i32::try_from(pid).ok());
            if let Some(group) = group.and_then(Pid::from_raw) {
                let _ = rustix::process::kill_process_group(group, Signal::KILL);
            }
            let _ = child.wait().await;
            (None, cancelled)
        }
    };
    Ok(Exit {
        code,
        cancelled,
        elapsed: started.elapsed(),
    })
}

#[cfg(test)]
mod tests {
    use super::*;

    fn args(cmd: &Command) -> Vec<String> {
        let std = cmd.as_std();
        std::iter::once(std.get_program())
            .chain(std.get_args())
            .map(|a| a.to_string_lossy().into_owned())
            .collect()
    }

    #[test]
    fn executors() {
        assert_eq!(Executor::parse("shell", "x"), Some(Executor::Shell));
        assert_eq!(Executor::parse("vm", "x"), None);
        let at = Path::new("/tmp/run/steps/2");
        assert_eq!(Executor::Shell.step_dir(2, at), "/tmp/run/steps/2");
        let docker = Executor::parse("docker", "x").unwrap();
        assert_eq!(docker.step_dir(2, at), format!("{STEPS}/2"));
        let vars = [("SECRET".to_string(), "v".to_string())];
        let shell = Session::Shell {
            dir: "/w".into(),
            vars: vars.to_vec(),
        };
        let own = [("KEY".to_string(), "k".to_string())];
        let cmd = shell.command("make", &own);
        assert_eq!(args(&cmd), ["sh", "-ec", "make"]);
        let env: Vec<_> = cmd.as_std().get_envs().collect();
        assert!(env.contains(&("SECRET".as_ref(), Some("v".as_ref()))));
        assert!(env.contains(&("KEY".as_ref(), Some("k".as_ref()))));
        let docker = Session::Docker {
            docker: "docker".into(),
            id: "c1".into(),
            volumes: Vec::new(),
        };
        assert_eq!(
            args(&docker.command("make", &[])).join(" "),
            "docker exec c1 sh -ec make"
        );
        let cmd = docker.command("make", &own);
        assert_eq!(args(&cmd).join(" "), "docker exec -e KEY c1 sh -ec make");
        let env: Vec<_> = cmd.as_std().get_envs().collect();
        assert!(env.contains(&("KEY".as_ref(), Some("k".as_ref()))));
        // A secret passed to a shared step under the name its step file
        // asks for reaches the client by that name alone.
        let step = knotless::workflow::Step {
            secrets: Some([("SSH_KEY".to_string(), "DEPLOY_KEY".to_string())].into()),
            ..Default::default()
        };
        let granted = [("DEPLOY_KEY".to_string(), "k".to_string())];
        let cmd = docker.command("make", &step.seen(&granted));
        assert_eq!(
            args(&cmd).join(" "),
            "docker exec -e SSH_KEY c1 sh -ec make"
        );
        let env: Vec<_> = cmd.as_std().get_envs().collect();
        assert!(env.contains(&("SSH_KEY".as_ref(), Some("k".as_ref()))));
        assert!(!env.iter().any(|(k, _)| *k == "DEPLOY_KEY"));
    }

    /// A stand-in docker client, in `dir`, that writes what it is asked
    /// to `dir/calls`: it starts container `c1` unless the image is
    /// `bad` or `slow`, makes every volume but `knotless-overlay-bad`,
    /// lists overlay volumes `o1` and `o2`, of runs `r1` and `r0`, and an
    /// old cache volume `old`, has network `have`, makes any other but
    /// `gone` and `raced` (made meanwhile by another, as its making
    /// fails), and runs what else it is asked to exec here.
    fn fake_docker(dir: &Path) -> PathBuf {
        let path = dir.join("docker");
        let calls = dir.join("calls");
        let script = format!(
            "#!/bin/sh\necho \"$*\" >> {calls}\ncase \"$*\" in\n  'network inspect have') exit 0;;\n  'network inspect raced') test -e {raced}; exit $?;;\n  'network inspect '*) exit 1;;\n  'network create '*' raced') touch {raced}; exit 1;;\n  'network create '*' gone') echo 'no network' >&2; exit 1;;\n  'network create '*) exit 0;;\n  *' bad tail '*) echo 'no such image' >&2; exit 1;;\n  *' slow tail '*) sleep 5;;\n  run\\ *) echo c1;;\n  *' knotless-overlay-bad') echo 'no overlay' >&2; exit 1;;\n  *' for p in /proc/'*) echo 1; echo 7; echo 9; echo x;;\n  'exec c1 kill -9 '*) ;;\n  exec\\ *) shift 2; exec \"$@\";;\n  'volume ls -q --filter label='*) echo o1; echo o2;;\n  'volume ls -q --filter name='*) echo old;;\n  'volume ls --filter '*) echo r1; echo r1; echo r0;;\n  'ps -aq') echo c7; echo c8; echo c9; echo c10;;\n  inspect\\ *) echo 'c7|did:left|r7|/x;'; echo 'c8|did:left|r8|/w/work/a/src;'; echo 'c9|||/w/work/b/src;/w/work/b/outputs;'; echo 'c10|did:other||/w/workshop;';;\nesac\n",
            calls = calls.display(),
            raced = dir.join("raced").display()
        );
        // Copied into place by another process: a file this one wrote can
        // be held open for writing by a test's fork meanwhile, and running
        // it then fails, "Text file busy".
        let written = dir.join("docker.sh");
        std::fs::write(&written, script).unwrap();
        let copied = std::process::Command::new("cp")
            .arg(&written)
            .arg(&path)
            .status()
            .unwrap();
        assert!(copied.success());
        use std::os::unix::fs::PermissionsExt;
        std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o755)).unwrap();
        path
    }

    #[tokio::test]
    async fn what_a_step_leaves_running_is_ended() {
        let dir = tempfile::tempdir().unwrap();
        let docker = fake_docker(dir.path());
        let log_path = dir.path().join("log");
        let log = std::fs::File::create(&log_path).unwrap();
        let session = Session::Docker {
            docker,
            id: "c1".into(),
            volumes: Vec::new(),
        };
        let running = session.running().await.unwrap();
        assert_eq!(running, BTreeSet::from([1, 7, 9]));
        let before = BTreeSet::from([1, 7]);
        session
            .end_left(Some(&before), "deploy", &log)
            .await
            .unwrap();
        session
            .end_left(Some(&running), "quiet", &log)
            .await
            .unwrap();
        session.end_left(None, "unwatched", &log).await.unwrap();
        let calls = std::fs::read_to_string(dir.path().join("calls")).unwrap();
        assert!(calls.contains("exec c1 kill -9 9\n"), "{calls}");
        assert_eq!(calls.matches("kill").count(), 1, "{calls}");
        let said = std::fs::read_to_string(&log_path).unwrap();
        assert_eq!(
            said,
            "==> ended 1 of what deploy left running, as it was given secrets\n"
        );
        let shell = Session::Shell {
            dir: "/w".into(),
            vars: Vec::new(),
        };
        assert_eq!(shell.running().await, None);
        shell.end_left(Some(&before), "s", &log).await.unwrap();
    }

    fn overlay_of(volume: &str, path: &str) -> Cache {
        Cache {
            volume: volume.into(),
            path: path.into(),
            run: "r1".into(),
            lower: "/c/bases/t/0".into(),
            upper: format!("/c/runs/r1/{volume}/upper").into(),
            work: format!("/c/runs/r1/{volume}/work").into(),
        }
    }

    #[tokio::test]
    async fn one_container_a_run() {
        let dir = tempfile::tempdir().unwrap();
        let docker = fake_docker(dir.path());
        let log = std::fs::File::create(dir.path().join("log")).unwrap();
        let vars = [("SECRET".to_string(), "v".to_string())];
        let executor = Executor::parse("docker", "debian:stable-slim").unwrap();
        let time = Duration::from_secs(10);
        let src = Path::new("/w/src");
        let caches = [overlay_of("x", "/work/target"), overlay_of("y", "/root")];
        let setup = |image, vars, caches| Setup {
            dir: src,
            vars,
            image,
            caches,
            outputs: Path::new("/w/outputs"),
            steps: None,
            owner: Owner {
                runner: "did:r",
                run: "k1",
            },
            behind: false,
            vouched: false,
        };
        // A pull's, behind the rest for CPU.
        let behind = Setup {
            behind: true,
            steps: Some(Path::new("/w/steps")),
            ..setup(Some("rust:1"), &vars, &caches)
        };
        let session = executor
            .start_with(&docker, &behind, &log, time)
            .await
            .unwrap();
        let ran = run(
            session.command("echo ran", &[]),
            &log,
            time,
            std::future::pending(),
        );
        assert!(ran.await.unwrap().success());
        // Its overlays go with its container.
        session.stop().await;
        // Containers a runner left are removed: its own, but for a run
        // going, and any mounting a path under its work directory, as one
        // from before runs were labeled does. One that left none removes
        // none, and one with no docker to ask removes nothing.
        let work = Some(Path::new("/w/work"));
        let going = Mutex::new(HashSet::from(["r8".to_string()]));
        executor
            .remove_left_with(&docker, "did:left", &going, work)
            .await;
        let none = Mutex::default();
        executor
            .remove_left_with(&docker, "did:none", &none, None)
            .await;
        let gone = Path::new("/nonexistent/docker");
        executor
            .remove_left_with(gone, "did:left", &none, work)
            .await;
        Executor::Shell.remove_left("did:left", &none, None).await;
        let calls = std::fs::read_to_string(dir.path().join("calls")).unwrap();
        let inspect = "inspect --format {{.Id}}|{{index .Config.Labels \"dev.knotless.runner\"}}|\
             {{index .Config.Labels \"dev.knotless.run\"}}|{{range .Mounts}}{{.Source}};{{end}} \
             c7 c8 c9 c10";
        assert_eq!(
            calls,
            format!(
                "volume create --driver local --label dev.knotless.overlay=r1 --opt type=overlay \
             --opt device=overlay --opt o=lowerdir=/c/bases/t/0,upperdir=/c/runs/r1/x/upper,\
             workdir=/c/runs/r1/x/work,userxattr x\n\
             volume create --driver local --label dev.knotless.overlay=r1 --opt type=overlay \
             --opt device=overlay --opt o=lowerdir=/c/bases/t/0,upperdir=/c/runs/r1/y/upper,\
             workdir=/c/runs/r1/y/work,userxattr y\n\
             run -d --init --label dev.knotless.runner=did:r --label dev.knotless.run=k1 -v /w/src:/work -v /w/outputs:/outputs -v /w/steps:/steps:ro -w /work \
             --cpu-shares 256 --mount type=volume,src=x,dst=/work/target \
             --mount type=volume,src=y,dst=/root \
             -e SECRET -e KNOTLESS_RUN_OUTPUTS rust:1 tail -f /dev/null\n\
             exec c1 sh -ec echo ran\n\
             rm -f c1\n\
             volume rm -f x y\n\
             ps -aq\n{inspect}\n\
             rm -f c7 c9\n\
             ps -aq\n{inspect}\n"
            )
        );
        assert_eq!(
            std::fs::read_to_string(dir.path().join("log")).unwrap(),
            "ran\n"
        );
        // An image that will not start, or not in time, is an error, and
        // the overlays made for it go.
        std::fs::remove_file(dir.path().join("calls")).unwrap();
        let err = executor
            .start_with(&docker, &setup(Some("bad"), &[], &caches[..1]), &log, time)
            .await
            .unwrap_err();
        assert_eq!(err.to_string(), "bad did not start: see the log");
        let calls = std::fs::read_to_string(dir.path().join("calls")).unwrap();
        assert!(
            calls.ends_with("tail -f /dev/null\nvolume rm -f x\n"),
            "{calls}"
        );
        // One whose overlay is not made does not start.
        std::fs::remove_file(dir.path().join("calls")).unwrap();
        let bad = [
            caches[0].clone(),
            overlay_of("knotless-overlay-bad", "/root"),
        ];
        let err = executor
            .start_with(&docker, &setup(Some("rust:1"), &[], &bad), &log, time)
            .await
            .unwrap_err();
        assert_eq!(
            err.to_string(),
            "the overlay at /root was not made: see the log"
        );
        let calls = std::fs::read_to_string(dir.path().join("calls")).unwrap();
        assert!(!calls.contains("run -d"), "{calls}");
        assert!(
            calls.ends_with(" knotless-overlay-bad\nvolume rm -f x\n"),
            "{calls}"
        );
        let err = executor
            .start_with(
                &docker,
                &setup(Some("slow"), &[], &[]),
                &log,
                Duration::from_millis(100),
            )
            .await
            .unwrap_err();
        assert_eq!(err.to_string(), "slow did not start in 0s");
        let gone = Path::new("/nonexistent/docker");
        let err = executor
            .start_with(gone, &setup(None, &[], &[]), &log, time)
            .await
            .unwrap_err();
        assert_eq!(err.to_string(), "starting a container");
        let err = executor
            .start_with(gone, &setup(None, &[], &caches), &log, time)
            .await
            .unwrap_err();
        assert_eq!(err.to_string(), "making a cache's overlay");
        // The shell's is the checkout itself.
        let shell = Executor::Shell
            .start(&setup(None, &vars, &caches), &log, time)
            .await
            .unwrap();
        assert_eq!(
            shell,
            Session::Shell {
                dir: src.into(),
                vars: vec![vars[0].clone(), (OUTPUTS.into(), "/w/outputs".into())]
            }
        );
        shell.stop().await;
    }

    #[tokio::test]
    async fn an_earlier_runners_caches_go() {
        let dir = tempfile::tempdir().unwrap();
        let docker = fake_docker(dir.path());
        let executor = Executor::parse("docker", "debian:stable-slim").unwrap();
        // The runs whose overlays are still mounted stay, each once.
        assert_eq!(executor.clean_caches_with(&docker).await, ["r0", "r1"]);
        assert!(Executor::Shell.clean_caches().await.is_empty());
        let calls = std::fs::read_to_string(dir.path().join("calls")).unwrap();
        assert_eq!(
            calls,
            "volume ls -q --filter label=dev.knotless.overlay\n\
             volume rm -f o1 o2\n\
             volume ls -q --filter name=knotless-cache-\n\
             volume rm -f old\n\
             volume ls --filter label=dev.knotless.overlay --format {{.Label \"dev.knotless.overlay\"}}\n"
        );
    }

    #[tokio::test]
    async fn a_run_within_its_limits() {
        let dir = tempfile::tempdir().unwrap();
        let docker = fake_docker(dir.path());
        let log = std::fs::File::create(dir.path().join("log")).unwrap();
        let time = Duration::from_secs(10);
        let caches = [overlay_of("x", "/work/target")];
        let setup = Setup {
            dir: Path::new("/w/src"),
            vars: &[],
            image: None,
            caches: &caches,
            outputs: Path::new("/w/outputs"),
            steps: None,
            owner: Owner {
                runner: "did:r",
                run: "k1",
            },
            behind: false,
            vouched: false,
        };
        let within = |network: &str| Executor::Docker {
            image: "rust:1".into(),
            limits: Limits {
                cpus: Some("1.5".into()),
                memory: Some(4 << 30),
                pids: Some(512),
                network: Some(network.into()),
                runtime: Some("runsc".into()),
                trusted_runtime: Some("runc".into()),
                resolv: Some("/r".into()),
            },
        };
        let calls = || {
            let path = dir.path().join("calls");
            let text = std::fs::read_to_string(&path).unwrap();
            std::fs::remove_file(&path).unwrap();
            text
        };
        let create = "network create --driver bridge --opt \
             com.docker.network.bridge.enable_icc=false";
        let run = "run -d --init --label dev.knotless.runner=did:r --label dev.knotless.run=k1 \
             -v /w/src:/work -v /w/outputs:/outputs -w /work \
             --cpus 1.5 --memory 4294967296 --memory-swap 4294967296 --pids-limit 512";
        let mount = "--mount type=volume,src=x,dst=/work/target -e KNOTLESS_RUN_OUTPUTS \
             rust:1 tail -f /dev/null";
        // A network there already is joined as it is; one missing is made,
        // its containers kept apart; one made meanwhile by another run's
        // start is joined too.
        for (network, asked) in [
            ("have", "network inspect have\n".to_string()),
            ("new", format!("network inspect new\n{create} new\n")),
            (
                "raced",
                format!("network inspect raced\n{create} raced\nnetwork inspect raced\n"),
            ),
        ] {
            let session = within(network)
                .start_with(&docker, &setup, &log, time)
                .await
                .unwrap();
            session.stop().await;
            let calls = calls();
            let started = format!(
                "{asked}{run} --network {network} --runtime runsc \
                 -v /r:/etc/resolv.conf:ro {mount}\n"
            );
            assert!(calls.contains(&started), "{calls}");
        }
        // Code vouched for runs under the trusted runtime instead.
        let vouched = Setup {
            vouched: true,
            ..setup
        };
        let session = within("have")
            .start_with(&docker, &vouched, &log, time)
            .await
            .unwrap();
        session.stop().await;
        let ran = calls();
        assert!(ran.contains(" --runtime runc -v /r:"), "{ran}");
        // One that cannot be made starts nothing, and its overlays go.
        let err = within("gone")
            .start_with(&docker, &setup, &log, time)
            .await
            .unwrap_err();
        assert_eq!(
            err.to_string(),
            "the network gone was not made: see the log"
        );
        let calls = calls();
        assert!(!calls.contains("run -d"), "{calls}");
        assert!(
            calls.ends_with("network inspect gone\nvolume rm -f x\n"),
            "{calls}"
        );
        // With no docker to ask, it is not known to be there.
        let gone = Path::new("/nonexistent/docker");
        let err = within("have")
            .start_with(
                gone,
                &Setup {
                    caches: &[],
                    ..setup
                },
                &log,
                time,
            )
            .await
            .unwrap_err();
        assert_eq!(err.to_string(), "asking for a network");
        let text = std::fs::read_to_string(dir.path().join("log")).unwrap();
        assert_eq!(text, "no network\n");
    }

    #[tokio::test]
    async fn runs_and_times_out() {
        let dir = tempfile::tempdir().unwrap();
        let log = std::fs::File::create(dir.path().join("log")).unwrap();
        let vars = [("WHO".to_string(), "step".to_string())];
        let shell = Session::Shell {
            dir: dir.path().into(),
            vars: vars.to_vec(),
        };
        let cmd = shell.command("echo hi $WHO; echo err >&2; exit 3", &[]);
        let never = std::future::pending::<()>;
        let exit = run(cmd, &log, Duration::from_secs(10), never())
            .await
            .unwrap();
        assert_eq!(exit.code, Some(3));
        assert!(!exit.success());
        // A child of the shell is killed with it.
        let cmd = shell.command("sleep 30 & sleep 30", &[]);
        let exit = run(cmd, &log, Duration::from_millis(200), never())
            .await
            .unwrap();
        assert_eq!((exit.code, exit.cancelled), (None, false));
        assert!(exit.elapsed < Duration::from_secs(5));
        // Cancelled, likewise.
        let cmd = shell.command("sleep 30 & sleep 30", &[]);
        let soon = tokio::time::sleep(Duration::from_millis(200));
        let exit = run(cmd, &log, Duration::from_secs(10), soon).await.unwrap();
        assert_eq!((exit.code, exit.cancelled), (None, true));
        assert!(exit.elapsed < Duration::from_secs(5));
        let text = std::fs::read_to_string(dir.path().join("log")).unwrap();
        assert_eq!(text, "hi step\nerr\n");
        let mut missing = shell.command("true", &[]);
        missing.current_dir(dir.path().join("gone"));
        assert!(
            run(missing, &log, Duration::from_secs(1), never())
                .await
                .is_err()
        );
    }
}