diff --git a/ROADMAP.md b/ROADMAP.md index edaac0eb..83e86c9c 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -707,7 +707,7 @@ and reconciler, collaborators, branches) and F2b (smart-HTTP git transport via ` `pre-receive` ref policy, admission control, threat-model pass #1) and F3a (content reads: refs, tags, commits, trees, contents, archives) and F3b (contents writes: create, update, delete as server-made commits) and F4a (branch protection enforced on push, contents API and branch delete; commit statuses) and F4b (pull requests, reviews, comments, retention -refs) are built. Hosted repos are +refs) and F4c (merge with a crash-safe journal, required-check gating, kill -9 and protection-matrix tests) are built. Hosted repos are bare Git and get none of the `.sc/` features. Other follow-up work is tracked as GitHub issues and listed under Deferred diff --git a/crates/forge/conformance/departures.json b/crates/forge/conformance/departures.json index 22d1d980..d6739fc5 100644 --- a/crates/forge/conformance/departures.json +++ b/crates/forge/conformance/departures.json @@ -1004,10 +1004,12 @@ "match": "GET /api/v1/repos/acme/life/pulls/", "expect": { "ignore_fields": [ - "head.sha" + "head.sha", + "additions", + "changed_files" ] }, - "reason": "head.sha is the frozen merge-time sha in sc-forge; the notes below pin that" + "reason": "head.sha is the frozen merge-time sha in sc-forge; the notes below pin that; additions and changed_files are those frozen at merge (Gitea follows the recreated branch)" }, { "scenario": "O", @@ -1525,5 +1527,47 @@ ] }, "reason": "sc-forge's hook words the refusal 'refusing to delete the default branch'; Gitea says 'gitea: branch main is the default branch and cannot be deleted'. Both end in 'pre-receive hook declined'" + }, + { + "scenario": "Q", + "exchange": 11, + "kind": "git", + "match": "force-protected-push-enabled", + "expect": { + "ok": false, + "message_contains": [ + "force-push", + "pre-receive hook declined" + ] + }, + "reason": "sc-forge's hook words the refusal 'refusing to force-push protected branch '; Gitea says 'gitea: branch main is protected from force push'. Both end in 'pre-receive hook declined'" + }, + { + "scenario": "Q", + "exchange": 12, + "kind": "git", + "match": "delete-protected-push-enabled", + "expect": { + "ok": false, + "message_contains": [ + "default branch", + "pre-receive hook declined" + ] + }, + "reason": "sc-forge's hook words the refusal 'refusing to delete the default branch'; Gitea says 'gitea: branch main is the default branch and cannot be deleted'. Both end in 'pre-receive hook declined'" + }, + { + "scenario": "Q", + "exchange": 14, + "kind": "git", + "match": "delete-protected-push-disabled", + "expect": { + "ok": false, + "message_contains": [ + "default branch", + "pre-receive hook declined" + ] + }, + "reason": "sc-forge's hook words the refusal 'refusing to delete the default branch'; Gitea says 'gitea: branch main is the default branch and cannot be deleted'. Both end in 'pre-receive hook declined'" } ] diff --git a/crates/forge/conformance/forge-check.sh b/crates/forge/conformance/forge-check.sh index fbc54aae..22b26501 100755 --- a/crates/forge/conformance/forge-check.sh +++ b/crates/forge/conformance/forge-check.sh @@ -25,4 +25,4 @@ for _ in $(seq 50); do done python3 -I "$here/run.py" check --base-url "http://127.0.0.1:$port" --admin-token "$token" \ --fixtures "$here/fixtures/1.22.6" --departures "$here/departures.json" --pending "$here/slice-pending.json" \ - --only A,B,C,D,E,F,H,I,J,K,L,M,N,O,P,Q,R,V,W "$@" || { echo "--- server log"; cat "$data/serve.log"; exit 1; } + --only A,B,C,D,E,F,G,H,I,J,K,L,M,N,O,P,Q,R,T,U,V,W "$@" || { echo "--- server log"; cat "$data/serve.log"; exit 1; } diff --git a/crates/forge/conformance/slice-pending.json b/crates/forge/conformance/slice-pending.json index c33d16be..2e73343f 100644 --- a/crates/forge/conformance/slice-pending.json +++ b/crates/forge/conformance/slice-pending.json @@ -10,57 +10,5 @@ "GET /api/v1/orgs/acme/repos?limit=50" ] ] - }, - "Q": { - "stop": [ - 8, - "origin/main~1" - ], - "defer": [ - [ - 5, - "POST /api/v1/repos/acme/bypass/pulls//merge" - ] - ] - }, - "O": { - "defer": [ - [ - 9, - "POST /api/v1/repos/acme/life/pulls//merge" - ], - [ - 19, - "POST /api/v1/repos/acme/life/pulls//merge" - ], - [ - 20, - "GET /api/v1/repos/acme/life/pulls/" - ], - [ - 21, - "merged_pr_head_sha_frozen" - ], - [ - 22, - "merged_pr_head_sha_follows_new_branch" - ], - [ - 23, - "PATCH /api/v1/repos/acme/life/pulls/" - ], - [ - 25, - "GET /api/v1/repos/acme/life/pulls/" - ], - [ - 26, - "PATCH /api/v1/repos/acme/life/pulls/" - ], - [ - 27, - "GET /api/v1/repos/acme/life/pulls/" - ] - ] } } diff --git a/crates/forge/src/cgi.rs b/crates/forge/src/cgi.rs index 2a992d8e..918eaf83 100644 --- a/crates/forge/src/cgi.rs +++ b/crates/forge/src/cgi.rs @@ -811,9 +811,6 @@ impl Supervise { let status = child.wait(); let ok = reason == "exited" && status.as_ref().is_ok_and(|s| s.success()); let _ = verdict.send(ok); - // The child is gone: release the transport registration, so a repository delete waiting - // for it is not also waiting on the sync below (which takes the shared lifecycle lock). - drop(reg); let err = tail.and_then(|t| t.join().ok()).unwrap_or_default(); if !ok { eprintln!( @@ -824,6 +821,10 @@ impl Supervise { } else if receive && marker.exists() { if let Ok(conn) = crate::db::connect(&app.data_dir) { let _ = repos::touch(&conn, repo_id, crate::timefmt::now()); + // The child is gone and the stamp is written: release the transport registration, so + // a repository delete waiting for it is not also waiting on the sync below (which + // takes the shared lifecycle lock). + drop(reg); crate::pulls::sync(&app.life, &conn, repo_id, crate::timefmt::now()); } } diff --git a/crates/forge/src/cli.rs b/crates/forge/src/cli.rs index 9a7a3592..07ebf007 100644 --- a/crates/forge/src/cli.rs +++ b/crates/forge/src/cli.rs @@ -308,6 +308,9 @@ fn run_serve(a: ServeArgs) -> Result<()> { &std::env::current_exe().context("cannot find the sc-forge binary")?, )?; let life = scl_forge::lifecycle::Lifecycle::new(&a.data_dir, git)?; + // A killed predecessor may have left git children running and their ref locks behind; they + // go before anything below looks at a repository (we hold the data-directory lock). + scl_forge::orphans::sweep(&std::fs::canonicalize(&a.data_dir)?, Some(life.git.bin())); // Resolve whatever a crash left, before the listener is bound. let report = life.reconcile(&db::connect(&a.data_dir)?, scl_forge::timefmt::now())?; if report != scl_forge::lifecycle::Report::default() { diff --git a/crates/forge/src/content_write.rs b/crates/forge/src/content_write.rs index 00a9a007..c21e5da8 100644 --- a/crates/forge/src/content_write.rs +++ b/crates/forge/src/content_write.rs @@ -186,7 +186,7 @@ pub fn write_contents(ctx: &mut Ctx) -> Resp { /// The address git will record: surrounding blanks are stripped by git, so strip them here too and the /// response matches the commit. -fn commit_email(user: &crate::store::User) -> String { +pub(crate) fn commit_email(user: &crate::store::User) -> String { let e = user.email.trim(); if e.is_empty() { format!("{}@noreply.localhost", user.login) diff --git a/crates/forge/src/db.rs b/crates/forge/src/db.rs index e0a1eab8..a64b7042 100644 --- a/crates/forge/src/db.rs +++ b/crates/forge/src/db.rs @@ -163,6 +163,33 @@ const MIGRATIONS: &[&str] = &[ updated_unix INTEGER NOT NULL ); CREATE INDEX issue_comments_by_issue ON issue_comments (repo_id, number, id);", + // 5: the merge journal. A row is written before the base branch moves; `refs/forge/merge//landed` + // is written right after, and a crash is resolved from the marker or, failing that, from the merge + // commit being part of the base's history (see `merges.rs`). `recovered` marks a row finished by + // recovery rather than by the request that started it. + // The first statement drops cached PR statistics computed before unrelated histories were + // compared against the base tip (they are derived data and are recomputed on demand). + "DELETE FROM mergeability_cache; + CREATE TABLE merge_intents ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + repo_id INTEGER NOT NULL REFERENCES repos(id) ON DELETE CASCADE, + pull_number INTEGER NOT NULL, + base_ref TEXT NOT NULL, + head_ref TEXT NOT NULL, + base_oid TEXT NOT NULL, + head_oid TEXT NOT NULL, + merge_oid TEXT NOT NULL, + merge_base TEXT NOT NULL DEFAULT '', + changed_files INTEGER NOT NULL, + additions INTEGER NOT NULL, + deletions INTEGER NOT NULL, + merger_id INTEGER NOT NULL REFERENCES users(id), + merged_unix INTEGER NOT NULL, + recovered INTEGER NOT NULL DEFAULT 0, + state TEXT NOT NULL CHECK (state IN ('pending', 'done', 'aborted')), + created_unix INTEGER NOT NULL + ); + CREATE UNIQUE INDEX merge_intents_one_pending ON merge_intents (repo_id, pull_number) WHERE state = 'pending';", ]; /// Every table the current schema creates; `/healthz` probes each, so a missing or unreadable one @@ -179,6 +206,7 @@ pub const TABLES: &[&str] = &[ "mergeability_cache", "reviews", "issue_comments", + "merge_intents", ]; /// Database failures a caller reports to the operator. diff --git a/crates/forge/src/lib.rs b/crates/forge/src/lib.rs index 69c71ff5..c9019083 100644 --- a/crates/forge/src/lib.rs +++ b/crates/forge/src/lib.rs @@ -23,7 +23,9 @@ pub mod hook; pub mod http; pub mod lifecycle; pub mod limits; +pub mod merges; pub mod names; +pub mod orphans; pub mod policy; pub mod protection; pub mod pulls; diff --git a/crates/forge/src/merges.rs b/crates/forge/src/merges.rs new file mode 100644 index 00000000..37ec290d --- /dev/null +++ b/crates/forge/src/merges.rs @@ -0,0 +1,814 @@ +//! Merging a pull request: `POST /pulls/{n}/merge`, the merge journal and its crash recovery. +//! +//! Git and SQLite cannot commit together, so a merge is journaled. Under the repository's write +//! gate the handler (1) settles any pending intent, (2) re-reads the PR, the protection rules and +//! the required statuses, (3) computes the merge commit with `merge-tree --write-tree` and +//! `commit-tree`, (4) writes a `pending` intent row, (5) runs ONE `update-ref --stdin` transaction +//! that `verify`s the head and moves the base branch compare-and-swap, (6) records the landing in +//! `refs/forge/merge//landed`, and (7) marks the intent done and the PR merged in one +//! SQLite transaction. +//! +//! Git's files backend commits several refs one after another, so no single transaction can both +//! move the base and create a marker crash-atomically. The protocol is therefore built from steps +//! that are each atomic and from evidence that survives a crash between them. A pending intent +//! has landed when (a) its `landed` marker exists (definitive, even if the base was rewritten +//! afterwards), or (b) its merge commit is an ancestor of the base branch (the crash fell between +//! (5) and (6); at startup nobody can have rewritten the base yet). Otherwise it did not land, and +//! is aborted unless a lock file shows a git writer still holding the base ref (the intent then +//! stays pending until that writer is gone). An unreadable answer never aborts: the intent stays +//! pending. + +use crate::api::{body, str_field}; +use crate::gitproc::{Git, GitError, Identity, Spec}; +use crate::gitwrite as gw; +use crate::http::{error, Resp}; +use crate::pulls::{self, ComputeError}; +use crate::pulls_api::{find, gated, missing, needs_write, number, tip}; +use crate::refname; +use crate::repo_api::acc; +use crate::repos::{Access, Perm}; +use crate::routes::Ctx; +use crate::statuses; +use crate::timefmt::now; +use rusqlite::{params, Connection, OptionalExtension, Transaction, TransactionBehavior}; +use std::path::Path; +use std::time::Duration; + +/// Rounds a merge takes when the base branch moves under it, then it answers "try again later". +const ATTEMPTS: usize = 3; +/// Conflicting paths named in a refusal. +const NAMED_PATHS: usize = 10; + +/// The marker ref that says intent `id` landed. +pub fn landed_ref(id: i64) -> String { + format!("refs/forge/merge/{id}/landed") +} + +/// One journal row. +#[derive(Clone, Debug)] +pub struct Intent { + pub id: i64, + pub repo_id: i64, + pub pull_number: i64, + pub base_ref: String, + pub head_ref: String, + pub base_oid: String, + pub head_oid: String, + pub merge_oid: String, + pub merge_base: String, + pub changed_files: i64, + pub additions: i64, + pub deletions: i64, + pub merger_id: i64, + pub merged_unix: i64, +} + +const COLS: &str = + "id, repo_id, pull_number, base_ref, head_ref, base_oid, head_oid, merge_oid, merge_base, \ + changed_files, additions, deletions, merger_id, merged_unix"; + +fn row(r: &rusqlite::Row) -> rusqlite::Result { + Ok(Intent { + id: r.get(0)?, + repo_id: r.get(1)?, + pull_number: r.get(2)?, + base_ref: r.get(3)?, + head_ref: r.get(4)?, + base_oid: r.get(5)?, + head_oid: r.get(6)?, + merge_oid: r.get(7)?, + merge_base: r.get(8)?, + changed_files: r.get(9)?, + additions: r.get(10)?, + deletions: r.get(11)?, + merger_id: r.get(12)?, + merged_unix: r.get(13)?, + }) +} + +/// Pending intents of the repository. +pub fn pending(conn: &Connection, repo_id: i64) -> rusqlite::Result> { + let mut st = conn.prepare(&format!( + "SELECT {COLS} FROM merge_intents WHERE repo_id = ?1 AND state = 'pending' ORDER BY id" + ))?; + let rows = st.query_map([repo_id], row)?.collect(); + rows +} + +/// Is a merge of this repository unfinished? +pub fn has_pending(conn: &Connection, repo_id: i64) -> bool { + conn.query_row( + "SELECT 1 FROM merge_intents WHERE repo_id = ?1 AND state = 'pending' LIMIT 1", + [repo_id], + |_| Ok(()), + ) + .optional() + .map(|r| r.is_some()) + .unwrap_or(false) +} + +fn insert(conn: &Connection, i: &Intent, now: i64) -> rusqlite::Result { + conn.execute( + "INSERT INTO merge_intents (repo_id, pull_number, base_ref, head_ref, base_oid, head_oid, merge_oid, \ + merge_base, changed_files, additions, deletions, merger_id, merged_unix, state, created_unix) \ + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, 'pending', ?14)", + params![ + i.repo_id, + i.pull_number, + i.base_ref, + i.head_ref, + i.base_oid, + i.head_oid, + i.merge_oid, + i.merge_base, + i.changed_files, + i.additions, + i.deletions, + i.merger_id, + i.merged_unix, + now + ], + )?; + Ok(conn.last_insert_rowid()) +} + +/// A retry builds a new merge commit on the new base (and, when the client named no head, from +/// whatever the head is by then), with its own merge base, statistics and time: the journal says +/// all of it, the head included, before the ref moves, so recovery writes exactly what a +/// normal completion would. +fn retarget(conn: &Connection, i: &Intent) -> rusqlite::Result<()> { + conn.execute( + "UPDATE merge_intents SET base_oid = ?1, merge_oid = ?2, merge_base = ?3, changed_files = ?4, \ + additions = ?5, deletions = ?6, merged_unix = ?7, head_oid = ?9 WHERE id = ?8 AND state = 'pending'", + params![ + i.base_oid, + i.merge_oid, + i.merge_base, + i.changed_files, + i.additions, + i.deletions, + i.merged_unix, + i.id, + i.head_oid + ], + )?; + Ok(()) +} + +fn abort(conn: &Connection, id: i64) -> rusqlite::Result<()> { + conn.execute( + "UPDATE merge_intents SET state = 'aborted' WHERE id = ?1 AND state = 'pending'", + [id], + )?; + Ok(()) +} + +/// The merge landed: the PR becomes merged and the intent done, in one transaction. Idempotent: a +/// PR already merged by this very commit is fine. A PR in any other state is an inconsistency and +/// fails the transaction, so the intent (and its evidence) stays pending instead of being consumed. +fn finalize_rows(conn: &Connection, i: &Intent, recovered: bool) -> rusqlite::Result<()> { + let tx = Transaction::new_unchecked(conn, TransactionBehavior::Immediate)?; + let changed = tx.execute( + "UPDATE pulls SET state = 'merged', merge_commit_sha = ?1, final_head_sha = ?2, head_sha = ?2, \ + base_sha = ?3, merged_unix = ?4, closed_unix = ?4, updated_unix = ?4, merged_by = ?5, \ + merge_base = ?6, changed_files = ?7, additions = ?8, deletions = ?9, head_gone = 0 \ + WHERE repo_id = ?10 AND number = ?11 AND state = 'open' AND base_ref = ?12 AND head_ref = ?13", + params![ + i.merge_oid, + i.head_oid, + i.base_oid, + i.merged_unix, + i.merger_id, + i.merge_base, + i.changed_files, + i.additions, + i.deletions, + i.repo_id, + i.pull_number, + i.base_ref, + i.head_ref + ], + )?; + if changed == 0 { + let same: bool = tx + .query_row( + "SELECT 1 FROM pulls WHERE repo_id = ?1 AND number = ?2 AND state = 'merged' AND merge_commit_sha = ?3", + params![i.repo_id, i.pull_number, i.merge_oid], + |_| Ok(()), + ) + .optional()? + .is_some(); + if !same { + return Err(rusqlite::Error::StatementChangedRows(0)); + } + } + tx.execute( + "UPDATE merge_intents SET state = 'done', recovered = ?2 WHERE id = ?1 AND state = 'pending'", + params![i.id, recovered], + )?; + tx.commit() +} + +/// Finish a landed merge: keep the head reachable, mark the rows, drop the marker. The SQLite step +/// is retried a few times while the caller still holds the gate; if it still fails the intent stays +/// pending (its evidence is intact) and the next gated operation or read completes it. +fn finish( + conn: &Connection, + git: &Git, + dir: &Path, + i: &Intent, + recovered: bool, +) -> Result<(), ComputeError> { + if let Err(e) = pulls::retain(git, dir, i.pull_number, &i.head_oid) { + eprintln!( + "sc-forge: could not retain the head of merged pull request {}: {e}", + i.pull_number + ); + } + let mut last = None; + for k in 0..3 { + match finalize_rows(conn, i, recovered) { + Ok(()) => { + last = None; + break; + } + Err(e) => { + last = Some(e); + std::thread::sleep(Duration::from_millis(20 * (k + 1))); + } + } + } + if let Some(e) = last { + return Err(e.into()); + } + let _ = crate::repos::touch(conn, i.repo_id, i.merged_unix); + // The marker has done its job. + let _ = git.run(&Spec { + repo: Some(dir), + args: &["update-ref", "-d", &landed_ref(i.id)], + ..Default::default() + }); + Ok(()) +} + +/// Does the commit exist in the repository? +fn commit_exists(git: &Git, dir: &Path, oid: &str) -> Result { + let out = git.run(&Spec { + repo: Some(dir), + args: &["cat-file", "-e", &format!("{oid}^{{commit}}")], + ..Default::default() + })?; + match out.status.code() { + Some(0) => Ok(true), + Some(1) => Ok(false), + // Anything else is "could not tell", which never aborts an intent. + _ => Err(GitError::Failed( + String::from_utf8_lossy(&out.stderr).trim().to_string(), + )), + } +} + +/// Has the intent's merge landed? The marker says so definitively; failing that, the merge commit +/// being part of the base branch's history does. `Err` is "could not tell", never "no". +fn is_landed(git: &Git, dir: &Path, i: &Intent) -> Result { + if gw::read_ref(git, dir, &landed_ref(i.id))?.is_some() { + return Ok(true); + } + let Some(tip) = gw::read_ref(git, dir, &refname::branch_ref(&i.base_ref))? else { + return Ok(false); + }; + if !commit_exists(git, dir, &i.merge_oid)? { + return Ok(false); + } + gw::is_ancestor(git, dir, &i.merge_oid, &tip) +} + +/// Record that the intent landed. Best effort: ancestry carries the same fact until it is written. +fn mark_landed(git: &Git, dir: &Path, i: &Intent) { + let zero = "0".repeat(i.merge_oid.len()); + let _ = git.run(&Spec { + repo: Some(dir), + args: &["update-ref", &landed_ref(i.id), &i.merge_oid, &zero], + ..Default::default() + }); +} + +/// Is a git writer holding the base ref right now (its lock file exists)? At run time that is a +/// live writer (a push, or our own publish): the intent waits rather than being judged under it. +/// The lock of a *dead* writer is cleared at startup (`orphans::sweep`), not guessed at here. +fn base_is_locked(dir: &Path, base_ref: &str) -> bool { + dir.join("refs") + .join("heads") + .join(format!("{base_ref}.lock")) + .exists() +} + +/// Settle the pending intents of the repository, with the write gate held (or at startup, when +/// nothing else runs). Landed ones are finalized from the journal; ones that did not land are +/// aborted; ones whose outcome cannot be told yet (a git writer holds the base, or git failed) stay +/// pending. Returns how many were finalized. +pub fn recover_locked( + conn: &Connection, + git: &Git, + dir: &Path, + repo_id: i64, +) -> Result { + let mut done = 0; + for i in pending(conn, repo_id)? { + if is_landed(git, dir, &i)? { + mark_landed(git, dir, &i); + finish(conn, git, dir, &i, true)?; + done += 1; + } else if !base_is_locked(dir, &i.base_ref) { + // A writer that committed between the first look and the lock check has also dropped its + // lock: look once more before an abort is final. + if is_landed(git, dir, &i)? { + mark_landed(git, dir, &i); + finish(conn, git, dir, &i, true)?; + done += 1; + } else { + abort(conn, i.id)?; + } + } + } + Ok(done) +} + +/// Remove marker refs that belong to no pending intent (a finished merge whose cleanup was lost). +pub fn drop_stale_receipts( + conn: &Connection, + git: &Git, + dir: &Path, + repo_id: i64, +) -> Result<(), ComputeError> { + let live: Vec = pending(conn, repo_id)? + .iter() + .map(|i| landed_ref(i.id)) + .collect(); + for name in pulls::refs_under(git, dir, "refs/forge/merge")?.keys() { + if !live.contains(name) { + let _ = git.run(&Spec { + repo: Some(dir), + args: &["update-ref", "-d", name], + ..Default::default() + }); + } + } + Ok(()) +} + +// ---- git steps ---- + +/// The result of `merge-tree --write-tree`. +enum Tree { + Clean(String), + Conflict(Vec), +} + +fn merge_tree(git: &Git, dir: &Path, base: &str, head: &str) -> Result { + let out = git.run(&Spec { + repo: Some(dir), + args: &[ + "merge-tree", + "--write-tree", + "--name-only", + "--no-messages", + "-z", + "--end-of-options", + base, + head, + ], + ..Default::default() + })?; + let text = String::from_utf8_lossy(&out.stdout).to_string(); + let mut toks = text.split('\0'); + let tree = toks.next().unwrap_or("").trim().to_string(); + match out.status.code() { + Some(0) if !tree.is_empty() => Ok(Tree::Clean(tree)), + Some(1) => { + let mut names: Vec = Vec::new(); + for t in toks { + if t.is_empty() { + break; + } + if !names.iter().any(|n| n == t) { + names.push(t.to_string()); + } + } + Ok(Tree::Conflict(names)) + } + _ => Err(GitError::Failed( + String::from_utf8_lossy(&out.stderr).trim().to_string(), + )), + } +} + +fn commit_tree( + git: &Git, + dir: &Path, + tree: &str, + base: &str, + head: &str, + message: &str, + id: &Identity, +) -> Result { + let out = git.run(&Spec { + repo: Some(dir), + args: &["commit-tree", tree, "-p", base, "-p", head], + stdin: Some(message.as_bytes()), + identity: Some(id), + ..Default::default() + })?; + if out.ok() { + Ok(out.text()) + } else { + Err(GitError::Failed( + String::from_utf8_lossy(&out.stderr).trim().to_string(), + )) + } +} + +/// The ref transaction: the head is verified and the base moves compare-and-swap, together. One +/// ref changes, so it commits or it does not. `Ok(false)`: it did not commit, or we cannot tell +/// (a timeout); the caller asks [`is_landed`]. +fn publish(git: &Git, dir: &Path, i: &Intent) -> Result { + let script = format!( + "start\nverify {head} {head_oid}\nupdate {base} {merge} {base_oid}\nprepare\ncommit\n", + head = refname::branch_ref(&i.head_ref), + head_oid = i.head_oid, + base = refname::branch_ref(&i.base_ref), + merge = i.merge_oid, + base_oid = i.base_oid, + ); + match git.run(&Spec { + repo: Some(dir), + args: &["update-ref", "--stdin"], + stdin: Some(script.as_bytes()), + ..Default::default() + }) { + Ok(o) => Ok(o.ok()), + // Killed by the runner's deadline. SIGKILL leaves the locks of the refs it had prepared; they + // stay until the next restart's cleanup, because a lock cannot be told from a live writer's + // at run time (the intent stays pending while the base is locked). + Err(GitError::Timeout { .. }) => Ok(false), + Err(e) => Err(e), + } +} + +/// Test-only knob: `SC_FORGE_MERGE_PAUSE_MS` stretches the two crash windows of a merge (after the +/// journal row, and after the base moved) so a test can SIGKILL the server inside them. Unset in +/// production. +fn pause_for_tests() { + if let Some(ms) = std::env::var("SC_FORGE_MERGE_PAUSE_MS") + .ok() + .and_then(|v| v.parse::().ok()) + { + std::thread::sleep(Duration::from_millis(ms.min(5_000))); + } +} + +// ---- the handler ---- + +fn try_later(ctx: &Ctx) -> Resp { + error(&ctx.app.base, 405, "Please try again later") +} + +fn stale_head(ctx: &Ctx) -> Resp { + error(&ctx.app.base, 409, "head out of date") +} + +fn internal(ctx: &Ctx, e: E) -> Resp { + ctx.app.internal(&e) +} + +fn conflict(ctx: &Ctx, paths: &[String]) -> Resp { + let shown: Vec<&str> = paths.iter().take(NAMED_PATHS).map(String::as_str).collect(); + let more = paths.len().saturating_sub(shown.len()); + let mut list = shown.join(", "); + if more > 0 { + list.push_str(&format!(" and {more} more")); + } + error(&ctx.app.base, 409, &format!("Merge conflict: {list}")) +} + +/// `POST /repos/{o}/{r}/pulls/{n}/merge`. +pub fn merge_pull(ctx: &mut Ctx) -> Resp { + let a = match acc(ctx) { + Ok(a) => a.clone(), + Err(r) => return r, + }; + if a.perm < Perm::Write { + return needs_write(ctx); + } + let Some(n) = number(ctx.param("n")) else { + return missing(ctx); + }; + let b = match body(ctx) { + Ok(b) => b, + Err(r) => return r, + }; + match str_field(&b, "Do").unwrap_or_default().as_str() { + "merge" => {} + "" => return error(&ctx.app.base, 422, "[Do]: Required"), + "squash" | "rebase" | "rebase-merge" | "fast-forward-only" | "manually-merged" => { + return error(&ctx.app.base, 422, "only Do=merge is supported") + } + _ => return error(&ctx.app.base, 422, "[Do]: In"), + } + let want_head = str_field(&b, "head_commit_id").filter(|s| !s.is_empty()); + gated(ctx, &a, || run(ctx, &a, n, want_head.as_deref())) +} + +fn run(ctx: &Ctx, a: &Access, n: i64, want_head: Option<&str>) -> Resp { + // `gated` has settled whatever an earlier merge left; nothing else can be pending now. + let mut intent_id: Option = None; + let outcome = attempt_loop(ctx, a, n, want_head, &mut intent_id); + // Any exit that did not land leaves the intent for the same judgement recovery makes; a + // landed one is already done, an unknown one stays pending. + if intent_id.is_some() { + let (git, dir) = (&ctx.app.life.git, ctx.app.life.repo_dir(a.repo.id)); + if let Err(e) = recover_locked(ctx.conn, git, &dir, a.repo.id) { + eprintln!( + "sc-forge: could not settle a merge of repository {}: {e}", + a.repo.id + ); + } + } + outcome +} + +fn attempt_loop( + ctx: &Ctx, + a: &Access, + n: i64, + want_head: Option<&str>, + intent_id: &mut Option, +) -> Resp { + let (git, dir) = (&ctx.app.life.git, ctx.app.life.repo_dir(a.repo.id)); + let user = &ctx.who.user; + for round in 0..ATTEMPTS { + // Re-read everything each round: the PR, the rules, both tips, the statuses. + let p = match find(ctx, a, n) { + Ok(p) => p, + Err(r) => return r, + }; + match p.state.as_str() { + "merged" => return error(&ctx.app.base, 405, "The PR is already merged"), + "closed" => return missing(ctx), + _ => {} + } + let (base_oid, head_oid) = match (tip(ctx, a, &p.base_ref), tip(ctx, a, &p.head_ref)) { + (Ok(Some(b)), Ok(Some(h))) => (b, h), + (Ok(_), Ok(_)) => return missing(ctx), + (Err(r), _) | (_, Err(r)) => return r, + }; + if want_head.is_some_and(|w| !w.eq_ignore_ascii_case(&head_oid)) { + return stale_head(ctx); + } + if gw::is_ancestor(git, &dir, &head_oid, &base_oid).unwrap_or(false) { + return try_later(ctx); + } + let rules = match crate::protection::load(ctx.conn, a.repo.id) { + Ok(r) => r, + Err(e) => return internal(ctx, e), + }; + if let Some(rule) = rules + .for_branch(&p.base_ref) + .filter(|r| r.enable_status_check) + { + let latest = match statuses::latest(ctx.conn, a.repo.id, &head_oid) { + Ok(l) => l, + Err(e) => return internal(ctx, e), + }; + let required = rule.contexts.clone().unwrap_or_default(); + // A rule that requires checks but names none cannot be satisfied: fail closed. + if required.is_empty() || !required.iter().all(|c| statuses::satisfied(&latest, c)) { + return error( + &ctx.app.base, + 405, + "not allowed to merge [reason: Not all required status checks successful]", + ); + } + } + let facts = match pulls::compute(ctx.conn, git, &dir, a.repo.id, &base_oid, &head_oid) { + Ok(c) => c, + Err(ComputeError::Db(e)) => return internal(ctx, e), + Err(ComputeError::Git(e)) => return crate::content_api::read_err(ctx, a, e), + }; + let Some(merge_base) = facts.merge_base.clone() else { + return error(&ctx.app.base, 409, "Merge conflict: unrelated histories"); + }; + let tree = match merge_tree(git, &dir, &base_oid, &head_oid) { + Ok(Tree::Clean(t)) => t, + Ok(Tree::Conflict(paths)) => return conflict(ctx, &paths), + Err(e) => return crate::content_api::read_err(ctx, a, e), + }; + let when = now(); + let email = crate::content_write::commit_email(user); + let Some(identity) = Identity::new(&user.login, &email).map(|i| i.at(when)) else { + return internal(ctx, "unusable commit identity"); + }; + // One line, whatever the title holds: a newline would add body or trailer lines (a fake + // `Signed-off-by`) to a commit written under the merger's identity, and a NUL would make + // `commit-tree` fail for good. + let title: String = p + .title + .chars() + .map(|c| if c.is_control() { ' ' } else { c }) + .collect(); + let message = format!( + "Merge pull request '{}' (#{}) from {} into {}\n", + title, p.number, p.head_ref, p.base_ref + ); + let merge_oid = + match commit_tree(git, &dir, &tree, &base_oid, &head_oid, &message, &identity) { + Ok(o) => o, + Err(e) => return crate::content_api::read_err(ctx, a, e), + }; + let mut intent = Intent { + id: 0, + repo_id: a.repo.id, + pull_number: p.number, + base_ref: p.base_ref.clone(), + head_ref: p.head_ref.clone(), + base_oid: base_oid.clone(), + head_oid: head_oid.clone(), + merge_oid: merge_oid.clone(), + merge_base, + changed_files: facts.changed_files, + additions: facts.additions, + deletions: facts.deletions, + merger_id: user.id, + merged_unix: when, + }; + // The journal first, and again (every field) whenever the merge commit changes. + match *intent_id { + None => match insert(ctx.conn, &intent, when) { + Ok(id) => { + *intent_id = Some(id); + intent.id = id; + } + Err(e) => return internal(ctx, e), + }, + Some(id) => { + intent.id = id; + if let Err(e) = retarget(ctx.conn, &intent) { + return internal(ctx, e); + } + } + } + pause_for_tests(); + // Whether the transaction committed is asked of git, not read off an exit status; an + // answer that cannot be had leaves the intent pending and the request failed. + let committed = publish(git, &dir, &intent).unwrap_or(false); + let landed = committed + || match is_landed(git, &dir, &intent) { + Ok(l) => l, + Err(e) => return crate::content_api::read_err(ctx, a, e), + }; + if landed { + mark_landed(git, &dir, &intent); + pause_for_tests(); + return match finish(ctx.conn, git, &dir, &intent, false) { + Ok(()) => { + let mut r = Resp::json(200, &serde_json::Value::Null); + r.body = Vec::new(); + r + } + Err(e) => internal(ctx, e), + }; + } + // Not published. Why: the head moved (the client's id is stale), the base moved (rebuild on + // the new tip), or something transient. + let (head2, base2) = (tip(ctx, a, &p.head_ref), tip(ctx, a, &p.base_ref)); + if let (Ok(Some(h2)), Ok(Some(b2))) = (&head2, &base2) { + if *h2 != head_oid { + return stale_head(ctx); + } + if *b2 == base_oid && round + 1 == ATTEMPTS { + return try_later(ctx); + } + } + std::thread::sleep(Duration::from_millis(30)); + } + try_later(ctx) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn fixture() -> (Connection, std::path::PathBuf, i64, i64) { + static N: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0); + let d = std::env::temp_dir().join(format!( + "scl-forge-merges-{}-{}", + std::process::id(), + N.fetch_add(1, std::sync::atomic::Ordering::SeqCst) + )); + let _ = std::fs::remove_dir_all(&d); + let conn = crate::db::open_for_admin(&d).unwrap(); + let owner = crate::store::create_account( + &conn, + &crate::store::NewAccount { + login: "o".into(), + visibility: "public".into(), + ..Default::default() + }, + 1, + ) + .unwrap(); + let repo = crate::repos::insert_creating(&conn, owner.id, "r", "main", "", "", 1).unwrap(); + (conn, d, repo.id, owner.id) + } + + fn intent(repo: i64, user: i64, n: i64) -> Intent { + Intent { + id: 0, + repo_id: repo, + pull_number: n, + base_ref: "main".into(), + head_ref: "f".into(), + base_oid: "b".repeat(40), + head_oid: "a".repeat(40), + merge_oid: "c".repeat(40), + merge_base: "d".repeat(40), + changed_files: 2, + additions: 3, + deletions: 1, + merger_id: user, + merged_unix: 77, + } + } + + #[test] + fn finalizing_writes_every_field_a_normal_merge_does_and_is_idempotent() { + let (c, d, repo, user) = fixture(); + pulls::insert(&c, repo, "f", "main", "b", "h", "t", "", user, 5).unwrap(); + let id = insert(&c, &intent(repo, user, 1), 6).unwrap(); + assert!(has_pending(&c, repo)); + let mut i = intent(repo, user, 1); + i.id = id; + finalize_rows(&c, &i, false).unwrap(); + finalize_rows(&c, &i, false).unwrap(); + let p = pulls::find(&c, repo, 1).unwrap().unwrap(); + assert_eq!(p.state, "merged"); + assert_eq!(p.merge_commit.as_deref(), Some(i.merge_oid.as_str())); + assert_eq!(p.final_head.as_deref(), Some(i.head_oid.as_str())); + assert_eq!( + (p.merged, p.merged_by, p.closed), + (Some(77), Some(user), Some(77)) + ); + assert_eq!( + (p.changed_files, p.additions, p.deletions), + (Some(2), Some(3), Some(1)) + ); + assert_eq!(p.merge_base.as_deref(), Some(i.merge_base.as_str())); + assert!(!has_pending(&c, repo)); + std::fs::remove_dir_all(&d).unwrap(); + assert!(!d.exists()); + } + + #[test] + fn one_pending_intent_per_pull_and_retarget_changes_only_the_commit() { + let (c, d, repo, user) = fixture(); + pulls::insert(&c, repo, "f", "main", "b", "h", "t", "", user, 5).unwrap(); + let id = insert(&c, &intent(repo, user, 1), 6).unwrap(); + assert!(insert(&c, &intent(repo, user, 1), 7).is_err()); + let mut again = intent(repo, user, 1); + again.id = id; + again.base_oid = "e".repeat(40); + again.merge_oid = "f".repeat(40); + again.merge_base = "9".repeat(40); + (again.changed_files, again.additions, again.deletions) = (5, 6, 7); + again.merged_unix = 99; + again.head_oid = "8".repeat(40); + retarget(&c, &again).unwrap(); + let p = pending(&c, repo).unwrap(); + assert_eq!( + ( + p[0].base_oid.clone(), + p[0].merge_oid.clone(), + p[0].merge_base.clone() + ), + ("e".repeat(40), "f".repeat(40), "9".repeat(40)) + ); + assert_eq!( + ( + p[0].changed_files, + p[0].additions, + p[0].deletions, + p[0].merged_unix + ), + (5, 6, 7, 99), + "a retry persists every fact it recomputed" + ); + assert_eq!(p[0].head_oid, "8".repeat(40), "and the head it built on"); + abort(&c, id).unwrap(); + assert!( + insert(&c, &intent(repo, user, 1), 8).is_ok(), + "an aborted intent frees the slot" + ); + std::fs::remove_dir_all(&d).unwrap(); + assert!(!d.exists()); + } +} diff --git a/crates/forge/src/orphans.rs b/crates/forge/src/orphans.rs new file mode 100644 index 00000000..c55e4c2d --- /dev/null +++ b/crates/forge/src/orphans.rs @@ -0,0 +1,359 @@ +//! Startup cleanup after a crash: git processes the dead server left behind, and the ref locks they +//! held. +//! +//! Git children run in their own process groups, so a SIGKILLed `sc-forge` leaves its `update-ref` +//! or `receive-pack` children running. Left alone they could finish a ref update after startup +//! recovery has judged a merge "not landed". The data directory is exclusively locked by the +//! server (see `lifecycle::DirLock`), so at startup every git process that names a path under this +//! data directory's `repos/` belongs to a dead predecessor; they are killed, and the `*.lock` files +//! a killed git leaves in `refs/` (nothing else holds them now) are removed, before any recovery +//! looks at the repositories. At run time nothing here runs: a live lock then belongs to a live +//! writer. + +use std::path::Path; +use std::time::{Duration, Instant}; + +/// What the cleanup found. +#[derive(Debug, Default, PartialEq, Eq)] +pub struct Swept { + pub killed: usize, + pub locks: usize, +} + +/// Separates arguments and environment entries in what [`processes`] returns on Linux, where they +/// are NUL-delimited and a value may hold spaces. +const SEP: char = '\u{1f}'; + +/// `(pid, argv and environment joined by the separator, the separator)` of every process, from +/// `/proc` where it exists, else from BSD `ps` (whose `-E` flag appends the environment of the +/// caller's own processes, space separated: a data directory with spaces cannot be told apart there, +/// and development hosts are the only users). The environment matters: the writer behind an HTTP +/// push is `git receive-pack --stateless-rpc .` run in the repository's directory, named by nothing +/// on its command line, but it inherited the `FORGE_DATA_DIR` the transport sets. +fn processes() -> Option> { + if Path::new("/proc/self").exists() { + let mut out = Vec::new(); + let rd = std::fs::read_dir("/proc").ok()?; + for e in rd.flatten() { + let Some(pid) = e.file_name().to_str().and_then(|s| s.parse::().ok()) else { + continue; + }; + let join = |raw: Vec| { + raw.split(|b| *b == 0) + .filter(|a| !a.is_empty()) + .map(|a| String::from_utf8_lossy(a).to_string()) + .collect::>() + .join(&SEP.to_string()) + }; + if let Ok(raw) = std::fs::read(e.path().join("cmdline")) { + let mut cmd = join(raw); + if let Ok(env) = std::fs::read(e.path().join("environ")) { + cmd.push(SEP); + cmd.push_str(&join(env)); + } + out.push((pid, cmd, SEP)); + } + } + return Some(out); + } + let ps = ["/bin/ps", "/usr/bin/ps"] + .into_iter() + .find(|p| Path::new(p).exists()); + let ps = ps?; + let o = std::process::Command::new(ps) + .args(["-axEww", "-o", "pid=,command="]) + .output() + .ok() + .filter(|o| o.status.success())?; + Some( + String::from_utf8_lossy(&o.stdout) + .lines() + .filter_map(|l| { + let l = l.trim_start(); + let (pid, cmd) = l.split_once(' ')?; + Some((pid.parse().ok()?, cmd.trim_start().to_string(), ' ')) + }) + .collect(), + ) +} + +/// Is `cmd` a git invocation: the configured git executable (`SC_FORGE_GIT` may name it anything), +/// or an executable called `git` or `git-*` (what git itself spawns, `git-receive-pack` and the like)? +fn is_git(cmd: &str, sep: char, configured: Option<&Path>) -> bool { + let exe = cmd.split(sep).next().unwrap_or(""); + if configured.is_some_and(|c| c == Path::new(exe)) { + return true; + } + Path::new(exe) + .file_name() + .and_then(|n| n.to_str()) + .is_some_and(|n| n == "git" || n.starts_with("git-")) +} + +/// Does `cmd` (arguments and environment entries joined by `sep`) contain `needle` as a whole token +/// start: preceded by the separator, `=` or the start, and, for an environment value (`exact_end`), +/// followed by the separator or the end? Another instance's `/srv/forge-secondary` or +/// `/srv/forge secondary` must not match `/srv/forge`, nor `/mnt/srv/forge/repos/` match +/// `/srv/forge/repos/`. +fn mentions(cmd: &str, needle: &str, exact_end: bool, sep: char) -> bool { + let mut from = 0; + while let Some(i) = cmd[from..].find(needle) { + let at = from + i; + let end = at + needle.len(); + let before_ok = at == 0 || cmd[..at].ends_with(sep) || cmd[..at].ends_with('='); + let after_ok = !exact_end || end == cmd.len() || cmd[end..].starts_with(sep); + if before_ok && after_ok { + return true; + } + from = at + cmd[at..].chars().next().map_or(1, char::len_utf8); + } + false +} + +/// Dead but not yet reaped by its parent (a zombie holds nothing, least of all a lock). +fn is_zombie(pid: i32) -> bool { + if let Ok(stat) = std::fs::read_to_string(format!("/proc/{pid}/stat")) { + // `pid (comm) S ...`: the state follows the last closing parenthesis. + return stat + .rsplit_once(')') + .and_then(|(_, rest)| rest.trim_start().chars().next()) + == Some('Z'); + } + ["/bin/ps", "/usr/bin/ps"] + .into_iter() + .find(|p| Path::new(p).exists()) + .and_then(|ps| { + std::process::Command::new(ps) + .args(["-o", "stat=", "-p", &pid.to_string()]) + .output() + .ok() + }) + .is_some_and(|o| { + String::from_utf8_lossy(&o.stdout) + .trim_start() + .starts_with('Z') + }) +} + +fn alive(pid: i32) -> bool { + // SAFETY: signal 0 only checks that the process exists. + let exists = unsafe { libc::kill(pid, 0) == 0 }; + exists && !is_zombie(pid) +} + +/// Kill git processes that work on this data directory's repositories (not ours: `me`), then wait +/// briefly for them to be gone. Returns `(killed, survivors)`, or `None` when the process table +/// could not be read at all. The match is on the command line +/// (`--git-dir=/repos/...` for our `update-ref`s, the repository path for `receive-pack` and +/// `upload-pack`), or by the `FORGE_DATA_DIR` they inherited from the HTTP transport: the processes +/// that can move a ref. Helpers that only read or write objects +/// (`index-pack`, `unpack-objects`) are not named by path and are left to finish and exit. +pub fn kill_orphan_git(data_dir: &Path, me: i32, git_bin: Option<&Path>) -> Option<(usize, usize)> { + let needle = format!("{}/repos/", data_dir.display()); + let env_needle = format!("FORGE_DATA_DIR={}", data_dir.display()); + let mut killed = std::collections::BTreeSet::new(); + // A dying `http-backend` may have forked its `receive-pack` between two looks: pass until a + // look finds nothing alive (at most a few seconds). + let until = Instant::now() + Duration::from_secs(5); + loop { + // A look that could not be taken is not an empty look: nothing is concluded from it. + let found: Vec = processes()? + .into_iter() + .filter(|(pid, cmd, sep)| { + *pid != me + && (mentions(cmd, &needle, false, *sep) + || mentions(cmd, &env_needle, true, *sep)) + && is_git(cmd, *sep, git_bin) + && alive(*pid) + }) + .map(|(pid, _, _)| pid) + .collect(); + if found.is_empty() { + return Some((killed.len(), 0)); + } + for pid in &found { + killed.insert(*pid); + // SAFETY: SIGKILL to a process we just found by its command line. + unsafe { libc::kill(*pid, libc::SIGKILL) }; + } + if Instant::now() >= until { + let survivors = found.iter().filter(|p| alive(**p)).count(); + return Some((killed.len(), survivors)); + } + std::thread::sleep(Duration::from_millis(30)); + } +} + +fn remove_locks(dir: &Path, depth: usize) -> usize { + let Ok(rd) = std::fs::read_dir(dir) else { + return 0; + }; + let mut n = 0; + for e in rd.flatten() { + let p = e.path(); + let Ok(t) = e.file_type() else { continue }; + if t.is_dir() && depth < 32 { + n += remove_locks(&p, depth + 1); + } else if t.is_file() + && p.extension().is_some_and(|x| x == "lock") + && std::fs::remove_file(&p).is_ok() + { + eprintln!( + "sc-forge: removed the ref lock {} left by a killed git", + p.display() + ); + n += 1; + } + } + n +} + +/// Remove the leftover ref locks of every repository (`refs/**/*.lock`, `packed-refs.lock`, +/// `HEAD.lock`). Only after [`kill_orphan_git`], and only at startup. +pub fn remove_stale_locks(data_dir: &Path) -> usize { + let mut n = 0; + let Ok(rd) = std::fs::read_dir(data_dir.join("repos")) else { + return 0; + }; + for e in rd.flatten() { + let repo = e.path(); + if !repo.is_dir() { + continue; + } + n += remove_locks(&repo.join("refs"), 0); + for f in ["packed-refs.lock", "HEAD.lock"] { + if std::fs::remove_file(repo.join(f)).is_ok() { + n += 1; + } + } + } + n +} + +/// The startup cleanup. `data_dir` must be the canonical data directory; `git_bin` is the git +/// executable the server runs. Locks are removed only +/// when no matching process is left alive: a lock of a writer that could not be killed is not ours +/// to take. +pub fn sweep(data_dir: &Path, git_bin: Option<&Path>) -> Swept { + let Some((killed, survivors)) = kill_orphan_git(data_dir, std::process::id() as i32, git_bin) + else { + eprintln!( + "sc-forge: could not read the process table; leaving any orphaned git writers and ref locks alone" + ); + return Swept::default(); + }; + let locks = if survivors == 0 { + remove_stale_locks(data_dir) + } else { + eprintln!( + "sc-forge: {survivors} orphaned git process(es) would not die; leaving the ref locks in place" + ); + 0 + }; + if killed + locks > 0 { + eprintln!("sc-forge: startup cleanup killed {killed} orphaned git process(es) and removed {locks} stale lock(s)"); + } + Swept { killed, locks } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn only_git_executables_count() { + let sp = ' '; + assert!(is_git( + "/usr/bin/git --git-dir=/d/repos/1.git update-ref --stdin", + sp, + None + )); + assert!(is_git("git-receive-pack /d/repos/1.git", sp, None)); + assert!(!is_git("/bin/sleep 100 /d/repos/1.git", sp, None)); + assert!(!is_git("/usr/bin/gitk /d/repos/1.git", sp, None)); + // a differently named git, configured through SC_FORGE_GIT, counts too + let custom = Path::new("/opt/vcs/bin/scm"); + assert!(is_git( + "/opt/vcs/bin/scm --git-dir=/d/repos/1.git update-ref", + sp, + Some(custom) + )); + assert!(!is_git( + "/opt/vcs/bin/scm --git-dir=/d/repos/1.git update-ref", + sp, + None + )); + } + + #[test] + fn a_data_directory_matches_only_as_a_whole_token() { + let env = "FORGE_DATA_DIR=/srv/forge"; + let sp = ' '; + assert!(mentions( + "git receive-pack . HOME=/h FORGE_DATA_DIR=/srv/forge PATH=/bin", + env, + true, + sp + )); + assert!(mentions("FORGE_DATA_DIR=/srv/forge", env, true, sp)); + assert!(!mentions( + "git x FORGE_DATA_DIR=/srv/forge-secondary PATH=/bin", + env, + true, + sp + )); + assert!(!mentions("git x XFORGE_DATA_DIR=/srv/forge", env, true, sp)); + let repos = "/srv/forge/repos/"; + assert!(mentions( + "git --git-dir=/srv/forge/repos/7.git update-ref", + repos, + false, + sp + )); + assert!(!mentions( + "git --git-dir=/mnt/srv/forge/repos/7.git update-ref", + repos, + false, + sp + )); + assert!(!mentions( + "git --git-dir=/srv/forge-secondary/repos/7.git x", + repos, + false, + sp + )); + // with real separators (Linux), a space inside a value is not a boundary + let u = SEP; + let spaced = format!("git{u}x{u}FORGE_DATA_DIR=/srv/forge secondary{u}PATH=/bin"); + assert!( + !mentions(&spaced, env, true, u), + "`/srv/forge secondary` is another directory" + ); + let exact = format!("git{u}x{u}FORGE_DATA_DIR=/srv/forge{u}PATH=/bin"); + assert!(mentions(&exact, env, true, u)); + } + + #[test] + fn locks_under_refs_are_removed_and_nothing_else() { + let d = std::env::temp_dir().join(format!("scl-forge-orphans-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&d); + let repo = d.join("repos").join("1.git"); + std::fs::create_dir_all(repo.join("refs/heads/feat")).unwrap(); + for f in [ + "refs/heads/main.lock", + "refs/heads/feat/x.lock", + "refs/heads/main", + "packed-refs.lock", + "config.lock", + ] { + std::fs::write(repo.join(f), b"").unwrap(); + } + assert_eq!(remove_stale_locks(&d), 3); + assert!(repo.join("refs/heads/main").exists() && repo.join("config.lock").exists()); + assert!( + !repo.join("refs/heads/main.lock").exists() && !repo.join("packed-refs.lock").exists() + ); + std::fs::remove_dir_all(&d).unwrap(); + assert!(!d.exists()); + } +} diff --git a/crates/forge/src/pulls.rs b/crates/forge/src/pulls.rs index de449b2f..08df78f5 100644 --- a/crates/forge/src/pulls.rs +++ b/crates/forge/src/pulls.rs @@ -324,7 +324,8 @@ pub fn compute( return Ok(c); } let mb = merge_base(git, dir, base, head)?; - let (files, stats) = gitobj::changes(git, dir, mb.as_deref(), head)?; + // Unrelated histories: the changes are counted against the base tip (recorded T#0). + let (files, stats) = gitobj::changes(git, dir, Some(mb.as_deref().unwrap_or(base)), head)?; // Unrelated histories cannot merge; `merge-tree` would die on them. let mergeable = mb.is_some() && merges_cleanly(git, dir, base, head)?; let c = Compute { @@ -530,6 +531,12 @@ pub fn sync_locked( repo_id: i64, now: i64, ) -> Result { + // A merge that landed is finalized before any pull request is closed or moved on; one that + // cannot be settled yet (a git writer still holds its base) blocks the sync. + crate::merges::recover_locked(conn, git, dir, repo_id)?; + if crate::merges::has_pending(conn, repo_id) { + return Ok(0); + } let open = open_pulls(conn, repo_id)?; if open.is_empty() { return Ok(0); @@ -561,6 +568,9 @@ pub fn reconcile_repo( repo_id: i64, now: i64, ) -> Result<(), ComputeError> { + // A merge a crash interrupted is settled first: its receipt decides, then stray receipts go. + crate::merges::recover_locked(conn, git, dir, repo_id)?; + crate::merges::drop_stale_receipts(conn, git, dir, repo_id)?; sync_locked(conn, git, dir, repo_id, now)?; let kept = refs_under(git, dir, "refs/forge/pull")?; let mut st = diff --git a/crates/forge/src/pulls_api.rs b/crates/forge/src/pulls_api.rs index d4599f41..7ab4ed8c 100644 --- a/crates/forge/src/pulls_api.rs +++ b/crates/forge/src/pulls_api.rs @@ -35,11 +35,11 @@ fn enc_path(p: &str) -> String { .join("/") } -fn missing(ctx: &Ctx) -> Resp { +pub(crate) fn missing(ctx: &Ctx) -> Resp { not_found(&ctx.app.base, "The target couldn't be found.", None) } -fn needs_write(ctx: &Ctx) -> Resp { +pub(crate) fn needs_write(ctx: &Ctx) -> Resp { error( &ctx.app.base, 403, @@ -48,11 +48,11 @@ fn needs_write(ctx: &Ctx) -> Resp { } /// The PR number out of a path parameter (`12`, or `12.diff` / `12.patch` where allowed). -fn number(raw: &str) -> Option { +pub(crate) fn number(raw: &str) -> Option { raw.parse::().ok().filter(|n| *n >= 1) } -fn tip(ctx: &Ctx, a: &Access, branch: &str) -> Result, Resp> { +pub(crate) fn tip(ctx: &Ctx, a: &Access, branch: &str) -> Result, Resp> { if !refname::valid_branch(branch) { return Ok(None); } @@ -128,13 +128,15 @@ fn live(ctx: &Ctx, a: &Access, p: &Pull) -> Result { pulls::ComputeError::Db(e) => ctx.app.internal(&e), pulls::ComputeError::Git(e) => read_err(ctx, a, e), })?; + let merge_base = c.merge_base.or_else(|| Some(base_sha.clone())); Ok(Live { base_sha, head_sha, head_ref, // Recorded: a closed PR keeps reading mergeable. mergeable: p.state != "open" || c.mergeable, - merge_base: c.merge_base, + // Unrelated histories have no merge base; Gitea names the base tip (recorded T#0) and so do the diffs. + merge_base, changed_files: c.changed_files, additions: c.additions, deletions: c.deletions, @@ -191,7 +193,7 @@ fn pull_json( "merged": merged, "merged_at": p.merged.filter(|_| merged).map(rfc3339), "merge_commit_sha": if merged { json!(p.merge_commit) } else { Value::Null }, - "merged_by": match p.merged_by.filter(|_| merged) { Some(id) => user(id)?, None => Value::Null }, + "merged_by": match p.merged_by.filter(|_| merged) { Some(id) => nested_user(ctx, id)?, None => Value::Null }, "allow_maintainer_edit": false, "base": {"label": p.base_ref, "ref": p.base_ref, "sha": l.base_sha, "repo_id": a.repo.id, "repo": repo}, "head": {"label": p.head_ref, "ref": l.head_ref, "sha": l.head_sha, "repo_id": a.repo.id, "repo": repo}, @@ -204,6 +206,14 @@ fn pull_json( })) } +/// The account in `merged_by`: the nested form Gitea shows there (login-derived email, inactive). +fn nested_user(ctx: &Ctx, id: i64) -> Result { + let u = store::find_by_id(ctx.conn, id) + .map_err(|e| ctx.app.internal(&e))? + .ok_or_else(|| ctx.app.internal(&"pull request without merger"))?; + Ok(crate::api::nested_user_json(&ctx.app.base, &u, true)) +} + fn repo_value(ctx: &Ctx, a: &Access) -> Result { // A repository with a pull request has branches, so it is not empty; `repo_json` would take // the shared lifecycle lock a second time. @@ -220,7 +230,7 @@ pub(crate) fn head_of(ctx: &Ctx, a: &Access, p: &Pull) -> Result { live(ctx, a, p).map(|l| l.head_sha) } -fn find(ctx: &Ctx, a: &Access, n: i64) -> Result { +pub(crate) fn find(ctx: &Ctx, a: &Access, n: i64) -> Result { match pulls::find(ctx.conn, a.repo.id, n) { Ok(Some(p)) => Ok(p), Ok(None) => Err(missing(ctx)), @@ -229,12 +239,27 @@ fn find(ctx: &Ctx, a: &Access, n: i64) -> Result { } /// Run `f` under the write gate. -fn gated(ctx: &Ctx, a: &Access, f: impl FnOnce() -> Resp) -> Resp { +pub(crate) fn gated(ctx: &Ctx, a: &Access, f: impl FnOnce() -> Resp) -> Resp { match ctx .app .life - .with_write_row(ctx.conn, a.repo.id, GATE_WAIT, |_| f()) - { + .with_write_row(ctx.conn, a.repo.id, GATE_WAIT, |_| { + // A merge that landed but was not finalized is settled before anything else may change + // the pull request (a close or retarget would otherwise outrun its finalization). If it + // cannot be settled right now, nothing proceeds. + let (git, dir) = (&ctx.app.life.git, dir(ctx, a)); + let settled = crate::merges::recover_locked(ctx.conn, git, &dir, a.repo.id) + .map(|_| !crate::merges::has_pending(ctx.conn, a.repo.id)); + match settled { + Ok(true) => f(), + Ok(false) => error( + &ctx.app.base, + 503, + "a merge of this repository is still being settled, try again", + ), + Err(e) => ctx.app.internal(&e), + } + }) { Ok(WriteRun::Ran(r)) => r, Ok(WriteRun::Busy) => error( &ctx.app.base, @@ -249,6 +274,15 @@ fn gated(ctx: &Ctx, a: &Access, f: impl FnOnce() -> Resp) -> Resp { /// Before a read: if a push deleted the head of an open PR and its own sync has not run yet, /// run it now (outside any lifecycle lock; the sync takes the write gate). fn converge(ctx: &Ctx, a: &Access) { + // An unfinished merge is completed if the gate is free right now (never waited for). + if crate::merges::has_pending(ctx.conn, a.repo.id) { + let _ = ctx + .app + .life + .with_write_row(ctx.conn, a.repo.id, Duration::from_millis(50), |_| { + crate::merges::recover_locked(ctx.conn, &ctx.app.life.git, &dir(ctx, a), a.repo.id) + }); + } if pulls::has_vanished_head(ctx.conn, &ctx.app.life.git, &dir(ctx, a), a.repo.id) { pulls::sync(&ctx.app.life, ctx.conn, a.repo.id, now()); } @@ -333,7 +367,9 @@ pub fn create_pull(ctx: &mut Ctx) -> Resp { Ok(None) => {} Err(e) => return ctx.app.internal(&e), } - if head_tip == base_tip { + // Recorded: a head already contained in the base (even the same commit) is accepted; only a + // branch compared with itself has "no changes". + if head == base { return error( &ctx.app.base, 422, @@ -645,6 +681,8 @@ pub fn is_merged(ctx: &mut Ctx) -> Resp { Ok(a) => a.clone(), Err(r) => return r, }; + // An unfinished merge is settled before the answer, so a poller sees it. + converge(ctx, &a); let Some(n) = number(ctx.param("n")) else { return missing(ctx); }; diff --git a/crates/forge/src/routes.rs b/crates/forge/src/routes.rs index 7cee9a5a..ce90b3b0 100644 --- a/crates/forge/src/routes.rs +++ b/crates/forge/src/routes.rs @@ -10,7 +10,8 @@ use crate::limits::{Backoff, Gate, Proxies}; use crate::repos::{self, Access, Perm}; use crate::scope::{self, Cat, Level, Need}; use crate::{ - api, content_api, content_write, db, protection, pulls_api, repo_api, reviews, statuses, store, + api, content_api, content_write, db, merges, protection, pulls_api, repo_api, reviews, + statuses, store, }; use rusqlite::Connection; use std::path::PathBuf; @@ -630,6 +631,14 @@ pub static ROUTES: &[Route] = &[ Repo(Perm::Read), reviews::create_comment, ), + r( + "POST", + "/repos/{o}/{r}/pulls/{n}/merge", + Token, + n(Cat::Repository, Level::Write), + Repo(Perm::Read), + merges::merge_pull, + ), ]; /// Captured path parameters, in pattern order. diff --git a/crates/forge/tests/e2e_crash.rs b/crates/forge/tests/e2e_crash.rs new file mode 100644 index 00000000..8c1a23d5 --- /dev/null +++ b/crates/forge/tests/e2e_crash.rs @@ -0,0 +1,536 @@ +//! F4c: `kill -9` during merges. A real `sc-forge serve` process merges pull requests while it is +//! killed at an arbitrary moment and restarted; after each restart the merge journal and the +//! reconciler must leave the forge consistent: no pending intent, no receipt ref, and every pull +//! request either wholly merged (its merge commit on the base, two parents) or wholly open (its +//! head not on the base). +#![cfg(unix)] + +use serde_json::{json, Value}; +use std::io::{Read, Write}; +use std::net::TcpStream; +use std::path::{Path, PathBuf}; +use std::process::{Child, Command, Stdio}; +use std::time::{Duration, Instant}; + +const BIN: &str = env!("CARGO_BIN_EXE_sc-forge"); +/// Pull requests opened per round. +const PRS: i64 = 3; + +struct Server { + child: Option, + port: u16, +} + +impl Server { + fn start(data: &Path, port: u16) -> Server { + let child = Command::new(BIN) + .args(["serve", "--data-dir"]) + .arg(data) + .args([ + "--listen", + &format!("127.0.0.1:{port}"), + "--public-url", + &format!("http://127.0.0.1:{port}"), + ]) + // stretch the two crash windows of a merge so the kills can land inside them + .env("SC_FORGE_MERGE_PAUSE_MS", "300") + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("start sc-forge"); + let s = Server { + child: Some(child), + port, + }; + let t0 = Instant::now(); + while t0.elapsed() < Duration::from_secs(30) { + if matches!(s.req("GET", "/api/healthz", None, None), Some((200, _))) { + return s; + } + std::thread::sleep(Duration::from_millis(50)); + } + panic!("sc-forge did not come up"); + } + + /// SIGKILL: no drain, no cleanup. + fn kill9(&mut self) { + if let Some(mut c) = self.child.take() { + let _ = c.kill(); + let _ = c.wait(); + } + } + + fn req( + &self, + method: &str, + path: &str, + token: Option<&str>, + body: Option<&Value>, + ) -> Option<(u16, String)> { + let mut s = TcpStream::connect(("127.0.0.1", self.port)).ok()?; + s.set_read_timeout(Some(Duration::from_secs(20))).ok()?; + let b = body.map(Value::to_string).unwrap_or_default(); + let mut head = format!("{method} {path} HTTP/1.1\r\nHost: x\r\nConnection: close\r\n"); + if let Some(t) = token { + head += &format!("Authorization: token {t}\r\n"); + } + if body.is_some() { + head += &format!( + "Content-Type: application/json\r\nContent-Length: {}\r\n", + b.len() + ); + } + s.write_all(format!("{head}\r\n{b}").as_bytes()).ok()?; + let mut buf = Vec::new(); + s.read_to_end(&mut buf).ok()?; + let text = String::from_utf8_lossy(&buf).to_string(); + let (h, rest) = text.split_once("\r\n\r\n")?; + let status = h.split(' ').nth(1)?.parse().ok()?; + Some((status, rest.to_string())) + } +} + +impl Drop for Server { + fn drop(&mut self) { + self.kill9(); + } +} + +fn free_port() -> u16 { + std::net::TcpListener::bind("127.0.0.1:0") + .unwrap() + .local_addr() + .unwrap() + .port() +} + +fn admin(data: &Path, args: &[&str]) -> String { + let out = Command::new(BIN) + .arg("admin") + .args(args) + .arg("--data-dir") + .arg(data) + .output() + .unwrap(); + assert!( + out.status.success(), + "admin {args:?}: {}", + String::from_utf8_lossy(&out.stderr) + ); + String::from_utf8_lossy(&out.stdout).trim().to_string() +} + +fn b64(s: &str) -> String { + const T: &[u8] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"; + let mut out = String::new(); + for c in s.as_bytes().chunks(3) { + let n = (u32::from(c[0]) << 16) + | (u32::from(*c.get(1).unwrap_or(&0)) << 8) + | u32::from(*c.get(2).unwrap_or(&0)); + for i in 0..4 { + out.push(if i <= c.len() { + T[(n >> (18 - 6 * i) & 63) as usize] as char + } else { + '=' + }); + } + } + out +} + +/// Kills a process on every exit of a test (panic included), so an orphan stand-in never outlives it. +struct KillOnDrop(i32); +impl Drop for KillOnDrop { + fn drop(&mut self) { + // SAFETY: SIGKILL to a child this test spawned. + unsafe { libc::kill(self.0, libc::SIGKILL) }; + } +} + +struct Tmp(PathBuf); +impl Drop for Tmp { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.0); + } +} + +fn git(data: &Path, repo: i64, args: &[&str]) -> Option { + let g = scl_forge::gitproc::Git::new(data, scl_forge::gitproc::find_git().unwrap(), 2).unwrap(); + let dir = data.join("repos").join(format!("{repo}.git")); + let out = g + .run(&scl_forge::gitproc::Spec { + repo: Some(&dir), + args, + ..Default::default() + }) + .unwrap(); + out.ok().then(|| out.text()) +} + +#[test] +fn kill_nine_during_merges_leaves_a_consistent_forge() { + let data = std::env::temp_dir().join(format!("scl-forge-crash-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&data); + let _tmp = Tmp(data.clone()); + admin( + &data, + &[ + "user", + "create", + "--username", + "siteadmin", + "--password", + &format!("pw-{}-0123456789", std::process::id()), + "--email", + "a@example.invalid", + "--admin", + "--must-change-password=false", + ], + ); + let tok = admin( + &data, + &[ + "user", + "generate-access-token", + "--username", + "siteadmin", + "--token-name", + "t", + "--scopes", + "all", + "--raw", + ], + ); + let tok = tok.as_str(); + + let mut srv = Server::start(&data, free_port()); + let (st, body) = srv + .req( + "POST", + "/api/v1/user/repos", + Some(tok), + Some(&json!({"name": "c", "default_branch": "main", "auto_init": true})), + ) + .unwrap(); + assert_eq!(st, 201, "{body}"); + let repo: i64 = serde_json::from_str::(&body).unwrap()["id"] + .as_i64() + .unwrap(); + + // Each round: open fresh pull requests, merge them all at once, kill the server after a delay + // that lands somewhere inside the merges, restart, and check. The merge is paused for 300 ms + // after its journal row and again after the base moved, so kills land inside both windows + // often; rounds go on until the database shows that both kinds of interruption really + // happened (an intent aborted by recovery, an intent finished by recovery), at least five + // rounds and at most sixteen. + let delays = [250u64, 600, 950, 1300, 1700, 2100]; + let mut total = 0; + let (mut aborted, mut recovered) = (0, 0); + for round in 0..16 { + let first = total + 1; + for _ in 0..PRS { + let i = total; + total += 1; + let (st, b) = srv + .req( + "POST", + &format!("/api/v1/repos/siteadmin/c/contents/m{i}.txt"), + Some(tok), + Some(&json!({"content": b64("x\n"), "message": format!("m{i}"), "branch": "main", "new_branch": format!("b{i}")})), + ) + .unwrap(); + assert_eq!(st, 201, "round {round}: {b}"); + let (st, b) = srv + .req( + "POST", + "/api/v1/repos/siteadmin/c/pulls", + Some(tok), + Some(&json!({"title": format!("PR {i}"), "head": format!("b{i}"), "base": "main"})), + ) + .unwrap(); + assert_eq!(st, 201, "round {round}: {b}"); + } + let port = srv.port; + std::thread::scope(|sc| { + for n in first..=total { + sc.spawn(move || { + let c = Server { child: None, port }; + let _ = c.req( + "POST", + &format!("/api/v1/repos/siteadmin/c/pulls/{n}/merge"), + Some(tok), + Some(&json!({"Do": "merge"})), + ); + }); + } + std::thread::sleep(Duration::from_millis(delays[round % delays.len()])); + srv.kill9(); + }); + // restart: `serve` repairs the journal before it accepts anything + srv = Server::start(&data, free_port()); + check(&srv, tok, &data, repo, total); + let conn = scl_forge::db::connect(&data).unwrap(); + aborted = count( + &conn, + "SELECT COUNT(*) FROM merge_intents WHERE state = 'aborted'", + ); + recovered = count( + &conn, + "SELECT COUNT(*) FROM merge_intents WHERE state = 'done' AND recovered = 1", + ); + eprintln!("round {round}: {aborted} intents aborted by recovery, {recovered} finished by recovery so far"); + if round >= 4 && aborted >= 1 && recovered >= 1 { + break; + } + } + assert!( + aborted >= 1 && recovered >= 1, + "the kills never interrupted a merge inside the journal windows ({aborted} aborted, {recovered} recovered): the test proved nothing" + ); + + // whatever is still open merges normally now + for n in 1..=total { + let (_, b) = srv + .req( + "GET", + &format!("/api/v1/repos/siteadmin/c/pulls/{n}"), + Some(tok), + None, + ) + .unwrap(); + if serde_json::from_str::(&b).unwrap()["merged"] != json!(true) { + let (st, b) = srv + .req( + "POST", + &format!("/api/v1/repos/siteadmin/c/pulls/{n}/merge"), + Some(tok), + Some(&json!({"Do": "merge"})), + ) + .unwrap(); + assert!(st == 200 || st == 405, "{st}: {b}"); + } + } + check(&srv, tok, &data, repo, total); + for i in 0..total { + assert!( + git( + &data, + repo, + &["cat-file", "-e", &format!("refs/heads/main:m{i}.txt")] + ) + .is_some(), + "m{i}.txt is on main at the end" + ); + } + drop(srv); + std::fs::remove_dir_all(&data).unwrap(); + assert!(!data.exists()); +} + +fn count(conn: &rusqlite::Connection, sql: &str) -> i64 { + conn.query_row(sql, [], |r| r.get(0)).unwrap() +} + +/// The invariants after a restart. +fn check(srv: &Server, tok: &str, data: &Path, repo: i64, total: i64) { + let conn = scl_forge::db::connect(data).unwrap(); + assert!( + !scl_forge::merges::has_pending(&conn, repo), + "a pending intent survived the restart" + ); + assert_eq!( + git(data, repo, &["for-each-ref", "refs/forge/merge"]).unwrap(), + "", + "a receipt survived the restart" + ); + let main = git(data, repo, &["rev-parse", "refs/heads/main"]).unwrap(); + let mut merged = 0; + let aborted: i64 = conn + .query_row( + "SELECT COUNT(*) FROM merge_intents WHERE state = 'aborted'", + [], + |r| r.get(0), + ) + .unwrap(); + for n in 1..=total { + let (st, b) = srv + .req( + "GET", + &format!("/api/v1/repos/siteadmin/c/pulls/{n}"), + Some(tok), + None, + ) + .unwrap(); + assert_eq!(st, 200, "{b}"); + let pr: Value = serde_json::from_str(&b).unwrap(); + let head = git( + data, + repo, + &["rev-parse", &format!("refs/heads/b{}", n - 1)], + ) + .unwrap(); + let on_main = git(data, repo, &["merge-base", "--is-ancestor", &head, &main]).is_some(); + let file_on_main = git( + data, + repo, + &["cat-file", "-e", &format!("{main}:m{}.txt", n - 1)], + ) + .is_some(); + if pr["merged"] == json!(true) { + merged += 1; + let m = pr["merge_commit_sha"].as_str().unwrap(); + assert_eq!( + git(data, repo, &["rev-list", "--parents", "-n", "1", m]) + .unwrap() + .split(' ') + .count(), + 3, + "PR {n}: a merge commit has two parents" + ); + assert!( + on_main && file_on_main, + "PR {n} is merged, so its commits are on main" + ); + assert_eq!(pr["head"]["sha"], head.as_str()); + } else { + assert!( + !on_main && !file_on_main, + "PR {n} is open but its head is already on main (a landed merge was not finalized)" + ); + } + } + eprintln!("after the restart: {merged} of {total} merged, {aborted} intents aborted so far"); +} + +#[test] +fn a_restart_kills_an_orphaned_git_writer_and_clears_its_lock() { + let data = std::env::temp_dir().join(format!("scl-forge-orphan-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&data); + let _tmp = Tmp(data.clone()); + admin( + &data, + &[ + "user", + "create", + "--username", + "siteadmin", + "--password", + &format!("pw-{}-0123456789", std::process::id()), + "--email", + "a@example.invalid", + "--admin", + "--must-change-password=false", + ], + ); + let tok = admin( + &data, + &[ + "user", + "generate-access-token", + "--username", + "siteadmin", + "--token-name", + "t", + "--scopes", + "all", + "--raw", + ], + ); + let mut srv = Server::start(&data, free_port()); + let (st, body) = srv + .req( + "POST", + "/api/v1/user/repos", + Some(&tok), + Some(&json!({"name": "c", "default_branch": "main", "auto_init": true})), + ) + .unwrap(); + assert_eq!(st, 201, "{body}"); + let repo: i64 = serde_json::from_str::(&body).unwrap()["id"] + .as_i64() + .unwrap(); + let main = git(&data, repo, &["rev-parse", "refs/heads/main"]).unwrap(); + srv.kill9(); + + // What a SIGKILLed server leaves: an `update-ref` that has locked its ref and waits for the + // rest of its script (its stdin stays open because this test holds the pipe). + let canon = std::fs::canonicalize(&data).unwrap(); + let repo_dir = canon.join("repos").join(format!("{repo}.git")); + let mut orphan = Command::new("git") + .env_clear() + .env("PATH", "/usr/bin:/bin:/usr/local/bin:/opt/homebrew/bin") + .env("HOME", &canon) + .arg(format!("--git-dir={}", repo_dir.display())) + .args(["update-ref", "--stdin"]) + .stdin(Stdio::piped()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .unwrap(); + let _orphan_guard = KillOnDrop(orphan.id() as i32); + let mut pipe = orphan.stdin.take().unwrap(); + let zero = "0".repeat(main.len()); + pipe.write_all(format!("start\nupdate refs/heads/zz {main} {zero}\nprepare\n").as_bytes()) + .unwrap(); + pipe.flush().unwrap(); + // and the writer behind an HTTP push: `receive-pack --stateless-rpc .` in the repository, named + // by nothing on its command line, but carrying the transport's FORGE_DATA_DIR + let mut http_orphan = Command::new("git") + .env_clear() + .env("PATH", "/usr/bin:/bin:/usr/local/bin:/opt/homebrew/bin") + .env("HOME", &canon) + .env("FORGE_DATA_DIR", &canon) + .args(["receive-pack", "--stateless-rpc", "."]) + .current_dir(&repo_dir) + .stdin(Stdio::piped()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .unwrap(); + let _http_pipe = http_orphan.stdin.take().unwrap(); + let _http_guard = KillOnDrop(http_orphan.id() as i32); + let lock = repo_dir.join("refs/heads/zz.lock"); + let t0 = Instant::now(); + while !lock.exists() && t0.elapsed() < Duration::from_secs(10) { + std::thread::sleep(Duration::from_millis(20)); + } + assert!(lock.exists(), "the orphan holds its lock"); + assert!(orphan.try_wait().unwrap().is_none(), "and is still running"); + + // the restart's startup cleanup kills it and clears the lock, before recovery looks at anything + let srv = Server::start(&data, free_port()); + let t0 = Instant::now(); + let status = loop { + if let Some(s) = orphan.try_wait().unwrap() { + break s; + } + assert!( + t0.elapsed() < Duration::from_secs(10), + "the orphan was not killed" + ); + std::thread::sleep(Duration::from_millis(20)); + }; + assert!(!status.success()); + let t0 = Instant::now(); + while http_orphan.try_wait().unwrap().is_none() { + assert!( + t0.elapsed() < Duration::from_secs(10), + "the HTTP writer was not killed" + ); + std::thread::sleep(Duration::from_millis(20)); + } + assert!(!lock.exists(), "its lock was removed"); + assert!( + git( + &data, + repo, + &["rev-parse", "--verify", "-q", "refs/heads/zz"] + ) + .is_none(), + "and it published nothing" + ); + drop(pipe); + drop(srv); + std::fs::remove_dir_all(&data).unwrap(); + assert!(!data.exists()); +} diff --git a/crates/forge/tests/e2e_merge.rs b/crates/forge/tests/e2e_merge.rs new file mode 100644 index 00000000..d7970b92 --- /dev/null +++ b/crates/forge/tests/e2e_merge.rs @@ -0,0 +1,646 @@ +//! F4c: merging a pull request, its refusals, concurrency and the crash-recovery journal. +#![cfg(unix)] + +mod common; +use common::*; +use serde_json::{json, Value}; + +struct W { + s: Srv, + auth: String, + id: i64, +} + +fn world() -> W { + let s = Srv::start(); + let auth = token(&s.admin); + let r = s.post( + "/api/v1/user/repos", + &auth, + json!({"name": "w", "default_branch": "main", "auto_init": true}), + ); + assert_eq!(r.status, 201, "{}", r.msg()); + let id = r.json()["id"].as_i64().unwrap(); + W { s, auth, id } +} + +impl W { + fn api(&self, method: &str, path: &str, body: Option) -> Resp { + self.as_(&self.auth, method, path, body) + } + fn as_(&self, auth: &str, method: &str, path: &str, body: Option) -> Resp { + self.s.send( + method, + &format!("/api/v1/repos/siteadmin/w{path}"), + Some(auth), + body.as_ref(), + ) + } + fn git(&self, args: &[&str]) -> String { + self.s.git(self.id, args, None) + } + fn git_try(&self, args: &[&str]) -> Option { + self.s.git_try(self.id, args) + } + fn tip(&self, branch: &str) -> String { + self.git(&["rev-parse", &format!("refs/heads/{branch}")]) + } + /// A commit adding `file` with `text` on `branch` (created from `from` when given). + fn commit(&self, branch: &str, file: &str, text: &str, from: Option<&str>) -> String { + let mut body = json!({"content": b64(text), "message": format!("add {file}"), "branch": from.unwrap_or(branch)}); + if from.is_some() { + body["new_branch"] = json!(branch); + } + let r = self.api("POST", &format!("/contents/{file}"), Some(body)); + assert_eq!(r.status, 201, "{}", r.msg()); + self.tip(branch) + } + fn open(&self, head: &str, base: &str) -> i64 { + let r = self.api( + "POST", + "/pulls", + Some(json!({"title": format!("PR {head}"), "head": head, "base": base})), + ); + assert_eq!(r.status, 201, "{}", r.msg()); + r.json()["number"].as_i64().unwrap() + } + fn merge(&self, n: i64, body: Value) -> Resp { + self.api("POST", &format!("/pulls/{n}/merge"), Some(body)) + } + fn no_leftovers(&self) { + let conn = scl_forge::db::connect(&self.s.dir).unwrap(); + assert!( + !scl_forge::merges::has_pending(&conn, self.id), + "a pending intent is left" + ); + assert_eq!( + self.git(&["for-each-ref", "refs/forge/merge"]), + "", + "a receipt is left" + ); + } +} + +#[test] +fn a_clean_merge_makes_a_two_parent_commit_and_a_merged_pull() { + let w = world(); + let head = w.commit("f", "f.txt", "f\n", Some("main")); + let base = w.tip("main"); + let n = w.open("f", "main"); + assert_eq!(w.api("GET", &format!("/pulls/{n}/merge"), None).status, 404); + let r = w.merge(n, json!({"Do": "merge", "head_commit_id": head})); + assert_eq!( + (r.status, r.raw.len()), + (200, 0), + "{}", + String::from_utf8_lossy(&r.raw) + ); + let m = w.tip("main"); + assert_ne!(m, base); + let parents = w.git(&["rev-list", "--parents", "-n", "1", &m]); + assert_eq!( + parents, + format!("{m} {base} {head}"), + "first parent is the base, second the head" + ); + assert_eq!( + w.git(&["log", "-1", "--format=%s", &m]), + "Merge pull request 'PR f' (#1) from f into main" + ); + assert_eq!(w.git(&["log", "-1", "--format=%an", &m]), "siteadmin"); + assert_eq!(w.git(&["show", &format!("{m}:f.txt")]), "f"); + let pr = w.api("GET", &format!("/pulls/{n}"), None).json(); + assert_eq!( + (pr["merged"].clone(), pr["state"].clone()), + (json!(true), json!("closed")) + ); + assert_eq!(pr["merge_commit_sha"], m.as_str()); + assert_eq!(pr["merged_by"]["login"], "siteadmin"); + assert_eq!(pr["head"]["sha"], head.as_str()); + assert_eq!(pr["mergeable"], false); + assert!(pr["merged_at"].is_string() && pr["closed_at"].is_string()); + assert_eq!(w.api("GET", &format!("/pulls/{n}/merge"), None).status, 204); + assert_eq!( + w.git(&["rev-parse", "refs/forge/pull/1/head"]), + head, + "the head stays retained" + ); + w.no_leftovers(); + // the branch keeps moving; the merged PR does not follow + w.commit("f", "g.txt", "g\n", None); + assert_eq!( + w.api("GET", &format!("/pulls/{n}"), None).json()["head"]["sha"], + head.as_str() + ); + let again = w.merge(n, json!({"Do": "merge"})); + assert_eq!( + (again.status, again.msg().as_str()), + (405, "The PR is already merged") + ); +} + +#[test] +fn refusals_are_the_recorded_ones() { + let w = world(); + w.commit("a", "a.txt", "a\n", Some("main")); + let n = w.open("a", "main"); + let msg = |r: Resp| (r.status, r.msg()); + assert_eq!( + msg(w.merge(n, json!({"Do": "bogus"}))), + (422, "[Do]: In".into()) + ); + assert_eq!(msg(w.merge(n, json!({}))), (422, "[Do]: Required".into())); + for style in ["squash", "rebase", "rebase-merge", "fast-forward-only"] { + let r = w.merge(n, json!({"Do": style})); + assert!( + r.status == 422 && r.msg().contains("merge"), + "{style}: {}", + r.msg() + ); + } + assert_eq!( + msg(w.merge(n, json!({"Do": "merge", "head_commit_id": "0".repeat(40)}))), + (409, "head out of date".into()) + ); + assert_eq!(w.merge(99, json!({"Do": "merge"})).status, 404); + // a read-only collaborator may not merge + let pw = w.s.mk_user("reader"); + let t = token(&w.s.mk_token("reader", &pw, "t", &["all"])); + w.api( + "PUT", + "/collaborators/reader", + Some(json!({"permission": "read"})), + ); + assert_eq!( + w.as_( + &t, + "POST", + &format!("/pulls/{n}/merge"), + Some(json!({"Do": "merge"})) + ) + .status, + 403 + ); + // a conflict names the paths + w.commit("c1", "x.txt", "one\n", Some("main")); + w.commit("c2", "x.txt", "two\n", Some("main")); + let c1 = w.open("c1", "main"); + let c2 = w.open("c2", "main"); + assert_eq!(w.merge(c1, json!({"Do": "merge"})).status, 200); + let r = w.merge(c2, json!({"Do": "merge"})); + assert_eq!((r.status, r.msg()), (409, "Merge conflict: x.txt".into())); + // a head already contained in the base: nothing to merge + let tip = w.tip("main"); + w.git(&["update-ref", "refs/heads/inside", &tip]); + w.commit("tmp", "t.txt", "t\n", Some("main")); + let r = w.api( + "POST", + "/pulls", + Some(json!({"title": "t", "head": "inside", "base": "main"})), + ); + assert_eq!(r.status, 201, "a contained head may be opened: {}", r.msg()); + let inside = r.json()["number"].as_i64().unwrap(); + let r = w.merge(inside, json!({"Do": "merge"})); + assert_eq!( + (r.status, r.msg().as_str()), + (405, "Please try again later") + ); + // unrelated histories + let tree = w.git(&["mktree"]); + let root = { + let out = w.s.git(w.id, &["commit-tree", &tree], Some(b"orphan\n")); + w.git(&["update-ref", "refs/heads/orph", &out]); + out + }; + let _ = root; + let o = w.api( + "POST", + "/pulls", + Some(json!({"title": "o", "head": "orph", "base": "main"})), + ); + assert_eq!(o.status, 201, "{}", o.msg()); + let r = w.merge(o.json()["number"].as_i64().unwrap(), json!({"Do": "merge"})); + assert_eq!( + (r.status, r.msg()), + (409, "Merge conflict: unrelated histories".into()) + ); + // closed and head-deleted pulls are not found + w.commit("d", "d.txt", "d\n", Some("main")); + let d = w.open("d", "main"); + w.api( + "PATCH", + &format!("/pulls/{d}"), + Some(json!({"state": "closed"})), + ); + assert_eq!(w.merge(d, json!({"Do": "merge"})).status, 404); + w.commit("e", "e.txt", "e\n", Some("main")); + let e = w.open("e", "main"); + w.api("DELETE", "/branches/e", None); + assert_eq!(w.merge(e, json!({"Do": "merge"})).status, 404); + w.no_leftovers(); +} + +#[test] +fn required_checks_are_read_at_merge_time_and_a_merge_ignores_enable_push() { + let w = world(); + let head = w.commit("r", "r.txt", "r\n", Some("main")); + let n = w.open("r", "main"); + // the rule appears after the PR exists: the merge sees it + let rule = w.api("POST", "/branch_protections", Some(json!({"branch_name": "main", "enable_push": false, "enable_status_check": true, "status_check_contexts": ["ci/test"]}))); + assert_eq!(rule.status, 201); + let refused = |w: &W| { + let r = w.merge(n, json!({"Do": "merge"})); + assert_eq!(r.status, 405, "{}", r.msg()); + assert!( + r.msg() + .contains("Not all required status checks successful"), + "{}", + r.msg() + ); + }; + refused(&w); + for (state, ctx, sha) in [ + ("success", "ci/other", head.clone()), + ("failure", "ci/test", head.clone()), + ("pending", "ci/test", head.clone()), + ("warning", "ci/test", head.clone()), + ("error", "ci/test", head.clone()), + ("success", "ci/test", w.tip("main")), + ] { + let r = w.api( + "POST", + &format!("/statuses/{sha}"), + Some(json!({"state": state, "context": ctx})), + ); + assert_eq!(r.status, 201); + refused(&w); + } + w.api( + "POST", + &format!("/statuses/{head}"), + Some(json!({"state": "skipped", "context": "ci/test"})), + ); + let r = w.merge(n, json!({"Do": "merge"})); + assert_eq!( + r.status, + 200, + "skipped satisfies the check, and enable_push=false does not block a merge: {}", + r.msg() + ); + // the protected branch still refuses a direct contents write + let denied = w.api( + "POST", + "/contents/z.txt", + Some(json!({"content": b64("z"), "message": "m", "branch": "main"})), + ); + assert_eq!(denied.status, 403); + w.no_leftovers(); +} + +#[test] +fn concurrent_merges_into_one_base_all_land_and_nothing_is_left_over() { + let w = world(); + let n = 6; + for i in 0..n { + w.commit(&format!("m{i}"), &format!("m{i}.txt"), "x\n", Some("main")); + w.open(&format!("m{i}"), "main"); + } + let results: Vec<(u16, String)> = std::thread::scope(|sc| { + let hs: Vec<_> = (1..=n) + .map(|i| { + let w = &w; + sc.spawn(move || { + let r = w.merge(i, json!({"Do": "merge"})); + (r.status, String::from_utf8_lossy(&r.raw).to_string()) + }) + }) + .collect(); + hs.into_iter().map(|h| h.join().unwrap()).collect() + }); + for (status, body) in &results { + assert!(matches!(*status, 200 | 503), "unexpected {status}: {body}"); + } + // anything the gate timed out on merges afterwards + for i in 1..=n { + let p = w.api("GET", &format!("/pulls/{i}"), None).json(); + if p["merged"] != json!(true) { + assert_eq!(w.merge(i, json!({"Do": "merge"})).status, 200); + } + } + for i in 0..n { + assert!( + w.git_try(&["cat-file", "-e", &format!("refs/heads/main:m{i}.txt")]) + .is_some(), + "m{i}.txt is on main" + ); + } + assert_eq!( + w.git(&["rev-list", "--merges", "--count", "main"]), + n.to_string() + ); + w.no_leftovers(); +} + +#[test] +fn a_writer_moving_the_base_during_merges_loses_nothing() { + let w = world(); + let n = 4; + for i in 0..n { + w.commit(&format!("m{i}"), &format!("m{i}.txt"), "x\n", Some("main")); + w.open(&format!("m{i}"), "main"); + } + let landed = std::sync::Mutex::new(Vec::::new()); + std::thread::scope(|sc| { + // an independent writer advances main with plumbing, compare-and-swap, like a push would + sc.spawn(|| { + // twelve moves, spaced so that a merge round (a handful of git processes) can win between two + for i in 1..=12 { + let old = w.tip("main"); + let tree = w.git(&["rev-parse", &format!("{old}^{{tree}}")]); + let c = w.s.git( + w.id, + &["commit-tree", &tree, "-p", &old], + Some(format!("w{i}\n").as_bytes()), + ); + if w.git_try(&["update-ref", "refs/heads/main", &c, &old]) + .is_some() + { + landed.lock().unwrap().push(c); + } + std::thread::sleep(std::time::Duration::from_millis(400)); + } + }); + for i in 1..=n { + let mut ok = false; + for _ in 0..40 { + let r = w.merge(i, json!({"Do": "merge"})); + if r.status == 200 { + ok = true; + break; + } + assert!(matches!(r.status, 405 | 503), "{}: {}", r.status, r.msg()); + } + assert!(ok, "merge {i} never landed"); + } + }); + let main = w.tip("main"); + for c in landed.lock().unwrap().iter() { + assert!( + w.git_try(&["merge-base", "--is-ancestor", c, &main]) + .is_some(), + "writer commit {c} was lost" + ); + } + for i in 0..n { + assert!(w + .git_try(&["cat-file", "-e", &format!("{main}:m{i}.txt")]) + .is_some()); + } + w.no_leftovers(); +} + +// ---- crash recovery: the journal decides, not guesswork ---- + +/// How far a hand-built merge got before the "crash". +#[derive(Clone, Copy, PartialEq)] +enum Crash { + /// The intent row exists; the base never moved. + NotPublished, + /// The base moved; the `landed` marker was not written. + PublishedOnly, + /// The base moved and the marker exists. + PublishedMarked, +} + +/// Do by hand what the handler does, up to the crash point. +fn crashed_merge(w: &W, n: i64, head: &str, how: Crash) -> (i64, String, String) { + let base = w.tip("main"); + let tree = w.git(&["merge-tree", "--write-tree", &base, head]); + let tree = tree.lines().next().unwrap().to_string(); + let merge = w.s.git( + w.id, + &["commit-tree", &tree, "-p", &base, "-p", head], + Some(b"Merge by hand\n"), + ); + let conn = scl_forge::db::connect(&w.s.dir).unwrap(); + let uid: i64 = conn + .query_row("SELECT id FROM users WHERE login = 'siteadmin'", [], |r| { + r.get(0) + }) + .unwrap(); + conn.execute( + "INSERT INTO merge_intents (repo_id, pull_number, base_ref, head_ref, base_oid, head_oid, merge_oid, merge_base, changed_files, additions, deletions, merger_id, merged_unix, state, created_unix) \ + VALUES (?1, ?2, 'main', 'f', ?3, ?4, ?5, ?6, 1, 1, 0, ?7, 12345, 'pending', 12345)", + rusqlite::params![w.id, n, base, head, merge, base, uid], + ) + .unwrap(); + let id = conn.last_insert_rowid(); + if how != Crash::NotPublished { + let script = format!( + "start\nverify refs/heads/f {head}\nupdate refs/heads/main {merge} {base}\nprepare\ncommit\n" + ); + w.s.git(w.id, &["update-ref", "--stdin"], Some(script.as_bytes())); + } + if how == Crash::PublishedMarked { + let zero = "0".repeat(40); + w.git(&[ + "update-ref", + &format!("refs/forge/merge/{id}/landed"), + &merge, + &zero, + ]); + } + (id, base, merge) +} + +fn recover(w: &W) { + let conn = scl_forge::db::connect(&w.s.dir).unwrap(); + let dir = w.s.dir.join("repos").join(format!("{}.git", w.id)); + scl_forge::merges::recover_locked(&conn, &w.s.runner, &dir, w.id).unwrap(); + scl_forge::merges::drop_stale_receipts(&conn, &w.s.runner, &dir, w.id).unwrap(); +} + +fn pull_state(w: &W, n: i64) -> String { + let conn = scl_forge::db::connect(&w.s.dir).unwrap(); + conn.query_row("SELECT state FROM pulls WHERE number = ?1", [n], |r| { + r.get(0) + }) + .unwrap() +} + +fn intent_state(w: &W, id: i64) -> String { + let conn = scl_forge::db::connect(&w.s.dir).unwrap(); + conn.query_row("SELECT state FROM merge_intents WHERE id = ?1", [id], |r| { + r.get(0) + }) + .unwrap() +} + +#[test] +fn a_merge_that_landed_before_the_crash_is_finalized_with_every_field() { + let w = world(); + let head = w.commit("f", "f.txt", "f\n", Some("main")); + let n = w.open("f", "main"); + let (_, _, merge) = crashed_merge(&w, n, &head, Crash::PublishedMarked); + assert_eq!(pull_state(&w, n), "open", "the database does not know yet"); + recover(&w); + let pr = w.api("GET", &format!("/pulls/{n}"), None).json(); + assert_eq!(pr["merged"], true); + assert_eq!(pr["merge_commit_sha"], merge.as_str()); + assert_eq!(pr["merged_by"]["login"], "siteadmin"); + assert_eq!(pr["head"]["sha"], head.as_str()); + assert_eq!(pr["changed_files"], 1); + w.no_leftovers(); + assert_eq!( + w.merge(n, json!({"Do": "merge"})).status, + 405, + "and a retry sees the original outcome" + ); +} + +#[test] +fn a_crash_between_the_base_moving_and_its_marker_is_recognized_by_the_base_history() { + let w = world(); + let head = w.commit("f", "f.txt", "f\n", Some("main")); + let n = w.open("f", "main"); + let (id, _, merge) = crashed_merge(&w, n, &head, Crash::PublishedOnly); + assert_eq!(w.git(&["for-each-ref", "refs/forge/merge"]), ""); + recover(&w); + assert_eq!(intent_state(&w, id), "done"); + let pr = w.api("GET", &format!("/pulls/{n}"), None).json(); + assert_eq!( + (pr["merged"].clone(), pr["merge_commit_sha"].clone()), + (json!(true), json!(merge)) + ); + w.no_leftovers(); +} + +#[test] +fn a_landed_merge_with_its_marker_is_still_merged_after_the_base_was_force_pushed_away() { + let w = world(); + let head = w.commit("f", "f.txt", "f\n", Some("main")); + let n = w.open("f", "main"); + let (_, base, merge) = crashed_merge(&w, n, &head, Crash::PublishedMarked); + // an unprotected base may be rewritten before recovery runs: the marker, not ancestry, decides + w.git(&["update-ref", "refs/heads/main", &base]); + assert!(w + .git_try(&["merge-base", "--is-ancestor", &merge, "main"]) + .is_none()); + recover(&w); + let pr = w.api("GET", &format!("/pulls/{n}"), None).json(); + assert_eq!(pr["merged"], true); + assert_eq!(pr["merge_commit_sha"], merge.as_str()); + w.no_leftovers(); +} + +#[test] +fn an_intent_that_never_published_is_aborted_and_the_pull_can_merge_again() { + let w = world(); + let head = w.commit("f", "f.txt", "f\n", Some("main")); + let n = w.open("f", "main"); + let (id, base, _) = crashed_merge(&w, n, &head, Crash::NotPublished); + assert_eq!(w.tip("main"), base, "nothing was published"); + recover(&w); + assert_eq!(intent_state(&w, id), "aborted"); + assert_eq!(pull_state(&w, n), "open"); + assert_eq!(w.merge(n, json!({"Do": "merge"})).status, 200); + w.no_leftovers(); +} + +#[test] +fn a_git_writer_still_holding_the_base_keeps_the_intent_pending() { + let w = world(); + let head = w.commit("f", "f.txt", "f\n", Some("main")); + let n = w.open("f", "main"); + let (id, _, _) = crashed_merge(&w, n, &head, Crash::NotPublished); + // an orphan `update-ref` from before a crash would hold this lock while it finishes + let lock = + w.s.dir + .join("repos") + .join(format!("{}.git", w.id)) + .join("refs/heads/main.lock"); + std::fs::write(&lock, b"").unwrap(); + recover(&w); + assert_eq!( + intent_state(&w, id), + "pending", + "not judged while the base is held" + ); + // and nothing may change the pull request meanwhile + let r = w.api( + "PATCH", + &format!("/pulls/{n}"), + Some(json!({"state": "closed"})), + ); + assert_eq!(r.status, 503, "{}", r.msg()); + assert_eq!(pull_state(&w, n), "open"); + std::fs::remove_file(&lock).unwrap(); + recover(&w); + assert_eq!(intent_state(&w, id), "aborted"); +} + +#[test] +fn a_close_cannot_overtake_a_landed_merge() { + let w = world(); + let head = w.commit("f", "f.txt", "f\n", Some("main")); + let n = w.open("f", "main"); + crashed_merge(&w, n, &head, Crash::PublishedMarked); + // the pull request is still open in the database; the PATCH settles the merge first + let r = w.api( + "PATCH", + &format!("/pulls/{n}"), + Some(json!({"state": "closed"})), + ); + assert_eq!(r.status, 422, "{}", r.msg()); + assert_eq!(pull_state(&w, n), "merged"); + w.no_leftovers(); +} + +#[test] +fn a_read_settles_an_unfinished_merge() { + let w = world(); + let head = w.commit("f", "f.txt", "f\n", Some("main")); + let n = w.open("f", "main"); + let (_, _, merge) = crashed_merge(&w, n, &head, Crash::PublishedOnly); + // the next read of the pull settles it (the gate is free) + let pr = w.api("GET", &format!("/pulls/{n}"), None).json(); + assert_eq!(pr["merged"], true, "{pr}"); + assert_eq!(pr["merge_commit_sha"], merge.as_str()); + w.no_leftovers(); +} + +#[test] +fn stray_markers_are_removed_by_the_startup_repair() { + let w = world(); + let tip = w.tip("main"); + w.git(&["update-ref", "refs/forge/merge/77/landed", &tip]); + recover(&w); + assert_eq!(w.git(&["for-each-ref", "refs/forge/merge"]), ""); +} + +#[test] +fn a_hostile_title_cannot_add_lines_to_the_merge_commit() { + let w = world(); + w.commit("t", "t.txt", "t\n", Some("main")); + let r = w.api( + "POST", + "/pulls", + Some(json!({"title": "x\n\nSigned-off-by: Eve \u{1}", "head": "t", "base": "main"})), + ); + assert_eq!(r.status, 201, "{}", r.msg()); + assert_eq!(w.merge(1, json!({"Do": "merge"})).status, 200); + let m = w.tip("main"); + let body = w.git(&["log", "-1", "--format=%B", &m]); + assert_eq!(body.lines().count(), 1, "{body:?}"); + assert!(!body.contains("\n"), "{body:?}"); +} + +#[test] +fn the_merge_status_probe_settles_an_unfinished_merge() { + let w = world(); + let head = w.commit("f", "f.txt", "f\n", Some("main")); + let n = w.open("f", "main"); + crashed_merge(&w, n, &head, Crash::PublishedMarked); + assert_eq!(w.api("GET", &format!("/pulls/{n}/merge"), None).status, 204); + w.no_leftovers(); +} diff --git a/crates/forge/tests/e2e_protection.rs b/crates/forge/tests/e2e_protection.rs index f10c16ca..0eb561f8 100644 --- a/crates/forge/tests/e2e_protection.rs +++ b/crates/forge/tests/e2e_protection.rs @@ -632,3 +632,142 @@ fn combined_status_does_not_deadlock_against_exclusive_repository_operations() { ); }); } + +// ---- the bypass matrix: every write path against every rule setting and role ---- + +impl W { + fn as_(&self, auth: &str, method: &str, path: &str, body: Option) -> Resp { + self.s.send( + method, + &format!("/api/v1/repos/siteadmin/w{path}"), + Some(auth), + body.as_ref(), + ) + } +} + +/// One cell group of the matrix: a protected non-default branch `rel` with `enable_push = push`, +/// and `who` (an administrator or a write collaborator) trying each way to move or remove it. +fn matrix(push: bool, who: &str) { + let w = world(); + let c = Client::new(&format!("matrix-{push}-{who}")); + let auth = if who == "admin" { + w.auth.clone() + } else { + let pw = w.s.mk_user("writer"); + let t = token(&w.s.mk_token("writer", &pw, "t", &["all"])); + let r = w.api( + "PUT", + "/collaborators/writer", + Some(json!({"permission": "write"})), + ); + assert_eq!(r.status, 204); + t + }; + let work = c.dir.join("work"); + c.ok( + &c.dir, + Some(&w.auth), + &["clone", "-q", &w.url(), work.to_str().unwrap()], + ); + c.ok(&work, None, &["checkout", "-q", "-b", "rel"]); + c.ok(&work, Some(&w.auth), &["push", "-q", "origin", "rel"]); + w.rule("rel", push); + let tag = format!("[push={push} who={who}]"); + + // a pull request into `rel`, opened before any of the attempts + let r = w.api( + "POST", + "/contents/t.txt", + Some( + json!({"content": b64s("t\n"), "message": "m", "branch": "rel", "new_branch": "topic"}), + ), + ); + assert_eq!(r.status, 201, "{tag} {}", r.msg()); + let pr = w.as_( + &auth, + "POST", + "/pulls", + Some(json!({"title": "t", "head": "topic", "base": "rel"})), + ); + assert_eq!(pr.status, 201, "{tag} {}", pr.msg()); + let n = pr.json()["number"].as_i64().unwrap(); + + let write = |path: &str| { + w.as_( + &auth, + "POST", + &format!("/contents/{path}"), + Some(json!({"content": b64s("x\n"), "message": "m", "branch": "rel"})), + ) + .status + }; + // direct pushes + c.commit(&work, "a.txt", "a\n"); + let (ff_ok, _) = c.git_in(&work, Some(&auth), &["push", "origin", "rel"]); + assert_eq!(ff_ok, push, "{tag} fast-forward push"); + let tip = w.git(&["rev-parse", "refs/heads/rel"]); + if push { + assert_ne!(tip, w.git(&["rev-parse", "refs/heads/main"])); + } + c.ok(&work, None, &["reset", "-q", "--hard", "HEAD~1"]); + c.commit(&work, "b.txt", "b\n"); + let (force_ok, _) = c.git_in(&work, Some(&auth), &["push", "-f", "origin", "rel"]); + assert!(!force_ok, "{tag} force push"); + let (del_ok, _) = c.git_in(&work, Some(&auth), &["push", "origin", ":rel"]); + assert!(!del_ok, "{tag} push delete"); + assert_eq!( + w.git(&["rev-parse", "refs/heads/rel"]), + tip, + "{tag} neither refused push moved rel" + ); + // the API + assert_eq!(write("c.txt") == 201, push, "{tag} contents write"); + assert_eq!( + w.as_(&auth, "DELETE", "/branches/rel", None).status, + 403, + "{tag} branch delete" + ); + let nb = w.as_(&auth, "POST", "/contents/d.txt", Some(json!({"content": b64s("d\n"), "message": "m", "branch": "rel", "new_branch": "off-rel"}))); + assert_eq!( + nb.status, + 201, + "{tag} a new branch cut from a protected one is allowed: {}", + nb.msg() + ); + // a merge is the one way in when pushes are off; it works for both settings + let tip = w.git(&["rev-parse", "refs/heads/rel"]); + let m = w.as_( + &auth, + "POST", + &format!("/pulls/{n}/merge"), + Some(json!({"Do": "merge"})), + ); + assert_eq!(m.status, 200, "{tag} merge: {}", m.msg()); + assert_ne!( + w.git(&["rev-parse", "refs/heads/rel"]), + tip, + "{tag} the merge moved rel" + ); + assert_eq!(w.git(&["for-each-ref", "refs/forge/merge"]), "", "{tag}"); +} + +#[test] +fn bypass_matrix_pushes_off_administrator() { + matrix(false, "admin"); +} + +#[test] +fn bypass_matrix_pushes_off_writer() { + matrix(false, "writer"); +} + +#[test] +fn bypass_matrix_pushes_on_administrator() { + matrix(true, "admin"); +} + +#[test] +fn bypass_matrix_pushes_on_writer() { + matrix(true, "writer"); +} diff --git a/docs/THREAT-MODEL.md b/docs/THREAT-MODEL.md index e495f7ed..24723577 100644 --- a/docs/THREAT-MODEL.md +++ b/docs/THREAT-MODEL.md @@ -1050,6 +1050,53 @@ and the `pulls` unit tests. - **Cost.** Mergeability and diff statistics run `git` per PR on a cache miss; a page lists at most 50 PRs. A repository with many open PRs and a huge history makes a list slow, not wrong; the runner's concurrency and timeouts bound it. +### Merge, recovery and the protection matrix (F4c) + +Threat-model pass #2, last part. Tests: `crates/forge/tests/e2e_merge.rs`, `e2e_crash.rs` (a real `sc-forge serve` killed with +SIGKILL during merges, restarted and checked), and the `bypass_matrix_*` tests in `e2e_protection.rs`. + +- **The matrix.** Against a protected non-default branch, for `enable_push` off and on and for an administrator and a write + collaborator: a fast-forward push and a contents write succeed only with pushes on; a forced push, a push deletion and the + API branch delete never do; a new branch cut from the protected one is allowed; a pull request merge into it works under both + settings and is the only way in when pushes are off. Receive-pack, the contents API and the branch API go through the + same `policy` functions on rules read from SQLite; the merge deliberately does not (it ignores `enable_push`) and reads the same rules + only for the required status contexts. +- **Required checks gate the merge, not a push** (recorded, W#8). They are read for the head commit under the write gate + immediately before the intent is written; a status posted by a repository writer for that commit counts (statuses are not + attestations). There is no administrator override. A rule that requires checks but lists none refuses every merge. +- **A merge cannot publish a stale decision.** Rule changes, PR changes and merges serialize on the gate and every merge round + re-reads the PR, rules, both tips and the statuses. The head is `verify`d inside the same ref transaction that moves the base, + so "the head I checked is the head I merged" is atomic; a head that moved answers 409 and is never substituted. +- **Crash consistency.** Git and SQLite cannot commit together, and Git's files backend commits several refs one after another, so + the protocol uses only single-ref atomic steps and evidence that survives a crash between them: the journal row precedes the + base compare-and-swap (one ref, with the head `verify`d in the same transaction), the `landed` marker follows it, and recovery + trusts, in order, the marker, the merge commit being in the base's history, and a lock file on the base (a live writer holding it + keeps the intent pending; git children that outlived a SIGKILLed server are killed, and their locks removed, by the startup + cleanup before any recovery runs, `startup_cleanup`/`a_restart_kills_an_orphaned_git_writer_and_clears_its_lock`); only a definite "no" aborts, and a git error leaves the intent pending. Recovery also + runs before any pull-request mutation or sync, so a close cannot overtake a landed merge, and finalization never runs against a + pull request that is not open (`a_close_cannot_overtake_a_landed_merge`, + `a_git_writer_still_holding_the_base_keeps_the_intent_pending`, + `a_crash_between_the_base_moving_and_its_marker_is_recognized_by_the_base_history`, + `a_landed_merge_with_its_marker_is_still_merged_after_the_base_was_force_pushed_away`). `kill_nine_during_merges_leaves_a_consistent_forge` + SIGKILLs a real server inside the merge windows (a test-only environment knob, `SC_FORGE_MERGE_PAUSE_MS`, stretches the two windows + that matter), restarts it, and asserts after every restart: no pending intent, no marker, every pull request merged with a + two-parent commit on the base or open with its head not on the base; it keeps going until the database shows an intent aborted + by recovery and one finished by recovery. +- **Operator note.** A merge's `update-ref` killed by the runner's deadline can leave `refs/heads/.lock` and `refs/heads/.lock` + (the transaction locks the verified head as well as the base). Nothing removes a lock at run time, because a lock cannot be told from + a live push's: the intent stays pending while the base is locked, pushes and merges to those branches wait, and the next restart's + startup cleanup removes every `refs/**/*.lock` once no matching git process is left alive (it leaves them, and says so, if one cannot + be killed). That cleanup passes over the process table repeatedly until it finds nothing new (an `http-backend` can fork a + `receive-pack` while it is being killed), and matches git processes by command line (`--git-dir=/repos/...`) or by the + `FORGE_DATA_DIR` the HTTP transport sets, on whole tokens (another instance's `/srv/forge-secondary` is not `/srv/forge`; on Linux a + space inside a value is not a boundary). Migration 5 was rewritten while the feature was unreleased (no database at version 5 exists outside + CI); a later change to it needs a new migration. +- **Known windows.** A rule created while a push is between `pre-receive` and its ref update is not stopped (the push does not + take the gate). A merge computes its tree before it takes the base lock in git, so a base that moves under it only costs a retry, + not a wrong merge (the compare-and-swap refuses). Objects of aborted or lost merges stay unreferenced until F5's gc. +- **Identity.** The merge commit is authored and committed by the authenticated user (name and stored email, or + `@noreply.localhost`), as for contents writes; a client cannot choose it. + ## Reporting Security reports go to **toni@git-agentic.com** or GitHub private vulnerability diff --git a/docs/superpowers/plans/2026-10-24-p38-f4-pulls-statuses-merge-protection.md b/docs/superpowers/plans/2026-10-24-p38-f4-pulls-statuses-merge-protection.md index 87a3bb3f..ee05a815 100644 --- a/docs/superpowers/plans/2026-10-24-p38-f4-pulls-statuses-merge-protection.md +++ b/docs/superpowers/plans/2026-10-24-p38-f4-pulls-statuses-merge-protection.md @@ -317,3 +317,16 @@ and the base`; PR object keys `additions`, `deletions`, `merge_base`, `comments` 200; other users' PENDING reviews are hidden from lists but counted in `X-Total-Count`; `official` follows the recorded transitions (F#28 vs F#33), computed at read time; list `sort=oldest|newest|recentupdate` and `Link` headers (P#0-4, P#16-18). + +## Amendment after the F4c diff review (Sol and Codex; supersedes R2 where they differ) + +R2 put the receipt ref into the same `update-ref --stdin` transaction as the base move. Git's files backend commits the refs of a +transaction one after another, so a crash or a killed helper between two of them leaves the base moved without its receipt (or the +reverse): the receipt predicate alone then aborts a landed merge or finalizes one that never landed. The as-built protocol uses only +atomic single-ref steps: the transaction `verify`s the head and moves the base (one ref), then `refs/forge/merge//landed` is +written; recovery trusts the marker, then the merge commit being an ancestor of the base, and keeps an intent pending while a lock +file shows a git writer holding the base; startup first kills orphaned git children of the data directory and removes their ref locks +(`orphans.rs`), so nothing can publish after recovery has judged. Also changed: every retry persists all +recomputed facts (R3 parity), recovery runs before pull-request mutations and syncs and finalization refuses to consume a pending +intent against a PR that is not open, an unreadable git answer never aborts an intent, and the kill -9 test must witness an aborted +and a recovered intent. diff --git a/docs/superpowers/specs/2026-10-22-p38-forge-design.md b/docs/superpowers/specs/2026-10-22-p38-forge-design.md index 22623a0b..43b1dc43 100644 --- a/docs/superpowers/specs/2026-10-22-p38-forge-design.md +++ b/docs/superpowers/specs/2026-10-22-p38-forge-design.md @@ -905,6 +905,52 @@ F3b is the write half of the content API (plan: `docs/superpowers/plans/2026-10- deterministic interfering writer); the F4 `refs/forge/pull//head` refresh for a contents write to a PR branch; portable-name collision rules; a disk reservation sized to the exact rebuilt trees. +### F4c decisions (as built) + +F4c adds migration 5 (`merge_intents`) and `POST /pulls/{n}/merge`; the gate now replays G, T, U, O and Q completely (the F4 +slices together leave only K's two deferred org-listing exchanges pending). The merge below replaces the corresponding +steps of "Merge" and "Crash recovery" above where they differ (plan decisions R1-R5 after the Opus and Sol review). + +- **Order, all under the repository's write gate.** Settle any pending intent; re-read the PR (merged: 405 `The PR is + already merged`; closed: 404), both branch tips (a missing one: 404), the protection rules and the required statuses; refuse a + stale `head_commit_id` (409 `head out of date`), a head already contained in the base (405 `Please try again later`), unmet + required contexts (405 `not allowed to merge [reason: Not all required status checks successful]`; a rule that requires + checks but names none refuses), unrelated histories (409 `Merge conflict: unrelated histories`) and conflicts (409 `Merge + conflict: `, from `merge-tree --write-tree --name-only -z`). Rule changes, PR changes and merges are all gate-ordered, so + the check cannot be overtaken; statuses do not take the gate and count as of the read. Only `Do: merge` is supported (other + styles 422 `only Do=merge is supported`, unknown 422 `[Do]: In`, missing `[Do]: Required`). A merge ignores `enable_push`. +- **The journal.** A `pending` `merge_intents` row (PR, branch names, base/head/merge ids, merge base, the three stats, the merger, + the merge time) is written before anything is published, and rewritten in full before each retry. The merge + commit is `commit-tree -p base -p head` on the `merge-tree` tree, authored and committed by the merging user at the intent's + time, message `Merge pull request '' (#<n>) from <head> into <base>`. +- **Publication** is one `update-ref --stdin` transaction that `verify`s the head and moves the base compare-and-swap (one ref + changes, so it is atomic on Git's files backend; a transaction that also created a marker would not be, since the files backend + commits refs one after another), followed by the marker `refs/forge/merge/<id>/landed`. Whether the transaction committed is asked of + Git (marker, then ancestry), never read off an exit status. Not published: a moved head answers 409 `head out of date` (never + silently substituted), a moved base rebuilds on the new tip (3 rounds, then 405 `Please try again later`); every retry + rewrites the whole intent row (merge commit, merge base, stats, time) first. +- **Recovery** judges a pending intent in this order: marker present = landed (even if the base was force-pushed away + afterwards); else the merge commit being an ancestor of the base = landed (the crash fell between the transaction and the + marker; at startup nobody can have rewritten the base yet); else, if a lock file shows a git writer holding the base ref, the + intent stays pending (at run time that is a live writer); else aborted. Git children run in their own process groups and outlive a + SIGKILLed server, so startup, which holds the exclusive data-directory lock, first kills every git process that names a path under + this data directory's `repos/` and removes the ref locks they left (`orphans.rs`), before any recovery runs; an orphan therefore + cannot publish after recovery has judged. An answer git cannot give never + aborts. Landed means finalized from the row (the PR merged with every field a normal merge writes, the head retained, the marker + dropped); finalization refuses to run against a PR that is not open (the evidence is kept pending). Recovery runs before every + gated pull-request operation and sync (a close, a retarget or a head-deletion sync cannot overtake a landed merge; if the + intent cannot be settled the operation answers 503), at startup before the listener is bound, and inline on a read when the gate + is free (never waited for). Stray markers are removed at startup. The only unresolvable case is a crash between the transaction + and the marker combined with a rewrite of the base before recovery, which would read as "not landed" (the PR stays open, the base + does not contain the merge: consistent, if surprising). (The F0 text recovered by ancestry alone; an earlier draft of this section + put a receipt into the transaction, which the files backend cannot make atomic.) +- **Merged pulls** read `state: closed, merged: true`, `merge_commit_sha`, `merged_by` (nested account form), `head.sha` and the + stats frozen at merge, `base.sha` the merge commit, `mergeable: false`. `GET /pulls/{n}/merge` is 204. `delete_branch_after_merge` + is ignored. +- **Not built.** Squash, rebase and fast-forward merges; merge messages and `force_merge` options; merging a PR whose base branch is + gone (404). Review state does not gate a merge (only required status contexts do). Under a base that moves faster than a merge + round (about a second with durable fsync settings) a merge can starve and answer 405; the client retries. + ### F4b decisions (as built) F4b adds migration 4 (`pulls`, `mergeability_cache`, `reviews`, `issue_comments`) and gates F, O (the merge-dependent exchanges