diff --git a/services/ci/src/commit_probe.rs b/services/ci/src/commit_probe.rs new file mode 100644 index 00000000..c29635f6 --- /dev/null +++ b/services/ci/src/commit_probe.rs @@ -0,0 +1,186 @@ +//! Probe whether a commit can be checked out from a repository's storage. +//! +//! A CI run is only meaningful for a commit its workspace can actually +//! check out. The workspace is a clone of the repository's bare storage, so +//! a commit the storage does not hold — or holds only under a ref the +//! clone does not fetch — yields a run that can only fail, with +//! `fatal: unable to read tree ` as its reason. +//! +//! Two trigger paths race for every push: an external relay that learns +//! about a push before the mirror has received the objects, and the +//! mirror's own receive path, which arrives later. Accepting the first +//! without a presence check creates a doomed run for every push and a +//! second, real run behind it. + +use std::path::Path; +use std::time::Duration; + +/// The state of a commit in a repository's storage. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum CommitAvailability { + /// The commit exists and a ref the workspace clone fetches reaches it. + Available, + /// The repository does not hold the commit at all. + Missing, + /// The repository holds the commit, but no fetched ref reaches it, so a + /// clone of this repository cannot check it out. + Unreachable, +} + +impl CommitAvailability { + /// Whether the commit can be checked out. + pub fn is_available(self) -> bool { + matches!(self, CommitAvailability::Available) + } +} + +/// Probe the storage at `git_path` for `commit`. +/// +/// Availability is decided by the same rule the workspace clone obeys: the +/// commit must be reachable from a branch or tag, because a clone fetches +/// `refs/heads/*` and `refs/tags/*` and nothing else. A commit parked +/// under another namespace is present in the object store and still +/// unreachable at checkout, so object existence alone is not enough. +pub async fn probe( + git_path: &Path, + commit: &str, + budget: Duration, +) -> anyhow::Result { + let revision = format!("{commit}^{{commit}}"); + let exists = crate::run_git_output( + git_command(git_path, &["cat-file", "-e", &revision]), + "commit presence", + budget, + ) + .await?; + if !exists.status.success() { + return Ok(CommitAvailability::Missing); + } + let containing = crate::run_git_output( + git_command( + git_path, + &[ + "for-each-ref", + "--contains", + &revision, + "--format=%(refname)", + "refs/heads", + "refs/tags", + ], + ), + "commit reachability", + budget, + ) + .await?; + if !containing.status.success() { + // `for-each-ref` refuses a revision it cannot resolve. The object + // exists (the presence check said so), so this is a reachability + // question the probe cannot answer: treat it as unreachable rather + // than as an error, so no run is created for a commit the clone + // cannot reach. + return Ok(CommitAvailability::Unreachable); + } + let refs = String::from_utf8_lossy(&containing.stdout); + if refs.lines().any(|line| !line.trim().is_empty()) { + Ok(CommitAvailability::Available) + } else { + Ok(CommitAvailability::Unreachable) + } +} + +/// A `git` invocation rooted at `git_path`. +fn git_command(git_path: &Path, args: &[&str]) -> tokio::process::Command { + let mut command = tokio::process::Command::new("git"); + command.arg("-C").arg(git_path).args(args); + command +} + +#[cfg(test)] +mod tests { + use super::*; + use std::process::Command; + + fn run_git(args: &[&str], cwd: &Path) -> String { + let output = Command::new("git") + .args(args) + .current_dir(cwd) + .env("GIT_CONFIG_NOSYSTEM", "1") + .output() + .expect("spawn git"); + assert!( + output.status.success(), + "git {} failed: {}", + args.join(" "), + String::from_utf8_lossy(&output.stderr) + ); + String::from_utf8_lossy(&output.stdout).trim().to_string() + } + + /// A bare repository with one branch and one commit. + fn seed_repository(root: &Path) -> (std::path::PathBuf, String) { + std::fs::create_dir_all(root).expect("create seed root"); + let bare = root.join("probe.git"); + let seed = root.join("seed"); + run_git( + &[ + "init", + "--bare", + "--initial-branch=main", + bare.to_str().expect("bare path"), + ], + root, + ); + std::fs::create_dir_all(&seed).expect("create seed dir"); + run_git(&["init", "--initial-branch=main"], &seed); + run_git(&["config", "user.email", "probe@example.com"], &seed); + run_git(&["config", "user.name", "Probe"], &seed); + std::fs::write(seed.join("file.txt"), "content").expect("write file"); + run_git(&["add", "."], &seed); + run_git(&["commit", "-m", "probe commit"], &seed); + run_git( + &["remote", "add", "origin", bare.to_str().expect("bare path")], + &seed, + ); + run_git(&["push", "origin", "main"], &seed); + let head = run_git(&["rev-parse", "HEAD"], &seed); + (bare, head) + } + + #[tokio::test] + async fn a_commit_on_a_branch_is_available() { + let root = std::env::temp_dir().join(format!("gf-probe-{}", uuid::Uuid::new_v4())); + let (bare, head) = seed_repository(&root); + let availability = probe(&bare, &head, Duration::from_secs(30)) + .await + .expect("probe"); + assert_eq!(availability, CommitAvailability::Available); + let _ = std::fs::remove_dir_all(&root); + } + + #[tokio::test] + async fn an_unknown_commit_is_missing() { + let root = std::env::temp_dir().join(format!("gf-probe-{}", uuid::Uuid::new_v4())); + let (bare, _) = seed_repository(&root); + let availability = probe(&bare, &"deadbeef".repeat(5), Duration::from_secs(30)) + .await + .expect("probe"); + assert_eq!(availability, CommitAvailability::Missing); + let _ = std::fs::remove_dir_all(&root); + } + + #[tokio::test] + async fn a_commit_only_a_non_fetched_namespace_reaches_is_unreachable() { + let root = std::env::temp_dir().join(format!("gf-probe-{}", uuid::Uuid::new_v4())); + let (bare, head) = seed_repository(&root); + // Park the commit under a namespace a clone does not fetch. The + // object stays in the store, so a presence check alone would call it + // available — and the clone would still fail to check it out. + run_git(&["update-ref", "refs/ci/probe", &head], &bare); + run_git(&["update-ref", "-d", "refs/heads/main"], &bare); + let availability = probe(&bare, &head, Duration::from_secs(30)) + .await + .expect("probe"); + assert_eq!(availability, CommitAvailability::Unreachable); + let _ = std::fs::remove_dir_all(&root); + } +} diff --git a/services/ci/src/main.rs b/services/ci/src/main.rs index 85b022de..adc570aa 100644 --- a/services/ci/src/main.rs +++ b/services/ci/src/main.rs @@ -37,6 +37,8 @@ use std::time::Duration; use tokio::time::timeout; use tower_http::trace::TraceLayer; +mod commit_probe; + type PipelineCache = HashMap; type PipelineRegistry = HashMap>; @@ -87,6 +89,21 @@ const ENGINE_REDRIVE_INTERVAL_SECS: u64 = 300; /// terminal and visible (`failed`) instead of retried. const TRIGGER_MAX_ATTEMPTS: i64 = 5; +/// How long a commit-presence probe may spend in `git` before it is killed. +/// +/// The probe runs on the trigger path, so it must not be able to wedge the +/// endpoint: a repository with a slow object store answers "unknown" +/// instead of holding the request open. +const TRIGGER_PROBE_TIMEOUT_SECS: u64 = 15; + +/// How long a trigger for a commit the repository does not hold yet waits +/// for the objects to arrive before it fails terminally. +/// +/// The wait is bounded so a commit that never arrives cannot leave a +/// trigger pending forever: the row fails visibly, with a cause, instead +/// of lingering. +const TRIGGER_DEFER_MAX_AGE_SECS: i64 = 30 * 60; + struct TriggerState { event_bus: Arc, /// Durable store for trigger requests. `None` keeps the development @@ -509,6 +526,33 @@ async fn main() -> anyhow::Result<()> { .await; continue; } + // A trigger for a commit the repository does not hold + // yet must not be published: the run would fail at + // checkout with `unable to read tree`. The relay's + // trigger for a push the mirror has not received yet + // waits here until the objects arrive. + if let Some(state) = + probe_trigger_commit(Some(pool), request.repo_id, &request.new_hash).await + { + if !state.is_available() { + if defer_budget_exhausted(&request) { + let _ = gitforge_db::queries::TriggerRequestQueries::mark_failed( + pool, + request.id, + "the commit never became available in the repository", + request.run_id, + ) + .await; + continue; + } + tracing::warn!( + trigger = %request.id, + repo = %request.repo_id, + "deferring trigger: the commit is not yet available in the repository" + ); + continue; + } + } tracing::warn!( trigger = %request.id, repo = %request.repo_id, @@ -914,6 +958,57 @@ async fn get_trigger_request_status( .into_response() } +/// Probe the repository's storage for `commit`. +/// +/// `None` means the probe could not be decided — no durable store to +/// resolve the repository from, the repository is not registered, or `git` +/// itself failed. The caller must not read that as "the commit is +/// unavailable": the probe decides the one failure mode it can decide, and +/// an undecidable probe must not silence a trigger that would have worked. +async fn probe_trigger_commit( + pool: Option<&gitforge_db::Pool>, + repo_id: gitforge_common::RepoId, + commit: &str, +) -> Option { + let pool = pool?; + let repository = match gitforge_db::queries::RepoQueries::get(pool, repo_id).await { + Ok(Some(repository)) => repository, + Ok(None) => { + tracing::warn!( + %repo_id, + "commit presence probe skipped: the repository is not registered" + ); + return None; + } + Err(error) => { + tracing::warn!( + %error, + %repo_id, + "commit presence probe skipped: the repository lookup failed" + ); + return None; + } + }; + match commit_probe::probe( + std::path::Path::new(&repository.git_path), + commit, + Duration::from_secs(TRIGGER_PROBE_TIMEOUT_SECS), + ) + .await + { + Ok(state) => Some(state), + Err(error) => { + tracing::warn!(%error, %repo_id, "commit presence probe failed"); + None + } + } +} + +/// Whether a deferred trigger has waited past its budget. +fn defer_budget_exhausted(request: &gitforge_db::models::PipelineTriggerRequest) -> bool { + (Utc::now() - request.created_at).num_seconds() > TRIGGER_DEFER_MAX_AGE_SECS +} + async fn trigger_pipeline( Extension(trigger_state): Extension>, Json(request): Json, @@ -944,6 +1039,18 @@ async fn trigger_pipeline( // Captured before `request` is consumed by the event payload below. let request_pipeline_id = request.pipeline_id; + // A trigger for a commit the repository does not hold cannot produce a + // runnable workspace: the clone fails with `unable to read tree`, and + // the run is a doomed one. Two trigger paths race for every push — an + // external relay that learns of the push before the mirror received the + // objects, and the mirror's own receive path, which arrives later — so + // the early trigger is recorded durably and published by the sweep once + // the commit is actually checkable. The mirror's trigger for the same + // commit dedups onto that row, and exactly one run is created. + let commit_state = + probe_trigger_commit(trigger_state.db.as_ref(), repo_id, &request.new_hash).await; + let commit_deferred = matches!(commit_state, Some(state) if !state.is_available()); + let working_dir = match request.working_dir { Some(path) => match validate_workspace_path(&path) { Ok(path) => Some(path), @@ -1021,6 +1128,29 @@ async fn trigger_pipeline( None }; + // The commit is not (yet) in the repository's storage. The trigger + // request stays pending and the durable sweep publishes it once the + // objects arrive; publishing now would create a run that can only fail + // at checkout. + if commit_deferred { + tracing::warn!( + %repo_id, + new_hash = %request.new_hash, + "trigger deferred: the commit is not yet available in the repository" + ); + return ( + StatusCode::ACCEPTED, + Json(serde_json::json!({ + "status": "queued", + "deferred": true, + "reason": "commit_not_available", + "trigger_id": trigger_request.map(|request| request.id.to_string()), + "repo_id": repo_id.to_string(), + "new_hash": request.new_hash, + })), + ); + } + let mut event = EventEnvelope::new( EventType::PushReceived, EventPayload::PushReceived(PushReceivedPayload { @@ -2081,18 +2211,18 @@ fn parse_workspace_prep_timeout(value: Option) -> Duration { /// `output()` future is dropped, and without the kill flag the child would /// outlive its future and keep cloning — the exact unbounded state this /// helper exists to prevent. -async fn run_git_output( +pub(crate) async fn run_git_output( mut command: tokio::process::Command, step: &str, budget: Duration, ) -> anyhow::Result { command.kill_on_drop(true); match timeout(budget, command.output()).await { - Ok(output) => output.map_err(|error| { - anyhow::anyhow!("workspace prep step '{step}' failed to spawn: {error}") - }), + Ok(output) => { + output.map_err(|error| anyhow::anyhow!("git step '{step}' failed to spawn: {error}")) + } Err(_) => Err(anyhow::anyhow!( - "workspace prep step '{step}' exceeded its {}s budget and was killed", + "git step '{step}' exceeded its {}s budget and was killed", budget.as_secs() )), } diff --git a/services/ci/tests/ci_trigger_flow.rs b/services/ci/tests/ci_trigger_flow.rs index 9c25a87f..da23af6e 100644 --- a/services/ci/tests/ci_trigger_flow.rs +++ b/services/ci/tests/ci_trigger_flow.rs @@ -328,3 +328,78 @@ async fn test_trigger_requires_token_and_runs_committed_pipeline() { common::shutdown_gracefully(&mut service.child).await; } + +/// A trigger for a commit the repository does not hold must not create a +/// run: the workspace clone would fail with `unable to read tree` and the +/// run would be doomed from the moment it was created. The request is +/// recorded durably and deferred until the objects arrive, so the sweep +/// can publish it later — and the racing second trigger for the same +/// commit dedups onto that row instead of minting a second run. +#[tokio::test] +async fn test_trigger_for_an_absent_commit_is_deferred_and_creates_no_run() { + let mut service = spawn_ci().await; + + let absent_hash = "d".repeat(40); + let trigger_body = format!( + "{{\"repo_id\":\"{}\",\"ref_name\":\"refs/heads/main\",\ + \"old_hash\":\"{}\",\"new_hash\":\"{}\",\"pusher_id\":\"{}\",\ + \"working_dir\":null}}", + service.repo_id, + "0".repeat(40), + absent_hash, + service.user_id + ); + + let (status, body) = post_trigger( + service.scheduler_port, + Some("harness-trigger-token"), + &trigger_body, + ) + .await; + assert_eq!(status, reqwest::StatusCode::ACCEPTED, "body: {body}"); + let payload: serde_json::Value = serde_json::from_str(&body).expect("parse trigger response"); + assert_eq!(payload["status"], "queued", "body: {body}"); + assert_eq!(payload["deferred"], true, "body: {body}"); + assert_eq!(payload["reason"], "commit_not_available", "body: {body}"); + assert!( + payload["pipeline_run_id"].is_null(), + "a deferred trigger must not report a run: {body}" + ); + let trigger_id = payload["trigger_id"] + .as_str() + .expect("trigger_id in response") + .to_string(); + + // The deferred request is durable and still pending: the sweep probes + // again once the objects arrive. + let (status, body) = get_trigger_status( + service.scheduler_port, + Some("harness-trigger-token"), + &trigger_id, + ) + .await; + assert_eq!(status, reqwest::StatusCode::OK, "body: {body}"); + let receipt: serde_json::Value = serde_json::from_str(&body).expect("parse trigger receipt"); + assert_eq!(receipt["status"], "pending", "body: {body}"); + assert!( + receipt["pipeline_run_id"].is_null(), + "a deferred trigger must not have a run: {body}" + ); + + // The decisive assertion: no run exists for the absent commit. + let pool = gitforge_db::Pool::new(&service.db_path.display().to_string()) + .await + .expect("reopen service database"); + let runs = gitforge_db::queries::PipelineRunQueries::list(&pool) + .await + .expect("list runs"); + assert!( + runs.is_empty(), + "a deferred trigger must not create a run: {:?}", + runs.iter() + .map(|run| (run.commit_hash.clone(), run.status.clone())) + .collect::>() + ); + + common::shutdown_gracefully(&mut service.child).await; +}