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