onevcs_testing/repository.rs
1//! The repository side: one implementation of [`Vcs`] over either store.
2//!
3//! What it is not: git. It answers the questions the interface asks, records what
4//! it was asked, and emits the events the real implementation emits. What it
5//! cannot do is tell you whether a tree is dirty or whether a merge conflicts,
6//! because there is no tree — a journey that needs those drives the real `Git`.
7//!
8//! Publishing is the one operation that reaches past the repository: the host side
9//! of it is *performed*, against the [`Hosting`] the publication was handed, and
10//! the repository side of it is neither performed nor claimed. What that leaves
11//! out is written where each piece is left out.
12
13use 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
32/// The base a session is cut from when the request names none.
33///
34/// The real implementation asks the origin for its default branch, and this
35/// provider has no origin to ask.
36pub const DEFAULT_BASE: &str = "main";
37
38/// The policy a publication takes when nothing was seeded and nothing requested.
39///
40/// The policy the contract's own `default:` names, which is what the real
41/// implementation resolves to for a registry with no rules file.
42pub const DEFAULT_PUBLICATION: MergePolicy = MergePolicy::ChangeOpen;
43
44/// The approvals a publication takes when nothing was seeded: the contract's own
45/// `default:`, as [`DEFAULT_PUBLICATION`] is.
46pub const DEFAULT_APPROVALS: Approvals = Approvals::Required;
47
48/// The repository side of a run, over whichever store holds its state.
49///
50/// The two flavours below are this one behaviour with a different store under it,
51/// so neither can learn something the other does not know.
52#[derive(Debug)]
53pub struct Repository<T> {
54 store: T,
55 root: PathBuf,
56 trees: Trees,
57}
58
59/// What a session's worktree path means.
60#[derive(Debug, Clone, Copy, PartialEq, Eq)]
61enum Trees {
62 /// The path is named and nothing is created there: the in-memory provider
63 /// touches no filesystem beyond the event stream.
64 Named,
65 /// The directory is created, so a journey has somewhere to write work.
66 Created,
67}
68
69/// A repository provider that keeps its state in this process: no disk, no
70/// visibility to a second process, and the fastest of the two.
71///
72/// Its sessions name a worktree under the system temporary directory and **do not
73/// create it** — nothing here touches the filesystem except the event stream, which
74/// is the record a journey reads.
75pub type MemoryVcs = Repository<MemoryStore<VcsState>>;
76
77/// A repository provider that keeps its state in one JSON document, so several
78/// `onevcs` invocations see one another's effects.
79///
80/// Its sessions name a worktree beside that document **and create it**, so a
81/// journey that writes a file into a session's tree has somewhere to write it.
82pub type FileVcs = Repository<FileStore<VcsState>>;
83
84impl MemoryVcs {
85 /// A repository provider knowing nothing.
86 pub fn new() -> Self {
87 Self::seeded(VcsState::default())
88 }
89
90 /// A repository provider that starts from a scenario.
91 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 /// Everything it knows.
100 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 /// A repository provider keeping its state at `path`: whatever is already
115 /// there, or nothing.
116 ///
117 /// Attaching rather than replacing, so a second provider over the same path
118 /// picks up what the first one left — which is what a journey driving several
119 /// invocations reaches for this flavour to get.
120 pub fn create(path: impl Into<PathBuf>) -> Result<Self> {
121 Self::over(FileStore::attach(path, &VcsState::default())?)
122 }
123
124 /// A repository provider that starts from a scenario, keeping its state at
125 /// `path` and replacing whatever was there.
126 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 /// Everything it knows, read back out of its document.
145 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 // Consecutive and predictable, so a journey can name the session it is
177 // about to open — the one thing a digest-shaped token takes away.
178 let token = SessionToken(format!("s-testing-{}", state.sessions.len() + 1));
179 let run_root = root.join(&token.0);
180 // Both names are checked before they are recorded, because both go on to
181 // spell a ref for whoever holds the session — and a provider that
182 // accepted a name git refuses would let a journey pass where the real
183 // run stops.
184 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 // Never a conflict: a provider has no base to merge into a continued
190 // branch, so every session it opens opens clean — which is also why its
191 // `session-opened` below never carries the `conflict` key the real one
192 // writes over a merge it left in progress, and why `refuse_conflicts`
193 // has nothing here to refuse.
194 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 // Synthetic, and named anyway: a consumer reading this event
221 // reads the same keys whichever implementation produced it.
222 "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 // A run root of its own, which is the one placement a provider
226 // that keeps no pool can make — and the one every real session
227 // gets on a host that configures none.
228 "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 stopped_because: format!("session {} was left open", s.token.0),
270 recover_command: recover_command(&s.branch, &s.worktree, provenance),
271 // Preserved work this provider was handed is work nobody published:
272 // there is no base here whose history could record a landing, and no
273 // tree to compare content against, so the one answer it can give is
274 // the one it knows.
275 landed: Landed::No,
276 // Both answered where the row is *read* rather than frozen in here:
277 // whether a session still holds its branch is a fact about the session
278 // now, and there is no tree here to count a diff's lines against.
279 held_by: None,
280 net_negative: None,
281 // The session that preserved it is the session that answers for it,
282 // and what it was opened with is what the row carries.
283 session: Some(s.token.clone()),
284 labels: state
285 .session_labels
286 .get(&s.token)
287 .cloned()
288 .unwrap_or_default(),
289 // A preservation onto an origin is a thing only a real repository can
290 // do — there is no git here and no remote to push to — so this provider
291 // answers what a scenario wrote down and nothing else. A hand-written
292 // state may still seed one, exactly as it seeds a hold.
293 on_origin: None,
294 // Whether a branch may be retired is decided from every copy of it this
295 // host holds, its base on the origin, and this host's records — none of
296 // which a provider has — so this provider classifies nothing, and a row
297 // it answers carries no classification rather than an invented one.
298 retirement: None,
299 };
300 // Preserving the same branch twice replaces its row rather than listing
301 // it twice, which is what `recoverable` does across the checkouts a
302 // branch is reachable from.
303 state.preserved.retain(|kept| {
304 kept.identity != row.identity || kept.branch.branch != row.branch.branch
305 });
306 state.preserved.push(row);
307 let emission = Emission {
308 stream: s.token.0.clone(),
309 // No identity label, because the real implementation carries none
310 // here: the label is stamped where a session is opened, and work is
311 // preserved against a stream a later process opened fresh. Claiming
312 // it would be drift in the direction that looks like more
313 // information.
314 identity: None,
315 kind: EventKind::CommitPreserved,
316 payload: object(json!({
317 "branch": s.branch,
318 "sha": events::stable_sha(&[&s.token.0, &s.branch, spell(provenance)]),
319 "provenance": spell(provenance),
320 })),
321 };
322 Ok((branch, emission))
323 })?;
324 events::emit(&emission);
325 Ok(branch)
326 }
327
328 fn session(&self, token: &SessionToken) -> Result<SessionRecord> {
329 self.store.with(|state| {
330 let session = state::session_of(state, token)
331 .cloned()
332 .ok_or_else(|| unknown_session(token))?;
333 let identity = state::identity_for(state, token)?;
334 // Read off what was preserved rather than remembered separately: the
335 // real implementation reads the branch, so a session whose work was
336 // preserved behind an incomplete-step marker answers the same here.
337 let provenance = state
338 .preserved
339 .iter()
340 .find(|row| row.identity == identity && row.branch.branch == session.branch)
341 .map_or(Provenance::Complete, |row| row.branch.provenance);
342 Ok(SessionRecord {
343 lifecycle: if state.closed_sessions.contains(token) {
344 Lifecycle::Closed
345 } else {
346 Lifecycle::Open
347 },
348 session,
349 identity,
350 provenance,
351 // A retry link is a fact about run clones under a state root, and
352 // this provider keeps none: its sessions are records in a store, so
353 // there is no second copy of a branch for one to tell apart.
354 retried_by: None,
355 })
356 })
357 }
358
359 fn close_session(&self, token: &SessionToken) -> Result<Session> {
360 self.store.with(|state| {
361 let session = state::session_of(state, token)
362 .cloned()
363 .ok_or_else(|| unknown_session(token))?;
364 let emission = Emission {
365 stream: token.0.clone(),
366 // No identity label, because the real implementation carries none
367 // here: closing opens the stream fresh, and the label is stamped
368 // where a session is opened.
369 identity: None,
370 kind: EventKind::SessionClosed,
371 payload: object(json!({"token": token.0, "branch": session.branch})),
372 };
373 // Match the real provider's observable ordering: a follower reads the
374 // stream before consulting this state, so the terminator must exist
375 // before `Closed` can be returned to that concurrent reader.
376 events::emit(&emission);
377 state.closed_sessions.insert(token.clone());
378 Ok(session)
379 })
380 }
381
382 /// Publish a session's branch, as far as a provider honestly can.
383 ///
384 /// The host side is performed rather than described: a change request is really
385 /// opened against the [`Hosting`] this was handed, really adopted when one is
386 /// already open, and really merged under the policy — so the six host methods
387 /// are exercised and what the host recorded is what a journey reads back.
388 ///
389 /// The repository side is not, and none of it is claimed. There is no origin to
390 /// fetch from, no tree to build in, no push, and no lock to queue behind, so no
391 /// `fetch`, `push`, `lock-wait`, `lock-acquired`, or `merge-queued` event is
392 /// emitted. What is emitted is what was decided: the change that was opened, and
393 /// the merge that landed.
394 fn publish(
395 &self,
396 token: &SessionToken,
397 request: &PublishRequest,
398 hosting: &dyn Hosting,
399 ) -> Result<Publication> {
400 let (publication, emissions) = self.store.with(|state| {
401 let session = state::session_of(state, token)
402 .cloned()
403 .ok_or_else(|| unknown_session(token))?;
404 let identity = state::identity_for(state, token)?;
405 // The publication rule the crate next door states, applied rather than
406 // restated: a reason that would not render as itself is refused here
407 // exactly where the real implementation refuses it, before any host is
408 // asked and before anything is recorded.
409 if let Some(reason) = &request.draft {
410 reason.checked()?;
411 }
412 let resolved = state.policy.unwrap_or(DEFAULT_PUBLICATION);
413 let policy = match request.policy {
414 Some(requested) => resolved.narrow(requested)?,
415 None => resolved,
416 };
417 let published = |outcome, emissions| {
418 (
419 Publication {
420 session: token.clone(),
421 branch: session.branch.clone(),
422 policy,
423 outcome,
424 },
425 emissions,
426 )
427 };
428 // A session that has already landed has nothing the base does not carry,
429 // which is what the real implementation reports for the same reason. One
430 // whose change request is merely open or queued has *not* landed, and
431 // publishing it again adopts that change rather than opening a second —
432 // so it falls through to the host, as it does there.
433 if state.publications.iter().any(|earlier| {
434 earlier.session == *token && matches!(earlier.outcome, PublishOutcome::Merged(_))
435 }) {
436 let (publication, emissions) =
437 published(PublishOutcome::NothingToPublish, Vec::new());
438 state.publications.push(publication.clone());
439 return Ok((publication, emissions));
440 }
441
442 let (outcome, emissions) = if policy == MergePolicy::LocalDirect {
443 // The same refusal the real publication makes at the same boundary: a
444 // local-direct publication opens no change request, so there is
445 // nothing to draft and the work would land carrying the very pin the
446 // draft exists to hold back.
447 match &request.draft {
448 Some(reason) => (
449 failed(&Error::Invalid {
450 reason: format!(
451 "{branch:?} was asked to publish as {asked}, and this identity \
452 publishes with local-direct, which opens no change request at \
453 all",
454 branch = session.branch,
455 asked = asked_for(reason),
456 ),
457 }),
458 Vec::new(),
459 ),
460 None => record_local_landing(&identity, &session, token),
461 }
462 } else {
463 match slug(&identity) {
464 Some(slug) => {
465 // Recorded as it happens, as next door: a publication that
466 // fails after opening, drafting or lifting keeps that record.
467 let mut emissions = Vec::new();
468 let outcome = match publish_as_change(
469 hosting,
470 &Publishing {
471 slug: &slug,
472 identity: &identity,
473 session: &session,
474 policy,
475 approvals: state.approvals.unwrap_or(DEFAULT_APPROVALS),
476 drafts: state.drafts.unwrap_or_default(),
477 request,
478 token,
479 },
480 &mut emissions,
481 ) {
482 Ok(outcome) => outcome,
483 // Once a publication has started, what stops it is an outcome
484 // rather than a refusal — the same split the real
485 // implementation keeps, so a caller reads one shape.
486 Err(error) => failed(&error),
487 };
488 (outcome, emissions)
489 }
490 None => (refusal(&identity), Vec::new()),
491 }
492 };
493 let (publication, emissions) = published(outcome, emissions);
494 state.publications.push(publication.clone());
495 Ok((publication, emissions))
496 })?;
497 for emission in &emissions {
498 events::emit(emission);
499 }
500 Ok(publication)
501 }
502
503 /// The same rows [`Vcs::recoverable`] answers with.
504 ///
505 /// This provider is *handed* its preserved work, and what it is handed is work
506 /// nobody published: there is no base here whose history could record a landing
507 /// and no tree to compare content against, so it withholds nothing and the wider
508 /// question has the same answer as the narrower one.
509 fn preserved(&self, scope: Scope) -> Result<Vec<Recoverable>> {
510 self.recoverable(scope)
511 }
512
513 fn recoverable(&self, scope: Scope) -> Result<Vec<Recoverable>> {
514 self.store.with(|state| {
515 let wanted = match &scope {
516 Scope::All => None,
517 Scope::Repo(repo) => Some(
518 state::identity_of(state, repo)
519 .map(|identity| identity.origin.clone())
520 .ok_or_else(|| Error::Invalid {
521 reason: format!(
522 "{repo:?} does not name a repository this provider knows; {}",
523 state::known(state)
524 ),
525 })?,
526 ),
527 };
528 // Newest first, as the real implementation reports them.
529 Ok(state
530 .preserved
531 .iter()
532 .rev()
533 .filter(|row| wanted.as_ref().is_none_or(|key| *key == row.identity))
534 .cloned()
535 .map(|row| held(state, row))
536 .collect())
537 })
538 }
539}
540
541/// Say which live session still holds a row's branch, when one of this provider's
542/// does.
543///
544/// Read here rather than written when the work was preserved, because closing the
545/// session is what ends the hold and a row that had frozen the answer would go on
546/// saying somebody is in there. What makes it *live* is the same thing that makes it
547/// live for the real implementation in-process: the session is open, and the process
548/// that opened it is this one. A row a scenario seeded the answer into keeps it — a
549/// hand-written state says what it means to say.
550fn held(state: &VcsState, row: Recoverable) -> Recoverable {
551 if row.held_by.is_some() {
552 return row;
553 }
554 let holder = state.sessions.iter().find(|session| {
555 session.branch == row.branch.branch
556 && state.session_identities.get(&session.token) == Some(&row.identity)
557 && !state.closed_sessions.contains(&session.token)
558 });
559 Recoverable {
560 held_by: holder.map(|session| HeldBy {
561 token: Some(session.token.clone()),
562 worktree: session.worktree.clone(),
563 holding: Holding::OwnerRunning,
564 }),
565 ..row
566 }
567}
568
569/// The refusal a session this provider never opened meets.
570fn unknown_session(token: &SessionToken) -> Error {
571 Error::Invalid {
572 reason: format!(
573 "no session {:?} is open; `onevcs session open` prints a token",
574 token.0
575 ),
576 }
577}
578
579/// The `owner/name` slug an identity key spells, when it is a GitHub one.
580///
581/// The host is checked rather than assumed, exactly as the real implementation
582/// checks it: a GitLab origin has the same three segments, and a provider that
583/// published one anyway would let a journey pass where the real run answers that
584/// nobody has implemented that host.
585fn slug(identity: &str) -> Option<String> {
586 let mut parts = identity.split('/');
587 let (host, owner, name) = (parts.next()?, parts.next()?, parts.next()?);
588 if parts.next().is_some() || host != DEFAULT_HOST || owner.is_empty() || name.is_empty() {
589 return None;
590 }
591 Some(format!("{owner}/{name}"))
592}
593
594/// Record that a `local-direct` publication landed — record, and nothing more.
595///
596/// Landing one is entirely repository-side work, a squash built detached and
597/// pushed, which this provider does not perform and must not be named as if it
598/// did. What it does is decide the outcome and emit the completion that says so.
599fn record_local_landing(
600 identity: &str,
601 session: &Session,
602 token: &SessionToken,
603) -> (PublishOutcome, Vec<Emission>) {
604 let sha = events::stable_sha(&["publish", &token.0, &session.branch]);
605 let emission = Emission {
606 stream: token.0.clone(),
607 identity: Some(identity.to_owned()),
608 kind: EventKind::MergeCompleted,
609 // Every record of a landing says when the base received it and at which
610 // commit, as `onevcs`'s own does: here, the moment this provider decided it.
611 payload: object(json!({
612 "identity": identity,
613 "sha": sha,
614 "base": session.base,
615 "landed_at": events::timestamp(),
616 "landing": sha,
617 })),
618 };
619 (PublishOutcome::Merged(Sha(sha)), vec![emission])
620}
621
622/// Everything one provider publication knows about itself.
623struct Publishing<'a> {
624 slug: &'a str,
625 identity: &'a str,
626 session: &'a Session,
627 policy: MergePolicy,
628 approvals: Approvals,
629 drafts: Drafts,
630 request: &'a PublishRequest,
631 token: &'a SessionToken,
632}
633
634impl Publishing<'_> {
635 /// One event on the session's own stream, under this identity.
636 fn emission(&self, kind: EventKind, payload: Value) -> Emission {
637 Emission {
638 stream: self.token.0.clone(),
639 identity: Some(self.identity.to_owned()),
640 kind,
641 payload: object(payload),
642 }
643 }
644}
645
646/// Publish as a change request: open the session's change on the host, or adopt
647/// the one it already holds, watch its required checks, and then do with it what
648/// the policy asks.
649///
650/// **The draft lifecycle, as next door.** A publication carrying no reason opens its
651/// change request as a draft while the checks run — or, adopting a draft, records it
652/// as awaiting them rather than lifting it on the spot — and once they are green lifts
653/// it before asking for any merge, or keeps it under `change-open` with approvals
654/// required. `drafts: {disabled: true}` opens ready and lifts an adopted draft at once.
655///
656/// **What this provider cannot do is wait.** It has no clock, so each phase of the
657/// watch is **one reading** of the host: a reading that has not settled is the bound
658/// elapsing (`checks-unsettled`), and a draft on which no required check has run is
659/// the grace window elapsing — it is lifted early, and the next reading is the host's
660/// answer after the lift, which is where [`HostState::checks_after_lift`](crate::HostState::checks_after_lift)
661/// comes in. A consumer drives each row of the lifecycle by seeding the reading it
662/// wants rather than by timing one.
663fn publish_as_change(
664 hosting: &dyn Hosting,
665 publishing: &Publishing<'_>,
666 emissions: &mut Vec<Emission>,
667) -> Result<PublishOutcome> {
668 let Publishing {
669 slug,
670 session,
671 policy,
672 request,
673 ..
674 } = *publishing;
675 let host = hosting.for_repo(slug)?;
676 // Who the host believes is calling travels with the change, as it does in the
677 // real publication and for the same reason.
678 let author = host.authenticated_user()?;
679 let lifecycle = !publishing.drafts.disabled.unwrap_or(false);
680 // Checked before anything is opened, as the real publication checks it before
681 // anything is pushed.
682 if lifecycle {
683 grace_seconds()?;
684 }
685 let existing = host.find_changes(&session.branch, &session.base)?;
686 // Whether the host already held it, which is what decides whether there can be a
687 // draft to lift — and, as next door, keeps a publication that opens its own change
688 // request from asking this host anything extra.
689 let adopted = !existing.is_empty();
690 let change = match existing.into_iter().next() {
691 Some(change) => change,
692 None => host.open_change(ChangeSpec {
693 head: session.branch.clone(),
694 base: session.base.clone(),
695 // A requested title has been checked by the conversion that built it, so
696 // this provider cannot accept one the real publication would refuse. The
697 // real implementation takes the subject from the branch's commits when no
698 // title was requested, and a provider has no commits to read — so an
699 // unrequested title names the branch instead.
700 title: request
701 .title
702 .as_deref()
703 .map_or_else(|| format!("Publish {}", session.branch), str::to_owned),
704 // Verbatim, and nothing when there is none — the real publication
705 // composes no body either, so a provider that composed one would be a
706 // consumer's suite proving a change request nobody opens.
707 body: request.body.clone(),
708 // The reason travels to the host and no further, as it does there: what
709 // the host does with it is open the change as a draft.
710 draft: request.draft.clone(),
711 draft_awaiting_checks: lifecycle && request.draft.is_none(),
712 })?,
713 };
714 emissions.push(publishing.emission(
715 EventKind::ChangeOpened,
716 json!({
717 "url": change.url.to_string(),
718 "host": "github",
719 "id": change.id.0,
720 "base": change.base,
721 "author": author,
722 }),
723 ));
724 if let Some(reason) = &request.draft {
725 // The same question the real publication asks, for the same reason: a host
726 // that ignored the request, or a change request already open for review,
727 // would otherwise be reported as held back while it can land.
728 if !host.is_draft(&change)? {
729 return Err(Error::Invalid {
730 reason: format!(
731 "{url} is open for review on the host, and this publication asked for \
732 {asked}. A change that is open can land, so reporting it as a draft would \
733 say the work is held back when nothing is holding it",
734 url = change.url,
735 asked = asked_for(reason),
736 ),
737 });
738 }
739 // The publication record, and the only place the reason is written: the real
740 // implementation writes nothing of it into the change request, and a provider
741 // that did would be a consumer's suite proving a body nobody renders. The
742 // payload is the reason's own serialized form — its `kind` and every field of
743 // that kind — under the change request it holds, exactly as next door.
744 let mut drafted = object(json!({
745 "url": change.url.to_string(),
746 "id": change.id.0,
747 "base": change.base,
748 }));
749 if let serde_json::Value::Object(fields) =
750 serde_json::to_value(reason).expect("a draft reason serializes")
751 {
752 drafted.extend(fields);
753 }
754 emissions.push(publishing.emission(EventKind::ChangeDrafted, Value::Object(drafted)));
755 // Under every policy: a draft is unmergeable in that state, so nothing below
756 // asks this host to merge it.
757 return Ok(PublishOutcome::ChangeDraft(change.url.clone()));
758 }
759 // Three answers about whether the host holds it as a draft, told apart as next
760 // door: a host that was never taught to draft one is holding nothing, and a host
761 // that *could not say* is a refusal rather than a change nobody is holding.
762 // Asked of a change this publication just opened only under the lifecycle, which
763 // asked for a draft: without it, one opened moments ago without a reason is one
764 // nobody drafted, and a host written against the earlier surface is asked nothing.
765 let is_draft = match if lifecycle || adopted {
766 host.is_draft(&change)
767 } else {
768 Ok(false)
769 } {
770 Ok(draft) => draft,
771 Err(Error::NotImplemented { .. }) => false,
772 Err(unreadable) => return Err(unreadable),
773 };
774 let drafted = if lifecycle {
775 if is_draft {
776 emissions.push(publishing.emission(
777 EventKind::ChangeDrafted,
778 json!({
779 "url": change.url.to_string(),
780 "id": change.id.0,
781 "base": change.base,
782 "kind": "awaiting-checks",
783 }),
784 ));
785 }
786 is_draft
787 } else {
788 // Without the lifecycle, publishing without a reason lifts an adopted draft
789 // on the spot, as it always did; one opened moments ago is one nobody drafted.
790 if adopted && is_draft {
791 lift(host.as_ref(), publishing, &change, emissions)?;
792 }
793 false
794 };
795
796 // Every change policy watches — except that `change-auto` on a change nobody
797 // drafted arms the host's own merge and leaves the checks to it, as next door.
798 // Next door's merge watch still refuses a required check that already concluded
799 // red rather than arm a merge over it, so this reading does too; one still
800 // running arms it, and the host's hold is what the outcome reports.
801 let still_draft = if drafted || policy != MergePolicy::ChangeAuto {
802 watch(host.as_ref(), publishing, &change, drafted, emissions)?
803 } else {
804 let answered = host.change_checks(&change)?;
805 let checks: Vec<&Check> = answered.checks.iter().collect();
806 if let Some(failed) = red(&checks, &standings(&checks, &Declared::Unknown)) {
807 return Err(failed);
808 }
809 false
810 };
811 if still_draft {
812 if policy == MergePolicy::ChangeOpen && publishing.approvals == Approvals::Required {
813 emissions.push(publishing.emission(
814 EventKind::DraftKeptForReview,
815 json!({
816 "url": change.url.to_string(),
817 "id": change.id.0,
818 "base": change.base,
819 }),
820 ));
821 return Ok(PublishOutcome::ChangeReviewDraft(change.url.clone()));
822 }
823 // Lifted before any merge is asked for: this host, like GitHub, will neither
824 // merge a draft nor arm its own merge on one.
825 lift(host.as_ref(), publishing, &change, emissions)?;
826 }
827
828 if policy == MergePolicy::ChangeOpen {
829 return Ok(PublishOutcome::ChangeOpen(change.url.clone()));
830 }
831 Ok(match host.merge(&change, policy)? {
832 MergeOutcome::Merged(sha) => {
833 // When the host says the base received it, as `onevcs`'s own record of a
834 // landing carries it. Where the host cannot say there is no landing commit
835 // in this world to read a time from either, so the record says nobody can
836 // time it rather than claim the moment this provider saw the merge.
837 let landed_at = host
838 .merge_time(&change)
839 .ok()
840 .flatten()
841 .and_then(|spelled| events::moment(&spelled));
842 emissions.push(publishing.emission(
843 EventKind::ChangeMerged,
844 json!({
845 "url": change.url.to_string(),
846 "sha": sha.0,
847 "landed_at": landed_at,
848 "landing": sha.0,
849 }),
850 ));
851 emissions.push(publishing.emission(
852 EventKind::MergeCompleted,
853 json!({
854 "identity": publishing.identity,
855 "sha": sha.0,
856 "landed_at": landed_at,
857 "landing": sha.0,
858 }),
859 ));
860 PublishOutcome::Merged(sha)
861 }
862 MergeOutcome::Queued => PublishOutcome::Queued(change.url.clone()),
863 MergeOutcome::Open => PublishOutcome::ChangeOpen(change.url.clone()),
864 })
865}
866
867/// Take a change out of its draft on the host, and record the lift.
868fn lift(
869 host: &dyn RemoteHost,
870 publishing: &Publishing<'_>,
871 change: &ChangeRequest,
872 emissions: &mut Vec<Emission>,
873) -> Result<()> {
874 host.ready_for_review(change)?;
875 emissions.push(publishing.emission(
876 EventKind::DraftLifted,
877 json!({"url": change.url.to_string(), "id": change.id.0}),
878 ));
879 Ok(())
880}
881
882/// Which checks a merge requires, as the host said: nothing, these names, or unknown.
883enum Declared {
884 Nothing,
885 Names(BTreeSet<String>),
886 Unknown,
887}
888
889/// Each required check's name and the state its entries add up to — `None` where the
890/// host reported none — ranked as next door: red, then running, then passed, then no
891/// verdict, then skipped.
892fn standings(checks: &[&Check], declared: &Declared) -> Vec<(String, Option<CheckState>)> {
893 let names: BTreeSet<String> = match declared {
894 Declared::Names(names) => names.clone(),
895 Declared::Nothing => BTreeSet::new(),
896 Declared::Unknown => checks
897 .iter()
898 .filter(|check| check.required)
899 .map(|check| check.name.clone())
900 .collect(),
901 };
902 let rank = |state: CheckState| match state {
903 CheckState::Failed => 0,
904 CheckState::Pending => 1,
905 CheckState::Passed => 2,
906 CheckState::NoVerdict => 3,
907 CheckState::Skipped => 4,
908 };
909 names
910 .into_iter()
911 .map(|name| {
912 let state = checks
913 .iter()
914 .filter(|check| check.name == name)
915 .map(|check| check.state())
916 .min_by_key(|state| rank(*state));
917 (name, state)
918 })
919 .collect()
920}
921
922/// The refusal for the first required check that concluded red.
923fn red(checks: &[&Check], standing: &[(String, Option<CheckState>)]) -> Option<Error> {
924 let (name, _) = standing
925 .iter()
926 .find(|(_, state)| *state == Some(CheckState::Failed))?;
927 let check = checks
928 .iter()
929 .find(|check| &check.name == name && check.state() == CheckState::Failed)?;
930 Some(Error::ChecksFailed {
931 reason: format!(
932 "required check {:?} concluded {}.",
933 check.name,
934 check
935 .conclusion
936 .as_deref()
937 .unwrap_or("without a conclusion")
938 ),
939 })
940}
941
942/// Whether every required check lets the change through: passed, or skipped.
943fn through(standing: &[(String, Option<CheckState>)]) -> bool {
944 standing
945 .iter()
946 .all(|(_, state)| matches!(state, Some(CheckState::Passed | CheckState::Skipped)))
947}
948
949/// The skipped required checks, by name.
950fn skipped(standing: &[(String, Option<CheckState>)]) -> Vec<String> {
951 standing
952 .iter()
953 .filter(|(_, state)| *state == Some(CheckState::Skipped))
954 .map(|(name, _)| name.clone())
955 .collect()
956}
957
958/// The watch, one reading per phase: `Ok(true)` is green on a draft still standing,
959/// `Ok(false)` green on a ready change. See [`publish_as_change`] for what a reading
960/// stands for.
961fn watch(
962 host: &dyn RemoteHost,
963 publishing: &Publishing<'_>,
964 change: &ChangeRequest,
965 drafted: bool,
966 emissions: &mut Vec<Emission>,
967) -> Result<bool> {
968 // `unread`: the requirement was read from the host's own per-check marking because
969 // its declaration could not be read, which the record says, as next door.
970 let settled = |emissions: &mut Vec<Emission>, skipped: Vec<String>, unread: Option<&str>| {
971 let verdict = if skipped.is_empty() {
972 "passed"
973 } else {
974 "passed-with-skipped"
975 };
976 let mut payload = json!({
977 "url": change.url.to_string(),
978 "id": change.id.0,
979 "head": change.head_sha.0,
980 "verdict": verdict,
981 "skipped": skipped,
982 });
983 if let Some(because) = unread {
984 eprintln!(
985 "onevcs: warning: the required checks on {} were read from the host's own \
986 per-check marking, because the declaration could not be read: {because}",
987 change.url
988 );
989 payload["requirement"] = json!({"read_from": "host-marking", "because": because});
990 }
991 emissions.push(publishing.emission(EventKind::ChecksSettled, payload));
992 };
993 let unsettled = |standing: &[(String, Option<CheckState>)]| Error::ChecksUnsettled {
994 reason: format!(
995 "the host had not settled its required checks on {}; still unsettled: {}",
996 change.url,
997 standing
998 .iter()
999 .filter(|(_, state)| !matches!(state, Some(CheckState::Passed)))
1000 .map(|(name, _)| format!("{name:?}"))
1001 .collect::<Vec<_>>()
1002 .join(", ")
1003 ),
1004 };
1005 let answered = host.change_checks(change)?;
1006 let complete = answered.complete();
1007 let checks: Vec<&Check> = answered.checks.iter().collect();
1008
1009 if !drafted {
1010 let standing = standings(&checks, &Declared::Unknown);
1011 if let Some(failed) = red(&checks, &standing) {
1012 return Err(failed);
1013 }
1014 if !through(&standing) {
1015 return Err(unsettled(&standing));
1016 }
1017 settled(emissions, skipped(&standing), None);
1018 return Ok(false);
1019 }
1020
1021 // The draft: what the host requires, and a skipped check has not run.
1022 let (mut declared, mut unread) = match host.required_checks_on(&change.base) {
1023 Ok(answer) if answer.checks.is_empty() && answer.complete() => (Declared::Nothing, None),
1024 Ok(answer) if !answer.checks.is_empty() => (Declared::Names(answer.checks), None),
1025 Ok(_) => (
1026 Declared::Unknown,
1027 Some(format!(
1028 "the host answered only in part about which checks a merge into {} requires",
1029 change.base
1030 )),
1031 ),
1032 Err(refused) => (
1033 Declared::Unknown,
1034 Some(format!(
1035 "the host would not say which checks a merge into {} requires: {refused}",
1036 change.base
1037 )),
1038 ),
1039 };
1040 if matches!(declared, Declared::Unknown)
1041 && complete
1042 && !checks.is_empty()
1043 && checks.iter().all(|check| !check.required)
1044 {
1045 declared = Declared::Nothing;
1046 unread = None;
1047 }
1048 if matches!(declared, Declared::Nothing) {
1049 settled(emissions, Vec::new(), None);
1050 return Ok(true);
1051 }
1052 let unread = unread.as_deref();
1053 let standing = standings(&checks, &declared);
1054 if let Some(failed) = red(&checks, &standing) {
1055 return Err(failed);
1056 }
1057 if !standing.is_empty()
1058 && standing
1059 .iter()
1060 .all(|(_, state)| *state == Some(CheckState::Passed))
1061 {
1062 settled(emissions, Vec::new(), unread);
1063 return Ok(true);
1064 }
1065 let not_run: Vec<String> = standing
1066 .iter()
1067 .filter(|(_, state)| matches!(state, None | Some(CheckState::Skipped)))
1068 .map(|(name, _)| name.clone())
1069 .collect();
1070 let running = standing
1071 .iter()
1072 .any(|(_, state)| matches!(state, Some(CheckState::Pending | CheckState::NoVerdict)));
1073 let unseen = matches!(declared, Declared::Unknown) && standing.is_empty();
1074 // The grace window, elapsed: where the host would not say what it requires and
1075 // checks ran on the draft with none skipped, its own marking is the answer.
1076 if unseen
1077 && !checks.is_empty()
1078 && checks
1079 .iter()
1080 .all(|check| check.state() != CheckState::Skipped)
1081 {
1082 settled(emissions, Vec::new(), unread);
1083 return Ok(true);
1084 }
1085 if running || (not_run.is_empty() && !unseen) {
1086 return Err(unsettled(&standing));
1087 }
1088
1089 // The grace window elapsed with nothing required run: lift early, say so, and read
1090 // the host again as a ready change whose checks must run after the lift.
1091 let snapshot: Vec<Check> = answered.checks.clone();
1092 lift(host, publishing, change, emissions)?;
1093 let warned = publishing.drafts.warn_on_early_lift.unwrap_or(true);
1094 let grace = grace_seconds()?;
1095 if warned {
1096 eprintln!(
1097 "onevcs: warning: {} was lifted out of its draft before its required checks ran on \
1098 it ({}): none had run within {grace}s, which is what a workflow that skips drafts \
1099 looks like",
1100 change.url,
1101 not_run.join(", ")
1102 );
1103 }
1104 emissions.push(publishing.emission(
1105 EventKind::DraftLiftedEarly,
1106 json!({
1107 "url": change.url.to_string(),
1108 "id": change.id.0,
1109 "base": change.base,
1110 "awaited": not_run,
1111 "grace_seconds": grace,
1112 "warned": warned,
1113 }),
1114 ));
1115 let after = host.change_checks(change)?.checks;
1116 let from_the_draft = |check: &Check| !ran_after_the_lift(check, &snapshot);
1117 let counted: Vec<&Check> = after
1118 .iter()
1119 .filter(|check| !(from_the_draft(check) && check.state() == CheckState::Skipped))
1120 .collect();
1121 let standing = standings(&counted, &declared);
1122 if let Some(failed) = red(&counted, &standing) {
1123 return Err(failed);
1124 }
1125 let rerun = after.iter().any(|check| {
1126 !from_the_draft(check)
1127 && (standing.is_empty() || standing.iter().any(|(name, _)| *name == check.name))
1128 });
1129 if !rerun {
1130 return Err(Error::ChecksUnsettled {
1131 reason: format!(
1132 "the required checks on {} did not re-run after its draft was lifted: no run \
1133 the host attached after the lift appeared, and the runs from while it was a \
1134 draft were skipped, which is not a verdict. The likely cause is a workflow that \
1135 does not trigger on `ready_for_review`",
1136 change.url
1137 ),
1138 });
1139 }
1140 if !through(&standing) {
1141 return Err(unsettled(&standing));
1142 }
1143 settled(emissions, skipped(&standing), unread);
1144 Ok(false)
1145}
1146
1147/// Whether `check` is a run the host attached after the lift rather than the draft's
1148/// run of it in `snapshot`, decided as the real publication decides it: by the run's
1149/// own identity, never by its conclusion or its status. A check the draft never
1150/// reported is new, and one it did report is new only where it reports a `started_at`
1151/// no draft-era run of that check reported, running or settled — so a consumer seeds a
1152/// re-run by giving it a start of its own.
1153fn ran_after_the_lift(check: &Check, snapshot: &[Check]) -> bool {
1154 let earlier: Vec<&Check> = snapshot
1155 .iter()
1156 .filter(|seen| seen.name == check.name)
1157 .collect();
1158 if earlier.is_empty() {
1159 return true;
1160 }
1161 check.started_at.as_ref().is_some_and(|started| {
1162 earlier
1163 .iter()
1164 .all(|seen| seen.started_at.as_ref() != Some(started))
1165 })
1166}
1167
1168/// The default grace window of the real publication, which is `onevcs`'s own and
1169/// private to it. A copy, so it is gated: `lifecycle.rs`'s
1170/// `the_grace_window_an_early_lift_records_is_the_one_onevcs_defaults_to` reads the
1171/// constant out of `onevcs`'s source and holds this to it.
1172const DEFAULT_DRAFT_GRACE_SECONDS: f64 = 120.0;
1173
1174/// The knob a real publication waits its grace window by.
1175const DRAFT_GRACE_ENV: &str = "ONEVCS_DRAFT_CHECKS_GRACE_SECONDS";
1176
1177/// The grace window a real publication would have waited out, as the operator set it
1178/// or the default — recorded on `draft-lifted-early` so the payload reads as the real
1179/// one does, though nothing here waits.
1180///
1181/// Validated as the real publication validates it, and refused by name where it is
1182/// not a finite number of seconds above zero: a provider that quietly took the default
1183/// would let a consumer's suite pass on a setting the real one refuses.
1184fn grace_seconds() -> Result<f64> {
1185 let Some(raw) = std::env::var_os(DRAFT_GRACE_ENV) else {
1186 return Ok(DEFAULT_DRAFT_GRACE_SECONDS);
1187 };
1188 let raw = raw.to_string_lossy().into_owned();
1189 let seconds: f64 = raw.trim().parse().map_err(|_| Error::Invalid {
1190 reason: format!("{DRAFT_GRACE_ENV} must be a number of seconds, not {raw:?}"),
1191 })?;
1192 if !seconds.is_finite() || seconds <= 0.0 {
1193 return Err(Error::Invalid {
1194 reason: format!(
1195 "{DRAFT_GRACE_ENV} must be a finite number of seconds above zero, not {raw:?}"
1196 ),
1197 });
1198 }
1199 Ok(seconds)
1200}
1201
1202/// The draft a publication asked for, as the two refusals next door spell it: "a
1203/// draft awaiting <repository> <target>", or "a draft held by the session that opened
1204/// it (<why>)". A restatement, so the refusals here read as the real ones do.
1205fn asked_for(reason: &DraftReason) -> String {
1206 match reason {
1207 DraftReason::AwaitingRelease {
1208 awaiting, target, ..
1209 } => format!("a draft awaiting {awaiting} {target}"),
1210 DraftReason::Held { because } => {
1211 format!("a draft held by the session that opened it ({because})")
1212 }
1213 }
1214}
1215
1216/// What a publication answers for an identity no change request can be opened
1217/// against.
1218///
1219/// Two failures rather than one, as the real implementation keeps them: an
1220/// identity that is not hosted at all is asking for the wrong policy, while a
1221/// hosted one on a host this build does not speak for is asking for an
1222/// implementation that has not arrived.
1223fn refusal(identity: &str) -> PublishOutcome {
1224 failed(&if identity.split('/').count() == 3 {
1225 Error::NotImplemented {
1226 operation: "RemoteHost for a host other than github.com",
1227 }
1228 } else {
1229 Error::Invalid {
1230 reason: format!(
1231 "identity {identity:?} is not a hosted repository, so it cannot publish a \
1232 change request; a local identity publishes with local-direct"
1233 ),
1234 }
1235 })
1236}
1237
1238/// One failure, as the outcome a publication that started and did not land is.
1239fn failed(error: &Error) -> PublishOutcome {
1240 PublishOutcome::Failed {
1241 // Through the crate's own mapping, so the kind a caller branches on is the
1242 // one the real implementation would report for the same failure.
1243 kind: FailureKind::of(error),
1244 reason: error.to_string(),
1245 // A provider has no execution checkout, so there is nowhere a branch could
1246 // have been handed back to and nothing to report about one.
1247 retained: None,
1248 }
1249}
1250
1251/// The argv that lands a preserved branch, as `recoverable` reports it.
1252///
1253/// Both verbs take the branch by name and the repository by path, the way the real
1254/// implementation reports them: the provenance decides which one, and neither
1255/// depends on the directory a reader happens to be standing in.
1256// llmlint: ignore[names_match_behavior] named for the public
1257// `Recoverable::recover_command` field it fills, which `onevcs` publishes and this
1258// crate must answer identically; a provider whose helper were named otherwise would
1259// read as filling something else.
1260fn recover_command(branch: &str, checkout: &Path, provenance: Provenance) -> Vec<String> {
1261 let verb = match provenance {
1262 Provenance::IncompleteStep => "recover",
1263 Provenance::Complete => "publish-branch",
1264 };
1265 vec![
1266 "onevcs".to_owned(),
1267 verb.to_owned(),
1268 branch.to_owned(),
1269 "--repo".to_owned(),
1270 checkout.display().to_string(),
1271 ]
1272}
1273
1274/// How a provenance kind is spelled in an event payload.
1275fn spell(provenance: Provenance) -> &'static str {
1276 match provenance {
1277 Provenance::Complete => "complete",
1278 Provenance::IncompleteStep => "incomplete-step",
1279 }
1280}
1281
1282fn object(value: Value) -> Map<String, Value> {
1283 value.as_object().cloned().unwrap_or_default()
1284}