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::testing::EmissionResult;
21use onevcs::{
22 ArmedMerge, ChangeRequest, ChangeSpec, Check, CheckState, DraftReason, Error, EventKind,
23 FailureKind, HeldBy, Holding, Hosting, Identity, Landed, Lifecycle, MergeOutcome, MergePolicy,
24 PreservedBranch, Provenance, Publication, PublishOutcome, PublishRequest, Recoverable,
25 RemoteHost, Result, Scope, Session, SessionRecord, SessionRequest, SessionToken, Sha, Vcs,
26};
27
28use crate::events::{self, Emission};
29use crate::remote::DEFAULT_HOST;
30use crate::state::{self, VcsState};
31use crate::store::{FileStore, MemoryStore, Store};
32
33pub const DEFAULT_BASE: &str = "main";
38
39pub const DEFAULT_PUBLICATION: MergePolicy = MergePolicy::ChangeOpen;
44
45pub const DEFAULT_APPROVALS: Approvals = Approvals::Required;
48
49#[derive(Debug)]
54pub struct Repository<T> {
55 store: T,
56 root: PathBuf,
57 trees: Trees,
58}
59
60#[derive(Debug, Clone, Copy, PartialEq, Eq)]
62enum Trees {
63 Named,
66 Created,
68}
69
70pub type MemoryVcs = Repository<MemoryStore<VcsState>>;
77
78pub type FileVcs = Repository<FileStore<VcsState>>;
84
85impl MemoryVcs {
86 pub fn new() -> Self {
88 Self::seeded(VcsState::default())
89 }
90
91 pub fn seeded(state: VcsState) -> Self {
93 Self {
94 store: MemoryStore::new(state),
95 root: std::env::temp_dir().join("onevcs-testing-memory"),
96 trees: Trees::Named,
97 }
98 }
99
100 pub fn state(&self) -> VcsState {
102 self.store
103 .snapshot()
104 .expect("an in-memory store always answers")
105 }
106}
107
108impl Default for MemoryVcs {
109 fn default() -> Self {
110 Self::new()
111 }
112}
113
114impl FileVcs {
115 pub fn create(path: impl Into<PathBuf>) -> Result<Self> {
122 Self::over(FileStore::attach(path, &VcsState::default())?)
123 }
124
125 pub fn seeded(path: impl Into<PathBuf>, state: VcsState) -> Result<Self> {
128 Self::over(FileStore::replace(path, &state)?)
129 }
130
131 fn over(store: FileStore<VcsState>) -> Result<Self> {
132 let root = store
133 .path()
134 .parent()
135 .filter(|parent| !parent.as_os_str().is_empty())
136 .unwrap_or_else(|| Path::new("."))
137 .join("worktrees");
138 Ok(Self {
139 store,
140 root,
141 trees: Trees::Created,
142 })
143 }
144
145 pub fn state(&self) -> Result<VcsState> {
147 self.store.snapshot()
148 }
149}
150
151impl<T: Store<VcsState>> Vcs for Repository<T> {
152 fn resolve_identity(&self, origin_or_path: &str) -> Result<Identity> {
153 self.store.with(|state| {
154 state::identity_of(state, origin_or_path)
155 .cloned()
156 .ok_or_else(|| Error::Invalid {
157 reason: format!(
158 "{origin_or_path:?} does not name a repository this provider knows; {}",
159 state::known(state)
160 ),
161 })
162 })
163 }
164
165 fn open_session(&self, req: SessionRequest) -> Result<Session> {
166 let root = self.root.clone();
167 let (session, emission) = self.store.with(|state| {
168 let identity = state::identity_of(state, &req.repo)
169 .cloned()
170 .ok_or_else(|| Error::Invalid {
171 reason: format!(
172 "{:?} does not name a repository this provider knows; {}",
173 req.repo,
174 state::known(state)
175 ),
176 })?;
177 let token = SessionToken(format!("s-testing-{}", state.sessions.len() + 1));
180 let run_root = root.join(&token.0);
181 let base = req.base.clone().unwrap_or_else(|| DEFAULT_BASE.to_owned());
186 state::named_branch(&base, "the base")?;
187 for (key, value) in &req.labels {
188 state::label_pair(key, value)?;
189 }
190 let session = Session {
196 worktree: run_root.join("worktree"),
197 branch: state::requested_branch(&req, &token)?,
198 base,
199 token: token.clone(),
200 conflict: None,
201 };
202 state.sessions.push(session.clone());
203 state
204 .session_identities
205 .insert(token.clone(), identity.origin.clone());
206 if !req.labels.is_empty() {
207 state
208 .session_labels
209 .insert(token.clone(), req.labels.clone());
210 }
211 let emission = Emission {
212 stream: token.0.clone(),
213 identity: Some(identity.origin.clone()),
214 kind: EventKind::SessionOpened,
215 result: EmissionResult::Success,
216 payload: object(json!({
217 "token": token.0,
218 "identity": identity.origin,
219 "branch": session.branch,
220 "base": session.base,
221 "worktree": session.worktree.display().to_string(),
222 "clone": run_root.join("clone").display().to_string(),
225 "execution_checkout": run_root.join("checkout").display().to_string(),
226 "publication_checkout": run_root.join("checkout").display().to_string(),
227 "placement": {"kind": "run-root"},
231 })),
232 };
233 Ok((session, emission))
234 })?;
235 if self.trees == Trees::Created {
236 std::fs::create_dir_all(&session.worktree).map_err(|e| Error::Invalid {
237 reason: format!("cannot create {}: {e}", session.worktree.display()),
238 })?;
239 }
240 events::emit(&emission);
241 Ok(session)
242 }
243
244 fn adopt_session(&self, token: SessionToken) -> Result<Session> {
245 self.store.with(|state| {
246 state::session_of(state, &token)
247 .cloned()
248 .ok_or_else(|| Error::Invalid {
249 reason: format!(
250 "no session {:?} is open; `onevcs session open` prints a token",
251 token.0
252 ),
253 })
254 })
255 }
256
257 fn preserve(&self, s: &Session, provenance: Provenance) -> Result<PreservedBranch> {
258 let (branch, emission) = self.store.with(|state| {
259 let identity = state::identity_for(state, &s.token)?;
260 let branch = PreservedBranch {
261 branch: s.branch.clone(),
262 base: s.base.clone(),
263 provenance,
264 change_url: None,
265 change_base: None,
266 };
267 let row = Recoverable {
268 identity: identity.clone(),
269 branch: branch.clone(),
270 checkout: s.worktree.clone(),
271 tip: None,
272 stopped_because: format!("session {} was left open", s.token.0),
273 recover_command: recover_command(&s.branch, &s.worktree, provenance),
274 landed: Landed::No,
279 held_by: None,
283 net_negative: None,
284 session: Some(s.token.clone()),
287 labels: state
288 .session_labels
289 .get(&s.token)
290 .cloned()
291 .unwrap_or_default(),
292 on_origin: None,
297 retirement: None,
302 };
303 state.preserved.retain(|kept| {
307 kept.identity != row.identity || kept.branch.branch != row.branch.branch
308 });
309 state.preserved.push(row);
310 let emission = Emission {
311 stream: s.token.0.clone(),
312 identity: None,
318 kind: EventKind::CommitPreserved,
319 result: EmissionResult::Success,
320 payload: object(json!({
321 "branch": s.branch,
322 "sha": events::stable_sha(&[&s.token.0, &s.branch, spell(provenance)]),
323 "provenance": spell(provenance),
324 })),
325 };
326 Ok((branch, emission))
327 })?;
328 events::emit(&emission);
329 Ok(branch)
330 }
331
332 fn session(&self, token: &SessionToken) -> Result<SessionRecord> {
333 self.store.with(|state| {
334 let session = state::session_of(state, token)
335 .cloned()
336 .ok_or_else(|| unknown_session(token))?;
337 let identity = state::identity_for(state, token)?;
338 let provenance = state
342 .preserved
343 .iter()
344 .find(|row| row.identity == identity && row.branch.branch == session.branch)
345 .map_or(Provenance::Complete, |row| row.branch.provenance);
346 Ok(SessionRecord {
347 lifecycle: if state.closed_sessions.contains(token) {
348 Lifecycle::Closed
349 } else {
350 Lifecycle::Open
351 },
352 session,
353 identity,
354 provenance,
355 retried_by: None,
359 })
360 })
361 }
362
363 fn close_session(&self, token: &SessionToken) -> Result<Session> {
364 self.store.with(|state| {
365 let session = state::session_of(state, token)
366 .cloned()
367 .ok_or_else(|| unknown_session(token))?;
368 let emission = Emission {
369 stream: token.0.clone(),
370 identity: None,
374 kind: EventKind::SessionClosed,
375 result: EmissionResult::Success,
376 payload: object(json!({"token": token.0, "branch": session.branch})),
377 };
378 events::emit(&emission);
382 state.closed_sessions.insert(token.clone());
383 Ok(session)
384 })
385 }
386
387 fn publish(
400 &self,
401 token: &SessionToken,
402 request: &PublishRequest,
403 hosting: &dyn Hosting,
404 ) -> Result<Publication> {
405 let (publication, emissions) = self.store.with(|state| {
406 let session = state::session_of(state, token)
407 .cloned()
408 .ok_or_else(|| unknown_session(token))?;
409 let identity = state::identity_for(state, token)?;
410 if let Some(reason) = &request.draft {
415 reason.checked()?;
416 }
417 let resolved = state.policy.unwrap_or(DEFAULT_PUBLICATION);
418 let policy = match request.policy {
419 Some(requested) => resolved.narrow(requested)?,
420 None => resolved,
421 };
422 let published = |outcome, emissions| {
423 (
424 Publication {
425 session: token.clone(),
426 branch: session.branch.clone(),
427 policy,
428 outcome,
429 },
430 emissions,
431 )
432 };
433 if request.defer_checks || request.adopt.is_some() {
437 let approvals = state.approvals.unwrap_or(DEFAULT_APPROVALS);
438 let adopt = deferred(
439 state, &session, &identity, request, policy, approvals, token,
440 )?;
441 let number = adopt
442 .path_segments()
443 .and_then(|mut segments| segments.nth(3))
444 .unwrap_or_default()
445 .to_owned();
446 let emission = Emission {
447 stream: token.0.clone(),
448 identity: Some(identity.clone()),
449 kind: EventKind::ChangeOpened,
450 result: EmissionResult::Success,
451 payload: object(json!({
452 "url": adopt.to_string(),
453 "host": "github",
454 "id": number,
455 "base": session.base,
456 })),
457 };
458 let (publication, emissions) = published(
459 PublishOutcome::ChangeReviewDraftUnchecked(adopt),
460 vec![emission],
461 );
462 state.publications.push(publication.clone());
463 return Ok((publication, emissions));
464 }
465 if state.publications.iter().any(|earlier| {
471 earlier.session == *token && matches!(earlier.outcome, PublishOutcome::Merged(_))
472 }) {
473 let (publication, emissions) =
474 published(PublishOutcome::NothingToPublish, Vec::new());
475 state.publications.push(publication.clone());
476 return Ok((publication, emissions));
477 }
478
479 let (outcome, emissions) = if policy == MergePolicy::LocalDirect {
480 match &request.draft {
485 Some(reason) => (
486 failed(&Error::Invalid {
487 reason: format!(
488 "{branch:?} was asked to publish as {asked}, and this identity \
489 publishes with local-direct, which opens no change request at \
490 all",
491 branch = session.branch,
492 asked = asked_for(reason),
493 ),
494 }),
495 Vec::new(),
496 ),
497 None => record_local_landing(&identity, &session, token),
498 }
499 } else {
500 match slug(&identity) {
501 Some(slug) => {
502 let mut emissions = Vec::new();
505 let outcome = match publish_as_change(
506 hosting,
507 &Publishing {
508 slug: &slug,
509 identity: &identity,
510 session: &session,
511 policy,
512 approvals: state.approvals.unwrap_or(DEFAULT_APPROVALS),
513 drafts: state.drafts.unwrap_or_default(),
514 request,
515 token,
516 },
517 &mut emissions,
518 ) {
519 Ok(outcome) => outcome,
520 Err(error) => failed(&error),
524 };
525 (outcome, emissions)
526 }
527 None => (refusal(&identity), Vec::new()),
528 }
529 };
530 let (publication, emissions) = published(outcome, emissions);
531 state.publications.push(publication.clone());
532 Ok((publication, emissions))
533 })?;
534 for emission in &emissions {
535 events::emit(emission);
536 }
537 Ok(publication)
538 }
539
540 fn preserved(&self, scope: Scope) -> Result<Vec<Recoverable>> {
547 self.recoverable(scope)
548 }
549
550 fn recoverable(&self, scope: Scope) -> Result<Vec<Recoverable>> {
551 self.store.with(|state| {
552 let wanted = match &scope {
553 Scope::All => None,
554 Scope::Repo(repo) => Some(
555 state::identity_of(state, repo)
556 .map(|identity| identity.origin.clone())
557 .ok_or_else(|| Error::Invalid {
558 reason: format!(
559 "{repo:?} does not name a repository this provider knows; {}",
560 state::known(state)
561 ),
562 })?,
563 ),
564 };
565 Ok(state
567 .preserved
568 .iter()
569 .rev()
570 .filter(|row| wanted.as_ref().is_none_or(|key| *key == row.identity))
571 .cloned()
572 .map(|row| held(state, row))
573 .collect())
574 })
575 }
576}
577
578fn held(state: &VcsState, row: Recoverable) -> Recoverable {
588 if row.held_by.is_some() {
589 return row;
590 }
591 let holder = state.sessions.iter().find(|session| {
592 session.branch == row.branch.branch
593 && state.session_identities.get(&session.token) == Some(&row.identity)
594 && !state.closed_sessions.contains(&session.token)
595 });
596 Recoverable {
597 held_by: holder.map(|session| HeldBy {
598 token: Some(session.token.clone()),
599 worktree: session.worktree.clone(),
600 holding: Holding::OwnerRunning,
601 }),
602 ..row
603 }
604}
605
606fn unknown_session(token: &SessionToken) -> Error {
608 Error::Invalid {
609 reason: format!(
610 "no session {:?} is open; `onevcs session open` prints a token",
611 token.0
612 ),
613 }
614}
615
616fn slug(identity: &str) -> Option<String> {
623 let mut parts = identity.split('/');
624 let (host, owner, name) = (parts.next()?, parts.next()?, parts.next()?);
625 if parts.next().is_some() || host != DEFAULT_HOST || owner.is_empty() || name.is_empty() {
626 return None;
627 }
628 Some(format!("{owner}/{name}"))
629}
630
631fn deferred(
639 state: &VcsState,
640 session: &Session,
641 identity: &str,
642 request: &PublishRequest,
643 policy: MergePolicy,
644 approvals: Approvals,
645 token: &SessionToken,
646) -> Result<url::Url> {
647 let refused = |reason: String| Error::Invalid { reason };
648 let adopt = match (request.defer_checks, &request.adopt) {
649 (true, Some(adopt)) => adopt,
650 (true, None) => {
651 return Err(refused(
652 "--defer-checks publishes into a draft that is already open: name it with \
653 --adopt <URL>. Nothing was pushed"
654 .to_owned(),
655 ))
656 }
657 (false, _) => {
658 return Err(refused(
659 "--adopt names the draft a --defer-checks publication pushes into, and this \
660 publication does not defer its checks. Nothing was pushed"
661 .to_owned(),
662 ))
663 }
664 };
665 if request.draft.is_some() {
666 return Err(refused(
667 "--defer-checks pushes into a draft held for review, and a draft reason asks for a \
668 different draft. Nothing was pushed"
669 .to_owned(),
670 ));
671 }
672 if policy != MergePolicy::ChangeOpen || approvals != Approvals::Required {
673 return Err(refused(format!(
674 "--defer-checks is for a change request held for review — a change-open identity \
675 whose approvals are required — and {identity} does not hold this one for review. \
676 Nothing was pushed"
677 )));
678 }
679 let segments: Vec<&str> = adopt
683 .path_segments()
684 .map(|segments| segments.collect())
685 .unwrap_or_default();
686 let named = match segments.as_slice() {
687 [owner, name, "pull", number, ..]
688 if adopt.scheme() == "https"
689 && !number.is_empty()
690 && number.chars().all(|c| c.is_ascii_digit()) =>
691 {
692 adopt
693 .host_str()
694 .map(|host| format!("{}/{owner}/{name}", host.to_ascii_lowercase()))
695 }
696 _ => None,
697 };
698 if named.as_deref() != Some(identity) || slug(identity).is_none() {
699 return Err(refused(format!(
700 "{adopt} is not a change request of {identity}: a change request is named as \
701 https://{DEFAULT_HOST}/<owner>/<name>/pull/<number>. Nothing was pushed"
702 )));
703 }
704 let recorded = state.publications.iter().rev().find_map(|earlier| {
705 let carried = match &earlier.outcome {
706 PublishOutcome::ChangeOpen(url)
707 | PublishOutcome::ChangeDraft(url)
708 | PublishOutcome::ChangeReviewDraft(url)
709 | PublishOutcome::ChangeReviewDraftUnchecked(url)
710 | PublishOutcome::Queued(url) => Some(url),
711 _ => None,
712 };
713 (carried == Some(adopt)).then(|| earlier.branch.clone())
714 });
715 match recorded {
716 Some(branch) if branch == session.branch => Ok(adopt.clone()),
717 Some(branch) => Err(refused(format!(
718 "{adopt} carries {branch:?} by this provider's newest record of it, and session {} \
719 is on {:?}. Nothing was pushed",
720 token.0, session.branch
721 ))),
722 None => Err(Error::UnrecordedChangeHead {
723 change: adopt.to_string(),
724 reason: format!(
725 "this provider holds no record of which branch {adopt} carries, and a \
726 --defer-checks publication asks the host nothing. Nothing was pushed"
727 ),
728 }),
729 }
730}
731
732fn record_local_landing(
738 identity: &str,
739 session: &Session,
740 token: &SessionToken,
741) -> (PublishOutcome, Vec<Emission>) {
742 let sha = events::stable_sha(&["publish", &token.0, &session.branch]);
743 let emission = Emission {
744 stream: token.0.clone(),
745 identity: Some(identity.to_owned()),
746 kind: EventKind::MergeCompleted,
747 result: EmissionResult::Success,
750 payload: object(json!({
751 "identity": identity,
752 "sha": sha,
753 "base": session.base,
754 "landed_at": events::timestamp(),
755 "landing": sha,
756 })),
757 };
758 (PublishOutcome::Merged(Sha(sha)), vec![emission])
759}
760
761struct Publishing<'a> {
763 slug: &'a str,
764 identity: &'a str,
765 session: &'a Session,
766 policy: MergePolicy,
767 approvals: Approvals,
768 drafts: Drafts,
769 request: &'a PublishRequest,
770 token: &'a SessionToken,
771}
772
773impl Publishing<'_> {
774 fn emission(&self, kind: EventKind, payload: Value) -> Emission {
776 Emission {
777 stream: self.token.0.clone(),
778 identity: Some(self.identity.to_owned()),
779 kind,
780 result: EmissionResult::Success,
781 payload: object(payload),
782 }
783 }
784}
785
786fn publish_as_change(
804 hosting: &dyn Hosting,
805 publishing: &Publishing<'_>,
806 emissions: &mut Vec<Emission>,
807) -> Result<PublishOutcome> {
808 let Publishing {
809 slug,
810 session,
811 policy,
812 request,
813 ..
814 } = *publishing;
815 let host = hosting.for_repo(slug)?;
816 let author = host.authenticated_user()?;
819 let lifecycle = !publishing.drafts.disabled.unwrap_or(false);
820 if lifecycle {
823 grace_seconds()?;
824 }
825 let existing = host.find_changes(&session.branch, &session.base)?;
826 let adopted = !existing.is_empty();
830 let change = match existing.into_iter().next() {
831 Some(change) => change,
832 None => host.open_change(ChangeSpec {
833 head: session.branch.clone(),
834 base: session.base.clone(),
835 title: request
841 .title
842 .as_deref()
843 .map_or_else(|| format!("Publish {}", session.branch), str::to_owned),
844 body: request.body.clone(),
848 draft: request.draft.clone(),
851 draft_awaiting_checks: lifecycle && request.draft.is_none(),
852 })?,
853 };
854 emissions.push(publishing.emission(
855 EventKind::ChangeOpened,
856 json!({
857 "url": change.url.to_string(),
858 "host": "github",
859 "id": change.id.0,
860 "base": change.base,
861 "author": author,
862 }),
863 ));
864 if let Some(reason) = &request.draft {
865 if !host.is_draft(&change)? {
869 return Err(Error::Invalid {
870 reason: format!(
871 "{url} is open for review on the host, and this publication asked for \
872 {asked}. A change that is open can land, so reporting it as a draft would \
873 say the work is held back when nothing is holding it",
874 url = change.url,
875 asked = asked_for(reason),
876 ),
877 });
878 }
879 let mut drafted = object(json!({
885 "url": change.url.to_string(),
886 "id": change.id.0,
887 "base": change.base,
888 }));
889 if let serde_json::Value::Object(fields) =
890 serde_json::to_value(reason).expect("a draft reason serializes")
891 {
892 drafted.extend(fields);
893 }
894 emissions.push(publishing.emission(EventKind::ChangeDrafted, Value::Object(drafted)));
895 return Ok(PublishOutcome::ChangeDraft(change.url.clone()));
898 }
899 let is_draft = match if lifecycle || adopted {
906 host.is_draft(&change)
907 } else {
908 Ok(false)
909 } {
910 Ok(draft) => draft,
911 Err(Error::NotImplemented { .. }) => false,
912 Err(unreadable) => return Err(unreadable),
913 };
914 let drafted = if lifecycle {
915 if is_draft {
916 emissions.push(publishing.emission(
917 EventKind::ChangeDrafted,
918 json!({
919 "url": change.url.to_string(),
920 "id": change.id.0,
921 "base": change.base,
922 "kind": "awaiting-checks",
923 }),
924 ));
925 }
926 is_draft
927 } else {
928 if adopted && is_draft {
931 lift(host.as_ref(), publishing, &change, emissions)?;
932 }
933 false
934 };
935
936 let still_draft = if drafted || policy != MergePolicy::ChangeAuto {
944 match watch(host.as_ref(), publishing, &change, drafted, emissions)? {
945 Watched::Draft => true,
946 Watched::Ready => false,
947 Watched::Elapsed { error, skipped } => {
950 return match policy {
951 MergePolicy::ChangeAuto | MergePolicy::ChangeDirect => arm_on_elapse(
952 host.as_ref(),
953 publishing,
954 &change,
955 emissions,
956 error,
957 &skipped,
958 ),
959 MergePolicy::ChangeOpen | MergePolicy::LocalDirect => Err(error),
960 };
961 }
962 }
963 } else {
964 let answered = host.change_checks(&change)?;
965 let checks: Vec<&Check> = answered.checks.iter().collect();
966 let declared = given(
967 declared_required(host.as_ref(), &change),
968 answered.complete(),
969 &checks,
970 );
971 let standing = standings(&checks, &declared);
972 if let Some(failed) = red(&checks, &standing) {
973 return Err(failed);
974 }
975 if through(&standing, &declared) {
976 record_settled(
977 publishing,
978 &change,
979 emissions,
980 skipped(&standing),
981 declared.unread(),
982 );
983 }
984 false
985 };
986 if still_draft {
987 if policy == MergePolicy::ChangeOpen && publishing.approvals == Approvals::Required {
988 emissions.push(publishing.emission(
989 EventKind::DraftKeptForReview,
990 json!({
991 "url": change.url.to_string(),
992 "id": change.id.0,
993 "base": change.base,
994 }),
995 ));
996 return Ok(PublishOutcome::ChangeReviewDraft(change.url.clone()));
997 }
998 lift(host.as_ref(), publishing, &change, emissions)?;
1001 }
1002
1003 if policy == MergePolicy::ChangeOpen {
1004 return Ok(PublishOutcome::ChangeOpen(change.url.clone()));
1005 }
1006 Ok(match host.merge(&change, policy)? {
1007 MergeOutcome::Merged(sha) => merged(host.as_ref(), publishing, &change, emissions, &sha),
1008 MergeOutcome::Queued => PublishOutcome::Queued(change.url.clone()),
1009 MergeOutcome::Open => PublishOutcome::ChangeOpen(change.url.clone()),
1010 })
1011}
1012
1013fn merged(
1015 host: &dyn RemoteHost,
1016 publishing: &Publishing<'_>,
1017 change: &ChangeRequest,
1018 emissions: &mut Vec<Emission>,
1019 sha: &Sha,
1020) -> PublishOutcome {
1021 let landed_at = host
1026 .merge_time(change)
1027 .ok()
1028 .flatten()
1029 .and_then(|spelled| events::moment(&spelled));
1030 emissions.push(publishing.emission(
1031 EventKind::ChangeMerged,
1032 json!({
1033 "url": change.url.to_string(),
1034 "sha": sha.0,
1035 "landed_at": landed_at,
1036 "landing": sha.0,
1037 }),
1038 ));
1039 emissions.push(publishing.emission(
1040 EventKind::MergeCompleted,
1041 json!({
1042 "identity": publishing.identity,
1043 "sha": sha.0,
1044 "landed_at": landed_at,
1045 "landing": sha.0,
1046 }),
1047 ));
1048 PublishOutcome::Merged(sha.clone())
1049}
1050
1051fn arm_on_elapse(
1058 host: &dyn RemoteHost,
1059 publishing: &Publishing<'_>,
1060 change: &ChangeRequest,
1061 emissions: &mut Vec<Emission>,
1062 elapsed: Error,
1063 skipped: &[String],
1064) -> Result<PublishOutcome> {
1065 let Error::ChecksUnsettled { reason } = elapsed else {
1066 return Err(elapsed);
1067 };
1068 let unarmed = |why: String| Error::ChecksUnsettled {
1069 reason: format!(
1070 "{reason}. {why}, so {url} is left as it was — nothing lifted it or armed its merge",
1071 url = change.url
1072 ),
1073 };
1074 if !skipped.is_empty() {
1077 return Err(unarmed(format!(
1078 "Required check {} concluded skipped, which the host's auto-merge would read as \
1079 passed",
1080 skipped.join(", ")
1081 )));
1082 }
1083 match host.allows_auto_merge() {
1084 Ok(true) => {}
1085 Ok(false) => {
1086 return Err(unarmed(
1087 "The repository does not allow auto-merge".to_owned(),
1088 ))
1089 }
1090 Err(unread) => {
1091 return Err(unarmed(format!(
1092 "Whether the repository allows auto-merge could not be read ({unread})"
1093 )))
1094 }
1095 }
1096 let is_draft = match host.is_draft(change) {
1097 Ok(draft) => draft,
1098 Err(Error::NotImplemented { .. }) => false,
1099 Err(unreadable) => return Err(unreadable),
1100 };
1101 if is_draft {
1102 lift(host, publishing, change, emissions)?;
1103 }
1104 if let MergeOutcome::Merged(sha) = host.merge(change, MergePolicy::ChangeAuto)? {
1105 return Ok(merged(host, publishing, change, emissions, &sha));
1106 }
1107 emissions.push(publishing.emission(
1108 EventKind::MergeArmed,
1109 json!({"url": change.url.to_string(), "head": change.head_sha.0}),
1110 ));
1111 Ok(PublishOutcome::Failed {
1112 kind: FailureKind::ChecksUnsettled,
1113 reason: format!(
1114 "{}. Its merge is armed: the host lands {url} once its required checks pass, and \
1115 publishing the session again watches it",
1116 Error::ChecksUnsettled { reason },
1117 url = change.url,
1118 ),
1119 retained: None,
1120 armed: Some(ArmedMerge {
1121 change: change.url.clone(),
1122 head: change.head_sha.0.clone(),
1123 }),
1124 })
1125}
1126
1127fn lift(
1129 host: &dyn RemoteHost,
1130 publishing: &Publishing<'_>,
1131 change: &ChangeRequest,
1132 emissions: &mut Vec<Emission>,
1133) -> Result<()> {
1134 host.ready_for_review(change)?;
1135 emissions.push(publishing.emission(
1136 EventKind::DraftLifted,
1137 json!({"url": change.url.to_string(), "id": change.id.0}),
1138 ));
1139 Ok(())
1140}
1141
1142#[derive(Clone, PartialEq, Eq)]
1145enum Declared {
1146 Nothing,
1147 Names(BTreeSet<String>),
1148 Unknown(String),
1149}
1150
1151impl Declared {
1152 fn unread(&self) -> Option<&str> {
1154 match self {
1155 Declared::Unknown(because) => Some(because),
1156 Declared::Nothing | Declared::Names(_) => None,
1157 }
1158 }
1159}
1160
1161fn declared_required(host: &dyn RemoteHost, change: &ChangeRequest) -> Declared {
1165 match host.required_checks_on(&change.base) {
1166 Ok(answer) if answer.checks.is_empty() && answer.complete() => Declared::Nothing,
1167 Ok(answer) if !answer.checks.is_empty() => Declared::Names(answer.checks),
1168 Ok(_) => Declared::Unknown(format!(
1169 "the host answered only in part about which checks a merge into {} requires",
1170 change.base
1171 )),
1172 Err(refused) => Declared::Unknown(format!(
1173 "the host would not say which checks a merge into {} requires: {refused}",
1174 change.base
1175 )),
1176 }
1177}
1178
1179fn given(declared: Declared, complete: bool, checks: &[&Check]) -> Declared {
1183 if matches!(declared, Declared::Unknown(_))
1184 && complete
1185 && !checks.is_empty()
1186 && checks.iter().all(|check| !check.required)
1187 {
1188 return Declared::Nothing;
1189 }
1190 declared
1191}
1192
1193fn standings(checks: &[&Check], declared: &Declared) -> Vec<(String, Option<CheckState>)> {
1197 let names: BTreeSet<String> = match declared {
1198 Declared::Names(names) => names.clone(),
1199 Declared::Nothing => BTreeSet::new(),
1200 Declared::Unknown(_) => checks
1201 .iter()
1202 .filter(|check| check.required)
1203 .map(|check| check.name.clone())
1204 .collect(),
1205 };
1206 let rank = |state: CheckState| match state {
1207 CheckState::Failed => 0,
1208 CheckState::Pending => 1,
1209 CheckState::Passed => 2,
1210 CheckState::NoVerdict => 3,
1211 CheckState::Skipped => 4,
1212 };
1213 names
1214 .into_iter()
1215 .map(|name| {
1216 let state = checks
1217 .iter()
1218 .filter(|check| check.name == name)
1219 .map(|check| check.state())
1220 .min_by_key(|state| rank(*state));
1221 (name, state)
1222 })
1223 .collect()
1224}
1225
1226fn red(checks: &[&Check], standing: &[(String, Option<CheckState>)]) -> Option<Error> {
1228 let (name, _) = standing
1229 .iter()
1230 .find(|(_, state)| *state == Some(CheckState::Failed))?;
1231 let check = checks
1232 .iter()
1233 .find(|check| &check.name == name && check.state() == CheckState::Failed)?;
1234 Some(Error::ChecksFailed {
1235 reason: format!(
1236 "required check {:?} concluded {}.",
1237 check.name,
1238 check
1239 .conclusion
1240 .as_deref()
1241 .unwrap_or("without a conclusion")
1242 ),
1243 })
1244}
1245
1246fn through(standing: &[(String, Option<CheckState>)], declared: &Declared) -> bool {
1250 (*declared == Declared::Nothing || !standing.is_empty())
1251 && standing
1252 .iter()
1253 .all(|(_, state)| matches!(state, Some(CheckState::Passed | CheckState::Skipped)))
1254}
1255
1256fn unsettled_named(
1261 checks: &[&Check],
1262 standing: &[(String, Option<CheckState>)],
1263 declared: &Declared,
1264) -> String {
1265 let pending: Vec<String> = standing
1266 .iter()
1267 .filter_map(|(name, state)| match state {
1268 Some(CheckState::Pending) => Some(format!("{name:?}")),
1269 None => Some(format!("{name:?} (no run yet)")),
1270 _ => None,
1271 })
1272 .collect();
1273 let no_verdict: Vec<String> = standing
1274 .iter()
1275 .filter(|(_, state)| *state == Some(CheckState::NoVerdict))
1276 .map(|(name, _)| {
1277 let conclusion = checks
1278 .iter()
1279 .find(|check| &check.name == name && check.state() == CheckState::NoVerdict)
1280 .and_then(|check| check.conclusion.as_deref())
1281 .unwrap_or("unknown");
1282 format!("{name:?}: {conclusion}")
1283 })
1284 .collect();
1285 let skipped: Vec<String> = skipped(standing)
1286 .iter()
1287 .map(|name| format!("{name:?}"))
1288 .collect();
1289 let named = if !pending.is_empty() {
1290 let verdicts = if no_verdict.is_empty() {
1291 String::new()
1292 } else {
1293 format!("; completed with no verdict: {}", no_verdict.join(", "))
1294 };
1295 format!("still unsettled: {}{verdicts}", pending.join(", "))
1296 } else if !no_verdict.is_empty() {
1297 format!("completed with no verdict: {}", no_verdict.join(", "))
1298 } else if !skipped.is_empty() {
1299 format!(
1300 "skipped on the draft, so not run on it: {}",
1301 skipped.join(", ")
1302 )
1303 } else if standing.is_empty() {
1304 match declared {
1305 Declared::Nothing => "the host declared no required check on it at all".to_owned(),
1306 _ => "it has marked no check required on it".to_owned(),
1307 }
1308 } else {
1309 "every required check it declared had settled".to_owned()
1310 };
1311 let read_from = declared
1312 .unread()
1313 .map(|because| {
1314 format!(
1315 ". Which checks it requires could not be read, so they were read from the \
1316 checks it marked required: {because}"
1317 )
1318 })
1319 .unwrap_or_default();
1320 format!("{named}{read_from}")
1321}
1322
1323fn skipped(standing: &[(String, Option<CheckState>)]) -> Vec<String> {
1325 standing
1326 .iter()
1327 .filter(|(_, state)| *state == Some(CheckState::Skipped))
1328 .map(|(name, _)| name.clone())
1329 .collect()
1330}
1331
1332fn record_settled(
1338 publishing: &Publishing<'_>,
1339 change: &ChangeRequest,
1340 emissions: &mut Vec<Emission>,
1341 skipped: Vec<String>,
1342 unread: Option<&str>,
1343) {
1344 let verdict = if skipped.is_empty() {
1345 "passed"
1346 } else {
1347 "passed-with-skipped"
1348 };
1349 let mut payload = json!({
1350 "url": change.url.to_string(),
1351 "id": change.id.0,
1352 "head": change.head_sha.0,
1353 "verdict": verdict,
1354 "skipped": skipped,
1355 });
1356 if let Some(because) = unread {
1357 eprintln!(
1358 "onevcs: warning: the required checks on {} were read from the host's own \
1359 per-check marking, because the declaration could not be read: {because}",
1360 change.url
1361 );
1362 payload["requirement"] = json!({"read_from": "host-marking", "because": because});
1363 }
1364 let mut settled = publishing.emission(EventKind::ChecksSettled, payload);
1365 settled.result = if verdict == "passed" {
1366 EmissionResult::Passed
1367 } else {
1368 EmissionResult::PassedWithSkipped
1369 };
1370 emissions.push(settled);
1371}
1372
1373fn watch(
1376 host: &dyn RemoteHost,
1377 publishing: &Publishing<'_>,
1378 change: &ChangeRequest,
1379 drafted: bool,
1380 emissions: &mut Vec<Emission>,
1381) -> Result<Watched> {
1382 let settled = |emissions: &mut Vec<Emission>, skipped: Vec<String>, unread: Option<&str>| {
1383 record_settled(publishing, change, emissions, skipped, unread);
1384 };
1385 let elapsed_naming = |checks: &[&Check],
1386 standing: &[(String, Option<CheckState>)],
1387 declared: &Declared,
1388 skipped: Vec<String>| Watched::Elapsed {
1389 error: Error::ChecksUnsettled {
1390 reason: format!(
1391 "the host had not settled its required checks on {}; {}",
1392 change.url,
1393 unsettled_named(checks, standing, declared)
1394 ),
1395 },
1396 skipped,
1397 };
1398 let unsettled =
1399 |checks: &[&Check], standing: &[(String, Option<CheckState>)], declared: &Declared| {
1400 elapsed_naming(checks, standing, declared, skipped(standing))
1401 };
1402 let answered = host.change_checks(change)?;
1403 let complete = answered.complete();
1404 let checks: Vec<&Check> = answered.checks.iter().collect();
1405 let declared = given(declared_required(host, change), complete, &checks);
1407 let unread = declared.unread();
1408
1409 if !drafted {
1410 let standing = standings(&checks, &declared);
1411 if let Some(failed) = red(&checks, &standing) {
1412 return Err(failed);
1413 }
1414 if !through(&standing, &declared) {
1415 return Ok(unsettled(&checks, &standing, &declared));
1416 }
1417 settled(emissions, skipped(&standing), unread);
1418 return Ok(Watched::Ready);
1419 }
1420
1421 if matches!(declared, Declared::Nothing) {
1423 settled(emissions, Vec::new(), None);
1424 return Ok(Watched::Draft);
1425 }
1426 let standing = standings(&checks, &declared);
1427 if let Some(failed) = red(&checks, &standing) {
1428 return Err(failed);
1429 }
1430 if !standing.is_empty()
1431 && standing
1432 .iter()
1433 .all(|(_, state)| *state == Some(CheckState::Passed))
1434 {
1435 settled(emissions, Vec::new(), unread);
1436 return Ok(Watched::Draft);
1437 }
1438 let not_run: Vec<String> = standing
1439 .iter()
1440 .filter(|(_, state)| matches!(state, None | Some(CheckState::Skipped)))
1441 .map(|(name, _)| name.clone())
1442 .collect();
1443 let running = standing
1444 .iter()
1445 .any(|(_, state)| matches!(state, Some(CheckState::Pending | CheckState::NoVerdict)));
1446 let unseen = matches!(declared, Declared::Unknown(_)) && standing.is_empty();
1447 if unseen
1450 && !checks.is_empty()
1451 && checks
1452 .iter()
1453 .all(|check| check.state() != CheckState::Skipped)
1454 {
1455 settled(emissions, Vec::new(), unread);
1456 return Ok(Watched::Draft);
1457 }
1458 if running || (not_run.is_empty() && !unseen) || still_running(host, change, &checks) {
1459 return Ok(unsettled(&checks, &standing, &declared));
1460 }
1461
1462 let snapshot: Vec<Check> = answered.checks.clone();
1465 lift(host, publishing, change, emissions)?;
1466 let warned = publishing.drafts.warn_on_early_lift.unwrap_or(true);
1467 let grace = grace_seconds()?;
1468 if warned {
1469 eprintln!(
1470 "onevcs: warning: {} was lifted out of its draft before its required checks ran on \
1471 it ({}): none had run within {grace}s and nothing on its head was still running, \
1472 which is what a workflow that skips drafts looks like. {DRAFT_CI}",
1473 change.url,
1474 not_run.join(", ")
1475 );
1476 }
1477 emissions.push(publishing.emission(
1478 EventKind::DraftLiftedEarly,
1479 json!({
1480 "url": change.url.to_string(),
1481 "id": change.id.0,
1482 "base": change.base,
1483 "awaited": not_run,
1484 "grace_seconds": grace,
1485 "warned": warned,
1486 }),
1487 ));
1488 let after = host.change_checks(change)?.checks;
1489 let from_the_draft = |check: &Check| !ran_after_the_lift(check, &snapshot);
1490 let counted: Vec<&Check> = after
1491 .iter()
1492 .filter(|check| !(from_the_draft(check) && check.state() == CheckState::Skipped))
1493 .collect();
1494 let standing = standings(&counted, &declared);
1495 if let Some(failed) = red(&counted, &standing) {
1496 return Err(failed);
1497 }
1498 let rerun = after.iter().any(|check| {
1499 !from_the_draft(check)
1500 && (standing.is_empty() || standing.iter().any(|(name, _)| *name == check.name))
1501 });
1502 let reported: Vec<&Check> = after.iter().collect();
1506 let elapsed = || {
1507 let mut named = skipped(&standing);
1508 for name in skipped(&standings(&reported, &declared)) {
1509 if !named.contains(&name) {
1510 named.push(name);
1511 }
1512 }
1513 elapsed_naming(&counted, &standing, &declared, named)
1514 };
1515 if !rerun {
1516 if still_running(host, change, &reported) {
1519 return Ok(elapsed());
1520 }
1521 return Err(Error::ChecksUnsettled {
1522 reason: format!(
1523 "the required checks on {} did not re-run after its draft was lifted: no run \
1524 the host attached after the lift appeared and nothing on its head is still \
1525 running, while the runs from while it was a draft were skipped or never \
1526 started, which is not a verdict. {DRAFT_CI}",
1527 change.url
1528 ),
1529 });
1530 }
1531 if !standing
1534 .iter()
1535 .all(|(_, state)| matches!(state, Some(CheckState::Passed | CheckState::Skipped)))
1536 {
1537 return Ok(elapsed());
1538 }
1539 settled(emissions, skipped(&standing), unread);
1540 Ok(Watched::Ready)
1541}
1542
1543const DRAFT_CI: &str = "Run the required checks on drafts: their results attach to the head \
1546 commit, so the change is then kept a draft until they are green, and a workflow that runs \
1547 on drafts needs no `ready_for_review` trigger. Only a workflow that skips its required \
1548 jobs on drafts needs `ready_for_review` to run them after a lift, at the cost of one more \
1549 run per lift";
1550
1551fn still_running(host: &dyn RemoteHost, change: &ChangeRequest, checks: &[&Check]) -> bool {
1557 checks.iter().any(|check| !check.settled())
1558 || host
1559 .head_activity(change)
1560 .is_ok_and(|answer| answer.head == change.head_sha && !answer.in_progress.is_empty())
1561}
1562
1563enum Watched {
1565 Draft,
1567 Ready,
1569 Elapsed {
1572 error: Error,
1574 skipped: Vec<String>,
1577 },
1578}
1579
1580fn ran_after_the_lift(check: &Check, snapshot: &[Check]) -> bool {
1587 let earlier: Vec<&Check> = snapshot
1588 .iter()
1589 .filter(|seen| seen.name == check.name)
1590 .collect();
1591 if earlier.is_empty() {
1592 return true;
1593 }
1594 check.started_at.as_ref().is_some_and(|started| {
1595 earlier
1596 .iter()
1597 .all(|seen| seen.started_at.as_ref() != Some(started))
1598 })
1599}
1600
1601const DEFAULT_DRAFT_GRACE_SECONDS: f64 = 120.0;
1606
1607const DRAFT_GRACE_ENV: &str = "ONEVCS_DRAFT_CHECKS_GRACE_SECONDS";
1609
1610fn grace_seconds() -> Result<f64> {
1618 let Some(raw) = std::env::var_os(DRAFT_GRACE_ENV) else {
1619 return Ok(DEFAULT_DRAFT_GRACE_SECONDS);
1620 };
1621 let raw = raw.to_string_lossy().into_owned();
1622 let seconds: f64 = raw.trim().parse().map_err(|_| Error::Invalid {
1623 reason: format!("{DRAFT_GRACE_ENV} must be a number of seconds, not {raw:?}"),
1624 })?;
1625 if !seconds.is_finite() || seconds <= 0.0 {
1626 return Err(Error::Invalid {
1627 reason: format!(
1628 "{DRAFT_GRACE_ENV} must be a finite number of seconds above zero, not {raw:?}"
1629 ),
1630 });
1631 }
1632 Ok(seconds)
1633}
1634
1635fn asked_for(reason: &DraftReason) -> String {
1639 match reason {
1640 DraftReason::AwaitingRelease {
1641 awaiting, target, ..
1642 } => format!("a draft awaiting {awaiting} {target}"),
1643 DraftReason::Held { because } => {
1644 format!("a draft held by the session that opened it ({because})")
1645 }
1646 }
1647}
1648
1649fn refusal(identity: &str) -> PublishOutcome {
1657 failed(&if identity.split('/').count() == 3 {
1658 Error::NotImplemented {
1659 operation: "RemoteHost for a host other than github.com",
1660 }
1661 } else {
1662 Error::Invalid {
1663 reason: format!(
1664 "identity {identity:?} is not a hosted repository, so it cannot publish a \
1665 change request; a local identity publishes with local-direct"
1666 ),
1667 }
1668 })
1669}
1670
1671fn failed(error: &Error) -> PublishOutcome {
1673 PublishOutcome::Failed {
1674 kind: FailureKind::of(error),
1677 reason: error.to_string(),
1678 retained: None,
1681 armed: None,
1682 }
1683}
1684
1685fn recover_command(branch: &str, checkout: &Path, provenance: Provenance) -> Vec<String> {
1695 let verb = match provenance {
1696 Provenance::IncompleteStep => "recover",
1697 Provenance::Complete => "publish-branch",
1698 };
1699 vec![
1700 "onevcs".to_owned(),
1701 verb.to_owned(),
1702 branch.to_owned(),
1703 "--repo".to_owned(),
1704 checkout.display().to_string(),
1705 ]
1706}
1707
1708fn spell(provenance: Provenance) -> &'static str {
1710 match provenance {
1711 Provenance::Complete => "complete",
1712 Provenance::IncompleteStep => "incomplete-step",
1713 }
1714}
1715
1716fn object(value: Value) -> Map<String, Value> {
1717 value.as_object().cloned().unwrap_or_default()
1718}