fix(node): merge rollback, honest create rollback, purge row recheck

- merge_pr snapshots the pre-merge target tip and uses
  release_compensating to rewind it via bounded update-ref on a
  refused publish, matching the close_issue pattern; regression proves
  the ref returns to its prior OID after a refused upload (P1)
- create_repo rollback names residual state instead of claiming a full
  rollback when the row delete fails or dir removal fails (P2)
- purge execute re-checks row absence via get_repo_by_id immediately
  before remove_dir_all and refuses the disk delete when a row
  reappeared after commit, closing the post-commit mirror-ingest
  window (P2)
- sweep_evicts_expired_keys_from_every_limiter now seeds and overrides
  the write_rate_limiter TTL (P3)
This commit is contained in:
beardthelion committed 2026-11-11 18:45:01 -06:00
1 parent 1187dcab79
commit 261f7ab3c6
4 files changed
+205 -10

No files matched your search

+23 -1
View File
@@ -533,7 +533,29 @@ pub async fn run_purge_spam(
summary.disk_failed += 1;
}
Ok(p) => {
if let Err(e) = std::fs::remove_dir_all(p) {
// Post-commit window: delete_repo_by_id's slug lock ends
// at commit, so a concurrent mirror ingest can insert a
// new row for this slug before remove_dir_all runs.
// Re-check immediately before the destructive op and
// abort the disk delete when a fresh row landed rather
// than wipe disk under a live row. A probe error fails
// closed, the same as the lock-phase checks.
let row_gone = match db.get_repo_by_id(&c.id).await {
Ok(None) => true,
Ok(Some(_)) => {
warn!(repo = %c.id,
"purge-spam: a repo row reappeared after delete — refusing to remove disk under a live row");
false
}
Err(e) => {
warn!(repo = %c.id, err = %e,
"purge-spam: could not re-check row absence — refusing to remove (fail-closed)");
false
}
};
if !row_gone {
summary.disk_failed += 1;
} else if let Err(e) = std::fs::remove_dir_all(p) {
warn!(repo = %c.id, path = %p.display(), err = %e,
"purge-spam: deleted DB row but could not remove on-disk repo dir");
summary.disk_failed += 1;
+157 -1
View File
@@ -226,6 +226,18 @@ pub async fn merge_pr(
let disk_path = guard.path().to_path_buf();
let merger_did = auth.0;
// Snapshot the target ref's pre-merge OID so a refused durable upload can
// rewind the local merge, mirroring the close_issue / create_issue
// compensating pattern. A read failure or an absent ref (a first-commit
// merge onto an unborn branch) yields no rollback closure: the merge still
// applies and a refused publish still surfaces as 5xx.
let target_ref = format!("refs/heads/{}", pr.target_branch);
let git_timeout = std::time::Duration::from_secs(state.config.git_service_timeout_secs);
let prior_tip = store::read_ref_bounded(&state.git_bin, &disk_path, &target_ref, git_timeout)
.await
.ok()
.flatten();
let merge_result = store::merge_branch(
&disk_path,
&pr.target_branch,
@@ -238,7 +250,27 @@ pub async fn merge_pr(
// Short-circuit on a refused publish before the PR is marked merged and
// before the webhook fires. Both are irreversible announcements of a merge
// commit that only exists on this node's disk.
guard.release(merge_result.is_ok()).await.into_result()?;
let release_result = match (merge_result.is_ok(), prior_tip) {
(true, Some(tip)) => {
let git_bin = state.git_bin.clone();
let deadline = std::time::Instant::now() + git_timeout;
guard
.release_compensating(true, move |path| {
crate::git::visibility_pack::run_bounded_git(
&git_bin,
&["update-ref", &target_ref, &tip],
path,
b"",
deadline,
)
.map(|_| ())
})
.await
.into_result()
}
_ => guard.release(merge_result.is_ok()).await.into_result(),
};
release_result?;
let merge_sha = merge_result.map_err(|e| AppError::Git(e.to_string()))?;
@@ -438,6 +470,130 @@ pub async fn list_comments(
Ok(Json(serde_json::json!({ "comments": comments })))
}
#[cfg(test)]
mod merge_rollback_tests {
use super::*;
use axum::response::IntoResponse;
use sqlx::PgPool;
fn git(dir: &std::path::Path, args: &[&str]) -> String {
let out = std::process::Command::new("git")
.args(["-C", &*dir.to_string_lossy()])
.args(args)
.output()
.unwrap();
assert!(
out.status.success(),
"git {args:?}: {}",
String::from_utf8_lossy(&out.stderr)
);
String::from_utf8_lossy(&out.stdout).trim().to_string()
}
/// Build a bare repo whose `main` and `feature` refs diverge: one root
/// commit on main, one commit on feature, so merge_branch creates a real
/// merge commit that a refused publish must rewind.
fn seeded_merge_repo(dir: &std::path::Path) -> (String, String) {
crate::git::store::init_bare(dir).expect("bare repo");
let empty_tree = git(dir, &["mktree"]);
let root = git(dir, &["commit-tree", &empty_tree, "-m", "root"]);
git(dir, &["update-ref", "refs/heads/main", &root]);
let tip2 = git(
dir,
&["commit-tree", &empty_tree, "-m", "feature", "-p", &root],
);
git(dir, &["update-ref", "refs/heads/feature", &tip2]);
(root, tip2)
}
/// On a definite publish refusal, the local merge commit must be rewound:
/// the client got a 5xx, so the repo cannot show a merge the mirror never
/// learned about (same contract create_issue's rollback enforces).
#[sqlx::test]
async fn merge_pr_rewinds_local_merge_when_publish_refuses(pool: PgPool) {
// Endpoint that answers HEAD with an etag but refuses every write.
let app = axum::Router::new().route(
"/{*key}",
axum::routing::any(|method: axum::http::Method| async move {
if method == axum::http::Method::HEAD {
let mut resp = axum::http::StatusCode::OK.into_response();
resp.headers_mut()
.insert("etag", axum::http::HeaderValue::from_static("\"gen-1\""));
resp
} else {
axum::http::StatusCode::INTERNAL_SERVER_ERROR.into_response()
}
}),
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let endpoint = format!("http://127.0.0.1:{}", listener.local_addr().unwrap().port());
tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
let owner = "did:key:zMERGEROLLBACKAAAAAAAAAAAAAAAAAAAAAAAA";
let repo_name = "rollback-merge";
let mut state = crate::test_support::test_state(pool.clone()).await;
let disk = tempfile::TempDir::new().unwrap();
let repo_path = disk.path().join("repo.git");
let (main_tip, _feature_tip) = seeded_merge_repo(&repo_path);
let mut seeded = crate::db::RepoRecord {
id: Uuid::new_v4().to_string(),
name: repo_name.to_string(),
owner_did: owner.to_string(),
description: None,
is_public: true,
default_branch: "main".to_string(),
created_at: Utc::now(),
updated_at: Utc::now(),
disk_path: String::new(),
forked_from: None,
machine_id: None,
};
seeded.disk_path = repo_path.to_string_lossy().to_string();
let repo_id = seeded.id.clone();
state.db.create_repo(&seeded).await.expect("seed repo");
let now = Utc::now().to_rfc3339();
state
.db
.create_pr(&PullRequest {
id: Uuid::new_v4().to_string(),
repo_id,
number: 1,
title: "rb".to_string(),
body: None,
author_did: owner.to_string(),
source_branch: "feature".to_string(),
target_branch: "main".to_string(),
status: "open".to_string(),
merged_by_did: None,
merged_at: None,
created_at: now.clone(),
updated_at: now,
})
.await
.expect("seed PR");
state.repo_store = crate::git::repo_store::RepoStore::for_testing_with_tigris(
disk.path().to_path_buf(),
crate::git::repo_store::build_lock_pool(&pool, 2, std::time::Duration::from_secs(1)),
crate::git::tigris::TigrisClient::for_testing_with_endpoint("test-bucket", &endpoint),
);
let res = merge_pr(
State(state),
Extension(AuthenticatedDid(owner.to_string())),
Path((owner.to_string(), repo_name.to_string(), 1)),
)
.await;
assert!(res.is_err(), "a refused publish must surface an error");
let tip_after = git(&repo_path, &["rev-parse", "refs/heads/main"]);
assert_eq!(
tip_after, main_tip,
"a refused publish must rewind the local merge to the pre-merge tip"
);
}
}
/// #173 F1 follow-up: merge holds no admission permit, so an exhausted write-lock
/// pool must shed 503 + Retry-After.
#[cfg(test)]
+23 -8
View File
@@ -373,15 +373,30 @@ pub async fn create_repo(
if let Err(e) = landed {
tracing::warn!(repo = %req.name, err = %e,
"create: initial archive upload failed — rolling back row and dir");
if let Err(dbe) = state.db.delete_repo_by_id(&record.id).await {
tracing::warn!(repo = %req.name, err = %dbe,
"create rollback: could not delete the repo row");
}
let _ = std::fs::remove_dir_all(&disk_path);
let row_deleted = match state.db.delete_repo_by_id(&record.id).await {
Ok(_) => true,
Err(dbe) => {
tracing::warn!(repo = %req.name, err = %dbe,
"create rollback: could not delete the repo row");
false
}
};
let dir_removed = std::fs::remove_dir_all(&disk_path).is_ok();
repo_lock.release().await;
return Err(AppError::Git(format!(
"repo created but initial archive upload failed and was rolled back: {e}"
)));
let residual = match (row_deleted, dir_removed) {
(true, true) => None,
(false, true) => Some("the repo row remains"),
(true, false) => Some("the on-disk directory remains"),
(false, false) => Some("the repo row and on-disk directory both remain"),
};
return Err(match residual {
None => AppError::Git(format!(
"initial archive upload failed; the repo was rolled back: {e}"
)),
Some(what) => AppError::Git(format!(
"initial archive upload failed and rollback is incomplete: {what}; operator cleanup required: {e}"
)),
});
}
}
+2
View File
@@ -1595,6 +1595,7 @@ mod rate_limiter_sweep_tests {
let mut state = crate::test_support::test_state_lazy();
state.rate_limiter = RateLimiter::new(10, window);
state.create_ip_rate_limiter = RateLimiter::new(10, window);
state.write_rate_limiter = RateLimiter::new(10, window);
state.push_rate_limiter = RateLimiter::new(10, window);
state.close_issue_rate_limiter = RateLimiter::new(10, window);
state.sync_trigger_rate_limiter = RateLimiter::new(10, window);
@@ -1608,6 +1609,7 @@ mod rate_limiter_sweep_tests {
[
s.rate_limiter.clone(),
s.create_ip_rate_limiter.clone(),
s.write_rate_limiter.clone(),
s.push_rate_limiter.clone(),
s.close_issue_rate_limiter.clone(),
s.sync_trigger_rate_limiter.clone(),