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
58const 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.hold_machine(Some(format!("{HOLD_PREFIX}{reason}")));
1159 match q.put(&mut t) {
1160 Ok(()) => st.held_task = Some(t.id.clone()),
1161 Err(e) => tracing::warn!("could not hold task {} for {run}: {e:#}", t.id),
1162 }
1163 }
1164 return;
1165 }
1166 }
1167 }
1168
1169 #[must_use]
1176 fn unhold_task(&self, st: &mut WatchState) -> bool {
1177 let Some(id) = st.held_task.take() else {
1178 return true;
1179 };
1180 let q = crate::queue::Queue::at(self.home.join("queue"));
1181 let mut t = match q.get(&id) {
1182 Ok(t) => t,
1183 Err(_) if matches!(q.path_of(&id).try_exists(), Ok(false)) => return true,
1186 Err(e) => {
1187 tracing::warn!("could not read task {id} to restore it after the release: {e:#}");
1188 st.held_task = Some(id);
1189 return false;
1190 }
1191 };
1192 let ours = t.status == crate::queue::TaskStatus::Held
1193 && t.hold_source == Some(crate::queue::HoldSource::Machine)
1194 && t.hold_reason
1195 .as_deref()
1196 .is_some_and(|r| r.starts_with(HOLD_PREFIX));
1197 if ours {
1198 t.succeed();
1199 if let Err(e) = q.put(&mut t) {
1200 tracing::warn!("could not restore task {id} after the release: {e:#}");
1201 st.held_task = Some(id);
1202 return false;
1203 }
1204 }
1205 true
1206 }
1207
1208 fn complete(&self, pr: &str, st: &mut WatchState) {
1212 if self.unhold_task(st) {
1213 self.finish(pr, st);
1214 } else {
1215 self.save(pr, st);
1216 }
1217 }
1218
1219 fn raise_released(&self, pr: &str, url: &str, job: &Job) {
1220 notices::raise_in(
1221 &self.home,
1222 Notice::info(&released_key(pr), format!("Released {} ({pr})", job.tag())).link(
1223 Link::Url {
1224 url: url.to_owned(),
1225 },
1226 ),
1227 );
1228 }
1229
1230 fn file_job_question(&self, pr: &str, st: &mut WatchState, job: &Job) {
1232 let last = job
1233 .log
1234 .last()
1235 .map(|l| format!("\nLast step: {}\n\n{}\n", l.name, l.tail))
1236 .unwrap_or_default();
1237 let mut q = Question::new(
1238 String::new(),
1239 crate::bump::NOTICE_NODE.to_owned(),
1240 "release-watch".to_owned(),
1241 format!("Release {} of {pr} is on hold: retry?", job.tag()),
1242 format!(
1243 "Pull request: {}\nWhy it stopped: {}\n{last}\n\
1244 Full output is under the magi home's release-local directory.\n\n\
1245 - `{RETRY}`: run it again from where it stopped (the tag is not \
1246 recreated and finished commands are skipped).\n\
1247 - `{LEAVE_IT}`: stop watching; release by hand.\n\n\
1248 Nothing is retried on its own; silence is a hold.\n",
1249 job.pr_url,
1250 job.failed.clone().unwrap_or_default()
1251 ),
1252 vec![RETRY.to_owned(), LEAVE_IT.to_owned()],
1253 );
1254 match self.questions().put(&mut q) {
1255 Ok(()) => st.question = Some(q.id.clone()),
1256 Err(e) => tracing::warn!("could not file the question for {pr}: {e:#}"),
1257 }
1258 }
1259
1260 fn settle_job_question(&self, pr: &str, st: &mut WatchState, job: &mut Job) -> bool {
1264 let why = job.failed.clone().unwrap_or_default();
1265 if let Some(id) = st.question.clone() {
1266 match self.questions().get(&id) {
1267 Ok(q) if q.status.open() => return false,
1268 Ok(q) => {
1269 st.question = None;
1270 match &q.answer {
1271 Some(Answer::Choice(c)) if c == RETRY => {
1272 job.resume();
1273 st.held = None;
1274 return true;
1275 }
1276 Some(Answer::Choice(c)) if c == LEAVE_IT => {
1277 st.ignored = true;
1278 let _ = Notices::at(self.home.join("notifications"))
1279 .dismiss(¬ices::id_of(¬ice_key(pr)));
1280 }
1281 _ => st.held = Some(why),
1283 }
1284 return false;
1285 }
1286 Err(_) => st.question = None,
1287 }
1288 }
1289 if !st.ignored && st.held.as_deref() != Some(why.as_str()) {
1290 self.raise(pr, &st.url, failed_message(pr));
1291 self.file_job_question(pr, st, job);
1292 }
1293 false
1294 }
1295
1296 fn apply_answer(&self, pr: &str, st: &mut WatchState, snap: &RollupView, q: &Question) {
1298 st.question = None;
1299 if st.applied.contains(&q.id) {
1300 return;
1301 }
1302 st.applied.push(q.id.clone());
1303 if st.applied.len() > 20 {
1304 st.applied.remove(0);
1305 }
1306 match &q.answer {
1307 Some(Answer::Choice(c)) if c == RERUN_AGAIN => {
1308 st.held = None;
1309 for c in snap.checks.iter().filter(|c| is_failed(c.verdict)) {
1310 if let Some(r) = &c.run {
1311 st.reruns.remove(r);
1312 }
1313 }
1314 }
1315 Some(Answer::Choice(c)) if c == LEAVE_IT => {
1316 st.ignored = true;
1317 if let Err(e) = Notices::at(self.home.join("notifications"))
1318 .dismiss(¬ices::id_of(¬ice_key(pr)))
1319 {
1320 tracing::warn!("could not dismiss the notice for {pr}: {e:#}");
1321 }
1322 }
1323 _ => st.held = Some(fingerprint(snap)),
1325 }
1326 }
1327
1328 fn finish(&self, pr: &str, st: &WatchState) {
1330 let _ =
1331 Notices::at(self.home.join("notifications")).dismiss(¬ices::id_of(¬ice_key(pr)));
1332 if let Some(id) = &st.question {
1333 let _ = self.questions().update(id, |q| {
1334 q.abandon("the release pull request is no longer open");
1335 Ok(())
1336 });
1337 }
1338 let _ = std::fs::remove_file(self.state_path(pr));
1339 }
1340}
1341
1342pub(crate) async fn run(
1346 watcher: Watcher,
1347 settings: impl Fn() -> (Vec<PathBuf>, u64),
1348 stop: crate::daemon::Stop,
1349) {
1350 let halt = {
1351 let stop = stop.clone();
1352 move || stop.stopped()
1353 };
1354 while !stop.stopped() {
1355 let (repos, minutes) = settings();
1356 if minutes > 0 {
1357 let now = jiff::Timestamp::now().as_second();
1358 watcher
1359 .lap(&repos, (minutes as i64).saturating_mul(60), now, &halt)
1360 .await;
1361 }
1362 let mut slept = Duration::ZERO;
1363 while slept < LAP && !stop.stopped() {
1364 tokio::time::sleep(Duration::from_secs(1)).await;
1365 slept += Duration::from_secs(1);
1366 }
1367 }
1368}
1369
1370#[cfg(test)]
1371mod tests {
1372 use super::*;
1373 use crate::land::CheckView;
1374 use anyhow::Context;
1375 use std::sync::Mutex;
1376
1377 const URL: &str = "https://github.com/o/r/pull/7";
1378
1379 fn check(name: &str, v: Verdict, run: Option<&str>) -> CheckView {
1380 CheckView {
1381 name: name.to_owned(),
1382 verdict: v,
1383 run: run.map(str::to_owned),
1384 url: run.map(|r| format!("https://github.com/o/r/actions/runs/{r}/job/1")),
1385 }
1386 }
1387
1388 fn snap(state: PrLifecycle, head: &str, checks: Vec<CheckView>) -> RollupView {
1389 RollupView {
1390 url: URL.to_owned(),
1391 number: 7,
1392 state,
1393 head: head.to_owned(),
1394 checks,
1395 }
1396 }
1397
1398 fn open(checks: Vec<CheckView>) -> RollupView {
1399 snap(PrLifecycle::Open, "h1", checks)
1400 }
1401
1402 fn st_at(progress: i64) -> WatchState {
1403 WatchState {
1404 progress_at: progress,
1405 ..WatchState::default()
1406 }
1407 }
1408
1409 #[test]
1410 fn unreadable_is_unknown_and_merged_or_closed_is_done() {
1411 assert_eq!(decide(None, &st_at(0), 10, 60), Step::Unknown);
1412 for s in [PrLifecycle::Merged, PrLifecycle::Closed] {
1413 assert_eq!(
1414 decide(Some(&snap(s, "h", vec![])), &st_at(0), 10, 60),
1415 Step::Done
1416 );
1417 }
1418 }
1419
1420 #[test]
1421 fn a_failed_run_is_rerun_even_while_another_run_is_pending() {
1422 let s = open(vec![
1423 check("win", Verdict::Fail, Some("11")),
1424 check("lint", Verdict::Pending, Some("12")),
1425 ]);
1426 assert_eq!(
1427 decide(Some(&s), &st_at(0), 1, 3600),
1428 Step::Rerun(vec!["11".into()])
1429 );
1430 }
1431
1432 #[test]
1433 fn a_run_with_a_job_still_pending_is_not_complete() {
1434 let s = open(vec![
1435 check("win", Verdict::Fail, Some("11")),
1436 check("mac", Verdict::Pending, Some("11")),
1437 ]);
1438 assert_eq!(decide(Some(&s), &st_at(0), 1, 3600), Step::Wait);
1439 }
1440
1441 #[test]
1442 fn one_rerun_per_run_then_grace_then_escalation() {
1443 let s = open(vec![check("win", Verdict::Fail, Some("11"))]);
1444 let mut st = st_at(0);
1445 st.reruns.insert("11".into(), 100);
1446 assert_eq!(decide(Some(&s), &st, 100 + RERUN_GRACE - 1, 0), Step::Wait);
1447 assert_eq!(
1448 decide(Some(&s), &st, 100 + RERUN_GRACE, 0),
1449 Step::Escalate(Why::StillRed)
1450 );
1451 }
1452
1453 #[test]
1454 fn a_failure_with_no_run_id_goes_straight_to_a_human_once_settled() {
1455 let s = open(vec![check("ci/ext", Verdict::Fail, None)]);
1456 assert_eq!(
1457 decide(Some(&s), &st_at(0), 1, 0),
1458 Step::Escalate(Why::StillRed)
1459 );
1460 let s = open(vec![
1461 check("ci/ext", Verdict::Fail, None),
1462 check("x", Verdict::Pending, Some("5")),
1463 ]);
1464 assert_eq!(decide(Some(&s), &st_at(0), 1, 0), Step::Wait);
1465 }
1466
1467 #[test]
1468 fn no_progress_for_the_bounded_time_escalates_and_zero_disables_it() {
1469 let s = open(vec![check("a", Verdict::Pass, Some("1"))]);
1470 assert_eq!(decide(Some(&s), &st_at(0), 3599, 3600), Step::Wait);
1471 assert_eq!(
1472 decide(Some(&s), &st_at(0), 3600, 3600),
1473 Step::Escalate(Why::Stalled)
1474 );
1475 assert_eq!(decide(Some(&s), &st_at(0), 99_999, 0), Step::Wait);
1476 }
1477
1478 #[test]
1479 fn held_and_ignored_stay_quiet_until_the_fingerprint_moves() {
1480 let s = open(vec![check("a", Verdict::Fail, None)]);
1481 let mut st = st_at(0);
1482 st.held = Some(fingerprint(&s));
1483 assert_eq!(decide(Some(&s), &st, 99_999, 60), Step::Wait);
1484 let moved = open(vec![
1485 check("a", Verdict::Pass, None),
1486 check("b", Verdict::Fail, None),
1487 ]);
1488 assert_eq!(
1489 decide(Some(&moved), &st, 99_999, 60),
1490 Step::Escalate(Why::StillRed)
1491 );
1492 st.ignored = true;
1493 assert_eq!(decide(Some(&moved), &st, 99_999, 60), Step::Wait);
1494 }
1495
1496 #[test]
1497 fn observe_restarts_the_clock_only_on_change_and_forgets_reruns_on_a_new_head() {
1498 let mut st = st_at(0);
1499 let a = open(vec![check("a", Verdict::Pending, Some("1"))]);
1500 observe(&mut st, &a, 10);
1501 assert_eq!(st.progress_at, 10);
1502 observe(&mut st, &a, 50);
1503 assert_eq!(st.progress_at, 10);
1504 st.reruns.insert("1".into(), 5);
1505 observe(
1506 &mut st,
1507 &snap(PrLifecycle::Open, "h2", a.checks.clone()),
1508 60,
1509 );
1510 assert!(st.reruns.is_empty());
1511 assert_eq!(st.progress_at, 60);
1512 }
1513
1514 #[test]
1515 fn pr_key_names_owner_repo_and_number() {
1516 assert_eq!(pr_key(URL).as_deref(), Some("o/r#7"));
1517 assert_eq!(pr_key("https://github.com/o/r/issues/7"), None);
1518 assert_eq!(pr_key("nonsense"), None);
1519 }
1520
1521 #[test]
1522 fn notice_wording_does_not_vary_between_polls() {
1523 assert_eq!(rerun_message("o/r#7"), rerun_message("o/r#7"));
1524 assert!(
1525 !human_message("o/r#7")
1526 .chars()
1527 .any(|c| c.is_ascii_digit() && c != '7')
1528 );
1529 }
1530
1531 #[test]
1532 fn the_release_question_is_not_claimed_by_other_machinery() {
1533 let q = Question::new(
1534 String::new(),
1535 crate::bump::NOTICE_NODE.into(),
1536 "release-watch".into(),
1537 "s".into(),
1538 String::new(),
1539 vec![HOLD.into()],
1540 );
1541 assert_eq!(
1543 crate::deputy::kind_of(&q),
1544 Some(crate::deputy::Kind::Release)
1545 );
1546 let n = Notice::warn("release-pr:o/r#7", "m").about([String::new()]);
1547 assert!(!notices::covers(&q, &n));
1548 let mut plain = q.clone();
1550 plain.seat = "bump".into();
1551 assert_eq!(crate::deputy::kind_of(&plain), None);
1552 }
1553
1554 #[derive(Default)]
1555 struct Fake {
1556 snap: Mutex<Option<RollupView>>,
1557 reruns: Mutex<Vec<String>>,
1558 local: Mutex<bool>,
1559 info: Mutex<Option<PrInfo>>,
1560 merges: Mutex<Vec<String>>,
1561 approval: Mutex<Option<bool>>,
1563 merge_fails: Mutex<bool>,
1564 config_fails: Mutex<bool>,
1566 }
1567
1568 impl ReleaseForge for std::sync::Arc<Fake> {
1569 fn list<'a>(&'a self, _: &'a Path) -> Fut<'a, Result<Vec<(String, String)>>> {
1570 Box::pin(async { Ok(vec![("chore/release-v1.0.0".to_owned(), URL.to_owned())]) })
1571 }
1572 fn snapshot<'a>(&'a self, _: &'a Path, _: &'a str) -> Fut<'a, Result<RollupView>> {
1573 let s = self.snap.lock().unwrap().clone();
1574 Box::pin(async move { s.context("unreadable") })
1575 }
1576 fn rerun<'a>(&'a self, _: &'a Path, run: &'a str) -> Fut<'a, Result<()>> {
1577 self.reruns.lock().unwrap().push(run.to_owned());
1578 Box::pin(async { Ok(()) })
1579 }
1580 fn config<'a>(&'a self, _: &'a Path) -> Fut<'a, Result<crate::config::Config>> {
1581 if *self.config_fails.lock().unwrap() {
1582 return Box::pin(async { bail!("unreadable config") });
1583 }
1584 let mut c = crate::config::Config::default();
1585 if *self.local.lock().unwrap() {
1586 c.release.mode = crate::config::ReleaseMode::Local;
1587 c.release.commands = vec!["true".to_owned()];
1588 }
1589 Box::pin(async move { Ok(c) })
1590 }
1591 fn info<'a>(&'a self, _: &'a Path, _: &'a str) -> Fut<'a, Result<PrInfo>> {
1592 let i = self.info.lock().unwrap().clone();
1593 Box::pin(async move { i.context("no info") })
1594 }
1595 fn merge<'a>(&'a self, _: &'a Path, _: &'a str, head: &'a str) -> Fut<'a, Result<bool>> {
1596 self.merges.lock().unwrap().push(head.to_owned());
1597 let fails = *self.merge_fails.lock().unwrap();
1598 Box::pin(async move { if fails { bail!("refused") } else { Ok(true) } })
1599 }
1600 fn land_approval<'a>(&'a self, _: &'a Path) -> Fut<'a, Result<bool>> {
1601 let a = *self.approval.lock().unwrap();
1602 Box::pin(async move { a.context("unreadable config") })
1603 }
1604 }
1605
1606 fn rig() -> (tempfile::TempDir, std::sync::Arc<Fake>, Watcher) {
1607 let dir = tempfile::tempdir().unwrap();
1608 let fake = std::sync::Arc::new(Fake::default());
1609 *fake.approval.lock().unwrap() = Some(true);
1611 let w = Watcher::new(Box::new(fake.clone()), dir.path().to_path_buf());
1612 (dir, fake, w)
1613 }
1614
1615 fn red() -> RollupView {
1616 open(vec![check("win", Verdict::Fail, Some("11"))])
1617 }
1618
1619 #[tokio::test]
1620 async fn reruns_once_survives_a_restart_then_escalates_with_one_question() {
1621 let (dir, fake, w) = rig();
1622 *fake.snap.lock().unwrap() = Some(red());
1623 let repo = PathBuf::from("/nowhere");
1624 let no = || false;
1625 w.lap(std::slice::from_ref(&repo), 3600, 1000, &no).await;
1626 assert_eq!(*fake.reruns.lock().unwrap(), vec!["11".to_owned()]);
1627
1628 let w2 = Watcher::new(Box::new(fake.clone()), dir.path().to_path_buf());
1630 w2.lap(
1631 std::slice::from_ref(&repo),
1632 3600,
1633 1000 + RERUN_GRACE - 1,
1634 &no,
1635 )
1636 .await;
1637 assert_eq!(fake.reruns.lock().unwrap().len(), 1);
1638 assert!(w2.questions().list().is_empty());
1639
1640 w2.lap(
1641 std::slice::from_ref(&repo),
1642 3600,
1643 1000 + RERUN_GRACE + 1,
1644 &no,
1645 )
1646 .await;
1647 w2.lap(
1648 std::slice::from_ref(&repo),
1649 3600,
1650 1000 + RERUN_GRACE + 400,
1651 &no,
1652 )
1653 .await;
1654 assert_eq!(fake.reruns.lock().unwrap().len(), 1);
1655 let qs = w2.questions().list();
1656 assert_eq!(qs.len(), 1, "asked once, not every lap");
1657 assert_eq!(qs[0].node, crate::bump::NOTICE_NODE);
1658 assert!(qs[0].detail.contains("win") && qs[0].detail.contains(URL));
1659 assert_eq!(qs[0].choices, vec![RERUN_AGAIN, HOLD, LEAVE_IT]);
1660 let ns = Notices::at(dir.path().join("notifications")).list();
1661 assert_eq!(ns.len(), 1);
1662 assert_eq!(ns[0].key, "release-pr:o/r#7");
1663 assert_eq!(
1664 ns[0].link,
1665 Some(Link::Url {
1666 url: URL.to_owned(),
1667 })
1668 );
1669 }
1670
1671 async fn escalated() -> (tempfile::TempDir, std::sync::Arc<Fake>, Watcher, String) {
1672 let (dir, fake, w) = rig();
1673 *fake.snap.lock().unwrap() = Some(red());
1674 let repo = PathBuf::from("/nowhere");
1675 let no = || false;
1676 w.lap(std::slice::from_ref(&repo), 3600, 1000, &no).await;
1677 w.lap(&[repo], 3600, 2000, &no).await;
1678 let id = w.questions().list()[0].id.clone();
1679 (dir, fake, w, id)
1680 }
1681
1682 fn answer(w: &Watcher, id: &str, choice: &str) {
1683 w.questions()
1684 .update(id, |q| q.answer(Answer::Choice(choice.to_owned())))
1685 .unwrap();
1686 }
1687
1688 #[tokio::test]
1689 async fn rerun_again_reruns_exactly_once_more() {
1690 let (_d, fake, w, id) = escalated().await;
1691 answer(&w, &id, RERUN_AGAIN);
1692 let repo = PathBuf::from("/nowhere");
1693 let no = || false;
1694 w.lap(std::slice::from_ref(&repo), 3600, 3000, &no).await;
1695 w.lap(&[repo], 3600, 3001, &no).await;
1696 assert_eq!(fake.reruns.lock().unwrap().len(), 2);
1697 }
1698
1699 #[tokio::test]
1700 async fn hold_and_silence_do_not_reask_and_leave_it_dismisses() {
1701 let (d, fake, w, id) = escalated().await;
1702 answer(&w, &id, HOLD);
1703 let repo = PathBuf::from("/nowhere");
1704 let no = || false;
1705 for t in [3000, 90_000, 200_000] {
1706 w.lap(std::slice::from_ref(&repo), 3600, t, &no).await;
1707 }
1708 assert_eq!(w.questions().list().len(), 1, "held: no second question");
1709 assert_eq!(fake.reruns.lock().unwrap().len(), 1);
1710
1711 let (d2, _f2, w2, id2) = escalated().await;
1712 answer(&w2, &id2, LEAVE_IT);
1713 w2.lap(std::slice::from_ref(&repo), 3600, 3000, &no).await;
1714 let ns = Notices::at(d2.path().join("notifications")).list();
1715 assert!(ns.is_empty(), "dismissed notices are hidden");
1716 drop(d);
1717 }
1718
1719 fn with_deputy(w: &Watcher, id: &str, home: &Path) -> String {
1721 let seat = crate::agent::SeatState::new(&crate::ask::deputy_seat_key(id), "a", 1);
1722 let key = seat.key.clone();
1723 let brief = deputy_brief(&w.questions().get(id).unwrap(), home);
1724 w.questions()
1725 .update(id, |q| {
1726 let mut d = crate::ask::Deputy::new(brief);
1727 d.seat = Some(seat);
1728 q.deputy = Some(d);
1729 Ok(())
1730 })
1731 .unwrap();
1732 key
1733 }
1734
1735 #[tokio::test]
1736 async fn the_brief_names_the_pull_request_and_what_leave_it_really_does() {
1737 let (d, _f, w, id) = escalated().await;
1738 let q = w.questions().get(&id).unwrap();
1739 let b = deputy_brief(&q, d.path());
1740 assert!(b.contains(URL), "{b}");
1741 assert!(
1742 b.contains("`leave it`") && b.contains("does NOT close"),
1743 "{b}"
1744 );
1745 assert!(b.contains("`rerun again`") && b.contains("`hold`"), "{b}");
1746 assert!(!b.contains("`merge`"), "no merge on an escalation: {b}");
1747
1748 let empty = tempfile::tempdir().unwrap();
1750 let b = deputy_brief(&q, empty.path());
1751 assert!(b.contains("could not be found") && !b.contains(URL), "{b}");
1752 }
1753
1754 #[tokio::test]
1755 async fn a_settled_leave_it_is_applied_once_by_the_watcher_and_closes_nothing() {
1756 let (d, fake, w, id) = escalated().await;
1757 let seat = with_deputy(&w, &id, d.path());
1758 let say = "クローズしていいよ。private repo だから、何回やっても失敗しちゃうから";
1759 w.questions().update(&id, |q| q.say(say)).unwrap();
1760 w.questions()
1761 .update(&id, |q| {
1762 q.settle_by_deputy(&seat, LEAVE_IT, "クローズしていいよ", None)
1763 })
1764 .unwrap();
1765 let q = w.questions().get(&id).unwrap();
1766 assert_eq!(q.answer, Some(Answer::Choice(LEAVE_IT.to_owned())));
1767
1768 let repo = PathBuf::from("/nowhere");
1769 let no = || false;
1770 for t in [3000, 3100] {
1771 w.lap(std::slice::from_ref(&repo), 3600, t, &no).await;
1772 }
1773 let st = w.stored().pop().unwrap();
1774 assert!(st.ignored, "watching stopped");
1775 assert_eq!(st.applied, vec![id.clone()], "applied exactly once");
1776 assert!(
1777 Notices::at(d.path().join("notifications"))
1778 .list()
1779 .is_empty()
1780 );
1781 assert_eq!(fake.reruns.lock().unwrap().len(), 1, "no extra rerun");
1782 assert_eq!(w.questions().list().len(), 1, "no second question");
1783 }
1784
1785 #[tokio::test]
1786 async fn the_task_of_a_release_question_comes_from_the_watch_record() {
1787 let (d, _f, w, id) = escalated().await;
1788 let q = w.questions().get(&id).unwrap();
1789 let queue = crate::queue::Queue::at(d.path().join("queue"));
1790 assert!(
1791 task_of_question(d.path(), &q, &queue).is_none(),
1792 "no run known"
1793 );
1794 assert!(task_of_question(d.path(), &q, &queue).is_none());
1795 let mut st = w.stored().pop().unwrap();
1796 st.run = Some("run-1".into());
1797 w.save("o/r#7", &st);
1798 assert!(task_of_question(d.path(), &q, &queue).is_none());
1800 let mut t = crate::queue::Task::new(
1801 "t".into(),
1802 "i".into(),
1803 PathBuf::from("/nowhere"),
1804 crate::queue::Source::Human,
1805 );
1806 t.runs.push("run-1".into());
1807 queue.put(&mut t).unwrap();
1808 assert_eq!(task_of_question(d.path(), &q, &queue).unwrap().id, t.id);
1809 }
1810
1811 #[tokio::test]
1812 async fn merged_clears_the_notice_the_question_and_the_record() {
1813 let (d, fake, w, _id) = escalated().await;
1814 *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Merged, "h1", vec![]));
1815 let repo = PathBuf::from("/nowhere");
1816 w.lap(&[repo], 3600, 5000, &(|| false)).await;
1817 assert!(
1818 Notices::at(d.path().join("notifications"))
1819 .list()
1820 .is_empty()
1821 );
1822 assert!(w.questions().list().iter().all(|q| !q.status.open()));
1823 assert!(w.stored().is_empty());
1824 }
1825
1826 #[tokio::test]
1827 async fn an_unreadable_forge_changes_nothing() {
1828 let (d, fake, w) = rig();
1829 *fake.snap.lock().unwrap() = None;
1830 w.lap(&[PathBuf::from("/nowhere")], 60, 99_999, &(|| false))
1831 .await;
1832 assert!(fake.reruns.lock().unwrap().is_empty());
1833 assert!(w.questions().list().is_empty());
1834 assert!(!d.path().join("release-watch").exists());
1835 }
1836
1837 #[tokio::test]
1838 async fn no_rerun_when_the_record_cannot_be_saved() {
1839 let (d, fake, w) = rig();
1840 *fake.snap.lock().unwrap() = Some(red());
1841 std::fs::write(d.path().join("release-watch"), "x").unwrap();
1843 w.lap(&[PathBuf::from("/nowhere")], 3600, 1000, &(|| false))
1844 .await;
1845 assert!(fake.reruns.lock().unwrap().is_empty());
1846 }
1847
1848 #[tokio::test]
1849 async fn a_park_stops_the_lap_before_any_call() {
1850 let (_d, fake, w) = rig();
1851 *fake.snap.lock().unwrap() = Some(red());
1852 w.lap(&[PathBuf::from("/nowhere")], 60, 1000, &(|| true))
1853 .await;
1854 assert!(fake.reruns.lock().unwrap().is_empty());
1855 }
1856
1857 fn local_st(asked: Option<&str>, held: Option<&str>) -> WatchState {
1860 WatchState {
1861 asked_head: asked.map(str::to_owned),
1862 held_head: held.map(str::to_owned),
1863 ..WatchState::default()
1864 }
1865 }
1866
1867 #[test]
1868 fn local_decisions_never_wait_on_ci_and_bind_approval_to_the_head() {
1869 use Approved::*;
1870 let none = local_st(None, None);
1871 assert_eq!(decide_local("h1", &none, NoQuestion, true), LocalStep::Ask);
1872 assert_eq!(decide_local("h1", &none, Open, true), LocalStep::Wait);
1873 let asked = local_st(Some("h1"), None);
1874 assert_eq!(
1875 decide_local("h1", &asked, Merge, true),
1876 LocalStep::Merge("h1".to_owned())
1877 );
1878 assert_eq!(
1879 decide_local("h1", &asked, Hold, true),
1880 LocalStep::Hold("h1".to_owned())
1881 );
1882 assert_eq!(decide_local("h2", &asked, Merge, true), LocalStep::Ask);
1884 assert_eq!(decide_local("h2", &asked, Hold, true), LocalStep::Ask);
1885 let held = local_st(Some("h1"), Some("h1"));
1887 assert_eq!(decide_local("h1", &held, NoQuestion, true), LocalStep::Wait);
1888 assert_eq!(decide_local("h2", &held, NoQuestion, true), LocalStep::Ask);
1889 let mut ign = local_st(None, None);
1890 ign.ignored = true;
1891 assert_eq!(decide_local("h1", &ign, NoQuestion, true), LocalStep::Wait);
1892 }
1893
1894 #[test]
1895 fn without_land_approval_an_unasked_head_merges_but_stops_stay_stopped() {
1896 use Approved::*;
1897 let none = local_st(None, None);
1898 assert_eq!(
1899 decide_local("h1", &none, NoQuestion, false),
1900 LocalStep::Merge("h1".to_owned())
1901 );
1902 assert_eq!(decide_local("h1", &none, Open, false), LocalStep::Wait);
1904 let held = local_st(Some("h1"), Some("h1"));
1905 assert_eq!(
1906 decide_local("h1", &held, NoQuestion, false),
1907 LocalStep::Wait
1908 );
1909 assert_eq!(
1910 decide_local("h2", &held, NoQuestion, false),
1911 LocalStep::Merge("h2".to_owned())
1912 );
1913 let mut ign = local_st(None, None);
1914 ign.ignored = true;
1915 assert_eq!(decide_local("h1", &ign, NoQuestion, false), LocalStep::Wait);
1916 let asked = local_st(Some("h1"), None);
1918 assert_eq!(
1919 decide_local("h1", &asked, Hold, false),
1920 LocalStep::Hold("h1".to_owned())
1921 );
1922 }
1923
1924 #[tokio::test]
1925 async fn a_registered_local_pr_still_asks_but_never_merges_while_the_config_is_unreadable() {
1926 let (dir, fake, w) = rig();
1927 *fake.config_fails.lock().unwrap() = true;
1928 *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Open, "h1", vec![]));
1929 let repo = PathBuf::from("/nowhere");
1930 let no = || false;
1931 let pr = pr_key(URL).unwrap();
1932 register(dir.path(), &repo, URL, "run1");
1933 assert!(w.load(&pr).local);
1935 w.lap(std::slice::from_ref(&repo), 60, 1, &no).await;
1936 w.lap(std::slice::from_ref(&repo), 60, 2, &no).await;
1937 assert!(fake.merges.lock().unwrap().is_empty());
1938 let qs = w.questions().list();
1939 assert_eq!(qs.len(), 1);
1940 assert_eq!(qs[0].choices, vec![land::APPROVE, land::HOLD]);
1941 answer(&w, &qs[0].id, land::APPROVE);
1943 w.lap(std::slice::from_ref(&repo), 60, 3, &no).await;
1944 assert!(fake.merges.lock().unwrap().is_empty());
1945 assert_eq!(w.load(&pr).question.as_deref(), Some(qs[0].id.as_str()));
1946 *fake.config_fails.lock().unwrap() = false;
1948 *fake.local.lock().unwrap() = true;
1949 w.lap(std::slice::from_ref(&repo), 60, 4, &no).await;
1950 assert_eq!(*fake.merges.lock().unwrap(), vec!["h1".to_owned()]);
1951 }
1952
1953 #[tokio::test]
1954 async fn a_local_record_written_before_the_flag_existed_is_still_local() {
1955 let (_d, fake, w) = rig();
1956 *fake.config_fails.lock().unwrap() = true;
1957 *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Open, "h1", vec![]));
1958 let repo = PathBuf::from("/nowhere");
1959 let no = || false;
1960 let pr = pr_key(URL).unwrap();
1961 let old = WatchState {
1962 repo: repo.to_string_lossy().into_owned(),
1963 url: URL.to_owned(),
1964 run: Some("run1".to_owned()),
1965 ..WatchState::default()
1966 };
1967 assert!(!old.local);
1968 w.save(&pr, &old);
1969 w.lap(std::slice::from_ref(&repo), 60, 1, &no).await;
1970 assert!(fake.merges.lock().unwrap().is_empty());
1971 let qs = w.questions().list();
1972 assert_eq!(qs.len(), 1);
1973 assert_eq!(qs[0].choices, vec![land::APPROVE, land::HOLD]);
1974 }
1975
1976 #[tokio::test]
1977 async fn an_unregistered_pr_is_an_actions_release_when_the_config_is_unreadable() {
1978 let (_d, fake, w) = rig();
1979 *fake.config_fails.lock().unwrap() = true;
1980 *fake.snap.lock().unwrap() = Some(red());
1981 let repo = PathBuf::from("/nowhere");
1982 let no = || false;
1983 w.lap(std::slice::from_ref(&repo), 3600, 1000, &no).await;
1984 assert_eq!(*fake.reruns.lock().unwrap(), vec!["11".to_owned()]);
1985 }
1986
1987 #[tokio::test]
1988 async fn a_merged_local_pr_waits_with_its_record_while_the_config_is_unreadable() {
1989 let (dir, fake, w) = rig();
1990 *fake.config_fails.lock().unwrap() = true;
1991 *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Merged, "h1", vec![]));
1992 let repo = PathBuf::from("/nowhere");
1993 let no = || false;
1994 let pr = pr_key(URL).unwrap();
1995 register(dir.path(), &repo, URL, "run1");
1996 w.lap(std::slice::from_ref(&repo), 60, 1, &no).await;
1997 assert!(w.state_path(&pr).exists(), "the record is kept");
1998 }
1999
2000 #[tokio::test]
2001 async fn land_approval_off_merges_the_observed_head_without_a_question() {
2002 let (_d, fake, w) = rig();
2003 *fake.local.lock().unwrap() = true;
2004 *fake.approval.lock().unwrap() = Some(false);
2005 *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Open, "h1", vec![]));
2006 let repo = PathBuf::from("/nowhere");
2007 let no = || false;
2008 w.lap(std::slice::from_ref(&repo), 60, 1, &no).await;
2009 assert_eq!(*fake.merges.lock().unwrap(), vec!["h1".to_owned()]);
2010 assert!(w.questions().list().is_empty(), "no approval question");
2011 }
2012
2013 #[tokio::test]
2014 async fn land_approval_on_or_unreadable_still_asks_and_never_merges() {
2015 for approval in [Some(true), None] {
2016 let (_d, fake, w) = rig();
2017 *fake.local.lock().unwrap() = true;
2018 *fake.approval.lock().unwrap() = approval;
2019 *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Open, "h1", vec![]));
2020 let repo = PathBuf::from("/nowhere");
2021 let no = || false;
2022 w.lap(std::slice::from_ref(&repo), 60, 1, &no).await;
2023 w.lap(std::slice::from_ref(&repo), 60, 2, &no).await;
2024 assert!(fake.merges.lock().unwrap().is_empty(), "silence is a hold");
2025 let qs = w.questions().list();
2026 assert_eq!(qs.len(), 1);
2027 assert_eq!(qs[0].choices, vec![land::APPROVE, land::HOLD]);
2028 }
2029 }
2030
2031 #[tokio::test]
2032 async fn land_approval_off_does_not_merge_on_an_unreadable_forge_or_retry_a_refusal() {
2033 let (_d, fake, w) = rig();
2034 *fake.local.lock().unwrap() = true;
2035 *fake.approval.lock().unwrap() = Some(false);
2036 let repo = PathBuf::from("/nowhere");
2037 let no = || false;
2038 *fake.snap.lock().unwrap() = None;
2040 w.lap(std::slice::from_ref(&repo), 60, 1, &no).await;
2041 assert!(fake.merges.lock().unwrap().is_empty());
2042 *fake.merge_fails.lock().unwrap() = true;
2044 *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Open, "h1", vec![]));
2045 w.lap(std::slice::from_ref(&repo), 60, 2, &no).await;
2046 w.lap(std::slice::from_ref(&repo), 60, 3, &no).await;
2047 assert_eq!(fake.merges.lock().unwrap().len(), 1);
2048 assert!(w.questions().list().is_empty());
2049 *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Open, "h2", vec![]));
2051 w.lap(std::slice::from_ref(&repo), 60, 4, &no).await;
2052 assert_eq!(
2053 *fake.merges.lock().unwrap(),
2054 vec!["h1".to_owned(), "h2".to_owned()]
2055 );
2056 }
2057
2058 #[test]
2059 fn pr_info_is_read_from_gh_json() {
2060 let i = parse_info(
2061 r#"{"headRefName":"chore/release-v1.2.3","title":"chore: release v1.2.3","mergeCommit":{"oid":"abc"}}"#,
2062 )
2063 .unwrap();
2064 assert_eq!(i.branch, "chore/release-v1.2.3");
2065 assert_eq!(i.merge_commit.as_deref(), Some("abc"));
2066 let open = parse_info(r#"{"headRefName":"b","title":"t","mergeCommit":null}"#).unwrap();
2067 assert_eq!(open.merge_commit, None);
2068 }
2069
2070 #[tokio::test]
2071 async fn a_local_release_asks_once_without_ci_reruns_or_stall_escalation() {
2072 let (_d, fake, w) = rig();
2073 *fake.local.lock().unwrap() = true;
2074 *fake.snap.lock().unwrap() = Some(red());
2077 let repo = PathBuf::from("/nowhere");
2078 let no = || false;
2079 w.lap(std::slice::from_ref(&repo), 60, 1_000_000, &no).await;
2080 w.lap(std::slice::from_ref(&repo), 60, 2_000_000, &no).await;
2081 assert!(fake.reruns.lock().unwrap().is_empty());
2082 let qs = w.questions().list();
2083 assert_eq!(qs.len(), 1, "one approval question, not one per lap");
2084 assert_eq!(qs[0].choices, vec![land::APPROVE, land::HOLD]);
2085 assert!(fake.merges.lock().unwrap().is_empty(), "silence is a hold");
2086 }
2087
2088 #[tokio::test]
2089 async fn merge_is_pinned_to_the_approved_head_and_a_moved_head_asks_again() {
2090 let (_d, fake, w) = rig();
2091 *fake.local.lock().unwrap() = true;
2092 *fake.snap.lock().unwrap() = Some(open(vec![]));
2093 let repo = PathBuf::from("/nowhere");
2094 let no = || false;
2095 w.lap(std::slice::from_ref(&repo), 60, 1, &no).await;
2096 let id = w.questions().list()[0].id.clone();
2097 *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Open, "h2", vec![]));
2099 answer(&w, &id, land::APPROVE);
2100 w.lap(std::slice::from_ref(&repo), 60, 2, &no).await;
2101 assert!(fake.merges.lock().unwrap().is_empty());
2102 assert_eq!(w.questions().list().len(), 2, "asked again about h2");
2103 let id2 = w
2105 .questions()
2106 .list()
2107 .into_iter()
2108 .find(|q| q.status.open())
2109 .unwrap()
2110 .id;
2111 answer(&w, &id2, land::APPROVE);
2112 w.lap(std::slice::from_ref(&repo), 60, 3, &no).await;
2113 assert_eq!(*fake.merges.lock().unwrap(), vec!["h2".to_owned()]);
2114 }
2115
2116 #[tokio::test]
2117 async fn a_failed_release_holds_with_one_notice_and_one_question_and_never_retries() {
2118 let (d, fake, w) = rig();
2119 *fake.local.lock().unwrap() = true;
2120 *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Merged, "h1", vec![]));
2121 *fake.info.lock().unwrap() = Some(PrInfo {
2122 branch: "chore/release-v1.0.0".to_owned(),
2123 merge_commit: Some("deadbeef".to_owned()),
2124 title: "chore: release v1.0.0".to_owned(),
2125 });
2126 let repo = d.path().join("nowhere");
2128 let no = || false;
2129 w.lap(std::slice::from_ref(&repo), 60, 1, &no).await;
2130 let st = w.load("o/r#7");
2131 let job = st.job.expect("the job is recorded");
2132 assert!(job.failed.is_some() && !job.tag_done && !job.finished);
2133 let qs = w.questions().list();
2134 assert_eq!(qs.len(), 1);
2135 assert_eq!(qs[0].choices, vec![RETRY, LEAVE_IT]);
2136 w.lap(std::slice::from_ref(&repo), 60, 2, &no).await;
2138 assert_eq!(w.questions().list().len(), 1);
2139 let ns = Notices::at(d.path().join("notifications")).list();
2140 assert_eq!(ns.len(), 1);
2141 assert_eq!(
2142 ns[0].link,
2143 Some(Link::Url {
2144 url: URL.to_owned(),
2145 })
2146 );
2147 }
2148
2149 #[tokio::test]
2150 async fn an_escalated_pull_request_records_the_title_version_as_acceptable() {
2151 let (d, fake, w) = rig();
2152 *fake.local.lock().unwrap() = true;
2153 *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Merged, "h1", vec![]));
2154 *fake.info.lock().unwrap() = Some(PrInfo {
2155 branch: "chore/release-v0.1.4".to_owned(),
2156 merge_commit: Some("deadbeef".to_owned()),
2157 title: "chore: release v0.2.0 (minor bump)".to_owned(),
2158 });
2159 let repo = d.path().join("nowhere");
2160 w.lap(std::slice::from_ref(&repo), 60, 1, &(|| false)).await;
2161 let job = w.load("o/r#7").job.expect("the job is recorded");
2162 assert_eq!(job.version, "0.1.4");
2163 assert_eq!(job.accepted, vec!["0.2.0".to_owned()]);
2164 }
2165
2166 #[tokio::test]
2167 async fn a_failed_release_holds_the_task_whose_run_opened_the_pr() {
2168 let (d, fake, w) = rig();
2169 *fake.local.lock().unwrap() = true;
2170 *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Merged, "h1", vec![]));
2171 *fake.info.lock().unwrap() = Some(PrInfo {
2172 branch: "chore/release-v1.0.0".to_owned(),
2173 merge_commit: Some("deadbeef".to_owned()),
2174 title: "t".to_owned(),
2175 });
2176 let repo = d.path().join("nowhere");
2177 let q = crate::queue::Queue::at(d.path().join("queue"));
2178 let mut t = crate::queue::Task::new(
2179 "t".to_owned(),
2180 "i".to_owned(),
2181 repo.clone(),
2182 crate::queue::Source::Human,
2183 );
2184 t.runs.push("run1".to_owned());
2185 t.status = crate::queue::TaskStatus::Done;
2186 q.put(&mut t).unwrap();
2187 register(d.path(), &repo, URL, "run1");
2188 w.lap(std::slice::from_ref(&repo), 60, 1, &(|| false)).await;
2189 let held = q.get(&t.id).unwrap();
2190 assert_eq!(held.status, crate::queue::TaskStatus::Held);
2191 assert!(
2192 held.hold_reason
2193 .unwrap_or_default()
2194 .contains("release v1.0.0")
2195 );
2196 }
2197
2198 #[tokio::test]
2199 async fn unhold_gives_the_task_back_only_when_the_hold_is_ours() {
2200 let (d, _fake, w) = rig();
2201 let q = crate::queue::Queue::at(d.path().join("queue"));
2202 let mut t = crate::queue::Task::new(
2203 "t".to_owned(),
2204 "i".to_owned(),
2205 PathBuf::from("/r"),
2206 crate::queue::Source::Human,
2207 );
2208 t.runs.push("run1".to_owned());
2209 t.status = crate::queue::TaskStatus::Done;
2210 q.put(&mut t).unwrap();
2211 let mut st = WatchState {
2212 run: Some("run1".to_owned()),
2213 ..WatchState::default()
2214 };
2215 w.hold_task(&mut st, "failed");
2216 assert_eq!(q.get(&t.id).unwrap().status, crate::queue::TaskStatus::Held);
2217 w.hold_task(&mut st, "failed again");
2219 assert!(w.unhold_task(&mut st));
2220 assert_eq!(q.get(&t.id).unwrap().status, crate::queue::TaskStatus::Done);
2221 assert!(st.held_task.is_none());
2222
2223 let mut h = q.get(&t.id).unwrap();
2225 st.held_task = Some(h.id.clone());
2226 h.hold_manual(Some("mine".to_owned()));
2227 q.put(&mut h).unwrap();
2228 assert!(w.unhold_task(&mut st));
2229 assert_eq!(q.get(&t.id).unwrap().status, crate::queue::TaskStatus::Held);
2230 }
2231
2232 #[tokio::test]
2233 async fn a_restart_after_the_failure_was_saved_still_holds_the_task() {
2234 let (d, fake, w) = rig();
2235 *fake.local.lock().unwrap() = true;
2236 *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Merged, "h1", vec![]));
2237 let repo = d.path().join("nowhere");
2238 let q = crate::queue::Queue::at(d.path().join("queue"));
2239 let mut t = crate::queue::Task::new(
2240 "t".to_owned(),
2241 "i".to_owned(),
2242 repo.clone(),
2243 crate::queue::Source::Human,
2244 );
2245 t.runs.push("run1".to_owned());
2246 t.status = crate::queue::TaskStatus::Done;
2247 q.put(&mut t).unwrap();
2248 let mut job = Job::new("1.0.0", URL, "deadbeef");
2250 job.failed = Some("command 1 exited 3".to_owned());
2251 let st = WatchState {
2252 repo: repo.to_string_lossy().into_owned(),
2253 url: URL.to_owned(),
2254 run: Some("run1".to_owned()),
2255 job: Some(job),
2256 ..WatchState::default()
2257 };
2258 w.save("o/r#7", &st);
2259 w.lap(std::slice::from_ref(&repo), 60, 1, &(|| false)).await;
2260 assert_eq!(q.get(&t.id).unwrap().status, crate::queue::TaskStatus::Held);
2261 }
2262
2263 #[tokio::test]
2264 async fn a_restart_after_the_job_finished_still_gives_the_task_back() {
2265 let (d, fake, w) = rig();
2266 *fake.local.lock().unwrap() = true;
2267 *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Merged, "h1", vec![]));
2268 let repo = d.path().join("nowhere");
2269 let q = crate::queue::Queue::at(d.path().join("queue"));
2270 let mut t = crate::queue::Task::new(
2271 "t".to_owned(),
2272 "i".to_owned(),
2273 repo.clone(),
2274 crate::queue::Source::Human,
2275 );
2276 t.runs.push("run1".to_owned());
2277 t.hold_machine(Some(format!("{HOLD_PREFIX}release v1.0.0 failed")));
2278 q.put(&mut t).unwrap();
2279 let mut job = Job::new("1.0.0", URL, "deadbeef");
2281 job.finished = true;
2282 let st = WatchState {
2283 repo: repo.to_string_lossy().into_owned(),
2284 url: URL.to_owned(),
2285 run: Some("run1".to_owned()),
2286 job: Some(job),
2287 held_task: Some(t.id.clone()),
2288 ..WatchState::default()
2289 };
2290 w.save("o/r#7", &st);
2291 w.lap(std::slice::from_ref(&repo), 60, 1, &(|| false)).await;
2292 assert_eq!(q.get(&t.id).unwrap().status, crate::queue::TaskStatus::Done);
2293 assert!(!w.state_path("o/r#7").exists());
2294 }
2295
2296 #[tokio::test]
2297 async fn a_pull_request_registered_at_open_is_picked_up_even_if_unlisted() {
2298 let (d, _fake, w) = rig();
2299 register(d.path(), Path::new("/r"), URL, "run1");
2300 let st = w.load("o/r#7");
2301 assert_eq!(st.url, URL);
2302 assert_eq!(st.repo, "/r");
2303 let mut live = st.clone();
2305 live.head = "keep".to_owned();
2306 w.save("o/r#7", &live);
2307 register(d.path(), Path::new("/r"), URL, "run1");
2308 assert_eq!(w.load("o/r#7").head, "keep");
2309 }
2310
2311 #[tokio::test]
2312 async fn a_finished_release_is_kept_on_the_run_and_a_missing_run_still_completes() {
2313 let (d, fake, w) = rig();
2314 *fake.local.lock().unwrap() = true;
2315 *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Merged, "h1", vec![]));
2316 let repo = d.path().join("nowhere");
2317 let mut run = RunState::new(
2318 repo.clone(),
2319 "main".to_owned(),
2320 "0123456789abcdef".to_owned(),
2321 "t".to_owned(),
2322 crate::config::Config::default(),
2323 );
2324 run.id = "run1".to_owned();
2325 run.save_under(d.path()).unwrap();
2326 let mut job = Job::new("1.0.0", URL, "deadbeef");
2327 job.finished = true;
2328 job.log.push(release_local::StepLog {
2329 name: "cmd".to_owned(),
2330 code: Some(0),
2331 tail: "ok".to_owned(),
2332 output: None,
2333 });
2334 let mk = |run: &str| WatchState {
2335 repo: repo.to_string_lossy().into_owned(),
2336 url: URL.to_owned(),
2337 run: Some(run.to_owned()),
2338 job: Some(job.clone()),
2339 ..WatchState::default()
2340 };
2341 w.save("o/r#7", &mk("run1"));
2342 w.lap(std::slice::from_ref(&repo), 60, 1, &(|| false)).await;
2343 assert!(!w.state_path("o/r#7").exists());
2344 let kept = RunState::load_under("run1", d.path()).unwrap();
2345 assert_eq!(kept.release_bump.unwrap().release, Some(job.clone()));
2346
2347 w.save("o/r#7", &mk("gone"));
2349 w.lap(std::slice::from_ref(&repo), 60, 1, &(|| false)).await;
2350 assert!(!w.state_path("o/r#7").exists());
2351 }
2352
2353 #[tokio::test]
2354 async fn raise_released_attaches_url_link() {
2355 let (d, _fake, w) = rig();
2356 let job = Job::new("1.0.0", URL, "deadbeef");
2357 w.raise_released("o/r#7", URL, &job);
2358 let ns = Notices::at(d.path().join("notifications")).list();
2359 assert_eq!(ns.len(), 1);
2360 assert_eq!(ns[0].key, "release-done:o/r#7");
2361 assert_eq!(
2362 ns[0].link,
2363 Some(Link::Url {
2364 url: URL.to_owned(),
2365 })
2366 );
2367 }
2368}