From 48e88972f02e76a4ac2014ba015d2ae2ab1d0959 Mon Sep 17 00:00:00 2001 From: beardthelion <56458543+beardthelion@users.noreply.github.com> Date: Thu, 12 Nov 2026 10:54:37 -0600 Subject: [PATCH] fix(node): harden status surface writers and close-path head freeze Review-driven fixes on the merged surface: - close_pr freezes the source head it decided against, resolving the recorded push for the branch (then disk) when head_commit is unset; a PR closed before its first status read no longer reports head_resolved=false forever. - insert_status_claim_capped bounds its advisory-lock wait with SET LOCAL lock_timeout and retries on transient classes, matching the push-events writer's contract. - record_push_events retries only transient failures (the inner bound already retries lock waits; a permanent error's second run is a guaranteed re-failure) and update_open_pr_heads retries once on transient and escalates to error naming the lost branch@sha rows. - parse_ref_updates lowercases old/new SHAs so pushed values match the claim side's lowercase normalization. - combined_state folds unrecognized stored states to failure, never silently to success. --- crates/gitlawb-node/src/api/pulls.rs | 25 +++++++- crates/gitlawb-node/src/api/repos.rs | 58 ++++++++++++----- crates/gitlawb-node/src/api/status/mod.rs | 6 +- crates/gitlawb-node/src/api/status/tests.rs | 2 +- crates/gitlawb-node/src/db/mod.rs | 70 ++++++++++++++++++--- 5 files changed, 132 insertions(+), 29 deletions(-) diff --git a/crates/gitlawb-node/src/api/pulls.rs b/crates/gitlawb-node/src/api/pulls.rs index 15652960..027ce39c 100644 --- a/crates/gitlawb-node/src/api/pulls.rs +++ b/crates/gitlawb-node/src/api/pulls.rs @@ -306,7 +306,30 @@ pub async fn close_pr( )); } - state.db.close_pr(&pr.id).await?; + // Freeze the head this close decided against, the same snapshot merge_pr + // takes on its path: a closed pull request's rollup serves only the stored + // head, so a PR closed before its first status read would report + // head_resolved=false forever. The recorded push for the branch is the + // canonical answer; the disk read covers a push whose event write failed. + // An unresolvable head keeps whatever the push path last stored. + let branch_ref = store::local_branch_ref(&pr.source_branch); + let closed_head = if pr.head_commit.is_some() { + None + } else { + state + .db + .latest_push_sha_for_ref(&record.id, &branch_ref) + .await + .ok() + .flatten() + .or_else(|| { + store::read_ref(std::path::Path::new(&record.disk_path), &branch_ref) + .ok() + .flatten() + }) + .map(|sha| sha.to_ascii_lowercase()) + }; + state.db.close_pr(&pr.id, closed_head.as_deref()).await?; webhooks::fire_event( std::sync::Arc::clone(&state.db), diff --git a/crates/gitlawb-node/src/api/repos.rs b/crates/gitlawb-node/src/api/repos.rs index 91c54c91..0dcaf23f 100644 --- a/crates/gitlawb-node/src/api/repos.rs +++ b/crates/gitlawb-node/src/api/repos.rs @@ -3745,12 +3745,34 @@ async fn update_open_pr_heads(db: &crate::db::Db, repo_id: &str, ref_updates: &[ // Chunked for the same reason the push events are: bounded per statement, // and every branch git accepted still gets its head moved. for chunk in heads.chunks(PUSH_WRITE_CHUNK) { - if let Err(e) = db.set_open_pr_heads_batch(repo_id, chunk).await { - tracing::warn!( + let mut outcome = db.set_open_pr_heads_batch(repo_id, chunk).await; + if let Err(first) = &outcome { + if crate::db::is_transient_db_write(first) { + tracing::warn!( + err = %first, + repo_id = %repo_id, + branches = chunk.len(), + "stored-head update hit a transient failure; retrying once" + ); + outcome = db.set_open_pr_heads_batch(repo_id, chunk).await; + } + } + if let Err(e) = outcome { + // Error, not a warning, and it names the rows: a stale head is not + // self-healing — `set_pr_head_if_absent` only fills NULL heads — so + // the missed update is permanent until the next push to that + // branch, and the SHAs are what make it repairable by hand. + let lost = chunk + .iter() + .map(|(branch, sha)| format!("{branch}@{sha}")) + .collect::>() + .join(" "); + tracing::error!( err = %e, repo_id = %repo_id, branches = chunk.len(), - "failed to update stored pull request heads for a pushed branch" + lost = %lost, + "stored pull request heads lost; rollups will serve a stale tip" ); } } @@ -3794,17 +3816,19 @@ pub(crate) async fn record_push_events( for chunk in events.chunks(PUSH_WRITE_CHUNK) { let mut outcome = db.insert_repo_push_events(chunk).await; if let Err(first) = &outcome { - // One retry, because the write now waits on a per-repo lock and the - // failures that wait produces are transient by construction: a lock - // wait that expired, or a pool acquire that timed out behind the - // queue. The retry is bounded because this runs inline on the push. - tracing::warn!( - err = %first, - repo_id = %repo_id, - refs = chunk.len(), - "recording push events failed; retrying once" - ); - outcome = db.insert_repo_push_events(chunk).await; + // One retry, only for the transient classes the write itself can + // outlive: a lock wait that expired, or a pool acquire that timed + // out behind the queue. A permanent error's second run is a + // guaranteed re-failure that only adds latency to the push. + if crate::db::is_transient_db_write(first) { + tracing::warn!( + err = %first, + repo_id = %repo_id, + refs = chunk.len(), + "recording push events hit a transient failure; retrying once" + ); + outcome = db.insert_repo_push_events(chunk).await; + } } if let Err(e) = outcome { // Error, not a warning, and it names the rows. receive-pack has @@ -5609,8 +5633,8 @@ fn parse_ref_updates(body: &[u8]) -> Result> { ))); } let candidate = RefUpdate { - old_sha: parts[0].to_string(), - new_sha: parts[1].to_string(), + old_sha: parts[0].to_ascii_lowercase(), + new_sha: parts[1].to_ascii_lowercase(), ref_name: parts[2].to_string(), }; if seen.insert(( @@ -8486,7 +8510,7 @@ mod tests { .await .unwrap(); let pr = db.get_pr("repo-1", 1).await.unwrap().unwrap(); - db.close_pr(&pr.id).await.unwrap(); + db.close_pr(&pr.id, None).await.unwrap(); update_open_pr_heads( &db, diff --git a/crates/gitlawb-node/src/api/status/mod.rs b/crates/gitlawb-node/src/api/status/mod.rs index f1141f66..5fe6e0d5 100644 --- a/crates/gitlawb-node/src/api/status/mod.rs +++ b/crates/gitlawb-node/src/api/status/mod.rs @@ -572,8 +572,12 @@ fn combined_state(statuses: &[StatusEntry]) -> &'static str { "failure" } else if statuses.iter().any(|s| s.state == "pending") { "pending" - } else { + } else if statuses.iter().all(|s| s.state == "success") { "success" + } else { + // A stored state outside the vocabulary is not a pass: an + // unrecognized verdict folds toward failure, never silently to success. + "failure" } } diff --git a/crates/gitlawb-node/src/api/status/tests.rs b/crates/gitlawb-node/src/api/status/tests.rs index 8e229107..ada29e57 100644 --- a/crates/gitlawb-node/src/api/status/tests.rs +++ b/crates/gitlawb-node/src/api/status/tests.rs @@ -2368,7 +2368,7 @@ async fn closed_pr_with_no_stored_head_does_not_resolve_the_branch(pool: PgPool) state.db.create_repo(&repo).await.unwrap(); let pr = seed_pr(&repo.id, 1, "feature"); state.db.create_pr(&pr).await.unwrap(); - state.db.close_pr(&pr.id).await.unwrap(); + state.db.close_pr(&pr.id, None).await.unwrap(); // The branch is alive and would resolve if the fallback ran. seed_branch_head(&state, &repo, "feature", SHA_B).await; diff --git a/crates/gitlawb-node/src/db/mod.rs b/crates/gitlawb-node/src/db/mod.rs index e0963848..cf65ef27 100644 --- a/crates/gitlawb-node/src/db/mod.rs +++ b/crates/gitlawb-node/src/db/mod.rs @@ -391,7 +391,8 @@ pub struct Webhook { #[derive(Debug, Clone, Serialize, Deserialize)] pub struct RepoPushEvent { pub id: String, - /// Database-assigned ordering key, and the poll cursor itself. Ignored on + /// Database-assigned ordering key the poll cursor's row id resolves to. + /// Ignored on /// insert (see [`Db::insert_repo_push_events`]): nothing the writer supplies /// can order these rows, because `created_at` is stamped before the insert /// and a row stamped later can commit earlier. @@ -4125,13 +4126,22 @@ impl Db { Ok(()) } - pub async fn close_pr(&self, pr_id: &str) -> Result<()> { + /// `head_sha` freezes the source head the close decided against: a closed + /// pull request's rollup serves only the stored head, so nothing resolved + /// later can stand in for it. A `None` keeps whatever the push path last + /// recorded — the same fall-back-to-stored trade `merge_pr` makes when the + /// ref cannot be resolved. + pub async fn close_pr(&self, pr_id: &str, head_sha: Option<&str>) -> Result<()> { let now = Utc::now().to_rfc3339(); - sqlx::query("UPDATE pull_requests SET status='closed', updated_at=$1 WHERE id=$2") - .bind(&now) - .bind(pr_id) - .execute(&self.pool) - .await?; + sqlx::query( + "UPDATE pull_requests SET status='closed', updated_at=$1, \ + head_commit=COALESCE($3, head_commit) WHERE id=$2", + ) + .bind(&now) + .bind(pr_id) + .bind(head_sha) + .execute(&self.pool) + .await?; Ok(()) } @@ -4452,9 +4462,41 @@ impl Db { &self, claim: &StatusClaim, caps: &ClaimCaps, + ) -> Result { + let mut attempt = 1; + loop { + match self.insert_status_claim_capped_attempt(claim, caps).await { + Ok(outcome) => return Ok(outcome), + Err(e) if attempt < PUSH_LOCK_ATTEMPTS && is_transient_db_write(&e) => { + tracing::debug!( + attempt, + repo = %claim.repo_id, + "status-claim write timed out waiting for the repo lock; retrying" + ); + attempt += 1; + } + Err(e) => return Err(e), + } + } + } + + /// One attempt at the capped insert: bound every lock wait in the + /// transaction, take the repo advisory lock, then run the cap checks and + /// the insert. `SET LOCAL lock_timeout` is what turns a wedged lock holder + /// from an unbounded wait into a retryable error. + async fn insert_status_claim_capped_attempt( + &self, + claim: &StatusClaim, + caps: &ClaimCaps, ) -> Result { let mut tx = self.pool.begin().await?; + sqlx::query(&format!( + "SET LOCAL lock_timeout = '{PUSH_LOCK_TIMEOUT_MS}ms'" + )) + .execute(&mut *tx) + .await?; + sqlx::query("SELECT pg_advisory_xact_lock(hashtext($1)::bigint)") .bind(&claim.repo_id) .execute(&mut *tx) @@ -8756,6 +8798,16 @@ fn is_lock_timeout(err: &sqlx::Error) -> bool { .is_some_and(|code| code == LOCK_NOT_AVAILABLE) } +/// The two transient classes a bounded inline write may retry: a lock wait the +/// timeout cancelled (`lock_not_available`), and a pool acquire that timed out +/// behind the queue. Every other error is permanent for this attempt. +pub(crate) fn is_transient_db_write(err: &anyhow::Error) -> bool { + match err.downcast_ref::() { + Some(e) => is_lock_timeout(e) || matches!(e, sqlx::Error::PoolTimedOut), + None => false, + } +} + impl Db { /// Record one local push event. Idempotent on the row id. /// @@ -12243,7 +12295,7 @@ mod migration_tests { let closed = sample_pr("repo-1", 3); db.create_pr(&closed).await.unwrap(); set_head_directly(&db, &closed.id, SHA_A).await; - db.close_pr(&closed.id).await.unwrap(); + db.close_pr(&closed.id, None).await.unwrap(); // #4 merged on `feature`, frozen at SHA_A. let merged = sample_pr("repo-1", 4); db.create_pr(&merged).await.unwrap(); @@ -12377,7 +12429,7 @@ mod migration_tests { // Absent head but closed: refused. let closed = sample_pr("repo-1", 2); db.create_pr(&closed).await.unwrap(); - db.close_pr(&closed.id).await.unwrap(); + db.close_pr(&closed.id, None).await.unwrap(); assert_eq!( db.set_pr_head_if_absent(&closed.id, SHA_A).await.unwrap(), 0