gated-exec
Source: crates/lvcs/src/runner/gated.rs.
/// One gated run, as the shell fills it from the runner actor's `Prepare`
/// and its gated action.
///
/// `run` is the run's key, `frontier` the frontier it runs on (the
/// actor's, deleted documents and all), `channel` the action channel
/// it serves, `action` the gated action, and `trigger` what started it: a
/// run request (`manual`) or the action's channel trigger (`channel`,
/// decision 329). A credential's value is
/// never held here: the executor reads each from the process environment
/// as the step starts.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct GatedJob {
/// The run's key, its `LVCS_RUN`.
pub(crate) run: RunKey,
/// The frontier it runs on, whose id is its `LVCS_FRONTIER`.
pub(crate) frontier: Frontier,
/// The channel it serves, its `LVCS_CHANNEL`.
pub(crate) channel: String,
/// The gated action.
pub(crate) action: Action,
/// What started it, its `LVCS_TRIGGER`.
pub(crate) trigger: RunTrigger,
}
/// What one of a run's reads gave: a document's bytes and executable bit
/// at the frontier (row 307), or the document deleted there, which the run
/// sees absent (decision 145).
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum Input {
/// The document's bytes and executable bit.
File(TreeFile),
/// The document is deleted at its heads in the frontier.
Deleted,
}
/// How a gated run ended: `passed` when its command exited 0 within
/// `GATED_TIMEOUT_SECS`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct GatedResult {
/// Whether its command exited 0 within its timeout.
pub(crate) passed: bool,
}
/// The executor's order (decision 168): the run holding it, from its reads
/// to its end, and the runs the actor asked to prepare meanwhile, oldest
/// first, each holding only its frontier and action.
#[derive(Debug, Default)]
pub(crate) struct Executor {
/// The run holding the executor.
current: Option<RunKey>,
/// The runs waiting their turn, oldest first.
waiting: VecDeque<GatedJob>,
}
impl Executor {
/// The actor asked to prepare `job`: it takes the executor and comes
/// back when no run holds it, else it waits its turn. A run already
/// holding or waiting for the executor is asked once.
pub(crate) fn ask(&mut self, job: GatedJob) -> Option<GatedJob> {
let asked = self.current == Some(job.run)
|| self.waiting.iter().any(|waiting| waiting.run == job.run);
if asked {
return None;
}
if self.current.is_some() {
self.waiting.push_back(job);
return None;
}
self.current = Some(job.run);
Some(job)
}
/// Run `run` ended: when it held the executor, the oldest waiting run
/// takes it and comes back; a waiting one leaves the queue.
pub(crate) fn end(&mut self, run: RunKey) -> Option<GatedJob> {
if self.current != Some(run) {
self.waiting.retain(|waiting| waiting.run != run);
return None;
}
self.current = None;
let next = self.waiting.pop_front()?;
self.current = Some(next.run);
Some(next)
}
}
/// The documents the shell reads through `ReadAt` at `job`'s frontier:
/// every file document its `paths` match.
///
/// # Errors
///
/// When a `paths` pattern doesn't parse, which fails the run.
pub(crate) fn reads(job: &GatedJob) -> io::Result<BTreeSet<String>> {
let paths = PathSet::new(&job.action.inputs.paths, false).map_err(|err| {
let reason = match err {
PathSetError::Pattern { reason, .. } | PathSetError::Build { reason } => reason,
};
io::Error::other(format!("a paths pattern doesn't parse: {reason}"))
})?;
Ok(job
.frontier
.docs
.keys()
.filter(|address| address::kind_of(address) == Some(DocKind::File))
.filter(|address| paths.matches(address))
.cloned()
.collect())
}
/// Write run `job`'s fresh directory ([`run_dir`], 0700, owned by the
/// daemon's uid) from what its reads gave.
///
/// Its `work/` holds each document [`reads`] names with its bytes, mode
/// 0755 when its document's executable bit is set and 0644 otherwise, as
/// the mirror gives it (decision 85, row 307), a deleted one left absent, beside a fresh `home/`, `cargo/`, and `tmp/`.
/// The run directory comes back.
///
/// # Errors
///
/// When a document [`reads`] names has no answer in `inputs` (a document
/// refused the read, which fails the run), an address isn't a plain
/// relative path, or the directory or a file can't be written.
pub(crate) fn prepare(
layout: &Layout,
job: &GatedJob,
inputs: &BTreeMap<String, Input>,
) -> io::Result<PathBuf> {
let addresses = reads(job)?;
if addresses
.iter()
.any(|address| !inputs.contains_key(address))
{
return Err(io::Error::other("a document refused the run's read"));
}
let dir = run_dir(layout, &job.run);
super::fresh_run_dir(&dir)?;
let work = dir.join(WORK);
for address in &addresses {
if let Some(Input::File(file)) = inputs.get(address) {
write_input(&work, address, file)?;
}
}
Ok(dir)
}
/// Run `job`'s gated action in its prepared directory, journaled as
/// `journal`, recording a `run` span with `outcome` `ok` when it
/// passed and `error` otherwise.
///
/// Its profile sets `network` and, for every step, `LVCS_FRONTIER` (the
/// frontier's id), `LVCS_CHANNEL`, `LVCS_RUN` (the run key), and
/// `LVCS_TRIGGER` (`manual` or `channel`, the `run_trigger` vocabulary); each
/// secret its `env` names is filled from this process's environment as
/// its step starts. A name missing there, or not Unicode, fails the step
/// before it starts, and its `log` says which; no value is ever logged,
/// journaled, put in a span, or written by LVCS under the project root.
/// Dropping `kill`'s switch cancels the run (decision 249, revised): the
/// step under way is killed with its PID namespace, and no later step
/// starts.
pub(crate) fn run(
layout: &Layout,
job: &GatedJob,
journal: RunJournal,
kill: KillWatch,
progress: super::steps::Progress,
) -> GatedResult {
let span = tracing::info_span!("run", outcome = Empty);
let _entered = span.enter();
let dir = run_dir(layout, &job.run);
let passed = match profile(job, kill, progress) {
Ok(profile) => super::steps::execute(&dir, &job.action, &profile, journal)
.inspect_err(|err| tracing::error!(error = %err, "a gated run couldn't run"))
.is_ok_and(|ran| ran.passed),
Err(reason) => {
let line = format!("lvcs: run: {reason}\n");
if let Err(err) = super::fs_write_log(&dir, line.as_bytes()) {
tracing::debug!(error = %err, "a gated run's log couldn't be written");
}
false
}
};
let outcome = if passed { Outcome::Ok } else { Outcome::Error };
span.record("outcome", outcome.as_str());
GatedResult { passed }
}
/// Remove run `run`'s `work/`, `home/`, `cargo/`, and `tmp/`, its `log`
/// staying (decision 168); a missing run directory has nothing to remove.
///
/// # Errors
///
/// When a tree can't be removed.
pub(crate) fn end(layout: &Layout, run: &RunKey) -> io::Result<()> {
super::prune_run(&run_dir(layout, run))
}