fix(ci): defer a trigger until the commit is checkable

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 <sha>` 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<CommitAvailability> {
+    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);
+    }
+}

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<gitforge_common::RepoId, PipelineDefinition>;
 type PipelineRegistry = HashMap<gitforge_common::PipelineRunId, Arc<CiEngine>>;
 
@@ -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<dyn EventBus>,
     /// 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<commit_probe::CommitAvailability> {
+    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<Arc<TriggerState>>,
     Json(request): Json<PipelineTriggerRequest>,
@@ -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<String>) -> 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<std::process::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}")
-        }),
+        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()
         )),
     }