onevcs/vcs.rs
1//! The repository side of the seam.
2
3use std::collections::BTreeSet;
4use std::path::{Path, PathBuf};
5
6use crate::error::{self, Error, Result};
7use crate::event::EventKind;
8use crate::host::{Hosting, Sha};
9use crate::landed::{self, Landed};
10use crate::publish::{Publication, PublishRequest};
11use crate::registry::Identity;
12use crate::session::{
13 HeldBy, Holding, Lifecycle, LineChange, Liveness, NetNegative, OnOrigin, PreservedBranch,
14 Provenance, Recoverable, Scope, Selection, Session, SessionRecord, SessionRequest,
15 SessionToken,
16};
17use crate::stream::Stream;
18use crate::workspace::{self, object};
19use crate::{git, label, lock, provenance, publish, store};
20
21use serde_json::json;
22use url::Url;
23
24/// Everything `onevcs` does to a repository, independent of which version
25/// control system is underneath.
26///
27/// The session record belongs here rather than beside it. It was written by [`Git`]
28/// directly for three releases, and the cost was exact: a session a supplied
29/// implementation had just opened was refused by `publish` and by `session close`
30/// as a session nobody opened, so the seam was declared and those commands went
31/// around it.
32pub trait Vcs {
33 /// Resolve an origin URL or a checkout path to the repository identity it
34 /// belongs to.
35 fn resolve_identity(&self, origin_or_path: &str) -> Result<Identity>;
36
37 /// Open a session: a clone of an execution checkout and an isolated worktree cut
38 /// from it — a warm pool slot where the host keeps one, else cut for this run —
39 /// held under an occupancy lease.
40 fn open_session(&self, req: SessionRequest) -> Result<Session>;
41
42 /// Re-attach to a session that already exists, claiming its free occupancy
43 /// lease.
44 fn adopt_session(&self, token: SessionToken) -> Result<Session>;
45
46 /// What this implementation recorded about a session it opened.
47 fn session(&self, token: &SessionToken) -> Result<SessionRecord>;
48
49 /// Release a session's worktree and its lease, keeping its branch.
50 fn close_session(&self, token: &SessionToken) -> Result<Session>;
51
52 /// Commit the session's work onto a branch that outlives it, recording why.
53 fn preserve(&self, s: &Session, provenance: Provenance) -> Result<PreservedBranch>;
54
55 /// Verify a session's branch and land it under its repository's policy.
56 ///
57 /// The one operation that reaches both interfaces — the repository side lands
58 /// the change and the host side opens and merges the change request — so the
59 /// host factory travels with the request rather than being reached for.
60 fn publish(
61 &self,
62 token: &SessionToken,
63 request: &PublishRequest,
64 hosting: &dyn Hosting,
65 ) -> Result<Publication>;
66
67 /// Every preserved-but-unpublished branch in scope, and what would land each.
68 ///
69 /// Unpublished is decided from history rather than inferred from content, and a
70 /// branch whose work reached the base is not one of these — including one the
71 /// base has moved a long way past since. The row that offered such a branch a
72 /// paste-ready `publish-branch` is what this excludes: following it re-opens a
73 /// change request for work the base already carries. A branch nothing can decide
74 /// about is still here, because it may be work nobody published; what it does not
75 /// carry is a command anybody should paste without looking.
76 fn recoverable(&self, scope: Scope) -> Result<Vec<Recoverable>>;
77
78 /// Every preserved branch in scope, whatever became of its work, each saying
79 /// what did become of it and what says so.
80 ///
81 /// [`recoverable`](Self::recoverable) is this without the ones whose work is on
82 /// the base, and is what somebody asking "what is left to publish" wants. This is
83 /// what somebody asking "what became of all of it" wants, and it is the one place
84 /// a withheld branch is reported rather than silently dropped — an exclusion
85 /// nobody can see is how preserved work goes missing.
86 ///
87 /// Required rather than defaulted to [`recoverable`](Self::recoverable): the
88 /// default would answer the *narrower* question under this one's name, and an
89 /// implementation whose wider answer is the same one has only to say so.
90 fn preserved(&self, scope: Scope) -> Result<Vec<Recoverable>>;
91
92 /// [`recoverable`](Self::recoverable), narrowed to the sessions a
93 /// [`Selection`] names.
94 ///
95 /// Defaulted rather than required, and the default is the whole meaning: a row is
96 /// answered with when some session record of this host naming its branch is one
97 /// the selection picked — by token, by carrying every label pair asked for, or by
98 /// both. So every implementation narrows identically, and what an override buys is
99 /// not a different answer but a cheaper one: an implementation that knows *where*
100 /// it looks can decline to look, which is what [`Git`] does with it. A selection
101 /// that asks nothing is the whole report, which is what every caller before this
102 /// existed asked for.
103 fn recoverable_matching(
104 &self,
105 scope: Scope,
106 selection: &Selection,
107 ) -> Result<Vec<Recoverable>> {
108 retain(self.recoverable(scope)?, selection)
109 }
110
111 /// [`preserved`](Self::preserved), narrowed the same way and for the same reason.
112 fn preserved_matching(&self, scope: Scope, selection: &Selection) -> Result<Vec<Recoverable>> {
113 retain(self.preserved(scope)?, selection)
114 }
115}
116
117/// The git implementation of [`Vcs`].
118///
119/// Stateless by design: everything it needs is the registry and the workspaces
120/// under the one state root, so two processes driving it see the same host.
121#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
122pub struct Git;
123
124impl Vcs for Git {
125 fn resolve_identity(&self, origin_or_path: &str) -> Result<Identity> {
126 let registry = store::load()?;
127 Ok(store::resolve(®istry, origin_or_path)?.identity)
128 }
129
130 fn open_session(&self, req: SessionRequest) -> Result<Session> {
131 let registry = store::load()?;
132 let (record, _stream) = workspace::open(®istry, &req)?;
133 Ok(record.session())
134 }
135
136 fn adopt_session(&self, token: SessionToken) -> Result<Session> {
137 let (record, _stream, _preserved) = workspace::adopt(&token.0)?;
138 Ok(record.session())
139 }
140
141 fn session(&self, token: &SessionToken) -> Result<SessionRecord> {
142 let record = workspace::load(&token.0)?;
143 let trailers = provenance::configured()?;
144 // Where the branch is now. A session's clone holds it for the session's
145 // life, and a run root's clone keeps holding it after the close — but a slot's
146 // clone lets a copied branch go when the slot is returned, and the execution
147 // checkout the hand-back copied it to is then where the branch lives.
148 let held_in = match git::branch_exists(&record.clone, &record.branch) {
149 true => record.clone.clone(),
150 false => record.execution_checkout.clone(),
151 };
152 let target = publish::standing_target(&record)?;
153 let base = base_ref(&held_in, target.base());
154 Ok(SessionRecord {
155 session: record.session(),
156 identity: record.identity.clone(),
157 lifecycle: record.state,
158 // Read off the branch rather than remembered: an adoption that met a
159 // dirty tree writes the marker, and a recovery clears it, so what the
160 // branch carries now is the only answer that stays true.
161 provenance: provenance::provenance_of(&held_in, &base, &record.branch, &trailers)?,
162 retried_by: record
163 .retried_by
164 .as_ref()
165 .map(|token| SessionToken(token.to_string())),
166 })
167 }
168
169 fn close_session(&self, token: &SessionToken) -> Result<Session> {
170 let record = workspace::close(&token.0)?;
171 Ok(record.session())
172 }
173
174 fn preserve(&self, s: &Session, provenance: Provenance) -> Result<PreservedBranch> {
175 let record = workspace::load(&s.token.0)?;
176 let mut stream = Stream::open(&s.token.0)?;
177 preserve_into(&record, &mut stream, provenance)
178 }
179
180 fn publish(
181 &self,
182 token: &SessionToken,
183 request: &PublishRequest,
184 hosting: &dyn Hosting,
185 ) -> Result<Publication> {
186 publish::run_for_session(token, request, hosting)
187 }
188
189 fn recoverable(&self, scope: Scope) -> Result<Vec<Recoverable>> {
190 collect(&scope, Reporting::UnpublishedOnly)
191 }
192
193 fn preserved(&self, scope: Scope) -> Result<Vec<Recoverable>> {
194 collect(&scope, Reporting::Everything)
195 }
196
197 fn recoverable_matching(
198 &self,
199 scope: Scope,
200 selection: &Selection,
201 ) -> Result<Vec<Recoverable>> {
202 collect_matching(&scope, Reporting::UnpublishedOnly, selection)
203 }
204
205 fn preserved_matching(&self, scope: Scope, selection: &Selection) -> Result<Vec<Recoverable>> {
206 collect_matching(&scope, Reporting::Everything, selection)
207 }
208}
209
210/// Commit whatever the worktree holds onto the branch, and hand the branch back to
211/// the execution checkout.
212///
213/// The session's clone is disposable, so a branch that is not copied out is lost
214/// with it. The copy is fast-forward only, to protect a concurrent session holding
215/// the same branch name, and a refusal is reported rather than swallowed: the
216/// branch a caller is then told about would name something nothing outside this
217/// session carries.
218pub fn preserve_into(
219 record: &workspace::Record,
220 stream: &mut Stream,
221 kind: Provenance,
222) -> Result<PreservedBranch> {
223 let trailers = provenance::configured()?;
224 if git::is_dirty(&record.worktree)? {
225 git::add_all(&record.worktree)?;
226 let message = match kind {
227 Provenance::Complete => format!("chore: preserve work on {}", record.branch),
228 Provenance::IncompleteStep => provenance::incomplete_message(
229 &format!("work on {}", record.branch),
230 record.change_base.as_deref(),
231 &trailers,
232 ),
233 };
234 let sha = git::commit(&record.worktree, &message)?;
235 stream.emit(
236 EventKind::CommitPreserved,
237 object(json!({
238 "branch": record.branch,
239 "sha": sha,
240 "provenance": spell_provenance(kind),
241 })),
242 );
243 }
244
245 let copied = git::copy_branch(&record.clone, &record.execution_checkout, &record.branch)?;
246 if !copied {
247 return Err(Error::Invalid {
248 reason: format!(
249 "the execution checkout {} refused branch {:?}; it holds work this session's \
250 clone does not, so nothing outside the session carries this branch",
251 record.execution_checkout.display(),
252 record.branch
253 ),
254 });
255 }
256
257 let base = judging_base(record)?;
258 Ok(PreservedBranch {
259 branch: record.branch.to_string(),
260 base: record.base.to_string(),
261 provenance: provenance::provenance_of(&record.clone, &base, &record.branch, &trailers)?,
262 change_url: None,
263 change_base: record.change_base.as_ref().map(ToString::to_string),
264 })
265}
266
267/// How a provenance kind is spelled in an event payload.
268pub fn spell_provenance(kind: Provenance) -> &'static str {
269 match kind {
270 Provenance::Complete => "complete",
271 Provenance::IncompleteStep => "incomplete-step",
272 }
273}
274
275/// The ref a session's branch is judged against: the base its publication resolves
276/// from what the clone holds now — the branch below for a stacked session, until the
277/// clone has seen the root carry it — as [`base_ref`] spells it.
278fn judging_base(record: &workspace::Record) -> Result<String> {
279 let target = publish::standing_target(record)?;
280 Ok(base_ref(&record.clone, target.base()))
281}
282
283/// The ref a branch's commits are counted against: the remote-tracking base when
284/// the repository has one, and the local base otherwise.
285pub fn base_ref(repo: &Path, base: &str) -> String {
286 let remote = format!("origin/{base}");
287 if git::ref_exists(repo, &format!("refs/remotes/{remote}")) {
288 remote
289 } else {
290 base.to_owned()
291 }
292}
293
294/// What a branch held in `repo` is judged against: the identity's base *now*,
295/// named by a commit `repo` can reach.
296///
297/// A checkout's remote-tracking refs are frozen at its last fetch, so its own
298/// `origin/main` can be many merges behind. Judged against that, a branch whose work
299/// landed weeks ago still looks like work nobody published — and a name whose meaning
300/// is spent still looks like a name that means something. Which is why the repository
301/// is asked *through* the object store of the checkout every publication
302/// fast-forwards: the commit the base stands at is one that checkout has by
303/// construction, so a copy that never fetched it can still read it. What is left over
304/// — a repository that cannot reach the base even so — is judged against its own view
305/// and reported as behind, which is what keeps the tiers below a record from closing
306/// the question from a history that stops short of the evidence.
307// The two states this answers with are deliberately one type, as `base_ref` beside it
308// and `Landing::compared_change_base` already are: what comes back is a *comparison
309// target*, and git resolves a ref name and a commit id identically at every call that
310// takes one. Distinguishing them in the type would only oblige each of those call
311// sites to collapse the distinction again, and the thing that must not be confused
312// with either — a branch name this crate writes — is `Ref`, which neither of these is.
313// llmlint: ignore[invalid_states_unrepresentable] a comparison target git resolves either way
314pub fn judged_against<'a>(
315 repo: impl Into<git::Asked<'a>>,
316 base: &str,
317 current: Option<&Sha>,
318) -> String {
319 let repo = repo.into();
320 match current {
321 Some(sha) if git::has_commit(repo, sha) => sha.0.clone(),
322 _ => base_ref(repo.path(), base),
323 }
324}
325
326/// The commit a checkout's base ref stands at.
327///
328/// Asked of the publication checkout, which is the one every publication
329/// fast-forwards and therefore the freshest view of the base this host keeps — not
330/// a guarantee of the newest there is, which only the remote can answer for.
331pub fn base_commit(checkout: &Path, base: &str) -> Option<Sha> {
332 git::tip(checkout, &base_ref(checkout, base)).map(Sha)
333}
334
335/// Which branches a report is asked to carry.
336#[derive(Debug, Clone, Copy, PartialEq, Eq)]
337pub enum Reporting {
338 /// Every branch whose work is not known to have reached the base, which is what
339 /// [`Vcs::recoverable`] answers.
340 UnpublishedOnly,
341 /// Every preserved branch, whatever became of its work — the ones that reached
342 /// the base included, which is what [`Vcs::preserved`] answers.
343 Everything,
344}
345
346/// Which preserved branches a [`Selection`] asks about, read out of this host's
347/// session records, and where a process has to look to answer.
348///
349/// The three sets are one question asked three ways, and the last two are what makes
350/// a filter cheap rather than merely narrow: a report that selected its rows at the
351/// end would still have walked every checkout of every identity and decided every
352/// branch in them, which is exactly the cost a filtered read exists to avoid. So a
353/// selection names sessions, the sessions name branches, and the branches name the
354/// few places they can be — and everything else is never opened.
355struct Narrowed {
356 /// The identities the selected sessions belong to.
357 identities: BTreeSet<String>,
358 /// The `(identity, branch)` pairs they hold or held.
359 branches: BTreeSet<(String, String)>,
360 /// The checkouts that can be holding one of those branches: each selected
361 /// session's own clone, and the execution checkout its branch is handed back to
362 /// when it closes. A branch this host knows about is in one of the two.
363 checkouts: BTreeSet<PathBuf>,
364}
365
366impl Narrowed {
367 /// Whether a branch is one of the ones asked about.
368 fn wants(&self, identity: &str, branch: &str) -> bool {
369 self.branches
370 .contains(&(identity.to_owned(), branch.to_owned()))
371 }
372}
373
374/// What a selection asks for, or `None` where it asks nothing.
375///
376/// A session token no record on this host names is refused **by name**: "nothing of
377/// that session is left to publish" and "there is no such session" are different
378/// answers, and a consumer sequencing its own work behind the first must never be
379/// handed it in place of the second. A label pair nothing carries is the first of
380/// those and answers an empty report, because a run whose branches all landed is
381/// exactly the run that has nothing here.
382fn narrowed(selection: &Selection, sessions: &[workspace::Record]) -> Result<Option<Narrowed>> {
383 if selection.is_empty() {
384 return Ok(None);
385 }
386 for token in &selection.sessions {
387 if !sessions.iter().any(|record| *record.token == *token.0) {
388 return Err(error::invalid(format!(
389 "no session record on this host names {}; \
390 `onevcs session holders REPO` lists the sessions of one repository",
391 token.0
392 )));
393 }
394 }
395 let mut narrowed = Narrowed {
396 identities: BTreeSet::new(),
397 branches: BTreeSet::new(),
398 checkouts: BTreeSet::new(),
399 };
400 for record in sessions {
401 let named = selection.sessions.is_empty()
402 || selection
403 .sessions
404 .iter()
405 .any(|token| *record.token == *token.0);
406 if !named || !label::matches(&record.labels, &selection.labels) {
407 continue;
408 }
409 narrowed.identities.insert(record.identity.clone());
410 narrowed
411 .branches
412 .insert((record.identity.clone(), record.branch.to_string()));
413 narrowed.checkouts.insert(record.clone.clone());
414 narrowed.checkouts.insert(record.execution_checkout.clone());
415 }
416 Ok(Some(narrowed))
417}
418
419/// The rows of an answer a selection asks about, for an implementation that has
420/// already made the whole answer.
421///
422/// The one definition of what a selection *means*, so the narrowing [`Git`] does to
423/// its scan and the narrowing every other implementation gets by default cannot come
424/// to disagree about which rows an answer holds.
425pub(crate) fn retain(rows: Vec<Recoverable>, selection: &Selection) -> Result<Vec<Recoverable>> {
426 let Some(narrowed) = narrowed(selection, &workspace::all()?)? else {
427 return Ok(rows);
428 };
429 Ok(rows
430 .into_iter()
431 .filter(|row| narrowed.wants(&row.identity, &row.branch.branch))
432 .collect())
433}
434
435/// Every preserved branch in scope, newest first, and whether its work landed.
436///
437/// Read-only in the strongest sense: it opens repositories to ask questions, writes
438/// nothing, and takes no lease, so it is safe to run beside live work — which is
439/// exactly when somebody reaches for it.
440pub fn collect(scope: &Scope, reporting: Reporting) -> Result<Vec<Recoverable>> {
441 collect_matching(scope, reporting, &Selection::default())
442}
443
444/// The same report, narrowed to what a [`Selection`] asks about before it is made.
445///
446/// Every git read inside one invocation is answered once ([`git::memoized`]), because
447/// this asks the same questions of the same repositories over and over — one base tip
448/// per checkout read by every branch in it, one branch log read by four provenance
449/// readers — and each asking used to be a process. Each identity is scanned under a
450/// memo of its own, on its own thread, since no two identities share a repository to
451/// ask about. The memo is this call's and ends with it: the repositories a library caller holds move between invocations, and a
452/// fact remembered across one would be this report answering from a tree that has
453/// changed.
454pub fn collect_matching(
455 scope: &Scope,
456 reporting: Reporting,
457 selection: &Selection,
458) -> Result<Vec<Recoverable>> {
459 git::memoized(|| collected(scope, reporting, selection))
460}
461
462fn collected(
463 scope: &Scope,
464 reporting: Reporting,
465 selection: &Selection,
466) -> Result<Vec<Recoverable>> {
467 let registry = store::load()?;
468 let (rules, _source) = crate::policy::load(®istry)?;
469 let trailers = provenance::from_rules(&rules);
470 let sessions = workspace::all()?;
471 // Which sessions were asked about, and therefore which identities, checkouts and
472 // branch names the scan below may stop at. Read before anything is opened, so a
473 // token naming no record is refused before a single repository is.
474 let narrowed = narrowed(selection, &sessions)?;
475 // What this host's own runs recorded, read once: the change request each branch
476 // opened and any landing seen for it. A gap in the streams is not reported here —
477 // this report has nowhere to say one — and costs only certainty: a branch whose
478 // record could not be read falls to a lower tier and is judged from the base's own
479 // history instead.
480 let streams = crate::status::recorded_streams(&mut Vec::new())?;
481 let wanted = match scope {
482 Scope::All => None,
483 Scope::Repo(repo) => Some(store::resolve(®istry, repo)?.key),
484 };
485
486 // Once per identity rather than once per checkout of one, because the places a
487 // branch of it can be are a property of the identity — and they are read from
488 // the one list the verbs that go on to *land* a branch read, so this report
489 // cannot come to offer branches nothing can reach, or miss ones something can.
490 let mut identities: Vec<&str> = registry
491 .checkouts
492 .values()
493 .map(|checkout| checkout.identity.as_str())
494 .collect();
495 identities.sort_unstable();
496 identities.dedup();
497 identities.retain(|&identity| {
498 if wanted.as_ref().is_some_and(|key| key != identity) {
499 return false;
500 }
501 // An identity none of the selected sessions belongs to holds none of the
502 // branches asked about, so nothing of it is opened at all — not its
503 // publication checkout, not its base, not one of its clones.
504 if narrowed
505 .as_ref()
506 .is_some_and(|only| !only.identities.contains(identity))
507 {
508 return false;
509 }
510 true
511 });
512 let scan = Scan {
513 registry: ®istry,
514 streams: &streams,
515 sessions: &sessions,
516 trailers: &trailers,
517 narrowed: &narrowed,
518 reporting,
519 };
520 // Each identity is scanned on its own thread, and the answers are joined in the
521 // order the identities are named. That changes no verdict and no row: every
522 // question below is asked of one identity's own checkouts, and every key the scan
523 // deduplicates on names the identity, so no identity's answer reads another's.
524 // What it changes is the clock — the report is a long series of git reads against
525 // repositories that share nothing, and a host with a dozen identities used to wait
526 // for each of them in turn.
527 let mut rows: Vec<(Option<u64>, Recoverable)> = Vec::new();
528 // The branches whose work reached the base, kept aside rather than dropped: they
529 // are what `preserved` adds, and a copy of a name whose work the base already
530 // carries must not answer for a copy of it elsewhere that still holds work, so
531 // they join the answer only where no such copy did.
532 let mut withheld_rows: Vec<(Option<u64>, Recoverable)> = Vec::new();
533 let mut seen: Vec<(String, String)> = Vec::new();
534 for scanned in concurrently(&identities, |identity| {
535 git::memoized(|| scanned(identity, &scan))
536 }) {
537 let scanned = scanned?;
538 rows.extend(scanned.rows);
539 withheld_rows.extend(scanned.withheld_rows);
540 seen.extend(scanned.seen);
541 }
542 // A landed copy answers only where nothing holding work under that name did.
543 let mut kept: Vec<(String, String)> = Vec::new();
544 for (at, row) in withheld_rows {
545 let key = (row.identity.clone(), row.branch.branch.clone());
546 if seen.contains(&key) || kept.contains(&key) {
547 continue;
548 }
549 kept.push(key);
550 rows.push((at, row));
551 }
552 rows.sort_by_key(|(at, row)| {
553 (
554 std::cmp::Reverse(at.unwrap_or(0)),
555 row.branch.branch.clone(),
556 )
557 });
558 Ok(rows.into_iter().map(|(_, row)| row).collect())
559}
560
561/// What one `recoverable` read holds for the whole of its scan, lent to each
562/// identity's.
563struct Scan<'a> {
564 registry: &'a crate::registry::Registry,
565 streams: &'a [crate::status::Recorded],
566 sessions: &'a [workspace::Record],
567 trailers: &'a provenance::Trailers,
568 narrowed: &'a Option<Narrowed>,
569 reporting: Reporting,
570}
571
572/// One identity's share of the report: the rows it answers with, the rows it
573/// withheld because their work landed, and the names it answered for.
574struct Scanned {
575 rows: Vec<(Option<u64>, Recoverable)>,
576 withheld_rows: Vec<(Option<u64>, Recoverable)>,
577 seen: Vec<(String, String)>,
578}
579
580/// Scan every checkout of one identity for its preserved branches.
581fn scanned(identity: &str, scan: &Scan<'_>) -> Result<Scanned> {
582 let Scan {
583 registry,
584 streams,
585 sessions,
586 trailers,
587 narrowed,
588 reporting,
589 } = *scan;
590 let mut rows: Vec<(Option<u64>, Recoverable)> = Vec::new();
591 let mut withheld_rows: Vec<(Option<u64>, Recoverable)> = Vec::new();
592 let mut seen: Vec<(String, String)> = Vec::new();
593 // Every copy of a name this report has already put the question to, with the
594 // commit that copy stands at. A branch of one identity lives in as many clones as
595 // ever held it — a busy host keeps dozens — and deciding it is the expensive part
596 // of this report: the landing tiers, the provenance reads and, at the bottom, a
597 // content comparison, all of which answer the same for two copies standing at the
598 // same commit. So the second copy of one is not asked. The tip is in the key
599 // rather than assumed away, because two clones of a name that has been retried do
600 // *not* hold the same work, and a verdict borrowed across that difference would be
601 // this report answering about commits it never looked at.
602 let mut decided: BTreeSet<(String, String, String)> = BTreeSet::new();
603 let resolution = store::resolve(registry, identity)?;
604 let publication = resolution.publication.clone();
605 let current = git::default_branch(&publication, "origin")
606 .ok()
607 .and_then(|base| base_commit(&publication, &base));
608 // Every publication fast-forwards this checkout, so it is where a landing's
609 // evidence is — and lending its objects is what lets a checkout that has not
610 // fetched since read the commit that carries them.
611 let lent = git::objects_dir(&publication).ok();
612 // The branches this host itself put on the origin with `onevcs preserve`, read
613 // once per identity: see `reported_branches` for why the listing below needs
614 // them.
615 let preserved_here = crate::status::preserved_branches(streams, identity);
616 for repo in workspace::checkouts_of(registry, &resolution)? {
617 // A checkout none of the selected sessions can be holding a branch in is
618 // not opened: that is the difference between a filter that narrows the
619 // answer and one that narrows the work, and on a host with forty retained
620 // clones per identity it is the whole difference.
621 if narrowed
622 .as_ref()
623 .is_some_and(|only| !only.checkouts.contains(&repo))
624 {
625 continue;
626 }
627 if !git::is_repo(&repo) {
628 continue;
629 }
630 let base = match git::default_branch(&repo, "origin") {
631 Ok(base) => base,
632 Err(_) => continue,
633 };
634 let asked = git::Asked::borrowing(&repo, lent.as_deref());
635 let compared = judged_against(asked, &base, current.as_ref());
636 // Only the names asked about are counted against their remote-tracking
637 // refs, which is a process per branch per checkout that a filtered read
638 // has no reason to spend — and the ones `onevcs preserve` put on the
639 // origin are listed whatever those refs say: see `reported_branches`.
640 let listed = reported_branches(&repo, &preserved_here, |branch| {
641 narrowed
642 .as_ref()
643 .is_none_or(|only| only.wants(identity, branch))
644 })?;
645 for (branch, tip) in listed {
646 let key = (identity.to_owned(), branch.clone());
647 if seen.contains(&key) {
648 continue;
649 }
650 // The clone of a session something superseded holds the work that was
651 // taken over rather than the work that went on, so it answers for this
652 // name no more here than it does in `onevcs status` — and a row from it
653 // is a paste-ready publication of commits a later session already
654 // replaced.
655 if superseded_copy(sessions, &repo, identity, &branch) {
656 continue;
657 }
658 // A copy of this name standing at this commit has already been
659 // decided, in another checkout of the same identity — and the answer
660 // is a property of the two, so asking again would spend the whole
661 // decision to be told what is already known. Which copy answers is
662 // unchanged: the first one reached is the one whose row survives the
663 // deduplication below, and it is now also the only one asked.
664 if !decided.insert((identity.to_owned(), branch.clone(), tip.clone())) {
665 continue;
666 }
667 // Unpublished by ref is not the same as unfinished: publication
668 // squashes, so a branch that landed is never an ancestor of the base
669 // afterwards. What answers the question is what the base's own history
670 // records about this branch — and, only where it records nothing, what
671 // the base carries of what the branch changed.
672 let recorded = crate::status::recorded_for(
673 streams,
674 identity,
675 &branch,
676 session_holding(sessions, identity, &branch),
677 );
678 let recorded = landed::Recorded {
679 change: recorded
680 .change
681 .or_else(|| change_url_of(asked, &compared, &branch, trailers)),
682 ..recorded
683 };
684 let change_url = recorded.change.clone();
685 // A chain of retries this host cannot follow leaves nothing decided
686 // about the branch — the same answer `onevcs status` gives, through
687 // the same reading of the same records, because a row that said `no`
688 // here and `unknown` there would be the disagreement this report
689 // exists to end. Asked before the tiers rather than after them, because
690 // whatever they found would be replaced by it: one such branch on the
691 // consuming host spent seventy-five content merges on a verdict that
692 // was then discarded.
693 let verdict = if unfollowable_chain(sessions, identity, &branch) {
694 Landed::Unknown
695 } else {
696 landed::decide(
697 asked,
698 &compared,
699 current.as_ref(),
700 &branch,
701 &recorded,
702 trailers,
703 )?
704 };
705 // Withheld unless every branch was asked for, and only where the
706 // work *reached the base*: that is the row whose command must not be
707 // pasted. A row nothing can decide about is the opposite case — it
708 // may be work nobody published, so withholding it is how preserved
709 // work goes missing — and it is listed, saying so, with no line that
710 // reads as "paste this".
711 let withheld = verdict.is_landed();
712 if withheld && reporting == Reporting::UnpublishedOnly {
713 continue;
714 }
715 // Marked seen only once it is a row this report is answering with, so
716 // that one repository's spent copy of a name cannot answer for
717 // another's: a branch published out of the checkout and re-cut in a
718 // later run has both, and the first has nothing left in it.
719 if !withheld {
720 seen.push(key);
721 }
722 let row = preserved_row(
723 &Preserved {
724 identity,
725 repo: asked,
726 publication: &publication,
727 base: &base,
728 compared: &compared,
729 branch: &branch,
730 change_url,
731 verdict,
732 // Read through the same reader `onevcs status` reads it
733 // through, so the two reports cannot come to disagree about
734 // where one branch is.
735 on_origin: crate::status::preserved_for(
736 streams,
737 identity,
738 &branch,
739 session_holding(sessions, identity, &branch),
740 ),
741 },
742 sessions,
743 trailers,
744 )?;
745 if withheld {
746 withheld_rows.push(row);
747 } else {
748 rows.push(row);
749 }
750 }
751 }
752 Ok(Scanned {
753 rows,
754 withheld_rows,
755 seen,
756 })
757}
758
759/// `work` asked of every item, on at most as many threads as this host runs at
760/// once, and answered in the items' own order.
761///
762/// Scoped threads, so each borrows what the caller holds and none outlives the
763/// call; a bounded number of them, because an item here is a stream of git
764/// processes and a host with more identities than cores gains nothing from
765/// starting them all at once.
766fn concurrently<T: Sync, R: Send>(items: &[T], work: impl Fn(&T) -> R + Sync) -> Vec<R> {
767 let workers = std::thread::available_parallelism()
768 .map_or(1, std::num::NonZeroUsize::get)
769 .min(items.len());
770 if workers <= 1 {
771 return items.iter().map(&work).collect();
772 }
773 let next = std::sync::atomic::AtomicUsize::new(0);
774 let mut answered: Vec<(usize, R)> = std::thread::scope(|scope| {
775 let handles: Vec<_> = (0..workers)
776 .map(|_| {
777 scope.spawn(|| {
778 let mut mine = Vec::new();
779 loop {
780 let at = next.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
781 let Some(item) = items.get(at) else {
782 return mine;
783 };
784 mine.push((at, work(item)));
785 }
786 })
787 })
788 .collect();
789 handles
790 .into_iter()
791 .flat_map(|handle| match handle.join() {
792 Ok(mine) => mine,
793 Err(panic) => std::panic::resume_unwind(panic),
794 })
795 .collect()
796 });
797 answered.sort_by_key(|(at, _)| *at);
798 answered.into_iter().map(|(_, answer)| answer).collect()
799}
800
801/// One preserved branch, and everything a row about it is read out of.
802///
803/// One value rather than seven arguments, for the reason `status`'s own decision
804/// takes one: every field is read by the one function below, and two of them
805/// transposed would produce a row that reads perfectly and names another branch's
806/// repository.
807struct Preserved<'a> {
808 identity: &'a str,
809 /// The repository holding the branch, asked through whatever store lends it the
810 /// base — so every read below judges against the base the row names.
811 repo: git::Asked<'a>,
812 publication: &'a Path,
813 base: &'a str,
814 compared: &'a str,
815 branch: &'a str,
816 change_url: Option<Url>,
817 verdict: Landed,
818 on_origin: Option<OnOrigin>,
819}
820
821/// The row one preserved branch answers with.
822fn preserved_row(
823 preserved: &Preserved<'_>,
824 sessions: &[workspace::Record],
825 trailers: &provenance::Trailers,
826) -> Result<(Option<u64>, Recoverable)> {
827 let Preserved {
828 identity,
829 repo,
830 publication,
831 base,
832 compared,
833 branch,
834 ref change_url,
835 ref verdict,
836 ref on_origin,
837 } = *preserved;
838 // A marker under a prefix this host does not read is still a marker:
839 // reporting the branch as complete is what would let somebody hand
840 // interrupted work to the verb that publishes a finished one.
841 let unrecognized = provenance::unrecognized(repo, compared, branch, trailers)?;
842 let kind = match unrecognized.first() {
843 Some(_) => Provenance::IncompleteStep,
844 None => provenance::provenance_of(repo, compared, branch, trailers)?,
845 };
846 let incomplete = kind == Provenance::IncompleteStep;
847 let change_base = provenance::recorded_change_base(repo, compared, branch, trailers)?;
848 // Asked before the row says anything about why the work stopped,
849 // because a branch a live session is still writing to has not stopped.
850 let held_by = held_by(sessions, identity, branch)?;
851 let mut stopped = match verdict {
852 // Nothing stopped it: it finished. Which tier says so travels with the
853 // sentence, because "it landed" is exactly the claim that used to be an
854 // inference and was wrong whenever the base had moved.
855 Landed::Yes { evidence } => format!(
856 "nothing stopped: this branch's work reached {base}, and {tier} ({commit}) says so",
857 tier = verdict.tier(),
858 commit = evidence.commit(),
859 ),
860 _ => match &held_by {
861 Some(held) => format!(
862 "nothing has stopped: session {token} still holds this branch and \
863 {because}. Its work is being made in {worktree}",
864 token = held.token.0,
865 because = held.holding.because(),
866 worktree = held.worktree.display(),
867 ),
868 None => sessions
869 .iter()
870 .find(|record| *record.branch == *branch)
871 .map(|record| {
872 if record.state == Lifecycle::Open {
873 format!("session {} was left open", record.token)
874 } else {
875 format!("session {} closed without publishing", record.token)
876 }
877 })
878 .unwrap_or_else(|| {
879 "no session record names this branch; it stopped before recording one"
880 .to_owned()
881 }),
882 },
883 };
884 if let Landed::InPart { evidence, unlanded } = verdict {
885 stopped.push_str(&format!(
886 ". Part of this branch's work reached {base}, and {tier} ({commit}) says so — the \
887 {unlanded} commit(s) it has gained since are what is left to publish",
888 tier = verdict.tier(),
889 commit = evidence.commit(),
890 ));
891 }
892 if *verdict == Landed::Unknown {
893 stopped.push_str(&format!(
894 ". Whether it landed cannot be decided from history: nothing records that it \
895 reached {base} — no landing, no change request's number in the base's history, \
896 and no landing trailer — and comparing content settles nothing here, so {base} \
897 may already carry this work"
898 ));
899 }
900 if let Some(prefix) = unrecognized.first() {
901 stopped.push_str(&format!(
902 ". Its provenance is written under the trailer prefix {prefix:?}, which \
903 this host is not configured to read: set trailer_prefix in the rules \
904 file to {prefix:?} before publishing it"
905 ));
906 }
907 // The verb its provenance earns, taking the repository by path so
908 // that the command runs wherever the row is read. A branch whose work
909 // reached the base earns none: the row is read to be pasted, and pasting
910 // one for finished work re-opens a change request for what the base has.
911 let verb = if incomplete {
912 "recover"
913 } else {
914 "publish-branch"
915 };
916 // llmlint: ignore[names_match_behavior] the name is the public
917 // `Recoverable::recover_command` field this fills, which the recorded
918 // surface in docs/inferred-surface.md fixes; renaming that field is a
919 // break of the published surface, and a local that disagreed with it
920 // would be the drift. Which verb the argv holds is the two lines above.
921 let recover_command = match verdict.is_landed() {
922 true => Vec::new(),
923 false => vec![
924 "onevcs".to_owned(),
925 verb.to_owned(),
926 branch.to_owned(),
927 "--repo".to_owned(),
928 publication.display().to_string(),
929 ],
930 };
931 let answering = latest_session(sessions, identity, branch);
932 Ok((
933 git::committed_at(repo.path(), branch),
934 Recoverable {
935 identity: identity.to_owned(),
936 branch: PreservedBranch {
937 branch: branch.to_owned(),
938 base: base.to_owned(),
939 provenance: kind,
940 change_url: change_url
941 .clone()
942 .or_else(|| change_url_of(repo, compared, branch, trailers)),
943 change_base,
944 },
945 checkout: repo.path().to_path_buf(),
946 landed: verdict.clone(),
947 stopped_because: stopped,
948 recover_command,
949 held_by,
950 net_negative: net_negative(repo, compared, branch)?,
951 session: answering.map(|record| SessionToken(record.token.to_string())),
952 labels: answering
953 .map(|record| record.labels.clone())
954 .unwrap_or_default(),
955 on_origin: on_origin.clone(),
956 },
957 ))
958}
959
960/// The session that answers for a branch: the newest record naming it.
961///
962/// The end of its chain of retries — a record nothing superseded — and an open one
963/// over a closed one where two chains end apart, which is the preference `status`
964/// makes when it picks whose evidence is the branch's. Where every record of the
965/// branch has been superseded, which is a chain this host cannot follow, the same
966/// preference is applied to all of them rather than answering nobody: the row still
967/// names a session somebody can look up, and its landing is already `unknown`.
968/// Ties are broken by token, so two reads answer the same record.
969pub(crate) fn latest_session<'a>(
970 sessions: &'a [workspace::Record],
971 identity: &str,
972 branch: &str,
973) -> Option<&'a workspace::Record> {
974 let named: Vec<&workspace::Record> = sessions
975 .iter()
976 .filter(|record| record.identity == identity && *record.branch == *branch)
977 .collect();
978 let ends: Vec<&workspace::Record> = named
979 .iter()
980 .copied()
981 .filter(|record| record.retried_by.is_none())
982 .collect();
983 let candidates = if ends.is_empty() { named } else { ends };
984 candidates
985 .into_iter()
986 .max_by_key(|record| (record.state == Lifecycle::Open, record.token.to_string()))
987}
988
989/// The branches of one repository this report answers about.
990///
991/// [`git::unpublished_branches_among`] is the question it has always asked — which local
992/// branches hold commits no `origin` remote-tracking ref has — and the preserved names
993/// are a union with it rather than a change to it. They have to be, because a
994/// preserving push updates the pushing repository's own `origin/<branch>`: measured
995/// against that, a branch `onevcs preserve` had just put somewhere safe would read as
996/// published and vanish from this report, which is the opposite of what preserving it
997/// promised. Being on the origin under its own name is not being published — the work
998/// has not reached the base — so the row stays, with the same `recover_command` it
999/// carried before and the origin named beside it.
1000///
1001/// Nothing else about which branches this report covers moves: a name no
1002/// `branch-preserved` record of this identity carries is listed exactly as it was, and
1003/// the other readers of `unpublished_branches` — the close, the reclaim, and the
1004/// sweep's retention rule — go on asking whether letting a clone go would lose work,
1005/// where a branch the origin carries genuinely loses none.
1006///
1007/// Each name comes with the commit it stands at, and the union costs no process of
1008/// its own: both halves are answered by the one listing of this checkout's refs, and
1009/// a name `keep` declines is in neither.
1010fn reported_branches(
1011 repo: &Path,
1012 preserved: &BTreeSet<String>,
1013 keep: impl Fn(&str) -> bool,
1014) -> Result<Vec<(String, String)>> {
1015 git::unpublished_branches_among(repo, keep, preserved)
1016}
1017
1018/// Whether this copy of a branch belongs to a session something superseded.
1019fn superseded_copy(
1020 sessions: &[workspace::Record],
1021 repo: &Path,
1022 identity: &str,
1023 branch: &str,
1024) -> bool {
1025 sessions.iter().any(|record| {
1026 record.retried_by.is_some()
1027 && record.clone == repo
1028 && record.identity == identity
1029 && *record.branch == *branch
1030 })
1031}
1032
1033/// Whether any session of this branch names a chain of retries this host cannot
1034/// follow to an end.
1035fn unfollowable_chain(sessions: &[workspace::Record], identity: &str, branch: &str) -> bool {
1036 sessions
1037 .iter()
1038 .filter(|record| record.identity == identity && *record.branch == *branch)
1039 .any(|record| workspace::newest(record).is_err())
1040}
1041
1042/// The token of an open session holding this branch, for reading its stream.
1043fn session_holding<'a>(
1044 sessions: &'a [workspace::Record],
1045 identity: &str,
1046 branch: &str,
1047) -> Option<&'a str> {
1048 sessions
1049 .iter()
1050 .find(|record| record.identity == identity && *record.branch == *branch)
1051 .map(|record| record.token.as_ref())
1052}
1053
1054/// The live session still writing to a preserved branch, when one is.
1055///
1056/// Two ways of being live, because the two are true at different times and a report
1057/// that knew only one would offer somebody a branch mid-flight. A consumer holding a
1058/// [`Session`] keeps the process that opened it, which is the question
1059/// [`Liveness`] already answers; the CLI takes an occupancy lease per command and
1060/// outlives none of them, so what says a command is in there *now* is the lease
1061/// itself. Either one means the same thing about the branch.
1062///
1063/// Only an open session is asked about: closing one hands its branch back and means
1064/// finished, and its run root going on being occupied afterwards says nothing about
1065/// work nobody is doing on that branch any more.
1066fn held_by(sessions: &[workspace::Record], identity: &str, branch: &str) -> Result<Option<HeldBy>> {
1067 for record in sessions {
1068 if record.identity != identity
1069 || *record.branch != *branch
1070 || record.state != Lifecycle::Open
1071 {
1072 continue;
1073 }
1074 let holding = match record.liveness() {
1075 Liveness::Live => Some(Holding::OwnerRunning),
1076 Liveness::Stale => {
1077 lock::is_occupied(&record.lease())?.then_some(Holding::RunRootOccupied)
1078 }
1079 };
1080 if let Some(holding) = holding {
1081 return Ok(Some(HeldBy {
1082 token: SessionToken(record.token.to_string()),
1083 worktree: record.worktree.clone(),
1084 holding,
1085 }));
1086 }
1087 }
1088 Ok(None)
1089}
1090
1091/// What a branch would land, when it removes more lines than it adds.
1092///
1093/// Measured from the commit the branch forked from rather than from `compared`
1094/// itself: what the branch did is what it did to the tree it started on, and against
1095/// a base that has moved on every line that base gained would read as a line this
1096/// branch removed and never touched. A branch sharing no history with the base is
1097/// not measured — there is no point it forked from to measure against.
1098fn net_negative<'a>(
1099 repo: impl Into<git::Asked<'a>>,
1100 compared: &str,
1101 branch: &str,
1102) -> Result<Option<NetNegative>> {
1103 let repo = repo.into();
1104 let Some(fork) = git::merge_base(repo, compared, branch)? else {
1105 return Ok(None);
1106 };
1107 let counted = git::line_change(repo, &fork, branch)?;
1108 // Which counts are net-negative is `NetNegative`'s own rule, asked here rather
1109 // than restated: a second spelling of it is how a row comes to be marked by one
1110 // rule and read back under another.
1111 Ok(NetNegative::new(LineChange {
1112 added: counted.added,
1113 removed: counted.removed,
1114 }))
1115}
1116
1117/// The change request a preserved branch recorded, when one was opened for it.
1118pub fn change_url_of<'a>(
1119 repo: impl Into<git::Asked<'a>>,
1120 base: &str,
1121 branch: &str,
1122 trailers: &provenance::Trailers,
1123) -> Option<Url> {
1124 let commits = git::log_messages(repo, base, branch).ok()?;
1125 commits
1126 .iter()
1127 .rev()
1128 .flat_map(|commit| commit.message.lines())
1129 .filter_map(|line| line.trim().strip_prefix(trailers.change_url()))
1130 .find_map(|value| Url::parse(value.trim()).ok())
1131}