use std::collections::BTreeMap;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use parking_lot::{Condvar, Mutex};
use crate::config::{Config, IndexKind};
use crate::db::standing_roots::{self, StandingRootRecord};
use crate::root_cache::{ArtifactPublishEpoch, WriterLease};
use crate::scoped_key::{
reject_duplicate_artifact_keys, resolve_standing_root, ResolvedStandingRoot,
};
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct StandingRootEntry {
pub literal_path: String,
pub resolved_target: PathBuf,
pub resolved_git_toplevel: Option<PathBuf>,
pub scoped_relative_path: Option<PathBuf>,
pub artifact_key: String,
pub indexes: Vec<IndexKind>,
pub config_order: usize,
}
#[derive(Debug)]
pub enum StandingRootsError {
Resolution(crate::scoped_key::ScopedKeyError),
Database(standing_roots::StandingRootError),
OpenDatabase(crate::db::OpenError),
UnknownEntry { literal_path: String },
}
impl std::fmt::Display for StandingRootsError {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Resolution(error) => {
write!(formatter, "standing root resolution failed: {error}")
}
Self::Database(error) => write!(formatter, "standing roots database error: {error}"),
Self::OpenDatabase(error) => {
write!(formatter, "could not open standing roots database: {error}")
}
Self::UnknownEntry { literal_path } => {
write!(
formatter,
"standing root {literal_path:?} has no active lifecycle"
)
}
}
}
}
impl std::error::Error for StandingRootsError {}
impl From<crate::scoped_key::ScopedKeyError> for StandingRootsError {
fn from(error: crate::scoped_key::ScopedKeyError) -> Self {
Self::Resolution(error)
}
}
impl From<standing_roots::StandingRootError> for StandingRootsError {
fn from(error: standing_roots::StandingRootError) -> Self {
Self::Database(error)
}
}
impl From<crate::db::OpenError> for StandingRootsError {
fn from(error: crate::db::OpenError) -> Self {
Self::OpenDatabase(error)
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct StandingReconcileReport {
pub active_entries: Vec<StandingRootEntry>,
pub added: Vec<String>,
pub removed: Vec<String>,
pub replaced: Vec<String>,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum StandingRouteError {
NoContainingEntry,
KindUnavailable {
deepest_literal_path: String,
deepest_selection: Vec<IndexKind>,
},
StrictVerificationRequired {
literal_path: String,
kind: IndexKind,
},
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct StandingRoute {
pub entry: StandingRootEntry,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct StandingPublicationAdmission {
pub admission_epoch: u64,
publication_epoch: u64,
}
pub struct StandingBuildAdmission {
lifecycle: Arc<ArtifactLifecycle>,
pub publication: StandingPublicationAdmission,
}
impl StandingBuildAdmission {
pub fn checkpoint(&self) -> bool {
self.lifecycle.checkpoint()
}
pub fn cancellation_requested(&self) -> bool {
self.lifecycle.cancellation_requested()
}
}
impl Drop for StandingBuildAdmission {
fn drop(&mut self) {
self.lifecycle.finish_build();
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct BindTransition {
pub admission_epoch: u64,
pub cancelled_builds: usize,
}
#[derive(Clone, Default)]
pub struct StandingRoots {
inner: Arc<StandingRootsInner>,
}
#[derive(Default)]
struct StandingRootsInner {
database_path: Mutex<Option<PathBuf>>,
state: Mutex<StandingRootsState>,
}
#[derive(Default)]
struct StandingRootsState {
entries: BTreeMap<String, ManagedEntry>,
observed_first_snapshot: bool,
}
struct ManagedEntry {
entry: StandingRootEntry,
lifecycle: Arc<ArtifactLifecycle>,
}
struct ArtifactLifecycle {
state: Mutex<ArtifactLifecycleState>,
checkpoint_acknowledged: Condvar,
admission_epoch: AtomicU64,
publication_epoch: ArtifactPublishEpoch,
}
#[derive(Default)]
struct ArtifactLifecycleState {
standing_admission_open: bool,
bind_pending: bool,
session_bound: bool,
active_builds: usize,
cancellation_requested: bool,
checkpoint_acknowledged: bool,
}
impl Default for ArtifactLifecycle {
fn default() -> Self {
Self {
state: Mutex::new(ArtifactLifecycleState {
standing_admission_open: true,
..ArtifactLifecycleState::default()
}),
checkpoint_acknowledged: Condvar::new(),
admission_epoch: AtomicU64::new(0),
publication_epoch: ArtifactPublishEpoch::default(),
}
}
}
impl ArtifactLifecycle {
fn admit(self: &Arc<Self>) -> Option<StandingBuildAdmission> {
let mut state = self.state.lock();
if !state.standing_admission_open || state.bind_pending || state.session_bound {
return None;
}
state.active_builds += 1;
Some(StandingBuildAdmission {
lifecycle: Arc::clone(self),
publication: StandingPublicationAdmission {
admission_epoch: self.admission_epoch.load(Ordering::SeqCst),
publication_epoch: self.publication_epoch.current(),
},
})
}
fn revoke_and_mint(&self, bind_pending: bool) -> u64 {
let mut state = self.state.lock();
state.bind_pending = bind_pending;
state.session_bound = bind_pending;
state.standing_admission_open = false;
state.cancellation_requested = true;
state.checkpoint_acknowledged = state.active_builds == 0;
let admission_epoch = self
.admission_epoch
.fetch_add(1, Ordering::SeqCst)
.wrapping_add(1);
self.publication_epoch.next();
if state.checkpoint_acknowledged {
self.checkpoint_acknowledged.notify_all();
}
admission_epoch
}
fn bind_pending(&self) -> bool {
self.state.lock().bind_pending
}
fn resume_unbound(&self) {
let mut state = self.state.lock();
state.bind_pending = false;
state.session_bound = false;
state.standing_admission_open = true;
state.cancellation_requested = false;
state.checkpoint_acknowledged = false;
}
fn checkpoint(&self) -> bool {
let mut state = self.state.lock();
if !state.cancellation_requested {
return false;
}
state.checkpoint_acknowledged = true;
self.checkpoint_acknowledged.notify_all();
true
}
fn cancellation_requested(&self) -> bool {
self.state.lock().cancellation_requested
}
fn finish_build(&self) {
let mut state = self.state.lock();
state.active_builds = state.active_builds.saturating_sub(1);
if state.cancellation_requested && state.active_builds == 0 {
state.checkpoint_acknowledged = true;
self.checkpoint_acknowledged.notify_all();
}
}
fn wait_for_checkpoint_acknowledgement(&self, timeout: Duration) -> bool {
let mut state = self.state.lock();
if state.checkpoint_acknowledged {
return true;
}
self.checkpoint_acknowledged.wait_for(&mut state, timeout);
state.checkpoint_acknowledged
}
}
impl StandingRoots {
pub fn reconcile(
&self,
config: &Config,
) -> Result<StandingReconcileReport, StandingRootsError> {
let entries = resolve_entries(config)?;
let configured_db_path =
crate::bash_background::storage_dir(config.storage_dir.as_deref()).join("aft.db");
let db_path = {
let mut current = self.inner.database_path.lock();
let path = if entries.is_empty() {
current.clone().unwrap_or(configured_db_path)
} else {
configured_db_path
};
*current = Some(path.clone());
path
};
let mut conn = crate::db::open(&db_path)?;
let mut state = self.inner.state.lock();
let previous_first_snapshot = state.observed_first_snapshot;
let previous = std::mem::take(&mut state.entries);
let mut next = BTreeMap::new();
let mut added = Vec::new();
let mut removed = Vec::new();
let mut replaced = Vec::new();
for entry in &entries {
let resolved = resolve_standing_root(&entry.literal_path)?;
let record = StandingRootRecord {
literal_path: entry.literal_path.clone(),
resolved_target: resolved.resolved_target,
resolved_git_toplevel: resolved.resolved_git_toplevel,
scoped_relative_path: resolved.scoped_relative_path,
};
standing_roots::ensure_standing_root(&mut conn, &record, &entry.indexes)?;
if let Some(old) = previous.get(&entry.literal_path) {
let changed = old.entry.resolved_target != entry.resolved_target
|| old.entry.resolved_git_toplevel != entry.resolved_git_toplevel
|| old.entry.scoped_relative_path != entry.scoped_relative_path
|| old.entry.artifact_key != entry.artifact_key
|| old.entry.indexes != entry.indexes;
if changed {
old.lifecycle.revoke_and_mint(false);
old.lifecycle.resume_unbound();
replaced.push(entry.literal_path.clone());
}
next.insert(
entry.literal_path.clone(),
ManagedEntry {
entry: entry.clone(),
lifecycle: Arc::clone(&old.lifecycle),
},
);
} else {
added.push(entry.literal_path.clone());
next.insert(
entry.literal_path.clone(),
ManagedEntry {
entry: entry.clone(),
lifecycle: Arc::new(ArtifactLifecycle::default()),
},
);
}
}
for (literal_path, old) in &previous {
if !next.contains_key(literal_path) {
old.lifecycle.revoke_and_mint(false);
standing_roots::delete_standing_root(&conn, literal_path)?;
removed.push(literal_path.clone());
}
}
if !previous_first_snapshot && !entries.is_empty() {
for entry in next.values() {
mark_kinds_needing_strict_verify(
&mut conn,
&entry.entry.literal_path,
&entry.entry.indexes,
)?;
}
}
state.observed_first_snapshot |= !entries.is_empty();
state.entries = next;
Ok(StandingReconcileReport {
active_entries: entries,
added,
removed,
replaced,
})
}
pub fn entries(&self) -> Vec<StandingRootEntry> {
self.inner
.state
.lock()
.entries
.values()
.map(|entry| entry.entry.clone())
.collect()
}
pub fn admit_build(&self, literal_path: &str) -> Option<StandingBuildAdmission> {
self.inner
.state
.lock()
.entries
.get(literal_path)
.and_then(|entry| entry.lifecycle.admit())
}
pub fn begin_case_a_bind(
&self,
literal_path: &str,
) -> Result<BindTransition, StandingRootsError> {
let (lifecycle, indexes) = {
let state = self.inner.state.lock();
let entry = state.entries.get(literal_path).ok_or_else(|| {
StandingRootsError::UnknownEntry {
literal_path: literal_path.to_string(),
}
})?;
(Arc::clone(&entry.lifecycle), entry.entry.indexes.clone())
};
let admission_epoch = lifecycle.revoke_and_mint(true);
let mut conn = self.open_database()?;
mark_kinds_needing_strict_verify(&mut conn, literal_path, &indexes)?;
let cancelled_builds = lifecycle.state.lock().active_builds;
Ok(BindTransition {
admission_epoch,
cancelled_builds,
})
}
pub fn wait_for_case_a_checkpoint(
&self,
literal_path: &str,
) -> Result<bool, StandingRootsError> {
let lifecycle = self.lifecycle(literal_path)?;
Ok(lifecycle.wait_for_checkpoint_acknowledgement(Duration::from_secs(2)))
}
pub fn resume_after_session(
&self,
literal_path: &str,
session_maintained: &[IndexKind],
) -> Result<(), StandingRootsError> {
let (lifecycle, missing) = {
let state = self.inner.state.lock();
let entry = state.entries.get(literal_path).ok_or_else(|| {
StandingRootsError::UnknownEntry {
literal_path: literal_path.to_string(),
}
})?;
let missing = entry
.entry
.indexes
.iter()
.copied()
.filter(|kind| !session_maintained.contains(kind))
.collect::<Vec<_>>();
(Arc::clone(&entry.lifecycle), missing)
};
if !lifecycle.bind_pending() {
return Ok(());
}
if !missing.is_empty() {
let mut conn = self.open_database()?;
mark_kinds_needing_strict_verify(&mut conn, literal_path, &missing)?;
}
lifecycle.resume_unbound();
Ok(())
}
pub fn mark_observation_gap(
&self,
literal_path: &str,
kinds: &[IndexKind],
) -> Result<(), StandingRootsError> {
let mut conn = self.open_database()?;
mark_kinds_needing_strict_verify(&mut conn, literal_path, kinds)
}
pub fn record_strict_verification(
&self,
literal_path: &str,
kind: IndexKind,
) -> Result<(), StandingRootsError> {
let mut conn = self.open_database()?;
standing_roots::record_successful_strict_verification(
&mut conn,
literal_path,
kind,
now_ms(),
)?;
Ok(())
}
pub fn route_explicit_path(
&self,
query_path: &Path,
kind: IndexKind,
) -> Result<StandingRoute, StandingRouteError> {
let query_path = canonicalize_existing_ancestor(query_path);
let mut candidates = self
.entries()
.into_iter()
.filter(|entry| query_path.starts_with(&entry.resolved_target))
.collect::<Vec<_>>();
candidates.sort_by(|left, right| {
path_depth(&right.resolved_target)
.cmp(&path_depth(&left.resolved_target))
.then_with(|| left.config_order.cmp(&right.config_order))
});
let Some(deepest) = candidates.first().cloned() else {
return Err(StandingRouteError::NoContainingEntry);
};
let Some(entry) = candidates
.into_iter()
.find(|entry| entry.indexes.contains(&kind))
else {
return Err(StandingRouteError::KindUnavailable {
deepest_literal_path: deepest.literal_path,
deepest_selection: deepest.indexes,
});
};
let conn = self
.open_database()
.map_err(|_| StandingRouteError::NoContainingEntry)?;
if standing_roots::needs_strict_verify(&conn, &entry.literal_path, kind)
.ok()
.flatten()
.unwrap_or(true)
{
return Err(StandingRouteError::StrictVerificationRequired {
literal_path: entry.literal_path,
kind,
});
}
Ok(StandingRoute { entry })
}
pub fn publish_if_current<R>(
&self,
literal_path: &str,
admission: StandingPublicationAdmission,
writer_lease: &WriterLease,
fingerprint_is_current: impl FnOnce() -> bool,
generation_is_current: impl FnOnce() -> bool,
publish_and_rename: impl FnOnce() -> R,
) -> Result<Option<R>, StandingRootsError> {
let lifecycle = self.lifecycle(literal_path)?;
Ok(lifecycle
.publication_epoch
.run_if_current(admission.publication_epoch, || {
if lifecycle.admission_epoch.load(Ordering::SeqCst) != admission.admission_epoch
|| !lifecycle.state.lock().standing_admission_open
|| !writer_lease.verify().unwrap_or(false)
|| !fingerprint_is_current()
|| !generation_is_current()
{
return None;
}
Some(publish_and_rename())
})
.flatten())
}
fn lifecycle(&self, literal_path: &str) -> Result<Arc<ArtifactLifecycle>, StandingRootsError> {
self.inner
.state
.lock()
.entries
.get(literal_path)
.map(|entry| Arc::clone(&entry.lifecycle))
.ok_or_else(|| StandingRootsError::UnknownEntry {
literal_path: literal_path.to_string(),
})
}
fn open_database(&self) -> Result<rusqlite::Connection, StandingRootsError> {
let path = self
.inner
.database_path
.lock()
.clone()
.unwrap_or_else(|| crate::bash_background::storage_dir(None).join("aft.db"));
Ok(crate::db::open(&path)?)
}
}
fn resolve_entries(config: &Config) -> Result<Vec<StandingRootEntry>, StandingRootsError> {
let resolved = config
.index
.roots
.iter()
.map(|root| resolve_standing_root(&root.path))
.collect::<Result<Vec<ResolvedStandingRoot>, _>>()?;
reject_duplicate_artifact_keys(&resolved)?;
Ok(config
.index
.roots
.iter()
.zip(resolved)
.enumerate()
.map(|(config_order, (root, resolved))| StandingRootEntry {
literal_path: root.path.clone(),
resolved_target: PathBuf::from(resolved.resolved_target),
resolved_git_toplevel: resolved.resolved_git_toplevel.map(PathBuf::from),
scoped_relative_path: resolved.scoped_relative_path.map(PathBuf::from),
artifact_key: resolved.artifact_key,
indexes: root.indexes.clone(),
config_order,
})
.collect())
}
fn mark_kinds_needing_strict_verify(
conn: &mut rusqlite::Connection,
literal_path: &str,
kinds: &[IndexKind],
) -> Result<(), StandingRootsError> {
let tx = conn
.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)
.map_err(standing_roots::StandingRootError::from)?;
for kind in kinds {
let updated = tx
.execute(
"UPDATE standing_root_freshness
SET needs_strict_verify = 1, strict_verified_at = NULL
WHERE literal_path = ?1 AND index_kind = ?2",
rusqlite::params![literal_path, kind.as_str()],
)
.map_err(standing_roots::StandingRootError::from)?;
if updated != 1 {
return Err(StandingRootsError::Database(
standing_roots::StandingRootError::MissingFreshnessRow {
literal_path: literal_path.to_string(),
kind: *kind,
},
));
}
}
tx.commit()
.map_err(standing_roots::StandingRootError::from)?;
Ok(())
}
fn path_depth(path: &Path) -> usize {
path.components().count()
}
fn canonicalize_existing_ancestor(path: &Path) -> PathBuf {
if let Ok(canonical) = std::fs::canonicalize(path) {
return canonical;
}
let mut missing = Vec::new();
let mut ancestor = path;
while !ancestor.exists() {
let Some(name) = ancestor.file_name() else {
return path.to_path_buf();
};
missing.push(name.to_os_string());
let Some(parent) = ancestor.parent() else {
return path.to_path_buf();
};
ancestor = parent;
}
let mut canonical = std::fs::canonicalize(ancestor).unwrap_or_else(|_| ancestor.to_path_buf());
for component in missing.iter().rev() {
canonical.push(component);
}
canonical
}
fn now_ms() -> i64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_millis()
.try_into()
.unwrap_or(i64::MAX)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::config::{IndexConfig, IndexRootConfig};
use tempfile::tempdir;
fn config(storage: &Path, roots: Vec<IndexRootConfig>) -> Config {
Config {
storage_dir: Some(storage.to_path_buf()),
index: IndexConfig { roots },
..Config::default()
}
}
fn root(path: &Path, indexes: Vec<IndexKind>) -> IndexRootConfig {
IndexRootConfig {
path: path.to_string_lossy().into_owned(),
indexes,
}
}
#[test]
fn reconciliation_marks_restart_and_new_kind_for_strict_verification() {
let storage = tempdir().unwrap();
let first = tempdir().unwrap();
let roots = StandingRoots::default();
let mut cfg = config(
storage.path(),
vec![root(first.path(), vec![IndexKind::Search])],
);
roots.reconcile(&cfg).unwrap();
roots
.record_strict_verification(first.path().to_str().unwrap(), IndexKind::Search)
.unwrap();
cfg.index.roots[0].indexes = vec![IndexKind::Search, IndexKind::Callgraph];
let report = roots.reconcile(&cfg).unwrap();
assert_eq!(report.replaced, vec![first.path().to_string_lossy()]);
let conn = crate::db::open(&storage.path().join("aft.db")).unwrap();
assert!(standing_roots::needs_strict_verify(
&conn,
first.path().to_str().unwrap(),
IndexKind::Callgraph
)
.unwrap()
.unwrap());
}
#[test]
fn daemon_startup_empty_pass_marks_existing_snapshot_strict_on_first_configured_pass() {
let storage = tempdir().unwrap();
let root_dir = tempdir().unwrap();
let cfg = config(
storage.path(),
vec![root(root_dir.path(), vec![IndexKind::Search])],
);
let first = StandingRoots::default();
first.reconcile(&cfg).unwrap();
first
.record_strict_verification(root_dir.path().to_str().unwrap(), IndexKind::Search)
.unwrap();
let restarted = StandingRoots::default();
restarted.reconcile(&Config::default()).unwrap();
restarted.reconcile(&cfg).unwrap();
let conn = crate::db::open(&storage.path().join("aft.db")).unwrap();
assert!(standing_roots::needs_strict_verify(
&conn,
root_dir.path().to_str().unwrap(),
IndexKind::Search
)
.unwrap()
.unwrap());
}
#[test]
fn maintenance_snapshot_preserves_configuration_and_fixed_kind_order() {
let storage = tempdir().unwrap();
let first = tempdir().unwrap();
let second = tempdir().unwrap();
let roots = StandingRoots::default();
let cfg = config(
storage.path(),
vec![
root(first.path(), vec![IndexKind::Callgraph, IndexKind::Search]),
root(second.path(), vec![IndexKind::Semantic]),
],
);
let report = roots.reconcile(&cfg).unwrap();
let scheduled = report
.active_entries
.iter()
.flat_map(|entry| {
IndexKind::ALL
.into_iter()
.filter(move |kind| entry.indexes.contains(kind))
.map(move |kind| (entry.literal_path.clone(), kind))
})
.collect::<Vec<_>>();
assert_eq!(
scheduled,
vec![
(
first.path().to_string_lossy().into_owned(),
IndexKind::Search
),
(
first.path().to_string_lossy().into_owned(),
IndexKind::Callgraph
),
(
second.path().to_string_lossy().into_owned(),
IndexKind::Semantic
),
]
);
}
#[test]
fn route_falls_back_to_shallower_entry_and_discloses_it() {
let storage = tempdir().unwrap();
let outer = tempdir().unwrap();
let inner = outer.path().join("inner");
std::fs::create_dir(&inner).unwrap();
let roots = StandingRoots::default();
let cfg = config(
storage.path(),
vec![
root(outer.path(), vec![IndexKind::Search]),
root(&inner, vec![IndexKind::Callgraph]),
],
);
roots.reconcile(&cfg).unwrap();
roots
.record_strict_verification(outer.path().to_str().unwrap(), IndexKind::Search)
.unwrap();
let route = roots
.route_explicit_path(&inner.join("file.rs"), IndexKind::Search)
.unwrap();
assert_eq!(route.entry.literal_path, outer.path().to_string_lossy());
}
#[test]
fn publication_is_a_noop_after_bind_epoch_revocation() {
let storage = tempdir().unwrap();
let root_dir = tempdir().unwrap();
let roots = StandingRoots::default();
let cfg = config(
storage.path(),
vec![root(root_dir.path(), vec![IndexKind::Search])],
);
let mut report = roots.reconcile(&cfg).unwrap();
let entry = report.active_entries.remove(0);
let build = roots.admit_build(&entry.literal_path).unwrap();
crate::root_cache::configure_artifact_access(
&entry.resolved_target,
&entry.artifact_key,
false,
);
let cache_dir = storage.path().join("index").join(&entry.artifact_key);
let lease = crate::root_cache::WriterLease::acquire_shared(
crate::root_cache::RootCacheDomain::Index,
&cache_dir,
&entry.artifact_key,
&entry.resolved_target,
)
.unwrap()
.unwrap();
roots.begin_case_a_bind(&entry.literal_path).unwrap();
let published = std::sync::atomic::AtomicBool::new(false);
let result = roots
.publish_if_current(
&entry.literal_path,
build.publication,
&lease,
|| true,
|| true,
|| published.store(true, Ordering::SeqCst),
)
.unwrap();
assert!(result.is_none());
assert!(!published.load(Ordering::SeqCst));
}
#[test]
fn publication_fence_holds_epoch_mutex_through_final_rename() {
let storage = tempdir().unwrap();
let root_dir = tempdir().unwrap();
let roots = StandingRoots::default();
let cfg = config(
storage.path(),
vec![root(root_dir.path(), vec![IndexKind::Search])],
);
let mut report = roots.reconcile(&cfg).unwrap();
let entry = report.active_entries.remove(0);
let admission = roots.admit_build(&entry.literal_path).unwrap();
crate::root_cache::configure_artifact_access(
&entry.resolved_target,
&entry.artifact_key,
false,
);
let cache_dir = storage.path().join("index").join(&entry.artifact_key);
let lease = crate::root_cache::WriterLease::acquire_shared(
crate::root_cache::RootCacheDomain::Index,
&cache_dir,
&entry.artifact_key,
&entry.resolved_target,
)
.unwrap()
.unwrap();
let (rename_entered_tx, rename_entered_rx) = std::sync::mpsc::channel();
let (release_rename_tx, release_rename_rx) = std::sync::mpsc::channel();
let publish_roots = roots.clone();
let publish_literal = entry.literal_path.clone();
let publisher = std::thread::spawn(move || {
publish_roots
.publish_if_current(
&publish_literal,
admission.publication,
&lease,
|| true,
|| true,
|| {
rename_entered_tx.send(()).unwrap();
release_rename_rx.recv().unwrap();
},
)
.unwrap()
});
rename_entered_rx.recv().unwrap();
let (bind_done_tx, bind_done_rx) = std::sync::mpsc::channel();
let bind_roots = roots.clone();
let bind_literal = entry.literal_path.clone();
let binder = std::thread::spawn(move || {
bind_done_tx
.send(bind_roots.begin_case_a_bind(&bind_literal))
.unwrap();
});
assert!(bind_done_rx
.recv_timeout(Duration::from_millis(100))
.is_err());
release_rename_tx.send(()).unwrap();
assert!(publisher.join().unwrap().is_some());
bind_done_rx
.recv_timeout(Duration::from_secs(1))
.unwrap()
.unwrap();
binder.join().unwrap();
}
#[test]
fn superseded_snapshot_publication_is_a_noop() {
let storage = tempdir().unwrap();
let root_dir = tempdir().unwrap();
let roots = StandingRoots::default();
let mut cfg = config(
storage.path(),
vec![root(root_dir.path(), vec![IndexKind::Search])],
);
let mut initial = roots.reconcile(&cfg).unwrap();
let entry = initial.active_entries.remove(0);
let build = roots.admit_build(&entry.literal_path).unwrap();
crate::root_cache::configure_artifact_access(
&entry.resolved_target,
&entry.artifact_key,
false,
);
cfg.index.roots[0].indexes = vec![IndexKind::Semantic];
let report = roots.reconcile(&cfg).unwrap();
assert_eq!(report.replaced, vec![entry.literal_path.clone()]);
let cache_dir = storage.path().join("index").join(&entry.artifact_key);
let lease = crate::root_cache::WriterLease::acquire_shared(
crate::root_cache::RootCacheDomain::Index,
&cache_dir,
&entry.artifact_key,
&entry.resolved_target,
)
.unwrap()
.unwrap();
assert!(roots
.publish_if_current(
&entry.literal_path,
build.publication,
&lease,
|| true,
|| true,
|| (),
)
.unwrap()
.is_none());
}
#[test]
fn crash_before_freshness_clear_blocks_verify_on_query_until_transactional_clear() {
let storage = tempdir().unwrap();
let root_dir = tempdir().unwrap();
let cfg = config(
storage.path(),
vec![root(root_dir.path(), vec![IndexKind::Search])],
);
let first = StandingRoots::default();
first.reconcile(&cfg).unwrap();
first
.record_strict_verification(root_dir.path().to_str().unwrap(), IndexKind::Search)
.unwrap();
first
.mark_observation_gap(root_dir.path().to_str().unwrap(), &[IndexKind::Search])
.unwrap();
drop(first);
let restarted = StandingRoots::default();
restarted.reconcile(&cfg).unwrap();
assert!(matches!(
restarted.route_explicit_path(&root_dir.path().join("missing.rs"), IndexKind::Search),
Err(StandingRouteError::StrictVerificationRequired { .. })
));
restarted
.record_strict_verification(root_dir.path().to_str().unwrap(), IndexKind::Search)
.unwrap();
assert!(restarted
.route_explicit_path(&root_dir.path().join("missing.rs"), IndexKind::Search)
.is_ok());
}
#[test]
fn verify_on_query_retains_observation_gap_between_passes() {
let storage = tempdir().unwrap();
let root_dir = tempdir().unwrap();
let roots = StandingRoots::default();
let cfg = config(
storage.path(),
vec![root(root_dir.path(), vec![IndexKind::Search])],
);
let literal = root_dir.path().to_str().unwrap();
roots.reconcile(&cfg).unwrap();
roots
.record_strict_verification(literal, IndexKind::Search)
.unwrap();
roots
.mark_observation_gap(literal, &[IndexKind::Search])
.unwrap();
assert!(matches!(
roots.route_explicit_path(&root_dir.path().join("query.rs"), IndexKind::Search),
Err(StandingRouteError::StrictVerificationRequired { .. })
));
}
#[test]
fn suspension_edit_resume_requires_strict_verification_before_query() {
let storage = tempdir().unwrap();
let root_dir = tempdir().unwrap();
let roots = StandingRoots::default();
let cfg = config(
storage.path(),
vec![root(root_dir.path(), vec![IndexKind::Search])],
);
roots.reconcile(&cfg).unwrap();
roots
.record_strict_verification(root_dir.path().to_str().unwrap(), IndexKind::Search)
.unwrap();
let build = roots
.admit_build(root_dir.path().to_str().unwrap())
.unwrap();
roots
.begin_case_a_bind(root_dir.path().to_str().unwrap())
.unwrap();
assert!(build.checkpoint());
std::fs::write(root_dir.path().join("edited.rs"), "fn changed() {}\n").unwrap();
roots
.resume_after_session(root_dir.path().to_str().unwrap(), &[])
.unwrap();
assert!(matches!(
roots.route_explicit_path(&root_dir.path().join("edited.rs"), IndexKind::Search),
Err(StandingRouteError::StrictVerificationRequired { .. })
));
roots
.record_strict_verification(root_dir.path().to_str().unwrap(), IndexKind::Search)
.unwrap();
assert!(roots
.route_explicit_path(&root_dir.path().join("edited.rs"), IndexKind::Search)
.is_ok());
}
#[test]
fn shared_key_handoff_preserves_session_proven_kind_and_marks_other_kind() {
let storage = tempdir().unwrap();
let root_dir = tempdir().unwrap();
let roots = StandingRoots::default();
let cfg = config(
storage.path(),
vec![root(
root_dir.path(),
vec![IndexKind::Search, IndexKind::Semantic],
)],
);
roots.reconcile(&cfg).unwrap();
let literal = root_dir.path().to_str().unwrap();
roots
.record_strict_verification(literal, IndexKind::Search)
.unwrap();
roots
.record_strict_verification(literal, IndexKind::Semantic)
.unwrap();
roots.begin_case_a_bind(literal).unwrap();
roots
.record_strict_verification(literal, IndexKind::Search)
.unwrap();
roots
.resume_after_session(literal, &[IndexKind::Search])
.unwrap();
let conn = crate::db::open(&storage.path().join("aft.db")).unwrap();
assert!(
!standing_roots::needs_strict_verify(&conn, literal, IndexKind::Search)
.unwrap()
.unwrap()
);
assert!(
standing_roots::needs_strict_verify(&conn, literal, IndexKind::Semantic)
.unwrap()
.unwrap()
);
}
#[test]
fn configuration_add_modify_and_remove_mint_boundaries_and_delete_rows() {
let storage = tempdir().unwrap();
let root_dir = tempdir().unwrap();
let roots = StandingRoots::default();
let mut cfg = config(
storage.path(),
vec![root(root_dir.path(), vec![IndexKind::Search])],
);
let literal = root_dir.path().to_str().unwrap().to_string();
assert_eq!(roots.reconcile(&cfg).unwrap().added, vec![literal.clone()]);
let admitted = roots.admit_build(&literal).unwrap();
cfg.index.roots[0].indexes = vec![IndexKind::Callgraph];
let report = roots.reconcile(&cfg).unwrap();
assert_eq!(report.replaced, vec![literal.clone()]);
let replacement = roots
.admit_build(&literal)
.expect("replacement reopens only a new admission");
assert_ne!(
admitted.publication.admission_epoch,
replacement.publication.admission_epoch
);
let removed = roots.reconcile(&Config::default()).unwrap();
assert_eq!(removed.removed, vec![literal.clone()]);
let conn = crate::db::open(&storage.path().join("aft.db")).unwrap();
assert!(standing_roots::get_standing_root(&conn, &literal)
.unwrap()
.is_none());
}
#[test]
fn bounded_join_proceeds_after_two_seconds_without_checkpoint() {
let storage = tempdir().unwrap();
let root_dir = tempdir().unwrap();
let roots = StandingRoots::default();
let cfg = config(
storage.path(),
vec![root(root_dir.path(), vec![IndexKind::Search])],
);
roots.reconcile(&cfg).unwrap();
let _build = roots
.admit_build(root_dir.path().to_str().unwrap())
.unwrap();
roots
.begin_case_a_bind(root_dir.path().to_str().unwrap())
.unwrap();
let before = std::time::Instant::now();
assert!(!roots
.wait_for_case_a_checkpoint(root_dir.path().to_str().unwrap())
.unwrap());
assert!(before.elapsed() >= Duration::from_secs(2));
assert!(before.elapsed() < Duration::from_millis(2300));
}
#[test]
fn bind_revokes_admission_and_checkpoint_join_is_bounded() {
let storage = tempdir().unwrap();
let root_dir = tempdir().unwrap();
let roots = StandingRoots::default();
let cfg = config(
storage.path(),
vec![root(root_dir.path(), vec![IndexKind::Search])],
);
roots.reconcile(&cfg).unwrap();
let build = roots
.admit_build(root_dir.path().to_str().unwrap())
.unwrap();
let before = std::time::Instant::now();
roots
.begin_case_a_bind(root_dir.path().to_str().unwrap())
.unwrap();
assert!(build.checkpoint());
assert!(roots
.wait_for_case_a_checkpoint(root_dir.path().to_str().unwrap())
.unwrap());
assert!(before.elapsed() < Duration::from_secs(2));
assert!(roots
.admit_build(root_dir.path().to_str().unwrap())
.is_none());
}
}