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.
This commit is contained in:
beardthelion committed 2026-11-12 10:54:37 -06:00
1 parent 41bbe70853
commit 48e88972f0
5 files changed
+132 -29

No files matched your search

+24 -1
View File
@@ -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),
+41 -17
View File
@@ -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::<Vec<_>>()
.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<Vec<RefUpdate>> {
)));
}
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,
+5 -1
View File
@@ -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"
}
}
+1 -1
View File
@@ -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;
+61 -9
View File
@@ -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<ClaimInsert> {
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<ClaimInsert> {
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::<sqlx::Error>() {
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