use std::collections::{HashMap, HashSet};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
use std::sync::{Arc, Condvar, Mutex, OnceLock, RwLock};
use std::thread::{self, JoinHandle};
use std::time::{Duration, Instant};
use crossbeam_channel::{Receiver, Sender, select};
use rayon::iter::{IntoParallelIterator, ParallelIterator};
use crate::base;
use crate::cell::{Cell, Generation, Settled, Timestamp, Unknown};
use crate::default_branch;
use crate::discovery::{self, SetSpec};
use crate::entity::{
ActionReceipt, DefaultBranch, DeleteRisk, DirtyCounts, EntityKey, EntityState, Head, Kind,
OwnWork, Presence, RunningStep, Skip, StepOutcome, StepResult, SyncState, WorktreeState,
};
use crate::environment;
use crate::executor;
use crate::filter::{Applicability, Filter, Partition};
use crate::git;
use crate::landing;
#[cfg(any(test, feature = "test-util"))]
use crate::liveness;
use crate::patch_equivalence;
use crate::poll;
use crate::snapshot::Snapshot;
#[allow(dead_code)] const FIRST_FRAME_NAMES_BUDGET_MS: u64 = 50;
#[allow(dead_code)] const FIRST_FRAME_CHEAP_COLUMNS_BUDGET_MS: u64 = 200;
#[derive(Debug, Clone)]
pub struct RepoOverride {
pub path: PathBuf,
pub default_branch: Option<String>,
pub excluded: bool,
}
#[derive(Debug, Clone)]
pub struct Step {
pub argv: Vec<String>,
pub shell: bool,
pub interactive: bool,
pub env: Vec<(String, String)>,
}
#[derive(Debug, Clone)]
pub struct ActionSpec {
pub label: Arc<str>,
pub name: Option<Arc<str>>,
pub steps: Vec<Step>,
pub concurrency: u32,
pub when: Option<Filter>,
}
#[derive(Debug, Clone)]
struct ResolvedOverride {
path: PathBuf,
common_dir: PathBuf,
default_branch: Option<String>,
}
#[derive(Debug, Clone)]
struct ResolvedExclusion {
path: PathBuf,
common_dir: PathBuf,
excluded: bool,
}
trait ResolvedEntry {
fn path(&self) -> &Path;
fn common_dir(&self) -> &Path;
}
impl ResolvedEntry for ResolvedOverride {
fn path(&self) -> &Path {
&self.path
}
fn common_dir(&self) -> &Path {
&self.common_dir
}
}
impl ResolvedEntry for ResolvedExclusion {
fn path(&self) -> &Path {
&self.path
}
fn common_dir(&self) -> &Path {
&self.common_dir
}
}
fn resolve_entries(overrides: &[RepoOverride]) -> (Vec<ResolvedOverride>, Vec<ResolvedExclusion>) {
overrides
.iter()
.filter_map(|entry| {
let common_dir = git::common_dir_of(&entry.path).ok()?;
Some((
ResolvedOverride {
path: entry.path.clone(),
common_dir: common_dir.to_path_buf(),
default_branch: entry.default_branch.clone(),
},
ResolvedExclusion {
path: entry.path.clone(),
common_dir: common_dir.to_path_buf(),
excluded: entry.excluded,
},
))
})
.unzip()
}
fn find_entry<'a, T: ResolvedEntry>(
entries: &'a [T],
path: &Path,
common_dir: &Path,
) -> Option<&'a T> {
entries
.iter()
.find(|entry| entry.path() == path)
.or_else(|| {
entries
.iter()
.find(|entry| entry.common_dir() == common_dir)
})
}
fn excluded_by(exclusions: &[ResolvedExclusion], path: &Path, common_dir: &Path) -> bool {
find_entry(exclusions, path, common_dir).is_some_and(|entry| entry.excluded)
}
fn dispatches_kind(kind: Kind, show_submodules: bool) -> bool {
match kind {
Kind::Repo | Kind::Worktree => true,
Kind::Submodule => show_submodules,
}
}
#[derive(Debug, Clone)]
pub struct FetchSpec {
pub enabled: bool,
pub interval: Duration,
pub concurrency: usize,
}
#[derive(Debug, Clone, Copy)]
pub struct AutoUpdateSpec {
pub enabled: bool,
}
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct FetchFailures {
pub failed: Vec<(PathBuf, String)>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum AutoUpdateAttempt {
NotClean,
NoUpstream,
NotBehind,
NotFastForward,
Updated,
Failed(String),
}
#[derive(Debug, Clone)]
pub struct CoreSpec {
pub set: SetSpec,
pub overrides: Vec<RepoOverride>,
pub poll_interval: Duration,
pub status_stale_after: Duration,
pub generation_deadline: Duration,
pub show_submodules: bool,
pub fetch: FetchSpec,
pub auto_update: AutoUpdateSpec,
}
struct InFlight {
generation: u64,
cancel: Arc<AtomicBool>,
}
struct Table {
generation: u64,
discovered_at: Timestamp,
entities: Vec<EntityState>,
index: HashMap<EntityKey, usize>,
in_flight: HashMap<EntityKey, InFlight>,
generation_started_at: HashMap<u64, Instant>,
repos: HashMap<EntityKey, Arc<gix::ThreadSafeRepository>>,
poll_fingerprints: HashMap<EntityKey, poll::GitdirFingerprint>,
}
enum ClockControl {
Pause,
Resume,
Shutdown,
}
pub struct Core {
table: Arc<RwLock<Table>>,
overrides: Arc<Vec<ResolvedOverride>>,
exclusions: Arc<RwLock<Vec<ResolvedExclusion>>>,
set: SetSpec,
discovery_manual: Arc<AtomicBool>,
discovery_warn_after: Duration,
discovery_abandon_after: Arc<AtomicU64>,
show_submodules: Arc<AtomicBool>,
settle_gate: Arc<SettleGate>,
control: Sender<ClockControl>,
clock_thread: Option<JoinHandle<()>>,
discovery_warning: Arc<Mutex<Option<String>>>,
#[allow(dead_code)] default_branch_chain_reads: Arc<AtomicUsize>,
#[allow(dead_code)] patch_identity_reads: Arc<AtomicUsize>,
#[allow(dead_code)] patch_scan_bounds: Arc<Mutex<Vec<Option<gix::ObjectId>>>>,
action_running: Arc<AtomicBool>,
action_control: Arc<Mutex<Option<Arc<executor::RunControl>>>>,
#[allow(dead_code)] dispatch_log: Arc<Mutex<Vec<EntityKey>>>,
#[allow(dead_code)] phase_c_gates: Arc<Mutex<HashMap<EntityKey, PhaseCGateHandle>>>,
status_stale_after: Duration,
#[allow(dead_code)] poll_reprobed: Arc<Mutex<Vec<EntityKey>>>,
#[allow(dead_code)] poll_sweep_count: Arc<AtomicUsize>,
#[allow(dead_code)] fetch_cycle_count: Arc<AtomicUsize>,
network_default_branch: Arc<Mutex<HashMap<PathBuf, Arc<str>>>>,
fetch_failures: Arc<Mutex<FetchFailures>>,
turnstile: Arc<DispatchTurnstile>,
discovery_gate: Option<DiscoveryGate>,
}
#[derive(Default)]
struct PhaseCGate {
cheap_landed: bool,
may_proceed: bool,
finished: bool,
}
type PhaseCGateHandle = Arc<(Mutex<PhaseCGate>, Condvar)>;
impl Core {
pub fn start(spec: CoreSpec) -> Core {
Self::start_watched(spec).core
}
fn start_watched(spec: CoreSpec) -> StartForTest {
let interval = spec.poll_interval.max(Duration::from_nanos(1));
let ticks = crossbeam_channel::tick(interval);
let alive = Arc::new(AtomicBool::new(true));
let fetch_start = FetchStart {
enabled: spec.fetch.enabled,
concurrency: spec.fetch.concurrency.max(1),
ticks: if spec.fetch.enabled {
crossbeam_channel::tick(spec.fetch.interval.max(Duration::from_nanos(1)))
} else {
crossbeam_channel::never()
},
};
start_internal(
spec,
Duration::from_secs(1),
discovery::ABANDON_AFTER,
ticks,
fetch_start,
alive,
None,
)
}
#[cfg(any(test, feature = "test-util"))]
pub fn start_discovered(spec: CoreSpec) -> Core {
let mut started = Self::start_watched(spec);
if let Some(handle) = started.initial_discovery.take() {
handle
.join()
.expect("the first discovery thread should not panic");
}
started.core
}
pub fn refresh(&self, order: &[EntityKey]) -> Generation {
self.refresh_handles().dispatch(order)
}
pub fn refresh_all(&self) -> Generation {
self.refresh_handles().dispatch_over_everything()
}
pub fn rederive_default_branches(&self, keys: &[EntityKey]) -> Generation {
let generation = {
let mut table = self.table.write().unwrap();
table.generation += 1;
Generation::new(table.generation)
};
let dispatched: Vec<RederiveCandidate> = {
let mut table = self.table.write().unwrap();
let mut dispatched = Vec::new();
for key in keys {
let Some(&idx) = table.index.get(key) else {
continue;
};
table.entities[idx].default_branch.begin_probe();
let common_dir = Arc::clone(&table.entities[idx].common_dir);
let override_branch = find_entry(&self.overrides, key.path(), &common_dir)
.and_then(|entry| entry.default_branch.clone());
let repo = table.repos.get(key).cloned();
let kind = table.entities[idx].kind;
dispatched.push(RederiveCandidate {
key: key.clone(),
path: key.path().to_path_buf(),
common_dir,
repo,
override_branch,
kind,
});
}
dispatched
};
if dispatched.is_empty() {
return generation;
}
begin_probes_owed(&self.settle_gate, dispatched.len());
let table = Arc::clone(&self.table);
let settle_gate = Arc::clone(&self.settle_gate);
let network_default_branch = Arc::clone(&self.network_default_branch);
thread::spawn(move || {
let common_dirs: HashSet<Arc<Path>> = dispatched
.iter()
.map(|candidate| Arc::clone(&candidate.common_dir))
.collect();
probe_network_default_branches(&common_dirs, &network_default_branch);
let chain_cache: ChainFactsCache = Mutex::new(HashMap::new());
let chain_reads = AtomicUsize::new(0);
let never_cancelled = AtomicBool::new(false);
for candidate in dispatched {
let RederiveCandidate {
key,
path,
common_dir,
repo,
override_branch,
kind,
} = candidate;
let network_branch = network_branch_for(&network_default_branch, &common_dir);
let resolution = probe_default_branch_memoised(
&path,
repo.as_deref(),
&common_dir,
DefaultBranchHints {
override_branch: override_branch.as_deref(),
network_branch: network_branch.as_deref(),
},
kind,
&never_cancelled,
&ChainFactsMemo {
cache: &chain_cache,
reads: &chain_reads,
},
);
{
let mut table = table.write().unwrap();
if let (Some(&idx), Some(resolution)) = (table.index.get(&key), resolution) {
table.entities[idx].apply_default_branch_resolution(generation, resolution);
}
}
complete_one(&settle_gate);
}
});
generation
}
fn refresh_handles(&self) -> RefreshHandles {
RefreshHandles {
table: Arc::clone(&self.table),
overrides: Arc::clone(&self.overrides),
exclusions: Arc::clone(&self.exclusions),
set: self.set.clone(),
discovery_manual: Arc::clone(&self.discovery_manual),
discovery_warn_after: self.discovery_warn_after,
discovery_abandon_after: Arc::clone(&self.discovery_abandon_after),
discovery_warning: Arc::clone(&self.discovery_warning),
show_submodules: Arc::clone(&self.show_submodules),
settle_gate: Arc::clone(&self.settle_gate),
default_branch_chain_reads: Arc::clone(&self.default_branch_chain_reads),
patch_identity_reads: Arc::clone(&self.patch_identity_reads),
patch_scan_bounds: Arc::clone(&self.patch_scan_bounds),
dispatch_log: Arc::clone(&self.dispatch_log),
phase_c_gates: Arc::clone(&self.phase_c_gates),
network_default_branch: Arc::clone(&self.network_default_branch),
turnstile: Arc::clone(&self.turnstile),
discovery_gate: self.discovery_gate.clone(),
}
}
pub fn probe_now(&self, key: &EntityKey) -> EntityState {
let never_cancelled = Arc::new(AtomicBool::new(false));
let (cached_repo, common_dir_hint, probes_state, probes_base, kind) = {
let table = self.table.read().unwrap();
let repo = table.repos.get(key).cloned();
let common_dir = table
.index
.get(key)
.map(|&idx| Arc::clone(&table.entities[idx].common_dir));
let probes_state = table
.index
.get(key)
.map(|&idx| table.entities[idx].probes_state())
.unwrap_or(false);
let probes_base = table
.index
.get(key)
.map(|&idx| table.entities[idx].probes_base())
.unwrap_or(true);
let kind = table
.index
.get(key)
.map(|&idx| table.entities[idx].kind)
.unwrap_or(Kind::Repo);
(repo, common_dir, probes_state, probes_base, kind)
};
let common_dir_hint = common_dir_hint.unwrap_or_else(|| Arc::from(key.path().join(".git")));
let override_branch = find_entry(&self.overrides, key.path(), &common_dir_hint)
.and_then(|entry| entry.default_branch.clone());
let excluded = excluded_by(
&self.exclusions.read().unwrap(),
key.path(),
&common_dir_hint,
);
let branch_outcome =
probe_branch(key.path(), cached_repo.as_deref(), kind, &never_cancelled);
let sync_outcome = probe_sync(
key.path(),
cached_repo.as_deref(),
branch_outcome.as_ref().map(|(settled, ..)| settled),
kind,
&never_cancelled,
);
let default_branch_outcome = probe_default_branch(
key.path(),
cached_repo.as_deref(),
DefaultBranchHints {
override_branch: override_branch.as_deref(),
network_branch: network_branch_for(&self.network_default_branch, &common_dir_hint)
.as_deref(),
},
kind,
&never_cancelled,
);
let base_outcome = if probes_base {
probe_base(
key.path(),
cached_repo.as_deref(),
branch_outcome.as_ref().map(|(settled, ..)| settled),
default_branch_outcome.as_ref().map(|r| &r.settled),
&never_cancelled,
)
} else {
None
};
let state_outcome = if probes_state {
let patch_cache: PatchIdentityCache = Mutex::new(HashMap::new());
let patch_reads = AtomicUsize::new(0);
let patch_scan_bounds = Mutex::new(Vec::new());
let gate = BoundGate::new(1);
let mut report = GateReport::new(&gate);
let memo = PatchEquivalenceMemo {
cache: &patch_cache,
reads: &patch_reads,
scan_bounds: &patch_scan_bounds,
};
probe_worktree_state(
key.path(),
cached_repo.as_deref(),
default_branch_outcome.as_ref().map(|r| &r.settled),
&common_dir_hint,
&never_cancelled,
&memo,
&mut report,
)
} else {
None
};
let dirty_outcome =
probe_status(key.path(), cached_repo.as_deref(), kind, &never_cancelled);
let mut table = self.table.write().unwrap();
let generation = Generation::new(table.generation);
let idx = match table.index.get(key).copied() {
Some(idx) => idx,
None => {
let name = display_name(key.path());
table.entities.push(EntityState::new(
key.clone(),
name,
common_dir_hint,
Kind::Repo,
));
let idx = table.entities.len() - 1;
table.index.insert(key.clone(), idx);
idx
}
};
table.entities[idx].excluded = excluded;
if let Some((settled, in_progress, recent)) = branch_outcome {
table.entities[idx].apply_branch_probe(generation, settled, in_progress, recent);
}
if let Some(settled) = sync_outcome {
table.entities[idx].sync.settle(generation, settled);
}
if let Some(settled) = base_outcome {
table.entities[idx].base.settle(generation, settled);
}
if let Some(resolution) = default_branch_outcome {
table.entities[idx].apply_default_branch_resolution(generation, resolution);
}
if let Some(settled) = state_outcome {
table.entities[idx].state.settle(generation, settled);
}
if let Some(settled) = dirty_outcome {
table.entities[idx].dirty.settle(generation, settled);
}
table.entities[idx].clone()
}
pub fn snapshot(&self) -> Snapshot {
let table = self.table.read().unwrap();
let mut entities = table.entities.clone();
for entity in &mut entities {
entity.age_status_cells(self.status_stale_after);
}
Snapshot {
generation: Generation::new(table.generation),
discovered_at: table.discovered_at,
entities,
}
}
pub fn try_settle(&self, within: Duration) -> Result<Snapshot, Snapshot> {
let (lock, cvar) = &*self.settle_gate;
let guard = lock.lock().unwrap();
let (guard, timeout) = cvar
.wait_timeout_while(guard, within, |counts| !counts.is_settled())
.unwrap();
drop(guard);
let snapshot = self.snapshot();
if timeout.timed_out() {
Err(snapshot)
} else {
Ok(snapshot)
}
}
#[cfg(any(test, feature = "test-util"))]
pub fn settle(&self) -> Snapshot {
self.settle_within(liveness::BACKSTOP)
}
#[cfg(any(test, feature = "test-util"))]
fn settle_within(&self, deadline: Duration) -> Snapshot {
self.try_settle(deadline).unwrap_or_else(|_| {
let (probes, dispatches) = {
let counts = self.settle_gate.0.lock().unwrap();
(counts.probes, counts.dispatches)
};
liveness::expired(
deadline,
"everything this Core has in flight to land",
&format!("{probes} probe(s) and {dispatches} dispatch(es) still outstanding"),
)
})
}
pub fn delete_risk(&self, key: &EntityKey) -> Result<DeleteRisk, git::ProbeError> {
let repo = git::open_thread_safe(key.path())?.to_thread_local();
let dirty = git::dirty_counts(&repo, Arc::new(AtomicBool::new(false)))?;
let staged = git::staged_changes(&repo)?;
let (unpushed_commits, unpushed_branches) = git::unpushed(&repo)?;
let linked_worktrees = git::linked_worktrees(&repo)?;
Ok(DeleteRisk {
uncommitted: dirty.total() > 0 || staged,
unpushed_commits,
unpushed_branches,
linked_worktrees,
})
}
pub fn worktree_admin_dir(&self, key: &EntityKey) -> Result<PathBuf, git::ProbeError> {
let repo = git::open_thread_safe(key.path())?.to_thread_local();
Ok(git::worktree_admin_dir(&repo))
}
pub fn linked_worktree_paths(&self, key: &EntityKey) -> Result<Vec<PathBuf>, git::ProbeError> {
let repo = git::open_thread_safe(key.path())?.to_thread_local();
git::linked_worktree_paths(&repo)
}
pub fn ignored_directories_for_deletion(
&self,
path: &Path,
) -> Result<Vec<PathBuf>, git::ProbeError> {
let repo = git::open_thread_safe(path)?.to_thread_local();
git::ignored_directories_for_deletion(&repo)
}
pub fn attempt_auto_update(&self, key: &EntityKey) -> AutoUpdateAttempt {
match crate::auto_update::attempt(key.path()) {
crate::auto_update::Outcome::Ineligible(crate::auto_update::Ineligible::NotClean) => {
AutoUpdateAttempt::NotClean
}
crate::auto_update::Outcome::Ineligible(crate::auto_update::Ineligible::NoUpstream) => {
AutoUpdateAttempt::NoUpstream
}
crate::auto_update::Outcome::Ineligible(crate::auto_update::Ineligible::NotBehind) => {
AutoUpdateAttempt::NotBehind
}
crate::auto_update::Outcome::Ineligible(
crate::auto_update::Ineligible::NotFastForward,
) => AutoUpdateAttempt::NotFastForward,
crate::auto_update::Outcome::Updated { .. } => AutoUpdateAttempt::Updated,
crate::auto_update::Outcome::Failed(error) => AutoUpdateAttempt::Failed(error),
}
}
pub fn run_action_for_entity_blocking(
&self,
action: &ActionSpec,
key: &EntityKey,
) -> Option<ActionReceipt> {
let entity = {
let table = self.table.read().unwrap();
let idx = *table.index.get(key)?;
table.entities[idx].clone()
};
let control = executor::RunControl::new();
Some(run_action_for_entity(&entity, action, &control, &|_| {}))
}
pub fn management_handle(&self) -> ManagementHandle {
ManagementHandle {
table: Arc::clone(&self.table),
}
}
pub fn dismiss(&self, key: &EntityKey) {
let mut table = self.table.write().unwrap();
if let Some(idx) = table.index.remove(key) {
table.entities.remove(idx);
for position in table.index.values_mut() {
if *position > idx {
*position -= 1;
}
}
}
table.poll_fingerprints.remove(key);
if let Some(in_flight) = table.in_flight.remove(key) {
in_flight.cancel.store(true, Ordering::Release);
drop(table);
complete_one(&self.settle_gate);
}
}
fn partition_operable(&self, order: &[EntityKey]) -> (Vec<EntityState>, Vec<EntityState>) {
let table = self.table.read().unwrap();
order
.iter()
.filter_map(|key| table.index.get(key).map(|&idx| table.entities[idx].clone()))
.partition(|entity| !entity.excluded)
}
pub fn operable_count(&self, order: &[EntityKey]) -> usize {
self.partition_operable(order).0.len()
}
pub fn vanished_count(&self) -> usize {
self.table
.read()
.unwrap()
.entities
.iter()
.filter(|entity| entity.presence == Presence::Vanished)
.count()
}
pub fn applicability(&self, order: &[EntityKey], when: &Filter) -> Applicability {
when.applicability(self.partition_operable(order).0.iter())
}
pub fn action_running(&self) -> bool {
self.action_running.load(Ordering::Acquire)
}
pub fn refresh_running(&self) -> bool {
let (lock, _cvar) = &*self.settle_gate;
!lock.lock().unwrap().is_settled()
}
pub fn run_action(&self, action: ActionSpec, order: &[EntityKey]) -> bool {
if self
.action_running
.compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
.is_err()
{
return false;
}
cancel_in_flight(&self.table, &self.settle_gate);
let (operable, excluded) = self.partition_operable(order);
let write_skip_receipts = |entities: &[EntityState], skip: Skip| {
if entities.is_empty() {
return;
}
let finished_at = Timestamp::now();
let mut table = self.table.write().unwrap();
for entity in entities {
if let Some(&idx) = table.index.get(&entity.key) {
table.entities[idx].last_action = Some(ActionReceipt {
label: Arc::clone(&action.label),
steps: Arc::from(Vec::new()),
skip: Some(skip),
finished_at,
running: None,
});
}
}
};
write_skip_receipts(&excluded, Skip::Excluded);
let included = match &action.when {
Some(when) => {
let Partition {
applicable,
inapplicable,
unresolved,
} = when.partition(operable);
write_skip_receipts(&inapplicable, Skip::Inapplicable);
write_skip_receipts(&unresolved, Skip::Unresolved);
applicable
}
None => operable,
};
let table_handle = Arc::clone(&self.table);
let action_running = Arc::clone(&self.action_running);
let refresh_handles = self.refresh_handles();
let control = executor::RunControl::new();
*self.action_control.lock().unwrap() = Some(Arc::clone(&control));
let action_control = Arc::clone(&self.action_control);
let concurrency = action.concurrency.max(1) as usize;
thread::spawn(move || {
let pool = rayon::ThreadPoolBuilder::new()
.num_threads(concurrency)
.build()
.expect("build the Action fan-out's own dedicated pool");
let fan_out = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
pool.install(|| {
included.into_par_iter().for_each(|entity| {
let write_receipt = |receipt: ActionReceipt| {
let mut table = table_handle.write().unwrap();
if let Some(&idx) = table.index.get(&entity.key) {
table.entities[idx].last_action = Some(receipt);
}
};
let receipt =
run_action_for_entity(&entity, &action, &control, &write_receipt);
write_receipt(receipt);
});
});
}));
action_running.store(false, Ordering::Release);
*action_control.lock().unwrap() = None;
let Ok(()) = fan_out else {
return;
};
let all_keys: Vec<EntityKey> = table_handle
.read()
.unwrap()
.entities
.iter()
.map(|entity| entity.key.clone())
.collect();
refresh_handles.dispatch(&all_keys);
});
true
}
pub fn hold_action(&self) {
if let Some(control) = self.action_control.lock().unwrap().as_ref() {
control.hold();
}
}
pub fn continue_action(&self) {
if let Some(control) = self.action_control.lock().unwrap().as_ref() {
control.continue_run();
}
}
pub fn stop_action(&self) {
if let Some(control) = self.action_control.lock().unwrap().as_ref() {
control.cancel();
}
}
pub fn pause(&self) {
let _ = self.control.send(ClockControl::Pause);
}
pub fn resume(&self) {
let _ = self.control.send(ClockControl::Resume);
}
pub fn discovery_warning(&self) -> Option<String> {
self.discovery_warning.lock().unwrap().clone()
}
pub fn fetch_failures(&self) -> FetchFailures {
self.fetch_failures.lock().unwrap().clone()
}
pub fn set_show_submodules(&self, show_submodules: bool) {
self.show_submodules
.store(show_submodules, Ordering::Release);
}
pub fn record_own_work(&self, label: &str, results: &[(EntityKey, OwnWork, Duration)]) {
let label: Arc<str> = Arc::from(label);
let finished_at = Timestamp::now();
let mut table = self.table.write().unwrap();
for (key, work, elapsed) in results {
let Some(&idx) = table.index.get(key) else {
continue;
};
table.entities[idx].last_action = Some(ActionReceipt {
label: Arc::clone(&label),
steps: Arc::from(vec![StepResult {
label: Arc::clone(&label),
outcome: StepOutcome::OwnWork(work.clone()),
output: Arc::from(&b""[..]),
elapsed: *elapsed,
elision: None,
shell: false,
interactive: false,
}]),
skip: None,
finished_at,
running: None,
});
}
}
pub fn set_exclusions(&self, overrides: &[RepoOverride]) {
let (_, resolved) = resolve_entries(overrides);
{
let mut exclusions = self.exclusions.write().unwrap();
*exclusions = resolved.clone();
}
let mut table = self.table.write().unwrap();
for entity in &mut table.entities {
entity.excluded = excluded_by(&resolved, entity.key.path(), &entity.common_dir);
}
}
}
#[derive(Clone)]
pub struct ManagementHandle {
table: Arc<RwLock<Table>>,
}
impl ManagementHandle {
pub fn worktree_admin_dir(&self, key: &EntityKey) -> Result<PathBuf, git::ProbeError> {
let repo = git::open_thread_safe(key.path())?.to_thread_local();
Ok(git::worktree_admin_dir(&repo))
}
pub fn linked_worktree_paths(&self, key: &EntityKey) -> Result<Vec<PathBuf>, git::ProbeError> {
let repo = git::open_thread_safe(key.path())?.to_thread_local();
git::linked_worktree_paths(&repo)
}
pub fn ignored_directories_for_deletion(
&self,
path: &Path,
) -> Result<Vec<PathBuf>, git::ProbeError> {
let repo = git::open_thread_safe(path)?.to_thread_local();
git::ignored_directories_for_deletion(&repo)
}
pub fn attempt_auto_update(&self, key: &EntityKey) -> AutoUpdateAttempt {
match crate::auto_update::attempt(key.path()) {
crate::auto_update::Outcome::Ineligible(crate::auto_update::Ineligible::NotClean) => {
AutoUpdateAttempt::NotClean
}
crate::auto_update::Outcome::Ineligible(crate::auto_update::Ineligible::NoUpstream) => {
AutoUpdateAttempt::NoUpstream
}
crate::auto_update::Outcome::Ineligible(crate::auto_update::Ineligible::NotBehind) => {
AutoUpdateAttempt::NotBehind
}
crate::auto_update::Outcome::Ineligible(
crate::auto_update::Ineligible::NotFastForward,
) => AutoUpdateAttempt::NotFastForward,
crate::auto_update::Outcome::Updated { .. } => AutoUpdateAttempt::Updated,
crate::auto_update::Outcome::Failed(error) => AutoUpdateAttempt::Failed(error),
}
}
pub fn run_action_for_entity_blocking(
&self,
action: &ActionSpec,
key: &EntityKey,
) -> Option<ActionReceipt> {
let entity = {
let table = self.table.read().unwrap();
let idx = *table.index.get(key)?;
table.entities[idx].clone()
};
let control = executor::RunControl::new();
Some(run_action_for_entity(&entity, action, &control, &|_| {}))
}
}
#[derive(Clone)]
struct RefreshHandles {
table: Arc<RwLock<Table>>,
overrides: Arc<Vec<ResolvedOverride>>,
exclusions: Arc<RwLock<Vec<ResolvedExclusion>>>,
set: SetSpec,
discovery_manual: Arc<AtomicBool>,
discovery_warn_after: Duration,
discovery_abandon_after: Arc<AtomicU64>,
discovery_warning: Arc<Mutex<Option<String>>>,
show_submodules: Arc<AtomicBool>,
settle_gate: Arc<SettleGate>,
default_branch_chain_reads: Arc<AtomicUsize>,
patch_identity_reads: Arc<AtomicUsize>,
patch_scan_bounds: Arc<Mutex<Vec<Option<gix::ObjectId>>>>,
dispatch_log: Arc<Mutex<Vec<EntityKey>>>,
phase_c_gates: Arc<Mutex<HashMap<EntityKey, PhaseCGateHandle>>>,
network_default_branch: Arc<Mutex<HashMap<PathBuf, Arc<str>>>>,
turnstile: Arc<DispatchTurnstile>,
discovery_gate: Option<DiscoveryGate>,
}
#[derive(Default)]
struct DispatchTurnstile {
serving: Mutex<u64>,
ready: Condvar,
next: AtomicU64,
}
impl DispatchTurnstile {
fn reserve(&self) -> u64 {
self.next.fetch_add(1, Ordering::AcqRel)
}
fn take(&self, ticket: u64) -> DispatchTurn<'_> {
let serving = self.serving.lock().unwrap();
drop(
self.ready
.wait_while(serving, |serving| *serving != ticket)
.unwrap(),
);
DispatchTurn {
turnstile: self,
ticket,
}
}
}
struct DispatchTurn<'a> {
turnstile: &'a DispatchTurnstile,
ticket: u64,
}
impl Drop for DispatchTurn<'_> {
fn drop(&mut self) {
let mut serving = self.turnstile.serving.lock().unwrap();
*serving = self.ticket + 1;
self.turnstile.ready.notify_all();
}
}
impl RefreshHandles {
fn dispatch(&self, order: &[EntityKey]) -> Generation {
let (generation, ticket) = self.reserve_generation();
begin_dispatch(&self.settle_gate);
let handles = self.clone();
let order = order.to_vec();
thread::spawn(move || {
let _turn = handles.turnstile.take(ticket);
handles.run_generation(&order, generation);
finish_dispatch(&handles.settle_gate);
});
generation
}
fn dispatch_over_everything(&self) -> Generation {
let (generation, ticket) = self.reserve_generation();
begin_dispatch(&self.settle_gate);
let handles = self.clone();
thread::spawn(move || {
let _turn = handles.turnstile.take(ticket);
handles.rediscover();
let order: Vec<EntityKey> = handles
.table
.read()
.unwrap()
.entities
.iter()
.map(|entity| entity.key.clone())
.collect();
handles.dispatch_probes(&order, generation);
finish_dispatch(&handles.settle_gate);
});
generation
}
fn reserve_generation(&self) -> (Generation, u64) {
let mut table = self.table.write().unwrap();
table.generation += 1;
(Generation::new(table.generation), self.turnstile.reserve())
}
fn run_generation(&self, order: &[EntityKey], generation: Generation) {
self.rediscover();
self.dispatch_probes(order, generation);
}
fn rediscover(&self) {
if !self.discovery_manual.load(Ordering::Acquire) {
self.rerun_discovery();
}
}
fn dispatch_probes(&self, order: &[EntityKey], generation: Generation) {
self.default_branch_chain_reads.store(0, Ordering::Release);
self.patch_identity_reads.store(0, Ordering::Release);
self.patch_scan_bounds.lock().unwrap().clear();
self.dispatch_log.lock().unwrap().clear();
let generation_number = generation.value();
let mut table = self.table.write().unwrap();
table
.generation_started_at
.insert(generation_number, Instant::now());
let show_submodules = self.show_submodules.load(Ordering::Acquire);
let mut dispatched = Vec::new();
for key in order {
let Some(&idx) = table.index.get(key) else {
continue;
};
if !dispatches_kind(table.entities[idx].kind, show_submodules) {
continue;
}
if let Some(previous) = table.in_flight.remove(key) {
previous.cancel.store(true, Ordering::Release);
}
let cancel = Arc::new(AtomicBool::new(false));
table.in_flight.insert(
key.clone(),
InFlight {
generation: generation_number,
cancel: Arc::clone(&cancel),
},
);
begin_probes(&mut table.entities[idx]);
dispatched.push((key.clone(), cancel));
}
if dispatched.is_empty() {
return;
}
begin_probes_owed(&self.settle_gate, dispatched.len());
let repos: Vec<Option<Arc<gix::ThreadSafeRepository>>> = dispatched
.iter()
.map(|(key, _)| table.repos.get(key).cloned())
.collect();
let override_branches: Vec<Option<String>> = dispatched
.iter()
.map(|(key, _)| {
let idx = table.index[key];
let common_dir = &table.entities[idx].common_dir;
find_entry(&self.overrides, key.path(), common_dir)
.and_then(|entry| entry.default_branch.clone())
})
.collect();
let network_branches: Vec<Option<Arc<str>>> = dispatched
.iter()
.map(|(key, _)| {
let idx = table.index[key];
let common_dir = &table.entities[idx].common_dir;
network_branch_for(&self.network_default_branch, common_dir)
})
.collect();
let common_dirs: Vec<Arc<Path>> = dispatched
.iter()
.map(|(key, _)| Arc::clone(&table.entities[table.index[key]].common_dir))
.collect();
let probes_state: Vec<bool> = dispatched
.iter()
.map(|(key, _)| table.entities[table.index[key]].probes_state())
.collect();
let probes_base: Vec<bool> = dispatched
.iter()
.map(|(key, _)| table.entities[table.index[key]].probes_base())
.collect();
let kinds: Vec<Kind> = dispatched
.iter()
.map(|(key, _)| table.entities[table.index[key]].kind)
.collect();
drop(table);
let chain_cache: Arc<ChainFactsCache> = Arc::new(Mutex::new(HashMap::new()));
let patch_cache: Arc<PatchIdentityCache> = Arc::new(Mutex::new(HashMap::new()));
let bound_gates: Arc<HashMap<Arc<Path>, BoundGate>> = Arc::new({
let mut counts: HashMap<Arc<Path>, usize> = HashMap::new();
for (common_dir, probes_state) in common_dirs.iter().zip(&probes_state) {
if *probes_state {
*counts.entry(Arc::clone(common_dir)).or_insert(0) += 1;
}
}
counts
.into_iter()
.map(|(dir, count)| (dir, BoundGate::new(count)))
.collect()
});
for (
(
(
(((((key, cancel), repo), override_branch), network_branch), common_dir),
probes_state,
),
probes_base,
),
kind,
) in dispatched
.into_iter()
.zip(repos)
.zip(override_branches)
.zip(network_branches)
.zip(common_dirs)
.zip(probes_state)
.zip(probes_base)
.zip(kinds)
{
self.dispatch_log.lock().unwrap().push(key.clone());
let path = key.path().to_path_buf();
let table_handle = Arc::clone(&self.table);
let settle_gate = Arc::clone(&self.settle_gate);
let chain_cache = Arc::clone(&chain_cache);
let chain_reads = Arc::clone(&self.default_branch_chain_reads);
let patch_cache = Arc::clone(&patch_cache);
let patch_reads = Arc::clone(&self.patch_identity_reads);
let patch_scan_bounds = Arc::clone(&self.patch_scan_bounds);
let bound_gates = Arc::clone(&bound_gates);
let held_gate = self.phase_c_gates.lock().unwrap().get(&key).cloned();
rayon::spawn(move || {
let branch_outcome = probe_branch(&path, repo.as_deref(), kind, &cancel);
let sync_outcome = probe_sync(
&path,
repo.as_deref(),
branch_outcome.as_ref().map(|(settled, ..)| settled),
kind,
&cancel,
);
let default_branch_outcome = probe_default_branch_memoised(
&path,
repo.as_deref(),
&common_dir,
DefaultBranchHints {
override_branch: override_branch.as_deref(),
network_branch: network_branch.as_deref(),
},
kind,
&cancel,
&ChainFactsMemo {
cache: &chain_cache,
reads: &chain_reads,
},
);
let base_outcome = if probes_base {
probe_base(
&path,
repo.as_deref(),
branch_outcome.as_ref().map(|(settled, ..)| settled),
default_branch_outcome.as_ref().map(|r| &r.settled),
&cancel,
)
} else {
None
};
apply_cheap_probe_outcomes(
&table_handle,
&key,
generation,
CheapProbeOutcomes {
branch: branch_outcome,
sync: sync_outcome,
base: base_outcome,
default_branch: default_branch_outcome.clone(),
},
);
if let Some(gate) = &held_gate {
let (lock, cvar) = &**gate;
let mut state = lock.lock().unwrap();
state.cheap_landed = true;
cvar.notify_all();
state = cvar.wait_while(state, |state| !state.may_proceed).unwrap();
drop(state);
}
let state_outcome = if probes_state {
let gate = bound_gates
.get(&common_dir)
.expect("every probes_state entity's common dir has a gate sized for it");
let mut report = GateReport::new(gate);
let memo = PatchEquivalenceMemo {
cache: &patch_cache,
reads: &patch_reads,
scan_bounds: &patch_scan_bounds,
};
probe_worktree_state(
&path,
repo.as_deref(),
default_branch_outcome.as_ref().map(|r| &r.settled),
&common_dir,
&cancel,
&memo,
&mut report,
)
} else {
None
};
let dirty_outcome = probe_status(&path, repo.as_deref(), kind, &cancel);
apply_probe_outcome(
&table_handle,
&settle_gate,
&key,
generation,
ProbeOutcomes {
state: state_outcome,
dirty: dirty_outcome,
},
);
if let Some(gate) = &held_gate {
let (lock, cvar) = &**gate;
let mut state = lock.lock().unwrap();
state.finished = true;
cvar.notify_all();
}
});
}
}
fn rerun_discovery(&self) {
let repos_cache: HashMap<EntityKey, Arc<gix::ThreadSafeRepository>> =
self.table.read().unwrap().repos.clone();
wait_for_discovery_gate(self.discovery_gate.as_ref());
let (watch, _watcher) = spawn_discovery_watcher(
self.set.roots.clone(),
&self.discovery_warning,
self.discovery_warn_after,
);
let discovery = run_watched_discovery(
&watch,
&self.set,
&self.discovery_warning,
Duration::from_nanos(self.discovery_abandon_after.load(Ordering::Acquire)),
);
if discovery.abandoned {
self.discovery_manual.store(true, Ordering::Release);
}
let (discovered, gitmodules_failures) =
discovery::resolve_with_cache(&self.set, &discovery.entities, &repos_cache);
let exclusions = self.exclusions.read().unwrap().clone();
let mut table = self.table.write().unwrap();
table.discovered_at = Timestamp::now();
let cancelled = merge_discovery(&mut table, &exclusions, discovered, gitmodules_failures);
drop(table);
if cancelled > 0 {
complete_many(&self.settle_gate, cancelled);
}
}
}
impl Drop for Core {
fn drop(&mut self) {
cancel_in_flight(&self.table, &self.settle_gate);
let _ = self.control.send(ClockControl::Shutdown);
if let Some(handle) = self.clock_thread.take() {
let _ = handle.join();
}
}
}
pub(crate) struct StartForTest {
pub core: Core,
#[allow(dead_code)] pub clock_alive: Arc<AtomicBool>,
#[allow(dead_code)] pub discovery_watcher: JoinHandle<()>,
#[allow(dead_code)] pub initial_discovery: Option<JoinHandle<()>>,
}
#[cfg(test)]
impl StartForTest {
fn discovered(mut self) -> Self {
if let Some(handle) = self.initial_discovery.take() {
handle
.join()
.expect("the first discovery thread should not panic");
}
self
}
}
impl Core {
#[cfg(any(test, feature = "test-util"))]
pub fn begin_untracked_probe_for_test(&self, key: &EntityKey) -> Arc<AtomicBool> {
let mut table = self.table.write().unwrap();
table.generation += 1;
let generation_number = table.generation;
table
.generation_started_at
.insert(generation_number, Instant::now());
if let Some(&idx) = table.index.get(key) {
begin_probes(&mut table.entities[idx]);
}
let cancel = Arc::new(AtomicBool::new(false));
table.in_flight.insert(
key.clone(),
InFlight {
generation: generation_number,
cancel: Arc::clone(&cancel),
},
);
begin_probes_owed(&self.settle_gate, 1);
cancel
}
}
#[cfg(test)]
pub(crate) struct SharedGeneration {
pub generation: Generation,
pub cancels: HashMap<EntityKey, Arc<AtomicBool>>,
}
#[cfg(test)]
impl Core {
pub(crate) fn cached_repo_handle_for_test(
&self,
key: &EntityKey,
) -> Option<Arc<gix::ThreadSafeRepository>> {
self.table.read().unwrap().repos.get(key).cloned()
}
pub(crate) fn default_branch_chain_reads_for_test(&self) -> usize {
self.default_branch_chain_reads.load(Ordering::Acquire)
}
pub(crate) fn patch_identity_reads_for_test(&self) -> usize {
self.patch_identity_reads.load(Ordering::Acquire)
}
pub(crate) fn patch_scan_bounds_for_test(&self) -> Vec<Option<gix::ObjectId>> {
self.patch_scan_bounds.lock().unwrap().clone()
}
pub(crate) fn dispatch_log_for_test(&self) -> Vec<EntityKey> {
self.dispatch_log.lock().unwrap().clone()
}
pub(crate) fn poll_once_for_test(&self) {
run_poll_sweep(
&self.table,
&self.overrides,
&self.show_submodules,
&self.poll_reprobed,
&self.poll_sweep_count,
&self.network_default_branch,
);
}
pub(crate) fn poll_reprobed_for_test(&self) -> Vec<EntityKey> {
self.poll_reprobed.lock().unwrap().clone()
}
pub(crate) fn poll_sweep_count_for_test(&self) -> usize {
self.poll_sweep_count.load(Ordering::Acquire)
}
pub(crate) fn hold_phase_c_for_test(&self, key: &EntityKey) {
self.phase_c_gates.lock().unwrap().insert(
key.clone(),
Arc::new((Mutex::new(PhaseCGate::default()), Condvar::new())),
);
}
pub(crate) fn wait_phase_c_landed_for_test(&self, key: &EntityKey) {
let gate = self
.phase_c_gates
.lock()
.unwrap()
.get(key)
.cloned()
.expect("hold_phase_c_for_test must be called before waiting on its gate");
let (lock, cvar) = &*gate;
let guard = lock.lock().unwrap();
drop(cvar.wait_while(guard, |state| !state.cheap_landed).unwrap());
}
pub(crate) fn release_phase_c_for_test(&self, key: &EntityKey) {
let gate = self
.phase_c_gates
.lock()
.unwrap()
.get(key)
.cloned()
.expect("hold_phase_c_for_test must be called before releasing its gate");
let (lock, cvar) = &*gate;
let mut state = lock.lock().unwrap();
state.may_proceed = true;
cvar.notify_all();
}
pub(crate) fn wait_phase_c_finished_for_test(&self, key: &EntityKey) {
let gate = self
.phase_c_gates
.lock()
.unwrap()
.get(key)
.cloned()
.expect("hold_phase_c_for_test must be called before waiting on its gate");
let (lock, cvar) = &*gate;
let guard = lock.lock().unwrap();
drop(cvar.wait_while(guard, |state| !state.finished).unwrap());
}
pub(crate) fn wait_dispatched_for_test(&self) {
let (lock, cvar) = &*self.settle_gate;
let guard = lock.lock().unwrap();
drop(
cvar.wait_while(guard, |counts| counts.dispatches > 0)
.unwrap(),
);
}
pub(crate) fn settle_gate_count_for_test(&self) -> usize {
self.settle_gate.0.lock().unwrap().probes
}
pub(crate) fn start_for_test(
spec: CoreSpec,
warn_after: Duration,
ticks: Receiver<Instant>,
) -> StartForTest {
Self::start_for_test_with_discovery_abandon(
spec,
warn_after,
discovery::ABANDON_AFTER,
ticks,
)
}
pub(crate) fn start_for_test_with_discovery_abandon(
spec: CoreSpec,
warn_after: Duration,
discovery_abandon_after: Duration,
ticks: Receiver<Instant>,
) -> StartForTest {
Self::start_for_test_gated(spec, warn_after, discovery_abandon_after, ticks, None)
}
pub(crate) fn start_for_test_gated(
spec: CoreSpec,
warn_after: Duration,
discovery_abandon_after: Duration,
ticks: Receiver<Instant>,
discovery_gate: Option<DiscoveryGate>,
) -> StartForTest {
let alive = Arc::new(AtomicBool::new(true));
start_internal(
spec,
warn_after,
discovery_abandon_after,
ticks,
FetchStart {
enabled: false,
concurrency: 1,
ticks: crossbeam_channel::never(),
},
alive,
discovery_gate,
)
}
pub(crate) fn start_for_test_with_fetch(
spec: CoreSpec,
warn_after: Duration,
ticks: Receiver<Instant>,
fetch_ticks: Receiver<Instant>,
) -> StartForTest {
let alive = Arc::new(AtomicBool::new(true));
let fetch_start = FetchStart {
enabled: spec.fetch.enabled,
concurrency: spec.fetch.concurrency.max(1),
ticks: fetch_ticks,
};
start_internal(
spec,
warn_after,
discovery::ABANDON_AFTER,
ticks,
fetch_start,
alive,
None,
)
}
pub(crate) fn fetch_cycle_count_for_test(&self) -> usize {
self.fetch_cycle_count.load(Ordering::Acquire)
}
#[cfg(test)]
pub(crate) fn set_discovery_abandon_after_for_test(&self, after: Duration) {
self.discovery_abandon_after
.store(after.as_nanos() as u64, Ordering::Release);
}
pub(crate) fn discovery_manual_for_test(&self) -> bool {
self.discovery_manual.load(Ordering::Acquire)
}
pub(crate) fn begin_shared_generation_for_test(&self, keys: &[EntityKey]) -> SharedGeneration {
let mut table = self.table.write().unwrap();
table.generation += 1;
let generation_number = table.generation;
table
.generation_started_at
.insert(generation_number, Instant::now());
let mut cancels = HashMap::new();
for key in keys {
if let Some(&idx) = table.index.get(key) {
table.entities[idx].branch.begin_probe();
}
let cancel = Arc::new(AtomicBool::new(false));
table.in_flight.insert(
key.clone(),
InFlight {
generation: generation_number,
cancel: Arc::clone(&cancel),
},
);
cancels.insert(key.clone(), cancel);
}
SharedGeneration {
generation: Generation::new(generation_number),
cancels,
}
}
pub(crate) fn apply_probe_result_for_test(
&self,
key: &EntityKey,
generation: Generation,
settled: Settled<Head>,
) {
apply_cheap_probe_outcomes(
&self.table,
key,
generation,
CheapProbeOutcomes {
branch: Some((settled, None, Vec::new())),
sync: None,
base: None,
default_branch: None,
},
);
}
pub(crate) fn set_last_action_for_test(
&self,
key: &EntityKey,
receipt: crate::entity::ActionReceipt,
) {
let mut table = self.table.write().unwrap();
if let Some(&idx) = table.index.get(key) {
table.entities[idx].last_action = Some(receipt);
}
}
}
fn run_action_for_entity(
entity: &EntityState,
action: &ActionSpec,
control: &Arc<executor::RunControl>,
report: &dyn Fn(ActionReceipt),
) -> ActionReceipt {
let base_env = environment::environment(entity, action.name.as_deref());
let mut failed = false;
let mut cancelled = false;
let mut results: Vec<StepResult> = Vec::with_capacity(action.steps.len());
for step in &action.steps {
if failed || cancelled || control.is_cancelled() {
cancelled = cancelled || control.is_cancelled();
results.push(StepResult {
label: Arc::from(step.argv.join(" ")),
outcome: if cancelled {
StepOutcome::Cancelled
} else {
StepOutcome::NotRun
},
output: Arc::from(&b""[..]),
elapsed: Duration::ZERO,
elision: None,
shell: step.shell,
interactive: step.interactive,
});
continue;
}
let label: Arc<str> = Arc::from(step.argv.join(" "));
report(ActionReceipt {
label: Arc::clone(&action.label),
steps: Arc::from(results.clone()),
skip: None,
finished_at: Timestamp::now(),
running: Some(RunningStep {
label: Arc::clone(&label),
started_at: Timestamp::now(),
shell: step.shell,
interactive: step.interactive,
}),
});
let mut env = base_env.clone();
env.extend(
step.env
.iter()
.map(|(name, value)| (name.clone(), Some(value.clone()))),
);
let mut result = executor::run_step(
&step.argv,
step.shell,
step.interactive,
entity.key.path(),
&env,
control,
);
if control.is_cancelled() {
result.outcome = StepOutcome::Cancelled;
cancelled = true;
} else {
failed = result.outcome.is_failure();
}
results.push(result);
}
ActionReceipt {
label: Arc::clone(&action.label),
steps: Arc::from(results),
skip: None,
finished_at: Timestamp::now(),
running: None,
}
}
type DiscoveryGate = Arc<(Mutex<bool>, Condvar)>;
fn wait_for_discovery_gate(gate: Option<&DiscoveryGate>) {
let Some(gate) = gate else {
return;
};
let (lock, cvar) = &**gate;
let open = lock.lock().unwrap();
drop(cvar.wait_while(open, |open| !*open).unwrap());
}
#[cfg(test)]
fn set_discovery_gate(gate: &DiscoveryGate, open: bool) {
let (lock, cvar) = &**gate;
*lock.lock().unwrap() = open;
cvar.notify_all();
}
struct DiscoveryWatch {
progress: Arc<AtomicUsize>,
finished: Arc<AtomicBool>,
}
fn spawn_discovery_watcher(
roots: Vec<PathBuf>,
discovery_warning: &Arc<Mutex<Option<String>>>,
warn_after: Duration,
) -> (DiscoveryWatch, JoinHandle<()>) {
let progress = Arc::new(AtomicUsize::new(0));
let finished = Arc::new(AtomicBool::new(false));
let watcher = thread::spawn({
let progress = Arc::clone(&progress);
let finished = Arc::clone(&finished);
let warning_slot = Arc::clone(discovery_warning);
move || {
if let Some(message) = watch_for_slow_discovery(progress, finished, roots, warn_after) {
*warning_slot.lock().unwrap() = Some(message);
}
}
});
(DiscoveryWatch { progress, finished }, watcher)
}
fn run_watched_discovery(
watch: &DiscoveryWatch,
set: &SetSpec,
discovery_warning: &Arc<Mutex<Option<String>>>,
abandon_after: Duration,
) -> discovery::Discovery {
let discovery =
discovery::discover_watched_with_deadline(set, Arc::clone(&watch.progress), abandon_after);
watch.finished.store(true, Ordering::Release);
if discovery.abandoned {
*discovery_warning.lock().unwrap() =
Some(abandoned_discovery_message(discovery.directories_visited));
}
discovery
}
fn start_internal(
spec: CoreSpec,
warn_after: Duration,
discovery_abandon_after: Duration,
ticks: Receiver<Instant>,
fetch_start: FetchStart,
alive: Arc<AtomicBool>,
discovery_gate: Option<DiscoveryGate>,
) -> StartForTest {
let FetchStart {
enabled: fetch_enabled,
concurrency: fetch_concurrency,
ticks: fetch_ticks,
} = fetch_start;
let discovery_warning = Arc::new(Mutex::new(None));
let discovery_manual = Arc::new(AtomicBool::new(false));
let (overrides, resolved_exclusions) = resolve_entries(&spec.overrides);
let overrides = Arc::new(overrides);
let exclusions = Arc::new(RwLock::new(resolved_exclusions));
let show_submodules = Arc::new(AtomicBool::new(spec.show_submodules));
let table = Arc::new(RwLock::new(Table {
generation: 0,
discovered_at: Timestamp::now(),
entities: Vec::new(),
index: HashMap::new(),
in_flight: HashMap::new(),
generation_started_at: HashMap::new(),
repos: HashMap::new(),
poll_fingerprints: HashMap::new(),
}));
let settle_gate: Arc<SettleGate> =
Arc::new((Mutex::new(SettleCounts::default()), Condvar::new()));
let poll_reprobed = Arc::new(Mutex::new(Vec::new()));
let poll_sweep_count = Arc::new(AtomicUsize::new(0));
let network_default_branch = Arc::new(Mutex::new(HashMap::new()));
let (control, control_rx) = crossbeam_channel::unbounded();
let poll_handles = PollHandles {
overrides: Arc::clone(&overrides),
show_submodules: Arc::clone(&show_submodules),
poll_reprobed: Arc::clone(&poll_reprobed),
poll_sweep_count: Arc::clone(&poll_sweep_count),
network_default_branch: Arc::clone(&network_default_branch),
};
let discovery_abandon_after_atomic =
Arc::new(AtomicU64::new(discovery_abandon_after.as_nanos() as u64));
let default_branch_chain_reads = Arc::new(AtomicUsize::new(0));
let patch_identity_reads = Arc::new(AtomicUsize::new(0));
let patch_scan_bounds = Arc::new(Mutex::new(Vec::new()));
let dispatch_log = Arc::new(Mutex::new(Vec::new()));
let phase_c_gates = Arc::new(Mutex::new(HashMap::new()));
let fetch_cycle_count = Arc::new(AtomicUsize::new(0));
let fetch_failures = Arc::new(Mutex::new(FetchFailures::default()));
let turnstile = Arc::new(DispatchTurnstile::default());
let fetch_refresh_handles = RefreshHandles {
table: Arc::clone(&table),
overrides: Arc::clone(&overrides),
exclusions: Arc::clone(&exclusions),
set: spec.set.clone(),
discovery_manual: Arc::clone(&discovery_manual),
discovery_warn_after: warn_after,
discovery_abandon_after: Arc::clone(&discovery_abandon_after_atomic),
discovery_warning: Arc::clone(&discovery_warning),
show_submodules: Arc::clone(&show_submodules),
settle_gate: Arc::clone(&settle_gate),
default_branch_chain_reads: Arc::clone(&default_branch_chain_reads),
patch_identity_reads: Arc::clone(&patch_identity_reads),
patch_scan_bounds: Arc::clone(&patch_scan_bounds),
dispatch_log: Arc::clone(&dispatch_log),
phase_c_gates: Arc::clone(&phase_c_gates),
network_default_branch: Arc::clone(&network_default_branch),
turnstile: Arc::clone(&turnstile),
discovery_gate: discovery_gate.clone(),
};
let auto_update_enabled = spec.auto_update.enabled;
let fetch_schedule = FetchSchedule {
concurrency: fetch_concurrency,
ticks: fetch_ticks,
refresh: fetch_refresh_handles.clone(),
cycle_count: Arc::clone(&fetch_cycle_count),
failures: Arc::clone(&fetch_failures),
auto_update_enabled,
};
let clock_thread = spawn_clock_thread(
Arc::clone(&table),
poll_handles,
fetch_schedule,
Arc::clone(&settle_gate),
spec.generation_deadline,
ClockChannels {
control: control_rx,
ticks,
alive: Arc::clone(&alive),
},
);
let (startup_generation, startup_ticket) = fetch_refresh_handles.reserve_generation();
begin_dispatch(&settle_gate);
let (watch, discovery_watcher) =
spawn_discovery_watcher(spec.set.roots.clone(), &discovery_warning, warn_after);
let initial_discovery = thread::spawn({
let set = spec.set.clone();
let discovery_warning = Arc::clone(&discovery_warning);
let discovery_manual = Arc::clone(&discovery_manual);
let exclusions = Arc::clone(&exclusions);
let table = Arc::clone(&table);
let settle_gate = Arc::clone(&settle_gate);
let fetch_refresh_handles = fetch_refresh_handles.clone();
let fetch_cycle_count = Arc::clone(&fetch_cycle_count);
let fetch_failures = Arc::clone(&fetch_failures);
let discovery_gate = discovery_gate.clone();
move || {
let turn = fetch_refresh_handles.turnstile.take(startup_ticket);
wait_for_discovery_gate(discovery_gate.as_ref());
let discovery =
run_watched_discovery(&watch, &set, &discovery_warning, discovery_abandon_after);
if discovery.abandoned {
discovery_manual.store(true, Ordering::Release);
}
let (discovered, gitmodules_failures) = discovery::resolve(&set, &discovery.entities);
let resolved_exclusions = exclusions.read().unwrap().clone();
let order: Vec<EntityKey> = {
let mut table = table.write().unwrap();
merge_discovery(
&mut table,
&resolved_exclusions,
discovered,
gitmodules_failures,
);
table.discovered_at = Timestamp::now();
table
.entities
.iter()
.map(|entity| entity.key.clone())
.collect()
};
fetch_refresh_handles.dispatch_probes(&order, startup_generation);
finish_dispatch(&settle_gate);
drop(turn);
if fetch_enabled {
let table = Arc::clone(&table);
thread::spawn(move || {
run_fetch_cycle(
&table,
fetch_concurrency,
&fetch_refresh_handles,
&fetch_cycle_count,
&fetch_failures,
auto_update_enabled,
);
});
}
}
});
StartForTest {
core: Core {
table,
overrides,
exclusions,
set: spec.set,
discovery_manual,
discovery_warn_after: warn_after,
discovery_abandon_after: discovery_abandon_after_atomic,
show_submodules,
settle_gate,
control,
clock_thread: Some(clock_thread),
discovery_warning,
default_branch_chain_reads,
patch_identity_reads,
patch_scan_bounds,
action_running: Arc::new(AtomicBool::new(false)),
action_control: Arc::new(Mutex::new(None)),
dispatch_log,
phase_c_gates,
status_stale_after: spec.status_stale_after,
poll_reprobed,
poll_sweep_count,
fetch_cycle_count,
network_default_branch,
fetch_failures,
turnstile,
discovery_gate,
},
clock_alive: alive,
discovery_watcher,
initial_discovery: Some(initial_discovery),
}
}
struct PollHandles {
overrides: Arc<Vec<ResolvedOverride>>,
show_submodules: Arc<AtomicBool>,
poll_reprobed: Arc<Mutex<Vec<EntityKey>>>,
poll_sweep_count: Arc<AtomicUsize>,
network_default_branch: Arc<Mutex<HashMap<PathBuf, Arc<str>>>>,
}
struct FetchStart {
enabled: bool,
concurrency: usize,
ticks: Receiver<Instant>,
}
struct FetchSchedule {
concurrency: usize,
ticks: Receiver<Instant>,
refresh: RefreshHandles,
cycle_count: Arc<AtomicUsize>,
failures: Arc<Mutex<FetchFailures>>,
auto_update_enabled: bool,
}
struct ClockChannels {
control: Receiver<ClockControl>,
ticks: Receiver<Instant>,
alive: Arc<AtomicBool>,
}
fn spawn_clock_thread(
table: Arc<RwLock<Table>>,
poll: PollHandles,
fetch: FetchSchedule,
settle_gate: Arc<SettleGate>,
generation_deadline: Duration,
channels: ClockChannels,
) -> JoinHandle<()> {
let ClockChannels {
control,
ticks,
alive,
} = channels;
thread::spawn(move || {
let mut paused = false;
loop {
select! {
recv(control) -> message => match message {
Ok(ClockControl::Pause) => {
paused = true;
cancel_in_flight(&table, &settle_gate);
}
Ok(ClockControl::Resume) => paused = false,
Ok(ClockControl::Shutdown) | Err(_) => break,
},
recv(ticks) -> tick => {
if tick.is_err() {
break;
}
if !paused {
run_poll_sweep(
&table,
&poll.overrides,
&poll.show_submodules,
&poll.poll_reprobed,
&poll.poll_sweep_count,
&poll.network_default_branch,
);
sweep_deadline(&table, &settle_gate, generation_deadline);
}
}
recv(fetch.ticks) -> tick => {
if tick.is_err() {
break;
}
if !paused {
run_fetch_cycle(
&table,
fetch.concurrency,
&fetch.refresh,
&fetch.cycle_count,
&fetch.failures,
fetch.auto_update_enabled,
);
}
}
}
}
alive.store(false, Ordering::Release);
})
}
fn run_fetch_cycle(
table: &Arc<RwLock<Table>>,
concurrency: usize,
refresh: &RefreshHandles,
cycle_count: &Arc<AtomicUsize>,
failures: &Arc<Mutex<FetchFailures>>,
auto_update_enabled: bool,
) {
cycle_count.fetch_add(1, Ordering::Release);
let common_dirs = distinct_fetchable_common_dirs(table);
let failed: Mutex<Vec<(PathBuf, String)>> = Mutex::new(Vec::new());
crate::fetch::run_bounded(common_dirs, concurrency.max(1), |common_dir| {
let cancel = AtomicBool::new(false);
match crate::fetch::fetch_and_prune(&common_dir, &cancel) {
Ok(outcome) => {
if let Some(crate::fetch::AdvertisedDefaultBranch::Branch(name)) =
outcome.advertised_default_branch
{
refresh
.network_default_branch
.lock()
.unwrap()
.insert(common_dir.clone(), Arc::from(name));
}
}
Err(error) => {
failed
.lock()
.unwrap()
.push((common_dir.clone(), error.to_string()));
}
}
});
*failures.lock().unwrap() = FetchFailures {
failed: failed.into_inner().unwrap(),
};
if auto_update_enabled {
for repo_path in repos_eligible_for_auto_update_attempt(table) {
let _ = crate::auto_update::attempt(&repo_path);
}
}
let all_keys: Vec<EntityKey> = table
.read()
.unwrap()
.entities
.iter()
.map(|entity| entity.key.clone())
.collect();
refresh.dispatch(&all_keys);
}
fn repos_eligible_for_auto_update_attempt(table: &Arc<RwLock<Table>>) -> Vec<PathBuf> {
table
.read()
.unwrap()
.entities
.iter()
.filter(|entity| entity.kind == Kind::Repo && !entity.excluded)
.map(|entity| entity.key.path().to_path_buf())
.collect()
}
fn distinct_fetchable_common_dirs(table: &Arc<RwLock<Table>>) -> Vec<PathBuf> {
let table = table.read().unwrap();
let mut seen: HashMap<PathBuf, bool> = HashMap::new();
for entity in &table.entities {
let common_dir = entity.common_dir.to_path_buf();
let operable = seen.entry(common_dir).or_insert(false);
*operable = *operable || !entity.excluded;
}
seen.into_iter()
.filter(|(_, operable)| *operable)
.map(|(common_dir, _)| common_dir)
.collect()
}
fn probe_network_default_branches(
common_dirs: &HashSet<Arc<Path>>,
network_default_branch: &Mutex<HashMap<PathBuf, Arc<str>>>,
) {
for common_dir in common_dirs {
if let Ok(Some(crate::fetch::AdvertisedDefaultBranch::Branch(name))) =
crate::fetch::probe_remote_head(common_dir)
{
network_default_branch
.lock()
.unwrap()
.insert(common_dir.to_path_buf(), Arc::from(name));
}
}
}
struct RederiveCandidate {
key: EntityKey,
path: PathBuf,
common_dir: Arc<Path>,
repo: Option<Arc<gix::ThreadSafeRepository>>,
override_branch: Option<String>,
kind: Kind,
}
struct PollCandidate {
key: EntityKey,
path: PathBuf,
common_dir: Arc<Path>,
kind: Kind,
cached_repo: Option<Arc<gix::ThreadSafeRepository>>,
probes_base: bool,
}
fn run_poll_sweep(
table: &Arc<RwLock<Table>>,
overrides: &Arc<Vec<ResolvedOverride>>,
show_submodules: &Arc<AtomicBool>,
poll_reprobed: &Arc<Mutex<Vec<EntityKey>>>,
poll_sweep_count: &Arc<AtomicUsize>,
network_default_branch: &Mutex<HashMap<PathBuf, Arc<str>>>,
) {
poll_sweep_count.fetch_add(1, Ordering::Release);
poll_reprobed.lock().unwrap().clear();
let show_submodules = show_submodules.load(Ordering::Acquire);
let candidates: Vec<PollCandidate> = {
let table = table.read().unwrap();
table
.entities
.iter()
.filter(|entity| dispatches_kind(entity.kind, show_submodules))
.map(|entity| PollCandidate {
key: entity.key.clone(),
path: entity.key.path().to_path_buf(),
common_dir: Arc::clone(&entity.common_dir),
kind: entity.kind,
cached_repo: table.repos.get(&entity.key).cloned(),
probes_base: entity.probes_base(),
})
.collect()
};
for candidate in candidates {
let opened;
let repo = match candidate.cached_repo.as_deref() {
Some(repo) => Some(repo),
None => match git::open_thread_safe(&candidate.path) {
Ok(repo) => {
opened = repo;
Some(&opened)
}
Err(_) => None,
},
};
let gitdir = repo
.map(|repo| repo.git_dir().to_path_buf())
.unwrap_or_else(|| candidate.common_dir.to_path_buf());
let current = poll::fingerprint(&gitdir);
let moved = {
let mut table = table.write().unwrap();
let previous = table
.poll_fingerprints
.insert(candidate.key.clone(), current);
previous.is_some_and(|previous| poll::moved(&previous, ¤t))
};
if !moved {
continue;
}
{
let mut table = table.write().unwrap();
if let Some(&idx) = table.index.get(&candidate.key) {
table.entities[idx].force_stale_status_cells();
}
}
let override_branch = find_entry(overrides, &candidate.path, &candidate.common_dir)
.and_then(|entry| entry.default_branch.clone());
let never_cancelled = AtomicBool::new(false);
let chain_cache: ChainFactsCache = Mutex::new(HashMap::new());
let chain_reads = AtomicUsize::new(0);
let branch_outcome = probe_branch(&candidate.path, repo, candidate.kind, &never_cancelled);
let sync_outcome = probe_sync(
&candidate.path,
repo,
branch_outcome.as_ref().map(|(settled, ..)| settled),
candidate.kind,
&never_cancelled,
);
let default_branch_outcome = probe_default_branch_memoised(
&candidate.path,
repo,
&candidate.common_dir,
DefaultBranchHints {
override_branch: override_branch.as_deref(),
network_branch: network_branch_for(network_default_branch, &candidate.common_dir)
.as_deref(),
},
candidate.kind,
&never_cancelled,
&ChainFactsMemo {
cache: &chain_cache,
reads: &chain_reads,
},
);
let base_outcome = if candidate.probes_base {
probe_base(
&candidate.path,
repo,
branch_outcome.as_ref().map(|(settled, ..)| settled),
default_branch_outcome.as_ref().map(|r| &r.settled),
&never_cancelled,
)
} else {
None
};
let generation = {
let mut table = table.write().unwrap();
table.generation += 1;
Generation::new(table.generation)
};
apply_cheap_probe_outcomes(
table,
&candidate.key,
generation,
CheapProbeOutcomes {
branch: branch_outcome,
sync: sync_outcome,
base: base_outcome,
default_branch: default_branch_outcome,
},
);
poll_reprobed.lock().unwrap().push(candidate.key);
}
}
fn cancel_in_flight(table: &Arc<RwLock<Table>>, settle_gate: &Arc<SettleGate>) {
let mut table = table.write().unwrap();
let cancelled = table.in_flight.len();
for in_flight in table.in_flight.values() {
in_flight.cancel.store(true, Ordering::Release);
}
table.in_flight.clear();
table.generation_started_at.clear();
drop(table);
if cancelled > 0 {
complete_many(settle_gate, cancelled);
}
}
trait TimeoutableCell {
fn is_in_flight(&self) -> bool;
fn time_out(&mut self, generation: Generation);
}
impl<T> TimeoutableCell for Cell<T> {
fn is_in_flight(&self) -> bool {
Cell::is_in_flight(self)
}
fn time_out(&mut self, generation: Generation) {
self.settle(generation, Settled::Unknown(Unknown::TimedOut));
}
}
fn sweep_deadline(table: &Arc<RwLock<Table>>, settle_gate: &Arc<SettleGate>, deadline: Duration) {
let mut table = table.write().unwrap();
let now = Instant::now();
let mut timed_out = Vec::new();
for (key, in_flight) in table.in_flight.iter() {
let started = table
.generation_started_at
.get(&in_flight.generation)
.copied()
.unwrap_or(now);
if now.duration_since(started) >= deadline {
timed_out.push((key.clone(), Generation::new(in_flight.generation)));
}
}
for (key, generation) in &timed_out {
if let Some(&idx) = table.index.get(key) {
let EntityState {
key: _,
name: _,
common_dir: _,
kind: _,
branch,
sync,
base,
dirty,
state,
default_branch,
diagnostics: _,
last_action: _,
presence: _,
excluded: _,
in_progress_operation: _,
recent_commits: _,
} = &mut table.entities[idx];
let cells: [&mut dyn TimeoutableCell; 6] =
[branch, sync, base, dirty, state, default_branch];
for cell in cells {
if cell.is_in_flight() {
cell.time_out(*generation);
}
}
}
table.in_flight.remove(key);
}
let live_generations: std::collections::HashSet<u64> =
table.in_flight.values().map(|f| f.generation).collect();
table
.generation_started_at
.retain(|generation, _| live_generations.contains(generation));
drop(table);
if !timed_out.is_empty() {
complete_many(settle_gate, timed_out.len());
}
}
fn begin_probes(entity: &mut EntityState) {
let probes_state = entity.probes_state();
let EntityState {
key: _,
name: _,
common_dir: _,
kind: _,
branch,
sync: _,
base: _,
dirty,
state,
default_branch,
diagnostics: _,
last_action: _,
presence: _,
excluded: _,
in_progress_operation: _,
recent_commits: _,
} = entity;
branch.begin_probe();
default_branch.begin_probe();
dirty.begin_probe();
if probes_state {
state.begin_probe();
}
}
type SettleGate = (Mutex<SettleCounts>, Condvar);
#[derive(Default)]
struct SettleCounts {
probes: usize,
dispatches: usize,
}
impl SettleCounts {
fn is_settled(&self) -> bool {
let SettleCounts { probes, dispatches } = self;
*probes == 0 && *dispatches == 0
}
}
fn begin_dispatch(settle_gate: &SettleGate) {
let (lock, _cvar) = settle_gate;
lock.lock().unwrap().dispatches += 1;
}
fn finish_dispatch(settle_gate: &SettleGate) {
let (lock, cvar) = settle_gate;
let mut counts = lock.lock().unwrap();
counts.dispatches = counts.dispatches.saturating_sub(1);
drop(counts);
cvar.notify_all();
}
fn begin_probes_owed(settle_gate: &SettleGate, owed: usize) {
let (lock, _cvar) = settle_gate;
lock.lock().unwrap().probes += owed;
}
fn complete_one(settle_gate: &SettleGate) {
complete_many(settle_gate, 1);
}
fn complete_many(settle_gate: &SettleGate, finished: usize) {
let (lock, cvar) = settle_gate;
let mut counts = lock.lock().unwrap();
counts.probes = counts.probes.saturating_sub(finished);
if counts.is_settled() {
cvar.notify_all();
}
}
const RECENT_COMMITS_LIMIT: usize = 5;
fn submodule_open_failure<T>(kind: Kind, error: git::ProbeError) -> Settled<T> {
match kind {
Kind::Repo | Kind::Worktree => Settled::Failed(error),
Kind::Submodule => Settled::Unknown(Unknown::SubmoduleUninitialized),
}
}
fn probe_branch(
path: &Path,
repo: Option<&gix::ThreadSafeRepository>,
kind: Kind,
cancel: &AtomicBool,
) -> Option<(
Settled<Head>,
Option<git::InProgressOperation>,
Vec<git::RecentCommit>,
)> {
if cancel.load(Ordering::Acquire) {
return None;
}
let opened;
let repo = match repo {
Some(repo) => repo,
None => match git::open_thread_safe(path) {
Ok(repo) => {
opened = repo;
&opened
}
Err(error) => return Some((submodule_open_failure(kind, error), None, Vec::new())),
},
};
let local = repo.to_thread_local();
let settled = match git::head_shape(&local) {
Ok(head) => Settled::Known {
value: head,
at: Timestamp::now(),
stale: false,
},
Err(error) => Settled::Failed(error),
};
let in_progress = git::in_progress_operation(&local);
let recent = git::recent_commits(&local, RECENT_COMMITS_LIMIT);
Some((settled, in_progress, recent))
}
fn probe_sync(
path: &Path,
repo: Option<&gix::ThreadSafeRepository>,
branch_settled: Option<&Settled<Head>>,
kind: Kind,
cancel: &AtomicBool,
) -> Option<Settled<SyncState>> {
if cancel.load(Ordering::Acquire) {
return None;
}
let head = match branch_settled? {
Settled::Known {
value,
at: _,
stale: _,
} => Some(value),
Settled::Failed(error) => return Some(Settled::Failed(error.clone())),
Settled::Unknown(_) | Settled::NotApplicable => None,
};
let opened;
let repo = match repo {
Some(repo) => repo,
None => match git::open_thread_safe(path) {
Ok(repo) => {
opened = repo;
&opened
}
Err(error) => return Some(submodule_open_failure(kind, error)),
},
};
let local = repo.to_thread_local();
let settled = match git::resolve_sync(&local, head) {
Ok(value) => Settled::Known {
value,
at: Timestamp::now(),
stale: false,
},
Err(error) => Settled::Failed(error),
};
Some(settled)
}
fn probe_base(
path: &Path,
repo: Option<&gix::ThreadSafeRepository>,
branch_settled: Option<&Settled<Head>>,
default_branch_settled: Option<&Settled<DefaultBranch>>,
cancel: &AtomicBool,
) -> Option<Settled<u32>> {
if cancel.load(Ordering::Acquire) {
return None;
}
let head = match branch_settled? {
Settled::Known {
value,
at: _,
stale: _,
} => value,
Settled::Failed(error) => return Some(Settled::Failed(error.clone())),
Settled::Unknown(_) | Settled::NotApplicable => return None,
};
let default_branch_settled = default_branch_settled?;
let opened;
let repo = match repo {
Some(repo) => repo,
None => match git::open_thread_safe(path) {
Ok(repo) => {
opened = repo;
&opened
}
Err(error) => return Some(Settled::Failed(error)),
},
};
let local = repo.to_thread_local();
Some(base::probe(&local, head, default_branch_settled))
}
fn probe_status(
path: &Path,
repo: Option<&gix::ThreadSafeRepository>,
kind: Kind,
cancel: &Arc<AtomicBool>,
) -> Option<Settled<DirtyCounts>> {
if cancel.load(Ordering::Acquire) {
return None;
}
let opened;
let repo = match repo {
Some(repo) => repo,
None => match git::open_thread_safe(path) {
Ok(repo) => {
opened = repo;
&opened
}
Err(error) => return Some(submodule_open_failure(kind, error)),
},
};
let local = repo.to_thread_local();
classify_status_result(git::dirty_counts(&local, Arc::clone(cancel)), cancel)
}
fn classify_status_result(
result: Result<DirtyCounts, git::ProbeError>,
cancel: &AtomicBool,
) -> Option<Settled<DirtyCounts>> {
match result {
Ok(_) if cancel.load(Ordering::Acquire) => None,
Ok(value) => Some(Settled::Known {
value,
at: Timestamp::now(),
stale: false,
}),
Err(_) if cancel.load(Ordering::Acquire) => None,
Err(error) => Some(Settled::Failed(error)),
}
}
struct DefaultBranchHints<'a> {
override_branch: Option<&'a str>,
network_branch: Option<&'a str>,
}
fn network_branch_for(
network_default_branch: &Mutex<HashMap<PathBuf, Arc<str>>>,
common_dir: &Path,
) -> Option<Arc<str>> {
network_default_branch
.lock()
.unwrap()
.get(common_dir)
.cloned()
}
fn supersede_with_network(
mut resolution: default_branch::Resolution,
network_branch: Option<&str>,
) -> default_branch::Resolution {
if let Some(name) = network_branch {
resolution.settled = Settled::Known {
value: DefaultBranch::new(name.into()),
at: Timestamp::now(),
stale: false,
};
}
resolution
}
fn probe_default_branch(
path: &Path,
repo: Option<&gix::ThreadSafeRepository>,
hints: DefaultBranchHints<'_>,
kind: Kind,
cancel: &AtomicBool,
) -> Option<default_branch::Resolution> {
if cancel.load(Ordering::Acquire) {
return None;
}
let opened;
let repo = match repo {
Some(repo) => repo,
None => match git::open_thread_safe(path) {
Ok(repo) => {
opened = repo;
&opened
}
Err(error) => {
return Some(match kind {
Kind::Repo | Kind::Worktree => default_branch::Resolution::failed(error),
Kind::Submodule => default_branch::Resolution::submodule_uninitialized(),
});
}
},
};
Some(supersede_with_network(
default_branch::resolve(&repo.to_thread_local(), hints.override_branch),
hints.network_branch,
))
}
struct BoundGate {
state: Mutex<BoundGateState>,
condvar: Condvar,
bound: OnceLock<Option<gix::ObjectId>>,
}
struct BoundGateState {
remaining: usize,
candidates: Vec<gix::ObjectId>,
}
impl BoundGate {
fn new(remaining: usize) -> Self {
Self {
state: Mutex::new(BoundGateState {
remaining,
candidates: Vec::new(),
}),
condvar: Condvar::new(),
bound: OnceLock::new(),
}
}
fn report(&self, candidate: Option<gix::ObjectId>) {
let mut state = self.state.lock().unwrap();
if let Some(candidate) = candidate {
state.candidates.push(candidate);
}
state.remaining -= 1;
if state.remaining == 0 {
self.condvar.notify_all();
}
}
fn deepest(&self, repo: &gix::Repository) -> Option<gix::ObjectId> {
let mut state = self.state.lock().unwrap();
while state.remaining != 0 {
state = self.condvar.wait(state).unwrap();
}
let candidates = std::mem::take(&mut state.candidates);
*self
.bound
.get_or_init(|| deepest_merge_base(repo, &candidates))
}
}
fn deepest_merge_base(
repo: &gix::Repository,
candidates: &[gix::ObjectId],
) -> Option<gix::ObjectId> {
let mut candidates = candidates.iter().copied();
let mut deepest = candidates.next()?;
for candidate in candidates {
deepest = git::checked_merge_base(repo, deepest, candidate)
.ok()
.flatten()
.unwrap_or(deepest);
}
Some(deepest)
}
struct GateReport<'a> {
gate: &'a BoundGate,
reported: bool,
}
impl<'a> GateReport<'a> {
fn new(gate: &'a BoundGate) -> Self {
Self {
gate,
reported: false,
}
}
fn report_now(&mut self, candidate: Option<gix::ObjectId>) {
self.gate.report(candidate);
self.reported = true;
}
}
impl Drop for GateReport<'_> {
fn drop(&mut self) {
if !self.reported {
self.gate.report(None);
}
}
}
struct PatchEquivalenceMemo<'a> {
cache: &'a PatchIdentityCache,
reads: &'a AtomicUsize,
scan_bounds: &'a Mutex<Vec<Option<gix::ObjectId>>>,
}
fn probe_worktree_state(
path: &Path,
repo: Option<&gix::ThreadSafeRepository>,
default_branch_settled: Option<&Settled<DefaultBranch>>,
common_dir: &Arc<Path>,
cancel: &AtomicBool,
memo: &PatchEquivalenceMemo<'_>,
report: &mut GateReport<'_>,
) -> Option<Settled<WorktreeState>> {
if cancel.load(Ordering::Acquire) {
return None;
}
let default_branch_settled = default_branch_settled?;
let opened;
let repo = match repo {
Some(repo) => repo,
None => match git::open_thread_safe(path) {
Ok(repo) => {
opened = repo;
&opened
}
Err(error) => return Some(Settled::Failed(error)),
},
};
let local = repo.to_thread_local();
match landing::probe(&local, default_branch_settled) {
landing::Outcome::Settle(settled) => Some(settled),
landing::Outcome::Outstanding(outstanding) => {
probe_patch_equivalence(&local, &outstanding, common_dir, cancel, memo, report)
}
}
}
fn probe_patch_equivalence(
repo: &gix::Repository,
outstanding: &landing::Outstanding,
common_dir: &Arc<Path>,
cancel: &AtomicBool,
memo: &PatchEquivalenceMemo<'_>,
report: &mut GateReport<'_>,
) -> Option<Settled<WorktreeState>> {
if cancel.load(Ordering::Acquire) {
return None;
}
let landing::Outstanding {
entity_tip,
default_tip,
merge_base,
} = *outstanding;
let Some(merge_base) = merge_base else {
report.report_now(None);
return Some(patch_equivalence::probe(
repo,
entity_tip,
None,
&patch_equivalence::PatchIdentitySet::new(),
));
};
report.report_now(Some(merge_base));
let bound = report.gate.deepest(repo);
let shared = match patch_identities_for(memo.cache, common_dir, memo.reads, || {
memo.scan_bounds.lock().unwrap().push(bound);
patch_equivalence::scan_default_branch(repo, default_tip, bound)
}) {
Ok(shared) => shared,
Err(error) => return Some(Settled::Failed(error)),
};
Some(patch_equivalence::probe(
repo,
entity_tip,
Some(merge_base),
&shared,
))
}
type PatchIdentityCache = Mutex<
HashMap<Arc<Path>, Arc<OnceLock<Result<patch_equivalence::PatchIdentitySet, git::ProbeError>>>>,
>;
fn patch_identities_for(
cache: &PatchIdentityCache,
common_dir: &Arc<Path>,
reads: &AtomicUsize,
compute: impl FnOnce() -> Result<patch_equivalence::PatchIdentitySet, git::ProbeError>,
) -> Result<patch_equivalence::PatchIdentitySet, git::ProbeError> {
let cell = {
let mut cache = cache.lock().unwrap();
Arc::clone(
cache
.entry(Arc::clone(common_dir))
.or_insert_with(|| Arc::new(OnceLock::new())),
)
};
cell.get_or_init(|| {
reads.fetch_add(1, Ordering::Relaxed);
compute()
})
.clone()
}
type ChainFactsCache = Mutex<HashMap<Arc<Path>, Arc<OnceLock<default_branch::ChainFacts>>>>;
fn chain_facts_for(
cache: &ChainFactsCache,
common_dir: &Arc<Path>,
reads: &AtomicUsize,
compute: impl FnOnce() -> default_branch::ChainFacts,
) -> default_branch::ChainFacts {
let cell = {
let mut cache = cache.lock().unwrap();
Arc::clone(
cache
.entry(Arc::clone(common_dir))
.or_insert_with(|| Arc::new(OnceLock::new())),
)
};
cell.get_or_init(|| {
reads.fetch_add(1, Ordering::Relaxed);
compute()
})
.clone()
}
struct ChainFactsMemo<'a> {
cache: &'a ChainFactsCache,
reads: &'a AtomicUsize,
}
fn probe_default_branch_memoised(
path: &Path,
repo: Option<&gix::ThreadSafeRepository>,
common_dir: &Arc<Path>,
hints: DefaultBranchHints<'_>,
kind: Kind,
cancel: &AtomicBool,
memo: &ChainFactsMemo<'_>,
) -> Option<default_branch::Resolution> {
if cancel.load(Ordering::Acquire) {
return None;
}
let opened;
let repo = match repo {
Some(repo) => repo,
None => match git::open_thread_safe(path) {
Ok(repo) => {
opened = repo;
&opened
}
Err(error) => {
return Some(match kind {
Kind::Repo | Kind::Worktree => default_branch::Resolution::failed(error),
Kind::Submodule => default_branch::Resolution::submodule_uninitialized(),
});
}
},
};
let local = repo.to_thread_local();
let facts = chain_facts_for(memo.cache, common_dir, memo.reads, || {
default_branch::ChainFacts::resolve(&local)
});
Some(supersede_with_network(
default_branch::resolve_with_facts(&facts, hints.override_branch),
hints.network_branch,
))
}
struct CheapProbeOutcomes {
branch: Option<(
Settled<Head>,
Option<git::InProgressOperation>,
Vec<git::RecentCommit>,
)>,
sync: Option<Settled<SyncState>>,
base: Option<Settled<u32>>,
default_branch: Option<default_branch::Resolution>,
}
fn apply_cheap_probe_outcomes(
table: &Arc<RwLock<Table>>,
key: &EntityKey,
generation: Generation,
outcomes: CheapProbeOutcomes,
) {
let CheapProbeOutcomes {
branch: branch_outcome,
sync: sync_outcome,
base: base_outcome,
default_branch: default_branch_outcome,
} = outcomes;
let mut table = table.write().unwrap();
if let Some(&idx) = table.index.get(key) {
if let Some((settled, in_progress, recent)) = branch_outcome {
table.entities[idx].apply_branch_probe(generation, settled, in_progress, recent);
}
if let Some(settled) = sync_outcome {
table.entities[idx].sync.settle(generation, settled);
}
if let Some(settled) = base_outcome {
table.entities[idx].base.settle(generation, settled);
}
if let Some(resolution) = default_branch_outcome {
table.entities[idx].apply_default_branch_resolution(generation, resolution);
}
}
}
struct ProbeOutcomes {
state: Option<Settled<WorktreeState>>,
dirty: Option<Settled<DirtyCounts>>,
}
fn apply_probe_outcome(
table: &Arc<RwLock<Table>>,
settle_gate: &Arc<SettleGate>,
key: &EntityKey,
generation: Generation,
outcomes: ProbeOutcomes,
) {
let ProbeOutcomes {
state: state_outcome,
dirty: dirty_outcome,
} = outcomes;
let mut table = table.write().unwrap();
if let Some(&idx) = table.index.get(key) {
if let Some(settled) = state_outcome {
table.entities[idx].state.settle(generation, settled);
}
if let Some(settled) = dirty_outcome {
table.entities[idx].dirty.settle(generation, settled);
}
}
if table
.in_flight
.get(key)
.is_some_and(|in_flight| in_flight.generation == generation.value())
{
table.in_flight.remove(key);
}
drop(table);
complete_one(settle_gate);
}
fn merge_discovery(
table: &mut Table,
exclusions: &[ResolvedExclusion],
discovered: Vec<discovery::DiscoveredEntity>,
gitmodules_failures: Vec<(EntityKey, String)>,
) -> usize {
let mut found: HashSet<EntityKey> = HashSet::with_capacity(discovered.len());
for discovered in discovered {
found.insert(discovered.key.clone());
match table.index.get(&discovered.key).copied() {
Some(idx) => {
table.entities[idx].presence = Presence::Present;
if let Some(repo) = discovered.repo {
table.repos.insert(discovered.key.clone(), repo);
}
}
None => {
let name = discovered
.display_name_override
.clone()
.unwrap_or_else(|| display_name(discovered.key.path()));
let mut entity = EntityState::new(
discovered.key.clone(),
name,
Arc::clone(&discovered.common_dir),
discovered.kind,
);
entity.excluded =
excluded_by(exclusions, discovered.key.path(), &discovered.common_dir);
if let Some(repo) = discovered.repo {
table.repos.insert(discovered.key.clone(), repo);
}
let idx = table.entities.len();
table.index.insert(discovered.key, idx);
table.entities.push(entity);
}
}
}
let now_failing: HashMap<EntityKey, String> = gitmodules_failures.into_iter().collect();
for key in &found {
if let Some(&idx) = table.index.get(key) {
table.entities[idx].diagnostics.gitmodules_failed = now_failing
.get(key)
.map(|message| Arc::from(message.as_str()));
}
}
let missing: Vec<EntityKey> = table
.index
.keys()
.filter(|key| !found.contains(*key))
.cloned()
.collect();
let mut cancelled = 0usize;
for key in missing {
if let Some(&idx) = table.index.get(&key) {
table.entities[idx].mark_vanished();
}
if let Some(in_flight) = table.in_flight.remove(&key) {
in_flight.cancel.store(true, Ordering::Release);
cancelled += 1;
}
}
cancelled
}
fn display_name(path: &Path) -> Arc<str> {
Arc::from(
path.file_name()
.and_then(|name| name.to_str())
.unwrap_or("?"),
)
}
fn watch_for_slow_discovery(
progress: Arc<AtomicUsize>,
finished: Arc<AtomicBool>,
roots: Vec<PathBuf>,
warn_after: Duration,
) -> Option<String> {
thread::sleep(warn_after);
if finished.load(Ordering::Acquire) {
return None;
}
Some(still_walking_message(
progress.load(Ordering::Acquire),
&roots,
))
}
fn still_walking_message(directories_visited: usize, roots: &[PathBuf]) -> String {
let roots = roots
.iter()
.map(|root| root.display().to_string())
.collect::<Vec<_>>()
.join(", ");
format!("discovery: still walking, {directories_visited} directories reached under {roots}")
}
fn abandoned_discovery_message(directories_visited: usize) -> String {
format!("discovery: stopped at {directories_visited} directories")
}
#[allow(dead_code)] pub(crate) fn run_while_not_cancelled(
cancel: &AtomicBool,
mut step: impl FnMut() -> bool,
) -> usize {
let mut ran = 0;
while !cancel.load(Ordering::Acquire) {
if !step() {
break;
}
ran += 1;
}
ran
}
#[cfg(test)]
mod tests {
use std::fs;
use std::process::Command;
use super::*;
use crate::entity::{AheadBehind, DefaultBranchStopped, WorktreeState};
use crate::liveness::{BACKSTOP, FIXTURE_LIFETIME, wait_for};
use crate::snapshot::{RowSummary, summary};
use crate::test_support::{git, head_sha, loose_object_count};
fn init_repo_with_a_commit(path: &Path) {
fs::create_dir_all(path).expect("create repo dir");
gix::init(path).expect("init repo");
let status = Command::new("git")
.arg("-C")
.arg(path)
.args(["-c", "user.email=test@example.com", "-c", "user.name=Test"])
.args(["commit", "--allow-empty", "-m", "first"])
.status()
.expect("run git commit");
assert!(status.success());
}
fn commit_a_change(path: &Path, message: &str) {
let gitdir = gitdir_of(path);
let before = poll::fingerprint(&gitdir);
std::fs::write(path.join(format!("{message}.txt")), message.as_bytes())
.expect("write a file to commit");
let added = Command::new("git")
.arg("-C")
.arg(path)
.args(["add", "-A"])
.status()
.expect("run git add");
assert!(added.success());
commit(path, message, &["-m", message]);
assert!(
poll::moved(&before, &poll::fingerprint(&gitdir)),
"committing in {} moved none of the polled paths under {}, so this fixture cannot \
show the poll anything",
path.display(),
gitdir.display()
);
}
fn gitdir_of(work_dir: &Path) -> PathBuf {
let output = Command::new("git")
.arg("-C")
.arg(work_dir)
.args(["rev-parse", "--absolute-git-dir"])
.output()
.expect("run git rev-parse");
assert!(
output.status.success(),
"resolve the gitdir of {}",
work_dir.display()
);
PathBuf::from(
std::str::from_utf8(&output.stdout)
.expect("a utf-8 gitdir path")
.trim(),
)
}
fn commit(path: &Path, message: &str, args: &[&str]) {
let status = Command::new("git")
.arg("-C")
.arg(path)
.args(["-c", "user.email=test@example.com", "-c", "user.name=Test"])
.arg("commit")
.args(args)
.status()
.unwrap_or_else(|error| panic!("run git commit {message}: {error}"));
assert!(status.success());
}
fn fetch_spec_for_test() -> FetchSpec {
FetchSpec {
enabled: false,
interval: Duration::from_secs(3600),
concurrency: 4,
}
}
fn auto_update_spec_for_test() -> AutoUpdateSpec {
AutoUpdateSpec { enabled: false }
}
fn spec(roots: Vec<PathBuf>) -> CoreSpec {
CoreSpec {
set: SetSpec {
name: "test".to_string(),
roots,
include: Vec::new(),
exclude: Vec::new(),
},
overrides: Vec::new(),
poll_interval: Duration::from_secs(3600),
status_stale_after: Duration::from_secs(3600),
generation_deadline: Duration::from_secs(3600),
show_submodules: false,
fetch: fetch_spec_for_test(),
auto_update: auto_update_spec_for_test(),
}
}
#[test]
fn core_spec_carries_no_scoping_field_scope_is_never_a_dial() {
let CoreSpec {
set: _,
overrides: _,
poll_interval: _,
status_stale_after: _,
generation_deadline: _,
show_submodules: _,
fetch: _,
auto_update: _,
} = spec(Vec::new());
}
fn root_of(dir: &tempfile::TempDir) -> PathBuf {
dir.path().canonicalize().expect("canonicalize temp dir")
}
fn settle_launch(core: &Core) -> Snapshot {
let launched = core.settle();
assert_eq!(
core.settle_gate_count_for_test(),
0,
"launch's own Generation never settled, so nothing after this is starting from \
the point it claims to"
);
launched
}
fn started_and_settled(spec: CoreSpec) -> (Core, Snapshot) {
let core = Core::start_discovered(spec);
let launched = settle_launch(&core);
(core, launched)
}
fn backdate_polled_entries(work_dir: &Path) {
let gitdir = gitdir_of(work_dir);
let past = std::time::SystemTime::now() - Duration::from_secs(10);
let mut touched = 0;
for name in poll::POLLED_GITDIR_ENTRIES {
let path = gitdir.join(name);
if path.exists() {
set_mtime_to(&path, past);
touched += 1;
}
}
assert!(
touched > 0,
"backdated nothing under {}; the gitdir holds none of the polled entries and the \
baseline this sets up would not be older than what follows",
gitdir.display()
);
}
fn set_mtime_to(path: &Path, at: std::time::SystemTime) {
use std::os::unix::ffi::OsStrExt;
let secs = at
.duration_since(std::time::SystemTime::UNIX_EPOCH)
.expect("a time after the epoch")
.as_secs() as libc::time_t;
let times = [
libc::timespec {
tv_sec: secs,
tv_nsec: 0,
},
libc::timespec {
tv_sec: secs,
tv_nsec: 0,
},
];
let c_path =
std::ffi::CString::new(path.as_os_str().as_bytes()).expect("a path with no NUL");
let rc = unsafe { libc::utimensat(libc::AT_FDCWD, c_path.as_ptr(), times.as_ptr(), 0) };
assert_eq!(
rc,
0,
"set mtime on {}: {}",
path.display(),
std::io::Error::last_os_error()
);
}
fn step(argv: &[&str]) -> Step {
Step {
argv: argv.iter().map(|s| s.to_string()).collect(),
shell: false,
interactive: false,
env: Vec::new(),
}
}
fn shell_step(command: &str) -> Step {
Step {
argv: vec![command.to_string()],
shell: true,
interactive: false,
env: Vec::new(),
}
}
fn interactive_shell_step(command: &str) -> Step {
Step {
argv: vec![command.to_string()],
shell: true,
interactive: true,
env: Vec::new(),
}
}
fn action(label: &str, steps: Vec<Step>) -> ActionSpec {
ActionSpec {
label: Arc::from(label),
name: Some(Arc::from(label)),
steps,
concurrency: 4,
when: None,
}
}
fn action_with_when(label: &str, steps: Vec<Step>, when: &str) -> ActionSpec {
ActionSpec {
when: Some(Filter::parse(when)),
..action(label, steps)
}
}
#[test]
fn refresh_and_settle_populate_real_cells_without_the_caller_spawning_a_thread() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let core = Core::start_discovered(spec(vec![root]));
let keys: Vec<EntityKey> = core
.snapshot()
.entities
.iter()
.map(|entity| entity.key.clone())
.collect();
assert_eq!(keys.len(), 1);
core.refresh(&keys);
let settled = core.settle();
let entity = &settled.entities[0];
match entity.branch.settled() {
Some(Settled::Known {
value: Head::Branch { .. },
at: _,
stale: _,
}) => {}
other => panic!("expected an attached branch, got {other:?}"),
}
}
fn spec_refresh_md() -> String {
let manifest_dir = Path::new(env!("CARGO_MANIFEST_DIR"));
std::fs::read_to_string(manifest_dir.join("../../docs/spec/refresh.md"))
.expect("read docs/spec/refresh.md")
}
fn spec_first_frame_budgets_ms(spec: &str) -> (u64, u64) {
let anchor = "rows with names on screen within ";
let after = spec
.split(anchor)
.nth(1)
.expect("the first-frame budget sentence is present");
let mut parts = after.splitn(2, "ms, every cheap column filled within ");
let names: u64 = parts
.next()
.expect("a names-on-screen budget")
.parse()
.expect("the names-on-screen budget is an integer");
let after_cheap = parts.next().expect("a cheap-column budget and beyond");
let cheap_columns: u64 = after_cheap
.split("ms,")
.next()
.expect("a cheap-column budget")
.parse()
.expect("the cheap-column budget is an integer");
(names, cheap_columns)
}
#[test]
fn first_frame_budget_constants_match_the_spec_of_record() {
let spec = spec_refresh_md();
let (names_ms, cheap_columns_ms) = spec_first_frame_budgets_ms(&spec);
assert_eq!(names_ms, FIRST_FRAME_NAMES_BUDGET_MS);
assert_eq!(cheap_columns_ms, FIRST_FRAME_CHEAP_COLUMNS_BUDGET_MS);
}
#[test]
fn every_dispatched_entity_gets_its_dirty_cell_settled_not_a_subset() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
const ENTITY_COUNT: usize = 16;
for index in 0..ENTITY_COUNT {
init_repo_with_a_commit(&root.join(format!("repo-{index}")));
}
let core = Core::start_discovered(spec(vec![root]));
let keys: Vec<EntityKey> = core
.snapshot()
.entities
.iter()
.map(|entity| entity.key.clone())
.collect();
assert_eq!(keys.len(), ENTITY_COUNT, "expected every repo discovered");
core.refresh(&keys);
let settled = core.settle();
for entity in &settled.entities {
assert!(
matches!(
entity.dirty.settled(),
Some(Settled::Known {
value: _,
at: _,
stale: _
})
),
"entity {:?} was left without a settled dirty cell, which is exactly what a \
visibility-scoped dispatch would leave behind on the entities it skipped: \
got {:?}",
entity.name,
entity.dirty.settled()
);
}
}
#[test]
fn cheap_outcomes_land_before_a_held_phase_c_settles() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let (core, launched) = started_and_settled(spec(vec![root]));
let key = launched.entities[0].key.clone();
assert_eq!(
dirty_total(&launched.entities[0]),
0,
"the fixture starts clean, which is the value the held phase C must still be \
reading once the working tree below has moved"
);
git(&repo, &["checkout", "-b", "held"]);
fs::write(repo.join("untracked.txt"), b"uncommitted")
.expect("write an untracked file into the fixture");
core.hold_phase_c_for_test(&key);
core.refresh(std::slice::from_ref(&key));
core.wait_phase_c_landed_for_test(&key);
let mid_flight = core.snapshot();
let entity = mid_flight
.entities
.iter()
.find(|entity| entity.key == key)
.expect("entity present");
assert!(
matches!(
entity.branch.settled(),
Some(Settled::Known {
value: Head::Branch { name, .. },
at: _,
stale: _
}) if &**name == "held"
),
"the cheap branch cell must carry this Generation's own answer while phase C is \
still held open, got {:?}",
entity.branch.settled()
);
assert!(
entity.dirty.is_in_flight() && dirty_total(entity) == 0,
"phase C is deliberately held open here; a bundled apply would already have \
written this cell's new count alongside branch, got {:?}",
entity.dirty.settled()
);
core.release_phase_c_for_test(&key);
core.wait_phase_c_finished_for_test(&key);
let settled = core.snapshot();
let entity = settled
.entities
.iter()
.find(|entity| entity.key == key)
.expect("entity present");
assert_eq!(
dirty_total(entity),
1,
"phase C must settle its own count once released, got {:?}",
entity.dirty.settled()
);
}
fn dirty_total(entity: &EntityState) -> u32 {
match entity.dirty.settled() {
Some(Settled::Known {
value,
at: _,
stale: _,
}) => value.total(),
other => panic!("expected a settled dirty count, got {other:?}"),
}
}
#[test]
fn splitting_the_probe_write_signals_settle_gate_exactly_once_per_entity() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
init_repo_with_a_commit(&root.join("a"));
init_repo_with_a_commit(&root.join("b"));
let (core, snapshot) = started_and_settled(spec(vec![root]));
let key_a = snapshot
.entities
.iter()
.find(|entity| &*entity.name == "a")
.expect("entity a present")
.key
.clone();
let key_b = snapshot
.entities
.iter()
.find(|entity| &*entity.name == "b")
.expect("entity b present")
.key
.clone();
core.hold_phase_c_for_test(&key_a);
core.hold_phase_c_for_test(&key_b);
core.refresh(&[key_a.clone(), key_b.clone()]);
core.wait_dispatched_for_test();
assert_eq!(
core.settle_gate_count_for_test(),
2,
"dispatching two entities must add exactly two to the settle gate"
);
core.wait_phase_c_landed_for_test(&key_a);
core.wait_phase_c_landed_for_test(&key_b);
assert_eq!(
core.settle_gate_count_for_test(),
2,
"the cheap apply must never touch the settle gate: both entities' cheap \
outcomes have landed and neither has finished phase C yet"
);
core.release_phase_c_for_test(&key_a);
core.wait_phase_c_finished_for_test(&key_a);
assert_eq!(
core.settle_gate_count_for_test(),
1,
"exactly one entity finished, so the gate must fall by exactly one, not two \
(double-counted) and not zero (left short)"
);
core.release_phase_c_for_test(&key_b);
core.wait_phase_c_finished_for_test(&key_b);
assert_eq!(
core.settle_gate_count_for_test(),
0,
"both entities finished, so the gate must be fully drained"
);
}
fn registered_gate(core: &Core, key: &EntityKey) -> PhaseCGateHandle {
core.phase_c_gates
.lock()
.unwrap()
.get(key)
.cloned()
.expect("hold_phase_c_for_test must be called before reading its gate")
}
fn release_gate(gate: &PhaseCGateHandle) {
let (lock, cvar) = &**gate;
lock.lock().unwrap().may_proceed = true;
cvar.notify_all();
}
fn gate_is_finished(gate: &PhaseCGateHandle) -> bool {
gate.0.lock().unwrap().finished
}
#[test]
fn a_probe_signals_the_phase_c_gate_its_own_generation_was_dispatched_against() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
init_repo_with_a_commit(&root.join("repo"));
let (core, launched) = started_and_settled(spec(vec![root]));
let key = launched.entities[0].key.clone();
core.hold_phase_c_for_test(&key);
let dispatched_against = registered_gate(&core, &key);
core.refresh(std::slice::from_ref(&key));
core.wait_phase_c_landed_for_test(&key);
core.hold_phase_c_for_test(&key);
let registered_later = registered_gate(&core, &key);
release_gate(&dispatched_against);
wait_for(
"the held probe to signal the gate its own Generation was dispatched against",
|| gate_is_finished(&dispatched_against),
);
assert!(
!gate_is_finished(®istered_later),
"a gate registered after this Generation dispatched must never be marked \
finished by it: a test waiting on that gate would return before this \
Generation had applied its outcome or decremented the settle gate"
);
}
#[test]
fn a_probe_finishing_clears_only_its_own_generations_in_flight_entry() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
init_repo_with_a_commit(&root.join("repo"));
let (core, launched) = started_and_settled(spec(vec![root]));
let key = launched.entities[0].key.clone();
core.hold_phase_c_for_test(&key);
core.refresh(std::slice::from_ref(&key));
core.wait_phase_c_landed_for_test(&key);
let superseding = core.begin_shared_generation_for_test(std::slice::from_ref(&key));
core.release_phase_c_for_test(&key);
core.wait_phase_c_finished_for_test(&key);
core.refresh(std::slice::from_ref(&key));
core.wait_dispatched_for_test();
assert!(
superseding.cancels[&key].load(Ordering::Acquire),
"a probe from a Generation that has already been superseded must leave the \
live Generation's in-flight entry alone, or the Generation after it has \
nothing to interrupt"
);
}
#[test]
fn refresh_dispatches_phase_c_in_exactly_the_order_it_is_given() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
const ENTITY_COUNT: usize = 6;
for index in 0..ENTITY_COUNT {
init_repo_with_a_commit(&root.join(format!("repo-{index}")));
}
let (core, launched) = started_and_settled(spec(vec![root]));
let discovery_order: Vec<EntityKey> = launched
.entities
.iter()
.map(|entity| entity.key.clone())
.collect();
assert_eq!(
discovery_order.len(),
ENTITY_COUNT,
"expected every repo discovered"
);
let cursor = discovery_order[3].clone();
let visible = [discovery_order[1].clone(), discovery_order[4].clone()];
let mut three_tier_order = vec![cursor.clone()];
three_tier_order.extend(visible.iter().cloned());
for key in &discovery_order {
if *key != cursor && !visible.contains(key) {
three_tier_order.push(key.clone());
}
}
assert_eq!(
three_tier_order.len(),
ENTITY_COUNT,
"sanity check: the hand-built order must cover every discovered entity exactly \
once"
);
core.refresh(&three_tier_order);
core.settle();
assert_eq!(
core.dispatch_log_for_test(),
three_tier_order,
"refresh must dispatch phase C in exactly the order it was given: the cursor \
row, then the visible rows, then the rest in discovery order"
);
}
#[test]
fn refresh_reuses_the_cached_repository_handle_rather_than_reopening_it() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let core = Core::start_discovered(spec(vec![root]));
let key = core.snapshot().entities[0].key.clone();
let before = core
.cached_repo_handle_for_test(&key)
.expect("discovery should have cached a handle");
core.refresh(std::slice::from_ref(&key));
core.settle();
let after = core
.cached_repo_handle_for_test(&key)
.expect("the cached handle should still be there after a refresh");
assert!(
Arc::ptr_eq(&before, &after),
"a refresh must reuse the cached handle, not replace it with a new one"
);
}
#[test]
fn refresh_running_reads_true_the_instant_refresh_returns_and_false_once_it_settles() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
init_repo_with_a_commit(&root.join("repo"));
let core = Core::start_discovered(spec(vec![root]));
core.settle();
assert!(
!core.refresh_running(),
"sanity: nothing outstanding once startup has settled"
);
let keys: Vec<EntityKey> = core
.snapshot()
.entities
.iter()
.map(|entity| entity.key.clone())
.collect();
core.refresh(&keys);
assert!(
core.refresh_running(),
"refresh reserves its Generation and records the dispatch debt before it \
returns, so this must already read true"
);
core.settle();
assert!(
!core.refresh_running(),
"settle blocks until nothing is outstanding, so this must read false once it \
returns"
);
}
#[test]
fn probing_a_key_with_no_cached_handle_still_opens_the_repository_itself() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let empty_root = root_of(&tempfile::tempdir().expect("temp dir"));
let core = Core::start_discovered(spec(vec![empty_root]));
let key = EntityKey::new(Arc::from(repo.as_path()));
assert!(core.cached_repo_handle_for_test(&key).is_none());
let entity = core.probe_now(&key);
assert!(matches!(
entity.branch.settled(),
Some(Settled::Known {
value: Head::Branch { .. },
at: _,
stale: _
})
));
}
#[test]
fn an_empty_order_dispatches_nothing_and_settle_returns_immediately() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let (core, _launched) = started_and_settled(spec(vec![root]));
assert!(
!core.dispatch_log_for_test().is_empty(),
"launch dispatched nothing, so an empty log below would say nothing about the \
empty order"
);
core.refresh(&[]);
core.wait_dispatched_for_test();
assert_eq!(
core.dispatch_log_for_test(),
Vec::new(),
"an empty order must dispatch no probe"
);
let settled = core
.try_settle(Duration::from_millis(50))
.expect("an empty order raises no probe, so the settle gate is already at zero");
assert!(!settled.entities[0].branch.is_in_flight());
}
fn one_probe_owed_that_never_lands(
dir: &tempfile::TempDir,
) -> (Core, crossbeam_channel::Sender<Instant>) {
let root = root_of(dir);
init_repo_with_a_commit(&root.join("repo"));
let (tick_tx, tick_rx) = crossbeam_channel::unbounded::<Instant>();
let core = Core::start_for_test(spec(vec![root]), Duration::from_secs(3600), tick_rx)
.discovered()
.core;
let key = settle_launch(&core).entities[0].key.clone();
core.begin_untracked_probe_for_test(&key);
(core, tick_tx)
}
#[test]
#[should_panic(expected = "waiting for everything this Core has in flight to land")]
fn a_settle_that_expires_reports_at_the_wait_rather_than_returning_the_table() {
let dir = tempfile::tempdir().expect("temp dir");
let (core, _tick_tx) = one_probe_owed_that_never_lands(&dir);
core.settle_within(Duration::from_millis(20));
}
#[test]
fn try_settle_hands_an_expiry_back_as_an_error_carrying_the_table_it_gave_up_on() {
let dir = tempfile::tempdir().expect("temp dir");
let (core, _tick_tx) = one_probe_owed_that_never_lands(&dir);
let unsettled = core
.try_settle(Duration::from_millis(20))
.expect_err("a probe nothing will ever complete cannot settle");
assert!(
unsettled.entities[0].branch.is_in_flight(),
"the Err arm must still carry the table as it stood, so a caller that degrades \
deliberately has something to degrade with"
);
}
#[test]
fn try_settle_hands_a_generation_that_really_landed_back_as_ok() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
init_repo_with_a_commit(&root.join("repo"));
let (core, launched) = started_and_settled(spec(vec![root]));
let key = launched.entities[0].key.clone();
core.refresh(std::slice::from_ref(&key));
let settled = core
.try_settle(BACKSTOP)
.expect("a dispatched Generation must land inside the backstop");
assert!(!settled.entities[0].branch.is_in_flight());
}
#[test]
fn probe_now_settles_the_sync_cell_as_well_as_the_branch_it_depends_on() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let core = Core::start_discovered(spec(vec![root]));
let key = core.snapshot().entities[0].key.clone();
let entity = core.probe_now(&key);
assert!(
matches!(
entity.sync.settled(),
Some(Settled::Known {
value: SyncState::NoRemote,
at: _,
stale: _
})
),
"expected probe_now to settle sync, got {:?}",
entity.sync.settled()
);
}
#[test]
fn probe_now_settles_the_base_cell_as_well_as_the_branch_it_depends_on() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let core = Core::start_discovered(spec(vec![root]));
let key = core.snapshot().entities[0].key.clone();
let entity = core.probe_now(&key);
assert!(
matches!(entity.base.settled(), Some(Settled::NotApplicable)),
"expected probe_now to settle base Not applicable for a Repo with no remote, \
got {:?}",
entity.base.settled()
);
}
#[test]
fn refresh_settles_a_real_base_count_against_the_resolved_default_branch() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
git(
&repo,
&[
"remote",
"add",
"origin",
"https://example.invalid/repo.git",
],
);
let root_sha = head_sha(&repo);
git(&repo, &["commit", "--allow-empty", "-m", "second"]);
let tip_sha = head_sha(&repo);
git(&repo, &["reset", "--hard", &root_sha]);
git(&repo, &["update-ref", "refs/remotes/origin/main", &tip_sha]);
let core = Core::start_discovered(spec(vec![root]));
let key = core.snapshot().entities[0].key.clone();
core.refresh(std::slice::from_ref(&key));
let settled = core.settle();
assert!(
matches!(
settled.entities[0].base.settled(),
Some(Settled::Known {
value: 1,
at: _,
stale: _
})
),
"expected a real refresh to settle base's live count against the resolved \
default branch, got {:?}",
settled.entities[0].base.settled()
);
}
#[test]
fn probe_now_settles_the_dirty_cell_with_the_counts_it_probed() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
fs::write(repo.join("untracked.txt"), "x").expect("write untracked file");
let core = Core::start_discovered(spec(vec![root]));
let key = core.snapshot().entities[0].key.clone();
let entity = core.probe_now(&key);
assert!(
matches!(
entity.dirty.settled(),
Some(Settled::Known {
value: DirtyCounts {
modified: 0,
untracked: 1,
deleted: 0,
},
at: _,
stale: _
})
),
"expected probe_now to settle dirty with the one untracked path, got {:?}",
entity.dirty.settled()
);
}
#[test]
fn probe_now_updates_the_entity_synchronously_with_no_refresh_call() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let core = Core::start_discovered(spec(vec![root]));
let key = core.snapshot().entities[0].key.clone();
let entity = core.probe_now(&key);
assert!(matches!(
entity.branch.settled(),
Some(Settled::Known {
value: Head::Branch { .. },
at: _,
stale: _
})
));
}
#[test]
fn the_display_name_agrees_between_discovery_and_probe_nows_fallback_insert() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("named-repo");
init_repo_with_a_commit(&repo);
let core = Core::start_discovered(spec(vec![root]));
let discovered = core.snapshot().entities[0].clone();
assert_eq!(&*discovered.name, "named-repo");
core.dismiss(&discovered.key);
assert!(core.snapshot().entities.is_empty());
let reinserted = core.probe_now(&discovered.key);
assert_eq!(
reinserted.name, discovered.name,
"the name discovery assigned and the name probe_now's fallback insert \
assigns for the same path must be byte-identical"
);
}
#[test]
fn dismiss_removes_the_entity_from_the_snapshot() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let core = Core::start_discovered(spec(vec![root]));
let key = core.snapshot().entities[0].key.clone();
core.dismiss(&key);
assert!(core.snapshot().entities.is_empty());
}
#[test]
fn an_entitys_steps_run_in_order_and_a_failure_marks_every_later_step_not_run() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let marker = repo.join("step-three-ran");
let core = Core::start_discovered(spec(vec![root]));
let key = core.snapshot().entities[0].key.clone();
let steps = vec![
step(&["true"]),
step(&["sh", "-c", "exit 7"]),
step(&["touch", "step-three-ran"]),
];
let started = core.run_action(action("reinstall", steps), std::slice::from_ref(&key));
assert!(started);
wait_for("the fan-out to finish and write a receipt", || {
!core.action_running()
});
let receipt = core.snapshot().entities[0]
.last_action
.clone()
.expect("receipt written");
assert_eq!(receipt.steps.len(), 3);
assert_eq!(receipt.steps[0].outcome, StepOutcome::Ok);
assert_eq!(receipt.steps[1].outcome, StepOutcome::Failed(7));
assert_eq!(
receipt.steps[2].outcome,
StepOutcome::NotRun,
"a step after a failure must be recorded NotRun, not silently dropped or run anyway"
);
assert!(
!marker.exists(),
"the third step's own `touch` must never have run: its marker file exists, so \
the step ran despite being recorded NotRun"
);
}
#[test]
fn steps_run_in_the_order_theyre_declared_not_some_other_order() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let order_log = repo.join("order.log");
let core = Core::start_discovered(spec(vec![root]));
let key = core.snapshot().entities[0].key.clone();
let steps = vec![
step(&["sh", "-c", "printf 1 >> order.log"]),
step(&["sh", "-c", "printf 2 >> order.log"]),
step(&["sh", "-c", "printf 3 >> order.log"]),
];
let started = core.run_action(action("ordering", steps), std::slice::from_ref(&key));
assert!(started);
wait_for("the fan-out to finish and write a receipt", || {
!core.action_running()
});
let receipt = core.snapshot().entities[0]
.last_action
.clone()
.expect("receipt written");
assert_eq!(receipt.steps.len(), 3);
assert!(
receipt
.steps
.iter()
.all(|result| result.outcome == StepOutcome::Ok),
"every step here always exits zero; this test isolates ordering from gating"
);
let content = fs::read_to_string(&order_log).expect("order.log written by the steps");
assert_eq!(
content, "123",
"the file's content pins actual execution order; running the steps out of \
declaration order would produce a different digit sequence here even though \
every step still succeeds"
);
}
#[test]
fn a_still_running_actions_finished_step_and_its_currently_executing_one_are_both_visible_before_the_whole_run_ends()
{
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let core = Core::start_discovered(spec(vec![root]));
let key = core.snapshot().entities[0].key.clone();
let steps = vec![step(&["true"]), step(&["sh", "-c", "sleep 0.5"])];
let started = core.run_action(action("reinstall", steps), std::slice::from_ref(&key));
assert!(started);
wait_for(
"a receipt naming the second step running before the run finished",
|| {
core.snapshot().entities[0]
.last_action
.as_ref()
.and_then(|receipt| receipt.running.as_ref())
.is_some_and(|running| running.label.contains("sleep"))
},
);
let mid_run = core.snapshot().entities[0]
.last_action
.clone()
.expect("receipt written");
assert_eq!(
mid_run.steps.len(),
1,
"the first, already-finished step must already be in `steps`"
);
assert_eq!(mid_run.steps[0].outcome, StepOutcome::Ok);
let running = mid_run.running.expect("a step must be recorded running");
assert!(
running.label.contains("sleep"),
"expected the running step's own label, got {:?}",
running.label
);
wait_for("the fan-out to finish", || !core.action_running());
let finished = core.snapshot().entities[0]
.last_action
.clone()
.expect("receipt written");
assert!(
finished.running.is_none(),
"a finished receipt must carry no running step"
);
assert_eq!(finished.steps.len(), 2);
}
#[test]
fn a_shell_true_step_runs_through_shell_c_with_repon_as_its_own_dollar_zero() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let core = Core::start_discovered(spec(vec![root]));
let key = core.snapshot().entities[0].key.clone();
let steps = vec![shell_step("echo \"[$0]\"")];
let started = core.run_action(action("shell-step", steps), std::slice::from_ref(&key));
assert!(started);
wait_for("the fan-out to finish and write a receipt", || {
!core.action_running()
});
let receipt = core.snapshot().entities[0]
.last_action
.clone()
.expect("receipt written");
assert_eq!(receipt.steps.len(), 1);
assert_eq!(receipt.steps[0].outcome, StepOutcome::Ok);
assert_eq!(&*receipt.steps[0].output, b"[repon]\n");
assert!(
receipt.steps[0].shell,
"the receipt's own StepResult::shell must carry the mode the step ran under"
);
}
#[test]
fn an_interactive_shell_true_step_runs_through_run_action_with_interactive_on_its_receipt() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let core = Core::start_discovered(spec(vec![root]));
let key = core.snapshot().entities[0].key.clone();
let steps = vec![interactive_shell_step("true")];
let started = core.run_action(
action("interactive-step", steps),
std::slice::from_ref(&key),
);
assert!(started);
wait_for("the fan-out to finish and write a receipt", || {
!core.action_running()
});
let receipt = core.snapshot().entities[0]
.last_action
.clone()
.expect("receipt written");
assert_eq!(receipt.steps.len(), 1);
assert_eq!(receipt.steps[0].outcome, StepOutcome::Ok);
assert!(
receipt.steps[0].shell,
"an interactive step is still a shell step"
);
assert!(
receipt.steps[0].interactive,
"the receipt's own StepResult::interactive must carry the mode the step ran under"
);
}
#[test]
fn an_argv_step_runs_through_run_action_with_shell_false_on_its_receipt() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let core = Core::start_discovered(spec(vec![root]));
let key = core.snapshot().entities[0].key.clone();
let steps = vec![Step {
argv: vec!["true".to_string()],
shell: false,
interactive: false,
env: Vec::new(),
}];
let started = core.run_action(action("argv-step", steps), std::slice::from_ref(&key));
assert!(started);
wait_for("the fan-out to finish and write a receipt", || {
!core.action_running()
});
let receipt = core.snapshot().entities[0]
.last_action
.clone()
.expect("receipt written");
assert!(!receipt.steps[0].shell);
}
#[test]
fn starting_an_action_cancels_any_generation_already_in_flight() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let core = Core::start_discovered(spec(vec![root]));
let key = core.snapshot().entities[0].key.clone();
let in_flight = core.begin_shared_generation_for_test(std::slice::from_ref(&key));
let cancel = in_flight
.cancels
.get(&key)
.expect("the in-flight entity has a cancel flag")
.clone();
assert!(!cancel.load(Ordering::Acquire));
let started = core.run_action(
action("reinstall", vec![step(&["true"])]),
std::slice::from_ref(&key),
);
assert!(started);
assert!(
cancel.load(Ordering::Acquire),
"starting an Action must cancel a Generation already in flight, not share \
execution with it"
);
wait_for("the fan-out and its completion refresh to drain", || {
!core.action_running()
});
}
#[test]
fn a_finished_action_starts_exactly_one_generation_over_every_known_entity() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let acted_on = root.join("acted-on");
let untouched = root.join("untouched");
init_repo_with_a_commit(&acted_on);
init_repo_with_a_commit(&untouched);
let (core, before) = started_and_settled(spec(vec![root]));
let acted_key = before
.entities
.iter()
.find(|entity| entity.key.path() == acted_on)
.expect("the acted-on entity is discovered")
.key
.clone();
let started = core.run_action(
action("reinstall", vec![step(&["true"])]),
std::slice::from_ref(&acted_key),
);
assert!(started);
wait_for(
"the completion Generation to probe every known entity, including the one the \
Action never touched",
|| {
let snapshot = core.snapshot();
snapshot.generation != before.generation
&& snapshot.entities.iter().all(|entity| {
matches!(
entity.branch.settled(),
Some(Settled::Known {
value: _,
at: _,
stale: _
})
)
})
},
);
assert_eq!(
core.settle().generation,
before.generation.successor(),
"completion must start exactly one Generation: not zero (no refresh at all) and \
not two (a double refresh)"
);
}
#[test]
fn an_excluded_row_swept_into_an_action_gets_a_not_applicable_receipt_and_no_other_path_does() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let excluded_repo = root.join("excluded");
let normal_repo = root.join("normal");
init_repo_with_a_commit(&excluded_repo);
init_repo_with_a_commit(&normal_repo);
let core = Core::start_discovered(spec_with_overrides(
vec![root],
vec![RepoOverride {
path: excluded_repo.clone(),
default_branch: None,
excluded: true,
}],
));
let snapshot = core.snapshot();
let find = |path: &Path| {
snapshot
.entities
.iter()
.find(|entity| entity.key.path() == path)
.unwrap_or_else(|| panic!("entity at {path:?} present"))
.key
.clone()
};
let excluded_key = find(&excluded_repo);
let normal_key = find(&normal_repo);
assert!(
snapshot
.entities
.iter()
.find(|entity| entity.key == excluded_key)
.unwrap()
.excluded
);
let started = core.run_action(
action("reinstall", vec![step(&["sh", "-c", "exit 3"])]),
&[excluded_key.clone(), normal_key.clone()],
);
assert!(started);
wait_for("the fan-out to finish", || !core.action_running());
let after = core.snapshot();
let receipt_of = |key: &EntityKey| {
after
.entities
.iter()
.find(|entity| entity.key == *key)
.unwrap()
.last_action
.clone()
.unwrap()
};
let excluded_receipt = receipt_of(&excluded_key);
assert!(excluded_receipt.not_applicable());
assert!(excluded_receipt.steps.is_empty());
let normal_receipt = receipt_of(&normal_key);
assert!(
!normal_receipt.not_applicable(),
"a row that actually ran a step, even a failing one, must never read as \
not_applicable: an excluded row is the one legitimate producer of that outcome"
);
assert!(!normal_receipt.steps.is_empty());
assert!(normal_receipt.failed());
}
#[test]
fn operable_count_matches_how_many_entities_run_action_actually_runs_a_step_against() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let excluded_repo = root.join("excluded");
let normal_repo = root.join("normal");
init_repo_with_a_commit(&excluded_repo);
init_repo_with_a_commit(&normal_repo);
let core = Core::start_discovered(spec_with_overrides(
vec![root],
vec![RepoOverride {
path: excluded_repo.clone(),
default_branch: None,
excluded: true,
}],
));
let snapshot = core.snapshot();
let find = |path: &Path| {
snapshot
.entities
.iter()
.find(|entity| entity.key.path() == path)
.unwrap_or_else(|| panic!("entity at {path:?} present"))
.key
.clone()
};
let order = [find(&excluded_repo), find(&normal_repo)];
assert_eq!(
core.operable_count(&order),
1,
"one of the two rows is excluded, so exactly one is operable"
);
let started = core.run_action(action("reinstall", vec![step(&["true"])]), &order);
assert!(started);
wait_for("every entity in the order to carry a receipt", || {
let snapshot = core.snapshot();
order.iter().all(|key| {
snapshot
.entities
.iter()
.find(|entity| entity.key == *key)
.and_then(|entity| entity.last_action.as_ref())
.is_some()
})
});
let after = core.snapshot();
let actually_ran = after
.entities
.iter()
.filter(|entity| order.contains(&entity.key))
.filter(|entity| {
entity
.last_action
.as_ref()
.is_some_and(|receipt| !receipt.not_applicable())
})
.count();
assert_eq!(
core.operable_count(&order),
actually_ran,
"operable_count must report exactly how many rows run_action actually ran a \
step against, not merely how many keys resolved"
);
}
#[test]
fn run_action_for_entity_blocking_returns_the_finished_receipt_on_the_calling_thread() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let marker = repo.join("hook-ran");
let core = Core::start_discovered(spec_with_overrides(vec![root], Vec::new()));
let key = core
.snapshot()
.entities
.iter()
.find(|entity| entity.key.path() == repo)
.expect("the repo is discovered")
.key
.clone();
let receipt = core
.run_action_for_entity_blocking(
&action("hook", vec![step(&["touch", "hook-ran"])]),
&key,
)
.expect("the entity is known");
assert!(
marker.exists(),
"the step must have already run by the time this call returns"
);
assert_eq!(receipt.steps.len(), 1);
assert_eq!(receipt.steps[0].outcome, StepOutcome::Ok);
}
#[test]
fn run_action_for_entity_blocking_answers_none_for_an_unknown_key() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let core = Core::start_discovered(spec_with_overrides(vec![root.clone()], Vec::new()));
let unknown = EntityKey::new(Arc::from(root.join("never-discovered").as_path()));
assert!(
core.run_action_for_entity_blocking(&action("hook", vec![step(&["true"])]), &unknown)
.is_none()
);
}
#[test]
fn run_action_skips_a_row_its_when_predicate_disproves_rather_than_running_it_anyway() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let proved_repo = root.join("alpha");
let disproved_repo = root.join("beta");
init_repo_with_a_commit(&proved_repo);
init_repo_with_a_commit(&disproved_repo);
let core = Core::start_discovered(spec(vec![root]));
let snapshot = core.snapshot();
let find = |path: &Path| {
snapshot
.entities
.iter()
.find(|entity| entity.key.path() == path)
.unwrap_or_else(|| panic!("entity at {path:?} present"))
.key
.clone()
};
let proved_key = find(&proved_repo);
let disproved_key = find(&disproved_repo);
let order = [proved_key.clone(), disproved_key.clone()];
let started = core.run_action(
action_with_when(
"reinstall",
vec![step(&["sh", "-c", "exit 3"])],
"name:alpha",
),
&order,
);
assert!(started);
wait_for("the fan-out to finish", || !core.action_running());
let after = core.snapshot();
let receipt_of = |key: &EntityKey| {
after
.entities
.iter()
.find(|entity| entity.key == *key)
.unwrap()
.last_action
.clone()
.unwrap()
};
let proved_receipt = receipt_of(&proved_key);
assert_eq!(
proved_receipt.skip, None,
"the row the predicate proved must actually run"
);
assert!(proved_receipt.failed(), "its own step still ran and failed");
let disproved_receipt = receipt_of(&disproved_key);
assert!(
disproved_receipt.inapplicable(),
"the row the predicate disproved must be skipped rather than run"
);
assert!(disproved_receipt.steps.is_empty());
assert!(
!disproved_receipt.failed(),
"a skipped row never ran a step, so it cannot have failed one"
);
}
#[test]
fn applicability_subtracts_an_excluded_row_before_the_predicate_reads_it() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let excluded_repo = root.join("excluded");
let normal_repo = root.join("normal");
init_repo_with_a_commit(&excluded_repo);
init_repo_with_a_commit(&normal_repo);
let core = Core::start_discovered(spec_with_overrides(
vec![root],
vec![RepoOverride {
path: excluded_repo.clone(),
default_branch: None,
excluded: true,
}],
));
let order: Vec<EntityKey> = core
.snapshot()
.entities
.iter()
.map(|entity| entity.key.clone())
.collect();
assert_eq!(order.len(), 2, "the fixture must discover both repos");
let counts = core.applicability(&order, &Filter::parse("kind:repo"));
assert_eq!(
counts.total(),
core.operable_count(&order),
"the predicate must be counted over exactly the rows `operable_count` keeps"
);
assert_eq!(
counts,
Applicability {
applicable: 1,
inapplicable: 0,
unresolved: 0,
}
);
}
#[test]
fn operable_count_silently_drops_a_key_that_no_longer_resolves() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let core = Core::start_discovered(spec(vec![root]));
let real_key = core.snapshot().entities[0].key.clone();
let unknown_key = EntityKey::new(Arc::from(dir.path().join("never-discovered")));
assert_eq!(core.operable_count(&[real_key, unknown_key]), 1);
}
#[test]
fn only_one_action_fan_out_runs_at_a_time_a_second_call_is_rejected_while_one_is_live() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let core = Core::start_discovered(spec(vec![root]));
let key = core.snapshot().entities[0].key.clone();
let slow = action("first", vec![step(&["sh", "-c", "sleep 0.3"])]);
let fast = action("second", vec![step(&["true"])]);
let first_started = core.run_action(slow, std::slice::from_ref(&key));
let second_started = core.run_action(fast, std::slice::from_ref(&key));
assert!(first_started);
assert!(
!second_started,
"a second run_action call must be rejected while the first is still in flight"
);
wait_for("the accepted first fan-out to finish", || {
!core.action_running()
});
let receipt = core.snapshot().entities[0].last_action.clone().unwrap();
assert_eq!(
&*receipt.label, "first",
"the surviving receipt must be the accepted first run's, never the rejected second"
);
}
#[test]
fn hold_action_genuinely_pauses_a_running_steps_progress_and_continue_action_resumes_it() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let core = Core::start_discovered(spec(vec![root]));
let key = core.snapshot().entities[0].key.clone();
let two_seconds = action("brief", vec![step(&["sh", "-c", "sleep 2"])]);
assert!(core.run_action(two_seconds, std::slice::from_ref(&key)));
wait_for("the two-second step to actually start running", || {
core.snapshot().entities[0]
.last_action
.as_ref()
.is_some_and(|receipt| receipt.running.is_some())
});
for _ in 0..20 {
core.hold_action();
thread::sleep(Duration::from_millis(20));
}
thread::sleep(Duration::from_millis(1_800));
assert!(
core.action_running(),
"a genuinely held step must not have finished on its own well past its own 2s \
sleep; a no-op hold_action would already show this false here"
);
core.continue_action();
wait_for("continue_action to let the held step finish", || {
!core.action_running()
});
let receipt = core.snapshot().entities[0].last_action.clone().unwrap();
assert_eq!(receipt.steps[0].outcome, StepOutcome::Ok);
}
#[test]
fn hold_continue_and_stop_action_are_no_ops_with_no_fan_out_running() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let core = Core::start_discovered(spec(vec![root]));
core.hold_action();
core.continue_action();
core.stop_action();
assert!(!core.action_running());
}
#[test]
fn stop_action_escalates_from_sigterm_to_sigkill_against_a_trapping_step() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let core = Core::start_discovered(spec(vec![root]));
let key = core.snapshot().entities[0].key.clone();
let sleep_past_the_backstop = format!("trap '' TERM; sleep {}", FIXTURE_LIFETIME.as_secs());
let trapping = action(
"trapping",
vec![step(&["sh", "-c", &sleep_past_the_backstop])],
);
assert!(core.run_action(trapping, std::slice::from_ref(&key)));
wait_for("the trapping step to actually start running", || {
core.snapshot().entities[0]
.last_action
.as_ref()
.is_some_and(|receipt| receipt.running.is_some())
});
thread::sleep(Duration::from_millis(100));
core.stop_action();
wait_for(
"a SIGTERM-trapping step to come down from the follow-up SIGKILL",
|| !core.action_running(),
);
let receipt = core.snapshot().entities[0].last_action.clone().unwrap();
assert_eq!(receipt.steps.len(), 1);
assert_eq!(
receipt.steps[0].outcome,
StepOutcome::Cancelled,
"a step running when the run was cancelled must read Cancelled, never Failed"
);
}
#[test]
fn cancelled_and_not_run_are_distinct_outcomes_shown_together_in_one_run() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
init_repo_with_a_commit(&root.join("fail"));
init_repo_with_a_commit(&root.join("slow"));
let core = Core::start_discovered(spec(vec![root]));
let snapshot = core.snapshot();
let fail_key = snapshot
.entities
.iter()
.find(|entity| &*entity.name == "fail")
.expect("the fail entity is present")
.key
.clone();
let slow_key = snapshot
.entities
.iter()
.find(|entity| &*entity.name == "slow")
.expect("the slow entity is present")
.key
.clone();
let branch_on_the_entity_name = format!(
"case \"$(basename \"$PWD\")\" in fail) exit 1 ;; *) sleep {} ;; esac",
FIXTURE_LIFETIME.as_secs()
);
let steps = vec![
step(&["sh", "-c", &branch_on_the_entity_name]),
step(&["true"]),
];
let mut action_spec = action("mixed", steps);
action_spec.concurrency = 2;
assert!(core.run_action(action_spec, &[fail_key.clone(), slow_key.clone()]));
wait_for(
"`fail` finished and `slow` still running before cancelling",
|| {
let snapshot = core.snapshot();
let fail_done = snapshot
.entities
.iter()
.find(|entity| entity.key == fail_key)
.and_then(|entity| entity.last_action.as_ref())
.is_some_and(|receipt| receipt.steps.len() == 2);
let slow_running = snapshot
.entities
.iter()
.find(|entity| entity.key == slow_key)
.and_then(|entity| entity.last_action.as_ref())
.is_some_and(|receipt| receipt.running.is_some());
fail_done && slow_running
},
);
core.stop_action();
wait_for("the fan-out to finish once cancelled", || {
!core.action_running()
});
let snapshot = core.snapshot();
let fail_receipt = snapshot
.entities
.iter()
.find(|entity| entity.key == fail_key)
.and_then(|entity| entity.last_action.clone())
.expect("fail's own receipt");
assert_eq!(fail_receipt.steps[0].outcome, StepOutcome::Failed(1));
assert_eq!(
fail_receipt.steps[1].outcome,
StepOutcome::NotRun,
"blocked by fail's own earlier failure, not by the later cancellation"
);
let slow_receipt = snapshot
.entities
.iter()
.find(|entity| entity.key == slow_key)
.and_then(|entity| entity.last_action.clone())
.expect("slow's own receipt");
assert_eq!(
slow_receipt.steps[0].outcome,
StepOutcome::Cancelled,
"a step running when the run was cancelled must read Cancelled"
);
assert_eq!(
slow_receipt.steps[1].outcome,
StepOutcome::Cancelled,
"a step that had not started when the run was cancelled must also read \
Cancelled, never NotRun, which stays reserved for an earlier failure"
);
}
#[test]
fn a_panicking_fan_out_still_resets_action_running_so_a_later_action_can_start() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let (core, launched) = started_and_settled(spec(vec![root]));
let key = launched.entities[0].key.clone();
let started = core.run_action(
action("boom", vec![step(&["sh", "-c", "sleep 0.3"])]),
std::slice::from_ref(&key),
);
assert!(started);
let table = Arc::clone(&core.table);
thread::spawn(move || {
let _guard = table.write().unwrap();
panic!("deliberately poison the table lock for this test");
})
.join()
.expect_err("the poisoning thread must itself panic to poison the lock");
wait_for(
"a panicking fan-out to reset action_running rather than leave it stuck true",
|| !core.action_running.load(Ordering::Acquire),
);
core.table.clear_poison();
let second_started = core.run_action(
action("second", vec![step(&["true"])]),
std::slice::from_ref(&key),
);
assert!(
second_started,
"a later Action must be able to start once the panicking one has finished"
);
wait_for("the second Action to run to completion", || {
core.snapshot()
.entities
.iter()
.find(|entity| entity.key == key)
.and_then(|entity| entity.last_action.as_ref())
.is_some_and(|receipt| &*receipt.label == "second")
});
}
fn assert_vanished_with_stale_branch(entity: &EntityState, expected_branch: &str) {
assert_eq!(entity.presence, crate::entity::Presence::Vanished);
match entity.branch.settled() {
Some(Settled::Known {
value: Head::Branch { name, .. },
stale: true,
at: _,
}) => assert_eq!(
&**name, expected_branch,
"a Vanished entity must keep its last known branch value"
),
other => panic!(
"expected the branch cell to keep its Known value and go stale, got {other:?}"
),
}
}
#[test]
fn a_repo_removed_from_disk_stays_in_the_table_vanished_with_its_last_values() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let core = Core::start_discovered(spec(vec![root]));
let key = core.snapshot().entities[0].key.clone();
core.refresh(std::slice::from_ref(&key));
let before = core.settle();
let branch_name = match before.entities[0].branch.settled() {
Some(Settled::Known {
value: Head::Branch { name, .. },
at: _,
stale: _,
}) => name.to_string(),
other => panic!("expected the first refresh to settle a branch, got {other:?}"),
};
fs::remove_dir_all(&repo).expect("remove the repo from disk");
core.refresh(&[]);
let after = core.settle();
assert_eq!(
after.entities.len(),
1,
"a vanished entity must stay in the snapshot, not disappear from it"
);
assert_vanished_with_stale_branch(&after.entities[0], &branch_name);
}
#[test]
fn a_vanished_entitys_action_receipt_survives_the_vanished_staleness_pass_untouched() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let core = Core::start_discovered(spec(vec![root]));
let key = core.snapshot().entities[0].key.clone();
let receipt = crate::entity::ActionReceipt {
label: Arc::from("reinstall"),
steps: Arc::from(vec![crate::entity::StepResult {
label: Arc::from("pnpm install"),
outcome: crate::entity::StepOutcome::Ok,
output: Arc::from(&b""[..]),
elapsed: Duration::from_millis(1),
elision: None,
shell: false,
interactive: false,
}]),
skip: None,
finished_at: Timestamp::now(),
running: None,
};
core.set_last_action_for_test(&key, receipt.clone());
fs::remove_dir_all(&repo).expect("remove the repo from disk");
core.refresh(&[]);
let after = core.settle();
let entity = &after.entities[0];
assert_eq!(entity.presence, crate::entity::Presence::Vanished);
assert_eq!(entity.last_action, Some(receipt));
}
#[test]
fn two_snapshots_of_an_entity_share_its_last_actions_label_and_steps_by_pointer() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let core = Core::start_discovered(spec(vec![root]));
let key = core.snapshot().entities[0].key.clone();
let receipt = crate::entity::ActionReceipt {
label: Arc::from("reinstall"),
steps: Arc::from(vec![crate::entity::StepResult {
label: Arc::from("pnpm install"),
outcome: crate::entity::StepOutcome::Failed(1),
output: Arc::from(&b""[..]),
elapsed: Duration::from_millis(1),
elision: None,
shell: false,
interactive: false,
}]),
skip: None,
finished_at: Timestamp::now(),
running: None,
};
core.set_last_action_for_test(&key, receipt);
let first = core.snapshot();
let second = core.snapshot();
let first_receipt = first.entities[0]
.last_action
.as_ref()
.expect("receipt was set");
let second_receipt = second.entities[0]
.last_action
.as_ref()
.expect("receipt was set");
assert!(
Arc::ptr_eq(&first_receipt.label, &second_receipt.label),
"two snapshots of the same receipt must share the label's allocation, not \
re-copy it"
);
assert!(
Arc::ptr_eq(&first_receipt.steps, &second_receipt.steps),
"two snapshots of the same receipt must share the steps slice's allocation, not \
re-copy it, which is also what shares every step's own captured output"
);
}
#[test]
fn a_submodule_removed_from_gitmodules_vanishes_by_the_same_rule_as_a_repo() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let parent = root.join("parent");
init_repo_with_a_commit(&parent);
fs::write(
parent.join(".gitmodules"),
"[submodule \"lib\"]\n\tpath = vendor/lib\n\turl = https://example.com/lib.git\n",
)
.expect("write .gitmodules");
let submodule_path = parent.join("vendor").join("lib");
init_repo_with_a_commit(&submodule_path);
let mut core_spec = spec(vec![root]);
core_spec.show_submodules = true;
let core = Core::start_discovered(core_spec);
let snapshot = core.snapshot();
let submodule_key = snapshot
.entities
.iter()
.find(|entity| matches!(entity.kind, Kind::Submodule))
.expect("submodule discovered")
.key
.clone();
core.refresh(std::slice::from_ref(&submodule_key));
let before = core.settle();
let submodule_before = before
.entities
.iter()
.find(|entity| entity.key == submodule_key)
.expect("submodule present");
let branch_name = match submodule_before.branch.settled() {
Some(Settled::Known {
value: Head::Branch { name, .. },
at: _,
stale: _,
}) => name.to_string(),
other => {
panic!("expected the submodule's first refresh to settle a branch, got {other:?}")
}
};
fs::write(parent.join(".gitmodules"), "").expect("clear .gitmodules");
core.refresh(&[]);
let after = core.settle();
let submodule_after = after
.entities
.iter()
.find(|entity| entity.key == submodule_key)
.expect("the vanished submodule must stay in the snapshot");
assert_vanished_with_stale_branch(submodule_after, &branch_name);
}
#[test]
fn dismissal_persists_nothing_across_a_fresh_core() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let first_core = Core::start_discovered(spec(vec![root.clone()]));
let key = first_core.snapshot().entities[0].key.clone();
first_core.dismiss(&key);
assert!(first_core.snapshot().entities.is_empty());
drop(first_core);
let second_core = Core::start_discovered(spec(vec![root]));
let snapshot = second_core.snapshot();
assert_eq!(
snapshot.entities.len(),
1,
"a fresh Core must discover the repo again"
);
assert_eq!(
snapshot.entities[0].presence,
crate::entity::Presence::Present,
"nothing from the dismissing Core's lifetime may be persisted, so the \
repo must come back Present, never restored as Vanished"
);
}
#[test]
fn a_repo_that_moves_reads_as_vanished_plus_new() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let original_path = root.join("original-name");
init_repo_with_a_commit(&original_path);
let core = Core::start_discovered(spec(vec![root.clone()]));
let original_key = core.snapshot().entities[0].key.clone();
core.refresh(std::slice::from_ref(&original_key));
let before = core.settle();
let branch_name = match before.entities[0].branch.settled() {
Some(Settled::Known {
value: Head::Branch { name, .. },
at: _,
stale: _,
}) => name.to_string(),
other => panic!("expected the first refresh to settle a branch, got {other:?}"),
};
let moved_path = root.join("new-name");
fs::rename(&original_path, &moved_path).expect("move the repo on disk");
core.refresh(&[]);
let after = core.settle();
assert_eq!(
after.entities.len(),
2,
"a moved entity must read as the old key vanished plus a new one present, \
never as one renamed entity"
);
let old_entity = after
.entities
.iter()
.find(|entity| entity.key == original_key)
.expect("the old key must stay in the table");
assert_vanished_with_stale_branch(old_entity, &branch_name);
let new_entity = after
.entities
.iter()
.find(|entity| entity.key != original_key)
.expect("a new entity at the moved path must be present");
assert_eq!(new_entity.presence, crate::entity::Presence::Present);
assert_eq!(new_entity.key.path(), moved_path);
}
#[test]
fn a_vanished_repo_recreated_on_disk_reads_present_on_the_next_refresh() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let core = Core::start_discovered(spec(vec![root]));
let key = core.snapshot().entities[0].key.clone();
fs::remove_dir_all(&repo).expect("remove the repo from disk");
core.refresh(&[]);
let vanished = core.settle();
assert_eq!(
vanished.entities[0].presence,
crate::entity::Presence::Vanished,
"the repo must read Vanished once removed from disk"
);
init_repo_with_a_commit(&repo);
core.refresh(&[]);
let recreated = core.settle();
let entity = recreated
.entities
.iter()
.find(|entity| entity.key == key)
.expect("the recreated repo must still resolve to the same entity key");
assert_eq!(
entity.presence,
crate::entity::Presence::Present,
"an entity discovery finds again after it vanished must read Present, \
not stay stuck Vanished forever"
);
}
#[test]
fn a_new_repo_created_after_start_is_discovered_by_the_next_refresh() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
init_repo_with_a_commit(&root.join("first"));
let core = Core::start_discovered(spec(vec![root.clone()]));
assert_eq!(core.snapshot().entities.len(), 1);
init_repo_with_a_commit(&root.join("second"));
core.refresh(&[]);
let after = core.settle();
assert_eq!(
after.entities.len(),
2,
"a new repo created after start must be found by the next refresh's own discovery"
);
let new_key = after
.entities
.iter()
.find(|entity| &*entity.name == "second")
.expect("the newly discovered repo must be named by the walk")
.key
.clone();
core.refresh(std::slice::from_ref(&new_key));
let probed = core.settle();
let new_entity = probed
.entities
.iter()
.find(|entity| entity.key == new_key)
.expect("the newly discovered repo must still be present");
assert!(
matches!(
new_entity.branch.settled(),
Some(Settled::Known {
value: _,
at: _,
stale: _
})
),
"a refresh naming the newly discovered repo's key must actually probe \
it and settle its branch cell, got {:?}",
new_entity.branch.settled()
);
}
#[test]
fn an_abandoned_discovery_stops_riding_later_refreshes() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let decoys = root.join("decoys");
for i in 0..4_000 {
fs::create_dir(decoys.join(format!("decoy-{i}")))
.or_else(|_| fs::create_dir_all(decoys.join(format!("decoy-{i}"))))
.expect("create decoy dir");
}
let (_tick_tx, tick_rx) = crossbeam_channel::unbounded::<Instant>();
let started = Core::start_for_test_with_discovery_abandon(
spec(vec![root.clone()]),
Duration::from_secs(3600),
Duration::from_micros(500),
tick_rx,
)
.discovered();
let core = started.core;
assert!(
core.discovery_manual_for_test(),
"walking 4,000 decoy directories against a 500 microsecond deadline \
must have abandoned and taken the Set manual"
);
fs::remove_dir_all(&decoys).expect("remove decoy directories");
init_repo_with_a_commit(&root.join("second"));
core.refresh(&[]);
let after = core.settle();
assert!(
!after
.entities
.iter()
.any(|entity| &*entity.name == "second"),
"once discovery has abandoned, a later refresh must not re-run it, so a \
repo created afterward, on a tree that would now resolve quickly, \
must still never appear"
);
}
#[test]
fn a_refresh_triggered_discovery_abandon_sets_manual_and_warns() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
init_repo_with_a_commit(&root.join("first"));
let (_tick_tx, tick_rx) = crossbeam_channel::unbounded::<Instant>();
let started = Core::start_for_test_with_discovery_abandon(
spec(vec![root.clone()]),
Duration::from_secs(3600),
Duration::from_secs(3600),
tick_rx,
)
.discovered();
let core = started.core;
assert!(
!core.discovery_manual_for_test(),
"an hour-long deadline must leave the first walk automatic"
);
let decoys = root.join("decoys");
for i in 0..4_000 {
fs::create_dir(decoys.join(format!("decoy-{i}")))
.or_else(|_| fs::create_dir_all(decoys.join(format!("decoy-{i}"))))
.expect("create decoy dir");
}
core.set_discovery_abandon_after_for_test(Duration::from_micros(500));
core.refresh(&[]);
core.wait_dispatched_for_test();
assert!(
core.discovery_manual_for_test(),
"refresh's own rerun_discovery must abandon against the newly-grown \
tree and take the Set manual, the same as an abandon at start does"
);
let warning = core.discovery_warning();
assert!(
warning
.as_deref()
.is_some_and(|message| message.starts_with("discovery: stopped at")),
"refresh's rerun_discovery must leave the abandoned-discovery warning \
behind, not merely flip the manual flag: got {warning:?}"
);
}
#[test]
fn a_fresh_core_over_different_roots_is_unaffected_by_another_cores_abandoned_discovery() {
let abandoned_dir = tempfile::tempdir().expect("temp dir");
let abandoned_root = root_of(&abandoned_dir);
init_repo_with_a_commit(&abandoned_root.join("first"));
let (_tick_tx, tick_rx) = crossbeam_channel::unbounded::<Instant>();
let started = Core::start_for_test_with_discovery_abandon(
spec(vec![abandoned_root]),
Duration::from_secs(3600),
Duration::ZERO,
tick_rx,
)
.discovered();
started.core.refresh(&[]);
started.core.settle();
assert!(
started.core.discovery_manual_for_test(),
"the zero-length abandon deadline must have already taken this Core manual"
);
drop(started.core);
let fresh_dir = tempfile::tempdir().expect("temp dir");
let fresh_root = root_of(&fresh_dir);
init_repo_with_a_commit(&fresh_root.join("first"));
let fresh_core = Core::start_discovered(spec(vec![fresh_root.clone()]));
assert_eq!(fresh_core.snapshot().entities.len(), 1);
init_repo_with_a_commit(&fresh_root.join("second"));
fresh_core.refresh(&[]);
let after = fresh_core.settle();
assert_eq!(
after.entities.len(),
2,
"a fresh Core, standing in for the Set's roots changing, must discover \
normally regardless of an earlier, unrelated Core having gone manual"
);
}
#[test]
fn dropping_the_core_joins_the_dedicated_thread_before_returning() {
let (tick_tx, tick_rx) = crossbeam_channel::unbounded::<Instant>();
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let started =
Core::start_for_test(spec(vec![root]), Duration::from_secs(3600), tick_rx).discovered();
assert!(started.clock_alive.load(Ordering::Acquire));
drop(started.core);
assert!(
!started.clock_alive.load(Ordering::Acquire),
"the dedicated thread should have exited, and cleared this flag, before drop returned"
);
drop(tick_tx);
}
#[test]
fn the_deadline_sweep_runs_only_when_a_tick_arrives() {
let (tick_tx, tick_rx) = crossbeam_channel::unbounded::<Instant>();
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let mut spec = spec(vec![root]);
spec.generation_deadline = Duration::ZERO;
let started = Core::start_for_test(spec, Duration::from_secs(3600), tick_rx).discovered();
let core = started.core;
let key = settle_launch(&core).entities[0].key.clone();
core.begin_untracked_probe_for_test(&key);
let before = core.snapshot();
assert!(
matches!(
before.entities[0].branch.settled(),
Some(Settled::Known {
value: _,
at: _,
stale: _
})
),
"the cell still holds launch's own answer here, so the Unknown below is the \
sweep's write rather than a cell that was already empty"
);
assert!(before.entities[0].branch.is_in_flight());
tick_tx.send(Instant::now()).expect("send one tick");
let after = core.settle();
assert!(matches!(
after.entities[0].branch.settled(),
Some(Settled::Unknown(Unknown::TimedOut))
));
}
#[test]
fn a_real_tick_through_the_dedicated_thread_reaches_the_poll_sweep_and_reprobes_a_moved_entity()
{
let (tick_tx, tick_rx) = crossbeam_channel::unbounded::<Instant>();
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let started =
Core::start_for_test(spec(vec![root]), Duration::from_secs(3600), tick_rx).discovered();
let core = started.core;
let key = core.snapshot().entities[0].key.clone();
backdate_polled_entries(&repo);
tick_tx
.send(Instant::now())
.expect("send the baseline tick");
wait_for(
"a tick sent on the real channel to reach the poll sweep",
|| core.poll_sweep_count_for_test() >= 1,
);
assert!(core.poll_reprobed_for_test().is_empty());
commit_a_change(&repo, "second");
tick_tx
.send(Instant::now())
.expect("send the movement tick");
wait_for(
"the real tick channel to reach the poll sweep and reprobe the moved entity",
|| core.poll_reprobed_for_test() == vec![key.clone()],
);
drop(tick_tx);
}
#[test]
fn poll_reprobe_touches_only_the_moved_entity_and_never_runs_a_status_probe() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo_a = root.join("repo-a");
let repo_b = root.join("repo-b");
init_repo_with_a_commit(&repo_a);
init_repo_with_a_commit(&repo_b);
let core = Core::start_discovered(spec(vec![root]));
let snapshot = core.snapshot();
let key_a = snapshot
.entities
.iter()
.find(|entity| entity.key.path() == repo_a)
.expect("repo-a discovered")
.key
.clone();
let key_b = snapshot
.entities
.iter()
.find(|entity| entity.key.path() == repo_b)
.expect("repo-b discovered")
.key
.clone();
core.refresh(&[key_a.clone(), key_b.clone()]);
let landed = core.settle();
let entity_of = |snapshot: &Snapshot, key: &EntityKey| {
snapshot
.entities
.iter()
.find(|entity| &entity.key == key)
.expect("entity present")
.clone()
};
let a_before = entity_of(&landed, &key_a);
let b_before = entity_of(&landed, &key_b);
let branch_at = |entity: &EntityState| match entity.branch.settled() {
Some(Settled::Known {
at,
value: _,
stale: _,
}) => *at,
other => panic!("expected a landed branch, got {other:?}"),
};
let dirty_state = |entity: &EntityState| match entity.dirty.settled() {
Some(Settled::Known { value, at, stale }) => (*value, *at, *stale),
other => panic!("expected a landed dirty count, got {other:?}"),
};
let (a_dirty_value_before, a_dirty_at_before, a_dirty_stale_before) =
dirty_state(&a_before);
assert!(
!a_dirty_stale_before,
"the fresh refresh must land dirty as not stale"
);
backdate_polled_entries(&repo_a);
backdate_polled_entries(&repo_b);
core.poll_once_for_test();
assert!(
core.poll_reprobed_for_test().is_empty(),
"a first sweep has nothing to compare against, so it must report no movement"
);
commit_a_change(&repo_a, "second");
core.poll_once_for_test();
assert_eq!(
core.poll_reprobed_for_test(),
vec![key_a.clone()],
"only the entity whose gitdir actually moved must be re-probed"
);
let after = core.snapshot();
let a_after = entity_of(&after, &key_a);
let b_after = entity_of(&after, &key_b);
assert_ne!(
branch_at(&a_after),
branch_at(&a_before),
"the moved entity's branch must carry a fresh timestamp from the re-probe"
);
let (a_dirty_value_after, a_dirty_at_after, a_dirty_stale_after) = dirty_state(&a_after);
assert_eq!(
a_dirty_value_after, a_dirty_value_before,
"no status probe ran, so dirty's value must be exactly what the last real refresh \
landed"
);
assert_eq!(
a_dirty_at_after, a_dirty_at_before,
"no status probe ran, so dirty's timestamp must be untouched, only its stale flag \
set"
);
assert!(
a_dirty_stale_after,
"the moved entity's dirty cell must go stale on poll evidence"
);
assert_eq!(
branch_at(&b_after),
branch_at(&b_before),
"the untouched entity's branch must be exactly as the prior refresh left it"
);
let (b_dirty_value_after, b_dirty_at_after, b_dirty_stale_after) = dirty_state(&b_after);
let (b_dirty_value_before, b_dirty_at_before, b_dirty_stale_before) =
dirty_state(&b_before);
assert_eq!(b_dirty_value_after, b_dirty_value_before);
assert_eq!(b_dirty_at_after, b_dirty_at_before);
assert_eq!(
b_dirty_stale_after, b_dirty_stale_before,
"an entity the sweep found unmoved must never go stale"
);
}
#[test]
fn poll_detects_an_attached_commit_through_index_while_head_itself_never_moves() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let core = Core::start_discovered(spec(vec![root]));
let key = core.snapshot().entities[0].key.clone();
backdate_polled_entries(&repo);
core.poll_once_for_test();
assert!(core.poll_reprobed_for_test().is_empty());
let head_path = repo.join(".git").join("HEAD");
let head_mtime_before = fs::metadata(&head_path)
.expect("stat HEAD")
.modified()
.expect("HEAD mtime");
commit_a_change(&repo, "second");
let head_mtime_after = fs::metadata(&head_path)
.expect("stat HEAD")
.modified()
.expect("HEAD mtime");
assert_eq!(
head_mtime_before, head_mtime_after,
"a commit on an attached HEAD must never touch HEAD itself"
);
core.poll_once_for_test();
assert_eq!(
core.poll_reprobed_for_test(),
vec![key],
"the poll must still detect the attached commit, through index rather than HEAD"
);
}
#[test]
fn poll_detects_a_detached_commit_through_the_per_worktree_head_file() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let parent = root.join("parent");
init_repo_with_a_commit(&parent);
let worktree_path = root.join("detached-worktree");
let status = Command::new("git")
.arg("-C")
.arg(&parent)
.args([
"worktree",
"add",
"--detach",
worktree_path.to_str().expect("utf8 path"),
])
.status()
.expect("run git worktree add");
assert!(status.success());
let core = Core::start_discovered(spec(vec![root]));
let snapshot = core.snapshot();
let worktree_key = snapshot
.entities
.iter()
.find(|entity| matches!(entity.kind, Kind::Worktree))
.expect("worktree discovered")
.key
.clone();
backdate_polled_entries(&parent);
backdate_polled_entries(&worktree_path);
core.poll_once_for_test();
assert!(core.poll_reprobed_for_test().is_empty());
let worktree_head_path = parent
.join(".git")
.join("worktrees")
.join("detached-worktree")
.join("HEAD");
let head_mtime_before = fs::metadata(&worktree_head_path)
.expect("stat the per-worktree HEAD")
.modified()
.expect("HEAD mtime");
commit_a_change(&worktree_path, "on the detached worktree");
let head_mtime_after = fs::metadata(&worktree_head_path)
.expect("stat the per-worktree HEAD")
.modified()
.expect("HEAD mtime");
assert_ne!(
head_mtime_before, head_mtime_after,
"a commit on a detached HEAD must write the new object id straight into its own \
HEAD file"
);
core.poll_once_for_test();
assert_eq!(
core.poll_reprobed_for_test(),
vec![worktree_key],
"the poll must detect the detached commit via the per-worktree HEAD file"
);
}
#[test]
fn snapshot_ages_a_freshly_landed_dirty_cell_stale_once_status_stale_after_has_elapsed() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let mut short_lived = spec(vec![root]);
short_lived.status_stale_after = Duration::from_nanos(1);
let core = Core::start_discovered(short_lived);
let key = core.snapshot().entities[0].key.clone();
core.refresh(std::slice::from_ref(&key));
core.settle();
let aged = core.snapshot();
match aged.entities[0].dirty.settled() {
Some(Settled::Known {
stale: true,
value: _,
at: _,
}) => {}
other => panic!(
"expected a landed dirty cell to have already aged past a one-nanosecond \
threshold, got {other:?}"
),
}
}
#[test]
fn snapshot_leaves_a_freshly_landed_dirty_cell_fresh_under_a_large_status_stale_after() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let core = Core::start_discovered(spec(vec![root]));
let key = core.snapshot().entities[0].key.clone();
core.refresh(std::slice::from_ref(&key));
core.settle();
let fresh = core.snapshot();
match fresh.entities[0].dirty.settled() {
Some(Settled::Known {
stale: false,
value: _,
at: _,
}) => {}
other => panic!("expected a freshly landed dirty cell to stay fresh, got {other:?}"),
}
}
#[test]
fn hidden_submodules_are_never_polled_but_shown_ones_are() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let parent = root.join("parent");
init_repo_with_a_commit(&parent);
fs::write(
parent.join(".gitmodules"),
"[submodule \"lib\"]\n\tpath = vendor/lib\n\turl = https://example.com/lib.git\n",
)
.expect("write .gitmodules");
let submodule_path = parent.join("vendor").join("lib");
init_repo_with_a_commit(&submodule_path);
let mut hidden_spec = spec(vec![root.clone()]);
hidden_spec.show_submodules = false;
let hidden_core = Core::start_discovered(hidden_spec);
let hidden_submodule_key = hidden_core
.snapshot()
.entities
.iter()
.find(|entity| matches!(entity.kind, Kind::Submodule))
.expect("the submodule is discovered regardless of show_submodules")
.key
.clone();
backdate_polled_entries(&submodule_path);
hidden_core.poll_once_for_test();
commit_a_change(&submodule_path, "into the hidden submodule");
hidden_core.poll_once_for_test();
assert!(
!hidden_core
.poll_reprobed_for_test()
.contains(&hidden_submodule_key),
"a hidden Submodule must never be re-probed by the poll, since it was never \
polled at all"
);
drop(hidden_core);
let mut shown_spec = spec(vec![root]);
shown_spec.show_submodules = true;
let shown_core = Core::start_discovered(shown_spec);
let submodule_key = shown_core
.snapshot()
.entities
.iter()
.find(|entity| matches!(entity.kind, Kind::Submodule))
.expect("the submodule is discovered regardless of show_submodules")
.key
.clone();
backdate_polled_entries(&submodule_path);
shown_core.poll_once_for_test();
commit_a_change(&submodule_path, "into the shown submodule");
shown_core.poll_once_for_test();
assert_eq!(
shown_core.poll_reprobed_for_test(),
vec![submodule_key],
"a shown Submodule must be polled and re-probed exactly like any other row"
);
}
#[test]
fn pause_cancels_every_in_flight_entity_and_releases_a_pending_settle() {
let (tick_tx, tick_rx) = crossbeam_channel::unbounded::<Instant>();
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let started =
Core::start_for_test(spec(vec![root]), Duration::from_secs(3600), tick_rx).discovered();
let core = started.core;
let key = settle_launch(&core).entities[0].key.clone();
let cancel = core.begin_untracked_probe_for_test(&key);
assert!(!cancel.load(Ordering::Acquire));
core.pause();
let settled = core.settle();
assert!(
cancel.load(Ordering::Acquire),
"pause should cancel the entity that was in flight"
);
assert!(settled.entities[0].branch.is_in_flight());
drop(tick_tx);
}
#[test]
fn a_launch_is_one_generation_over_every_row_its_own_walk_found() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
init_repo_with_a_commit(&root.join("first"));
init_repo_with_a_commit(&root.join("second"));
let (_core, launched) = started_and_settled(spec(vec![root]));
assert_eq!(
launched.generation,
Generation::default().successor(),
"a launch must settle on the first Generation a fresh `Core` mints; a second \
walk of the same tree would be a second Generation"
);
let mut named: Vec<String> = launched
.entities
.iter()
.filter(|entity| entity.branch.settled().is_some())
.map(|entity| entity.name.to_string())
.collect();
named.sort();
assert_eq!(
named,
vec!["first".to_string(), "second".to_string()],
"that one Generation must cover every row its own walk found, or the walk it \
saved would have to be paid by a second one"
);
}
#[test]
fn dropping_a_core_cancels_every_entity_it_still_has_in_flight() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
init_repo_with_a_commit(&root.join("repo"));
let (core, launched) = started_and_settled(spec(vec![root]));
let key = launched.entities[0].key.clone();
let cancel = core.begin_untracked_probe_for_test(&key);
assert!(!cancel.load(Ordering::Acquire));
drop(core);
assert!(
cancel.load(Ordering::Acquire),
"a dropped Core must cancel the Generation it still has in flight rather than \
leave it running against a Set nothing will read again"
);
}
#[test]
fn a_selection_scoped_refresh_supersedes_only_the_entity_it_covers() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
init_repo_with_a_commit(&root.join("a"));
init_repo_with_a_commit(&root.join("b"));
let (core, snapshot) = started_and_settled(spec(vec![root]));
let key_a = snapshot
.entities
.iter()
.find(|entity| &*entity.name == "a")
.expect("entity a discovered")
.key
.clone();
let key_b = snapshot
.entities
.iter()
.find(|entity| &*entity.name == "b")
.expect("entity b discovered")
.key
.clone();
let older = core.begin_shared_generation_for_test(&[key_a.clone(), key_b.clone()]);
let newer = core.refresh(std::slice::from_ref(&key_a));
assert_eq!(
newer,
older.generation.successor(),
"the Selection-scoped refresh must be the Generation immediately after the one \
still in flight, with nothing minted in between"
);
core.wait_dispatched_for_test();
assert!(
older.cancels[&key_a].load(Ordering::Acquire),
"the entity the new Generation covers must have its old interrupt flag set"
);
assert!(
!older.cancels[&key_b].load(Ordering::Acquire),
"an entity the new Generation does not cover must be left running, untouched"
);
let after_refresh = core.settle();
let a_after_gen2 = after_refresh
.entities
.iter()
.find(|entity| entity.key == key_a)
.expect("entity a present");
assert!(
matches!(
a_after_gen2.branch.settled(),
Some(Settled::Known {
value: Head::Branch { .. },
at: _,
stale: _
})
),
"the newer Generation's real probe should have written A's cell by now"
);
core.apply_probe_result_for_test(
&key_a,
older.generation,
Settled::Known {
value: Head::Branch {
name: Arc::from("stale-from-generation-one"),
commit: gix::hash::Kind::Sha1.null(),
},
at: Timestamp::now(),
stale: false,
},
);
let after_stale_write = core.snapshot();
let a_final = after_stale_write
.entities
.iter()
.find(|entity| entity.key == key_a)
.expect("entity a present");
match a_final.branch.settled() {
Some(Settled::Known {
value: Head::Branch { name, .. },
at: _,
stale: _,
}) => assert_ne!(
&**name, "stale-from-generation-one",
"a lower-Generation result must be dropped at the cell it would write"
),
other => panic!("expected A to still hold the newer Generation's value, got {other:?}"),
}
core.apply_probe_result_for_test(
&key_b,
older.generation,
Settled::Known {
value: Head::Branch {
name: Arc::from("b-generation-one-result"),
commit: gix::hash::Kind::Sha1.null(),
},
at: Timestamp::now(),
stale: false,
},
);
let final_snapshot = core.snapshot();
let b_final = final_snapshot
.entities
.iter()
.find(|entity| entity.key == key_b)
.expect("entity b present");
match b_final.branch.settled() {
Some(Settled::Known {
value: Head::Branch { name, .. },
at: _,
stale: _,
}) => assert_eq!(
&**name, "b-generation-one-result",
"an entity the new Generation never covered must still accept its own result"
),
other => {
panic!("expected B's un-superseded older result to be accepted, got {other:?}")
}
}
}
#[test]
fn the_deadline_sweep_keeps_already_settled_cells_and_only_times_out_what_is_still_loading() {
let (tick_tx, tick_rx) = crossbeam_channel::unbounded::<Instant>();
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
init_repo_with_a_commit(&root.join("a"));
init_repo_with_a_commit(&root.join("b"));
let mut spec = spec(vec![root]);
spec.generation_deadline = Duration::ZERO;
let started = Core::start_for_test(spec, Duration::from_secs(3600), tick_rx).discovered();
let core = started.core;
let snapshot = settle_launch(&core);
let key_a = snapshot
.entities
.iter()
.find(|entity| &*entity.name == "a")
.expect("entity a discovered")
.key
.clone();
let key_b = snapshot
.entities
.iter()
.find(|entity| &*entity.name == "b")
.expect("entity b discovered")
.key
.clone();
let a_settled = core.probe_now(&key_a);
let a_value_before = match a_settled.branch.settled() {
Some(Settled::Known {
value: Head::Branch { name, .. },
at: _,
stale: _,
}) => Arc::clone(name),
other => panic!("expected A's synchronous probe to settle a branch, got {other:?}"),
};
let cancel_b = core.begin_untracked_probe_for_test(&key_b);
let before_tick = core.snapshot();
let b_before = before_tick
.entities
.iter()
.find(|entity| entity.key == key_b)
.expect("entity b present");
assert!(
b_before.branch.is_in_flight(),
"B must be mid-flight when the sweep fires; that is the only shape the sweep \
may touch"
);
assert!(
matches!(
b_before.branch.settled(),
Some(Settled::Known {
value: _,
at: _,
stale: _
})
),
"B still carries launch's own answer here, so the Unknown below is a write the \
sweep made rather than a cell that was already empty, got {:?}",
b_before.branch.settled()
);
tick_tx.send(Instant::now()).expect("send one tick");
let after_sweep = core.settle();
let a_after = after_sweep
.entities
.iter()
.find(|entity| entity.key == key_a)
.expect("entity a present");
match a_after.branch.settled() {
Some(Settled::Known {
value: Head::Branch { name, .. },
at: _,
stale: _,
}) => assert_eq!(
name, &a_value_before,
"an already-settled cell must keep its value when the deadline sweep runs, not be blanked"
),
other => panic!("expected A's settled value to survive the sweep, got {other:?}"),
}
let b_after = after_sweep
.entities
.iter()
.find(|entity| entity.key == key_b)
.expect("entity b present");
assert!(matches!(
b_after.branch.settled(),
Some(Settled::Unknown(Unknown::TimedOut))
));
assert!(
!cancel_b.load(Ordering::Acquire),
"the deadline sweep marks a cell Unknown; it never sets the entity's own \
cancel flag, since the underlying probe (nonexistent here) is left to keep running"
);
}
#[test]
fn the_deadline_sweep_times_out_a_worktrees_outstanding_state_but_leaves_a_repos_not_applicable_one_alone()
{
let (tick_tx, tick_rx) = crossbeam_channel::unbounded::<Instant>();
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let parent = root.join("parent");
init_repo_with_a_commit(&parent);
let worktree_path = root.join("feature-worktree");
git(
&parent,
&[
"worktree",
"add",
"-b",
"feature",
worktree_path.to_str().expect("utf8 path"),
],
);
let mut spec = spec(vec![root]);
spec.generation_deadline = Duration::ZERO;
let started = Core::start_for_test(spec, Duration::from_secs(3600), tick_rx).discovered();
let core = started.core;
let snapshot = settle_launch(&core);
let repo_key = snapshot
.entities
.iter()
.find(|entity| matches!(entity.kind, Kind::Repo))
.expect("repo entity present")
.key
.clone();
let worktree_key = snapshot
.entities
.iter()
.find(|entity| matches!(entity.kind, Kind::Worktree))
.expect("worktree entity present")
.key
.clone();
core.begin_untracked_probe_for_test(&repo_key);
core.begin_untracked_probe_for_test(&worktree_key);
tick_tx.send(Instant::now()).expect("send one tick");
let after_sweep = core.settle();
let worktree_after = after_sweep
.entities
.iter()
.find(|entity| entity.key == worktree_key)
.expect("worktree entity present");
assert!(
matches!(
worktree_after.state.settled(),
Some(Settled::Unknown(Unknown::TimedOut))
),
"expected the outstanding state cell to time out, got {:?}",
worktree_after.state.settled()
);
let repo_after = after_sweep
.entities
.iter()
.find(|entity| entity.key == repo_key)
.expect("repo entity present");
assert!(
matches!(repo_after.state.settled(), Some(Settled::NotApplicable)),
"a Repo's Not applicable state must survive the sweep untouched, got {:?}",
repo_after.state.settled()
);
}
#[test]
fn the_deadline_sweeps_poll_never_touches_an_entitys_action_receipt() {
let (tick_tx, tick_rx) = crossbeam_channel::unbounded::<Instant>();
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let mut spec = spec(vec![root]);
spec.generation_deadline = Duration::ZERO;
let started = Core::start_for_test(spec, Duration::from_secs(3600), tick_rx).discovered();
let core = started.core;
let key = settle_launch(&core).entities[0].key.clone();
let receipt = crate::entity::ActionReceipt {
label: Arc::from("reinstall"),
steps: Arc::from(vec![crate::entity::StepResult {
label: Arc::from("pnpm install"),
outcome: crate::entity::StepOutcome::Ok,
output: Arc::from(&b""[..]),
elapsed: Duration::from_millis(1),
elision: None,
shell: false,
interactive: false,
}]),
skip: None,
finished_at: Timestamp::now(),
running: None,
};
core.set_last_action_for_test(&key, receipt.clone());
core.begin_untracked_probe_for_test(&key);
tick_tx.send(Instant::now()).expect("send one tick");
let after = core.settle();
let entity = after
.entities
.iter()
.find(|entity| entity.key == key)
.expect("entity present");
assert!(
matches!(
entity.branch.settled(),
Some(Settled::Unknown(Unknown::TimedOut))
),
"sanity check: the sweep must have actually timed out the in-flight cell, got {:?}",
entity.branch.settled()
);
assert_eq!(entity.last_action, Some(receipt));
}
#[test]
fn a_cancelled_probe_never_opens_the_repository_at_all() {
let cancel = AtomicBool::new(true);
let outcome = probe_branch(
Path::new("/nonexistent/nowhere-at-all"),
None,
Kind::Repo,
&cancel,
);
assert!(
outcome.is_none(),
"a probe observing cancellation before its first read must do no work \
at all, not attempt the read and fail having tried it"
);
}
#[test]
fn classify_status_result_drops_an_error_once_cancel_reads_true() {
let cancel = AtomicBool::new(true);
let outcome = classify_status_result(
Err(crate::git::ProbeError::Status(Arc::from("boom"))),
&cancel,
);
assert!(
outcome.is_none(),
"an error alongside a cancel flag already set must read as cancelled, not \
Failed, got {outcome:?}"
);
}
#[test]
fn classify_status_result_settles_failed_when_cancel_never_fired() {
let cancel = AtomicBool::new(false);
let outcome = classify_status_result(
Err(crate::git::ProbeError::Status(Arc::from("boom"))),
&cancel,
);
assert!(
matches!(outcome, Some(Settled::Failed(git::ProbeError::Status(_)))),
"a genuine error with no cancellation must settle Failed, got {outcome:?}"
);
}
#[test]
fn classify_status_result_drops_an_ok_once_cancel_reads_true() {
let cancel = AtomicBool::new(true);
let outcome = classify_status_result(Ok(DirtyCounts::default()), &cancel);
assert!(
outcome.is_none(),
"an Ok value that raced ahead of a cancel flag now set must read as cancelled, \
not be settled Known, got {outcome:?}"
);
}
#[test]
fn classify_status_result_settles_known_when_cancel_never_fired() {
let cancel = AtomicBool::new(false);
let counts = DirtyCounts {
modified: 1,
untracked: 2,
deleted: 3,
};
let outcome = classify_status_result(Ok(counts), &cancel);
assert!(
matches!(
outcome,
Some(Settled::Known {
value,
at: _,
stale: _
}) if value == counts
),
"a genuine completed read with no cancellation must settle Known, got {outcome:?}"
);
}
#[test]
fn a_linked_worktree_is_its_own_entity_and_never_doubles_as_a_repo() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let parent = root.join("parent");
init_repo_with_a_commit(&parent);
let worktree_path = root.join("feature-worktree");
let status = Command::new("git")
.arg("-C")
.arg(&parent)
.args([
"worktree",
"add",
"-b",
"feature",
worktree_path.to_str().expect("utf8 path"),
])
.status()
.expect("run git worktree add");
assert!(status.success());
let core = Core::start_discovered(spec(vec![root]));
let snapshot = core.snapshot();
assert_eq!(
snapshot.entities.len(),
2,
"expected the parent plus one Worktree, not two Repos"
);
let repo_count = snapshot
.entities
.iter()
.filter(|entity| matches!(entity.kind, Kind::Repo))
.count();
let worktree_count = snapshot
.entities
.iter()
.filter(|entity| matches!(entity.kind, Kind::Worktree))
.count();
assert_eq!(
repo_count, 1,
"the parent must be counted as exactly one Repo"
);
assert_eq!(
worktree_count, 1,
"the linked worktree must be counted as exactly one Worktree"
);
let worktree_entity = snapshot
.entities
.iter()
.find(|entity| matches!(entity.kind, Kind::Worktree))
.expect("worktree entity present");
let repo_entity = snapshot
.entities
.iter()
.find(|entity| matches!(entity.kind, Kind::Repo))
.expect("repo entity present");
assert_eq!(worktree_entity.common_dir, repo_entity.common_dir);
let repo_branch = core.probe_now(&repo_entity.key);
let worktree_branch = core.probe_now(&worktree_entity.key);
match (
repo_branch.branch.settled(),
worktree_branch.branch.settled(),
) {
(
Some(Settled::Known {
value:
Head::Branch {
name: repo_name, ..
},
at: _,
stale: _,
}),
Some(Settled::Known {
value:
Head::Branch {
name: worktree_name,
..
},
at: _,
stale: _,
}),
) => {
assert_ne!(repo_name, worktree_name);
assert_eq!(&**worktree_name, "feature");
}
other => panic!("expected both entities to read an attached branch, got {other:?}"),
}
}
#[test]
fn a_worktrees_branch_that_is_an_ancestor_of_the_default_branch_reads_merged_after_a_refresh() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let parent = root.join("parent");
init_repo_with_a_commit(&parent);
git(
&parent,
&[
"remote",
"add",
"origin",
"https://example.invalid/repo.git",
],
);
let sha = head_sha(&parent);
git(&parent, &["update-ref", "refs/remotes/origin/main", &sha]);
let worktree_path = root.join("feature-worktree");
git(
&parent,
&[
"worktree",
"add",
"-b",
"feature",
worktree_path.to_str().expect("utf8 path"),
],
);
let core = Core::start_discovered(spec(vec![root]));
let keys: Vec<EntityKey> = core
.snapshot()
.entities
.iter()
.map(|entity| entity.key.clone())
.collect();
core.refresh(&keys);
let settled = core.settle();
let worktree_entity = settled
.entities
.iter()
.find(|entity| matches!(entity.kind, Kind::Worktree))
.expect("worktree entity present");
assert!(
matches!(
worktree_entity.state.settled(),
Some(Settled::Known {
value: WorktreeState::Merged,
at: _,
stale: _
})
),
"expected the worktree, at the same commit as the default branch, to read Merged, got {:?}",
worktree_entity.state.settled()
);
}
#[test]
fn a_squash_merged_worktree_branch_reads_merged_after_a_refresh() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let parent = root.join("parent");
init_repo_with_a_commit(&parent);
git(
&parent,
&[
"remote",
"add",
"origin",
"https://example.invalid/repo.git",
],
);
let worktree_path = root.join("feature-worktree");
git(
&parent,
&[
"worktree",
"add",
"-b",
"feature",
worktree_path.to_str().expect("utf8 path"),
],
);
fs::write(worktree_path.join("a.txt"), "one\n").expect("write a.txt");
git(&worktree_path, &["add", "a.txt"]);
git(&worktree_path, &["commit", "-m", "add a"]);
fs::write(worktree_path.join("b.txt"), "two\n").expect("write b.txt");
git(&worktree_path, &["add", "b.txt"]);
git(&worktree_path, &["commit", "-m", "add b"]);
let feature_sha = head_sha(&worktree_path);
git(&parent, &["merge", "--squash", "feature"]);
git(&parent, &["commit", "-m", "squashed feature"]);
let main_sha = head_sha(&parent);
git(
&parent,
&["update-ref", "refs/remotes/origin/main", &main_sha],
);
git(&parent, &["config", "branch.feature.remote", "origin"]);
git(
&parent,
&["config", "branch.feature.merge", "refs/heads/feature"],
);
git(
&parent,
&["update-ref", "refs/remotes/origin/feature", &feature_sha],
);
let core = Core::start_discovered(spec(vec![root]));
let keys: Vec<EntityKey> = core
.snapshot()
.entities
.iter()
.map(|entity| entity.key.clone())
.collect();
core.refresh(&keys);
let settled = core.settle();
let worktree_entity = settled
.entities
.iter()
.find(|entity| matches!(entity.kind, Kind::Worktree))
.expect("worktree entity present");
assert!(
matches!(
worktree_entity.state.settled(),
Some(Settled::Known {
value: WorktreeState::Merged,
at: _,
stale: _
})
),
"expected a squash-merged worktree branch to read Merged, got {:?}",
worktree_entity.state.settled()
);
}
#[test]
fn patch_equivalence_never_runs_for_an_entity_ancestry_already_settled() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let parent = root.join("parent");
init_repo_with_a_commit(&parent);
git(
&parent,
&[
"remote",
"add",
"origin",
"https://example.invalid/repo.git",
],
);
let sha = head_sha(&parent);
git(&parent, &["update-ref", "refs/remotes/origin/main", &sha]);
let worktree_path = root.join("feature-worktree");
git(
&parent,
&[
"worktree",
"add",
"-b",
"feature",
worktree_path.to_str().expect("utf8 path"),
],
);
let (core, launched) = started_and_settled(spec(vec![root]));
let keys: Vec<EntityKey> = launched
.entities
.iter()
.map(|entity| entity.key.clone())
.collect();
core.refresh(&keys);
let settled = core.settle();
let worktree_entity = settled
.entities
.iter()
.find(|entity| matches!(entity.kind, Kind::Worktree))
.expect("worktree entity present");
assert!(
matches!(
worktree_entity.state.settled(),
Some(Settled::Known {
value: WorktreeState::Merged,
at: _,
stale: _
})
),
"expected ancestry alone to settle Merged here, got {:?}",
worktree_entity.state.settled()
);
assert_eq!(
core.patch_identity_reads_for_test(),
0,
"ancestry already settled this entity, so patch equivalence's shared \
scan must never run for its common dir at all"
);
}
#[test]
fn a_full_refresh_reaching_patch_equivalence_writes_no_loose_objects() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let parent = root.join("parent");
init_repo_with_a_commit(&parent);
git(
&parent,
&[
"remote",
"add",
"origin",
"https://example.invalid/repo.git",
],
);
let worktree_path = root.join("feature-worktree");
git(
&parent,
&[
"worktree",
"add",
"-b",
"feature",
worktree_path.to_str().expect("utf8 path"),
],
);
fs::write(worktree_path.join("a.txt"), "one\n").expect("write a.txt");
git(&worktree_path, &["add", "a.txt"]);
git(&worktree_path, &["commit", "-m", "add a"]);
fs::write(worktree_path.join("b.txt"), "two\n").expect("write b.txt");
git(&worktree_path, &["add", "b.txt"]);
git(&worktree_path, &["commit", "-m", "add b"]);
let feature_sha = head_sha(&worktree_path);
git(&parent, &["merge", "--squash", "feature"]);
git(&parent, &["commit", "-m", "squashed feature"]);
let main_sha = head_sha(&parent);
git(
&parent,
&["update-ref", "refs/remotes/origin/main", &main_sha],
);
git(&parent, &["config", "branch.feature.remote", "origin"]);
git(
&parent,
&["config", "branch.feature.merge", "refs/heads/feature"],
);
git(
&parent,
&["update-ref", "refs/remotes/origin/feature", &feature_sha],
);
let core = Core::start_discovered(spec(vec![root]));
let keys: Vec<EntityKey> = core
.snapshot()
.entities
.iter()
.map(|entity| entity.key.clone())
.collect();
let before = loose_object_count(&parent);
core.refresh(&keys);
let settled = core.settle();
let after = loose_object_count(&parent);
let worktree_entity = settled
.entities
.iter()
.find(|entity| matches!(entity.kind, Kind::Worktree))
.expect("worktree entity present");
assert!(
matches!(
worktree_entity.state.settled(),
Some(Settled::Known {
value: WorktreeState::Merged,
at: _,
stale: _
})
),
"expected this refresh to actually reach patch equivalence and settle \
Merged, got {:?}",
worktree_entity.state.settled()
);
assert_eq!(
before, after,
"a full refresh reaching patch equivalence must never write a loose \
object to the repository"
);
}
#[test]
fn a_diverged_worktree_with_a_live_upstream_and_genuinely_unmerged_work_settles_active_after_a_refresh()
{
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let parent = root.join("parent");
init_repo_with_a_commit(&parent);
let base_sha = head_sha(&parent);
git(
&parent,
&[
"remote",
"add",
"origin",
"https://example.invalid/repo.git",
],
);
git(
&parent,
&["update-ref", "refs/remotes/origin/main", &base_sha],
);
let worktree_path = root.join("feature-worktree");
git(
&parent,
&[
"worktree",
"add",
"-b",
"feature",
worktree_path.to_str().expect("utf8 path"),
],
);
fs::write(worktree_path.join("feature.txt"), "unmerged work\n").expect("write feature.txt");
git(&worktree_path, &["add", "feature.txt"]);
git(&worktree_path, &["commit", "-m", "unmerged"]);
let feature_sha = head_sha(&worktree_path);
git(&parent, &["config", "branch.feature.remote", "origin"]);
git(
&parent,
&["config", "branch.feature.merge", "refs/heads/feature"],
);
git(
&parent,
&["update-ref", "refs/remotes/origin/feature", &feature_sha],
);
let core = Core::start_discovered(spec(vec![root]));
let keys: Vec<EntityKey> = core
.snapshot()
.entities
.iter()
.map(|entity| entity.key.clone())
.collect();
core.refresh(&keys);
let settled = core.settle();
let worktree_entity = settled
.entities
.iter()
.find(|entity| matches!(entity.kind, Kind::Worktree))
.expect("worktree entity present");
assert!(
matches!(
worktree_entity.state.settled(),
Some(Settled::Known {
value: WorktreeState::Active,
at: _,
stale: _
})
),
"expected genuinely unmerged work with a live upstream to settle Active, got {:?}",
worktree_entity.state.settled()
);
}
#[test]
fn a_submodule_is_in_the_snapshot_even_though_hidden_by_the_default_preference() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let parent = root.join("parent");
init_repo_with_a_commit(&parent);
fs::write(
parent.join(".gitmodules"),
"[submodule \"lib\"]\n\tpath = vendor/lib\n\turl = https://example.com/lib.git\n",
)
.expect("write .gitmodules");
fs::create_dir_all(parent.join("vendor").join("lib")).expect("create submodule dir");
let core = Core::start_discovered(spec(vec![root]));
let snapshot = core.snapshot();
assert!(
snapshot
.entities
.iter()
.any(|entity| matches!(entity.kind, Kind::Submodule)),
"a discovered Submodule must be in the snapshot even while show_submodules is off"
);
}
#[test]
fn a_submodules_state_and_base_cells_stay_unknown_through_a_real_refresh() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let parent = root.join("parent");
init_repo_with_a_commit(&parent);
fs::write(
parent.join(".gitmodules"),
"[submodule \"lib\"]\n\tpath = vendor/lib\n\turl = https://example.com/lib.git\n",
)
.expect("write .gitmodules");
let submodule = parent.join("vendor").join("lib");
init_repo_with_a_commit(&submodule);
git(
&submodule,
&["remote", "add", "origin", "https://example.invalid/lib.git"],
);
let root_sha = head_sha(&submodule);
git(&submodule, &["commit", "--allow-empty", "-m", "second"]);
let tip_sha = head_sha(&submodule);
git(&submodule, &["reset", "--hard", &root_sha]);
git(
&submodule,
&["update-ref", "refs/remotes/origin/main", &tip_sha],
);
let mut core_spec = spec(vec![root]);
core_spec.show_submodules = true;
let core = Core::start_discovered(core_spec);
let key = core
.snapshot()
.entities
.iter()
.find(|entity| matches!(entity.kind, Kind::Submodule))
.expect("a discovered Submodule")
.key
.clone();
core.refresh(std::slice::from_ref(&key));
let settled = core.settle();
let submodule_entity = settled
.entities
.iter()
.find(|entity| entity.key == key)
.expect("the Submodule entity");
assert!(
matches!(
submodule_entity.base.settled(),
Some(Settled::Unknown(Unknown::NoDefaultBranch))
),
"expected a Submodule's base to stay Unknown through a real refresh, \
got {:?}",
submodule_entity.base.settled()
);
assert!(
matches!(
submodule_entity.state.settled(),
Some(Settled::Unknown(Unknown::NoDefaultBranch))
),
"expected a Submodule's state to stay Unknown through a real refresh, \
rather than settling Merged off an untrusted default branch, got {:?}",
submodule_entity.state.settled()
);
}
#[test]
fn a_submodules_entity_name_is_its_relative_path_not_its_basename() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let parent = root.join("parent");
init_repo_with_a_commit(&parent);
fs::write(
parent.join(".gitmodules"),
"[submodule \"lib\"]\n\tpath = vendor/lib\n\turl = https://example.com/lib.git\n",
)
.expect("write .gitmodules");
fs::create_dir_all(parent.join("vendor").join("lib")).expect("create submodule dir");
let core = Core::start_discovered(spec(vec![root]));
let submodule = core
.snapshot()
.entities
.into_iter()
.find(|entity| matches!(entity.kind, Kind::Submodule))
.expect("a discovered Submodule");
assert_eq!(
submodule.name.as_ref(),
"vendor/lib",
"expected the declared relative path, not the basename `lib`"
);
}
#[test]
fn an_uninitialised_submodules_probed_cells_settle_unknown_not_failed() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let parent = root.join("parent");
init_repo_with_a_commit(&parent);
fs::write(
parent.join(".gitmodules"),
"[submodule \"lib\"]\n\tpath = vendor/lib\n\turl = https://example.com/lib.git\n",
)
.expect("write .gitmodules");
let mut core_spec = spec(vec![root]);
core_spec.show_submodules = true;
let core = Core::start_discovered(core_spec);
let key = core
.snapshot()
.entities
.iter()
.find(|entity| matches!(entity.kind, Kind::Submodule))
.expect("a discovered Submodule")
.key
.clone();
core.refresh(std::slice::from_ref(&key));
let settled = core.settle();
let submodule = settled
.entities
.iter()
.find(|entity| entity.key == key)
.expect("the Submodule entity");
assert!(
matches!(
submodule.branch.settled(),
Some(Settled::Unknown(Unknown::SubmoduleUninitialized))
),
"expected branch to settle Unknown(SubmoduleUninitialized), got {:?}",
submodule.branch.settled()
);
assert!(
matches!(
submodule.sync.settled(),
Some(Settled::Unknown(Unknown::SubmoduleUninitialized))
),
"expected sync to settle Unknown(SubmoduleUninitialized), got {:?}",
submodule.sync.settled()
);
assert!(
matches!(
submodule.dirty.settled(),
Some(Settled::Unknown(Unknown::SubmoduleUninitialized))
),
"expected dirty to settle Unknown(SubmoduleUninitialized), got {:?}",
submodule.dirty.settled()
);
assert_eq!(
summary(submodule),
RowSummary::Unknown,
"expected the row's own gutter fold to read Unknown, not Failed"
);
}
#[test]
fn dispatch_skips_probing_a_hidden_submodule_while_probing_the_same_one_shown() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let parent = root.join("parent");
init_repo_with_a_commit(&parent);
fs::write(
parent.join(".gitmodules"),
"[submodule \"lib\"]\n\tpath = vendor/lib\n\turl = https://example.com/lib.git\n",
)
.expect("write .gitmodules");
init_repo_with_a_commit(&parent.join("vendor").join("lib"));
let core = Core::start_discovered(spec(vec![root]));
let key = core
.snapshot()
.entities
.iter()
.find(|entity| matches!(entity.kind, Kind::Submodule))
.expect("a discovered Submodule")
.key
.clone();
core.refresh(std::slice::from_ref(&key));
let while_hidden = core.settle();
let hidden_entity = while_hidden
.entities
.iter()
.find(|entity| entity.key == key)
.expect("submodule entity");
assert!(
hidden_entity.branch.settled().is_none(),
"a Submodule dispatched while hidden must never even reach probe_branch, \
so its cell stays never-settled rather than holding any value at all, got {:?}",
hidden_entity.branch.settled()
);
core.set_show_submodules(true);
core.refresh(std::slice::from_ref(&key));
let while_shown = core.settle();
let shown_entity = while_shown
.entities
.iter()
.find(|entity| entity.key == key)
.expect("submodule entity");
assert!(
matches!(
shown_entity.branch.settled(),
Some(Settled::Known {
value: _,
at: _,
stale: _
})
),
"expected the same Submodule's branch to settle a real value once shown, got {:?}",
shown_entity.branch.settled()
);
}
#[test]
fn toggling_show_submodules_starts_no_new_generation_and_dispatches_nothing() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
init_repo_with_a_commit(&root.join("repo-a"));
let (core, launched) = started_and_settled(spec(vec![root]));
let before = launched.generation;
let dispatched_before = core.dispatch_log_for_test();
assert!(
!dispatched_before.is_empty(),
"launch dispatched nothing, so the comparison below would hold however much a \
toggle dispatched"
);
core.set_show_submodules(true);
core.set_show_submodules(false);
assert_eq!(
core.snapshot().generation,
before,
"toggling show_submodules must start no Generation of its own"
);
assert_eq!(
core.dispatch_log_for_test(),
dispatched_before,
"toggling show_submodules must dispatch no probe of its own, leaving the last \
Generation's own log exactly as it found it"
);
}
#[test]
fn a_malformed_gitmodules_file_still_fails_the_parent_while_submodules_are_hidden() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let parent = root.join("parent");
init_repo_with_a_commit(&parent);
fs::write(
parent.join(".gitmodules"),
"[submodule \"lib\"\n\tpath = lib\n",
)
.expect("write malformed .gitmodules");
let core = Core::start_discovered(spec(vec![root]));
let key = core
.snapshot()
.entities
.iter()
.find(|entity| entity.key.path() == parent)
.expect("the parent entity")
.key
.clone();
core.refresh(std::slice::from_ref(&key));
let settled = core.settle();
let parent_entity = settled
.entities
.iter()
.find(|entity| entity.key == key)
.expect("the parent entity");
assert_eq!(
summary(parent_entity),
RowSummary::Failed,
"expected the parent to fold Failed even with Submodules hidden"
);
assert!(
parent_entity.diagnostics.gitmodules_failed.is_some(),
"expected the failure recorded in Diagnostics for the detail pane"
);
assert!(
!settled
.entities
.iter()
.any(|entity| matches!(entity.kind, Kind::Submodule)),
"an unparseable .gitmodules yields no Submodule rows for that parent"
);
}
#[test]
fn count_matches_a_plain_discoverys_entity_count() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
init_repo_with_a_commit(&root.join("one"));
init_repo_with_a_commit(&root.join("two"));
let set = SetSpec {
name: "test".to_string(),
roots: vec![root],
include: Vec::new(),
exclude: Vec::new(),
};
assert_eq!(discovery::count(&set), 2);
}
#[test]
fn the_slow_discovery_watcher_warns_with_the_count_reached_and_the_roots() {
let progress = Arc::new(AtomicUsize::new(42));
let finished = Arc::new(AtomicBool::new(false));
let roots = vec![PathBuf::from("/repos/a"), PathBuf::from("/repos/b")];
let warning = watch_for_slow_discovery(progress, finished, roots, Duration::from_millis(1));
let message = warning.expect("a walk that has not finished should warn");
assert!(message.contains("42"));
assert!(message.contains("/repos/a"));
assert!(message.contains("/repos/b"));
}
#[test]
fn the_slow_discovery_watcher_is_silent_once_the_walk_has_already_finished() {
let progress = Arc::new(AtomicUsize::new(7));
let finished = Arc::new(AtomicBool::new(true));
let warning =
watch_for_slow_discovery(progress, finished, Vec::new(), Duration::from_millis(1));
assert!(warning.is_none());
}
#[test]
fn a_fast_discovery_leaves_no_warning_once_the_watcher_has_run() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
init_repo_with_a_commit(&root.join("repo"));
let (_tick_tx, tick_rx) = crossbeam_channel::unbounded::<Instant>();
let started =
Core::start_for_test(spec(vec![root]), Duration::from_secs(1), tick_rx).discovered();
started
.discovery_watcher
.join()
.expect("watcher thread should not panic");
assert!(started.core.discovery_warning().is_none());
}
fn gate_opened_on_signal(open: bool) -> (DiscoveryGate, Sender<()>, JoinHandle<()>) {
let gate: DiscoveryGate = Arc::new((Mutex::new(open), Condvar::new()));
let (returned_tx, returned_rx) = crossbeam_channel::bounded::<()>(1);
let opener = thread::spawn({
let gate = Arc::clone(&gate);
move || {
let _ = returned_rx.recv_timeout(crate::liveness::BACKSTOP);
set_discovery_gate(&gate, true);
}
});
(gate, returned_tx, opener)
}
#[test]
fn start_returns_against_an_empty_table_and_the_rows_land_when_discovery_does() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let (_tick_tx, tick_rx) = crossbeam_channel::unbounded::<Instant>();
let (gate, start_returned, opener) = gate_opened_on_signal(false);
let started = Core::start_for_test_gated(
spec(vec![root]),
Duration::from_secs(3600),
discovery::ABANDON_AFTER,
tick_rx,
Some(Arc::clone(&gate)),
);
let at_start = started.core.snapshot();
let key = EntityKey::new(Arc::from(repo.as_path()));
started.core.hold_phase_c_for_test(&key);
start_returned.send(()).expect("the opener is listening");
opener.join().expect("the opener thread should not panic");
let started = started.discovered();
assert!(
at_start.entities.is_empty(),
"`Core::start` must return before discovery has finished, against the empty \
table a consumer draws its first frame from, got {:?}",
at_start
.entities
.iter()
.map(|entity| entity.name.to_string())
.collect::<Vec<_>>()
);
let landed = started.core.snapshot();
assert_eq!(
landed
.entities
.iter()
.map(|entity| entity.name.to_string())
.collect::<Vec<_>>(),
vec!["repo".to_string()],
"the row must land on the table as soon as discovery does"
);
assert!(
landed.entities[0].dirty.settled().is_none() && landed.entities[0].dirty.is_in_flight(),
"discovery lands the row alone: launch's own Generation is already covering it \
and its Cells stay unsettled until that Generation answers, which is what the \
spinner sits behind"
);
started.core.release_phase_c_for_test(&key);
started.core.wait_phase_c_finished_for_test(&key);
}
#[test]
fn refresh_all_covers_every_row_its_own_discovery_found() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
init_repo_with_a_commit(&root.join("repo"));
let (core, launched) = started_and_settled(spec(vec![root.clone()]));
assert_eq!(
launched
.entities
.iter()
.map(|entity| entity.name.to_string())
.collect::<Vec<_>>(),
vec!["repo".to_string()],
"launch's own walk must have landed and covered exactly the one row that \
existed when it ran"
);
init_repo_with_a_commit(&root.join("late"));
assert_eq!(
core.refresh_all(),
launched.generation.successor(),
"`refresh_all` must be the Generation immediately after the one already on the \
table"
);
let settled = core.settle();
let mut named: Vec<String> = settled
.entities
.iter()
.filter(|entity| entity.branch.settled().is_some())
.map(|entity| entity.name.to_string())
.collect();
named.sort();
assert_eq!(
named,
vec!["late".to_string(), "repo".to_string()],
"the Generation must cover every row its own discovery found, including one the \
caller had no key for"
);
}
#[test]
fn refresh_returns_before_its_own_generations_discovery_has_run() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
init_repo_with_a_commit(&root.join("repo"));
let (_tick_tx, tick_rx) = crossbeam_channel::unbounded::<Instant>();
let (gate, walk_may_run, opener) = gate_opened_on_signal(true);
let started = Core::start_for_test_gated(
spec(vec![root.clone()]),
Duration::from_secs(3600),
discovery::ABANDON_AFTER,
tick_rx,
Some(Arc::clone(&gate)),
)
.discovered();
let core = started.core;
let launched = settle_launch(&core);
let keys: Vec<EntityKey> = launched
.entities
.iter()
.map(|entity| entity.key.clone())
.collect();
init_repo_with_a_commit(&root.join("late"));
set_discovery_gate(&gate, false);
let generation = core.refresh(&keys);
let while_held = core.snapshot();
let dispatched_while_held = core.settle_gate_count_for_test();
walk_may_run.send(()).expect("the opener is listening");
opener.join().expect("the opener thread should not panic");
assert_eq!(
generation,
launched.generation.successor(),
"`refresh` must return its own Generation's number, the one immediately after \
the table's, before that Generation has done any of its work"
);
assert!(
!while_held
.entities
.iter()
.any(|entity| &*entity.name == "late"),
"`refresh` must return before its own Generation's walk has run, so a Repo \
created after the previous walk is not on the table it returned against"
);
assert_eq!(
dispatched_while_held, 0,
"`refresh` returned before its Generation reached the table at all, so nothing \
is dispatched yet"
);
core.wait_dispatched_for_test();
let settled = core.settle();
assert!(
settled
.entities
.iter()
.any(|entity| &*entity.name == "late"),
"the deferred Generation must still run its own walk once it is let through: \
deferred, never dropped"
);
}
#[test]
fn a_dispatch_body_waits_for_every_earlier_reserved_generation() {
let turnstile = Arc::new(DispatchTurnstile::default());
let earlier = turnstile.reserve();
let later = turnstile.reserve();
let order = Arc::new(Mutex::new(Vec::new()));
let earlier_body = thread::spawn({
let turnstile = Arc::clone(&turnstile);
let order = Arc::clone(&order);
move || {
let _turn = turnstile.take(earlier);
order.lock().unwrap().push(earlier);
}
});
{
let _turn = turnstile.take(later);
order.lock().unwrap().push(later);
}
earlier_body
.join()
.expect("the earlier body should not panic");
assert_eq!(
*order.lock().unwrap(),
vec![earlier, later],
"a dispatch body must run in the order its Generation was reserved"
);
}
#[test]
fn run_while_not_cancelled_stops_at_the_next_check_rather_than_running_forever() {
let cancel = Arc::new(AtomicBool::new(false));
let worker_cancel = Arc::clone(&cancel);
let (step_started_tx, step_started_rx) = crossbeam_channel::bounded::<()>(0);
let (proceed_tx, proceed_rx) = crossbeam_channel::bounded::<()>(0);
let worker = thread::spawn(move || {
run_while_not_cancelled(&worker_cancel, || {
step_started_tx.send(()).expect("test should be listening");
proceed_rx.recv().is_ok()
})
});
for _ in 0..2 {
step_started_rx
.recv()
.expect("worker should announce each step");
proceed_tx.send(()).expect("let the step finish");
}
step_started_rx
.recv()
.expect("worker should announce its third step");
cancel.store(true, Ordering::Release);
proceed_tx.send(()).expect("let the third step finish");
let ran = worker.join().expect("worker thread should not panic");
assert_eq!(
ran, 3,
"expected cancellation to stop the loop after its third step"
);
}
fn benchmark_identity_phase(
population: Vec<crate::discovery::DiscoveredEntity>,
) -> (Duration, Vec<Duration>) {
let (tx, rx) = crossbeam_channel::unbounded();
let started = Instant::now();
crate::fanout::scatter(population, tx, |entity| {
let task_started = Instant::now();
let repo = match &entity.repo {
Some(repo) => repo.to_thread_local(),
None => match git::open_thread_safe(entity.key.path()) {
Ok(repo) => repo.to_thread_local(),
Err(_) => return None,
},
};
let _ = git::head_shape(&repo);
Some(task_started.elapsed())
});
let wall = started.elapsed();
let durations: Vec<Duration> = rx.into_iter().flatten().collect();
(wall, durations)
}
fn real_corpus_roots() -> Vec<PathBuf> {
let Some(home) = std::env::var_os("HOME") else {
return Vec::new();
};
let home = PathBuf::from(home);
["dev", "dev-misc"]
.into_iter()
.map(|leaf| home.join(leaf))
.filter(|root| root.is_dir())
.collect()
}
fn generated_fixture_corpus(size: usize) -> tempfile::TempDir {
let root = tempfile::tempdir().expect("temp dir for generated fixture corpus");
for i in 0..size {
let repo = root.path().join(format!("fixture-repo-{i}"));
fs::create_dir_all(&repo).expect("create fixture repo dir");
gix::init(&repo).expect("init fixture repo");
let status = Command::new("git")
.arg("-C")
.arg(&repo)
.args(["-c", "user.email=test@example.com", "-c", "user.name=Test"])
.args(["commit", "--allow-empty", "-m", &format!("commit {i}")])
.status()
.expect("run git commit");
assert!(status.success());
}
root
}
fn percentile(sorted: &[Duration], p: usize) -> Duration {
let index = (sorted.len() - 1) * p / 100;
sorted[index]
}
fn extra_excluded_names() -> Vec<String> {
parse_excluded_names(&std::env::var("REPON_BENCHMARK_EXCLUDE_NAMES").unwrap_or_default())
}
fn parse_excluded_names(raw: &str) -> Vec<String> {
raw.split(',')
.map(str::trim)
.filter(|name| !name.is_empty())
.map(str::to_string)
.collect()
}
fn discover_population(
roots: Vec<PathBuf>,
excluded_names: &[String],
) -> (Vec<crate::discovery::DiscoveredEntity>, Duration) {
let set = SetSpec {
name: "identity-probe-benchmark".to_string(),
roots,
include: Vec::new(),
exclude: Vec::new(),
};
let started = Instant::now();
let discovery = discovery::discover(&set);
let (discovered, _) = discovery::resolve(&set, &discovery.entities);
let elapsed = started.elapsed();
let population = discovered
.into_iter()
.filter(|entity| {
!entity.key.path().components().any(|component| {
excluded_names
.iter()
.any(|name| component.as_os_str() == name.as_str())
})
})
.collect();
(population, elapsed)
}
#[test]
fn a_boundary_whose_path_matches_an_excluded_name_is_left_out_of_the_population() {
let fixture = generated_fixture_corpus(3);
let excluded = vec!["fixture-repo-1".to_string()];
let (population, _) = discover_population(vec![fixture.path().to_path_buf()], &excluded);
assert_eq!(population.len(), 2);
assert!(
population
.iter()
.all(|entity| entity.key.path().file_name().unwrap() != "fixture-repo-1"),
"the excluded name must never appear in the population discovery returns"
);
}
#[test]
fn excluded_names_parses_a_comma_separated_list_and_ignores_blanks() {
assert_eq!(
parse_excluded_names("foo, bar ,,baz"),
vec!["foo".to_string(), "bar".to_string(), "baz".to_string()]
);
assert!(parse_excluded_names("").is_empty());
assert!(parse_excluded_names(" ").is_empty());
}
#[test]
#[ignore = "hand-run against the owner's real corpus; see docs/spec/refresh.md for the recorded figures"]
fn identity_probe_benchmark() {
let excluded_names = extra_excluded_names();
let mut _fixture: Option<tempfile::TempDir> = None;
let (real_population, real_discovery_wall) =
discover_population(real_corpus_roots(), &excluded_names);
let (population, using_fixture, discovery_wall) = if real_population.len() >= 20 {
(real_population, false, real_discovery_wall)
} else {
println!(
"real corpus absent or too small to be meaningful ({} entities); \
using a generated fixture instead",
real_population.len()
);
let fixture = generated_fixture_corpus(300);
let (population, fixture_discovery_wall) =
discover_population(vec![fixture.path().to_path_buf()], &excluded_names);
_fixture = Some(fixture);
(population, true, fixture_discovery_wall)
};
let population_size = population.len();
assert!(
population_size > 0,
"neither a real corpus root nor the generated fixture produced any entities"
);
let (wall, mut durations) = benchmark_identity_phase(population);
durations.sort();
println!(
"identity probe benchmark: corpus = {}, population = {population_size}",
if using_fixture {
"generated fixture"
} else {
"real corpus"
}
);
println!(
"discovery + first open (serial, every entity's own gix::open): {discovery_wall:?}"
);
println!("identity phase, warm, parallel (HEAD re-read from the cached handle): {wall:?}");
println!(
"identity phase per entity: p50 {:?}, p90 {:?}, max {:?}",
percentile(&durations, 50),
percentile(&durations, 90),
durations.last().copied().unwrap_or_default(),
);
}
fn spec_with_overrides(roots: Vec<PathBuf>, overrides: Vec<RepoOverride>) -> CoreSpec {
let mut spec = spec(roots);
spec.overrides = overrides;
spec
}
#[test]
fn a_per_repo_override_resolves_the_default_branch_at_rung_one_through_a_real_refresh() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
git(
&repo,
&[
"remote",
"add",
"origin",
"https://example.invalid/repo.git",
],
);
let sha = head_sha(&repo);
git(&repo, &["update-ref", "refs/remotes/origin/main", &sha]);
let remote_refs_dir = repo
.join(".git")
.join("refs")
.join("remotes")
.join("origin");
fs::create_dir_all(&remote_refs_dir).expect("create refs/remotes/origin dir");
fs::write(
remote_refs_dir.join("HEAD"),
"ref: refs/remotes/origin/main\n",
)
.expect("write HEAD");
let core = Core::start_discovered(spec_with_overrides(
vec![root],
vec![RepoOverride {
path: repo.clone(),
default_branch: Some("develop".to_string()),
excluded: false,
}],
));
let key = core.snapshot().entities[0].key.clone();
core.refresh(std::slice::from_ref(&key));
let settled = core.settle();
let entity = &settled.entities[0];
match entity.default_branch.settled() {
Some(Settled::Known {
value,
at: _,
stale: _,
}) => assert_eq!(
value.name(),
"origin/develop",
"the override must win even though origin/HEAD names a different branch"
),
other => panic!("expected the override's own answer, got {other:?}"),
}
assert_eq!(
entity.diagnostics.default_branch_rung,
Some(1),
"an override must be recorded as rung 1"
);
}
#[test]
fn a_per_repo_override_also_resolves_through_probe_now() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let core = Core::start_discovered(spec_with_overrides(
vec![root],
vec![RepoOverride {
path: repo.clone(),
default_branch: Some("release".to_string()),
excluded: false,
}],
));
let key = core.snapshot().entities[0].key.clone();
let entity = core.probe_now(&key);
match entity.default_branch.settled() {
Some(Settled::Known {
value,
at: _,
stale: _,
}) => assert_eq!(value.name(), "release"),
other => panic!("expected the override's own answer, got {other:?}"),
}
assert_eq!(entity.diagnostics.default_branch_rung, Some(1));
}
#[test]
fn reaching_rung_four_with_no_remote_at_all_records_why() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let core = Core::start_discovered(spec(vec![root]));
let key = core.snapshot().entities[0].key.clone();
core.refresh(std::slice::from_ref(&key));
let settled = core.settle();
let entity = &settled.entities[0];
assert_eq!(entity.diagnostics.default_branch_rung, Some(4));
assert_eq!(
entity.diagnostics.default_branch_stopped,
Some(DefaultBranchStopped::NoRemote)
);
}
#[test]
fn reaching_rung_four_with_two_unnamed_remotes_records_why() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
git(
&repo,
&[
"remote",
"add",
"fork-one",
"https://example.invalid/one.git",
],
);
git(
&repo,
&[
"remote",
"add",
"fork-two",
"https://example.invalid/two.git",
],
);
let core = Core::start_discovered(spec(vec![root]));
let key = core.snapshot().entities[0].key.clone();
core.refresh(std::slice::from_ref(&key));
let settled = core.settle();
let entity = &settled.entities[0];
assert_eq!(entity.diagnostics.default_branch_rung, Some(4));
assert_eq!(
entity.diagnostics.default_branch_stopped,
Some(DefaultBranchStopped::AmbiguousRemote)
);
}
#[test]
fn reaching_rung_four_with_a_chosen_remote_and_no_matching_ref_records_why() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
git(
&repo,
&[
"remote",
"add",
"origin",
"https://example.invalid/repo.git",
],
);
let sha = head_sha(&repo);
git(&repo, &["update-ref", "refs/remotes/origin/feature", &sha]);
let core = Core::start_discovered(spec(vec![root]));
let key = core.snapshot().entities[0].key.clone();
core.refresh(std::slice::from_ref(&key));
let settled = core.settle();
let entity = &settled.entities[0];
assert_eq!(entity.diagnostics.default_branch_rung, Some(4));
assert_eq!(
entity.diagnostics.default_branch_stopped,
Some(DefaultBranchStopped::NameListExhausted)
);
}
#[test]
fn a_repo_with_nothing_to_resolve_settles_unknown_never_failed() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let core = Core::start_discovered(spec(vec![root]));
let key = core.snapshot().entities[0].key.clone();
core.refresh(std::slice::from_ref(&key));
let settled = core.settle();
let entity = &settled.entities[0];
assert!(matches!(
entity.default_branch.settled(),
Some(Settled::Unknown(Unknown::NoDefaultBranch))
));
assert_eq!(entity.diagnostics.default_branch_rung, Some(4));
}
#[test]
fn a_stale_remote_head_is_recorded_in_diagnostics_through_a_real_refresh() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
git(
&repo,
&[
"remote",
"add",
"origin",
"https://example.invalid/repo.git",
],
);
let sha = head_sha(&repo);
git(&repo, &["update-ref", "refs/remotes/origin/trunk", &sha]);
let remote_refs_dir = repo
.join(".git")
.join("refs")
.join("remotes")
.join("origin");
fs::create_dir_all(&remote_refs_dir).expect("create refs/remotes/origin dir");
fs::write(
remote_refs_dir.join("HEAD"),
"ref: refs/remotes/origin/main\n",
)
.expect("write HEAD");
let core = Core::start_discovered(spec(vec![root]));
let key = core.snapshot().entities[0].key.clone();
core.refresh(std::slice::from_ref(&key));
let settled = core.settle();
let entity = &settled.entities[0];
match entity.default_branch.settled() {
Some(Settled::Known {
value,
at: _,
stale: _,
}) => {
assert_eq!(value.name(), "origin/trunk")
}
other => panic!("expected the name list's answer, got {other:?}"),
}
assert!(
entity.diagnostics.default_branch_rung_two_stale,
"a stale origin/HEAD target must be recorded on the entity's diagnostics"
);
}
#[test]
fn a_resolvable_remote_head_is_not_recorded_as_stale() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
git(
&repo,
&[
"remote",
"add",
"origin",
"https://example.invalid/repo.git",
],
);
let sha = head_sha(&repo);
git(&repo, &["update-ref", "refs/remotes/origin/main", &sha]);
let remote_refs_dir = repo
.join(".git")
.join("refs")
.join("remotes")
.join("origin");
fs::create_dir_all(&remote_refs_dir).expect("create refs/remotes/origin dir");
fs::write(
remote_refs_dir.join("HEAD"),
"ref: refs/remotes/origin/main\n",
)
.expect("write HEAD");
let core = Core::start_discovered(spec(vec![root]));
let key = core.snapshot().entities[0].key.clone();
core.refresh(std::slice::from_ref(&key));
let settled = core.settle();
let entity = &settled.entities[0];
assert!(!entity.diagnostics.default_branch_rung_two_stale);
}
#[test]
fn one_override_on_a_repos_path_covers_a_worktree_sharing_its_common_dir() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let parent = root.join("parent");
init_repo_with_a_commit(&parent);
let worktree = root.join("worktree");
git(
&parent,
&[
"worktree",
"add",
"-b",
"feature",
worktree.to_str().expect("utf8 path"),
],
);
let core = Core::start_discovered(spec_with_overrides(
vec![root],
vec![RepoOverride {
path: parent.clone(),
default_branch: None,
excluded: true,
}],
));
let snapshot = core.snapshot();
for entity in &snapshot.entities {
assert!(
entity.excluded,
"both the Repo and its Worktree must inherit the entry declared on the Repo's own path, entity: {:?}",
entity.key
);
}
assert_eq!(
snapshot.entities.len(),
2,
"expected the parent plus its worktree"
);
}
#[test]
fn an_entry_naming_a_worktrees_own_path_beats_the_inherited_one() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let parent = root.join("parent");
init_repo_with_a_commit(&parent);
let worktree_own = root.join("worktree-own");
let worktree_inherits = root.join("worktree-inherits");
git(
&parent,
&[
"worktree",
"add",
"-b",
"feature-own",
worktree_own.to_str().expect("utf8 path"),
],
);
git(
&parent,
&[
"worktree",
"add",
"-b",
"feature-inherits",
worktree_inherits.to_str().expect("utf8 path"),
],
);
let core = Core::start_discovered(spec_with_overrides(
vec![root],
vec![
RepoOverride {
path: parent.clone(),
default_branch: None,
excluded: true,
},
RepoOverride {
path: worktree_own.clone(),
default_branch: None,
excluded: false,
},
],
));
let snapshot = core.snapshot();
let find = |path: &Path| {
snapshot
.entities
.iter()
.find(|entity| entity.key.path() == path)
.unwrap_or_else(|| panic!("entity at {path:?} present"))
};
assert!(
find(&parent).excluded,
"the parent Repo has no entry of its own and inherits the excluding one"
);
assert!(
!find(&worktree_own).excluded,
"the Worktree named directly by its own path must use its own entry, not the inherited one"
);
assert!(
find(&worktree_inherits).excluded,
"a sibling Worktree with no entry of its own still inherits the Repo's entry"
);
}
#[test]
fn an_override_on_the_parents_path_never_excludes_its_submodule() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let parent = root.join("parent");
init_repo_with_a_commit(&parent);
fs::write(
parent.join(".gitmodules"),
"[submodule \"lib\"]\n\tpath = vendor/lib\n\turl = https://example.com/lib.git\n",
)
.expect("write .gitmodules");
fs::create_dir_all(parent.join("vendor").join("lib")).expect("create submodule dir");
let core = Core::start_discovered(spec_with_overrides(
vec![root],
vec![RepoOverride {
path: parent.clone(),
default_branch: None,
excluded: true,
}],
));
let snapshot = core.snapshot();
let submodule = snapshot
.entities
.iter()
.find(|entity| matches!(entity.kind, Kind::Submodule))
.expect("the submodule is still discovered and listed");
assert!(
!submodule.excluded,
"an entry naming only the parent's path must never reach a Submodule, \
whose own common dir differs from its parent's"
);
}
#[test]
fn the_default_branch_chain_is_memoised_once_per_common_dir_per_generation() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let parent = root.join("parent");
init_repo_with_a_commit(&parent);
for name in ["wt-a", "wt-b", "wt-c"] {
let worktree = root.join(name);
git(
&parent,
&[
"worktree",
"add",
"-b",
name,
worktree.to_str().expect("utf8 path"),
],
);
}
let other_repo = root.join("other");
init_repo_with_a_commit(&other_repo);
let (core, launched) = started_and_settled(spec(vec![root]));
let keys: Vec<EntityKey> = launched
.entities
.iter()
.map(|entity| entity.key.clone())
.collect();
assert_eq!(
keys.len(),
5,
"expected the parent, its three worktrees and the unrelated repo"
);
core.refresh(&keys);
core.settle();
assert_eq!(
core.default_branch_chain_reads_for_test(),
2,
"four entities span exactly two common dirs; a memoised chain reads \
each common dir once, not once per entity"
);
core.refresh(&keys);
core.settle();
assert_eq!(
core.default_branch_chain_reads_for_test(),
2,
"the memo lives inside one Generation's dispatch; the next Generation \
recomputes rather than inheriting it"
);
}
#[test]
fn patch_equivalence_is_memoised_once_per_common_dir_per_generation() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let parent = root.join("parent");
init_repo_with_a_commit(&parent);
git(
&parent,
&[
"remote",
"add",
"origin",
"https://example.invalid/repo.git",
],
);
let base_sha = head_sha(&parent);
git(
&parent,
&["update-ref", "refs/remotes/origin/main", &base_sha],
);
for name in ["feature-x", "feature-y"] {
let worktree = root.join(name);
git(
&parent,
&[
"worktree",
"add",
"-b",
name,
worktree.to_str().expect("utf8 path"),
],
);
fs::write(worktree.join(format!("{name}.txt")), "unmerged\n")
.expect("write worktree file");
git(&worktree, &["add", "."]);
git(&worktree, &["commit", "-m", "unmerged work"]);
let tip_sha = head_sha(&worktree);
git(
&parent,
&["config", &format!("branch.{name}.remote"), "origin"],
);
git(
&parent,
&[
"config",
&format!("branch.{name}.merge"),
&format!("refs/heads/{name}"),
],
);
git(
&parent,
&[
"update-ref",
&format!("refs/remotes/origin/{name}"),
&tip_sha,
],
);
}
let other_parent = root.join("other");
init_repo_with_a_commit(&other_parent);
git(
&other_parent,
&[
"remote",
"add",
"origin",
"https://example.invalid/other.git",
],
);
let other_base_sha = head_sha(&other_parent);
git(
&other_parent,
&["update-ref", "refs/remotes/origin/main", &other_base_sha],
);
let other_worktree = root.join("other-feature");
git(
&other_parent,
&[
"worktree",
"add",
"-b",
"other-feature",
other_worktree.to_str().expect("utf8 path"),
],
);
fs::write(other_worktree.join("other.txt"), "unmerged\n").expect("write worktree file");
git(&other_worktree, &["add", "."]);
git(&other_worktree, &["commit", "-m", "unmerged work"]);
let other_tip_sha = head_sha(&other_worktree);
git(
&other_parent,
&["config", "branch.other-feature.remote", "origin"],
);
git(
&other_parent,
&[
"config",
"branch.other-feature.merge",
"refs/heads/other-feature",
],
);
git(
&other_parent,
&[
"update-ref",
"refs/remotes/origin/other-feature",
&other_tip_sha,
],
);
let (core, launched) = started_and_settled(spec(vec![root]));
let keys: Vec<EntityKey> = launched
.entities
.iter()
.map(|entity| entity.key.clone())
.collect();
assert_eq!(
keys.len(),
5,
"expected two parents plus their three worktrees"
);
core.refresh(&keys);
let settled = core.settle();
let worktree_states: Vec<_> = settled
.entities
.iter()
.filter(|entity| matches!(entity.kind, Kind::Worktree))
.map(|entity| entity.state.settled())
.collect();
assert_eq!(worktree_states.len(), 3, "expected three worktree rows");
for settled_state in &worktree_states {
assert!(
matches!(
settled_state,
Some(Settled::Known {
value: WorktreeState::Active,
at: _,
stale: _
})
),
"expected every worktree's genuinely unmerged work to settle Active, got {settled_state:?}"
);
}
assert_eq!(
core.patch_identity_reads_for_test(),
2,
"two worktrees share one common dir and must scan its default-branch \
history once between them, not once per entity; the unrelated repo's \
own worktree pays for a second scan"
);
core.refresh(&keys);
core.settle();
assert_eq!(
core.patch_identity_reads_for_test(),
2,
"the memo lives inside one Generation's dispatch; the next Generation \
recomputes rather than inheriting it"
);
}
#[test]
fn an_entity_whose_merge_base_is_deeper_than_its_siblings_widens_the_shared_scan() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let parent = root.join("parent");
init_repo_with_a_commit(&parent);
git(
&parent,
&[
"remote",
"add",
"origin",
"https://example.invalid/repo.git",
],
);
let deep_fork_sha = head_sha(&parent);
git(&parent, &["branch", "feature-deep"]);
let deep_worktree = root.join("feature-deep");
git(
&parent,
&[
"worktree",
"add",
deep_worktree.to_str().expect("utf8 path"),
"feature-deep",
],
);
fs::write(deep_worktree.join("deep.txt"), "deep work\n").expect("write deep.txt");
git(&deep_worktree, &["add", "."]);
git(&deep_worktree, &["commit", "-m", "deep work"]);
let deep_tip_sha = head_sha(&deep_worktree);
git(&parent, &["merge", "--squash", "feature-deep"]);
git(&parent, &["commit", "-m", "squashed deep"]);
let shallow_fork_sha = head_sha(&parent);
git(&parent, &["branch", "feature-shallow"]);
let shallow_worktree = root.join("feature-shallow");
git(
&parent,
&[
"worktree",
"add",
shallow_worktree.to_str().expect("utf8 path"),
"feature-shallow",
],
);
fs::write(shallow_worktree.join("shallow.txt"), "shallow work\n")
.expect("write shallow.txt");
git(&shallow_worktree, &["add", "."]);
git(&shallow_worktree, &["commit", "-m", "shallow work"]);
let shallow_tip_sha = head_sha(&shallow_worktree);
git(&parent, &["merge", "--squash", "feature-shallow"]);
git(&parent, &["commit", "-m", "squashed shallow"]);
let main_tip_sha = head_sha(&parent);
assert_ne!(
deep_fork_sha, shallow_fork_sha,
"the two siblings must fork at genuinely different commits"
);
git(
&parent,
&["update-ref", "refs/remotes/origin/main", &main_tip_sha],
);
for (name, tip_sha) in [
("feature-deep", &deep_tip_sha),
("feature-shallow", &shallow_tip_sha),
] {
git(
&parent,
&["config", &format!("branch.{name}.remote"), "origin"],
);
git(
&parent,
&[
"config",
&format!("branch.{name}.merge"),
&format!("refs/heads/{name}"),
],
);
git(
&parent,
&[
"update-ref",
&format!("refs/remotes/origin/{name}"),
tip_sha,
],
);
}
let (core, snapshot) = started_and_settled(spec(vec![root]));
let deep_key = snapshot
.entities
.iter()
.find(|entity| entity.key.path() == deep_worktree)
.expect("feature-deep worktree discovered")
.key
.clone();
let shallow_key = snapshot
.entities
.iter()
.find(|entity| entity.key.path() == shallow_worktree)
.expect("feature-shallow worktree discovered")
.key
.clone();
let parent_key = snapshot
.entities
.iter()
.find(|entity| entity.key.path() == parent)
.expect("parent repo discovered")
.key
.clone();
let order = vec![parent_key, shallow_key.clone(), deep_key.clone()];
core.refresh(&order);
let settled = core.settle();
let state_of = |key: &EntityKey| {
settled
.entities
.iter()
.find(|entity| &entity.key == key)
.and_then(|entity| entity.state.settled())
.cloned()
};
assert!(
matches!(
state_of(&deep_key),
Some(Settled::Known {
value: WorktreeState::Merged,
at: _,
stale: _
})
),
"expected the deepest sibling's own squash commit to be found once the scan is \
bounded by the deepest merge base, got {:?}",
state_of(&deep_key)
);
assert!(
matches!(
state_of(&shallow_key),
Some(Settled::Known {
value: WorktreeState::Merged,
at: _,
stale: _
})
),
"expected the shallow sibling to settle Merged too, got {:?}",
state_of(&shallow_key)
);
assert_eq!(
core.patch_identity_reads_for_test(),
1,
"both worktrees share one common dir and must still scan its default-branch \
history once between them, not once per entity"
);
assert_eq!(
core.patch_scan_bounds_for_test(),
vec![Some(id(&deep_fork_sha))],
"the one shared scan that ran must have been bounded by the deepest sibling's own \
merge base, not the shallower one's"
);
}
fn id(sha: &str) -> gix::ObjectId {
gix::ObjectId::from_hex(sha.as_bytes()).expect("parse sha")
}
#[test]
fn bound_gate_deepest_folds_every_candidate_regardless_of_report_order() {
let dir = tempfile::tempdir().expect("temp dir");
let repo_path = root_of(&dir).join("repo");
init_repo_with_a_commit(&repo_path);
let deep_sha = id(&head_sha(&repo_path));
fs::write(repo_path.join("child.txt"), "child\n").expect("write child.txt");
git(&repo_path, &["add", "."]);
git(&repo_path, &["commit", "-m", "child of deep"]);
let shallow_sha = id(&head_sha(&repo_path));
let repo = gix::open(&repo_path).expect("open repo");
let gate = BoundGate::new(2);
gate.report(Some(shallow_sha));
gate.report(Some(deep_sha));
assert_eq!(
gate.deepest(&repo),
Some(deep_sha),
"the deepest candidate must win even though the shallower one reported first"
);
}
#[test]
fn probe_patch_equivalence_bounds_the_scan_by_the_gates_deepest_not_its_own_merge_base() {
let dir = tempfile::tempdir().expect("temp dir");
let repo_path = root_of(&dir).join("repo");
init_repo_with_a_commit(&repo_path);
let deep_sha = id(&head_sha(&repo_path));
fs::write(repo_path.join("child.txt"), "child\n").expect("write child.txt");
git(&repo_path, &["add", "."]);
git(&repo_path, &["commit", "-m", "child of deep"]);
let shallow_sha_hex = head_sha(&repo_path);
let shallow_sha = id(&shallow_sha_hex);
fs::write(repo_path.join("tip.txt"), "tip\n").expect("write tip.txt");
git(&repo_path, &["add", "."]);
git(&repo_path, &["commit", "-m", "default tip"]);
let default_tip_hex = head_sha(&repo_path);
let repo = gix::open(&repo_path).expect("open repo");
let outstanding = landing::Outstanding {
entity_tip: shallow_sha,
default_tip: id(&default_tip_hex),
merge_base: Some(shallow_sha),
};
let common_dir: Arc<Path> = Arc::from(repo_path.join(".git"));
let cancel = AtomicBool::new(false);
let patch_cache: PatchIdentityCache = Mutex::new(HashMap::new());
let patch_reads = AtomicUsize::new(0);
let patch_scan_bounds: Mutex<Vec<Option<gix::ObjectId>>> = Mutex::new(Vec::new());
let memo = PatchEquivalenceMemo {
cache: &patch_cache,
reads: &patch_reads,
scan_bounds: &patch_scan_bounds,
};
let gate = BoundGate::new(2);
gate.report(Some(deep_sha));
let mut report = GateReport::new(&gate);
probe_patch_equivalence(
&repo,
&outstanding,
&common_dir,
&cancel,
&memo,
&mut report,
);
assert_eq!(
patch_scan_bounds.lock().unwrap().as_slice(),
[Some(deep_sha)],
"the scan must be bounded by the deepest sibling's merge base, not shallow's own \
({shallow_sha:?})"
);
}
#[test]
fn probe_patch_equivalence_diffs_from_the_merge_base_it_was_handed() {
let dir = tempfile::tempdir().expect("temp dir");
let repo_path = root_of(&dir).join("repo");
init_repo_with_a_commit(&repo_path);
let fork_point_hex = head_sha(&repo_path);
git(&repo_path, &["checkout", "-b", "feature"]);
fs::write(repo_path.join("a.txt"), "one\n").expect("write a.txt");
git(&repo_path, &["add", "a.txt"]);
git(&repo_path, &["commit", "-m", "add a"]);
let mid_sha = id(&head_sha(&repo_path));
fs::write(repo_path.join("b.txt"), "two\n").expect("write b.txt");
git(&repo_path, &["add", "b.txt"]);
git(&repo_path, &["commit", "-m", "add b"]);
let feature_sha = id(&head_sha(&repo_path));
git(&repo_path, &["checkout", "-B", "main", &fork_point_hex]);
git(&repo_path, &["merge", "--squash", "feature"]);
git(&repo_path, &["commit", "-m", "squashed feature"]);
let main_sha = id(&head_sha(&repo_path));
let repo = gix::open(&repo_path).expect("open repo");
let outstanding = landing::Outstanding {
entity_tip: feature_sha,
default_tip: main_sha,
merge_base: Some(mid_sha),
};
let common_dir: Arc<Path> = Arc::from(repo_path.join(".git"));
let cancel = AtomicBool::new(false);
let patch_cache: PatchIdentityCache = Mutex::new(HashMap::new());
let patch_reads = AtomicUsize::new(0);
let patch_scan_bounds: Mutex<Vec<Option<gix::ObjectId>>> = Mutex::new(Vec::new());
let memo = PatchEquivalenceMemo {
cache: &patch_cache,
reads: &patch_reads,
scan_bounds: &patch_scan_bounds,
};
let gate = BoundGate::new(1);
let mut report = GateReport::new(&gate);
let settled = probe_patch_equivalence(
&repo,
&outstanding,
&common_dir,
&cancel,
&memo,
&mut report,
);
assert!(
matches!(
settled,
Some(Settled::Known {
value: WorktreeState::Active,
at: _,
stale: _
})
),
"the range must be measured from the handed-in base ({mid_sha:?}), whose only \
change the squash commit does not match, got {settled:?}"
);
}
#[test]
fn bound_gate_deepest_with_no_candidates_leaves_the_scan_unbounded() {
let dir = tempfile::tempdir().expect("temp dir");
let repo_path = root_of(&dir).join("repo");
gix::init(&repo_path).expect("init repo");
let repo = gix::open(&repo_path).expect("open repo");
let gate = BoundGate::new(2);
gate.report(None);
gate.report(None);
assert_eq!(
gate.deepest(&repo),
None,
"no contributed candidate must leave the scan unbounded"
);
}
#[test]
fn an_outstanding_entity_with_no_shared_history_settles_active_without_the_shared_scan() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let parent = root.join("parent");
init_repo_with_a_commit(&parent);
git(&parent, &["branch", "-M", "main"]);
git(
&parent,
&[
"remote",
"add",
"origin",
"https://example.invalid/repo.git",
],
);
let main_sha = head_sha(&parent);
git(
&parent,
&["update-ref", "refs/remotes/origin/main", &main_sha],
);
git(&parent, &["checkout", "--orphan", "unrelated"]);
git(
&parent,
&["commit", "--allow-empty", "-m", "unrelated root"],
);
let unrelated_sha = head_sha(&parent);
git(&parent, &["checkout", "main"]);
let worktree = root.join("unrelated");
git(
&parent,
&[
"worktree",
"add",
worktree.to_str().expect("utf8 path"),
"unrelated",
],
);
git(&parent, &["config", "branch.unrelated.remote", "origin"]);
git(
&parent,
&["config", "branch.unrelated.merge", "refs/heads/unrelated"],
);
git(
&parent,
&[
"update-ref",
"refs/remotes/origin/unrelated",
&unrelated_sha,
],
);
let (core, snapshot) = started_and_settled(spec(vec![root]));
let worktree_key = snapshot
.entities
.iter()
.find(|entity| entity.key.path() == worktree)
.expect("unrelated worktree discovered")
.key
.clone();
core.refresh(std::slice::from_ref(&worktree_key));
let settled = core.settle();
let state = settled
.entities
.iter()
.find(|entity| entity.key == worktree_key)
.and_then(|entity| entity.state.settled())
.cloned();
assert!(
matches!(
state,
Some(Settled::Known {
value: WorktreeState::Active,
at: _,
stale: _
})
),
"expected an Outstanding entity with no shared history to settle Active via the \
bypass, got {state:?}"
);
assert_eq!(
core.patch_identity_reads_for_test(),
0,
"the bypass must settle without ever running the shared scan"
);
}
fn add_origin_remote(path: &Path) {
git(
path,
&[
"remote",
"add",
"origin",
"https://example.invalid/repo.git",
],
);
}
fn set_upstream(path: &Path, branch: &str, upstream_sha: &str) {
git(
path,
&["config", &format!("branch.{branch}.remote"), "origin"],
);
git(
path,
&[
"config",
&format!("branch.{branch}.merge"),
&format!("refs/heads/{branch}"),
],
);
git(
path,
&[
"update-ref",
&format!("refs/remotes/origin/{branch}"),
upstream_sha,
],
);
}
fn refresh_and_settle(core: &Core) -> crate::snapshot::Snapshot {
let keys: Vec<EntityKey> = core
.snapshot()
.entities
.iter()
.map(|entity| entity.key.clone())
.collect();
core.refresh(&keys);
core.settle()
}
fn sync_of<'a>(
snapshot: &'a crate::snapshot::Snapshot,
path: &Path,
) -> Option<&'a Settled<SyncState>> {
snapshot
.entities
.iter()
.find(|entity| entity.key.path() == path)
.unwrap_or_else(|| panic!("no entity for {}", path.display()))
.sync
.settled()
}
#[test]
fn an_attached_branch_ahead_of_its_upstream_reads_the_ahead_count() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let fork_sha = head_sha(&repo);
add_origin_remote(&repo);
set_upstream(&repo, "main", &fork_sha);
git(&repo, &["commit", "--allow-empty", "-m", "local work"]);
let core = Core::start_discovered(spec(vec![root]));
let settled = refresh_and_settle(&core);
match sync_of(&settled, &repo) {
Some(Settled::Known {
value: SyncState::Tracking(AheadBehind { ahead, behind }),
at: _,
stale: _,
}) => {
assert_eq!(*ahead, 1);
assert_eq!(*behind, 0);
}
other => panic!("expected 1 ahead, 0 behind, got {other:?}"),
}
}
#[test]
fn an_attached_branch_behind_its_upstream_reads_the_behind_count() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
git(&repo, &["checkout", "-b", "temp"]);
git(&repo, &["commit", "--allow-empty", "-m", "upstream work"]);
let upstream_sha = head_sha(&repo);
git(&repo, &["checkout", "main"]);
git(&repo, &["branch", "-D", "temp"]);
add_origin_remote(&repo);
set_upstream(&repo, "main", &upstream_sha);
let core = Core::start_discovered(spec(vec![root]));
let settled = refresh_and_settle(&core);
match sync_of(&settled, &repo) {
Some(Settled::Known {
value: SyncState::Tracking(AheadBehind { ahead, behind }),
at: _,
stale: _,
}) => {
assert_eq!(*ahead, 0);
assert_eq!(*behind, 1);
}
other => panic!("expected 0 ahead, 1 behind, got {other:?}"),
}
}
#[test]
fn an_attached_branch_level_with_its_upstream_reads_in_sync() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let sha = head_sha(&repo);
add_origin_remote(&repo);
set_upstream(&repo, "main", &sha);
let core = Core::start_discovered(spec(vec![root]));
let settled = refresh_and_settle(&core);
match sync_of(&settled, &repo) {
Some(Settled::Known {
value:
SyncState::Tracking(AheadBehind {
ahead: 0,
behind: 0,
}),
at: _,
stale: _,
}) => {}
other => panic!("expected level with its upstream, got {other:?}"),
}
}
#[test]
fn an_attached_branch_tracking_nothing_reads_no_upstream() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
add_origin_remote(&repo);
let core = Core::start_discovered(spec(vec![root]));
let settled = refresh_and_settle(&core);
match sync_of(&settled, &repo) {
Some(Settled::Known {
value: SyncState::NoUpstream,
at: _,
stale: _,
}) => {}
other => panic!("expected no upstream configured, got {other:?}"),
}
}
#[test]
fn a_detached_row_reads_no_upstream() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let first_sha = head_sha(&repo);
git(&repo, &["commit", "--allow-empty", "-m", "second"]);
git(&repo, &["checkout", "--detach", &first_sha]);
add_origin_remote(&repo);
let core = Core::start_discovered(spec(vec![root]));
let settled = refresh_and_settle(&core);
match sync_of(&settled, &repo) {
Some(Settled::Known {
value: SyncState::NoUpstream,
at: _,
stale: _,
}) => {}
other => panic!("expected a detached row to read no upstream, got {other:?}"),
}
}
#[test]
fn a_repo_with_no_remote_reads_no_remote_on_itself_and_every_worktree() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let parent = root.join("parent");
init_repo_with_a_commit(&parent);
let worktree = root.join("feature");
git(
&parent,
&[
"worktree",
"add",
"-b",
"feature",
worktree.to_str().expect("utf8 path"),
],
);
let core = Core::start_discovered(spec(vec![root]));
let settled = refresh_and_settle(&core);
assert_eq!(
settled.entities.len(),
2,
"expected the parent Repo and its one linked Worktree"
);
for path in [&parent, &worktree] {
match sync_of(&settled, path) {
Some(Settled::Known {
value: SyncState::NoRemote,
at: _,
stale: _,
}) => {}
other => panic!(
"expected {} to read no remote at all, got {other:?}",
path.display()
),
}
}
}
#[test]
fn sync_is_computed_for_every_entity_dispatched_this_generation_not_only_one() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let parent = root.join("parent");
init_repo_with_a_commit(&parent);
let fork_sha = head_sha(&parent);
add_origin_remote(&parent);
let ahead_worktree = root.join("feature-ahead");
git(
&parent,
&[
"worktree",
"add",
"-b",
"feature-ahead",
ahead_worktree.to_str().expect("utf8 path"),
],
);
set_upstream(&parent, "feature-ahead", &fork_sha);
git(
&ahead_worktree,
&["commit", "--allow-empty", "-m", "unpushed"],
);
let behind_worktree = root.join("feature-behind");
git(
&parent,
&[
"worktree",
"add",
"-b",
"feature-behind",
behind_worktree.to_str().expect("utf8 path"),
],
);
git(
&behind_worktree,
&["commit", "--allow-empty", "-m", "on the remote only"],
);
let ahead_of_behind_sha = head_sha(&behind_worktree);
git(&behind_worktree, &["reset", "--hard", "HEAD~1"]);
set_upstream(&parent, "feature-behind", &ahead_of_behind_sha);
let core = Core::start_discovered(spec(vec![root]));
let settled = refresh_and_settle(&core);
match sync_of(&settled, &ahead_worktree) {
Some(Settled::Known {
value:
SyncState::Tracking(AheadBehind {
ahead: 1,
behind: 0,
}),
at: _,
stale: _,
}) => {}
other => panic!("expected feature-ahead to read 1 ahead, got {other:?}"),
}
match sync_of(&settled, &behind_worktree) {
Some(Settled::Known {
value:
SyncState::Tracking(AheadBehind {
ahead: 0,
behind: 1,
}),
at: _,
stale: _,
}) => {}
other => panic!("expected feature-behind to read 1 behind, got {other:?}"),
}
}
#[test]
fn sync_recomputes_on_a_second_generation_not_only_the_first() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let fork_sha = head_sha(&repo);
add_origin_remote(&repo);
set_upstream(&repo, "main", &fork_sha);
let core = Core::start_discovered(spec(vec![root]));
let first = refresh_and_settle(&core);
match sync_of(&first, &repo) {
Some(Settled::Known {
value:
SyncState::Tracking(AheadBehind {
ahead: 0,
behind: 0,
}),
at: _,
stale: _,
}) => {}
other => panic!("expected the first Generation level with its upstream, got {other:?}"),
}
git(
&repo,
&[
"commit",
"--allow-empty",
"-m",
"second Generation's own work",
],
);
let second = refresh_and_settle(&core);
match sync_of(&second, &repo) {
Some(Settled::Known {
value:
SyncState::Tracking(AheadBehind {
ahead: 1,
behind: 0,
}),
at: _,
stale: _,
}) => {}
other => panic!(
"expected the second Generation to recompute and read 1 ahead, got {other:?}"
),
}
}
#[test]
fn worktrees_now_behind_a_moved_default_branch_are_reported_by_name() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let sha_a = head_sha(&repo);
add_origin_remote(&repo);
set_upstream(&repo, "main", &sha_a);
let behind_path = root.join("wt-behind");
git(
&repo,
&[
"worktree",
"add",
"-b",
"topic-behind",
behind_path.to_str().expect("utf8 path"),
"main",
],
);
git(&repo, &["checkout", "-b", "scratch"]);
git(&repo, &["commit", "--allow-empty", "-m", "second"]);
let sha_b = head_sha(&repo);
git(&repo, &["checkout", "main"]);
git(&repo, &["update-ref", "refs/remotes/origin/main", &sha_b]);
git(&repo, &["branch", "-D", "scratch"]);
let caught_up_path = root.join("wt-caught-up");
git(
&repo,
&[
"worktree",
"add",
"-b",
"topic-caught-up",
caught_up_path.to_str().expect("utf8 path"),
&sha_b,
],
);
let core = Core::start_discovered(spec(vec![root]));
let snapshot = refresh_and_settle(&core);
let base_of = |name: &str| -> u32 {
let entity = snapshot
.entities
.iter()
.find(|entity| &*entity.name == name)
.unwrap_or_else(|| panic!("no entity named {name} in {snapshot:?}"));
match entity.base.settled() {
Some(Settled::Known {
value,
at: _,
stale: _,
}) => *value,
other => panic!("expected a known base count for {name}, got {other:?}"),
}
};
assert!(
base_of("wt-behind") > 0,
"a Worktree branched before the default branch moved must be reported behind"
);
assert_eq!(
base_of("wt-caught-up"),
0,
"a Worktree branched from the new tip must not be reported behind"
);
}
mod fetch_scheduler {
use super::*;
use crate::liveness::wait_for_or;
fn fetch_spec(enabled: bool, root: PathBuf) -> CoreSpec {
let mut spec = spec(vec![root]);
spec.fetch = FetchSpec {
enabled,
interval: Duration::from_secs(3600),
concurrency: 4,
};
spec
}
fn seeded_remote() -> tempfile::TempDir {
let remote = tempfile::tempdir().expect("temp dir");
crate::test_support::init_bare(remote.path());
crate::test_support::push_new_commit(remote.path(), "README.md", "seed\n");
remote
}
fn clone_into(remote: &Path, dest: &Path) {
let status = Command::new("git")
.arg("clone")
.arg(remote)
.arg(dest)
.status()
.expect("run git clone");
assert!(status.success());
crate::test_support::set_identity(dest);
}
#[test]
fn enabling_the_periodic_fetch_runs_one_cycle_before_any_tick_arrives() {
let remote = seeded_remote();
let root = tempfile::tempdir().expect("temp dir");
let root_path = root_of(&root);
clone_into(remote.path(), &root_path.join("parent"));
let fetch_ticks: Receiver<Instant> = crossbeam_channel::never();
let started = Core::start_for_test_with_fetch(
fetch_spec(true, root_path),
Duration::from_secs(3600),
crossbeam_channel::never(),
fetch_ticks,
)
.discovered();
let core = started.core;
wait_for(
"the periodic fetch to run its first cycle without waiting for a tick",
|| core.fetch_cycle_count_for_test() >= 1,
);
}
#[test]
fn a_tick_on_the_fetch_channel_runs_another_cycle() {
let remote = seeded_remote();
let root = tempfile::tempdir().expect("temp dir");
let root_path = root_of(&root);
clone_into(remote.path(), &root_path.join("parent"));
let (fetch_tick_tx, fetch_tick_rx) = crossbeam_channel::unbounded();
let started = Core::start_for_test_with_fetch(
fetch_spec(true, root_path),
Duration::from_secs(3600),
crossbeam_channel::never(),
fetch_tick_rx,
)
.discovered();
let core = started.core;
wait_for("the immediate cycle to have run first", || {
core.fetch_cycle_count_for_test() >= 1
});
fetch_tick_tx
.send(Instant::now())
.expect("send a fetch tick");
wait_for("a tick on the fetch channel to run a second cycle", || {
core.fetch_cycle_count_for_test() >= 2
});
}
fn break_remote(repo: &Path) {
let status = Command::new("git")
.arg("-C")
.arg(repo)
.args(["remote", "set-url", "origin", "/nonexistent-remote-282"])
.status()
.expect("run git remote set-url");
assert!(status.success());
}
#[test]
fn a_cycle_in_which_every_fetch_succeeds_reports_no_failures() {
let remote = seeded_remote();
let root = tempfile::tempdir().expect("temp dir");
let root_path = root_of(&root);
clone_into(remote.path(), &root_path.join("parent"));
let fetch_ticks: Receiver<Instant> = crossbeam_channel::never();
let started = Core::start_for_test_with_fetch(
fetch_spec(true, root_path),
Duration::from_secs(3600),
crossbeam_channel::never(),
fetch_ticks,
)
.discovered();
let core = started.core;
wait_for("the periodic fetch to run its first cycle", || {
core.fetch_cycle_count_for_test() >= 1
});
assert!(
core.fetch_failures().failed.is_empty(),
"a cycle where every fetch succeeds must report no failures, got: {:?}",
core.fetch_failures().failed
);
}
#[test]
fn a_repository_that_cannot_be_fetched_is_counted_while_its_sibling_still_fetches() {
let good_remote = seeded_remote();
let bad_remote = seeded_remote();
let root = tempfile::tempdir().expect("temp dir");
let root_path = root_of(&root);
let good = root_path.join("good");
let bad = root_path.join("bad");
clone_into(good_remote.path(), &good);
clone_into(bad_remote.path(), &bad);
break_remote(&bad);
crate::test_support::push_new_commit(good_remote.path(), "second.txt", "second\n");
let good_remote_tip = rev_parse(good_remote.path(), "refs/heads/main");
let fetch_ticks: Receiver<Instant> = crossbeam_channel::never();
let started = Core::start_for_test_with_fetch(
fetch_spec(true, root_path),
Duration::from_secs(3600),
crossbeam_channel::never(),
fetch_ticks,
)
.discovered();
let core = started.core;
wait_for(
"the cycle to run and count the one repository it could not fetch",
|| core.fetch_failures().failed.len() == 1,
);
let failures = core.fetch_failures();
assert_eq!(
failures.failed.len(),
1,
"exactly one repository failed, so exactly one failure must be counted, \
got: {:?}",
failures.failed
);
assert!(
failures.failed[0].0.to_string_lossy().contains("bad"),
"the counted failure must name the repository that actually failed, \
got: {:?}",
failures.failed
);
wait_for(
"the sibling repository to still fetch despite the other one failing",
|| rev_parse(&good, "refs/remotes/origin/main") == good_remote_tip,
);
}
fn push_new_commit_on_branch(remote: &Path, branch: &str, name: &str, contents: &str) {
let contributor = tempfile::tempdir().expect("temp dir");
let status = Command::new("git")
.arg("clone")
.arg("--branch")
.arg(branch)
.arg(remote)
.arg(contributor.path())
.status()
.expect("run git clone");
assert!(status.success());
std::fs::write(contributor.path().join(name), contents).expect("write fixture file");
git(contributor.path(), &["add", name]);
git(contributor.path(), &["commit", "-m", "extra work on topic"]);
git(contributor.path(), &["push", "origin", branch]);
}
#[test]
fn a_finished_fetch_prunes_and_starts_its_own_generation_that_lands_gone() {
let remote = seeded_remote();
let root = tempfile::tempdir().expect("temp dir");
let root_path = root_of(&root);
let parent = root_path.join("parent");
clone_into(remote.path(), &parent);
git(remote.path(), &["branch", "topic"]);
push_new_commit_on_branch(remote.path(), "topic", "topic.txt", "extra work\n");
git(&parent, &["fetch", "origin"]);
let worktree_path = root_path.join("topic-worktree");
git(
&parent,
&[
"worktree",
"add",
"-b",
"topic",
worktree_path.to_str().expect("utf8 path"),
"origin/topic",
],
);
git(remote.path(), &["branch", "-D", "topic"]);
let fetch_ticks: Receiver<Instant> = crossbeam_channel::never();
let started = Core::start_for_test_with_fetch(
fetch_spec(true, root_path),
Duration::from_secs(3600),
crossbeam_channel::never(),
fetch_ticks,
)
.discovered();
let core = started.core;
wait_for_or(
"a finished fetch's own Generation to land the pruned Worktree as Gone \
without the test ever calling refresh",
|| {
core.snapshot()
.entities
.iter()
.filter(|entity| matches!(entity.kind, Kind::Worktree))
.any(|entity| {
matches!(
entity.state.settled(),
Some(Settled::Known {
value: WorktreeState::Gone,
at: _,
stale: _,
})
)
})
},
|| {
format!(
"snapshot: {:?}",
core.snapshot()
.entities
.iter()
.map(|entity| (entity.kind, entity.state.settled().cloned()))
.collect::<Vec<_>>()
)
},
);
}
fn spec_with_auto_update(
fetch_enabled: bool,
auto_update_enabled: bool,
root: PathBuf,
) -> CoreSpec {
let mut spec = fetch_spec(fetch_enabled, root);
spec.auto_update = AutoUpdateSpec {
enabled: auto_update_enabled,
};
spec
}
fn rev_parse(path: &Path, rev: &str) -> String {
let output = Command::new("git")
.arg("-C")
.arg(path)
.args(["rev-parse", rev])
.output()
.expect("run git rev-parse");
assert!(output.status.success(), "git rev-parse {rev} failed");
String::from_utf8(output.stdout)
.expect("utf8 sha")
.trim()
.to_string()
}
#[test]
fn auto_update_is_off_by_default_even_with_fetch_enabled() {
let remote = seeded_remote();
let root = tempfile::tempdir().expect("temp dir");
let root_path = root_of(&root);
let parent = root_path.join("parent");
clone_into(remote.path(), &parent);
let before = rev_parse(&parent, "refs/heads/main");
crate::test_support::push_new_commit(remote.path(), "second.txt", "second\n");
let fetch_ticks: Receiver<Instant> = crossbeam_channel::never();
let started = Core::start_for_test_with_fetch(
spec_with_auto_update(true, false, root_path),
Duration::from_secs(3600),
crossbeam_channel::never(),
fetch_ticks,
)
.discovered();
let core = started.core;
wait_for(
"the periodic fetch to still run its immediate cycle",
|| core.fetch_cycle_count_for_test() >= 1,
);
assert_eq!(
rev_parse(&parent, "refs/heads/main"),
before,
"an eligible branch must not move while auto_update.enabled is false, \
even though fetch.enabled is true"
);
}
#[test]
fn auto_update_enabled_rides_the_immediate_fetch_cycle_with_no_timer_of_its_own() {
let remote = seeded_remote();
let root = tempfile::tempdir().expect("temp dir");
let root_path = root_of(&root);
let parent = root_path.join("parent");
clone_into(remote.path(), &parent);
crate::test_support::push_new_commit(remote.path(), "second.txt", "second\n");
let remote_tip = rev_parse(remote.path(), "refs/heads/main");
let fetch_ticks: Receiver<Instant> = crossbeam_channel::never();
let started = Core::start_for_test_with_fetch(
spec_with_auto_update(true, true, root_path),
Duration::from_secs(3600),
crossbeam_channel::never(),
fetch_ticks,
)
.discovered();
let _core = started.core;
wait_for(
"the eligible branch to fast-forward on the immediate cycle alone, with no \
fetch tick and no auto-update tick of its own",
|| rev_parse(&parent, "refs/heads/main") == remote_tip,
);
}
}
mod attempt_auto_update {
use super::*;
fn seeded_remote() -> tempfile::TempDir {
let remote = tempfile::tempdir().expect("temp dir");
crate::test_support::init_bare(remote.path());
crate::test_support::push_new_commit(remote.path(), "README.md", "seed\n");
remote
}
fn clone_into(remote: &Path, dest: &Path) {
let status = Command::new("git")
.arg("clone")
.arg(remote)
.arg(dest)
.status()
.expect("run git clone");
assert!(status.success());
crate::test_support::set_identity(dest);
}
fn discover_repo(root: &Path) -> (Core, EntityKey) {
let core = Core::start_discovered(spec(vec![root.to_path_buf()]));
let key = core
.settle()
.entities
.into_iter()
.find(|entity| entity.kind == Kind::Repo)
.expect("the Repo row is discovered")
.key;
(core, key)
}
#[test]
fn an_eligible_repo_fast_forwards_through_the_wrapper_too() {
let remote = seeded_remote();
let root = tempfile::tempdir().expect("temp dir");
let root_path = root_of(&root);
let repo = root_path.join("repo");
clone_into(remote.path(), &repo);
crate::test_support::push_new_commit(remote.path(), "second.txt", "second\n");
crate::test_support::git(&repo, &["fetch", "origin"]);
let (core, key) = discover_repo(&root_path);
assert_eq!(core.attempt_auto_update(&key), AutoUpdateAttempt::Updated);
assert!(
repo.join("second.txt").exists(),
"the fast-forward must reach the working tree through the wrapper too"
);
}
#[test]
fn a_dirty_repo_is_reported_not_clean_through_the_wrapper_too() {
let remote = seeded_remote();
let root = tempfile::tempdir().expect("temp dir");
let root_path = root_of(&root);
let repo = root_path.join("repo");
clone_into(remote.path(), &repo);
crate::test_support::push_new_commit(remote.path(), "second.txt", "second\n");
crate::test_support::git(&repo, &["fetch", "origin"]);
fs::write(repo.join("stray.txt"), "uncommitted\n").expect("write a stray file");
let (core, key) = discover_repo(&root_path);
assert_eq!(core.attempt_auto_update(&key), AutoUpdateAttempt::NotClean);
}
#[test]
fn an_up_to_date_repo_is_reported_not_behind_through_the_wrapper_too() {
let remote = seeded_remote();
let root = tempfile::tempdir().expect("temp dir");
let root_path = root_of(&root);
let repo = root_path.join("repo");
clone_into(remote.path(), &repo);
crate::test_support::git(&repo, &["fetch", "origin"]);
let (core, key) = discover_repo(&root_path);
assert_eq!(core.attempt_auto_update(&key), AutoUpdateAttempt::NotBehind);
}
#[test]
fn an_unpublished_local_commit_is_reported_not_fast_forward_through_the_wrapper_too() {
let remote = seeded_remote();
let root = tempfile::tempdir().expect("temp dir");
let root_path = root_of(&root);
let repo = root_path.join("repo");
clone_into(remote.path(), &repo);
crate::test_support::push_new_commit(remote.path(), "second.txt", "second\n");
crate::test_support::git(&repo, &["fetch", "origin"]);
crate::test_support::commit_file(&repo, "local-only.txt", "never pushed\n");
let (core, key) = discover_repo(&root_path);
assert_eq!(
core.attempt_auto_update(&key),
AutoUpdateAttempt::NotFastForward
);
}
#[test]
fn a_branch_with_no_upstream_is_reported_through_the_wrapper_too() {
let remote = seeded_remote();
let root = tempfile::tempdir().expect("temp dir");
let root_path = root_of(&root);
let repo = root_path.join("repo");
clone_into(remote.path(), &repo);
crate::test_support::git(&repo, &["checkout", "-b", "untracked-branch"]);
let (core, key) = discover_repo(&root_path);
assert_eq!(
core.attempt_auto_update(&key),
AutoUpdateAttempt::NoUpstream
);
}
}
mod network_default_branch {
use super::*;
fn seeded_remote() -> tempfile::TempDir {
let remote = tempfile::tempdir().expect("temp dir");
crate::test_support::init_bare(remote.path());
crate::test_support::push_new_commit(remote.path(), "README.md", "seed\n");
remote
}
fn clone_into(remote: &Path, dest: &Path) {
let status = Command::new("git")
.arg("clone")
.arg(remote)
.arg(dest)
.status()
.expect("run git clone");
assert!(status.success());
crate::test_support::set_identity(dest);
}
fn set_remote_head(path: &Path, branch: &str) {
git(
path,
&["symbolic-ref", "HEAD", &format!("refs/heads/{branch}")],
);
}
fn rev_parse(path: &Path, rev: &str) -> String {
let output = Command::new("git")
.arg("-C")
.arg(path)
.args(["rev-parse", rev])
.output()
.expect("run git rev-parse");
assert!(output.status.success());
String::from_utf8(output.stdout)
.expect("utf8 sha")
.trim()
.to_string()
}
fn default_branch_name(entity: &EntityState) -> Option<String> {
match entity.default_branch.settled() {
Some(Settled::Known {
value,
at: _,
stale: _,
}) => Some(value.name().to_string()),
_ => None,
}
}
#[test]
fn the_local_chain_answers_first_and_only_a_later_network_round_trip_supersedes_it() {
let remote = seeded_remote();
let root = tempfile::tempdir().expect("temp dir");
let root_path = root_of(&root);
let repo_path = root_path.join("repo");
clone_into(remote.path(), &repo_path);
git(remote.path(), &["branch", "trunk"]);
set_remote_head(remote.path(), "trunk");
let core = Core::start_discovered(spec(vec![root_path]));
let key = core.snapshot().entities[0].key.clone();
core.refresh(std::slice::from_ref(&key));
let settled = core.settle();
assert_eq!(
default_branch_name(&settled.entities[0]),
Some("origin/main".to_string()),
"a plain refresh must answer from the local chain alone, unaffected by the \
remote's own current (but not yet asked) truth"
);
core.rederive_default_branches(std::slice::from_ref(&key));
let settled = core.settle();
assert_eq!(
default_branch_name(&settled.entities[0]),
Some("origin/trunk".to_string()),
"once the network round trip actually ran, its own differing answer must \
supersede the local chain's"
);
}
#[test]
fn rederive_default_branches_never_fetches_and_leaves_a_row_outside_it_untouched() {
let remote = seeded_remote();
let root = tempfile::tempdir().expect("temp dir");
let root_path = root_of(&root);
let selected_path = root_path.join("selected");
let outside_path = root_path.join("outside");
clone_into(remote.path(), &selected_path);
init_repo_with_a_commit(&outside_path);
git(remote.path(), &["branch", "trunk"]);
crate::test_support::push_new_commit(remote.path(), "second.txt", "second\n");
set_remote_head(remote.path(), "trunk");
let before_tracking = rev_parse(&selected_path, "refs/remotes/origin/main");
let core = Core::start_discovered(spec(vec![root_path]));
let snapshot = core.snapshot();
let selected_key = snapshot
.entities
.iter()
.find(|entity| entity.key.path() == selected_path)
.expect("discovered the selected repo")
.key
.clone();
let outside_key = snapshot
.entities
.iter()
.find(|entity| entity.key.path() == outside_path)
.expect("discovered the outside repo")
.key
.clone();
core.refresh(&[selected_key.clone(), outside_key.clone()]);
let settled = core.settle();
let outside_before = format!(
"{:?}",
settled
.entities
.iter()
.find(|entity| entity.key == outside_key)
.expect("outside entity present")
);
core.rederive_default_branches(std::slice::from_ref(&selected_key));
let settled = core.settle();
let selected_after = settled
.entities
.iter()
.find(|entity| entity.key == selected_key)
.expect("selected entity present");
assert_eq!(
default_branch_name(selected_after),
Some("origin/trunk".to_string()),
"the rederive must have reached the remote's own current, differing answer"
);
let after_tracking = rev_parse(&selected_path, "refs/remotes/origin/main");
assert_eq!(
before_tracking, after_tracking,
"a rederive must never fetch: the remote-tracking ref must not have moved \
even though the remote gained a new commit"
);
let outside_after = format!(
"{:?}",
settled
.entities
.iter()
.find(|entity| entity.key == outside_key)
.expect("outside entity present")
);
assert_eq!(
outside_before, outside_after,
"a row outside the rederive's own keys must be left exactly as it was, not \
only on its default_branch cell"
);
}
}
#[test]
fn set_exclusions_excludes_a_row_already_in_the_table_with_no_rebuild() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let core = Core::start_discovered(spec(vec![root]));
let snapshot = core.settle();
let key = snapshot.entities[0].key.clone();
let generation_before = snapshot.generation;
assert!(
!snapshot.entities[0].excluded,
"nothing excludes it to start with"
);
assert_eq!(core.operable_count(std::slice::from_ref(&key)), 1);
core.set_exclusions(&[RepoOverride {
path: repo.clone(),
default_branch: None,
excluded: true,
}]);
let after = core.snapshot();
assert!(
after.entities[0].excluded,
"the row the write named is excluded in the very next snapshot"
);
assert_eq!(
core.operable_count(&[key]),
0,
"an excluded row is subtracted from what an operation may reach"
);
assert_eq!(
after.generation, generation_before,
"re-applying an operate-time filter must start no Generation of its own"
);
}
#[test]
fn set_exclusions_clears_the_flag_when_the_entry_is_gone() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let core = Core::start_discovered(spec_with_overrides(
vec![root],
vec![RepoOverride {
path: repo.clone(),
default_branch: None,
excluded: true,
}],
));
assert!(
core.settle().entities[0].excluded,
"the starting override excludes it"
);
core.set_exclusions(&[]);
assert!(
!core.snapshot().entities[0].excluded,
"removing the entry unexcludes the row in the very next snapshot"
);
}
#[test]
fn set_exclusions_moves_exclude_alone_and_never_the_default_branch_override() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
crate::test_support::git(&repo, &["branch", "trunk"]);
let core = Core::start_discovered(spec(vec![root]));
let key = core.settle().entities[0].key.clone();
core.refresh(std::slice::from_ref(&key));
let before = format!("{:?}", core.settle().entities[0].default_branch.settled());
core.set_exclusions(&[RepoOverride {
path: repo.clone(),
default_branch: Some("trunk".to_string()),
excluded: true,
}]);
core.refresh(&[key]);
core.settle();
let after = core.snapshot();
assert!(after.entities[0].excluded, "exclude took effect");
assert_eq!(
format!("{:?}", after.entities[0].default_branch.settled()),
before,
"a default_branch override reaches a session only through a rebuilt Core"
);
}
#[test]
fn record_own_work_leaves_one_receipt_per_row_it_names_and_none_elsewhere() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
init_repo_with_a_commit(&root.join("repo-a"));
init_repo_with_a_commit(&root.join("repo-b"));
let core = Core::start_discovered(spec(vec![root]));
let entities = core.settle().entities;
let named = entities
.iter()
.find(|entity| &*entity.name == "repo-a")
.expect("repo-a is discovered")
.key
.clone();
core.record_own_work(
"ignore",
&[(
named.clone(),
OwnWork::Refused(Arc::from("refused, already ignored")),
Duration::from_millis(7),
)],
);
let after = core.snapshot().entities;
let receipt = after
.iter()
.find(|entity| entity.key == named)
.and_then(|entity| entity.last_action.clone())
.expect("the row it named carries a receipt");
assert_eq!(&*receipt.label, "ignore");
assert!(
!receipt.not_applicable(),
"a refusal is not an excluded row"
);
assert!(receipt.running.is_none(), "the work is already done");
assert_eq!(receipt.steps.len(), 1, "one act, not an ordered list");
assert_eq!(&*receipt.steps[0].label, "ignore");
assert_eq!(receipt.steps[0].elapsed, Duration::from_millis(7));
assert!(receipt.steps[0].output.is_empty(), "nothing to quote");
assert!(receipt.steps[0].elision.is_none());
assert_eq!(
receipt.steps[0].outcome,
StepOutcome::OwnWork(OwnWork::Refused(Arc::from("refused, already ignored"))),
);
assert!(
after
.iter()
.filter(|entity| entity.key != named)
.all(|entity| entity.last_action.is_none()),
"no row this did not name takes a receipt"
);
}
#[test]
fn record_own_work_skips_a_key_the_table_no_longer_holds() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
init_repo_with_a_commit(&root.join("repo-a"));
let core = Core::start_discovered(spec(vec![root]));
let entities = core.settle().entities;
let stranger = EntityKey::new(Arc::from(std::path::Path::new("/nowhere/at/all")));
core.record_own_work(
"delete",
&[(stranger, OwnWork::Did(Arc::from("gone")), Duration::ZERO)],
);
assert!(
core.snapshot()
.entities
.iter()
.all(|entity| entity.last_action.is_none()),
"an unknown key writes nothing anywhere"
);
assert_eq!(core.snapshot().entities.len(), entities.len());
}
#[test]
fn delete_risk_reads_all_three_facts_the_confirm_gate_names() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
fs::write(repo.join("uncommitted.txt"), "not staged\n").expect("write a stray file");
crate::test_support::git(
&repo,
&["worktree", "add", "-b", "sidecar", "../sidecar-worktree"],
);
let core = Core::start_discovered(spec(vec![root]));
let key = core
.settle()
.entities
.into_iter()
.find(|entity| entity.kind == Kind::Repo)
.expect("the Repo row is discovered")
.key;
let risk = core.delete_risk(&key).expect("read the risk");
assert!(risk.uncommitted, "the stray file makes the tree dirty");
assert!(
risk.unpushed_commits > 0 && risk.unpushed_branches > 0,
"no remote-tracking ref carries any of this Repo's commits, got {risk:?}"
);
assert_eq!(
risk.linked_worktrees, 1,
"the one linked Worktree pointing into this Repo is counted, got {risk:?}"
);
}
#[test]
fn every_kind_of_work_that_is_not_in_a_commit_makes_the_gate_say_uncommitted() {
for kind in ["modified", "deleted", "untracked", "staged"] {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
fs::write(repo.join("tracked.txt"), "first\n").expect("write a tracked file");
crate::test_support::git(&repo, &["add", "tracked.txt"]);
crate::test_support::git(&repo, &["commit", "-m", "add tracked"]);
let sha = crate::test_support::head_sha(&repo);
crate::test_support::git(&repo, &["update-ref", "refs/remotes/origin/main", &sha]);
match kind {
"modified" => fs::write(repo.join("tracked.txt"), "second\n").expect("modify it"),
"deleted" => fs::remove_file(repo.join("tracked.txt")).expect("delete it"),
"untracked" => fs::write(repo.join("stray.txt"), "new\n").expect("write a stray"),
"staged" => {
fs::write(repo.join("staged.txt"), "new\n").expect("write a new file");
crate::test_support::git(&repo, &["add", "staged.txt"]);
}
other => unreachable!("unhandled kind {other}"),
}
let core = Core::start_discovered(spec(vec![root]));
let key = core.settle().entities[0].key.clone();
let risk = core.delete_risk(&key).expect("read the risk");
assert!(
risk.uncommitted,
"a {kind} change is work that is not in a commit, got {risk:?}"
);
}
}
#[test]
fn staged_work_reads_clean_to_the_dirty_column_and_uncommitted_to_the_delete_gate() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let sha = crate::test_support::head_sha(&repo);
crate::test_support::git(&repo, &["update-ref", "refs/remotes/origin/main", &sha]);
fs::write(repo.join("staged.txt"), "staged\n").expect("write a new file");
crate::test_support::git(&repo, &["add", "staged.txt"]);
let core = Core::start_discovered(spec(vec![root]));
let key = core.settle().entities[0].key.clone();
let opened = git::open_thread_safe(repo.as_path())
.expect("open the repo")
.to_thread_local();
let dirty = git::dirty_counts(&opened, Arc::new(AtomicBool::new(false)))
.expect("read the dirty counts");
assert_eq!(
dirty.total(),
0,
"the dirty column stays an index-to-worktree comparison, got {dirty:?}"
);
let risk = core.delete_risk(&key).expect("read the risk");
assert!(
risk.uncommitted,
"a Repo whose only work is staged must never be listed plainly, got {risk:?}"
);
}
#[test]
fn unpushed_commits_and_unpushed_branches_are_counted_into_their_own_fields() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let sha = crate::test_support::head_sha(&repo);
crate::test_support::git(&repo, &["update-ref", "refs/remotes/origin/main", &sha]);
for nth in 0..3 {
fs::write(repo.join(format!("file-{nth}.txt")), "x\n").expect("write a file");
crate::test_support::git(&repo, &["add", "."]);
crate::test_support::git(&repo, &["commit", "-m", "unpushed"]);
}
crate::test_support::git(&repo, &["checkout", "."]);
let core = Core::start_discovered(spec(vec![root]));
let key = core.settle().entities[0].key.clone();
let risk = core.delete_risk(&key).expect("read the risk");
assert_eq!(
(risk.unpushed_commits, risk.unpushed_branches),
(3, 1),
"three commits on one branch, each in its own field, got {risk:?}"
);
}
#[test]
fn a_linked_worktree_outside_the_sets_roots_is_still_counted_by_the_gate() {
let dir = tempfile::tempdir().expect("temp dir");
let base = root_of(&dir);
let inside = base.join("inside");
let outside = base.join("outside");
fs::create_dir_all(&outside).expect("create the outside dir");
let repo = inside.join("repo");
init_repo_with_a_commit(&repo);
crate::test_support::git(
&repo,
&["worktree", "add", "-b", "sidecar", "../../outside/sidecar"],
);
assert!(
outside.join("sidecar").exists(),
"the harness really created a linked Worktree outside the Set's roots"
);
let core = Core::start_discovered(spec(vec![inside]));
let snapshot = core.settle();
assert!(
snapshot
.entities
.iter()
.all(|entity| entity.kind != Kind::Worktree),
"the Worktree is outside the roots and so is not discovered, got {:?}",
snapshot.entities.iter().map(|e| e.kind).collect::<Vec<_>>()
);
let key = snapshot
.entities
.into_iter()
.find(|entity| entity.kind == Kind::Repo)
.expect("the Repo row is discovered")
.key;
let risk = core.delete_risk(&key).expect("read the risk");
assert_eq!(
risk.linked_worktrees, 1,
"the gate must name the linked Worktree deleting this Repo would orphan, got {risk:?}"
);
}
#[test]
fn delete_risk_on_a_clean_fully_pushed_repo_with_no_worktrees_reports_nothing() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let sha = crate::test_support::head_sha(&repo);
crate::test_support::git(&repo, &["update-ref", "refs/remotes/origin/main", &sha]);
let core = Core::start_discovered(spec(vec![root]));
let key = core.settle().entities[0].key.clone();
let risk = core.delete_risk(&key).expect("read the risk");
assert_eq!(
risk,
DeleteRisk {
uncommitted: false,
unpushed_commits: 0,
unpushed_branches: 0,
linked_worktrees: 0,
}
);
}
#[test]
fn worktree_admin_dir_names_the_entry_git_worktree_list_forgets_once_it_is_removed() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let worktree = root.join("sidecar");
crate::test_support::git(
&repo,
&[
"worktree",
"add",
"-b",
"sidecar",
worktree.to_str().expect("utf8 path"),
],
);
let core = Core::start_discovered(spec(vec![root]));
let key = core
.settle()
.entities
.into_iter()
.find(|entity| entity.kind == Kind::Worktree)
.expect("the Worktree row is discovered")
.key;
let admin_dir = core.worktree_admin_dir(&key).expect("read the admin dir");
fs::remove_dir_all(&admin_dir).expect("remove the admin dir by hand");
let reopened = git::open_thread_safe(&repo)
.expect("reopen the repo")
.to_thread_local();
assert_eq!(
git::linked_worktrees(&reopened).expect("count"),
0,
"removing the admin dir alone must be what git's own register stops naming"
);
}
#[test]
fn worktree_admin_dir_errors_when_the_path_cannot_be_opened_as_a_repository() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let not_a_repo = root.join("plain-directory");
fs::create_dir_all(¬_a_repo).expect("create it");
let core = Core::start_discovered(spec(vec![root]));
core.settle();
let key = EntityKey::new(Arc::from(not_a_repo.as_path()));
assert!(core.worktree_admin_dir(&key).is_err());
}
#[test]
fn linked_worktree_paths_names_every_linked_worktrees_own_directory() {
let dir = tempfile::tempdir().expect("temp dir");
let root = root_of(&dir);
let repo = root.join("repo");
init_repo_with_a_commit(&repo);
let first = root.join("first-worktree");
let second = root.join("second-worktree");
crate::test_support::git(
&repo,
&[
"worktree",
"add",
"-b",
"one",
first.to_str().expect("utf8 path"),
],
);
crate::test_support::git(
&repo,
&[
"worktree",
"add",
"-b",
"two",
second.to_str().expect("utf8 path"),
],
);
let core = Core::start_discovered(spec(vec![root]));
let key = core
.settle()
.entities
.into_iter()
.find(|entity| entity.kind == Kind::Repo)
.expect("the Repo row is discovered")
.key;
let mut paths = core
.linked_worktree_paths(&key)
.expect("read the linked worktree paths");
paths.sort();
let mut expected = vec![
first.canonicalize().expect("canonicalize first"),
second.canonicalize().expect("canonicalize second"),
];
expected.sort();
assert_eq!(paths, expected);
}
}