From b7c1b258411e026ff51dd77efce885c55b69fc79 Mon Sep 17 00:00:00 2001 From: Michael Kinney <14325925+aliasfoxkde@users.noreply.github.com> Date: Fri, 6 Nov 2026 05:56:00 -0600 Subject: [PATCH 1/2] fix(ci): refuse a trigger for a commit the repo cannot check out MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The trigger endpoint accepted any non-zero new_hash, recorded it, and published a run. When the commit was not in the repository's storage the run's workspace clone failed with `fatal: unable to read tree ` and the run was persisted as failed — a doomed run created for every push. 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. The relay's trigger was the one that lost the race, so it produced the doomed run while the mirror's trigger produced the real one behind it. The endpoint now probes the repository's storage before it publishes. The probe decides availability the way the workspace clone does: the commit must exist AND 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. A trigger for a commit that is not available yet is recorded durably and deferred: the response is 202 queued with `deferred: true`, and the durable sweep publishes it once the objects arrive. The mirror's trigger for the same commit dedups onto that row, so exactly one run is created. The deferral is bounded — after 30 minutes the row fails visibly with a cause instead of lingering forever. An undecidable probe (no durable store, an unregistered repository, a git failure) does not defer: the probe decides the one failure mode it can decide, and a probe that cannot answer must not silence a trigger that would have worked. Co-Authored-By: Claude Code --- services/ci/src/commit_probe.rs | 186 ++++++++++++++++++++++++++++++++ services/ci/src/main.rs | 137 ++++++++++++++++++++++- 2 files changed, 320 insertions(+), 3 deletions(-) create mode 100644 services/ci/src/commit_probe.rs 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..3d9a71ec 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(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,30 @@ 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, + })), + ) + .into_response(); + } + let mut event = EventEnvelope::new( EventType::PushReceived, EventPayload::PushReceived(PushReceivedPayload { @@ -2081,7 +2212,7 @@ 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, @@ -2089,10 +2220,10 @@ async fn run_git_output( 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}") + 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() )), } From d13f81daeb57be446a0946cb6d40dc4b9f69de3d Mon Sep 17 00:00:00 2001 From: Michael Kinney <14325925+aliasfoxkde@users.noreply.github.com> Date: Fri, 6 Nov 2026 06:01:24 -0600 Subject: [PATCH 2/2] fix(ci): defer a trigger until the commit is checkable The trigger endpoint accepted any non-zero new_hash and answered 202, so a relay trigger that reached the service before the mirror held the objects created a run whose workspace clone failed with 'unable to read tree'. That run was persisted as failed: terminal, visible, and useless. The endpoint now probes the repository's bare storage before publishing. Availability is decided the way the workspace clone decides it: the commit must be reachable from refs/heads/* or refs/tags/*, because the clone fetches those and nothing else. A commit present in the object store but parked under another namespace is still unreachable at checkout. An undecidable probe (no durable store, unregistered repository, git failure) is not treated as unavailable: the probe decides the one failure mode it can decide and must not silence a trigger that would have worked. A trigger for an unavailable commit is recorded durably and answered 202 with 'deferred': true. The durable sweep probes again and publishes once the objects arrive; the mirror's own trigger for the same commit dedups onto the same row, so exactly one run is created. The deferral is bounded at 30 minutes, after which the row fails visibly instead of waiting forever. Regression: test_trigger_for_an_absent_commit_is_deferred_and_creates_no_run asserts the 202/deferred contract, the pending durable row, and that no pipeline run exists for the absent commit. Co-Authored-By: Claude Code --- services/ci/src/main.rs | 11 ++-- services/ci/tests/ci_trigger_flow.rs | 75 ++++++++++++++++++++++++++++ 2 files changed, 80 insertions(+), 6 deletions(-) diff --git a/services/ci/src/main.rs b/services/ci/src/main.rs index 3d9a71ec..adc570aa 100644 --- a/services/ci/src/main.rs +++ b/services/ci/src/main.rs @@ -532,7 +532,7 @@ async fn main() -> anyhow::Result<()> { // trigger for a push the mirror has not received yet // waits here until the objects arrive. if let Some(state) = - probe_trigger_commit(pool, request.repo_id, &request.new_hash).await + probe_trigger_commit(Some(pool), request.repo_id, &request.new_hash).await { if !state.is_available() { if defer_budget_exhausted(&request) { @@ -1148,8 +1148,7 @@ async fn trigger_pipeline( "repo_id": repo_id.to_string(), "new_hash": request.new_hash, })), - ) - .into_response(); + ); } let mut event = EventEnvelope::new( @@ -2219,9 +2218,9 @@ pub(crate) async fn run_git_output( ) -> anyhow::Result { command.kill_on_drop(true); match timeout(budget, command.output()).await { - Ok(output) => output.map_err(|error| { - anyhow::anyhow!("git 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!( "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; +}