graft-mirror: pr-batch never adopts a batch PR or branch it did not create (WI-2585)

changes/WI-2585.md

@@ -0,0 +1 @@
+- `graft-mirror pr-batch` no longer adopts a PR it did not open. A new batch skips every number whose `graft/batch-<n>` branch or PR already exists on GitHub, and a PR is adopted only when it is open at the batch's own head; any other is refused and the batch rejected, so its lands stay pending. Previously a reset mirror DB could settle new lands as merged in an old merged PR. Repair procedure: `docs/runbooks/batch-adoption-repair.md`.

crates/graft-mirror/src/batch/batcher.rs

@@ -34,6 +34,9 @@ use super::types::{
 /// An open batch's detail when the line has nothing past its tip.
 const NOTHING_NEW: &str = "waiting for review; no new lands to append";
 
+/// How many taken batch numbers a new batch skips before it gives up.
+const MAX_TAKEN_PROBES: u64 = 32;
+
 /// Batches one integration line into squashed GitHub PRs.
 pub struct Batcher<'a, A, F> {
     api: &'a A,
@@ -991,7 +994,7 @@ where
             }
         }
 
-        let number = self.db.next_batch_number()?;
+        let number = self.free_number().await?;
         let created_ms = (self.clock)();
         let branch = format!("{BRANCH_PREFIX}{number}");
         let line = clean_text(&self.config.line, FIELD_CHARS, false);
@@ -1595,11 +1598,58 @@ where
         Ok(())
     }
 
-    /// Opens the PR, or adopts the one already on the branch. Never merges
-    /// or queues it.
+    /// The next batch number whose branch graft is free to use: past the
+    /// database's last batch, and past any `graft/batch-<n>` that already
+    /// has a branch or a PR on GitHub (a reset database, trial leftovers).
+    /// Such a branch or PR is never reused or adopted (WI-2585).
+    async fn free_number(&mut self) -> Result<u64, MirrorError> {
+        let first = self.db.next_batch_number()?;
+        for number in first..first + MAX_TAKEN_PROBES {
+            let branch = format!("{BRANCH_PREFIX}{number}");
+            let there = self.mirror.ls_remote(&[branch.as_str()]).await?;
+            if there.iter().any(|(_, sha)| sha.is_some()) {
+                tracing::warn!(branch = %branch, "mirror batch: branch exists; skipping its number");
+                continue;
+            }
+            if let Some(pull) = self.forge.find_pull(&branch).await? {
+                tracing::warn!(
+                    branch = %branch, pr = pull.number,
+                    "mirror batch: branch has a PR graft did not open; skipping its number"
+                );
+                continue;
+            }
+            return Ok(number);
+        }
+        Err(config_err(format!(
+            "batch numbers {first} to {} all have a branch or PR on GitHub; \
+             remove stale {BRANCH_PREFIX}* branches or PRs",
+            first + MAX_TAKEN_PROBES - 1
+        )))
+    }
+
+    /// Opens the PR, or adopts the one this batch opened before a crash: open,
+    /// and at the batch's own head. Any other PR on the branch (merged,
+    /// closed, or at other commits) was not opened by this batch; it is never
+    /// adopted, and the batch is rejected so its lands stay pending.
     async fn open(&mut self, rec: &mut BatchRecord) -> Result<(), MirrorError> {
         let pull = match self.forge.find_pull(&rec.branch).await? {
-            Some(pull) => pull,
+            Some(pull) if pull.open && !pull.merged && pull.head_sha == rec.squash_sha.to_hex() => {
+                pull
+            }
+            // This includes the batch's own PR when a run crashed after
+            // create_pull and the PR was merged or closed before the retry:
+            // the batch is rejected and its lands are batched again. If that
+            // PR was merged, the next batch repeats changes main already has
+            // and the operator closes it; lands are never silently lost.
+            Some(pull) => {
+                return self.reject(
+                    rec,
+                    format!(
+                        "branch {} already has PR #{} that this batch did not open; not adopted",
+                        rec.branch, pull.number
+                    ),
+                );
+            }
             None => {
                 let new = crate::github::NewPull {
                     title: rec.title.clone(),

crates/graft-mirror/tests/batch.rs

@@ -47,6 +47,9 @@ struct FakeGitHub {
     merging: Mutex<Vec<u64>>,
     /// A branch update made immediately after the next PR read.
     race_push: Mutex<Option<(usize, String, String)>>,
+    /// `(branch, skip)`: after `skip` PR lookups for `branch` answer as
+    /// usual, a merged PR graft did not open appears on it.
+    foreign_pr: Mutex<Option<(String, usize)>>,
 }
 
 impl FakeGitHub {
@@ -58,6 +61,7 @@ impl FakeGitHub {
             comments: Mutex::new(Vec::new()),
             merging: Mutex::new(Vec::new()),
             race_push: Mutex::new(None),
+            foreign_pr: Mutex::new(None),
         }
     }
 
@@ -187,7 +191,30 @@ impl FakeGitHub {
 
 impl Forge for FakeGitHub {
     async fn find_pull(&self, branch: &str) -> Result<Option<Pull>, ForgeError> {
-        let prs = self.prs.lock().unwrap();
+        let mut prs = self.prs.lock().unwrap();
+        let mut foreign = self.foreign_pr.lock().unwrap();
+        if let Some((b, skip)) = foreign.as_mut()
+            && b == branch
+        {
+            if *skip == 0 {
+                let main = support::branch(&self.upstream, "main").unwrap();
+                let number = prs.len() as u64 + 1;
+                prs.push(FakePr {
+                    number,
+                    new: NewPull {
+                        title: "someone else's PR".to_owned(),
+                        body: String::new(),
+                        head: branch.to_owned(),
+                        base: "main".to_owned(),
+                    },
+                    open: false,
+                    merged: Some(main),
+                });
+                *foreign = None;
+            } else {
+                *skip -= 1;
+            }
+        }
         Ok(prs
             .iter()
             .find(|p| p.new.head == branch)
@@ -2036,3 +2063,152 @@ async fn an_append_whose_lease_fails_is_an_error() {
     );
     assert_eq!(support::branch(&w.upstream, &first.branch).unwrap(), main);
 }
+
+/// A PR left on GitHub for `head` that this mirror DB never opened (a trial,
+/// or a DB that was reset): merged into `main` as `merged`, its branch gone.
+fn leftover_merged_pr(w: &World, head: &str, merged: &str) -> u64 {
+    let mut prs = w.github.prs_mut();
+    let number = prs.len() as u64 + 1;
+    prs.push(FakePr {
+        number,
+        new: NewPull {
+            title: "an earlier batch".to_owned(),
+            body: String::new(),
+            head: head.to_owned(),
+            base: "main".to_owned(),
+        },
+        open: false,
+        merged: Some(merged.to_owned()),
+    });
+    number
+}
+
+/// WI-2585: a fresh DB on a GitHub repo that already has `graft/batch-1`
+/// merged (branch deleted) never adopts that PR; the lands go in a new PR.
+#[tokio::test]
+async fn a_leftover_merged_pr_is_never_adopted() {
+    let w = World::new().await;
+    w.start().await;
+    let main = support::branch(&w.upstream, "main").unwrap();
+    let old = leftover_merged_pr(&w, "graft/batch-1", &main);
+    let a = w.land(&[("a.txt", "a1\n")], "add a", 1).await;
+    let b = w.land(&[("b.txt", "b1\n")], "add b", 2).await;
+
+    let run = w.batcher().run().await.unwrap();
+    assert!(run.settled.is_none());
+    let rec = opened(run.outcome);
+    assert_eq!(rec.state, BatchState::Open, "{}", rec.detail);
+    assert_eq!((rec.number, rec.lands), (2, 2));
+    assert_eq!(rec.branch, "graft/batch-2");
+    assert!(rec.merged.is_none() && rec.merge_sha.is_none());
+    let new = rec.pr_number.unwrap();
+    assert_ne!(new, old);
+
+    let prs = w.github.prs();
+    let pr = prs.iter().find(|p| p.number == new).unwrap();
+    assert!(pr.open && pr.merged.is_none());
+    assert_eq!(pr.new.head, "graft/batch-2");
+    for id in [a, b] {
+        assert!(
+            pr.new.body.contains(&id.to_string()[..13]),
+            "{}",
+            pr.new.body
+        );
+    }
+    // Nothing reached main; the old PR is as it was.
+    assert_eq!(support::branch(&w.upstream, "main").unwrap(), main);
+    assert!(support::branch(&w.upstream, "graft/batch-1").is_none());
+    let old_pr = prs.iter().find(|p| p.number == old).unwrap();
+    assert!(!old_pr.open && old_pr.new.head == "graft/batch-1");
+
+    // The next run waits on the new PR; it does not settle anything.
+    match w.batcher().run().await.unwrap().outcome {
+        BatchOutcome::Waiting(r) => assert_eq!(r.pr_number, Some(new)),
+        other => panic!("expected to wait, got {other:?}"),
+    }
+}
+
+/// WI-2585: a leftover branch with other commits and a leftover merged PR
+/// (the trial's batch-1 and batch-2) are both skipped; the lands go in a
+/// fresh PR on a free number, and neither leftover is changed.
+#[tokio::test]
+async fn leftover_batch_branches_and_prs_are_skipped_for_a_fresh_number() {
+    let w = World::new().await;
+    w.start().await;
+    let main = support::branch(&w.upstream, "main").unwrap();
+    let tree = git(&w.upstream, &["rev-parse", &format!("{main}^{{tree}}")]);
+    let stray = git(
+        &w.upstream,
+        &["commit-tree", &tree, "-p", &main, "-m", "trial leftover"],
+    );
+    git(
+        &w.upstream,
+        &["update-ref", "refs/heads/graft/batch-1", &stray],
+    );
+    let old = leftover_merged_pr(&w, "graft/batch-2", &main);
+    w.land(&[("a.txt", "a1\n")], "add a", 1).await;
+    w.land(&[("b.txt", "b1\n")], "add b", 2).await;
+
+    let rec = opened(w.batcher().run().await.unwrap().outcome);
+    assert_eq!(rec.state, BatchState::Open, "{}", rec.detail);
+    assert_eq!((rec.number, rec.lands), (3, 2));
+    assert_eq!(rec.branch, "graft/batch-3");
+    assert!(rec.merged.is_none());
+    assert_ne!(rec.pr_number, Some(old));
+    let prs = w.github.prs();
+    let pr = prs
+        .iter()
+        .find(|p| Some(p.number) == rec.pr_number)
+        .unwrap();
+    assert!(pr.open && pr.merged.is_none());
+    assert_eq!(pr.new.head, "graft/batch-3");
+    assert_eq!(
+        w.changed(
+            &main,
+            &support::branch(&w.upstream, "graft/batch-3").unwrap()
+        ),
+        ["a.txt", "b.txt"]
+    );
+    assert_eq!(
+        support::branch(&w.upstream, "graft/batch-1").unwrap(),
+        stray
+    );
+    assert_eq!(support::branch(&w.upstream, "main").unwrap(), main);
+}
+
+/// WI-2585: a PR graft did not open that appears on the batch branch after
+/// the number was chosen and the branch pushed is not adopted. The batch is
+/// rejected unmerged, and the next run carries its lands in a fresh PR.
+#[tokio::test]
+async fn a_foreign_pr_on_the_pushed_branch_is_rejected_not_adopted() {
+    let w = World::new().await;
+    w.start().await;
+    let main = support::branch(&w.upstream, "main").unwrap();
+    let a = w.land(&[("a.txt", "a1\n")], "add a", 1).await;
+    // The number check reads graft/batch-1 once; the PR appears for open().
+    *w.github.foreign_pr.lock().unwrap() = Some(("graft/batch-1".to_owned(), 1));
+
+    let run = w.batcher().run().await.unwrap();
+    let rec = opened(run.outcome);
+    assert_eq!(rec.number, 1);
+    assert_eq!(rec.state, BatchState::Rejected);
+    assert!(rec.detail.contains("did not open"), "{}", rec.detail);
+    assert!(rec.merged.is_none() && rec.merge_sha.is_none());
+    assert_eq!(rec.pr_number, None, "the foreign PR is not recorded");
+    assert_eq!(w.github.prs().len(), 1, "no PR was opened over it");
+    assert_eq!(support::branch(&w.upstream, "main").unwrap(), main);
+
+    // The lands are still pending: the next run puts them in a new PR on a
+    // free number, past the taken branch.
+    let rec = opened(w.batcher().run().await.unwrap().outcome);
+    assert_eq!(rec.state, BatchState::Open, "{}", rec.detail);
+    assert_eq!((rec.number, rec.lands), (2, 1));
+    let prs = w.github.prs();
+    let pr = prs
+        .iter()
+        .find(|p| Some(p.number) == rec.pr_number)
+        .unwrap();
+    assert!(pr.open && pr.merged.is_none());
+    assert_eq!(pr.new.head, "graft/batch-2");
+    assert!(pr.new.body.contains(&a.to_string()[..13]));
+}