use std::collections::BTreeMap;
use std::num::NonZeroU64;
use std::path::{Path, PathBuf};
use serde::{Deserialize, Serialize};
use serde_json::{json, Map, Value};
use crate::error::{self, Error, Result};
use crate::event::EventKind;
use crate::registry::Registry;
use crate::remainder::Remainder;
use crate::session::{
Lifecycle, Liveness, OpenConflict, Session, SessionHolder, SessionRequest, SessionToken,
};
use crate::store::{self, Resolution};
use crate::stream::Stream;
use crate::workspaces::Sourced;
use crate::{branches, git, guidance, home, ids, label, lock, pool, processes, workspaces};
pub const RETAINED_DEAD_RUNS: usize = 3;
pub const RECORD_VERSION: u32 = 3;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(transparent)]
pub struct ProcessStart(NonZeroU64);
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(try_from = "String", into = "String")]
pub struct Token(String);
impl TryFrom<String> for Token {
type Error = String;
fn try_from(value: String) -> std::result::Result<Self, Self::Error> {
if ids::is_safe_name(&value) {
Ok(Token(value))
} else {
Err(format!("{value:?} is not a session token"))
}
}
}
impl From<Token> for String {
fn from(token: Token) -> Self {
token.0
}
}
impl std::ops::Deref for Token {
type Target = str;
fn deref(&self) -> &str {
&self.0
}
}
impl std::fmt::Display for Token {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(&self.0)
}
}
#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(try_from = "String", into = "String")]
pub struct Ref(String);
impl std::fmt::Debug for Ref {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
std::fmt::Debug::fmt(&self.0, f)
}
}
impl TryFrom<String> for Ref {
type Error = String;
fn try_from(value: String) -> std::result::Result<Self, Self::Error> {
if git::is_valid_branch_name(&value) {
Ok(Ref(value))
} else {
Err(format!("{value:?} is a name git would not accept"))
}
}
}
impl Ref {
pub fn from_git(name: impl Into<String>) -> Self {
Ref(name.into())
}
}
impl From<Ref> for String {
fn from(name: Ref) -> Self {
name.0
}
}
impl std::ops::Deref for Ref {
type Target = str;
fn deref(&self) -> &str {
&self.0
}
}
impl std::fmt::Display for Ref {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(&self.0)
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Record {
pub version: u32,
pub token: Token,
pub identity: String,
pub alias: String,
pub branch: Ref,
pub base: Ref,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub change_base: Option<Ref>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub stack_tip: Option<String>,
pub worktree: PathBuf,
pub clone: PathBuf,
pub run_root: PathBuf,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub slot: Option<u32>,
pub execution_checkout: PathBuf,
pub publication_checkout: PathBuf,
pub state: Lifecycle,
pub owner_pid: u32,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub owner_started: Option<ProcessStart>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub retried_by: Option<Token>,
#[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
pub labels: BTreeMap<String, String>,
#[serde(skip)]
pub carried: Remainder,
}
pub type Chain = std::result::Result<Record, String>;
pub fn newest(from: &Record) -> Chain {
let mut seen = vec![from.token.to_string()];
let mut record = from.clone();
while let Some(next) = record.retried_by.clone() {
if seen.iter().any(|token| *token == *next) {
return Err(format!(
"the session {first} was retried by {next}, and following that reaches {first} \
again; a chain of retries that closes on itself names no newest session",
first = from.token,
));
}
let followed = load(&next).map_err(|failure| {
format!(
"the session {token} was retried by {next}, and there is no such session on this \
host: {failure}",
token = record.token,
)
})?;
if followed.identity != record.identity {
return Err(format!(
"the session {token} in {here:?} was retried by {next} in {there:?}; a session \
continues a branch of its own repository, so nothing here answers for the other",
token = record.token,
here = record.identity,
there = followed.identity,
));
}
seen.push(next.to_string());
record = followed;
}
Ok(record)
}
pub fn record_retry(older: &Token, newer: &Token) -> Result<()> {
update(older, |record| record.retried_by = Some(newer.clone())).map(drop)
}
impl From<Record> for SessionHolder {
fn from(record: Record) -> Self {
let liveness = record.liveness();
Self {
token: SessionToken(record.token.to_string()),
identity: record.identity,
branch: record.branch.to_string(),
worktree: record.worktree,
owner_pid: record.owner_pid,
state: record.state,
liveness,
labels: record.labels,
}
}
}
#[cfg(windows)]
pub(crate) fn process_started(pid: u32) -> Option<ProcessStart> {
use windows_sys::Win32::Foundation::{CloseHandle, FILETIME, STILL_ACTIVE};
use windows_sys::Win32::System::Threading::{
GetExitCodeProcess, GetProcessTimes, OpenProcess, PROCESS_QUERY_LIMITED_INFORMATION,
};
if pid == 0 {
return None;
}
let handle = unsafe { OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, 0, pid) };
if handle.is_null() {
return None;
}
let mut exit_code = 0;
let mut created = FILETIME::default();
let mut exited = FILETIME::default();
let mut kernel = FILETIME::default();
let mut user = FILETIME::default();
let running = unsafe { GetExitCodeProcess(handle, &mut exit_code) } != 0
&& exit_code == STILL_ACTIVE as u32
&& unsafe { GetProcessTimes(handle, &mut created, &mut exited, &mut kernel, &mut user) }
!= 0;
unsafe { CloseHandle(handle) };
running
.then(|| (u64::from(created.dwHighDateTime) << 32) | u64::from(created.dwLowDateTime))
.and_then(NonZeroU64::new)
.map(ProcessStart)
}
#[cfg(target_os = "linux")]
pub(crate) fn process_started(pid: u32) -> Option<ProcessStart> {
if pid == 0 {
return None;
}
let stat = std::fs::read_to_string(format!("/proc/{pid}/stat")).ok()?;
let fields: Vec<_> = stat.rsplit_once(')')?.1.split_whitespace().collect();
let state = *fields.first()?;
if matches!(state, "Z" | "X") {
return None;
}
fields
.get(19)?
.parse()
.ok()
.and_then(NonZeroU64::new)
.map(ProcessStart)
}
#[cfg(target_os = "macos")]
pub(crate) fn process_started(pid: u32) -> Option<ProcessStart> {
use std::ffi::{c_int, c_void};
#[repr(C)]
#[derive(Default)]
struct ProcBsdInfo {
flags: u32,
status: u32,
xstatus: u32,
pid: u32,
ppid: u32,
uid: u32,
gid: u32,
ruid: u32,
rgid: u32,
svuid: u32,
svgid: u32,
rfu_1: u32,
comm: [u8; 16],
name: [u8; 32],
nfiles: u32,
pgid: u32,
pjobc: u32,
e_tdev: u32,
e_tpgid: u32,
nice: i32,
start_tvsec: u64,
start_tvusec: u64,
}
extern "C" {
fn proc_pidinfo(
pid: c_int,
flavor: c_int,
arg: u64,
buffer: *mut c_void,
size: c_int,
) -> c_int;
}
const PROC_PIDTBSDINFO: c_int = 3;
const SZOMB: u32 = 5;
let pid = c_int::try_from(pid).ok().filter(|pid| *pid > 0)?;
let mut info = ProcBsdInfo::default();
let size = c_int::try_from(std::mem::size_of::<ProcBsdInfo>()).ok()?;
let read = unsafe {
proc_pidinfo(
pid,
PROC_PIDTBSDINFO,
0,
(&mut info as *mut ProcBsdInfo).cast(),
size,
)
};
if read != size || info.status == SZOMB {
return None;
}
NonZeroU64::new(
info.start_tvsec
.saturating_mul(1_000_000)
.saturating_add(info.start_tvusec),
)
.map(ProcessStart)
}
impl Record {
pub fn owner_is_running(&self) -> bool {
self.owner_started
.is_some_and(|started| process_started(self.owner_pid) == Some(started))
}
pub fn liveness(&self) -> Liveness {
match self.state == Lifecycle::Open && self.owner_is_running() {
true => Liveness::Live,
false => Liveness::Stale,
}
}
pub fn session(&self) -> Session {
Session {
token: SessionToken(self.token.to_string()),
worktree: self.worktree.clone(),
branch: self.branch.to_string(),
base: self.base.to_string(),
conflict: None,
}
}
pub fn lease(&self) -> String {
occupancy_identity(&self.run_root)
}
pub fn tree_is_its_own(&self) -> Result<bool> {
if self.tree_is_always_its_own() {
return Ok(true);
}
Ok(self.tree_is_its_own_of(&OpenRoots::read()?))
}
pub(crate) fn tree_is_its_own_of(&self, open: &OpenRoots) -> bool {
if self.tree_is_always_its_own() {
return true;
}
!open.taken_from(&self.token, &self.run_root)
&& self.worktree.is_dir()
&& git::current_branch(&self.worktree).is_ok_and(|current| current == *self.branch)
}
fn tree_is_always_its_own(&self) -> bool {
self.slot.is_none() || self.state == Lifecycle::Open
}
}
#[derive(Debug, Clone, Default)]
pub(crate) struct OpenRoots(Vec<(Token, PathBuf)>);
impl OpenRoots {
pub(crate) fn of(records: &[Record]) -> Self {
Self(
records
.iter()
.filter(|record| record.state == Lifecycle::Open)
.map(|record| (record.token.clone(), record.run_root.clone()))
.collect(),
)
}
pub(crate) fn read() -> Result<Self> {
Ok(Self::of(&all()?))
}
fn taken_from(&self, token: &Token, run_root: &Path) -> bool {
self.0
.iter()
.any(|(held, root)| root == run_root && held != token)
}
}
pub fn occupancy_identity(run_root: &Path) -> String {
format!("run:{}", run_root.display())
}
pub(crate) fn record_path(token: &str) -> Result<PathBuf> {
Ok(home::sessions_dir()?.join(format!("{token}.json")))
}
pub fn load(token: &str) -> Result<Record> {
if !ids::is_safe_name(token) {
return Err(error::invalid(format!(
"{token:?} is not a session token; `onevcs session open` prints one"
)));
}
let path = record_path(token)?;
let raw = std::fs::read_to_string(&path).map_err(|failure| match failure.kind() {
std::io::ErrorKind::NotFound => Error::Invalid {
reason: format!("no session {token:?} is open; `onevcs session open` prints a token"),
},
_ => error::at("read the session record at", &path)(failure),
})?;
let document: Value =
serde_json::from_str(&raw).map_err(error::at("read the session record at", &path))?;
let mut record: Record = serde_json::from_value(document.clone())
.map_err(error::at("read the session record at", &path))?;
record.carried = Remainder::between(
&document,
&serde_json::to_value(&record).map_err(error::at("read the session record at", &path))?,
);
usable(&path, token, &record)?;
Ok(record)
}
fn usable(path: &Path, token: &str, record: &Record) -> Result<()> {
if record.version != RECORD_VERSION {
return Err(error::invalid(format!(
"the session record at {} declares version {}; this build reads version \
{RECORD_VERSION}",
path.display(),
record.version
)));
}
if *record.token != *token {
return Err(error::invalid(format!(
"the session record at {} is for {:?}, not for {token:?}",
path.display(),
record.token.to_string()
)));
}
for (what, value) in [
("worktree", &record.worktree),
("clone", &record.clone),
("run root", &record.run_root),
("execution checkout", &record.execution_checkout),
("publication checkout", &record.publication_checkout),
] {
if !value.is_absolute() {
return Err(error::invalid(format!(
"the session record at {} names a {what} at {}, which is not an absolute path",
path.display(),
value.display()
)));
}
}
Ok(())
}
fn record_lock(token: &str) -> Result<lock::Guard> {
let directory = home::locks_dir()?.join("sessions");
home::ensure_dir(&directory)?;
lock::exclusive_at(&directory.join(format!("{token}.lock")))
}
pub(crate) fn update(token: &str, change: impl FnOnce(&mut Record)) -> Result<Record> {
let _guard = record_lock(token)?;
let mut record = load(token)?;
change(&mut record);
save(&record)?;
Ok(record)
}
fn create(record: &Record) -> Result<()> {
let _guard = record_lock(&record.token)?;
let path = record_path(&record.token)?;
if path.exists() {
return Err(error::invalid(format!(
"a session record for {} already exists at {}; a token names one session",
record.token,
path.display()
)));
}
save(record)
}
fn save(record: &Record) -> Result<()> {
followable(record)?;
let path = record_path(&record.token)?;
let mut document = serde_json::to_value(record).map_err(error::at("serialize", &path))?;
record.carried.restore(&mut document);
let json = serde_json::to_string_pretty(&document).map_err(error::at("serialize", &path))?;
home::atomic_write(&path, &format!("{json}\n"))
}
fn followable(record: &Record) -> Result<()> {
let Some(next) = &record.retried_by else {
return Ok(());
};
let refuse = |what: String| {
Err(error::invalid(format!(
"the session {token} cannot record that {next} continued its branch: {what}",
token = record.token,
)))
};
match load(next) {
Err(failure) => return refuse(format!("{failure}")),
Ok(followed) if followed.identity != record.identity => {
return refuse(format!(
"that session belongs to {there:?} and this one to {here:?}, and a session \
continues a branch of its own repository",
there = followed.identity,
here = record.identity,
))
}
Ok(_) => {}
}
match newest(record) {
Ok(_) => Ok(()),
Err(broken) => refuse(broken),
}
}
pub fn all() -> Result<Vec<Record>> {
let directory = home::sessions_dir()?;
let entries = match std::fs::read_dir(&directory) {
Ok(entries) => entries,
Err(absent) if absent.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
Err(refused) => {
return Err(error::at("list the session records in", &directory)(
refused,
))
}
};
let mut records = Vec::new();
for entry in entries {
let entry = entry.map_err(error::at("list the session records in", &directory))?;
let Some(token) = entry
.file_name()
.to_string_lossy()
.strip_suffix(".json")
.map(str::to_owned)
else {
continue;
};
records.push(load(&token)?);
}
records.sort_by(|a, b| a.token.cmp(&b.token));
Ok(records)
}
pub fn holders(repo: &str) -> Result<Vec<SessionHolder>> {
let registry = store::load()?;
let resolution = store::resolve(®istry, repo)?;
let records = all()?;
let open = OpenRoots::of(&records);
let mut holders = Vec::new();
for record in records {
if record.identity == resolution.key && !spent(&record, &open)? {
holders.push(SessionHolder::from(record));
}
}
Ok(holders)
}
pub(crate) fn spent_among(listed: &[Record]) -> Result<Vec<Record>> {
let open = OpenRoots::of(listed);
let mut records = Vec::new();
for record in listed {
if spent(record, &open)? {
records.push(record.clone());
}
}
Ok(records)
}
fn spent(record: &Record, open: &OpenRoots) -> Result<bool> {
Ok(!record.owner_is_running()
&& (!record.tree_is_its_own_of(open) || processes::holding(&record.run_root).is_empty())
&& !holds_unpublished_work(record, open)?)
}
fn holds_unpublished_work(record: &Record, open: &OpenRoots) -> Result<bool> {
let reference = format!("refs/heads/{}", record.branch);
let carried = [&record.clone, &record.execution_checkout]
.into_iter()
.any(|repo| {
git::is_repo(repo) && !matches!(git::unpublished_ahead(repo, &reference, &[]), Ok(0))
});
Ok(carried
|| (record.tree_is_its_own_of(open)
&& record.worktree.is_dir()
&& git::is_dirty(&record.worktree).unwrap_or(true)))
}
pub(crate) fn forget(record: &Path) -> std::result::Result<(), String> {
match std::fs::remove_file(record) {
Ok(()) => Ok(()),
Err(gone) if gone.kind() == std::io::ErrorKind::NotFound => Ok(()),
Err(kept) => Err(format!("{kept}")),
}
}
pub fn checkouts_of(registry: &Registry, resolution: &Resolution) -> Result<Vec<PathBuf>> {
let mut searched: Vec<PathBuf> = vec![resolution.publication.clone()];
for checkout in registry.checkouts.values() {
if checkout.identity == resolution.key && !searched.contains(&checkout.path) {
searched.push(checkout.path.clone());
}
}
for record in all()? {
if record.identity == resolution.key && !searched.contains(&record.clone) {
searched.push(record.clone.clone());
}
}
Ok(searched)
}
pub(crate) fn identity_dir(identity: &str) -> Result<PathBuf> {
let flattened: String = identity
.chars()
.map(|c| {
if c.is_ascii_alphanumeric() || c == '-' || c == '_' || c == '.' {
c
} else {
'-'
}
})
.collect();
Ok(home::workspaces_dir()?.join(format!("{flattened}-{}", ids::short_digest(identity))))
}
pub fn open(
registry: &Registry,
request: &SessionRequest,
) -> Result<(Record, Stream, Option<OpenConflict>)> {
let resolution = store::resolve(registry, &request.repo)?;
let execution =
execution_checkout(registry, &resolution, request.execution_checkout.as_deref())?;
if !git::is_repo(&execution) {
return Err(Error::Invalid {
reason: format!("{} is not a git checkout", execution.display()),
});
}
label::validate(&request.labels)?;
if let (Some(pinned), Some(proposed)) = (&request.branch, &request.branch_name) {
return Err(both_named(pinned, proposed));
}
let prefix = branches::resolve_for(request.branch_prefix.as_deref())?;
let proposed = request
.branch_name
.as_deref()
.map(|name| stem(name, &prefix.value))
.transpose()?;
let named = |value: String| -> Result<Ref> {
Ref::try_from(value).map_err(|reason| Error::Invalid {
reason: format!("{reason}: it is not a valid branch name"),
})
};
let root = match request.base.as_deref() {
Some(_) => git::default_branch(&execution, "origin").ok(),
None => Some(git::default_branch(&execution, "origin")?),
};
let base = named(match request.base.as_deref() {
Some(base) => base.to_owned(),
None => root.clone().unwrap_or_default(),
})?;
let pinned = request
.branch
.as_deref()
.map(|branch| named(branch.to_owned()))
.transpose()?;
if pinned.as_ref() == Some(&base) {
return Err(same_branch_and_base(&base));
}
if let Some((held, lease)) = resumable(&resolution, pinned.as_ref(), &base, &execution)? {
return resume(&held, lease, &execution, request);
}
let token = ids::session_token();
let mut stream = Stream::open(&token)?;
stream.label("identity", &resolution.key);
refresh(&execution, &mut stream)?;
let branch = match (pinned, &proposed) {
(Some(branch), _) => branch,
(None, Some(composed)) => first_free(
registry,
&resolution,
&execution,
request.branch_name.as_deref().unwrap_or(composed),
composed,
)?,
(None, None) => named(format!("{}onevcs/{token}", prefix.value))?,
};
let continued = match request.branch.is_some() {
true => continuation(registry, &resolution, &execution, &branch, &base)?,
false => None,
};
let identity_root = identity_dir(&resolution.key)?;
let runs = identity_root.join("runs");
home::ensure_dir(&runs)?;
reclaim(&runs)?;
let policy = workspaces::resolve_for(
&resolution,
workspaces::Overrides {
pool: request.pool,
overflow: request.overflow,
},
)?;
let placement = match policy.configured || pool::pool_dir(&identity_root).is_dir() {
true => pool::place(&pool::Ask {
resolution: &resolution,
execution: &execution,
resolved: &policy,
pinned: request.branch.is_some().then_some(&branch),
identity_root: &identity_root,
})?,
false => pool::Placement::RunRoot { _serial: None },
};
let (run_root, slot, cut) = match &placement {
pool::Placement::Slot {
number, dir, cut, ..
} => (dir.clone(), Some(*number), *cut),
pool::Placement::RunRoot { .. } => (runs.join(&token), None, pool::Cut::Fresh),
};
let created = cut.created();
let clone = run_root.join("clone");
let worktree = run_root.join("worktree");
home::ensure_dir(&run_root)?;
let lease = match placement {
pool::Placement::RunRoot { .. } => Some(
lock::try_shared(&occupancy_identity(&run_root))?.ok_or_else(|| Error::Invalid {
reason: format!("the run root {} is already occupied", run_root.display()),
})?,
),
pool::Placement::Slot { .. } => None,
};
let origin = git::remote_url(&execution, "origin")
.unwrap_or_else(|_| execution.to_string_lossy().into_owned());
let (carried, placed) = match cut {
pool::Cut::Fresh => (
git::clone_sharing(&execution, &clone, &origin, &base)?,
Placed::NewWorktree,
),
pool::Cut::KeptClone => (
git::carry_remote_refs(&execution, &clone, &base)?,
Placed::NewWorktree,
),
pool::Cut::Warm => {
let carried = git::carry_remote_refs(&execution, &clone, &base)?;
settle_slot(&run_root, &policy.delete)?;
(carried, Placed::InPlace)
}
};
let Placing {
stack_tip,
conflict,
} = match cut_or_continue(&Cut {
clone: &clone,
worktree: &worktree,
branch: &branch,
base: &base,
root: root.as_deref(),
continued: continued.as_ref(),
publication: &resolution.publication,
placed,
on_conflict: OnMerge::asked_by(request),
}) {
Ok(placing) => placing,
Err(error) => {
drop(lease);
match cut {
pool::Cut::Fresh => {
let _ = std::fs::remove_dir_all(&run_root);
}
pool::Cut::KeptClone => {
let _ = std::fs::remove_dir_all(&worktree);
let _ = git::worktree_prune(&clone);
}
pool::Cut::Warm => restore_returned(&clone, &worktree, &branch, &base, &execution),
}
drop(placement);
return Err(error);
}
};
let record = Record {
version: RECORD_VERSION,
token: Token::try_from(token.clone()).map_err(error::invalid)?,
identity: resolution.key.clone(),
alias: resolution.alias.clone(),
branch,
base,
change_base: None,
stack_tip,
worktree,
clone,
run_root,
slot,
execution_checkout: execution,
publication_checkout: resolution.publication.clone(),
state: Lifecycle::Open,
owner_pid: std::process::id(),
owner_started: process_started(std::process::id()),
retried_by: None,
labels: request.labels.clone(),
carried: Remainder::default(),
};
create(&record)?;
drop(placement);
let reuse = match continued {
Some(_) => Reuse::Continued,
None => Reuse::Cut,
};
if continued.is_some() {
supersede(&record)?;
}
note_base(
&carried,
&record.base,
&record.worktree,
&record.execution_checkout,
);
let applied = match request.branch.is_none() && !prefix.value.is_empty() {
true => Some(&prefix),
false => None,
};
if let Some(conflict) = &conflict {
say_conflict(&record, conflict);
}
stream.emit(
EventKind::SessionOpened,
opened(&record, reuse, created, applied, conflict.as_ref()),
);
drop(lease);
Ok((record, stream, conflict))
}
fn say_conflict(record: &Record, conflict: &OpenConflict) {
eprintln!(
"onevcs: session {token} opened with a merge in progress: {base} conflicts with branch \
{branch:?} in {paths}. Conclude it in {worktree} — resolve, stage and commit the merge \
— before publishing; publication refuses the session until then",
token = record.token,
base = conflict.base_commit,
branch = record.branch,
paths = guidance::listed(&conflict.paths),
worktree = record.worktree.display(),
);
}
fn settle_slot(slot: &Path, delete: &[PathBuf]) -> Result<()> {
let worktree = slot.join("worktree");
if pool::returned(&worktree) {
return Ok(());
}
let current = git::current_branch(&worktree)?;
let previous = all()?
.into_iter()
.filter(|record| record.run_root == slot)
.max_by_key(|record| {
(
*record.branch == *current,
record_path(&record.token)
.ok()
.and_then(|path| std::fs::metadata(path).ok())
.and_then(|meta| meta.modified().ok())
.unwrap_or(std::time::UNIX_EPOCH),
)
});
match previous {
Some(record) => {
let mut stream = Stream::open(&record.token)?;
let handed = preserve_and_hand_back(&record, &mut stream)?;
let stray = stray_work(&record)?;
if !stray.is_empty() {
return Err(stranding(&record, &stray));
}
return_tree(&record, delete, handed.copied)
}
None => {
let clone = slot.join("clone");
let base = git::default_branch(&clone, "origin")
.map(Ref::from_git)
.unwrap_or_else(|_| Ref::from_git("HEAD"));
reset_onto_base(&worktree, &integrated_base(&clone, &base), delete)
}
}
}
fn restore_returned(clone: &Path, worktree: &Path, branch: &Ref, base: &Ref, lender: &Path) {
let _ = reset_onto_base(worktree, &integrated_base(clone, base), &[]);
if let Some(tip) = git::tip(clone, &format!("refs/heads/{branch}")) {
if git::refs_reach(lender, &tip) {
git::delete_ref(clone, &format!("refs/heads/{branch}"));
}
}
}
fn note_base(carried: &git::BaseCarried, base: &Ref, worktree: &Path, execution: &Path) {
let git::BaseCarried::Kept { local, origin } = carried else {
return;
};
let at = execution.to_string_lossy();
eprintln!(
"onevcs: warning: this session's base {base:?} stands at {local}, which origin/{base} at \
{origin} does not carry, so it was left where it was rather than moved off work nothing \
else holds. A `{base}` typed in {tree} names that commit; what onevcs computes is \
addressed from origin/{base} either way. The copy came from {at}, so publish it from \
there with `{push}` — or reconcile it against origin — and open this session again",
tree = worktree.display(),
push = guidance::command(["git", "-C", &at, "push", "origin", base]),
);
}
fn supersede(record: &Record) -> Result<()> {
for older in all()? {
if older.identity != record.identity
|| older.branch != record.branch
|| *older.token == *record.token
|| older.retried_by.is_some()
{
continue;
}
if let Err(failure) = record_retry(&older.token, &record.token) {
eprintln!(
"onevcs: warning: the session {older} is not recorded as continued by {newer}, so \
a report about {branch:?} cannot tell the two apart: {failure}",
older = older.token,
newer = record.token,
branch = record.branch,
);
}
}
Ok(())
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Reuse {
Cut,
Continued,
Resumed,
}
fn opened(
record: &Record,
reuse: Reuse,
created: bool,
prefix: Option<&Sourced<String>>,
conflict: Option<&OpenConflict>,
) -> Map<String, Value> {
let placement = match record.slot {
Some(number) => json!({"kind": "slot", "slot": number, "created": created}),
None => json!({"kind": "run-root"}),
};
let mut payload = object(json!({
"token": record.token,
"identity": record.identity,
"branch": record.branch,
"base": record.base,
"worktree": record.worktree.display().to_string(),
"clone": record.clone.display().to_string(),
"execution_checkout": record.execution_checkout.display().to_string(),
"publication_checkout": record.publication_checkout.display().to_string(),
"placement": placement,
}));
match reuse {
Reuse::Cut => {}
Reuse::Continued => {
payload.insert("continued".to_owned(), Value::Bool(true));
}
Reuse::Resumed => {
payload.insert("reused".to_owned(), Value::Bool(true));
}
}
if let Some(prefix) = prefix {
payload.insert(
"branch_prefix".to_owned(),
json!({"prefix": prefix.value, "from": prefix.from}),
);
}
if let Some(conflict) = conflict {
payload.insert("conflict".to_owned(), json!(conflict));
}
payload
}
fn refresh(execution: &Path, stream: &mut Stream) -> Result<()> {
if git::has_remote(execution, "origin") {
git::fetch(execution, "origin")?;
stream.emit(
EventKind::Fetch,
object(json!({"remote": "origin", "checkout": execution.display().to_string()})),
);
}
git::retain_objects_for_borrowers(execution)
}
fn resumable(
resolution: &Resolution,
branch: Option<&Ref>,
base: &Ref,
execution: &Path,
) -> Result<Option<(Record, lock::Guard)>> {
let Some(branch) = branch else {
return Ok(None);
};
let mut held = all()?.into_iter().filter(|record| {
record.identity == resolution.key
&& record.branch == *branch
&& record.base == *base
&& record.state == Lifecycle::Open
&& record.execution_checkout == execution
&& record.run_root.is_dir()
&& record.clone.is_dir()
});
let (Some(candidate), None) = (held.next(), held.next()) else {
return Ok(None);
};
Ok(lock::try_shared(&candidate.lease())?.map(|lease| (candidate, lease)))
}
fn resume(
held: &Record,
lease: lock::Guard,
execution: &Path,
request: &SessionRequest,
) -> Result<(Record, Stream, Option<OpenConflict>)> {
let labels = &request.labels;
let (mut record, mut stream, _preserved) = adopt_with(&held.token, OnMerge::asked_by(request))?;
drop(lease);
let conflict = match git::merge_in_progress(&record.worktree)? {
Some(merging) if !merging.unmerged.is_empty() => Some(OpenConflict {
paths: merging.unmerged,
base_commit: merging.merging,
branch_tip: git::head_sha(&record.worktree)?,
}),
_ => None,
};
if !labels.is_empty() {
record = update(&record.token, |stored| {
stored
.labels
.extend(labels.iter().map(|(k, v)| (k.clone(), v.clone())));
})?;
}
stream.label("identity", &record.identity);
refresh(execution, &mut stream)?;
let carried = git::carry_remote_refs(execution, &record.clone, &record.base)?;
note_base(&carried, &record.base, &record.worktree, execution);
if let Some(conflict) = &conflict {
say_conflict(&record, conflict);
}
stream.emit(
EventKind::SessionOpened,
opened(&record, Reuse::Resumed, false, None, conflict.as_ref()),
);
Ok((record, stream, conflict))
}
#[derive(Debug, Clone)]
struct Carried {
checkout: Option<PathBuf>,
tip: String,
}
impl Carried {
fn at(&self) -> String {
match &self.checkout {
Some(checkout) => checkout.display().to_string(),
None => "origin".to_owned(),
}
}
fn bring_into(&self, clone: &Path, branch: &Ref) -> Result<()> {
let Some(checkout) = &self.checkout else {
return Ok(());
};
git::fetch_into_ref(clone, &checkout.to_string_lossy(), branch, CONTINUED_REF)?;
Ok(())
}
}
#[derive(Debug, Clone)]
enum Continued {
Held(Carried),
Pushed(Carried),
Both {
held: Carried,
pushed: Carried,
},
}
const CONTINUED_REF: &str = "refs/onevcs/continued";
const RECONCILE_THEN: &str = "open this session again, which continues the copy that is left";
fn continuation(
registry: &Registry,
resolution: &Resolution,
execution: &Path,
branch: &Ref,
base: &Ref,
) -> Result<Option<Continued>> {
let anywhere = checkouts_of(registry, resolution)?
.into_iter()
.any(|repo| git::is_repo(&repo) && git::branch_exists(&repo, branch));
let held = match anywhere {
true => {
let checkout =
crate::branch::locate(registry, resolution, branch, base, RECONCILE_THEN)?;
git::tip(&checkout, &format!("refs/heads/{branch}")).map(|tip| Carried {
checkout: Some(checkout),
tip,
})
}
false => None,
};
let pushed = git::tip(execution, &format!("refs/remotes/origin/{branch}")).map(|tip| Carried {
checkout: None,
tip,
});
Ok(match (held, pushed) {
(Some(held), Some(pushed)) => Some(Continued::Both { held, pushed }),
(Some(held), None) => Some(Continued::Held(held)),
(None, Some(pushed)) => Some(Continued::Pushed(pushed)),
(None, None) => None,
})
}
fn opened_at(clone: &Path, branch: &Ref, continued: &Continued) -> Result<Carried> {
match continued {
Continued::Held(only) | Continued::Pushed(only) => {
only.bring_into(clone, branch)?;
Ok(only.clone())
}
Continued::Both { held, pushed } => {
held.bring_into(clone, branch)?;
pushed.bring_into(clone, branch)?;
if git::is_ancestor(clone, &pushed.tip, &held.tip)? {
return Ok(held.clone());
}
if git::is_ancestor(clone, &held.tip, &pushed.tip)? {
return Ok(pushed.clone());
}
Err(diverged(branch, held, pushed))
}
}
}
fn diverged(branch: &Ref, held: &Carried, pushed: &Carried) -> Error {
let at = held.at();
Error::Invalid {
reason: format!(
"branch {branch:?} stands at {here} in {at} and at {there} on {theirs}, and neither \
copy carries the other, so a session continuing it would have to leave one of them \
behind. Reconcile them where the branch is — `{fetch}` brings origin's copy in as \
FETCH_HEAD, to merge or rebase onto the one that is there — and then {RECONCILE_THEN}",
here = held.tip,
there = pushed.tip,
theirs = pushed.at(),
fetch = guidance::command(["git", "-C", &at, "fetch", "origin", branch]),
),
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Placed {
NewWorktree,
InPlace,
}
pub(crate) fn integrated_base(clone: &Path, base: &Ref) -> String {
let remote = format!("origin/{base}");
match git::ref_exists(clone, &format!("refs/remotes/{remote}")) {
true => remote,
false => base.to_string(),
}
}
struct Cut<'a> {
clone: &'a Path,
worktree: &'a Path,
branch: &'a Ref,
base: &'a Ref,
root: Option<&'a str>,
continued: Option<&'a Continued>,
publication: &'a Path,
placed: Placed,
on_conflict: OnMerge,
}
struct Placing {
stack_tip: Option<String>,
conflict: Option<OpenConflict>,
}
fn cut_or_continue(cut: &Cut<'_>) -> Result<Placing> {
let Cut {
clone,
worktree,
branch,
base,
root,
continued,
publication,
placed,
on_conflict,
} = *cut;
let integrated = integrated_base(clone, base);
let Some(continued) = continued else {
match placed {
Placed::NewWorktree => git::worktree_add(clone, worktree, branch, &integrated)?,
Placed::InPlace => git::checkout_new(worktree, branch, &integrated)?,
}
let stack_tip = match root {
Some(root) if *root != **base => Some(git::head_sha(worktree)?),
_ => None,
};
return Ok(Placing {
stack_tip,
conflict: None,
});
};
let opened = opened_at(clone, branch, continued)?;
git::update_ref(clone, &format!("refs/heads/{branch}"), &opened.tip)?;
git::delete_ref(clone, CONTINUED_REF);
git::detach_head(clone)?;
match placed {
Placed::NewWorktree => git::worktree_add_existing(clone, worktree, branch)?,
Placed::InPlace => git::checkout_existing(worktree, branch)?,
}
let stack_tip = match root {
Some(root) if *root != **base => git::merge_base(clone, &integrated, branch)?,
_ => None,
};
let conflict = integrate(
worktree,
&integrated,
branch,
&opened,
publication,
on_conflict,
)?;
Ok(Placing {
stack_tip,
conflict,
})
}
fn integrate(
worktree: &Path,
integrated: &str,
branch: &Ref,
opened: &Carried,
publication: &Path,
on_conflict: OnMerge,
) -> Result<Option<OpenConflict>> {
if on_conflict == OnMerge::Keep {
let message = crate::publish::merge_message(integrated, branch);
let git::Integrated::Conflicted(conflict) =
git::merge_leaving_conflict(worktree, integrated, &message)?
else {
return Ok(None);
};
let base_commit = git::tip(worktree, "MERGE_HEAD").ok_or_else(|| {
error::invalid(format!(
"the merge of {integrated} into {branch:?} in {} stopped on a conflict but \
names no MERGE_HEAD",
worktree.display()
))
})?;
return Ok(Some(OpenConflict {
paths: conflict.paths().to_vec(),
base_commit,
branch_tip: opened.tip.clone(),
}));
}
let crate::publish::Reconciled::Conflicted(_, conflict) =
crate::publish::reconcile(worktree, integrated, branch, None)?
else {
return Ok(None);
};
Err(Error::SyncConflict {
reason: format!(
"{integrated} conflicts with branch {branch:?} in {conflicting} at {tip}, which is the copy in {at} \
this session would continue, so opening it would leave a conflict in a worktree \
nobody asked to resolve. The branch is untouched. Resolve the conflict on it — check \
it out in {publication} and merge {integrated} into it — and open this session \
again, or land it as it stands with `{land}`",
conflicting = crate::guidance::listed(conflict.paths()),
tip = opened.tip,
at = opened.at(),
publication = publication.display(),
land = guidance::command([
"onevcs",
"publish-branch",
branch,
"--repo",
&publication.to_string_lossy(),
]),
),
})
}
fn same_branch_and_base(base: &Ref) -> Error {
Error::Invalid {
reason: format!(
"branch {base:?} is also this session's base, so it would be published into itself \
and every publication of it would answer that there is nothing to publish. Pass \
`--branch {base}` on its own — a branch that already exists is continued from its \
own tip — and pass `--base` only to name the branch this work is merged with and \
published into"
),
}
}
const SUFFIXES: u32 = 1_000;
fn both_named(pinned: &str, proposed: &str) -> Error {
Error::Invalid {
reason: format!(
"this request names both a branch to continue ({pinned:?}) and a name to cut \
({proposed:?}), which are two answers to one question. Pass `--branch {pinned}` \
to continue that branch — one that already exists is opened at its own tip — or \
`--branch-name {proposed}` to have a fresh branch cut from that name, never both"
),
}
}
fn stem(proposed: &str, prefix: &str) -> Result<String> {
let Some(sanitized) = branches::sanitize(proposed) else {
return Err(Error::Invalid {
reason: format!(
"the branch name {proposed:?} has nothing in it a branch name can be made \
of: every character git would refuse is replaced, and what is left is \
empty. Propose a name carrying at least one letter, digit or underscore"
),
});
};
let composed = format!("{prefix}{sanitized}");
if !git::is_valid_branch_name(&composed) {
return Err(Error::Invalid {
reason: format!(
"the branch name {proposed:?} becomes {composed:?} once it is sanitized and \
prefixed, and git would not accept that as a branch name"
),
});
}
Ok(composed)
}
fn first_free(
registry: &Registry,
resolution: &Resolution,
execution: &Path,
proposed: &str,
stem: &str,
) -> Result<Ref> {
let taken = taken_names(registry, resolution, execution)?;
for nth in 1..=SUFFIXES {
let candidate = match nth {
1 => stem.to_owned(),
nth => format!("{stem}-{nth}"),
};
if taken.contains(&candidate) {
continue;
}
return Ref::try_from(candidate).map_err(|reason| Error::Invalid {
reason: format!("{reason}: it is not a valid branch name"),
});
}
Err(Error::Invalid {
reason: format!(
"the branch name {proposed:?} is taken, and so is every name from {stem}-2 to \
{stem}-{SUFFIXES}: {identity} has a thousand branches of that name already. \
Propose a name that says which piece of work this is",
identity = resolution.key,
),
})
}
fn taken_names(
registry: &Registry,
resolution: &Resolution,
execution: &Path,
) -> Result<std::collections::BTreeSet<String>> {
let mut taken = std::collections::BTreeSet::new();
for repo in checkouts_of(registry, resolution)? {
if git::is_repo(&repo) {
taken.extend(git::branches(&repo).unwrap_or_default());
}
}
taken.extend(git::remote_branches(execution, "origin")?);
Ok(taken)
}
pub(crate) fn execution_checkout(
registry: &Registry,
resolution: &Resolution,
alias: Option<&str>,
) -> Result<PathBuf> {
let Some(alias) = alias else {
return Ok(resolution.publication.clone());
};
let checkout = registry
.checkouts
.get(alias)
.ok_or_else(|| Error::Invalid {
reason: format!("{alias:?} is not a registered checkout"),
})?;
if checkout.identity != resolution.key {
return Err(Error::Invalid {
reason: format!(
"execution checkout {alias:?} belongs to identity {:?}, not to {:?}",
checkout.identity, resolution.key
),
});
}
Ok(checkout.path.clone())
}
pub fn adopt(token: &str) -> Result<(Record, Stream, Option<String>)> {
adopt_with(token, OnMerge::Refuse)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum OnMerge {
Refuse,
Keep,
}
impl OnMerge {
fn asked_by(request: &SessionRequest) -> Self {
match request.refuse_conflicts {
true => Self::Refuse,
false => Self::Keep,
}
}
}
fn adopt_with(token: &str, on_merge: OnMerge) -> Result<(Record, Stream, Option<String>)> {
let record = load(token)?;
let ours = record.tree_is_its_own()?;
if let Some(number) = record.slot.filter(|_| !ours) {
return Err(error::invalid(format!(
"session {token:?} closed and returned slot {number} of {identity}, so there is no \
tree of its own to re-attach to; its branch {branch:?} was handed to {checkout}. \
Open a session pinned to that branch to continue it: `{open}`",
identity = record.identity,
branch = record.branch,
checkout = record.execution_checkout.display(),
open = guidance::command([
"onevcs",
"session",
"open",
&record.alias,
"--branch",
&record.branch
]),
)));
}
let mut stream = Stream::open(token)?;
let lease = lock::try_shared(&record.lease())?.ok_or_else(|| Error::Invalid {
reason: format!(
"session {token:?} is occupied by another process (opened by pid {}); \
wait for it or close the session",
record.owner_pid
),
})?;
if !record.clone.is_dir() {
return Err(error::invalid(format!(
"session {token:?} has been reclaimed: only the newest {RETAINED_DEAD_RUNS} \
abandoned sessions holding unpublished work are kept. Its branch {:?} was handed \
to {} before it went.",
record.branch,
record.execution_checkout.display()
)));
}
if !record.worktree.is_dir() {
git::worktree_prune(&record.clone)?;
git::worktree_add_existing(&record.clone, &record.worktree, &record.branch)?;
}
let mut preserved = None;
let merging = git::merge_in_progress(&record.worktree)?.is_some();
if merging && on_merge == OnMerge::Refuse {
crate::vcs::refuse_unfinished_merge(
&record.worktree,
&record.branch,
&format!(
"adopt it again with `{}`",
guidance::command(["onevcs", "session", "adopt", token])
),
)?;
}
if !merging && git::is_dirty(&record.worktree)? {
let branch = crate::vcs::preserve_into(
&record,
&mut stream,
crate::session::Provenance::IncompleteStep,
)?;
preserved = Some(branch.branch);
}
let record = update(token, |stored| {
stored.state = Lifecycle::Open;
stored.owner_pid = std::process::id();
stored.owner_started = process_started(std::process::id());
})?;
drop(lease);
Ok((record, stream, preserved))
}
pub(crate) const DETACHED_HEAD: &str = "HEAD";
const NAMED_SHA: usize = 12;
struct Stray {
branch: Ref,
commits: u64,
preserved: bool,
}
pub fn close(token: &str) -> Result<Record> {
let record = load(token)?;
let lease = lock::try_shared(&record.lease())?.ok_or_else(|| Error::Invalid {
reason: format!("session {token:?} is occupied by another process"),
})?;
let ours = record.tree_is_its_own()?;
if ours {
let occupants = occupants(&record);
if !occupants.is_empty() {
return Err(occupied(&record, &occupants));
}
}
let mut stream = Stream::open(token)?;
let mut closed = json!({"token": record.token, "branch": record.branch});
if ours && record.clone.is_dir() {
if let Err(failure) = release_the_tree(&record, &mut stream, &mut closed) {
let Some(landing) = released_from_the_record(&record) else {
return Err(failure);
};
closed["landed"] = json!(landing.commit);
closed["retained"] = json!(record.clone.to_string_lossy());
closed["unreadable_clone"] = json!(landing.said);
eprintln!(
"onevcs: warning: the run clone of session {token} at {clone} could not be read \
by git ({said}), and this session's work is recorded as landed at {commit} — so \
the session is closed from that record and the clone is left exactly as it is \
for whoever repairs or removes it",
token = record.token,
clone = record.clone.display(),
said = landing.said,
commit = landing.commit,
);
}
}
stream.emit(EventKind::SessionClosed, object(closed));
let record = update(token, |stored| stored.state = Lifecycle::Closed)?;
drop(lease);
Ok(record)
}
fn release_the_tree(record: &Record, stream: &mut Stream, closed: &mut Value) -> Result<()> {
let handed = preserve_and_hand_back(record, stream)?;
if let Some(preserved) = &handed.preserved {
closed["preserved"] = json!(preserved);
}
if !handed.copied {
closed["retained"] = json!(record.clone.to_string_lossy());
}
let stray = stray_work(record)?;
if !stray.is_empty() {
return Err(stranding(record, &stray));
}
match record.slot {
Some(number) => {
return_tree(record, &delete_paths_for(record)?, handed.copied)?;
closed["returned"] = json!({"slot": number});
}
None => {
if record.worktree.is_dir() {
git::worktree_remove(&record.clone, &record.worktree)?;
}
}
}
Ok(())
}
struct FromTheRecord {
commit: String,
said: String,
}
fn released_from_the_record(record: &Record) -> Option<FromTheRecord> {
let said = unreadable_objects(&record.clone)?;
let mut notes = Vec::new();
let streams = crate::status::recorded_streams(&mut notes).ok()?;
let commit = crate::status::recorded_for(
&streams,
&record.identity,
&record.branch,
Some(&record.token),
)
.landing?;
Some(FromTheRecord {
commit: commit.as_str().to_owned(),
said,
})
}
fn unreadable_objects(clone: &Path) -> Option<String> {
match git::run(
&["rev-list", "--objects", "--no-walk", "--all"],
Some(clone),
) {
Ok(output) if output.ok() => None,
Ok(output) => Some(output.diagnostic()),
Err(failure) => Some(failure.to_string()),
}
}
fn occupants(record: &Record) -> Vec<processes::Holder> {
processes::holding(&record.run_root)
}
struct HandedBack {
preserved: Option<String>,
copied: bool,
}
fn preserve_and_hand_back(record: &Record, stream: &mut Stream) -> Result<HandedBack> {
let mut preserved = None;
git::abort_merge(&record.worktree);
if record.worktree.is_dir() && git::is_dirty(&record.worktree)? {
let branch =
crate::vcs::preserve_into(record, stream, crate::session::Provenance::IncompleteStep)?;
preserved = Some(branch.branch);
}
Ok(HandedBack {
preserved,
copied: hand_back(record),
})
}
fn return_tree(record: &Record, delete: &[PathBuf], copied: bool) -> Result<()> {
let integrated = integrated_base(&record.clone, &record.base);
reset_onto_base(&record.worktree, &integrated, delete)?;
if copied {
git::delete_ref(&record.clone, &format!("refs/heads/{}", record.branch));
}
Ok(())
}
pub(crate) fn reset_onto_base(worktree: &Path, base: &str, delete: &[PathBuf]) -> Result<()> {
if !git::is_repo(worktree) {
return Err(error::invalid(format!(
"the slot worktree at {} is not a repository, so it cannot be returned; the next \
session to take the slot recreates it",
worktree.display()
)));
}
git::abort_merge(worktree);
git::checkout_detached(worktree)?;
git::reset_hard(worktree, base)?;
git::clean_untracked(worktree)?;
for path in delete {
let target = worktree.join(path);
if let Some(link) = symlink_on_the_way(worktree, path) {
return Err(error::invalid(format!(
"the delete entry {path:?} of the workspaces file passes through {}, which is a \
symbolic link, so deleting it could reach outside the slot worktree at {}; it \
was not deleted. Name the path the link points at, inside the worktree, or \
remove the entry",
link.display(),
worktree.display()
)));
}
let Ok(meta) = std::fs::symlink_metadata(&target) else {
continue;
};
let removed = match meta.is_dir() {
true => std::fs::remove_dir_all(&target),
false => std::fs::remove_file(&target),
};
removed.map_err(error::at("delete", &target))?;
}
Ok(())
}
fn symlink_on_the_way(worktree: &Path, path: &Path) -> Option<PathBuf> {
let mut walked = worktree.to_path_buf();
let components: Vec<_> = path.components().collect();
for component in components.iter().take(components.len().saturating_sub(1)) {
walked.push(component);
let is_link = std::fs::symlink_metadata(&walked)
.map(|meta| meta.file_type().is_symlink())
.unwrap_or(false);
if is_link {
return Some(walked);
}
}
None
}
fn delete_paths_for(record: &Record) -> Result<Vec<PathBuf>> {
let registry = store::load()?;
let Ok(resolution) = store::resolve(®istry, &record.identity) else {
return Ok(Vec::new());
};
Ok(workspaces::resolve_for(&resolution, workspaces::Overrides::default())?.delete)
}
fn occupied(record: &Record, holders: &[processes::Holder]) -> Error {
let inside: Vec<String> = holders.iter().map(ToString::to_string).collect();
let inside: Vec<&str> = inside.iter().map(String::as_str).collect();
error::invalid(format!(
"session {token:?} was not closed: {count} still working inside its run root {root} — \
{inside} — and removing its worktree {worktree} would take the directory out from \
under {them}. Let the dispatch that holds it finish, or stop it, and then re-run \
`{close}`",
token = &*record.token,
count = counted_processes(holders.len()),
root = record.run_root.display(),
worktree = record.worktree.display(),
inside = guidance::listed(&inside),
them = match holders.len() {
1 => "it",
_ => "them",
},
close = guidance::command(["onevcs", "session", "close", &record.token]),
))
}
fn counted_processes(holders: usize) -> String {
match holders {
1 => "one process is".to_owned(),
many => format!("{many} processes are"),
}
}
fn hand_back(record: &Record) -> bool {
git::copy_branch(&record.clone, &record.execution_checkout, &record.branch).unwrap_or(false)
}
fn stray_work(record: &Record) -> Result<Vec<Stray>> {
let mut found: Vec<(Ref, u64)> = Vec::new();
if record.worktree.is_dir() && git::current_branch(&record.worktree)? == DETACHED_HEAD {
let head = git::head_sha(&record.worktree)?;
if let Some(commits) = stranded(record, &head)? {
let branch = detached_name(&record.branch, &head);
git::update_ref(&record.clone, &format!("refs/heads/{branch}"), &head)?;
found.push((Ref::from_git(branch), commits));
}
}
let retained_by_others: Vec<String> = match record.slot {
Some(_) => all()?
.into_iter()
.filter(|other| {
other.identity == record.identity
&& *other.token != *record.token
&& other.state == Lifecycle::Closed
})
.map(|other| other.branch.to_string())
.collect(),
None => Vec::new(),
};
for branch in git::branches(&record.clone)? {
let branch = Ref::from_git(branch);
if *branch == *record.branch
|| found.iter().any(|(held, _)| **held == *branch)
|| retained_by_others.iter().any(|held| **held == *branch)
{
continue;
}
if let Some(commits) = stranded(record, &branch)? {
found.push((branch, commits));
}
}
Ok(found
.into_iter()
.map(|(branch, commits)| Stray {
preserved: git::copy_branch(&record.clone, &record.execution_checkout, &branch)
.unwrap_or(false),
branch,
commits,
})
.collect())
}
fn stranded(record: &Record, reference: &str) -> Result<Option<u64>> {
let commits = git::unpublished_ahead(&record.clone, reference, &[&*record.branch])?;
if commits == 0 {
return Ok(None);
}
let held = git::tip(&record.clone, reference)
.is_some_and(|tip| git::refs_reach(&record.execution_checkout, &tip));
Ok((!held).then_some(commits))
}
fn detached_name(branch: &Ref, head: &str) -> String {
let short: String = head.chars().take(NAMED_SHA).collect();
format!("{branch}-detached-{short}")
}
fn stranding(record: &Record, stray: &[Stray]) -> Error {
let names: Vec<&str> = stray.iter().map(|found| &*found.branch).collect();
let refused: Vec<&str> = stray
.iter()
.filter(|found| !found.preserved)
.map(|found| &*found.branch)
.collect();
let total = stray.iter().map(|found| found.commits).sum();
let checkout = record.execution_checkout.display();
let mut reason = format!(
"session {:?} was not closed: its worktree {} holds work its branch {:?} does not \
carry — {} on {} — and removing the worktree is what would have made it unreachable.",
&*record.token,
record.worktree.display(),
record.branch,
counted(total),
guidance::listed(&names),
);
if refused.is_empty() {
reason.push_str(&format!(" All of it has been copied into {checkout}."));
} else {
reason.push_str(&format!(
" {checkout} would not take {}, which is still only in {}; put it there under a \
name that checkout will accept with `{}`.",
guidance::listed(&refused),
record.clone.display(),
import_command(record, refused[0]),
));
}
reason.push_str(&format!(
" See what is preserved and the verb that lands each branch with `{}`, then re-run \
`{}`, which releases the worktree once the work is reachable outside it",
guidance::command(["onevcs", "recoverable"]),
guidance::command(["onevcs", "session", "close", &record.token]),
));
error::invalid(reason)
}
fn import_command(record: &Record, branch: &str) -> String {
guidance::command([
"onevcs",
"import",
branch,
"--repo",
&record.execution_checkout.to_string_lossy(),
"--from",
&record.clone.to_string_lossy(),
])
}
fn counted(commits: u64) -> String {
match commits {
1 => "1 commit".to_owned(),
many => format!("{many} commits"),
}
}
fn run_roots_of_open_sessions() -> Result<Vec<PathBuf>> {
Ok(all()?
.into_iter()
.filter(|record| record.state == Lifecycle::Open)
.map(|record| record.run_root)
.collect())
}
fn reclaim(runs: &Path) -> Result<()> {
let Ok(entries) = std::fs::read_dir(runs) else {
return Ok(());
};
let open = run_roots_of_open_sessions()?;
let mut holding_work: Vec<(std::time::SystemTime, PathBuf)> = Vec::new();
for entry in entries.flatten() {
let run_root = entry.path();
if !run_root.is_dir() {
continue;
}
if open.iter().any(|held| held == &run_root) {
continue;
}
let Some(exclusive) = lock::try_exclusive(&occupancy_identity(&run_root))? else {
continue;
};
drop(exclusive);
let clone = run_root.join("clone");
let unpublished = if git::is_repo(&clone) {
git::unpublished_branches(&clone).unwrap_or_default()
} else {
Vec::new()
};
if unpublished.is_empty() {
let _ = std::fs::remove_dir_all(&run_root);
} else {
let written = std::fs::metadata(&run_root)
.and_then(|meta| meta.modified())
.unwrap_or(std::time::UNIX_EPOCH);
holding_work.push((written, run_root));
}
}
holding_work.sort_by_key(|(written, _)| std::cmp::Reverse(*written));
for (_, reclaimed) in holding_work.into_iter().skip(RETAINED_DEAD_RUNS) {
let _ = std::fs::remove_dir_all(&reclaimed);
}
Ok(())
}
pub fn object(value: Value) -> Map<String, Value> {
value.as_object().cloned().unwrap_or_default()
}
#[cfg(test)]
mod process_tests {
use super::{process_started, ProcessStart, Record, Ref, Token, RECORD_VERSION};
use crate::session::{Lifecycle, Liveness, SessionHolder};
use std::num::NonZeroU64;
use std::path::PathBuf;
use std::process::Command;
fn holder_for(owner_started: Option<ProcessStart>) -> SessionHolder {
SessionHolder::from(Record {
version: RECORD_VERSION,
token: Token::try_from("s-process-test".to_owned()).expect("a token"),
identity: "identity".to_owned(),
alias: "alias".to_owned(),
branch: Ref::from_git("main"),
base: Ref::from_git("main"),
change_base: None,
stack_tip: None,
worktree: PathBuf::from("worktree"),
clone: PathBuf::from("clone"),
run_root: PathBuf::from("run"),
slot: None,
execution_checkout: PathBuf::from("execution"),
publication_checkout: PathBuf::from("publication"),
state: Lifecycle::Open,
owner_pid: std::process::id(),
owner_started,
retried_by: None,
labels: Default::default(),
carried: crate::remainder::Remainder::default(),
})
}
#[test]
fn process_identity_is_live_then_stale_after_the_child_is_reaped() {
let mut child = if cfg!(windows) {
Command::new("powershell")
.args(["-NoProfile", "-Command", "Start-Sleep -Seconds 30"])
.spawn()
} else {
Command::new("sh").args(["-c", "sleep 30"]).spawn()
}
.expect("spawn a child that remains alive");
let pid = child.id();
let started = process_started(pid).expect("the running child has an identity");
assert_eq!(process_started(pid), Some(started));
child.kill().expect("terminate the child");
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
while process_started(pid).is_some() && std::time::Instant::now() < deadline {
std::thread::yield_now();
}
assert_eq!(
process_started(pid),
None,
"an exited child is stale before its handle is reaped"
);
child.wait().expect("reap the child");
assert_eq!(process_started(pid), None);
assert_eq!(process_started(0), None);
}
#[test]
fn holder_requires_the_same_live_process_instance_not_only_the_same_pid() {
let started = process_started(std::process::id()).expect("this process is live");
assert!(matches!(holder_for(Some(started)).liveness, Liveness::Live));
let different_value = started.0.get().checked_add(1).unwrap_or(1);
let different = NonZeroU64::new(different_value)
.map(ProcessStart)
.expect("a nonzero different creation identity");
assert!(matches!(
holder_for(Some(different)).liveness,
Liveness::Stale
));
assert!(matches!(holder_for(None).liveness, Liveness::Stale));
}
}