1use 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
35pub const READ_MARKER_TOUCH_INTERVAL_MS: u64 = fs_lock::HEARTBEAT_INTERVAL_MS;
39pub const READ_MARKER_CROSS_HOST_STALE_MS: u64 = fs_lock::STALE_HEARTBEAT_MS * 5;
42const 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 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#[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();
122static 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#[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 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 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 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 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 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
276fn 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
322pub 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 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 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#[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 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 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(®istry_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(®istry_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(®istry_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
732pub(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 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 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}