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