1use 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
55const LAP: Duration = Duration::from_secs(300);
57
58pub(crate) const RERUN_GRACE: i64 = 180;
62
63pub const RERUN_AGAIN: &str = "rerun again";
66pub const HOLD: &str = "hold";
68pub const LEAVE_IT: &str = "leave it";
70pub const RETRY: &str = "retry";
72
73#[derive(Debug, Clone, PartialEq, Eq)]
75pub(crate) enum Step {
76 Done,
78 Wait,
80 Unknown,
82 Rerun(Vec<String>),
84 Escalate(Why),
86}
87
88#[derive(Debug, Clone, Copy, PartialEq, Eq)]
89pub(crate) enum Why {
90 StillRed,
92 Stalled,
94}
95
96#[derive(Debug, Clone, Default, Serialize, Deserialize)]
98#[serde(default)]
99pub(crate) struct WatchState {
100 pub repo: String,
102 pub url: String,
103 pub head: String,
105 pub reruns: BTreeMap<String, i64>,
107 pub fingerprint: String,
109 pub progress_at: i64,
110 pub question: Option<String>,
112 pub held: Option<String>,
114 pub ignored: bool,
116 pub applied: Vec<String>,
118 pub asked_head: Option<String>,
120 pub held_head: Option<String>,
123 pub run: Option<String>,
126 pub held_task: Option<String>,
129 pub job: Option<Job>,
131 pub local: bool,
134}
135
136impl WatchState {
137 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
149pub(crate) fn is_failed(v: Verdict) -> bool {
150 v == Verdict::Fail
151}
152
153pub(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
164pub(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
178pub(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 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
239pub(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
261fn 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 failed_workflows(snap: &RollupView) -> Vec<String> {
271 snap.checks
272 .iter()
273 .filter(|c| is_failed(c.verdict))
274 .filter_map(|c| c.run.clone())
275 .collect::<std::collections::BTreeSet<_>>()
276 .into_iter()
277 .collect()
278}
279
280fn maintenance_choices(snap: &RollupView) -> Vec<String> {
281 let mut choices = vec![HOLD.to_owned(), LEAVE_IT.to_owned()];
282 if !failed_workflows(snap).is_empty() {
283 choices.insert(0, RERUN_AGAIN.to_owned());
284 }
285 choices
286}
287
288fn question_detail(snap: &RollupView, why: Why, st: &WatchState) -> String {
289 let mut s = String::new();
290 s.push_str(&format!("Pull request: {}\n\n", snap.url));
291 s.push_str(match why {
292 Why::StillRed => "Checks are still red after the failed jobs were rerun once.\n\n",
293 Why::Stalled => "The pull request has made no progress for too long.\n\n",
294 });
295 let failed: Vec<_> = snap
296 .checks
297 .iter()
298 .filter(|c| is_failed(c.verdict))
299 .collect();
300 if failed.is_empty() {
301 s.push_str("No check is failing.\n");
302 } else {
303 s.push_str("Failed jobs:\n");
304 for c in failed {
305 match &c.url {
306 Some(u) => s.push_str(&format!("- {} ({u})\n", c.name)),
307 None => s.push_str(&format!("- {}\n", c.name)),
308 }
309 }
310 }
311 s.push_str(&format!(
312 "\nWorkflow runs rerun so far: {}.\n\n\
313 - `{RERUN_AGAIN}`: rerun the failed jobs once more.\n\
314 - `{HOLD}`: stay quiet until a check or the head changes.\n\
315 - `{LEAVE_IT}`: stop watching this pull request.\n\n\
316 Nothing is merged or pushed by magi; silence is a hold.\n",
317 st.reruns.len()
318 ));
319 if failed_workflows(snap).is_empty() {
320 s = s.replace(
321 &format!("- `{RERUN_AGAIN}`: rerun the failed jobs once more.\n"),
322 "",
323 );
324 }
325 s
326}
327
328#[derive(Debug, Clone, Copy, PartialEq, Eq)]
330pub(crate) enum Approved {
331 NoQuestion,
333 Open,
335 Merge,
337 Hold,
339}
340
341#[derive(Debug, Clone, PartialEq, Eq)]
343pub(crate) enum LocalStep {
344 Wait,
346 Ask,
348 Merge(String),
350 Hold(String),
352}
353
354pub(crate) fn decide_local(
362 head: &str,
363 st: &WatchState,
364 approved: Approved,
365 ask_first: bool,
366) -> LocalStep {
367 if st.ignored || st.held_head.as_deref() == Some(head) {
368 return LocalStep::Wait;
369 }
370 let about_this_head = st.asked_head.as_deref() == Some(head);
371 match approved {
372 Approved::Open => LocalStep::Wait,
373 Approved::NoQuestion if !ask_first => LocalStep::Merge(head.to_owned()),
374 Approved::NoQuestion => LocalStep::Ask,
375 Approved::Merge if about_this_head => LocalStep::Merge(head.to_owned()),
376 Approved::Hold if about_this_head => LocalStep::Hold(head.to_owned()),
377 Approved::Merge | Approved::Hold if !ask_first => LocalStep::Merge(head.to_owned()),
379 Approved::Merge | Approved::Hold => LocalStep::Ask,
380 }
381}
382
383pub(crate) fn register(home: &Path, repo: &Path, url: &str, run: &str) {
386 let Some(pr) = pr_key(url) else {
387 return;
388 };
389 let w = Watcher::new(Box::new(GhForge), home.to_path_buf());
390 if w.state_path(&pr).exists() {
391 return;
392 }
393 let st = WatchState {
394 repo: repo.to_string_lossy().into_owned(),
395 url: url.to_owned(),
396 run: Some(run.to_owned()),
397 local: true,
398 ..WatchState::default()
399 };
400 w.save(&pr, &st);
401}
402
403pub(crate) fn state_for_question(home: &Path, question_id: &str) -> Option<WatchState> {
406 Watcher::new(Box::new(GhForge), home.to_path_buf())
407 .stored()
408 .into_iter()
409 .find(|s| s.question.as_deref() == Some(question_id))
410}
411
412pub(crate) fn question_for_notice(home: &Path, key: &str) -> Option<String> {
414 let pr = key.strip_prefix("release-pr:")?;
415 let state = Watcher::new(Box::new(GhForge), home.to_path_buf()).load(pr);
416 (pr_key(&state.url).as_deref() == Some(pr))
417 .then_some(state.question)
418 .flatten()
419}
420
421pub fn task_of_question(
426 home: &Path,
427 q: &Question,
428 queue: &crate::queue::Queue,
429) -> Option<crate::queue::Task> {
430 let run = state_for_question(home, &q.id)?.run?;
431 queue.list().into_iter().find(|t| t.runs.contains(&run))
432}
433
434pub(crate) fn approval_target(home: &Path, q: &Question) -> Option<crate::merge_decision::Target> {
437 if q.node != crate::bump::NOTICE_NODE
438 || q.seat != "release-watch"
439 || !q.choices.iter().any(|c| c == land::APPROVE)
440 {
441 return None;
442 }
443 let state = state_for_question(home, &q.id)?;
444 Some(crate::merge_decision::Target {
445 pr: state.url,
446 head: state.asked_head?,
447 language: "en".into(),
448 })
449}
450
451pub(crate) fn deputy_brief(q: &Question, home: &Path) -> String {
453 let mut s = String::from(
454 "This question was filed by magi's release watcher about a release pull \
455 request. Whatever the owner picks, only the watcher acts on it, on its \
456 next lap; you apply nothing and never close, merge, rerun or push \
457 anything. Silence is a hold.",
458 );
459 match state_for_question(home, &q.id) {
460 Some(st) => {
461 s.push_str(&format!(
462 "\n\nPull request: {}\nCheckout: {}",
463 st.url, st.repo
464 ));
465 }
466 None => s.push_str(
467 "\n\nThe watcher's record for this question could not be found, so the \
468 pull request is known to you only through the question's own text. \
469 Say so to the owner rather than guessing.",
470 ),
471 }
472 if let Some(target) = &q.merge_target {
473 s.push_str(&format!("\n\nThis approval is bound to PR {} at head `{}`. A deputy merge requires the owner's whole latest message and quote to be exactly `merge {}`. A standing conversation can bind it as `{}: merge {}`. A generic batch instruction never authorizes merge.", target.pr, target.head, target.head, q.id, target.head));
474 }
475 s.push_str("\n\nWhat each option does:");
476 for c in &q.choices {
477 let what = match c.as_str() {
478 RERUN_AGAIN => {
479 "the daemon reruns the recorded failed workflows once if the PR head still matches"
480 }
481 HOLD if q.choices.iter().any(|c| c == land::APPROVE) => {
482 "leave the pull request open and unmerged; ask again only when the head moves"
483 }
484 HOLD => "stay quiet until a check or the head changes",
485 LEAVE_IT => {
486 "stop watching this pull request and dismiss its notice. It does NOT \
487 close the pull request"
488 }
489 RETRY => "run the failed release again from where it stopped",
490 land::APPROVE => {
491 "squash-merge exactly the head the question names - irreversible - \
492 then tag and release"
493 }
494 _ => "recorded as the answer",
495 };
496 s.push_str(&format!("\n- `{c}`: {what}"));
497 }
498 s.push_str(
499 "\n\nIf the owner's words ask for the pull request itself to be closed, that \
500 is something only they can do by hand; do not read it as a choice unless \
501 it plainly means stop watching.",
502 );
503 s
504}
505
506const HOLD_PREFIX: &str = "[release] ";
509
510fn approval_message(pr: &str) -> String {
511 format!("Release PR {pr} is waiting for your approval to merge")
512}
513
514fn failed_message(pr: &str) -> String {
515 format!("Release PR {pr} merged but the release is on hold")
516}
517
518fn released_key(pr: &str) -> String {
519 format!("release-done:{pr}")
520}
521
522type Fut<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;
523
524pub(crate) trait ReleaseForge: Send + Sync {
526 fn list<'a>(&'a self, repo: &'a Path) -> Fut<'a, Result<Vec<(String, String)>>>;
528 fn snapshot<'a>(&'a self, repo: &'a Path, url: &'a str) -> Fut<'a, Result<RollupView>>;
529 fn rerun<'a>(&'a self, repo: &'a Path, run: &'a str) -> Fut<'a, Result<()>>;
530 fn config<'a>(&'a self, _repo: &'a Path) -> Fut<'a, Result<crate::config::Config>> {
532 Box::pin(async { Ok(crate::config::Config::default()) })
533 }
534 fn land_approval<'a>(&'a self, _repo: &'a Path) -> Fut<'a, Result<bool>> {
537 Box::pin(async { Ok(true) })
538 }
539 fn info<'a>(&'a self, _repo: &'a Path, url: &'a str) -> Fut<'a, Result<PrInfo>> {
541 Box::pin(async move { bail!("no pull request info for {url}") })
542 }
543 fn merge<'a>(&'a self, _repo: &'a Path, url: &'a str, _head: &'a str) -> Fut<'a, Result<bool>> {
546 Box::pin(async move { bail!("cannot merge {url}") })
547 }
548}
549
550#[derive(Debug, Clone, Default, PartialEq, Eq)]
552pub(crate) struct PrInfo {
553 pub branch: String,
555 pub merge_commit: Option<String>,
557 pub title: String,
559}
560
561pub(crate) fn parse_info(json: &str) -> Result<PrInfo> {
563 let v: serde_json::Value = serde_json::from_str(json)?;
564 let text = |k: &str| {
565 v.get(k)
566 .and_then(|x| x.as_str())
567 .unwrap_or_default()
568 .to_owned()
569 };
570 Ok(PrInfo {
571 branch: text("headRefName"),
572 title: text("title"),
573 merge_commit: v
574 .get("mergeCommit")
575 .and_then(|m| m.get("oid"))
576 .and_then(|o| o.as_str())
577 .filter(|o| !o.is_empty())
578 .map(str::to_owned),
579 })
580}
581
582pub(crate) struct GhForge;
584
585impl ReleaseForge for GhForge {
586 fn list<'a>(&'a self, repo: &'a Path) -> Fut<'a, Result<Vec<(String, String)>>> {
587 Box::pin(crate::bump::list_open_release_prs(repo))
588 }
589
590 fn snapshot<'a>(&'a self, repo: &'a Path, url: &'a str) -> Fut<'a, Result<RollupView>> {
591 Box::pin(async move {
592 let args = [
593 "pr".to_owned(),
594 "view".to_owned(),
595 url.to_owned(),
596 "--json".to_owned(),
597 "url,number,state,headRefOid,statusCheckRollup".to_owned(),
598 ];
599 let (ok, out) = land::gh(repo, &args).await?;
600 if !ok {
601 bail!("gh pr view {url}: {out}");
602 }
603 land::parse_rollup(&out)
604 })
605 }
606
607 fn rerun<'a>(&'a self, repo: &'a Path, run: &'a str) -> Fut<'a, Result<()>> {
608 Box::pin(async move {
609 let args = [
610 "run".to_owned(),
611 "rerun".to_owned(),
612 run.to_owned(),
613 "--failed".to_owned(),
614 ];
615 let (ok, out) = land::gh(repo, &args).await?;
616 if !ok {
617 bail!("gh run rerun {run}: {out}");
618 }
619 Ok(())
620 })
621 }
622
623 fn config<'a>(&'a self, repo: &'a Path) -> Fut<'a, Result<crate::config::Config>> {
624 Box::pin(async move { Ok(crate::config::Config::discover(repo, None)?.0) })
625 }
626
627 fn land_approval<'a>(&'a self, repo: &'a Path) -> Fut<'a, Result<bool>> {
628 Box::pin(async move {
632 Ok(crate::config::Config::discover(repo, None)?
633 .0
634 .graph
635 .land_approval)
636 })
637 }
638
639 fn info<'a>(&'a self, repo: &'a Path, url: &'a str) -> Fut<'a, Result<PrInfo>> {
640 Box::pin(async move {
641 let args = [
642 "pr".to_owned(),
643 "view".to_owned(),
644 url.to_owned(),
645 "--json".to_owned(),
646 "headRefName,mergeCommit,title".to_owned(),
647 ];
648 let (ok, out) = land::gh(repo, &args).await?;
649 if !ok {
650 bail!("gh pr view {url}: {out}");
651 }
652 parse_info(&out)
653 })
654 }
655
656 fn merge<'a>(&'a self, repo: &'a Path, url: &'a str, head: &'a str) -> Fut<'a, Result<bool>> {
657 Box::pin(async move {
658 let title = self
659 .info(repo, url)
660 .await
661 .map(|i| i.title)
662 .unwrap_or_default();
663 let title = if title.trim().is_empty() {
664 "chore: release".to_owned()
665 } else {
666 title
667 };
668 let mut argv = crate::bump::bump_merge_argv(url, &title);
671 argv.push("--match-head-commit".to_owned());
672 argv.push(head.to_owned());
673 let (ok, out) = land::gh(repo, &argv).await?;
674 if ok {
675 return Ok(true);
676 }
677 let after = land::lifecycle(repo, url).await.ok();
680 if land::merged_after_all(&argv, &out, after).is_some() {
681 return Ok(true);
682 }
683 bail!("gh pr merge {url}: {out}")
684 })
685 }
686}
687
688pub(crate) struct Watcher {
690 forge: Box<dyn ReleaseForge>,
691 home: PathBuf,
692}
693
694impl Watcher {
695 pub(crate) fn new(forge: Box<dyn ReleaseForge>, home: PathBuf) -> Self {
696 Self { forge, home }
697 }
698
699 fn dir(&self) -> PathBuf {
700 self.home.join("release-watch")
701 }
702
703 fn state_path(&self, pr: &str) -> PathBuf {
704 self.dir().join(format!("{}.json", notices::id_of(pr)))
705 }
706
707 fn load(&self, pr: &str) -> WatchState {
708 std::fs::read_to_string(self.state_path(pr))
709 .ok()
710 .and_then(|s| serde_json::from_str(&s).ok())
711 .unwrap_or_default()
712 }
713
714 fn save(&self, pr: &str, st: &WatchState) -> bool {
715 let write = || -> Result<()> {
716 std::fs::create_dir_all(self.dir())?;
717 let path = self.state_path(pr);
718 let tmp = path.with_extension("json.tmp");
719 std::fs::write(&tmp, serde_json::to_string_pretty(st)?)?;
720 std::fs::rename(&tmp, &path)?;
721 Ok(())
722 };
723 match write() {
724 Ok(()) => true,
725 Err(e) => {
726 tracing::warn!("could not save the release watch for {pr}: {e:#}");
727 false
728 }
729 }
730 }
731
732 fn stored(&self) -> Vec<WatchState> {
734 let Ok(rd) = std::fs::read_dir(self.dir()) else {
735 return Vec::new();
736 };
737 rd.flatten()
738 .filter(|e| e.path().extension().is_some_and(|x| x == "json"))
739 .filter_map(|e| std::fs::read_to_string(e.path()).ok())
740 .filter_map(|s| serde_json::from_str(&s).ok())
741 .collect()
742 }
743
744 fn questions(&self) -> Questions {
745 Questions::at(self.home.join("questions"))
746 }
747
748 fn raise(&self, pr: &str, url: &str, message: String) {
749 notices::raise_in(
750 &self.home,
751 Notice::warn(¬ice_key(pr), message).link(Link::Url {
752 url: url.to_owned(),
753 }),
754 );
755 }
756
757 pub(crate) async fn lap(
760 &self,
761 repos: &[PathBuf],
762 stall_secs: i64,
763 now: i64,
764 halt: &(dyn Fn() -> bool + Sync),
765 ) {
766 let stored = self.stored();
767 let mut repos = repos.to_vec();
769 for s in &stored {
770 let p = PathBuf::from(&s.repo);
771 if !s.repo.is_empty() && !repos.contains(&p) {
772 repos.push(p);
773 }
774 }
775 for repo in &repos {
776 if halt() {
777 return;
778 }
779 let mut urls: Vec<String> = match self.forge.list(repo).await {
780 Ok(v) => v.into_iter().map(|(_, u)| u).collect(),
781 Err(e) => {
782 tracing::warn!(
783 "could not list release pull requests in {}: {e:#}",
784 repo.display()
785 );
786 Vec::new()
787 }
788 };
789 let here = repo.to_string_lossy();
792 for s in stored.iter().filter(|s| s.repo == here.as_ref()) {
793 if !urls.contains(&s.url) {
794 urls.push(s.url.clone());
795 }
796 }
797 for url in urls {
798 if halt() {
799 return;
800 }
801 self.watch(repo, &url, stall_secs, now, halt).await;
802 }
803 }
804 }
805
806 pub(crate) async fn watch(
808 &self,
809 repo: &Path,
810 url: &str,
811 stall_secs: i64,
812 now: i64,
813 halt: &(dyn Fn() -> bool + Sync),
814 ) {
815 let Some(pr) = pr_key(url) else {
816 tracing::warn!("not a pull request url: {url}");
817 return;
818 };
819 let mut st = self.load(&pr);
820 st.repo = repo.to_string_lossy().into_owned();
821 st.url = url.to_owned();
822
823 let snap = match self.forge.snapshot(repo, url).await {
824 Ok(s) => Some(s),
825 Err(e) => {
826 tracing::warn!("could not read {url}: {e:#}");
827 None
828 }
829 };
830 match self.forge.config(repo).await {
831 Ok(cfg) if cfg.release.is_local() => {
832 st.local = true;
833 self.watch_local(repo, &pr, st, snap, Some(&cfg)).await;
834 return;
835 }
836 Ok(_) => st.local = false,
837 Err(e) if st.is_local() => {
838 tracing::warn!(
839 "could not read the config of {}: {e:#}; watching {url} as a local release without it",
840 repo.display()
841 );
842 self.watch_local(repo, &pr, st, snap, None).await;
843 return;
844 }
845 Err(e) => tracing::warn!(
846 "could not read the config of {}: {e:#}; watching {url} as an Actions release",
847 repo.display()
848 ),
849 }
850 if decide(snap.as_ref(), &st, now, stall_secs) == Step::Done {
851 self.finish(&pr, &st);
852 return;
853 }
854 let Some(snap) = snap else {
855 return;
856 };
857
858 if let Some(id) = st.question.clone() {
860 match self.questions().get(&id) {
861 Ok(q) if q.status.open() => return,
862 Ok(q) => self.apply_answer(&pr, &mut st, &snap, &q, now),
863 Err(_) => st.question = None,
864 }
865 observe(&mut st, &snap, now);
867 self.save(&pr, &st);
868 } else {
869 observe(&mut st, &snap, now);
870 }
871
872 match decide(Some(&snap), &st, now, stall_secs) {
873 Step::Done | Step::Unknown | Step::Wait => {}
874 Step::Rerun(runs) => {
875 for r in &runs {
877 st.reruns.insert(r.clone(), now);
878 }
879 if !self.save(&pr, &st) || halt() {
882 return;
883 }
884 let mut ok = false;
885 for r in &runs {
886 match self.forge.rerun(repo, r).await {
887 Ok(()) => ok = true,
888 Err(e) => tracing::warn!("could not rerun run {r} of {url}: {e:#}"),
889 }
890 }
891 if ok {
892 self.raise(&pr, url, rerun_message(&pr));
893 }
894 }
895 Step::Escalate(why) => {
896 self.raise(&pr, url, human_message(&pr));
897 let mut q = Question::new(
898 String::new(),
899 crate::bump::NOTICE_NODE.to_owned(),
900 "release-watch".to_owned(),
901 format!("Release PR {pr} is stuck: what now?"),
902 question_detail(&snap, why, &st),
903 maintenance_choices(&snap),
904 );
905 if q.choices.iter().any(|c| c == RERUN_AGAIN) {
906 q.actions.insert(
907 RERUN_AGAIN.into(),
908 crate::ask::ChoiceAction::RerunChecks {
909 repo: repo.to_path_buf(),
910 pr: snap.url.clone(),
911 head: snap.head.clone(),
912 runs: failed_workflows(&snap).into_iter().take(20).collect(),
913 },
914 );
915 }
916 match self.questions().put_executable(&mut q) {
917 Ok(()) => st.question = Some(q.id.clone()),
918 Err(e) => tracing::warn!("could not file the question for {pr}: {e:#}"),
919 }
920 }
921 }
922 self.save(&pr, &st);
923 }
924
925 async fn watch_local(
927 &self,
928 repo: &Path,
929 pr: &str,
930 mut st: WatchState,
931 snap: Option<RollupView>,
932 cfg: Option<&crate::config::Config>,
933 ) {
934 let Some(snap) = snap else {
935 return;
936 };
937 match snap.state {
938 PrLifecycle::Closed => self.finish(pr, &st),
939 PrLifecycle::Merged => match cfg {
940 Some(cfg) => self.release_merged(repo, pr, st, cfg).await,
941 None => {
943 tracing::warn!("{pr} is merged but its config is unreadable; not released yet")
944 }
945 },
946 PrLifecycle::Open => {
947 let approved = match st.question.clone() {
948 None => Approved::NoQuestion,
949 Some(id) => match self.questions().get(&id) {
950 Ok(q) if q.status.open() => Approved::Open,
951 Ok(q) => match &q.answer {
952 Some(Answer::Choice(c))
953 if land::approval(Some(c)) == land::Approval::Merge =>
954 {
955 Approved::Merge
956 }
957 _ => Approved::Hold,
958 },
959 Err(_) => Approved::NoQuestion,
960 },
961 };
962 let ask_first = if cfg.is_none() {
963 true
964 } else {
965 match self.forge.land_approval(repo).await {
966 Ok(v) => v,
967 Err(e) => {
968 tracing::warn!("could not read graph.land_approval for {pr}: {e:#}");
969 true
970 }
971 }
972 };
973 let step = decide_local(&snap.head, &st, approved, ask_first);
974 if cfg.is_none() && matches!(step, LocalStep::Merge(_)) {
975 tracing::warn!("not merging {pr} while its config is unreadable");
977 return;
978 }
979 if matches!(
980 step,
981 LocalStep::Merge(_) | LocalStep::Hold(_) | LocalStep::Ask
982 ) {
983 st.question = None;
985 }
986 match step {
987 LocalStep::Wait => {}
988 LocalStep::Hold(head) => st.held_head = Some(head),
989 LocalStep::Ask => {
990 st.held_head = None;
991 st.asked_head = Some(snap.head.clone());
992 self.raise(pr, &snap.url, approval_message(pr));
993 let mut q = Question::new(
994 String::new(),
995 crate::bump::NOTICE_NODE.to_owned(),
996 "release-watch".to_owned(),
997 format!("Release PR {pr}: merge it and release?"),
998 format!(
999 "Pull request: {}\nHead: {}\n\n\
1000 `[release] mode = \"local\"`: no CI is awaited. `{}` merges \
1001 exactly this head, then magi tags the merge commit and runs the \
1002 configured release commands. `{}` leaves the pull request \
1003 open. Silence is a hold.\n",
1004 snap.url,
1005 snap.head,
1006 land::APPROVE,
1007 land::HOLD
1008 ),
1009 vec![land::APPROVE.to_owned(), land::HOLD.to_owned()],
1010 );
1011 q.merge_target = Some(crate::merge_decision::Target {
1012 pr: snap.url.clone(),
1013 head: snap.head.clone(),
1014 language: "en".into(),
1015 });
1016 match self.questions().put_executable(&mut q) {
1017 Ok(()) => st.question = Some(q.id.clone()),
1018 Err(e) => tracing::warn!("could not file the question for {pr}: {e:#}"),
1019 }
1020 }
1021 LocalStep::Merge(head) => {
1022 match self.forge.merge(repo, &snap.url, &head).await {
1023 Ok(true) => {
1024 self.save(pr, &st);
1026 if let Some(cfg) = cfg {
1027 self.release_merged(repo, pr, st, cfg).await;
1028 }
1029 return;
1030 }
1031 res => {
1032 match res {
1033 Err(e) => {
1034 tracing::warn!("could not merge {}: {e:#}", snap.url)
1035 }
1036 _ => tracing::warn!("{} did not merge", snap.url),
1037 }
1038 st.held_head = Some(head);
1039 self.raise(
1040 pr,
1041 &snap.url,
1042 format!(
1043 "Release PR {pr} could not be merged; merge it by hand"
1044 ),
1045 );
1046 }
1047 }
1048 }
1049 }
1050 self.save(pr, &st);
1051 }
1052 }
1053 }
1054
1055 async fn release_merged(
1057 &self,
1058 repo: &Path,
1059 pr: &str,
1060 mut st: WatchState,
1061 cfg: &crate::config::Config,
1062 ) {
1063 if st.ignored {
1064 return self.finish(pr, &st);
1065 }
1066 if st.job.is_none() {
1067 let info = match self.forge.info(repo, &st.url).await {
1068 Ok(i) => i,
1069 Err(e) => {
1070 tracing::warn!("could not read the merged {}: {e:#}", st.url);
1071 return;
1072 }
1073 };
1074 let (Some(version), Some(commit)) = (
1075 release_local::version_from_branch(&info.branch),
1076 info.merge_commit,
1077 ) else {
1078 if release_local::version_from_branch(&info.branch).is_none() {
1081 self.finish(pr, &st);
1082 }
1083 return;
1084 };
1085 let mut job = Job::new(&version, &st.url, &commit);
1086 job.accepted = release_local::version_from_title(&info.title)
1089 .into_iter()
1090 .filter(|t| *t != version)
1091 .collect();
1092 st.job = Some(job);
1093 if !self.save(pr, &st) {
1094 return;
1095 }
1096 }
1097 let Some(mut job) = st.job.take() else {
1098 return;
1099 };
1100 if job.finished {
1101 st.job = Some(job);
1104 if !self.record_on_run(&st) {
1105 self.save(pr, &st);
1106 return;
1107 }
1108 return self.complete(pr, &mut st);
1109 }
1110 if job.interrupted() {
1111 job.failed = Some(format!(
1112 "magi stopped while `{}` was running; it may have partly run, so it is not repeated on its own",
1113 job.running.clone().unwrap_or_default()
1114 ));
1115 }
1116 if let Some(why) = job.failed.clone() {
1117 self.hold_task(&mut st, &format!("release {} is on hold: {why}", job.tag()));
1120 let retry = self.settle_job_question(pr, &mut st, &mut job);
1122 if !retry {
1123 st.job = Some(job);
1124 self.save(pr, &st);
1125 return;
1126 }
1127 }
1128 let shell = cfg.shell();
1129 let env = release_local::Env {
1130 repo,
1131 home: &self.home,
1132 key: pr,
1133 remote: &cfg.merge.remote,
1134 shell: &shell,
1135 release: &cfg.release,
1136 };
1137 let result = {
1138 let mut save = |j: &Job| {
1139 let mut copy = st.clone();
1140 copy.job = Some(j.clone());
1141 self.save(pr, ©)
1142 };
1143 release_local::run_job(&env, &mut job, &mut save).await
1144 };
1145 match result {
1146 Ok(()) => {
1147 self.raise_released(pr, &st.url, &job);
1148 st.job = Some(job);
1149 if self.record_on_run(&st) {
1150 self.complete(pr, &mut st);
1151 } else {
1152 self.save(pr, &st);
1155 }
1156 }
1157 Err(e) => {
1158 job.failed = Some(format!("{e:#}"));
1159 self.hold_task(&mut st, &format!("release {} failed: {e:#}", job.tag()));
1160 self.raise(pr, &st.url, failed_message(pr));
1161 self.file_job_question(pr, &mut st, &job);
1162 st.job = Some(job);
1163 self.save(pr, &st);
1164 self.record_on_run(&st);
1165 }
1166 }
1167 }
1168
1169 fn record_on_run(&self, st: &WatchState) -> bool {
1174 let (Some(run), Some(job)) = (st.run.as_deref(), st.job.as_ref()) else {
1175 return true;
1176 };
1177 let mut state = match RunState::load_under(run, &self.home) {
1178 Ok(s) => s,
1179 Err(e) => {
1180 tracing::warn!("could not record the release on run {run}: {e:#}");
1181 return true;
1182 }
1183 };
1184 let bump = state.release_bump.get_or_insert_with(Default::default);
1185 if bump.release.as_ref() == Some(job) {
1186 return true;
1187 }
1188 bump.release = Some(job.clone());
1189 state.event(
1190 "release",
1191 format!(
1192 "release {} {}",
1193 job.tag(),
1194 if job.finished {
1195 "finished"
1196 } else if let Some(why) = &job.failed {
1197 why
1198 } else {
1199 "stopped"
1200 }
1201 ),
1202 );
1203 match state.save_under(&self.home) {
1204 Ok(()) => true,
1205 Err(e) => {
1206 tracing::warn!("could not save the release on run {run}: {e:#}");
1207 false
1208 }
1209 }
1210 }
1211
1212 fn hold_task(&self, st: &mut WatchState, reason: &str) {
1216 if st.held_task.is_some() {
1217 return;
1218 }
1219 let Some(run) = st.run.clone() else {
1220 return;
1221 };
1222 let q = crate::queue::Queue::at(self.home.join("queue"));
1223 for mut t in q.list() {
1224 if t.runs.contains(&run) {
1225 if t.status != crate::queue::TaskStatus::Held {
1226 t.notice_cause("release-failed", Some(&run), None);
1227 t.hold_machine(Some(format!("{HOLD_PREFIX}{reason}")));
1228 match q.put(&mut t) {
1229 Ok(()) => st.held_task = Some(t.id.clone()),
1230 Err(e) => tracing::warn!("could not hold task {} for {run}: {e:#}", t.id),
1231 }
1232 }
1233 return;
1234 }
1235 }
1236 }
1237
1238 #[must_use]
1245 fn unhold_task(&self, st: &mut WatchState) -> bool {
1246 let Some(id) = st.held_task.take() else {
1247 return true;
1248 };
1249 let q = crate::queue::Queue::at(self.home.join("queue"));
1250 let mut t = match q.get(&id) {
1251 Ok(t) => t,
1252 Err(_) if matches!(q.path_of(&id).try_exists(), Ok(false)) => return true,
1255 Err(e) => {
1256 tracing::warn!("could not read task {id} to restore it after the release: {e:#}");
1257 st.held_task = Some(id);
1258 return false;
1259 }
1260 };
1261 let ours = t.status == crate::queue::TaskStatus::Held
1262 && t.hold_source == Some(crate::queue::HoldSource::Machine)
1263 && t.hold_reason
1264 .as_deref()
1265 .is_some_and(|r| r.starts_with(HOLD_PREFIX));
1266 if ours {
1267 t.succeed();
1268 if let Err(e) = q.put(&mut t) {
1269 tracing::warn!("could not restore task {id} after the release: {e:#}");
1270 st.held_task = Some(id);
1271 return false;
1272 }
1273 }
1274 true
1275 }
1276
1277 fn complete(&self, pr: &str, st: &mut WatchState) {
1281 if self.unhold_task(st) {
1282 self.finish(pr, st);
1283 } else {
1284 self.save(pr, st);
1285 }
1286 }
1287
1288 fn raise_released(&self, pr: &str, url: &str, job: &Job) {
1289 notices::raise_in(
1290 &self.home,
1291 Notice::info(&released_key(pr), format!("Released {} ({pr})", job.tag())).link(
1292 Link::Url {
1293 url: url.to_owned(),
1294 },
1295 ),
1296 );
1297 }
1298
1299 fn file_job_question(&self, pr: &str, st: &mut WatchState, job: &Job) {
1301 let last = job
1302 .log
1303 .last()
1304 .map(|l| format!("\nLast step: {}\n\n{}\n", l.name, l.tail))
1305 .unwrap_or_default();
1306 let mut q = Question::new(
1307 String::new(),
1308 crate::bump::NOTICE_NODE.to_owned(),
1309 "release-watch".to_owned(),
1310 format!("Release {} of {pr} is on hold: retry?", job.tag()),
1311 format!(
1312 "Pull request: {}\nWhy it stopped: {}\n{last}\n\
1313 Full output is under the magi home's release-local directory.\n\n\
1314 - `{RETRY}`: run it again from where it stopped (the tag is not \
1315 recreated and finished commands are skipped).\n\
1316 - `{LEAVE_IT}`: stop watching; release by hand.\n\n\
1317 Nothing is retried on its own; silence is a hold.\n",
1318 job.pr_url,
1319 job.failed.clone().unwrap_or_default()
1320 ),
1321 vec![RETRY.to_owned(), LEAVE_IT.to_owned()],
1322 );
1323 match self.questions().put_executable(&mut q) {
1324 Ok(()) => st.question = Some(q.id.clone()),
1325 Err(e) => tracing::warn!("could not file the question for {pr}: {e:#}"),
1326 }
1327 }
1328
1329 fn settle_job_question(&self, pr: &str, st: &mut WatchState, job: &mut Job) -> bool {
1333 let why = job.failed.clone().unwrap_or_default();
1334 if let Some(id) = st.question.clone() {
1335 match self.questions().get(&id) {
1336 Ok(q) if q.status.open() => return false,
1337 Ok(q) => {
1338 st.question = None;
1339 match &q.answer {
1340 Some(Answer::Choice(c)) if c == RETRY => {
1341 job.resume();
1342 st.held = None;
1343 return true;
1344 }
1345 Some(Answer::Choice(c)) if c == LEAVE_IT => {
1346 st.ignored = true;
1347 let _ = Notices::at(self.home.join("notifications"))
1348 .dismiss(¬ices::id_of(¬ice_key(pr)));
1349 }
1350 _ => st.held = Some(why),
1352 }
1353 return false;
1354 }
1355 Err(_) => st.question = None,
1356 }
1357 }
1358 if !st.ignored && st.held.as_deref() != Some(why.as_str()) {
1359 self.raise(pr, &st.url, failed_message(pr));
1360 self.file_job_question(pr, st, job);
1361 }
1362 false
1363 }
1364
1365 fn apply_answer(
1367 &self,
1368 pr: &str,
1369 st: &mut WatchState,
1370 snap: &RollupView,
1371 q: &Question,
1372 now: i64,
1373 ) {
1374 st.question = None;
1375 if st.applied.contains(&q.id) {
1376 return;
1377 }
1378 st.applied.push(q.id.clone());
1379 if st.applied.len() > 20 {
1380 st.applied.remove(0);
1381 }
1382 match &q.answer {
1383 Some(Answer::Choice(c)) if c == RERUN_AGAIN => {
1384 st.held = None;
1385 if let Some(crate::ask::ChoiceAction::RerunChecks { runs, .. }) = q.chosen_action()
1386 {
1387 for run in runs {
1390 st.reruns.insert(run.clone(), now);
1391 }
1392 return;
1393 }
1394 for c in snap.checks.iter().filter(|c| is_failed(c.verdict)) {
1395 if let Some(r) = &c.run {
1396 st.reruns.remove(r);
1397 }
1398 }
1399 }
1400 Some(Answer::Choice(c)) if c == LEAVE_IT => {
1401 st.ignored = true;
1402 if let Err(e) = Notices::at(self.home.join("notifications"))
1403 .dismiss(¬ices::id_of(¬ice_key(pr)))
1404 {
1405 tracing::warn!("could not dismiss the notice for {pr}: {e:#}");
1406 }
1407 }
1408 _ => st.held = Some(fingerprint(snap)),
1410 }
1411 }
1412
1413 fn finish(&self, pr: &str, st: &WatchState) {
1415 let _ =
1416 Notices::at(self.home.join("notifications")).dismiss(¬ices::id_of(¬ice_key(pr)));
1417 if let Some(id) = &st.question {
1418 let _ = self.questions().update(id, |q| {
1419 q.abandon("the release pull request is no longer open");
1420 Ok(())
1421 });
1422 }
1423 let _ = std::fs::remove_file(self.state_path(pr));
1424 }
1425}
1426
1427pub(crate) async fn run(
1431 watcher: Watcher,
1432 settings: impl Fn() -> (Vec<PathBuf>, u64),
1433 stop: crate::daemon::Stop,
1434) {
1435 let halt = {
1436 let stop = stop.clone();
1437 move || stop.stopped()
1438 };
1439 while !stop.stopped() {
1440 let (repos, minutes) = settings();
1441 if minutes > 0 {
1442 let now = jiff::Timestamp::now().as_second();
1443 watcher
1444 .lap(&repos, (minutes as i64).saturating_mul(60), now, &halt)
1445 .await;
1446 }
1447 let mut slept = Duration::ZERO;
1448 while slept < LAP && !stop.stopped() {
1449 tokio::time::sleep(Duration::from_secs(1)).await;
1450 slept += Duration::from_secs(1);
1451 }
1452 }
1453}
1454
1455#[cfg(test)]
1456mod tests {
1457 use super::*;
1458 use crate::land::CheckView;
1459 use anyhow::Context;
1460 use std::sync::Mutex;
1461
1462 const URL: &str = "https://github.com/o/r/pull/7";
1463
1464 fn check(name: &str, v: Verdict, run: Option<&str>) -> CheckView {
1465 CheckView {
1466 name: name.to_owned(),
1467 verdict: v,
1468 run: run.map(str::to_owned),
1469 url: run.map(|r| format!("https://github.com/o/r/actions/runs/{r}/job/1")),
1470 }
1471 }
1472
1473 fn snap(state: PrLifecycle, head: &str, checks: Vec<CheckView>) -> RollupView {
1474 RollupView {
1475 url: URL.to_owned(),
1476 number: 7,
1477 state,
1478 head: head.to_owned(),
1479 checks,
1480 }
1481 }
1482
1483 fn open(checks: Vec<CheckView>) -> RollupView {
1484 snap(PrLifecycle::Open, "h1", checks)
1485 }
1486
1487 fn st_at(progress: i64) -> WatchState {
1488 WatchState {
1489 progress_at: progress,
1490 ..WatchState::default()
1491 }
1492 }
1493
1494 #[test]
1495 fn release_notice_uses_saved_question_and_checks_its_current_status() {
1496 let dir = tempfile::tempdir().unwrap();
1497 let watcher = Watcher::new(Box::new(GhForge), dir.path().to_path_buf());
1498 let questions = Questions::at(dir.path().join("questions"));
1499 let mut q = Question::new(
1500 String::new(),
1501 crate::bump::NOTICE_NODE.into(),
1502 "release-watch".into(),
1503 "Release?".into(),
1504 String::new(),
1505 Vec::new(),
1506 );
1507 questions.put(&mut q).unwrap();
1508 let st = WatchState {
1509 url: "https://example.invalid/o/r/pull/7".into(),
1510 question: Some(q.id.clone()),
1511 ..WatchState::default()
1512 };
1513 assert!(watcher.save("o/r#7", &st));
1514 let n = Notice::warn("release-pr:o/r#7", "Release PR is waiting for approval.");
1515 assert_eq!(
1516 notices::answerable_question(&n, &questions),
1517 Some(q.id.clone())
1518 );
1519 questions
1520 .update(&q.id, |q| {
1521 q.abandon("obsolete approval");
1522 Ok(())
1523 })
1524 .unwrap();
1525 assert_eq!(notices::answerable_question(&n, &questions), None);
1526 assert_eq!(question_for_notice(dir.path(), "release-pr:o/r#8"), None);
1527 }
1528
1529 #[test]
1530 fn unreadable_is_unknown_and_merged_or_closed_is_done() {
1531 assert_eq!(decide(None, &st_at(0), 10, 60), Step::Unknown);
1532 for s in [PrLifecycle::Merged, PrLifecycle::Closed] {
1533 assert_eq!(
1534 decide(Some(&snap(s, "h", vec![])), &st_at(0), 10, 60),
1535 Step::Done
1536 );
1537 }
1538 }
1539
1540 #[test]
1541 fn a_failed_run_is_rerun_even_while_another_run_is_pending() {
1542 let s = open(vec![
1543 check("win", Verdict::Fail, Some("11")),
1544 check("lint", Verdict::Pending, Some("12")),
1545 ]);
1546 assert_eq!(
1547 decide(Some(&s), &st_at(0), 1, 3600),
1548 Step::Rerun(vec!["11".into()])
1549 );
1550 }
1551
1552 #[test]
1553 fn a_run_with_a_job_still_pending_is_not_complete() {
1554 let s = open(vec![
1555 check("win", Verdict::Fail, Some("11")),
1556 check("mac", Verdict::Pending, Some("11")),
1557 ]);
1558 assert_eq!(decide(Some(&s), &st_at(0), 1, 3600), Step::Wait);
1559 }
1560
1561 #[test]
1562 fn one_rerun_per_run_then_grace_then_escalation() {
1563 let s = open(vec![check("win", Verdict::Fail, Some("11"))]);
1564 let mut st = st_at(0);
1565 st.reruns.insert("11".into(), 100);
1566 assert_eq!(decide(Some(&s), &st, 100 + RERUN_GRACE - 1, 0), Step::Wait);
1567 assert_eq!(
1568 decide(Some(&s), &st, 100 + RERUN_GRACE, 0),
1569 Step::Escalate(Why::StillRed)
1570 );
1571 }
1572
1573 #[test]
1574 fn a_failure_with_no_run_id_goes_straight_to_a_human_once_settled() {
1575 let s = open(vec![check("ci/ext", Verdict::Fail, None)]);
1576 assert_eq!(
1577 decide(Some(&s), &st_at(0), 1, 0),
1578 Step::Escalate(Why::StillRed)
1579 );
1580 let s = open(vec![
1581 check("ci/ext", Verdict::Fail, None),
1582 check("x", Verdict::Pending, Some("5")),
1583 ]);
1584 assert_eq!(decide(Some(&s), &st_at(0), 1, 0), Step::Wait);
1585 }
1586
1587 #[test]
1588 fn no_progress_for_the_bounded_time_escalates_and_zero_disables_it() {
1589 let s = open(vec![check("a", Verdict::Pass, Some("1"))]);
1590 assert_eq!(decide(Some(&s), &st_at(0), 3599, 3600), Step::Wait);
1591 assert_eq!(
1592 decide(Some(&s), &st_at(0), 3600, 3600),
1593 Step::Escalate(Why::Stalled)
1594 );
1595 assert_eq!(decide(Some(&s), &st_at(0), 99_999, 0), Step::Wait);
1596 }
1597
1598 #[test]
1599 fn held_and_ignored_stay_quiet_until_the_fingerprint_moves() {
1600 let s = open(vec![check("a", Verdict::Fail, None)]);
1601 let mut st = st_at(0);
1602 st.held = Some(fingerprint(&s));
1603 assert_eq!(decide(Some(&s), &st, 99_999, 60), Step::Wait);
1604 let moved = open(vec![
1605 check("a", Verdict::Pass, None),
1606 check("b", Verdict::Fail, None),
1607 ]);
1608 assert_eq!(
1609 decide(Some(&moved), &st, 99_999, 60),
1610 Step::Escalate(Why::StillRed)
1611 );
1612 st.ignored = true;
1613 assert_eq!(decide(Some(&moved), &st, 99_999, 60), Step::Wait);
1614 }
1615
1616 #[test]
1617 fn observe_restarts_the_clock_only_on_change_and_forgets_reruns_on_a_new_head() {
1618 let mut st = st_at(0);
1619 let a = open(vec![check("a", Verdict::Pending, Some("1"))]);
1620 observe(&mut st, &a, 10);
1621 assert_eq!(st.progress_at, 10);
1622 observe(&mut st, &a, 50);
1623 assert_eq!(st.progress_at, 10);
1624 st.reruns.insert("1".into(), 5);
1625 observe(
1626 &mut st,
1627 &snap(PrLifecycle::Open, "h2", a.checks.clone()),
1628 60,
1629 );
1630 assert!(st.reruns.is_empty());
1631 assert_eq!(st.progress_at, 60);
1632 }
1633
1634 #[test]
1635 fn pr_key_names_owner_repo_and_number() {
1636 assert_eq!(pr_key(URL).as_deref(), Some("o/r#7"));
1637 assert_eq!(pr_key("https://github.com/o/r/issues/7"), None);
1638 assert_eq!(pr_key("nonsense"), None);
1639 }
1640
1641 #[test]
1642 fn notice_wording_does_not_vary_between_polls() {
1643 assert_eq!(rerun_message("o/r#7"), rerun_message("o/r#7"));
1644 assert!(
1645 !human_message("o/r#7")
1646 .chars()
1647 .any(|c| c.is_ascii_digit() && c != '7')
1648 );
1649 }
1650
1651 #[test]
1652 fn the_release_question_is_not_claimed_by_other_machinery() {
1653 let q = Question::new(
1654 String::new(),
1655 crate::bump::NOTICE_NODE.into(),
1656 "release-watch".into(),
1657 "s".into(),
1658 String::new(),
1659 vec![HOLD.into()],
1660 );
1661 assert_eq!(
1663 crate::deputy::kind_of(&q),
1664 Some(crate::deputy::Kind::Release)
1665 );
1666 let n = Notice::warn("release-pr:o/r#7", "m").about([String::new()]);
1667 assert!(!notices::covers(&q, &n));
1668 let mut plain = q.clone();
1670 plain.seat = "bump".into();
1671 assert_eq!(crate::deputy::kind_of(&plain), None);
1672 }
1673
1674 #[derive(Default)]
1675 struct Fake {
1676 snap: Mutex<Option<RollupView>>,
1677 reruns: Mutex<Vec<String>>,
1678 local: Mutex<bool>,
1679 info: Mutex<Option<PrInfo>>,
1680 merges: Mutex<Vec<String>>,
1681 approval: Mutex<Option<bool>>,
1683 merge_fails: Mutex<bool>,
1684 config_fails: Mutex<bool>,
1686 }
1687
1688 impl ReleaseForge for std::sync::Arc<Fake> {
1689 fn list<'a>(&'a self, _: &'a Path) -> Fut<'a, Result<Vec<(String, String)>>> {
1690 Box::pin(async { Ok(vec![("chore/release-v1.0.0".to_owned(), URL.to_owned())]) })
1691 }
1692 fn snapshot<'a>(&'a self, _: &'a Path, _: &'a str) -> Fut<'a, Result<RollupView>> {
1693 let s = self.snap.lock().unwrap().clone();
1694 Box::pin(async move { s.context("unreadable") })
1695 }
1696 fn rerun<'a>(&'a self, _: &'a Path, run: &'a str) -> Fut<'a, Result<()>> {
1697 self.reruns.lock().unwrap().push(run.to_owned());
1698 Box::pin(async { Ok(()) })
1699 }
1700 fn config<'a>(&'a self, _: &'a Path) -> Fut<'a, Result<crate::config::Config>> {
1701 if *self.config_fails.lock().unwrap() {
1702 return Box::pin(async { bail!("unreadable config") });
1703 }
1704 let mut c = crate::config::Config::default();
1705 if *self.local.lock().unwrap() {
1706 c.release.mode = crate::config::ReleaseMode::Local;
1707 c.release.commands = vec!["true".to_owned()];
1708 }
1709 Box::pin(async move { Ok(c) })
1710 }
1711 fn info<'a>(&'a self, _: &'a Path, _: &'a str) -> Fut<'a, Result<PrInfo>> {
1712 let i = self.info.lock().unwrap().clone();
1713 Box::pin(async move { i.context("no info") })
1714 }
1715 fn merge<'a>(&'a self, _: &'a Path, _: &'a str, head: &'a str) -> Fut<'a, Result<bool>> {
1716 self.merges.lock().unwrap().push(head.to_owned());
1717 let fails = *self.merge_fails.lock().unwrap();
1718 Box::pin(async move { if fails { bail!("refused") } else { Ok(true) } })
1719 }
1720 fn land_approval<'a>(&'a self, _: &'a Path) -> Fut<'a, Result<bool>> {
1721 let a = *self.approval.lock().unwrap();
1722 Box::pin(async move { a.context("unreadable config") })
1723 }
1724 }
1725
1726 fn rig() -> (tempfile::TempDir, std::sync::Arc<Fake>, Watcher) {
1727 let dir = tempfile::tempdir().unwrap();
1728 let fake = std::sync::Arc::new(Fake::default());
1729 *fake.approval.lock().unwrap() = Some(true);
1731 let w = Watcher::new(Box::new(fake.clone()), dir.path().to_path_buf());
1732 (dir, fake, w)
1733 }
1734
1735 fn red() -> RollupView {
1736 open(vec![check("win", Verdict::Fail, Some("11"))])
1737 }
1738
1739 #[tokio::test]
1740 async fn reruns_once_survives_a_restart_then_escalates_with_one_question() {
1741 let (dir, fake, w) = rig();
1742 *fake.snap.lock().unwrap() = Some(red());
1743 let repo = PathBuf::from("/nowhere");
1744 let no = || false;
1745 w.lap(std::slice::from_ref(&repo), 3600, 1000, &no).await;
1746 assert_eq!(*fake.reruns.lock().unwrap(), vec!["11".to_owned()]);
1747
1748 let w2 = Watcher::new(Box::new(fake.clone()), dir.path().to_path_buf());
1750 w2.lap(
1751 std::slice::from_ref(&repo),
1752 3600,
1753 1000 + RERUN_GRACE - 1,
1754 &no,
1755 )
1756 .await;
1757 assert_eq!(fake.reruns.lock().unwrap().len(), 1);
1758 assert!(w2.questions().list().is_empty());
1759
1760 w2.lap(
1761 std::slice::from_ref(&repo),
1762 3600,
1763 1000 + RERUN_GRACE + 1,
1764 &no,
1765 )
1766 .await;
1767 w2.lap(
1768 std::slice::from_ref(&repo),
1769 3600,
1770 1000 + RERUN_GRACE + 400,
1771 &no,
1772 )
1773 .await;
1774 assert_eq!(fake.reruns.lock().unwrap().len(), 1);
1775 let qs = w2.questions().list();
1776 assert_eq!(qs.len(), 1, "asked once, not every lap");
1777 assert_eq!(qs[0].node, crate::bump::NOTICE_NODE);
1778 assert!(qs[0].detail.contains("win") && qs[0].detail.contains(URL));
1779 assert_eq!(qs[0].choices, vec![RERUN_AGAIN, HOLD, LEAVE_IT]);
1780 let ns = Notices::at(dir.path().join("notifications")).list();
1781 assert_eq!(ns.len(), 1);
1782 assert_eq!(ns[0].key, "release-pr:o/r#7");
1783 assert_eq!(
1784 ns[0].link,
1785 Some(Link::Url {
1786 url: URL.to_owned(),
1787 })
1788 );
1789 }
1790
1791 async fn escalated() -> (tempfile::TempDir, std::sync::Arc<Fake>, Watcher, String) {
1792 let (dir, fake, w) = rig();
1793 *fake.snap.lock().unwrap() = Some(red());
1794 let repo = PathBuf::from("/nowhere");
1795 let no = || false;
1796 w.lap(std::slice::from_ref(&repo), 3600, 1000, &no).await;
1797 w.lap(&[repo], 3600, 2000, &no).await;
1798 let id = w.questions().list()[0].id.clone();
1799 (dir, fake, w, id)
1800 }
1801
1802 fn answer(w: &Watcher, id: &str, choice: &str) {
1803 w.questions()
1804 .update(id, |q| q.answer(Answer::Choice(choice.to_owned())))
1805 .unwrap();
1806 }
1807
1808 #[tokio::test]
1809 async fn legacy_rerun_again_reruns_exactly_once_more() {
1810 let (_d, fake, w, id) = escalated().await;
1811 w.questions()
1812 .update(&id, |q| {
1813 q.actions.clear();
1814 Ok(())
1815 })
1816 .unwrap();
1817 answer(&w, &id, RERUN_AGAIN);
1818 let repo = PathBuf::from("/nowhere");
1819 let no = || false;
1820 w.lap(std::slice::from_ref(&repo), 3600, 3000, &no).await;
1821 w.lap(&[repo], 3600, 3001, &no).await;
1822 assert_eq!(fake.reruns.lock().unwrap().len(), 2);
1823 }
1824
1825 #[tokio::test]
1826 async fn typed_rerun_is_left_to_the_daemon_instead_of_the_watcher() {
1827 let (_d, fake, w, id) = escalated().await;
1828 let q = w.questions().get(&id).unwrap();
1829 assert!(
1830 matches!(q.actions.get(RERUN_AGAIN), Some(crate::ask::ChoiceAction::RerunChecks { head, runs, .. })
1831 if head == "h1" && runs == &["11"])
1832 );
1833 answer(&w, &id, RERUN_AGAIN);
1834 w.lap(&[PathBuf::from("/nowhere")], 3600, 3000, &|| false)
1835 .await;
1836 assert_eq!(
1837 fake.reruns.lock().unwrap().len(),
1838 1,
1839 "only the earlier automatic rerun"
1840 );
1841 }
1842
1843 #[test]
1844 fn no_failed_workflow_means_no_rerun_choice() {
1845 let snap = open(vec![check("external", Verdict::Fail, None)]);
1846 assert_eq!(maintenance_choices(&snap), vec![HOLD, LEAVE_IT]);
1847 }
1848
1849 #[tokio::test]
1850 async fn hold_and_silence_do_not_reask_and_leave_it_dismisses() {
1851 let (d, fake, w, id) = escalated().await;
1852 answer(&w, &id, HOLD);
1853 let repo = PathBuf::from("/nowhere");
1854 let no = || false;
1855 for t in [3000, 90_000, 200_000] {
1856 w.lap(std::slice::from_ref(&repo), 3600, t, &no).await;
1857 }
1858 assert_eq!(w.questions().list().len(), 1, "held: no second question");
1859 assert_eq!(fake.reruns.lock().unwrap().len(), 1);
1860
1861 let (d2, _f2, w2, id2) = escalated().await;
1862 answer(&w2, &id2, LEAVE_IT);
1863 w2.lap(std::slice::from_ref(&repo), 3600, 3000, &no).await;
1864 let ns = Notices::at(d2.path().join("notifications")).list();
1865 assert!(ns.is_empty(), "dismissed notices are hidden");
1866 drop(d);
1867 }
1868
1869 fn with_deputy(w: &Watcher, id: &str, home: &Path) -> String {
1871 let seat = crate::agent::SeatState::new(&crate::ask::deputy_seat_key(id), "a", 1);
1872 let key = seat.key.clone();
1873 let brief = deputy_brief(&w.questions().get(id).unwrap(), home);
1874 w.questions()
1875 .update(id, |q| {
1876 let mut d = crate::ask::Deputy::new(brief);
1877 d.seat = Some(seat);
1878 q.deputy = Some(d);
1879 Ok(())
1880 })
1881 .unwrap();
1882 key
1883 }
1884
1885 #[tokio::test]
1886 async fn the_brief_names_the_pull_request_and_what_leave_it_really_does() {
1887 let (d, _f, w, id) = escalated().await;
1888 let q = w.questions().get(&id).unwrap();
1889 let b = deputy_brief(&q, d.path());
1890 assert!(b.contains(URL), "{b}");
1891 assert!(
1892 b.contains("`leave it`") && b.contains("does NOT close"),
1893 "{b}"
1894 );
1895 assert!(b.contains("`rerun again`") && b.contains("`hold`"), "{b}");
1896 assert!(!b.contains("`merge`"), "no merge on an escalation: {b}");
1897
1898 let empty = tempfile::tempdir().unwrap();
1900 let b = deputy_brief(&q, empty.path());
1901 assert!(b.contains("could not be found") && !b.contains(URL), "{b}");
1902 }
1903
1904 #[tokio::test]
1905 async fn a_settled_leave_it_is_applied_once_by_the_watcher_and_closes_nothing() {
1906 let (d, fake, w, id) = escalated().await;
1907 let seat = with_deputy(&w, &id, d.path());
1908 let say = "クローズしていいよ。private repo だから、何回やっても失敗しちゃうから";
1909 w.questions().update(&id, |q| q.say(say)).unwrap();
1910 w.questions()
1911 .update(&id, |q| {
1912 q.settle_by_deputy(&seat, LEAVE_IT, "クローズしていいよ", None)
1913 })
1914 .unwrap();
1915 let q = w.questions().get(&id).unwrap();
1916 assert_eq!(q.answer, Some(Answer::Choice(LEAVE_IT.to_owned())));
1917
1918 let repo = PathBuf::from("/nowhere");
1919 let no = || false;
1920 for t in [3000, 3100] {
1921 w.lap(std::slice::from_ref(&repo), 3600, t, &no).await;
1922 }
1923 let st = w.stored().pop().unwrap();
1924 assert!(st.ignored, "watching stopped");
1925 assert_eq!(st.applied, vec![id.clone()], "applied exactly once");
1926 assert!(
1927 Notices::at(d.path().join("notifications"))
1928 .list()
1929 .is_empty()
1930 );
1931 assert_eq!(fake.reruns.lock().unwrap().len(), 1, "no extra rerun");
1932 assert_eq!(w.questions().list().len(), 1, "no second question");
1933 }
1934
1935 #[tokio::test]
1936 async fn the_task_of_a_release_question_comes_from_the_watch_record() {
1937 let (d, _f, w, id) = escalated().await;
1938 let q = w.questions().get(&id).unwrap();
1939 let queue = crate::queue::Queue::at(d.path().join("queue"));
1940 assert!(
1941 task_of_question(d.path(), &q, &queue).is_none(),
1942 "no run known"
1943 );
1944 assert!(task_of_question(d.path(), &q, &queue).is_none());
1945 let mut st = w.stored().pop().unwrap();
1946 st.run = Some("run-1".into());
1947 w.save("o/r#7", &st);
1948 assert!(task_of_question(d.path(), &q, &queue).is_none());
1950 let mut t = crate::queue::Task::new(
1951 "t".into(),
1952 "i".into(),
1953 PathBuf::from("/nowhere"),
1954 crate::queue::Source::Human,
1955 );
1956 t.runs.push("run-1".into());
1957 queue.put(&mut t).unwrap();
1958 assert_eq!(task_of_question(d.path(), &q, &queue).unwrap().id, t.id);
1959 }
1960
1961 #[tokio::test]
1962 async fn merged_clears_the_notice_the_question_and_the_record() {
1963 let (d, fake, w, _id) = escalated().await;
1964 *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Merged, "h1", vec![]));
1965 let repo = PathBuf::from("/nowhere");
1966 w.lap(&[repo], 3600, 5000, &(|| false)).await;
1967 assert!(
1968 Notices::at(d.path().join("notifications"))
1969 .list()
1970 .is_empty()
1971 );
1972 assert!(w.questions().list().iter().all(|q| !q.status.open()));
1973 assert!(w.stored().is_empty());
1974 }
1975
1976 #[tokio::test]
1977 async fn an_unreadable_forge_changes_nothing() {
1978 let (d, fake, w) = rig();
1979 *fake.snap.lock().unwrap() = None;
1980 w.lap(&[PathBuf::from("/nowhere")], 60, 99_999, &(|| false))
1981 .await;
1982 assert!(fake.reruns.lock().unwrap().is_empty());
1983 assert!(w.questions().list().is_empty());
1984 assert!(!d.path().join("release-watch").exists());
1985 }
1986
1987 #[tokio::test]
1988 async fn no_rerun_when_the_record_cannot_be_saved() {
1989 let (d, fake, w) = rig();
1990 *fake.snap.lock().unwrap() = Some(red());
1991 std::fs::write(d.path().join("release-watch"), "x").unwrap();
1993 w.lap(&[PathBuf::from("/nowhere")], 3600, 1000, &(|| false))
1994 .await;
1995 assert!(fake.reruns.lock().unwrap().is_empty());
1996 }
1997
1998 #[tokio::test]
1999 async fn a_park_stops_the_lap_before_any_call() {
2000 let (_d, fake, w) = rig();
2001 *fake.snap.lock().unwrap() = Some(red());
2002 w.lap(&[PathBuf::from("/nowhere")], 60, 1000, &(|| true))
2003 .await;
2004 assert!(fake.reruns.lock().unwrap().is_empty());
2005 }
2006
2007 fn local_st(asked: Option<&str>, held: Option<&str>) -> WatchState {
2010 WatchState {
2011 asked_head: asked.map(str::to_owned),
2012 held_head: held.map(str::to_owned),
2013 ..WatchState::default()
2014 }
2015 }
2016
2017 #[test]
2018 fn local_decisions_never_wait_on_ci_and_bind_approval_to_the_head() {
2019 use Approved::*;
2020 let none = local_st(None, None);
2021 assert_eq!(decide_local("h1", &none, NoQuestion, true), LocalStep::Ask);
2022 assert_eq!(decide_local("h1", &none, Open, true), LocalStep::Wait);
2023 let asked = local_st(Some("h1"), None);
2024 assert_eq!(
2025 decide_local("h1", &asked, Merge, true),
2026 LocalStep::Merge("h1".to_owned())
2027 );
2028 assert_eq!(
2029 decide_local("h1", &asked, Hold, true),
2030 LocalStep::Hold("h1".to_owned())
2031 );
2032 assert_eq!(decide_local("h2", &asked, Merge, true), LocalStep::Ask);
2034 assert_eq!(decide_local("h2", &asked, Hold, true), LocalStep::Ask);
2035 let held = local_st(Some("h1"), Some("h1"));
2037 assert_eq!(decide_local("h1", &held, NoQuestion, true), LocalStep::Wait);
2038 assert_eq!(decide_local("h2", &held, NoQuestion, true), LocalStep::Ask);
2039 let mut ign = local_st(None, None);
2040 ign.ignored = true;
2041 assert_eq!(decide_local("h1", &ign, NoQuestion, true), LocalStep::Wait);
2042 }
2043
2044 #[test]
2045 fn without_land_approval_an_unasked_head_merges_but_stops_stay_stopped() {
2046 use Approved::*;
2047 let none = local_st(None, None);
2048 assert_eq!(
2049 decide_local("h1", &none, NoQuestion, false),
2050 LocalStep::Merge("h1".to_owned())
2051 );
2052 assert_eq!(decide_local("h1", &none, Open, false), LocalStep::Wait);
2054 let held = local_st(Some("h1"), Some("h1"));
2055 assert_eq!(
2056 decide_local("h1", &held, NoQuestion, false),
2057 LocalStep::Wait
2058 );
2059 assert_eq!(
2060 decide_local("h2", &held, NoQuestion, false),
2061 LocalStep::Merge("h2".to_owned())
2062 );
2063 let mut ign = local_st(None, None);
2064 ign.ignored = true;
2065 assert_eq!(decide_local("h1", &ign, NoQuestion, false), LocalStep::Wait);
2066 let asked = local_st(Some("h1"), None);
2068 assert_eq!(
2069 decide_local("h1", &asked, Hold, false),
2070 LocalStep::Hold("h1".to_owned())
2071 );
2072 }
2073
2074 #[tokio::test]
2075 async fn a_registered_local_pr_still_asks_but_never_merges_while_the_config_is_unreadable() {
2076 let (dir, fake, w) = rig();
2077 *fake.config_fails.lock().unwrap() = true;
2078 *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Open, "h1", vec![]));
2079 let repo = PathBuf::from("/nowhere");
2080 let no = || false;
2081 let pr = pr_key(URL).unwrap();
2082 register(dir.path(), &repo, URL, "run1");
2083 assert!(w.load(&pr).local);
2085 w.lap(std::slice::from_ref(&repo), 60, 1, &no).await;
2086 w.lap(std::slice::from_ref(&repo), 60, 2, &no).await;
2087 assert!(fake.merges.lock().unwrap().is_empty());
2088 let qs = w.questions().list();
2089 assert_eq!(qs.len(), 1);
2090 assert_eq!(qs[0].choices, vec![land::APPROVE, land::HOLD]);
2091 answer(&w, &qs[0].id, land::APPROVE);
2093 w.lap(std::slice::from_ref(&repo), 60, 3, &no).await;
2094 assert!(fake.merges.lock().unwrap().is_empty());
2095 assert_eq!(w.load(&pr).question.as_deref(), Some(qs[0].id.as_str()));
2096 *fake.config_fails.lock().unwrap() = false;
2098 *fake.local.lock().unwrap() = true;
2099 w.lap(std::slice::from_ref(&repo), 60, 4, &no).await;
2100 assert_eq!(*fake.merges.lock().unwrap(), vec!["h1".to_owned()]);
2101 }
2102
2103 #[tokio::test]
2104 async fn a_local_record_written_before_the_flag_existed_is_still_local() {
2105 let (_d, fake, w) = rig();
2106 *fake.config_fails.lock().unwrap() = true;
2107 *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Open, "h1", vec![]));
2108 let repo = PathBuf::from("/nowhere");
2109 let no = || false;
2110 let pr = pr_key(URL).unwrap();
2111 let old = WatchState {
2112 repo: repo.to_string_lossy().into_owned(),
2113 url: URL.to_owned(),
2114 run: Some("run1".to_owned()),
2115 ..WatchState::default()
2116 };
2117 assert!(!old.local);
2118 w.save(&pr, &old);
2119 w.lap(std::slice::from_ref(&repo), 60, 1, &no).await;
2120 assert!(fake.merges.lock().unwrap().is_empty());
2121 let qs = w.questions().list();
2122 assert_eq!(qs.len(), 1);
2123 assert_eq!(qs[0].choices, vec![land::APPROVE, land::HOLD]);
2124 }
2125
2126 #[tokio::test]
2127 async fn an_unregistered_pr_is_an_actions_release_when_the_config_is_unreadable() {
2128 let (_d, fake, w) = rig();
2129 *fake.config_fails.lock().unwrap() = true;
2130 *fake.snap.lock().unwrap() = Some(red());
2131 let repo = PathBuf::from("/nowhere");
2132 let no = || false;
2133 w.lap(std::slice::from_ref(&repo), 3600, 1000, &no).await;
2134 assert_eq!(*fake.reruns.lock().unwrap(), vec!["11".to_owned()]);
2135 }
2136
2137 #[tokio::test]
2138 async fn a_merged_local_pr_waits_with_its_record_while_the_config_is_unreadable() {
2139 let (dir, fake, w) = rig();
2140 *fake.config_fails.lock().unwrap() = true;
2141 *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Merged, "h1", vec![]));
2142 let repo = PathBuf::from("/nowhere");
2143 let no = || false;
2144 let pr = pr_key(URL).unwrap();
2145 register(dir.path(), &repo, URL, "run1");
2146 w.lap(std::slice::from_ref(&repo), 60, 1, &no).await;
2147 assert!(w.state_path(&pr).exists(), "the record is kept");
2148 }
2149
2150 #[tokio::test]
2151 async fn land_approval_off_merges_the_observed_head_without_a_question() {
2152 let (_d, fake, w) = rig();
2153 *fake.local.lock().unwrap() = true;
2154 *fake.approval.lock().unwrap() = Some(false);
2155 *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Open, "h1", vec![]));
2156 let repo = PathBuf::from("/nowhere");
2157 let no = || false;
2158 w.lap(std::slice::from_ref(&repo), 60, 1, &no).await;
2159 assert_eq!(*fake.merges.lock().unwrap(), vec!["h1".to_owned()]);
2160 assert!(w.questions().list().is_empty(), "no approval question");
2161 }
2162
2163 #[tokio::test]
2164 async fn land_approval_on_or_unreadable_still_asks_and_never_merges() {
2165 for approval in [Some(true), None] {
2166 let (_d, fake, w) = rig();
2167 *fake.local.lock().unwrap() = true;
2168 *fake.approval.lock().unwrap() = approval;
2169 *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Open, "h1", vec![]));
2170 let repo = PathBuf::from("/nowhere");
2171 let no = || false;
2172 w.lap(std::slice::from_ref(&repo), 60, 1, &no).await;
2173 w.lap(std::slice::from_ref(&repo), 60, 2, &no).await;
2174 assert!(fake.merges.lock().unwrap().is_empty(), "silence is a hold");
2175 let qs = w.questions().list();
2176 assert_eq!(qs.len(), 1);
2177 assert_eq!(qs[0].choices, vec![land::APPROVE, land::HOLD]);
2178 }
2179 }
2180
2181 #[tokio::test]
2182 async fn land_approval_off_does_not_merge_on_an_unreadable_forge_or_retry_a_refusal() {
2183 let (_d, fake, w) = rig();
2184 *fake.local.lock().unwrap() = true;
2185 *fake.approval.lock().unwrap() = Some(false);
2186 let repo = PathBuf::from("/nowhere");
2187 let no = || false;
2188 *fake.snap.lock().unwrap() = None;
2190 w.lap(std::slice::from_ref(&repo), 60, 1, &no).await;
2191 assert!(fake.merges.lock().unwrap().is_empty());
2192 *fake.merge_fails.lock().unwrap() = true;
2194 *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Open, "h1", vec![]));
2195 w.lap(std::slice::from_ref(&repo), 60, 2, &no).await;
2196 w.lap(std::slice::from_ref(&repo), 60, 3, &no).await;
2197 assert_eq!(fake.merges.lock().unwrap().len(), 1);
2198 assert!(w.questions().list().is_empty());
2199 *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Open, "h2", vec![]));
2201 w.lap(std::slice::from_ref(&repo), 60, 4, &no).await;
2202 assert_eq!(
2203 *fake.merges.lock().unwrap(),
2204 vec!["h1".to_owned(), "h2".to_owned()]
2205 );
2206 }
2207
2208 #[test]
2209 fn pr_info_is_read_from_gh_json() {
2210 let i = parse_info(
2211 r#"{"headRefName":"chore/release-v1.2.3","title":"chore: release v1.2.3","mergeCommit":{"oid":"abc"}}"#,
2212 )
2213 .unwrap();
2214 assert_eq!(i.branch, "chore/release-v1.2.3");
2215 assert_eq!(i.merge_commit.as_deref(), Some("abc"));
2216 let open = parse_info(r#"{"headRefName":"b","title":"t","mergeCommit":null}"#).unwrap();
2217 assert_eq!(open.merge_commit, None);
2218 }
2219
2220 #[tokio::test]
2221 async fn a_local_release_asks_once_without_ci_reruns_or_stall_escalation() {
2222 let (_d, fake, w) = rig();
2223 *fake.local.lock().unwrap() = true;
2224 *fake.snap.lock().unwrap() = Some(red());
2227 let repo = PathBuf::from("/nowhere");
2228 let no = || false;
2229 w.lap(std::slice::from_ref(&repo), 60, 1_000_000, &no).await;
2230 w.lap(std::slice::from_ref(&repo), 60, 2_000_000, &no).await;
2231 assert!(fake.reruns.lock().unwrap().is_empty());
2232 let qs = w.questions().list();
2233 assert_eq!(qs.len(), 1, "one approval question, not one per lap");
2234 assert_eq!(qs[0].choices, vec![land::APPROVE, land::HOLD]);
2235 assert!(fake.merges.lock().unwrap().is_empty(), "silence is a hold");
2236 }
2237
2238 #[tokio::test]
2239 async fn merge_is_pinned_to_the_approved_head_and_a_moved_head_asks_again() {
2240 let (_d, fake, w) = rig();
2241 *fake.local.lock().unwrap() = true;
2242 *fake.snap.lock().unwrap() = Some(open(vec![]));
2243 let repo = PathBuf::from("/nowhere");
2244 let no = || false;
2245 w.lap(std::slice::from_ref(&repo), 60, 1, &no).await;
2246 let original = &w.questions().list()[0];
2247 assert_eq!(original.merge_target.as_ref().unwrap().head, "h1");
2248 let id = original.id.clone();
2249 w.questions()
2250 .update(&id, |q| {
2251 q.merge_target = None;
2252 Ok(())
2253 })
2254 .unwrap();
2255 assert_eq!(
2256 w.questions()
2257 .get(&id)
2258 .unwrap()
2259 .merge_target
2260 .as_ref()
2261 .unwrap()
2262 .head,
2263 "h1",
2264 "legacy binding comes from the matching persisted watch record"
2265 );
2266 *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Open, "h2", vec![]));
2268 answer(&w, &id, land::APPROVE);
2269 w.lap(std::slice::from_ref(&repo), 60, 2, &no).await;
2270 assert!(fake.merges.lock().unwrap().is_empty());
2271 assert_eq!(w.questions().list().len(), 2, "asked again about h2");
2272 let id2 = w
2274 .questions()
2275 .list()
2276 .into_iter()
2277 .find(|q| q.status.open())
2278 .unwrap()
2279 .id;
2280 answer(&w, &id2, land::APPROVE);
2281 w.lap(std::slice::from_ref(&repo), 60, 3, &no).await;
2282 assert_eq!(*fake.merges.lock().unwrap(), vec!["h2".to_owned()]);
2283 }
2284
2285 #[tokio::test]
2286 async fn a_failed_release_holds_with_one_notice_and_one_question_and_never_retries() {
2287 let (d, fake, w) = rig();
2288 *fake.local.lock().unwrap() = true;
2289 *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Merged, "h1", vec![]));
2290 *fake.info.lock().unwrap() = Some(PrInfo {
2291 branch: "chore/release-v1.0.0".to_owned(),
2292 merge_commit: Some("deadbeef".to_owned()),
2293 title: "chore: release v1.0.0".to_owned(),
2294 });
2295 let repo = d.path().join("nowhere");
2297 let no = || false;
2298 w.lap(std::slice::from_ref(&repo), 60, 1, &no).await;
2299 let st = w.load("o/r#7");
2300 let job = st.job.expect("the job is recorded");
2301 assert!(job.failed.is_some() && !job.tag_done && !job.finished);
2302 let qs = w.questions().list();
2303 assert_eq!(qs.len(), 1);
2304 assert_eq!(qs[0].choices, vec![RETRY, LEAVE_IT]);
2305 w.lap(std::slice::from_ref(&repo), 60, 2, &no).await;
2307 assert_eq!(w.questions().list().len(), 1);
2308 let ns = Notices::at(d.path().join("notifications")).list();
2309 assert_eq!(ns.len(), 1);
2310 assert_eq!(
2311 ns[0].link,
2312 Some(Link::Url {
2313 url: URL.to_owned(),
2314 })
2315 );
2316 }
2317
2318 #[tokio::test]
2319 async fn an_escalated_pull_request_records_the_title_version_as_acceptable() {
2320 let (d, fake, w) = rig();
2321 *fake.local.lock().unwrap() = true;
2322 *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Merged, "h1", vec![]));
2323 *fake.info.lock().unwrap() = Some(PrInfo {
2324 branch: "chore/release-v0.1.4".to_owned(),
2325 merge_commit: Some("deadbeef".to_owned()),
2326 title: "chore: release v0.2.0 (minor bump)".to_owned(),
2327 });
2328 let repo = d.path().join("nowhere");
2329 w.lap(std::slice::from_ref(&repo), 60, 1, &(|| false)).await;
2330 let job = w.load("o/r#7").job.expect("the job is recorded");
2331 assert_eq!(job.version, "0.1.4");
2332 assert_eq!(job.accepted, vec!["0.2.0".to_owned()]);
2333 }
2334
2335 #[tokio::test]
2336 async fn a_failed_release_holds_the_task_whose_run_opened_the_pr() {
2337 let (d, fake, w) = rig();
2338 *fake.local.lock().unwrap() = true;
2339 *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Merged, "h1", vec![]));
2340 *fake.info.lock().unwrap() = Some(PrInfo {
2341 branch: "chore/release-v1.0.0".to_owned(),
2342 merge_commit: Some("deadbeef".to_owned()),
2343 title: "t".to_owned(),
2344 });
2345 let repo = d.path().join("nowhere");
2346 let q = crate::queue::Queue::at(d.path().join("queue"));
2347 let mut t = crate::queue::Task::new(
2348 "t".to_owned(),
2349 "i".to_owned(),
2350 repo.clone(),
2351 crate::queue::Source::Human,
2352 );
2353 t.runs.push("run1".to_owned());
2354 t.status = crate::queue::TaskStatus::Done;
2355 q.put(&mut t).unwrap();
2356 register(d.path(), &repo, URL, "run1");
2357 w.lap(std::slice::from_ref(&repo), 60, 1, &(|| false)).await;
2358 let held = q.get(&t.id).unwrap();
2359 assert_eq!(held.status, crate::queue::TaskStatus::Held);
2360 assert!(
2361 held.hold_reason
2362 .unwrap_or_default()
2363 .contains("release v1.0.0")
2364 );
2365 }
2366
2367 #[tokio::test]
2368 async fn unhold_gives_the_task_back_only_when_the_hold_is_ours() {
2369 let (d, _fake, w) = rig();
2370 let q = crate::queue::Queue::at(d.path().join("queue"));
2371 let mut t = crate::queue::Task::new(
2372 "t".to_owned(),
2373 "i".to_owned(),
2374 PathBuf::from("/r"),
2375 crate::queue::Source::Human,
2376 );
2377 t.runs.push("run1".to_owned());
2378 t.status = crate::queue::TaskStatus::Done;
2379 q.put(&mut t).unwrap();
2380 let mut st = WatchState {
2381 run: Some("run1".to_owned()),
2382 ..WatchState::default()
2383 };
2384 w.hold_task(&mut st, "failed");
2385 assert_eq!(q.get(&t.id).unwrap().status, crate::queue::TaskStatus::Held);
2386 w.hold_task(&mut st, "failed again");
2388 assert!(w.unhold_task(&mut st));
2389 assert_eq!(q.get(&t.id).unwrap().status, crate::queue::TaskStatus::Done);
2390 assert!(st.held_task.is_none());
2391
2392 let mut h = q.get(&t.id).unwrap();
2394 st.held_task = Some(h.id.clone());
2395 h.hold_manual(Some("mine".to_owned()));
2396 q.put(&mut h).unwrap();
2397 assert!(w.unhold_task(&mut st));
2398 assert_eq!(q.get(&t.id).unwrap().status, crate::queue::TaskStatus::Held);
2399 }
2400
2401 #[tokio::test]
2402 async fn a_restart_after_the_failure_was_saved_still_holds_the_task() {
2403 let (d, fake, w) = rig();
2404 *fake.local.lock().unwrap() = true;
2405 *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Merged, "h1", vec![]));
2406 let repo = d.path().join("nowhere");
2407 let q = crate::queue::Queue::at(d.path().join("queue"));
2408 let mut t = crate::queue::Task::new(
2409 "t".to_owned(),
2410 "i".to_owned(),
2411 repo.clone(),
2412 crate::queue::Source::Human,
2413 );
2414 t.runs.push("run1".to_owned());
2415 t.status = crate::queue::TaskStatus::Done;
2416 q.put(&mut t).unwrap();
2417 let mut job = Job::new("1.0.0", URL, "deadbeef");
2419 job.failed = Some("command 1 exited 3".to_owned());
2420 let st = WatchState {
2421 repo: repo.to_string_lossy().into_owned(),
2422 url: URL.to_owned(),
2423 run: Some("run1".to_owned()),
2424 job: Some(job),
2425 ..WatchState::default()
2426 };
2427 w.save("o/r#7", &st);
2428 w.lap(std::slice::from_ref(&repo), 60, 1, &(|| false)).await;
2429 assert_eq!(q.get(&t.id).unwrap().status, crate::queue::TaskStatus::Held);
2430 }
2431
2432 #[tokio::test]
2433 async fn a_restart_after_the_job_finished_still_gives_the_task_back() {
2434 let (d, fake, w) = rig();
2435 *fake.local.lock().unwrap() = true;
2436 *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Merged, "h1", vec![]));
2437 let repo = d.path().join("nowhere");
2438 let q = crate::queue::Queue::at(d.path().join("queue"));
2439 let mut t = crate::queue::Task::new(
2440 "t".to_owned(),
2441 "i".to_owned(),
2442 repo.clone(),
2443 crate::queue::Source::Human,
2444 );
2445 t.runs.push("run1".to_owned());
2446 t.hold_machine(Some(format!("{HOLD_PREFIX}release v1.0.0 failed")));
2447 q.put(&mut t).unwrap();
2448 let mut job = Job::new("1.0.0", URL, "deadbeef");
2450 job.finished = true;
2451 let st = WatchState {
2452 repo: repo.to_string_lossy().into_owned(),
2453 url: URL.to_owned(),
2454 run: Some("run1".to_owned()),
2455 job: Some(job),
2456 held_task: Some(t.id.clone()),
2457 ..WatchState::default()
2458 };
2459 w.save("o/r#7", &st);
2460 w.lap(std::slice::from_ref(&repo), 60, 1, &(|| false)).await;
2461 assert_eq!(q.get(&t.id).unwrap().status, crate::queue::TaskStatus::Done);
2462 assert!(!w.state_path("o/r#7").exists());
2463 }
2464
2465 #[tokio::test]
2466 async fn a_pull_request_registered_at_open_is_picked_up_even_if_unlisted() {
2467 let (d, _fake, w) = rig();
2468 register(d.path(), Path::new("/r"), URL, "run1");
2469 let st = w.load("o/r#7");
2470 assert_eq!(st.url, URL);
2471 assert_eq!(st.repo, "/r");
2472 let mut live = st.clone();
2474 live.head = "keep".to_owned();
2475 w.save("o/r#7", &live);
2476 register(d.path(), Path::new("/r"), URL, "run1");
2477 assert_eq!(w.load("o/r#7").head, "keep");
2478 }
2479
2480 #[tokio::test]
2481 async fn a_finished_release_is_kept_on_the_run_and_a_missing_run_still_completes() {
2482 let (d, fake, w) = rig();
2483 *fake.local.lock().unwrap() = true;
2484 *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Merged, "h1", vec![]));
2485 let repo = d.path().join("nowhere");
2486 let mut run = RunState::new(
2487 repo.clone(),
2488 "main".to_owned(),
2489 "0123456789abcdef".to_owned(),
2490 "t".to_owned(),
2491 crate::config::Config::default(),
2492 );
2493 run.id = "run1".to_owned();
2494 run.save_under(d.path()).unwrap();
2495 let mut job = Job::new("1.0.0", URL, "deadbeef");
2496 job.finished = true;
2497 job.log.push(release_local::StepLog {
2498 name: "cmd".to_owned(),
2499 code: Some(0),
2500 tail: "ok".to_owned(),
2501 output: None,
2502 });
2503 let mk = |run: &str| WatchState {
2504 repo: repo.to_string_lossy().into_owned(),
2505 url: URL.to_owned(),
2506 run: Some(run.to_owned()),
2507 job: Some(job.clone()),
2508 ..WatchState::default()
2509 };
2510 w.save("o/r#7", &mk("run1"));
2511 w.lap(std::slice::from_ref(&repo), 60, 1, &(|| false)).await;
2512 assert!(!w.state_path("o/r#7").exists());
2513 let kept = RunState::load_under("run1", d.path()).unwrap();
2514 assert_eq!(kept.release_bump.unwrap().release, Some(job.clone()));
2515
2516 w.save("o/r#7", &mk("gone"));
2518 w.lap(std::slice::from_ref(&repo), 60, 1, &(|| false)).await;
2519 assert!(!w.state_path("o/r#7").exists());
2520 }
2521
2522 #[tokio::test]
2523 async fn raise_released_attaches_url_link() {
2524 let (d, _fake, w) = rig();
2525 let job = Job::new("1.0.0", URL, "deadbeef");
2526 w.raise_released("o/r#7", URL, &job);
2527 let ns = Notices::at(d.path().join("notifications")).list();
2528 assert_eq!(ns.len(), 1);
2529 assert_eq!(ns[0].key, "release-done:o/r#7");
2530 assert_eq!(
2531 ns[0].link,
2532 Some(Link::Url {
2533 url: URL.to_owned(),
2534 })
2535 );
2536 }
2537}