1use std::path::{Path, PathBuf};
14
15use serde_json::{json, Map, Value};
16
17use std::collections::BTreeSet;
18
19use onevcs::rules::{Approvals, Drafts};
20use onevcs::{
21 ChangeRequest, ChangeSpec, Check, CheckState, DraftReason, Error, EventKind, FailureKind,
22 HeldBy, Holding, Hosting, Identity, Landed, Lifecycle, MergeOutcome, MergePolicy,
23 PreservedBranch, Provenance, Publication, PublishOutcome, PublishRequest, Recoverable,
24 RemoteHost, Result, Scope, Session, SessionRecord, SessionRequest, SessionToken, Sha, Vcs,
25};
26
27use crate::events::{self, Emission};
28use crate::remote::DEFAULT_HOST;
29use crate::state::{self, VcsState};
30use crate::store::{FileStore, MemoryStore, Store};
31
32pub const DEFAULT_BASE: &str = "main";
37
38pub const DEFAULT_PUBLICATION: MergePolicy = MergePolicy::ChangeOpen;
43
44pub const DEFAULT_APPROVALS: Approvals = Approvals::Required;
47
48#[derive(Debug)]
53pub struct Repository<T> {
54 store: T,
55 root: PathBuf,
56 trees: Trees,
57}
58
59#[derive(Debug, Clone, Copy, PartialEq, Eq)]
61enum Trees {
62 Named,
65 Created,
67}
68
69pub type MemoryVcs = Repository<MemoryStore<VcsState>>;
76
77pub type FileVcs = Repository<FileStore<VcsState>>;
83
84impl MemoryVcs {
85 pub fn new() -> Self {
87 Self::seeded(VcsState::default())
88 }
89
90 pub fn seeded(state: VcsState) -> Self {
92 Self {
93 store: MemoryStore::new(state),
94 root: std::env::temp_dir().join("onevcs-testing-memory"),
95 trees: Trees::Named,
96 }
97 }
98
99 pub fn state(&self) -> VcsState {
101 self.store
102 .snapshot()
103 .expect("an in-memory store always answers")
104 }
105}
106
107impl Default for MemoryVcs {
108 fn default() -> Self {
109 Self::new()
110 }
111}
112
113impl FileVcs {
114 pub fn create(path: impl Into<PathBuf>) -> Result<Self> {
121 Self::over(FileStore::attach(path, &VcsState::default())?)
122 }
123
124 pub fn seeded(path: impl Into<PathBuf>, state: VcsState) -> Result<Self> {
127 Self::over(FileStore::replace(path, &state)?)
128 }
129
130 fn over(store: FileStore<VcsState>) -> Result<Self> {
131 let root = store
132 .path()
133 .parent()
134 .filter(|parent| !parent.as_os_str().is_empty())
135 .unwrap_or_else(|| Path::new("."))
136 .join("worktrees");
137 Ok(Self {
138 store,
139 root,
140 trees: Trees::Created,
141 })
142 }
143
144 pub fn state(&self) -> Result<VcsState> {
146 self.store.snapshot()
147 }
148}
149
150impl<T: Store<VcsState>> Vcs for Repository<T> {
151 fn resolve_identity(&self, origin_or_path: &str) -> Result<Identity> {
152 self.store.with(|state| {
153 state::identity_of(state, origin_or_path)
154 .cloned()
155 .ok_or_else(|| Error::Invalid {
156 reason: format!(
157 "{origin_or_path:?} does not name a repository this provider knows; {}",
158 state::known(state)
159 ),
160 })
161 })
162 }
163
164 fn open_session(&self, req: SessionRequest) -> Result<Session> {
165 let root = self.root.clone();
166 let (session, emission) = self.store.with(|state| {
167 let identity = state::identity_of(state, &req.repo)
168 .cloned()
169 .ok_or_else(|| Error::Invalid {
170 reason: format!(
171 "{:?} does not name a repository this provider knows; {}",
172 req.repo,
173 state::known(state)
174 ),
175 })?;
176 let token = SessionToken(format!("s-testing-{}", state.sessions.len() + 1));
179 let run_root = root.join(&token.0);
180 let base = req.base.clone().unwrap_or_else(|| DEFAULT_BASE.to_owned());
185 state::named_branch(&base, "the base")?;
186 for (key, value) in &req.labels {
187 state::label_pair(key, value)?;
188 }
189 let session = Session {
195 worktree: run_root.join("worktree"),
196 branch: state::requested_branch(&req, &token)?,
197 base,
198 token: token.clone(),
199 conflict: None,
200 };
201 state.sessions.push(session.clone());
202 state
203 .session_identities
204 .insert(token.clone(), identity.origin.clone());
205 if !req.labels.is_empty() {
206 state
207 .session_labels
208 .insert(token.clone(), req.labels.clone());
209 }
210 let emission = Emission {
211 stream: token.0.clone(),
212 identity: Some(identity.origin.clone()),
213 kind: EventKind::SessionOpened,
214 payload: object(json!({
215 "token": token.0,
216 "identity": identity.origin,
217 "branch": session.branch,
218 "base": session.base,
219 "worktree": session.worktree.display().to_string(),
220 "clone": run_root.join("clone").display().to_string(),
223 "execution_checkout": run_root.join("checkout").display().to_string(),
224 "publication_checkout": run_root.join("checkout").display().to_string(),
225 "placement": {"kind": "run-root"},
229 })),
230 };
231 Ok((session, emission))
232 })?;
233 if self.trees == Trees::Created {
234 std::fs::create_dir_all(&session.worktree).map_err(|e| Error::Invalid {
235 reason: format!("cannot create {}: {e}", session.worktree.display()),
236 })?;
237 }
238 events::emit(&emission);
239 Ok(session)
240 }
241
242 fn adopt_session(&self, token: SessionToken) -> Result<Session> {
243 self.store.with(|state| {
244 state::session_of(state, &token)
245 .cloned()
246 .ok_or_else(|| Error::Invalid {
247 reason: format!(
248 "no session {:?} is open; `onevcs session open` prints a token",
249 token.0
250 ),
251 })
252 })
253 }
254
255 fn preserve(&self, s: &Session, provenance: Provenance) -> Result<PreservedBranch> {
256 let (branch, emission) = self.store.with(|state| {
257 let identity = state::identity_for(state, &s.token)?;
258 let branch = PreservedBranch {
259 branch: s.branch.clone(),
260 base: s.base.clone(),
261 provenance,
262 change_url: None,
263 change_base: None,
264 };
265 let row = Recoverable {
266 identity: identity.clone(),
267 branch: branch.clone(),
268 checkout: s.worktree.clone(),
269 tip: None,
270 stopped_because: format!("session {} was left open", s.token.0),
271 recover_command: recover_command(&s.branch, &s.worktree, provenance),
272 landed: Landed::No,
277 held_by: None,
281 net_negative: None,
282 session: Some(s.token.clone()),
285 labels: state
286 .session_labels
287 .get(&s.token)
288 .cloned()
289 .unwrap_or_default(),
290 on_origin: None,
295 retirement: None,
300 };
301 state.preserved.retain(|kept| {
305 kept.identity != row.identity || kept.branch.branch != row.branch.branch
306 });
307 state.preserved.push(row);
308 let emission = Emission {
309 stream: s.token.0.clone(),
310 identity: None,
316 kind: EventKind::CommitPreserved,
317 payload: object(json!({
318 "branch": s.branch,
319 "sha": events::stable_sha(&[&s.token.0, &s.branch, spell(provenance)]),
320 "provenance": spell(provenance),
321 })),
322 };
323 Ok((branch, emission))
324 })?;
325 events::emit(&emission);
326 Ok(branch)
327 }
328
329 fn session(&self, token: &SessionToken) -> Result<SessionRecord> {
330 self.store.with(|state| {
331 let session = state::session_of(state, token)
332 .cloned()
333 .ok_or_else(|| unknown_session(token))?;
334 let identity = state::identity_for(state, token)?;
335 let provenance = state
339 .preserved
340 .iter()
341 .find(|row| row.identity == identity && row.branch.branch == session.branch)
342 .map_or(Provenance::Complete, |row| row.branch.provenance);
343 Ok(SessionRecord {
344 lifecycle: if state.closed_sessions.contains(token) {
345 Lifecycle::Closed
346 } else {
347 Lifecycle::Open
348 },
349 session,
350 identity,
351 provenance,
352 retried_by: None,
356 })
357 })
358 }
359
360 fn close_session(&self, token: &SessionToken) -> Result<Session> {
361 self.store.with(|state| {
362 let session = state::session_of(state, token)
363 .cloned()
364 .ok_or_else(|| unknown_session(token))?;
365 let emission = Emission {
366 stream: token.0.clone(),
367 identity: None,
371 kind: EventKind::SessionClosed,
372 payload: object(json!({"token": token.0, "branch": session.branch})),
373 };
374 events::emit(&emission);
378 state.closed_sessions.insert(token.clone());
379 Ok(session)
380 })
381 }
382
383 fn publish(
396 &self,
397 token: &SessionToken,
398 request: &PublishRequest,
399 hosting: &dyn Hosting,
400 ) -> Result<Publication> {
401 let (publication, emissions) = self.store.with(|state| {
402 let session = state::session_of(state, token)
403 .cloned()
404 .ok_or_else(|| unknown_session(token))?;
405 let identity = state::identity_for(state, token)?;
406 if let Some(reason) = &request.draft {
411 reason.checked()?;
412 }
413 let resolved = state.policy.unwrap_or(DEFAULT_PUBLICATION);
414 let policy = match request.policy {
415 Some(requested) => resolved.narrow(requested)?,
416 None => resolved,
417 };
418 let published = |outcome, emissions| {
419 (
420 Publication {
421 session: token.clone(),
422 branch: session.branch.clone(),
423 policy,
424 outcome,
425 },
426 emissions,
427 )
428 };
429 if request.defer_checks || request.adopt.is_some() {
433 let approvals = state.approvals.unwrap_or(DEFAULT_APPROVALS);
434 let adopt = deferred(
435 state, &session, &identity, request, policy, approvals, token,
436 )?;
437 let number = adopt
438 .path_segments()
439 .and_then(|mut segments| segments.nth(3))
440 .unwrap_or_default()
441 .to_owned();
442 let emission = Emission {
443 stream: token.0.clone(),
444 identity: Some(identity.clone()),
445 kind: EventKind::ChangeOpened,
446 payload: object(json!({
447 "url": adopt.to_string(),
448 "host": "github",
449 "id": number,
450 "base": session.base,
451 })),
452 };
453 let (publication, emissions) = published(
454 PublishOutcome::ChangeReviewDraftUnchecked(adopt),
455 vec![emission],
456 );
457 state.publications.push(publication.clone());
458 return Ok((publication, emissions));
459 }
460 if state.publications.iter().any(|earlier| {
466 earlier.session == *token && matches!(earlier.outcome, PublishOutcome::Merged(_))
467 }) {
468 let (publication, emissions) =
469 published(PublishOutcome::NothingToPublish, Vec::new());
470 state.publications.push(publication.clone());
471 return Ok((publication, emissions));
472 }
473
474 let (outcome, emissions) = if policy == MergePolicy::LocalDirect {
475 match &request.draft {
480 Some(reason) => (
481 failed(&Error::Invalid {
482 reason: format!(
483 "{branch:?} was asked to publish as {asked}, and this identity \
484 publishes with local-direct, which opens no change request at \
485 all",
486 branch = session.branch,
487 asked = asked_for(reason),
488 ),
489 }),
490 Vec::new(),
491 ),
492 None => record_local_landing(&identity, &session, token),
493 }
494 } else {
495 match slug(&identity) {
496 Some(slug) => {
497 let mut emissions = Vec::new();
500 let outcome = match publish_as_change(
501 hosting,
502 &Publishing {
503 slug: &slug,
504 identity: &identity,
505 session: &session,
506 policy,
507 approvals: state.approvals.unwrap_or(DEFAULT_APPROVALS),
508 drafts: state.drafts.unwrap_or_default(),
509 request,
510 token,
511 },
512 &mut emissions,
513 ) {
514 Ok(outcome) => outcome,
515 Err(error) => failed(&error),
519 };
520 (outcome, emissions)
521 }
522 None => (refusal(&identity), Vec::new()),
523 }
524 };
525 let (publication, emissions) = published(outcome, emissions);
526 state.publications.push(publication.clone());
527 Ok((publication, emissions))
528 })?;
529 for emission in &emissions {
530 events::emit(emission);
531 }
532 Ok(publication)
533 }
534
535 fn preserved(&self, scope: Scope) -> Result<Vec<Recoverable>> {
542 self.recoverable(scope)
543 }
544
545 fn recoverable(&self, scope: Scope) -> Result<Vec<Recoverable>> {
546 self.store.with(|state| {
547 let wanted = match &scope {
548 Scope::All => None,
549 Scope::Repo(repo) => Some(
550 state::identity_of(state, repo)
551 .map(|identity| identity.origin.clone())
552 .ok_or_else(|| Error::Invalid {
553 reason: format!(
554 "{repo:?} does not name a repository this provider knows; {}",
555 state::known(state)
556 ),
557 })?,
558 ),
559 };
560 Ok(state
562 .preserved
563 .iter()
564 .rev()
565 .filter(|row| wanted.as_ref().is_none_or(|key| *key == row.identity))
566 .cloned()
567 .map(|row| held(state, row))
568 .collect())
569 })
570 }
571}
572
573fn held(state: &VcsState, row: Recoverable) -> Recoverable {
583 if row.held_by.is_some() {
584 return row;
585 }
586 let holder = state.sessions.iter().find(|session| {
587 session.branch == row.branch.branch
588 && state.session_identities.get(&session.token) == Some(&row.identity)
589 && !state.closed_sessions.contains(&session.token)
590 });
591 Recoverable {
592 held_by: holder.map(|session| HeldBy {
593 token: Some(session.token.clone()),
594 worktree: session.worktree.clone(),
595 holding: Holding::OwnerRunning,
596 }),
597 ..row
598 }
599}
600
601fn unknown_session(token: &SessionToken) -> Error {
603 Error::Invalid {
604 reason: format!(
605 "no session {:?} is open; `onevcs session open` prints a token",
606 token.0
607 ),
608 }
609}
610
611fn slug(identity: &str) -> Option<String> {
618 let mut parts = identity.split('/');
619 let (host, owner, name) = (parts.next()?, parts.next()?, parts.next()?);
620 if parts.next().is_some() || host != DEFAULT_HOST || owner.is_empty() || name.is_empty() {
621 return None;
622 }
623 Some(format!("{owner}/{name}"))
624}
625
626fn deferred(
634 state: &VcsState,
635 session: &Session,
636 identity: &str,
637 request: &PublishRequest,
638 policy: MergePolicy,
639 approvals: Approvals,
640 token: &SessionToken,
641) -> Result<url::Url> {
642 let refused = |reason: String| Error::Invalid { reason };
643 let adopt = match (request.defer_checks, &request.adopt) {
644 (true, Some(adopt)) => adopt,
645 (true, None) => {
646 return Err(refused(
647 "--defer-checks publishes into a draft that is already open: name it with \
648 --adopt <URL>. Nothing was pushed"
649 .to_owned(),
650 ))
651 }
652 (false, _) => {
653 return Err(refused(
654 "--adopt names the draft a --defer-checks publication pushes into, and this \
655 publication does not defer its checks. Nothing was pushed"
656 .to_owned(),
657 ))
658 }
659 };
660 if request.draft.is_some() {
661 return Err(refused(
662 "--defer-checks pushes into a draft held for review, and a draft reason asks for a \
663 different draft. Nothing was pushed"
664 .to_owned(),
665 ));
666 }
667 if policy != MergePolicy::ChangeOpen || approvals != Approvals::Required {
668 return Err(refused(format!(
669 "--defer-checks is for a change request held for review — a change-open identity \
670 whose approvals are required — and {identity} does not hold this one for review. \
671 Nothing was pushed"
672 )));
673 }
674 let segments: Vec<&str> = adopt
678 .path_segments()
679 .map(|segments| segments.collect())
680 .unwrap_or_default();
681 let named = match segments.as_slice() {
682 [owner, name, "pull", number, ..]
683 if adopt.scheme() == "https"
684 && !number.is_empty()
685 && number.chars().all(|c| c.is_ascii_digit()) =>
686 {
687 adopt
688 .host_str()
689 .map(|host| format!("{}/{owner}/{name}", host.to_ascii_lowercase()))
690 }
691 _ => None,
692 };
693 if named.as_deref() != Some(identity) || slug(identity).is_none() {
694 return Err(refused(format!(
695 "{adopt} is not a change request of {identity}: a change request is named as \
696 https://{DEFAULT_HOST}/<owner>/<name>/pull/<number>. Nothing was pushed"
697 )));
698 }
699 let recorded = state.publications.iter().rev().find_map(|earlier| {
700 let carried = match &earlier.outcome {
701 PublishOutcome::ChangeOpen(url)
702 | PublishOutcome::ChangeDraft(url)
703 | PublishOutcome::ChangeReviewDraft(url)
704 | PublishOutcome::ChangeReviewDraftUnchecked(url)
705 | PublishOutcome::Queued(url) => Some(url),
706 _ => None,
707 };
708 (carried == Some(adopt)).then(|| earlier.branch.clone())
709 });
710 match recorded {
711 Some(branch) if branch == session.branch => Ok(adopt.clone()),
712 Some(branch) => Err(refused(format!(
713 "{adopt} carries {branch:?} by this provider's newest record of it, and session {} \
714 is on {:?}. Nothing was pushed",
715 token.0, session.branch
716 ))),
717 None => Err(Error::UnrecordedChangeHead {
718 change: adopt.to_string(),
719 reason: format!(
720 "this provider holds no record of which branch {adopt} carries, and a \
721 --defer-checks publication asks the host nothing. Nothing was pushed"
722 ),
723 }),
724 }
725}
726
727fn record_local_landing(
733 identity: &str,
734 session: &Session,
735 token: &SessionToken,
736) -> (PublishOutcome, Vec<Emission>) {
737 let sha = events::stable_sha(&["publish", &token.0, &session.branch]);
738 let emission = Emission {
739 stream: token.0.clone(),
740 identity: Some(identity.to_owned()),
741 kind: EventKind::MergeCompleted,
742 payload: object(json!({
745 "identity": identity,
746 "sha": sha,
747 "base": session.base,
748 "landed_at": events::timestamp(),
749 "landing": sha,
750 })),
751 };
752 (PublishOutcome::Merged(Sha(sha)), vec![emission])
753}
754
755struct Publishing<'a> {
757 slug: &'a str,
758 identity: &'a str,
759 session: &'a Session,
760 policy: MergePolicy,
761 approvals: Approvals,
762 drafts: Drafts,
763 request: &'a PublishRequest,
764 token: &'a SessionToken,
765}
766
767impl Publishing<'_> {
768 fn emission(&self, kind: EventKind, payload: Value) -> Emission {
770 Emission {
771 stream: self.token.0.clone(),
772 identity: Some(self.identity.to_owned()),
773 kind,
774 payload: object(payload),
775 }
776 }
777}
778
779fn publish_as_change(
797 hosting: &dyn Hosting,
798 publishing: &Publishing<'_>,
799 emissions: &mut Vec<Emission>,
800) -> Result<PublishOutcome> {
801 let Publishing {
802 slug,
803 session,
804 policy,
805 request,
806 ..
807 } = *publishing;
808 let host = hosting.for_repo(slug)?;
809 let author = host.authenticated_user()?;
812 let lifecycle = !publishing.drafts.disabled.unwrap_or(false);
813 if lifecycle {
816 grace_seconds()?;
817 }
818 let existing = host.find_changes(&session.branch, &session.base)?;
819 let adopted = !existing.is_empty();
823 let change = match existing.into_iter().next() {
824 Some(change) => change,
825 None => host.open_change(ChangeSpec {
826 head: session.branch.clone(),
827 base: session.base.clone(),
828 title: request
834 .title
835 .as_deref()
836 .map_or_else(|| format!("Publish {}", session.branch), str::to_owned),
837 body: request.body.clone(),
841 draft: request.draft.clone(),
844 draft_awaiting_checks: lifecycle && request.draft.is_none(),
845 })?,
846 };
847 emissions.push(publishing.emission(
848 EventKind::ChangeOpened,
849 json!({
850 "url": change.url.to_string(),
851 "host": "github",
852 "id": change.id.0,
853 "base": change.base,
854 "author": author,
855 }),
856 ));
857 if let Some(reason) = &request.draft {
858 if !host.is_draft(&change)? {
862 return Err(Error::Invalid {
863 reason: format!(
864 "{url} is open for review on the host, and this publication asked for \
865 {asked}. A change that is open can land, so reporting it as a draft would \
866 say the work is held back when nothing is holding it",
867 url = change.url,
868 asked = asked_for(reason),
869 ),
870 });
871 }
872 let mut drafted = object(json!({
878 "url": change.url.to_string(),
879 "id": change.id.0,
880 "base": change.base,
881 }));
882 if let serde_json::Value::Object(fields) =
883 serde_json::to_value(reason).expect("a draft reason serializes")
884 {
885 drafted.extend(fields);
886 }
887 emissions.push(publishing.emission(EventKind::ChangeDrafted, Value::Object(drafted)));
888 return Ok(PublishOutcome::ChangeDraft(change.url.clone()));
891 }
892 let is_draft = match if lifecycle || adopted {
899 host.is_draft(&change)
900 } else {
901 Ok(false)
902 } {
903 Ok(draft) => draft,
904 Err(Error::NotImplemented { .. }) => false,
905 Err(unreadable) => return Err(unreadable),
906 };
907 let drafted = if lifecycle {
908 if is_draft {
909 emissions.push(publishing.emission(
910 EventKind::ChangeDrafted,
911 json!({
912 "url": change.url.to_string(),
913 "id": change.id.0,
914 "base": change.base,
915 "kind": "awaiting-checks",
916 }),
917 ));
918 }
919 is_draft
920 } else {
921 if adopted && is_draft {
924 lift(host.as_ref(), publishing, &change, emissions)?;
925 }
926 false
927 };
928
929 let still_draft = if drafted || policy != MergePolicy::ChangeAuto {
937 watch(host.as_ref(), publishing, &change, drafted, emissions)?
938 } else {
939 let answered = host.change_checks(&change)?;
940 let checks: Vec<&Check> = answered.checks.iter().collect();
941 let declared = given(
942 declared_required(host.as_ref(), &change),
943 answered.complete(),
944 &checks,
945 );
946 let standing = standings(&checks, &declared);
947 if let Some(failed) = red(&checks, &standing) {
948 return Err(failed);
949 }
950 if through(&standing, &declared) {
951 record_settled(
952 publishing,
953 &change,
954 emissions,
955 skipped(&standing),
956 declared.unread(),
957 );
958 }
959 false
960 };
961 if still_draft {
962 if policy == MergePolicy::ChangeOpen && publishing.approvals == Approvals::Required {
963 emissions.push(publishing.emission(
964 EventKind::DraftKeptForReview,
965 json!({
966 "url": change.url.to_string(),
967 "id": change.id.0,
968 "base": change.base,
969 }),
970 ));
971 return Ok(PublishOutcome::ChangeReviewDraft(change.url.clone()));
972 }
973 lift(host.as_ref(), publishing, &change, emissions)?;
976 }
977
978 if policy == MergePolicy::ChangeOpen {
979 return Ok(PublishOutcome::ChangeOpen(change.url.clone()));
980 }
981 Ok(match host.merge(&change, policy)? {
982 MergeOutcome::Merged(sha) => {
983 let landed_at = host
988 .merge_time(&change)
989 .ok()
990 .flatten()
991 .and_then(|spelled| events::moment(&spelled));
992 emissions.push(publishing.emission(
993 EventKind::ChangeMerged,
994 json!({
995 "url": change.url.to_string(),
996 "sha": sha.0,
997 "landed_at": landed_at,
998 "landing": sha.0,
999 }),
1000 ));
1001 emissions.push(publishing.emission(
1002 EventKind::MergeCompleted,
1003 json!({
1004 "identity": publishing.identity,
1005 "sha": sha.0,
1006 "landed_at": landed_at,
1007 "landing": sha.0,
1008 }),
1009 ));
1010 PublishOutcome::Merged(sha)
1011 }
1012 MergeOutcome::Queued => PublishOutcome::Queued(change.url.clone()),
1013 MergeOutcome::Open => PublishOutcome::ChangeOpen(change.url.clone()),
1014 })
1015}
1016
1017fn lift(
1019 host: &dyn RemoteHost,
1020 publishing: &Publishing<'_>,
1021 change: &ChangeRequest,
1022 emissions: &mut Vec<Emission>,
1023) -> Result<()> {
1024 host.ready_for_review(change)?;
1025 emissions.push(publishing.emission(
1026 EventKind::DraftLifted,
1027 json!({"url": change.url.to_string(), "id": change.id.0}),
1028 ));
1029 Ok(())
1030}
1031
1032#[derive(Clone, PartialEq, Eq)]
1035enum Declared {
1036 Nothing,
1037 Names(BTreeSet<String>),
1038 Unknown(String),
1039}
1040
1041impl Declared {
1042 fn unread(&self) -> Option<&str> {
1044 match self {
1045 Declared::Unknown(because) => Some(because),
1046 Declared::Nothing | Declared::Names(_) => None,
1047 }
1048 }
1049}
1050
1051fn declared_required(host: &dyn RemoteHost, change: &ChangeRequest) -> Declared {
1055 match host.required_checks_on(&change.base) {
1056 Ok(answer) if answer.checks.is_empty() && answer.complete() => Declared::Nothing,
1057 Ok(answer) if !answer.checks.is_empty() => Declared::Names(answer.checks),
1058 Ok(_) => Declared::Unknown(format!(
1059 "the host answered only in part about which checks a merge into {} requires",
1060 change.base
1061 )),
1062 Err(refused) => Declared::Unknown(format!(
1063 "the host would not say which checks a merge into {} requires: {refused}",
1064 change.base
1065 )),
1066 }
1067}
1068
1069fn given(declared: Declared, complete: bool, checks: &[&Check]) -> Declared {
1073 if matches!(declared, Declared::Unknown(_))
1074 && complete
1075 && !checks.is_empty()
1076 && checks.iter().all(|check| !check.required)
1077 {
1078 return Declared::Nothing;
1079 }
1080 declared
1081}
1082
1083fn standings(checks: &[&Check], declared: &Declared) -> Vec<(String, Option<CheckState>)> {
1087 let names: BTreeSet<String> = match declared {
1088 Declared::Names(names) => names.clone(),
1089 Declared::Nothing => BTreeSet::new(),
1090 Declared::Unknown(_) => checks
1091 .iter()
1092 .filter(|check| check.required)
1093 .map(|check| check.name.clone())
1094 .collect(),
1095 };
1096 let rank = |state: CheckState| match state {
1097 CheckState::Failed => 0,
1098 CheckState::Pending => 1,
1099 CheckState::Passed => 2,
1100 CheckState::NoVerdict => 3,
1101 CheckState::Skipped => 4,
1102 };
1103 names
1104 .into_iter()
1105 .map(|name| {
1106 let state = checks
1107 .iter()
1108 .filter(|check| check.name == name)
1109 .map(|check| check.state())
1110 .min_by_key(|state| rank(*state));
1111 (name, state)
1112 })
1113 .collect()
1114}
1115
1116fn red(checks: &[&Check], standing: &[(String, Option<CheckState>)]) -> Option<Error> {
1118 let (name, _) = standing
1119 .iter()
1120 .find(|(_, state)| *state == Some(CheckState::Failed))?;
1121 let check = checks
1122 .iter()
1123 .find(|check| &check.name == name && check.state() == CheckState::Failed)?;
1124 Some(Error::ChecksFailed {
1125 reason: format!(
1126 "required check {:?} concluded {}.",
1127 check.name,
1128 check
1129 .conclusion
1130 .as_deref()
1131 .unwrap_or("without a conclusion")
1132 ),
1133 })
1134}
1135
1136fn through(standing: &[(String, Option<CheckState>)], declared: &Declared) -> bool {
1140 (*declared == Declared::Nothing || !standing.is_empty())
1141 && standing
1142 .iter()
1143 .all(|(_, state)| matches!(state, Some(CheckState::Passed | CheckState::Skipped)))
1144}
1145
1146fn unsettled_named(
1151 checks: &[&Check],
1152 standing: &[(String, Option<CheckState>)],
1153 declared: &Declared,
1154) -> String {
1155 let pending: Vec<String> = standing
1156 .iter()
1157 .filter_map(|(name, state)| match state {
1158 Some(CheckState::Pending) => Some(format!("{name:?}")),
1159 None => Some(format!("{name:?} (no run yet)")),
1160 _ => None,
1161 })
1162 .collect();
1163 let no_verdict: Vec<String> = standing
1164 .iter()
1165 .filter(|(_, state)| *state == Some(CheckState::NoVerdict))
1166 .map(|(name, _)| {
1167 let conclusion = checks
1168 .iter()
1169 .find(|check| &check.name == name && check.state() == CheckState::NoVerdict)
1170 .and_then(|check| check.conclusion.as_deref())
1171 .unwrap_or("unknown");
1172 format!("{name:?}: {conclusion}")
1173 })
1174 .collect();
1175 let skipped: Vec<String> = skipped(standing)
1176 .iter()
1177 .map(|name| format!("{name:?}"))
1178 .collect();
1179 let named = if !pending.is_empty() {
1180 let verdicts = if no_verdict.is_empty() {
1181 String::new()
1182 } else {
1183 format!("; completed with no verdict: {}", no_verdict.join(", "))
1184 };
1185 format!("still unsettled: {}{verdicts}", pending.join(", "))
1186 } else if !no_verdict.is_empty() {
1187 format!("completed with no verdict: {}", no_verdict.join(", "))
1188 } else if !skipped.is_empty() {
1189 format!(
1190 "skipped on the draft, so not run on it: {}",
1191 skipped.join(", ")
1192 )
1193 } else if standing.is_empty() {
1194 match declared {
1195 Declared::Nothing => "the host declared no required check on it at all".to_owned(),
1196 _ => "it has marked no check required on it".to_owned(),
1197 }
1198 } else {
1199 "every required check it declared had settled".to_owned()
1200 };
1201 let read_from = declared
1202 .unread()
1203 .map(|because| {
1204 format!(
1205 ". Which checks it requires could not be read, so they were read from the \
1206 checks it marked required: {because}"
1207 )
1208 })
1209 .unwrap_or_default();
1210 format!("{named}{read_from}")
1211}
1212
1213fn skipped(standing: &[(String, Option<CheckState>)]) -> Vec<String> {
1215 standing
1216 .iter()
1217 .filter(|(_, state)| *state == Some(CheckState::Skipped))
1218 .map(|(name, _)| name.clone())
1219 .collect()
1220}
1221
1222fn record_settled(
1228 publishing: &Publishing<'_>,
1229 change: &ChangeRequest,
1230 emissions: &mut Vec<Emission>,
1231 skipped: Vec<String>,
1232 unread: Option<&str>,
1233) {
1234 let verdict = if skipped.is_empty() {
1235 "passed"
1236 } else {
1237 "passed-with-skipped"
1238 };
1239 let mut payload = json!({
1240 "url": change.url.to_string(),
1241 "id": change.id.0,
1242 "head": change.head_sha.0,
1243 "verdict": verdict,
1244 "skipped": skipped,
1245 });
1246 if let Some(because) = unread {
1247 eprintln!(
1248 "onevcs: warning: the required checks on {} were read from the host's own \
1249 per-check marking, because the declaration could not be read: {because}",
1250 change.url
1251 );
1252 payload["requirement"] = json!({"read_from": "host-marking", "because": because});
1253 }
1254 emissions.push(publishing.emission(EventKind::ChecksSettled, payload));
1255}
1256
1257fn watch(
1261 host: &dyn RemoteHost,
1262 publishing: &Publishing<'_>,
1263 change: &ChangeRequest,
1264 drafted: bool,
1265 emissions: &mut Vec<Emission>,
1266) -> Result<bool> {
1267 let settled = |emissions: &mut Vec<Emission>, skipped: Vec<String>, unread: Option<&str>| {
1268 record_settled(publishing, change, emissions, skipped, unread);
1269 };
1270 let unsettled = |checks: &[&Check],
1271 standing: &[(String, Option<CheckState>)],
1272 declared: &Declared| Error::ChecksUnsettled {
1273 reason: format!(
1274 "the host had not settled its required checks on {}; {}",
1275 change.url,
1276 unsettled_named(checks, standing, declared)
1277 ),
1278 };
1279 let answered = host.change_checks(change)?;
1280 let complete = answered.complete();
1281 let checks: Vec<&Check> = answered.checks.iter().collect();
1282 let declared = given(declared_required(host, change), complete, &checks);
1284 let unread = declared.unread();
1285
1286 if !drafted {
1287 let standing = standings(&checks, &declared);
1288 if let Some(failed) = red(&checks, &standing) {
1289 return Err(failed);
1290 }
1291 if !through(&standing, &declared) {
1292 return Err(unsettled(&checks, &standing, &declared));
1293 }
1294 settled(emissions, skipped(&standing), unread);
1295 return Ok(false);
1296 }
1297
1298 if matches!(declared, Declared::Nothing) {
1300 settled(emissions, Vec::new(), None);
1301 return Ok(true);
1302 }
1303 let standing = standings(&checks, &declared);
1304 if let Some(failed) = red(&checks, &standing) {
1305 return Err(failed);
1306 }
1307 if !standing.is_empty()
1308 && standing
1309 .iter()
1310 .all(|(_, state)| *state == Some(CheckState::Passed))
1311 {
1312 settled(emissions, Vec::new(), unread);
1313 return Ok(true);
1314 }
1315 let not_run: Vec<String> = standing
1316 .iter()
1317 .filter(|(_, state)| matches!(state, None | Some(CheckState::Skipped)))
1318 .map(|(name, _)| name.clone())
1319 .collect();
1320 let running = standing
1321 .iter()
1322 .any(|(_, state)| matches!(state, Some(CheckState::Pending | CheckState::NoVerdict)));
1323 let unseen = matches!(declared, Declared::Unknown(_)) && standing.is_empty();
1324 if unseen
1327 && !checks.is_empty()
1328 && checks
1329 .iter()
1330 .all(|check| check.state() != CheckState::Skipped)
1331 {
1332 settled(emissions, Vec::new(), unread);
1333 return Ok(true);
1334 }
1335 if running || (not_run.is_empty() && !unseen) {
1336 return Err(unsettled(&checks, &standing, &declared));
1337 }
1338
1339 let snapshot: Vec<Check> = answered.checks.clone();
1342 lift(host, publishing, change, emissions)?;
1343 let warned = publishing.drafts.warn_on_early_lift.unwrap_or(true);
1344 let grace = grace_seconds()?;
1345 if warned {
1346 eprintln!(
1347 "onevcs: warning: {} was lifted out of its draft before its required checks ran on \
1348 it ({}): none had run within {grace}s, which is what a workflow that skips drafts \
1349 looks like",
1350 change.url,
1351 not_run.join(", ")
1352 );
1353 }
1354 emissions.push(publishing.emission(
1355 EventKind::DraftLiftedEarly,
1356 json!({
1357 "url": change.url.to_string(),
1358 "id": change.id.0,
1359 "base": change.base,
1360 "awaited": not_run,
1361 "grace_seconds": grace,
1362 "warned": warned,
1363 }),
1364 ));
1365 let after = host.change_checks(change)?.checks;
1366 let from_the_draft = |check: &Check| !ran_after_the_lift(check, &snapshot);
1367 let counted: Vec<&Check> = after
1368 .iter()
1369 .filter(|check| !(from_the_draft(check) && check.state() == CheckState::Skipped))
1370 .collect();
1371 let standing = standings(&counted, &declared);
1372 if let Some(failed) = red(&counted, &standing) {
1373 return Err(failed);
1374 }
1375 let rerun = after.iter().any(|check| {
1376 !from_the_draft(check)
1377 && (standing.is_empty() || standing.iter().any(|(name, _)| *name == check.name))
1378 });
1379 if !rerun {
1380 return Err(Error::ChecksUnsettled {
1381 reason: format!(
1382 "the required checks on {} did not re-run after its draft was lifted: no run \
1383 the host attached after the lift appeared, and the runs from while it was a \
1384 draft were skipped, which is not a verdict. The likely cause is a workflow that \
1385 does not trigger on `ready_for_review`",
1386 change.url
1387 ),
1388 });
1389 }
1390 if !standing
1393 .iter()
1394 .all(|(_, state)| matches!(state, Some(CheckState::Passed | CheckState::Skipped)))
1395 {
1396 return Err(unsettled(&counted, &standing, &declared));
1397 }
1398 settled(emissions, skipped(&standing), unread);
1399 Ok(false)
1400}
1401
1402fn ran_after_the_lift(check: &Check, snapshot: &[Check]) -> bool {
1409 let earlier: Vec<&Check> = snapshot
1410 .iter()
1411 .filter(|seen| seen.name == check.name)
1412 .collect();
1413 if earlier.is_empty() {
1414 return true;
1415 }
1416 check.started_at.as_ref().is_some_and(|started| {
1417 earlier
1418 .iter()
1419 .all(|seen| seen.started_at.as_ref() != Some(started))
1420 })
1421}
1422
1423const DEFAULT_DRAFT_GRACE_SECONDS: f64 = 120.0;
1428
1429const DRAFT_GRACE_ENV: &str = "ONEVCS_DRAFT_CHECKS_GRACE_SECONDS";
1431
1432fn grace_seconds() -> Result<f64> {
1440 let Some(raw) = std::env::var_os(DRAFT_GRACE_ENV) else {
1441 return Ok(DEFAULT_DRAFT_GRACE_SECONDS);
1442 };
1443 let raw = raw.to_string_lossy().into_owned();
1444 let seconds: f64 = raw.trim().parse().map_err(|_| Error::Invalid {
1445 reason: format!("{DRAFT_GRACE_ENV} must be a number of seconds, not {raw:?}"),
1446 })?;
1447 if !seconds.is_finite() || seconds <= 0.0 {
1448 return Err(Error::Invalid {
1449 reason: format!(
1450 "{DRAFT_GRACE_ENV} must be a finite number of seconds above zero, not {raw:?}"
1451 ),
1452 });
1453 }
1454 Ok(seconds)
1455}
1456
1457fn asked_for(reason: &DraftReason) -> String {
1461 match reason {
1462 DraftReason::AwaitingRelease {
1463 awaiting, target, ..
1464 } => format!("a draft awaiting {awaiting} {target}"),
1465 DraftReason::Held { because } => {
1466 format!("a draft held by the session that opened it ({because})")
1467 }
1468 }
1469}
1470
1471fn refusal(identity: &str) -> PublishOutcome {
1479 failed(&if identity.split('/').count() == 3 {
1480 Error::NotImplemented {
1481 operation: "RemoteHost for a host other than github.com",
1482 }
1483 } else {
1484 Error::Invalid {
1485 reason: format!(
1486 "identity {identity:?} is not a hosted repository, so it cannot publish a \
1487 change request; a local identity publishes with local-direct"
1488 ),
1489 }
1490 })
1491}
1492
1493fn failed(error: &Error) -> PublishOutcome {
1495 PublishOutcome::Failed {
1496 kind: FailureKind::of(error),
1499 reason: error.to_string(),
1500 retained: None,
1503 }
1504}
1505
1506fn recover_command(branch: &str, checkout: &Path, provenance: Provenance) -> Vec<String> {
1516 let verb = match provenance {
1517 Provenance::IncompleteStep => "recover",
1518 Provenance::Complete => "publish-branch",
1519 };
1520 vec![
1521 "onevcs".to_owned(),
1522 verb.to_owned(),
1523 branch.to_owned(),
1524 "--repo".to_owned(),
1525 checkout.display().to_string(),
1526 ]
1527}
1528
1529fn spell(provenance: Provenance) -> &'static str {
1531 match provenance {
1532 Provenance::Complete => "complete",
1533 Provenance::IncompleteStep => "incomplete-step",
1534 }
1535}
1536
1537fn object(value: Value) -> Map<String, Value> {
1538 value.as_object().cloned().unwrap_or_default()
1539}