Skip to main content

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(&registry, 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(&registry, &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(&registry)?;
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(&registry, 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: &registry,
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}