1use std::collections::{BTreeMap, BTreeSet};
30use std::fs::File;
31use std::path::{Path, PathBuf};
32use std::time::Duration;
33
34use serde::{Deserialize, Serialize};
35
36use crate::agent::Agent;
37use crate::answer;
38use crate::checkin;
39use crate::clock;
40use crate::config::{Config, ListenCfg, OnMention, OwnPr, ReviewOrIgnore};
41use crate::error::Result;
42use crate::model::{IssueRun, ItemKind, Login, PrSweepRow, ReviewRequested, Status};
43use crate::proc;
44use crate::repo::{ApiResponse, Repo};
45use crate::review_only::{self, RecheckMode, ReviewOpts};
46use crate::spend;
47use crate::{bail, log, logdim, logwarn, spar_err};
48
49const FLOOR_SECS: u64 = 15;
51const SEEN_FOR_SECS: u64 = 7 * 86_400;
53const MAX_PAGES: usize = 50;
56const MAX_FAILURES: u32 = 5;
58
59pub const STATE_VERSION: u32 = 1;
60
61#[derive(Debug, Clone, Default)]
63pub struct Opts {
64 pub items: Vec<i64>,
66 pub once: bool,
68 pub dry_run: bool,
70 pub since: Option<Duration>,
72}
73
74#[derive(Debug, Clone, Default, Deserialize, PartialEq, Eq)]
81pub struct Thread {
82 #[serde(default)]
83 pub id: String,
84 #[serde(default)]
85 pub reason: String,
86 #[serde(default)]
87 pub updated_at: String,
88 #[serde(default)]
89 pub subject: Subject,
90}
91
92#[derive(Debug, Clone, Default, Deserialize, PartialEq, Eq)]
93pub struct Subject {
94 #[serde(default)]
95 pub title: String,
96 #[serde(default)]
99 pub url: Option<String>,
100 #[serde(default)]
101 pub latest_comment_url: Option<String>,
102 #[serde(default, rename = "type")]
104 pub kind: String,
105}
106
107#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
112#[serde(rename_all = "snake_case")]
113pub enum Trigger {
114 Pushed,
116 Replied,
118 Mentioned,
119 ReviewRequested,
120 Assigned,
121 Closed,
122}
123
124impl Trigger {
125 fn noun(self) -> &'static str {
127 match self {
128 Trigger::Pushed => "a push",
129 Trigger::Replied => "a reply",
130 Trigger::Mentioned => "a mention",
131 Trigger::ReviewRequested => "a review request",
132 Trigger::Assigned => "an assignment",
133 Trigger::Closed => "a close",
134 }
135 }
136}
137
138#[derive(Debug, Clone, PartialEq, Eq)]
140pub struct Event {
141 pub number: i64,
142 pub kind: ItemKind,
143 pub trigger: Trigger,
144 pub who: Option<String>,
146 pub head: Option<(String, String)>,
148 pub commits: Option<u64>,
150 pub from_thread: bool,
154}
155
156#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
160pub struct Pending {
161 pub kind: ItemKind,
162 pub first_seen: u64,
163 pub last_seen: u64,
164 pub triggers: BTreeSet<Trigger>,
165 #[serde(default)]
166 pub who: Vec<String>,
167 #[serde(default)]
168 pub head_from: Option<String>,
169 #[serde(default)]
170 pub head_to: Option<String>,
171}
172
173impl Pending {
174 pub fn because(&self) -> String {
176 let mut parts: Vec<String> = self.triggers.iter().map(|t| t.noun().to_string()).collect();
177 if parts.is_empty() {
178 parts.push("nothing in particular".into());
179 }
180 match parts.len() {
181 1 => parts.remove(0),
182 2 => format!("{} and {}", parts[0], parts[1]),
183 _ => {
184 let last = parts.pop().unwrap_or_default();
185 format!("{}, and {last}", parts.join(", "))
186 }
187 }
188 }
189}
190
191#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
193pub struct Watched {
194 pub kind: ItemKind,
195 #[serde(default)]
196 pub title: String,
197 #[serde(default)]
198 pub author: String,
199 #[serde(default)]
202 pub head: Option<String>,
203 #[serde(default)]
204 pub requested_noticed: bool,
205 #[serde(default)]
206 pub assigned_noticed: bool,
207 #[serde(default)]
208 pub failures: u32,
209 #[serde(default)]
210 pub retry_after: u64,
211 #[serde(default)]
212 pub said_capped: bool,
213 #[serde(default)]
214 pub said_resume: bool,
215 #[serde(default)]
216 pub last_dispatched: Option<u64>,
217 #[serde(default)]
222 pub held: BTreeSet<Trigger>,
223}
224
225impl Watched {
226 fn new(kind: ItemKind) -> Self {
227 Self {
228 kind,
229 title: String::new(),
230 author: String::new(),
231 head: None,
232 requested_noticed: false,
233 assigned_noticed: false,
234 failures: 0,
235 retry_after: 0,
236 said_capped: false,
237 said_resume: false,
238 last_dispatched: None,
239 held: BTreeSet::new(),
240 }
241 }
242}
243
244#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
247pub struct ListenState {
248 pub version: u32,
249 #[serde(default)]
251 pub watermark: String,
252 #[serde(default)]
255 pub last_modified: Option<String>,
256 #[serde(default)]
259 pub seen: BTreeMap<String, String>,
260 #[serde(default)]
261 pub items: BTreeMap<i64, Watched>,
262 #[serde(default)]
263 pub pending: BTreeMap<i64, Pending>,
264}
265
266impl ListenState {
267 fn watched(&mut self, number: i64, kind: ItemKind) -> &mut Watched {
268 self.items
269 .entry(number)
270 .or_insert_with(|| Watched::new(kind))
271 }
272
273 fn note(&mut self, event: &Event, now: u64) {
276 let held = std::mem::take(&mut self.watched(event.number, event.kind).held);
277 merge(&mut self.pending, event, now);
278 if let Some(p) = self.pending.get_mut(&event.number) {
279 p.triggers.extend(held);
280 }
281 }
282
283 fn give_up(&mut self, number: i64, kind: ItemKind) {
287 if let Some(p) = self.pending.remove(&number) {
288 self.watched(number, kind).held.extend(p.triggers);
289 }
290 }
291}
292
293pub fn parse_duration(text: &str) -> Result<Duration> {
299 let bad = || {
300 spar_err!(
301 "`--since` takes a number and a unit, like 30m, 2h or 1d, not {:?}",
302 text.trim()
303 )
304 };
305 let mut total = 0u64;
306 let mut digits = String::new();
307 let mut any = false;
308 for c in text.trim().chars() {
309 if c.is_ascii_digit() {
310 digits.push(c);
311 continue;
312 }
313 let n: u64 = digits.parse().map_err(|_| bad())?;
314 digits.clear();
315 let unit = match c {
316 's' => 1,
317 'm' => 60,
318 'h' => 3600,
319 'd' => 86_400,
320 _ => return Err(bad()),
321 };
322 total = total.saturating_add(n.saturating_mul(unit));
323 any = true;
324 }
325 if !digits.is_empty() || !any {
326 return Err(bad());
327 }
328 Ok(Duration::from_secs(total))
329}
330
331pub fn poll_interval(cfg: &ListenCfg, header: Option<&str>) -> u64 {
334 let asked = header
335 .and_then(|h| h.trim().parse::<u64>().ok())
336 .unwrap_or(0);
337 cfg.interval.max(asked).max(FLOOR_SECS)
338}
339
340pub fn rate_limit_wait(resp: &ApiResponse, now: u64) -> Option<u64> {
343 if resp.status != 403 && resp.status != 429 {
344 return None;
345 }
346 let retry_after = resp
347 .headers
348 .get("retry-after")
349 .and_then(|v| v.trim().parse::<u64>().ok());
350 let exhausted = resp
351 .headers
352 .get("x-ratelimit-remaining")
353 .is_some_and(|v| v.trim() == "0");
354 let reset = resp
355 .headers
356 .get("x-ratelimit-reset")
357 .and_then(|v| v.trim().parse::<u64>().ok())
358 .map(|at| at.saturating_sub(now));
359 let wait = match (retry_after, exhausted, reset) {
360 (Some(secs), _, _) => secs,
361 (None, true, Some(secs)) => secs,
362 _ => return None,
363 };
364 Some(wait.clamp(60, 3600))
365}
366
367fn status_message(resp: &ApiResponse) -> String {
370 serde_json::from_str::<serde_json::Value>(&resp.body)
371 .ok()
372 .and_then(|v| {
373 v.get("message")
374 .and_then(|m| m.as_str())
375 .map(|m| format!("HTTP {}: {m}", resp.status))
376 })
377 .unwrap_or_else(|| format!("HTTP {}", resp.status))
378}
379
380pub fn next_link(headers: &BTreeMap<String, String>) -> Option<String> {
382 let link = headers.get("link")?;
383 link.split(',').find_map(|part| {
384 let (url, rel) = part.split_once(';')?;
385 if !rel.contains("rel=\"next\"") {
386 return None;
387 }
388 let url = url.trim().trim_start_matches('<').trim_end_matches('>');
389 (!url.is_empty()).then(|| url.to_string())
390 })
391}
392
393pub fn number_from_url(url: &str) -> Option<i64> {
395 url.trim_end_matches('/')
396 .rsplit('/')
397 .next()
398 .and_then(|last| last.parse().ok())
399}
400
401pub fn events_from_threads(threads: &[Thread], state: &mut ListenState) -> Vec<Event> {
404 let mut out = Vec::new();
405 for thread in threads {
406 if state.seen.get(&thread.id) == Some(&thread.updated_at) {
407 continue;
408 }
409 if !thread.updated_at.is_empty() {
410 state
411 .seen
412 .insert(thread.id.clone(), thread.updated_at.clone());
413 if thread.updated_at > state.watermark {
414 state.watermark = thread.updated_at.clone();
415 }
416 }
417 let kind = match thread.subject.kind.as_str() {
418 "PullRequest" => ItemKind::Pr,
419 "Issue" => ItemKind::Issue,
420 _ => continue,
421 };
422 let Some(number) = thread.subject.url.as_deref().and_then(number_from_url) else {
423 continue;
424 };
425 let trigger = match thread.reason.as_str() {
426 "review_requested" => Trigger::ReviewRequested,
427 "assign" => Trigger::Assigned,
428 "mention" | "team_mention" => Trigger::Mentioned,
429 _ => Trigger::Replied,
433 };
434 out.push(Event {
435 number,
436 kind,
437 trigger,
438 who: None,
439 head: None,
440 commits: None,
441 from_thread: true,
442 });
443 }
444 out
445}
446
447pub fn events_from_sweep(
453 rows: &[PrSweepRow],
454 viewer: &str,
455 state: &mut ListenState,
456 named: &BTreeSet<i64>,
457 min_number: i64,
458 record_head: &dyn Fn(i64) -> Option<String>,
459 repo_mode: bool,
460) -> Vec<Event> {
461 let mut out = Vec::new();
462 let mut present: BTreeSet<i64> = BTreeSet::new();
463 for row in rows {
464 present.insert(row.number);
468 if row.number < min_number && !named.contains(&row.number) {
469 continue;
470 }
471 let watched = state.watched(row.number, ItemKind::Pr);
472 if !row.title.is_empty() {
473 watched.title = row.title.clone();
474 }
475 if let Some(author) = &row.author {
476 watched.author = author.login.clone();
477 }
478 if !row.state.eq_ignore_ascii_case("OPEN") {
479 out.push(Event {
480 number: row.number,
481 kind: ItemKind::Pr,
482 trigger: Trigger::Closed,
483 who: None,
484 head: None,
485 commits: None,
486 from_thread: false,
487 });
488 continue;
489 }
490 let from = match &watched.head {
491 Some(old) => Some(old.clone()),
492 None => record_head(row.number),
493 };
494 if let Some(old) = from {
495 if !row.head_ref_oid.is_empty() && !old.eq_ignore_ascii_case(&row.head_ref_oid) {
496 out.push(Event {
497 number: row.number,
498 kind: ItemKind::Pr,
499 trigger: Trigger::Pushed,
500 who: None,
501 head: Some((old, row.head_ref_oid.clone())),
502 commits: None,
503 from_thread: false,
504 });
505 }
506 }
507 if !row.head_ref_oid.is_empty() {
508 watched.head = Some(row.head_ref_oid.clone());
509 }
510 let asked = row
511 .review_requests
512 .iter()
513 .any(|l| crate::comments::same_login(&l.login, viewer));
514 if !asked {
519 watched.requested_noticed = false;
520 } else if !watched.requested_noticed {
521 watched.requested_noticed = true;
522 out.push(Event {
523 number: row.number,
524 kind: ItemKind::Pr,
525 trigger: Trigger::ReviewRequested,
526 who: None,
527 head: None,
528 commits: None,
529 from_thread: false,
530 });
531 }
532 let assigned = row
533 .assignees
534 .iter()
535 .any(|l| crate::comments::same_login(&l.login, viewer));
536 if assigned && !watched.assigned_noticed {
537 watched.assigned_noticed = true;
538 out.push(Event {
539 number: row.number,
540 kind: ItemKind::Pr,
541 trigger: Trigger::Assigned,
542 who: None,
543 head: None,
544 commits: None,
545 from_thread: false,
546 });
547 }
548 }
549 if repo_mode {
550 let gone: Vec<i64> = state
553 .items
554 .iter()
555 .filter(|(n, w)| w.kind == ItemKind::Pr && !present.contains(n))
556 .map(|(n, _)| *n)
557 .collect();
558 for number in gone {
559 out.push(Event {
560 number,
561 kind: ItemKind::Pr,
562 trigger: Trigger::Closed,
563 who: None,
564 head: None,
565 commits: None,
566 from_thread: false,
567 });
568 }
569 }
570 out
571}
572
573pub fn merge(pending: &mut BTreeMap<i64, Pending>, event: &Event, now: u64) {
575 let p = pending.entry(event.number).or_insert_with(|| Pending {
576 kind: event.kind,
577 first_seen: now,
578 last_seen: now,
579 triggers: BTreeSet::new(),
580 who: Vec::new(),
581 head_from: None,
582 head_to: None,
583 });
584 p.last_seen = now;
585 p.triggers.insert(event.trigger);
586 if let Some(who) = &event.who {
587 if !p.who.contains(who) {
588 p.who.push(who.clone());
589 }
590 }
591 if let Some((from, to)) = &event.head {
592 if p.head_from.is_none() {
593 p.head_from = Some(from.clone());
594 }
595 p.head_to = Some(to.clone());
596 }
597}
598
599pub fn due(p: &Pending, now: u64, cfg: &ListenCfg, once: bool) -> bool {
602 once || now.saturating_sub(p.last_seen) >= cfg.settle
603 || now.saturating_sub(p.first_seen) >= cfg.max_wait
604}
605
606pub fn next_due(
610 state: &ListenState,
611 cfg: &ListenCfg,
612 acts_on: &dyn Fn(i64) -> bool,
613) -> Option<u64> {
614 state
615 .pending
616 .iter()
617 .filter(|(n, _)| acts_on(**n))
618 .map(|(n, p)| {
619 let settled = (p.last_seen + cfg.settle).min(p.first_seen + cfg.max_wait);
620 state
621 .items
622 .get(n)
623 .map_or(settled, |w| settled.max(w.retry_after))
624 })
625 .min()
626}
627
628#[derive(Debug, Clone, PartialEq, Eq)]
630pub enum Action {
631 Recheck(i64),
633 Review,
635 Checkin { reply_only: bool },
637 Answer { draft: bool },
640 AnswersCapped,
643 Capped(i64),
646 BelongsToResume,
648 Drop,
650 Ignore(&'static str),
652}
653
654impl Action {
655 pub fn phrase(&self, number: i64) -> String {
657 match self {
658 Action::Recheck(pr) if *pr == number => "take a second look".to_string(),
659 Action::Recheck(pr) => format!("take a second look at #{pr}"),
660 Action::Review => "review it".to_string(),
661 Action::Checkin { reply_only: false } => "check in on it".to_string(),
662 Action::Checkin { reply_only: true } => "answer the comments on it".to_string(),
663 Action::Answer { draft: false } => "answer what it asks of you".to_string(),
664 Action::Answer { draft: true } => "draft an answer to what it asks of you".to_string(),
665 Action::Capped(_) | Action::AnswersCapped | Action::BelongsToResume => {
666 "leave it".to_string()
667 }
668 Action::Drop => "drop it".to_string(),
669 Action::Ignore(_) => "ignore it".to_string(),
670 }
671 }
672
673 fn runs(&self) -> bool {
674 matches!(
675 self,
676 Action::Recheck(_) | Action::Review | Action::Checkin { .. } | Action::Answer { .. }
677 )
678 }
679}
680
681#[derive(Debug, Clone, Default, PartialEq, Eq)]
684pub struct Facts {
685 pub number: i64,
686 pub named: bool,
687 pub open: bool,
688 pub author_is_viewer: bool,
691 pub has_posted_record: bool,
693 pub recheck_capped: bool,
694 pub has_custody_state: bool,
696 pub below_floor: bool,
697 pub linked_pr_with_record: Option<i64>,
699 pub review_pending: bool,
702 pub request_answered: bool,
705 pub mentioned: bool,
709 pub asks_open: bool,
713 pub answers_capped: bool,
716}
717
718pub fn review_pending(
722 requests: &[Login],
723 viewer: &str,
724 in_team: &dyn Fn(&str) -> Option<bool>,
725) -> bool {
726 requests
727 .iter()
728 .any(|r| crate::comments::same_login(&r.login, viewer))
729 || requests
730 .iter()
731 .filter(|r| r.login.is_empty())
732 .any(|team| in_team(&team.slug).unwrap_or(true))
733}
734
735pub fn request_answered(
744 made: &[ReviewRequested],
745 viewer: &str,
746 in_team: &dyn Fn(&str) -> Option<bool>,
747 reviewed_at: &str,
748) -> bool {
749 let Some(reviewed) = clock::parse_rfc3339(reviewed_at) else {
750 return false;
751 };
752 made.iter()
753 .filter(|r| review_pending(std::slice::from_ref(&r.who), viewer, in_team))
754 .filter_map(|r| clock::parse_rfc3339(&r.at))
755 .max()
756 .is_some_and(|newest| newest <= reviewed)
757}
758
759pub fn settle(p: &mut Pending, f: &Facts) -> bool {
766 let stale = p.kind == ItemKind::Pr
767 && (!f.review_pending || f.request_answered)
768 && p.triggers.remove(&Trigger::ReviewRequested);
769 if stale {
770 p.triggers.insert(Trigger::Replied);
771 }
772 stale
773}
774
775pub fn settle_mention(p: &mut Pending, f: &Facts) -> bool {
783 let asked = match p.kind {
784 ItemKind::Pr => !f.has_posted_record || f.mentioned,
785 ItemKind::Issue if f.linked_pr_with_record.is_some() => f.mentioned,
786 ItemKind::Issue => f.asks_open,
787 };
788 let stale = !asked && p.triggers.remove(&Trigger::Mentioned);
789 if stale {
790 p.triggers.insert(Trigger::Replied);
791 }
792 stale
793}
794
795pub fn plan(p: &Pending, f: &Facts, cfg: &ListenCfg) -> Action {
797 use Trigger::*;
798 if !f.open || p.triggers.contains(&Closed) {
799 return Action::Drop;
800 }
801 if f.below_floor && !f.named {
802 return Action::Ignore("it is below min_number and was not named");
803 }
804 match p.kind {
805 ItemKind::Issue => {
806 if !p.triggers.contains(&Mentioned) {
810 return Action::Ignore("nothing on the issue asked for spar");
811 }
812 match f.linked_pr_with_record {
815 Some(_) if f.author_is_viewer => Action::Ignore("its pull request is your own"),
816 Some(pr) if f.recheck_capped => Action::Capped(pr),
817 Some(pr) => Action::Recheck(pr),
818 None => match cfg.on_mention {
822 OnMention::Ignore => Action::Ignore("on_mention is ignore"),
823 _ if !f.asks_open => Action::Ignore("a mention spar has already judged"),
824 _ if f.answers_capped => Action::AnswersCapped,
825 OnMention::Answer => Action::Answer { draft: false },
826 OnMention::Draft => Action::Answer { draft: true },
827 },
828 }
829 }
830 ItemKind::Pr => {
831 if f.author_is_viewer {
836 let spoken_to = p
837 .triggers
838 .iter()
839 .any(|t| matches!(t, Replied | Mentioned | ReviewRequested));
840 if !spoken_to {
841 return Action::Ignore("a push to your own pull request");
842 }
843 return match cfg.on_own_pr {
844 OwnPr::Checkin => Action::Checkin { reply_only: false },
845 OwnPr::Reply => Action::Checkin { reply_only: true },
846 OwnPr::Ignore => Action::Ignore("on_own_pr is ignore"),
847 };
848 }
849 if f.has_posted_record {
850 let asked =
854 p.triggers.contains(&ReviewRequested) || p.triggers.contains(&Mentioned);
855 if !asked {
856 return Action::Ignore(
857 "what is neither a review request nor a mention, the only things that \
858 ask for a second look",
859 );
860 }
861 return if f.recheck_capped {
862 Action::Capped(f.number)
863 } else {
864 Action::Recheck(f.number)
865 };
866 }
867 if f.has_custody_state {
868 return Action::BelongsToResume;
869 }
870 if f.named {
871 return Action::Review;
872 }
873 if p.triggers.contains(&ReviewRequested) {
874 return match cfg.on_review_requested {
875 ReviewOrIgnore::Review => Action::Review,
876 ReviewOrIgnore::Ignore => Action::Ignore("on_review_requested is ignore"),
877 };
878 }
879 if p.triggers.contains(&Assigned) {
880 return match cfg.on_assigned {
881 ReviewOrIgnore::Review => Action::Review,
882 ReviewOrIgnore::Ignore => Action::Ignore("on_assigned is ignore"),
883 };
884 }
885 if p.triggers.contains(&Mentioned) {
886 return Action::Review;
887 }
888 Action::Ignore("a push or a comment on a pull request spar has not reviewed")
889 }
890 }
891}
892
893pub fn prune_seen(state: &mut ListenState, now: u64) {
895 state.seen.retain(|_, at| {
896 clock::parse_rfc3339(at).is_none_or(|t| now.saturating_sub(t) < SEEN_FOR_SECS)
897 });
898}
899
900pub fn backoff_secs(failures: u32) -> u64 {
902 (60u64 << failures.saturating_sub(1).min(10)).min(900)
903}
904
905pub fn dispatch_backoff_secs(failures: u32) -> u64 {
907 (900u64 << failures.saturating_sub(1).min(10)).min(6 * 3600)
908}
909
910pub fn retry_wait(moved: bool, failures: u32, cfg: &ListenCfg) -> u64 {
914 if moved {
915 cfg.settle.max(FLOOR_SECS)
916 } else {
917 dispatch_backoff_secs(failures)
918 }
919}
920
921#[derive(Debug, Default)]
923struct Backoff {
924 failures: u32,
925 said: bool,
926}
927
928impl Backoff {
929 fn failed(&mut self, what: &str, why: &str) -> u64 {
930 self.failures += 1;
931 let wait = backoff_secs(self.failures);
932 let line = format!(
933 "could not {what}: {why}. Trying again in {}.",
934 spend::clock(Duration::from_secs(wait))
935 );
936 if self.said {
937 logdim!("{line}");
938 } else {
939 logwarn!("{line}");
940 self.said = true;
941 }
942 wait
943 }
944
945 fn recovered(&mut self) {
946 if self.said {
947 log!("gh is answering again");
948 }
949 *self = Self::default();
950 }
951}
952
953#[derive(Debug, Serialize, Deserialize)]
960struct LockFile {
961 pid: u32,
962 started: String,
963}
964
965#[derive(Debug)]
974struct Lock {
975 path: PathBuf,
976 _file: File,
978}
979
980impl Lock {
981 fn acquire(repo: &Repo, now: u64) -> Result<Lock> {
982 Lock::at(repo.listen_lock_path(), now)
983 }
984
985 fn at(path: PathBuf, now: u64) -> Result<Lock> {
986 if let Some(parent) = path.parent() {
987 std::fs::create_dir_all(parent)?;
988 }
989 for _ in 0..3 {
994 let Some(mut file) = open_locked(&path)? else {
995 let pid = std::fs::read_to_string(&path)
996 .ok()
997 .and_then(|text| serde_json::from_str::<LockFile>(&text).ok())
998 .map(|h| h.pid.to_string())
999 .unwrap_or_else(|| "unknown".into());
1000 bail!(
1001 "another `spar listen` (pid {pid}) is already watching this checkout. \
1002 Stop it first.{}",
1003 if cfg!(unix) {
1004 String::new()
1005 } else {
1006 format!(" If that process is gone, delete {}.", path.display())
1007 }
1008 );
1009 };
1010 if !same_file(&file, &path) {
1011 continue;
1012 }
1013 file.set_len(0)?;
1014 serde_json::to_writer_pretty(
1015 &mut file,
1016 &LockFile {
1017 pid: std::process::id(),
1018 started: clock::rfc3339(now),
1019 },
1020 )?;
1021 return Ok(Lock { path, _file: file });
1022 }
1023 bail!("could not take the listen lock at {}", path.display())
1024 }
1025}
1026
1027impl Drop for Lock {
1028 fn drop(&mut self) {
1029 let _ = std::fs::remove_file(&self.path);
1032 }
1033}
1034
1035#[cfg(unix)]
1037fn open_locked(path: &Path) -> Result<Option<File>> {
1038 use std::os::unix::io::AsRawFd;
1039 let file = std::fs::OpenOptions::new()
1040 .read(true)
1041 .write(true)
1042 .create(true)
1043 .truncate(false)
1044 .open(path)?;
1045 if unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) } == 0 {
1046 return Ok(Some(file));
1047 }
1048 let e = std::io::Error::last_os_error();
1049 if e.raw_os_error() == Some(libc::EWOULDBLOCK) {
1050 Ok(None)
1051 } else {
1052 Err(e.into())
1053 }
1054}
1055
1056#[cfg(not(unix))]
1059fn open_locked(path: &Path) -> Result<Option<File>> {
1060 match std::fs::OpenOptions::new()
1061 .write(true)
1062 .create_new(true)
1063 .open(path)
1064 {
1065 Ok(file) => Ok(Some(file)),
1066 Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => Ok(None),
1067 Err(e) => Err(e.into()),
1068 }
1069}
1070
1071#[cfg(unix)]
1072fn same_file(file: &File, path: &Path) -> bool {
1073 use std::os::unix::fs::MetadataExt;
1074 match (file.metadata(), std::fs::metadata(path)) {
1075 (Ok(open), Ok(named)) => open.dev() == named.dev() && open.ino() == named.ino(),
1076 _ => false,
1077 }
1078}
1079
1080#[cfg(not(unix))]
1081fn same_file(_file: &File, _path: &Path) -> bool {
1082 true
1083}
1084
1085enum Polled {
1090 Nothing(u64),
1092 Threads(Vec<Thread>, u64),
1093 RateLimited(u64),
1095 Failed(String),
1096}
1097
1098fn poll(repo: &Repo, cfg: &ListenCfg, state: &mut ListenState, now: u64) -> Polled {
1099 let endpoint = format!(
1100 "repos/{{owner}}/{{repo}}/notifications?participating=true&all=true&per_page=100&since={}",
1101 state.watermark
1102 );
1103 let last_modified = state.last_modified.clone();
1104 let headers: Vec<(&str, &str)> = last_modified
1105 .as_deref()
1106 .map(|lm| vec![("If-Modified-Since", lm)])
1107 .unwrap_or_default();
1108 let resp = match repo.gh_api_raw(&endpoint, &headers) {
1109 Ok(resp) => resp,
1110 Err(e) => return Polled::Failed(e.last_line().to_string()),
1111 };
1112 let interval = poll_interval(cfg, resp.headers.get("x-poll-interval").map(String::as_str));
1113 match resp.status {
1114 304 => Polled::Nothing(interval),
1115 200 => match every_page(repo, &resp, &state.watermark, now) {
1116 Ok(threads) => {
1117 state.last_modified = resp
1120 .headers
1121 .get("last-modified")
1122 .or_else(|| resp.headers.get("date"))
1123 .cloned();
1124 Polled::Threads(threads, interval)
1125 }
1126 Err(failed) => failed,
1127 },
1128 401 => Polled::Failed("gh is not authenticated; run `gh auth status`".into()),
1129 _ => refused(&resp, now),
1130 }
1131}
1132
1133fn refused(resp: &ApiResponse, now: u64) -> Polled {
1135 match rate_limit_wait(resp, now) {
1136 Some(wait) => Polled::RateLimited(wait),
1137 None => Polled::Failed(status_message(resp)),
1138 }
1139}
1140
1141fn every_page(
1148 repo: &Repo,
1149 first: &ApiResponse,
1150 watermark: &str,
1151 now: u64,
1152) -> std::result::Result<Vec<Thread>, Polled> {
1153 let unexpected =
1154 |e: serde_json::Error| Polled::Failed(format!("unexpected notification data: {e}"));
1155 let mut threads: Vec<Thread> = serde_json::from_str(&first.body).map_err(unexpected)?;
1156 let mut next = next_link(&first.headers);
1157 let mut pages = 1;
1158 while let Some(url) = next.take() {
1159 if pages >= MAX_PAGES {
1160 return Err(Polled::Failed(format!(
1161 "more than {MAX_PAGES} pages of notifications since {watermark}. Run once with \
1162 --since, like --since 1d, to start from later and skip the rest"
1163 )));
1164 }
1165 let more = repo
1166 .gh_api_raw(&url, &[])
1167 .map_err(|e| Polled::Failed(e.last_line().to_string()))?;
1168 if more.status != 200 {
1169 return Err(refused(&more, now));
1170 }
1171 threads.extend(serde_json::from_str::<Vec<Thread>>(&more.body).map_err(unexpected)?);
1172 next = next_link(&more.headers);
1173 pages += 1;
1174 }
1175 Ok(threads)
1176}
1177
1178#[derive(Default)]
1180struct Swept {
1181 rows: Vec<PrSweepRow>,
1182 closed_issues: Vec<i64>,
1184}
1185
1186fn sweep_rows(
1189 repo: &Repo,
1190 cfg: &Config,
1191 repo_mode: bool,
1192 named_prs: &[i64],
1193 named_issues: &[i64],
1194 state: &ListenState,
1195) -> Result<Swept> {
1196 if repo_mode {
1197 return Ok(Swept {
1198 rows: repo.sweep_open_prs()?,
1199 closed_issues: Vec::new(),
1200 });
1201 }
1202 let watched = |n: &&i64| state.items.contains_key(*n);
1203 let mut swept = Swept::default();
1204 for number in named_prs.iter().filter(watched) {
1205 swept.rows.push(repo.sweep_pr(*number)?);
1206 }
1207 for issue in named_issues.iter().filter(watched) {
1208 if !repo.read_issue(*issue)?.state.eq_ignore_ascii_case("OPEN") {
1212 swept.closed_issues.push(*issue);
1213 continue;
1214 }
1215 if let Ok(Some(pr)) = repo.try_open_pr_for_issue(*issue, cfg.base_branch()) {
1218 if repo.read_review_record(pr.number).is_some_and(|r| r.posted) {
1219 swept.rows.push(repo.sweep_pr(pr.number)?);
1220 }
1221 }
1222 }
1223 Ok(swept)
1224}
1225
1226fn facts(
1229 repo: &Repo,
1230 cfg: &Config,
1231 number: i64,
1232 kind: ItemKind,
1233 named: bool,
1234 mention: bool,
1235) -> Result<Facts> {
1236 let viewer = repo.viewer_login()?.to_string();
1237 let below_floor = number < cfg.loop_cfg.min_number;
1238 let in_team = |team: &str| repo.team_has_member(team, &viewer);
1239 match kind {
1240 ItemKind::Pr => {
1241 let row = repo.sweep_pr(number)?;
1242 let record = repo.read_review_record(number);
1243 let posted = record.as_ref().is_some_and(|r| r.posted);
1244 let pending = review_pending(&row.review_requests, &viewer, &in_team);
1245 let answered = pending
1248 && record.as_ref().filter(|r| r.posted).is_some_and(|r| {
1249 repo.review_requests_made(number).is_some_and(|made| {
1250 request_answered(&made, &viewer, &in_team, &r.reviewed_at)
1251 })
1252 });
1253 let mentioned = match record.as_ref().filter(|r| r.posted) {
1256 Some(r) if mention => {
1257 let mut rows = repo.try_pr_review_comments(number)?;
1258 rows.extend(repo.try_pr_reviews(number)?);
1259 rows.extend(repo.try_issue_comments(number)?);
1260 crate::comments::mentioned_in(&rows, &viewer, &r.reviewed_at, &in_team)
1261 }
1262 _ => false,
1263 };
1264 Ok(Facts {
1265 number,
1266 named,
1267 open: row.state.eq_ignore_ascii_case("OPEN"),
1268 author_is_viewer: row
1269 .author
1270 .as_ref()
1271 .is_some_and(|a| crate::comments::same_login(&a.login, &viewer)),
1272 has_posted_record: posted,
1273 recheck_capped: posted
1274 && record
1275 .as_ref()
1276 .is_some_and(|r| r.rechecks >= cfg.listen.max_rechecks),
1277 has_custody_state: repo.state_path(number).exists(),
1278 below_floor,
1279 linked_pr_with_record: None,
1280 review_pending: pending,
1281 request_answered: answered,
1282 mentioned,
1283 asks_open: false,
1284 answers_capped: false,
1285 })
1286 }
1287 ItemKind::Issue => {
1288 let issue = repo.read_issue(number)?;
1289 let linked = repo
1290 .try_open_pr_for_issue(number, cfg.base_branch())
1291 .ok()
1292 .flatten()
1293 .and_then(|pr| {
1294 repo.read_review_record(pr.number)
1295 .filter(|r| r.posted)
1296 .map(|r| (pr.number, r))
1297 });
1298 let (author_is_viewer, recheck_capped) = match &linked {
1300 Some((pr, record)) => {
1301 let row = repo.sweep_pr(*pr)?;
1302 (
1303 row.author
1304 .as_ref()
1305 .is_some_and(|a| crate::comments::same_login(&a.login, &viewer)),
1306 record.rechecks >= cfg.listen.max_rechecks,
1307 )
1308 }
1309 None => (false, false),
1310 };
1311 let mentioned = match &linked {
1312 Some((_, record)) if mention => crate::comments::mentioned_in(
1313 &repo.try_issue_comments(number)?,
1314 &viewer,
1315 &record.reviewed_at,
1316 &in_team,
1317 ),
1318 _ => false,
1319 };
1320 let (asks_open, answers_capped) = match &linked {
1321 None if mention => {
1322 let (open, automatic) = answer::outstanding(repo, number, &viewer, &in_team)?;
1323 (open > 0, automatic >= cfg.listen.max_answers)
1324 }
1325 _ => (false, false),
1326 };
1327 Ok(Facts {
1328 number,
1329 named,
1330 open: issue.state.eq_ignore_ascii_case("OPEN"),
1331 author_is_viewer,
1332 has_posted_record: false,
1333 recheck_capped,
1334 has_custody_state: false,
1335 below_floor,
1336 linked_pr_with_record: linked.map(|(pr, _)| pr),
1337 review_pending: false,
1338 request_answered: false,
1339 mentioned,
1340 asks_open,
1341 answers_capped,
1342 })
1343 }
1344 }
1345}
1346
1347fn dispatch(
1348 agents: &[Agent],
1349 cfg: &Config,
1350 repo: &Repo,
1351 action: &Action,
1352 number: i64,
1353 p: &Pending,
1354) -> IssueRun {
1355 match action {
1356 Action::Recheck(pr) => review_only::recheck_pr(
1357 agents,
1358 cfg,
1359 repo,
1360 *pr,
1361 RecheckMode {
1362 dry_run: false,
1363 requested: p.triggers.contains(&Trigger::ReviewRequested),
1364 verdict_on_fallback: cfg.listen.verdicts,
1365 },
1366 ),
1367 Action::Review => review_only::review_pr(
1368 agents,
1369 cfg,
1370 repo,
1371 number,
1372 ReviewOpts {
1373 dry_run: false,
1374 verdict: cfg.listen.verdicts,
1375 },
1376 ),
1377 Action::Checkin { reply_only } => checkin::checkin_pr(
1378 agents,
1379 cfg,
1380 repo,
1381 number,
1382 &checkin::Mode {
1383 dry_run: false,
1384 reply_only: *reply_only,
1385 trust: cfg.loop_cfg.checkin_trust,
1386 again: false,
1387 resolve: cfg.loop_cfg.checkin_resolve,
1388 posts: checkin::posts(cfg),
1389 },
1390 ),
1391 Action::Answer { draft } => answer::answer_issue(
1392 agents,
1393 cfg,
1394 repo,
1395 number,
1396 &answer::Mode {
1397 dry_run: false,
1398 draft: *draft,
1399 again: false,
1400 auto: true,
1401 posts: checkin::posts(cfg),
1402 },
1403 ),
1404 other => {
1405 let mut run = IssueRun::new(number, format!("#{number}"));
1406 run.status = Status::Error;
1407 run.notes
1408 .push(format!("nothing to run: {}", other.phrase(number)));
1409 run
1410 }
1411 }
1412}
1413
1414fn describe(repo: &Repo, event: &Event) -> String {
1416 let n = event.number;
1417 match event.trigger {
1418 Trigger::Pushed => {
1419 let (from, to) = event.head.clone().unwrap_or_default();
1420 let short = |s: &str| s.chars().take(7).collect::<String>();
1421 let commits = repo
1422 .commits_between(&from, &to)
1423 .map(|c| format!(", {c} new commit{} since the review", plural(c as usize)))
1424 .unwrap_or_default();
1425 format!(
1426 "#{n}: head moved {} to {}{commits}",
1427 short(&from),
1428 short(&to)
1429 )
1430 }
1431 Trigger::Replied => match &event.who {
1432 Some(who) => format!("#{n}: {who} commented"),
1433 None => format!("#{n}: somebody commented"),
1434 },
1435 Trigger::Mentioned => match &event.who {
1436 Some(who) => format!("#{n}: mentioned by {who}"),
1437 None => format!("#{n}: mentioned"),
1438 },
1439 Trigger::ReviewRequested if event.from_thread => {
1442 format!("#{n}: news on a thread that asked for your review")
1443 }
1444 Trigger::ReviewRequested => format!("#{n}: your review was requested"),
1445 Trigger::Assigned => format!("#{n}: assigned to you"),
1446 Trigger::Closed => format!("#{n}: closed, no longer watched"),
1447 }
1448}
1449
1450fn plural(n: usize) -> &'static str {
1451 if n == 1 {
1452 ""
1453 } else {
1454 "s"
1455 }
1456}
1457
1458enum Ending {
1463 Interrupted,
1464 Once { failed: bool },
1465 Done,
1466}
1467
1468pub fn run(agents: &[Agent], cfg: &Config, repo: &Repo, opts: Opts) -> Result<i32> {
1470 proc::stop_on_signal()?;
1471 if !cfg!(unix) {
1472 logdim!(
1473 "Ctrl-C ends spar at once on this platform; an agent call in flight is not stopped"
1474 );
1475 }
1476 let now = clock::now_secs();
1477 let lock = Lock::acquire(repo, now)?;
1478 let viewer = repo.viewer_login()?.to_string();
1479 let name = repo
1480 .repo_name()
1481 .unwrap_or_else(|| repo.root().display().to_string());
1482
1483 let named: BTreeSet<i64> = opts.items.iter().copied().collect();
1486 let mut named_prs = Vec::new();
1487 let mut named_issues = Vec::new();
1488 for number in &named {
1489 match repo.item_kind(*number)? {
1490 ItemKind::Pr => named_prs.push(*number),
1491 ItemKind::Issue => named_issues.push(*number),
1492 }
1493 }
1494 let repo_mode = named.is_empty();
1495 let mut watching: BTreeSet<i64> = named.clone();
1500 let save = |state: &ListenState| {
1503 if !opts.dry_run {
1504 write_state(repo, state);
1505 }
1506 };
1507
1508 let mut state = read_state(repo);
1509 if state.watermark.is_empty() || opts.since.is_some() {
1510 let back = opts.since.map(|d| d.as_secs()).unwrap_or(0);
1511 state.watermark = clock::rfc3339(now.saturating_sub(back));
1512 state.last_modified = None;
1513 log!("watching from {}", state.watermark);
1514 }
1515 for number in &named_prs {
1516 state.watched(*number, ItemKind::Pr);
1517 }
1518 for number in &named_issues {
1519 state.watched(*number, ItemKind::Issue);
1520 }
1521 log!(
1522 "listening to {name}{}: polling every {}s, sweeping {} every {}s. Ctrl-C to stop.",
1523 if repo_mode {
1524 String::new()
1525 } else {
1526 format!(
1527 " for {}",
1528 named
1529 .iter()
1530 .map(|n| format!("#{n}"))
1531 .collect::<Vec<_>>()
1532 .join(", ")
1533 )
1534 },
1535 cfg.listen.interval.max(FLOOR_SECS),
1536 if repo_mode {
1537 "open pull requests"
1538 } else {
1539 "the named items"
1540 },
1541 cfg.listen.sweep
1542 );
1543
1544 let started = now;
1545 let mut next_poll = now;
1546 let mut next_sweep = now;
1547 let mut backoff = Backoff::default();
1548 let mut events_seen = 0usize;
1549 let mut dispatches = 0usize;
1550 let mut tick_failed = false;
1551 let mut last_prune = now;
1552
1553 let ending = loop {
1554 if proc::interrupted() {
1555 break Ending::Interrupted;
1556 }
1557 let now = clock::now_secs();
1558 let mut events: Vec<Event> = Vec::new();
1559
1560 if now >= next_poll {
1561 match poll(repo, &cfg.listen, &mut state, now) {
1562 Polled::Nothing(interval) => {
1563 backoff.recovered();
1564 next_poll = now + interval;
1565 }
1566 Polled::Threads(threads, interval) => {
1567 backoff.recovered();
1568 events.extend(events_from_threads(&threads, &mut state));
1569 next_poll = now + interval;
1570 }
1571 Polled::RateLimited(wait) => {
1572 logwarn!(
1573 "rate limited by GitHub, waiting {} until it resets",
1574 spend::clock(Duration::from_secs(wait))
1575 );
1576 next_poll = now + wait;
1577 tick_failed = true;
1578 }
1579 Polled::Failed(why) => {
1580 next_poll = now + backoff.failed("read notifications", &why);
1581 tick_failed = true;
1582 }
1583 }
1584 }
1585
1586 if now >= next_sweep {
1587 match sweep_rows(repo, cfg, repo_mode, &named_prs, &named_issues, &state) {
1588 Ok(Swept {
1589 rows,
1590 closed_issues,
1591 }) => {
1592 backoff.recovered();
1593 if !repo_mode {
1594 watching.extend(rows.iter().map(|r| r.number));
1595 }
1596 events.extend(closed_issues.into_iter().map(|number| Event {
1597 number,
1598 kind: ItemKind::Issue,
1599 trigger: Trigger::Closed,
1600 who: None,
1601 head: None,
1602 commits: None,
1603 from_thread: false,
1604 }));
1605 let record_head = |n: i64| {
1606 repo.read_review_record(n)
1607 .filter(|r| r.posted)
1608 .map(|r| r.head)
1609 };
1610 let swept = events_from_sweep(
1611 &rows,
1612 &viewer,
1613 &mut state,
1614 &named,
1615 cfg.loop_cfg.min_number,
1616 &record_head,
1617 repo_mode,
1618 );
1619 events.extend(swept);
1620 next_sweep = now + cfg.listen.sweep;
1621 }
1622 Err(e) => {
1623 next_sweep = now + backoff.failed("sweep the pull requests", e.last_line());
1624 tick_failed = true;
1625 }
1626 }
1627 }
1628
1629 for event in &events {
1633 if !repo_mode && !watching.contains(&event.number) {
1634 continue;
1635 }
1636 if event.trigger != Trigger::Closed && event.who.is_none() {
1637 }
1641 events_seen += 1;
1642 log!("{}", describe(repo, event));
1643 if event.trigger == Trigger::Closed {
1644 state.items.remove(&event.number);
1645 state.pending.remove(&event.number);
1646 continue;
1647 }
1648 state.note(event, now);
1649 }
1650 save(&state);
1651
1652 let due_now: Vec<i64> = state
1653 .pending
1654 .iter()
1655 .filter(|(n, p)| {
1656 (repo_mode || watching.contains(n))
1657 && due(p, now, &cfg.listen, opts.once)
1658 && state.items.get(n).is_none_or(|w| w.retry_after <= now)
1659 })
1660 .map(|(n, _)| *n)
1661 .collect();
1662 for number in due_now {
1663 let Some(mut p) = state.pending.get(&number).cloned() else {
1664 continue;
1665 };
1666 let named_item = named.contains(&number);
1667 let mention = p.triggers.contains(&Trigger::Mentioned);
1668 let f = match facts(repo, cfg, number, p.kind, named_item, mention) {
1669 Ok(f) => f,
1670 Err(e) => {
1671 let w = state.watched(number, p.kind);
1672 w.failures += 1;
1673 w.retry_after = now + dispatch_backoff_secs(w.failures);
1674 logwarn!("#{number}: could not read it: {}", e.last_line());
1675 tick_failed = true;
1676 save(&state);
1677 continue;
1678 }
1679 };
1680 if settle(&mut p, &f) {
1681 if f.request_answered {
1682 log!(
1683 "#{number}: the review asked of you is the one on record, so that was \
1684 only an update"
1685 );
1686 } else {
1687 log!("#{number}: no review is asked of you now, so that was only an update");
1688 }
1689 }
1690 if settle_mention(&mut p, &f) {
1691 if p.kind == ItemKind::Issue && f.linked_pr_with_record.is_none() {
1692 log!(
1693 "#{number}: nothing on it asks you anything spar has not judged, so that \
1694 was only an update"
1695 );
1696 } else {
1697 log!(
1698 "#{number}: nobody mentioned you since the review on record, so that was \
1699 only an update"
1700 );
1701 }
1702 }
1703 if p.triggers.contains(&Trigger::ReviewRequested) || f.request_answered {
1704 state.watched(number, p.kind).requested_noticed = true;
1709 }
1710 let action = plan(&p, &f, &cfg.listen);
1711 match &action {
1712 Action::Ignore(why) => {
1713 logdim!("#{number}: ignoring {why}");
1714 state.pending.remove(&number);
1715 }
1716 Action::Drop => {
1717 log!("#{number}: closed, no longer watched");
1718 state.items.remove(&number);
1719 state.pending.remove(&number);
1720 }
1721 Action::Capped(pr) => {
1722 let w = state.watched(number, p.kind);
1723 if !w.said_capped {
1724 let what = if *pr == number {
1725 "rechecked".to_string()
1726 } else {
1727 format!("its pull request #{pr} was rechecked")
1728 };
1729 log!(
1730 "#{number}: {what} {} time{} already; run `spar review {pr}` to look \
1731 again",
1732 cfg.listen.max_rechecks,
1733 plural(cfg.listen.max_rechecks as usize)
1734 );
1735 w.said_capped = true;
1736 }
1737 state.pending.remove(&number);
1738 }
1739 Action::AnswersCapped => {
1740 let w = state.watched(number, p.kind);
1741 if !w.said_capped {
1742 if cfg.listen.max_answers == 0 {
1743 log!(
1744 "#{number}: max_answers is 0, so spar listen answers nothing; run \
1745 `spar answer {number}` to answer it"
1746 );
1747 } else {
1748 log!(
1749 "#{number}: answered {} time{} already; run `spar answer {number}` \
1750 to answer again",
1751 cfg.listen.max_answers,
1752 plural(cfg.listen.max_answers as usize)
1753 );
1754 }
1755 w.said_capped = true;
1756 }
1757 state.pending.remove(&number);
1758 }
1759 Action::BelongsToResume => {
1760 let w = state.watched(number, p.kind);
1761 if !w.said_resume {
1762 log!("#{number}: in custody, `spar resume {number}` continues it");
1763 w.said_resume = true;
1764 }
1765 state.pending.remove(&number);
1766 }
1767 action if action.runs() => {
1768 if opts.dry_run {
1769 log!(
1770 "#{number}: would {} after {}, dry run so nothing runs",
1771 action.phrase(number),
1772 p.because()
1773 );
1774 state.pending.remove(&number);
1775 } else {
1776 log!(
1777 "#{number}: about to {} after {}",
1778 action.phrase(number),
1779 p.because()
1780 );
1781 let acted_on = match action {
1782 Action::Recheck(pr) => *pr,
1783 _ => number,
1784 };
1785 let record_before = repo.read_review_record(acted_on);
1786 let failed_before = repo.write_summary().failed;
1787 let run = dispatch(agents, cfg, repo, action, number, &p);
1788 dispatches += 1;
1789 let spent = spend::drain();
1790 if let Err(e) = repo.record_spend(&spent) {
1791 logdim!("could not record what this run spent: {e}");
1792 }
1793 crate::cli::print_outcome(&run, &spent);
1794 if proc::interrupted() {
1795 log!("interrupted while working on #{number}; its worktree was kept");
1796 save(&state);
1797 break;
1798 }
1799 let record_after = repo.read_review_record(acted_on);
1805 if record_after != record_before {
1806 if let Some(record) = record_after {
1807 state.watched(acted_on, ItemKind::Pr).head = Some(record.head);
1808 }
1809 }
1810 let unposted = repo.write_summary().failed > failed_before;
1814 let w = state.watched(number, p.kind);
1815 let moved = run.status == Status::Moved && !unposted;
1819 if run.status == Status::Error || moved || unposted {
1820 w.failures += 1;
1821 let wait = retry_wait(moved, w.failures, &cfg.listen);
1822 w.retry_after = clock::now_secs() + wait;
1825 let gave_up = w.failures >= MAX_FAILURES;
1826 if gave_up {
1827 logwarn!("#{number}: giving up on it until it changes again");
1828 state.give_up(number, p.kind);
1829 } else {
1830 if moved {
1831 log!(
1832 "#{number}: its head moved while spar read it, so it runs \
1833 again on the new head in {}",
1834 spend::clock(Duration::from_secs(wait))
1835 );
1836 } else if unposted {
1837 logwarn!(
1838 "#{number}: not everything it wrote reached GitHub, so it \
1839 runs again in {}",
1840 spend::clock(Duration::from_secs(wait))
1841 );
1842 }
1843 if let Some(p) = state.pending.get_mut(&number) {
1844 p.first_seen = now;
1845 }
1846 }
1847 if !moved || gave_up {
1850 tick_failed = true;
1851 }
1852 } else {
1853 w.failures = 0;
1854 w.retry_after = 0;
1855 w.last_dispatched = Some(now);
1856 state.pending.remove(&number);
1857 }
1858 }
1859 }
1860 _ => {}
1861 }
1862 save(&state);
1863 }
1864 if proc::interrupted() {
1865 break Ending::Interrupted;
1866 }
1867
1868 if opts.once {
1869 break Ending::Once {
1870 failed: tick_failed,
1871 };
1872 }
1873 if !repo_mode && named.iter().all(|n| !state.items.contains_key(n)) {
1874 log!("every named item is closed, nothing left to watch");
1875 break Ending::Done;
1876 }
1877 if now.saturating_sub(last_prune) >= 3600 {
1878 prune_seen(&mut state, now);
1879 last_prune = now;
1880 save(&state);
1881 }
1882
1883 let acts_on = |n: i64| repo_mode || watching.contains(&n);
1884 let next = [next_poll, next_sweep]
1885 .into_iter()
1886 .chain(next_due(&state, &cfg.listen, &acts_on))
1887 .min()
1888 .unwrap_or(now + FLOOR_SECS);
1889 let wait = next.saturating_sub(clock::now_secs()).clamp(1, 3600);
1890 if proc::sleep_unless_interrupted(Duration::from_secs(wait)) {
1891 break Ending::Interrupted;
1892 }
1893 };
1894
1895 save(&state);
1896 drop(lock);
1897 log!(
1898 "listened for {}: {events_seen} event{}, {dispatches} dispatch{}",
1899 spend::clock(Duration::from_secs(
1900 clock::now_secs().saturating_sub(started)
1901 )),
1902 plural(events_seen),
1903 if dispatches == 1 { "" } else { "es" }
1904 );
1905 let writes = crate::cli::report_writes(repo);
1906 Ok(match ending {
1907 Ending::Interrupted => 130.max(writes),
1908 Ending::Once { failed: true } => 1.max(writes),
1909 Ending::Once { failed: false } | Ending::Done => writes,
1910 })
1911}
1912
1913fn read_state(repo: &Repo) -> ListenState {
1914 let path = repo.listen_state_path();
1915 match std::fs::read_to_string(&path) {
1916 Ok(text) => match serde_json::from_str::<ListenState>(&text) {
1917 Ok(state) => state,
1918 Err(e) => {
1919 logdim!("could not read {}: {e}. Starting fresh.", path.display());
1920 ListenState::default()
1921 }
1922 },
1923 Err(_) => ListenState::default(),
1924 }
1925}
1926
1927fn write_state(repo: &Repo, state: &ListenState) {
1928 let mut state = state.clone();
1929 state.version = STATE_VERSION;
1930 if let Err(e) = crate::repo::write_json_atomic(&repo.listen_state_path(), &state) {
1931 logdim!("could not write the listen state: {e}");
1932 }
1933}
1934
1935#[cfg(test)]
1936mod tests {
1937 use super::*;
1938 use crate::model::Author;
1939
1940 fn cfg() -> ListenCfg {
1941 ListenCfg::default()
1942 }
1943
1944 fn user(login: &str) -> Login {
1945 Login {
1946 login: login.into(),
1947 ..Login::default()
1948 }
1949 }
1950
1951 fn team(slug: &str) -> Login {
1952 Login {
1953 slug: slug.into(),
1954 ..Login::default()
1955 }
1956 }
1957
1958 fn thread(id: &str, reason: &str, kind: &str, number: i64, at: &str) -> Thread {
1959 Thread {
1960 id: id.into(),
1961 reason: reason.into(),
1962 updated_at: at.into(),
1963 subject: Subject {
1964 title: format!("thing {number}"),
1965 url: Some(format!(
1966 "https://api.github.com/repos/o/r/{}/{number}",
1967 if kind == "PullRequest" {
1968 "pulls"
1969 } else {
1970 "issues"
1971 }
1972 )),
1973 latest_comment_url: None,
1974 kind: kind.into(),
1975 },
1976 }
1977 }
1978
1979 fn row(number: i64, head: &str, author: &str, state: &str) -> PrSweepRow {
1980 PrSweepRow {
1981 number,
1982 title: format!("PR {number}"),
1983 head_ref_oid: head.into(),
1984 author: Some(Author {
1985 login: author.into(),
1986 }),
1987 state: state.into(),
1988 ..Default::default()
1989 }
1990 }
1991
1992 fn pending(kind: ItemKind, triggers: &[Trigger], now: u64) -> Pending {
1993 Pending {
1994 kind,
1995 first_seen: now,
1996 last_seen: now,
1997 triggers: triggers.iter().copied().collect(),
1998 who: Vec::new(),
1999 head_from: None,
2000 head_to: None,
2001 }
2002 }
2003
2004 fn open_pr() -> Facts {
2005 Facts {
2006 number: 7,
2007 open: true,
2008 ..Default::default()
2009 }
2010 }
2011
2012 fn response(status: u16, headers: &[(&str, &str)]) -> ApiResponse {
2013 ApiResponse {
2014 status,
2015 headers: headers
2016 .iter()
2017 .map(|(k, v)| (k.to_string(), v.to_string()))
2018 .collect(),
2019 body: String::new(),
2020 }
2021 }
2022
2023 #[test]
2026 fn the_poll_interval_is_the_largest_of_the_config_the_header_and_the_floor() {
2027 let mut c = cfg();
2028 assert_eq!(60, poll_interval(&c, None));
2029 assert_eq!(90, poll_interval(&c, Some("90")));
2030 assert_eq!(60, poll_interval(&c, Some("30")));
2031 assert_eq!(60, poll_interval(&c, Some("soon")));
2032 c.interval = 5;
2033 assert_eq!(15, poll_interval(&c, None));
2034 }
2035
2036 #[test]
2037 fn a_rate_limit_answer_says_how_long_to_wait_and_caps_it_at_an_hour() {
2038 let now = 1_000_000;
2039 assert_eq!(
2040 Some(120),
2041 rate_limit_wait(&response(429, &[("retry-after", "120")]), now)
2042 );
2043 assert_eq!(
2044 Some(600),
2045 rate_limit_wait(
2046 &response(
2047 403,
2048 &[
2049 ("x-ratelimit-remaining", "0"),
2050 ("x-ratelimit-reset", "1000600")
2051 ]
2052 ),
2053 now
2054 )
2055 );
2056 assert_eq!(
2057 Some(3600),
2058 rate_limit_wait(&response(403, &[("retry-after", "99999")]), now)
2059 );
2060 assert_eq!(
2061 Some(60),
2062 rate_limit_wait(&response(429, &[("retry-after", "1")]), now)
2063 );
2064 }
2065
2066 #[test]
2067 fn a_forbidden_answer_without_a_rate_limit_header_is_a_failure_not_a_wait() {
2068 assert_eq!(None, rate_limit_wait(&response(403, &[]), 0));
2069 assert_eq!(
2070 None,
2071 rate_limit_wait(&response(403, &[("x-ratelimit-remaining", "40")]), 0)
2072 );
2073 assert_eq!(
2074 None,
2075 rate_limit_wait(&response(200, &[("retry-after", "5")]), 0)
2076 );
2077 }
2078
2079 #[test]
2080 fn next_link_reads_the_next_page_and_nothing_else() {
2081 let mut headers = BTreeMap::new();
2082 headers.insert(
2083 "link".to_string(),
2084 "<https://api.github.com/x?page=2>; rel=\"next\", <https://api.github.com/x?page=9>; rel=\"last\"".to_string(),
2085 );
2086 assert_eq!(
2087 Some("https://api.github.com/x?page=2".to_string()),
2088 next_link(&headers)
2089 );
2090 headers.insert(
2091 "link".to_string(),
2092 "<https://api.github.com/x?page=1>; rel=\"prev\"".to_string(),
2093 );
2094 assert_eq!(None, next_link(&headers));
2095 assert_eq!(None, next_link(&BTreeMap::new()));
2096 }
2097
2098 #[test]
2101 fn a_thread_becomes_an_event_keyed_by_its_number_and_reason() {
2102 let mut state = ListenState::default();
2103 let threads = vec![
2104 thread(
2105 "1",
2106 "review_requested",
2107 "PullRequest",
2108 7,
2109 "2026-10-20T10:00:00Z",
2110 ),
2111 thread("2", "mention", "Issue", 42, "2026-10-20T10:01:00Z"),
2112 thread("3", "comment", "PullRequest", 8, "2026-10-20T10:02:00Z"),
2113 thread("4", "assign", "PullRequest", 9, "2026-10-20T10:03:00Z"),
2114 thread("5", "team_mention", "Issue", 43, "2026-10-20T10:04:00Z"),
2115 thread(
2116 "6",
2117 "state_change",
2118 "PullRequest",
2119 10,
2120 "2026-10-20T10:05:00Z",
2121 ),
2122 ];
2123 let events = events_from_threads(&threads, &mut state);
2124 let got: Vec<(i64, ItemKind, Trigger)> = events
2125 .iter()
2126 .map(|e| (e.number, e.kind, e.trigger))
2127 .collect();
2128 assert_eq!(
2129 vec![
2130 (7, ItemKind::Pr, Trigger::ReviewRequested),
2131 (42, ItemKind::Issue, Trigger::Mentioned),
2132 (8, ItemKind::Pr, Trigger::Replied),
2133 (9, ItemKind::Pr, Trigger::Assigned),
2134 (43, ItemKind::Issue, Trigger::Mentioned),
2135 (10, ItemKind::Pr, Trigger::Replied),
2136 ],
2137 got
2138 );
2139 }
2140
2141 #[test]
2142 fn a_thread_seen_at_the_same_update_time_is_not_an_event_twice() {
2143 let mut state = ListenState::default();
2144 let same = vec![thread(
2145 "1",
2146 "comment",
2147 "PullRequest",
2148 7,
2149 "2026-10-20T10:00:00Z",
2150 )];
2151 assert_eq!(1, events_from_threads(&same, &mut state).len());
2152 assert_eq!(0, events_from_threads(&same, &mut state).len());
2153 let moved = vec![thread(
2154 "1",
2155 "comment",
2156 "PullRequest",
2157 7,
2158 "2026-10-20T10:05:00Z",
2159 )];
2160 assert_eq!(1, events_from_threads(&moved, &mut state).len());
2161 }
2162
2163 #[test]
2164 fn a_thread_about_a_release_or_discussion_is_ignored() {
2165 let mut state = ListenState::default();
2166 let threads = vec![
2167 thread("1", "subscribed", "Release", 1, "2026-10-20T10:00:00Z"),
2168 thread("2", "mention", "Discussion", 2, "2026-10-20T10:01:00Z"),
2169 Thread {
2170 subject: Subject {
2171 url: None,
2172 kind: "PullRequest".into(),
2173 ..Default::default()
2174 },
2175 ..thread("3", "comment", "PullRequest", 3, "2026-10-20T10:02:00Z")
2176 },
2177 ];
2178 assert!(events_from_threads(&threads, &mut state).is_empty());
2179 assert_eq!(3, state.seen.len());
2182 }
2183
2184 #[test]
2185 fn the_watermark_advances_to_the_newest_update_seen() {
2186 let mut state = ListenState {
2187 watermark: "2026-10-20T09:00:00Z".into(),
2188 ..Default::default()
2189 };
2190 let threads = vec![
2191 thread("1", "comment", "PullRequest", 7, "2026-10-20T10:02:00Z"),
2192 thread("2", "comment", "PullRequest", 8, "2026-10-20T10:01:00Z"),
2193 ];
2194 events_from_threads(&threads, &mut state);
2195 assert_eq!("2026-10-20T10:02:00Z", state.watermark);
2196 }
2197
2198 fn no_record(_: i64) -> Option<String> {
2201 None
2202 }
2203
2204 #[test]
2205 fn a_sweep_notices_a_head_that_moved_and_says_it_once() {
2206 let mut state = ListenState::default();
2207 let named = BTreeSet::new();
2208 let first = events_from_sweep(
2209 &[row(7, "aaa", "alice", "OPEN")],
2210 "me",
2211 &mut state,
2212 &named,
2213 0,
2214 &no_record,
2215 true,
2216 );
2217 assert!(first.is_empty(), "{first:?}");
2218 let moved = events_from_sweep(
2219 &[row(7, "bbb", "alice", "OPEN")],
2220 "me",
2221 &mut state,
2222 &named,
2223 0,
2224 &no_record,
2225 true,
2226 );
2227 assert_eq!(1, moved.len());
2228 assert_eq!(Trigger::Pushed, moved[0].trigger);
2229 assert_eq!(Some(("aaa".to_string(), "bbb".to_string())), moved[0].head);
2230 let again = events_from_sweep(
2231 &[row(7, "bbb", "alice", "OPEN")],
2232 "me",
2233 &mut state,
2234 &named,
2235 0,
2236 &no_record,
2237 true,
2238 );
2239 assert!(again.is_empty(), "{again:?}");
2240 assert_eq!("alice", state.items[&7].author);
2241 }
2242
2243 #[test]
2244 fn the_first_sweep_treats_a_reviewed_pull_request_that_already_moved_as_due() {
2245 let mut state = ListenState::default();
2246 let record = |n: i64| (n == 7).then(|| "aaa".to_string());
2247 let events = events_from_sweep(
2248 &[row(7, "bbb", "alice", "OPEN"), row(8, "ccc", "bob", "OPEN")],
2249 "me",
2250 &mut state,
2251 &BTreeSet::new(),
2252 0,
2253 &record,
2254 true,
2255 );
2256 assert_eq!(1, events.len());
2257 assert_eq!(7, events[0].number);
2258 assert_eq!(Trigger::Pushed, events[0].trigger);
2259 }
2260
2261 #[test]
2262 fn a_sweep_notices_a_review_request_once_not_every_five_minutes() {
2263 let mut state = ListenState::default();
2264 let mut asked = row(7, "aaa", "alice", "OPEN");
2265 asked.review_requests = vec![user("ME")];
2266 asked.assignees = vec![user("me")];
2267 let rows = [asked];
2268 let first = events_from_sweep(
2269 &rows,
2270 "me",
2271 &mut state,
2272 &BTreeSet::new(),
2273 0,
2274 &no_record,
2275 true,
2276 );
2277 let triggers: BTreeSet<Trigger> = first.iter().map(|e| e.trigger).collect();
2278 assert_eq!(
2279 [Trigger::ReviewRequested, Trigger::Assigned]
2280 .into_iter()
2281 .collect::<BTreeSet<_>>(),
2282 triggers
2283 );
2284 let second = events_from_sweep(
2285 &rows,
2286 "me",
2287 &mut state,
2288 &BTreeSet::new(),
2289 0,
2290 &no_record,
2291 true,
2292 );
2293 assert!(second.is_empty(), "{second:?}");
2294 }
2295
2296 #[test]
2297 fn a_sweep_notices_a_review_requested_again_after_a_review() {
2298 let mut state = ListenState::default();
2299 let mut asked = row(7, "aaa", "alice", "OPEN");
2300 asked.review_requests = vec![user("me")];
2301 let asked = [asked];
2302 let reviewed = [row(7, "aaa", "alice", "OPEN")];
2303 let mut sweep = |rows: &[PrSweepRow]| -> Vec<Trigger> {
2304 events_from_sweep(
2305 rows,
2306 "me",
2307 &mut state,
2308 &BTreeSet::new(),
2309 0,
2310 &no_record,
2311 true,
2312 )
2313 .iter()
2314 .map(|e| e.trigger)
2315 .collect()
2316 };
2317 assert_eq!(vec![Trigger::ReviewRequested], sweep(&asked));
2318 assert!(sweep(&reviewed).is_empty());
2320 assert!(sweep(&reviewed).is_empty());
2321 assert_eq!(vec![Trigger::ReviewRequested], sweep(&asked));
2323 assert!(sweep(&asked).is_empty());
2324 }
2325
2326 #[test]
2327 fn a_sweep_drops_a_pull_request_that_closed_or_vanished() {
2328 let mut state = ListenState::default();
2329 let named = BTreeSet::new();
2330 events_from_sweep(
2331 &[row(7, "aaa", "alice", "OPEN"), row(8, "bbb", "bob", "OPEN")],
2332 "me",
2333 &mut state,
2334 &named,
2335 0,
2336 &no_record,
2337 true,
2338 );
2339 let events = events_from_sweep(
2340 &[row(7, "aaa", "alice", "MERGED")],
2341 "me",
2342 &mut state,
2343 &named,
2344 0,
2345 &no_record,
2346 true,
2347 );
2348 let closed: Vec<i64> = events
2349 .iter()
2350 .filter(|e| e.trigger == Trigger::Closed)
2351 .map(|e| e.number)
2352 .collect();
2353 assert_eq!(vec![7, 8], closed);
2354
2355 let mut state = ListenState::default();
2358 let named: BTreeSet<i64> = [7, 8].into_iter().collect();
2359 events_from_sweep(
2360 &[row(7, "aaa", "alice", "OPEN"), row(8, "bbb", "bob", "OPEN")],
2361 "me",
2362 &mut state,
2363 &named,
2364 0,
2365 &no_record,
2366 false,
2367 );
2368 let events = events_from_sweep(
2369 &[row(7, "aaa", "alice", "OPEN")],
2370 "me",
2371 &mut state,
2372 &named,
2373 0,
2374 &no_record,
2375 false,
2376 );
2377 assert!(events.is_empty(), "{events:?}");
2378 }
2379
2380 #[test]
2381 fn a_sweep_ignores_a_number_below_the_floor_unless_it_was_named() {
2382 let mut state = ListenState::default();
2383 let named: BTreeSet<i64> = [3].into_iter().collect();
2384 let record = |_: i64| Some("old".to_string());
2385 let events = events_from_sweep(
2386 &[
2387 row(3, "new", "alice", "OPEN"),
2388 row(4, "new", "alice", "OPEN"),
2389 row(500, "new", "alice", "OPEN"),
2390 ],
2391 "me",
2392 &mut state,
2393 &named,
2394 480,
2395 &record,
2396 true,
2397 );
2398 let numbers: Vec<i64> = events.iter().map(|e| e.number).collect();
2399 assert_eq!(vec![3, 500], numbers);
2400 assert!(!state.items.contains_key(&4));
2401 }
2402
2403 #[test]
2406 fn a_pull_request_below_the_floor_that_is_still_open_is_not_called_closed() {
2407 let mut state = ListenState::default();
2408 state.watched(4, ItemKind::Pr);
2409 let events = events_from_sweep(
2410 &[
2411 row(4, "new", "alice", "OPEN"),
2412 row(500, "new", "alice", "OPEN"),
2413 ],
2414 "me",
2415 &mut state,
2416 &BTreeSet::new(),
2417 480,
2418 &no_record,
2419 true,
2420 );
2421 assert!(
2422 events.iter().all(|e| e.trigger != Trigger::Closed),
2423 "{events:?}"
2424 );
2425 let gone = events_from_sweep(
2426 &[row(500, "new", "alice", "OPEN")],
2427 "me",
2428 &mut state,
2429 &BTreeSet::new(),
2430 480,
2431 &no_record,
2432 true,
2433 );
2434 assert_eq!(
2435 vec![(4, Trigger::Closed)],
2436 gone.iter()
2437 .map(|e| (e.number, e.trigger))
2438 .collect::<Vec<_>>()
2439 );
2440 }
2441
2442 #[test]
2445 fn a_review_request_nobody_is_still_making_is_only_an_update() {
2446 let threads = [thread(
2449 "1",
2450 "review_requested",
2451 "PullRequest",
2452 7,
2453 "2026-10-21T14:51:16Z",
2454 )];
2455 let events = events_from_threads(&threads, &mut ListenState::default());
2456 let mut waiting = BTreeMap::new();
2457 for e in &events {
2458 merge(&mut waiting, e, 0);
2459 }
2460 let reviewed = Facts {
2461 has_posted_record: true,
2462 ..open_pr()
2463 };
2464 let mut p = waiting.remove(&7).expect("pending");
2465 assert!(settle(&mut p, &reviewed));
2466 assert_eq!(
2467 [Trigger::Replied].into_iter().collect::<BTreeSet<_>>(),
2468 p.triggers
2469 );
2470 assert!(matches!(plan(&p, &reviewed, &cfg()), Action::Ignore(_)));
2471
2472 let mut p = pending(ItemKind::Pr, &[Trigger::ReviewRequested], 0);
2474 assert!(settle(&mut p, &open_pr()));
2475 assert!(matches!(plan(&p, &open_pr(), &cfg()), Action::Ignore(_)));
2476
2477 let asked = Facts {
2479 review_pending: true,
2480 ..open_pr()
2481 };
2482 let mut p = pending(ItemKind::Pr, &[Trigger::ReviewRequested], 0);
2483 assert!(!settle(&mut p, &asked));
2484 assert!(p.triggers.contains(&Trigger::ReviewRequested));
2485 assert_eq!(Action::Review, plan(&p, &asked, &cfg()));
2486
2487 let answered = Facts {
2490 has_posted_record: true,
2491 review_pending: true,
2492 request_answered: true,
2493 ..open_pr()
2494 };
2495 let mut p = pending(ItemKind::Pr, &[Trigger::ReviewRequested], 0);
2496 assert!(settle(&mut p, &answered));
2497 assert_eq!(
2498 [Trigger::Replied].into_iter().collect::<BTreeSet<_>>(),
2499 p.triggers
2500 );
2501 assert!(matches!(plan(&p, &answered, &cfg()), Action::Ignore(_)));
2502
2503 let mut p = pending(ItemKind::Pr, &[Trigger::Pushed], 0);
2505 assert!(!settle(&mut p, &open_pr()));
2506 assert_eq!(
2507 [Trigger::Pushed].into_iter().collect::<BTreeSet<_>>(),
2508 p.triggers
2509 );
2510 }
2511
2512 #[test]
2513 fn a_review_request_counts_for_the_viewer_or_a_team_the_viewer_is_in() {
2514 let teams = |team: &str| match team {
2515 "example/ours" => Some(true),
2516 "example/theirs" => Some(false),
2517 _ => None,
2518 };
2519 let pending = |requests: &[Login]| review_pending(requests, "me", &teams);
2520 assert!(!pending(&[]));
2521 assert!(pending(&[user("ME")]));
2522 assert!(!pending(&[user("alice")]));
2523 assert!(pending(&[team("example/ours")]));
2524 assert!(!pending(&[team("example/theirs"), user("alice")]));
2527 assert!(pending(&[team("example/theirs"), team("example/ours")]));
2528 assert!(pending(&[team("example/unknown")]));
2530 let never = |_: &str| -> Option<bool> { panic!("looked a team up") };
2532 assert!(review_pending(
2533 &[team("example/ours"), user("me")],
2534 "me",
2535 &never
2536 ));
2537 }
2538
2539 #[test]
2543 fn a_request_made_before_the_review_on_record_was_answered_by_it() {
2544 let teams = |team: &str| match team {
2545 "example/ours" => Some(true),
2546 _ => Some(false),
2547 };
2548 let made = |at: &str, who: Login| ReviewRequested { at: at.into(), who };
2549 let answered = |requests: &[ReviewRequested]| {
2550 request_answered(requests, "me", &teams, "2026-10-20T12:00:00Z")
2551 };
2552 assert!(answered(&[made("2026-10-20T11:00:00Z", user("me"))]));
2553 assert!(answered(&[made("2026-10-20T12:00:00Z", user("ME"))]));
2554 assert!(answered(&[made(
2555 "2026-10-20T11:00:00Z",
2556 team("example/ours")
2557 )]));
2558 assert!(!answered(&[
2560 made("2026-10-20T11:00:00Z", user("me")),
2561 made("2026-10-20T13:00:00Z", user("me")),
2562 ]));
2563 assert!(!answered(&[
2564 made("2026-10-20T11:00:00Z", user("me")),
2565 made("2026-10-20T13:00:00Z", team("example/ours")),
2566 ]));
2567 assert!(answered(&[
2569 made("2026-10-20T11:00:00Z", user("me")),
2570 made("2026-10-20T13:00:00Z", user("alice")),
2571 made("2026-10-20T13:00:00Z", team("example/theirs")),
2572 ]));
2573 assert!(!answered(&[]));
2575 assert!(!answered(&[made("2026-10-20T11:00:00Z", user("alice"))]));
2576 assert!(!answered(&[made("", user("me"))]));
2577 assert!(!request_answered(
2578 &[made("2026-10-20T11:00:00Z", user("me"))],
2579 "me",
2580 &teams,
2581 ""
2582 ));
2583 }
2584
2585 #[test]
2586 fn a_review_request_or_a_mention_on_a_reviewed_pull_request_is_a_recheck() {
2587 let f = Facts {
2588 has_posted_record: true,
2589 ..open_pr()
2590 };
2591 for triggers in [
2592 &[Trigger::ReviewRequested][..],
2593 &[Trigger::Mentioned],
2594 &[Trigger::Pushed, Trigger::Replied, Trigger::ReviewRequested],
2595 ] {
2596 assert_eq!(
2597 Action::Recheck(7),
2598 plan(&pending(ItemKind::Pr, triggers, 0), &f, &cfg()),
2599 "{triggers:?}"
2600 );
2601 }
2602 }
2603
2604 #[test]
2608 fn a_push_a_reply_or_an_assignment_on_a_reviewed_pull_request_asks_for_nothing() {
2609 let f = Facts {
2610 has_posted_record: true,
2611 ..open_pr()
2612 };
2613 for triggers in [
2614 &[Trigger::Pushed][..],
2615 &[Trigger::Replied],
2616 &[Trigger::Assigned],
2617 &[Trigger::Pushed, Trigger::Replied],
2618 ] {
2619 assert!(
2620 matches!(
2621 plan(&pending(ItemKind::Pr, triggers, 0), &f, &cfg()),
2622 Action::Ignore(_)
2623 ),
2624 "{triggers:?}"
2625 );
2626 }
2627 }
2628
2629 #[test]
2630 fn a_mention_from_before_the_review_on_record_is_only_an_update() {
2631 let reviewed = Facts {
2632 has_posted_record: true,
2633 ..open_pr()
2634 };
2635 let mut p = pending(ItemKind::Pr, &[Trigger::Mentioned], 0);
2636 assert!(settle_mention(&mut p, &reviewed));
2637 assert_eq!(
2638 [Trigger::Replied].into_iter().collect::<BTreeSet<_>>(),
2639 p.triggers
2640 );
2641 assert!(matches!(plan(&p, &reviewed, &cfg()), Action::Ignore(_)));
2642
2643 let asked = Facts {
2645 mentioned: true,
2646 ..reviewed.clone()
2647 };
2648 let mut p = pending(ItemKind::Pr, &[Trigger::Mentioned], 0);
2649 assert!(!settle_mention(&mut p, &asked));
2650 assert_eq!(Action::Recheck(7), plan(&p, &asked, &cfg()));
2651
2652 let mut p = pending(ItemKind::Pr, &[Trigger::Mentioned], 0);
2655 assert!(!settle_mention(&mut p, &open_pr()));
2656 assert_eq!(Action::Review, plan(&p, &open_pr(), &cfg()));
2657
2658 let linked = Facts {
2660 number: 42,
2661 open: true,
2662 linked_pr_with_record: Some(108),
2663 ..Default::default()
2664 };
2665 let mut p = pending(ItemKind::Issue, &[Trigger::Mentioned], 0);
2666 assert!(settle_mention(&mut p, &linked));
2667 assert!(matches!(plan(&p, &linked, &cfg()), Action::Ignore(_)));
2668 let asked = Facts {
2669 mentioned: true,
2670 ..linked
2671 };
2672 let mut p = pending(ItemKind::Issue, &[Trigger::Mentioned], 0);
2673 assert!(!settle_mention(&mut p, &asked));
2674 assert_eq!(Action::Recheck(108), plan(&p, &asked, &cfg()));
2675 }
2676
2677 #[test]
2681 fn an_issue_thread_moved_by_spars_own_answer_is_only_an_update() {
2682 let judged = Facts {
2683 number: 229,
2684 open: true,
2685 ..Default::default()
2686 };
2687 let mut p = pending(ItemKind::Issue, &[Trigger::Mentioned], 0);
2688 assert!(settle_mention(&mut p, &judged));
2689 assert_eq!(
2690 [Trigger::Replied].into_iter().collect::<BTreeSet<_>>(),
2691 p.triggers
2692 );
2693 assert!(matches!(plan(&p, &judged, &cfg()), Action::Ignore(_)));
2694
2695 let asked = Facts {
2696 asks_open: true,
2697 ..judged
2698 };
2699 let mut p = pending(ItemKind::Issue, &[Trigger::Mentioned], 0);
2700 assert!(!settle_mention(&mut p, &asked));
2701 assert_eq!(Action::Answer { draft: false }, plan(&p, &asked, &cfg()));
2702 }
2703
2704 #[test]
2705 fn a_review_request_or_an_assignment_on_an_unreviewed_pull_request_follows_the_config() {
2706 let f = open_pr();
2707 let mut c = cfg();
2708 assert_eq!(
2709 Action::Review,
2710 plan(
2711 &pending(ItemKind::Pr, &[Trigger::ReviewRequested], 0),
2712 &f,
2713 &c
2714 )
2715 );
2716 assert_eq!(
2717 Action::Review,
2718 plan(&pending(ItemKind::Pr, &[Trigger::Assigned], 0), &f, &c)
2719 );
2720 assert_eq!(
2721 Action::Review,
2722 plan(&pending(ItemKind::Pr, &[Trigger::Mentioned], 0), &f, &c)
2723 );
2724 c.on_review_requested = ReviewOrIgnore::Ignore;
2725 c.on_assigned = ReviewOrIgnore::Ignore;
2726 assert!(matches!(
2727 plan(
2728 &pending(ItemKind::Pr, &[Trigger::ReviewRequested], 0),
2729 &f,
2730 &c
2731 ),
2732 Action::Ignore(_)
2733 ));
2734 assert!(matches!(
2735 plan(&pending(ItemKind::Pr, &[Trigger::Assigned], 0), &f, &c),
2736 Action::Ignore(_)
2737 ));
2738 assert_eq!(
2740 Action::Review,
2741 plan(&pending(ItemKind::Pr, &[Trigger::Mentioned], 0), &f, &c)
2742 );
2743 }
2744
2745 #[test]
2746 fn a_push_to_an_unreviewed_pull_request_is_ignored_unless_it_was_named() {
2747 let f = open_pr();
2748 assert!(matches!(
2749 plan(&pending(ItemKind::Pr, &[Trigger::Pushed], 0), &f, &cfg()),
2750 Action::Ignore(_)
2751 ));
2752 assert!(matches!(
2753 plan(&pending(ItemKind::Pr, &[Trigger::Replied], 0), &f, &cfg()),
2754 Action::Ignore(_)
2755 ));
2756 let named = Facts {
2757 named: true,
2758 ..open_pr()
2759 };
2760 assert_eq!(
2761 Action::Review,
2762 plan(
2763 &pending(ItemKind::Pr, &[Trigger::Pushed], 0),
2764 &named,
2765 &cfg()
2766 )
2767 );
2768 assert_eq!(
2769 Action::Review,
2770 plan(
2771 &pending(ItemKind::Pr, &[Trigger::Replied], 0),
2772 &named,
2773 &cfg()
2774 )
2775 );
2776 }
2777
2778 #[test]
2779 fn your_own_pull_request_follows_on_own_pr_and_a_push_to_it_is_ignored() {
2780 let f = Facts {
2781 author_is_viewer: true,
2782 has_posted_record: true,
2783 ..open_pr()
2784 };
2785 let mut c = cfg();
2786 assert_eq!(
2787 Action::Checkin { reply_only: false },
2788 plan(&pending(ItemKind::Pr, &[Trigger::Replied], 0), &f, &c)
2789 );
2790 assert_eq!(
2791 Action::Checkin { reply_only: false },
2792 plan(
2793 &pending(ItemKind::Pr, &[Trigger::ReviewRequested], 0),
2794 &f,
2795 &c
2796 )
2797 );
2798 assert!(matches!(
2799 plan(&pending(ItemKind::Pr, &[Trigger::Pushed], 0), &f, &c),
2800 Action::Ignore(_)
2801 ));
2802 c.on_own_pr = OwnPr::Reply;
2803 assert_eq!(
2804 Action::Checkin { reply_only: true },
2805 plan(&pending(ItemKind::Pr, &[Trigger::Mentioned], 0), &f, &c)
2806 );
2807 c.on_own_pr = OwnPr::Ignore;
2808 assert!(matches!(
2809 plan(&pending(ItemKind::Pr, &[Trigger::Replied], 0), &f, &c),
2810 Action::Ignore(_)
2811 ));
2812 }
2813
2814 #[test]
2815 fn a_pull_request_in_custody_belongs_to_resume_and_a_capped_one_waits_for_a_person() {
2816 let custody = Facts {
2817 has_custody_state: true,
2818 ..open_pr()
2819 };
2820 assert_eq!(
2821 Action::BelongsToResume,
2822 plan(
2823 &pending(ItemKind::Pr, &[Trigger::Pushed], 0),
2824 &custody,
2825 &cfg()
2826 )
2827 );
2828 let capped = Facts {
2829 has_posted_record: true,
2830 recheck_capped: true,
2831 ..open_pr()
2832 };
2833 assert_eq!(
2834 Action::Capped(7),
2835 plan(
2836 &pending(ItemKind::Pr, &[Trigger::ReviewRequested], 0),
2837 &capped,
2838 &cfg()
2839 )
2840 );
2841 assert!(matches!(
2842 plan(
2843 &pending(ItemKind::Pr, &[Trigger::Pushed], 0),
2844 &capped,
2845 &cfg()
2846 ),
2847 Action::Ignore(_)
2848 ));
2849 }
2850
2851 #[test]
2855 fn an_issue_mention_keeps_to_its_pull_requests_cap_and_owner() {
2856 let linked = Facts {
2857 number: 42,
2858 open: true,
2859 linked_pr_with_record: Some(108),
2860 ..Default::default()
2861 };
2862 let capped = Facts {
2863 recheck_capped: true,
2864 ..linked.clone()
2865 };
2866 assert_eq!(
2867 Action::Capped(108),
2868 plan(
2869 &pending(ItemKind::Issue, &[Trigger::Mentioned], 0),
2870 &capped,
2871 &cfg()
2872 )
2873 );
2874 let own = Facts {
2875 author_is_viewer: true,
2876 ..linked
2877 };
2878 assert!(matches!(
2879 plan(
2880 &pending(ItemKind::Issue, &[Trigger::Mentioned], 0),
2881 &own,
2882 &cfg()
2883 ),
2884 Action::Ignore(_)
2885 ));
2886 }
2887
2888 #[test]
2889 fn an_issue_mention_without_a_pull_request_is_answered_and_with_one_rechecks_it() {
2890 let issue = Facts {
2891 number: 42,
2892 open: true,
2893 ..Default::default()
2894 };
2895 assert!(matches!(
2897 plan(
2898 &pending(ItemKind::Issue, &[Trigger::Mentioned], 0),
2899 &issue,
2900 &cfg()
2901 ),
2902 Action::Ignore(_)
2903 ));
2904 let asks = Facts {
2905 asks_open: true,
2906 ..issue.clone()
2907 };
2908 assert_eq!(
2909 Action::Answer { draft: false },
2910 plan(
2911 &pending(ItemKind::Issue, &[Trigger::Mentioned], 0),
2912 &asks,
2913 &cfg()
2914 )
2915 );
2916 let mut c = cfg();
2917 c.on_mention = OnMention::Draft;
2918 assert_eq!(
2919 Action::Answer { draft: true },
2920 plan(
2921 &pending(ItemKind::Issue, &[Trigger::Mentioned], 0),
2922 &asks,
2923 &c
2924 )
2925 );
2926 c.on_mention = OnMention::Ignore;
2927 assert_eq!(
2928 Action::Ignore("on_mention is ignore"),
2929 plan(
2930 &pending(ItemKind::Issue, &[Trigger::Mentioned], 0),
2931 &asks,
2932 &c
2933 )
2934 );
2935 let capped = Facts {
2936 answers_capped: true,
2937 ..asks.clone()
2938 };
2939 assert_eq!(
2940 Action::AnswersCapped,
2941 plan(
2942 &pending(ItemKind::Issue, &[Trigger::Mentioned], 0),
2943 &capped,
2944 &cfg()
2945 )
2946 );
2947 assert!(matches!(
2948 plan(
2949 &pending(ItemKind::Issue, &[Trigger::Replied], 0),
2950 &issue,
2951 &cfg()
2952 ),
2953 Action::Ignore(_)
2954 ));
2955 let linked = Facts {
2956 linked_pr_with_record: Some(108),
2957 ..issue.clone()
2958 };
2959 assert_eq!(
2960 Action::Recheck(108),
2961 plan(
2962 &pending(ItemKind::Issue, &[Trigger::Mentioned], 0),
2963 &linked,
2964 &cfg()
2965 )
2966 );
2967 assert!(matches!(
2971 plan(
2972 &pending(ItemKind::Issue, &[Trigger::Assigned], 0),
2973 &linked,
2974 &cfg()
2975 ),
2976 Action::Ignore(_)
2977 ));
2978 let named = Facts {
2979 named: true,
2980 ..linked
2981 };
2982 assert!(matches!(
2983 plan(
2984 &pending(ItemKind::Issue, &[Trigger::Replied], 0),
2985 &named,
2986 &cfg()
2987 ),
2988 Action::Ignore(_)
2989 ));
2990 }
2991
2992 #[test]
2993 fn a_closed_item_is_dropped_and_one_below_the_floor_is_ignored_unless_named() {
2994 let closed = Facts {
2995 open: false,
2996 has_posted_record: true,
2997 ..open_pr()
2998 };
2999 assert_eq!(
3000 Action::Drop,
3001 plan(
3002 &pending(ItemKind::Pr, &[Trigger::Pushed], 0),
3003 &closed,
3004 &cfg()
3005 )
3006 );
3007 let f = Facts {
3008 has_posted_record: true,
3009 ..open_pr()
3010 };
3011 assert_eq!(
3012 Action::Drop,
3013 plan(
3014 &pending(ItemKind::Pr, &[Trigger::Pushed, Trigger::Closed], 0),
3015 &f,
3016 &cfg()
3017 )
3018 );
3019 let low = Facts {
3020 below_floor: true,
3021 has_posted_record: true,
3022 ..open_pr()
3023 };
3024 assert!(matches!(
3025 plan(&pending(ItemKind::Pr, &[Trigger::Pushed], 0), &low, &cfg()),
3026 Action::Ignore(_)
3027 ));
3028 let named_low = Facts { named: true, ..low };
3029 assert_eq!(
3030 Action::Recheck(7),
3031 plan(
3032 &pending(ItemKind::Pr, &[Trigger::ReviewRequested], 0),
3033 &named_low,
3034 &cfg()
3035 )
3036 );
3037 }
3038
3039 #[test]
3042 fn an_item_is_due_once_quiet_for_the_settle_time_or_at_max_wait_anyway() {
3043 let c = cfg();
3044 let mut p = pending(ItemKind::Pr, &[Trigger::Pushed], 1000);
3045 assert!(!due(&p, 1000, &c, false));
3046 assert!(!due(&p, 1000 + c.settle - 1, &c, false));
3047 assert!(due(&p, 1000 + c.settle, &c, false));
3048 p.last_seen = 1000 + c.max_wait - 30;
3050 assert!(!due(&p, 1000 + c.max_wait - 1, &c, false));
3051 assert!(due(&p, 1000 + c.max_wait, &c, false));
3052 assert!(due(&p, 1000, &c, true), "--once waits for nothing");
3053 }
3054
3055 #[test]
3056 fn events_merge_into_one_pending_record_per_item_and_the_next_due_is_the_earliest() {
3057 let c = cfg();
3058 let mut pending = BTreeMap::new();
3059 let push = Event {
3060 number: 7,
3061 kind: ItemKind::Pr,
3062 trigger: Trigger::Pushed,
3063 who: None,
3064 head: Some(("aaa".into(), "bbb".into())),
3065 commits: None,
3066 from_thread: false,
3067 };
3068 merge(&mut pending, &push, 1000);
3069 let reply = Event {
3070 trigger: Trigger::Replied,
3071 who: Some("alice".into()),
3072 head: Some(("bbb".into(), "ccc".into())),
3073 ..push.clone()
3074 };
3075 merge(&mut pending, &reply, 1060);
3076 let other = Event {
3077 number: 8,
3078 ..push.clone()
3079 };
3080 merge(&mut pending, &other, 1100);
3081 assert_eq!(2, pending.len());
3082 let p = &pending[&7];
3083 assert_eq!(1000, p.first_seen);
3084 assert_eq!(1060, p.last_seen);
3085 assert_eq!(Some("aaa".to_string()), p.head_from);
3086 assert_eq!(Some("ccc".to_string()), p.head_to);
3087 assert_eq!(vec!["alice".to_string()], p.who);
3088 assert_eq!("a push and a reply", p.because());
3089 let mut state = ListenState {
3090 pending,
3091 ..Default::default()
3092 };
3093 let all = |_: i64| true;
3094 assert_eq!(Some(1060 + c.settle), next_due(&state, &c, &all));
3095 assert_eq!(None, next_due(&ListenState::default(), &c, &all));
3096
3097 state.watched(7, ItemKind::Pr).retry_after = 5000;
3100 assert_eq!(Some(1100 + c.settle), next_due(&state, &c, &all));
3101 state.watched(8, ItemKind::Pr).retry_after = 6000;
3102 assert_eq!(Some(5000), next_due(&state, &c, &all));
3103 assert_eq!(Some(6000), next_due(&state, &c, &|n| n == 8));
3105 }
3106
3107 #[test]
3110 fn since_accepts_minutes_hours_days_and_combinations() {
3111 assert_eq!(Duration::from_secs(1800), parse_duration("30m").unwrap());
3112 assert_eq!(Duration::from_secs(7200), parse_duration("2h").unwrap());
3113 assert_eq!(Duration::from_secs(86_400), parse_duration("1d").unwrap());
3114 assert_eq!(
3115 Duration::from_secs(5400),
3116 parse_duration(" 1h30m ").unwrap()
3117 );
3118 assert_eq!(Duration::from_secs(45), parse_duration("45s").unwrap());
3119 }
3120
3121 #[test]
3122 fn since_rejects_a_bare_number_and_says_what_it_takes() {
3123 for bad in ["", "30", "2 hours", "h", "1w", "1h30"] {
3124 let err = parse_duration(bad).unwrap_err().to_string();
3125 assert!(err.contains("like 30m, 2h or 1d"), "{bad:?}: {err}");
3126 }
3127 }
3128
3129 #[test]
3130 fn listen_state_round_trips_through_json_and_old_seen_entries_are_pruned() {
3131 let mut state = ListenState {
3132 version: STATE_VERSION,
3133 watermark: "2026-10-20T10:00:00Z".into(),
3134 last_modified: Some("Tue, 20 Oct 2026 10:00:00 GMT".into()),
3135 ..Default::default()
3136 };
3137 state
3138 .seen
3139 .insert("old".into(), "2026-10-01T10:00:00Z".into());
3140 state
3141 .seen
3142 .insert("new".into(), "2026-10-20T09:00:00Z".into());
3143 state.seen.insert("junk".into(), "yesterday".into());
3144 state.watched(7, ItemKind::Pr).head = Some("aaa".into());
3145 state
3146 .pending
3147 .insert(7, pending(ItemKind::Pr, &[Trigger::Pushed], 5));
3148 let text = serde_json::to_string_pretty(&state).unwrap();
3149 assert!(text.contains("\"pushed\""), "{text}");
3150 assert!(text.contains("\"kind\": \"pr\""), "{text}");
3151 let back: ListenState = serde_json::from_str(&text).unwrap();
3152 assert_eq!(state, back);
3153
3154 let now = clock::parse_rfc3339("2026-10-20T10:00:00Z").unwrap();
3155 prune_seen(&mut state, now);
3156 let left: Vec<&String> = state.seen.keys().collect();
3157 assert_eq!(vec!["junk", "new"], left);
3158 }
3159
3160 #[test]
3161 fn gh_failures_back_off_from_a_minute_to_fifteen_and_dispatches_from_fifteen_to_six_hours() {
3162 assert_eq!(60, backoff_secs(1));
3163 assert_eq!(120, backoff_secs(2));
3164 assert_eq!(480, backoff_secs(4));
3165 assert_eq!(900, backoff_secs(5));
3166 assert_eq!(900, backoff_secs(40));
3167 assert_eq!(900, dispatch_backoff_secs(1));
3168 assert_eq!(3600, dispatch_backoff_secs(3));
3169 assert_eq!(6 * 3600, dispatch_backoff_secs(9));
3170 }
3171
3172 #[test]
3176 fn a_head_that_moved_runs_again_once_it_settles_and_a_failure_backs_off() {
3177 let c = cfg();
3178 for failures in [1, 2, 4] {
3179 assert_eq!(c.settle, retry_wait(true, failures, &c));
3180 assert_eq!(
3181 dispatch_backoff_secs(failures),
3182 retry_wait(false, failures, &c)
3183 );
3184 }
3185 let quick = ListenCfg { settle: 1, ..cfg() };
3186 assert_eq!(FLOOR_SECS, retry_wait(true, 1, &quick));
3187 }
3188
3189 #[test]
3193 fn giving_up_holds_what_the_item_was_waiting_on_until_it_changes() {
3194 let mut state = ListenState::default();
3195 let asked = Event {
3196 number: 7,
3197 kind: ItemKind::Pr,
3198 trigger: Trigger::ReviewRequested,
3199 who: None,
3200 head: None,
3201 commits: None,
3202 from_thread: true,
3203 };
3204 state.note(&asked, 0);
3205 state.give_up(7, ItemKind::Pr);
3206 assert!(state.pending.is_empty(), "{state:?}");
3207 assert_eq!(
3208 [Trigger::ReviewRequested]
3209 .into_iter()
3210 .collect::<BTreeSet<_>>(),
3211 state.items[&7].held
3212 );
3213
3214 let shown = Event {
3215 trigger: Trigger::Pushed,
3216 head: Some(("aaa".into(), "bbb".into())),
3217 from_thread: false,
3218 ..asked.clone()
3219 };
3220 state.note(&shown, 600);
3221 let p = &state.pending[&7];
3222 assert_eq!(
3223 [Trigger::Pushed, Trigger::ReviewRequested]
3224 .into_iter()
3225 .collect::<BTreeSet<_>>(),
3226 p.triggers
3227 );
3228 assert!(state.items[&7].held.is_empty(), "{state:?}");
3229 let asked_now = Facts {
3230 review_pending: true,
3231 ..open_pr()
3232 };
3233 assert_eq!(Action::Review, plan(p, &asked_now, &cfg()));
3234 state.pending.clear();
3236 state.note(&shown, 700);
3237 assert_eq!(
3238 [Trigger::Pushed].into_iter().collect::<BTreeSet<_>>(),
3239 state.pending[&7].triggers
3240 );
3241 }
3242
3243 #[test]
3244 fn number_from_url_reads_the_last_segment() {
3245 assert_eq!(
3246 Some(108),
3247 number_from_url("https://api.github.com/repos/o/r/pulls/108")
3248 );
3249 assert_eq!(
3250 Some(42),
3251 number_from_url("https://api.github.com/repos/o/r/issues/42/")
3252 );
3253 assert_eq!(
3254 None,
3255 number_from_url("https://api.github.com/repos/o/r/pulls")
3256 );
3257 }
3258
3259 #[cfg(unix)]
3260 #[test]
3261 fn a_held_lock_refuses_a_second_listener_however_old_and_a_left_file_is_taken_over() {
3262 let dir = std::env::temp_dir().join(format!(
3263 "spar-listen-lock-{}-{}",
3264 std::process::id(),
3265 clock::now_secs()
3266 ));
3267 std::fs::create_dir_all(&dir).unwrap();
3268 let path = dir.join("listen.lock");
3269
3270 let held = Lock::at(path.clone(), 1000).expect("first lock");
3273 for now in [1000, 1000 + 86_400] {
3274 let err = Lock::at(path.clone(), now).unwrap_err().to_string();
3275 assert!(err.contains("already watching"), "{err}");
3276 assert!(err.contains(&std::process::id().to_string()), "{err}");
3277 }
3278 assert!(path.exists(), "a refused listener leaves the lock alone");
3279 drop(held);
3280 assert!(!path.exists(), "dropping the lock removes the file");
3281
3282 std::fs::write(
3285 &path,
3286 serde_json::to_string(&LockFile {
3287 pid: std::process::id(),
3288 started: "x".into(),
3289 })
3290 .unwrap(),
3291 )
3292 .unwrap();
3293 let taken = Lock::at(path.clone(), 1000).expect("a left file is taken over");
3294 drop(taken);
3295 let _ = std::fs::remove_dir_all(&dir);
3296 }
3297}