Skip to main content

aft/
root_cache.rs

1//! These files coordinate safe access to a project-root cache: a writer lease
2//! ensures only one process updates the cache at a time, and read-marker files
3//! let cleanup see which readers are still using the cache.
4//!
5//! Writer leases are stored at `<storage>/callgraph/<artifact_cache_key>/writer.lease`
6//! and `<storage>/inspect/<project_scope_key>/writer.lease`. They use the
7//! `fs_lock` JSON format with a `writer_epoch` nonce so a writer can detect if
8//! another process has taken over before publishing changes or starting SQLite
9//! write transactions.
10//!
11//! Read markers track active SQLite readers so cache cleanup can tell when it is
12//! safe to remove old data. They are stored under
13//! `<cache-domain>/readers/<generation-label>/<pid>.<hostname>.<created_at_ms>.<seq>.json`;
14//! the JSON records the process identity and creation time, mtime is used as a
15//! heartbeat for cleanup across hosts, and the PID is used for cleanup on the
16//! same host. Marker files are created `0600` so they do not expose checkout
17//! activity or let another local user delete a protected marker.
18
19use std::collections::{HashMap, HashSet};
20use std::ffi::OsStr;
21use std::fs::{self, File, OpenOptions};
22use std::io::{self, Write};
23use std::path::{Path, PathBuf};
24use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
25use std::sync::{Arc, Mutex, OnceLock, TryLockError, Weak};
26use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
27
28use serde::{Deserialize, Serialize};
29
30use crate::fs_lock;
31
32static MARKER_SEQ: AtomicU64 = AtomicU64::new(0);
33static LIVE_SCOPE_KEYS: OnceLock<Mutex<HashMap<(PathBuf, String), usize>>> = OnceLock::new();
34
35/// Read-marker heartbeats refresh no more often than the filesystem lock
36/// heartbeat. Active readers piggyback this on normal read paths instead of
37/// spawning a thread per connection.
38pub const READ_MARKER_TOUCH_INTERVAL_MS: u64 = fs_lock::HEARTBEAT_INTERVAL_MS;
39/// Cross-host markers cannot use local PID liveness, so they expire after the
40/// same conservative 5x stale-heartbeat window used by filesystem locks.
41pub const READ_MARKER_CROSS_HOST_STALE_MS: u64 = fs_lock::STALE_HEARTBEAT_MS * 5;
42// Process start timestamps are not always millisecond-precise across OS APIs.
43// A one-second grace keeps a marker created immediately after process launch
44// attached to that process while still identifying clear PID reuse.
45const PROCESS_START_TIME_GRACE_MS: u64 = 1_000;
46
47#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
48pub enum RootCacheDomain {
49    Callgraph,
50    Inspect,
51    /// Search-index publication for an unbound standing root. This remains a
52    /// root-keyed WriterLease domain, separate from the session-owned cache.
53    Index,
54}
55
56impl RootCacheDomain {
57    pub fn as_str(self) -> &'static str {
58        match self {
59            RootCacheDomain::Callgraph => "callgraph",
60            RootCacheDomain::Inspect => "inspect",
61            RootCacheDomain::Index => "index",
62        }
63    }
64}
65
66/// Serializes artifact supersession with the final disk publication step.
67/// Advancing an epoch either happens before a stale worker checks and prevents
68/// its publish, or after that worker has fully published the still-current
69/// generation. This closes the check-then-publish race of a bare atomic epoch.
70#[derive(Clone, Default)]
71pub struct ArtifactPublishEpoch {
72    current: Arc<parking_lot::Mutex<u64>>,
73}
74
75impl ArtifactPublishEpoch {
76    pub fn next(&self) -> u64 {
77        let mut current = self.current.lock();
78        *current = current.wrapping_add(1);
79        *current
80    }
81
82    pub fn current(&self) -> u64 {
83        *self.current.lock()
84    }
85
86    pub fn is_current(&self, expected: u64) -> bool {
87        *self.current.lock() == expected
88    }
89
90    pub fn run_if_current<R>(&self, expected: u64, publish: impl FnOnce() -> R) -> Option<R> {
91        let current = self.current.lock();
92        if *current != expected {
93            return None;
94        }
95        Some(publish())
96    }
97}
98
99pub struct WriterLease {
100    domain: RootCacheDomain,
101    key: String,
102    path: PathBuf,
103    epoch: String,
104    guard: Mutex<fs_lock::LockGuard>,
105}
106
107#[derive(Clone, Debug, Eq, Hash, PartialEq)]
108struct ProcessLeaseKey {
109    domain: RootCacheDomain,
110    cache_dir: PathBuf,
111}
112
113#[derive(Clone, Debug, Eq, Hash, PartialEq)]
114struct WriterLeaseAcquisitionKey {
115    domain: RootCacheDomain,
116    key: String,
117    project_root: PathBuf,
118}
119
120static PROCESS_LEASES: OnceLock<Mutex<HashMap<ProcessLeaseKey, Weak<WriterLease>>>> =
121    OnceLock::new();
122// Same-root callers share this short-lived gate so only one thread performs the
123// filesystem lease attempt, while different roots do not wait on the registry
124// mutex during stat/probe/create/heartbeat work.
125static PROCESS_LEASE_ACQUISITIONS: OnceLock<Mutex<HashMap<ProcessLeaseKey, Weak<Mutex<()>>>>> =
126    OnceLock::new();
127static WRITER_LEASE_ACQUISITION_COUNTS: OnceLock<Mutex<HashMap<WriterLeaseAcquisitionKey, usize>>> =
128    OnceLock::new();
129static WRITER_LEASE_ACQUISITION_COUNTER_ENABLED: AtomicBool = AtomicBool::new(false);
130static CONFIGURED_ARTIFACT_ACCESS: OnceLock<Mutex<HashMap<PathBuf, ArtifactAccess>>> =
131    OnceLock::new();
132static WARNED_BORROW_ONLY_WRITES: OnceLock<Mutex<HashSet<(PathBuf, PathBuf)>>> = OnceLock::new();
133
134/// Root-scoped capability that distinguishes shared repository artifacts from
135/// mutable state private to one checkout.
136#[derive(Clone, Debug)]
137pub struct ArtifactAccess {
138    project_root: PathBuf,
139    shared_key: String,
140    private_key: String,
141    borrow_only_shared: bool,
142    registered: bool,
143}
144
145impl ArtifactAccess {
146    fn configured(project_root: &Path, shared_key: &str, borrow_only_shared: bool) -> Self {
147        let project_root = canonical_root(project_root);
148        Self {
149            private_key: crate::path_identity::project_scope_key(&project_root),
150            project_root,
151            shared_key: shared_key.to_string(),
152            borrow_only_shared,
153            registered: true,
154        }
155    }
156
157    /// Resolve the capability registered during configure. Unregistered roots
158    /// receive a borrow-only capability without probing Git on the store path.
159    pub fn for_root(project_root: &Path) -> Self {
160        let project_root = canonical_root(project_root);
161        if let Some(access) = configured_artifact_access()
162            .lock()
163            .ok()
164            .and_then(|access| access.get(&project_root).cloned())
165        {
166            return access;
167        }
168        // Unregistered root: fail closed without spawning git probes. Configure
169        // registers every bound root before any store acquisition, so landing
170        // here means a direct artifact-API caller on an unconfigured root —
171        // treating it as borrow-only keeps shared artifacts safe and keeps this
172        // path subprocess-free (a git probe here has unbounded latency and can
173        // run on latency-critical threads).
174        crate::slog_warn!(
175            "artifact access requested for unconfigured root {}; defaulting to borrow-only",
176            project_root.display()
177        );
178        let shared_key = crate::path_identity::project_scope_key(&project_root);
179        Self {
180            private_key: shared_key.clone(),
181            project_root,
182            shared_key,
183            borrow_only_shared: true,
184            registered: false,
185        }
186    }
187
188    /// Return whether this root may write the keyed artifact, logging the first
189    /// denial for each concrete path so read-only degradation stays observable.
190    pub fn allows_write(&self, artifact_key: &str, write_path: &Path) -> bool {
191        let writes_keyed_dir = write_path
192            .ancestors()
193            .any(|ancestor| ancestor.file_name() == Some(OsStr::new(artifact_key)));
194        if !self.borrow_only_shared
195            || artifact_key != self.shared_key
196            || artifact_key == self.private_key
197            || !writes_keyed_dir
198        {
199            return true;
200        }
201        self.warn_write_denied_once(write_path);
202        false
203    }
204
205    fn allows_writer_lease(
206        &self,
207        domain: RootCacheDomain,
208        artifact_key: &str,
209        write_path: &Path,
210    ) -> bool {
211        self.allows_writer_lease_inner(domain, artifact_key, write_path)
212    }
213
214    fn allows_writer_lease_inner(
215        &self,
216        domain: RootCacheDomain,
217        artifact_key: &str,
218        write_path: &Path,
219    ) -> bool {
220        // Classify the layout from a comparison-local canonical copy rather
221        // than trusting key equality. A key derived while the root's final
222        // component was absent can encode a symlink alias (for example,
223        // /var versus /private/var) that disappears after the root exists.
224        let comparison_cache_dir = write_path.parent().map(canonical_process_lease_dir);
225        let writes_keyed_dir = comparison_cache_dir.as_deref().and_then(Path::file_name)
226            == Some(OsStr::new(artifact_key));
227        let writes_root_keyed_dir = comparison_cache_dir
228            .as_deref()
229            .and_then(Path::parent)
230            .and_then(Path::file_name)
231            == Some(OsStr::new(domain.as_str()));
232        if !self.borrow_only_shared {
233            return true;
234        }
235        if domain != RootCacheDomain::Callgraph
236            && writes_keyed_dir
237            && artifact_key == self.private_key
238        {
239            return true;
240        }
241        if !self.registered && !writes_root_keyed_dir {
242            // Caller-owned scratch paths are outside the shared root-keyed
243            // cache layout, so this capability does not govern them.
244            return true;
245        }
246        self.warn_write_denied_once(write_path);
247        false
248    }
249
250    fn warn_write_denied_once(&self, write_path: &Path) {
251        let warning_key = (self.project_root.clone(), write_path.to_path_buf());
252        let should_warn = WARNED_BORROW_ONLY_WRITES
253            .get_or_init(|| Mutex::new(HashSet::new()))
254            .lock()
255            .map(|mut warned| {
256                if warned.len() >= 4_096 {
257                    warned.clear();
258                }
259                warned.insert(warning_key)
260            })
261            .unwrap_or(false);
262        if should_warn {
263            crate::slog_warn!(
264                "borrow-only root denied shared artifact write at {} (configured shared key {})",
265                write_path.display(),
266                self.shared_key
267            );
268        }
269    }
270}
271
272fn configured_artifact_access() -> &'static Mutex<HashMap<PathBuf, ArtifactAccess>> {
273    CONFIGURED_ARTIFACT_ACCESS.get_or_init(|| Mutex::new(HashMap::new()))
274}
275
276/// Track scope keys belonging to roots currently bound in this process. The
277/// inspect sweep snapshots this registry at publication time so a live actor's
278/// cache cannot be mistaken for a reclaimed worktree cache.
279fn live_scope_keys() -> &'static Mutex<HashMap<(PathBuf, String), usize>> {
280    LIVE_SCOPE_KEYS.get_or_init(|| Mutex::new(HashMap::new()))
281}
282
283pub(crate) fn register_live_scope(storage_root: &Path, project_root: &Path) {
284    let storage_root = canonical_root(storage_root);
285    let scope_key = crate::path_identity::project_scope_key(project_root);
286    let mut scopes = match live_scope_keys().lock() {
287        Ok(scopes) => scopes,
288        Err(poisoned) => poisoned.into_inner(),
289    };
290    *scopes.entry((storage_root, scope_key)).or_default() += 1;
291}
292
293pub(crate) fn unregister_live_scope(storage_root: &Path, project_root: &Path) {
294    let storage_root = canonical_root(storage_root);
295    let scope_key = crate::path_identity::project_scope_key(project_root);
296    let mut scopes = match live_scope_keys().lock() {
297        Ok(scopes) => scopes,
298        Err(poisoned) => poisoned.into_inner(),
299    };
300    let key = (storage_root, scope_key);
301    if let Some(count) = scopes.get_mut(&key) {
302        *count = count.saturating_sub(1);
303        if *count == 0 {
304            scopes.remove(&key);
305        }
306    }
307}
308
309pub(crate) fn live_scope_keys_for_storage(storage_root: &Path) -> HashSet<String> {
310    let storage_root = canonical_root(storage_root);
311    let scopes = match live_scope_keys().lock() {
312        Ok(scopes) => scopes,
313        Err(poisoned) => poisoned.into_inner(),
314    };
315    scopes
316        .iter()
317        .filter(|((root, _), count)| root == &storage_root && **count > 0)
318        .map(|((_, key), _)| key.clone())
319        .collect()
320}
321
322/// Register the worktree topology already detected by configure so artifact
323/// APIs can enforce it without repeating a Git subprocess on every write path.
324pub fn configure_artifact_access(project_root: &Path, shared_key: &str, borrow_only_shared: bool) {
325    let access = ArtifactAccess::configured(project_root, shared_key, borrow_only_shared);
326    if let Ok(mut configured) = configured_artifact_access().lock() {
327        // Bounded, but never a wholesale clear: dropping still-live roots'
328        // capabilities would silently flip them onto the fail-closed
329        // (borrow-only) fallback. Evict an arbitrary other entry instead —
330        // any evicted-but-live root re-registers on its next configure.
331        if configured.len() >= 4_096 && !configured.contains_key(&access.project_root) {
332            if let Some(evict) = configured.keys().next().cloned() {
333                configured.remove(&evict);
334            }
335        }
336        configured.insert(access.project_root.clone(), access);
337    }
338}
339
340fn canonical_root(project_root: &Path) -> PathBuf {
341    // Nearest-existing-ancestor canonicalization, not a raw-spelling fallback:
342    // registration can precede the directory's creation (and lookups can follow
343    // it), and on macOS the raw tempdir spelling (/var/...) differs from the
344    // canonical one (/private/var/...). Keying the access registry by whichever
345    // spelling happened to win the existence race made a registered root miss
346    // its own entry and fail closed.
347    canonical_process_lease_dir(project_root)
348}
349
350fn process_leases() -> &'static Mutex<HashMap<ProcessLeaseKey, Weak<WriterLease>>> {
351    PROCESS_LEASES.get_or_init(|| Mutex::new(HashMap::new()))
352}
353
354fn process_lease_acquisitions() -> &'static Mutex<HashMap<ProcessLeaseKey, Weak<Mutex<()>>>> {
355    PROCESS_LEASE_ACQUISITIONS.get_or_init(|| Mutex::new(HashMap::new()))
356}
357
358fn writer_lease_acquisition_counts() -> &'static Mutex<HashMap<WriterLeaseAcquisitionKey, usize>> {
359    WRITER_LEASE_ACQUISITION_COUNTS.get_or_init(|| Mutex::new(HashMap::new()))
360}
361
362fn record_writer_lease_acquisition(domain: RootCacheDomain, key: &str, project_root: &Path) {
363    if !WRITER_LEASE_ACQUISITION_COUNTER_ENABLED.load(Ordering::Relaxed) {
364        return;
365    }
366    let project_root =
367        std::fs::canonicalize(project_root).unwrap_or_else(|_| project_root.to_path_buf());
368    let acquisition_key = WriterLeaseAcquisitionKey {
369        domain,
370        key: key.to_string(),
371        project_root,
372    };
373    if let Ok(mut counts) = writer_lease_acquisition_counts().lock() {
374        *counts.entry(acquisition_key).or_default() += 1;
375    }
376}
377
378/// Turn the acquisition counter on for the test process without clearing it.
379///
380/// There is deliberately no reset: the counter map is keyed by
381/// (domain, key, project_root) and every consuming test uses its own fresh
382/// temp-dir root, so each test's key starts at zero by construction. A global
383/// clear here was the only cross-test coupling - under parallel execution it
384/// wiped entries between another test's setup and its assertion, producing
385/// isolated-green/parallel-red flakes.
386#[doc(hidden)]
387pub fn enable_writer_lease_acquisition_counts_for_test() {
388    WRITER_LEASE_ACQUISITION_COUNTER_ENABLED.store(true, Ordering::Relaxed);
389}
390
391#[doc(hidden)]
392pub fn writer_lease_acquisition_count_for_test(
393    domain: RootCacheDomain,
394    key: &str,
395    project_root: &Path,
396) -> usize {
397    let project_root =
398        std::fs::canonicalize(project_root).unwrap_or_else(|_| project_root.to_path_buf());
399    writer_lease_acquisition_counts()
400        .lock()
401        .ok()
402        .and_then(|counts| {
403            counts
404                .get(&WriterLeaseAcquisitionKey {
405                    domain,
406                    key: key.to_string(),
407                    project_root,
408                })
409                .copied()
410        })
411        .unwrap_or(0)
412}
413
414fn shared_process_lease(
415    registry_key: &ProcessLeaseKey,
416) -> Result<Option<Arc<WriterLease>>, fs_lock::AcquireError> {
417    let mut leases = process_leases().lock().map_err(|_| {
418        fs_lock::AcquireError::Io(io::Error::other("process lease registry poisoned"))
419    })?;
420    if let Some(existing) = leases.get(registry_key).and_then(Weak::upgrade) {
421        if existing.verify()? {
422            return Ok(Some(existing));
423        }
424        leases.remove(registry_key);
425    }
426    Ok(None)
427}
428
429fn process_lease_acquisition_lock(
430    registry_key: &ProcessLeaseKey,
431) -> Result<Arc<Mutex<()>>, fs_lock::AcquireError> {
432    let mut acquisitions = process_lease_acquisitions().lock().map_err(|_| {
433        fs_lock::AcquireError::Io(io::Error::other(
434            "process lease acquisition registry poisoned",
435        ))
436    })?;
437    if let Some(existing) = acquisitions.get(registry_key).and_then(Weak::upgrade) {
438        return Ok(existing);
439    }
440    if acquisitions.len() > 1024 {
441        acquisitions.retain(|_, lock| lock.strong_count() > 0);
442    }
443    let lock = Arc::new(Mutex::new(()));
444    acquisitions.insert(registry_key.clone(), Arc::downgrade(&lock));
445    Ok(lock)
446}
447
448#[cfg(test)]
449type AcquireSharedHook = Arc<dyn Fn(RootCacheDomain, &Path, &str) + Send + Sync + 'static>;
450
451#[cfg(test)]
452static ACQUIRE_SHARED_HOOK: OnceLock<Mutex<Option<AcquireSharedHook>>> = OnceLock::new();
453
454#[cfg(test)]
455fn set_acquire_shared_hook_for_test(hook: Option<AcquireSharedHook>) {
456    *ACQUIRE_SHARED_HOOK
457        .get_or_init(|| Mutex::new(None))
458        .lock()
459        .expect("acquire shared hook mutex") = hook;
460}
461
462#[cfg(test)]
463fn run_acquire_shared_hook_for_test(domain: RootCacheDomain, cache_dir: &Path, key: &str) {
464    let hook = ACQUIRE_SHARED_HOOK
465        .get_or_init(|| Mutex::new(None))
466        .lock()
467        .expect("acquire shared hook mutex")
468        .clone();
469    if let Some(hook) = hook {
470        hook(domain, cache_dir, key);
471    }
472}
473
474#[cfg(not(test))]
475fn run_acquire_shared_hook_for_test(_domain: RootCacheDomain, _cache_dir: &Path, _key: &str) {}
476
477impl std::fmt::Debug for WriterLease {
478    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
479        formatter
480            .debug_struct("WriterLease")
481            .field("domain", &self.domain)
482            .field("key", &self.key)
483            .field("path", &self.path)
484            .field("epoch", &self.epoch)
485            .finish_non_exhaustive()
486    }
487}
488
489impl WriterLease {
490    /// Acquire a shared writer lease.
491    ///
492    /// Same-process callers wait without a timeout because the current lease
493    /// attempt may need filesystem work that takes several seconds. The
494    /// filesystem lock acquisition itself is nonblocking.
495    pub fn acquire_shared(
496        domain: RootCacheDomain,
497        cache_dir: &Path,
498        key: &str,
499        project_root: &Path,
500    ) -> Result<Option<Arc<Self>>, fs_lock::AcquireError> {
501        Self::acquire_shared_inner(domain, cache_dir, key, project_root, None)
502    }
503
504    /// Acquire a process-shared writer lease without waiting past `timeout`.
505    ///
506    /// The deadline covers both the same-process single-flight gate and the
507    /// filesystem lease. A bounded caller must not become unbounded merely
508    /// because another thread is currently performing the filesystem attempt.
509    pub fn acquire_shared_with_timeout(
510        domain: RootCacheDomain,
511        cache_dir: &Path,
512        key: &str,
513        project_root: &Path,
514        timeout: Duration,
515    ) -> Result<Option<Arc<Self>>, fs_lock::AcquireError> {
516        Self::acquire_shared_inner(domain, cache_dir, key, project_root, Some(timeout))
517    }
518
519    fn acquire_shared_inner(
520        domain: RootCacheDomain,
521        cache_dir: &Path,
522        key: &str,
523        project_root: &Path,
524        timeout: Option<Duration>,
525    ) -> Result<Option<Arc<Self>>, fs_lock::AcquireError> {
526        let deadline = timeout.map(|timeout| Instant::now() + timeout);
527        let access = ArtifactAccess::for_root(project_root);
528        if !access.allows_writer_lease(domain, key, &writer_lease_path(cache_dir)) {
529            return Ok(None);
530        }
531        let registry_key = ProcessLeaseKey {
532            domain,
533            cache_dir: canonical_process_lease_dir(cache_dir),
534        };
535        if let Some(existing) = shared_process_lease(&registry_key)? {
536            record_writer_lease_acquisition(domain, key, project_root);
537            return Ok(Some(existing));
538        }
539
540        let acquisition_lock = process_lease_acquisition_lock(&registry_key)?;
541        let _acquisition_guard = if let Some(deadline) = deadline {
542            loop {
543                match acquisition_lock.try_lock() {
544                    Ok(guard) => break guard,
545                    Err(TryLockError::Poisoned(_)) => {
546                        return Err(fs_lock::AcquireError::Io(io::Error::other(
547                            "process lease acquisition poisoned",
548                        )));
549                    }
550                    Err(TryLockError::WouldBlock) => {
551                        let now = Instant::now();
552                        if now >= deadline {
553                            return Err(fs_lock::AcquireError::Timeout);
554                        }
555                        std::thread::sleep(
556                            Duration::from_millis(10).min(deadline.saturating_duration_since(now)),
557                        );
558                    }
559                }
560            }
561        } else {
562            acquisition_lock.lock().map_err(|_| {
563                fs_lock::AcquireError::Io(io::Error::other("process lease acquisition poisoned"))
564            })?
565        };
566
567        if let Some(existing) = shared_process_lease(&registry_key)? {
568            record_writer_lease_acquisition(domain, key, project_root);
569            return Ok(Some(existing));
570        }
571
572        run_acquire_shared_hook_for_test(domain, cache_dir, key);
573
574        let filesystem_timeout = deadline
575            .map(|deadline| deadline.saturating_duration_since(Instant::now()))
576            .unwrap_or(Duration::ZERO);
577        let lease = Arc::new(Self::acquire(domain, cache_dir, key, filesystem_timeout)?);
578        process_leases()
579            .lock()
580            .map_err(|_| {
581                fs_lock::AcquireError::Io(io::Error::other("process lease registry poisoned"))
582            })?
583            .insert(registry_key, Arc::downgrade(&lease));
584        record_writer_lease_acquisition(domain, key, project_root);
585        Ok(Some(lease))
586    }
587
588    fn acquire(
589        domain: RootCacheDomain,
590        cache_dir: &Path,
591        key: &str,
592        timeout: Duration,
593    ) -> Result<Self, fs_lock::AcquireError> {
594        if !storage_allows_root_keyed(cache_dir)? {
595            return Err(fs_lock::AcquireError::Io(io::Error::new(
596                io::ErrorKind::PermissionDenied,
597                format!(
598                    "refusing root-keyed {} writer lease on a network filesystem at {}",
599                    domain.as_str(),
600                    cache_dir.display()
601                ),
602            )));
603        }
604        if let Some(storage_root) = cache_dir.parent().and_then(Path::parent) {
605            crate::legacy_partitions::guard_new_layout_write_path(
606                storage_root,
607                cache_dir,
608                "root-keyed writer lease",
609            )?;
610        }
611        fs::create_dir_all(cache_dir)?;
612        let guard = fs_lock::try_acquire(&writer_lease_path(cache_dir), timeout)?;
613        if !guard.verify_writer_epoch()? {
614            return Err(fs_lock::AcquireError::Io(io::Error::other(
615                "writer lease epoch changed immediately after acquisition",
616            )));
617        }
618        let path = guard.path().to_path_buf();
619        let epoch = guard.writer_epoch().to_string();
620        Ok(Self {
621            domain,
622            key: key.to_string(),
623            path,
624            epoch,
625            guard: Mutex::new(guard),
626        })
627    }
628
629    pub fn verify(&self) -> io::Result<bool> {
630        self.guard
631            .lock()
632            .map_err(|_| io::Error::other("writer lease mutex poisoned"))?
633            .verify_writer_epoch()
634    }
635
636    pub fn epoch(&self) -> &str {
637        &self.epoch
638    }
639
640    pub fn domain(&self) -> RootCacheDomain {
641        self.domain
642    }
643
644    pub fn key(&self) -> &str {
645        &self.key
646    }
647
648    pub fn path(&self) -> &Path {
649        &self.path
650    }
651}
652
653pub fn writer_lease_path(cache_dir: &Path) -> PathBuf {
654    cache_dir.join("writer.lease")
655}
656
657#[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)]
658pub struct ReadMarkerMetadata {
659    pub pid: u32,
660    pub hostname: String,
661    pub created_at_ms: u64,
662}
663
664#[derive(Debug)]
665pub struct ReadMarker {
666    path: PathBuf,
667    metadata: ReadMarkerMetadata,
668    last_touched_at_ms: AtomicU64,
669}
670
671impl ReadMarker {
672    pub fn create(cache_dir: &Path, generation_label: &str) -> io::Result<Self> {
673        let metadata = ReadMarkerMetadata {
674            pid: std::process::id(),
675            hostname: current_hostname(),
676            created_at_ms: now_ms(),
677        };
678        let dir = read_marker_dir(cache_dir, generation_label);
679        fs::create_dir_all(&dir)?;
680        let seq = MARKER_SEQ.fetch_add(1, Ordering::Relaxed);
681        let path = dir.join(format!(
682            "{}.{}.{}.{}.json",
683            metadata.pid,
684            sanitize_marker_component(&metadata.hostname),
685            metadata.created_at_ms,
686            seq
687        ));
688        write_marker_file(&path, &metadata)?;
689        let last_touched_at_ms = AtomicU64::new(metadata.created_at_ms);
690        Ok(Self {
691            path,
692            metadata,
693            last_touched_at_ms,
694        })
695    }
696
697    pub fn touch(&self) -> io::Result<()> {
698        write_marker_file(&self.path, &self.metadata)?;
699        self.last_touched_at_ms.store(now_ms(), Ordering::Relaxed);
700        Ok(())
701    }
702
703    pub fn touch_if_due(&self) -> io::Result<()> {
704        let now = now_ms();
705        let last = self.last_touched_at_ms.load(Ordering::Relaxed);
706        if now.saturating_sub(last) < READ_MARKER_TOUCH_INTERVAL_MS {
707            return Ok(());
708        }
709        self.touch()
710    }
711
712    pub fn path(&self) -> &Path {
713        &self.path
714    }
715
716    pub fn metadata(&self) -> &ReadMarkerMetadata {
717        &self.metadata
718    }
719}
720
721impl Drop for ReadMarker {
722    fn drop(&mut self) {
723        let _ = fs::remove_file(&self.path);
724        fs_lock::sync_parent(&self.path);
725    }
726}
727
728pub fn read_marker_dir(cache_dir: &Path, generation_label: &str) -> PathBuf {
729    cache_dir.join("readers").join(generation_label)
730}
731
732/// Sweep every generation marker below a cache directory, reusing the same
733/// PID-authoritative and cross-host-TTL liveness rules as generation GC.
734pub(crate) fn sweep_all_read_markers(cache_dir: &Path) -> ReadMarkerSweep {
735    let readers = cache_dir.join("readers");
736    let entries = match fs::read_dir(&readers) {
737        Ok(entries) => entries,
738        Err(error) if error.kind() == io::ErrorKind::NotFound => return ReadMarkerSweep::default(),
739        Err(_) => {
740            return ReadMarkerSweep {
741                protected: true,
742                removed_stale: 0,
743            }
744        }
745    };
746
747    let mut sweep = ReadMarkerSweep::default();
748    for entry in entries {
749        let Ok(entry) = entry else {
750            sweep.protected = true;
751            continue;
752        };
753        let path = entry.path();
754        let file_type = match entry.file_type() {
755            Ok(file_type) => file_type,
756            Err(_) => {
757                sweep.protected = true;
758                continue;
759            }
760        };
761        if !file_type.is_dir() {
762            continue;
763        }
764        let Some(label) = path.file_name().and_then(|name| name.to_str()) else {
765            sweep.protected = true;
766            continue;
767        };
768        let generation = sweep_read_markers(cache_dir, label);
769        sweep.protected |= generation.protected;
770        sweep.removed_stale += generation.removed_stale;
771    }
772    sweep
773}
774
775#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
776pub struct ReadMarkerSweep {
777    pub protected: bool,
778    pub removed_stale: usize,
779}
780
781pub fn protected_read_marker_exists(cache_dir: &Path, generation_label: &str) -> bool {
782    read_marker_protection(cache_dir, generation_label, false).protected
783}
784
785pub fn sweep_read_markers(cache_dir: &Path, generation_label: &str) -> ReadMarkerSweep {
786    read_marker_protection(cache_dir, generation_label, true)
787}
788
789fn read_marker_protection(
790    cache_dir: &Path,
791    generation_label: &str,
792    remove_stale: bool,
793) -> ReadMarkerSweep {
794    let dir = read_marker_dir(cache_dir, generation_label);
795    let entries = match fs::read_dir(&dir) {
796        Ok(entries) => entries,
797        Err(error) if error.kind() == io::ErrorKind::NotFound => return ReadMarkerSweep::default(),
798        Err(_) => {
799            return ReadMarkerSweep {
800                protected: true,
801                removed_stale: 0,
802            };
803        }
804    };
805
806    let hostname = current_hostname();
807    let now = now_ms();
808    let mut sweep = ReadMarkerSweep::default();
809    for entry in entries.flatten() {
810        let path = entry.path();
811        match marker_file_is_protected(&path, now, &hostname) {
812            MarkerProtection::Protected => sweep.protected = true,
813            MarkerProtection::Stale | MarkerProtection::Malformed => {
814                if remove_stale && fs::remove_file(&path).is_ok() {
815                    fs_lock::sync_parent(&path);
816                    sweep.removed_stale += 1;
817                }
818            }
819        }
820    }
821    sweep
822}
823
824#[derive(Clone, Copy, Debug, PartialEq, Eq)]
825enum MarkerProtection {
826    Protected,
827    Stale,
828    Malformed,
829}
830
831fn marker_file_is_protected(path: &Path, now: u64, current_host: &str) -> MarkerProtection {
832    let bytes = match fs::read(path) {
833        Ok(bytes) => bytes,
834        Err(error) if error.kind() == io::ErrorKind::NotFound => return MarkerProtection::Stale,
835        Err(_) => return MarkerProtection::Protected,
836    };
837    let metadata: ReadMarkerMetadata = match serde_json::from_slice(&bytes) {
838        Ok(metadata) => metadata,
839        Err(_) => return MarkerProtection::Malformed,
840    };
841    if metadata.hostname != current_host {
842        let Ok(file_metadata) = fs::metadata(path) else {
843            return MarkerProtection::Protected;
844        };
845        let mtime_ms = file_metadata
846            .modified()
847            .ok()
848            .map(system_time_ms)
849            .unwrap_or(now);
850        let age_ms = now.saturating_sub(mtime_ms);
851        return if age_ms <= READ_MARKER_CROSS_HOST_STALE_MS {
852            MarkerProtection::Protected
853        } else {
854            MarkerProtection::Stale
855        };
856    }
857
858    if !fs_lock::process_alive(metadata.pid) {
859        return MarkerProtection::Stale;
860    }
861    if marker_matches_live_process_instance(&metadata) {
862        MarkerProtection::Protected
863    } else {
864        MarkerProtection::Stale
865    }
866}
867
868fn marker_matches_live_process_instance(metadata: &ReadMarkerMetadata) -> bool {
869    // Same-host PID liveness is authoritative for the process instance. When the
870    // OS can tell us the live PID started after this marker was created, the PID
871    // has been reused and the marker belongs to a dead prior process; otherwise
872    // a live PID protects the marker without consulting marker mtime.
873    process_start_time_ms(metadata.pid)
874        .map(|started_at_ms| {
875            started_at_ms
876                <= metadata
877                    .created_at_ms
878                    .saturating_add(PROCESS_START_TIME_GRACE_MS)
879        })
880        .unwrap_or(true)
881}
882
883fn write_marker_file(path: &Path, metadata: &ReadMarkerMetadata) -> io::Result<()> {
884    let tmp = path.with_file_name(format!(
885        ".{}.tmp.{}.{}",
886        path.file_name()
887            .and_then(|name| name.to_str())
888            .unwrap_or("reader"),
889        std::process::id(),
890        now_nanos()
891    ));
892    let result = (|| {
893        let mut file = open_private_file(&tmp)?;
894        serde_json::to_writer(&mut file, metadata).map_err(io::Error::other)?;
895        file.write_all(b"\n")?;
896        file.sync_all()?;
897        drop(file);
898        fs_lock::rename_over(&tmp, path)?;
899        fs_lock::sync_parent(path);
900        Ok(())
901    })();
902    if result.is_err() {
903        let _ = fs::remove_file(&tmp);
904    }
905    result
906}
907
908#[cfg(unix)]
909fn open_private_file(path: &Path) -> io::Result<File> {
910    use std::os::unix::fs::OpenOptionsExt;
911
912    OpenOptions::new()
913        .write(true)
914        .create_new(true)
915        .mode(0o600)
916        .open(path)
917}
918
919#[cfg(not(unix))]
920fn open_private_file(path: &Path) -> io::Result<File> {
921    OpenOptions::new().write(true).create_new(true).open(path)
922}
923
924fn sanitize_marker_component(value: &str) -> String {
925    value
926        .chars()
927        .map(|ch| {
928            if ch.is_ascii_alphanumeric() || matches!(ch, '-' | '_') {
929                ch
930            } else {
931                '_'
932            }
933        })
934        .collect()
935}
936
937#[cfg(test)]
938static FORCE_NETWORK_FS_FOR_TEST: AtomicBool = AtomicBool::new(false);
939
940#[cfg(test)]
941pub fn set_force_network_fs_for_test(enabled: bool) {
942    FORCE_NETWORK_FS_FOR_TEST.store(enabled, Ordering::SeqCst);
943}
944
945pub fn storage_allows_root_keyed(path: &Path) -> io::Result<bool> {
946    #[cfg(test)]
947    if FORCE_NETWORK_FS_FOR_TEST.load(Ordering::SeqCst) {
948        return Ok(false);
949    }
950
951    let probe = existing_ancestor(path);
952    filesystem_is_local(&probe)
953}
954
955fn canonical_process_lease_dir(path: &Path) -> PathBuf {
956    if let Ok(canonical) = std::fs::canonicalize(path) {
957        return canonical;
958    }
959
960    let normalized = lexical_normalize(path);
961    let mut missing_components = Vec::new();
962    let mut current = normalized.as_path();
963    while !current.exists() {
964        let Some(name) = current.file_name() else {
965            return normalized;
966        };
967        missing_components.push(name.to_os_string());
968        let Some(parent) = current.parent() else {
969            return normalized;
970        };
971        current = parent;
972    }
973
974    let mut canonical = std::fs::canonicalize(current).unwrap_or_else(|_| current.to_path_buf());
975    for component in missing_components.iter().rev() {
976        canonical.push(component);
977    }
978    canonical
979}
980
981fn lexical_normalize(path: &Path) -> PathBuf {
982    let mut normalized = PathBuf::new();
983    for component in path.components() {
984        match component {
985            std::path::Component::CurDir => {}
986            std::path::Component::ParentDir => {
987                normalized.pop();
988            }
989            other => normalized.push(other.as_os_str()),
990        }
991    }
992    normalized
993}
994
995fn existing_ancestor(path: &Path) -> PathBuf {
996    let mut current = path;
997    loop {
998        if current.exists() {
999            return current.to_path_buf();
1000        }
1001        let Some(parent) = current.parent() else {
1002            return PathBuf::from(".");
1003        };
1004        current = parent;
1005    }
1006}
1007
1008#[cfg(target_os = "macos")]
1009fn filesystem_is_local(path: &Path) -> io::Result<bool> {
1010    use std::ffi::CString;
1011    use std::os::unix::ffi::OsStrExt;
1012
1013    let c_path = CString::new(path.as_os_str().as_bytes())
1014        .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "path contains NUL byte"))?;
1015    let mut stat: libc::statfs = unsafe { std::mem::zeroed() };
1016    if unsafe { libc::statfs(c_path.as_ptr(), &mut stat) } != 0 {
1017        return Err(io::Error::last_os_error());
1018    }
1019    let nul = stat
1020        .f_fstypename
1021        .iter()
1022        .position(|byte| *byte == 0)
1023        .unwrap_or(stat.f_fstypename.len());
1024    let fs_type = String::from_utf8_lossy(
1025        &stat.f_fstypename[..nul]
1026            .iter()
1027            .map(|byte| *byte as u8)
1028            .collect::<Vec<_>>(),
1029    )
1030    .to_ascii_lowercase();
1031    Ok(!matches!(
1032        fs_type.as_str(),
1033        "nfs" | "smbfs" | "afpfs" | "webdav" | "fusefs"
1034    ))
1035}
1036
1037#[cfg(all(unix, not(target_os = "macos")))]
1038fn filesystem_is_local(path: &Path) -> io::Result<bool> {
1039    use std::ffi::CString;
1040    use std::os::unix::ffi::OsStrExt;
1041
1042    let c_path = CString::new(path.as_os_str().as_bytes())
1043        .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "path contains NUL byte"))?;
1044    let mut stat: libc::statfs = unsafe { std::mem::zeroed() };
1045    if unsafe { libc::statfs(c_path.as_ptr(), &mut stat) } != 0 {
1046        return Err(io::Error::last_os_error());
1047    }
1048    let fs_type = stat.f_type as i64;
1049    const NFS_SUPER_MAGIC: i64 = 0x6969;
1050    const SMB_SUPER_MAGIC: i64 = 0x517B;
1051    const CIFS_MAGIC_NUMBER: i64 = 0xFF534D42;
1052    Ok(!matches!(
1053        fs_type,
1054        NFS_SUPER_MAGIC | SMB_SUPER_MAGIC | CIFS_MAGIC_NUMBER
1055    ))
1056}
1057
1058#[cfg(not(unix))]
1059fn filesystem_is_local(_path: &Path) -> io::Result<bool> {
1060    Ok(true)
1061}
1062
1063fn now_ms() -> u64 {
1064    system_time_ms(SystemTime::now())
1065}
1066
1067fn system_time_ms(time: SystemTime) -> u64 {
1068    time.duration_since(UNIX_EPOCH)
1069        .unwrap_or(Duration::ZERO)
1070        .as_millis() as u64
1071}
1072
1073fn now_nanos() -> u128 {
1074    SystemTime::now()
1075        .duration_since(UNIX_EPOCH)
1076        .unwrap_or(Duration::ZERO)
1077        .as_nanos()
1078}
1079
1080#[cfg(test)]
1081static PROCESS_START_TIME_OVERRIDES: OnceLock<Mutex<HashMap<u32, Option<u64>>>> = OnceLock::new();
1082
1083#[cfg(test)]
1084fn set_process_start_time_for_test(pid: u32, started_at_ms: Option<u64>) {
1085    PROCESS_START_TIME_OVERRIDES
1086        .get_or_init(|| Mutex::new(HashMap::new()))
1087        .lock()
1088        .expect("process start override mutex")
1089        .insert(pid, started_at_ms);
1090}
1091
1092#[cfg(test)]
1093fn clear_process_start_time_for_test(pid: u32) {
1094    if let Some(overrides) = PROCESS_START_TIME_OVERRIDES.get() {
1095        overrides
1096            .lock()
1097            .expect("process start override mutex")
1098            .remove(&pid);
1099    }
1100}
1101
1102#[cfg(test)]
1103fn process_start_time_override(pid: u32) -> Option<Option<u64>> {
1104    PROCESS_START_TIME_OVERRIDES
1105        .get()
1106        .and_then(|overrides| overrides.lock().ok()?.get(&pid).copied())
1107}
1108
1109#[cfg(target_os = "linux")]
1110pub(crate) fn process_start_time_ms(pid: u32) -> Option<u64> {
1111    #[cfg(test)]
1112    if let Some(override_value) = process_start_time_override(pid) {
1113        return override_value;
1114    }
1115
1116    let stat = fs::read_to_string(format!("/proc/{pid}/stat")).ok()?;
1117    let after_comm = stat.rsplit_once(") ")?.1;
1118    let fields = after_comm.split_whitespace().collect::<Vec<_>>();
1119    let start_ticks = fields.get(19)?.parse::<u64>().ok()?;
1120    let boot_time_secs = fs::read_to_string("/proc/stat")
1121        .ok()?
1122        .lines()
1123        .find_map(|line| line.strip_prefix("btime ")?.parse::<u64>().ok())?;
1124    let ticks_per_second = unsafe { libc::sysconf(libc::_SC_CLK_TCK) };
1125    if ticks_per_second <= 0 {
1126        return None;
1127    }
1128    let ticks_per_second = ticks_per_second as u64;
1129    Some(
1130        boot_time_secs
1131            .saturating_mul(1_000)
1132            .saturating_add(start_ticks.saturating_mul(1_000) / ticks_per_second),
1133    )
1134}
1135
1136#[cfg(target_os = "macos")]
1137pub(crate) fn process_start_time_ms(pid: u32) -> Option<u64> {
1138    #[cfg(test)]
1139    if let Some(override_value) = process_start_time_override(pid) {
1140        return override_value;
1141    }
1142
1143    let mut info: libc::proc_bsdinfo = unsafe { std::mem::zeroed() };
1144    let info_size = std::mem::size_of::<libc::proc_bsdinfo>() as libc::c_int;
1145    let bytes = unsafe {
1146        libc::proc_pidinfo(
1147            pid as libc::c_int,
1148            libc::PROC_PIDTBSDINFO,
1149            0,
1150            (&mut info as *mut libc::proc_bsdinfo).cast(),
1151            info_size,
1152        )
1153    };
1154    if bytes != info_size {
1155        return None;
1156    }
1157    Some(
1158        info.pbi_start_tvsec
1159            .saturating_mul(1_000)
1160            .saturating_add(info.pbi_start_tvusec / 1_000),
1161    )
1162}
1163
1164#[cfg(not(any(target_os = "linux", target_os = "macos")))]
1165pub(crate) fn process_start_time_ms(pid: u32) -> Option<u64> {
1166    #[cfg(test)]
1167    if let Some(override_value) = process_start_time_override(pid) {
1168        return override_value;
1169    }
1170    let _ = pid;
1171    None
1172}
1173
1174#[cfg(unix)]
1175fn current_hostname() -> String {
1176    let mut buffer = [0u8; 256];
1177    let result = unsafe { libc::gethostname(buffer.as_mut_ptr().cast(), buffer.len()) };
1178    if result == 0 {
1179        let len = buffer
1180            .iter()
1181            .position(|byte| *byte == 0)
1182            .unwrap_or(buffer.len());
1183        if len > 0 {
1184            return String::from_utf8_lossy(&buffer[..len]).into_owned();
1185        }
1186    }
1187    "unknown-host".to_string()
1188}
1189
1190#[cfg(windows)]
1191fn current_hostname() -> String {
1192    std::env::var("COMPUTERNAME").unwrap_or_else(|_| "unknown-host".to_string())
1193}
1194
1195#[cfg(all(not(unix), not(windows)))]
1196fn current_hostname() -> String {
1197    "unknown-host".to_string()
1198}
1199
1200#[cfg(test)]
1201mod tests {
1202    use super::*;
1203
1204    #[test]
1205    fn read_marker_file_is_private_and_touchable() {
1206        let dir = tempfile::tempdir().unwrap();
1207        let marker = ReadMarker::create(dir.path(), "current").unwrap();
1208        assert!(marker.path().is_file());
1209        marker.touch().unwrap();
1210        let bytes = fs::read(marker.path()).unwrap();
1211        let parsed: ReadMarkerMetadata = serde_json::from_slice(&bytes).unwrap();
1212        assert_eq!(parsed.pid, std::process::id());
1213        #[cfg(unix)]
1214        {
1215            use std::os::unix::fs::PermissionsExt;
1216            assert_eq!(
1217                fs::metadata(marker.path()).unwrap().permissions().mode() & 0o777,
1218                0o600
1219            );
1220        }
1221    }
1222
1223    #[test]
1224    fn same_host_live_marker_ignores_stale_mtime() {
1225        let dir = tempfile::tempdir().unwrap();
1226        let marker = ReadMarker::create(dir.path(), "current").unwrap();
1227        filetime::set_file_mtime(marker.path(), filetime::FileTime::from_unix_time(0, 0)).unwrap();
1228
1229        assert!(protected_read_marker_exists(dir.path(), "current"));
1230    }
1231
1232    #[test]
1233    fn sweep_removes_dead_same_host_marker() {
1234        let dir = tempfile::tempdir().unwrap();
1235        let marker_path = read_marker_dir(dir.path(), "old").join("dead.json");
1236        let metadata = ReadMarkerMetadata {
1237            pid: 0,
1238            hostname: current_hostname(),
1239            created_at_ms: now_ms(),
1240        };
1241        fs::create_dir_all(marker_path.parent().unwrap()).unwrap();
1242        write_marker_file(&marker_path, &metadata).unwrap();
1243
1244        let sweep = sweep_read_markers(dir.path(), "old");
1245
1246        assert!(!sweep.protected);
1247        assert_eq!(sweep.removed_stale, 1);
1248        assert!(!marker_path.exists());
1249    }
1250
1251    #[test]
1252    fn sweep_removes_reused_pid_marker_when_created_at_predates_process_start() {
1253        let dir = tempfile::tempdir().unwrap();
1254        let marker_path = read_marker_dir(dir.path(), "old").join("reused.json");
1255        let pid = std::process::id();
1256        let metadata = ReadMarkerMetadata {
1257            pid,
1258            hostname: current_hostname(),
1259            created_at_ms: 1_000,
1260        };
1261        fs::create_dir_all(marker_path.parent().unwrap()).unwrap();
1262        write_marker_file(&marker_path, &metadata).unwrap();
1263        set_process_start_time_for_test(pid, Some(10_000));
1264
1265        let sweep = sweep_read_markers(dir.path(), "old");
1266        clear_process_start_time_for_test(pid);
1267
1268        assert!(!sweep.protected);
1269        assert_eq!(sweep.removed_stale, 1);
1270        assert!(!marker_path.exists());
1271    }
1272
1273    #[test]
1274    fn sweep_removes_expired_cross_host_marker() {
1275        let dir = tempfile::tempdir().unwrap();
1276        let marker_path = read_marker_dir(dir.path(), "old").join("cross-host.json");
1277        let metadata = ReadMarkerMetadata {
1278            pid: 123,
1279            hostname: format!("other-{}", current_hostname()),
1280            created_at_ms: now_ms(),
1281        };
1282        fs::create_dir_all(marker_path.parent().unwrap()).unwrap();
1283        write_marker_file(&marker_path, &metadata).unwrap();
1284        let stale_time = SystemTime::now()
1285            .checked_sub(Duration::from_millis(
1286                READ_MARKER_CROSS_HOST_STALE_MS.saturating_add(1_000),
1287            ))
1288            .unwrap_or(UNIX_EPOCH);
1289        filetime::set_file_mtime(
1290            &marker_path,
1291            filetime::FileTime::from_system_time(stale_time),
1292        )
1293        .unwrap();
1294
1295        let sweep = sweep_read_markers(dir.path(), "old");
1296
1297        assert!(!sweep.protected);
1298        assert_eq!(sweep.removed_stale, 1);
1299        assert!(!marker_path.exists());
1300    }
1301
1302    #[cfg(unix)]
1303    #[test]
1304    fn process_lease_dir_canonicalizes_existing_ancestor_before_cache_dir_exists() {
1305        let dir = tempfile::tempdir().unwrap();
1306        let real = dir.path().join("real");
1307        let link = dir.path().join("link");
1308        fs::create_dir_all(&real).unwrap();
1309        std::os::unix::fs::symlink(&real, &link).unwrap();
1310
1311        let missing_cache_dir = link.join("inspect").join("project");
1312        let before_create = canonical_process_lease_dir(&missing_cache_dir);
1313        fs::create_dir_all(&missing_cache_dir).unwrap();
1314        let after_create = canonical_process_lease_dir(&missing_cache_dir);
1315
1316        assert_eq!(before_create, after_create);
1317        assert!(before_create.starts_with(std::fs::canonicalize(&real).unwrap()));
1318    }
1319
1320    #[test]
1321    fn borrow_only_root_never_receives_existing_shared_writer_capability() {
1322        let storage = tempfile::tempdir().unwrap();
1323        let parent_root = tempfile::tempdir().unwrap();
1324        let worktree_root = tempfile::tempdir().unwrap();
1325        let shared_key = "shared-artifact-key";
1326        let cache_dir = storage.path().join("callgraph").join(shared_key);
1327        configure_artifact_access(parent_root.path(), shared_key, false);
1328        configure_artifact_access(worktree_root.path(), shared_key, true);
1329
1330        let parent_lease = WriterLease::acquire_shared(
1331            RootCacheDomain::Callgraph,
1332            &cache_dir,
1333            shared_key,
1334            parent_root.path(),
1335        )
1336        .unwrap()
1337        .expect("parent writer lease");
1338        enable_writer_lease_acquisition_counts_for_test();
1339
1340        let worktree_lease = WriterLease::acquire_shared(
1341            RootCacheDomain::Callgraph,
1342            &cache_dir,
1343            shared_key,
1344            worktree_root.path(),
1345        )
1346        .unwrap();
1347        let unkeyed_worktree_lease = WriterLease::acquire_shared(
1348            RootCacheDomain::Callgraph,
1349            &storage.path().join("scratch"),
1350            shared_key,
1351            worktree_root.path(),
1352        )
1353        .unwrap();
1354
1355        assert!(worktree_lease.is_none());
1356        assert!(unkeyed_worktree_lease.is_none());
1357        assert!(parent_lease.verify().unwrap());
1358        assert_eq!(
1359            writer_lease_acquisition_count_for_test(
1360                RootCacheDomain::Callgraph,
1361                shared_key,
1362                worktree_root.path(),
1363            ),
1364            0
1365        );
1366    }
1367
1368    #[test]
1369    fn unregistered_root_cannot_acquire_keyed_shared_writer_lease() {
1370        let storage = tempfile::tempdir().unwrap();
1371        let root = tempfile::tempdir().unwrap();
1372        let shared_key = "unregistered-shared-key";
1373        let cache_dir = storage.path().join("callgraph").join(shared_key);
1374        enable_writer_lease_acquisition_counts_for_test();
1375
1376        let lease = WriterLease::acquire_shared(
1377            RootCacheDomain::Callgraph,
1378            &cache_dir,
1379            shared_key,
1380            root.path(),
1381        )
1382        .unwrap();
1383
1384        assert!(lease.is_none());
1385        assert_eq!(
1386            writer_lease_acquisition_count_for_test(
1387                RootCacheDomain::Callgraph,
1388                shared_key,
1389                root.path(),
1390            ),
1391            0
1392        );
1393        assert!(!writer_lease_path(&cache_dir).exists());
1394    }
1395
1396    #[cfg(unix)]
1397    #[test]
1398    fn unregistered_root_denies_symlink_aliased_stale_key() {
1399        let storage = tempfile::tempdir().unwrap();
1400        let roots = tempfile::tempdir().unwrap();
1401        let real_parent = roots.path().join("real");
1402        let alias_parent = roots.path().join("alias");
1403        fs::create_dir_all(&real_parent).unwrap();
1404        std::os::unix::fs::symlink(&real_parent, &alias_parent).unwrap();
1405
1406        let root = alias_parent.join("root");
1407        let stale_key = crate::search_index::artifact_cache_key(&root);
1408        fs::create_dir_all(&root).unwrap();
1409        let canonical_key = crate::search_index::artifact_cache_key(&root);
1410        assert_ne!(stale_key, canonical_key, "fixture must exercise the alias");
1411
1412        let cache_dir = storage.path().join("callgraph").join(&stale_key);
1413        enable_writer_lease_acquisition_counts_for_test();
1414        let lease = WriterLease::acquire_shared(
1415            RootCacheDomain::Callgraph,
1416            &cache_dir,
1417            &canonical_key,
1418            &root,
1419        )
1420        .unwrap();
1421
1422        assert!(lease.is_none());
1423        assert_eq!(
1424            writer_lease_acquisition_count_for_test(
1425                RootCacheDomain::Callgraph,
1426                &canonical_key,
1427                &root,
1428            ),
1429            0
1430        );
1431        assert!(!writer_lease_path(&cache_dir).exists());
1432    }
1433
1434    #[cfg(unix)]
1435    #[test]
1436    fn registration_before_directory_creation_survives_alias_canonicalization() {
1437        // Regression for the concurrent-acquire flake: configure registered the
1438        // root before its directory existed (raw alias spelling), the first
1439        // lease's create_dir_all made later lookups canonicalize to the real
1440        // spelling, and the registered root missed its own capability entry.
1441        let storage = tempfile::tempdir().unwrap();
1442        let roots = tempfile::tempdir().unwrap();
1443        let real_parent = roots.path().join("real");
1444        let alias_parent = roots.path().join("alias");
1445        fs::create_dir_all(&real_parent).unwrap();
1446        std::os::unix::fs::symlink(&real_parent, &alias_parent).unwrap();
1447
1448        let root = alias_parent.join("pre-registered");
1449        configure_artifact_access(&root, "pre-registered", false);
1450        fs::create_dir_all(&root).unwrap();
1451
1452        let cache_dir = storage.path().join("inspect").join("pre-registered");
1453        let lease = WriterLease::acquire_shared(
1454            RootCacheDomain::Inspect,
1455            &cache_dir,
1456            "pre-registered",
1457            &root,
1458        )
1459        .unwrap();
1460        assert!(
1461            lease.is_some(),
1462            "a root registered before its directory existed must keep its capability"
1463        );
1464    }
1465
1466    #[test]
1467    fn borrow_only_root_keeps_private_project_scope_writable() {
1468        let storage = tempfile::tempdir().unwrap();
1469        let worktree_root = tempfile::tempdir().unwrap();
1470        let shared_key = "shared-artifact-key";
1471        let private_key = crate::path_identity::project_scope_key(worktree_root.path());
1472        let cache_dir = storage.path().join("inspect").join(&private_key);
1473        configure_artifact_access(worktree_root.path(), shared_key, true);
1474
1475        let lease = WriterLease::acquire_shared(
1476            RootCacheDomain::Inspect,
1477            &cache_dir,
1478            &private_key,
1479            worktree_root.path(),
1480        )
1481        .unwrap()
1482        .expect("private inspect writer lease");
1483
1484        assert!(lease.verify().unwrap());
1485    }
1486
1487    #[test]
1488    fn writer_lease_acquire_shared_does_not_serialize_different_roots() {
1489        let dir = tempfile::tempdir().unwrap();
1490        let blocked_cache_dir = dir.path().join("callgraph").join("blocked");
1491        let free_cache_dir = dir.path().join("callgraph").join("free");
1492        configure_artifact_access(&blocked_cache_dir, "blocked", false);
1493        configure_artifact_access(&free_cache_dir, "free", false);
1494        let (blocked_tx, blocked_rx) = std::sync::mpsc::channel();
1495        let (release_tx, release_rx) = std::sync::mpsc::channel();
1496        let release_rx = Arc::new(Mutex::new(release_rx));
1497
1498        struct HookGuard;
1499        impl Drop for HookGuard {
1500            fn drop(&mut self) {
1501                set_acquire_shared_hook_for_test(None);
1502            }
1503        }
1504
1505        set_acquire_shared_hook_for_test(Some(Arc::new(move |_, _, key| {
1506            if key == "blocked" {
1507                blocked_tx.send(()).unwrap();
1508                release_rx.lock().unwrap().recv().unwrap();
1509            }
1510        })));
1511        let _hook_guard = HookGuard;
1512
1513        let blocked_handle = std::thread::spawn(move || {
1514            WriterLease::acquire_shared(
1515                RootCacheDomain::Callgraph,
1516                &blocked_cache_dir,
1517                "blocked",
1518                &blocked_cache_dir,
1519            )
1520            .map_err(|error| error.to_string())
1521            .and_then(|lease| lease.ok_or_else(|| "writer lease unexpectedly denied".to_string()))
1522        });
1523        blocked_rx
1524            .recv_timeout(Duration::from_secs(5))
1525            .expect("blocked root should reach acquisition hook");
1526
1527        let (free_tx, free_rx) = std::sync::mpsc::channel();
1528        let free_handle = std::thread::spawn(move || {
1529            let result = WriterLease::acquire_shared(
1530                RootCacheDomain::Callgraph,
1531                &free_cache_dir,
1532                "free",
1533                &free_cache_dir,
1534            )
1535            .map_err(|error| error.to_string())
1536            .and_then(|lease| lease.ok_or_else(|| "writer lease unexpectedly denied".to_string()));
1537            free_tx.send(result).unwrap();
1538        });
1539        let free_lease = free_rx
1540            .recv_timeout(Duration::from_secs(5))
1541            .expect("free root should not wait behind another root's acquisition")
1542            .expect("free root should acquire while another root is in acquisition");
1543        assert!(free_lease.verify().unwrap());
1544        free_handle
1545            .join()
1546            .expect("free root thread should not panic");
1547
1548        release_tx.send(()).unwrap();
1549        let blocked_lease = blocked_handle
1550            .join()
1551            .expect("blocked root thread should not panic")
1552            .expect("blocked root should acquire after release");
1553        assert!(blocked_lease.verify().unwrap());
1554    }
1555
1556    #[test]
1557    fn writer_lease_acquire_shared_reuses_single_process_lease_concurrently() {
1558        let dir = tempfile::tempdir().unwrap();
1559        let cache_dir = dir.path().join("inspect").join("project");
1560        configure_artifact_access(&cache_dir, "project", false);
1561        let barrier = std::sync::Arc::new(std::sync::Barrier::new(8));
1562        let mut handles = Vec::new();
1563        for _ in 0..8 {
1564            let barrier = std::sync::Arc::clone(&barrier);
1565            let cache_dir = cache_dir.clone();
1566            handles.push(std::thread::spawn(move || {
1567                barrier.wait();
1568                WriterLease::acquire_shared(
1569                    RootCacheDomain::Inspect,
1570                    &cache_dir,
1571                    "project",
1572                    &cache_dir,
1573                )
1574                .map_err(|error| error.to_string())
1575                .and_then(|lease| {
1576                    lease.ok_or_else(|| "writer lease unexpectedly denied".to_string())
1577                })
1578            }));
1579        }
1580
1581        let leases = handles
1582            .into_iter()
1583            .map(|handle| handle.join().unwrap().unwrap())
1584            .collect::<Vec<_>>();
1585        let epoch = leases[0].epoch().to_string();
1586        let path = leases[0].path().to_path_buf();
1587        for lease in &leases {
1588            assert_eq!(lease.epoch(), epoch);
1589            assert_eq!(lease.path(), path.as_path());
1590            assert!(lease.verify().unwrap());
1591        }
1592    }
1593
1594    #[test]
1595    fn nfs_guard_test_seam_fails_closed() {
1596        set_force_network_fs_for_test(true);
1597        assert!(!storage_allows_root_keyed(Path::new(".")).unwrap());
1598        set_force_network_fs_for_test(false);
1599    }
1600}