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, 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, message: String) {
697 notices::raise_in(&self.home, Notice::warn(¬ice_key(pr), message));
698 }
699
700 pub(crate) async fn lap(
703 &self,
704 repos: &[PathBuf],
705 stall_secs: i64,
706 now: i64,
707 halt: &(dyn Fn() -> bool + Sync),
708 ) {
709 let stored = self.stored();
710 let mut repos = repos.to_vec();
712 for s in &stored {
713 let p = PathBuf::from(&s.repo);
714 if !s.repo.is_empty() && !repos.contains(&p) {
715 repos.push(p);
716 }
717 }
718 for repo in &repos {
719 if halt() {
720 return;
721 }
722 let mut urls: Vec<String> = match self.forge.list(repo).await {
723 Ok(v) => v.into_iter().map(|(_, u)| u).collect(),
724 Err(e) => {
725 tracing::warn!(
726 "could not list release pull requests in {}: {e:#}",
727 repo.display()
728 );
729 Vec::new()
730 }
731 };
732 let here = repo.to_string_lossy();
735 for s in stored.iter().filter(|s| s.repo == here.as_ref()) {
736 if !urls.contains(&s.url) {
737 urls.push(s.url.clone());
738 }
739 }
740 for url in urls {
741 if halt() {
742 return;
743 }
744 self.watch(repo, &url, stall_secs, now, halt).await;
745 }
746 }
747 }
748
749 pub(crate) async fn watch(
751 &self,
752 repo: &Path,
753 url: &str,
754 stall_secs: i64,
755 now: i64,
756 halt: &(dyn Fn() -> bool + Sync),
757 ) {
758 let Some(pr) = pr_key(url) else {
759 tracing::warn!("not a pull request url: {url}");
760 return;
761 };
762 let mut st = self.load(&pr);
763 st.repo = repo.to_string_lossy().into_owned();
764 st.url = url.to_owned();
765
766 let snap = match self.forge.snapshot(repo, url).await {
767 Ok(s) => Some(s),
768 Err(e) => {
769 tracing::warn!("could not read {url}: {e:#}");
770 None
771 }
772 };
773 match self.forge.config(repo).await {
774 Ok(cfg) if cfg.release.is_local() => {
775 st.local = true;
776 self.watch_local(repo, &pr, st, snap, Some(&cfg)).await;
777 return;
778 }
779 Ok(_) => st.local = false,
780 Err(e) if st.is_local() => {
781 tracing::warn!(
782 "could not read the config of {}: {e:#}; watching {url} as a local release without it",
783 repo.display()
784 );
785 self.watch_local(repo, &pr, st, snap, None).await;
786 return;
787 }
788 Err(e) => tracing::warn!(
789 "could not read the config of {}: {e:#}; watching {url} as an Actions release",
790 repo.display()
791 ),
792 }
793 if decide(snap.as_ref(), &st, now, stall_secs) == Step::Done {
794 self.finish(&pr, &st);
795 return;
796 }
797 let Some(snap) = snap else {
798 return;
799 };
800
801 if let Some(id) = st.question.clone() {
803 match self.questions().get(&id) {
804 Ok(q) if q.status.open() => return,
805 Ok(q) => self.apply_answer(&pr, &mut st, &snap, &q),
806 Err(_) => st.question = None,
807 }
808 observe(&mut st, &snap, now);
810 self.save(&pr, &st);
811 } else {
812 observe(&mut st, &snap, now);
813 }
814
815 match decide(Some(&snap), &st, now, stall_secs) {
816 Step::Done | Step::Unknown | Step::Wait => {}
817 Step::Rerun(runs) => {
818 for r in &runs {
820 st.reruns.insert(r.clone(), now);
821 }
822 if !self.save(&pr, &st) || halt() {
825 return;
826 }
827 let mut ok = false;
828 for r in &runs {
829 match self.forge.rerun(repo, r).await {
830 Ok(()) => ok = true,
831 Err(e) => tracing::warn!("could not rerun run {r} of {url}: {e:#}"),
832 }
833 }
834 if ok {
835 self.raise(&pr, rerun_message(&pr));
836 }
837 }
838 Step::Escalate(why) => {
839 self.raise(&pr, human_message(&pr));
840 let mut q = Question::new(
841 String::new(),
842 crate::bump::NOTICE_NODE.to_owned(),
843 "release-watch".to_owned(),
844 format!("Release PR {pr} is stuck: what now?"),
845 question_detail(&snap, why, &st),
846 vec![RERUN_AGAIN.to_owned(), HOLD.to_owned(), LEAVE_IT.to_owned()],
847 );
848 match self.questions().put(&mut q) {
849 Ok(()) => st.question = Some(q.id.clone()),
850 Err(e) => tracing::warn!("could not file the question for {pr}: {e:#}"),
851 }
852 }
853 }
854 self.save(&pr, &st);
855 }
856
857 async fn watch_local(
859 &self,
860 repo: &Path,
861 pr: &str,
862 mut st: WatchState,
863 snap: Option<RollupView>,
864 cfg: Option<&crate::config::Config>,
865 ) {
866 let Some(snap) = snap else {
867 return;
868 };
869 match snap.state {
870 PrLifecycle::Closed => self.finish(pr, &st),
871 PrLifecycle::Merged => match cfg {
872 Some(cfg) => self.release_merged(repo, pr, st, cfg).await,
873 None => {
875 tracing::warn!("{pr} is merged but its config is unreadable; not released yet")
876 }
877 },
878 PrLifecycle::Open => {
879 let approved = match st.question.clone() {
880 None => Approved::NoQuestion,
881 Some(id) => match self.questions().get(&id) {
882 Ok(q) if q.status.open() => Approved::Open,
883 Ok(q) => match &q.answer {
884 Some(Answer::Choice(c))
885 if land::approval(Some(c)) == land::Approval::Merge =>
886 {
887 Approved::Merge
888 }
889 _ => Approved::Hold,
890 },
891 Err(_) => Approved::NoQuestion,
892 },
893 };
894 let ask_first = if cfg.is_none() {
895 true
896 } else {
897 match self.forge.land_approval(repo).await {
898 Ok(v) => v,
899 Err(e) => {
900 tracing::warn!("could not read graph.land_approval for {pr}: {e:#}");
901 true
902 }
903 }
904 };
905 let step = decide_local(&snap.head, &st, approved, ask_first);
906 if cfg.is_none() && matches!(step, LocalStep::Merge(_)) {
907 tracing::warn!("not merging {pr} while its config is unreadable");
909 return;
910 }
911 if matches!(
912 step,
913 LocalStep::Merge(_) | LocalStep::Hold(_) | LocalStep::Ask
914 ) {
915 st.question = None;
917 }
918 match step {
919 LocalStep::Wait => {}
920 LocalStep::Hold(head) => st.held_head = Some(head),
921 LocalStep::Ask => {
922 st.held_head = None;
923 st.asked_head = Some(snap.head.clone());
924 self.raise(pr, approval_message(pr));
925 let mut q = Question::new(
926 String::new(),
927 crate::bump::NOTICE_NODE.to_owned(),
928 "release-watch".to_owned(),
929 format!("Release PR {pr}: merge it and release?"),
930 format!(
931 "Pull request: {}\nHead: {}\n\n\
932 `[release] mode = \"local\"`: no CI is awaited. `{}` merges \
933 exactly this head, then magi tags the merge commit and runs the \
934 configured release commands. `{}` leaves the pull request \
935 open. Silence is a hold.\n",
936 snap.url,
937 snap.head,
938 land::APPROVE,
939 land::HOLD
940 ),
941 vec![land::APPROVE.to_owned(), land::HOLD.to_owned()],
942 );
943 match self.questions().put(&mut q) {
944 Ok(()) => st.question = Some(q.id.clone()),
945 Err(e) => tracing::warn!("could not file the question for {pr}: {e:#}"),
946 }
947 }
948 LocalStep::Merge(head) => {
949 match self.forge.merge(repo, &snap.url, &head).await {
950 Ok(true) => {
951 self.save(pr, &st);
953 if let Some(cfg) = cfg {
954 self.release_merged(repo, pr, st, cfg).await;
955 }
956 return;
957 }
958 res => {
959 match res {
960 Err(e) => {
961 tracing::warn!("could not merge {}: {e:#}", snap.url)
962 }
963 _ => tracing::warn!("{} did not merge", snap.url),
964 }
965 st.held_head = Some(head);
966 self.raise(
967 pr,
968 format!(
969 "Release PR {pr} could not be merged; merge it by hand"
970 ),
971 );
972 }
973 }
974 }
975 }
976 self.save(pr, &st);
977 }
978 }
979 }
980
981 async fn release_merged(
983 &self,
984 repo: &Path,
985 pr: &str,
986 mut st: WatchState,
987 cfg: &crate::config::Config,
988 ) {
989 if st.ignored {
990 return self.finish(pr, &st);
991 }
992 if st.job.is_none() {
993 let info = match self.forge.info(repo, &st.url).await {
994 Ok(i) => i,
995 Err(e) => {
996 tracing::warn!("could not read the merged {}: {e:#}", st.url);
997 return;
998 }
999 };
1000 let (Some(version), Some(commit)) = (
1001 release_local::version_from_branch(&info.branch),
1002 info.merge_commit,
1003 ) else {
1004 if release_local::version_from_branch(&info.branch).is_none() {
1007 self.finish(pr, &st);
1008 }
1009 return;
1010 };
1011 let mut job = Job::new(&version, &st.url, &commit);
1012 job.accepted = release_local::version_from_title(&info.title)
1015 .into_iter()
1016 .filter(|t| *t != version)
1017 .collect();
1018 st.job = Some(job);
1019 if !self.save(pr, &st) {
1020 return;
1021 }
1022 }
1023 let Some(mut job) = st.job.take() else {
1024 return;
1025 };
1026 if job.finished {
1027 st.job = Some(job);
1030 if !self.record_on_run(&st) {
1031 self.save(pr, &st);
1032 return;
1033 }
1034 return self.complete(pr, &mut st);
1035 }
1036 if job.interrupted() {
1037 job.failed = Some(format!(
1038 "magi stopped while `{}` was running; it may have partly run, so it is not repeated on its own",
1039 job.running.clone().unwrap_or_default()
1040 ));
1041 }
1042 if let Some(why) = job.failed.clone() {
1043 self.hold_task(&mut st, &format!("release {} is on hold: {why}", job.tag()));
1046 let retry = self.settle_job_question(pr, &mut st, &mut job);
1048 if !retry {
1049 st.job = Some(job);
1050 self.save(pr, &st);
1051 return;
1052 }
1053 }
1054 let shell = cfg.shell();
1055 let env = release_local::Env {
1056 repo,
1057 home: &self.home,
1058 key: pr,
1059 remote: &cfg.merge.remote,
1060 shell: &shell,
1061 release: &cfg.release,
1062 };
1063 let result = {
1064 let mut save = |j: &Job| {
1065 let mut copy = st.clone();
1066 copy.job = Some(j.clone());
1067 self.save(pr, ©)
1068 };
1069 release_local::run_job(&env, &mut job, &mut save).await
1070 };
1071 match result {
1072 Ok(()) => {
1073 self.raise_released(pr, &job);
1074 st.job = Some(job);
1075 if self.record_on_run(&st) {
1076 self.complete(pr, &mut st);
1077 } else {
1078 self.save(pr, &st);
1081 }
1082 }
1083 Err(e) => {
1084 job.failed = Some(format!("{e:#}"));
1085 self.hold_task(&mut st, &format!("release {} failed: {e:#}", job.tag()));
1086 self.raise(pr, failed_message(pr));
1087 self.file_job_question(pr, &mut st, &job);
1088 st.job = Some(job);
1089 self.save(pr, &st);
1090 self.record_on_run(&st);
1091 }
1092 }
1093 }
1094
1095 fn record_on_run(&self, st: &WatchState) -> bool {
1100 let (Some(run), Some(job)) = (st.run.as_deref(), st.job.as_ref()) else {
1101 return true;
1102 };
1103 let mut state = match RunState::load_under(run, &self.home) {
1104 Ok(s) => s,
1105 Err(e) => {
1106 tracing::warn!("could not record the release on run {run}: {e:#}");
1107 return true;
1108 }
1109 };
1110 let bump = state.release_bump.get_or_insert_with(Default::default);
1111 if bump.release.as_ref() == Some(job) {
1112 return true;
1113 }
1114 bump.release = Some(job.clone());
1115 state.event(
1116 "release",
1117 format!(
1118 "release {} {}",
1119 job.tag(),
1120 if job.finished {
1121 "finished"
1122 } else if let Some(why) = &job.failed {
1123 why
1124 } else {
1125 "stopped"
1126 }
1127 ),
1128 );
1129 match state.save_under(&self.home) {
1130 Ok(()) => true,
1131 Err(e) => {
1132 tracing::warn!("could not save the release on run {run}: {e:#}");
1133 false
1134 }
1135 }
1136 }
1137
1138 fn hold_task(&self, st: &mut WatchState, reason: &str) {
1142 if st.held_task.is_some() {
1143 return;
1144 }
1145 let Some(run) = st.run.clone() else {
1146 return;
1147 };
1148 let q = crate::queue::Queue::at(self.home.join("queue"));
1149 for mut t in q.list() {
1150 if t.runs.contains(&run) {
1151 if t.status != crate::queue::TaskStatus::Held {
1152 t.hold_machine(Some(format!("{HOLD_PREFIX}{reason}")));
1153 match q.put(&mut t) {
1154 Ok(()) => st.held_task = Some(t.id.clone()),
1155 Err(e) => tracing::warn!("could not hold task {} for {run}: {e:#}", t.id),
1156 }
1157 }
1158 return;
1159 }
1160 }
1161 }
1162
1163 #[must_use]
1170 fn unhold_task(&self, st: &mut WatchState) -> bool {
1171 let Some(id) = st.held_task.take() else {
1172 return true;
1173 };
1174 let q = crate::queue::Queue::at(self.home.join("queue"));
1175 let mut t = match q.get(&id) {
1176 Ok(t) => t,
1177 Err(_) if matches!(q.path_of(&id).try_exists(), Ok(false)) => return true,
1180 Err(e) => {
1181 tracing::warn!("could not read task {id} to restore it after the release: {e:#}");
1182 st.held_task = Some(id);
1183 return false;
1184 }
1185 };
1186 let ours = t.status == crate::queue::TaskStatus::Held
1187 && t.hold_source == Some(crate::queue::HoldSource::Machine)
1188 && t.hold_reason
1189 .as_deref()
1190 .is_some_and(|r| r.starts_with(HOLD_PREFIX));
1191 if ours {
1192 t.succeed();
1193 if let Err(e) = q.put(&mut t) {
1194 tracing::warn!("could not restore task {id} after the release: {e:#}");
1195 st.held_task = Some(id);
1196 return false;
1197 }
1198 }
1199 true
1200 }
1201
1202 fn complete(&self, pr: &str, st: &mut WatchState) {
1206 if self.unhold_task(st) {
1207 self.finish(pr, st);
1208 } else {
1209 self.save(pr, st);
1210 }
1211 }
1212
1213 fn raise_released(&self, pr: &str, job: &Job) {
1214 notices::raise_in(
1215 &self.home,
1216 Notice::info(&released_key(pr), format!("Released {} ({pr})", job.tag())),
1217 );
1218 }
1219
1220 fn file_job_question(&self, pr: &str, st: &mut WatchState, job: &Job) {
1222 let last = job
1223 .log
1224 .last()
1225 .map(|l| format!("\nLast step: {}\n\n{}\n", l.name, l.tail))
1226 .unwrap_or_default();
1227 let mut q = Question::new(
1228 String::new(),
1229 crate::bump::NOTICE_NODE.to_owned(),
1230 "release-watch".to_owned(),
1231 format!("Release {} of {pr} is on hold: retry?", job.tag()),
1232 format!(
1233 "Pull request: {}\nWhy it stopped: {}\n{last}\n\
1234 Full output is under the magi home's release-local directory.\n\n\
1235 - `{RETRY}`: run it again from where it stopped (the tag is not \
1236 recreated and finished commands are skipped).\n\
1237 - `{LEAVE_IT}`: stop watching; release by hand.\n\n\
1238 Nothing is retried on its own; silence is a hold.\n",
1239 job.pr_url,
1240 job.failed.clone().unwrap_or_default()
1241 ),
1242 vec![RETRY.to_owned(), LEAVE_IT.to_owned()],
1243 );
1244 match self.questions().put(&mut q) {
1245 Ok(()) => st.question = Some(q.id.clone()),
1246 Err(e) => tracing::warn!("could not file the question for {pr}: {e:#}"),
1247 }
1248 }
1249
1250 fn settle_job_question(&self, pr: &str, st: &mut WatchState, job: &mut Job) -> bool {
1254 let why = job.failed.clone().unwrap_or_default();
1255 if let Some(id) = st.question.clone() {
1256 match self.questions().get(&id) {
1257 Ok(q) if q.status.open() => return false,
1258 Ok(q) => {
1259 st.question = None;
1260 match &q.answer {
1261 Some(Answer::Choice(c)) if c == RETRY => {
1262 job.resume();
1263 st.held = None;
1264 return true;
1265 }
1266 Some(Answer::Choice(c)) if c == LEAVE_IT => {
1267 st.ignored = true;
1268 let _ = Notices::at(self.home.join("notifications"))
1269 .dismiss(¬ices::id_of(¬ice_key(pr)));
1270 }
1271 _ => st.held = Some(why),
1273 }
1274 return false;
1275 }
1276 Err(_) => st.question = None,
1277 }
1278 }
1279 if !st.ignored && st.held.as_deref() != Some(why.as_str()) {
1280 self.raise(pr, failed_message(pr));
1281 self.file_job_question(pr, st, job);
1282 }
1283 false
1284 }
1285
1286 fn apply_answer(&self, pr: &str, st: &mut WatchState, snap: &RollupView, q: &Question) {
1288 st.question = None;
1289 if st.applied.contains(&q.id) {
1290 return;
1291 }
1292 st.applied.push(q.id.clone());
1293 if st.applied.len() > 20 {
1294 st.applied.remove(0);
1295 }
1296 match &q.answer {
1297 Some(Answer::Choice(c)) if c == RERUN_AGAIN => {
1298 st.held = None;
1299 for c in snap.checks.iter().filter(|c| is_failed(c.verdict)) {
1300 if let Some(r) = &c.run {
1301 st.reruns.remove(r);
1302 }
1303 }
1304 }
1305 Some(Answer::Choice(c)) if c == LEAVE_IT => {
1306 st.ignored = true;
1307 if let Err(e) = Notices::at(self.home.join("notifications"))
1308 .dismiss(¬ices::id_of(¬ice_key(pr)))
1309 {
1310 tracing::warn!("could not dismiss the notice for {pr}: {e:#}");
1311 }
1312 }
1313 _ => st.held = Some(fingerprint(snap)),
1315 }
1316 }
1317
1318 fn finish(&self, pr: &str, st: &WatchState) {
1320 let _ =
1321 Notices::at(self.home.join("notifications")).dismiss(¬ices::id_of(¬ice_key(pr)));
1322 if let Some(id) = &st.question {
1323 let _ = self.questions().update(id, |q| {
1324 q.abandon("the release pull request is no longer open");
1325 Ok(())
1326 });
1327 }
1328 let _ = std::fs::remove_file(self.state_path(pr));
1329 }
1330}
1331
1332pub(crate) async fn run(
1336 watcher: Watcher,
1337 settings: impl Fn() -> (Vec<PathBuf>, u64),
1338 stop: crate::daemon::Stop,
1339) {
1340 let halt = {
1341 let stop = stop.clone();
1342 move || stop.stopped()
1343 };
1344 while !stop.stopped() {
1345 let (repos, minutes) = settings();
1346 if minutes > 0 {
1347 let now = jiff::Timestamp::now().as_second();
1348 watcher
1349 .lap(&repos, (minutes as i64).saturating_mul(60), now, &halt)
1350 .await;
1351 }
1352 let mut slept = Duration::ZERO;
1353 while slept < LAP && !stop.stopped() {
1354 tokio::time::sleep(Duration::from_secs(1)).await;
1355 slept += Duration::from_secs(1);
1356 }
1357 }
1358}
1359
1360#[cfg(test)]
1361mod tests {
1362 use super::*;
1363 use crate::land::CheckView;
1364 use anyhow::Context;
1365 use std::sync::Mutex;
1366
1367 const URL: &str = "https://github.com/o/r/pull/7";
1368
1369 fn check(name: &str, v: Verdict, run: Option<&str>) -> CheckView {
1370 CheckView {
1371 name: name.to_owned(),
1372 verdict: v,
1373 run: run.map(str::to_owned),
1374 url: run.map(|r| format!("https://github.com/o/r/actions/runs/{r}/job/1")),
1375 }
1376 }
1377
1378 fn snap(state: PrLifecycle, head: &str, checks: Vec<CheckView>) -> RollupView {
1379 RollupView {
1380 url: URL.to_owned(),
1381 number: 7,
1382 state,
1383 head: head.to_owned(),
1384 checks,
1385 }
1386 }
1387
1388 fn open(checks: Vec<CheckView>) -> RollupView {
1389 snap(PrLifecycle::Open, "h1", checks)
1390 }
1391
1392 fn st_at(progress: i64) -> WatchState {
1393 WatchState {
1394 progress_at: progress,
1395 ..WatchState::default()
1396 }
1397 }
1398
1399 #[test]
1400 fn unreadable_is_unknown_and_merged_or_closed_is_done() {
1401 assert_eq!(decide(None, &st_at(0), 10, 60), Step::Unknown);
1402 for s in [PrLifecycle::Merged, PrLifecycle::Closed] {
1403 assert_eq!(
1404 decide(Some(&snap(s, "h", vec![])), &st_at(0), 10, 60),
1405 Step::Done
1406 );
1407 }
1408 }
1409
1410 #[test]
1411 fn a_failed_run_is_rerun_even_while_another_run_is_pending() {
1412 let s = open(vec![
1413 check("win", Verdict::Fail, Some("11")),
1414 check("lint", Verdict::Pending, Some("12")),
1415 ]);
1416 assert_eq!(
1417 decide(Some(&s), &st_at(0), 1, 3600),
1418 Step::Rerun(vec!["11".into()])
1419 );
1420 }
1421
1422 #[test]
1423 fn a_run_with_a_job_still_pending_is_not_complete() {
1424 let s = open(vec![
1425 check("win", Verdict::Fail, Some("11")),
1426 check("mac", Verdict::Pending, Some("11")),
1427 ]);
1428 assert_eq!(decide(Some(&s), &st_at(0), 1, 3600), Step::Wait);
1429 }
1430
1431 #[test]
1432 fn one_rerun_per_run_then_grace_then_escalation() {
1433 let s = open(vec![check("win", Verdict::Fail, Some("11"))]);
1434 let mut st = st_at(0);
1435 st.reruns.insert("11".into(), 100);
1436 assert_eq!(decide(Some(&s), &st, 100 + RERUN_GRACE - 1, 0), Step::Wait);
1437 assert_eq!(
1438 decide(Some(&s), &st, 100 + RERUN_GRACE, 0),
1439 Step::Escalate(Why::StillRed)
1440 );
1441 }
1442
1443 #[test]
1444 fn a_failure_with_no_run_id_goes_straight_to_a_human_once_settled() {
1445 let s = open(vec![check("ci/ext", Verdict::Fail, None)]);
1446 assert_eq!(
1447 decide(Some(&s), &st_at(0), 1, 0),
1448 Step::Escalate(Why::StillRed)
1449 );
1450 let s = open(vec![
1451 check("ci/ext", Verdict::Fail, None),
1452 check("x", Verdict::Pending, Some("5")),
1453 ]);
1454 assert_eq!(decide(Some(&s), &st_at(0), 1, 0), Step::Wait);
1455 }
1456
1457 #[test]
1458 fn no_progress_for_the_bounded_time_escalates_and_zero_disables_it() {
1459 let s = open(vec![check("a", Verdict::Pass, Some("1"))]);
1460 assert_eq!(decide(Some(&s), &st_at(0), 3599, 3600), Step::Wait);
1461 assert_eq!(
1462 decide(Some(&s), &st_at(0), 3600, 3600),
1463 Step::Escalate(Why::Stalled)
1464 );
1465 assert_eq!(decide(Some(&s), &st_at(0), 99_999, 0), Step::Wait);
1466 }
1467
1468 #[test]
1469 fn held_and_ignored_stay_quiet_until_the_fingerprint_moves() {
1470 let s = open(vec![check("a", Verdict::Fail, None)]);
1471 let mut st = st_at(0);
1472 st.held = Some(fingerprint(&s));
1473 assert_eq!(decide(Some(&s), &st, 99_999, 60), Step::Wait);
1474 let moved = open(vec![
1475 check("a", Verdict::Pass, None),
1476 check("b", Verdict::Fail, None),
1477 ]);
1478 assert_eq!(
1479 decide(Some(&moved), &st, 99_999, 60),
1480 Step::Escalate(Why::StillRed)
1481 );
1482 st.ignored = true;
1483 assert_eq!(decide(Some(&moved), &st, 99_999, 60), Step::Wait);
1484 }
1485
1486 #[test]
1487 fn observe_restarts_the_clock_only_on_change_and_forgets_reruns_on_a_new_head() {
1488 let mut st = st_at(0);
1489 let a = open(vec![check("a", Verdict::Pending, Some("1"))]);
1490 observe(&mut st, &a, 10);
1491 assert_eq!(st.progress_at, 10);
1492 observe(&mut st, &a, 50);
1493 assert_eq!(st.progress_at, 10);
1494 st.reruns.insert("1".into(), 5);
1495 observe(
1496 &mut st,
1497 &snap(PrLifecycle::Open, "h2", a.checks.clone()),
1498 60,
1499 );
1500 assert!(st.reruns.is_empty());
1501 assert_eq!(st.progress_at, 60);
1502 }
1503
1504 #[test]
1505 fn pr_key_names_owner_repo_and_number() {
1506 assert_eq!(pr_key(URL).as_deref(), Some("o/r#7"));
1507 assert_eq!(pr_key("https://github.com/o/r/issues/7"), None);
1508 assert_eq!(pr_key("nonsense"), None);
1509 }
1510
1511 #[test]
1512 fn notice_wording_does_not_vary_between_polls() {
1513 assert_eq!(rerun_message("o/r#7"), rerun_message("o/r#7"));
1514 assert!(
1515 !human_message("o/r#7")
1516 .chars()
1517 .any(|c| c.is_ascii_digit() && c != '7')
1518 );
1519 }
1520
1521 #[test]
1522 fn the_release_question_is_not_claimed_by_other_machinery() {
1523 let q = Question::new(
1524 String::new(),
1525 crate::bump::NOTICE_NODE.into(),
1526 "release-watch".into(),
1527 "s".into(),
1528 String::new(),
1529 vec![HOLD.into()],
1530 );
1531 assert_eq!(
1533 crate::deputy::kind_of(&q),
1534 Some(crate::deputy::Kind::Release)
1535 );
1536 let n = Notice::warn("release-pr:o/r#7", "m").about([String::new()]);
1537 assert!(!notices::covers(&q, &n));
1538 let mut plain = q.clone();
1540 plain.seat = "bump".into();
1541 assert_eq!(crate::deputy::kind_of(&plain), None);
1542 }
1543
1544 #[derive(Default)]
1545 struct Fake {
1546 snap: Mutex<Option<RollupView>>,
1547 reruns: Mutex<Vec<String>>,
1548 local: Mutex<bool>,
1549 info: Mutex<Option<PrInfo>>,
1550 merges: Mutex<Vec<String>>,
1551 approval: Mutex<Option<bool>>,
1553 merge_fails: Mutex<bool>,
1554 config_fails: Mutex<bool>,
1556 }
1557
1558 impl ReleaseForge for std::sync::Arc<Fake> {
1559 fn list<'a>(&'a self, _: &'a Path) -> Fut<'a, Result<Vec<(String, String)>>> {
1560 Box::pin(async { Ok(vec![("chore/release-v1.0.0".to_owned(), URL.to_owned())]) })
1561 }
1562 fn snapshot<'a>(&'a self, _: &'a Path, _: &'a str) -> Fut<'a, Result<RollupView>> {
1563 let s = self.snap.lock().unwrap().clone();
1564 Box::pin(async move { s.context("unreadable") })
1565 }
1566 fn rerun<'a>(&'a self, _: &'a Path, run: &'a str) -> Fut<'a, Result<()>> {
1567 self.reruns.lock().unwrap().push(run.to_owned());
1568 Box::pin(async { Ok(()) })
1569 }
1570 fn config<'a>(&'a self, _: &'a Path) -> Fut<'a, Result<crate::config::Config>> {
1571 if *self.config_fails.lock().unwrap() {
1572 return Box::pin(async { bail!("unreadable config") });
1573 }
1574 let mut c = crate::config::Config::default();
1575 if *self.local.lock().unwrap() {
1576 c.release.mode = crate::config::ReleaseMode::Local;
1577 c.release.commands = vec!["true".to_owned()];
1578 }
1579 Box::pin(async move { Ok(c) })
1580 }
1581 fn info<'a>(&'a self, _: &'a Path, _: &'a str) -> Fut<'a, Result<PrInfo>> {
1582 let i = self.info.lock().unwrap().clone();
1583 Box::pin(async move { i.context("no info") })
1584 }
1585 fn merge<'a>(&'a self, _: &'a Path, _: &'a str, head: &'a str) -> Fut<'a, Result<bool>> {
1586 self.merges.lock().unwrap().push(head.to_owned());
1587 let fails = *self.merge_fails.lock().unwrap();
1588 Box::pin(async move { if fails { bail!("refused") } else { Ok(true) } })
1589 }
1590 fn land_approval<'a>(&'a self, _: &'a Path) -> Fut<'a, Result<bool>> {
1591 let a = *self.approval.lock().unwrap();
1592 Box::pin(async move { a.context("unreadable config") })
1593 }
1594 }
1595
1596 fn rig() -> (tempfile::TempDir, std::sync::Arc<Fake>, Watcher) {
1597 let dir = tempfile::tempdir().unwrap();
1598 let fake = std::sync::Arc::new(Fake::default());
1599 *fake.approval.lock().unwrap() = Some(true);
1601 let w = Watcher::new(Box::new(fake.clone()), dir.path().to_path_buf());
1602 (dir, fake, w)
1603 }
1604
1605 fn red() -> RollupView {
1606 open(vec![check("win", Verdict::Fail, Some("11"))])
1607 }
1608
1609 #[tokio::test]
1610 async fn reruns_once_survives_a_restart_then_escalates_with_one_question() {
1611 let (dir, fake, w) = rig();
1612 *fake.snap.lock().unwrap() = Some(red());
1613 let repo = PathBuf::from("/nowhere");
1614 let no = || false;
1615 w.lap(std::slice::from_ref(&repo), 3600, 1000, &no).await;
1616 assert_eq!(*fake.reruns.lock().unwrap(), vec!["11".to_owned()]);
1617
1618 let w2 = Watcher::new(Box::new(fake.clone()), dir.path().to_path_buf());
1620 w2.lap(
1621 std::slice::from_ref(&repo),
1622 3600,
1623 1000 + RERUN_GRACE - 1,
1624 &no,
1625 )
1626 .await;
1627 assert_eq!(fake.reruns.lock().unwrap().len(), 1);
1628 assert!(w2.questions().list().is_empty());
1629
1630 w2.lap(
1631 std::slice::from_ref(&repo),
1632 3600,
1633 1000 + RERUN_GRACE + 1,
1634 &no,
1635 )
1636 .await;
1637 w2.lap(
1638 std::slice::from_ref(&repo),
1639 3600,
1640 1000 + RERUN_GRACE + 400,
1641 &no,
1642 )
1643 .await;
1644 assert_eq!(fake.reruns.lock().unwrap().len(), 1);
1645 let qs = w2.questions().list();
1646 assert_eq!(qs.len(), 1, "asked once, not every lap");
1647 assert_eq!(qs[0].node, crate::bump::NOTICE_NODE);
1648 assert!(qs[0].detail.contains("win") && qs[0].detail.contains(URL));
1649 assert_eq!(qs[0].choices, vec![RERUN_AGAIN, HOLD, LEAVE_IT]);
1650 let ns = Notices::at(dir.path().join("notifications")).list();
1651 assert_eq!(ns.len(), 1);
1652 assert_eq!(ns[0].key, "release-pr:o/r#7");
1653 }
1654
1655 async fn escalated() -> (tempfile::TempDir, std::sync::Arc<Fake>, Watcher, String) {
1656 let (dir, fake, w) = rig();
1657 *fake.snap.lock().unwrap() = Some(red());
1658 let repo = PathBuf::from("/nowhere");
1659 let no = || false;
1660 w.lap(std::slice::from_ref(&repo), 3600, 1000, &no).await;
1661 w.lap(&[repo], 3600, 2000, &no).await;
1662 let id = w.questions().list()[0].id.clone();
1663 (dir, fake, w, id)
1664 }
1665
1666 fn answer(w: &Watcher, id: &str, choice: &str) {
1667 w.questions()
1668 .update(id, |q| q.answer(Answer::Choice(choice.to_owned())))
1669 .unwrap();
1670 }
1671
1672 #[tokio::test]
1673 async fn rerun_again_reruns_exactly_once_more() {
1674 let (_d, fake, w, id) = escalated().await;
1675 answer(&w, &id, RERUN_AGAIN);
1676 let repo = PathBuf::from("/nowhere");
1677 let no = || false;
1678 w.lap(std::slice::from_ref(&repo), 3600, 3000, &no).await;
1679 w.lap(&[repo], 3600, 3001, &no).await;
1680 assert_eq!(fake.reruns.lock().unwrap().len(), 2);
1681 }
1682
1683 #[tokio::test]
1684 async fn hold_and_silence_do_not_reask_and_leave_it_dismisses() {
1685 let (d, fake, w, id) = escalated().await;
1686 answer(&w, &id, HOLD);
1687 let repo = PathBuf::from("/nowhere");
1688 let no = || false;
1689 for t in [3000, 90_000, 200_000] {
1690 w.lap(std::slice::from_ref(&repo), 3600, t, &no).await;
1691 }
1692 assert_eq!(w.questions().list().len(), 1, "held: no second question");
1693 assert_eq!(fake.reruns.lock().unwrap().len(), 1);
1694
1695 let (d2, _f2, w2, id2) = escalated().await;
1696 answer(&w2, &id2, LEAVE_IT);
1697 w2.lap(std::slice::from_ref(&repo), 3600, 3000, &no).await;
1698 let ns = Notices::at(d2.path().join("notifications")).list();
1699 assert!(ns.is_empty(), "dismissed notices are hidden");
1700 drop(d);
1701 }
1702
1703 fn with_deputy(w: &Watcher, id: &str, home: &Path) -> String {
1705 let seat = crate::agent::SeatState::new(&crate::ask::deputy_seat_key(id), "a", 1);
1706 let key = seat.key.clone();
1707 let brief = deputy_brief(&w.questions().get(id).unwrap(), home);
1708 w.questions()
1709 .update(id, |q| {
1710 let mut d = crate::ask::Deputy::new(brief);
1711 d.seat = Some(seat);
1712 q.deputy = Some(d);
1713 Ok(())
1714 })
1715 .unwrap();
1716 key
1717 }
1718
1719 #[tokio::test]
1720 async fn the_brief_names_the_pull_request_and_what_leave_it_really_does() {
1721 let (d, _f, w, id) = escalated().await;
1722 let q = w.questions().get(&id).unwrap();
1723 let b = deputy_brief(&q, d.path());
1724 assert!(b.contains(URL), "{b}");
1725 assert!(
1726 b.contains("`leave it`") && b.contains("does NOT close"),
1727 "{b}"
1728 );
1729 assert!(b.contains("`rerun again`") && b.contains("`hold`"), "{b}");
1730 assert!(!b.contains("`merge`"), "no merge on an escalation: {b}");
1731
1732 let empty = tempfile::tempdir().unwrap();
1734 let b = deputy_brief(&q, empty.path());
1735 assert!(b.contains("could not be found") && !b.contains(URL), "{b}");
1736 }
1737
1738 #[tokio::test]
1739 async fn a_settled_leave_it_is_applied_once_by_the_watcher_and_closes_nothing() {
1740 let (d, fake, w, id) = escalated().await;
1741 let seat = with_deputy(&w, &id, d.path());
1742 let say = "クローズしていいよ。private repo だから、何回やっても失敗しちゃうから";
1743 w.questions().update(&id, |q| q.say(say)).unwrap();
1744 w.questions()
1745 .update(&id, |q| {
1746 q.settle_by_deputy(&seat, LEAVE_IT, "クローズしていいよ", None)
1747 })
1748 .unwrap();
1749 let q = w.questions().get(&id).unwrap();
1750 assert_eq!(q.answer, Some(Answer::Choice(LEAVE_IT.to_owned())));
1751
1752 let repo = PathBuf::from("/nowhere");
1753 let no = || false;
1754 for t in [3000, 3100] {
1755 w.lap(std::slice::from_ref(&repo), 3600, t, &no).await;
1756 }
1757 let st = w.stored().pop().unwrap();
1758 assert!(st.ignored, "watching stopped");
1759 assert_eq!(st.applied, vec![id.clone()], "applied exactly once");
1760 assert!(
1761 Notices::at(d.path().join("notifications"))
1762 .list()
1763 .is_empty()
1764 );
1765 assert_eq!(fake.reruns.lock().unwrap().len(), 1, "no extra rerun");
1766 assert_eq!(w.questions().list().len(), 1, "no second question");
1767 }
1768
1769 #[tokio::test]
1770 async fn the_task_of_a_release_question_comes_from_the_watch_record() {
1771 let (d, _f, w, id) = escalated().await;
1772 let q = w.questions().get(&id).unwrap();
1773 let queue = crate::queue::Queue::at(d.path().join("queue"));
1774 assert!(
1775 task_of_question(d.path(), &q, &queue).is_none(),
1776 "no run known"
1777 );
1778 assert!(task_of_question(d.path(), &q, &queue).is_none());
1779 let mut st = w.stored().pop().unwrap();
1780 st.run = Some("run-1".into());
1781 w.save("o/r#7", &st);
1782 assert!(task_of_question(d.path(), &q, &queue).is_none());
1784 let mut t = crate::queue::Task::new(
1785 "t".into(),
1786 "i".into(),
1787 PathBuf::from("/nowhere"),
1788 crate::queue::Source::Human,
1789 );
1790 t.runs.push("run-1".into());
1791 queue.put(&mut t).unwrap();
1792 assert_eq!(task_of_question(d.path(), &q, &queue).unwrap().id, t.id);
1793 }
1794
1795 #[tokio::test]
1796 async fn merged_clears_the_notice_the_question_and_the_record() {
1797 let (d, fake, w, _id) = escalated().await;
1798 *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Merged, "h1", vec![]));
1799 let repo = PathBuf::from("/nowhere");
1800 w.lap(&[repo], 3600, 5000, &(|| false)).await;
1801 assert!(
1802 Notices::at(d.path().join("notifications"))
1803 .list()
1804 .is_empty()
1805 );
1806 assert!(w.questions().list().iter().all(|q| !q.status.open()));
1807 assert!(w.stored().is_empty());
1808 }
1809
1810 #[tokio::test]
1811 async fn an_unreadable_forge_changes_nothing() {
1812 let (d, fake, w) = rig();
1813 *fake.snap.lock().unwrap() = None;
1814 w.lap(&[PathBuf::from("/nowhere")], 60, 99_999, &(|| false))
1815 .await;
1816 assert!(fake.reruns.lock().unwrap().is_empty());
1817 assert!(w.questions().list().is_empty());
1818 assert!(!d.path().join("release-watch").exists());
1819 }
1820
1821 #[tokio::test]
1822 async fn no_rerun_when_the_record_cannot_be_saved() {
1823 let (d, fake, w) = rig();
1824 *fake.snap.lock().unwrap() = Some(red());
1825 std::fs::write(d.path().join("release-watch"), "x").unwrap();
1827 w.lap(&[PathBuf::from("/nowhere")], 3600, 1000, &(|| false))
1828 .await;
1829 assert!(fake.reruns.lock().unwrap().is_empty());
1830 }
1831
1832 #[tokio::test]
1833 async fn a_park_stops_the_lap_before_any_call() {
1834 let (_d, fake, w) = rig();
1835 *fake.snap.lock().unwrap() = Some(red());
1836 w.lap(&[PathBuf::from("/nowhere")], 60, 1000, &(|| true))
1837 .await;
1838 assert!(fake.reruns.lock().unwrap().is_empty());
1839 }
1840
1841 fn local_st(asked: Option<&str>, held: Option<&str>) -> WatchState {
1844 WatchState {
1845 asked_head: asked.map(str::to_owned),
1846 held_head: held.map(str::to_owned),
1847 ..WatchState::default()
1848 }
1849 }
1850
1851 #[test]
1852 fn local_decisions_never_wait_on_ci_and_bind_approval_to_the_head() {
1853 use Approved::*;
1854 let none = local_st(None, None);
1855 assert_eq!(decide_local("h1", &none, NoQuestion, true), LocalStep::Ask);
1856 assert_eq!(decide_local("h1", &none, Open, true), LocalStep::Wait);
1857 let asked = local_st(Some("h1"), None);
1858 assert_eq!(
1859 decide_local("h1", &asked, Merge, true),
1860 LocalStep::Merge("h1".to_owned())
1861 );
1862 assert_eq!(
1863 decide_local("h1", &asked, Hold, true),
1864 LocalStep::Hold("h1".to_owned())
1865 );
1866 assert_eq!(decide_local("h2", &asked, Merge, true), LocalStep::Ask);
1868 assert_eq!(decide_local("h2", &asked, Hold, true), LocalStep::Ask);
1869 let held = local_st(Some("h1"), Some("h1"));
1871 assert_eq!(decide_local("h1", &held, NoQuestion, true), LocalStep::Wait);
1872 assert_eq!(decide_local("h2", &held, NoQuestion, true), LocalStep::Ask);
1873 let mut ign = local_st(None, None);
1874 ign.ignored = true;
1875 assert_eq!(decide_local("h1", &ign, NoQuestion, true), LocalStep::Wait);
1876 }
1877
1878 #[test]
1879 fn without_land_approval_an_unasked_head_merges_but_stops_stay_stopped() {
1880 use Approved::*;
1881 let none = local_st(None, None);
1882 assert_eq!(
1883 decide_local("h1", &none, NoQuestion, false),
1884 LocalStep::Merge("h1".to_owned())
1885 );
1886 assert_eq!(decide_local("h1", &none, Open, false), LocalStep::Wait);
1888 let held = local_st(Some("h1"), Some("h1"));
1889 assert_eq!(
1890 decide_local("h1", &held, NoQuestion, false),
1891 LocalStep::Wait
1892 );
1893 assert_eq!(
1894 decide_local("h2", &held, NoQuestion, false),
1895 LocalStep::Merge("h2".to_owned())
1896 );
1897 let mut ign = local_st(None, None);
1898 ign.ignored = true;
1899 assert_eq!(decide_local("h1", &ign, NoQuestion, false), LocalStep::Wait);
1900 let asked = local_st(Some("h1"), None);
1902 assert_eq!(
1903 decide_local("h1", &asked, Hold, false),
1904 LocalStep::Hold("h1".to_owned())
1905 );
1906 }
1907
1908 #[tokio::test]
1909 async fn a_registered_local_pr_still_asks_but_never_merges_while_the_config_is_unreadable() {
1910 let (dir, fake, w) = rig();
1911 *fake.config_fails.lock().unwrap() = true;
1912 *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Open, "h1", vec![]));
1913 let repo = PathBuf::from("/nowhere");
1914 let no = || false;
1915 let pr = pr_key(URL).unwrap();
1916 register(dir.path(), &repo, URL, "run1");
1917 assert!(w.load(&pr).local);
1919 w.lap(std::slice::from_ref(&repo), 60, 1, &no).await;
1920 w.lap(std::slice::from_ref(&repo), 60, 2, &no).await;
1921 assert!(fake.merges.lock().unwrap().is_empty());
1922 let qs = w.questions().list();
1923 assert_eq!(qs.len(), 1);
1924 assert_eq!(qs[0].choices, vec![land::APPROVE, land::HOLD]);
1925 answer(&w, &qs[0].id, land::APPROVE);
1927 w.lap(std::slice::from_ref(&repo), 60, 3, &no).await;
1928 assert!(fake.merges.lock().unwrap().is_empty());
1929 assert_eq!(w.load(&pr).question.as_deref(), Some(qs[0].id.as_str()));
1930 *fake.config_fails.lock().unwrap() = false;
1932 *fake.local.lock().unwrap() = true;
1933 w.lap(std::slice::from_ref(&repo), 60, 4, &no).await;
1934 assert_eq!(*fake.merges.lock().unwrap(), vec!["h1".to_owned()]);
1935 }
1936
1937 #[tokio::test]
1938 async fn a_local_record_written_before_the_flag_existed_is_still_local() {
1939 let (_d, fake, w) = rig();
1940 *fake.config_fails.lock().unwrap() = true;
1941 *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Open, "h1", vec![]));
1942 let repo = PathBuf::from("/nowhere");
1943 let no = || false;
1944 let pr = pr_key(URL).unwrap();
1945 let old = WatchState {
1946 repo: repo.to_string_lossy().into_owned(),
1947 url: URL.to_owned(),
1948 run: Some("run1".to_owned()),
1949 ..WatchState::default()
1950 };
1951 assert!(!old.local);
1952 w.save(&pr, &old);
1953 w.lap(std::slice::from_ref(&repo), 60, 1, &no).await;
1954 assert!(fake.merges.lock().unwrap().is_empty());
1955 let qs = w.questions().list();
1956 assert_eq!(qs.len(), 1);
1957 assert_eq!(qs[0].choices, vec![land::APPROVE, land::HOLD]);
1958 }
1959
1960 #[tokio::test]
1961 async fn an_unregistered_pr_is_an_actions_release_when_the_config_is_unreadable() {
1962 let (_d, fake, w) = rig();
1963 *fake.config_fails.lock().unwrap() = true;
1964 *fake.snap.lock().unwrap() = Some(red());
1965 let repo = PathBuf::from("/nowhere");
1966 let no = || false;
1967 w.lap(std::slice::from_ref(&repo), 3600, 1000, &no).await;
1968 assert_eq!(*fake.reruns.lock().unwrap(), vec!["11".to_owned()]);
1969 }
1970
1971 #[tokio::test]
1972 async fn a_merged_local_pr_waits_with_its_record_while_the_config_is_unreadable() {
1973 let (dir, fake, w) = rig();
1974 *fake.config_fails.lock().unwrap() = true;
1975 *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Merged, "h1", vec![]));
1976 let repo = PathBuf::from("/nowhere");
1977 let no = || false;
1978 let pr = pr_key(URL).unwrap();
1979 register(dir.path(), &repo, URL, "run1");
1980 w.lap(std::slice::from_ref(&repo), 60, 1, &no).await;
1981 assert!(w.state_path(&pr).exists(), "the record is kept");
1982 }
1983
1984 #[tokio::test]
1985 async fn land_approval_off_merges_the_observed_head_without_a_question() {
1986 let (_d, fake, w) = rig();
1987 *fake.local.lock().unwrap() = true;
1988 *fake.approval.lock().unwrap() = Some(false);
1989 *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Open, "h1", vec![]));
1990 let repo = PathBuf::from("/nowhere");
1991 let no = || false;
1992 w.lap(std::slice::from_ref(&repo), 60, 1, &no).await;
1993 assert_eq!(*fake.merges.lock().unwrap(), vec!["h1".to_owned()]);
1994 assert!(w.questions().list().is_empty(), "no approval question");
1995 }
1996
1997 #[tokio::test]
1998 async fn land_approval_on_or_unreadable_still_asks_and_never_merges() {
1999 for approval in [Some(true), None] {
2000 let (_d, fake, w) = rig();
2001 *fake.local.lock().unwrap() = true;
2002 *fake.approval.lock().unwrap() = approval;
2003 *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Open, "h1", vec![]));
2004 let repo = PathBuf::from("/nowhere");
2005 let no = || false;
2006 w.lap(std::slice::from_ref(&repo), 60, 1, &no).await;
2007 w.lap(std::slice::from_ref(&repo), 60, 2, &no).await;
2008 assert!(fake.merges.lock().unwrap().is_empty(), "silence is a hold");
2009 let qs = w.questions().list();
2010 assert_eq!(qs.len(), 1);
2011 assert_eq!(qs[0].choices, vec![land::APPROVE, land::HOLD]);
2012 }
2013 }
2014
2015 #[tokio::test]
2016 async fn land_approval_off_does_not_merge_on_an_unreadable_forge_or_retry_a_refusal() {
2017 let (_d, fake, w) = rig();
2018 *fake.local.lock().unwrap() = true;
2019 *fake.approval.lock().unwrap() = Some(false);
2020 let repo = PathBuf::from("/nowhere");
2021 let no = || false;
2022 *fake.snap.lock().unwrap() = None;
2024 w.lap(std::slice::from_ref(&repo), 60, 1, &no).await;
2025 assert!(fake.merges.lock().unwrap().is_empty());
2026 *fake.merge_fails.lock().unwrap() = true;
2028 *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Open, "h1", vec![]));
2029 w.lap(std::slice::from_ref(&repo), 60, 2, &no).await;
2030 w.lap(std::slice::from_ref(&repo), 60, 3, &no).await;
2031 assert_eq!(fake.merges.lock().unwrap().len(), 1);
2032 assert!(w.questions().list().is_empty());
2033 *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Open, "h2", vec![]));
2035 w.lap(std::slice::from_ref(&repo), 60, 4, &no).await;
2036 assert_eq!(
2037 *fake.merges.lock().unwrap(),
2038 vec!["h1".to_owned(), "h2".to_owned()]
2039 );
2040 }
2041
2042 #[test]
2043 fn pr_info_is_read_from_gh_json() {
2044 let i = parse_info(
2045 r#"{"headRefName":"chore/release-v1.2.3","title":"chore: release v1.2.3","mergeCommit":{"oid":"abc"}}"#,
2046 )
2047 .unwrap();
2048 assert_eq!(i.branch, "chore/release-v1.2.3");
2049 assert_eq!(i.merge_commit.as_deref(), Some("abc"));
2050 let open = parse_info(r#"{"headRefName":"b","title":"t","mergeCommit":null}"#).unwrap();
2051 assert_eq!(open.merge_commit, None);
2052 }
2053
2054 #[tokio::test]
2055 async fn a_local_release_asks_once_without_ci_reruns_or_stall_escalation() {
2056 let (_d, fake, w) = rig();
2057 *fake.local.lock().unwrap() = true;
2058 *fake.snap.lock().unwrap() = Some(red());
2061 let repo = PathBuf::from("/nowhere");
2062 let no = || false;
2063 w.lap(std::slice::from_ref(&repo), 60, 1_000_000, &no).await;
2064 w.lap(std::slice::from_ref(&repo), 60, 2_000_000, &no).await;
2065 assert!(fake.reruns.lock().unwrap().is_empty());
2066 let qs = w.questions().list();
2067 assert_eq!(qs.len(), 1, "one approval question, not one per lap");
2068 assert_eq!(qs[0].choices, vec![land::APPROVE, land::HOLD]);
2069 assert!(fake.merges.lock().unwrap().is_empty(), "silence is a hold");
2070 }
2071
2072 #[tokio::test]
2073 async fn merge_is_pinned_to_the_approved_head_and_a_moved_head_asks_again() {
2074 let (_d, fake, w) = rig();
2075 *fake.local.lock().unwrap() = true;
2076 *fake.snap.lock().unwrap() = Some(open(vec![]));
2077 let repo = PathBuf::from("/nowhere");
2078 let no = || false;
2079 w.lap(std::slice::from_ref(&repo), 60, 1, &no).await;
2080 let id = w.questions().list()[0].id.clone();
2081 *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Open, "h2", vec![]));
2083 answer(&w, &id, land::APPROVE);
2084 w.lap(std::slice::from_ref(&repo), 60, 2, &no).await;
2085 assert!(fake.merges.lock().unwrap().is_empty());
2086 assert_eq!(w.questions().list().len(), 2, "asked again about h2");
2087 let id2 = w
2089 .questions()
2090 .list()
2091 .into_iter()
2092 .find(|q| q.status.open())
2093 .unwrap()
2094 .id;
2095 answer(&w, &id2, land::APPROVE);
2096 w.lap(std::slice::from_ref(&repo), 60, 3, &no).await;
2097 assert_eq!(*fake.merges.lock().unwrap(), vec!["h2".to_owned()]);
2098 }
2099
2100 #[tokio::test]
2101 async fn a_failed_release_holds_with_one_notice_and_one_question_and_never_retries() {
2102 let (d, fake, w) = rig();
2103 *fake.local.lock().unwrap() = true;
2104 *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Merged, "h1", vec![]));
2105 *fake.info.lock().unwrap() = Some(PrInfo {
2106 branch: "chore/release-v1.0.0".to_owned(),
2107 merge_commit: Some("deadbeef".to_owned()),
2108 title: "chore: release v1.0.0".to_owned(),
2109 });
2110 let repo = d.path().join("nowhere");
2112 let no = || false;
2113 w.lap(std::slice::from_ref(&repo), 60, 1, &no).await;
2114 let st = w.load("o/r#7");
2115 let job = st.job.expect("the job is recorded");
2116 assert!(job.failed.is_some() && !job.tag_done && !job.finished);
2117 let qs = w.questions().list();
2118 assert_eq!(qs.len(), 1);
2119 assert_eq!(qs[0].choices, vec![RETRY, LEAVE_IT]);
2120 w.lap(std::slice::from_ref(&repo), 60, 2, &no).await;
2122 assert_eq!(w.questions().list().len(), 1);
2123 assert_eq!(Notices::at(d.path().join("notifications")).list().len(), 1);
2124 }
2125
2126 #[tokio::test]
2127 async fn an_escalated_pull_request_records_the_title_version_as_acceptable() {
2128 let (d, fake, w) = rig();
2129 *fake.local.lock().unwrap() = true;
2130 *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Merged, "h1", vec![]));
2131 *fake.info.lock().unwrap() = Some(PrInfo {
2132 branch: "chore/release-v0.1.4".to_owned(),
2133 merge_commit: Some("deadbeef".to_owned()),
2134 title: "chore: release v0.2.0 (minor bump)".to_owned(),
2135 });
2136 let repo = d.path().join("nowhere");
2137 w.lap(std::slice::from_ref(&repo), 60, 1, &(|| false)).await;
2138 let job = w.load("o/r#7").job.expect("the job is recorded");
2139 assert_eq!(job.version, "0.1.4");
2140 assert_eq!(job.accepted, vec!["0.2.0".to_owned()]);
2141 }
2142
2143 #[tokio::test]
2144 async fn a_failed_release_holds_the_task_whose_run_opened_the_pr() {
2145 let (d, fake, w) = rig();
2146 *fake.local.lock().unwrap() = true;
2147 *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Merged, "h1", vec![]));
2148 *fake.info.lock().unwrap() = Some(PrInfo {
2149 branch: "chore/release-v1.0.0".to_owned(),
2150 merge_commit: Some("deadbeef".to_owned()),
2151 title: "t".to_owned(),
2152 });
2153 let repo = d.path().join("nowhere");
2154 let q = crate::queue::Queue::at(d.path().join("queue"));
2155 let mut t = crate::queue::Task::new(
2156 "t".to_owned(),
2157 "i".to_owned(),
2158 repo.clone(),
2159 crate::queue::Source::Human,
2160 );
2161 t.runs.push("run1".to_owned());
2162 t.status = crate::queue::TaskStatus::Done;
2163 q.put(&mut t).unwrap();
2164 register(d.path(), &repo, URL, "run1");
2165 w.lap(std::slice::from_ref(&repo), 60, 1, &(|| false)).await;
2166 let held = q.get(&t.id).unwrap();
2167 assert_eq!(held.status, crate::queue::TaskStatus::Held);
2168 assert!(
2169 held.hold_reason
2170 .unwrap_or_default()
2171 .contains("release v1.0.0")
2172 );
2173 }
2174
2175 #[tokio::test]
2176 async fn unhold_gives_the_task_back_only_when_the_hold_is_ours() {
2177 let (d, _fake, w) = rig();
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 PathBuf::from("/r"),
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 let mut st = WatchState {
2189 run: Some("run1".to_owned()),
2190 ..WatchState::default()
2191 };
2192 w.hold_task(&mut st, "failed");
2193 assert_eq!(q.get(&t.id).unwrap().status, crate::queue::TaskStatus::Held);
2194 w.hold_task(&mut st, "failed again");
2196 assert!(w.unhold_task(&mut st));
2197 assert_eq!(q.get(&t.id).unwrap().status, crate::queue::TaskStatus::Done);
2198 assert!(st.held_task.is_none());
2199
2200 let mut h = q.get(&t.id).unwrap();
2202 st.held_task = Some(h.id.clone());
2203 h.hold_manual(Some("mine".to_owned()));
2204 q.put(&mut h).unwrap();
2205 assert!(w.unhold_task(&mut st));
2206 assert_eq!(q.get(&t.id).unwrap().status, crate::queue::TaskStatus::Held);
2207 }
2208
2209 #[tokio::test]
2210 async fn a_restart_after_the_failure_was_saved_still_holds_the_task() {
2211 let (d, fake, w) = rig();
2212 *fake.local.lock().unwrap() = true;
2213 *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Merged, "h1", vec![]));
2214 let repo = d.path().join("nowhere");
2215 let q = crate::queue::Queue::at(d.path().join("queue"));
2216 let mut t = crate::queue::Task::new(
2217 "t".to_owned(),
2218 "i".to_owned(),
2219 repo.clone(),
2220 crate::queue::Source::Human,
2221 );
2222 t.runs.push("run1".to_owned());
2223 t.status = crate::queue::TaskStatus::Done;
2224 q.put(&mut t).unwrap();
2225 let mut job = Job::new("1.0.0", URL, "deadbeef");
2227 job.failed = Some("command 1 exited 3".to_owned());
2228 let st = WatchState {
2229 repo: repo.to_string_lossy().into_owned(),
2230 url: URL.to_owned(),
2231 run: Some("run1".to_owned()),
2232 job: Some(job),
2233 ..WatchState::default()
2234 };
2235 w.save("o/r#7", &st);
2236 w.lap(std::slice::from_ref(&repo), 60, 1, &(|| false)).await;
2237 assert_eq!(q.get(&t.id).unwrap().status, crate::queue::TaskStatus::Held);
2238 }
2239
2240 #[tokio::test]
2241 async fn a_restart_after_the_job_finished_still_gives_the_task_back() {
2242 let (d, fake, w) = rig();
2243 *fake.local.lock().unwrap() = true;
2244 *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Merged, "h1", vec![]));
2245 let repo = d.path().join("nowhere");
2246 let q = crate::queue::Queue::at(d.path().join("queue"));
2247 let mut t = crate::queue::Task::new(
2248 "t".to_owned(),
2249 "i".to_owned(),
2250 repo.clone(),
2251 crate::queue::Source::Human,
2252 );
2253 t.runs.push("run1".to_owned());
2254 t.hold_machine(Some(format!("{HOLD_PREFIX}release v1.0.0 failed")));
2255 q.put(&mut t).unwrap();
2256 let mut job = Job::new("1.0.0", URL, "deadbeef");
2258 job.finished = true;
2259 let st = WatchState {
2260 repo: repo.to_string_lossy().into_owned(),
2261 url: URL.to_owned(),
2262 run: Some("run1".to_owned()),
2263 job: Some(job),
2264 held_task: Some(t.id.clone()),
2265 ..WatchState::default()
2266 };
2267 w.save("o/r#7", &st);
2268 w.lap(std::slice::from_ref(&repo), 60, 1, &(|| false)).await;
2269 assert_eq!(q.get(&t.id).unwrap().status, crate::queue::TaskStatus::Done);
2270 assert!(!w.state_path("o/r#7").exists());
2271 }
2272
2273 #[tokio::test]
2274 async fn a_pull_request_registered_at_open_is_picked_up_even_if_unlisted() {
2275 let (d, _fake, w) = rig();
2276 register(d.path(), Path::new("/r"), URL, "run1");
2277 let st = w.load("o/r#7");
2278 assert_eq!(st.url, URL);
2279 assert_eq!(st.repo, "/r");
2280 let mut live = st.clone();
2282 live.head = "keep".to_owned();
2283 w.save("o/r#7", &live);
2284 register(d.path(), Path::new("/r"), URL, "run1");
2285 assert_eq!(w.load("o/r#7").head, "keep");
2286 }
2287
2288 #[tokio::test]
2289 async fn a_finished_release_is_kept_on_the_run_and_a_missing_run_still_completes() {
2290 let (d, fake, w) = rig();
2291 *fake.local.lock().unwrap() = true;
2292 *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Merged, "h1", vec![]));
2293 let repo = d.path().join("nowhere");
2294 let mut run = RunState::new(
2295 repo.clone(),
2296 "main".to_owned(),
2297 "0123456789abcdef".to_owned(),
2298 "t".to_owned(),
2299 crate::config::Config::default(),
2300 );
2301 run.id = "run1".to_owned();
2302 run.save_under(d.path()).unwrap();
2303 let mut job = Job::new("1.0.0", URL, "deadbeef");
2304 job.finished = true;
2305 job.log.push(release_local::StepLog {
2306 name: "cmd".to_owned(),
2307 code: Some(0),
2308 tail: "ok".to_owned(),
2309 output: None,
2310 });
2311 let mk = |run: &str| WatchState {
2312 repo: repo.to_string_lossy().into_owned(),
2313 url: URL.to_owned(),
2314 run: Some(run.to_owned()),
2315 job: Some(job.clone()),
2316 ..WatchState::default()
2317 };
2318 w.save("o/r#7", &mk("run1"));
2319 w.lap(std::slice::from_ref(&repo), 60, 1, &(|| false)).await;
2320 assert!(!w.state_path("o/r#7").exists());
2321 let kept = RunState::load_under("run1", d.path()).unwrap();
2322 assert_eq!(kept.release_bump.unwrap().release, Some(job.clone()));
2323
2324 w.save("o/r#7", &mk("gone"));
2326 w.lap(std::slice::from_ref(&repo), 60, 1, &(|| false)).await;
2327 assert!(!w.state_path("o/r#7").exists());
2328 }
2329}