Skip to main content

magi/
release_watch.rs

1//! Watches every open release-bump pull request until it lands.
2//!
3//! [`crate::bump`] opens a `chore/release-vX.Y.Z` pull request with auto-merge
4//! armed and then stops looking. A required check that goes red (a flaky
5//! windows or browser job) leaves auto-merge waiting forever, and nothing
6//! pages anybody because the notices there only fire when the pull request
7//! cannot be *opened*. This module is the part that keeps looking.
8//!
9//! It runs inside `magi serve` as its own task, like [`crate::waiter`], and is
10//! tied to no run: a pending bump is coalesced across runs and outlives the
11//! one that opened it. The forge is the source of truth (every open pull
12//! request whose head branch starts with `chore/release-v`, in every checkout
13//! magi knows); the pending marker is not consulted, so a missing one changes
14//! nothing.
15//!
16//! - **The decision is pure.** [`decide`] maps a checks snapshot and the
17//!   recorded [`WatchState`] to one [`Step`]. A forge answer that cannot be
18//!   read is `Unknown`: nothing moves, not even the stall clock.
19//! - **A failed run is rerun once, durably.** The run id is written to the
20//!   state file *before* `gh run rerun <id> --failed` is called, so a restart
21//!   never reruns twice; if the call then fails, the cost is one escalation to
22//!   a human, never a rerun loop.
23//! - **Escalation is a notice plus a question** ([`crate::bump::NOTICE_NODE`],
24//!   no `cwd`, no run). Nothing is ever merged, closed or pushed: the watcher
25//!   only reruns jobs. Silence is a hold. The owner's choice is applied by the
26//!   watcher itself on its next lap, recorded in `WatchState::applied` so an
27//!   answer is applied once.
28//! - **`[release] mode = "local"` replaces all of the above for that
29//!   repository's release pull requests.** No check is awaited, reran or
30//!   stalled on. The watcher asks the owner (`merge` / `hold`, bound to the
31//!   head it observed) unless `graph.land_approval` is off, in which case it
32//!   merges that head without asking. It merges with `--match-head-commit`, and after the merge
33//!   drives [`crate::release_local`] - tag, then the configured commands - with
34//!   its progress in the watch record. A failure is a notice and a question
35//!   (`retry` / `leave it`), never a blind retry. See [`decide_local`].
36//! - **Notice wording is fixed per stage** and carries only the pull request,
37//!   so a repeated poll never relights it; everything that varies (job names,
38//!   links) rides in the question.
39
40use std::collections::BTreeMap;
41use std::future::Future;
42use std::path::{Path, PathBuf};
43use std::pin::Pin;
44use std::time::Duration;
45
46use anyhow::{Result, bail};
47use serde::{Deserialize, Serialize};
48
49use crate::ask::{Answer, Question, Questions};
50use crate::land::{self, PrLifecycle, RollupView, Verdict};
51use crate::notices::{self, Link, Notice, Notices};
52use crate::release_local::{self, Job};
53use crate::run::RunState;
54
55/// Pause between laps.
56const LAP: Duration = Duration::from_secs(300);
57
58/// How long after a rerun the forge may still report the old failure before
59/// that is read as "still red". `gh run rerun` returns before the checks flip
60/// to pending.
61pub(crate) const RERUN_GRACE: i64 = 180;
62
63/// Choice: rerun the failed jobs once more. The three choices are matched
64/// verbatim when applied.
65pub const RERUN_AGAIN: &str = "rerun again";
66/// Choice: stay quiet until a check or the head changes.
67pub const HOLD: &str = "hold";
68/// Choice: stop watching this pull request.
69pub const LEAVE_IT: &str = "leave it";
70/// Choice (local mode): run the failed release again from where it stopped.
71pub const RETRY: &str = "retry";
72
73/// What one look at one pull request concludes.
74#[derive(Debug, Clone, PartialEq, Eq)]
75pub(crate) enum Step {
76    /// Merged or closed: stop watching.
77    Done,
78    /// Nothing to do yet.
79    Wait,
80    /// The forge answer could not be read; change nothing.
81    Unknown,
82    /// Rerun the failed jobs of these workflow runs, once each.
83    Rerun(Vec<String>),
84    /// A human is needed.
85    Escalate(Why),
86}
87
88#[derive(Debug, Clone, Copy, PartialEq, Eq)]
89pub(crate) enum Why {
90    /// Still red after the one rerun, or red with nothing to rerun.
91    StillRed,
92    /// No progress for longer than `[daemon] release_stall_minutes`.
93    Stalled,
94}
95
96/// Everything remembered about one watched pull request.
97#[derive(Debug, Clone, Default, Serialize, Deserialize)]
98#[serde(default)]
99pub(crate) struct WatchState {
100    /// Checkout the pull request belongs to.
101    pub repo: String,
102    pub url: String,
103    /// Head commit the rest of this record is about.
104    pub head: String,
105    /// Workflow runs already rerun for this head, with when (unix seconds).
106    pub reruns: BTreeMap<String, i64>,
107    /// Digest of the head and every check's verdict, and when it last changed.
108    pub fingerprint: String,
109    pub progress_at: i64,
110    /// The open question asked about this pull request.
111    pub question: Option<String>,
112    /// Fingerprint the owner chose to hold: no new question until it changes.
113    pub held: Option<String>,
114    /// The owner said to leave it.
115    pub ignored: bool,
116    /// Question ids whose answer was already applied.
117    pub applied: Vec<String>,
118    /// Local mode: the head the open (or last) approval question was about.
119    pub asked_head: Option<String>,
120    /// Local mode: the head the owner held or whose merge was refused; no new
121    /// approval question until the head moves.
122    pub held_head: Option<String>,
123    /// Local mode: the run that opened this release PR, so a failed release can
124    /// hold the task that run finished (the daemon marks it Done at the merge).
125    pub run: Option<String>,
126    /// Local mode: the task this watcher put on hold for a failed release, so a
127    /// successful retry can give it back (and a restart does not hold it twice).
128    pub held_task: Option<String>,
129    /// Local mode: the release after the merge.
130    pub job: Option<Job>,
131    /// Set when this pull request was last seen as a local-mode release, so a
132    /// lap that cannot read the config does not fall back to Actions rules.
133    pub local: bool,
134}
135
136impl WatchState {
137    /// Whether this record belongs to a local-mode release. Records written
138    /// before `local` existed lack the flag, but only local mode ever sets
139    /// `run`, `asked_head`, `held_head` or `job`.
140    fn is_local(&self) -> bool {
141        self.local
142            || self.run.is_some()
143            || self.asked_head.is_some()
144            || self.held_head.is_some()
145            || self.job.is_some()
146    }
147}
148
149fn is_failed(v: Verdict) -> bool {
150    v == Verdict::Fail
151}
152
153/// Digest of what "progress" means: the head and every check's verdict.
154pub(crate) fn fingerprint(snap: &RollupView) -> String {
155    let mut parts: Vec<String> = snap
156        .checks
157        .iter()
158        .map(|c| format!("{}={:?}", c.name, c.verdict))
159        .collect();
160    parts.sort();
161    format!("{}|{}", snap.head, parts.join(","))
162}
163
164/// Fold a fresh snapshot into the record: a new head forgets the old head's
165/// reruns, and a changed fingerprint restarts the stall clock.
166pub(crate) fn observe(st: &mut WatchState, snap: &RollupView, now: i64) {
167    if st.head != snap.head {
168        st.reruns.clear();
169        st.head = snap.head.clone();
170    }
171    let fp = fingerprint(snap);
172    if st.fingerprint != fp {
173        st.fingerprint = fp;
174        st.progress_at = now;
175    }
176}
177
178/// Decide what to do about a pull request. `snap` is `None` when the forge
179/// could not be read. `stall_secs == 0` disables the stall rule.
180pub(crate) fn decide(
181    snap: Option<&RollupView>,
182    st: &WatchState,
183    now: i64,
184    stall_secs: i64,
185) -> Step {
186    let Some(snap) = snap else {
187        return Step::Unknown;
188    };
189    if snap.state != PrLifecycle::Open {
190        return Step::Done;
191    }
192    if st.ignored || st.held.as_deref() == Some(fingerprint(snap).as_str()) {
193        return Step::Wait;
194    }
195
196    let any_pending = snap.checks.iter().any(|c| c.verdict == Verdict::Pending);
197    // A run is complete once none of its own checks is still pending; another
198    // run's pending job says nothing about it.
199    let complete = |run: &str| {
200        !snap
201            .checks
202            .iter()
203            .any(|c| c.verdict == Verdict::Pending && c.run.as_deref() == Some(run))
204    };
205    let mut fresh = Vec::new();
206    let mut spent_recently = false;
207    let mut spent = false;
208    let mut external = false;
209    for c in snap.checks.iter().filter(|c| is_failed(c.verdict)) {
210        match c.run.as_deref() {
211            None => external = true,
212            Some(run) if !complete(run) => {}
213            Some(run) => match st.reruns.get(run) {
214                None => {
215                    if !fresh.iter().any(|r: &String| r == run) {
216                        fresh.push(run.to_owned());
217                    }
218                }
219                Some(&at) if now - at < RERUN_GRACE => spent_recently = true,
220                Some(_) => spent = true,
221            },
222        }
223    }
224    if !fresh.is_empty() {
225        return Step::Rerun(fresh);
226    }
227    if spent_recently {
228        return Step::Wait;
229    }
230    if spent || (external && !any_pending) {
231        return Step::Escalate(Why::StillRed);
232    }
233    if stall_secs > 0 && now - st.progress_at >= stall_secs {
234        return Step::Escalate(Why::Stalled);
235    }
236    Step::Wait
237}
238
239/// `owner/repo#N` out of a pull request url, the stable name for notices.
240pub(crate) fn pr_key(url: &str) -> Option<String> {
241    let rest = url.split("://").nth(1)?;
242    let mut it = rest.split('/');
243    let _host = it.next()?;
244    let owner = it.next()?;
245    let repo = it.next()?;
246    if it.next()? != "pull" {
247        return None;
248    }
249    let n: String = it
250        .next()?
251        .chars()
252        .take_while(char::is_ascii_digit)
253        .collect();
254    (!n.is_empty()).then(|| format!("{owner}/{repo}#{n}"))
255}
256
257fn notice_key(pr: &str) -> String {
258    format!("release-pr:{pr}")
259}
260
261/// Fixed per stage: no counts, times or job names.
262fn rerun_message(pr: &str) -> String {
263    format!("Release PR {pr}: failed checks were rerun once")
264}
265
266fn human_message(pr: &str) -> String {
267    format!("Release PR {pr} is stuck and needs a human")
268}
269
270fn question_detail(snap: &RollupView, why: Why, st: &WatchState) -> String {
271    let mut s = String::new();
272    s.push_str(&format!("Pull request: {}\n\n", snap.url));
273    s.push_str(match why {
274        Why::StillRed => "Checks are still red after the failed jobs were rerun once.\n\n",
275        Why::Stalled => "The pull request has made no progress for too long.\n\n",
276    });
277    let failed: Vec<_> = snap
278        .checks
279        .iter()
280        .filter(|c| is_failed(c.verdict))
281        .collect();
282    if failed.is_empty() {
283        s.push_str("No check is failing.\n");
284    } else {
285        s.push_str("Failed jobs:\n");
286        for c in failed {
287            match &c.url {
288                Some(u) => s.push_str(&format!("- {} ({u})\n", c.name)),
289                None => s.push_str(&format!("- {}\n", c.name)),
290            }
291        }
292    }
293    s.push_str(&format!(
294        "\nWorkflow runs rerun so far: {}.\n\n\
295         - `{RERUN_AGAIN}`: rerun the failed jobs once more.\n\
296         - `{HOLD}`: stay quiet until a check or the head changes.\n\
297         - `{LEAVE_IT}`: stop watching this pull request.\n\n\
298         Nothing is merged or pushed by magi; silence is a hold.\n",
299        st.reruns.len()
300    ));
301    s
302}
303
304/// Where the owner's approval stands, as far as [`decide_local`] cares.
305#[derive(Debug, Clone, Copy, PartialEq, Eq)]
306pub(crate) enum Approved {
307    /// No question was filed (or it is gone).
308    NoQuestion,
309    /// Filed, not answered.
310    Open,
311    /// The owner said `merge`.
312    Merge,
313    /// The owner said anything else, or it was abandoned: silence is a hold.
314    Hold,
315}
316
317/// What one look at an open local-mode release pull request concludes.
318#[derive(Debug, Clone, PartialEq, Eq)]
319pub(crate) enum LocalStep {
320    /// Nothing to do now.
321    Wait,
322    /// Ask the owner whether to merge this head.
323    Ask,
324    /// Merge, pinned to this head.
325    Merge(String),
326    /// Remember the owner's no for this head and stay quiet until it moves.
327    Hold(String),
328}
329
330/// Decide an *open* local-mode pull request. No I/O, and no CI: nothing is
331/// awaited, rerun or stalled on, because nothing will ever report. The head is
332/// what the owner approved: an answer about an older head asks again.
333///
334/// `ask_first` is `graph.land_approval`. When it is off, a head nobody has been
335/// asked about is merged straight away, pinned to that head; an approval
336/// question that is already open is still respected.
337pub(crate) fn decide_local(
338    head: &str,
339    st: &WatchState,
340    approved: Approved,
341    ask_first: bool,
342) -> LocalStep {
343    if st.ignored || st.held_head.as_deref() == Some(head) {
344        return LocalStep::Wait;
345    }
346    let about_this_head = st.asked_head.as_deref() == Some(head);
347    match approved {
348        Approved::Open => LocalStep::Wait,
349        Approved::NoQuestion if !ask_first => LocalStep::Merge(head.to_owned()),
350        Approved::NoQuestion => LocalStep::Ask,
351        Approved::Merge if about_this_head => LocalStep::Merge(head.to_owned()),
352        Approved::Hold if about_this_head => LocalStep::Hold(head.to_owned()),
353        // The head moved after the question was filed.
354        Approved::Merge | Approved::Hold if !ask_first => LocalStep::Merge(head.to_owned()),
355        Approved::Merge | Approved::Hold => LocalStep::Ask,
356    }
357}
358
359/// Start watching a release pull request before the first lap sees it, so one
360/// merged within a lap of being opened still gets released. Never overwrites.
361pub(crate) fn register(home: &Path, repo: &Path, url: &str, run: &str) {
362    let Some(pr) = pr_key(url) else {
363        return;
364    };
365    let w = Watcher::new(Box::new(GhForge), home.to_path_buf());
366    if w.state_path(&pr).exists() {
367        return;
368    }
369    let st = WatchState {
370        repo: repo.to_string_lossy().into_owned(),
371        url: url.to_owned(),
372        run: Some(run.to_owned()),
373        local: true,
374        ..WatchState::default()
375    };
376    w.save(&pr, &st);
377}
378
379/// The watch record whose open question is `question_id`, read from the magi
380/// `home`. `None` when no record names it (never guessed from the text).
381pub(crate) fn state_for_question(home: &Path, question_id: &str) -> Option<WatchState> {
382    Watcher::new(Box::new(GhForge), home.to_path_buf())
383        .stored()
384        .into_iter()
385        .find(|s| s.question.as_deref() == Some(question_id))
386}
387
388/// The task a release-watch question's answer could touch: the one that
389/// finished the run which opened the release pull request (`WatchState::run`).
390/// `None` when the record or the run is not known - then there is nothing to
391/// check, and the settle goes on (an answer here never releases a task).
392pub fn task_of_question(
393    home: &Path,
394    q: &Question,
395    queue: &crate::queue::Queue,
396) -> Option<crate::queue::Task> {
397    let run = state_for_question(home, &q.id)?.run?;
398    queue.list().into_iter().find(|t| t.runs.contains(&run))
399}
400
401/// What a deputy is told about a release-watch question: which pull request,
402/// and what each offered choice really does - matched to [`Watcher::apply_answer`]
403/// and [`Watcher::settle_job_question`]. A missing watch record is said to be
404/// missing, never reconstructed.
405pub(crate) fn deputy_brief(q: &Question, home: &Path) -> String {
406    let mut s = String::from(
407        "This question was filed by magi's release watcher about a release pull \
408         request. Whatever the owner picks, only the watcher acts on it, on its \
409         next lap; you apply nothing and never close, merge, rerun or push \
410         anything. Silence is a hold.",
411    );
412    match state_for_question(home, &q.id) {
413        Some(st) => {
414            s.push_str(&format!(
415                "\n\nPull request: {}\nCheckout: {}",
416                st.url, st.repo
417            ));
418        }
419        None => s.push_str(
420            "\n\nThe watcher's record for this question could not be found, so the \
421             pull request is known to you only through the question's own text. \
422             Say so to the owner rather than guessing.",
423        ),
424    }
425    s.push_str("\n\nWhat each option does:");
426    for c in &q.choices {
427        let what = match c.as_str() {
428            RERUN_AGAIN => "forget the reruns already made and rerun the failed jobs once more",
429            HOLD if q.choices.iter().any(|c| c == land::APPROVE) => {
430                "leave the pull request open and unmerged; ask again only when the head moves"
431            }
432            HOLD => "stay quiet until a check or the head changes",
433            LEAVE_IT => {
434                "stop watching this pull request and dismiss its notice. It does NOT \
435                 close the pull request"
436            }
437            RETRY => "run the failed release again from where it stopped",
438            land::APPROVE => {
439                "squash-merge exactly the head the question names - irreversible - \
440                 then tag and release"
441            }
442            _ => "recorded as the answer",
443        };
444        s.push_str(&format!("\n- `{c}`: {what}"));
445    }
446    s.push_str(
447        "\n\nIf the owner's words ask for the pull request itself to be closed, that \
448         is something only they can do by hand; do not read it as a choice unless \
449         it plainly means stop watching.",
450    );
451    s
452}
453
454/// Starts the hold reason of a task held for a failed release, so only that
455/// hold is undone when the release later succeeds.
456const HOLD_PREFIX: &str = "[release] ";
457
458fn approval_message(pr: &str) -> String {
459    format!("Release PR {pr} is waiting for your approval to merge")
460}
461
462fn failed_message(pr: &str) -> String {
463    format!("Release PR {pr} merged but the release is on hold")
464}
465
466fn released_key(pr: &str) -> String {
467    format!("release-done:{pr}")
468}
469
470type Fut<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;
471
472/// Every forge access, so tests inject a fake instead of `gh`.
473pub(crate) trait ReleaseForge: Send + Sync {
474    /// `(branch, url)` of every open release-bump pull request in `repo`.
475    fn list<'a>(&'a self, repo: &'a Path) -> Fut<'a, Result<Vec<(String, String)>>>;
476    fn snapshot<'a>(&'a self, repo: &'a Path, url: &'a str) -> Fut<'a, Result<RollupView>>;
477    fn rerun<'a>(&'a self, repo: &'a Path, run: &'a str) -> Fut<'a, Result<()>>;
478    /// The repository's config, to read `[release]` and `[merge] remote`.
479    fn config<'a>(&'a self, _repo: &'a Path) -> Fut<'a, Result<crate::config::Config>> {
480        Box::pin(async { Ok(crate::config::Config::default()) })
481    }
482    /// `graph.land_approval` of the repository. An error means "unknown", and
483    /// the caller then asks.
484    fn land_approval<'a>(&'a self, _repo: &'a Path) -> Fut<'a, Result<bool>> {
485        Box::pin(async { Ok(true) })
486    }
487    /// Head branch and merge commit of a pull request (local mode).
488    fn info<'a>(&'a self, _repo: &'a Path, url: &'a str) -> Fut<'a, Result<PrInfo>> {
489        Box::pin(async move { bail!("no pull request info for {url}") })
490    }
491    /// Merge `url` (squash, delete branch) pinned to `head`. `Ok(true)` once
492    /// the forge says it merged, `Ok(false)` when it did not.
493    fn merge<'a>(&'a self, _repo: &'a Path, url: &'a str, _head: &'a str) -> Fut<'a, Result<bool>> {
494        Box::pin(async move { bail!("cannot merge {url}") })
495    }
496}
497
498/// What the forge says about a pull request's branch and merge.
499#[derive(Debug, Clone, Default, PartialEq, Eq)]
500pub(crate) struct PrInfo {
501    /// Head branch name, e.g. `chore/release-v1.2.3`.
502    pub branch: String,
503    /// The commit the pull request merged as, once merged.
504    pub merge_commit: Option<String>,
505    /// Pull request title.
506    pub title: String,
507}
508
509/// Parse `gh pr view --json headRefName,mergeCommit,title`. No I/O.
510pub(crate) fn parse_info(json: &str) -> Result<PrInfo> {
511    let v: serde_json::Value = serde_json::from_str(json)?;
512    let text = |k: &str| {
513        v.get(k)
514            .and_then(|x| x.as_str())
515            .unwrap_or_default()
516            .to_owned()
517    };
518    Ok(PrInfo {
519        branch: text("headRefName"),
520        title: text("title"),
521        merge_commit: v
522            .get("mergeCommit")
523            .and_then(|m| m.get("oid"))
524            .and_then(|o| o.as_str())
525            .filter(|o| !o.is_empty())
526            .map(str::to_owned),
527    })
528}
529
530/// The real forge: `gh`.
531pub(crate) struct GhForge;
532
533impl ReleaseForge for GhForge {
534    fn list<'a>(&'a self, repo: &'a Path) -> Fut<'a, Result<Vec<(String, String)>>> {
535        Box::pin(crate::bump::list_open_release_prs(repo))
536    }
537
538    fn snapshot<'a>(&'a self, repo: &'a Path, url: &'a str) -> Fut<'a, Result<RollupView>> {
539        Box::pin(async move {
540            let args = [
541                "pr".to_owned(),
542                "view".to_owned(),
543                url.to_owned(),
544                "--json".to_owned(),
545                "url,number,state,headRefOid,statusCheckRollup".to_owned(),
546            ];
547            let (ok, out) = land::gh(repo, &args).await?;
548            if !ok {
549                bail!("gh pr view {url}: {out}");
550            }
551            land::parse_rollup(&out)
552        })
553    }
554
555    fn rerun<'a>(&'a self, repo: &'a Path, run: &'a str) -> Fut<'a, Result<()>> {
556        Box::pin(async move {
557            let args = [
558                "run".to_owned(),
559                "rerun".to_owned(),
560                run.to_owned(),
561                "--failed".to_owned(),
562            ];
563            let (ok, out) = land::gh(repo, &args).await?;
564            if !ok {
565                bail!("gh run rerun {run}: {out}");
566            }
567            Ok(())
568        })
569    }
570
571    fn config<'a>(&'a self, repo: &'a Path) -> Fut<'a, Result<crate::config::Config>> {
572        Box::pin(async move { Ok(crate::config::Config::discover(repo, None)?.0) })
573    }
574
575    fn land_approval<'a>(&'a self, repo: &'a Path) -> Fut<'a, Result<bool>> {
576        // A second `discover` may read a newer generation of the config than
577        // the one `config` returned; only this one value rides on it, and an
578        // unreadable config makes the caller ask.
579        Box::pin(async move {
580            Ok(crate::config::Config::discover(repo, None)?
581                .0
582                .graph
583                .land_approval)
584        })
585    }
586
587    fn info<'a>(&'a self, repo: &'a Path, url: &'a str) -> Fut<'a, Result<PrInfo>> {
588        Box::pin(async move {
589            let args = [
590                "pr".to_owned(),
591                "view".to_owned(),
592                url.to_owned(),
593                "--json".to_owned(),
594                "headRefName,mergeCommit,title".to_owned(),
595            ];
596            let (ok, out) = land::gh(repo, &args).await?;
597            if !ok {
598                bail!("gh pr view {url}: {out}");
599            }
600            parse_info(&out)
601        })
602    }
603
604    fn merge<'a>(&'a self, repo: &'a Path, url: &'a str, head: &'a str) -> Fut<'a, Result<bool>> {
605        Box::pin(async move {
606            let title = self
607                .info(repo, url)
608                .await
609                .map(|i| i.title)
610                .unwrap_or_default();
611            let title = if title.trim().is_empty() {
612                "chore: release".to_owned()
613            } else {
614                title
615            };
616            // Same shape as the direct merge in `bump`, pinned to the head the
617            // owner approved: a push in between is refused by the forge.
618            let mut argv = crate::bump::bump_merge_argv(url, &title);
619            argv.push("--match-head-commit".to_owned());
620            argv.push(head.to_owned());
621            let (ok, out) = land::gh(repo, &argv).await?;
622            if ok {
623                return Ok(true);
624            }
625            // jj keeps HEAD detached, so `--delete-branch` exits non-zero after
626            // the merge happened: the forge decides.
627            let after = land::lifecycle(repo, url).await.ok();
628            if land::merged_after_all(&argv, &out, after).is_some() {
629                return Ok(true);
630            }
631            bail!("gh pr merge {url}: {out}")
632        })
633    }
634}
635
636/// The watcher: a forge, a magi home, and nothing else.
637pub(crate) struct Watcher {
638    forge: Box<dyn ReleaseForge>,
639    home: PathBuf,
640}
641
642impl Watcher {
643    pub(crate) fn new(forge: Box<dyn ReleaseForge>, home: PathBuf) -> Self {
644        Self { forge, home }
645    }
646
647    fn dir(&self) -> PathBuf {
648        self.home.join("release-watch")
649    }
650
651    fn state_path(&self, pr: &str) -> PathBuf {
652        self.dir().join(format!("{}.json", notices::id_of(pr)))
653    }
654
655    fn load(&self, pr: &str) -> WatchState {
656        std::fs::read_to_string(self.state_path(pr))
657            .ok()
658            .and_then(|s| serde_json::from_str(&s).ok())
659            .unwrap_or_default()
660    }
661
662    fn save(&self, pr: &str, st: &WatchState) -> bool {
663        let write = || -> Result<()> {
664            std::fs::create_dir_all(self.dir())?;
665            let path = self.state_path(pr);
666            let tmp = path.with_extension("json.tmp");
667            std::fs::write(&tmp, serde_json::to_string_pretty(st)?)?;
668            std::fs::rename(&tmp, &path)?;
669            Ok(())
670        };
671        match write() {
672            Ok(()) => true,
673            Err(e) => {
674                tracing::warn!("could not save the release watch for {pr}: {e:#}");
675                false
676            }
677        }
678    }
679
680    /// Stored records, for pull requests that left the open list.
681    fn stored(&self) -> Vec<WatchState> {
682        let Ok(rd) = std::fs::read_dir(self.dir()) else {
683            return Vec::new();
684        };
685        rd.flatten()
686            .filter(|e| e.path().extension().is_some_and(|x| x == "json"))
687            .filter_map(|e| std::fs::read_to_string(e.path()).ok())
688            .filter_map(|s| serde_json::from_str(&s).ok())
689            .collect()
690    }
691
692    fn questions(&self) -> Questions {
693        Questions::at(self.home.join("questions"))
694    }
695
696    fn raise(&self, pr: &str, url: &str, message: String) {
697        notices::raise_in(
698            &self.home,
699            Notice::warn(&notice_key(pr), message).link(Link::Url {
700                url: url.to_owned(),
701            }),
702        );
703    }
704
705    /// One pass over every repo. `stall_secs` is `[daemon] release_stall_minutes`
706    /// in seconds. Best-effort throughout.
707    pub(crate) async fn lap(
708        &self,
709        repos: &[PathBuf],
710        stall_secs: i64,
711        now: i64,
712        halt: &(dyn Fn() -> bool + Sync),
713    ) {
714        let stored = self.stored();
715        // A repo with a watch record is covered even when nothing else names it.
716        let mut repos = repos.to_vec();
717        for s in &stored {
718            let p = PathBuf::from(&s.repo);
719            if !s.repo.is_empty() && !repos.contains(&p) {
720                repos.push(p);
721            }
722        }
723        for repo in &repos {
724            if halt() {
725                return;
726            }
727            let mut urls: Vec<String> = match self.forge.list(repo).await {
728                Ok(v) => v.into_iter().map(|(_, u)| u).collect(),
729                Err(e) => {
730                    tracing::warn!(
731                        "could not list release pull requests in {}: {e:#}",
732                        repo.display()
733                    );
734                    Vec::new()
735                }
736            };
737            // A pull request that left the open list is kept until its end is
738            // confirmed, never assumed.
739            let here = repo.to_string_lossy();
740            for s in stored.iter().filter(|s| s.repo == here.as_ref()) {
741                if !urls.contains(&s.url) {
742                    urls.push(s.url.clone());
743                }
744            }
745            for url in urls {
746                if halt() {
747                    return;
748                }
749                self.watch(repo, &url, stall_secs, now, halt).await;
750            }
751        }
752    }
753
754    /// Look at one pull request and act once.
755    pub(crate) async fn watch(
756        &self,
757        repo: &Path,
758        url: &str,
759        stall_secs: i64,
760        now: i64,
761        halt: &(dyn Fn() -> bool + Sync),
762    ) {
763        let Some(pr) = pr_key(url) else {
764            tracing::warn!("not a pull request url: {url}");
765            return;
766        };
767        let mut st = self.load(&pr);
768        st.repo = repo.to_string_lossy().into_owned();
769        st.url = url.to_owned();
770
771        let snap = match self.forge.snapshot(repo, url).await {
772            Ok(s) => Some(s),
773            Err(e) => {
774                tracing::warn!("could not read {url}: {e:#}");
775                None
776            }
777        };
778        match self.forge.config(repo).await {
779            Ok(cfg) if cfg.release.is_local() => {
780                st.local = true;
781                self.watch_local(repo, &pr, st, snap, Some(&cfg)).await;
782                return;
783            }
784            Ok(_) => st.local = false,
785            Err(e) if st.is_local() => {
786                tracing::warn!(
787                    "could not read the config of {}: {e:#}; watching {url} as a local release without it",
788                    repo.display()
789                );
790                self.watch_local(repo, &pr, st, snap, None).await;
791                return;
792            }
793            Err(e) => tracing::warn!(
794                "could not read the config of {}: {e:#}; watching {url} as an Actions release",
795                repo.display()
796            ),
797        }
798        if decide(snap.as_ref(), &st, now, stall_secs) == Step::Done {
799            self.finish(&pr, &st);
800            return;
801        }
802        let Some(snap) = snap else {
803            return;
804        };
805
806        // The owner's word comes first.
807        if let Some(id) = st.question.clone() {
808            match self.questions().get(&id) {
809                Ok(q) if q.status.open() => return,
810                Ok(q) => self.apply_answer(&pr, &mut st, &snap, &q),
811                Err(_) => st.question = None,
812            }
813            // Whatever was decided, the record is saved before anything else.
814            observe(&mut st, &snap, now);
815            self.save(&pr, &st);
816        } else {
817            observe(&mut st, &snap, now);
818        }
819
820        match decide(Some(&snap), &st, now, stall_secs) {
821            Step::Done | Step::Unknown | Step::Wait => {}
822            Step::Rerun(runs) => {
823                // Durable before the call: at most once per failing run.
824                for r in &runs {
825                    st.reruns.insert(r.clone(), now);
826                }
827                // Without a durable record the call must not happen: a restart
828                // would rerun the same run again.
829                if !self.save(&pr, &st) || halt() {
830                    return;
831                }
832                let mut ok = false;
833                for r in &runs {
834                    match self.forge.rerun(repo, r).await {
835                        Ok(()) => ok = true,
836                        Err(e) => tracing::warn!("could not rerun run {r} of {url}: {e:#}"),
837                    }
838                }
839                if ok {
840                    self.raise(&pr, url, rerun_message(&pr));
841                }
842            }
843            Step::Escalate(why) => {
844                self.raise(&pr, url, human_message(&pr));
845                let mut q = Question::new(
846                    String::new(),
847                    crate::bump::NOTICE_NODE.to_owned(),
848                    "release-watch".to_owned(),
849                    format!("Release PR {pr} is stuck: what now?"),
850                    question_detail(&snap, why, &st),
851                    vec![RERUN_AGAIN.to_owned(), HOLD.to_owned(), LEAVE_IT.to_owned()],
852                );
853                match self.questions().put(&mut q) {
854                    Ok(()) => st.question = Some(q.id.clone()),
855                    Err(e) => tracing::warn!("could not file the question for {pr}: {e:#}"),
856                }
857            }
858        }
859        self.save(&pr, &st);
860    }
861
862    /// The local-mode path of [`Watcher::watch`]: approval, merge, release.
863    async fn watch_local(
864        &self,
865        repo: &Path,
866        pr: &str,
867        mut st: WatchState,
868        snap: Option<RollupView>,
869        cfg: Option<&crate::config::Config>,
870    ) {
871        let Some(snap) = snap else {
872            return;
873        };
874        match snap.state {
875            PrLifecycle::Closed => self.finish(pr, &st),
876            PrLifecycle::Merged => match cfg {
877                Some(cfg) => self.release_merged(repo, pr, st, cfg).await,
878                // The release commands live in the config: wait, keep the record.
879                None => {
880                    tracing::warn!("{pr} is merged but its config is unreadable; not released yet")
881                }
882            },
883            PrLifecycle::Open => {
884                let approved = match st.question.clone() {
885                    None => Approved::NoQuestion,
886                    Some(id) => match self.questions().get(&id) {
887                        Ok(q) if q.status.open() => Approved::Open,
888                        Ok(q) => match &q.answer {
889                            Some(Answer::Choice(c))
890                                if land::approval(Some(c)) == land::Approval::Merge =>
891                            {
892                                Approved::Merge
893                            }
894                            _ => Approved::Hold,
895                        },
896                        Err(_) => Approved::NoQuestion,
897                    },
898                };
899                let ask_first = if cfg.is_none() {
900                    true
901                } else {
902                    match self.forge.land_approval(repo).await {
903                        Ok(v) => v,
904                        Err(e) => {
905                            tracing::warn!("could not read graph.land_approval for {pr}: {e:#}");
906                            true
907                        }
908                    }
909                };
910                let step = decide_local(&snap.head, &st, approved, ask_first);
911                if cfg.is_none() && matches!(step, LocalStep::Merge(_)) {
912                    // Nothing to release with: keep the answer for a lap that can read the config.
913                    tracing::warn!("not merging {pr} while its config is unreadable");
914                    return;
915                }
916                if matches!(
917                    step,
918                    LocalStep::Merge(_) | LocalStep::Hold(_) | LocalStep::Ask
919                ) {
920                    // A settled (or stale) question is consumed exactly once.
921                    st.question = None;
922                }
923                match step {
924                    LocalStep::Wait => {}
925                    LocalStep::Hold(head) => st.held_head = Some(head),
926                    LocalStep::Ask => {
927                        st.held_head = None;
928                        st.asked_head = Some(snap.head.clone());
929                        self.raise(pr, &snap.url, approval_message(pr));
930                        let mut q = Question::new(
931                            String::new(),
932                            crate::bump::NOTICE_NODE.to_owned(),
933                            "release-watch".to_owned(),
934                            format!("Release PR {pr}: merge it and release?"),
935                            format!(
936                                "Pull request: {}\nHead: {}\n\n\
937                                 `[release] mode = \"local\"`: no CI is awaited. `{}` merges \
938                                 exactly this head, then magi tags the merge commit and runs the \
939                                 configured release commands. `{}` leaves the pull request \
940                                 open. Silence is a hold.\n",
941                                snap.url,
942                                snap.head,
943                                land::APPROVE,
944                                land::HOLD
945                            ),
946                            vec![land::APPROVE.to_owned(), land::HOLD.to_owned()],
947                        );
948                        match self.questions().put(&mut q) {
949                            Ok(()) => st.question = Some(q.id.clone()),
950                            Err(e) => tracing::warn!("could not file the question for {pr}: {e:#}"),
951                        }
952                    }
953                    LocalStep::Merge(head) => {
954                        match self.forge.merge(repo, &snap.url, &head).await {
955                            Ok(true) => {
956                                // Release at once rather than a lap later.
957                                self.save(pr, &st);
958                                if let Some(cfg) = cfg {
959                                    self.release_merged(repo, pr, st, cfg).await;
960                                }
961                                return;
962                            }
963                            res => {
964                                match res {
965                                    Err(e) => {
966                                        tracing::warn!("could not merge {}: {e:#}", snap.url)
967                                    }
968                                    _ => tracing::warn!("{} did not merge", snap.url),
969                                }
970                                st.held_head = Some(head);
971                                self.raise(
972                                    pr,
973                                    &snap.url,
974                                    format!(
975                                        "Release PR {pr} could not be merged; merge it by hand"
976                                    ),
977                                );
978                            }
979                        }
980                    }
981                }
982                self.save(pr, &st);
983            }
984        }
985    }
986
987    /// The pull request is merged: create the job once, then drive it.
988    async fn release_merged(
989        &self,
990        repo: &Path,
991        pr: &str,
992        mut st: WatchState,
993        cfg: &crate::config::Config,
994    ) {
995        if st.ignored {
996            return self.finish(pr, &st);
997        }
998        if st.job.is_none() {
999            let info = match self.forge.info(repo, &st.url).await {
1000                Ok(i) => i,
1001                Err(e) => {
1002                    tracing::warn!("could not read the merged {}: {e:#}", st.url);
1003                    return;
1004                }
1005            };
1006            let (Some(version), Some(commit)) = (
1007                release_local::version_from_branch(&info.branch),
1008                info.merge_commit,
1009            ) else {
1010                // Not a release branch, or the forge has not recorded the merge
1011                // commit yet: nothing to release (yet).
1012                if release_local::version_from_branch(&info.branch).is_none() {
1013                    self.finish(pr, &st);
1014                }
1015                return;
1016            };
1017            let mut job = Job::new(&version, &st.url, &commit);
1018            // An escalated pull request keeps its first branch name but gets a
1019            // new title; the manifest at the merge commit picks between them.
1020            job.accepted = release_local::version_from_title(&info.title)
1021                .into_iter()
1022                .filter(|t| *t != version)
1023                .collect();
1024            st.job = Some(job);
1025            if !self.save(pr, &st) {
1026                return;
1027            }
1028        }
1029        let Some(mut job) = st.job.take() else {
1030            return;
1031        };
1032        if job.finished {
1033            // Saved as finished, but a stop may have come before the task was
1034            // given back: do that now, and keep the record until it is.
1035            st.job = Some(job);
1036            if !self.record_on_run(&st) {
1037                self.save(pr, &st);
1038                return;
1039            }
1040            return self.complete(pr, &mut st);
1041        }
1042        if job.interrupted() {
1043            job.failed = Some(format!(
1044                "magi stopped while `{}` was running; it may have partly run, so it is not repeated on its own",
1045                job.running.clone().unwrap_or_default()
1046            ));
1047        }
1048        if let Some(why) = job.failed.clone() {
1049            // Reconcile first: a stop between saving the failure and holding
1050            // the task (or an interrupted step) must not leave the task Done.
1051            self.hold_task(&mut st, &format!("release {} is on hold: {why}", job.tag()));
1052            // Held: only the owner's `retry` runs it again.
1053            let retry = self.settle_job_question(pr, &mut st, &mut job);
1054            if !retry {
1055                st.job = Some(job);
1056                self.save(pr, &st);
1057                return;
1058            }
1059        }
1060        let shell = cfg.shell();
1061        let env = release_local::Env {
1062            repo,
1063            home: &self.home,
1064            key: pr,
1065            remote: &cfg.merge.remote,
1066            shell: &shell,
1067            release: &cfg.release,
1068        };
1069        let result = {
1070            let mut save = |j: &Job| {
1071                let mut copy = st.clone();
1072                copy.job = Some(j.clone());
1073                self.save(pr, &copy)
1074            };
1075            release_local::run_job(&env, &mut job, &mut save).await
1076        };
1077        match result {
1078            Ok(()) => {
1079                self.raise_released(pr, &st.url, &job);
1080                st.job = Some(job);
1081                if self.record_on_run(&st) {
1082                    self.complete(pr, &mut st);
1083                } else {
1084                    // Finished and saved as such: the next lap retries only
1085                    // the run record, never the commands.
1086                    self.save(pr, &st);
1087                }
1088            }
1089            Err(e) => {
1090                job.failed = Some(format!("{e:#}"));
1091                self.hold_task(&mut st, &format!("release {} failed: {e:#}", job.tag()));
1092                self.raise(pr, &st.url, failed_message(pr));
1093                self.file_job_question(pr, &mut st, &job);
1094                st.job = Some(job);
1095                self.save(pr, &st);
1096                self.record_on_run(&st);
1097            }
1098        }
1099    }
1100
1101    /// Copy the job onto the run that opened the release PR, so what the
1102    /// release did outlives the watcher's record. Returns whether the record
1103    /// may go: `false` only when a readable run could not be saved. No run, or
1104    /// a run that is gone, is warned about and counts as done.
1105    fn record_on_run(&self, st: &WatchState) -> bool {
1106        let (Some(run), Some(job)) = (st.run.as_deref(), st.job.as_ref()) else {
1107            return true;
1108        };
1109        let mut state = match RunState::load_under(run, &self.home) {
1110            Ok(s) => s,
1111            Err(e) => {
1112                tracing::warn!("could not record the release on run {run}: {e:#}");
1113                return true;
1114            }
1115        };
1116        let bump = state.release_bump.get_or_insert_with(Default::default);
1117        if bump.release.as_ref() == Some(job) {
1118            return true;
1119        }
1120        bump.release = Some(job.clone());
1121        state.event(
1122            "release",
1123            format!(
1124                "release {} {}",
1125                job.tag(),
1126                if job.finished {
1127                    "finished"
1128                } else if let Some(why) = &job.failed {
1129                    why
1130                } else {
1131                    "stopped"
1132                }
1133            ),
1134        );
1135        match state.save_under(&self.home) {
1136            Ok(()) => true,
1137            Err(e) => {
1138                tracing::warn!("could not save the release on run {run}: {e:#}");
1139                false
1140            }
1141        }
1142    }
1143
1144    /// Hold the task whose run opened this release PR: the daemon marked it
1145    /// Done when the change merged, but the release is not done. Best-effort;
1146    /// a task the owner already holds or that cannot be found is left alone.
1147    fn hold_task(&self, st: &mut WatchState, reason: &str) {
1148        if st.held_task.is_some() {
1149            return;
1150        }
1151        let Some(run) = st.run.clone() else {
1152            return;
1153        };
1154        let q = crate::queue::Queue::at(self.home.join("queue"));
1155        for mut t in q.list() {
1156            if t.runs.contains(&run) {
1157                if t.status != crate::queue::TaskStatus::Held {
1158                    t.notice_cause("release-failed", Some(&run), None);
1159                    t.hold_machine(Some(format!("{HOLD_PREFIX}{reason}")));
1160                    match q.put(&mut t) {
1161                        Ok(()) => st.held_task = Some(t.id.clone()),
1162                        Err(e) => tracing::warn!("could not hold task {} for {run}: {e:#}", t.id),
1163                    }
1164                }
1165                return;
1166            }
1167        }
1168    }
1169
1170    /// Give back the hold [`Watcher::hold_task`] put on a task, once the
1171    /// release succeeded. Only a machine hold carrying our marker is undone: a
1172    /// task the owner has since re-held or changed is left alone.
1173    ///
1174    /// Returns whether nothing is left to give back: `false` only when the
1175    /// write failed, in which case `held_task` is kept so a later lap retries.
1176    #[must_use]
1177    fn unhold_task(&self, st: &mut WatchState) -> bool {
1178        let Some(id) = st.held_task.take() else {
1179            return true;
1180        };
1181        let q = crate::queue::Queue::at(self.home.join("queue"));
1182        let mut t = match q.get(&id) {
1183            Ok(t) => t,
1184            // Only a task file confirmed absent is "nothing to restore";
1185            // any other failure may be transient, so keep the record.
1186            Err(_) if matches!(q.path_of(&id).try_exists(), Ok(false)) => return true,
1187            Err(e) => {
1188                tracing::warn!("could not read task {id} to restore it after the release: {e:#}");
1189                st.held_task = Some(id);
1190                return false;
1191            }
1192        };
1193        let ours = t.status == crate::queue::TaskStatus::Held
1194            && t.hold_source == Some(crate::queue::HoldSource::Machine)
1195            && t.hold_reason
1196                .as_deref()
1197                .is_some_and(|r| r.starts_with(HOLD_PREFIX));
1198        if ours {
1199            t.succeed();
1200            if let Err(e) = q.put(&mut t) {
1201                tracing::warn!("could not restore task {id} after the release: {e:#}");
1202                st.held_task = Some(id);
1203                return false;
1204            }
1205        }
1206        true
1207    }
1208
1209    /// The release is done (`st.job` is finished): give the task back, then
1210    /// drop the record. If the task cannot be restored the record stays, so
1211    /// the next lap tries again instead of leaving the task Held for good.
1212    fn complete(&self, pr: &str, st: &mut WatchState) {
1213        if self.unhold_task(st) {
1214            self.finish(pr, st);
1215        } else {
1216            self.save(pr, st);
1217        }
1218    }
1219
1220    fn raise_released(&self, pr: &str, url: &str, job: &Job) {
1221        notices::raise_in(
1222            &self.home,
1223            Notice::info(&released_key(pr), format!("Released {} ({pr})", job.tag())).link(
1224                Link::Url {
1225                    url: url.to_owned(),
1226                },
1227            ),
1228        );
1229    }
1230
1231    /// Ask what to do about a held job. One question per failure.
1232    fn file_job_question(&self, pr: &str, st: &mut WatchState, job: &Job) {
1233        let last = job
1234            .log
1235            .last()
1236            .map(|l| format!("\nLast step: {}\n\n{}\n", l.name, l.tail))
1237            .unwrap_or_default();
1238        let mut q = Question::new(
1239            String::new(),
1240            crate::bump::NOTICE_NODE.to_owned(),
1241            "release-watch".to_owned(),
1242            format!("Release {} of {pr} is on hold: retry?", job.tag()),
1243            format!(
1244                "Pull request: {}\nWhy it stopped: {}\n{last}\n\
1245                 Full output is under the magi home's release-local directory.\n\n\
1246                 - `{RETRY}`: run it again from where it stopped (the tag is not \
1247                 recreated and finished commands are skipped).\n\
1248                 - `{LEAVE_IT}`: stop watching; release by hand.\n\n\
1249                 Nothing is retried on its own; silence is a hold.\n",
1250                job.pr_url,
1251                job.failed.clone().unwrap_or_default()
1252            ),
1253            vec![RETRY.to_owned(), LEAVE_IT.to_owned()],
1254        );
1255        match self.questions().put(&mut q) {
1256            Ok(()) => st.question = Some(q.id.clone()),
1257            Err(e) => tracing::warn!("could not file the question for {pr}: {e:#}"),
1258        }
1259    }
1260
1261    /// Read the owner's word on a held job. `true` means `retry`: the job was
1262    /// resumed. Files the question when there is none yet and the owner has not
1263    /// already held this exact failure.
1264    fn settle_job_question(&self, pr: &str, st: &mut WatchState, job: &mut Job) -> bool {
1265        let why = job.failed.clone().unwrap_or_default();
1266        if let Some(id) = st.question.clone() {
1267            match self.questions().get(&id) {
1268                Ok(q) if q.status.open() => return false,
1269                Ok(q) => {
1270                    st.question = None;
1271                    match &q.answer {
1272                        Some(Answer::Choice(c)) if c == RETRY => {
1273                            job.resume();
1274                            st.held = None;
1275                            return true;
1276                        }
1277                        Some(Answer::Choice(c)) if c == LEAVE_IT => {
1278                            st.ignored = true;
1279                            let _ = Notices::at(self.home.join("notifications"))
1280                                .dismiss(&notices::id_of(&notice_key(pr)));
1281                        }
1282                        // Abandoned or anything else: this failure stays held.
1283                        _ => st.held = Some(why),
1284                    }
1285                    return false;
1286                }
1287                Err(_) => st.question = None,
1288            }
1289        }
1290        if !st.ignored && st.held.as_deref() != Some(why.as_str()) {
1291            self.raise(pr, &st.url, failed_message(pr));
1292            self.file_job_question(pr, st, job);
1293        }
1294        false
1295    }
1296
1297    /// Apply a settled question's outcome to the record, once.
1298    fn apply_answer(&self, pr: &str, st: &mut WatchState, snap: &RollupView, q: &Question) {
1299        st.question = None;
1300        if st.applied.contains(&q.id) {
1301            return;
1302        }
1303        st.applied.push(q.id.clone());
1304        if st.applied.len() > 20 {
1305            st.applied.remove(0);
1306        }
1307        match &q.answer {
1308            Some(Answer::Choice(c)) if c == RERUN_AGAIN => {
1309                st.held = None;
1310                for c in snap.checks.iter().filter(|c| is_failed(c.verdict)) {
1311                    if let Some(r) = &c.run {
1312                        st.reruns.remove(r);
1313                    }
1314                }
1315            }
1316            Some(Answer::Choice(c)) if c == LEAVE_IT => {
1317                st.ignored = true;
1318                if let Err(e) = Notices::at(self.home.join("notifications"))
1319                    .dismiss(&notices::id_of(&notice_key(pr)))
1320                {
1321                    tracing::warn!("could not dismiss the notice for {pr}: {e:#}");
1322                }
1323            }
1324            // `hold`, no answer, abandoned: quiet until something changes.
1325            _ => st.held = Some(fingerprint(snap)),
1326        }
1327    }
1328
1329    /// The pull request is confirmed merged or closed: clear everything.
1330    fn finish(&self, pr: &str, st: &WatchState) {
1331        let _ =
1332            Notices::at(self.home.join("notifications")).dismiss(&notices::id_of(&notice_key(pr)));
1333        if let Some(id) = &st.question {
1334            let _ = self.questions().update(id, |q| {
1335                q.abandon("the release pull request is no longer open");
1336                Ok(())
1337            });
1338        }
1339        let _ = std::fs::remove_file(self.state_path(pr));
1340    }
1341}
1342
1343/// The task `magi serve` spawns. `settings` returns the checkouts to cover and
1344/// `[daemon] release_stall_minutes`; it is re-read every lap so a config
1345/// change is picked up, and `0` minutes switches the lap off.
1346pub(crate) async fn run(
1347    watcher: Watcher,
1348    settings: impl Fn() -> (Vec<PathBuf>, u64),
1349    stop: crate::daemon::Stop,
1350) {
1351    let halt = {
1352        let stop = stop.clone();
1353        move || stop.stopped()
1354    };
1355    while !stop.stopped() {
1356        let (repos, minutes) = settings();
1357        if minutes > 0 {
1358            let now = jiff::Timestamp::now().as_second();
1359            watcher
1360                .lap(&repos, (minutes as i64).saturating_mul(60), now, &halt)
1361                .await;
1362        }
1363        let mut slept = Duration::ZERO;
1364        while slept < LAP && !stop.stopped() {
1365            tokio::time::sleep(Duration::from_secs(1)).await;
1366            slept += Duration::from_secs(1);
1367        }
1368    }
1369}
1370
1371#[cfg(test)]
1372mod tests {
1373    use super::*;
1374    use crate::land::CheckView;
1375    use anyhow::Context;
1376    use std::sync::Mutex;
1377
1378    const URL: &str = "https://github.com/o/r/pull/7";
1379
1380    fn check(name: &str, v: Verdict, run: Option<&str>) -> CheckView {
1381        CheckView {
1382            name: name.to_owned(),
1383            verdict: v,
1384            run: run.map(str::to_owned),
1385            url: run.map(|r| format!("https://github.com/o/r/actions/runs/{r}/job/1")),
1386        }
1387    }
1388
1389    fn snap(state: PrLifecycle, head: &str, checks: Vec<CheckView>) -> RollupView {
1390        RollupView {
1391            url: URL.to_owned(),
1392            number: 7,
1393            state,
1394            head: head.to_owned(),
1395            checks,
1396        }
1397    }
1398
1399    fn open(checks: Vec<CheckView>) -> RollupView {
1400        snap(PrLifecycle::Open, "h1", checks)
1401    }
1402
1403    fn st_at(progress: i64) -> WatchState {
1404        WatchState {
1405            progress_at: progress,
1406            ..WatchState::default()
1407        }
1408    }
1409
1410    #[test]
1411    fn unreadable_is_unknown_and_merged_or_closed_is_done() {
1412        assert_eq!(decide(None, &st_at(0), 10, 60), Step::Unknown);
1413        for s in [PrLifecycle::Merged, PrLifecycle::Closed] {
1414            assert_eq!(
1415                decide(Some(&snap(s, "h", vec![])), &st_at(0), 10, 60),
1416                Step::Done
1417            );
1418        }
1419    }
1420
1421    #[test]
1422    fn a_failed_run_is_rerun_even_while_another_run_is_pending() {
1423        let s = open(vec![
1424            check("win", Verdict::Fail, Some("11")),
1425            check("lint", Verdict::Pending, Some("12")),
1426        ]);
1427        assert_eq!(
1428            decide(Some(&s), &st_at(0), 1, 3600),
1429            Step::Rerun(vec!["11".into()])
1430        );
1431    }
1432
1433    #[test]
1434    fn a_run_with_a_job_still_pending_is_not_complete() {
1435        let s = open(vec![
1436            check("win", Verdict::Fail, Some("11")),
1437            check("mac", Verdict::Pending, Some("11")),
1438        ]);
1439        assert_eq!(decide(Some(&s), &st_at(0), 1, 3600), Step::Wait);
1440    }
1441
1442    #[test]
1443    fn one_rerun_per_run_then_grace_then_escalation() {
1444        let s = open(vec![check("win", Verdict::Fail, Some("11"))]);
1445        let mut st = st_at(0);
1446        st.reruns.insert("11".into(), 100);
1447        assert_eq!(decide(Some(&s), &st, 100 + RERUN_GRACE - 1, 0), Step::Wait);
1448        assert_eq!(
1449            decide(Some(&s), &st, 100 + RERUN_GRACE, 0),
1450            Step::Escalate(Why::StillRed)
1451        );
1452    }
1453
1454    #[test]
1455    fn a_failure_with_no_run_id_goes_straight_to_a_human_once_settled() {
1456        let s = open(vec![check("ci/ext", Verdict::Fail, None)]);
1457        assert_eq!(
1458            decide(Some(&s), &st_at(0), 1, 0),
1459            Step::Escalate(Why::StillRed)
1460        );
1461        let s = open(vec![
1462            check("ci/ext", Verdict::Fail, None),
1463            check("x", Verdict::Pending, Some("5")),
1464        ]);
1465        assert_eq!(decide(Some(&s), &st_at(0), 1, 0), Step::Wait);
1466    }
1467
1468    #[test]
1469    fn no_progress_for_the_bounded_time_escalates_and_zero_disables_it() {
1470        let s = open(vec![check("a", Verdict::Pass, Some("1"))]);
1471        assert_eq!(decide(Some(&s), &st_at(0), 3599, 3600), Step::Wait);
1472        assert_eq!(
1473            decide(Some(&s), &st_at(0), 3600, 3600),
1474            Step::Escalate(Why::Stalled)
1475        );
1476        assert_eq!(decide(Some(&s), &st_at(0), 99_999, 0), Step::Wait);
1477    }
1478
1479    #[test]
1480    fn held_and_ignored_stay_quiet_until_the_fingerprint_moves() {
1481        let s = open(vec![check("a", Verdict::Fail, None)]);
1482        let mut st = st_at(0);
1483        st.held = Some(fingerprint(&s));
1484        assert_eq!(decide(Some(&s), &st, 99_999, 60), Step::Wait);
1485        let moved = open(vec![
1486            check("a", Verdict::Pass, None),
1487            check("b", Verdict::Fail, None),
1488        ]);
1489        assert_eq!(
1490            decide(Some(&moved), &st, 99_999, 60),
1491            Step::Escalate(Why::StillRed)
1492        );
1493        st.ignored = true;
1494        assert_eq!(decide(Some(&moved), &st, 99_999, 60), Step::Wait);
1495    }
1496
1497    #[test]
1498    fn observe_restarts_the_clock_only_on_change_and_forgets_reruns_on_a_new_head() {
1499        let mut st = st_at(0);
1500        let a = open(vec![check("a", Verdict::Pending, Some("1"))]);
1501        observe(&mut st, &a, 10);
1502        assert_eq!(st.progress_at, 10);
1503        observe(&mut st, &a, 50);
1504        assert_eq!(st.progress_at, 10);
1505        st.reruns.insert("1".into(), 5);
1506        observe(
1507            &mut st,
1508            &snap(PrLifecycle::Open, "h2", a.checks.clone()),
1509            60,
1510        );
1511        assert!(st.reruns.is_empty());
1512        assert_eq!(st.progress_at, 60);
1513    }
1514
1515    #[test]
1516    fn pr_key_names_owner_repo_and_number() {
1517        assert_eq!(pr_key(URL).as_deref(), Some("o/r#7"));
1518        assert_eq!(pr_key("https://github.com/o/r/issues/7"), None);
1519        assert_eq!(pr_key("nonsense"), None);
1520    }
1521
1522    #[test]
1523    fn notice_wording_does_not_vary_between_polls() {
1524        assert_eq!(rerun_message("o/r#7"), rerun_message("o/r#7"));
1525        assert!(
1526            !human_message("o/r#7")
1527                .chars()
1528                .any(|c| c.is_ascii_digit() && c != '7')
1529        );
1530    }
1531
1532    #[test]
1533    fn the_release_question_is_not_claimed_by_other_machinery() {
1534        let q = Question::new(
1535            String::new(),
1536            crate::bump::NOTICE_NODE.into(),
1537            "release-watch".into(),
1538            "s".into(),
1539            String::new(),
1540            vec![HOLD.into()],
1541        );
1542        // The watcher's question is served by a deputy, but still covers nothing.
1543        assert_eq!(
1544            crate::deputy::kind_of(&q),
1545            Some(crate::deputy::Kind::Release)
1546        );
1547        let n = Notice::warn("release-pr:o/r#7", "m").about([String::new()]);
1548        assert!(!notices::covers(&q, &n));
1549        // A choice-less notice on the same node (seat `bump`) gets no deputy.
1550        let mut plain = q.clone();
1551        plain.seat = "bump".into();
1552        assert_eq!(crate::deputy::kind_of(&plain), None);
1553    }
1554
1555    #[derive(Default)]
1556    struct Fake {
1557        snap: Mutex<Option<RollupView>>,
1558        reruns: Mutex<Vec<String>>,
1559        local: Mutex<bool>,
1560        info: Mutex<Option<PrInfo>>,
1561        merges: Mutex<Vec<String>>,
1562        /// `graph.land_approval` as read; `None` is an unreadable config.
1563        approval: Mutex<Option<bool>>,
1564        merge_fails: Mutex<bool>,
1565        /// `Config::discover` itself fails.
1566        config_fails: Mutex<bool>,
1567    }
1568
1569    impl ReleaseForge for std::sync::Arc<Fake> {
1570        fn list<'a>(&'a self, _: &'a Path) -> Fut<'a, Result<Vec<(String, String)>>> {
1571            Box::pin(async { Ok(vec![("chore/release-v1.0.0".to_owned(), URL.to_owned())]) })
1572        }
1573        fn snapshot<'a>(&'a self, _: &'a Path, _: &'a str) -> Fut<'a, Result<RollupView>> {
1574            let s = self.snap.lock().unwrap().clone();
1575            Box::pin(async move { s.context("unreadable") })
1576        }
1577        fn rerun<'a>(&'a self, _: &'a Path, run: &'a str) -> Fut<'a, Result<()>> {
1578            self.reruns.lock().unwrap().push(run.to_owned());
1579            Box::pin(async { Ok(()) })
1580        }
1581        fn config<'a>(&'a self, _: &'a Path) -> Fut<'a, Result<crate::config::Config>> {
1582            if *self.config_fails.lock().unwrap() {
1583                return Box::pin(async { bail!("unreadable config") });
1584            }
1585            let mut c = crate::config::Config::default();
1586            if *self.local.lock().unwrap() {
1587                c.release.mode = crate::config::ReleaseMode::Local;
1588                c.release.commands = vec!["true".to_owned()];
1589            }
1590            Box::pin(async move { Ok(c) })
1591        }
1592        fn info<'a>(&'a self, _: &'a Path, _: &'a str) -> Fut<'a, Result<PrInfo>> {
1593            let i = self.info.lock().unwrap().clone();
1594            Box::pin(async move { i.context("no info") })
1595        }
1596        fn merge<'a>(&'a self, _: &'a Path, _: &'a str, head: &'a str) -> Fut<'a, Result<bool>> {
1597            self.merges.lock().unwrap().push(head.to_owned());
1598            let fails = *self.merge_fails.lock().unwrap();
1599            Box::pin(async move { if fails { bail!("refused") } else { Ok(true) } })
1600        }
1601        fn land_approval<'a>(&'a self, _: &'a Path) -> Fut<'a, Result<bool>> {
1602            let a = *self.approval.lock().unwrap();
1603            Box::pin(async move { a.context("unreadable config") })
1604        }
1605    }
1606
1607    fn rig() -> (tempfile::TempDir, std::sync::Arc<Fake>, Watcher) {
1608        let dir = tempfile::tempdir().unwrap();
1609        let fake = std::sync::Arc::new(Fake::default());
1610        // The repository's default: `graph.land_approval` on.
1611        *fake.approval.lock().unwrap() = Some(true);
1612        let w = Watcher::new(Box::new(fake.clone()), dir.path().to_path_buf());
1613        (dir, fake, w)
1614    }
1615
1616    fn red() -> RollupView {
1617        open(vec![check("win", Verdict::Fail, Some("11"))])
1618    }
1619
1620    #[tokio::test]
1621    async fn reruns_once_survives_a_restart_then_escalates_with_one_question() {
1622        let (dir, fake, w) = rig();
1623        *fake.snap.lock().unwrap() = Some(red());
1624        let repo = PathBuf::from("/nowhere");
1625        let no = || false;
1626        w.lap(std::slice::from_ref(&repo), 3600, 1000, &no).await;
1627        assert_eq!(*fake.reruns.lock().unwrap(), vec!["11".to_owned()]);
1628
1629        // A restart is a new Watcher over the same home: no second rerun.
1630        let w2 = Watcher::new(Box::new(fake.clone()), dir.path().to_path_buf());
1631        w2.lap(
1632            std::slice::from_ref(&repo),
1633            3600,
1634            1000 + RERUN_GRACE - 1,
1635            &no,
1636        )
1637        .await;
1638        assert_eq!(fake.reruns.lock().unwrap().len(), 1);
1639        assert!(w2.questions().list().is_empty());
1640
1641        w2.lap(
1642            std::slice::from_ref(&repo),
1643            3600,
1644            1000 + RERUN_GRACE + 1,
1645            &no,
1646        )
1647        .await;
1648        w2.lap(
1649            std::slice::from_ref(&repo),
1650            3600,
1651            1000 + RERUN_GRACE + 400,
1652            &no,
1653        )
1654        .await;
1655        assert_eq!(fake.reruns.lock().unwrap().len(), 1);
1656        let qs = w2.questions().list();
1657        assert_eq!(qs.len(), 1, "asked once, not every lap");
1658        assert_eq!(qs[0].node, crate::bump::NOTICE_NODE);
1659        assert!(qs[0].detail.contains("win") && qs[0].detail.contains(URL));
1660        assert_eq!(qs[0].choices, vec![RERUN_AGAIN, HOLD, LEAVE_IT]);
1661        let ns = Notices::at(dir.path().join("notifications")).list();
1662        assert_eq!(ns.len(), 1);
1663        assert_eq!(ns[0].key, "release-pr:o/r#7");
1664        assert_eq!(
1665            ns[0].link,
1666            Some(Link::Url {
1667                url: URL.to_owned(),
1668            })
1669        );
1670    }
1671
1672    async fn escalated() -> (tempfile::TempDir, std::sync::Arc<Fake>, Watcher, String) {
1673        let (dir, fake, w) = rig();
1674        *fake.snap.lock().unwrap() = Some(red());
1675        let repo = PathBuf::from("/nowhere");
1676        let no = || false;
1677        w.lap(std::slice::from_ref(&repo), 3600, 1000, &no).await;
1678        w.lap(&[repo], 3600, 2000, &no).await;
1679        let id = w.questions().list()[0].id.clone();
1680        (dir, fake, w, id)
1681    }
1682
1683    fn answer(w: &Watcher, id: &str, choice: &str) {
1684        w.questions()
1685            .update(id, |q| q.answer(Answer::Choice(choice.to_owned())))
1686            .unwrap();
1687    }
1688
1689    #[tokio::test]
1690    async fn rerun_again_reruns_exactly_once_more() {
1691        let (_d, fake, w, id) = escalated().await;
1692        answer(&w, &id, RERUN_AGAIN);
1693        let repo = PathBuf::from("/nowhere");
1694        let no = || false;
1695        w.lap(std::slice::from_ref(&repo), 3600, 3000, &no).await;
1696        w.lap(&[repo], 3600, 3001, &no).await;
1697        assert_eq!(fake.reruns.lock().unwrap().len(), 2);
1698    }
1699
1700    #[tokio::test]
1701    async fn hold_and_silence_do_not_reask_and_leave_it_dismisses() {
1702        let (d, fake, w, id) = escalated().await;
1703        answer(&w, &id, HOLD);
1704        let repo = PathBuf::from("/nowhere");
1705        let no = || false;
1706        for t in [3000, 90_000, 200_000] {
1707            w.lap(std::slice::from_ref(&repo), 3600, t, &no).await;
1708        }
1709        assert_eq!(w.questions().list().len(), 1, "held: no second question");
1710        assert_eq!(fake.reruns.lock().unwrap().len(), 1);
1711
1712        let (d2, _f2, w2, id2) = escalated().await;
1713        answer(&w2, &id2, LEAVE_IT);
1714        w2.lap(std::slice::from_ref(&repo), 3600, 3000, &no).await;
1715        let ns = Notices::at(d2.path().join("notifications")).list();
1716        assert!(ns.is_empty(), "dismissed notices are hidden");
1717        drop(d);
1718    }
1719
1720    /// Give the question a deputy seat the way `Deputies::attach` + a turn do.
1721    fn with_deputy(w: &Watcher, id: &str, home: &Path) -> String {
1722        let seat = crate::agent::SeatState::new(&crate::ask::deputy_seat_key(id), "a", 1);
1723        let key = seat.key.clone();
1724        let brief = deputy_brief(&w.questions().get(id).unwrap(), home);
1725        w.questions()
1726            .update(id, |q| {
1727                let mut d = crate::ask::Deputy::new(brief);
1728                d.seat = Some(seat);
1729                q.deputy = Some(d);
1730                Ok(())
1731            })
1732            .unwrap();
1733        key
1734    }
1735
1736    #[tokio::test]
1737    async fn the_brief_names_the_pull_request_and_what_leave_it_really_does() {
1738        let (d, _f, w, id) = escalated().await;
1739        let q = w.questions().get(&id).unwrap();
1740        let b = deputy_brief(&q, d.path());
1741        assert!(b.contains(URL), "{b}");
1742        assert!(
1743            b.contains("`leave it`") && b.contains("does NOT close"),
1744            "{b}"
1745        );
1746        assert!(b.contains("`rerun again`") && b.contains("`hold`"), "{b}");
1747        assert!(!b.contains("`merge`"), "no merge on an escalation: {b}");
1748
1749        // No record naming the question: said, not invented.
1750        let empty = tempfile::tempdir().unwrap();
1751        let b = deputy_brief(&q, empty.path());
1752        assert!(b.contains("could not be found") && !b.contains(URL), "{b}");
1753    }
1754
1755    #[tokio::test]
1756    async fn a_settled_leave_it_is_applied_once_by_the_watcher_and_closes_nothing() {
1757        let (d, fake, w, id) = escalated().await;
1758        let seat = with_deputy(&w, &id, d.path());
1759        let say = "クローズしていいよ。private repo だから、何回やっても失敗しちゃうから";
1760        w.questions().update(&id, |q| q.say(say)).unwrap();
1761        w.questions()
1762            .update(&id, |q| {
1763                q.settle_by_deputy(&seat, LEAVE_IT, "クローズしていいよ", None)
1764            })
1765            .unwrap();
1766        let q = w.questions().get(&id).unwrap();
1767        assert_eq!(q.answer, Some(Answer::Choice(LEAVE_IT.to_owned())));
1768
1769        let repo = PathBuf::from("/nowhere");
1770        let no = || false;
1771        for t in [3000, 3100] {
1772            w.lap(std::slice::from_ref(&repo), 3600, t, &no).await;
1773        }
1774        let st = w.stored().pop().unwrap();
1775        assert!(st.ignored, "watching stopped");
1776        assert_eq!(st.applied, vec![id.clone()], "applied exactly once");
1777        assert!(
1778            Notices::at(d.path().join("notifications"))
1779                .list()
1780                .is_empty()
1781        );
1782        assert_eq!(fake.reruns.lock().unwrap().len(), 1, "no extra rerun");
1783        assert_eq!(w.questions().list().len(), 1, "no second question");
1784    }
1785
1786    #[tokio::test]
1787    async fn the_task_of_a_release_question_comes_from_the_watch_record() {
1788        let (d, _f, w, id) = escalated().await;
1789        let q = w.questions().get(&id).unwrap();
1790        let queue = crate::queue::Queue::at(d.path().join("queue"));
1791        assert!(
1792            task_of_question(d.path(), &q, &queue).is_none(),
1793            "no run known"
1794        );
1795        assert!(task_of_question(d.path(), &q, &queue).is_none());
1796        let mut st = w.stored().pop().unwrap();
1797        st.run = Some("run-1".into());
1798        w.save("o/r#7", &st);
1799        // A known run with no task is still nothing to refuse on.
1800        assert!(task_of_question(d.path(), &q, &queue).is_none());
1801        let mut t = crate::queue::Task::new(
1802            "t".into(),
1803            "i".into(),
1804            PathBuf::from("/nowhere"),
1805            crate::queue::Source::Human,
1806        );
1807        t.runs.push("run-1".into());
1808        queue.put(&mut t).unwrap();
1809        assert_eq!(task_of_question(d.path(), &q, &queue).unwrap().id, t.id);
1810    }
1811
1812    #[tokio::test]
1813    async fn merged_clears_the_notice_the_question_and_the_record() {
1814        let (d, fake, w, _id) = escalated().await;
1815        *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Merged, "h1", vec![]));
1816        let repo = PathBuf::from("/nowhere");
1817        w.lap(&[repo], 3600, 5000, &(|| false)).await;
1818        assert!(
1819            Notices::at(d.path().join("notifications"))
1820                .list()
1821                .is_empty()
1822        );
1823        assert!(w.questions().list().iter().all(|q| !q.status.open()));
1824        assert!(w.stored().is_empty());
1825    }
1826
1827    #[tokio::test]
1828    async fn an_unreadable_forge_changes_nothing() {
1829        let (d, fake, w) = rig();
1830        *fake.snap.lock().unwrap() = None;
1831        w.lap(&[PathBuf::from("/nowhere")], 60, 99_999, &(|| false))
1832            .await;
1833        assert!(fake.reruns.lock().unwrap().is_empty());
1834        assert!(w.questions().list().is_empty());
1835        assert!(!d.path().join("release-watch").exists());
1836    }
1837
1838    #[tokio::test]
1839    async fn no_rerun_when_the_record_cannot_be_saved() {
1840        let (d, fake, w) = rig();
1841        *fake.snap.lock().unwrap() = Some(red());
1842        // A file where the directory should be makes every save fail.
1843        std::fs::write(d.path().join("release-watch"), "x").unwrap();
1844        w.lap(&[PathBuf::from("/nowhere")], 3600, 1000, &(|| false))
1845            .await;
1846        assert!(fake.reruns.lock().unwrap().is_empty());
1847    }
1848
1849    #[tokio::test]
1850    async fn a_park_stops_the_lap_before_any_call() {
1851        let (_d, fake, w) = rig();
1852        *fake.snap.lock().unwrap() = Some(red());
1853        w.lap(&[PathBuf::from("/nowhere")], 60, 1000, &(|| true))
1854            .await;
1855        assert!(fake.reruns.lock().unwrap().is_empty());
1856    }
1857
1858    // ---- [release] mode = "local" ----
1859
1860    fn local_st(asked: Option<&str>, held: Option<&str>) -> WatchState {
1861        WatchState {
1862            asked_head: asked.map(str::to_owned),
1863            held_head: held.map(str::to_owned),
1864            ..WatchState::default()
1865        }
1866    }
1867
1868    #[test]
1869    fn local_decisions_never_wait_on_ci_and_bind_approval_to_the_head() {
1870        use Approved::*;
1871        let none = local_st(None, None);
1872        assert_eq!(decide_local("h1", &none, NoQuestion, true), LocalStep::Ask);
1873        assert_eq!(decide_local("h1", &none, Open, true), LocalStep::Wait);
1874        let asked = local_st(Some("h1"), None);
1875        assert_eq!(
1876            decide_local("h1", &asked, Merge, true),
1877            LocalStep::Merge("h1".to_owned())
1878        );
1879        assert_eq!(
1880            decide_local("h1", &asked, Hold, true),
1881            LocalStep::Hold("h1".to_owned())
1882        );
1883        // The head moved after the owner answered: ask again, never merge.
1884        assert_eq!(decide_local("h2", &asked, Merge, true), LocalStep::Ask);
1885        assert_eq!(decide_local("h2", &asked, Hold, true), LocalStep::Ask);
1886        // A held head stays quiet until it moves; ignored stays quiet.
1887        let held = local_st(Some("h1"), Some("h1"));
1888        assert_eq!(decide_local("h1", &held, NoQuestion, true), LocalStep::Wait);
1889        assert_eq!(decide_local("h2", &held, NoQuestion, true), LocalStep::Ask);
1890        let mut ign = local_st(None, None);
1891        ign.ignored = true;
1892        assert_eq!(decide_local("h1", &ign, NoQuestion, true), LocalStep::Wait);
1893    }
1894
1895    #[test]
1896    fn without_land_approval_an_unasked_head_merges_but_stops_stay_stopped() {
1897        use Approved::*;
1898        let none = local_st(None, None);
1899        assert_eq!(
1900            decide_local("h1", &none, NoQuestion, false),
1901            LocalStep::Merge("h1".to_owned())
1902        );
1903        // An open question is respected, a held head and ignored stay quiet.
1904        assert_eq!(decide_local("h1", &none, Open, false), LocalStep::Wait);
1905        let held = local_st(Some("h1"), Some("h1"));
1906        assert_eq!(
1907            decide_local("h1", &held, NoQuestion, false),
1908            LocalStep::Wait
1909        );
1910        assert_eq!(
1911            decide_local("h2", &held, NoQuestion, false),
1912            LocalStep::Merge("h2".to_owned())
1913        );
1914        let mut ign = local_st(None, None);
1915        ign.ignored = true;
1916        assert_eq!(decide_local("h1", &ign, NoQuestion, false), LocalStep::Wait);
1917        // The owner's recorded no for this head still holds.
1918        let asked = local_st(Some("h1"), None);
1919        assert_eq!(
1920            decide_local("h1", &asked, Hold, false),
1921            LocalStep::Hold("h1".to_owned())
1922        );
1923    }
1924
1925    #[tokio::test]
1926    async fn a_registered_local_pr_still_asks_but_never_merges_while_the_config_is_unreadable() {
1927        let (dir, fake, w) = rig();
1928        *fake.config_fails.lock().unwrap() = true;
1929        *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Open, "h1", vec![]));
1930        let repo = PathBuf::from("/nowhere");
1931        let no = || false;
1932        let pr = pr_key(URL).unwrap();
1933        register(dir.path(), &repo, URL, "run1");
1934        // `register` goes through the real forge; make sure the record exists.
1935        assert!(w.load(&pr).local);
1936        w.lap(std::slice::from_ref(&repo), 60, 1, &no).await;
1937        w.lap(std::slice::from_ref(&repo), 60, 2, &no).await;
1938        assert!(fake.merges.lock().unwrap().is_empty());
1939        let qs = w.questions().list();
1940        assert_eq!(qs.len(), 1);
1941        assert_eq!(qs[0].choices, vec![land::APPROVE, land::HOLD]);
1942        // The owner says merge: nothing merges and the answer is not consumed.
1943        answer(&w, &qs[0].id, land::APPROVE);
1944        w.lap(std::slice::from_ref(&repo), 60, 3, &no).await;
1945        assert!(fake.merges.lock().unwrap().is_empty());
1946        assert_eq!(w.load(&pr).question.as_deref(), Some(qs[0].id.as_str()));
1947        // The config is readable again: the kept answer merges.
1948        *fake.config_fails.lock().unwrap() = false;
1949        *fake.local.lock().unwrap() = true;
1950        w.lap(std::slice::from_ref(&repo), 60, 4, &no).await;
1951        assert_eq!(*fake.merges.lock().unwrap(), vec!["h1".to_owned()]);
1952    }
1953
1954    #[tokio::test]
1955    async fn a_local_record_written_before_the_flag_existed_is_still_local() {
1956        let (_d, fake, w) = rig();
1957        *fake.config_fails.lock().unwrap() = true;
1958        *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Open, "h1", vec![]));
1959        let repo = PathBuf::from("/nowhere");
1960        let no = || false;
1961        let pr = pr_key(URL).unwrap();
1962        let old = WatchState {
1963            repo: repo.to_string_lossy().into_owned(),
1964            url: URL.to_owned(),
1965            run: Some("run1".to_owned()),
1966            ..WatchState::default()
1967        };
1968        assert!(!old.local);
1969        w.save(&pr, &old);
1970        w.lap(std::slice::from_ref(&repo), 60, 1, &no).await;
1971        assert!(fake.merges.lock().unwrap().is_empty());
1972        let qs = w.questions().list();
1973        assert_eq!(qs.len(), 1);
1974        assert_eq!(qs[0].choices, vec![land::APPROVE, land::HOLD]);
1975    }
1976
1977    #[tokio::test]
1978    async fn an_unregistered_pr_is_an_actions_release_when_the_config_is_unreadable() {
1979        let (_d, fake, w) = rig();
1980        *fake.config_fails.lock().unwrap() = true;
1981        *fake.snap.lock().unwrap() = Some(red());
1982        let repo = PathBuf::from("/nowhere");
1983        let no = || false;
1984        w.lap(std::slice::from_ref(&repo), 3600, 1000, &no).await;
1985        assert_eq!(*fake.reruns.lock().unwrap(), vec!["11".to_owned()]);
1986    }
1987
1988    #[tokio::test]
1989    async fn a_merged_local_pr_waits_with_its_record_while_the_config_is_unreadable() {
1990        let (dir, fake, w) = rig();
1991        *fake.config_fails.lock().unwrap() = true;
1992        *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Merged, "h1", vec![]));
1993        let repo = PathBuf::from("/nowhere");
1994        let no = || false;
1995        let pr = pr_key(URL).unwrap();
1996        register(dir.path(), &repo, URL, "run1");
1997        w.lap(std::slice::from_ref(&repo), 60, 1, &no).await;
1998        assert!(w.state_path(&pr).exists(), "the record is kept");
1999    }
2000
2001    #[tokio::test]
2002    async fn land_approval_off_merges_the_observed_head_without_a_question() {
2003        let (_d, fake, w) = rig();
2004        *fake.local.lock().unwrap() = true;
2005        *fake.approval.lock().unwrap() = Some(false);
2006        *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Open, "h1", vec![]));
2007        let repo = PathBuf::from("/nowhere");
2008        let no = || false;
2009        w.lap(std::slice::from_ref(&repo), 60, 1, &no).await;
2010        assert_eq!(*fake.merges.lock().unwrap(), vec!["h1".to_owned()]);
2011        assert!(w.questions().list().is_empty(), "no approval question");
2012    }
2013
2014    #[tokio::test]
2015    async fn land_approval_on_or_unreadable_still_asks_and_never_merges() {
2016        for approval in [Some(true), None] {
2017            let (_d, fake, w) = rig();
2018            *fake.local.lock().unwrap() = true;
2019            *fake.approval.lock().unwrap() = approval;
2020            *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Open, "h1", vec![]));
2021            let repo = PathBuf::from("/nowhere");
2022            let no = || false;
2023            w.lap(std::slice::from_ref(&repo), 60, 1, &no).await;
2024            w.lap(std::slice::from_ref(&repo), 60, 2, &no).await;
2025            assert!(fake.merges.lock().unwrap().is_empty(), "silence is a hold");
2026            let qs = w.questions().list();
2027            assert_eq!(qs.len(), 1);
2028            assert_eq!(qs[0].choices, vec![land::APPROVE, land::HOLD]);
2029        }
2030    }
2031
2032    #[tokio::test]
2033    async fn land_approval_off_does_not_merge_on_an_unreadable_forge_or_retry_a_refusal() {
2034        let (_d, fake, w) = rig();
2035        *fake.local.lock().unwrap() = true;
2036        *fake.approval.lock().unwrap() = Some(false);
2037        let repo = PathBuf::from("/nowhere");
2038        let no = || false;
2039        // Unreadable forge answer: nothing moves.
2040        *fake.snap.lock().unwrap() = None;
2041        w.lap(std::slice::from_ref(&repo), 60, 1, &no).await;
2042        assert!(fake.merges.lock().unwrap().is_empty());
2043        // A refused merge holds that head: one attempt, no retry loop.
2044        *fake.merge_fails.lock().unwrap() = true;
2045        *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Open, "h1", vec![]));
2046        w.lap(std::slice::from_ref(&repo), 60, 2, &no).await;
2047        w.lap(std::slice::from_ref(&repo), 60, 3, &no).await;
2048        assert_eq!(fake.merges.lock().unwrap().len(), 1);
2049        assert!(w.questions().list().is_empty());
2050        // A moved head is judged afresh.
2051        *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Open, "h2", vec![]));
2052        w.lap(std::slice::from_ref(&repo), 60, 4, &no).await;
2053        assert_eq!(
2054            *fake.merges.lock().unwrap(),
2055            vec!["h1".to_owned(), "h2".to_owned()]
2056        );
2057    }
2058
2059    #[test]
2060    fn pr_info_is_read_from_gh_json() {
2061        let i = parse_info(
2062            r#"{"headRefName":"chore/release-v1.2.3","title":"chore: release v1.2.3","mergeCommit":{"oid":"abc"}}"#,
2063        )
2064        .unwrap();
2065        assert_eq!(i.branch, "chore/release-v1.2.3");
2066        assert_eq!(i.merge_commit.as_deref(), Some("abc"));
2067        let open = parse_info(r#"{"headRefName":"b","title":"t","mergeCommit":null}"#).unwrap();
2068        assert_eq!(open.merge_commit, None);
2069    }
2070
2071    #[tokio::test]
2072    async fn a_local_release_asks_once_without_ci_reruns_or_stall_escalation() {
2073        let (_d, fake, w) = rig();
2074        *fake.local.lock().unwrap() = true;
2075        // Red checks that would rerun in Actions mode, and a stall clock far
2076        // past the limit: neither matters here.
2077        *fake.snap.lock().unwrap() = Some(red());
2078        let repo = PathBuf::from("/nowhere");
2079        let no = || false;
2080        w.lap(std::slice::from_ref(&repo), 60, 1_000_000, &no).await;
2081        w.lap(std::slice::from_ref(&repo), 60, 2_000_000, &no).await;
2082        assert!(fake.reruns.lock().unwrap().is_empty());
2083        let qs = w.questions().list();
2084        assert_eq!(qs.len(), 1, "one approval question, not one per lap");
2085        assert_eq!(qs[0].choices, vec![land::APPROVE, land::HOLD]);
2086        assert!(fake.merges.lock().unwrap().is_empty(), "silence is a hold");
2087    }
2088
2089    #[tokio::test]
2090    async fn merge_is_pinned_to_the_approved_head_and_a_moved_head_asks_again() {
2091        let (_d, fake, w) = rig();
2092        *fake.local.lock().unwrap() = true;
2093        *fake.snap.lock().unwrap() = Some(open(vec![]));
2094        let repo = PathBuf::from("/nowhere");
2095        let no = || false;
2096        w.lap(std::slice::from_ref(&repo), 60, 1, &no).await;
2097        let id = w.questions().list()[0].id.clone();
2098        // The head moves before the owner answers.
2099        *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Open, "h2", vec![]));
2100        answer(&w, &id, land::APPROVE);
2101        w.lap(std::slice::from_ref(&repo), 60, 2, &no).await;
2102        assert!(fake.merges.lock().unwrap().is_empty());
2103        assert_eq!(w.questions().list().len(), 2, "asked again about h2");
2104        // Answering the new question merges exactly h2.
2105        let id2 = w
2106            .questions()
2107            .list()
2108            .into_iter()
2109            .find(|q| q.status.open())
2110            .unwrap()
2111            .id;
2112        answer(&w, &id2, land::APPROVE);
2113        w.lap(std::slice::from_ref(&repo), 60, 3, &no).await;
2114        assert_eq!(*fake.merges.lock().unwrap(), vec!["h2".to_owned()]);
2115    }
2116
2117    #[tokio::test]
2118    async fn a_failed_release_holds_with_one_notice_and_one_question_and_never_retries() {
2119        let (d, fake, w) = rig();
2120        *fake.local.lock().unwrap() = true;
2121        *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Merged, "h1", vec![]));
2122        *fake.info.lock().unwrap() = Some(PrInfo {
2123            branch: "chore/release-v1.0.0".to_owned(),
2124            merge_commit: Some("deadbeef".to_owned()),
2125            title: "chore: release v1.0.0".to_owned(),
2126        });
2127        // The checkout is not a repository, so the release cannot start.
2128        let repo = d.path().join("nowhere");
2129        let no = || false;
2130        w.lap(std::slice::from_ref(&repo), 60, 1, &no).await;
2131        let st = w.load("o/r#7");
2132        let job = st.job.expect("the job is recorded");
2133        assert!(job.failed.is_some() && !job.tag_done && !job.finished);
2134        let qs = w.questions().list();
2135        assert_eq!(qs.len(), 1);
2136        assert_eq!(qs[0].choices, vec![RETRY, LEAVE_IT]);
2137        // Later laps change nothing: no retry, no second question.
2138        w.lap(std::slice::from_ref(&repo), 60, 2, &no).await;
2139        assert_eq!(w.questions().list().len(), 1);
2140        let ns = Notices::at(d.path().join("notifications")).list();
2141        assert_eq!(ns.len(), 1);
2142        assert_eq!(
2143            ns[0].link,
2144            Some(Link::Url {
2145                url: URL.to_owned(),
2146            })
2147        );
2148    }
2149
2150    #[tokio::test]
2151    async fn an_escalated_pull_request_records_the_title_version_as_acceptable() {
2152        let (d, fake, w) = rig();
2153        *fake.local.lock().unwrap() = true;
2154        *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Merged, "h1", vec![]));
2155        *fake.info.lock().unwrap() = Some(PrInfo {
2156            branch: "chore/release-v0.1.4".to_owned(),
2157            merge_commit: Some("deadbeef".to_owned()),
2158            title: "chore: release v0.2.0 (minor bump)".to_owned(),
2159        });
2160        let repo = d.path().join("nowhere");
2161        w.lap(std::slice::from_ref(&repo), 60, 1, &(|| false)).await;
2162        let job = w.load("o/r#7").job.expect("the job is recorded");
2163        assert_eq!(job.version, "0.1.4");
2164        assert_eq!(job.accepted, vec!["0.2.0".to_owned()]);
2165    }
2166
2167    #[tokio::test]
2168    async fn a_failed_release_holds_the_task_whose_run_opened_the_pr() {
2169        let (d, fake, w) = rig();
2170        *fake.local.lock().unwrap() = true;
2171        *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Merged, "h1", vec![]));
2172        *fake.info.lock().unwrap() = Some(PrInfo {
2173            branch: "chore/release-v1.0.0".to_owned(),
2174            merge_commit: Some("deadbeef".to_owned()),
2175            title: "t".to_owned(),
2176        });
2177        let repo = d.path().join("nowhere");
2178        let q = crate::queue::Queue::at(d.path().join("queue"));
2179        let mut t = crate::queue::Task::new(
2180            "t".to_owned(),
2181            "i".to_owned(),
2182            repo.clone(),
2183            crate::queue::Source::Human,
2184        );
2185        t.runs.push("run1".to_owned());
2186        t.status = crate::queue::TaskStatus::Done;
2187        q.put(&mut t).unwrap();
2188        register(d.path(), &repo, URL, "run1");
2189        w.lap(std::slice::from_ref(&repo), 60, 1, &(|| false)).await;
2190        let held = q.get(&t.id).unwrap();
2191        assert_eq!(held.status, crate::queue::TaskStatus::Held);
2192        assert!(
2193            held.hold_reason
2194                .unwrap_or_default()
2195                .contains("release v1.0.0")
2196        );
2197    }
2198
2199    #[tokio::test]
2200    async fn unhold_gives_the_task_back_only_when_the_hold_is_ours() {
2201        let (d, _fake, w) = rig();
2202        let q = crate::queue::Queue::at(d.path().join("queue"));
2203        let mut t = crate::queue::Task::new(
2204            "t".to_owned(),
2205            "i".to_owned(),
2206            PathBuf::from("/r"),
2207            crate::queue::Source::Human,
2208        );
2209        t.runs.push("run1".to_owned());
2210        t.status = crate::queue::TaskStatus::Done;
2211        q.put(&mut t).unwrap();
2212        let mut st = WatchState {
2213            run: Some("run1".to_owned()),
2214            ..WatchState::default()
2215        };
2216        w.hold_task(&mut st, "failed");
2217        assert_eq!(q.get(&t.id).unwrap().status, crate::queue::TaskStatus::Held);
2218        // A restart reconciling again does not hold twice.
2219        w.hold_task(&mut st, "failed again");
2220        assert!(w.unhold_task(&mut st));
2221        assert_eq!(q.get(&t.id).unwrap().status, crate::queue::TaskStatus::Done);
2222        assert!(st.held_task.is_none());
2223
2224        // The owner re-held it by hand with their own reason: left alone.
2225        let mut h = q.get(&t.id).unwrap();
2226        st.held_task = Some(h.id.clone());
2227        h.hold_manual(Some("mine".to_owned()));
2228        q.put(&mut h).unwrap();
2229        assert!(w.unhold_task(&mut st));
2230        assert_eq!(q.get(&t.id).unwrap().status, crate::queue::TaskStatus::Held);
2231    }
2232
2233    #[tokio::test]
2234    async fn a_restart_after_the_failure_was_saved_still_holds_the_task() {
2235        let (d, fake, w) = rig();
2236        *fake.local.lock().unwrap() = true;
2237        *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Merged, "h1", vec![]));
2238        let repo = d.path().join("nowhere");
2239        let q = crate::queue::Queue::at(d.path().join("queue"));
2240        let mut t = crate::queue::Task::new(
2241            "t".to_owned(),
2242            "i".to_owned(),
2243            repo.clone(),
2244            crate::queue::Source::Human,
2245        );
2246        t.runs.push("run1".to_owned());
2247        t.status = crate::queue::TaskStatus::Done;
2248        q.put(&mut t).unwrap();
2249        // The state a crash leaves: failure persisted, task never held.
2250        let mut job = Job::new("1.0.0", URL, "deadbeef");
2251        job.failed = Some("command 1 exited 3".to_owned());
2252        let st = WatchState {
2253            repo: repo.to_string_lossy().into_owned(),
2254            url: URL.to_owned(),
2255            run: Some("run1".to_owned()),
2256            job: Some(job),
2257            ..WatchState::default()
2258        };
2259        w.save("o/r#7", &st);
2260        w.lap(std::slice::from_ref(&repo), 60, 1, &(|| false)).await;
2261        assert_eq!(q.get(&t.id).unwrap().status, crate::queue::TaskStatus::Held);
2262    }
2263
2264    #[tokio::test]
2265    async fn a_restart_after_the_job_finished_still_gives_the_task_back() {
2266        let (d, fake, w) = rig();
2267        *fake.local.lock().unwrap() = true;
2268        *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Merged, "h1", vec![]));
2269        let repo = d.path().join("nowhere");
2270        let q = crate::queue::Queue::at(d.path().join("queue"));
2271        let mut t = crate::queue::Task::new(
2272            "t".to_owned(),
2273            "i".to_owned(),
2274            repo.clone(),
2275            crate::queue::Source::Human,
2276        );
2277        t.runs.push("run1".to_owned());
2278        t.hold_machine(Some(format!("{HOLD_PREFIX}release v1.0.0 failed")));
2279        q.put(&mut t).unwrap();
2280        // The state a crash leaves: finished saved, task never given back.
2281        let mut job = Job::new("1.0.0", URL, "deadbeef");
2282        job.finished = true;
2283        let st = WatchState {
2284            repo: repo.to_string_lossy().into_owned(),
2285            url: URL.to_owned(),
2286            run: Some("run1".to_owned()),
2287            job: Some(job),
2288            held_task: Some(t.id.clone()),
2289            ..WatchState::default()
2290        };
2291        w.save("o/r#7", &st);
2292        w.lap(std::slice::from_ref(&repo), 60, 1, &(|| false)).await;
2293        assert_eq!(q.get(&t.id).unwrap().status, crate::queue::TaskStatus::Done);
2294        assert!(!w.state_path("o/r#7").exists());
2295    }
2296
2297    #[tokio::test]
2298    async fn a_pull_request_registered_at_open_is_picked_up_even_if_unlisted() {
2299        let (d, _fake, w) = rig();
2300        register(d.path(), Path::new("/r"), URL, "run1");
2301        let st = w.load("o/r#7");
2302        assert_eq!(st.url, URL);
2303        assert_eq!(st.repo, "/r");
2304        // Never overwrites a live record.
2305        let mut live = st.clone();
2306        live.head = "keep".to_owned();
2307        w.save("o/r#7", &live);
2308        register(d.path(), Path::new("/r"), URL, "run1");
2309        assert_eq!(w.load("o/r#7").head, "keep");
2310    }
2311
2312    #[tokio::test]
2313    async fn a_finished_release_is_kept_on_the_run_and_a_missing_run_still_completes() {
2314        let (d, fake, w) = rig();
2315        *fake.local.lock().unwrap() = true;
2316        *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Merged, "h1", vec![]));
2317        let repo = d.path().join("nowhere");
2318        let mut run = RunState::new(
2319            repo.clone(),
2320            "main".to_owned(),
2321            "0123456789abcdef".to_owned(),
2322            "t".to_owned(),
2323            crate::config::Config::default(),
2324        );
2325        run.id = "run1".to_owned();
2326        run.save_under(d.path()).unwrap();
2327        let mut job = Job::new("1.0.0", URL, "deadbeef");
2328        job.finished = true;
2329        job.log.push(release_local::StepLog {
2330            name: "cmd".to_owned(),
2331            code: Some(0),
2332            tail: "ok".to_owned(),
2333            output: None,
2334        });
2335        let mk = |run: &str| WatchState {
2336            repo: repo.to_string_lossy().into_owned(),
2337            url: URL.to_owned(),
2338            run: Some(run.to_owned()),
2339            job: Some(job.clone()),
2340            ..WatchState::default()
2341        };
2342        w.save("o/r#7", &mk("run1"));
2343        w.lap(std::slice::from_ref(&repo), 60, 1, &(|| false)).await;
2344        assert!(!w.state_path("o/r#7").exists());
2345        let kept = RunState::load_under("run1", d.path()).unwrap();
2346        assert_eq!(kept.release_bump.unwrap().release, Some(job.clone()));
2347
2348        // A run that is gone must not keep the watcher record forever.
2349        w.save("o/r#7", &mk("gone"));
2350        w.lap(std::slice::from_ref(&repo), 60, 1, &(|| false)).await;
2351        assert!(!w.state_path("o/r#7").exists());
2352    }
2353
2354    #[tokio::test]
2355    async fn raise_released_attaches_url_link() {
2356        let (d, _fake, w) = rig();
2357        let job = Job::new("1.0.0", URL, "deadbeef");
2358        w.raise_released("o/r#7", URL, &job);
2359        let ns = Notices::at(d.path().join("notifications")).list();
2360        assert_eq!(ns.len(), 1);
2361        assert_eq!(ns[0].key, "release-done:o/r#7");
2362        assert_eq!(
2363            ns[0].link,
2364            Some(Link::Url {
2365                url: URL.to_owned(),
2366            })
2367        );
2368    }
2369}