use std::collections::{BTreeMap, BTreeSet};
use std::path::{Path, PathBuf};
use crate::error::{self, Error, Result};
use crate::event::EventKind;
use crate::host::{Hosting, Sha};
use crate::landed::{self, Landed};
use crate::publish::{Publication, PublishRequest};
use crate::registry::Identity;
use crate::session::{
HeldBy, Holding, Lifecycle, LineChange, Liveness, NetNegative, OnOrigin, PreservedBranch,
Provenance, Recoverable, Scope, Selection, Session, SessionRecord, SessionRequest,
SessionToken,
};
use crate::stream::Stream;
use crate::workspace::{self, object};
use crate::{git, label, lock, provenance, publish, store};
use serde_json::json;
use url::Url;
pub trait Vcs {
fn resolve_identity(&self, origin_or_path: &str) -> Result<Identity>;
fn open_session(&self, req: SessionRequest) -> Result<Session>;
fn adopt_session(&self, token: SessionToken) -> Result<Session>;
fn session(&self, token: &SessionToken) -> Result<SessionRecord>;
fn close_session(&self, token: &SessionToken) -> Result<Session>;
fn preserve(&self, s: &Session, provenance: Provenance) -> Result<PreservedBranch>;
fn publish(
&self,
token: &SessionToken,
request: &PublishRequest,
hosting: &dyn Hosting,
) -> Result<Publication>;
fn recoverable(&self, scope: Scope) -> Result<Vec<Recoverable>>;
fn preserved(&self, scope: Scope) -> Result<Vec<Recoverable>>;
fn recoverable_matching(
&self,
scope: Scope,
selection: &Selection,
) -> Result<Vec<Recoverable>> {
retain(self.recoverable(scope)?, selection)
}
fn preserved_matching(&self, scope: Scope, selection: &Selection) -> Result<Vec<Recoverable>> {
retain(self.preserved(scope)?, selection)
}
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct Git;
impl Vcs for Git {
fn resolve_identity(&self, origin_or_path: &str) -> Result<Identity> {
let registry = store::load()?;
Ok(store::resolve(®istry, origin_or_path)?.identity)
}
fn open_session(&self, req: SessionRequest) -> Result<Session> {
let registry = store::load()?;
let (record, _stream) = workspace::open(®istry, &req)?;
Ok(record.session())
}
fn adopt_session(&self, token: SessionToken) -> Result<Session> {
let (record, _stream, _preserved) = workspace::adopt(&token.0)?;
Ok(record.session())
}
fn session(&self, token: &SessionToken) -> Result<SessionRecord> {
let record = workspace::load(&token.0)?;
let trailers = provenance::configured()?;
let held_in = match git::branch_exists(&record.clone, &record.branch) {
true => record.clone.clone(),
false => record.execution_checkout.clone(),
};
if !git::branch_exists(&held_in, &record.branch) && crate::retire::retired(&record) {
return Ok(SessionRecord {
session: record.session(),
identity: record.identity.clone(),
lifecycle: record.state,
provenance: Provenance::Complete,
retried_by: record
.retried_by
.as_ref()
.map(|token| SessionToken(token.to_string())),
});
}
let target = publish::standing_target(&record)?;
let base = base_ref(&held_in, target.base());
Ok(SessionRecord {
session: record.session(),
identity: record.identity.clone(),
lifecycle: record.state,
provenance: provenance::provenance_of(&held_in, &base, &record.branch, &trailers)?,
retried_by: record
.retried_by
.as_ref()
.map(|token| SessionToken(token.to_string())),
})
}
fn close_session(&self, token: &SessionToken) -> Result<Session> {
let record = workspace::close(&token.0)?;
crate::retire::after_close(&record);
Ok(record.session())
}
fn preserve(&self, s: &Session, provenance: Provenance) -> Result<PreservedBranch> {
let record = workspace::load(&s.token.0)?;
let mut stream = Stream::open(&s.token.0)?;
preserve_into(&record, &mut stream, provenance)
}
fn publish(
&self,
token: &SessionToken,
request: &PublishRequest,
hosting: &dyn Hosting,
) -> Result<Publication> {
publish::run_for_session(token, request, hosting)
}
fn recoverable(&self, scope: Scope) -> Result<Vec<Recoverable>> {
collect(&scope, Reporting::UnpublishedOnly)
}
fn preserved(&self, scope: Scope) -> Result<Vec<Recoverable>> {
collect(&scope, Reporting::Everything)
}
fn recoverable_matching(
&self,
scope: Scope,
selection: &Selection,
) -> Result<Vec<Recoverable>> {
collect_matching(&scope, Reporting::UnpublishedOnly, selection)
}
fn preserved_matching(&self, scope: Scope, selection: &Selection) -> Result<Vec<Recoverable>> {
collect_matching(&scope, Reporting::Everything, selection)
}
}
pub fn preserve_into(
record: &workspace::Record,
stream: &mut Stream,
kind: Provenance,
) -> Result<PreservedBranch> {
let trailers = provenance::configured()?;
if git::is_dirty(&record.worktree)? {
git::add_all(&record.worktree)?;
let message = match kind {
Provenance::Complete => format!("chore: preserve work on {}", record.branch),
Provenance::IncompleteStep => provenance::incomplete_message(
&format!("work on {}", record.branch),
record.change_base.as_deref(),
&trailers,
),
};
let sha = git::commit(&record.worktree, &message)?;
stream.emit(
EventKind::CommitPreserved,
object(json!({
"branch": record.branch,
"sha": sha,
"provenance": spell_provenance(kind),
})),
);
}
let copied = git::copy_branch(&record.clone, &record.execution_checkout, &record.branch)?;
if !copied {
return Err(Error::Invalid {
reason: format!(
"the execution checkout {} refused branch {:?}; it holds work this session's \
clone does not, so nothing outside the session carries this branch",
record.execution_checkout.display(),
record.branch
),
});
}
let base = judging_base(record)?;
Ok(PreservedBranch {
branch: record.branch.to_string(),
base: record.base.to_string(),
provenance: provenance::provenance_of(&record.clone, &base, &record.branch, &trailers)?,
change_url: None,
change_base: record.change_base.as_ref().map(ToString::to_string),
})
}
pub fn spell_provenance(kind: Provenance) -> &'static str {
match kind {
Provenance::Complete => "complete",
Provenance::IncompleteStep => "incomplete-step",
}
}
fn judging_base(record: &workspace::Record) -> Result<String> {
let target = publish::standing_target(record)?;
Ok(base_ref(&record.clone, target.base()))
}
pub fn base_ref(repo: &Path, base: &str) -> String {
let remote = format!("origin/{base}");
if git::ref_exists(repo, &format!("refs/remotes/{remote}")) {
remote
} else {
base.to_owned()
}
}
pub fn judged_against<'a>(
repo: impl Into<git::Asked<'a>>,
base: &str,
current: Option<&Sha>,
) -> String {
let repo = repo.into();
match current {
Some(sha) if git::has_commit(repo, sha) => sha.0.clone(),
_ => base_ref(repo.path(), base),
}
}
pub fn base_commit(checkout: &Path, base: &str) -> Option<Sha> {
git::tip(checkout, &base_ref(checkout, base)).map(Sha)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Reporting {
UnpublishedOnly,
Everything,
}
struct Narrowed {
identities: BTreeSet<String>,
branches: BTreeSet<(String, String)>,
checkouts: BTreeSet<PathBuf>,
}
impl Narrowed {
fn wants(&self, identity: &str, branch: &str) -> bool {
self.branches
.contains(&(identity.to_owned(), branch.to_owned()))
}
}
fn narrowed(selection: &Selection, sessions: &[workspace::Record]) -> Result<Option<Narrowed>> {
if selection.is_empty() {
return Ok(None);
}
for token in &selection.sessions {
if !sessions.iter().any(|record| *record.token == *token.0) {
return Err(error::invalid(format!(
"no session record on this host names {}; \
`onevcs session holders REPO` lists the sessions of one repository",
token.0
)));
}
}
let mut narrowed = Narrowed {
identities: BTreeSet::new(),
branches: BTreeSet::new(),
checkouts: BTreeSet::new(),
};
for record in sessions {
let named = selection.sessions.is_empty()
|| selection
.sessions
.iter()
.any(|token| *record.token == *token.0);
if !named || !label::matches(&record.labels, &selection.labels) {
continue;
}
narrowed.identities.insert(record.identity.clone());
narrowed
.branches
.insert((record.identity.clone(), record.branch.to_string()));
narrowed.checkouts.insert(record.clone.clone());
narrowed.checkouts.insert(record.execution_checkout.clone());
}
Ok(Some(narrowed))
}
pub(crate) fn retain(rows: Vec<Recoverable>, selection: &Selection) -> Result<Vec<Recoverable>> {
let Some(narrowed) = narrowed(selection, &workspace::all()?)? else {
return Ok(rows);
};
Ok(rows
.into_iter()
.filter(|row| narrowed.wants(&row.identity, &row.branch.branch))
.collect())
}
pub fn collect(scope: &Scope, reporting: Reporting) -> Result<Vec<Recoverable>> {
collect_matching(scope, reporting, &Selection::default())
}
pub fn collect_matching(
scope: &Scope,
reporting: Reporting,
selection: &Selection,
) -> Result<Vec<Recoverable>> {
git::memoized(|| collected(scope, reporting, selection))
}
fn collected(
scope: &Scope,
reporting: Reporting,
selection: &Selection,
) -> Result<Vec<Recoverable>> {
let registry = store::load()?;
let (rules, _source) = crate::policy::load(®istry)?;
let trailers = provenance::from_rules(&rules);
let sessions = workspace::all()?;
let narrowed = narrowed(selection, &sessions)?;
let streams = crate::status::recorded_streams(&mut Vec::new())?;
let wanted = match scope {
Scope::All => None,
Scope::Repo(repo) => Some(store::resolve(®istry, repo)?.key),
};
let mut identities: Vec<&str> = registry
.checkouts
.values()
.map(|checkout| checkout.identity.as_str())
.collect();
identities.sort_unstable();
identities.dedup();
identities.retain(|&identity| {
if wanted.as_ref().is_some_and(|key| key != identity) {
return false;
}
if narrowed
.as_ref()
.is_some_and(|only| !only.identities.contains(identity))
{
return false;
}
true
});
let scan = Scan {
registry: ®istry,
streams: &streams,
sessions: &sessions,
trailers: &trailers,
narrowed: &narrowed,
reporting,
};
let mut rows: Vec<(Option<u64>, Recoverable)> = Vec::new();
let mut withheld_rows: Vec<(Option<u64>, Recoverable)> = Vec::new();
let mut seen: Vec<(String, String)> = Vec::new();
for scanned in concurrently(&identities, |identity| {
git::memoized(|| scanned(identity, &scan))
}) {
let scanned = scanned?;
rows.extend(scanned.rows);
withheld_rows.extend(scanned.withheld_rows);
seen.extend(scanned.seen);
}
let mut kept: Vec<(String, String)> = Vec::new();
for (at, row) in withheld_rows {
let key = (row.identity.clone(), row.branch.branch.clone());
if seen.contains(&key) || kept.contains(&key) {
continue;
}
kept.push(key);
rows.push((at, row));
}
rows.sort_by_key(|(at, row)| {
(
std::cmp::Reverse(at.unwrap_or(0)),
row.branch.branch.clone(),
)
});
Ok(rows.into_iter().map(|(_, row)| row).collect())
}
struct Scan<'a> {
registry: &'a crate::registry::Registry,
streams: &'a [crate::status::Recorded],
sessions: &'a [workspace::Record],
trailers: &'a provenance::Trailers,
narrowed: &'a Option<Narrowed>,
reporting: Reporting,
}
struct Scanned {
rows: Vec<(Option<u64>, Recoverable)>,
withheld_rows: Vec<(Option<u64>, Recoverable)>,
seen: Vec<(String, String)>,
}
fn scanned(identity: &str, scan: &Scan<'_>) -> Result<Scanned> {
let Scan {
registry,
streams,
sessions,
trailers,
narrowed,
reporting,
} = *scan;
let mut rows: Vec<(Option<u64>, Recoverable)> = Vec::new();
let mut withheld_rows: Vec<(Option<u64>, Recoverable)> = Vec::new();
let mut seen: Vec<(String, String)> = Vec::new();
let mut decided: BTreeSet<(String, String, String)> = BTreeSet::new();
let resolution = store::resolve(registry, identity)?;
let publication = resolution.publication.clone();
let current = git::default_branch(&publication, "origin")
.ok()
.and_then(|base| base_commit(&publication, &base));
let lent = git::objects_dir(&publication).ok();
let preserved_here = crate::status::preserved_branches(streams, identity);
let census = crate::retire::Census::read(
registry,
identity,
sessions,
streams,
trailers,
crate::retire::Reach::Offline,
narrowed.as_ref().map(|only| &only.checkouts),
)?;
let mut classified: BTreeMap<String, Option<crate::retire::Retirement>> = BTreeMap::new();
for repo in workspace::checkouts_of(registry, &resolution)? {
if narrowed
.as_ref()
.is_some_and(|only| !only.checkouts.contains(&repo))
{
continue;
}
if !git::is_repo(&repo) {
continue;
}
let base = match git::default_branch(&repo, "origin") {
Ok(base) => base,
Err(_) => continue,
};
let asked = git::Asked::borrowing(&repo, lent.as_deref());
let compared = judged_against(asked, &base, current.as_ref());
let listed = reported_branches(&repo, &preserved_here, |branch| {
narrowed
.as_ref()
.is_none_or(|only| only.wants(identity, branch))
})?;
for (branch, tip) in listed {
let key = (identity.to_owned(), branch.clone());
if seen.contains(&key) {
continue;
}
if superseded_copy(sessions, &repo, identity, &branch) {
continue;
}
if !decided.insert((identity.to_owned(), branch.clone(), tip.clone())) {
continue;
}
let recorded = crate::status::recorded_for(
streams,
identity,
&branch,
session_holding(sessions, identity, &branch),
);
let recorded = landed::Recorded {
change: recorded
.change
.or_else(|| change_url_of(asked, &compared, &branch, trailers)),
..recorded
};
let change_url = recorded.change.clone();
let verdict = if unfollowable_chain(sessions, identity, &branch) {
Landed::Unknown
} else {
landed::decide(
asked,
&compared,
current.as_ref(),
&branch,
&recorded,
trailers,
)?
};
let withheld = verdict.is_landed();
if withheld && reporting == Reporting::UnpublishedOnly {
continue;
}
let retirement = classified
.entry(branch.clone())
.or_insert_with(|| crate::retire::classify_offline(&census, &branch))
.clone();
let retirable = retirement
.as_ref()
.is_some_and(|found| found.class == crate::retire::RetirementClass::Retirable);
if retirable && reporting == Reporting::UnpublishedOnly {
continue;
}
if !withheld {
seen.push(key);
}
let row = preserved_row(
&Preserved {
identity,
repo: asked,
publication: &publication,
base: &base,
compared: &compared,
branch: &branch,
change_url,
verdict,
on_origin: crate::status::preserved_for(
streams,
identity,
&branch,
session_holding(sessions, identity, &branch),
),
retirement,
},
sessions,
trailers,
)?;
if withheld {
withheld_rows.push(row);
} else {
rows.push(row);
}
}
}
Ok(Scanned {
rows,
withheld_rows,
seen,
})
}
fn concurrently<T: Sync, R: Send>(items: &[T], work: impl Fn(&T) -> R + Sync) -> Vec<R> {
let workers = std::thread::available_parallelism()
.map_or(1, std::num::NonZeroUsize::get)
.min(items.len());
if workers <= 1 {
return items.iter().map(&work).collect();
}
let next = std::sync::atomic::AtomicUsize::new(0);
let mut answered: Vec<(usize, R)> = std::thread::scope(|scope| {
let handles: Vec<_> = (0..workers)
.map(|_| {
scope.spawn(|| {
let mut mine = Vec::new();
loop {
let at = next.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
let Some(item) = items.get(at) else {
return mine;
};
mine.push((at, work(item)));
}
})
})
.collect();
handles
.into_iter()
.flat_map(|handle| match handle.join() {
Ok(mine) => mine,
Err(panic) => std::panic::resume_unwind(panic),
})
.collect()
});
answered.sort_by_key(|(at, _)| *at);
answered.into_iter().map(|(_, answer)| answer).collect()
}
struct Preserved<'a> {
identity: &'a str,
repo: git::Asked<'a>,
publication: &'a Path,
base: &'a str,
compared: &'a str,
branch: &'a str,
change_url: Option<Url>,
verdict: Landed,
on_origin: Option<OnOrigin>,
retirement: Option<crate::retire::Retirement>,
}
fn preserved_row(
preserved: &Preserved<'_>,
sessions: &[workspace::Record],
trailers: &provenance::Trailers,
) -> Result<(Option<u64>, Recoverable)> {
let Preserved {
identity,
repo,
publication,
base,
compared,
branch,
ref change_url,
ref verdict,
ref on_origin,
ref retirement,
} = *preserved;
let unrecognized = provenance::unrecognized(repo, compared, branch, trailers)?;
let kind = match unrecognized.first() {
Some(_) => Provenance::IncompleteStep,
None => provenance::provenance_of(repo, compared, branch, trailers)?,
};
let incomplete = kind == Provenance::IncompleteStep;
let change_base = provenance::recorded_change_base(repo, compared, branch, trailers)?;
let held_by = held_by(sessions, identity, branch)?;
let mut stopped = match verdict {
Landed::Yes { evidence } => format!(
"nothing stopped: this branch's work reached {base}, and {tier} ({commit}) says so",
tier = verdict.tier(),
commit = evidence.commit(),
),
_ => match &held_by {
Some(held) => format!(
"nothing has stopped: session {token} still holds this branch and \
{because}. Its work is being made in {worktree}",
token = held.token.0,
because = held.holding.because(),
worktree = held.worktree.display(),
),
None => sessions
.iter()
.find(|record| *record.branch == *branch)
.map(|record| {
if record.state == Lifecycle::Open {
format!("session {} was left open", record.token)
} else {
format!("session {} closed without publishing", record.token)
}
})
.unwrap_or_else(|| {
"no session record names this branch; it stopped before recording one"
.to_owned()
}),
},
};
if let Landed::InPart { evidence, unlanded } = verdict {
stopped.push_str(&format!(
". Part of this branch's work reached {base}, and {tier} ({commit}) says so — the \
{unlanded} commit(s) it has gained since are what is left to publish",
tier = verdict.tier(),
commit = evidence.commit(),
));
}
if *verdict == Landed::Unknown {
stopped.push_str(&format!(
". Whether it landed cannot be decided from history: nothing records that it \
reached {base} — no landing, no change request's number in the base's history, \
and no landing trailer — and comparing content settles nothing here, so {base} \
may already carry this work"
));
}
if let Some(prefix) = unrecognized.first() {
stopped.push_str(&format!(
". Its provenance is written under the trailer prefix {prefix:?}, which \
this host is not configured to read: set trailer_prefix in the rules \
file to {prefix:?} before publishing it"
));
}
let verb = if incomplete {
"recover"
} else {
"publish-branch"
};
let recover_command = match verdict.is_landed() {
true => Vec::new(),
false => vec![
"onevcs".to_owned(),
verb.to_owned(),
branch.to_owned(),
"--repo".to_owned(),
publication.display().to_string(),
],
};
let answering = latest_session(sessions, identity, branch);
Ok((
git::committed_at(repo.path(), branch),
Recoverable {
identity: identity.to_owned(),
branch: PreservedBranch {
branch: branch.to_owned(),
base: base.to_owned(),
provenance: kind,
change_url: change_url
.clone()
.or_else(|| change_url_of(repo, compared, branch, trailers)),
change_base,
},
checkout: repo.path().to_path_buf(),
landed: verdict.clone(),
stopped_because: stopped,
recover_command,
held_by,
net_negative: net_negative(repo, compared, branch)?,
session: answering.map(|record| SessionToken(record.token.to_string())),
labels: answering
.map(|record| record.labels.clone())
.unwrap_or_default(),
on_origin: on_origin.clone(),
retirement: retirement.clone(),
},
))
}
pub(crate) fn latest_session<'a>(
sessions: &'a [workspace::Record],
identity: &str,
branch: &str,
) -> Option<&'a workspace::Record> {
let named: Vec<&workspace::Record> = sessions
.iter()
.filter(|record| record.identity == identity && *record.branch == *branch)
.collect();
let ends: Vec<&workspace::Record> = named
.iter()
.copied()
.filter(|record| record.retried_by.is_none())
.collect();
let candidates = if ends.is_empty() { named } else { ends };
candidates
.into_iter()
.max_by_key(|record| (record.state == Lifecycle::Open, record.token.to_string()))
}
fn reported_branches(
repo: &Path,
preserved: &BTreeSet<String>,
keep: impl Fn(&str) -> bool,
) -> Result<Vec<(String, String)>> {
git::unpublished_branches_among(repo, keep, preserved)
}
fn superseded_copy(
sessions: &[workspace::Record],
repo: &Path,
identity: &str,
branch: &str,
) -> bool {
sessions.iter().any(|record| {
record.retried_by.is_some()
&& record.clone == repo
&& record.identity == identity
&& *record.branch == *branch
})
}
fn unfollowable_chain(sessions: &[workspace::Record], identity: &str, branch: &str) -> bool {
sessions
.iter()
.filter(|record| record.identity == identity && *record.branch == *branch)
.any(|record| workspace::newest(record).is_err())
}
fn session_holding<'a>(
sessions: &'a [workspace::Record],
identity: &str,
branch: &str,
) -> Option<&'a str> {
sessions
.iter()
.find(|record| record.identity == identity && *record.branch == *branch)
.map(|record| record.token.as_ref())
}
fn held_by(sessions: &[workspace::Record], identity: &str, branch: &str) -> Result<Option<HeldBy>> {
for record in sessions {
if record.identity != identity
|| *record.branch != *branch
|| record.state != Lifecycle::Open
{
continue;
}
let holding = match record.liveness() {
Liveness::Live => Some(Holding::OwnerRunning),
Liveness::Stale => {
lock::is_occupied(&record.lease())?.then_some(Holding::RunRootOccupied)
}
};
if let Some(holding) = holding {
return Ok(Some(HeldBy {
token: SessionToken(record.token.to_string()),
worktree: record.worktree.clone(),
holding,
}));
}
}
Ok(None)
}
fn net_negative<'a>(
repo: impl Into<git::Asked<'a>>,
compared: &str,
branch: &str,
) -> Result<Option<NetNegative>> {
let repo = repo.into();
let Some(fork) = git::merge_base(repo, compared, branch)? else {
return Ok(None);
};
let counted = git::line_change(repo, &fork, branch)?;
Ok(NetNegative::new(LineChange {
added: counted.added,
removed: counted.removed,
}))
}
pub fn change_url_of<'a>(
repo: impl Into<git::Asked<'a>>,
base: &str,
branch: &str,
trailers: &provenance::Trailers,
) -> Option<Url> {
let commits = git::log_messages(repo, base, branch).ok()?;
commits
.iter()
.rev()
.flat_map(|commit| commit.message.lines())
.filter_map(|line| line.trim().strip_prefix(trailers.change_url()))
.find_map(|value| Url::parse(value.trim()).ok())
}