feat: run checks asynchronously after push instead of blocking the connection
commit
78f343ffeat: run checks asynchronously after push instead of blocking the connection
The post-receive hook no longer runs checks on the push connection. It records the run as queued and drops a job into a shared queue; a server-owned worker drains the queue, runs each check in the repo’s Sprite, and advances the run in place (running, then each check’s result) so the Checks tab fills in like CI. Infra failures finalize the run as error rather than leaving it stuck, and a hung check is killed after a timeout so it can’t wedge the worker for every repo.
feat: queue pushes in post-receive and drain them from a server-owned worker feat: report checks with live queued/running/pass/fail/error statuses feat: bound each check run with a 30-minute timeout refactor: store each check as a CheckDef struct on refs/meta/checks Assisted-by: Claude:claude-opus-4-8
Reviews
No reviews of this commit yet — record a verdict below.
Start a review
crates/git-ents-server/src/checks.rs
@@ -1,39 +1,51 @@
-//! The `post-receive` check runner: a git hook that runs the configured checks
-//! against a push inside a Fly.io [Sprite].
+//! Asynchronous check running: a `post-receive` hook that *queues* a push and a
+//! server-owned worker that runs the configured checks against it in a Fly.io
+//! [Sprite].
//!
-//! Where the `pre-receive` verifier gates the push synchronously, checks run
-//! *after* the refs are in. The runner reads the pushed ref updates git feeds
-//! the hook on stdin, loads the check set from `refs/meta/checks`, and for each
-//! updated branch runs every check in a Sprite — a persistent, hardware-isolated
-//! sandbox. One Sprite is kept per repository so its filesystem (and any build
-//! cache a check leaves behind) survives between pushes; the pushed tree is
-//! synced into it before the checks run. Results are reported on the hook's
-//! stdout, which git relays to the pusher.
+//! Checks run *after* the refs are in and off the push connection. The hook
+//! ([`post_receive`]) does almost nothing: it reads the pushed ref updates git
+//! feeds it on stdin and drops a job file into the shared queue directory, so
+//! the push returns immediately. The long-running server drains that queue from
+//! a dedicated worker ([`worker`]); for each job it loads the check set from
+//! `refs/meta/checks` and runs every check in a Sprite — a persistent,
+//! hardware-isolated sandbox. One Sprite is kept per repository so its
+//! filesystem (and any build cache a check leaves behind) survives between
+//! pushes; the pushed tree is synced into it before the checks run. Results are
+//! recorded as run refs (and surfaced on the Checks tab), and logged to the
+//! server's own output rather than relayed to the pusher.
//!
//! The Sprite is driven through the `sprite` CLI. The CLI authenticates from a
-//! config file rather than the environment, so the runner first hands it the
+//! config file rather than the environment, so the worker first hands it the
//! `SPRITES_TOKEN` the server passes down via `sprite auth setup`; only then
//! does an organization become configured.
//!
//! [Sprite]: https://sprites.dev
use std::io::{Read, Write};
-use std::path::Path;
+use std::path::{Path, PathBuf};
use std::process::{Command, Stdio};
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
use git_ents::checks::{self, Check, RunOutcome};
/// Where the pushed tree is unpacked inside the Sprite.
const WORKDIR: &str = "/work";
-/// Run the configured checks against the push git is reporting, returning
-/// `Ok(())` once results have been printed. The ref updates are read from the
-/// stdin git populates for a `post-receive` hook (`<old> <new> <ref>` lines).
+/// The environment variable through which the server hands the hook the queue
+/// directory; the worker is given the same path directly.
+pub const QUEUE_ENV: &str = "GIT_ENTS_CHECKS_QUEUE";
+
+/// How often the worker scans the queue directory for new jobs.
+const POLL: Duration = Duration::from_secs(2);
+
+/// Queue the push git is reporting for asynchronous checking, returning `Ok(())`
+/// once the jobs are enqueued. The ref updates are read from the stdin git
+/// populates for a `post-receive` hook (`<old> <new> <ref>` lines).
///
-/// A `post-receive` exit code cannot undo refs that are already in, so a failed
-/// check is reported rather than turned into an error: the function returns
-/// `Err` only when the runner itself could not run (unreadable check set, an
-/// unreachable Sprite), never merely because a check failed.
+/// The hook does no check work itself: it writes one job file per updated branch
+/// into the shared queue directory ([`QUEUE_ENV`]) and returns, so the push is
+/// never blocked on a Sprite. The server's [`worker`] picks the jobs up.
pub fn post_receive() -> Result<(), String> {
let repo = std::env::current_dir().map_err(|e| format!("cannot resolve repository: {e}"))?;
@@ -46,32 +58,197 @@
return Ok(());
}
- let checks = checks::load(&repo).map_err(|e| format!("could not read checks: {e}"))?;
- if checks.is_empty() {
+ // An empty check set leaves nothing to queue.
+ let runnable = checks::load(&repo).map_err(|e| format!("could not read checks: {e}"))?;
+ if runnable.is_empty() {
return Ok(());
}
- let sprite = sprite_name(&repo);
- ensure_auth()?;
- ensure_sprite(&sprite)?;
+ let Some(queue) = std::env::var_os(QUEUE_ENV).map(PathBuf::from) else {
+ eprintln!("checks: {QUEUE_ENV} is not set; skipping asynchronous checks");
+ return Ok(());
+ };
for update in updates {
+ enqueue(&queue, &repo, &update)?;
+ // Record the run as `queued` straight away so it shows up on the Checks
+ // tab the moment the push lands, before the worker picks it up; a
+ // recording hiccup is reported but never fails the hook.
+ let queued = statuses(&runnable, "queued");
+ if let Err(e) = checks::record(&repo, update.new, &queued) {
+ eprintln!(
+ "checks: could not record queued run for {}: {e}",
+ update.new
+ );
+ }
println!(
- "checks: running {} check(s) on {}",
- checks.len(),
+ "checks: queued {} check(s) on {}",
+ runnable.len(),
update.ref_name
);
- sync_tree(&repo, &sprite, update.new)?;
- let outcomes = run_checks(&sprite, &checks);
- // Persist the run as a ref (`refs/checks/<commit>`); a recording hiccup
- // is reported but never fails the hook.
- if let Err(e) = checks::record(&repo, update.new, &outcomes) {
- eprintln!("checks: could not record run for {}: {e}", update.new);
- }
}
Ok(())
}
+/// Every check's [`RunOutcome`] set to one shared `status` — the queued/running
+/// snapshot a run starts from before per-check results land.
+fn statuses(checks: &[Check], status: &str) -> Vec<RunOutcome> {
+ checks
+ .iter()
+ .map(|check| RunOutcome {
+ name: check.name.clone(),
+ outcome: status.to_owned(),
+ })
+ .collect()
+}
+
+/// Run the worker that drains the queue directory, running and recording the
+/// checks for each queued push. Runs for the life of the server; the blocking
+/// Sprite work is offloaded so it never stalls the async runtime.
+pub async fn worker(queue: PathBuf) {
+ if let Err(e) = std::fs::create_dir_all(&queue) {
+ eprintln!("checks: could not create queue directory {queue:?}: {e}");
+ return;
+ }
+ let mut tick = tokio::time::interval(POLL);
+ loop {
+ tick.tick().await;
+ let queue = queue.clone();
+ let _drained = tokio::task::spawn_blocking(move || drain(&queue)).await;
+ }
+}
+
+/// Process every job currently in the queue directory once, deleting each job
+/// file after it is handled (whether it ran cleanly, failed, or was malformed) —
+/// a poison job is dropped rather than retried forever.
+fn drain(queue: &Path) {
+ let Ok(entries) = std::fs::read_dir(queue) else {
+ return;
+ };
+ for entry in entries.flatten() {
+ let path = entry.path();
+ if path.extension().is_none_or(|ext| ext != "job") {
+ continue;
+ }
+ if let Some(job) = read_job(&path)
+ && let Err(e) = process_job(&job)
+ {
+ eprintln!("checks: {e}");
+ }
+ let _removed = std::fs::remove_file(&path);
+ }
+}
+
+/// Run the checks for one queued push in its repository's Sprite, advancing the
+/// recorded run as it goes: `running` while the Sprite is prepared, then each
+/// check flipped to its result as it finishes. An infra failure (an unreachable
+/// Sprite, a tree that will not sync) finalizes the run as `error` rather than
+/// leaving it stuck at `running`, then returns `Err`. Returns `Ok` even when a
+/// check fails — a failing check is a recorded result, not an error.
+fn process_job(job: &Job) -> Result<(), String> {
+ let runnable = checks::load(&job.repo).map_err(|e| format!("could not read checks: {e}"))?;
+ if runnable.is_empty() {
+ return Ok(());
+ }
+
+ let mut outcomes = statuses(&runnable, "running");
+ let sprite = sprite_name(&job.repo);
+ if let Err(e) = ensure_auth().and_then(|()| ensure_sprite(&sprite)) {
+ finalize_error(&job.repo, &job.new, &mut outcomes);
+ return Err(e);
+ }
+
+ eprintln!(
+ "checks: running {} check(s) on {}",
+ runnable.len(),
+ job.ref_name
+ );
+ advance(&job.repo, &job.new, &outcomes);
+ if let Err(e) = sync_tree(&job.repo, &sprite, &job.new) {
+ finalize_error(&job.repo, &job.new, &mut outcomes);
+ return Err(e);
+ }
+
+ for (index, check) in runnable.iter().enumerate() {
+ let result = run_one(&sprite, check);
+ if let Some(outcome) = outcomes.get_mut(index) {
+ outcome.outcome = result.to_owned();
+ }
+ advance(&job.repo, &job.new, &outcomes);
+ }
+ Ok(())
+}
+
+/// Advance the recorded run for `new` to `outcomes`; a recording hiccup is
+/// logged but never derails the worker.
+fn advance(repo: &Path, new: &str, outcomes: &[RunOutcome]) {
+ if let Err(e) = checks::update_run(repo, new, outcomes) {
+ eprintln!("checks: could not record run for {new}: {e}");
+ }
+}
+
+/// Mark every check in `outcomes` `error` and record it — the terminal state for
+/// a run the worker could not carry out.
+fn finalize_error(repo: &Path, new: &str, outcomes: &mut [RunOutcome]) {
+ for outcome in outcomes.iter_mut() {
+ outcome.outcome = "error".to_owned();
+ }
+ advance(repo, new, outcomes);
+}
+
+/// One queued push: the repository to check, the new tip to check, and the ref
+/// it updated (carried only for logging).
+struct Job {
+ repo: PathBuf,
+ new: String,
+ ref_name: String,
+}
+
+/// Write a job for `update` into `queue` as a three-line file (`repo`, new oid,
+/// ref). The file is written under a `.tmp` name and renamed into place so the
+/// worker never observes a half-written job.
+fn enqueue(queue: &Path, repo: &Path, update: &Update) -> Result<(), String> {
+ std::fs::create_dir_all(queue)
+ .map_err(|e| format!("could not create queue directory {queue:?}: {e}"))?;
+ let stem = job_stem();
+ let tmp = queue.join(format!("{stem}.tmp"));
+ let final_path = queue.join(format!("{stem}.job"));
+ let body = format!("{}\n{}\n{}\n", repo.display(), update.new, update.ref_name);
+ std::fs::write(&tmp, body).map_err(|e| format!("could not write job: {e}"))?;
+ std::fs::rename(&tmp, &final_path).map_err(|e| format!("could not enqueue job: {e}"))?;
+ Ok(())
+}
+
+/// A unique job file stem (`<nanos>-<pid>-<counter>`) so concurrent pushes never
+/// collide on a queue file name.
+fn job_stem() -> String {
+ static COUNTER: AtomicU64 = AtomicU64::new(0);
+ let nanos = SystemTime::now()
+ .duration_since(UNIX_EPOCH)
+ .map(|d| d.as_nanos())
+ .unwrap_or(0);
+ let n = COUNTER.fetch_add(1, Ordering::Relaxed);
+ format!("{nanos}-{}-{n}", std::process::id())
+}
+
+/// Parse a queued job file (`repo`, new oid, ref, one per line), or `None` when
+/// it is malformed.
+fn read_job(path: &Path) -> Option<Job> {
+ let contents = std::fs::read_to_string(path).ok()?;
+ let mut lines = contents.lines();
+ let repo = PathBuf::from(lines.next()?);
+ let new = lines.next()?.to_owned();
+ let ref_name = lines.next()?.to_owned();
+ if new.is_empty() {
+ return None;
+ }
+ Some(Job {
+ repo,
+ new,
+ ref_name,
+ })
+}
+
/// One ref git reported as updated by the push.
struct Update<'a> {
new: &'a str,
@@ -196,52 +373,80 @@
}
}
-/// Run each check in the Sprite's [`WORKDIR`], printing a `PASS`/`FAIL` line per
-/// check and echoing the output of any that fail so the pusher sees why. Returns
-/// each check's outcome (`pass`/`fail`/`error`) for recording.
-fn run_checks(sprite: &str, checks: &[Check]) -> Vec<RunOutcome> {
- let mut outcomes = Vec::with_capacity(checks.len());
- for check in checks {
- let output = Command::new("sprite")
- .args([
- "exec",
- "-s",
- sprite,
- "--dir",
- WORKDIR,
- "--",
- "sh",
- "-c",
- &check.command,
- ])
- .output();
- let outcome = match output {
- Ok(output) if output.status.success() => {
- println!("checks: PASS {}", check.name);
- "pass"
- }
- Ok(output) => {
- println!("checks: FAIL {} ({})", check.name, check.command);
- let logs = String::from_utf8_lossy(&output.stderr);
- let logs = if logs.trim().is_empty() {
- String::from_utf8_lossy(&output.stdout)
- } else {
- logs
- };
- for line in logs.lines() {
- println!("checks: {line}");
- }
- "fail"
- }
- Err(e) => {
- println!("checks: ERROR {} (could not run: {e})", check.name);
- "error"
- }
+/// How long a single check may run before the worker abandons it. A runaway
+/// check that outlived this — a hung build, a command blocked on input — is
+/// killed and recorded `error` rather than wedging the worker (and with it every
+/// other repository's checks) on the one blocking-pool thread the queue drains
+/// on.
+const CHECK_TIMEOUT: Duration = Duration::from_secs(30 * 60);
+
+/// Run one check in the Sprite's [`WORKDIR`], logging a `PASS`/`FAIL` line and
+/// echoing the output on failure. Returns its outcome (`pass`/`fail`/`error`); a
+/// check that exceeds [`CHECK_TIMEOUT`] or cannot be captured is `error`.
+fn run_one(sprite: &str, check: &Check) -> &'static str {
+ let child = Command::new("sprite")
+ .args([
+ "exec",
+ "-s",
+ sprite,
+ "--dir",
+ WORKDIR,
+ "--",
+ "sh",
+ "-c",
+ &check.command,
+ ])
+ .stdout(Stdio::piped())
+ .stderr(Stdio::piped())
+ .spawn();
+ let child = match child {
+ Ok(child) => child,
+ Err(e) => {
+ eprintln!("checks: ERROR {} (could not run: {e})", check.name);
+ return "error";
+ }
+ };
+ let Some(output) = wait_bounded(child, CHECK_TIMEOUT) else {
+ eprintln!(
+ "checks: ERROR {} (timed out after {:?} or could not be captured)",
+ check.name, CHECK_TIMEOUT
+ );
+ return "error";
+ };
+ if output.status.success() {
+ eprintln!("checks: PASS {}", check.name);
+ "pass"
+ } else {
+ eprintln!("checks: FAIL {} ({})", check.name, check.command);
+ let logs = String::from_utf8_lossy(&output.stderr);
+ let logs = if logs.trim().is_empty() {
+ String::from_utf8_lossy(&output.stdout)
+ } else {
+ logs
};
- outcomes.push(RunOutcome {
- name: check.name.clone(),
- outcome: outcome.to_owned(),
- });
+ for line in logs.lines() {
+ eprintln!("checks: {line}");
+ }
+ "fail"
+ }
+}
+
+/// Wait up to `timeout` for `child` to finish, returning its captured output, or
+/// `None` if it timed out or could not be waited on. On timeout the process is
+/// killed by pid — `sprite exec` is a local proxy for the remote command, so
+/// killing it frees the worker even though the in-Sprite command may run on.
+fn wait_bounded(child: std::process::Child, timeout: Duration) -> Option<std::process::Output> {
+ let pid = child.id();
+ let (tx, rx) = std::sync::mpsc::channel();
+ std::thread::spawn(move || {
+ let _sent = tx.send(child.wait_with_output());
+ });
+ match rx.recv_timeout(timeout) {
+ Ok(Ok(output)) => Some(output),
+ Ok(Err(_failed)) => None,
+ Err(_timeout) => {
+ let _killed = Command::new("kill").args(["-9", &pid.to_string()]).status();
+ None
+ }
}
- outcomes
}
crates/git-ents-server/src/http.rs
@@ -75,6 +75,9 @@
.env("PATH_INFO", &path_info)
.env("QUERY_STRING", &query_string)
.env("REQUEST_METHOD", method.as_str())
+ // Hand the `post-receive` hook the queue it drops jobs into; it inherits
+ // this through the receive-pack process tree git spawns.
+ .env(crate::checks::QUEUE_ENV, &state.checks_queue)
.stdin(Stdio::piped())
.stdout(Stdio::piped())
// Surface backend diagnostics in the server's own logs rather than
@@ -510,6 +513,7 @@
init_lock: std::sync::Arc::new(tokio::sync::Mutex::new(())),
cert_nonce_seed: cert_nonce_seed.map(str::to_owned),
hooks_dir: hooks_dir.map(PathBuf::from),
+ checks_queue: PathBuf::from("/data/checks-queue"),
}
}
crates/git-ents-server/src/main.rs
@@ -48,6 +48,15 @@
#[arg(long, env = "GIT_ENTS_HOOKS_DIR")]
hooks_dir: Option<PathBuf>,
+ /// Directory where the `post-receive` hook queues pushes for the check
+ /// worker to run asynchronously.
+ #[arg(
+ long,
+ env = "GIT_ENTS_CHECKS_QUEUE",
+ default_value = "/data/checks-queue"
+ )]
+ checks_queue: PathBuf,
+
/// Stop after handling this many requests.
#[arg(long)]
max_requests: Option<usize>,
@@ -76,6 +85,9 @@
/// When set, injected as `core.hooksPath` so every served repo runs the
/// bundled `pre-receive` verifier.
pub(crate) hooks_dir: Option<PathBuf>,
+ /// Directory the `post-receive` hook queues pushes into and the check
+ /// worker drains; passed down to the hook via [`checks::QUEUE_ENV`].
+ pub(crate) checks_queue: PathBuf,
}
fn main() -> ExitCode {
@@ -126,8 +138,12 @@
init_lock: Arc::new(Mutex::new(())),
cert_nonce_seed: args.cert_nonce_seed,
hooks_dir: args.hooks_dir,
+ checks_queue: args.checks_queue,
};
+ // Drain queued pushes and run their checks for the life of the server.
+ tokio::spawn(checks::worker(state.checks_queue.clone()));
+
// The git smart-HTTP protocol streams whole packfiles through the request
// body, so the default 2 MiB cap would reject any non-trivial push.
let mut app = Router::new()
crates/git-ents/src/checks.rs
@@ -19,17 +19,26 @@
/// The ref whose tree holds the configured check set.
pub const CHECKS_REF: &str = "refs/meta/checks";
-/// The check document stored at [`CHECKS_REF`]: `checks/<name>` maps to the
-/// command run for that check.
+/// The check document stored at [`CHECKS_REF`]: `checks/<name>` maps to that
+/// check's definition (its command).
#[derive(Debug, Clone, PartialEq, Eq, Facet)]
struct Checks {
- checks: BTreeMap<String, String>,
+ checks: BTreeMap<String, CheckDef>,
+}
+
+/// One check's stored definition under `checks/<name>` in [`CHECKS_REF`]. A
+/// struct (rather than a bare command blob) so each check can grow per-check
+/// settings without a tree-format migration.
+#[derive(Debug, Clone, PartialEq, Eq, Facet)]
+struct CheckDef {
+ /// The shell command run for the check.
+ command: String,
}
/// One configured check recorded under `checks/` in [`CHECKS_REF`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Check {
- /// The `checks/<name>` the command is stored under — the check's name.
+ /// The `checks/<name>` the definition is stored under — the check's name.
pub name: String,
/// The shell command run for the check (e.g. `cargo fmt --check`).
pub command: String,
@@ -66,9 +75,9 @@
Ok(checks
.checks
.into_iter()
- .map(|(name, command)| Check {
+ .map(|(name, def)| Check {
name,
- command: command.trim_end().to_owned(),
+ command: def.command.trim_end().to_owned(),
})
.collect())
}
@@ -79,7 +88,14 @@
let document = Checks {
checks: checks
.iter()
- .map(|check| (check.name.clone(), check.command.clone()))
+ .map(|check| {
+ (
+ check.name.clone(),
+ CheckDef {
+ command: check.command.clone(),
+ },
+ )
+ })
.collect(),
};
let odb = open_odb(repo).ok_or(Error::Odb)?;
@@ -220,7 +236,8 @@
pub struct RunOutcome {
/// The check's name (its `checks/<name>` in [`CHECKS_REF`]).
pub name: String,
- /// The outcome recorded for it — `pass`, `fail`, or `error`.
+ /// The outcome recorded for it as a run progresses — `queued`, `running`,
+ /// then `pass`, `fail`, or `error`.
pub outcome: String,
}
@@ -263,6 +280,50 @@
update_named_ref(repo, &refname, &new_commit)
}
+/// Advance the latest run recorded for `commit` to `outcomes`, in place. Unlike
+/// [`record`], which appends a new run, this replaces the run ref's tip commit
+/// (re-parented on the prior run) so a single run's status can progress —
+/// `queued` → `running` → results — without appending a commit per transition.
+///
+/// When no run has been recorded yet the update starts one, so a worker that
+/// advances a run is self-healing even if the `queued` record never landed.
+pub fn update_run(repo: &Path, commit: &str, outcomes: &[RunOutcome]) -> Result<(), Error> {
+ let doc = RunDoc {
+ results: outcomes
+ .iter()
+ .map(|outcome| (outcome.name.clone(), outcome.outcome.clone()))
+ .collect(),
+ };
+ let odb = open_odb(repo).ok_or(Error::Odb)?;
+ let tree = facet_git_tree::serialize_into(&doc, &odb)?;
+ let refname = format!("{RUNS_NS}/{commit}");
+ let parent = ref_parent(repo, &refname);
+ let new_commit = commit_run(repo, &tree, parent.as_deref())?;
+ update_named_ref(repo, &refname, &new_commit)
+}
+
+/// The first parent of `refname`'s tip commit, or `None` when the tip is a root
+/// commit or the ref is absent.
+fn ref_parent(repo: &Path, refname: &str) -> Option<String> {
+ let spec = format!("{refname}^");
+ let output = Command::new("git")
+ .arg("-C")
+ .arg(repo)
+ .args(["rev-parse", "--verify", "--quiet", &spec])
+ .output()
+ .ok()?;
+ if !output.status.success() {
+ return None;
+ }
+ let hex = String::from_utf8(output.stdout).ok()?;
+ let hex = hex.trim();
+ if hex.is_empty() {
+ None
+ } else {
+ Some(hex.to_owned())
+ }
+}
+
/// List the recorded runs per commit, newest commit first. Each commit's runs
/// are the ref's commit chain, newest first, with the run time taken from each
/// commit's date.
crates/git-ents-server/src/web/pages.rs
@@ -630,8 +630,9 @@
}
/// The Checks tab. The check set lives on `refs/meta/checks` (managed with
-/// `git ents checks`); each push runs them in a Sprite. The Configuration card
-/// reflects the live set; runs are not yet recorded, so that panel is empty.
+/// `git ents checks`); each push queues them and a worker runs them in a Sprite.
+/// The Configuration card reflects the live set; Recent runs reflects the run
+/// log on `refs/meta/runs`, including in-flight `queued`/`running` runs.
pub(super) async fn checks_page(repo: &Path, meta: &RepoMeta) -> Markup {
let checks = load_checks(repo).await;
let runs = load_runs(repo).await;
@@ -643,8 +644,8 @@
div.page-header { h1.page-title { "Checks" } }
p.shell-note {
"Checks are configured on " code { "refs/meta/checks" }
- " (" code { "git ents checks list" } ") and run in a Sprite on each push; "
- "each run is recorded under " code { "refs/checks/<commit>" } "."
+ " (" code { "git ents checks list" } ") and run in a Sprite after each push; "
+ "each run is recorded under " code { "refs/meta/runs/<commit>" } "."
}
div.checks-grid {
div.card {