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, 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 self.inner.blob_cache.clear();
1389 self.inner.repo.store().clear_recent_caches();
1390 }
1391
1392 /// Spawn a background tree-walker that hydrates every file blob
1393 /// in the captured tree into the shared blob cache. The first
1394 /// kernel `read` after this finishes is served from memory at
1395 /// `Arc::clone` + `memcpy` cost — beats `std::fs::read` on every
1396 /// tier we benchmark.
1397 ///
1398 /// The returned [`PrewarmHandle`] is the caller's lever:
1399 /// * Drop it without calling anything → the prewarmer keeps
1400 /// running until natural completion or the mount drops
1401 /// (the workers hold `Weak<MountInner>` and self-terminate
1402 /// when the strong count hits zero).
1403 /// * `.cancel()` signals shutdown without joining.
1404 /// * `.wait()` joins all workers and returns the final stats.
1405 ///
1406 /// Workers stop early when the cache is ≥ 90% full to avoid
1407 /// churn-evicting work they just did. Blobs already cached
1408 /// (from a sibling mount sharing the same pool) are skipped
1409 /// cheaply — this is the fork-thread fast path.
1410 pub fn prewarm(&self) -> PrewarmHandle {
1411 PrewarmHandle::start(Arc::downgrade(&self.inner))
1412 }
1413
1414 fn load_blob_bytes(&self, hash: &ContentHash) -> Result<bytes::Bytes> {
1415 if let Some(hit) = self.inner.blob_cache.get(hash) {
1416 return Ok(hit);
1417 }
1418 let bytes = self
1419 .store()
1420 .get_blob_bytes(hash)?
1421 .ok_or_else(|| MountError::NotFound(format!("blob {hash}")))?;
1422 self.inner.blob_cache.insert(*hash, bytes.clone());
1423 Ok(bytes)
1424 }
1425
1426 /// Header-only size lookup. Avoids loading the full blob just to
1427 /// learn its size — the hot path for `ls -l`.
1428 fn blob_size(&self, hash: &ContentHash) -> Result<u64> {
1429 self.store()
1430 .blob_size(hash)?
1431 .ok_or_else(|| MountError::NotFound(format!("blob {hash}")))
1432 }
1433
1434 fn record_for(&self, id: NodeId) -> Result<NodeRecord> {
1435 self.inner
1436 .inodes
1437 .lock_or_poisoned()
1438 .get(id)
1439 .ok_or_else(|| MountError::Stale(format!("node {}", id.0)))
1440 }
1441
1442 fn intern(&self, record: NodeRecord) -> NodeId {
1443 self.inner.inodes.lock_or_poisoned().intern(record)
1444 }
1445
1446 /// Resolve a mount-relative path to a [`NodeId`]. Used by tests
1447 /// that don't go through `lookup` step-by-step.
1448 pub fn lookup_path(&self, path: impl AsRef<Path>) -> Result<NodeId> {
1449 let mut node = NodeId::ROOT;
1450 for component in path.as_ref().components() {
1451 match component {
1452 Component::CurDir | Component::RootDir => continue,
1453 Component::Prefix(_) => {
1454 return Err(MountError::NotFound(format!(
1455 "unsupported path component in {}",
1456 path.as_ref().display()
1457 )));
1458 }
1459 Component::ParentDir => {
1460 return Err(MountError::NotFound(format!(
1461 "parent traversal not supported: {}",
1462 path.as_ref().display()
1463 )));
1464 }
1465 Component::Normal(name) => {
1466 let entry = self
1467 .lookup(node, name)?
1468 .ok_or_else(|| MountError::NotFound(name.to_string_lossy().into_owned()))?;
1469 node = entry.node;
1470 }
1471 }
1472 }
1473 Ok(node)
1474 }
1475
1476 fn entry_from_tree_entry(&self, parent_path: &Path, tree_entry: &TreeEntry) -> Result<Entry> {
1477 let entry_path = join_child(parent_path, tree_entry.name());
1478 let (kind, size, unix_mode, record) = match tree_entry.target() {
1479 TreeEntryTarget::Tree { hash } => {
1480 // We deliberately load the subtree here so the entry
1481 // count (the conventional "size" for a directory)
1482 // matches what userspace expects from `stat`.
1483 let subtree = self.load_tree(hash)?;
1484 (
1485 NodeKind::Directory,
1486 subtree.entries().len() as u64,
1487 DIR_UNIX_MODE,
1488 NodeRecord::Dir {
1489 tree: *hash,
1490 path: entry_path,
1491 },
1492 )
1493 }
1494 TreeEntryTarget::Blob { hash, executable } => {
1495 let size = self.blob_size(hash)?;
1496 let mode = if *executable {
1497 FileMode::Executable
1498 } else {
1499 FileMode::Normal
1500 };
1501 (
1502 kind_for_mode(mode),
1503 size,
1504 mode.to_unix_mode(),
1505 NodeRecord::File {
1506 blob: *hash,
1507 mode,
1508 path: entry_path,
1509 },
1510 )
1511 }
1512 TreeEntryTarget::Symlink { hash } => {
1513 let size = self.blob_size(hash)?;
1514 (
1515 NodeKind::Symlink,
1516 size,
1517 FileMode::Symlink.to_unix_mode(),
1518 NodeRecord::Symlink { blob: *hash },
1519 )
1520 }
1521 TreeEntryTarget::Gitlink { target } => {
1522 let placeholder = gitlink_placeholder_bytes(target);
1523 let size = placeholder.len() as u64;
1524 (
1525 NodeKind::File,
1526 size,
1527 FileMode::Normal.to_unix_mode(),
1528 NodeRecord::Gitlink {
1529 placeholder,
1530 path: entry_path,
1531 },
1532 )
1533 }
1534 // Native child-spool edges are not yet exposed in the FUSE mount
1535 // (no consumer facet wires them in this phase). Refuse explicitly
1536 // rather than masquerade one as a gitlink placeholder. Native
1537 // spool operations do not produce mounted trees containing these
1538 // yet, so this path is not hit in practice today.
1539 TreeEntryTarget::Spoollink { .. } => {
1540 return Err(MountError::InvalidArgument(format!(
1541 "spoollink entry '{}' is not exposable in the mount yet",
1542 tree_entry.name()
1543 )));
1544 }
1545 };
1546 let node = self.intern(record);
1547 Ok(Entry {
1548 node,
1549 name: OsString::from(tree_entry.name()),
1550 kind,
1551 size,
1552 unix_mode,
1553 })
1554 }
1555
1556 /// Build an [`Entry`] from a [`PendingHit`]. `path` is the child's
1557 /// mount-relative path (used to intern the `PendingFile` /
1558 /// `PendingSymlink` record for warm/symlink hits); `name` is the
1559 /// leaf name of the returned entry. Returns `None` for
1560 /// [`PendingHit::Tombstone`] — the caller treats that as "entry
1561 /// hidden". Shared by `lookup` and `enumerate`.
1562 fn entry_from_pending_hit(&self, hit: PendingHit, path: &Path, name: &OsStr) -> Option<Entry> {
1563 match hit {
1564 PendingHit::Tombstone => None,
1565 PendingHit::Hot { node, size, mode } => Some(Entry {
1566 node,
1567 name: name.to_os_string(),
1568 kind: kind_for_mode(mode),
1569 size,
1570 unix_mode: mode.to_unix_mode(),
1571 }),
1572 PendingHit::Warm {
1573 blob: _,
1574 size,
1575 mode,
1576 } => {
1577 let node = self.intern(NodeRecord::PendingFile {
1578 path: path.to_path_buf(),
1579 mode,
1580 });
1581 Some(Entry {
1582 node,
1583 name: name.to_os_string(),
1584 kind: kind_for_mode(mode),
1585 size,
1586 unix_mode: mode.to_unix_mode(),
1587 })
1588 }
1589 PendingHit::Symlink { target_len } => {
1590 let node = self.intern(NodeRecord::PendingSymlink {
1591 path: path.to_path_buf(),
1592 });
1593 Some(Entry {
1594 node,
1595 name: name.to_os_string(),
1596 kind: NodeKind::Symlink,
1597 size: target_len,
1598 unix_mode: FileMode::Symlink.to_unix_mode(),
1599 })
1600 }
1601 }
1602 }
1603
1604 fn tree_for_record(&self, record: &NodeRecord) -> Result<Tree> {
1605 match record {
1606 NodeRecord::Root { tree } | NodeRecord::Dir { tree, .. } => self.load_tree(tree),
1607 // Pending-only dirs have no captured tree to load yet —
1608 // their content lives entirely in the pending tier.
1609 NodeRecord::PendingDir { .. } => Ok(Tree::new()),
1610 _ => Err(MountError::NotADirectory(format!("{record:?}"))),
1611 }
1612 }
1613
1614 /// Mount-relative path for a directory record. Root resolves to
1615 /// `""`, captured Dirs and pending dirs to their stored path.
1616 fn dir_path_of(&self, record: &NodeRecord) -> Option<PathBuf> {
1617 match record {
1618 NodeRecord::Root { .. } => Some(PathBuf::new()),
1619 NodeRecord::Dir { path, .. } | NodeRecord::PendingDir { path } => Some(path.clone()),
1620 _ => None,
1621 }
1622 }
1623
1624 /// Build the relative path of `node` from the mount root, used to
1625 /// rendezvous a NodeId with its pending-tier entry. Returns `None`
1626 /// for the root or for nodes that don't carry a path identity.
1627 fn path_of(&self, record: &NodeRecord) -> Option<PathBuf> {
1628 match record {
1629 NodeRecord::PendingFile { path, .. }
1630 | NodeRecord::File { path, .. }
1631 | NodeRecord::Gitlink { path, .. } => Some(path.clone()),
1632 NodeRecord::Dir { path, .. } | NodeRecord::PendingDir { path } => Some(path.clone()),
1633 NodeRecord::PendingSymlink { path } => Some(path.clone()),
1634 _ => None,
1635 }
1636 }
1637
1638 // --- Pending tier helpers ------------------------------------------------
1639
1640 fn promote_idle_buffers(&self) -> Result<()> {
1641 self.inner.sweep_idle_buffers()
1642 }
1643
1644 /// Promote the hot buffer for `node` (if any) to a CAS blob and
1645 /// record it in the pending tree. Routed from the FUSE `flush`
1646 /// callback (per-descriptor-close). Orphaned nodes deliberately
1647 /// do nothing here — see [`MountInner::flush_node`] for the
1648 /// lifecycle rationale.
1649 pub fn flush_node(&self, node: NodeId) -> Result<()> {
1650 self.inner.flush_node(node)
1651 }
1652
1653 /// Final close of `node` from a FUSE `release` callback. Decrements
1654 /// the open-handle refcount; on the last close, drops orphan
1655 /// state and (for non-orphans) promotes any surviving hot buffer.
1656 pub fn release_node(&self, node: NodeId) -> Result<()> {
1657 self.inner.release_node(node)
1658 }
1659
1660 /// Notify the mount that a new open handle for `node` was minted
1661 /// (FUSE `open` / `create` callback). Used to time the orphan
1662 /// cleanup against the *final* close (see
1663 /// [`Self::release_node`] / [`MountInner::release_node`]).
1664 ///
1665 /// Bumps the open count on the existing `NodeState`, minting a
1666 /// `Live { open_count: 1 }` entry if the node is untracked. An
1667 /// Orphan can also be opened (rare — only via an fh the kernel
1668 /// still holds across a re-lookup race); we bump its refcount so
1669 /// the final release fires correctly.
1670 pub fn on_open(&self, node: NodeId) -> Result<()> {
1671 let mut pending = self.inner.pending.lock_or_poisoned();
1672 let next = match pending.state.get(&node.0).copied() {
1673 None => NodeState::Live { open_count: 1 },
1674 Some(NodeState::Live { open_count }) => NodeState::Live {
1675 open_count: open_count.saturating_add(1),
1676 },
1677 Some(NodeState::Orphan { open_count }) => NodeState::Orphan {
1678 open_count: open_count.saturating_add(1),
1679 },
1680 };
1681 pending.state.insert(node.0, next);
1682 Ok(())
1683 }
1684
1685 /// Mark `path` as deleted in the pending tier. Subsequent
1686 /// `lookup`/`enumerate` calls will skip the underlying captured
1687 /// entry, and `capture()` will fold the deletion into the new
1688 /// state's tree (pruning empty parent dirs as needed).
1689 ///
1690 /// Low-level (path-based) helper — unlike [`Self::unlink_entry`]
1691 /// it does not honour POSIX open-unlinked semantics. Used by
1692 /// tests that bypass the FUSE-callback lifecycle. The
1693 /// NodeId-keyed buffers for the path's current owner are dropped
1694 /// (no orphan tracking).
1695 pub fn unlink_path(&self, path: impl AsRef<Path>) -> Result<()> {
1696 let path = path.as_ref().to_path_buf();
1697 // Resolve path → NodeId via the inode registry so we can drop
1698 // the per-NodeId warm/hot bytes.
1699 let bound_id = {
1700 let inodes = self.inner.inodes.lock_or_poisoned();
1701 inodes.by_path.get(&path).copied()
1702 };
1703 let mut pending = self.inner.pending.lock_or_poisoned();
1704 if let Some(node_id) = pending.hot_by_path.remove(&path) {
1705 pending.hot.remove(&node_id);
1706 pending.warm.remove(&node_id);
1707 pending.state.remove(&node_id);
1708 }
1709 if let Some(node_id) = bound_id {
1710 pending.hot.remove(&node_id);
1711 pending.warm.remove(&node_id);
1712 pending.state.remove(&node_id);
1713 }
1714 pending.symlinks.remove(&path);
1715 pending.child_index.remove(&path);
1716 pending.tombstones.insert(path.clone());
1717 drop(pending);
1718 if bound_id.is_some() {
1719 let mut inodes = self.inner.inodes.lock_or_poisoned();
1720 inodes.by_path.remove(&path);
1721 }
1722 Ok(())
1723 }
1724
1725 // --- Write-side overlay ops (heddle#180) -----------------------------------
1726 //
1727 // Each method below corresponds to one FUSE callback the kernel
1728 // emits on cargo / git / npm style workloads:
1729 //
1730 // create → `create_file` open(O_CREAT)
1731 // mkdir → `make_dir`
1732 // unlink → `unlink_entry`
1733 // rmdir → `rmdir_entry`
1734 // rename → `rename_entry`
1735 // setattr → `set_attrs` chmod / ftruncate / O_TRUNC
1736 // symlink → `create_symlink`
1737 // readlink→ `read_link`
1738 //
1739 // All mutations land in the per-thread overlay (pending tier):
1740 //
1741 // * `Pending::hot` / `Pending::warm` — file bytes (existing).
1742 // * `Pending::tombstones` — file deletions (existing).
1743 // * `Pending::dir_tombstones` — `rmdir` of a captured dir.
1744 // * `Pending::explicit_dirs` — empty mkdirs.
1745 // * `Pending::symlinks` — link target bytes.
1746 //
1747 // None of these touch the underlying CAS until `capture()` folds
1748 // the overlay into a real heddle state.
1749
1750 /// Open-or-create a regular file under `parent`, mirroring
1751 /// `open(O_CREAT[|O_EXCL])` from userspace.
1752 ///
1753 /// When the named entry doesn't exist, mints a fresh
1754 /// [`NodeRecord::PendingFile`] inode + an empty hot buffer so the
1755 /// new path is immediately visible to [`lookup`](Self::lookup) /
1756 /// [`attrs`](Self::attrs) and the first
1757 /// [`write`](Self::write) drops cleanly into the existing
1758 /// two-tier model.
1759 ///
1760 /// When the named entry already exists:
1761 /// * `exclusive=true` ⇒ [`MountError::AlreadyExists`] (errno
1762 /// `EEXIST`).
1763 /// * `exclusive=false` ⇒ returns the existing entry. The kernel
1764 /// follows up with `setattr(size=0)` for `O_TRUNC` callers,
1765 /// which we honour in [`set_attrs`](Self::set_attrs).
1766 pub fn create_file(
1767 &self,
1768 parent: NodeId,
1769 name: &OsStr,
1770 mode: FileMode,
1771 exclusive: bool,
1772 ) -> Result<Entry> {
1773 // R8: serialize against rename / mkdir / symlink so an
1774 // exclusivity check (O_EXCL or rename-noreplace) lands its
1775 // existence-test and its mutation under the same write-side
1776 // critical section.
1777 let _write_guard = self.inner.write_mu.lock_or_poisoned();
1778 let name_str = validate_entry_name(name)?;
1779 if let Some(existing) = self.lookup(parent, name)? {
1780 if exclusive {
1781 return Err(MountError::AlreadyExists(name_str.to_string()));
1782 }
1783 return Ok(existing);
1784 }
1785 let parent_record = self.record_for(parent)?;
1786 let parent_path = self
1787 .dir_path_of(&parent_record)
1788 .ok_or_else(|| MountError::NotADirectory(format!("{parent_record:?}")))?;
1789 let child_path = join_child(&parent_path, name_str);
1790
1791 {
1792 let mut pending = self.inner.pending.lock_or_poisoned();
1793 // A prior unlink left a tombstone — clear it; the file
1794 // exists again.
1795 pending.tombstones.remove(&child_path);
1796 // An overlay-only directory used to live here; it's gone.
1797 pending.explicit_dirs.remove(&child_path);
1798 }
1799
1800 let node = self.intern(NodeRecord::PendingFile {
1801 path: child_path.clone(),
1802 mode,
1803 });
1804
1805 // Seed an empty hot buffer so the freshly-minted inode reads
1806 // as a 0-byte file even before any `write` callback fires.
1807 // Mirrors what userspace expects from `open(O_CREAT)`: the
1808 // file exists at length 0 immediately on return.
1809 {
1810 let mut pending = self.inner.pending.lock_or_poisoned();
1811 pending.hot.insert(
1812 node.0,
1813 HotBuffer {
1814 path: child_path.clone(),
1815 mode,
1816 bytes: Vec::new(),
1817 last_touched: Instant::now(),
1818 revision: 0,
1819 },
1820 );
1821 pending.hot_by_path.insert(child_path.clone(), node.0);
1822 pending.child_index.insert(
1823 child_path,
1824 PendingChildKind::HotFile {
1825 node,
1826 size: 0,
1827 mode,
1828 },
1829 );
1830 }
1831
1832 Ok(Entry {
1833 node,
1834 name: name.to_os_string(),
1835 kind: kind_for_mode(mode),
1836 size: 0,
1837 unix_mode: mode.to_unix_mode(),
1838 })
1839 }
1840
1841 /// Create an empty directory under `parent`. Recorded as an
1842 /// [`Pending::explicit_dirs`] entry so the new path is visible to
1843 /// lookup/enumerate even when no child has been written yet.
1844 pub fn make_dir(&self, parent: NodeId, name: &OsStr) -> Result<Entry> {
1845 // R8: serialize with other write-side mutations.
1846 let _write_guard = self.inner.write_mu.lock_or_poisoned();
1847 let name_str = validate_entry_name(name)?;
1848 if self.lookup(parent, name)?.is_some() {
1849 return Err(MountError::AlreadyExists(name_str.to_string()));
1850 }
1851 let parent_record = self.record_for(parent)?;
1852 let parent_path = self
1853 .dir_path_of(&parent_record)
1854 .ok_or_else(|| MountError::NotADirectory(format!("{parent_record:?}")))?;
1855 let child_path = join_child(&parent_path, name_str);
1856
1857 {
1858 let mut pending = self.inner.pending.lock_or_poisoned();
1859 // A rmdir of this exact path now reverts to "present".
1860 pending.dir_tombstones.remove(&child_path);
1861 // Clear any colliding file tombstone too.
1862 pending.tombstones.remove(&child_path);
1863 pending.explicit_dirs.insert(child_path.clone());
1864 pending
1865 .child_index
1866 .insert(child_path.clone(), PendingChildKind::Dir);
1867 }
1868
1869 let node = self.intern(NodeRecord::PendingDir { path: child_path });
1870 Ok(Entry {
1871 node,
1872 name: name.to_os_string(),
1873 kind: NodeKind::Directory,
1874 size: 0,
1875 unix_mode: DIR_UNIX_MODE,
1876 })
1877 }
1878
1879 /// Delete a regular file (or symlink) named `name` under `parent`.
1880 ///
1881 /// POSIX open-unlinked semantics: the directory entry goes (path
1882 /// tombstoned, `inodes.by_path[path]` retired), but if any fd
1883 /// still references the inode, the bytes survive in `hot[node]` /
1884 /// `warm[node]` until the final `release`. Under the post-spike
1885 /// unified NodeId-keyed model
1886 /// (`docs/design/mount-posix-semantics.md` §1.2 T1/T2), this is a
1887 /// state transition only — no byte migration. Pre-spike code
1888 /// dropped `pending.hot[node_id]` here (Codex thread 3293307302
1889 /// r9) and migrated `warm[path]` into `orphan_warm[node]` (r8);
1890 /// both steps go away.
1891 pub fn unlink_entry(&self, parent: NodeId, name: &OsStr) -> Result<()> {
1892 // R8: serialize with other write-side mutations.
1893 let _write_guard = self.inner.write_mu.lock_or_poisoned();
1894 let name_str = validate_entry_name(name)?;
1895 let entry = self
1896 .lookup(parent, name)?
1897 .ok_or_else(|| MountError::NotFound(name_str.to_string()))?;
1898 if entry.kind == NodeKind::Directory {
1899 return Err(MountError::IsADirectory(name_str.to_string()));
1900 }
1901 let parent_record = self.record_for(parent)?;
1902 let parent_path = self
1903 .dir_path_of(&parent_record)
1904 .ok_or_else(|| MountError::NotADirectory(format!("{parent_record:?}")))?;
1905 let child_path = join_child(&parent_path, name_str);
1906 let node_id = entry.node.0;
1907
1908 {
1909 let mut pending = self.inner.pending.lock_or_poisoned();
1910 // Detach the path-level hot binding. The bytes follow the
1911 // NodeId, so `hot[node_id]` / `warm[node_id]` stay put —
1912 // the surviving fd reads them via the orphan branches in
1913 // `read` / `attrs` / `write` / `apply_truncate`.
1914 // (r9 fix: pre-spike code called `pending.hot.remove(&node_id)`
1915 // here and the unflushed bytes vanished.)
1916 pending.hot_by_path.remove(&child_path);
1917 // Transition T1: Live{open_count >= 1} → Orphan{open_count}.
1918 // The witness-gated retrofit (heddle#209) makes the FSM
1919 // check the gate: `bp.transition_to_orphan(node_id)`
1920 // returns `None` (without touching `state`) for any
1921 // non-`LiveNonZero` state, and the missing
1922 // `Witness<Orphan>` IS the short-circuit at this call
1923 // site.
1924 //
1925 // That subsumes two earlier defensive checks: Codex r12
1926 // thread 3293510317 (symlinks have no `open`/`release`
1927 // lifecycle, so they never enter `state` and the
1928 // transition never fires for them), and r11 finding
1929 // 3293575534 (orphaning a `Live { open_count: 0 }` node
1930 // creates a record nothing will ever reap — same shape,
1931 // same fix).
1932 pending.with_brand(|bp| {
1933 let _ = bp.transition_to_orphan(node_id);
1934 });
1935 // Symlinks are path-keyed; their overlay goes when the
1936 // directory entry goes.
1937 pending.symlinks.remove(&child_path);
1938 pending.child_index.remove(&child_path);
1939 pending.tombstones.insert(child_path.clone());
1940 }
1941 // Retire the path→inode mapping so a subsequent `create_file`
1942 // at the same name mints a fresh inode (POSIX unlink/recreate
1943 // isolation — open-unlinked temp files must not be aliased by
1944 // a replacement at the same path). The `by_id` record stays so
1945 // any still-open kernel handle keeps resolving until `forget`.
1946 {
1947 let mut inodes = self.inner.inodes.lock_or_poisoned();
1948 inodes.by_path.remove(&child_path);
1949 }
1950 Ok(())
1951 }
1952
1953 /// Remove the empty directory `name` under `parent`. Fails with
1954 /// `ENOTEMPTY` if any child resolves through the mount.
1955 pub fn rmdir_entry(&self, parent: NodeId, name: &OsStr) -> Result<()> {
1956 // R8: serialize with other write-side mutations.
1957 let _write_guard = self.inner.write_mu.lock_or_poisoned();
1958 let name_str = validate_entry_name(name)?;
1959 let entry = self
1960 .lookup(parent, name)?
1961 .ok_or_else(|| MountError::NotFound(name_str.to_string()))?;
1962 if entry.kind != NodeKind::Directory {
1963 return Err(MountError::NotADirectory(name_str.to_string()));
1964 }
1965 // Empty check via enumerate — already overlay-aware (hot,
1966 // warm, symlinks, captured-with-pending-overlay).
1967 let children = self.enumerate(entry.node)?;
1968 if !children.is_empty() {
1969 return Err(MountError::NotEmpty(name_str.to_string()));
1970 }
1971 let parent_record = self.record_for(parent)?;
1972 let parent_path = self
1973 .dir_path_of(&parent_record)
1974 .ok_or_else(|| MountError::NotADirectory(format!("{parent_record:?}")))?;
1975 let child_path = join_child(&parent_path, name_str);
1976
1977 {
1978 let mut pending = self.inner.pending.lock_or_poisoned();
1979 pending.explicit_dirs.remove(&child_path);
1980 pending.child_index.remove(&child_path);
1981 pending.dir_tombstones.insert(child_path.clone());
1982 }
1983 // Codex r12 thread 3293510310 (P1): retire the path → inode
1984 // mapping. Otherwise `Inodes::intern` would coalesce a
1985 // subsequent `create_file` / `make_dir` at this path onto the
1986 // removed directory's NodeId, rebinding a cached directory
1987 // inode to a different object type — the same stale-handle
1988 // class `unlink_entry` already guards against. The `by_id`
1989 // record stays so any kernel handle the FS still holds keeps
1990 // resolving until `forget`.
1991 {
1992 let mut inodes = self.inner.inodes.lock_or_poisoned();
1993 inodes.by_path.remove(&child_path);
1994 }
1995 Ok(())
1996 }
1997
1998 /// Move `(old_parent, old_name)` to `(new_parent, new_name)`.
1999 /// Handles file + symlink renames across any pair of overlay /
2000 /// captured paths, and overlay-only directory rename (a captured
2001 /// directory rename would require recursively rewriting the
2002 /// tombstone/warm map — out of scope for the cargo / git path).
2003 pub fn rename_entry(
2004 &self,
2005 old_parent: NodeId,
2006 old_name: &OsStr,
2007 new_parent: NodeId,
2008 new_name: &OsStr,
2009 ) -> Result<()> {
2010 self.rename_entry_with_options(
2011 old_parent,
2012 old_name,
2013 new_parent,
2014 new_name,
2015 RenameOptions::default(),
2016 )
2017 }
2018
2019 /// Same as [`Self::rename_entry`] but honours [`RenameOptions`].
2020 /// `no_replace` (Linux `RENAME_NOREPLACE`) refuses the rename when
2021 /// the destination already resolves; the check is performed inside
2022 /// the same write-side critical section as the mutation, so a
2023 /// concurrent writer cannot install the destination between the
2024 /// check and the rename.
2025 pub fn rename_entry_with_options(
2026 &self,
2027 old_parent: NodeId,
2028 old_name: &OsStr,
2029 new_parent: NodeId,
2030 new_name: &OsStr,
2031 options: RenameOptions,
2032 ) -> Result<()> {
2033 // R8 (Codex Thread 3293235163): the existence-check + the
2034 // directory-entry mutation must land under the same mutation
2035 // lock. Holding `write_mu` for the duration of this method
2036 // serializes the rename against every other write-side op
2037 // that could install the destination (create_file, make_dir,
2038 // create_symlink, another rename) — that's the atomicity the
2039 // POSIX NOREPLACE flag promises.
2040 let _write_guard = self.inner.write_mu.lock_or_poisoned();
2041
2042 let old_name_str = validate_entry_name(old_name)?;
2043 let new_name_str = validate_entry_name(new_name)?;
2044 let src = self
2045 .lookup(old_parent, old_name)?
2046 .ok_or_else(|| MountError::NotFound(format!("rename src {old_name_str}")))?;
2047 let old_parent_record = self.record_for(old_parent)?;
2048 let new_parent_record = self.record_for(new_parent)?;
2049 let old_parent_path = self
2050 .dir_path_of(&old_parent_record)
2051 .ok_or_else(|| MountError::NotADirectory(format!("{old_parent_record:?}")))?;
2052 let new_parent_path = self
2053 .dir_path_of(&new_parent_record)
2054 .ok_or_else(|| MountError::NotADirectory(format!("{new_parent_record:?}")))?;
2055 let old_path = join_child(&old_parent_path, old_name_str);
2056 let new_path = join_child(&new_parent_path, new_name_str);
2057 if old_path == new_path {
2058 return Ok(());
2059 }
2060
2061 // POSIX: destination of a different kind is an error. We also
2062 // honour NOREPLACE here while still holding `write_mu` so the
2063 // check + the subsequent move are atomic against concurrent
2064 // writers. `dst` is shadowed for the kind-mismatch arm and
2065 // hoisted into `displaced_inode_id` so the move primitives
2066 // can preserve the displaced inode's warm bytes (r8).
2067 let dst = self.lookup(new_parent, new_name)?;
2068 if dst.is_some() && options.no_replace {
2069 return Err(MountError::AlreadyExists(new_name_str.to_string()));
2070 }
2071 if let Some(ref d) = dst {
2072 match (src.kind, d.kind) {
2073 (NodeKind::Directory, NodeKind::Directory) => {
2074 let dst_children = self.enumerate(d.node)?;
2075 if !dst_children.is_empty() {
2076 return Err(MountError::NotEmpty(new_name_str.to_string()));
2077 }
2078 }
2079 (NodeKind::Directory, _) => {
2080 return Err(MountError::NotADirectory(new_name_str.to_string()));
2081 }
2082 (_, NodeKind::Directory) => {
2083 return Err(MountError::IsADirectory(new_name_str.to_string()));
2084 }
2085 _ => {}
2086 }
2087 }
2088 let displaced_inode_id = dst.as_ref().map(|d| d.node.0);
2089
2090 match src.kind {
2091 NodeKind::File => self.move_file(&old_path, &new_path, displaced_inode_id)?,
2092 NodeKind::Symlink => self.move_symlink(&old_path, &new_path, displaced_inode_id)?,
2093 NodeKind::Directory => self.move_overlay_dir(&old_path, &new_path)?,
2094 }
2095 // Maintain the inode↔path invariant for both the source and
2096 // destination: the kernel may have cached either dentry from
2097 // a prior lookup, and (for FUSE) it does not re-issue lookup
2098 // after `rename` — it just rewrites its own dentry → inode
2099 // table. So the source inode now resolves through dentry
2100 // `new_name`, and any read against it must serve the new
2101 // path's overlay state. Rewriting the source record's stored
2102 // path is what keeps that consistent. The dest's old inode
2103 // (which the kernel will issue `forget` for) gets dropped
2104 // from `by_path` so the next lookup mints a fresh id.
2105 {
2106 let mut inodes = self.inner.inodes.lock_or_poisoned();
2107 // Detach the destination's path mapping. The inode record
2108 // stays in `by_id` so any kernel handle the FS still holds
2109 // for the replaced file keeps resolving (the kernel cleans
2110 // up via `forget` on close). POSIX semantics: rename-over
2111 // must not invalidate an already-open dest descriptor.
2112 let displaced_dest = inodes.by_path.remove(&new_path);
2113 // Rewrite the source inode's stored path so subsequent
2114 // reads/attrs against it serve the new-path overlay.
2115 // The kernel keeps using the source's NodeId after rename
2116 // (it's just a dentry-table rewrite on its side) — without
2117 // this, every read against the rebased dentry sees the
2118 // stale path and returns ESTALE.
2119 let rebased_src = if let Some(src_id) = inodes.by_path.remove(&old_path) {
2120 if let Some(
2121 NodeRecord::PendingFile { path, .. }
2122 | NodeRecord::File { path, .. }
2123 | NodeRecord::Gitlink { path, .. }
2124 | NodeRecord::PendingSymlink { path }
2125 | NodeRecord::Dir { path, .. }
2126 | NodeRecord::PendingDir { path },
2127 ) = inodes.by_id.get_mut(&src_id)
2128 {
2129 *path = new_path.clone();
2130 }
2131 inodes.by_path.insert(new_path.clone(), src_id);
2132 // For a directory rename, also rebase every cached
2133 // descendant inode. The kernel may already hold dentry
2134 // → inode bindings for `old_path/<child>` from prior
2135 // lookups, and reads against those inodes would
2136 // otherwise resolve through the stale path (ESTALE on
2137 // PendingFile, or the wrong overlay on File). Walk
2138 // by_path once, collect the entries under the old
2139 // prefix, then rewrite both the mapping and the
2140 // NodeRecord's stored path.
2141 if src.kind == NodeKind::Directory {
2142 let descendants: Vec<(PathBuf, PathBuf, u64)> = inodes
2143 .by_path
2144 .iter()
2145 .filter_map(|(p, id)| {
2146 let tail = p.strip_prefix(&old_path).ok()?;
2147 if tail.as_os_str().is_empty() {
2148 return None;
2149 }
2150 Some((p.clone(), new_path.join(tail), *id))
2151 })
2152 .collect();
2153 for (old_key, new_key, id) in descendants {
2154 inodes.by_path.remove(&old_key);
2155 if let Some(
2156 NodeRecord::PendingFile { path, .. }
2157 | NodeRecord::File { path, .. }
2158 | NodeRecord::Gitlink { path, .. }
2159 | NodeRecord::PendingSymlink { path }
2160 | NodeRecord::Dir { path, .. }
2161 | NodeRecord::PendingDir { path },
2162 ) = inodes.by_id.get_mut(&id)
2163 {
2164 *path = new_key.clone();
2165 }
2166 inodes.by_path.insert(new_key, id);
2167 }
2168 }
2169 Some(src_id)
2170 } else {
2171 None
2172 };
2173 drop(inodes);
2174 // Reach into pending for two cleanups under one lock:
2175 // * The source's hot buffer (if any) carries the old
2176 // path; rebase it. Descendant hot-buffer paths are
2177 // already handled by `move_overlay_dir`'s
2178 // `hot_path_updates` pass.
2179 // * The displaced destination (if any) becomes an
2180 // orphan: its directory entry is gone but the inode
2181 // id may still be held by a kernel fd. Subsequent
2182 // `write` / `apply_truncate` / `set_attrs` /
2183 // `read` / `attrs` calls through that fd consult
2184 // `Pending::orphans` and take the per-NodeId branch
2185 // instead of the rebased path overlay. The companion
2186 // orphan branch in `flush_node` drops any preserved
2187 // buffer without warm-promoting.
2188 let mut pending = self.inner.pending.lock_or_poisoned();
2189 if let Some(src_id) = rebased_src
2190 && let Some(buf) = pending.hot.get_mut(&src_id)
2191 {
2192 buf.path = new_path.clone();
2193 buf.revision = buf.revision.wrapping_add(1);
2194 }
2195 match src.kind {
2196 NodeKind::Directory => pending.child_index.rebase_prefix(&old_path, &new_path),
2197 NodeKind::File => {
2198 let moved = pending.child_index.remove(&old_path);
2199 pending.child_index.remove(&new_path);
2200 let indexed = moved.or_else(|| {
2201 let id = rebased_src?;
2202 if let Some(buf) = pending.hot.get(&id) {
2203 Some(PendingChildKind::HotFile {
2204 node: NodeId(id),
2205 size: buf.bytes.len() as u64,
2206 mode: buf.mode,
2207 })
2208 } else {
2209 pending
2210 .warm
2211 .get(&id)
2212 .map(|entry| PendingChildKind::WarmFile {
2213 size: entry.size,
2214 mode: entry.mode,
2215 })
2216 }
2217 });
2218 if let Some(kind) = indexed {
2219 pending.child_index.insert(new_path.clone(), kind);
2220 }
2221 }
2222 NodeKind::Symlink => {
2223 pending.child_index.remove(&old_path);
2224 pending.child_index.remove(&new_path);
2225 if let Some(size) = pending
2226 .symlinks
2227 .get(&new_path)
2228 .map(|target| target.len() as u64)
2229 {
2230 pending
2231 .child_index
2232 .insert(new_path.clone(), PendingChildKind::Symlink { size });
2233 }
2234 }
2235 }
2236 if let Some(dest_id) = displaced_dest {
2237 // T3: the displaced destination transitions to Orphan
2238 // iff it's currently `Live { open_count >= 1 }`. Bytes
2239 // (hot[dest_id], warm[dest_id]) stay put so the
2240 // surviving fd keeps reading the inode's own data
2241 // (spike doc §1.2 T3).
2242 //
2243 // Closes Codex PR #182 r11 finding 3293575541 (heddle
2244 // #209): `bp.transition_to_orphan(dest_id)` returns
2245 // `None` (without touching `state`) for any
2246 // non-`LiveNonZero` displaced destination, and the
2247 // missing `Witness<Orphan>` IS the short-circuit at
2248 // this call site. Pre-retrofit this branch
2249 // unconditionally inserted `Orphan { open_count: 0 }`
2250 // for non-`Live` destinations — including symlinks,
2251 // which have no `open`/`release` lifecycle and would
2252 // never reap the entry, growing `state` under symlink
2253 // churn until capture / invalidate.
2254 pending.with_brand(|bp| {
2255 let _ = bp.transition_to_orphan(dest_id);
2256 });
2257 }
2258 }
2259 Ok(())
2260 }
2261
2262 /// Rename a regular file. Under the post-spike unified
2263 /// NodeId-keyed model
2264 /// (`docs/design/mount-posix-semantics.md` §2.4), the source's
2265 /// bytes follow its NodeId — no byte migration step. The displaced
2266 /// destination keeps its own `hot[id]` / `warm[id]` so the
2267 /// surviving fd reads its own data. The work here is path-level:
2268 /// retire the destination's path-keyed hot binding, rebase the
2269 /// source's hot buffer's `path` field (so a subsequent `flush`
2270 /// promotes under the new path), seed warm if the source had only
2271 /// captured-tree bytes (so capture can plant the file at the new
2272 /// path), and tombstone the old path.
2273 ///
2274 /// `displaced_inode_id` is no longer used as a side-channel for
2275 /// byte preservation — the caller (`rename_entry_with_options`)
2276 /// handles the orphan state transition independently.
2277 fn move_file(
2278 &self,
2279 old_path: &Path,
2280 new_path: &Path,
2281 displaced_inode_id: Option<u64>,
2282 ) -> Result<()> {
2283 // Snapshot whether the source has a hot buffer (drain it to
2284 // warm so the warm tier becomes authoritative for capture
2285 // under the new path) and whether the source is captured-only
2286 // (then synthesize a warm entry so capture plants the bytes
2287 // at new_path).
2288 let src_id_opt = self
2289 .inner
2290 .pending
2291 .lock_or_poisoned()
2292 .hot_by_path
2293 .get(old_path)
2294 .copied();
2295 if let Some(id) = src_id_opt {
2296 self.flush_node(NodeId(id))?;
2297 }
2298 // After the flush, the source's bytes (if any) live in
2299 // `warm[src_id]`. If the source had no warm entry — captured
2300 // only — synthesize one keyed by the source's NodeId so
2301 // capture-time tree fold plants the file under new_path. We
2302 // resolve src_id via the path → inode reverse-index (or via
2303 // the captured-tree walk for a captured-only source).
2304 let src_id = {
2305 let inodes = self.inner.inodes.lock_or_poisoned();
2306 inodes.by_path.get(old_path).copied()
2307 };
2308 let needs_synth = match src_id {
2309 Some(id) => !self.inner.pending.lock_or_poisoned().warm.contains_key(&id),
2310 None => true,
2311 };
2312 let captured_seed = if needs_synth {
2313 // Captured-only source: pull (blob, mode, size) from the
2314 // captured tree so the rename survives `capture`.
2315 Some(self.captured_file_at(old_path)?)
2316 } else {
2317 None
2318 };
2319
2320 let mut pending = self.inner.pending.lock_or_poisoned();
2321 // Detach the destination's path-keyed hot binding. The
2322 // displaced inode's bytes are keyed by NodeId — they stay put
2323 // for the surviving fd. POSIX rename-over: open destination
2324 // descriptors keep referencing the displaced inode until close.
2325 pending.hot_by_path.remove(new_path);
2326 // Symlinks are path-keyed; clear at both endpoints.
2327 pending.symlinks.remove(new_path);
2328 pending.symlinks.remove(old_path);
2329 // Source: if a hot buffer survived the flush above (only
2330 // possible if the source was Orphan, which can't happen for
2331 // a valid rename source — but be defensive), rebase its
2332 // path-binding.
2333 if let Some(id) = pending.hot_by_path.remove(old_path) {
2334 if let Some(buf) = pending.hot.get_mut(&id) {
2335 buf.path = new_path.to_path_buf();
2336 buf.revision = buf.revision.wrapping_add(1);
2337 }
2338 pending.hot_by_path.insert(new_path.to_path_buf(), id);
2339 }
2340 // Captured-only source: synthesize a warm entry so capture
2341 // plants the bytes at new_path. The entry is keyed by the
2342 // source's NodeId; capture-time tree fold resolves its
2343 // current path through `inodes.by_path`.
2344 if let (Some(id), Some((blob, mode, size))) = (src_id, captured_seed) {
2345 pending.warm.insert(id, PendingEntry { blob, mode, size });
2346 }
2347 // Path-level bookkeeping: tombstone old_path so the captured
2348 // tree's old entry is hidden; clear any tombstone at
2349 // new_path (rename made it valid again).
2350 pending.tombstones.insert(old_path.to_path_buf());
2351 pending.tombstones.remove(new_path);
2352 // The displaced inode is handled by the caller via the
2353 // NodeState transition; no byte work here.
2354 let _ = displaced_inode_id;
2355 Ok(())
2356 }
2357
2358 fn move_symlink(
2359 &self,
2360 old_path: &Path,
2361 new_path: &Path,
2362 displaced_inode_id: Option<u64>,
2363 ) -> Result<()> {
2364 // Resolve target bytes from the pending overlay or the
2365 // captured-tree blob — symlinks are path-keyed (not openable
2366 // for IO; no orphan story applies).
2367 let target_bytes = {
2368 let pending = self.inner.pending.lock_or_poisoned();
2369 pending.symlinks.get(old_path).cloned()
2370 };
2371 let target_bytes = match target_bytes {
2372 Some(b) => b,
2373 None => {
2374 let blob = self.captured_symlink_at(old_path)?;
2375 (*self.load_blob_bytes(&blob)?).to_vec()
2376 }
2377 };
2378 let mut pending = self.inner.pending.lock_or_poisoned();
2379 // Detach the displaced destination's path-keyed hot binding.
2380 // Its NodeId-keyed bytes stay put for the surviving fd; the
2381 // caller's NodeState transition handles the orphan tracking.
2382 pending.hot_by_path.remove(new_path);
2383 pending.symlinks.remove(new_path);
2384 pending.symlinks.remove(old_path);
2385 pending
2386 .symlinks
2387 .insert(new_path.to_path_buf(), target_bytes);
2388 pending.tombstones.remove(new_path);
2389 pending.tombstones.insert(old_path.to_path_buf());
2390 let _ = displaced_inode_id;
2391 Ok(())
2392 }
2393
2394 fn move_overlay_dir(&self, old_path: &Path, new_path: &Path) -> Result<()> {
2395 // We only support overlay-only directory renames here. If the
2396 // source dir has any captured-tree backing, refuse — a full
2397 // captured-tree rename would need to rewrite every descendant
2398 // tombstone entry.
2399 if self.captured_dir_exists(old_path)? {
2400 return Err(MountError::InvalidArgument(format!(
2401 "cross-tree directory rename {} → {} not supported by the overlay",
2402 old_path.display(),
2403 new_path.display()
2404 )));
2405 }
2406 let mut pending = self.inner.pending.lock_or_poisoned();
2407 // Path-keyed structures under `old_path/` need to be rebased
2408 // to `new_path/`. Warm bytes follow the NodeId (unified shape)
2409 // so warm[id] is unaffected by this rewrite — descendant
2410 // NodeRecord paths get rebased in `rename_entry_with_options`.
2411 fn rebase(p: &Path, old: &Path, new: &Path) -> Option<PathBuf> {
2412 let tail = p.strip_prefix(old).ok()?;
2413 Some(new.join(tail))
2414 }
2415 let mut new_explicit: BTreeSet<PathBuf> = BTreeSet::new();
2416 let mut new_symlinks: BTreeMap<PathBuf, Vec<u8>> = BTreeMap::new();
2417 let mut new_tombstones: BTreeSet<PathBuf> = BTreeSet::new();
2418 let mut new_hot_by_path: BTreeMap<PathBuf, u64> = BTreeMap::new();
2419 let mut hot_path_updates: Vec<(u64, PathBuf)> = Vec::new();
2420 for explicit in std::mem::take(&mut pending.explicit_dirs) {
2421 match rebase(&explicit, old_path, new_path) {
2422 Some(rebased) => {
2423 new_explicit.insert(rebased);
2424 }
2425 None => {
2426 if explicit != old_path {
2427 new_explicit.insert(explicit);
2428 }
2429 }
2430 }
2431 }
2432 for (path, target) in std::mem::take(&mut pending.symlinks) {
2433 match rebase(&path, old_path, new_path) {
2434 Some(rebased) => {
2435 new_symlinks.insert(rebased, target);
2436 }
2437 None => {
2438 new_symlinks.insert(path, target);
2439 }
2440 }
2441 }
2442 for path in std::mem::take(&mut pending.tombstones) {
2443 match rebase(&path, old_path, new_path) {
2444 Some(rebased) => {
2445 new_tombstones.insert(rebased);
2446 }
2447 None => {
2448 new_tombstones.insert(path);
2449 }
2450 }
2451 }
2452 for (path, id) in std::mem::take(&mut pending.hot_by_path) {
2453 match rebase(&path, old_path, new_path) {
2454 Some(rebased) => {
2455 hot_path_updates.push((id, rebased.clone()));
2456 new_hot_by_path.insert(rebased, id);
2457 }
2458 None => {
2459 new_hot_by_path.insert(path, id);
2460 }
2461 }
2462 }
2463 // Rewrite hot-buffer path fields to match.
2464 for (id, new_p) in hot_path_updates {
2465 if let Some(buf) = pending.hot.get_mut(&id) {
2466 buf.path = new_p;
2467 buf.revision = buf.revision.wrapping_add(1);
2468 }
2469 }
2470 // Ensure the destination directory itself is registered.
2471 new_explicit.insert(new_path.to_path_buf());
2472 pending.explicit_dirs = new_explicit;
2473 pending.symlinks = new_symlinks;
2474 pending.tombstones = new_tombstones;
2475 pending.hot_by_path = new_hot_by_path;
2476 Ok(())
2477 }
2478
2479 /// Resolve a captured-tree file at `path`; returns its
2480 /// `(blob, mode, size)`. Errors with `NotFound` if no captured
2481 /// entry exists.
2482 fn captured_file_at(&self, path: &Path) -> Result<(ContentHash, FileMode, u64)> {
2483 let entry = self.captured_tree_entry(path)?;
2484 let Some(hash) = entry.blob_hash() else {
2485 return Err(MountError::InvalidArgument(format!(
2486 "{} is not a mutable file in the captured tree",
2487 path.display()
2488 )));
2489 };
2490 let mode = entry.mode();
2491 let size = self.blob_size(&hash)?;
2492 Ok((hash, mode, size))
2493 }
2494
2495 fn captured_symlink_at(&self, path: &Path) -> Result<ContentHash> {
2496 let entry = self.captured_tree_entry(path)?;
2497 let Some(hash) = entry.symlink_hash() else {
2498 return Err(MountError::InvalidArgument(format!(
2499 "{} is not a symlink in the captured tree",
2500 path.display()
2501 )));
2502 };
2503 Ok(hash)
2504 }
2505
2506 fn captured_tree_entry(&self, path: &Path) -> Result<TreeEntry> {
2507 let root_record = self.record_for(NodeId::ROOT)?;
2508 let mut tree = self.tree_for_record(&root_record)?;
2509 let comps: Vec<&str> = path
2510 .components()
2511 .filter_map(|c| match c {
2512 Component::Normal(n) => n.to_str(),
2513 _ => None,
2514 })
2515 .collect();
2516 let (leaf, dirs) = comps
2517 .split_last()
2518 .ok_or_else(|| MountError::NotFound(path.display().to_string()))?;
2519 for d in dirs {
2520 let e = tree
2521 .get(d)
2522 .ok_or_else(|| MountError::NotFound(path.display().to_string()))?;
2523 if !e.is_tree() {
2524 return Err(MountError::NotADirectory(d.to_string()));
2525 }
2526 let Some(hash) = e.tree_hash() else {
2527 return Err(MountError::NotADirectory(d.to_string()));
2528 };
2529 tree = self.load_tree(&hash)?;
2530 }
2531 let entry = tree
2532 .get(leaf)
2533 .cloned()
2534 .ok_or_else(|| MountError::NotFound(path.display().to_string()))?;
2535 Ok(entry)
2536 }
2537
2538 fn captured_dir_exists(&self, path: &Path) -> Result<bool> {
2539 match self.captured_tree_entry(path) {
2540 Ok(e) => Ok(e.is_tree()),
2541 Err(MountError::NotFound(_)) => Ok(false),
2542 Err(e) => Err(e),
2543 }
2544 }
2545
2546 /// Apply attribute updates from a FUSE `setattr` / FSKit
2547 /// `setattr` / etc. Returns post-update [`Attrs`] for an
2548 /// inline reply.
2549 pub fn set_attrs(&self, node: NodeId, update: AttrUpdate) -> Result<Attrs> {
2550 // Codex r13 thread 3293733165 (P1): every mutating branch of
2551 // `set_attrs` must serialize against `rename` / `create` /
2552 // `unlink` / `rmdir` under `write_mu`. Without it, a
2553 // `setattr(size=...)` racing with a `rename` re-uses the
2554 // pre-rename pathname in `apply_truncate`'s phase-2
2555 // bookkeeping — `tombstones.remove(old)` clears the rename's
2556 // tombstone and `hot_by_path.insert(old, node)` resurrects
2557 // the file at the old name. The mode-mutation branch has the
2558 // same shape (touches `hot_by_path[path]` / `warm[id]` derived
2559 // from `inodes.by_path[path]`), so we hold the lock for the
2560 // whole mutating prologue.
2561 let _write_guard = self.inner.write_mu.lock_or_poisoned();
2562
2563 // Mode mutation: only meaningful for file-kind records.
2564 if let Some(raw_mode) = update.mode {
2565 // Codex r13 thread 3293733164 (P2): the Normal↔Executable
2566 // fold is gated on the user execute bit (S_IXUSR = 0o100)
2567 // only, not on any of the three execute bits. A
2568 // `chmod 0o010` (group execute only) must leave the record
2569 // as Normal — otherwise capture would persist a
2570 // `FileMode::Executable` and grant owner+other execute
2571 // bits the agent never requested.
2572 let new_mode = if (raw_mode & 0o100) != 0 {
2573 FileMode::Executable
2574 } else {
2575 FileMode::Normal
2576 };
2577 let mut inodes = self.inner.inodes.lock_or_poisoned();
2578 if let Some(NodeRecord::File { mode, .. } | NodeRecord::PendingFile { mode, .. }) =
2579 inodes.by_id.get_mut(&node.0)
2580 {
2581 *mode = new_mode;
2582 }
2583 drop(inodes);
2584 // Reflect the mode in any open hot buffer + warm-tier
2585 // entry so a subsequent `capture` keeps the new mode.
2586 let record = self.record_for(node)?;
2587 if let Some(path) = match &record {
2588 NodeRecord::File { path, .. } | NodeRecord::PendingFile { path, .. } => Some(path),
2589 _ => None,
2590 } {
2591 let path = path.clone();
2592 let mut pending = self.inner.pending.lock_or_poisoned();
2593 // Always flip the per-NodeId buffer's mode — that's
2594 // the orphan's own bookkeeping when fd-based, and
2595 // the live buffer for non-orphan callers.
2596 if let Some(buf) = pending.hot.get_mut(&node.0) {
2597 buf.mode = new_mode;
2598 buf.revision = buf.revision.wrapping_add(1);
2599 }
2600 // Orphan branch: `unlink_entry` / `rename_entry`
2601 // recorded this NodeId because the kernel still
2602 // holds an fd to it, but the directory entry is
2603 // gone (or rebound to a sibling). POSIX is explicit:
2604 // an fd-based attribute change applies only to the
2605 // file referenced by that fd. Touching
2606 // `hot_by_path[path]` would mutate the fresh inode
2607 // now living at the same name; touching
2608 // `warm[path]` would land the change on the sibling
2609 // at capture time.
2610 if !pending.is_orphan(node.0) {
2611 if let Some(other_id) = pending.hot_by_path.get(&path).copied()
2612 && let Some(buf) = pending.hot.get_mut(&other_id)
2613 {
2614 buf.mode = new_mode;
2615 buf.revision = buf.revision.wrapping_add(1);
2616 }
2617 // Warm is NodeId-keyed: rebind via inodes if the
2618 // path still resolves Live to a tracked NodeId.
2619 let warm_id = {
2620 let inodes = self.inner.inodes.lock_or_poisoned();
2621 inodes.by_path.get(&path).copied()
2622 };
2623 if let Some(id) = warm_id
2624 && let Some(entry) = pending.warm.get_mut(&id)
2625 {
2626 entry.mode = new_mode;
2627 }
2628 pending.child_index.update_file_mode(&path, new_mode);
2629 }
2630 }
2631 }
2632
2633 // Size mutation: O_TRUNC, ftruncate, etc.
2634 if let Some(new_size) = update.size {
2635 self.apply_truncate(node, new_size)?;
2636 }
2637 // uid/gid/mtime: accepted as no-ops. The overlay doesn't carry
2638 // per-node ownership / timestamps yet (capture re-derives both
2639 // from the agent's principal + mount mtime).
2640 self.attrs(node)
2641 }
2642
2643 fn apply_truncate(&self, node: NodeId, new_size: u64) -> Result<()> {
2644 let new_size = validate_truncate_size(new_size)?;
2645 let record = self.record_for(node)?;
2646 let (path, mode, captured_blob) = match &record {
2647 NodeRecord::File {
2648 path, mode, blob, ..
2649 } => (path.clone(), *mode, Some(*blob)),
2650 NodeRecord::PendingFile { path, mode } => (path.clone(), *mode, None),
2651 _ => {
2652 return Err(MountError::IsADirectory(format!(
2653 "setattr(size) on non-file {record:?}"
2654 )));
2655 }
2656 };
2657
2658 // Phase 1: under the lock, decide whether a buffer already
2659 // exists (resize in place), and otherwise record orphan-ness
2660 // + the seed source. Drop the lock for the CAS read.
2661 //
2662 // POSIX `ftruncate` on an open-unlinked / rename-displaced fd
2663 // (an orphan in our terminology) must touch only the
2664 // anonymous open inode. The orphan branch never resizes a
2665 // sibling buffer at the rebased path, never seeds from
2666 // `warm[path]` (now owned by the sibling), and in Phase 2
2667 // never republishes `hot_by_path[path]` nor clears the
2668 // tombstone.
2669 enum Phase1 {
2670 ResizedInPlace,
2671 NeedSeed {
2672 orphan: bool,
2673 seed: Option<ContentHash>,
2674 },
2675 }
2676 let phase1 = {
2677 // Resolve the path's current Live NodeId via the inode
2678 // registry — under the unified shape `warm` is
2679 // NodeId-keyed, and the Live owner of `path` is the
2680 // sibling we'd seed from when no per-inode buffer exists.
2681 let path_owner = {
2682 let inodes = self.inner.inodes.lock_or_poisoned();
2683 inodes.by_path.get(&path).copied()
2684 };
2685 let mut pending = self.inner.pending.lock_or_poisoned();
2686 let orphan = pending.is_orphan(node.0);
2687 let id = if pending.hot.contains_key(&node.0) {
2688 Some(node.0)
2689 } else if orphan {
2690 // Never resize a sibling buffer through the orphan
2691 // fd — that buffer belongs to a fresh inode at the
2692 // rebound name.
2693 None
2694 } else {
2695 pending.hot_by_path.get(&path).copied()
2696 };
2697 if let Some(id) = id
2698 && let Some(buf) = pending.hot.get_mut(&id)
2699 {
2700 buf.bytes.resize(new_size, 0);
2701 buf.last_touched = Instant::now();
2702 buf.revision = buf.revision.wrapping_add(1);
2703 let mode = buf.mode;
2704 if !orphan {
2705 pending.child_index.insert(
2706 path.clone(),
2707 PendingChildKind::HotFile {
2708 node: NodeId(id),
2709 size: new_size as u64,
2710 mode,
2711 },
2712 );
2713 }
2714 Phase1::ResizedInPlace
2715 } else {
2716 let seed = if orphan {
2717 // Orphan: only the inode's pre-displacement
2718 // content is valid. Under the unified shape its
2719 // own warm bytes live at `warm[node.0]`; fall
2720 // back to the captured blob (this inode's own,
2721 // not the sibling at the rebound name).
2722 pending.warm.get(&node.0).map(|e| e.blob).or(captured_blob)
2723 } else {
2724 // Live: the path's bytes live at `warm[id]` where
2725 // id is the Live owner via `inodes.by_path`.
2726 path_owner
2727 .and_then(|id| pending.warm.get(&id).map(|e| e.blob))
2728 .or(captured_blob)
2729 };
2730 Phase1::NeedSeed { orphan, seed }
2731 }
2732 };
2733 let (orphan, seed_blob) = match phase1 {
2734 Phase1::ResizedInPlace => return Ok(()),
2735 Phase1::NeedSeed { orphan, seed } => (orphan, seed),
2736 };
2737
2738 let mut bytes = match seed_blob {
2739 Some(hash) => (*self.load_blob_bytes(&hash)?).to_vec(),
2740 None => Vec::new(),
2741 };
2742 bytes.resize(new_size, 0);
2743 let mut pending = self.inner.pending.lock_or_poisoned();
2744 if orphan {
2745 // Per-NodeId buffer only. Skip the tombstone-clear and
2746 // the `hot_by_path` rebind — the directory entry must
2747 // stay gone (open-unlinked) or stay rebound to the
2748 // sibling (rename-over). The companion orphan branch in
2749 // `flush_node` drops this buffer on release without
2750 // warm-promoting it.
2751 pending.hot.insert(
2752 node.0,
2753 HotBuffer {
2754 path,
2755 mode,
2756 bytes,
2757 last_touched: Instant::now(),
2758 revision: 0,
2759 },
2760 );
2761 } else {
2762 pending.tombstones.remove(&path);
2763 pending.hot.insert(
2764 node.0,
2765 HotBuffer {
2766 path: path.clone(),
2767 mode,
2768 bytes,
2769 last_touched: Instant::now(),
2770 revision: 0,
2771 },
2772 );
2773 pending.hot_by_path.insert(path.clone(), node.0);
2774 pending.child_index.insert(
2775 path,
2776 PendingChildKind::HotFile {
2777 node,
2778 size: new_size as u64,
2779 mode,
2780 },
2781 );
2782 }
2783 Ok(())
2784 }
2785
2786 /// Create a symbolic link under `parent`. Target bytes are kept
2787 /// in the pending tier verbatim; `capture` writes them as a CAS
2788 /// blob and emits a `Symlink` tree entry.
2789 pub fn create_symlink(&self, parent: NodeId, name: &OsStr, target: &Path) -> Result<Entry> {
2790 // R8: serialize with other write-side mutations.
2791 let _write_guard = self.inner.write_mu.lock_or_poisoned();
2792 let name_str = validate_entry_name(name)?;
2793 if self.lookup(parent, name)?.is_some() {
2794 return Err(MountError::AlreadyExists(name_str.to_string()));
2795 }
2796 let parent_record = self.record_for(parent)?;
2797 let parent_path = self
2798 .dir_path_of(&parent_record)
2799 .ok_or_else(|| MountError::NotADirectory(format!("{parent_record:?}")))?;
2800 let child_path = join_child(&parent_path, name_str);
2801 let target_bytes = target.as_os_str().as_encoded_bytes().to_vec();
2802 let target_len = target_bytes.len() as u64;
2803
2804 {
2805 let mut pending = self.inner.pending.lock_or_poisoned();
2806 pending.tombstones.remove(&child_path);
2807 pending.symlinks.insert(child_path.clone(), target_bytes);
2808 pending.child_index.insert(
2809 child_path.clone(),
2810 PendingChildKind::Symlink { size: target_len },
2811 );
2812 }
2813 let node = self.intern(NodeRecord::PendingSymlink { path: child_path });
2814 Ok(Entry {
2815 node,
2816 name: name.to_os_string(),
2817 kind: NodeKind::Symlink,
2818 size: target_len,
2819 unix_mode: FileMode::Symlink.to_unix_mode(),
2820 })
2821 }
2822
2823 /// Read the target of a symlink `node`. Works for both overlay
2824 /// (`PendingSymlink`) and captured (`Symlink`) records.
2825 ///
2826 /// Codex r12 thread 3293510316 (P1): the prior implementation
2827 /// used `OsStr::from_encoded_bytes_unchecked` on bytes loaded
2828 /// from the object store, which is unsound — that API's safety
2829 /// contract requires bytes minted by `OsStr::as_encoded_bytes`
2830 /// in *this* process and Rust version, but captured-tree blobs
2831 /// can come from any process and version. The corrected path
2832 /// delegates to [`symlink_target_from_bytes`], which uses
2833 /// platform-safe APIs (`OsStrExt::from_bytes` on Unix, UTF-8
2834 /// validation on Windows).
2835 pub fn read_link(&self, node: NodeId) -> Result<OsString> {
2836 let record = self.record_for(node)?;
2837 match record {
2838 NodeRecord::PendingSymlink { path } => {
2839 let pending = self.inner.pending.lock_or_poisoned();
2840 let bytes = pending
2841 .symlinks
2842 .get(&path)
2843 .ok_or_else(|| MountError::Stale(format!("symlink {}", path.display())))?;
2844 symlink_target_from_bytes(bytes)
2845 }
2846 NodeRecord::Symlink { blob } => {
2847 let bytes = self.load_blob_bytes(&blob)?;
2848 symlink_target_from_bytes(&bytes)
2849 }
2850 other => Err(MountError::InvalidArgument(format!(
2851 "read_link on non-symlink record: {other:?}"
2852 ))),
2853 }
2854 }
2855
2856 /// Flush all hot buffers to CAS. Useful at the start of `capture`
2857 /// or when tests want a deterministic warm state.
2858 pub fn flush_all(&self) -> Result<()> {
2859 let ids: Vec<u64> = self
2860 .inner
2861 .pending
2862 .lock_or_poisoned()
2863 .hot
2864 .keys()
2865 .copied()
2866 .collect();
2867 for id in ids {
2868 self.flush_node(NodeId(id))?;
2869 }
2870 Ok(())
2871 }
2872
2873 /// Look up a path in the pending tier. Order: hot buffer (in-flight
2874 /// writes), then warm tier (promoted blob), then None (caller must
2875 /// fall back to the immutable state's tree).
2876 ///
2877 /// Under the unified NodeId-keyed model warm bytes live at
2878 /// `warm[id]`; the path → id resolution goes through
2879 /// `inodes.by_path` (lock order: pending ⊐ inodes).
2880 fn pending_lookup(&self, path: &Path) -> Option<PendingHit> {
2881 let pending = self.inner.pending.lock_or_poisoned();
2882 if pending.tombstones.contains(path) {
2883 return Some(PendingHit::Tombstone);
2884 }
2885 if let Some(target) = pending.symlinks.get(path) {
2886 return Some(PendingHit::Symlink {
2887 target_len: target.len() as u64,
2888 });
2889 }
2890 if let Some(node_id) = pending.hot_by_path.get(path)
2891 && let Some(buf) = pending.hot.get(node_id)
2892 {
2893 return Some(PendingHit::Hot {
2894 node: NodeId(*node_id),
2895 size: buf.bytes.len() as u64,
2896 mode: buf.mode,
2897 });
2898 }
2899 // Warm needs path → NodeId resolution. Acquire inodes inside
2900 // the pending lock (lock order: pending ⊐ inodes).
2901 let inodes = self.inner.inodes.lock_or_poisoned();
2902 let id = *inodes.by_path.get(path)?;
2903 let entry = pending.warm.get(&id)?;
2904 Some(PendingHit::Warm {
2905 blob: entry.blob,
2906 size: entry.size,
2907 mode: entry.mode,
2908 })
2909 }
2910
2911 /// True if the parent dir or any ancestor of `path` has been
2912 /// `rmdir`'d through the mount. Used by lookup/enumerate so the
2913 /// kernel never sees stale captured children of a directory the
2914 /// agent removed.
2915 fn ancestor_is_dir_tombstoned(&self, pending: &Pending, path: &Path) -> bool {
2916 let mut cursor = path.parent();
2917 while let Some(p) = cursor {
2918 if p.as_os_str().is_empty() {
2919 break;
2920 }
2921 if pending.dir_tombstones.contains(p) {
2922 return true;
2923 }
2924 cursor = p.parent();
2925 }
2926 false
2927 }
2928
2929 /// Does any pending entry sit *under* `dir` as a strict prefix?
2930 /// I.e. has an agent created `dir/something` even though `dir`
2931 /// itself isn't in the captured tree yet? An explicit `mkdir dir`
2932 /// also counts (so an empty mkdir survives without children).
2933 fn pending_dir_exists(&self, dir: &Path) -> bool {
2934 if dir.as_os_str().is_empty() {
2935 return false;
2936 }
2937 let pending = self.inner.pending.lock_or_poisoned();
2938 pending.child_index.dir_exists(dir)
2939 }
2940
2941 /// Direct children of `dir` that exist purely in the pending
2942 /// tier (created/written by the mount, not in the captured tree).
2943 /// Returns each immediate child as either a file (with hot or
2944 /// warm metadata) or an implicit directory (because some pending
2945 /// path is *under* this dir, e.g. `src/foo.rs` makes `src` an
2946 /// implicit dir of root). Tombstones suppress paths.
2947 fn pending_children_at(&self, dir: &Path) -> Vec<(String, PendingChildKind)> {
2948 let pending = self.inner.pending.lock_or_poisoned();
2949 pending.child_index.children_at(dir)
2950 }
2951
2952 /// Pre-index implementation retained as a test-only differential
2953 /// oracle and negative control. `work` counts pending entries
2954 /// inspected, making the former O(all pending) behavior explicit.
2955 #[cfg(test)]
2956 fn pending_dir_exists_full_scan(&self, dir: &Path) -> (bool, usize) {
2957 if dir.as_os_str().is_empty() {
2958 return (false, 0);
2959 }
2960 let pending = self.inner.pending.lock_or_poisoned();
2961 let mut work = 1;
2962 if pending.explicit_dirs.contains(dir) {
2963 return (true, work);
2964 }
2965 let probe = |path: &Path| {
2966 path.strip_prefix(dir)
2967 .ok()
2968 .and_then(|tail| tail.components().next())
2969 .is_some()
2970 };
2971 let inodes = self.inner.inodes.lock_or_poisoned();
2972 for id in pending.warm.keys() {
2973 work += 1;
2974 if pending.is_orphan(*id) {
2975 continue;
2976 }
2977 if let Some(path) = inodes.by_id.get(id).and_then(warm_path_of_record)
2978 && !pending.tombstones.contains(path)
2979 && probe(path)
2980 {
2981 return (true, work);
2982 }
2983 }
2984 drop(inodes);
2985 for path in pending.hot_by_path.keys() {
2986 work += 1;
2987 if !pending.tombstones.contains(path) && probe(path) {
2988 return (true, work);
2989 }
2990 }
2991 for path in pending.symlinks.keys() {
2992 work += 1;
2993 if probe(path) {
2994 return (true, work);
2995 }
2996 }
2997 (false, work)
2998 }
2999
3000 #[cfg(test)]
3001 fn pending_children_at_full_scan(
3002 &self,
3003 dir: &Path,
3004 ) -> (Vec<(String, PendingChildKind)>, usize) {
3005 let pending = self.inner.pending.lock_or_poisoned();
3006 let mut out: BTreeMap<String, PendingChildKind> = BTreeMap::new();
3007 let mut work = 0;
3008 let project = |path: &Path| -> Option<(String, bool)> {
3009 let suffix = if dir.as_os_str().is_empty() {
3010 Some(path)
3011 } else {
3012 path.strip_prefix(dir).ok()
3013 }?;
3014 let mut comps = suffix.components();
3015 let name = match comps.next()? {
3016 Component::Normal(name) => name.to_str()?.to_string(),
3017 _ => return None,
3018 };
3019 Some((name, comps.next().is_some()))
3020 };
3021
3022 for (path, node_id) in &pending.hot_by_path {
3023 work += 1;
3024 if pending.tombstones.contains(path) {
3025 continue;
3026 }
3027 let Some((name, is_dir)) = project(path) else {
3028 continue;
3029 };
3030 if is_dir {
3031 out.entry(name).or_insert(PendingChildKind::Dir);
3032 } else if let Some(buf) = pending.hot.get(node_id) {
3033 out.insert(
3034 name,
3035 PendingChildKind::HotFile {
3036 node: NodeId(*node_id),
3037 size: buf.bytes.len() as u64,
3038 mode: buf.mode,
3039 },
3040 );
3041 }
3042 }
3043
3044 let inodes = self.inner.inodes.lock_or_poisoned();
3045 for (id, entry) in &pending.warm {
3046 work += 1;
3047 if pending.is_orphan(*id) {
3048 continue;
3049 }
3050 let Some(path) = inodes.by_id.get(id).and_then(warm_path_of_record) else {
3051 continue;
3052 };
3053 if pending.tombstones.contains(path) {
3054 continue;
3055 }
3056 let Some((name, is_dir)) = project(path) else {
3057 continue;
3058 };
3059 if is_dir {
3060 out.entry(name).or_insert(PendingChildKind::Dir);
3061 } else {
3062 out.entry(name).or_insert(PendingChildKind::WarmFile {
3063 size: entry.size,
3064 mode: entry.mode,
3065 });
3066 }
3067 }
3068 drop(inodes);
3069
3070 for (path, target) in &pending.symlinks {
3071 work += 1;
3072 let Some((name, is_dir)) = project(path) else {
3073 continue;
3074 };
3075 if is_dir {
3076 out.entry(name).or_insert(PendingChildKind::Dir);
3077 } else {
3078 out.entry(name).or_insert(PendingChildKind::Symlink {
3079 size: target.len() as u64,
3080 });
3081 }
3082 }
3083 for path in &pending.explicit_dirs {
3084 work += 1;
3085 if let Some((name, _)) = project(path) {
3086 out.entry(name).or_insert(PendingChildKind::Dir);
3087 }
3088 }
3089 (out.into_iter().collect(), work)
3090 }
3091
3092 #[cfg(test)]
3093 pub(crate) fn pending_index_matches_full_scan(&self) -> std::result::Result<(), String> {
3094 let dirs = {
3095 let pending = self.inner.pending.lock_or_poisoned();
3096 let inodes = self.inner.inodes.lock_or_poisoned();
3097 let mut paths: Vec<PathBuf> = pending.hot_by_path.keys().cloned().collect();
3098 paths.extend(pending.symlinks.keys().cloned());
3099 paths.extend(pending.explicit_dirs.iter().cloned());
3100 paths.extend(
3101 pending
3102 .warm
3103 .keys()
3104 .filter(|id| !pending.is_orphan(**id))
3105 .filter_map(|id| inodes.by_id.get(id).and_then(warm_path_of_record))
3106 .map(Path::to_path_buf),
3107 );
3108 let mut dirs = BTreeSet::from([PathBuf::new()]);
3109 for path in paths {
3110 let mut cursor = path.parent();
3111 while let Some(dir) = cursor {
3112 dirs.insert(dir.to_path_buf());
3113 cursor = dir.parent();
3114 }
3115 if pending.explicit_dirs.contains(&path) {
3116 dirs.insert(path);
3117 }
3118 }
3119 dirs
3120 };
3121
3122 for dir in dirs {
3123 let indexed_exists = self.pending_dir_exists(&dir);
3124 let (scanned_exists, _) = self.pending_dir_exists_full_scan(&dir);
3125 if indexed_exists != scanned_exists {
3126 return Err(format!(
3127 "exists mismatch at {}: indexed={indexed_exists}, scan={scanned_exists}",
3128 dir.display()
3129 ));
3130 }
3131 let indexed_children = self.pending_children_at(&dir);
3132 let (scanned_children, _) = self.pending_children_at_full_scan(&dir);
3133 if indexed_children != scanned_children {
3134 return Err(format!(
3135 "children mismatch at {}: indexed={indexed_children:?}, scan={scanned_children:?}",
3136 dir.display()
3137 ));
3138 }
3139 }
3140 Ok(())
3141 }
3142
3143 #[cfg(test)]
3144 pub(crate) fn pending_index_work(&self, dir: &Path) -> (usize, usize, usize, usize) {
3145 let indexed = self.inner.pending.lock_or_poisoned();
3146 let indexed_exists_work = usize::from(!dir.as_os_str().is_empty());
3147 let indexed_children_work = indexed
3148 .child_index
3149 .by_dir
3150 .get(dir)
3151 .map_or(0, |children| children.names.len());
3152 drop(indexed);
3153 let (_, scan_exists_work) = self.pending_dir_exists_full_scan(dir);
3154 let (_, scan_children_work) = self.pending_children_at_full_scan(dir);
3155 (
3156 indexed_exists_work,
3157 indexed_children_work,
3158 scan_exists_work,
3159 scan_children_work,
3160 )
3161 }
3162
3163 #[cfg(test)]
3164 pub(crate) fn time_pending_index(&self, dir: &Path, iterations: usize) -> Duration {
3165 let start = Instant::now();
3166 for _ in 0..iterations {
3167 std::hint::black_box(self.pending_dir_exists(dir));
3168 std::hint::black_box(self.pending_children_at(dir));
3169 }
3170 start.elapsed()
3171 }
3172}
3173
3174/// Reject FUSE entry names that wouldn't survive a `TreeEntry`'s
3175/// validator. Delegates to [`objects::object::validate_tree_entry_name`]
3176/// so the mount's write-side reject set stays in lockstep with the
3177/// tree serializer's — Codex r13 thread 3293733163 (P2) caught the
3178/// drift where the overlay accepted backslash and control bytes that
3179/// the serializer later rejected at capture with a confusing
3180/// "invalid object" error. The NUL pre-check is here (not in the
3181/// shared validator) because `OsStr` on Unix can carry interior NUL
3182/// bytes that `to_str()` would otherwise round-trip through to the
3183/// validator as an unmarked control byte; we surface a more specific
3184/// error.
3185fn validate_entry_name(name: &OsStr) -> Result<&str> {
3186 let bytes = name.as_encoded_bytes();
3187 if bytes.contains(&0) {
3188 return Err(MountError::InvalidArgument(format!(
3189 "entry name {name:?} contains NUL"
3190 )));
3191 }
3192 let name_str = name.to_str().ok_or_else(|| {
3193 MountError::InvalidArgument(format!("entry name {name:?} is not valid UTF-8"))
3194 })?;
3195 objects::object::validate_tree_entry_name(name_str)
3196 .map_err(|e| MountError::InvalidArgument(e.to_string()))?;
3197 Ok(name_str)
3198}
3199
3200/// Mount-relative path for a warm-tier entry, derived from its
3201/// [`NodeRecord`]. The NodeId-keyed warm tier doesn't store the path
3202/// directly; capture-time tree fold / `pending_dir_exists` /
3203/// `pending_children_at` resolve it via the inode registry. Only
3204/// file-like records (`File`, `PendingFile`) carry warm bytes; the
3205/// other variants return `None`.
3206#[cfg(test)]
3207fn warm_path_of_record(record: &NodeRecord) -> Option<&Path> {
3208 match record {
3209 NodeRecord::File { path, .. } | NodeRecord::PendingFile { path, .. } => Some(path),
3210 _ => None,
3211 }
3212}
3213
3214/// Decode symlink target bytes back into an `OsString`. The Unix
3215/// branch uses `OsStrExt::from_bytes`, which is sound for any byte
3216/// sequence (the inverse of `OsStrExt::as_bytes`). The Windows branch
3217/// validates as UTF-8 and returns [`MountError::InvalidArgument`]
3218/// otherwise — `OsStr` on Windows is a process-internal encoding
3219/// (WTF-8 today, but not promised), so accepting arbitrary captured
3220/// bytes is unsound. Replaces a prior
3221/// `unsafe { OsStr::from_encoded_bytes_unchecked(bytes) }` call site
3222/// (Codex r12 thread 3293510316).
3223fn symlink_target_from_bytes(bytes: &[u8]) -> Result<OsString> {
3224 #[cfg(unix)]
3225 {
3226 use std::os::unix::ffi::OsStrExt;
3227 Ok(OsStr::from_bytes(bytes).to_os_string())
3228 }
3229 #[cfg(not(unix))]
3230 {
3231 match std::str::from_utf8(bytes) {
3232 Ok(s) => Ok(OsString::from(s)),
3233 Err(_) => Err(MountError::InvalidArgument(
3234 "captured symlink target bytes are not valid UTF-8".into(),
3235 )),
3236 }
3237 }
3238}
3239
3240/// Join a parent mount-relative path with a leaf name. Mirrors the
3241/// shape every write-side op uses, so the construction stays
3242/// consistent across the file.
3243#[inline]
3244fn join_child(parent: &Path, name: &str) -> PathBuf {
3245 if parent.as_os_str().is_empty() {
3246 PathBuf::from(name)
3247 } else {
3248 parent.join(name)
3249 }
3250}
3251
3252/// Copy `[offset, offset+buf.len())` from `src` into `buf`, returning
3253/// the number of bytes actually copied (0 when `offset` is past EOF,
3254/// or `min(buf.len(), src.len() - offset)` otherwise). Pulled out so
3255/// the `read` hot path is a single slice copy rather than a Vec
3256/// allocation per call.
3257#[inline]
3258fn copy_into(src: &[u8], offset: u64, buf: &mut [u8]) -> usize {
3259 let offset = offset as usize;
3260 if offset >= src.len() {
3261 return 0;
3262 }
3263 let take = std::cmp::min(buf.len(), src.len() - offset);
3264 buf[..take].copy_from_slice(&src[offset..offset + take]);
3265 take
3266}
3267
3268/// Pending-tier overlay for a captured-tree path. Consumed by `read`
3269/// to decide whether to serve the captured blob (`None` returned by
3270/// the lookup) or the pending overlay's bytes.
3271enum Overlay {
3272 /// Promoted warm-tier blob. Same path now points at this blob in
3273 /// the pending tier; the captured `File` record's blob is
3274 /// effectively stale until capture folds the warm tier in.
3275 Warm(ContentHash),
3276 /// Tombstoned through the mount. The kernel will get a stale
3277 /// inode reply; subsequent dentry refresh resolves the entry as
3278 /// gone.
3279 Gone,
3280}
3281
3282/// What `pending_lookup` found at a given path.
3283#[allow(dead_code)] // `blob` reserved for cross-mount dedup callers.
3284enum PendingHit {
3285 Hot {
3286 node: NodeId,
3287 size: u64,
3288 mode: FileMode,
3289 },
3290 Warm {
3291 blob: ContentHash,
3292 size: u64,
3293 mode: FileMode,
3294 },
3295 Symlink {
3296 target_len: u64,
3297 },
3298 Tombstone,
3299}
3300
3301impl<R: RefBackend, O: OpLogBackend, S: ObjectStore> MountInner<R, O, S> {
3302 /// Drain any hot buffer whose `last_touched` is older than
3303 /// `idle_after`. Mirrors `ContentAddressedMount::promote_idle_buffers`
3304 /// but is callable from the worker thread which only holds a
3305 /// `Weak<MountInner>`.
3306 fn sweep_idle_buffers(&self) -> Result<()> {
3307 let now = Instant::now();
3308 let idle_after = self.promotion.read_or_poisoned().idle_after;
3309 let to_promote: Vec<u64> = {
3310 let pending = self.pending.lock_or_poisoned();
3311 pending
3312 .hot
3313 .iter()
3314 .filter(|(_, buf)| now.saturating_duration_since(buf.last_touched) >= idle_after)
3315 .map(|(id, _)| *id)
3316 .collect()
3317 };
3318 for id in to_promote {
3319 let _ = self.flush_node(NodeId(id));
3320 }
3321 Ok(())
3322 }
3323
3324 /// Promote a single hot buffer to CAS. Inner-side flush so the
3325 /// sweep worker can drain idle buffers without bouncing back
3326 /// through `ContentAddressedMount`.
3327 ///
3328 /// Lifecycle note (R8 — Codex Thread 3293235165): FUSE `flush`
3329 /// fires on every descriptor close including each close of a
3330 /// `dup`-derived fd. For an orphaned node we must NOT touch the
3331 /// orphan marker here and must NOT drop the hot buffer (surviving
3332 /// fds need both). Only [`Self::release_node`] — invoked on the
3333 /// last-close-per-FUSE-open — clears the marker.
3334 fn flush_node(&self, node: NodeId) -> Result<()> {
3335 let (path, mode, bytes, revision) = {
3336 let pending = self.pending.lock_or_poisoned();
3337 // Orphan: keep the buffer alive across `flush` events.
3338 // POSIX open-unlinked semantics: bytes persist for the
3339 // surviving fds; the state survives so subsequent writes
3340 // through those fds keep taking the orphan branch (no
3341 // path republish, no warm promotion). The final clear
3342 // happens in `release_node`.
3343 if pending.is_orphan(node.0) {
3344 return Ok(());
3345 }
3346 let Some(buf) = pending.hot.get(&node.0) else {
3347 return Ok(());
3348 };
3349 // Keep the hot buffer published while CAS I/O runs. The
3350 // namespace index and `pending_lookup` therefore continue
3351 // to agree throughout promotion instead of exposing a
3352 // transient false-negative window.
3353 (buf.path.clone(), buf.mode, buf.bytes.clone(), buf.revision)
3354 };
3355 let size = bytes.len() as u64;
3356 let blob = Blob::new(bytes);
3357 let blob_oid = self
3358 .repo
3359 .store()
3360 .put_blob(&blob)
3361 .map_err(MountError::Store)?;
3362 debug!(?path, %blob_oid, size, "promoted hot buffer to CAS");
3363 let mut pending = self.pending.lock_or_poisoned();
3364 if pending.is_orphan(node.0) {
3365 return Ok(());
3366 }
3367 let Some(current) = pending.hot.get(&node.0) else {
3368 return Ok(());
3369 };
3370 let unchanged =
3371 current.revision == revision && current.path == path && current.mode == mode;
3372 // Warm is NodeId-keyed. The path-keyed tombstone clear below
3373 // is a separate concern (directory-entry level).
3374 pending.warm.insert(
3375 node.0,
3376 PendingEntry {
3377 blob: blob_oid,
3378 mode,
3379 size,
3380 },
3381 );
3382 if !unchanged {
3383 return Ok(());
3384 }
3385
3386 pending.hot.remove(&node.0);
3387 // Only retract the path mapping if it still points at us; an
3388 // unlink-then-recreate during CAS I/O may have rebound the
3389 // same path to a fresh inode.
3390 let owns_path = pending.hot_by_path.get(&path) == Some(&node.0);
3391 if owns_path {
3392 pending.hot_by_path.remove(&path);
3393 pending
3394 .child_index
3395 .insert(path.clone(), PendingChildKind::WarmFile { size, mode });
3396 // Promotion supersedes any prior tombstone only while this
3397 // inode still owns the directory entry.
3398 pending.tombstones.remove(&path);
3399 }
3400 Ok(())
3401 }
3402
3403 /// Final close of `node` from a FUSE `release` callback. Drives
3404 /// the per-NodeId lifecycle: decrement the open count carried on
3405 /// `state[node]`; on the final close of an Orphan, drop bytes
3406 /// and remove the state entry; on the final close of a Live
3407 /// node, promote any hot buffer to warm via `flush_node`. A
3408 /// release for an untracked but real node is a safe no-op; a
3409 /// release for an unknown NodeId is rejected with `NotFound`.
3410 ///
3411 /// The orphan branch never warm-promotes — an orphan's bytes are
3412 /// unreachable by path post-T1/T3, so promoting them would leak
3413 /// data into the captured tree at a now-tombstoned path.
3414 fn release_node(&self, node: NodeId) -> Result<()> {
3415 // Action determined under the lock so we don't re-read state
3416 // after dropping bytes.
3417 enum Outcome {
3418 /// Mid-life (non-final) close OR final close of a Live
3419 /// node. Either way: forward to `flush_node`. (`flush_node`
3420 /// is a no-op for Orphan, so the mid-life Orphan case is
3421 /// also safe to forward.)
3422 Flush,
3423 /// Final close of an Orphan. Dirty hot bytes must still
3424 /// cross the durability boundary before the anonymous
3425 /// inode is retired, but they must not be warm-promoted
3426 /// into the path-indexed pending tree.
3427 OrphanFinal { hot: Option<(PathBuf, Vec<u8>)> },
3428 /// No lifecycle state and no hot buffer. We validate
3429 /// outside the pending lock so double-release of a real
3430 /// inode is a no-op, while a bogus NodeId is rejected.
3431 MaybeUntrackedNoop,
3432 }
3433 let outcome = {
3434 let mut pending = self.pending.lock_or_poisoned();
3435 match pending.state.get(&node.0).copied() {
3436 None => {
3437 // Untracked release (no on_open was ever
3438 // recorded). Treat a tracked hot buffer as Live
3439 // final-close; otherwise validate the NodeId
3440 // outside this lock.
3441 if pending.hot.contains_key(&node.0) {
3442 Outcome::Flush
3443 } else {
3444 Outcome::MaybeUntrackedNoop
3445 }
3446 }
3447 Some(NodeState::Live { open_count }) => {
3448 let n = open_count.saturating_sub(1);
3449 if n == 0 {
3450 pending.state.remove(&node.0);
3451 } else {
3452 pending
3453 .state
3454 .insert(node.0, NodeState::Live { open_count: n });
3455 }
3456 Outcome::Flush
3457 }
3458 Some(NodeState::Orphan { open_count }) => {
3459 let n = open_count.saturating_sub(1);
3460 if n == 0 {
3461 // Final release of an Orphan — POSIX "inode
3462 // lives until last close" ends here. Snapshot
3463 // hot bytes so they can be persisted outside
3464 // the lock before retiring the anonymous state.
3465 let hot = pending
3466 .hot
3467 .get(&node.0)
3468 .map(|buf| (buf.path.clone(), buf.bytes.clone()));
3469 Outcome::OrphanFinal { hot }
3470 } else {
3471 pending
3472 .state
3473 .insert(node.0, NodeState::Orphan { open_count: n });
3474 // Mid-life Orphan release: forward to
3475 // flush_node (which no-ops for orphans).
3476 Outcome::Flush
3477 }
3478 }
3479 }
3480 };
3481 match outcome {
3482 Outcome::Flush => self.flush_node(node),
3483 Outcome::MaybeUntrackedNoop => {
3484 if !self.inodes.lock_or_poisoned().by_id.contains_key(&node.0) {
3485 return Err(MountError::NotFound(format!("node {}", node.0)));
3486 }
3487 Ok(())
3488 }
3489 Outcome::OrphanFinal { hot } => {
3490 if let Some((path, bytes)) = hot {
3491 let size = bytes.len() as u64;
3492 let blob = Blob::new(bytes);
3493 let blob_oid = self
3494 .repo
3495 .store()
3496 .put_blob(&blob)
3497 .map_err(MountError::Store)?;
3498 debug!(?path, %blob_oid, size, "persisted orphan hot buffer to CAS");
3499 }
3500 let mut pending = self.pending.lock_or_poisoned();
3501 pending.state.remove(&node.0);
3502 pending.hot.remove(&node.0);
3503 pending.warm.remove(&node.0);
3504 Ok(())
3505 }
3506 }
3507 }
3508}
3509
3510/// Spawn the safety-sweep worker, if one is requested by the
3511/// inner's promotion policy. The worker holds a `Weak<MountInner>`
3512/// so the mount can drop normally; on each tick it upgrades the
3513/// weak handle and drains any hot buffer that's been idle longer
3514/// than `idle_after`. A `None` `sweep_interval` returns `None`,
3515/// meaning event-driven promotion only.
3516fn spawn_sweep_worker<
3517 R: RefBackend + 'static,
3518 O: OpLogBackend + 'static,
3519 S: ObjectStore + 'static,
3520>(
3521 inner: &Arc<MountInner<R, O, S>>,
3522) -> Option<SweepHandle> {
3523 let interval = inner.promotion.read_or_poisoned().sweep_interval?;
3524 let weak = Arc::downgrade(inner);
3525 let state = Arc::new(SweepShutdown::new());
3526 let state_for_thread = Arc::clone(&state);
3527 let join = std::thread::Builder::new()
3528 .name("heddle-mount-sweep".into())
3529 .spawn(move || sweep_worker_loop(weak, state_for_thread, interval))
3530 .ok()?;
3531 Some(SweepHandle {
3532 state,
3533 join: Some(join),
3534 })
3535}
3536
3537/// Tick body for the safety-sweep worker. Parks on the shutdown
3538/// condvar until either the timer interval elapses (run a sweep) or
3539/// `signal_and_join` wakes us (exit). Also exits when the weak
3540/// `MountInner` reference can no longer be upgraded.
3541fn sweep_worker_loop<
3542 R: RefBackend + 'static,
3543 O: OpLogBackend + 'static,
3544 S: ObjectStore + 'static,
3545>(
3546 inner: std::sync::Weak<MountInner<R, O, S>>,
3547 state: Arc<SweepShutdown>,
3548 interval: Duration,
3549) {
3550 loop {
3551 // Wait returns true on shutdown, false on timeout — either
3552 // way we re-check the upgrade afterwards.
3553 if state.wait(interval) {
3554 return;
3555 }
3556 let Some(mount) = inner.upgrade() else {
3557 return;
3558 };
3559 if let Err(err) = mount.sweep_idle_buffers() {
3560 warn!(?err, "sweep worker hit error promoting idle buffers");
3561 }
3562 // Drop the strong-count immediately so the mount can drop
3563 // even if our next wait is still pending.
3564 drop(mount);
3565 }
3566}
3567
3568fn resolve_thread<R: RefBackend, O: OpLogBackend, S: ObjectStore>(
3569 repo: &Repository<R, O, S>,
3570 thread: &str,
3571) -> Result<MountState> {
3572 let thread_name = objects::object::ThreadName::from(thread);
3573 // `CoreRefBackend::get_thread` is async for the Postgres backend;
3574 // mount construction always runs off-runtime, so block here.
3575 let state_id = pollster::block_on(repo.refs().get_thread(&thread_name))?
3576 .ok_or_else(|| MountError::UnknownThread(thread.to_string()))?;
3577 let state = repo
3578 .store()
3579 .get_state(&state_id)?
3580 .ok_or_else(|| MountError::UnknownThread(thread.to_string()))?;
3581 Ok(MountState {
3582 state_id,
3583 tree: state.tree,
3584 })
3585}
3586
3587impl<R: RefBackend + 'static, O: OpLogBackend + 'static, S: ObjectStore + 'static> PlatformShell
3588 for ContentAddressedMount<R, O, S>
3589{
3590 fn lookup(&self, parent: NodeId, name: &OsStr) -> Result<Option<Entry>> {
3591 let record = self.record_for(parent)?;
3592 let parent_path = match self.dir_path_of(&record) {
3593 Some(p) => p,
3594 None => return Ok(None),
3595 };
3596 let Some(name_str) = name.to_str() else {
3597 return Ok(None);
3598 };
3599 let child_path = join_child(&parent_path, name_str);
3600
3601 // Pending tier wins over the immutable tree for files —
3602 // that's what makes "write then read" return the new bytes.
3603 match self.pending_lookup(&child_path) {
3604 Some(PendingHit::Tombstone) => return Ok(None),
3605 Some(hit) => {
3606 // Non-tombstone hits always yield an entry; tombstone
3607 // is handled above.
3608 if let Some(entry) = self.entry_from_pending_hit(hit, &child_path, name) {
3609 return Ok(Some(entry));
3610 }
3611 return Ok(None);
3612 }
3613 None => {}
3614 }
3615
3616 // Did an ancestor get rmdir'd? Then the captured-tree entry
3617 // is no longer addressable through this mount.
3618 {
3619 let pending = self.inner.pending.lock_or_poisoned();
3620 if pending.dir_tombstones.contains(&child_path)
3621 || self.ancestor_is_dir_tombstoned(&pending, &child_path)
3622 {
3623 return Ok(None);
3624 }
3625 }
3626
3627 // Captured tree wins over implicit pending dirs: if both
3628 // the captured tree has `nested/` AND the pending tier has
3629 // `nested/c.txt`, we want callers to descend through the
3630 // captured `Dir` record (which still overlays pending on
3631 // its way down) rather than through a `PendingDir` shell
3632 // that would hide the captured siblings.
3633 let parent_tree = self.tree_for_record(&record)?;
3634 if let Some(tree_entry) = parent_tree.get(name_str) {
3635 return Ok(Some(self.entry_from_tree_entry(&parent_path, tree_entry)?));
3636 }
3637
3638 // Implicit directory introduced by a deeper pending write
3639 // (e.g. write to `newdir/foo.rs` makes `newdir` resolvable
3640 // as a directory before capture).
3641 if self.pending_dir_exists(&child_path) {
3642 let node = self.intern(NodeRecord::PendingDir {
3643 path: child_path.clone(),
3644 });
3645 return Ok(Some(Entry {
3646 node,
3647 name: OsString::from(name_str),
3648 kind: NodeKind::Directory,
3649 size: self.pending_children_at(&child_path).len() as u64,
3650 unix_mode: DIR_UNIX_MODE,
3651 }));
3652 }
3653
3654 Ok(None)
3655 }
3656
3657 fn read(&self, node: NodeId, offset: u64, buf: &mut [u8]) -> Result<usize> {
3658 let record = self.record_for(node)?;
3659
3660 // Hot-tier fast path: if there's an in-flight buffer for
3661 // *this* NodeId, copy the requested slice directly under the
3662 // lock without cloning the whole buffer. Sub-microsecond on
3663 // small writes; avoids one `Vec::clone` per `read` callback.
3664 {
3665 let pending = self.inner.pending.lock_or_poisoned();
3666 if let Some(hot) = pending.hot.get(&node.0) {
3667 return Ok(copy_into(&hot.bytes, offset, buf));
3668 }
3669 }
3670
3671 match &record {
3672 NodeRecord::PendingFile { path, .. } => {
3673 // Same shape, keyed by path: another NodeId may own
3674 // the buffer (e.g. after rename/coalesce). Orphan
3675 // PendingFiles skip the path overlay — the path is
3676 // gone (open-unlinked) or rebound (rename-over) — but
3677 // the unified shape preserves the inode's own warm
3678 // bytes (if any) at `warm[node.0]`. With no warm
3679 // fallback there is no captured-tier source either,
3680 // so the read errors with Stale.
3681 let warm_blob = {
3682 let pending = self.inner.pending.lock_or_poisoned();
3683 if pending.is_orphan(node.0) {
3684 return match pending.warm.get(&node.0).map(|e| e.blob) {
3685 Some(blob) => {
3686 drop(pending);
3687 let bytes = self.load_blob_bytes(&blob)?;
3688 Ok(copy_into(&bytes, offset, buf))
3689 }
3690 None => Err(MountError::Stale(format!(
3691 "orphan pending file {} has no readable bytes",
3692 path.display()
3693 ))),
3694 };
3695 }
3696 if let Some(id) = pending.hot_by_path.get(path).copied()
3697 && let Some(hot) = pending.hot.get(&id)
3698 {
3699 return Ok(copy_into(&hot.bytes, offset, buf));
3700 }
3701 // Warm is NodeId-keyed; resolve path → id via the
3702 // inode registry.
3703 let inodes = self.inner.inodes.lock_or_poisoned();
3704 inodes
3705 .by_path
3706 .get(path)
3707 .copied()
3708 .and_then(|id| pending.warm.get(&id).map(|e| e.blob))
3709 };
3710 match warm_blob {
3711 Some(blob) => {
3712 let bytes = self.load_blob_bytes(&blob)?;
3713 Ok(copy_into(&bytes, offset, buf))
3714 }
3715 None => Err(MountError::Stale(format!(
3716 "pending file {}",
3717 path.display()
3718 ))),
3719 }
3720 }
3721 NodeRecord::File { blob, path, .. } => {
3722 // A captured-tree file whose path now has a pending
3723 // overlay (hot buffer on a sibling NodeId, warm-tier
3724 // promotion, or tombstone) must serve the overlay,
3725 // not the captured blob. Without this, a FUSE
3726 // `write → flush → read` round-trip through the
3727 // *same* kernel-cached NodeId silently returns the
3728 // pre-write bytes (the kernel reuses its dentry for
3729 // the duration of the entry TTL and never re-issues
3730 // `lookup`, so the inode record is never refreshed
3731 // from `File` to `PendingFile`).
3732 //
3733 // Priority: hot @ another NodeId → warm → tombstone
3734 // (ENOENT-shaped Stale) → captured blob.
3735 //
3736 // Orphan exception: an open-unlinked or
3737 // rename-displaced inode must skip the path overlay
3738 // entirely. `tombstones[path]` / `hot_by_path[path]`
3739 // / `warm[path]` now reflect a sibling at the same
3740 // name; serving them would let the open fd observe
3741 // (or even modify, via Overlay::Hot) bytes that
3742 // POSIX assigns to the sibling. Fall through to the
3743 // captured blob — that's the inode's own data.
3744 let overlay = {
3745 let pending = self.inner.pending.lock_or_poisoned();
3746 if pending.is_orphan(node.0) {
3747 pending
3748 .warm
3749 .get(&node.0)
3750 .map(|warm| Overlay::Warm(warm.blob))
3751 } else if pending.tombstones.contains(path) {
3752 Some(Overlay::Gone)
3753 } else if let Some(other_id) = pending.hot_by_path.get(path).copied()
3754 && let Some(hot) = pending.hot.get(&other_id)
3755 {
3756 return Ok(copy_into(&hot.bytes, offset, buf));
3757 } else {
3758 let inodes = self.inner.inodes.lock_or_poisoned();
3759 inodes.by_path.get(path).copied().and_then(|id| {
3760 pending.warm.get(&id).map(|warm| Overlay::Warm(warm.blob))
3761 })
3762 }
3763 };
3764 match overlay {
3765 Some(Overlay::Gone) => Err(MountError::Stale(format!(
3766 "file {} was unlinked through the mount",
3767 path.display()
3768 ))),
3769 Some(Overlay::Warm(blob)) => {
3770 let bytes = self.load_blob_bytes(&blob)?;
3771 Ok(copy_into(&bytes, offset, buf))
3772 }
3773 None => {
3774 let bytes = self.load_blob_bytes(blob)?;
3775 Ok(copy_into(&bytes, offset, buf))
3776 }
3777 }
3778 }
3779 NodeRecord::Gitlink { placeholder, .. } => Ok(copy_into(placeholder, offset, buf)),
3780 NodeRecord::Symlink { blob } => {
3781 let bytes = self.load_blob_bytes(blob)?;
3782 Ok(copy_into(&bytes, offset, buf))
3783 }
3784 _ => Err(MountError::NotFound(format!(
3785 "read on non-file node {}",
3786 node.0
3787 ))),
3788 }
3789 }
3790
3791 fn write(&self, node: NodeId, offset: u64, data: &[u8]) -> Result<usize> {
3792 let end = validate_write_extent(offset, data.len())?;
3793 let offset = usize::try_from(offset).map_err(|_| {
3794 MountError::InvalidArgument(format!("write offset {offset} does not fit in usize"))
3795 })?;
3796 // Determine the mount-relative path and mode to key the hot
3797 // buffer on. New files (`PendingFile`) carry their path
3798 // directly; pre-existing files identify by the parent's
3799 // tree entry. Any other node type rejects writes.
3800 let record = self.record_for(node)?;
3801 let (path, mode, captured_blob) = match &record {
3802 NodeRecord::PendingFile { path, mode } => (path.clone(), *mode, None),
3803 NodeRecord::File {
3804 path, mode, blob, ..
3805 } => (path.clone(), *mode, Some(*blob)),
3806 _ => return Err(MountError::ReadOnly),
3807 };
3808
3809 // Phase 1: under the lock, decide whether a buffer already
3810 // exists, and if not, what durable source we should seed it
3811 // from. Snapshot the seed source's blob oid (if any) and drop
3812 // the lock so we can do CAS IO without blocking other writers.
3813 //
3814 // POSIX `pwrite` preserves bytes outside the [offset, offset+len)
3815 // range. The kernel never re-issues those bytes on a partial
3816 // overwrite, so the hot buffer must already contain them when
3817 // we apply `data`. The seed sources, in priority order:
3818 //
3819 // 1. The warm tier — a previously-flushed write to this same
3820 // path in this mount session. This is the most recent
3821 // durable view and supersedes the captured tree.
3822 // 2. The captured tree's blob for this path — the underlying
3823 // file the agent is editing. Only applicable when the
3824 // record was minted from a captured tree entry (i.e.
3825 // `NodeRecord::File`); a `PendingFile` with no warm entry
3826 // means the agent already unlinked-and-recreated.
3827 // 3. Empty — no durable predecessor, so this write builds a
3828 // file from scratch.
3829 //
3830 // A tombstone for the path overrides everything: the agent
3831 // deleted the file and is now creating a fresh one.
3832 enum Seed {
3833 None,
3834 Blob(ContentHash),
3835 }
3836 let seed = {
3837 // Resolve the path's current Live owner via the inode
3838 // registry — warm bytes for the path live at
3839 // `warm[live_id]` under the unified shape.
3840 let path_owner = {
3841 let inodes = self.inner.inodes.lock_or_poisoned();
3842 inodes.by_path.get(&path).copied()
3843 };
3844 let pending = self.inner.pending.lock_or_poisoned();
3845 let orphan = pending.is_orphan(node.0);
3846 if pending.hot.contains_key(&node.0) {
3847 // The per-NodeId buffer is always authoritative —
3848 // both for live writes (this fd's accumulated bytes)
3849 // and for orphan writes (POSIX says the bytes belong
3850 // to the open handle).
3851 Seed::None
3852 } else if !orphan
3853 && pending
3854 .hot_by_path
3855 .get(&path)
3856 .is_some_and(|id| pending.hot.contains_key(id))
3857 {
3858 // Sibling at the same path has a buffer — coalesce
3859 // onto it. Orphans never look at the path's overlay
3860 // (the sibling at `hot_by_path[path]` is a different
3861 // inode, not us).
3862 Seed::None
3863 } else if orphan {
3864 // Orphan-aware seeding. The path's overlay belongs
3865 // to the sibling at the rebound name; this inode's
3866 // own bytes live at `warm[node.0]` (or in the
3867 // captured blob).
3868 pending
3869 .warm
3870 .get(&node.0)
3871 .map(|e| Seed::Blob(e.blob))
3872 .or_else(|| captured_blob.map(Seed::Blob))
3873 .unwrap_or(Seed::None)
3874 } else if pending.tombstones.contains(&path) {
3875 // Unlink-then-write through a fresh inode (POSIX
3876 // unlink+open(O_CREAT)): start from empty.
3877 Seed::None
3878 } else if let Some(entry) = path_owner.and_then(|id| pending.warm.get(&id)) {
3879 Seed::Blob(entry.blob)
3880 } else if let Some(blob) = captured_blob {
3881 Seed::Blob(blob)
3882 } else {
3883 Seed::None
3884 }
3885 };
3886 let seed_bytes = match seed {
3887 Seed::None => None,
3888 // The hot buffer is owned + mutated, so we materialize a
3889 // Vec here. One alloc + copy per first-write per file;
3890 // subsequent writes hit the existing buffer.
3891 Seed::Blob(hash) => Some((*self.load_blob_bytes(&hash)?).to_vec()),
3892 };
3893
3894 // Phase 2: re-acquire the lock, install or update the hot
3895 // buffer, apply the write. If a buffer materialized between
3896 // phases (e.g. a coalesce from another NodeId), prefer the
3897 // existing buffer's bytes — our `seed_bytes` are stale.
3898 let mut pending = self.inner.pending.lock_or_poisoned();
3899 // POSIX unlink+open semantics. Two write shapes share this
3900 // method and must be kept separate:
3901 //
3902 // * unlink-then-create (`unlink P; open(P, O_CREAT); write`)
3903 // — `create_file` minted a fresh `NodeId` and cleared the
3904 // tombstone for P. Our `node.0` is not in `orphans` and
3905 // the write republishes the name normally.
3906 //
3907 // * open-then-unlink (`open(P); unlink P; write through old
3908 // fd`) — `unlink_entry` recorded `node.0` in `orphans`
3909 // and left the tombstone in place. POSIX is explicit:
3910 // the inode lives behind the fd, but the directory entry
3911 // must stay gone. Republishing `hot_by_path[P] = node.0`
3912 // or clearing the tombstone would resurrect the pathname
3913 // for every other observer (lookup, enumerate, capture).
3914 // The orphan branch updates only the per-NodeId buffer;
3915 // `flush_node` reads the same `orphans` signal at
3916 // promotion time and drops the buffer instead of warming
3917 // it.
3918 let orphan = pending.is_orphan(node.0);
3919 if !orphan {
3920 // Coalesce two NodeIds for the same path onto the same buffer.
3921 if let Some(existing_id) = pending.hot_by_path.get(&path).copied()
3922 && existing_id != node.0
3923 && let Some(buf) = pending.hot.remove(&existing_id)
3924 {
3925 pending.hot.insert(node.0, buf);
3926 }
3927 pending.hot_by_path.insert(path.clone(), node.0);
3928 // A live hot buffer means the file exists again — clear
3929 // any tombstone for this path so subsequent
3930 // `pending_lookup` calls see the buffer instead of a
3931 // "deleted" sentinel. POSIX:
3932 // unlink+open(O_CREAT)+pwrite reborns the path. The seed
3933 // logic above already starts the buffer empty when a
3934 // tombstone is present, so we don't need to inspect the
3935 // tombstone here.
3936 pending.tombstones.remove(&path);
3937 }
3938 let buf = pending.hot.entry(node.0).or_insert_with(|| HotBuffer {
3939 path: path.clone(),
3940 mode,
3941 bytes: seed_bytes.unwrap_or_default(),
3942 last_touched: Instant::now(),
3943 revision: 0,
3944 });
3945 // POSIX `pwrite` past EOF zero-fills the gap.
3946 if buf.bytes.len() < end {
3947 buf.bytes.resize(end, 0);
3948 }
3949 buf.bytes[offset..end].copy_from_slice(data);
3950 buf.last_touched = Instant::now();
3951 buf.revision = buf.revision.wrapping_add(1);
3952 if !orphan {
3953 let indexed = PendingChildKind::HotFile {
3954 node,
3955 size: buf.bytes.len() as u64,
3956 mode: buf.mode,
3957 };
3958 pending.child_index.insert(path.clone(), indexed);
3959 }
3960 let written = data.len();
3961 drop(pending);
3962 // Cheap idle-promotion sweep — an agent that's gone quiet on
3963 // *other* files for longer than the policy window gets its
3964 // buffers drained without an explicit close.
3965 let _ = self.promote_idle_buffers();
3966 Ok(written)
3967 }
3968
3969 fn enumerate(&self, dir: NodeId) -> Result<Vec<Entry>> {
3970 let record = self.record_for(dir)?;
3971 let parent_path = match self.dir_path_of(&record) {
3972 Some(p) => p,
3973 None => return Err(MountError::NotADirectory(format!("{record:?}"))),
3974 };
3975 let tree = self.tree_for_record(&record)?;
3976 let mut by_name: BTreeMap<&str, Entry> = BTreeMap::new();
3977
3978 // If this directory itself is dir-tombstoned, enumerate
3979 // returns empty regardless of any captured children. (A
3980 // child rmdir doesn't affect us — only an ancestor or self
3981 // tombstone does.)
3982 {
3983 let pending = self.inner.pending.lock_or_poisoned();
3984 if pending.dir_tombstones.contains(&parent_path)
3985 || self.ancestor_is_dir_tombstoned(&pending, &parent_path)
3986 {
3987 return Ok(vec![]);
3988 }
3989 }
3990
3991 // Pass 1: captured-tree entries, with pending overlay.
3992 for tree_entry in tree.entries() {
3993 let entry_path = join_child(&parent_path, tree_entry.name());
3994 // Whole-subtree rmdir on a captured dir entry.
3995 {
3996 let pending = self.inner.pending.lock_or_poisoned();
3997 if pending.dir_tombstones.contains(&entry_path) {
3998 continue;
3999 }
4000 }
4001 match self.pending_lookup(&entry_path) {
4002 Some(PendingHit::Tombstone) => continue,
4003 Some(hit) => {
4004 if let Some(entry) =
4005 self.entry_from_pending_hit(hit, &entry_path, OsStr::new(tree_entry.name()))
4006 {
4007 by_name.insert(tree_entry.name(), entry);
4008 }
4009 continue;
4010 }
4011 None => {}
4012 }
4013 let entry = self.entry_from_tree_entry(&parent_path, tree_entry)?;
4014 by_name.insert(tree_entry.name(), entry);
4015 }
4016
4017 // Pass 2: pending-only children of `parent_path` (mount-only
4018 // files and implicit subdirectories the agent created).
4019 let mut pending_entries: Vec<Entry> = Vec::new();
4020 let pending_children = self.pending_children_at(&parent_path);
4021 for (name, kind) in pending_children {
4022 // Don't shadow a captured-tree entry (already handled in
4023 // pass 1 via pending_lookup).
4024 if by_name.contains_key(name.as_str()) {
4025 continue;
4026 }
4027 let full_path = join_child(&parent_path, &name);
4028 match kind {
4029 PendingChildKind::HotFile { node, size, mode } => {
4030 pending_entries.push(Entry {
4031 node,
4032 name: OsString::from(&name),
4033 kind: kind_for_mode(mode),
4034 size,
4035 unix_mode: mode.to_unix_mode(),
4036 });
4037 }
4038 PendingChildKind::WarmFile { size, mode } => {
4039 let node = self.intern(NodeRecord::PendingFile {
4040 path: full_path,
4041 mode,
4042 });
4043 pending_entries.push(Entry {
4044 node,
4045 name: OsString::from(&name),
4046 kind: kind_for_mode(mode),
4047 size,
4048 unix_mode: mode.to_unix_mode(),
4049 });
4050 }
4051 PendingChildKind::Dir => {
4052 let node = self.intern(NodeRecord::PendingDir { path: full_path });
4053 pending_entries.push(Entry {
4054 node,
4055 name: OsString::from(&name),
4056 kind: NodeKind::Directory,
4057 size: 0,
4058 unix_mode: DIR_UNIX_MODE,
4059 });
4060 }
4061 PendingChildKind::Symlink { size } => {
4062 let node = self.intern(NodeRecord::PendingSymlink { path: full_path });
4063 pending_entries.push(Entry {
4064 node,
4065 name: OsString::from(&name),
4066 kind: NodeKind::Symlink,
4067 size,
4068 unix_mode: FileMode::Symlink.to_unix_mode(),
4069 });
4070 }
4071 }
4072 }
4073 let mut entries: Vec<Entry> = by_name.into_values().collect();
4074 entries.extend(pending_entries);
4075 Ok(entries)
4076 }
4077
4078 fn attrs(&self, node: NodeId) -> Result<Attrs> {
4079 let record = self.record_for(node)?;
4080 let kind = record.kind();
4081 let unix_mode = record.unix_mode();
4082 let (size, nlink) = match &record {
4083 NodeRecord::Root { tree } | NodeRecord::Dir { tree, .. } => {
4084 let tree = self.load_tree(tree)?;
4085 // 2 = `.` + the parent's entry pointing at us. Heddle
4086 // doesn't model hard links, so we don't try to count
4087 // subdirectories' `..` entries.
4088 (tree.entries().len() as u64, 2)
4089 }
4090 NodeRecord::PendingDir { path } => {
4091 // Implicit dir — content lives entirely in the
4092 // pending tier. Size = direct-child count.
4093 (self.pending_children_at(path).len() as u64, 2)
4094 }
4095 NodeRecord::File { blob, path, .. } => {
4096 // Same overlay priority as `read`: hot @ this NodeId
4097 // → hot @ another NodeId for the same path → warm-tier
4098 // promotion → tombstone (stale) → captured blob.
4099 // Keeping `attrs` and `read` symmetric is mandatory:
4100 // `read` consults the warm tier for captured files
4101 // (so `WORLD` shadows `world`), and a stale `attrs`
4102 // that still reports the captured size would clip the
4103 // returned bytes in the kernel's read buffer.
4104 //
4105 // Orphan exception: same as `read`. An open-unlinked
4106 // or rename-displaced inode skips the path overlay
4107 // and reports the captured blob's size (or the
4108 // per-NodeId hot buffer's length, checked first).
4109 let overlay_size = {
4110 let pending = self.inner.pending.lock_or_poisoned();
4111 if let Some(buf) = pending.hot.get(&node.0) {
4112 Some(Some(buf.bytes.len() as u64))
4113 } else if pending.is_orphan(node.0) {
4114 // Prefer the orphan's own warm size (unified
4115 // shape: `warm[node.0]`). With no warm, fall
4116 // through to `blob_size(blob)` — the captured
4117 // size is the orphan's own.
4118 pending.warm.get(&node.0).map(|e| Some(e.size))
4119 } else if pending.tombstones.contains(path) {
4120 // Tombstoned via the mount: treat as
4121 // not-yet-collected. The path is gone but the
4122 // inode is still registered.
4123 Some(None)
4124 } else if let Some(other_id) = pending.hot_by_path.get(path).copied()
4125 && let Some(hot) = pending.hot.get(&other_id)
4126 {
4127 Some(Some(hot.bytes.len() as u64))
4128 } else {
4129 // Warm is NodeId-keyed; resolve path → id via
4130 // the inode registry.
4131 let inodes = self.inner.inodes.lock_or_poisoned();
4132 inodes
4133 .by_path
4134 .get(path)
4135 .copied()
4136 .and_then(|id| pending.warm.get(&id).map(|warm| Some(warm.size)))
4137 }
4138 };
4139 match overlay_size {
4140 Some(Some(size)) => (size, 1),
4141 Some(None) => {
4142 return Err(MountError::Stale(format!(
4143 "file {} was unlinked through the mount",
4144 path.display()
4145 )));
4146 }
4147 None => (self.blob_size(blob)?, 1),
4148 }
4149 }
4150 NodeRecord::Gitlink { placeholder, .. } => (placeholder.len() as u64, 1),
4151 NodeRecord::Symlink { blob } => (self.blob_size(blob)?, 1),
4152 NodeRecord::PendingFile { path, .. } => {
4153 // Orphan branch: a rename-displaced or
4154 // unlinked-but-still-open PendingFile reports either
4155 // its per-NodeId hot buffer length, or its own
4156 // `warm[node.0]` size. `pending_lookup` would
4157 // otherwise consult the rebound path overlay and
4158 // serve the sibling's size.
4159 let orphan_size = {
4160 let pending = self.inner.pending.lock_or_poisoned();
4161 if pending.is_orphan(node.0) {
4162 Some(
4163 pending
4164 .hot
4165 .get(&node.0)
4166 .map(|buf| buf.bytes.len() as u64)
4167 .or_else(|| pending.warm.get(&node.0).map(|e| e.size)),
4168 )
4169 } else {
4170 None
4171 }
4172 };
4173 if let Some(opt) = orphan_size {
4174 let size = opt.ok_or_else(|| {
4175 MountError::Stale(format!(
4176 "orphan pending file {} has no buffered bytes",
4177 path.display()
4178 ))
4179 })?;
4180 (size, 1)
4181 } else {
4182 let hit = self.pending_lookup(path).ok_or_else(|| {
4183 MountError::Stale(format!("pending file {}", path.display()))
4184 })?;
4185 let size = match hit {
4186 PendingHit::Hot { size, .. } | PendingHit::Warm { size, .. } => size,
4187 PendingHit::Symlink { target_len } => target_len,
4188 PendingHit::Tombstone => 0,
4189 };
4190 (size, 1)
4191 }
4192 }
4193 NodeRecord::PendingSymlink { path } => {
4194 let pending = self.inner.pending.lock_or_poisoned();
4195 let size = pending
4196 .symlinks
4197 .get(path)
4198 .map(|t| t.len() as u64)
4199 .ok_or_else(|| {
4200 MountError::Stale(format!("pending symlink {}", path.display()))
4201 })?;
4202 (size, 1)
4203 }
4204 };
4205 let _ = self.path_of(&record);
4206 Ok(Attrs {
4207 node,
4208 kind,
4209 size,
4210 unix_mode,
4211 nlink,
4212 mtime: self.inner.mounted_at,
4213 })
4214 }
4215
4216 fn invalidate(&self, node: NodeId) -> Result<()> {
4217 // Witness-gated discharge: `bp.kernel_forget_inode(node.0)`
4218 // returns:
4219 //
4220 // * `Some(warm_still_references)` — the FSM check passed
4221 // (state is `Released` or `Live { open_count == 0 }`);
4222 // `hot[node]` (with its `hot_by_path` reverse-index
4223 // cleanup) and `state[node]` have been dropped, and the
4224 // bool tells us whether `warm[node]` is still populated.
4225 // Retire the inode-side record iff warm doesn't reference
4226 // — otherwise capture still needs the NodeId → path chain
4227 // to plant the warm bytes back into the new tree.
4228 // * `None` — the FSM check failed (state is
4229 // `Live { open_count >= 1 }` or any `Orphan`); the bytes
4230 // are still referenced. The witness-gated retrofit
4231 // (heddle#211) makes the entire forget path short-circuit
4232 // here: `hot[node]` / `state[node]` are preserved and the
4233 // inode-side `forget` is skipped. The kernel will re-issue
4234 // `forget` once the surviving fd closes (or never, and the
4235 // next `release_node` retires the record). Closes Codex
4236 // r11 finding #3 — the pre-retrofit path removed
4237 // `hot[node]` before any FSM check, stranding an open
4238 // Orphan fd with no readable bytes.
4239 //
4240 // Warm preservation (Codex r12 threads 3293484634 /
4241 // 3293510311, P1): `apply_kernel_forget` intentionally
4242 // leaves `warm[node]` alone — warm is the only durable
4243 // pre-capture copy of flushed writes, and FUSE `forget` is
4244 // a kernel-side dcache eviction (not a close), so dropping
4245 // warm would silently lose the user's committed-in-session
4246 // data.
4247 let retire_inode_record = {
4248 let mut pending = self.inner.pending.lock_or_poisoned();
4249 pending.with_brand(|bp| {
4250 bp.kernel_forget_inode(node.0)
4251 .map(|warm_still_references| !warm_still_references)
4252 .unwrap_or(false)
4253 })
4254 };
4255 if retire_inode_record {
4256 self.inner.inodes.lock_or_poisoned().forget(node);
4257 }
4258 Ok(())
4259 }
4260
4261 fn flush(&self, node: NodeId) -> Result<()> {
4262 self.flush_node(node)
4263 }
4264
4265 fn release(&self, node: NodeId) -> Result<()> {
4266 self.release_node(node)
4267 }
4268
4269 fn on_open(&self, node: NodeId) -> Result<()> {
4270 ContentAddressedMount::on_open(self, node)
4271 }
4272
4273 fn create_file(
4274 &self,
4275 parent: NodeId,
4276 name: &OsStr,
4277 mode: FileMode,
4278 exclusive: bool,
4279 ) -> Result<Entry> {
4280 ContentAddressedMount::create_file(self, parent, name, mode, exclusive)
4281 }
4282
4283 fn make_dir(&self, parent: NodeId, name: &OsStr) -> Result<Entry> {
4284 ContentAddressedMount::make_dir(self, parent, name)
4285 }
4286
4287 fn unlink_entry(&self, parent: NodeId, name: &OsStr) -> Result<()> {
4288 ContentAddressedMount::unlink_entry(self, parent, name)
4289 }
4290
4291 fn rmdir_entry(&self, parent: NodeId, name: &OsStr) -> Result<()> {
4292 ContentAddressedMount::rmdir_entry(self, parent, name)
4293 }
4294
4295 fn rename_entry(
4296 &self,
4297 old_parent: NodeId,
4298 old_name: &OsStr,
4299 new_parent: NodeId,
4300 new_name: &OsStr,
4301 ) -> Result<()> {
4302 ContentAddressedMount::rename_entry(self, old_parent, old_name, new_parent, new_name)
4303 }
4304
4305 fn rename_entry_with_options(
4306 &self,
4307 old_parent: NodeId,
4308 old_name: &OsStr,
4309 new_parent: NodeId,
4310 new_name: &OsStr,
4311 options: RenameOptions,
4312 ) -> Result<()> {
4313 ContentAddressedMount::rename_entry_with_options(
4314 self, old_parent, old_name, new_parent, new_name, options,
4315 )
4316 }
4317
4318 fn set_attrs(&self, node: NodeId, update: AttrUpdate) -> Result<Attrs> {
4319 ContentAddressedMount::set_attrs(self, node, update)
4320 }
4321
4322 fn create_symlink(&self, parent: NodeId, name: &OsStr, target: &Path) -> Result<Entry> {
4323 ContentAddressedMount::create_symlink(self, parent, name, target)
4324 }
4325
4326 fn read_link(&self, node: NodeId) -> Result<OsString> {
4327 ContentAddressedMount::read_link(self, node)
4328 }
4329}
4330
4331impl<R: RefBackend + 'static, O: OpLogBackend + 'static, S: ObjectStore + 'static>
4332 ContentAddressedMount<R, O, S>
4333{
4334 /// Test-only accessor for the warm tier so unit tests can verify
4335 /// promotions landed without going through `read`. Returns paths
4336 /// resolved via the inode registry (warm is NodeId-keyed under
4337 /// the unified shape).
4338 #[cfg(test)]
4339 pub(crate) fn warm_keys(&self) -> Vec<PathBuf> {
4340 let pending = self.inner.pending.lock_or_poisoned();
4341 let inodes = self.inner.inodes.lock_or_poisoned();
4342 pending
4343 .warm
4344 .keys()
4345 .filter(|id| !pending.is_orphan(**id))
4346 .filter_map(|id| inodes.by_id.get(id).and_then(warm_path_of_record))
4347 .map(Path::to_path_buf)
4348 .collect()
4349 }
4350
4351 /// Test-only accessor: was `path` promoted to a CAS blob? Returns
4352 /// the blob oid so dedup tests can compare across mounts.
4353 #[cfg(test)]
4354 pub(crate) fn warm_blob(&self, path: impl AsRef<Path>) -> Option<ContentHash> {
4355 let path = path.as_ref();
4356 let id = self
4357 .inner
4358 .inodes
4359 .lock_or_poisoned()
4360 .by_path
4361 .get(path)
4362 .copied()?;
4363 self.inner
4364 .pending
4365 .lock_or_poisoned()
4366 .warm
4367 .get(&id)
4368 .map(|e| e.blob)
4369 }
4370
4371 /// Test-only accessor: are there any open hot-tier buffers?
4372 #[cfg(test)]
4373 pub(crate) fn hot_buffer_count(&self) -> usize {
4374 self.inner.pending.lock_or_poisoned().hot.len()
4375 }
4376
4377 /// Test-only accessor: snapshot of currently tombstoned paths.
4378 #[cfg(test)]
4379 #[allow(dead_code)]
4380 pub(crate) fn tombstones(&self) -> Vec<PathBuf> {
4381 self.inner
4382 .pending
4383 .lock_or_poisoned()
4384 .tombstones
4385 .iter()
4386 .cloned()
4387 .collect()
4388 }
4389
4390 /// Test-only accessor for the wrapped repository.
4391 #[cfg(test)]
4392 pub(crate) fn repo_handle(&self) -> &Repository<R, O, S> {
4393 &self.inner.repo
4394 }
4395
4396 /// Test-only accessor: is `node` currently marked as an orphaned
4397 /// inode (open-unlinked or rename-displaced with surviving fds)?
4398 #[cfg(test)]
4399 pub(crate) fn orphans_contains(&self, node: NodeId) -> bool {
4400 self.inner.pending.lock_or_poisoned().is_orphan(node.0)
4401 }
4402}
4403
4404/// Low-level test helpers. The mount doesn't yet expose a `create()`
4405/// entrypoint (the FUSE adapter will eventually wire that callback);
4406/// for now tests bypass the kernel-walk and install pending records
4407/// directly. The shape mirrors what `Filesystem::create` will do once
4408/// it lands.
4409#[cfg(test)]
4410pub(crate) mod test_helpers {
4411 use super::*;
4412
4413 /// Mint a fresh pending-file at any (possibly nested) mount-relative
4414 /// path. Path components are taken verbatim — the helper does no
4415 /// validation beyond path normalization.
4416 pub(crate) fn install_pending_file(
4417 mount: &ContentAddressedMount,
4418 name: &str,
4419 mode: FileMode,
4420 ) -> NodeId {
4421 let path = PathBuf::from(name);
4422 mount.intern(NodeRecord::PendingFile { path, mode })
4423 }
4424}