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