mount/core.rs
1// SPDX-License-Identifier: Apache-2.0
2//! Content-addressed mount core.
3//!
4//! [`ContentAddressedMount`] is the platform-agnostic implementation
5//! of [`PlatformShell`]. It speaks heddle: given a thread name, it
6//! resolves it to a state via [`refs::RefManager`], pulls the tree
7//! root from the object store, and answers filesystem queries by
8//! walking the Merkle DAG lazily.
9//!
10//! ## Two-tier write model
11//!
12//! Writes don't go through a generic in-memory page cache that drains
13//! to disk on `heddle capture`. They go straight into heddle's CAS as
14//! soon as the file is closed:
15//!
16//! 1. **Hot tier (in-memory partial buffers).** A `write(offset, bytes)`
17//! is keyed by [`NodeId`] and accumulates in a single `Vec<u8>` per
18//! open file. Reads of the same node during the buffer's lifetime
19//! serve from the buffer (so a `write -> read` round-trip in the
20//! same FUSE session sees the new bytes immediately).
21//!
22//! 2. **Warm tier (CAS-promoted blobs).** When the kernel signals end
23//! of file (`flush`/`close`), or after an idle threshold (the
24//! [`PromotionPolicy::idle_after`] window), we hash the buffer,
25//! write a blob via the same [`ObjectStore`] API that
26//! `heddle capture` uses, and record `path -> blob_oid` in a
27//! per-thread *pending tree*. The hot buffer is dropped.
28//!
29//! 3. **Pending tree.** A `BTreeMap<RelPath, PendingEntry>` plus a
30//! `BTreeSet<RelPath>` of deletions that overlay the immutable
31//! state's tree. `lookup`/`enumerate`/`read` consult the pending
32//! tier first so the mount serves "what the agent just wrote"
33//! rather than the parent state.
34//!
35//! ### Crash semantics
36//!
37//! The hot tier lives only in process memory; an unclean unmount
38//! discards in-flight writes. The warm tier is written to the heddle
39//! object store via the same atomic write path that `heddle capture`
40//! uses, so a promoted blob survives a crash even if the surrounding
41//! `capture()` call never completes — the next agent that captures
42//! the same content will hit the dedup fast path.
43//!
44//! ### Why this beats a worktree-walk capture
45//!
46//! `heddle capture` from a worktree currently walks every file,
47//! hashes its contents, and writes the blob if new. Mount writes do
48//! that work *during* the write itself, so capture-from-mount becomes:
49//! - drain pending tree into a real `Tree` object
50//! - record `State` referencing the tree
51//! - update the thread's HEAD
52//!
53//! No worktree walk, no re-hashing, no blob duplication across
54//! threads — two agents writing the same `import { foo } from 'bar'`
55//! to two different files write *one* blob.
56
57use std::{
58 collections::{BTreeMap, BTreeSet, VecDeque},
59 ffi::{OsStr, OsString},
60 path::{Component, Path, PathBuf},
61 sync::{
62 Arc, Mutex, RwLock, Weak,
63 atomic::{AtomicBool, AtomicU32, AtomicUsize, Ordering},
64 },
65 thread::JoinHandle,
66 time::{Duration, Instant, SystemTime},
67};
68
69use objects::{
70 object::{Blob, ContentHash, EntryType, FileMode, StateId, Tree, TreeEntry, TreeEntryTarget},
71 store::{FsStore, ObjectCacheControl, ObjectStore},
72 sync::{LockExt, RwLockExt},
73 util::gitlink_placeholder_bytes,
74};
75use oplog::{OpLog, OpLogBackend};
76use refs::{RefBackend, RefManager};
77use repo::Repository;
78use tracing::{debug, warn};
79
80use crate::{
81 cache::BlobCachePool,
82 error::{MountError, Result},
83 shell::{
84 AttrUpdate, Attrs, DIR_UNIX_MODE, Entry, NodeId, NodeKind, PlatformShell, RenameOptions,
85 kind_for_mode,
86 },
87};
88
89/// Default promotion idle window: a buffer with no writes for this
90/// long is eligible to be drained to CAS without an explicit
91/// flush/close. The kernel doesn't always issue `release` for short-
92/// lived files (e.g. when the agent process is killed mid-write), so
93/// the timer is the safety net.
94const DEFAULT_PROMOTION_IDLE: Duration = Duration::from_secs(2);
95
96/// Default cadence for the clock-driven safety-sweep. A worker thread
97/// wakes up every `sweep_interval` and promotes any hot buffer that's
98/// been idle longer than `idle_after`. Five seconds is well below
99/// human attention but well above the kernel's flush cadence, so it
100/// catches process-pause/agent-crash leaks without burning CPU.
101const DEFAULT_SWEEP_INTERVAL: Option<Duration> = Some(Duration::from_secs(5));
102
103/// Maximum hot-buffer size accepted by [`ContentAddressedMount::write`] and
104/// [`ContentAddressedMount::apply_truncate`]. Matches the 100 MiB cap in
105/// `repo::worktree_walk` so mount promotion cannot build blobs capture would
106/// reject anyway.
107/// One source of truth with the walker's per-blob capture cap.
108pub(crate) use repo::worktree_walk::MAX_FILE_SIZE as MAX_MOUNT_HOT_FILE_SIZE;
109
110/// Reject wire offsets/sizes at the trust boundary before they reach
111/// `Vec::resize` (overflow panic or multi-TiB allocation abort).
112fn validate_write_extent(offset: u64, data_len: usize) -> Result<usize> {
113 let data_len_u64 = u64::try_from(data_len).map_err(|_| {
114 MountError::InvalidArgument(format!("write length {data_len} does not fit in u64"))
115 })?;
116 let end = offset.checked_add(data_len_u64).ok_or_else(|| {
117 MountError::InvalidArgument(format!(
118 "write offset {offset} + length {data_len} overflows u64"
119 ))
120 })?;
121 if end > MAX_MOUNT_HOT_FILE_SIZE {
122 return Err(MountError::FileTooLarge(format!(
123 "write would extend file to {end} bytes (max {MAX_MOUNT_HOT_FILE_SIZE})"
124 )));
125 }
126 usize::try_from(end).map_err(|_| {
127 MountError::InvalidArgument(format!(
128 "write extent end {end} does not fit in usize on this platform"
129 ))
130 })
131}
132
133fn validate_truncate_size(new_size: u64) -> Result<usize> {
134 if new_size > MAX_MOUNT_HOT_FILE_SIZE {
135 return Err(MountError::FileTooLarge(format!(
136 "truncate to {new_size} bytes exceeds max {MAX_MOUNT_HOT_FILE_SIZE}"
137 )));
138 }
139 usize::try_from(new_size).map_err(|_| {
140 MountError::InvalidArgument(format!(
141 "truncate size {new_size} does not fit in usize on this platform"
142 ))
143 })
144}
145
146#[cfg(test)]
147mod validation_tests {
148 use super::*;
149
150 #[test]
151 fn validate_write_extent_accepts_in_bounds_writes() {
152 assert_eq!(validate_write_extent(0, 0).unwrap(), 0);
153 assert_eq!(validate_write_extent(0, 64).unwrap(), 64);
154 assert_eq!(
155 validate_write_extent(MAX_MOUNT_HOT_FILE_SIZE - 1, 1).unwrap(),
156 MAX_MOUNT_HOT_FILE_SIZE as usize
157 );
158 }
159
160 #[test]
161 fn validate_write_extent_rejects_overflow_and_oversize() {
162 let err = validate_write_extent(u64::MAX, 1).unwrap_err();
163 assert!(matches!(err, MountError::InvalidArgument(_)));
164 let err = validate_write_extent(MAX_MOUNT_HOT_FILE_SIZE, 1).unwrap_err();
165 assert!(matches!(err, MountError::FileTooLarge(_)));
166 let err = validate_write_extent(0, (MAX_MOUNT_HOT_FILE_SIZE as usize) + 1).unwrap_err();
167 assert!(matches!(err, MountError::FileTooLarge(_)));
168 }
169
170 #[test]
171 fn validate_truncate_size_accepts_in_bounds_and_rejects_oversize() {
172 assert_eq!(validate_truncate_size(0).unwrap(), 0);
173 assert_eq!(
174 validate_truncate_size(MAX_MOUNT_HOT_FILE_SIZE).unwrap(),
175 MAX_MOUNT_HOT_FILE_SIZE as usize
176 );
177 let err = validate_truncate_size(MAX_MOUNT_HOT_FILE_SIZE + 1).unwrap_err();
178 assert!(matches!(err, MountError::FileTooLarge(_)));
179 }
180}
181
182/// Tunables for when buffered writes get promoted to CAS.
183#[derive(Clone, Copy, Debug)]
184pub struct PromotionPolicy {
185 /// Drain buffers with no writes for at least this long. The check
186 /// runs opportunistically on every mutating call; agents that go
187 /// quiet without closing aren't left holding the buffer.
188 pub idle_after: Duration,
189 /// How often the clock-driven safety-sweep thread wakes up to
190 /// drain idle buffers. `None` disables the timer entirely (useful
191 /// for tests that want deterministic event-driven promotion).
192 pub sweep_interval: Option<Duration>,
193}
194
195impl Default for PromotionPolicy {
196 fn default() -> Self {
197 Self {
198 idle_after: DEFAULT_PROMOTION_IDLE,
199 sweep_interval: DEFAULT_SWEEP_INTERVAL,
200 }
201 }
202}
203
204/// The kind of node a registered inode points at.
205#[derive(Clone, Debug)]
206enum NodeRecord {
207 /// Root of the mount — the tree at the thread's current state.
208 Root {
209 tree: ContentHash,
210 },
211 /// A subdirectory resolved from the captured tree. `path` is the
212 /// mount-relative path of this directory; `tree` is the content
213 /// hash of its tree object. Carrying the path lets `lookup` /
214 /// `enumerate` consult the pending tier for nested writes.
215 Dir {
216 tree: ContentHash,
217 path: PathBuf,
218 },
219 /// A directory that exists only in the pending tier (the agent
220 /// created `newdir/foo.rs` and `newdir/` is not yet in any
221 /// captured tree). No backing tree hash exists yet — it lives
222 /// virtually in the pending map.
223 PendingDir {
224 path: PathBuf,
225 },
226 /// A file resolved from the captured tree. We carry `path` so
227 /// writes against this NodeId can route into the hot tier
228 /// without re-walking from the root.
229 File {
230 blob: ContentHash,
231 mode: FileMode,
232 path: PathBuf,
233 },
234 Gitlink {
235 placeholder: Vec<u8>,
236 path: PathBuf,
237 },
238 Symlink {
239 blob: ContentHash,
240 },
241 /// A file that exists only in the pending tier (created by the
242 /// mount, not yet captured into a state). Its content lives at
243 /// `path` in the [`Pending`] map.
244 PendingFile {
245 path: PathBuf,
246 mode: FileMode,
247 },
248 /// A symlink created through the mount. Target bytes live in
249 /// [`Pending::symlinks`]; we don't promote the symlink to a CAS
250 /// blob until [`ContentAddressedMount::capture`].
251 PendingSymlink {
252 path: PathBuf,
253 },
254}
255
256impl NodeRecord {
257 fn kind(&self) -> NodeKind {
258 match self {
259 NodeRecord::Root { .. } | NodeRecord::Dir { .. } | NodeRecord::PendingDir { .. } => {
260 NodeKind::Directory
261 }
262 NodeRecord::File { mode, .. } | NodeRecord::PendingFile { mode, .. } => {
263 kind_for_mode(*mode)
264 }
265 NodeRecord::Gitlink { .. } => NodeKind::File,
266 NodeRecord::Symlink { .. } | NodeRecord::PendingSymlink { .. } => NodeKind::Symlink,
267 }
268 }
269
270 fn unix_mode(&self) -> u32 {
271 match self {
272 NodeRecord::Root { .. } | NodeRecord::Dir { .. } | NodeRecord::PendingDir { .. } => {
273 DIR_UNIX_MODE
274 }
275 NodeRecord::File { mode, .. } | NodeRecord::PendingFile { mode, .. } => {
276 mode.to_unix_mode()
277 }
278 NodeRecord::Gitlink { .. } => FileMode::Normal.to_unix_mode(),
279 NodeRecord::Symlink { .. } | NodeRecord::PendingSymlink { .. } => {
280 FileMode::Symlink.to_unix_mode()
281 }
282 }
283 }
284}
285
286/// Inode registry — maps the opaque ids we hand out to platform
287/// adapters back to the underlying object hashes.
288#[derive(Default)]
289struct Inodes {
290 next: u64,
291 by_id: BTreeMap<u64, NodeRecord>,
292 /// Reverse index for tree records: a repeated lookup of the
293 /// same content hash returns the same NodeId. FUSE caches
294 /// inodes aggressively; handing out fresh ids per lookup
295 /// explodes the kernel-side dcache.
296 by_hash: BTreeMap<HashKey, u64>,
297 /// Reverse index for files (both captured and pending): keyed
298 /// by relative path. Two files with identical content but
299 /// different paths get distinct inode numbers — that's required
300 /// for the cross-thread dedup story (the *blob* is the same, the
301 /// *inode* must not be).
302 by_path: BTreeMap<PathBuf, u64>,
303}
304
305#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
306struct HashKey {
307 /// 0 = tree, 1 = blob (file), 2 = blob (symlink). Distinguishing
308 /// the same hash referenced as both a tree and a blob is paranoid
309 /// — content hashes are typed — but it's cheap and self-documenting.
310 kind: u8,
311 hash: ContentHash,
312}
313
314impl Inodes {
315 fn new(root_tree: ContentHash) -> Self {
316 let mut me = Self {
317 next: NodeId::ROOT.0 + 1,
318 by_id: BTreeMap::new(),
319 by_hash: BTreeMap::new(),
320 by_path: BTreeMap::new(),
321 };
322 me.by_id
323 .insert(NodeId::ROOT.0, NodeRecord::Root { tree: root_tree });
324 me.by_hash.insert(
325 HashKey {
326 kind: 0,
327 hash: root_tree,
328 },
329 NodeId::ROOT.0,
330 );
331 me
332 }
333
334 fn get(&self, id: NodeId) -> Option<NodeRecord> {
335 self.by_id.get(&id.0).cloned()
336 }
337
338 fn intern(&mut self, record: NodeRecord) -> NodeId {
339 match &record {
340 NodeRecord::Root { tree } => {
341 let key = HashKey {
342 kind: 0,
343 hash: *tree,
344 };
345 if let Some(&id) = self.by_hash.get(&key) {
346 return NodeId(id);
347 }
348 let id = self.next;
349 self.next += 1;
350 self.by_id.insert(id, record);
351 self.by_hash.insert(key, id);
352 NodeId(id)
353 }
354 NodeRecord::Dir { path, .. } | NodeRecord::PendingDir { path } => {
355 // Coalesce by path so the same directory hands back
356 // the same NodeId across lookups, even if the backing
357 // tree hash flips after a capture.
358 if let Some(&id) = self.by_path.get(path) {
359 self.by_id.insert(id, record);
360 return NodeId(id);
361 }
362 let id = self.next;
363 self.next += 1;
364 self.by_path.insert(path.clone(), id);
365 self.by_id.insert(id, record);
366 NodeId(id)
367 }
368 NodeRecord::File { path, .. }
369 | NodeRecord::Gitlink { path, .. }
370 | NodeRecord::PendingFile { path, .. }
371 | NodeRecord::PendingSymlink { path } => {
372 if let Some(&id) = self.by_path.get(path) {
373 // If the path's record is being upgraded
374 // (e.g. PendingFile -> File after capture, or
375 // a File whose blob hash flipped), refresh the
376 // backing record so subsequent reads see the
377 // new identity.
378 self.by_id.insert(id, record);
379 return NodeId(id);
380 }
381 let id = self.next;
382 self.next += 1;
383 self.by_path.insert(path.clone(), id);
384 self.by_id.insert(id, record);
385 NodeId(id)
386 }
387 NodeRecord::Symlink { blob } => {
388 let key = HashKey {
389 kind: 2,
390 hash: *blob,
391 };
392 if let Some(&id) = self.by_hash.get(&key) {
393 return NodeId(id);
394 }
395 let id = self.next;
396 self.next += 1;
397 self.by_id.insert(id, record);
398 self.by_hash.insert(key, id);
399 NodeId(id)
400 }
401 }
402 }
403
404 fn forget(&mut self, id: NodeId) {
405 if id == NodeId::ROOT {
406 // Root is a permanent fixture; the only way to retire it
407 // is to drop the whole mount.
408 return;
409 }
410 if let Some(record) = self.by_id.remove(&id.0) {
411 match record {
412 NodeRecord::Root { tree } => {
413 self.by_hash.remove(&HashKey {
414 kind: 0,
415 hash: tree,
416 });
417 }
418 NodeRecord::Dir { path, .. } | NodeRecord::PendingDir { path } => {
419 // Codex r12 thread 3293680448 (P1): only drop the
420 // path mapping if it still points at *this* inode.
421 // After unlink-then-recreate or rename-over, `path`
422 // may already be rebound to a live inode at a
423 // different NodeId; a blind `remove` would yank
424 // that fresh inode's binding too.
425 if self.by_path.get(&path) == Some(&id.0) {
426 self.by_path.remove(&path);
427 }
428 }
429 NodeRecord::File { path, .. }
430 | NodeRecord::Gitlink { path, .. }
431 | NodeRecord::PendingFile { path, .. }
432 | NodeRecord::PendingSymlink { path } => {
433 if self.by_path.get(&path) == Some(&id.0) {
434 self.by_path.remove(&path);
435 }
436 }
437 NodeRecord::Symlink { blob } => {
438 self.by_hash.remove(&HashKey {
439 kind: 2,
440 hash: blob,
441 });
442 }
443 }
444 }
445 }
446}
447
448/// A single in-flight write tier entry.
449struct HotBuffer {
450 /// Mount-relative path the buffer maps to.
451 path: PathBuf,
452 /// File mode (executable bit, etc.).
453 mode: FileMode,
454 /// Buffered bytes. Indexed by absolute file offset.
455 bytes: Vec<u8>,
456 /// Last write time, used by the idle-promotion check.
457 last_touched: Instant,
458 /// Monotonic mutation generation used to keep promotion from
459 /// retiring a buffer that changed while CAS I/O was in flight.
460 revision: u64,
461}
462
463/// A single warm-tier entry — a path that has been promoted to CAS
464/// but not yet folded into a state.
465#[derive(Clone, Debug)]
466struct PendingEntry {
467 blob: ContentHash,
468 mode: FileMode,
469 size: u64,
470}
471
472/// Direct-child summary stored by the pending namespace index.
473#[derive(Clone, Copy, Debug, PartialEq, Eq)]
474pub(crate) enum PendingChildKind {
475 HotFile {
476 node: NodeId,
477 size: u64,
478 mode: FileMode,
479 },
480 WarmFile {
481 size: u64,
482 mode: FileMode,
483 },
484 Symlink {
485 size: u64,
486 },
487 Dir,
488}
489
490#[derive(Default)]
491struct DirChildren {
492 names: BTreeSet<OsString>,
493}
494
495/// Incremental projection of the pending namespace.
496///
497/// `entries` owns the exact pending facts while `by_dir` projects
498/// those facts onto every ancestor edge. A path insertion/removal is
499/// therefore O(path depth), and a directory read only visits that
500/// directory's direct children.
501#[derive(Default)]
502struct PendingChildIndex {
503 entries: BTreeMap<PathBuf, PendingChildKind>,
504 by_dir: BTreeMap<PathBuf, DirChildren>,
505}
506
507impl PendingChildIndex {
508 fn insert(&mut self, path: PathBuf, kind: PendingChildKind) {
509 let is_new = self.entries.insert(path.clone(), kind).is_none();
510 if !is_new {
511 return;
512 }
513
514 let mut child = path.as_path();
515 while !child.as_os_str().is_empty() {
516 let Some(parent) = child.parent() else {
517 break;
518 };
519 let Some(name) = child.file_name() else {
520 break;
521 };
522 self.by_dir
523 .entry(parent.to_path_buf())
524 .or_default()
525 .names
526 .insert(name.to_os_string());
527 child = parent;
528 }
529 }
530
531 fn remove(&mut self, path: &Path) -> Option<PendingChildKind> {
532 let removed = self.entries.remove(path)?;
533 let mut child = path;
534 while !child.as_os_str().is_empty() {
535 let child_still_exists = self.entries.contains_key(child)
536 || self
537 .by_dir
538 .get(child)
539 .is_some_and(|children| !children.names.is_empty());
540 if child_still_exists {
541 break;
542 }
543
544 let Some(parent) = child.parent() else {
545 break;
546 };
547 let Some(name) = child.file_name() else {
548 break;
549 };
550 let remove_parent = if let Some(children) = self.by_dir.get_mut(parent) {
551 children.names.remove(name);
552 children.names.is_empty()
553 } else {
554 false
555 };
556 if remove_parent {
557 self.by_dir.remove(parent);
558 }
559 child = parent;
560 }
561 Some(removed)
562 }
563
564 fn update_file_mode(&mut self, path: &Path, mode: FileMode) {
565 if let Some(
566 PendingChildKind::HotFile { mode: current, .. }
567 | PendingChildKind::WarmFile { mode: current, .. },
568 ) = self.entries.get_mut(path)
569 {
570 *current = mode;
571 }
572 }
573
574 fn rebase_prefix(&mut self, old: &Path, new: &Path) {
575 self.remove(new);
576 let rebased: Vec<(PathBuf, PathBuf, PendingChildKind)> = self
577 .entries
578 .iter()
579 .filter_map(|(path, kind)| {
580 let tail = path.strip_prefix(old).ok()?;
581 Some((path.clone(), new.join(tail), *kind))
582 })
583 .collect();
584 for (old_path, _, _) in &rebased {
585 self.remove(old_path);
586 }
587 for (_, new_path, kind) in rebased {
588 self.insert(new_path, kind);
589 }
590 }
591
592 fn dir_exists(&self, dir: &Path) -> bool {
593 matches!(self.entries.get(dir), Some(PendingChildKind::Dir))
594 || self
595 .by_dir
596 .get(dir)
597 .is_some_and(|children| !children.names.is_empty())
598 }
599
600 fn children_at(&self, dir: &Path) -> Vec<(String, PendingChildKind)> {
601 let Some(children) = self.by_dir.get(dir) else {
602 return Vec::new();
603 };
604 children
605 .names
606 .iter()
607 .filter_map(|name| {
608 let name = name.to_str()?.to_string();
609 let path = join_child(dir, &name);
610 let kind = match self.entries.get(&path).copied() {
611 Some(PendingChildKind::Dir) | None => PendingChildKind::Dir,
612 Some(kind) => kind,
613 };
614 Some((name, kind))
615 })
616 .collect()
617 }
618}
619
620/// Per-NodeId lifecycle state. Tracks both whether an inode is still
621/// resolvable via `inodes.by_path` (Live) or has had its directory
622/// entry removed but is still held by an open fd (Orphan), and the
623/// open-handle refcount that drives the final-close cleanup.
624///
625/// Absence from [`Pending::state`] is the third state — Released —
626/// matching the spike model (`docs/design/mount-posix-semantics.md`
627/// §1.1). The type system makes "orphaned with no open count" and
628/// "open count without orphan flag" unrepresentable, replacing the
629/// old `orphans: BTreeSet<u64>` + `open_handles: BTreeMap<u64, u32>`
630/// pair with one map and forcing every callback that branches on
631/// lifecycle to `match`.
632#[derive(Clone, Copy, Debug, PartialEq, Eq)]
633pub(crate) enum NodeState {
634 /// Live: the inode owns a binding in `inodes.by_path`. The
635 /// refcount tracks how many FUSE `open` / `create` callbacks
636 /// have minted handles to it; `release` drives it back down.
637 /// At T1 (unlink with N ≥ 0), the count is carried over into
638 /// `Orphan { open_count }`.
639 Live { open_count: u32 },
640 /// Orphan: directory entry gone (`unlink_entry` or `rename`-over
641 /// of the displaced destination) but `open_count` kernel fds
642 /// still hold the NodeId. Bytes in `hot[node]` / `warm[node]`
643 /// outlive the transition; the cleanup happens on the final
644 /// `release` (count drops to 0 → state entry removed + bytes
645 /// dropped).
646 Orphan { open_count: u32 },
647}
648
649/// The two-tier write state for a mount.
650///
651/// Post-spike (`docs/design/mount-posix-semantics.md` §2.1) the cache
652/// is **NodeId-keyed throughout**: `hot[id]` and `warm[id]` carry the
653/// bytes; path-keyed helpers (`hot_by_path`, `tombstones`,
654/// `dir_tombstones`, `explicit_dirs`, `symlinks`) are
655/// directory-entry-level concepts only. The Live → Orphan transition
656/// never moves bytes — it just rewrites `state[id]` and the path-side
657/// bookkeeping. That collapse eliminates the cache-layer asymmetry
658/// (Bug Class A in the spike doc §4) that produced every Codex
659/// finding r6 → r9 on PR #182.
660#[derive(Default)]
661#[doc(hidden)]
662pub struct Pending<'brand> {
663 /// Hot tier: per-`NodeId` open-file buffers.
664 hot: BTreeMap<u64, HotBuffer>,
665 /// Reverse-index for the hot tier: which NodeId currently owns a
666 /// buffer for `path`. Path-keyed because FUSE `lookup` arrives
667 /// with paths, not NodeIds. Only one at a time — opening the
668 /// same file twice from different node ids resolves to the same
669 /// buffer because the inode registry coalesces by path for
670 /// pending files. Removed on `unlink_entry` / rebound on
671 /// `rename_entry`.
672 hot_by_path: BTreeMap<PathBuf, u64>,
673 /// Warm tier: per-`NodeId` promoted bytes. Bytes survive the
674 /// Live → Orphan transition without a migration step — that's
675 /// Decision A of the spike. `pending_lookup` resolves a path to
676 /// its current Live NodeId via `inodes.by_path` and then reads
677 /// `warm[id]`; orphan branches in `read` / `attrs` / `write` /
678 /// `apply_truncate` consult `warm[node]` directly.
679 warm: BTreeMap<u64, PendingEntry>,
680 /// Tombstones — paths the mount has deleted. Suppress the
681 /// underlying state's entry on reads. File-only; directories
682 /// use [`Self::dir_tombstones`].
683 tombstones: BTreeSet<PathBuf>,
684 /// Directory tombstones — captured-tree directories the mount
685 /// has `rmdir`'d. Distinct from file tombstones because the
686 /// capture-time tree fold has to drop the
687 /// whole subtree, not a single leaf.
688 dir_tombstones: BTreeSet<PathBuf>,
689 /// Directories the mount has `mkdir`'d into the overlay that
690 /// don't (yet) have any children. Without this, an empty
691 /// `mkdir target/` wouldn't survive across a `lookup` /
692 /// `enumerate` round-trip (nothing under it means
693 /// [`pending_dir_exists`] would return false).
694 explicit_dirs: BTreeSet<PathBuf>,
695 /// Symlinks created through the mount, keyed by mount-relative
696 /// path. The bytes are the target as the kernel handed them to
697 /// `symlink`; capture hashes them into a CAS blob. Symlinks are
698 /// not openable for IO; no orphan story applies.
699 symlinks: BTreeMap<PathBuf, Vec<u8>>,
700 /// Authoritative namespace projection for pending-only lookup and
701 /// enumeration. Updated in the same critical sections as the
702 /// path-keyed overlay maps.
703 child_index: PendingChildIndex,
704 /// Per-NodeId lifecycle state. See [`NodeState`]. Replaces the
705 /// pre-spike `orphans: BTreeSet<u64>` + `open_handles:
706 /// BTreeMap<u64, u32>` pair. Absence from this map is the third
707 /// state (Released) — entries are removed on final `release`,
708 /// `invalidate`, and `capture`.
709 state: BTreeMap<u64, NodeState>,
710 /// Invariant phantom: ties this `Pending` to a unique `'brand`
711 /// introduced by [`Pending::with_brand`]. Witnesses minted under
712 /// one `'brand` cannot be passed to methods on a `Pending`
713 /// carrying a different `'brand` — closes Codex PR #217 r2
714 /// finding `3293832936`. The `fn(&'brand ()) -> &'brand ()`
715 /// shape makes `'brand` invariant (neither covariant nor
716 /// contravariant), so the borrow checker refuses to unify two
717 /// fresh brands handed out by separate `with_brand` calls.
718 _brand: std::marker::PhantomData<fn(&'brand ()) -> &'brand ()>,
719}
720
721impl<'brand> Pending<'brand> {
722 /// True iff the NodeId is currently Orphan (directory entry gone
723 /// but `open_count >= 0` fds still reference the inode). Every
724 /// callback that branches on lifecycle goes through this helper
725 /// (or matches `state` directly) so the "implicitly assume Live"
726 /// failure mode is hard to write.
727 fn is_orphan(&self, id: u64) -> bool {
728 matches!(self.state.get(&id), Some(NodeState::Orphan { .. }))
729 }
730
731 /// Current open-handle refcount for the NodeId, or zero if
732 /// untracked. Used by [`MountInner::release_node`] to drive the
733 /// final-close cleanup and by [`unlink_entry`] / [`rename_entry`]
734 /// to carry the count over into `Orphan { open_count }` at T1/T3.
735 fn open_count(&self, id: u64) -> u32 {
736 match self.state.get(&id) {
737 Some(NodeState::Live { open_count } | NodeState::Orphan { open_count }) => *open_count,
738 None => 0,
739 }
740 }
741
742 /// Read-only access to the per-NodeId lifecycle entry. Sole
743 /// reachable point for the [`crate::pending`] witness constructors
744 /// to query the FSM without taking a `pub(crate)` dependency on
745 /// the underlying `state` field. Returning `Option<NodeState>` by
746 /// value keeps the field private to this module — callers cannot
747 /// mutate state through this handle.
748 pub(crate) fn lookup_state(&self, id: u64) -> Option<NodeState> {
749 self.state.get(&id).copied()
750 }
751
752 /// Witness-gated LiveNonZero → Orphan state transition. The
753 /// `&Witness<'_, 'brand, Orphan>` parameter is the type-level
754 /// proof that the caller has already gone through the FSM check —
755 /// `Witness::new` is module-private to [`crate::pending`], so the
756 /// only callers that can name this method's argument type are the
757 /// [`crate::pending::BrandedPending::transition_to_orphan`] body
758 /// (which constructs the witness after consuming a matching
759 /// `Witness<LiveNonZero>`) and code that already held a
760 /// `Witness<Orphan>` (in which case the state was already Orphan,
761 /// and re-inserting is a no-op on the discriminant). Direct
762 /// callers in this module have no way to mint a `Witness<Orphan>`,
763 /// so they cannot bypass the witness discipline.
764 pub(crate) fn apply_transition_to_orphan(
765 &mut self,
766 w: &crate::pending::Witness<'_, 'brand, crate::pending::Orphan>,
767 ) {
768 let id = w.id();
769 let open_count = self.open_count(id);
770 self.state.insert(id, NodeState::Orphan { open_count });
771 }
772
773 /// Witness-gated FUSE-forget discharge. The
774 /// `&KernelForgetWitness<'_, 'brand>` parameter is the type-level
775 /// proof that the caller has already gone through the discharge-
776 /// safety FSM check: the witness is constructed only inside
777 /// [`crate::pending::BrandedPending::kernel_forget_inode`], whose
778 /// body matches the same `None | Some(Live { open_count: 0 })`
779 /// pattern as [`crate::pending::BrandedPending::witness_kernel_forget`].
780 /// [`crate::pending::KernelForgetWitness::new`] is module-private
781 /// to [`crate::pending`], so the only callers that can name this
782 /// method's argument type are that one entry point (and code that
783 /// already held a witness — same brand-gating chain as
784 /// [`Self::apply_transition_to_orphan`]).
785 ///
786 /// Removes `hot[id]` (with its `hot_by_path` reverse-index
787 /// cleanup) and `state[id]`, then returns `true` iff `warm[id]`
788 /// is still populated — the caller in `MountInner::invalidate`
789 /// uses that bool to decide whether the inode-side `forget` is
790 /// safe to fire (warm is the durable pre-capture copy; if it's
791 /// there, capture still needs the NodeId → path chain).
792 ///
793 /// `warm` is intentionally preserved here per Codex r12 threads
794 /// 3293484634 / 3293510311 (P1): FUSE `forget` is a kernel-side
795 /// dcache eviction, not a close — dropping warm bytes silently
796 /// loses the user's committed-in-session data.
797 pub(crate) fn apply_kernel_forget(
798 &mut self,
799 w: &crate::pending::KernelForgetWitness<'_, 'brand>,
800 ) -> bool {
801 let id = w.id();
802 let mut removed_path = None;
803 if let Some(buf) = self.hot.remove(&id)
804 && self.hot_by_path.get(&buf.path) == Some(&id)
805 {
806 self.hot_by_path.remove(&buf.path);
807 removed_path = Some(buf.path);
808 }
809 self.state.remove(&id);
810 let warm_survives = self.warm.contains_key(&id);
811 if !warm_survives && let Some(path) = removed_path {
812 self.child_index.remove(&path);
813 }
814 warm_survives
815 }
816
817 /// Test-only: insert a per-NodeId lifecycle entry directly,
818 /// bypassing the FSM entry points. Used by the
819 /// [`crate::pending`] substrate tests to set up `Pending` states
820 /// without dragging in the full mount lifecycle. Gated behind
821 /// `cfg(test)` so it never reaches a release binary.
822 #[cfg(test)]
823 pub(crate) fn test_insert_state(&mut self, id: u64, state: NodeState) {
824 self.state.insert(id, state);
825 }
826
827 /// Test-only: insert a hot-tier buffer for `id` with the given
828 /// `path` and `bytes`. Used by the [`crate::pending`] tests to set
829 /// up scenarios that need hot-tier bytes alongside a lifecycle entry.
830 #[cfg(test)]
831 pub(crate) fn test_insert_hot(&mut self, id: u64, path: PathBuf, bytes: Vec<u8>) {
832 self.hot.insert(
833 id,
834 HotBuffer {
835 path,
836 mode: FileMode::Normal,
837 bytes,
838 last_touched: Instant::now(),
839 revision: 0,
840 },
841 );
842 }
843
844 /// Test-only: true iff `hot[id]` is currently populated. Mirror
845 /// of [`Self::test_insert_hot`] for assertion in
846 /// substrate lifecycle tests.
847 #[cfg(test)]
848 pub(crate) fn test_has_hot(&self, id: u64) -> bool {
849 self.hot.contains_key(&id)
850 }
851}
852
853/// In-mount overlay: a snapshot-time view of the parent state plus
854/// pending writes the agent has issued since.
855///
856/// Writes never modify the immutable state; they accumulate in
857/// [`Pending`] until [`ContentAddressedMount::capture`] folds them
858/// into a fresh state.
859pub struct ContentAddressedMount<
860 R: RefBackend + 'static = RefManager,
861 O: OpLogBackend + 'static = OpLog,
862 S: ObjectStore + 'static = FsStore,
863> {
864 inner: Arc<MountInner<R, O, S>>,
865 /// Background safety-sweep worker. Held in an `Option` so the
866 /// `Drop` impl can `take()` it, signal shutdown, and join cleanly
867 /// without needing to borrow `&mut self`.
868 sweeper: Mutex<Option<SweepHandle>>,
869}
870
871/// All shared state — held inside an `Arc` so the safety-sweep
872/// worker thread can hold a `Weak` reference, drain hot buffers
873/// idly, and exit on its own when the mount is dropped.
874///
875/// `promotion` is wrapped in an `RwLock` so `with_promotion_policy`
876/// can swap the active policy without having to rebuild the Arc.
877///
878/// # Lock ordering invariant
879///
880/// Three locks coexist inside `MountInner` (`state`, `pending`,
881/// `inodes`) and the call sites use them in nested combinations.
882/// To avoid deadlock, every code path that acquires more than one
883/// MUST follow this order, top-to-bottom:
884///
885/// ```text
886/// state (RwLock — read or write)
887/// │
888/// ▼
889/// pending (Mutex)
890/// │
891/// ▼
892/// inodes (Mutex)
893/// ```
894///
895/// Equivalently: never take `state` while holding `pending` or
896/// `inodes`; never take `pending` while holding `inodes`. The
897/// reverse direction (drop the inner first, then the outer) is the
898/// only safe unwind. `promotion` is independent of all three — it
899/// guards a config knob that's read everywhere but never co-locked
900/// with the others — so it can be sequenced freely.
901///
902/// The discipline is currently safe-by-convention: there's no
903/// lock-ordering enforcement at the type system level. When adding
904/// a new code path that touches more than one of these locks,
905/// audit against the diagram above before merging. The existing
906/// call sites that take all three in the right order are good
907/// templates — search for `state.write` / `state.read` and trace
908/// the subsequent `pending.lock()` / `inodes.lock()` to see the
909/// pattern in action.
910pub(crate) struct MountInner<R: RefBackend, O: OpLogBackend, S: ObjectStore> {
911 repo: Repository<R, O, S>,
912 thread: String,
913 state: RwLock<MountState>,
914 inodes: Mutex<Inodes>,
915 // Storage carries `Pending<'static>` as the long-lived shape;
916 // every actual witness-minting access goes through
917 // [`Pending::with_brand`], which re-borrows under a fresh
918 // invariant `'brand` introduced by HRTB and hands the closure a
919 // [`crate::pending::BrandedPending<'_, 'brand>`]. The `'static`
920 // slot can never be exposed as a witness brand because
921 // `Pending<'brand>` carries no witness constructors at all — the
922 // `witness_*` methods live on [`crate::pending::BrandedPending`],
923 // whose private field makes it unconstructible outside
924 // `with_brand`'s body. This closes the structural gap Codex
925 // flagged in r2 (`3293832936`) and the r3 follow-on
926 // (`3293898540`).
927 pending: Mutex<Pending<'static>>,
928 promotion: RwLock<PromotionPolicy>,
929 mounted_at: SystemTime,
930 /// Write-side serialization. Acquired by structural-mutation
931 /// methods (rename, create, mkdir, symlink) that need their
932 /// existence-check + mutation pair to land atomically against
933 /// other writers — see [`ContentAddressedMount::rename_entry_with_options`]
934 /// and the RENAME_NOREPLACE atomicity contract (Codex r8 Thread
935 /// 3293235163). Lock order: `write_mu` precedes every other lock
936 /// in [`MountInner`]; never take it while holding `state`,
937 /// `pending`, or `inodes`.
938 write_mu: Mutex<()>,
939 /// Shared materialised-blob cache. Without this every kernel
940 /// `read` syscall re-decompresses the full blob from the object
941 /// store, which makes chunked + mmap reads ~200× slower than
942 /// vanilla FS on multi-MB files (see
943 /// `crates/mount/benches/mount_read_paths.rs`). Held as an `Arc`
944 /// so multiple mounts in the same process share warm state —
945 /// forked-thread mounts inherit fully-warm cache for any blob
946 /// the parent already touched.
947 blob_cache: Arc<BlobCachePool>,
948}
949
950/// Owns the worker thread + its shutdown signal. Dropping this joins
951/// the worker.
952///
953/// Shutdown is event-driven via a `Condvar` rather than polling: the
954/// worker parks on `wait_timeout(interval)` and is woken either by
955/// the timer firing (run a sweep) or by `signal_and_join` flipping
956/// `shutdown` + notifying the condvar (exit immediately). Mount drop
957/// used to pay up to 50 ms per `Drop` for a polled-AtomicBool worker
958/// to notice — visible in any churn-y workload (the prewarm bench
959/// uncovered this) — and now pays only the per-OS thread join cost.
960struct SweepHandle {
961 state: Arc<SweepShutdown>,
962 join: Option<JoinHandle<()>>,
963}
964
965struct SweepShutdown {
966 shutdown: Mutex<bool>,
967 cv: std::sync::Condvar,
968}
969
970impl SweepShutdown {
971 fn new() -> Self {
972 Self {
973 shutdown: Mutex::new(false),
974 cv: std::sync::Condvar::new(),
975 }
976 }
977
978 fn signal(&self) {
979 *self.shutdown.lock_or_poisoned() = true;
980 self.cv.notify_all();
981 }
982
983 /// Park the calling thread for up to `dur`, returning early if
984 /// `shutdown` flips. Returns `true` when shutdown was requested.
985 fn wait(&self, dur: Duration) -> bool {
986 let guard = self.shutdown.lock_or_poisoned();
987 let (guard, _timeout) = self
988 .cv
989 .wait_timeout_while(guard, dur, |s| !*s)
990 .unwrap_or_else(std::sync::PoisonError::into_inner);
991 *guard
992 }
993}
994
995impl SweepHandle {
996 fn signal_and_join(&mut self) {
997 self.state.signal();
998 if let Some(handle) = self.join.take() {
999 // Best-effort: panics from a sweep iteration shouldn't
1000 // poison the mount drop. Worst case we leak the OS thread
1001 // for a few hundred ms while it finishes its current
1002 // promote_idle pass.
1003 let _ = handle.join();
1004 }
1005 }
1006}
1007
1008impl Drop for SweepHandle {
1009 fn drop(&mut self) {
1010 self.signal_and_join();
1011 }
1012}
1013
1014/// Number of parallel workers the pre-warmer spawns. Decompression
1015/// is CPU-bound and our blobs are independent, so this scales
1016/// linearly with cores. Picked low enough to leave headroom for
1017/// rustc (or whatever the agent is doing) — bumping past 4 wins on
1018/// idle machines but contends with compile workloads on every
1019/// laptop I tested.
1020const PREWARM_WORKERS: usize = 4;
1021
1022/// Stop hydrating new blobs once the cache is this fraction full.
1023/// Without a cap the workers would happily decompress more blobs
1024/// than fit, then immediately watch the LRU evict them — pure churn
1025/// with no hit-rate benefit. 90% leaves a small headroom for
1026/// concurrent user reads that arrive while we're still warming.
1027const PREWARM_FULL_FRACTION: u8 = 90;
1028
1029/// Cumulative outcome of a prewarm pass. Returned from
1030/// [`PrewarmHandle::wait`].
1031#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
1032pub struct PrewarmStats {
1033 /// Blob hashes the tree walk discovered (file + symlink entries
1034 /// across every reachable tree).
1035 pub hashes_discovered: u64,
1036 /// Hashes the workers tried to hydrate. Equal to `hashes_discovered`
1037 /// for a run that completed, less when workers exited early
1038 /// because the cache filled up or the caller cancelled.
1039 pub hashes_visited: u64,
1040 /// Hashes that hit the cache (sibling mount already warmed
1041 /// them — the fork-thread fast path).
1042 pub already_cached: u64,
1043 /// Hashes loaded from the object store and inserted into the
1044 /// cache by this pass.
1045 pub loaded: u64,
1046 /// Whether the pass terminated naturally vs. early-stopped on
1047 /// cache fill / cancel.
1048 pub completed: bool,
1049}
1050
1051/// Handle to a running prewarm pass. See [`ContentAddressedMount::prewarm`].
1052pub struct PrewarmHandle {
1053 cancel: Arc<AtomicBool>,
1054 join: Option<JoinHandle<PrewarmStats>>,
1055}
1056
1057impl PrewarmHandle {
1058 fn start<R: RefBackend + 'static, O: OpLogBackend + 'static, S: ObjectStore + 'static>(
1059 weak: Weak<MountInner<R, O, S>>,
1060 ) -> Self {
1061 let cancel = Arc::new(AtomicBool::new(false));
1062 let cancel_for_worker = Arc::clone(&cancel);
1063 let join = std::thread::Builder::new()
1064 .name("heddle-prewarm-coordinator".to_string())
1065 .spawn(move || prewarm_run(weak, cancel_for_worker))
1066 // If we can't even spawn the coordinator, surface that
1067 // as an empty completed run rather than panicking — the
1068 // mount stays fully usable, just lukewarm.
1069 .ok();
1070 Self { cancel, join }
1071 }
1072
1073 /// Signal cancellation. Workers exit at the next poll point.
1074 /// Non-blocking; pair with [`Self::wait`] if you want to be
1075 /// sure the threads have actually stopped.
1076 pub fn cancel(&self) {
1077 self.cancel.store(true, Ordering::SeqCst);
1078 }
1079
1080 /// Block until the prewarm pass finishes (naturally or via
1081 /// cancel) and return its stats. Returns the default-zero
1082 /// stats if the coordinator thread couldn't be spawned.
1083 pub fn wait(mut self) -> PrewarmStats {
1084 self.join
1085 .take()
1086 .and_then(|h| h.join().ok())
1087 .unwrap_or_default()
1088 }
1089}
1090
1091impl Drop for PrewarmHandle {
1092 fn drop(&mut self) {
1093 // Cancel-on-drop so leaking the handle doesn't keep workers
1094 // running past the point the caller stopped caring. Workers
1095 // also self-terminate when the mount drops (Weak upgrade
1096 // fails), so this is belt-and-braces.
1097 self.cancel.store(true, Ordering::SeqCst);
1098 if let Some(join) = self.join.take() {
1099 let _ = join.join();
1100 }
1101 }
1102}
1103
1104/// Coordinator thread body: walk the tree to collect blob hashes,
1105/// then fan the hashes out across [`PREWARM_WORKERS`] worker
1106/// threads. Returns aggregate stats.
1107fn prewarm_run<R: RefBackend + 'static, O: OpLogBackend + 'static, S: ObjectStore + 'static>(
1108 weak: Weak<MountInner<R, O, S>>,
1109 cancel: Arc<AtomicBool>,
1110) -> PrewarmStats {
1111 let Some(inner) = weak.upgrade() else {
1112 return PrewarmStats::default();
1113 };
1114
1115 // Phase 1: tree walk. Cheap (in-memory tree + recent-trees
1116 // cache) so we just do it on the coordinator thread.
1117 let mut stats = PrewarmStats::default();
1118 let mut hashes: Vec<ContentHash> = Vec::new();
1119 let root_tree = inner.state.read_or_poisoned().tree;
1120 let mut queue: VecDeque<ContentHash> = VecDeque::from([root_tree]);
1121 let mut seen_trees: std::collections::HashSet<ContentHash> = std::collections::HashSet::new();
1122 while let Some(tree_hash) = queue.pop_front() {
1123 if cancel.load(Ordering::Relaxed) {
1124 return stats;
1125 }
1126 if !seen_trees.insert(tree_hash) {
1127 continue;
1128 }
1129 let Ok(Some(tree)) = inner.repo.store().get_tree(&tree_hash) else {
1130 continue;
1131 };
1132 for entry in tree.entries() {
1133 match entry.entry_type() {
1134 EntryType::Tree => {
1135 if let Some(hash) = entry.tree_hash() {
1136 queue.push_back(hash);
1137 }
1138 }
1139 EntryType::Blob | EntryType::Symlink => {
1140 if let Some(hash) = entry.content_hash() {
1141 hashes.push(hash);
1142 }
1143 stats.hashes_discovered += 1;
1144 }
1145 // Neither gitlinks nor native child-spool edges carry a local
1146 // content hash to prefetch.
1147 EntryType::Gitlink | EntryType::Spoollink => {}
1148 }
1149 }
1150 }
1151 drop(inner);
1152
1153 if hashes.is_empty() {
1154 stats.completed = true;
1155 return stats;
1156 }
1157
1158 // Phase 2: fan out. Each worker pulls indices off a shared
1159 // atomic counter — no per-worker chunking required, naturally
1160 // load-balances across blobs of varying sizes.
1161 let hashes = Arc::new(hashes);
1162 let cursor = Arc::new(AtomicUsize::new(0));
1163 let visited = Arc::new(AtomicU32::new(0));
1164 let already = Arc::new(AtomicU32::new(0));
1165 let loaded = Arc::new(AtomicU32::new(0));
1166 let stop_full = Arc::new(AtomicBool::new(false));
1167
1168 let mut workers = Vec::with_capacity(PREWARM_WORKERS);
1169 for worker_id in 0..PREWARM_WORKERS {
1170 let weak = weak.clone();
1171 let cancel = Arc::clone(&cancel);
1172 let hashes = Arc::clone(&hashes);
1173 let cursor = Arc::clone(&cursor);
1174 let visited = Arc::clone(&visited);
1175 let already = Arc::clone(&already);
1176 let loaded = Arc::clone(&loaded);
1177 let stop_full = Arc::clone(&stop_full);
1178 let handle = std::thread::Builder::new()
1179 .name(format!("heddle-prewarm-{worker_id}"))
1180 .spawn(move || {
1181 loop {
1182 if cancel.load(Ordering::Relaxed) || stop_full.load(Ordering::Relaxed) {
1183 return;
1184 }
1185 let idx = cursor.fetch_add(1, Ordering::Relaxed);
1186 if idx >= hashes.len() {
1187 return;
1188 }
1189 let hash = hashes[idx];
1190 let Some(inner) = weak.upgrade() else {
1191 return;
1192 };
1193 visited.fetch_add(1, Ordering::Relaxed);
1194 if inner.blob_cache.get(&hash).is_some() {
1195 already.fetch_add(1, Ordering::Relaxed);
1196 continue;
1197 }
1198 // Cooperative fill-stop: if the cache is already
1199 // near-full when *we* are about to insert, drop
1200 // out. Lets the agent's reads stay hot rather
1201 // than us evicting our own work.
1202 let pool = &inner.blob_cache;
1203 let full_threshold = pool
1204 .cap_bytes()
1205 .saturating_mul(PREWARM_FULL_FRACTION as usize)
1206 / 100;
1207 if pool.resident_bytes() >= full_threshold {
1208 stop_full.store(true, Ordering::Relaxed);
1209 return;
1210 }
1211 match inner.repo.store().get_blob_bytes(&hash) {
1212 Ok(Some(bytes)) => {
1213 pool.insert(hash, bytes);
1214 loaded.fetch_add(1, Ordering::Relaxed);
1215 }
1216 Ok(None) | Err(_) => {
1217 // Best-effort: a missing or unreadable
1218 // blob is the user's problem to surface
1219 // on the real read path. The prewarmer
1220 // silently skips so a corrupted blob
1221 // doesn't take down the whole pass.
1222 }
1223 }
1224 }
1225 })
1226 .ok();
1227 if let Some(h) = handle {
1228 workers.push(h);
1229 }
1230 }
1231
1232 for w in workers {
1233 let _ = w.join();
1234 }
1235
1236 stats.hashes_visited = visited.load(Ordering::Relaxed) as u64;
1237 stats.already_cached = already.load(Ordering::Relaxed) as u64;
1238 stats.loaded = loaded.load(Ordering::Relaxed) as u64;
1239 stats.completed = !cancel.load(Ordering::Relaxed) && !stop_full.load(Ordering::Relaxed);
1240 stats
1241}
1242
1243impl<R: RefBackend + 'static, O: OpLogBackend + 'static, S: ObjectStore + 'static> Drop
1244 for ContentAddressedMount<R, O, S>
1245{
1246 fn drop(&mut self) {
1247 // Signal the worker before dropping the Arc<MountInner> so
1248 // it observes the shutdown promptly rather than waiting for
1249 // a Weak::upgrade failure on the next tick.
1250 if let Some(mut handle) = self.sweeper.lock_or_poisoned().take() {
1251 handle.signal_and_join();
1252 }
1253 }
1254}
1255
1256#[derive(Clone, Copy, Debug)]
1257struct MountState {
1258 state_id: StateId,
1259 tree: ContentHash,
1260}
1261
1262/// Knobs handed to [`ContentAddressedMount::with_options`]. The
1263/// default-constructed value is what [`ContentAddressedMount::new`]
1264/// uses internally; build one explicitly when the caller wants to
1265/// share a blob cache across mounts or tune the cache cap.
1266#[derive(Clone, Default)]
1267pub struct MountOptions {
1268 /// Shared blob cache. `None` means "give me a fresh pool with
1269 /// the default cap"; clone an existing `Arc<BlobCachePool>` here
1270 /// to share warm state with sibling mounts. The daemon pattern
1271 /// is to construct one pool at startup (sized from physical
1272 /// RAM) and hand the same `Arc` to every mount it spawns.
1273 pub blob_cache: Option<Arc<BlobCachePool>>,
1274}
1275
1276impl<R: RefBackend + 'static, O: OpLogBackend + 'static, S: ObjectStore + 'static>
1277 ContentAddressedMount<R, O, S>
1278{
1279 /// Open a writable mount of `thread` against `repo`.
1280 ///
1281 /// Resolves the thread once, up front, so every subsequent
1282 /// `lookup`/`read` walks from a fixed snapshot. Writes accumulate
1283 /// in the pending tier until [`Self::capture`] folds them into a
1284 /// new state. To advance to a newer state, call [`Self::refresh`].
1285 ///
1286 /// Equivalent to [`Self::with_options`] with default options:
1287 /// a fresh per-mount blob cache. Daemon callers that want
1288 /// cross-mount cache reuse should construct an
1289 /// [`Arc<BlobCachePool>`] once and use `with_options` instead.
1290 pub fn new(repo: Repository<R, O, S>, thread: impl Into<String>) -> Result<Self> {
1291 Self::with_options(repo, thread, MountOptions::default())
1292 }
1293
1294 /// Construct a mount with explicit options. Lets the caller share
1295 /// a blob cache across mounts in the same process — see
1296 /// [`MountOptions::blob_cache`].
1297 pub fn with_options(
1298 repo: Repository<R, O, S>,
1299 thread: impl Into<String>,
1300 options: MountOptions,
1301 ) -> Result<Self> {
1302 let thread = thread.into();
1303 let state = resolve_thread(&repo, &thread)?;
1304 let inodes = Mutex::new(Inodes::new(state.tree));
1305 let blob_cache = options
1306 .blob_cache
1307 .unwrap_or_else(|| Arc::new(BlobCachePool::with_default_capacity()));
1308 let inner = Arc::new(MountInner {
1309 repo,
1310 thread,
1311 state: RwLock::new(state),
1312 inodes,
1313 pending: Mutex::new(Pending::default()),
1314 promotion: RwLock::new(PromotionPolicy::default()),
1315 mounted_at: SystemTime::now(),
1316 blob_cache,
1317 write_mu: Mutex::new(()),
1318 });
1319 let sweeper = spawn_sweep_worker(&inner);
1320 Ok(Self {
1321 inner,
1322 sweeper: Mutex::new(sweeper),
1323 })
1324 }
1325
1326 /// Borrow the shared blob cache pool. Useful when the caller
1327 /// wants to spawn a [`BlobCachePool`]-aware pre-warmer or
1328 /// inspect cache stats.
1329 pub fn blob_cache_pool(&self) -> &Arc<BlobCachePool> {
1330 &self.inner.blob_cache
1331 }
1332
1333 /// Override the promotion policy. Re-spawns (or terminates) the
1334 /// safety-sweep worker to honour the new `sweep_interval`.
1335 /// Mostly useful for tests that want a tight idle window or to
1336 /// disable idle-promotion entirely.
1337 pub fn with_promotion_policy(self, policy: PromotionPolicy) -> Self {
1338 // Terminate any pre-existing worker before mutating policy
1339 // so we never have two workers racing on `pending`.
1340 if let Some(mut handle) = self.sweeper.lock_or_poisoned().take() {
1341 handle.signal_and_join();
1342 }
1343 // Swap the active policy in-place. The worker has been
1344 // joined above, so there's no concurrent reader.
1345 *self.inner.promotion.write_or_poisoned() = policy;
1346 // Spawn a fresh worker matching the new policy.
1347 let sweeper = spawn_sweep_worker(&self.inner);
1348 *self.sweeper.lock_or_poisoned() = sweeper;
1349 self
1350 }
1351
1352 /// Re-resolve the thread and adopt the new state. Existing
1353 /// inodes are *not* invalidated — callers who want a clean slate
1354 /// should drop the mount and recreate.
1355 pub fn refresh(&self) -> Result<()> {
1356 let next = resolve_thread(&self.inner.repo, &self.inner.thread)?;
1357 *self.inner.state.write_or_poisoned() = next;
1358 Ok(())
1359 }
1360
1361 /// The thread name this mount serves.
1362 pub fn thread(&self) -> &str {
1363 &self.inner.thread
1364 }
1365
1366 /// The state id this mount currently points at.
1367 pub fn current_state_id(&self) -> StateId {
1368 self.inner.state.read_or_poisoned().state_id
1369 }
1370
1371 fn store(&self) -> &S {
1372 self.inner.repo.store()
1373 }
1374
1375 fn load_tree(&self, hash: &ContentHash) -> Result<Tree> {
1376 self.store()
1377 .get_tree(hash)?
1378 .ok_or_else(|| MountError::NotFound(format!("tree {hash}")))
1379 }
1380
1381 /// Drop every cached blob — both the mount-side LRU and the
1382 /// underlying `ObjectStore`'s `recent_blobs`/`recent_trees`
1383 /// caches. The next `read` on each blob pays full I/O +
1384 /// decompression cost. Exposed for benchmarks that want to
1385 /// measure the true cold-cache path without rebuilding the
1386 /// whole mount.
1387 pub fn clear_blob_cache(&self)
1388 where
1389 S: ObjectCacheControl,
1390 {
1391 self.inner.blob_cache.clear();
1392 self.inner.repo.store().clear_recent_caches();
1393 }
1394
1395 /// Spawn a background tree-walker that hydrates every file blob
1396 /// in the captured tree into the shared blob cache. The first
1397 /// kernel `read` after this finishes is served from memory at
1398 /// `Arc::clone` + `memcpy` cost — beats `std::fs::read` on every
1399 /// tier we benchmark.
1400 ///
1401 /// The returned [`PrewarmHandle`] is the caller's lever:
1402 /// * Drop it without calling anything → the prewarmer keeps
1403 /// running until natural completion or the mount drops
1404 /// (the workers hold `Weak<MountInner>` and self-terminate
1405 /// when the strong count hits zero).
1406 /// * `.cancel()` signals shutdown without joining.
1407 /// * `.wait()` joins all workers and returns the final stats.
1408 ///
1409 /// Workers stop early when the cache is ≥ 90% full to avoid
1410 /// churn-evicting work they just did. Blobs already cached
1411 /// (from a sibling mount sharing the same pool) are skipped
1412 /// cheaply — this is the fork-thread fast path.
1413 pub fn prewarm(&self) -> PrewarmHandle {
1414 PrewarmHandle::start(Arc::downgrade(&self.inner))
1415 }
1416
1417 fn load_blob_bytes(&self, hash: &ContentHash) -> Result<bytes::Bytes> {
1418 if let Some(hit) = self.inner.blob_cache.get(hash) {
1419 return Ok(hit);
1420 }
1421 let bytes = self
1422 .store()
1423 .get_blob_bytes(hash)?
1424 .ok_or_else(|| MountError::NotFound(format!("blob {hash}")))?;
1425 self.inner.blob_cache.insert(*hash, bytes.clone());
1426 Ok(bytes)
1427 }
1428
1429 /// Header-only size lookup. Avoids loading the full blob just to
1430 /// learn its size — the hot path for `ls -l`.
1431 fn blob_size(&self, hash: &ContentHash) -> Result<u64> {
1432 self.store()
1433 .blob_size(hash)?
1434 .ok_or_else(|| MountError::NotFound(format!("blob {hash}")))
1435 }
1436
1437 fn record_for(&self, id: NodeId) -> Result<NodeRecord> {
1438 self.inner
1439 .inodes
1440 .lock_or_poisoned()
1441 .get(id)
1442 .ok_or_else(|| MountError::Stale(format!("node {}", id.0)))
1443 }
1444
1445 fn intern(&self, record: NodeRecord) -> NodeId {
1446 self.inner.inodes.lock_or_poisoned().intern(record)
1447 }
1448
1449 /// Resolve a mount-relative path to a [`NodeId`]. Used by tests
1450 /// that don't go through `lookup` step-by-step.
1451 pub fn lookup_path(&self, path: impl AsRef<Path>) -> Result<NodeId> {
1452 let mut node = NodeId::ROOT;
1453 for component in path.as_ref().components() {
1454 match component {
1455 Component::CurDir | Component::RootDir => continue,
1456 Component::Prefix(_) => {
1457 return Err(MountError::NotFound(format!(
1458 "unsupported path component in {}",
1459 path.as_ref().display()
1460 )));
1461 }
1462 Component::ParentDir => {
1463 return Err(MountError::NotFound(format!(
1464 "parent traversal not supported: {}",
1465 path.as_ref().display()
1466 )));
1467 }
1468 Component::Normal(name) => {
1469 let entry = self
1470 .lookup(node, name)?
1471 .ok_or_else(|| MountError::NotFound(name.to_string_lossy().into_owned()))?;
1472 node = entry.node;
1473 }
1474 }
1475 }
1476 Ok(node)
1477 }
1478
1479 fn entry_from_tree_entry(&self, parent_path: &Path, tree_entry: &TreeEntry) -> Result<Entry> {
1480 let entry_path = join_child(parent_path, tree_entry.name());
1481 let (kind, size, unix_mode, record) = match tree_entry.target() {
1482 TreeEntryTarget::Tree { hash } => {
1483 // We deliberately load the subtree here so the entry
1484 // count (the conventional "size" for a directory)
1485 // matches what userspace expects from `stat`.
1486 let subtree = self.load_tree(hash)?;
1487 (
1488 NodeKind::Directory,
1489 subtree.entries().len() as u64,
1490 DIR_UNIX_MODE,
1491 NodeRecord::Dir {
1492 tree: *hash,
1493 path: entry_path,
1494 },
1495 )
1496 }
1497 TreeEntryTarget::Blob { hash, executable } => {
1498 let size = self.blob_size(hash)?;
1499 let mode = if *executable {
1500 FileMode::Executable
1501 } else {
1502 FileMode::Normal
1503 };
1504 (
1505 kind_for_mode(mode),
1506 size,
1507 mode.to_unix_mode(),
1508 NodeRecord::File {
1509 blob: *hash,
1510 mode,
1511 path: entry_path,
1512 },
1513 )
1514 }
1515 TreeEntryTarget::Symlink { hash } => {
1516 let size = self.blob_size(hash)?;
1517 (
1518 NodeKind::Symlink,
1519 size,
1520 FileMode::Symlink.to_unix_mode(),
1521 NodeRecord::Symlink { blob: *hash },
1522 )
1523 }
1524 TreeEntryTarget::Gitlink { target } => {
1525 let placeholder = gitlink_placeholder_bytes(target);
1526 let size = placeholder.len() as u64;
1527 (
1528 NodeKind::File,
1529 size,
1530 FileMode::Normal.to_unix_mode(),
1531 NodeRecord::Gitlink {
1532 placeholder,
1533 path: entry_path,
1534 },
1535 )
1536 }
1537 // Native child-spool edges are not yet exposed in the FUSE mount
1538 // (no consumer facet wires them in this phase). Refuse explicitly
1539 // rather than masquerade one as a gitlink placeholder. Native
1540 // spool operations do not produce mounted trees containing these
1541 // yet, so this path is not hit in practice today.
1542 TreeEntryTarget::Spoollink { .. } => {
1543 return Err(MountError::InvalidArgument(format!(
1544 "spoollink entry '{}' is not exposable in the mount yet",
1545 tree_entry.name()
1546 )));
1547 }
1548 };
1549 let node = self.intern(record);
1550 Ok(Entry {
1551 node,
1552 name: OsString::from(tree_entry.name()),
1553 kind,
1554 size,
1555 unix_mode,
1556 })
1557 }
1558
1559 /// Build an [`Entry`] from a [`PendingHit`]. `path` is the child's
1560 /// mount-relative path (used to intern the `PendingFile` /
1561 /// `PendingSymlink` record for warm/symlink hits); `name` is the
1562 /// leaf name of the returned entry. Returns `None` for
1563 /// [`PendingHit::Tombstone`] — the caller treats that as "entry
1564 /// hidden". Shared by `lookup` and `enumerate`.
1565 fn entry_from_pending_hit(&self, hit: PendingHit, path: &Path, name: &OsStr) -> Option<Entry> {
1566 match hit {
1567 PendingHit::Tombstone => None,
1568 PendingHit::Hot { node, size, mode } => Some(Entry {
1569 node,
1570 name: name.to_os_string(),
1571 kind: kind_for_mode(mode),
1572 size,
1573 unix_mode: mode.to_unix_mode(),
1574 }),
1575 PendingHit::Warm {
1576 blob: _,
1577 size,
1578 mode,
1579 } => {
1580 let node = self.intern(NodeRecord::PendingFile {
1581 path: path.to_path_buf(),
1582 mode,
1583 });
1584 Some(Entry {
1585 node,
1586 name: name.to_os_string(),
1587 kind: kind_for_mode(mode),
1588 size,
1589 unix_mode: mode.to_unix_mode(),
1590 })
1591 }
1592 PendingHit::Symlink { target_len } => {
1593 let node = self.intern(NodeRecord::PendingSymlink {
1594 path: path.to_path_buf(),
1595 });
1596 Some(Entry {
1597 node,
1598 name: name.to_os_string(),
1599 kind: NodeKind::Symlink,
1600 size: target_len,
1601 unix_mode: FileMode::Symlink.to_unix_mode(),
1602 })
1603 }
1604 }
1605 }
1606
1607 fn tree_for_record(&self, record: &NodeRecord) -> Result<Tree> {
1608 match record {
1609 NodeRecord::Root { tree } | NodeRecord::Dir { tree, .. } => self.load_tree(tree),
1610 // Pending-only dirs have no captured tree to load yet —
1611 // their content lives entirely in the pending tier.
1612 NodeRecord::PendingDir { .. } => Ok(Tree::new()),
1613 _ => Err(MountError::NotADirectory(format!("{record:?}"))),
1614 }
1615 }
1616
1617 /// Mount-relative path for a directory record. Root resolves to
1618 /// `""`, captured Dirs and pending dirs to their stored path.
1619 fn dir_path_of(&self, record: &NodeRecord) -> Option<PathBuf> {
1620 match record {
1621 NodeRecord::Root { .. } => Some(PathBuf::new()),
1622 NodeRecord::Dir { path, .. } | NodeRecord::PendingDir { path } => Some(path.clone()),
1623 _ => None,
1624 }
1625 }
1626
1627 /// Build the relative path of `node` from the mount root, used to
1628 /// rendezvous a NodeId with its pending-tier entry. Returns `None`
1629 /// for the root or for nodes that don't carry a path identity.
1630 fn path_of(&self, record: &NodeRecord) -> Option<PathBuf> {
1631 match record {
1632 NodeRecord::PendingFile { path, .. }
1633 | NodeRecord::File { path, .. }
1634 | NodeRecord::Gitlink { path, .. } => Some(path.clone()),
1635 NodeRecord::Dir { path, .. } | NodeRecord::PendingDir { path } => Some(path.clone()),
1636 NodeRecord::PendingSymlink { path } => Some(path.clone()),
1637 _ => None,
1638 }
1639 }
1640
1641 // --- Pending tier helpers ------------------------------------------------
1642
1643 fn promote_idle_buffers(&self) -> Result<()> {
1644 self.inner.sweep_idle_buffers()
1645 }
1646
1647 /// Promote the hot buffer for `node` (if any) to a CAS blob and
1648 /// record it in the pending tree. Routed from the FUSE `flush`
1649 /// callback (per-descriptor-close). Orphaned nodes deliberately
1650 /// do nothing here — see [`MountInner::flush_node`] for the
1651 /// lifecycle rationale.
1652 pub fn flush_node(&self, node: NodeId) -> Result<()> {
1653 self.inner.flush_node(node)
1654 }
1655
1656 /// Final close of `node` from a FUSE `release` callback. Decrements
1657 /// the open-handle refcount; on the last close, drops orphan
1658 /// state and (for non-orphans) promotes any surviving hot buffer.
1659 pub fn release_node(&self, node: NodeId) -> Result<()> {
1660 self.inner.release_node(node)
1661 }
1662
1663 /// Notify the mount that a new open handle for `node` was minted
1664 /// (FUSE `open` / `create` callback). Used to time the orphan
1665 /// cleanup against the *final* close (see
1666 /// [`Self::release_node`] / [`MountInner::release_node`]).
1667 ///
1668 /// Bumps the open count on the existing `NodeState`, minting a
1669 /// `Live { open_count: 1 }` entry if the node is untracked. An
1670 /// Orphan can also be opened (rare — only via an fh the kernel
1671 /// still holds across a re-lookup race); we bump its refcount so
1672 /// the final release fires correctly.
1673 pub fn on_open(&self, node: NodeId) -> Result<()> {
1674 let mut pending = self.inner.pending.lock_or_poisoned();
1675 let next = match pending.state.get(&node.0).copied() {
1676 None => NodeState::Live { open_count: 1 },
1677 Some(NodeState::Live { open_count }) => NodeState::Live {
1678 open_count: open_count.saturating_add(1),
1679 },
1680 Some(NodeState::Orphan { open_count }) => NodeState::Orphan {
1681 open_count: open_count.saturating_add(1),
1682 },
1683 };
1684 pending.state.insert(node.0, next);
1685 Ok(())
1686 }
1687
1688 /// Mark `path` as deleted in the pending tier. Subsequent
1689 /// `lookup`/`enumerate` calls will skip the underlying captured
1690 /// entry, and `capture()` will fold the deletion into the new
1691 /// state's tree (pruning empty parent dirs as needed).
1692 ///
1693 /// Low-level (path-based) helper — unlike [`Self::unlink_entry`]
1694 /// it does not honour POSIX open-unlinked semantics. Used by
1695 /// tests that bypass the FUSE-callback lifecycle. The
1696 /// NodeId-keyed buffers for the path's current owner are dropped
1697 /// (no orphan tracking).
1698 pub fn unlink_path(&self, path: impl AsRef<Path>) -> Result<()> {
1699 let path = path.as_ref().to_path_buf();
1700 // Resolve path → NodeId via the inode registry so we can drop
1701 // the per-NodeId warm/hot bytes.
1702 let bound_id = {
1703 let inodes = self.inner.inodes.lock_or_poisoned();
1704 inodes.by_path.get(&path).copied()
1705 };
1706 let mut pending = self.inner.pending.lock_or_poisoned();
1707 if let Some(node_id) = pending.hot_by_path.remove(&path) {
1708 pending.hot.remove(&node_id);
1709 pending.warm.remove(&node_id);
1710 pending.state.remove(&node_id);
1711 }
1712 if let Some(node_id) = bound_id {
1713 pending.hot.remove(&node_id);
1714 pending.warm.remove(&node_id);
1715 pending.state.remove(&node_id);
1716 }
1717 pending.symlinks.remove(&path);
1718 pending.child_index.remove(&path);
1719 pending.tombstones.insert(path.clone());
1720 drop(pending);
1721 if bound_id.is_some() {
1722 let mut inodes = self.inner.inodes.lock_or_poisoned();
1723 inodes.by_path.remove(&path);
1724 }
1725 Ok(())
1726 }
1727
1728 // --- Write-side overlay ops (heddle#180) -----------------------------------
1729 //
1730 // Each method below corresponds to one FUSE callback the kernel
1731 // emits on cargo / git / npm style workloads:
1732 //
1733 // create → `create_file` open(O_CREAT)
1734 // mkdir → `make_dir`
1735 // unlink → `unlink_entry`
1736 // rmdir → `rmdir_entry`
1737 // rename → `rename_entry`
1738 // setattr → `set_attrs` chmod / ftruncate / O_TRUNC
1739 // symlink → `create_symlink`
1740 // readlink→ `read_link`
1741 //
1742 // All mutations land in the per-thread overlay (pending tier):
1743 //
1744 // * `Pending::hot` / `Pending::warm` — file bytes (existing).
1745 // * `Pending::tombstones` — file deletions (existing).
1746 // * `Pending::dir_tombstones` — `rmdir` of a captured dir.
1747 // * `Pending::explicit_dirs` — empty mkdirs.
1748 // * `Pending::symlinks` — link target bytes.
1749 //
1750 // None of these touch the underlying CAS until `capture()` folds
1751 // the overlay into a real heddle state.
1752
1753 /// Open-or-create a regular file under `parent`, mirroring
1754 /// `open(O_CREAT[|O_EXCL])` from userspace.
1755 ///
1756 /// When the named entry doesn't exist, mints a fresh
1757 /// [`NodeRecord::PendingFile`] inode + an empty hot buffer so the
1758 /// new path is immediately visible to [`lookup`](Self::lookup) /
1759 /// [`attrs`](Self::attrs) and the first
1760 /// [`write`](Self::write) drops cleanly into the existing
1761 /// two-tier model.
1762 ///
1763 /// When the named entry already exists:
1764 /// * `exclusive=true` ⇒ [`MountError::AlreadyExists`] (errno
1765 /// `EEXIST`).
1766 /// * `exclusive=false` ⇒ returns the existing entry. The kernel
1767 /// follows up with `setattr(size=0)` for `O_TRUNC` callers,
1768 /// which we honour in [`set_attrs`](Self::set_attrs).
1769 pub fn create_file(
1770 &self,
1771 parent: NodeId,
1772 name: &OsStr,
1773 mode: FileMode,
1774 exclusive: bool,
1775 ) -> Result<Entry> {
1776 // R8: serialize against rename / mkdir / symlink so an
1777 // exclusivity check (O_EXCL or rename-noreplace) lands its
1778 // existence-test and its mutation under the same write-side
1779 // critical section.
1780 let _write_guard = self.inner.write_mu.lock_or_poisoned();
1781 let name_str = validate_entry_name(name)?;
1782 if let Some(existing) = self.lookup(parent, name)? {
1783 if exclusive {
1784 return Err(MountError::AlreadyExists(name_str.to_string()));
1785 }
1786 return Ok(existing);
1787 }
1788 let parent_record = self.record_for(parent)?;
1789 let parent_path = self
1790 .dir_path_of(&parent_record)
1791 .ok_or_else(|| MountError::NotADirectory(format!("{parent_record:?}")))?;
1792 let child_path = join_child(&parent_path, name_str);
1793
1794 {
1795 let mut pending = self.inner.pending.lock_or_poisoned();
1796 // A prior unlink left a tombstone — clear it; the file
1797 // exists again.
1798 pending.tombstones.remove(&child_path);
1799 // An overlay-only directory used to live here; it's gone.
1800 pending.explicit_dirs.remove(&child_path);
1801 }
1802
1803 let node = self.intern(NodeRecord::PendingFile {
1804 path: child_path.clone(),
1805 mode,
1806 });
1807
1808 // Seed an empty hot buffer so the freshly-minted inode reads
1809 // as a 0-byte file even before any `write` callback fires.
1810 // Mirrors what userspace expects from `open(O_CREAT)`: the
1811 // file exists at length 0 immediately on return.
1812 {
1813 let mut pending = self.inner.pending.lock_or_poisoned();
1814 pending.hot.insert(
1815 node.0,
1816 HotBuffer {
1817 path: child_path.clone(),
1818 mode,
1819 bytes: Vec::new(),
1820 last_touched: Instant::now(),
1821 revision: 0,
1822 },
1823 );
1824 pending.hot_by_path.insert(child_path.clone(), node.0);
1825 pending.child_index.insert(
1826 child_path,
1827 PendingChildKind::HotFile {
1828 node,
1829 size: 0,
1830 mode,
1831 },
1832 );
1833 }
1834
1835 Ok(Entry {
1836 node,
1837 name: name.to_os_string(),
1838 kind: kind_for_mode(mode),
1839 size: 0,
1840 unix_mode: mode.to_unix_mode(),
1841 })
1842 }
1843
1844 /// Create an empty directory under `parent`. Recorded as an
1845 /// [`Pending::explicit_dirs`] entry so the new path is visible to
1846 /// lookup/enumerate even when no child has been written yet.
1847 pub fn make_dir(&self, parent: NodeId, name: &OsStr) -> Result<Entry> {
1848 // R8: serialize with other write-side mutations.
1849 let _write_guard = self.inner.write_mu.lock_or_poisoned();
1850 let name_str = validate_entry_name(name)?;
1851 if self.lookup(parent, name)?.is_some() {
1852 return Err(MountError::AlreadyExists(name_str.to_string()));
1853 }
1854 let parent_record = self.record_for(parent)?;
1855 let parent_path = self
1856 .dir_path_of(&parent_record)
1857 .ok_or_else(|| MountError::NotADirectory(format!("{parent_record:?}")))?;
1858 let child_path = join_child(&parent_path, name_str);
1859
1860 {
1861 let mut pending = self.inner.pending.lock_or_poisoned();
1862 // A rmdir of this exact path now reverts to "present".
1863 pending.dir_tombstones.remove(&child_path);
1864 // Clear any colliding file tombstone too.
1865 pending.tombstones.remove(&child_path);
1866 pending.explicit_dirs.insert(child_path.clone());
1867 pending
1868 .child_index
1869 .insert(child_path.clone(), PendingChildKind::Dir);
1870 }
1871
1872 let node = self.intern(NodeRecord::PendingDir { path: child_path });
1873 Ok(Entry {
1874 node,
1875 name: name.to_os_string(),
1876 kind: NodeKind::Directory,
1877 size: 0,
1878 unix_mode: DIR_UNIX_MODE,
1879 })
1880 }
1881
1882 /// Delete a regular file (or symlink) named `name` under `parent`.
1883 ///
1884 /// POSIX open-unlinked semantics: the directory entry goes (path
1885 /// tombstoned, `inodes.by_path[path]` retired), but if any fd
1886 /// still references the inode, the bytes survive in `hot[node]` /
1887 /// `warm[node]` until the final `release`. Under the post-spike
1888 /// unified NodeId-keyed model
1889 /// (`docs/design/mount-posix-semantics.md` §1.2 T1/T2), this is a
1890 /// state transition only — no byte migration. Pre-spike code
1891 /// dropped `pending.hot[node_id]` here (Codex thread 3293307302
1892 /// r9) and migrated `warm[path]` into `orphan_warm[node]` (r8);
1893 /// both steps go away.
1894 pub fn unlink_entry(&self, parent: NodeId, name: &OsStr) -> Result<()> {
1895 // R8: serialize with other write-side mutations.
1896 let _write_guard = self.inner.write_mu.lock_or_poisoned();
1897 let name_str = validate_entry_name(name)?;
1898 let entry = self
1899 .lookup(parent, name)?
1900 .ok_or_else(|| MountError::NotFound(name_str.to_string()))?;
1901 if entry.kind == NodeKind::Directory {
1902 return Err(MountError::IsADirectory(name_str.to_string()));
1903 }
1904 let parent_record = self.record_for(parent)?;
1905 let parent_path = self
1906 .dir_path_of(&parent_record)
1907 .ok_or_else(|| MountError::NotADirectory(format!("{parent_record:?}")))?;
1908 let child_path = join_child(&parent_path, name_str);
1909 let node_id = entry.node.0;
1910
1911 {
1912 let mut pending = self.inner.pending.lock_or_poisoned();
1913 // Detach the path-level hot binding. The bytes follow the
1914 // NodeId, so `hot[node_id]` / `warm[node_id]` stay put —
1915 // the surviving fd reads them via the orphan branches in
1916 // `read` / `attrs` / `write` / `apply_truncate`.
1917 // (r9 fix: pre-spike code called `pending.hot.remove(&node_id)`
1918 // here and the unflushed bytes vanished.)
1919 pending.hot_by_path.remove(&child_path);
1920 // Transition T1: Live{open_count >= 1} → Orphan{open_count}.
1921 // The witness-gated retrofit (heddle#209) makes the FSM
1922 // check the gate: `bp.transition_to_orphan(node_id)`
1923 // returns `None` (without touching `state`) for any
1924 // non-`LiveNonZero` state, and the missing
1925 // `Witness<Orphan>` IS the short-circuit at this call
1926 // site.
1927 //
1928 // That subsumes two earlier defensive checks: Codex r12
1929 // thread 3293510317 (symlinks have no `open`/`release`
1930 // lifecycle, so they never enter `state` and the
1931 // transition never fires for them), and r11 finding
1932 // 3293575534 (orphaning a `Live { open_count: 0 }` node
1933 // creates a record nothing will ever reap — same shape,
1934 // same fix).
1935 pending.with_brand(|bp| {
1936 let _ = bp.transition_to_orphan(node_id);
1937 });
1938 // Symlinks are path-keyed; their overlay goes when the
1939 // directory entry goes.
1940 pending.symlinks.remove(&child_path);
1941 pending.child_index.remove(&child_path);
1942 pending.tombstones.insert(child_path.clone());
1943 }
1944 // Retire the path→inode mapping so a subsequent `create_file`
1945 // at the same name mints a fresh inode (POSIX unlink/recreate
1946 // isolation — open-unlinked temp files must not be aliased by
1947 // a replacement at the same path). The `by_id` record stays so
1948 // any still-open kernel handle keeps resolving until `forget`.
1949 {
1950 let mut inodes = self.inner.inodes.lock_or_poisoned();
1951 inodes.by_path.remove(&child_path);
1952 }
1953 Ok(())
1954 }
1955
1956 /// Remove the empty directory `name` under `parent`. Fails with
1957 /// `ENOTEMPTY` if any child resolves through the mount.
1958 pub fn rmdir_entry(&self, parent: NodeId, name: &OsStr) -> Result<()> {
1959 // R8: serialize with other write-side mutations.
1960 let _write_guard = self.inner.write_mu.lock_or_poisoned();
1961 let name_str = validate_entry_name(name)?;
1962 let entry = self
1963 .lookup(parent, name)?
1964 .ok_or_else(|| MountError::NotFound(name_str.to_string()))?;
1965 if entry.kind != NodeKind::Directory {
1966 return Err(MountError::NotADirectory(name_str.to_string()));
1967 }
1968 // Empty check via enumerate — already overlay-aware (hot,
1969 // warm, symlinks, captured-with-pending-overlay).
1970 let children = self.enumerate(entry.node)?;
1971 if !children.is_empty() {
1972 return Err(MountError::NotEmpty(name_str.to_string()));
1973 }
1974 let parent_record = self.record_for(parent)?;
1975 let parent_path = self
1976 .dir_path_of(&parent_record)
1977 .ok_or_else(|| MountError::NotADirectory(format!("{parent_record:?}")))?;
1978 let child_path = join_child(&parent_path, name_str);
1979
1980 {
1981 let mut pending = self.inner.pending.lock_or_poisoned();
1982 pending.explicit_dirs.remove(&child_path);
1983 pending.child_index.remove(&child_path);
1984 pending.dir_tombstones.insert(child_path.clone());
1985 }
1986 // Codex r12 thread 3293510310 (P1): retire the path → inode
1987 // mapping. Otherwise `Inodes::intern` would coalesce a
1988 // subsequent `create_file` / `make_dir` at this path onto the
1989 // removed directory's NodeId, rebinding a cached directory
1990 // inode to a different object type — the same stale-handle
1991 // class `unlink_entry` already guards against. The `by_id`
1992 // record stays so any kernel handle the FS still holds keeps
1993 // resolving until `forget`.
1994 {
1995 let mut inodes = self.inner.inodes.lock_or_poisoned();
1996 inodes.by_path.remove(&child_path);
1997 }
1998 Ok(())
1999 }
2000
2001 /// Move `(old_parent, old_name)` to `(new_parent, new_name)`.
2002 /// Handles file + symlink renames across any pair of overlay /
2003 /// captured paths, and overlay-only directory rename (a captured
2004 /// directory rename would require recursively rewriting the
2005 /// tombstone/warm map — out of scope for the cargo / git path).
2006 pub fn rename_entry(
2007 &self,
2008 old_parent: NodeId,
2009 old_name: &OsStr,
2010 new_parent: NodeId,
2011 new_name: &OsStr,
2012 ) -> Result<()> {
2013 self.rename_entry_with_options(
2014 old_parent,
2015 old_name,
2016 new_parent,
2017 new_name,
2018 RenameOptions::default(),
2019 )
2020 }
2021
2022 /// Same as [`Self::rename_entry`] but honours [`RenameOptions`].
2023 /// `no_replace` (Linux `RENAME_NOREPLACE`) refuses the rename when
2024 /// the destination already resolves; the check is performed inside
2025 /// the same write-side critical section as the mutation, so a
2026 /// concurrent writer cannot install the destination between the
2027 /// check and the rename.
2028 pub fn rename_entry_with_options(
2029 &self,
2030 old_parent: NodeId,
2031 old_name: &OsStr,
2032 new_parent: NodeId,
2033 new_name: &OsStr,
2034 options: RenameOptions,
2035 ) -> Result<()> {
2036 // R8 (Codex Thread 3293235163): the existence-check + the
2037 // directory-entry mutation must land under the same mutation
2038 // lock. Holding `write_mu` for the duration of this method
2039 // serializes the rename against every other write-side op
2040 // that could install the destination (create_file, make_dir,
2041 // create_symlink, another rename) — that's the atomicity the
2042 // POSIX NOREPLACE flag promises.
2043 let _write_guard = self.inner.write_mu.lock_or_poisoned();
2044
2045 let old_name_str = validate_entry_name(old_name)?;
2046 let new_name_str = validate_entry_name(new_name)?;
2047 let src = self
2048 .lookup(old_parent, old_name)?
2049 .ok_or_else(|| MountError::NotFound(format!("rename src {old_name_str}")))?;
2050 let old_parent_record = self.record_for(old_parent)?;
2051 let new_parent_record = self.record_for(new_parent)?;
2052 let old_parent_path = self
2053 .dir_path_of(&old_parent_record)
2054 .ok_or_else(|| MountError::NotADirectory(format!("{old_parent_record:?}")))?;
2055 let new_parent_path = self
2056 .dir_path_of(&new_parent_record)
2057 .ok_or_else(|| MountError::NotADirectory(format!("{new_parent_record:?}")))?;
2058 let old_path = join_child(&old_parent_path, old_name_str);
2059 let new_path = join_child(&new_parent_path, new_name_str);
2060 if old_path == new_path {
2061 return Ok(());
2062 }
2063
2064 // POSIX: destination of a different kind is an error. We also
2065 // honour NOREPLACE here while still holding `write_mu` so the
2066 // check + the subsequent move are atomic against concurrent
2067 // writers. `dst` is shadowed for the kind-mismatch arm and
2068 // hoisted into `displaced_inode_id` so the move primitives
2069 // can preserve the displaced inode's warm bytes (r8).
2070 let dst = self.lookup(new_parent, new_name)?;
2071 if dst.is_some() && options.no_replace {
2072 return Err(MountError::AlreadyExists(new_name_str.to_string()));
2073 }
2074 if let Some(ref d) = dst {
2075 match (src.kind, d.kind) {
2076 (NodeKind::Directory, NodeKind::Directory) => {
2077 let dst_children = self.enumerate(d.node)?;
2078 if !dst_children.is_empty() {
2079 return Err(MountError::NotEmpty(new_name_str.to_string()));
2080 }
2081 }
2082 (NodeKind::Directory, _) => {
2083 return Err(MountError::NotADirectory(new_name_str.to_string()));
2084 }
2085 (_, NodeKind::Directory) => {
2086 return Err(MountError::IsADirectory(new_name_str.to_string()));
2087 }
2088 _ => {}
2089 }
2090 }
2091 let displaced_inode_id = dst.as_ref().map(|d| d.node.0);
2092
2093 match src.kind {
2094 NodeKind::File => self.move_file(&old_path, &new_path, displaced_inode_id)?,
2095 NodeKind::Symlink => self.move_symlink(&old_path, &new_path, displaced_inode_id)?,
2096 NodeKind::Directory => self.move_overlay_dir(&old_path, &new_path)?,
2097 }
2098 // Maintain the inode↔path invariant for both the source and
2099 // destination: the kernel may have cached either dentry from
2100 // a prior lookup, and (for FUSE) it does not re-issue lookup
2101 // after `rename` — it just rewrites its own dentry → inode
2102 // table. So the source inode now resolves through dentry
2103 // `new_name`, and any read against it must serve the new
2104 // path's overlay state. Rewriting the source record's stored
2105 // path is what keeps that consistent. The dest's old inode
2106 // (which the kernel will issue `forget` for) gets dropped
2107 // from `by_path` so the next lookup mints a fresh id.
2108 {
2109 let mut inodes = self.inner.inodes.lock_or_poisoned();
2110 // Detach the destination's path mapping. The inode record
2111 // stays in `by_id` so any kernel handle the FS still holds
2112 // for the replaced file keeps resolving (the kernel cleans
2113 // up via `forget` on close). POSIX semantics: rename-over
2114 // must not invalidate an already-open dest descriptor.
2115 let displaced_dest = inodes.by_path.remove(&new_path);
2116 // Rewrite the source inode's stored path so subsequent
2117 // reads/attrs against it serve the new-path overlay.
2118 // The kernel keeps using the source's NodeId after rename
2119 // (it's just a dentry-table rewrite on its side) — without
2120 // this, every read against the rebased dentry sees the
2121 // stale path and returns ESTALE.
2122 let rebased_src = if let Some(src_id) = inodes.by_path.remove(&old_path) {
2123 if let Some(
2124 NodeRecord::PendingFile { path, .. }
2125 | NodeRecord::File { path, .. }
2126 | NodeRecord::Gitlink { path, .. }
2127 | NodeRecord::PendingSymlink { path }
2128 | NodeRecord::Dir { path, .. }
2129 | NodeRecord::PendingDir { path },
2130 ) = inodes.by_id.get_mut(&src_id)
2131 {
2132 *path = new_path.clone();
2133 }
2134 inodes.by_path.insert(new_path.clone(), src_id);
2135 // For a directory rename, also rebase every cached
2136 // descendant inode. The kernel may already hold dentry
2137 // → inode bindings for `old_path/<child>` from prior
2138 // lookups, and reads against those inodes would
2139 // otherwise resolve through the stale path (ESTALE on
2140 // PendingFile, or the wrong overlay on File). Walk
2141 // by_path once, collect the entries under the old
2142 // prefix, then rewrite both the mapping and the
2143 // NodeRecord's stored path.
2144 if src.kind == NodeKind::Directory {
2145 let descendants: Vec<(PathBuf, PathBuf, u64)> = inodes
2146 .by_path
2147 .iter()
2148 .filter_map(|(p, id)| {
2149 let tail = p.strip_prefix(&old_path).ok()?;
2150 if tail.as_os_str().is_empty() {
2151 return None;
2152 }
2153 Some((p.clone(), new_path.join(tail), *id))
2154 })
2155 .collect();
2156 for (old_key, new_key, id) in descendants {
2157 inodes.by_path.remove(&old_key);
2158 if let Some(
2159 NodeRecord::PendingFile { path, .. }
2160 | NodeRecord::File { path, .. }
2161 | NodeRecord::Gitlink { path, .. }
2162 | NodeRecord::PendingSymlink { path }
2163 | NodeRecord::Dir { path, .. }
2164 | NodeRecord::PendingDir { path },
2165 ) = inodes.by_id.get_mut(&id)
2166 {
2167 *path = new_key.clone();
2168 }
2169 inodes.by_path.insert(new_key, id);
2170 }
2171 }
2172 Some(src_id)
2173 } else {
2174 None
2175 };
2176 drop(inodes);
2177 // Reach into pending for two cleanups under one lock:
2178 // * The source's hot buffer (if any) carries the old
2179 // path; rebase it. Descendant hot-buffer paths are
2180 // already handled by `move_overlay_dir`'s
2181 // `hot_path_updates` pass.
2182 // * The displaced destination (if any) becomes an
2183 // orphan: its directory entry is gone but the inode
2184 // id may still be held by a kernel fd. Subsequent
2185 // `write` / `apply_truncate` / `set_attrs` /
2186 // `read` / `attrs` calls through that fd consult
2187 // `Pending::orphans` and take the per-NodeId branch
2188 // instead of the rebased path overlay. The companion
2189 // orphan branch in `flush_node` drops any preserved
2190 // buffer without warm-promoting.
2191 let mut pending = self.inner.pending.lock_or_poisoned();
2192 if let Some(src_id) = rebased_src
2193 && let Some(buf) = pending.hot.get_mut(&src_id)
2194 {
2195 buf.path = new_path.clone();
2196 buf.revision = buf.revision.wrapping_add(1);
2197 }
2198 match src.kind {
2199 NodeKind::Directory => pending.child_index.rebase_prefix(&old_path, &new_path),
2200 NodeKind::File => {
2201 let moved = pending.child_index.remove(&old_path);
2202 pending.child_index.remove(&new_path);
2203 let indexed = moved.or_else(|| {
2204 let id = rebased_src?;
2205 if let Some(buf) = pending.hot.get(&id) {
2206 Some(PendingChildKind::HotFile {
2207 node: NodeId(id),
2208 size: buf.bytes.len() as u64,
2209 mode: buf.mode,
2210 })
2211 } else {
2212 pending
2213 .warm
2214 .get(&id)
2215 .map(|entry| PendingChildKind::WarmFile {
2216 size: entry.size,
2217 mode: entry.mode,
2218 })
2219 }
2220 });
2221 if let Some(kind) = indexed {
2222 pending.child_index.insert(new_path.clone(), kind);
2223 }
2224 }
2225 NodeKind::Symlink => {
2226 pending.child_index.remove(&old_path);
2227 pending.child_index.remove(&new_path);
2228 if let Some(size) = pending
2229 .symlinks
2230 .get(&new_path)
2231 .map(|target| target.len() as u64)
2232 {
2233 pending
2234 .child_index
2235 .insert(new_path.clone(), PendingChildKind::Symlink { size });
2236 }
2237 }
2238 }
2239 if let Some(dest_id) = displaced_dest {
2240 // T3: the displaced destination transitions to Orphan
2241 // iff it's currently `Live { open_count >= 1 }`. Bytes
2242 // (hot[dest_id], warm[dest_id]) stay put so the
2243 // surviving fd keeps reading the inode's own data
2244 // (spike doc §1.2 T3).
2245 //
2246 // Closes Codex PR #182 r11 finding 3293575541 (heddle
2247 // #209): `bp.transition_to_orphan(dest_id)` returns
2248 // `None` (without touching `state`) for any
2249 // non-`LiveNonZero` displaced destination, and the
2250 // missing `Witness<Orphan>` IS the short-circuit at
2251 // this call site. Pre-retrofit this branch
2252 // unconditionally inserted `Orphan { open_count: 0 }`
2253 // for non-`Live` destinations — including symlinks,
2254 // which have no `open`/`release` lifecycle and would
2255 // never reap the entry, growing `state` under symlink
2256 // churn until capture / invalidate.
2257 pending.with_brand(|bp| {
2258 let _ = bp.transition_to_orphan(dest_id);
2259 });
2260 }
2261 }
2262 Ok(())
2263 }
2264
2265 /// Rename a regular file. Under the post-spike unified
2266 /// NodeId-keyed model
2267 /// (`docs/design/mount-posix-semantics.md` §2.4), the source's
2268 /// bytes follow its NodeId — no byte migration step. The displaced
2269 /// destination keeps its own `hot[id]` / `warm[id]` so the
2270 /// surviving fd reads its own data. The work here is path-level:
2271 /// retire the destination's path-keyed hot binding, rebase the
2272 /// source's hot buffer's `path` field (so a subsequent `flush`
2273 /// promotes under the new path), seed warm if the source had only
2274 /// captured-tree bytes (so capture can plant the file at the new
2275 /// path), and tombstone the old path.
2276 ///
2277 /// `displaced_inode_id` is no longer used as a side-channel for
2278 /// byte preservation — the caller (`rename_entry_with_options`)
2279 /// handles the orphan state transition independently.
2280 fn move_file(
2281 &self,
2282 old_path: &Path,
2283 new_path: &Path,
2284 displaced_inode_id: Option<u64>,
2285 ) -> Result<()> {
2286 // Snapshot whether the source has a hot buffer (drain it to
2287 // warm so the warm tier becomes authoritative for capture
2288 // under the new path) and whether the source is captured-only
2289 // (then synthesize a warm entry so capture plants the bytes
2290 // at new_path).
2291 let src_id_opt = self
2292 .inner
2293 .pending
2294 .lock_or_poisoned()
2295 .hot_by_path
2296 .get(old_path)
2297 .copied();
2298 if let Some(id) = src_id_opt {
2299 self.flush_node(NodeId(id))?;
2300 }
2301 // After the flush, the source's bytes (if any) live in
2302 // `warm[src_id]`. If the source had no warm entry — captured
2303 // only — synthesize one keyed by the source's NodeId so
2304 // capture-time tree fold plants the file under new_path. We
2305 // resolve src_id via the path → inode reverse-index (or via
2306 // the captured-tree walk for a captured-only source).
2307 let src_id = {
2308 let inodes = self.inner.inodes.lock_or_poisoned();
2309 inodes.by_path.get(old_path).copied()
2310 };
2311 let needs_synth = match src_id {
2312 Some(id) => !self.inner.pending.lock_or_poisoned().warm.contains_key(&id),
2313 None => true,
2314 };
2315 let captured_seed = if needs_synth {
2316 // Captured-only source: pull (blob, mode, size) from the
2317 // captured tree so the rename survives `capture`.
2318 Some(self.captured_file_at(old_path)?)
2319 } else {
2320 None
2321 };
2322
2323 let mut pending = self.inner.pending.lock_or_poisoned();
2324 // Detach the destination's path-keyed hot binding. The
2325 // displaced inode's bytes are keyed by NodeId — they stay put
2326 // for the surviving fd. POSIX rename-over: open destination
2327 // descriptors keep referencing the displaced inode until close.
2328 pending.hot_by_path.remove(new_path);
2329 // Symlinks are path-keyed; clear at both endpoints.
2330 pending.symlinks.remove(new_path);
2331 pending.symlinks.remove(old_path);
2332 // Source: if a hot buffer survived the flush above (only
2333 // possible if the source was Orphan, which can't happen for
2334 // a valid rename source — but be defensive), rebase its
2335 // path-binding.
2336 if let Some(id) = pending.hot_by_path.remove(old_path) {
2337 if let Some(buf) = pending.hot.get_mut(&id) {
2338 buf.path = new_path.to_path_buf();
2339 buf.revision = buf.revision.wrapping_add(1);
2340 }
2341 pending.hot_by_path.insert(new_path.to_path_buf(), id);
2342 }
2343 // Captured-only source: synthesize a warm entry so capture
2344 // plants the bytes at new_path. The entry is keyed by the
2345 // source's NodeId; capture-time tree fold resolves its
2346 // current path through `inodes.by_path`.
2347 if let (Some(id), Some((blob, mode, size))) = (src_id, captured_seed) {
2348 pending.warm.insert(id, PendingEntry { blob, mode, size });
2349 }
2350 // Path-level bookkeeping: tombstone old_path so the captured
2351 // tree's old entry is hidden; clear any tombstone at
2352 // new_path (rename made it valid again).
2353 pending.tombstones.insert(old_path.to_path_buf());
2354 pending.tombstones.remove(new_path);
2355 // The displaced inode is handled by the caller via the
2356 // NodeState transition; no byte work here.
2357 let _ = displaced_inode_id;
2358 Ok(())
2359 }
2360
2361 fn move_symlink(
2362 &self,
2363 old_path: &Path,
2364 new_path: &Path,
2365 displaced_inode_id: Option<u64>,
2366 ) -> Result<()> {
2367 // Resolve target bytes from the pending overlay or the
2368 // captured-tree blob — symlinks are path-keyed (not openable
2369 // for IO; no orphan story applies).
2370 let target_bytes = {
2371 let pending = self.inner.pending.lock_or_poisoned();
2372 pending.symlinks.get(old_path).cloned()
2373 };
2374 let target_bytes = match target_bytes {
2375 Some(b) => b,
2376 None => {
2377 let blob = self.captured_symlink_at(old_path)?;
2378 (*self.load_blob_bytes(&blob)?).to_vec()
2379 }
2380 };
2381 let mut pending = self.inner.pending.lock_or_poisoned();
2382 // Detach the displaced destination's path-keyed hot binding.
2383 // Its NodeId-keyed bytes stay put for the surviving fd; the
2384 // caller's NodeState transition handles the orphan tracking.
2385 pending.hot_by_path.remove(new_path);
2386 pending.symlinks.remove(new_path);
2387 pending.symlinks.remove(old_path);
2388 pending
2389 .symlinks
2390 .insert(new_path.to_path_buf(), target_bytes);
2391 pending.tombstones.remove(new_path);
2392 pending.tombstones.insert(old_path.to_path_buf());
2393 let _ = displaced_inode_id;
2394 Ok(())
2395 }
2396
2397 fn move_overlay_dir(&self, old_path: &Path, new_path: &Path) -> Result<()> {
2398 // We only support overlay-only directory renames here. If the
2399 // source dir has any captured-tree backing, refuse — a full
2400 // captured-tree rename would need to rewrite every descendant
2401 // tombstone entry.
2402 if self.captured_dir_exists(old_path)? {
2403 return Err(MountError::InvalidArgument(format!(
2404 "cross-tree directory rename {} → {} not supported by the overlay",
2405 old_path.display(),
2406 new_path.display()
2407 )));
2408 }
2409 let mut pending = self.inner.pending.lock_or_poisoned();
2410 // Path-keyed structures under `old_path/` need to be rebased
2411 // to `new_path/`. Warm bytes follow the NodeId (unified shape)
2412 // so warm[id] is unaffected by this rewrite — descendant
2413 // NodeRecord paths get rebased in `rename_entry_with_options`.
2414 fn rebase(p: &Path, old: &Path, new: &Path) -> Option<PathBuf> {
2415 let tail = p.strip_prefix(old).ok()?;
2416 Some(new.join(tail))
2417 }
2418 let mut new_explicit: BTreeSet<PathBuf> = BTreeSet::new();
2419 let mut new_symlinks: BTreeMap<PathBuf, Vec<u8>> = BTreeMap::new();
2420 let mut new_tombstones: BTreeSet<PathBuf> = BTreeSet::new();
2421 let mut new_hot_by_path: BTreeMap<PathBuf, u64> = BTreeMap::new();
2422 let mut hot_path_updates: Vec<(u64, PathBuf)> = Vec::new();
2423 for explicit in std::mem::take(&mut pending.explicit_dirs) {
2424 match rebase(&explicit, old_path, new_path) {
2425 Some(rebased) => {
2426 new_explicit.insert(rebased);
2427 }
2428 None => {
2429 if explicit != old_path {
2430 new_explicit.insert(explicit);
2431 }
2432 }
2433 }
2434 }
2435 for (path, target) in std::mem::take(&mut pending.symlinks) {
2436 match rebase(&path, old_path, new_path) {
2437 Some(rebased) => {
2438 new_symlinks.insert(rebased, target);
2439 }
2440 None => {
2441 new_symlinks.insert(path, target);
2442 }
2443 }
2444 }
2445 for path in std::mem::take(&mut pending.tombstones) {
2446 match rebase(&path, old_path, new_path) {
2447 Some(rebased) => {
2448 new_tombstones.insert(rebased);
2449 }
2450 None => {
2451 new_tombstones.insert(path);
2452 }
2453 }
2454 }
2455 for (path, id) in std::mem::take(&mut pending.hot_by_path) {
2456 match rebase(&path, old_path, new_path) {
2457 Some(rebased) => {
2458 hot_path_updates.push((id, rebased.clone()));
2459 new_hot_by_path.insert(rebased, id);
2460 }
2461 None => {
2462 new_hot_by_path.insert(path, id);
2463 }
2464 }
2465 }
2466 // Rewrite hot-buffer path fields to match.
2467 for (id, new_p) in hot_path_updates {
2468 if let Some(buf) = pending.hot.get_mut(&id) {
2469 buf.path = new_p;
2470 buf.revision = buf.revision.wrapping_add(1);
2471 }
2472 }
2473 // Ensure the destination directory itself is registered.
2474 new_explicit.insert(new_path.to_path_buf());
2475 pending.explicit_dirs = new_explicit;
2476 pending.symlinks = new_symlinks;
2477 pending.tombstones = new_tombstones;
2478 pending.hot_by_path = new_hot_by_path;
2479 Ok(())
2480 }
2481
2482 /// Resolve a captured-tree file at `path`; returns its
2483 /// `(blob, mode, size)`. Errors with `NotFound` if no captured
2484 /// entry exists.
2485 fn captured_file_at(&self, path: &Path) -> Result<(ContentHash, FileMode, u64)> {
2486 let entry = self.captured_tree_entry(path)?;
2487 let Some(hash) = entry.blob_hash() else {
2488 return Err(MountError::InvalidArgument(format!(
2489 "{} is not a mutable file in the captured tree",
2490 path.display()
2491 )));
2492 };
2493 let mode = entry.mode();
2494 let size = self.blob_size(&hash)?;
2495 Ok((hash, mode, size))
2496 }
2497
2498 fn captured_symlink_at(&self, path: &Path) -> Result<ContentHash> {
2499 let entry = self.captured_tree_entry(path)?;
2500 let Some(hash) = entry.symlink_hash() else {
2501 return Err(MountError::InvalidArgument(format!(
2502 "{} is not a symlink in the captured tree",
2503 path.display()
2504 )));
2505 };
2506 Ok(hash)
2507 }
2508
2509 fn captured_tree_entry(&self, path: &Path) -> Result<TreeEntry> {
2510 let root_record = self.record_for(NodeId::ROOT)?;
2511 let mut tree = self.tree_for_record(&root_record)?;
2512 let comps: Vec<&str> = path
2513 .components()
2514 .filter_map(|c| match c {
2515 Component::Normal(n) => n.to_str(),
2516 _ => None,
2517 })
2518 .collect();
2519 let (leaf, dirs) = comps
2520 .split_last()
2521 .ok_or_else(|| MountError::NotFound(path.display().to_string()))?;
2522 for d in dirs {
2523 let e = tree
2524 .get(d)
2525 .ok_or_else(|| MountError::NotFound(path.display().to_string()))?;
2526 if !e.is_tree() {
2527 return Err(MountError::NotADirectory(d.to_string()));
2528 }
2529 let Some(hash) = e.tree_hash() else {
2530 return Err(MountError::NotADirectory(d.to_string()));
2531 };
2532 tree = self.load_tree(&hash)?;
2533 }
2534 let entry = tree
2535 .get(leaf)
2536 .cloned()
2537 .ok_or_else(|| MountError::NotFound(path.display().to_string()))?;
2538 Ok(entry)
2539 }
2540
2541 fn captured_dir_exists(&self, path: &Path) -> Result<bool> {
2542 match self.captured_tree_entry(path) {
2543 Ok(e) => Ok(e.is_tree()),
2544 Err(MountError::NotFound(_)) => Ok(false),
2545 Err(e) => Err(e),
2546 }
2547 }
2548
2549 /// Apply attribute updates from a FUSE `setattr` / FSKit
2550 /// `setattr` / etc. Returns post-update [`Attrs`] for an
2551 /// inline reply.
2552 pub fn set_attrs(&self, node: NodeId, update: AttrUpdate) -> Result<Attrs> {
2553 // Codex r13 thread 3293733165 (P1): every mutating branch of
2554 // `set_attrs` must serialize against `rename` / `create` /
2555 // `unlink` / `rmdir` under `write_mu`. Without it, a
2556 // `setattr(size=...)` racing with a `rename` re-uses the
2557 // pre-rename pathname in `apply_truncate`'s phase-2
2558 // bookkeeping — `tombstones.remove(old)` clears the rename's
2559 // tombstone and `hot_by_path.insert(old, node)` resurrects
2560 // the file at the old name. The mode-mutation branch has the
2561 // same shape (touches `hot_by_path[path]` / `warm[id]` derived
2562 // from `inodes.by_path[path]`), so we hold the lock for the
2563 // whole mutating prologue.
2564 let _write_guard = self.inner.write_mu.lock_or_poisoned();
2565
2566 // Mode mutation: only meaningful for file-kind records.
2567 if let Some(raw_mode) = update.mode {
2568 // Codex r13 thread 3293733164 (P2): the Normal↔Executable
2569 // fold is gated on the user execute bit (S_IXUSR = 0o100)
2570 // only, not on any of the three execute bits. A
2571 // `chmod 0o010` (group execute only) must leave the record
2572 // as Normal — otherwise capture would persist a
2573 // `FileMode::Executable` and grant owner+other execute
2574 // bits the agent never requested.
2575 let new_mode = if (raw_mode & 0o100) != 0 {
2576 FileMode::Executable
2577 } else {
2578 FileMode::Normal
2579 };
2580 let mut inodes = self.inner.inodes.lock_or_poisoned();
2581 if let Some(NodeRecord::File { mode, .. } | NodeRecord::PendingFile { mode, .. }) =
2582 inodes.by_id.get_mut(&node.0)
2583 {
2584 *mode = new_mode;
2585 }
2586 drop(inodes);
2587 // Reflect the mode in any open hot buffer + warm-tier
2588 // entry so a subsequent `capture` keeps the new mode.
2589 let record = self.record_for(node)?;
2590 if let Some(path) = match &record {
2591 NodeRecord::File { path, .. } | NodeRecord::PendingFile { path, .. } => Some(path),
2592 _ => None,
2593 } {
2594 let path = path.clone();
2595 let mut pending = self.inner.pending.lock_or_poisoned();
2596 // Always flip the per-NodeId buffer's mode — that's
2597 // the orphan's own bookkeeping when fd-based, and
2598 // the live buffer for non-orphan callers.
2599 if let Some(buf) = pending.hot.get_mut(&node.0) {
2600 buf.mode = new_mode;
2601 buf.revision = buf.revision.wrapping_add(1);
2602 }
2603 // Orphan branch: `unlink_entry` / `rename_entry`
2604 // recorded this NodeId because the kernel still
2605 // holds an fd to it, but the directory entry is
2606 // gone (or rebound to a sibling). POSIX is explicit:
2607 // an fd-based attribute change applies only to the
2608 // file referenced by that fd. Touching
2609 // `hot_by_path[path]` would mutate the fresh inode
2610 // now living at the same name; touching
2611 // `warm[path]` would land the change on the sibling
2612 // at capture time.
2613 if !pending.is_orphan(node.0) {
2614 if let Some(other_id) = pending.hot_by_path.get(&path).copied()
2615 && let Some(buf) = pending.hot.get_mut(&other_id)
2616 {
2617 buf.mode = new_mode;
2618 buf.revision = buf.revision.wrapping_add(1);
2619 }
2620 // Warm is NodeId-keyed: rebind via inodes if the
2621 // path still resolves Live to a tracked NodeId.
2622 let warm_id = {
2623 let inodes = self.inner.inodes.lock_or_poisoned();
2624 inodes.by_path.get(&path).copied()
2625 };
2626 if let Some(id) = warm_id
2627 && let Some(entry) = pending.warm.get_mut(&id)
2628 {
2629 entry.mode = new_mode;
2630 }
2631 pending.child_index.update_file_mode(&path, new_mode);
2632 }
2633 }
2634 }
2635
2636 // Size mutation: O_TRUNC, ftruncate, etc.
2637 if let Some(new_size) = update.size {
2638 self.apply_truncate(node, new_size)?;
2639 }
2640 // uid/gid/mtime: accepted as no-ops. The overlay doesn't carry
2641 // per-node ownership / timestamps yet (capture re-derives both
2642 // from the agent's principal + mount mtime).
2643 self.attrs(node)
2644 }
2645
2646 fn apply_truncate(&self, node: NodeId, new_size: u64) -> Result<()> {
2647 let new_size = validate_truncate_size(new_size)?;
2648 let record = self.record_for(node)?;
2649 let (path, mode, captured_blob) = match &record {
2650 NodeRecord::File {
2651 path, mode, blob, ..
2652 } => (path.clone(), *mode, Some(*blob)),
2653 NodeRecord::PendingFile { path, mode } => (path.clone(), *mode, None),
2654 _ => {
2655 return Err(MountError::IsADirectory(format!(
2656 "setattr(size) on non-file {record:?}"
2657 )));
2658 }
2659 };
2660
2661 // Phase 1: under the lock, decide whether a buffer already
2662 // exists (resize in place), and otherwise record orphan-ness
2663 // + the seed source. Drop the lock for the CAS read.
2664 //
2665 // POSIX `ftruncate` on an open-unlinked / rename-displaced fd
2666 // (an orphan in our terminology) must touch only the
2667 // anonymous open inode. The orphan branch never resizes a
2668 // sibling buffer at the rebased path, never seeds from
2669 // `warm[path]` (now owned by the sibling), and in Phase 2
2670 // never republishes `hot_by_path[path]` nor clears the
2671 // tombstone.
2672 enum Phase1 {
2673 ResizedInPlace,
2674 NeedSeed {
2675 orphan: bool,
2676 seed: Option<ContentHash>,
2677 },
2678 }
2679 let phase1 = {
2680 // Resolve the path's current Live NodeId via the inode
2681 // registry — under the unified shape `warm` is
2682 // NodeId-keyed, and the Live owner of `path` is the
2683 // sibling we'd seed from when no per-inode buffer exists.
2684 let path_owner = {
2685 let inodes = self.inner.inodes.lock_or_poisoned();
2686 inodes.by_path.get(&path).copied()
2687 };
2688 let mut pending = self.inner.pending.lock_or_poisoned();
2689 let orphan = pending.is_orphan(node.0);
2690 let id = if pending.hot.contains_key(&node.0) {
2691 Some(node.0)
2692 } else if orphan {
2693 // Never resize a sibling buffer through the orphan
2694 // fd — that buffer belongs to a fresh inode at the
2695 // rebound name.
2696 None
2697 } else {
2698 pending.hot_by_path.get(&path).copied()
2699 };
2700 if let Some(id) = id
2701 && let Some(buf) = pending.hot.get_mut(&id)
2702 {
2703 buf.bytes.resize(new_size, 0);
2704 buf.last_touched = Instant::now();
2705 buf.revision = buf.revision.wrapping_add(1);
2706 let mode = buf.mode;
2707 if !orphan {
2708 pending.child_index.insert(
2709 path.clone(),
2710 PendingChildKind::HotFile {
2711 node: NodeId(id),
2712 size: new_size as u64,
2713 mode,
2714 },
2715 );
2716 }
2717 Phase1::ResizedInPlace
2718 } else {
2719 let seed = if orphan {
2720 // Orphan: only the inode's pre-displacement
2721 // content is valid. Under the unified shape its
2722 // own warm bytes live at `warm[node.0]`; fall
2723 // back to the captured blob (this inode's own,
2724 // not the sibling at the rebound name).
2725 pending.warm.get(&node.0).map(|e| e.blob).or(captured_blob)
2726 } else {
2727 // Live: the path's bytes live at `warm[id]` where
2728 // id is the Live owner via `inodes.by_path`.
2729 path_owner
2730 .and_then(|id| pending.warm.get(&id).map(|e| e.blob))
2731 .or(captured_blob)
2732 };
2733 Phase1::NeedSeed { orphan, seed }
2734 }
2735 };
2736 let (orphan, seed_blob) = match phase1 {
2737 Phase1::ResizedInPlace => return Ok(()),
2738 Phase1::NeedSeed { orphan, seed } => (orphan, seed),
2739 };
2740
2741 let mut bytes = match seed_blob {
2742 Some(hash) => (*self.load_blob_bytes(&hash)?).to_vec(),
2743 None => Vec::new(),
2744 };
2745 bytes.resize(new_size, 0);
2746 let mut pending = self.inner.pending.lock_or_poisoned();
2747 if orphan {
2748 // Per-NodeId buffer only. Skip the tombstone-clear and
2749 // the `hot_by_path` rebind — the directory entry must
2750 // stay gone (open-unlinked) or stay rebound to the
2751 // sibling (rename-over). The companion orphan branch in
2752 // `flush_node` drops this buffer on release without
2753 // warm-promoting it.
2754 pending.hot.insert(
2755 node.0,
2756 HotBuffer {
2757 path,
2758 mode,
2759 bytes,
2760 last_touched: Instant::now(),
2761 revision: 0,
2762 },
2763 );
2764 } else {
2765 pending.tombstones.remove(&path);
2766 pending.hot.insert(
2767 node.0,
2768 HotBuffer {
2769 path: path.clone(),
2770 mode,
2771 bytes,
2772 last_touched: Instant::now(),
2773 revision: 0,
2774 },
2775 );
2776 pending.hot_by_path.insert(path.clone(), node.0);
2777 pending.child_index.insert(
2778 path,
2779 PendingChildKind::HotFile {
2780 node,
2781 size: new_size as u64,
2782 mode,
2783 },
2784 );
2785 }
2786 Ok(())
2787 }
2788
2789 /// Create a symbolic link under `parent`. Target bytes are kept
2790 /// in the pending tier verbatim; `capture` writes them as a CAS
2791 /// blob and emits a `Symlink` tree entry.
2792 pub fn create_symlink(&self, parent: NodeId, name: &OsStr, target: &Path) -> Result<Entry> {
2793 // R8: serialize with other write-side mutations.
2794 let _write_guard = self.inner.write_mu.lock_or_poisoned();
2795 let name_str = validate_entry_name(name)?;
2796 if self.lookup(parent, name)?.is_some() {
2797 return Err(MountError::AlreadyExists(name_str.to_string()));
2798 }
2799 let parent_record = self.record_for(parent)?;
2800 let parent_path = self
2801 .dir_path_of(&parent_record)
2802 .ok_or_else(|| MountError::NotADirectory(format!("{parent_record:?}")))?;
2803 let child_path = join_child(&parent_path, name_str);
2804 let target_bytes = target.as_os_str().as_encoded_bytes().to_vec();
2805 let target_len = target_bytes.len() as u64;
2806
2807 {
2808 let mut pending = self.inner.pending.lock_or_poisoned();
2809 pending.tombstones.remove(&child_path);
2810 pending.symlinks.insert(child_path.clone(), target_bytes);
2811 pending.child_index.insert(
2812 child_path.clone(),
2813 PendingChildKind::Symlink { size: target_len },
2814 );
2815 }
2816 let node = self.intern(NodeRecord::PendingSymlink { path: child_path });
2817 Ok(Entry {
2818 node,
2819 name: name.to_os_string(),
2820 kind: NodeKind::Symlink,
2821 size: target_len,
2822 unix_mode: FileMode::Symlink.to_unix_mode(),
2823 })
2824 }
2825
2826 /// Read the target of a symlink `node`. Works for both overlay
2827 /// (`PendingSymlink`) and captured (`Symlink`) records.
2828 ///
2829 /// Codex r12 thread 3293510316 (P1): the prior implementation
2830 /// used `OsStr::from_encoded_bytes_unchecked` on bytes loaded
2831 /// from the object store, which is unsound — that API's safety
2832 /// contract requires bytes minted by `OsStr::as_encoded_bytes`
2833 /// in *this* process and Rust version, but captured-tree blobs
2834 /// can come from any process and version. The corrected path
2835 /// delegates to [`symlink_target_from_bytes`], which uses
2836 /// platform-safe APIs (`OsStrExt::from_bytes` on Unix, UTF-8
2837 /// validation on Windows).
2838 pub fn read_link(&self, node: NodeId) -> Result<OsString> {
2839 let record = self.record_for(node)?;
2840 match record {
2841 NodeRecord::PendingSymlink { path } => {
2842 let pending = self.inner.pending.lock_or_poisoned();
2843 let bytes = pending
2844 .symlinks
2845 .get(&path)
2846 .ok_or_else(|| MountError::Stale(format!("symlink {}", path.display())))?;
2847 symlink_target_from_bytes(bytes)
2848 }
2849 NodeRecord::Symlink { blob } => {
2850 let bytes = self.load_blob_bytes(&blob)?;
2851 symlink_target_from_bytes(&bytes)
2852 }
2853 other => Err(MountError::InvalidArgument(format!(
2854 "read_link on non-symlink record: {other:?}"
2855 ))),
2856 }
2857 }
2858
2859 /// Flush all hot buffers to CAS. Useful at the start of `capture`
2860 /// or when tests want a deterministic warm state.
2861 pub fn flush_all(&self) -> Result<()> {
2862 let ids: Vec<u64> = self
2863 .inner
2864 .pending
2865 .lock_or_poisoned()
2866 .hot
2867 .keys()
2868 .copied()
2869 .collect();
2870 for id in ids {
2871 self.flush_node(NodeId(id))?;
2872 }
2873 Ok(())
2874 }
2875
2876 /// Look up a path in the pending tier. Order: hot buffer (in-flight
2877 /// writes), then warm tier (promoted blob), then None (caller must
2878 /// fall back to the immutable state's tree).
2879 ///
2880 /// Under the unified NodeId-keyed model warm bytes live at
2881 /// `warm[id]`; the path → id resolution goes through
2882 /// `inodes.by_path` (lock order: pending ⊐ inodes).
2883 fn pending_lookup(&self, path: &Path) -> Option<PendingHit> {
2884 let pending = self.inner.pending.lock_or_poisoned();
2885 if pending.tombstones.contains(path) {
2886 return Some(PendingHit::Tombstone);
2887 }
2888 if let Some(target) = pending.symlinks.get(path) {
2889 return Some(PendingHit::Symlink {
2890 target_len: target.len() as u64,
2891 });
2892 }
2893 if let Some(node_id) = pending.hot_by_path.get(path)
2894 && let Some(buf) = pending.hot.get(node_id)
2895 {
2896 return Some(PendingHit::Hot {
2897 node: NodeId(*node_id),
2898 size: buf.bytes.len() as u64,
2899 mode: buf.mode,
2900 });
2901 }
2902 // Warm needs path → NodeId resolution. Acquire inodes inside
2903 // the pending lock (lock order: pending ⊐ inodes).
2904 let inodes = self.inner.inodes.lock_or_poisoned();
2905 let id = *inodes.by_path.get(path)?;
2906 let entry = pending.warm.get(&id)?;
2907 Some(PendingHit::Warm {
2908 blob: entry.blob,
2909 size: entry.size,
2910 mode: entry.mode,
2911 })
2912 }
2913
2914 /// True if the parent dir or any ancestor of `path` has been
2915 /// `rmdir`'d through the mount. Used by lookup/enumerate so the
2916 /// kernel never sees stale captured children of a directory the
2917 /// agent removed.
2918 fn ancestor_is_dir_tombstoned(&self, pending: &Pending, path: &Path) -> bool {
2919 let mut cursor = path.parent();
2920 while let Some(p) = cursor {
2921 if p.as_os_str().is_empty() {
2922 break;
2923 }
2924 if pending.dir_tombstones.contains(p) {
2925 return true;
2926 }
2927 cursor = p.parent();
2928 }
2929 false
2930 }
2931
2932 /// Does any pending entry sit *under* `dir` as a strict prefix?
2933 /// I.e. has an agent created `dir/something` even though `dir`
2934 /// itself isn't in the captured tree yet? An explicit `mkdir dir`
2935 /// also counts (so an empty mkdir survives without children).
2936 fn pending_dir_exists(&self, dir: &Path) -> bool {
2937 if dir.as_os_str().is_empty() {
2938 return false;
2939 }
2940 let pending = self.inner.pending.lock_or_poisoned();
2941 pending.child_index.dir_exists(dir)
2942 }
2943
2944 /// Direct children of `dir` that exist purely in the pending
2945 /// tier (created/written by the mount, not in the captured tree).
2946 /// Returns each immediate child as either a file (with hot or
2947 /// warm metadata) or an implicit directory (because some pending
2948 /// path is *under* this dir, e.g. `src/foo.rs` makes `src` an
2949 /// implicit dir of root). Tombstones suppress paths.
2950 fn pending_children_at(&self, dir: &Path) -> Vec<(String, PendingChildKind)> {
2951 let pending = self.inner.pending.lock_or_poisoned();
2952 pending.child_index.children_at(dir)
2953 }
2954
2955 /// Pre-index implementation retained as a test-only differential
2956 /// oracle and negative control. `work` counts pending entries
2957 /// inspected, making the former O(all pending) behavior explicit.
2958 #[cfg(test)]
2959 fn pending_dir_exists_full_scan(&self, dir: &Path) -> (bool, usize) {
2960 if dir.as_os_str().is_empty() {
2961 return (false, 0);
2962 }
2963 let pending = self.inner.pending.lock_or_poisoned();
2964 let mut work = 1;
2965 if pending.explicit_dirs.contains(dir) {
2966 return (true, work);
2967 }
2968 let probe = |path: &Path| {
2969 path.strip_prefix(dir)
2970 .ok()
2971 .and_then(|tail| tail.components().next())
2972 .is_some()
2973 };
2974 let inodes = self.inner.inodes.lock_or_poisoned();
2975 for id in pending.warm.keys() {
2976 work += 1;
2977 if pending.is_orphan(*id) {
2978 continue;
2979 }
2980 if let Some(path) = inodes.by_id.get(id).and_then(warm_path_of_record)
2981 && !pending.tombstones.contains(path)
2982 && probe(path)
2983 {
2984 return (true, work);
2985 }
2986 }
2987 drop(inodes);
2988 for path in pending.hot_by_path.keys() {
2989 work += 1;
2990 if !pending.tombstones.contains(path) && probe(path) {
2991 return (true, work);
2992 }
2993 }
2994 for path in pending.symlinks.keys() {
2995 work += 1;
2996 if probe(path) {
2997 return (true, work);
2998 }
2999 }
3000 (false, work)
3001 }
3002
3003 #[cfg(test)]
3004 fn pending_children_at_full_scan(
3005 &self,
3006 dir: &Path,
3007 ) -> (Vec<(String, PendingChildKind)>, usize) {
3008 let pending = self.inner.pending.lock_or_poisoned();
3009 let mut out: BTreeMap<String, PendingChildKind> = BTreeMap::new();
3010 let mut work = 0;
3011 let project = |path: &Path| -> Option<(String, bool)> {
3012 let suffix = if dir.as_os_str().is_empty() {
3013 Some(path)
3014 } else {
3015 path.strip_prefix(dir).ok()
3016 }?;
3017 let mut comps = suffix.components();
3018 let name = match comps.next()? {
3019 Component::Normal(name) => name.to_str()?.to_string(),
3020 _ => return None,
3021 };
3022 Some((name, comps.next().is_some()))
3023 };
3024
3025 for (path, node_id) in &pending.hot_by_path {
3026 work += 1;
3027 if pending.tombstones.contains(path) {
3028 continue;
3029 }
3030 let Some((name, is_dir)) = project(path) else {
3031 continue;
3032 };
3033 if is_dir {
3034 out.entry(name).or_insert(PendingChildKind::Dir);
3035 } else if let Some(buf) = pending.hot.get(node_id) {
3036 out.insert(
3037 name,
3038 PendingChildKind::HotFile {
3039 node: NodeId(*node_id),
3040 size: buf.bytes.len() as u64,
3041 mode: buf.mode,
3042 },
3043 );
3044 }
3045 }
3046
3047 let inodes = self.inner.inodes.lock_or_poisoned();
3048 for (id, entry) in &pending.warm {
3049 work += 1;
3050 if pending.is_orphan(*id) {
3051 continue;
3052 }
3053 let Some(path) = inodes.by_id.get(id).and_then(warm_path_of_record) else {
3054 continue;
3055 };
3056 if pending.tombstones.contains(path) {
3057 continue;
3058 }
3059 let Some((name, is_dir)) = project(path) else {
3060 continue;
3061 };
3062 if is_dir {
3063 out.entry(name).or_insert(PendingChildKind::Dir);
3064 } else {
3065 out.entry(name).or_insert(PendingChildKind::WarmFile {
3066 size: entry.size,
3067 mode: entry.mode,
3068 });
3069 }
3070 }
3071 drop(inodes);
3072
3073 for (path, target) in &pending.symlinks {
3074 work += 1;
3075 let Some((name, is_dir)) = project(path) else {
3076 continue;
3077 };
3078 if is_dir {
3079 out.entry(name).or_insert(PendingChildKind::Dir);
3080 } else {
3081 out.entry(name).or_insert(PendingChildKind::Symlink {
3082 size: target.len() as u64,
3083 });
3084 }
3085 }
3086 for path in &pending.explicit_dirs {
3087 work += 1;
3088 if let Some((name, _)) = project(path) {
3089 out.entry(name).or_insert(PendingChildKind::Dir);
3090 }
3091 }
3092 (out.into_iter().collect(), work)
3093 }
3094
3095 #[cfg(test)]
3096 pub(crate) fn pending_index_matches_full_scan(&self) -> std::result::Result<(), String> {
3097 let dirs = {
3098 let pending = self.inner.pending.lock_or_poisoned();
3099 let inodes = self.inner.inodes.lock_or_poisoned();
3100 let mut paths: Vec<PathBuf> = pending.hot_by_path.keys().cloned().collect();
3101 paths.extend(pending.symlinks.keys().cloned());
3102 paths.extend(pending.explicit_dirs.iter().cloned());
3103 paths.extend(
3104 pending
3105 .warm
3106 .keys()
3107 .filter(|id| !pending.is_orphan(**id))
3108 .filter_map(|id| inodes.by_id.get(id).and_then(warm_path_of_record))
3109 .map(Path::to_path_buf),
3110 );
3111 let mut dirs = BTreeSet::from([PathBuf::new()]);
3112 for path in paths {
3113 let mut cursor = path.parent();
3114 while let Some(dir) = cursor {
3115 dirs.insert(dir.to_path_buf());
3116 cursor = dir.parent();
3117 }
3118 if pending.explicit_dirs.contains(&path) {
3119 dirs.insert(path);
3120 }
3121 }
3122 dirs
3123 };
3124
3125 for dir in dirs {
3126 let indexed_exists = self.pending_dir_exists(&dir);
3127 let (scanned_exists, _) = self.pending_dir_exists_full_scan(&dir);
3128 if indexed_exists != scanned_exists {
3129 return Err(format!(
3130 "exists mismatch at {}: indexed={indexed_exists}, scan={scanned_exists}",
3131 dir.display()
3132 ));
3133 }
3134 let indexed_children = self.pending_children_at(&dir);
3135 let (scanned_children, _) = self.pending_children_at_full_scan(&dir);
3136 if indexed_children != scanned_children {
3137 return Err(format!(
3138 "children mismatch at {}: indexed={indexed_children:?}, scan={scanned_children:?}",
3139 dir.display()
3140 ));
3141 }
3142 }
3143 Ok(())
3144 }
3145
3146 #[cfg(test)]
3147 pub(crate) fn pending_index_work(&self, dir: &Path) -> (usize, usize, usize, usize) {
3148 let indexed = self.inner.pending.lock_or_poisoned();
3149 let indexed_exists_work = usize::from(!dir.as_os_str().is_empty());
3150 let indexed_children_work = indexed
3151 .child_index
3152 .by_dir
3153 .get(dir)
3154 .map_or(0, |children| children.names.len());
3155 drop(indexed);
3156 let (_, scan_exists_work) = self.pending_dir_exists_full_scan(dir);
3157 let (_, scan_children_work) = self.pending_children_at_full_scan(dir);
3158 (
3159 indexed_exists_work,
3160 indexed_children_work,
3161 scan_exists_work,
3162 scan_children_work,
3163 )
3164 }
3165
3166 #[cfg(test)]
3167 pub(crate) fn time_pending_index(&self, dir: &Path, iterations: usize) -> Duration {
3168 let start = Instant::now();
3169 for _ in 0..iterations {
3170 std::hint::black_box(self.pending_dir_exists(dir));
3171 std::hint::black_box(self.pending_children_at(dir));
3172 }
3173 start.elapsed()
3174 }
3175}
3176
3177/// Reject FUSE entry names that wouldn't survive a `TreeEntry`'s
3178/// validator. Delegates to [`objects::object::validate_tree_entry_name`]
3179/// so the mount's write-side reject set stays in lockstep with the
3180/// tree serializer's — Codex r13 thread 3293733163 (P2) caught the
3181/// drift where the overlay accepted backslash and control bytes that
3182/// the serializer later rejected at capture with a confusing
3183/// "invalid object" error. The NUL pre-check is here (not in the
3184/// shared validator) because `OsStr` on Unix can carry interior NUL
3185/// bytes that `to_str()` would otherwise round-trip through to the
3186/// validator as an unmarked control byte; we surface a more specific
3187/// error.
3188fn validate_entry_name(name: &OsStr) -> Result<&str> {
3189 let bytes = name.as_encoded_bytes();
3190 if bytes.contains(&0) {
3191 return Err(MountError::InvalidArgument(format!(
3192 "entry name {name:?} contains NUL"
3193 )));
3194 }
3195 let name_str = name.to_str().ok_or_else(|| {
3196 MountError::InvalidArgument(format!("entry name {name:?} is not valid UTF-8"))
3197 })?;
3198 objects::object::validate_tree_entry_name(name_str)
3199 .map_err(|e| MountError::InvalidArgument(e.to_string()))?;
3200 Ok(name_str)
3201}
3202
3203/// Mount-relative path for a warm-tier entry, derived from its
3204/// [`NodeRecord`]. The NodeId-keyed warm tier doesn't store the path
3205/// directly; capture-time tree fold / `pending_dir_exists` /
3206/// `pending_children_at` resolve it via the inode registry. Only
3207/// file-like records (`File`, `PendingFile`) carry warm bytes; the
3208/// other variants return `None`.
3209#[cfg(test)]
3210fn warm_path_of_record(record: &NodeRecord) -> Option<&Path> {
3211 match record {
3212 NodeRecord::File { path, .. } | NodeRecord::PendingFile { path, .. } => Some(path),
3213 _ => None,
3214 }
3215}
3216
3217/// Decode symlink target bytes back into an `OsString`. The Unix
3218/// branch uses `OsStrExt::from_bytes`, which is sound for any byte
3219/// sequence (the inverse of `OsStrExt::as_bytes`). The Windows branch
3220/// validates as UTF-8 and returns [`MountError::InvalidArgument`]
3221/// otherwise — `OsStr` on Windows is a process-internal encoding
3222/// (WTF-8 today, but not promised), so accepting arbitrary captured
3223/// bytes is unsound. Replaces a prior
3224/// `unsafe { OsStr::from_encoded_bytes_unchecked(bytes) }` call site
3225/// (Codex r12 thread 3293510316).
3226fn symlink_target_from_bytes(bytes: &[u8]) -> Result<OsString> {
3227 #[cfg(unix)]
3228 {
3229 use std::os::unix::ffi::OsStrExt;
3230 Ok(OsStr::from_bytes(bytes).to_os_string())
3231 }
3232 #[cfg(not(unix))]
3233 {
3234 match std::str::from_utf8(bytes) {
3235 Ok(s) => Ok(OsString::from(s)),
3236 Err(_) => Err(MountError::InvalidArgument(
3237 "captured symlink target bytes are not valid UTF-8".into(),
3238 )),
3239 }
3240 }
3241}
3242
3243/// Join a parent mount-relative path with a leaf name. Mirrors the
3244/// shape every write-side op uses, so the construction stays
3245/// consistent across the file.
3246#[inline]
3247fn join_child(parent: &Path, name: &str) -> PathBuf {
3248 if parent.as_os_str().is_empty() {
3249 PathBuf::from(name)
3250 } else {
3251 parent.join(name)
3252 }
3253}
3254
3255/// Copy `[offset, offset+buf.len())` from `src` into `buf`, returning
3256/// the number of bytes actually copied (0 when `offset` is past EOF,
3257/// or `min(buf.len(), src.len() - offset)` otherwise). Pulled out so
3258/// the `read` hot path is a single slice copy rather than a Vec
3259/// allocation per call.
3260#[inline]
3261fn copy_into(src: &[u8], offset: u64, buf: &mut [u8]) -> usize {
3262 let offset = offset as usize;
3263 if offset >= src.len() {
3264 return 0;
3265 }
3266 let take = std::cmp::min(buf.len(), src.len() - offset);
3267 buf[..take].copy_from_slice(&src[offset..offset + take]);
3268 take
3269}
3270
3271/// Pending-tier overlay for a captured-tree path. Consumed by `read`
3272/// to decide whether to serve the captured blob (`None` returned by
3273/// the lookup) or the pending overlay's bytes.
3274enum Overlay {
3275 /// Promoted warm-tier blob. Same path now points at this blob in
3276 /// the pending tier; the captured `File` record's blob is
3277 /// effectively stale until capture folds the warm tier in.
3278 Warm(ContentHash),
3279 /// Tombstoned through the mount. The kernel will get a stale
3280 /// inode reply; subsequent dentry refresh resolves the entry as
3281 /// gone.
3282 Gone,
3283}
3284
3285/// What `pending_lookup` found at a given path.
3286#[allow(dead_code)] // `blob` reserved for cross-mount dedup callers.
3287enum PendingHit {
3288 Hot {
3289 node: NodeId,
3290 size: u64,
3291 mode: FileMode,
3292 },
3293 Warm {
3294 blob: ContentHash,
3295 size: u64,
3296 mode: FileMode,
3297 },
3298 Symlink {
3299 target_len: u64,
3300 },
3301 Tombstone,
3302}
3303
3304impl<R: RefBackend, O: OpLogBackend, S: ObjectStore> MountInner<R, O, S> {
3305 /// Drain any hot buffer whose `last_touched` is older than
3306 /// `idle_after`. Mirrors `ContentAddressedMount::promote_idle_buffers`
3307 /// but is callable from the worker thread which only holds a
3308 /// `Weak<MountInner>`.
3309 fn sweep_idle_buffers(&self) -> Result<()> {
3310 let now = Instant::now();
3311 let idle_after = self.promotion.read_or_poisoned().idle_after;
3312 let to_promote: Vec<u64> = {
3313 let pending = self.pending.lock_or_poisoned();
3314 pending
3315 .hot
3316 .iter()
3317 .filter(|(_, buf)| now.saturating_duration_since(buf.last_touched) >= idle_after)
3318 .map(|(id, _)| *id)
3319 .collect()
3320 };
3321 for id in to_promote {
3322 let _ = self.flush_node(NodeId(id));
3323 }
3324 Ok(())
3325 }
3326
3327 /// Promote a single hot buffer to CAS. Inner-side flush so the
3328 /// sweep worker can drain idle buffers without bouncing back
3329 /// through `ContentAddressedMount`.
3330 ///
3331 /// Lifecycle note (R8 — Codex Thread 3293235165): FUSE `flush`
3332 /// fires on every descriptor close including each close of a
3333 /// `dup`-derived fd. For an orphaned node we must NOT touch the
3334 /// orphan marker here and must NOT drop the hot buffer (surviving
3335 /// fds need both). Only [`Self::release_node`] — invoked on the
3336 /// last-close-per-FUSE-open — clears the marker.
3337 fn flush_node(&self, node: NodeId) -> Result<()> {
3338 let (path, mode, bytes, revision) = {
3339 let pending = self.pending.lock_or_poisoned();
3340 // Orphan: keep the buffer alive across `flush` events.
3341 // POSIX open-unlinked semantics: bytes persist for the
3342 // surviving fds; the state survives so subsequent writes
3343 // through those fds keep taking the orphan branch (no
3344 // path republish, no warm promotion). The final clear
3345 // happens in `release_node`.
3346 if pending.is_orphan(node.0) {
3347 return Ok(());
3348 }
3349 let Some(buf) = pending.hot.get(&node.0) else {
3350 return Ok(());
3351 };
3352 // Keep the hot buffer published while CAS I/O runs. The
3353 // namespace index and `pending_lookup` therefore continue
3354 // to agree throughout promotion instead of exposing a
3355 // transient false-negative window.
3356 (buf.path.clone(), buf.mode, buf.bytes.clone(), buf.revision)
3357 };
3358 let size = bytes.len() as u64;
3359 let blob = Blob::new(bytes);
3360 let blob_oid = self
3361 .repo
3362 .store()
3363 .put_blob(&blob)
3364 .map_err(MountError::Store)?;
3365 debug!(?path, %blob_oid, size, "promoted hot buffer to CAS");
3366 let mut pending = self.pending.lock_or_poisoned();
3367 if pending.is_orphan(node.0) {
3368 return Ok(());
3369 }
3370 let Some(current) = pending.hot.get(&node.0) else {
3371 return Ok(());
3372 };
3373 let unchanged =
3374 current.revision == revision && current.path == path && current.mode == mode;
3375 // Warm is NodeId-keyed. The path-keyed tombstone clear below
3376 // is a separate concern (directory-entry level).
3377 pending.warm.insert(
3378 node.0,
3379 PendingEntry {
3380 blob: blob_oid,
3381 mode,
3382 size,
3383 },
3384 );
3385 if !unchanged {
3386 return Ok(());
3387 }
3388
3389 pending.hot.remove(&node.0);
3390 // Only retract the path mapping if it still points at us; an
3391 // unlink-then-recreate during CAS I/O may have rebound the
3392 // same path to a fresh inode.
3393 let owns_path = pending.hot_by_path.get(&path) == Some(&node.0);
3394 if owns_path {
3395 pending.hot_by_path.remove(&path);
3396 pending
3397 .child_index
3398 .insert(path.clone(), PendingChildKind::WarmFile { size, mode });
3399 // Promotion supersedes any prior tombstone only while this
3400 // inode still owns the directory entry.
3401 pending.tombstones.remove(&path);
3402 }
3403 Ok(())
3404 }
3405
3406 /// Final close of `node` from a FUSE `release` callback. Drives
3407 /// the per-NodeId lifecycle: decrement the open count carried on
3408 /// `state[node]`; on the final close of an Orphan, drop bytes
3409 /// and remove the state entry; on the final close of a Live
3410 /// node, promote any hot buffer to warm via `flush_node`. A
3411 /// release for an untracked but real node is a safe no-op; a
3412 /// release for an unknown NodeId is rejected with `NotFound`.
3413 ///
3414 /// The orphan branch never warm-promotes — an orphan's bytes are
3415 /// unreachable by path post-T1/T3, so promoting them would leak
3416 /// data into the captured tree at a now-tombstoned path.
3417 fn release_node(&self, node: NodeId) -> Result<()> {
3418 // Action determined under the lock so we don't re-read state
3419 // after dropping bytes.
3420 enum Outcome {
3421 /// Mid-life (non-final) close OR final close of a Live
3422 /// node. Either way: forward to `flush_node`. (`flush_node`
3423 /// is a no-op for Orphan, so the mid-life Orphan case is
3424 /// also safe to forward.)
3425 Flush,
3426 /// Final close of an Orphan. Dirty hot bytes must still
3427 /// cross the durability boundary before the anonymous
3428 /// inode is retired, but they must not be warm-promoted
3429 /// into the path-indexed pending tree.
3430 OrphanFinal { hot: Option<(PathBuf, Vec<u8>)> },
3431 /// No lifecycle state and no hot buffer. We validate
3432 /// outside the pending lock so double-release of a real
3433 /// inode is a no-op, while a bogus NodeId is rejected.
3434 MaybeUntrackedNoop,
3435 }
3436 let outcome = {
3437 let mut pending = self.pending.lock_or_poisoned();
3438 match pending.state.get(&node.0).copied() {
3439 None => {
3440 // Untracked release (no on_open was ever
3441 // recorded). Treat a tracked hot buffer as Live
3442 // final-close; otherwise validate the NodeId
3443 // outside this lock.
3444 if pending.hot.contains_key(&node.0) {
3445 Outcome::Flush
3446 } else {
3447 Outcome::MaybeUntrackedNoop
3448 }
3449 }
3450 Some(NodeState::Live { open_count }) => {
3451 let n = open_count.saturating_sub(1);
3452 if n == 0 {
3453 pending.state.remove(&node.0);
3454 } else {
3455 pending
3456 .state
3457 .insert(node.0, NodeState::Live { open_count: n });
3458 }
3459 Outcome::Flush
3460 }
3461 Some(NodeState::Orphan { open_count }) => {
3462 let n = open_count.saturating_sub(1);
3463 if n == 0 {
3464 // Final release of an Orphan — POSIX "inode
3465 // lives until last close" ends here. Snapshot
3466 // hot bytes so they can be persisted outside
3467 // the lock before retiring the anonymous state.
3468 let hot = pending
3469 .hot
3470 .get(&node.0)
3471 .map(|buf| (buf.path.clone(), buf.bytes.clone()));
3472 Outcome::OrphanFinal { hot }
3473 } else {
3474 pending
3475 .state
3476 .insert(node.0, NodeState::Orphan { open_count: n });
3477 // Mid-life Orphan release: forward to
3478 // flush_node (which no-ops for orphans).
3479 Outcome::Flush
3480 }
3481 }
3482 }
3483 };
3484 match outcome {
3485 Outcome::Flush => self.flush_node(node),
3486 Outcome::MaybeUntrackedNoop => {
3487 if !self.inodes.lock_or_poisoned().by_id.contains_key(&node.0) {
3488 return Err(MountError::NotFound(format!("node {}", node.0)));
3489 }
3490 Ok(())
3491 }
3492 Outcome::OrphanFinal { hot } => {
3493 if let Some((path, bytes)) = hot {
3494 let size = bytes.len() as u64;
3495 let blob = Blob::new(bytes);
3496 let blob_oid = self
3497 .repo
3498 .store()
3499 .put_blob(&blob)
3500 .map_err(MountError::Store)?;
3501 debug!(?path, %blob_oid, size, "persisted orphan hot buffer to CAS");
3502 }
3503 let mut pending = self.pending.lock_or_poisoned();
3504 pending.state.remove(&node.0);
3505 pending.hot.remove(&node.0);
3506 pending.warm.remove(&node.0);
3507 Ok(())
3508 }
3509 }
3510 }
3511}
3512
3513/// Spawn the safety-sweep worker, if one is requested by the
3514/// inner's promotion policy. The worker holds a `Weak<MountInner>`
3515/// so the mount can drop normally; on each tick it upgrades the
3516/// weak handle and drains any hot buffer that's been idle longer
3517/// than `idle_after`. A `None` `sweep_interval` returns `None`,
3518/// meaning event-driven promotion only.
3519fn spawn_sweep_worker<
3520 R: RefBackend + 'static,
3521 O: OpLogBackend + 'static,
3522 S: ObjectStore + 'static,
3523>(
3524 inner: &Arc<MountInner<R, O, S>>,
3525) -> Option<SweepHandle> {
3526 let interval = inner.promotion.read_or_poisoned().sweep_interval?;
3527 let weak = Arc::downgrade(inner);
3528 let state = Arc::new(SweepShutdown::new());
3529 let state_for_thread = Arc::clone(&state);
3530 let join = std::thread::Builder::new()
3531 .name("heddle-mount-sweep".into())
3532 .spawn(move || sweep_worker_loop(weak, state_for_thread, interval))
3533 .ok()?;
3534 Some(SweepHandle {
3535 state,
3536 join: Some(join),
3537 })
3538}
3539
3540/// Tick body for the safety-sweep worker. Parks on the shutdown
3541/// condvar until either the timer interval elapses (run a sweep) or
3542/// `signal_and_join` wakes us (exit). Also exits when the weak
3543/// `MountInner` reference can no longer be upgraded.
3544fn sweep_worker_loop<
3545 R: RefBackend + 'static,
3546 O: OpLogBackend + 'static,
3547 S: ObjectStore + 'static,
3548>(
3549 inner: std::sync::Weak<MountInner<R, O, S>>,
3550 state: Arc<SweepShutdown>,
3551 interval: Duration,
3552) {
3553 loop {
3554 // Wait returns true on shutdown, false on timeout — either
3555 // way we re-check the upgrade afterwards.
3556 if state.wait(interval) {
3557 return;
3558 }
3559 let Some(mount) = inner.upgrade() else {
3560 return;
3561 };
3562 if let Err(err) = mount.sweep_idle_buffers() {
3563 warn!(?err, "sweep worker hit error promoting idle buffers");
3564 }
3565 // Drop the strong-count immediately so the mount can drop
3566 // even if our next wait is still pending.
3567 drop(mount);
3568 }
3569}
3570
3571fn resolve_thread<R: RefBackend, O: OpLogBackend, S: ObjectStore>(
3572 repo: &Repository<R, O, S>,
3573 thread: &str,
3574) -> Result<MountState> {
3575 let thread_name = objects::object::ThreadName::from(thread);
3576 // `CoreRefBackend::get_thread` is async for the Postgres backend;
3577 // mount construction always runs off-runtime, so block here.
3578 let state_id = pollster::block_on(repo.refs().get_thread(&thread_name))?
3579 .ok_or_else(|| MountError::UnknownThread(thread.to_string()))?;
3580 let state = repo
3581 .store()
3582 .get_state(&state_id)?
3583 .ok_or_else(|| MountError::UnknownThread(thread.to_string()))?;
3584 Ok(MountState {
3585 state_id,
3586 tree: state.tree,
3587 })
3588}
3589
3590impl<R: RefBackend + 'static, O: OpLogBackend + 'static, S: ObjectStore + 'static> PlatformShell
3591 for ContentAddressedMount<R, O, S>
3592{
3593 fn lookup(&self, parent: NodeId, name: &OsStr) -> Result<Option<Entry>> {
3594 let record = self.record_for(parent)?;
3595 let parent_path = match self.dir_path_of(&record) {
3596 Some(p) => p,
3597 None => return Ok(None),
3598 };
3599 let Some(name_str) = name.to_str() else {
3600 return Ok(None);
3601 };
3602 let child_path = join_child(&parent_path, name_str);
3603
3604 // Pending tier wins over the immutable tree for files —
3605 // that's what makes "write then read" return the new bytes.
3606 match self.pending_lookup(&child_path) {
3607 Some(PendingHit::Tombstone) => return Ok(None),
3608 Some(hit) => {
3609 // Non-tombstone hits always yield an entry; tombstone
3610 // is handled above.
3611 if let Some(entry) = self.entry_from_pending_hit(hit, &child_path, name) {
3612 return Ok(Some(entry));
3613 }
3614 return Ok(None);
3615 }
3616 None => {}
3617 }
3618
3619 // Did an ancestor get rmdir'd? Then the captured-tree entry
3620 // is no longer addressable through this mount.
3621 {
3622 let pending = self.inner.pending.lock_or_poisoned();
3623 if pending.dir_tombstones.contains(&child_path)
3624 || self.ancestor_is_dir_tombstoned(&pending, &child_path)
3625 {
3626 return Ok(None);
3627 }
3628 }
3629
3630 // Captured tree wins over implicit pending dirs: if both
3631 // the captured tree has `nested/` AND the pending tier has
3632 // `nested/c.txt`, we want callers to descend through the
3633 // captured `Dir` record (which still overlays pending on
3634 // its way down) rather than through a `PendingDir` shell
3635 // that would hide the captured siblings.
3636 let parent_tree = self.tree_for_record(&record)?;
3637 if let Some(tree_entry) = parent_tree.get(name_str) {
3638 return Ok(Some(self.entry_from_tree_entry(&parent_path, tree_entry)?));
3639 }
3640
3641 // Implicit directory introduced by a deeper pending write
3642 // (e.g. write to `newdir/foo.rs` makes `newdir` resolvable
3643 // as a directory before capture).
3644 if self.pending_dir_exists(&child_path) {
3645 let node = self.intern(NodeRecord::PendingDir {
3646 path: child_path.clone(),
3647 });
3648 return Ok(Some(Entry {
3649 node,
3650 name: OsString::from(name_str),
3651 kind: NodeKind::Directory,
3652 size: self.pending_children_at(&child_path).len() as u64,
3653 unix_mode: DIR_UNIX_MODE,
3654 }));
3655 }
3656
3657 Ok(None)
3658 }
3659
3660 fn read(&self, node: NodeId, offset: u64, buf: &mut [u8]) -> Result<usize> {
3661 let record = self.record_for(node)?;
3662
3663 // Hot-tier fast path: if there's an in-flight buffer for
3664 // *this* NodeId, copy the requested slice directly under the
3665 // lock without cloning the whole buffer. Sub-microsecond on
3666 // small writes; avoids one `Vec::clone` per `read` callback.
3667 {
3668 let pending = self.inner.pending.lock_or_poisoned();
3669 if let Some(hot) = pending.hot.get(&node.0) {
3670 return Ok(copy_into(&hot.bytes, offset, buf));
3671 }
3672 }
3673
3674 match &record {
3675 NodeRecord::PendingFile { path, .. } => {
3676 // Same shape, keyed by path: another NodeId may own
3677 // the buffer (e.g. after rename/coalesce). Orphan
3678 // PendingFiles skip the path overlay — the path is
3679 // gone (open-unlinked) or rebound (rename-over) — but
3680 // the unified shape preserves the inode's own warm
3681 // bytes (if any) at `warm[node.0]`. With no warm
3682 // fallback there is no captured-tier source either,
3683 // so the read errors with Stale.
3684 let warm_blob = {
3685 let pending = self.inner.pending.lock_or_poisoned();
3686 if pending.is_orphan(node.0) {
3687 return match pending.warm.get(&node.0).map(|e| e.blob) {
3688 Some(blob) => {
3689 drop(pending);
3690 let bytes = self.load_blob_bytes(&blob)?;
3691 Ok(copy_into(&bytes, offset, buf))
3692 }
3693 None => Err(MountError::Stale(format!(
3694 "orphan pending file {} has no readable bytes",
3695 path.display()
3696 ))),
3697 };
3698 }
3699 if let Some(id) = pending.hot_by_path.get(path).copied()
3700 && let Some(hot) = pending.hot.get(&id)
3701 {
3702 return Ok(copy_into(&hot.bytes, offset, buf));
3703 }
3704 // Warm is NodeId-keyed; resolve path → id via the
3705 // inode registry.
3706 let inodes = self.inner.inodes.lock_or_poisoned();
3707 inodes
3708 .by_path
3709 .get(path)
3710 .copied()
3711 .and_then(|id| pending.warm.get(&id).map(|e| e.blob))
3712 };
3713 match warm_blob {
3714 Some(blob) => {
3715 let bytes = self.load_blob_bytes(&blob)?;
3716 Ok(copy_into(&bytes, offset, buf))
3717 }
3718 None => Err(MountError::Stale(format!(
3719 "pending file {}",
3720 path.display()
3721 ))),
3722 }
3723 }
3724 NodeRecord::File { blob, path, .. } => {
3725 // A captured-tree file whose path now has a pending
3726 // overlay (hot buffer on a sibling NodeId, warm-tier
3727 // promotion, or tombstone) must serve the overlay,
3728 // not the captured blob. Without this, a FUSE
3729 // `write → flush → read` round-trip through the
3730 // *same* kernel-cached NodeId silently returns the
3731 // pre-write bytes (the kernel reuses its dentry for
3732 // the duration of the entry TTL and never re-issues
3733 // `lookup`, so the inode record is never refreshed
3734 // from `File` to `PendingFile`).
3735 //
3736 // Priority: hot @ another NodeId → warm → tombstone
3737 // (ENOENT-shaped Stale) → captured blob.
3738 //
3739 // Orphan exception: an open-unlinked or
3740 // rename-displaced inode must skip the path overlay
3741 // entirely. `tombstones[path]` / `hot_by_path[path]`
3742 // / `warm[path]` now reflect a sibling at the same
3743 // name; serving them would let the open fd observe
3744 // (or even modify, via Overlay::Hot) bytes that
3745 // POSIX assigns to the sibling. Fall through to the
3746 // captured blob — that's the inode's own data.
3747 let overlay = {
3748 let pending = self.inner.pending.lock_or_poisoned();
3749 if pending.is_orphan(node.0) {
3750 pending
3751 .warm
3752 .get(&node.0)
3753 .map(|warm| Overlay::Warm(warm.blob))
3754 } else if pending.tombstones.contains(path) {
3755 Some(Overlay::Gone)
3756 } else if let Some(other_id) = pending.hot_by_path.get(path).copied()
3757 && let Some(hot) = pending.hot.get(&other_id)
3758 {
3759 return Ok(copy_into(&hot.bytes, offset, buf));
3760 } else {
3761 let inodes = self.inner.inodes.lock_or_poisoned();
3762 inodes.by_path.get(path).copied().and_then(|id| {
3763 pending.warm.get(&id).map(|warm| Overlay::Warm(warm.blob))
3764 })
3765 }
3766 };
3767 match overlay {
3768 Some(Overlay::Gone) => Err(MountError::Stale(format!(
3769 "file {} was unlinked through the mount",
3770 path.display()
3771 ))),
3772 Some(Overlay::Warm(blob)) => {
3773 let bytes = self.load_blob_bytes(&blob)?;
3774 Ok(copy_into(&bytes, offset, buf))
3775 }
3776 None => {
3777 let bytes = self.load_blob_bytes(blob)?;
3778 Ok(copy_into(&bytes, offset, buf))
3779 }
3780 }
3781 }
3782 NodeRecord::Gitlink { placeholder, .. } => Ok(copy_into(placeholder, offset, buf)),
3783 NodeRecord::Symlink { blob } => {
3784 let bytes = self.load_blob_bytes(blob)?;
3785 Ok(copy_into(&bytes, offset, buf))
3786 }
3787 _ => Err(MountError::NotFound(format!(
3788 "read on non-file node {}",
3789 node.0
3790 ))),
3791 }
3792 }
3793
3794 fn write(&self, node: NodeId, offset: u64, data: &[u8]) -> Result<usize> {
3795 let end = validate_write_extent(offset, data.len())?;
3796 let offset = usize::try_from(offset).map_err(|_| {
3797 MountError::InvalidArgument(format!("write offset {offset} does not fit in usize"))
3798 })?;
3799 // Determine the mount-relative path and mode to key the hot
3800 // buffer on. New files (`PendingFile`) carry their path
3801 // directly; pre-existing files identify by the parent's
3802 // tree entry. Any other node type rejects writes.
3803 let record = self.record_for(node)?;
3804 let (path, mode, captured_blob) = match &record {
3805 NodeRecord::PendingFile { path, mode } => (path.clone(), *mode, None),
3806 NodeRecord::File {
3807 path, mode, blob, ..
3808 } => (path.clone(), *mode, Some(*blob)),
3809 _ => return Err(MountError::ReadOnly),
3810 };
3811
3812 // Phase 1: under the lock, decide whether a buffer already
3813 // exists, and if not, what durable source we should seed it
3814 // from. Snapshot the seed source's blob oid (if any) and drop
3815 // the lock so we can do CAS IO without blocking other writers.
3816 //
3817 // POSIX `pwrite` preserves bytes outside the [offset, offset+len)
3818 // range. The kernel never re-issues those bytes on a partial
3819 // overwrite, so the hot buffer must already contain them when
3820 // we apply `data`. The seed sources, in priority order:
3821 //
3822 // 1. The warm tier — a previously-flushed write to this same
3823 // path in this mount session. This is the most recent
3824 // durable view and supersedes the captured tree.
3825 // 2. The captured tree's blob for this path — the underlying
3826 // file the agent is editing. Only applicable when the
3827 // record was minted from a captured tree entry (i.e.
3828 // `NodeRecord::File`); a `PendingFile` with no warm entry
3829 // means the agent already unlinked-and-recreated.
3830 // 3. Empty — no durable predecessor, so this write builds a
3831 // file from scratch.
3832 //
3833 // A tombstone for the path overrides everything: the agent
3834 // deleted the file and is now creating a fresh one.
3835 enum Seed {
3836 None,
3837 Blob(ContentHash),
3838 }
3839 let seed = {
3840 // Resolve the path's current Live owner via the inode
3841 // registry — warm bytes for the path live at
3842 // `warm[live_id]` under the unified shape.
3843 let path_owner = {
3844 let inodes = self.inner.inodes.lock_or_poisoned();
3845 inodes.by_path.get(&path).copied()
3846 };
3847 let pending = self.inner.pending.lock_or_poisoned();
3848 let orphan = pending.is_orphan(node.0);
3849 if pending.hot.contains_key(&node.0) {
3850 // The per-NodeId buffer is always authoritative —
3851 // both for live writes (this fd's accumulated bytes)
3852 // and for orphan writes (POSIX says the bytes belong
3853 // to the open handle).
3854 Seed::None
3855 } else if !orphan
3856 && pending
3857 .hot_by_path
3858 .get(&path)
3859 .is_some_and(|id| pending.hot.contains_key(id))
3860 {
3861 // Sibling at the same path has a buffer — coalesce
3862 // onto it. Orphans never look at the path's overlay
3863 // (the sibling at `hot_by_path[path]` is a different
3864 // inode, not us).
3865 Seed::None
3866 } else if orphan {
3867 // Orphan-aware seeding. The path's overlay belongs
3868 // to the sibling at the rebound name; this inode's
3869 // own bytes live at `warm[node.0]` (or in the
3870 // captured blob).
3871 pending
3872 .warm
3873 .get(&node.0)
3874 .map(|e| Seed::Blob(e.blob))
3875 .or_else(|| captured_blob.map(Seed::Blob))
3876 .unwrap_or(Seed::None)
3877 } else if pending.tombstones.contains(&path) {
3878 // Unlink-then-write through a fresh inode (POSIX
3879 // unlink+open(O_CREAT)): start from empty.
3880 Seed::None
3881 } else if let Some(entry) = path_owner.and_then(|id| pending.warm.get(&id)) {
3882 Seed::Blob(entry.blob)
3883 } else if let Some(blob) = captured_blob {
3884 Seed::Blob(blob)
3885 } else {
3886 Seed::None
3887 }
3888 };
3889 let seed_bytes = match seed {
3890 Seed::None => None,
3891 // The hot buffer is owned + mutated, so we materialize a
3892 // Vec here. One alloc + copy per first-write per file;
3893 // subsequent writes hit the existing buffer.
3894 Seed::Blob(hash) => Some((*self.load_blob_bytes(&hash)?).to_vec()),
3895 };
3896
3897 // Phase 2: re-acquire the lock, install or update the hot
3898 // buffer, apply the write. If a buffer materialized between
3899 // phases (e.g. a coalesce from another NodeId), prefer the
3900 // existing buffer's bytes — our `seed_bytes` are stale.
3901 let mut pending = self.inner.pending.lock_or_poisoned();
3902 // POSIX unlink+open semantics. Two write shapes share this
3903 // method and must be kept separate:
3904 //
3905 // * unlink-then-create (`unlink P; open(P, O_CREAT); write`)
3906 // — `create_file` minted a fresh `NodeId` and cleared the
3907 // tombstone for P. Our `node.0` is not in `orphans` and
3908 // the write republishes the name normally.
3909 //
3910 // * open-then-unlink (`open(P); unlink P; write through old
3911 // fd`) — `unlink_entry` recorded `node.0` in `orphans`
3912 // and left the tombstone in place. POSIX is explicit:
3913 // the inode lives behind the fd, but the directory entry
3914 // must stay gone. Republishing `hot_by_path[P] = node.0`
3915 // or clearing the tombstone would resurrect the pathname
3916 // for every other observer (lookup, enumerate, capture).
3917 // The orphan branch updates only the per-NodeId buffer;
3918 // `flush_node` reads the same `orphans` signal at
3919 // promotion time and drops the buffer instead of warming
3920 // it.
3921 let orphan = pending.is_orphan(node.0);
3922 if !orphan {
3923 // Coalesce two NodeIds for the same path onto the same buffer.
3924 if let Some(existing_id) = pending.hot_by_path.get(&path).copied()
3925 && existing_id != node.0
3926 && let Some(buf) = pending.hot.remove(&existing_id)
3927 {
3928 pending.hot.insert(node.0, buf);
3929 }
3930 pending.hot_by_path.insert(path.clone(), node.0);
3931 // A live hot buffer means the file exists again — clear
3932 // any tombstone for this path so subsequent
3933 // `pending_lookup` calls see the buffer instead of a
3934 // "deleted" sentinel. POSIX:
3935 // unlink+open(O_CREAT)+pwrite reborns the path. The seed
3936 // logic above already starts the buffer empty when a
3937 // tombstone is present, so we don't need to inspect the
3938 // tombstone here.
3939 pending.tombstones.remove(&path);
3940 }
3941 let buf = pending.hot.entry(node.0).or_insert_with(|| HotBuffer {
3942 path: path.clone(),
3943 mode,
3944 bytes: seed_bytes.unwrap_or_default(),
3945 last_touched: Instant::now(),
3946 revision: 0,
3947 });
3948 // POSIX `pwrite` past EOF zero-fills the gap.
3949 if buf.bytes.len() < end {
3950 buf.bytes.resize(end, 0);
3951 }
3952 buf.bytes[offset..end].copy_from_slice(data);
3953 buf.last_touched = Instant::now();
3954 buf.revision = buf.revision.wrapping_add(1);
3955 if !orphan {
3956 let indexed = PendingChildKind::HotFile {
3957 node,
3958 size: buf.bytes.len() as u64,
3959 mode: buf.mode,
3960 };
3961 pending.child_index.insert(path.clone(), indexed);
3962 }
3963 let written = data.len();
3964 drop(pending);
3965 // Cheap idle-promotion sweep — an agent that's gone quiet on
3966 // *other* files for longer than the policy window gets its
3967 // buffers drained without an explicit close.
3968 let _ = self.promote_idle_buffers();
3969 Ok(written)
3970 }
3971
3972 fn enumerate(&self, dir: NodeId) -> Result<Vec<Entry>> {
3973 let record = self.record_for(dir)?;
3974 let parent_path = match self.dir_path_of(&record) {
3975 Some(p) => p,
3976 None => return Err(MountError::NotADirectory(format!("{record:?}"))),
3977 };
3978 let tree = self.tree_for_record(&record)?;
3979 let mut by_name: BTreeMap<&str, Entry> = BTreeMap::new();
3980
3981 // If this directory itself is dir-tombstoned, enumerate
3982 // returns empty regardless of any captured children. (A
3983 // child rmdir doesn't affect us — only an ancestor or self
3984 // tombstone does.)
3985 {
3986 let pending = self.inner.pending.lock_or_poisoned();
3987 if pending.dir_tombstones.contains(&parent_path)
3988 || self.ancestor_is_dir_tombstoned(&pending, &parent_path)
3989 {
3990 return Ok(vec![]);
3991 }
3992 }
3993
3994 // Pass 1: captured-tree entries, with pending overlay.
3995 for tree_entry in tree.entries() {
3996 let entry_path = join_child(&parent_path, tree_entry.name());
3997 // Whole-subtree rmdir on a captured dir entry.
3998 {
3999 let pending = self.inner.pending.lock_or_poisoned();
4000 if pending.dir_tombstones.contains(&entry_path) {
4001 continue;
4002 }
4003 }
4004 match self.pending_lookup(&entry_path) {
4005 Some(PendingHit::Tombstone) => continue,
4006 Some(hit) => {
4007 if let Some(entry) =
4008 self.entry_from_pending_hit(hit, &entry_path, OsStr::new(tree_entry.name()))
4009 {
4010 by_name.insert(tree_entry.name(), entry);
4011 }
4012 continue;
4013 }
4014 None => {}
4015 }
4016 let entry = self.entry_from_tree_entry(&parent_path, tree_entry)?;
4017 by_name.insert(tree_entry.name(), entry);
4018 }
4019
4020 // Pass 2: pending-only children of `parent_path` (mount-only
4021 // files and implicit subdirectories the agent created).
4022 let mut pending_entries: Vec<Entry> = Vec::new();
4023 let pending_children = self.pending_children_at(&parent_path);
4024 for (name, kind) in pending_children {
4025 // Don't shadow a captured-tree entry (already handled in
4026 // pass 1 via pending_lookup).
4027 if by_name.contains_key(name.as_str()) {
4028 continue;
4029 }
4030 let full_path = join_child(&parent_path, &name);
4031 match kind {
4032 PendingChildKind::HotFile { node, size, mode } => {
4033 pending_entries.push(Entry {
4034 node,
4035 name: OsString::from(&name),
4036 kind: kind_for_mode(mode),
4037 size,
4038 unix_mode: mode.to_unix_mode(),
4039 });
4040 }
4041 PendingChildKind::WarmFile { size, mode } => {
4042 let node = self.intern(NodeRecord::PendingFile {
4043 path: full_path,
4044 mode,
4045 });
4046 pending_entries.push(Entry {
4047 node,
4048 name: OsString::from(&name),
4049 kind: kind_for_mode(mode),
4050 size,
4051 unix_mode: mode.to_unix_mode(),
4052 });
4053 }
4054 PendingChildKind::Dir => {
4055 let node = self.intern(NodeRecord::PendingDir { path: full_path });
4056 pending_entries.push(Entry {
4057 node,
4058 name: OsString::from(&name),
4059 kind: NodeKind::Directory,
4060 size: 0,
4061 unix_mode: DIR_UNIX_MODE,
4062 });
4063 }
4064 PendingChildKind::Symlink { size } => {
4065 let node = self.intern(NodeRecord::PendingSymlink { path: full_path });
4066 pending_entries.push(Entry {
4067 node,
4068 name: OsString::from(&name),
4069 kind: NodeKind::Symlink,
4070 size,
4071 unix_mode: FileMode::Symlink.to_unix_mode(),
4072 });
4073 }
4074 }
4075 }
4076 let mut entries: Vec<Entry> = by_name.into_values().collect();
4077 entries.extend(pending_entries);
4078 Ok(entries)
4079 }
4080
4081 fn attrs(&self, node: NodeId) -> Result<Attrs> {
4082 let record = self.record_for(node)?;
4083 let kind = record.kind();
4084 let unix_mode = record.unix_mode();
4085 let (size, nlink) = match &record {
4086 NodeRecord::Root { tree } | NodeRecord::Dir { tree, .. } => {
4087 let tree = self.load_tree(tree)?;
4088 // 2 = `.` + the parent's entry pointing at us. Heddle
4089 // doesn't model hard links, so we don't try to count
4090 // subdirectories' `..` entries.
4091 (tree.entries().len() as u64, 2)
4092 }
4093 NodeRecord::PendingDir { path } => {
4094 // Implicit dir — content lives entirely in the
4095 // pending tier. Size = direct-child count.
4096 (self.pending_children_at(path).len() as u64, 2)
4097 }
4098 NodeRecord::File { blob, path, .. } => {
4099 // Same overlay priority as `read`: hot @ this NodeId
4100 // → hot @ another NodeId for the same path → warm-tier
4101 // promotion → tombstone (stale) → captured blob.
4102 // Keeping `attrs` and `read` symmetric is mandatory:
4103 // `read` consults the warm tier for captured files
4104 // (so `WORLD` shadows `world`), and a stale `attrs`
4105 // that still reports the captured size would clip the
4106 // returned bytes in the kernel's read buffer.
4107 //
4108 // Orphan exception: same as `read`. An open-unlinked
4109 // or rename-displaced inode skips the path overlay
4110 // and reports the captured blob's size (or the
4111 // per-NodeId hot buffer's length, checked first).
4112 let overlay_size = {
4113 let pending = self.inner.pending.lock_or_poisoned();
4114 if let Some(buf) = pending.hot.get(&node.0) {
4115 Some(Some(buf.bytes.len() as u64))
4116 } else if pending.is_orphan(node.0) {
4117 // Prefer the orphan's own warm size (unified
4118 // shape: `warm[node.0]`). With no warm, fall
4119 // through to `blob_size(blob)` — the captured
4120 // size is the orphan's own.
4121 pending.warm.get(&node.0).map(|e| Some(e.size))
4122 } else if pending.tombstones.contains(path) {
4123 // Tombstoned via the mount: treat as
4124 // not-yet-collected. The path is gone but the
4125 // inode is still registered.
4126 Some(None)
4127 } else if let Some(other_id) = pending.hot_by_path.get(path).copied()
4128 && let Some(hot) = pending.hot.get(&other_id)
4129 {
4130 Some(Some(hot.bytes.len() as u64))
4131 } else {
4132 // Warm is NodeId-keyed; resolve path → id via
4133 // the inode registry.
4134 let inodes = self.inner.inodes.lock_or_poisoned();
4135 inodes
4136 .by_path
4137 .get(path)
4138 .copied()
4139 .and_then(|id| pending.warm.get(&id).map(|warm| Some(warm.size)))
4140 }
4141 };
4142 match overlay_size {
4143 Some(Some(size)) => (size, 1),
4144 Some(None) => {
4145 return Err(MountError::Stale(format!(
4146 "file {} was unlinked through the mount",
4147 path.display()
4148 )));
4149 }
4150 None => (self.blob_size(blob)?, 1),
4151 }
4152 }
4153 NodeRecord::Gitlink { placeholder, .. } => (placeholder.len() as u64, 1),
4154 NodeRecord::Symlink { blob } => (self.blob_size(blob)?, 1),
4155 NodeRecord::PendingFile { path, .. } => {
4156 // Orphan branch: a rename-displaced or
4157 // unlinked-but-still-open PendingFile reports either
4158 // its per-NodeId hot buffer length, or its own
4159 // `warm[node.0]` size. `pending_lookup` would
4160 // otherwise consult the rebound path overlay and
4161 // serve the sibling's size.
4162 let orphan_size = {
4163 let pending = self.inner.pending.lock_or_poisoned();
4164 if pending.is_orphan(node.0) {
4165 Some(
4166 pending
4167 .hot
4168 .get(&node.0)
4169 .map(|buf| buf.bytes.len() as u64)
4170 .or_else(|| pending.warm.get(&node.0).map(|e| e.size)),
4171 )
4172 } else {
4173 None
4174 }
4175 };
4176 if let Some(opt) = orphan_size {
4177 let size = opt.ok_or_else(|| {
4178 MountError::Stale(format!(
4179 "orphan pending file {} has no buffered bytes",
4180 path.display()
4181 ))
4182 })?;
4183 (size, 1)
4184 } else {
4185 let hit = self.pending_lookup(path).ok_or_else(|| {
4186 MountError::Stale(format!("pending file {}", path.display()))
4187 })?;
4188 let size = match hit {
4189 PendingHit::Hot { size, .. } | PendingHit::Warm { size, .. } => size,
4190 PendingHit::Symlink { target_len } => target_len,
4191 PendingHit::Tombstone => 0,
4192 };
4193 (size, 1)
4194 }
4195 }
4196 NodeRecord::PendingSymlink { path } => {
4197 let pending = self.inner.pending.lock_or_poisoned();
4198 let size = pending
4199 .symlinks
4200 .get(path)
4201 .map(|t| t.len() as u64)
4202 .ok_or_else(|| {
4203 MountError::Stale(format!("pending symlink {}", path.display()))
4204 })?;
4205 (size, 1)
4206 }
4207 };
4208 let _ = self.path_of(&record);
4209 Ok(Attrs {
4210 node,
4211 kind,
4212 size,
4213 unix_mode,
4214 nlink,
4215 mtime: self.inner.mounted_at,
4216 })
4217 }
4218
4219 fn invalidate(&self, node: NodeId) -> Result<()> {
4220 // Witness-gated discharge: `bp.kernel_forget_inode(node.0)`
4221 // returns:
4222 //
4223 // * `Some(warm_still_references)` — the FSM check passed
4224 // (state is `Released` or `Live { open_count == 0 }`);
4225 // `hot[node]` (with its `hot_by_path` reverse-index
4226 // cleanup) and `state[node]` have been dropped, and the
4227 // bool tells us whether `warm[node]` is still populated.
4228 // Retire the inode-side record iff warm doesn't reference
4229 // — otherwise capture still needs the NodeId → path chain
4230 // to plant the warm bytes back into the new tree.
4231 // * `None` — the FSM check failed (state is
4232 // `Live { open_count >= 1 }` or any `Orphan`); the bytes
4233 // are still referenced. The witness-gated retrofit
4234 // (heddle#211) makes the entire forget path short-circuit
4235 // here: `hot[node]` / `state[node]` are preserved and the
4236 // inode-side `forget` is skipped. The kernel will re-issue
4237 // `forget` once the surviving fd closes (or never, and the
4238 // next `release_node` retires the record). Closes Codex
4239 // r11 finding #3 — the pre-retrofit path removed
4240 // `hot[node]` before any FSM check, stranding an open
4241 // Orphan fd with no readable bytes.
4242 //
4243 // Warm preservation (Codex r12 threads 3293484634 /
4244 // 3293510311, P1): `apply_kernel_forget` intentionally
4245 // leaves `warm[node]` alone — warm is the only durable
4246 // pre-capture copy of flushed writes, and FUSE `forget` is
4247 // a kernel-side dcache eviction (not a close), so dropping
4248 // warm would silently lose the user's committed-in-session
4249 // data.
4250 let retire_inode_record = {
4251 let mut pending = self.inner.pending.lock_or_poisoned();
4252 pending.with_brand(|bp| {
4253 bp.kernel_forget_inode(node.0)
4254 .map(|warm_still_references| !warm_still_references)
4255 .unwrap_or(false)
4256 })
4257 };
4258 if retire_inode_record {
4259 self.inner.inodes.lock_or_poisoned().forget(node);
4260 }
4261 Ok(())
4262 }
4263
4264 fn flush(&self, node: NodeId) -> Result<()> {
4265 self.flush_node(node)
4266 }
4267
4268 fn release(&self, node: NodeId) -> Result<()> {
4269 self.release_node(node)
4270 }
4271
4272 fn on_open(&self, node: NodeId) -> Result<()> {
4273 ContentAddressedMount::on_open(self, node)
4274 }
4275
4276 fn create_file(
4277 &self,
4278 parent: NodeId,
4279 name: &OsStr,
4280 mode: FileMode,
4281 exclusive: bool,
4282 ) -> Result<Entry> {
4283 ContentAddressedMount::create_file(self, parent, name, mode, exclusive)
4284 }
4285
4286 fn make_dir(&self, parent: NodeId, name: &OsStr) -> Result<Entry> {
4287 ContentAddressedMount::make_dir(self, parent, name)
4288 }
4289
4290 fn unlink_entry(&self, parent: NodeId, name: &OsStr) -> Result<()> {
4291 ContentAddressedMount::unlink_entry(self, parent, name)
4292 }
4293
4294 fn rmdir_entry(&self, parent: NodeId, name: &OsStr) -> Result<()> {
4295 ContentAddressedMount::rmdir_entry(self, parent, name)
4296 }
4297
4298 fn rename_entry(
4299 &self,
4300 old_parent: NodeId,
4301 old_name: &OsStr,
4302 new_parent: NodeId,
4303 new_name: &OsStr,
4304 ) -> Result<()> {
4305 ContentAddressedMount::rename_entry(self, old_parent, old_name, new_parent, new_name)
4306 }
4307
4308 fn rename_entry_with_options(
4309 &self,
4310 old_parent: NodeId,
4311 old_name: &OsStr,
4312 new_parent: NodeId,
4313 new_name: &OsStr,
4314 options: RenameOptions,
4315 ) -> Result<()> {
4316 ContentAddressedMount::rename_entry_with_options(
4317 self, old_parent, old_name, new_parent, new_name, options,
4318 )
4319 }
4320
4321 fn set_attrs(&self, node: NodeId, update: AttrUpdate) -> Result<Attrs> {
4322 ContentAddressedMount::set_attrs(self, node, update)
4323 }
4324
4325 fn create_symlink(&self, parent: NodeId, name: &OsStr, target: &Path) -> Result<Entry> {
4326 ContentAddressedMount::create_symlink(self, parent, name, target)
4327 }
4328
4329 fn read_link(&self, node: NodeId) -> Result<OsString> {
4330 ContentAddressedMount::read_link(self, node)
4331 }
4332}
4333
4334impl<R: RefBackend + 'static, O: OpLogBackend + 'static, S: ObjectStore + 'static>
4335 ContentAddressedMount<R, O, S>
4336{
4337 /// Test-only accessor for the warm tier so unit tests can verify
4338 /// promotions landed without going through `read`. Returns paths
4339 /// resolved via the inode registry (warm is NodeId-keyed under
4340 /// the unified shape).
4341 #[cfg(test)]
4342 pub(crate) fn warm_keys(&self) -> Vec<PathBuf> {
4343 let pending = self.inner.pending.lock_or_poisoned();
4344 let inodes = self.inner.inodes.lock_or_poisoned();
4345 pending
4346 .warm
4347 .keys()
4348 .filter(|id| !pending.is_orphan(**id))
4349 .filter_map(|id| inodes.by_id.get(id).and_then(warm_path_of_record))
4350 .map(Path::to_path_buf)
4351 .collect()
4352 }
4353
4354 /// Test-only accessor: was `path` promoted to a CAS blob? Returns
4355 /// the blob oid so dedup tests can compare across mounts.
4356 #[cfg(test)]
4357 pub(crate) fn warm_blob(&self, path: impl AsRef<Path>) -> Option<ContentHash> {
4358 let path = path.as_ref();
4359 let id = self
4360 .inner
4361 .inodes
4362 .lock_or_poisoned()
4363 .by_path
4364 .get(path)
4365 .copied()?;
4366 self.inner
4367 .pending
4368 .lock_or_poisoned()
4369 .warm
4370 .get(&id)
4371 .map(|e| e.blob)
4372 }
4373
4374 /// Test-only accessor: are there any open hot-tier buffers?
4375 #[cfg(test)]
4376 pub(crate) fn hot_buffer_count(&self) -> usize {
4377 self.inner.pending.lock_or_poisoned().hot.len()
4378 }
4379
4380 /// Test-only accessor: snapshot of currently tombstoned paths.
4381 #[cfg(test)]
4382 #[allow(dead_code)]
4383 pub(crate) fn tombstones(&self) -> Vec<PathBuf> {
4384 self.inner
4385 .pending
4386 .lock_or_poisoned()
4387 .tombstones
4388 .iter()
4389 .cloned()
4390 .collect()
4391 }
4392
4393 /// Test-only accessor for the wrapped repository.
4394 #[cfg(test)]
4395 pub(crate) fn repo_handle(&self) -> &Repository<R, O, S> {
4396 &self.inner.repo
4397 }
4398
4399 /// Test-only accessor: is `node` currently marked as an orphaned
4400 /// inode (open-unlinked or rename-displaced with surviving fds)?
4401 #[cfg(test)]
4402 pub(crate) fn orphans_contains(&self, node: NodeId) -> bool {
4403 self.inner.pending.lock_or_poisoned().is_orphan(node.0)
4404 }
4405}
4406
4407/// Low-level test helpers. The mount doesn't yet expose a `create()`
4408/// entrypoint (the FUSE adapter will eventually wire that callback);
4409/// for now tests bypass the kernel-walk and install pending records
4410/// directly. The shape mirrors what `Filesystem::create` will do once
4411/// it lands.
4412#[cfg(test)]
4413pub(crate) mod test_helpers {
4414 use super::*;
4415
4416 /// Mint a fresh pending-file at any (possibly nested) mount-relative
4417 /// path. Path components are taken verbatim — the helper does no
4418 /// validation beyond path normalization.
4419 pub(crate) fn install_pending_file(
4420 mount: &ContentAddressedMount,
4421 name: &str,
4422 mode: FileMode,
4423 ) -> NodeId {
4424 let path = PathBuf::from(name);
4425 mount.intern(NodeRecord::PendingFile { path, mode })
4426 }
4427}