Skip to main content

acyclic_fs/native_mount/
adapter.rs

1//! One native callback adapter for every embedded checkout consumer.
2
3use super::view_gate::{ViewGate, ViewReadLease, ViewWriteLease};
4use super::{
5    CaptureOptions, MountAttributePage, MountAttributeWriteMode, MountDirectoryEntry,
6    MountDirectoryPage, MountFilesystem, MountLookup, MountNode, MountNodeKind, MountOpenFile,
7    MountPath, MountPublication, MountRangeAllocation, MountSeekTarget, MountSourceError,
8    MountViewLease, NativeMountError, capture_root_identity, capture_subtree, seal_checkout,
9};
10use crate::kernel::{
11    AttributeClass, AttributeName, ExtentSeekTarget, FileKind, FileMetadata, FilePayload,
12    FileRecord, GenerationMutationError, LogicalName, NameEncoding, NamespacePath, RebaseDecision,
13};
14use crate::model::{FilesystemProfile, VolumeConfig, VolumeLimits};
15use crate::native_capture::{
16    MAX_NATIVE_EXACT_CAPTURE_PATHS, NativeViewBaseline, capture_paths_batched_with_baseline,
17    capture_subtrees_with_policy_and_baseline,
18};
19use crate::{
20    AsyncAuthorityStore, AsyncObjectStore, AuthoredMutation, ByteRange, CancellationToken,
21    Checkout, DetachedFile, FileId, FsError, NamedAttributeWriteMode, NativeRootIdentity,
22    OperationFailure, OperationId, VolumeId, WorkBudget,
23};
24use bytes::Bytes;
25use std::future::Future;
26use std::ops::{Deref, DerefMut};
27use std::path::Path;
28use std::sync::Arc;
29use std::sync::Mutex as StdMutex;
30use std::sync::OnceLock;
31use std::sync::Weak;
32use std::sync::atomic::{AtomicU8, AtomicU64, Ordering};
33use std::time::Duration;
34
35const CALLBACK_TIMEOUT: Duration = Duration::from_secs(30);
36const MAX_NATIVE_SUBTREE_CAPTURE_PATHS: u32 = 4_000_000;
37const CAPTURE_PENDING: u8 = 0;
38const CAPTURE_COMMITTED: u8 = 1;
39const CAPTURE_CANCELLED: u8 = 2;
40
41#[derive(Clone, Copy)]
42pub(super) enum HostCaptureScope {
43    ExactPaths,
44    Subtree,
45}
46
47// A lifecycle capture runs host traversal off the executor. Dropping its
48// caller cancels the scan unless the atomic commit decision already won.
49struct CaptureCommitGate {
50    cancellation: CancellationToken,
51    state: AtomicU8,
52}
53
54struct CancelCaptureOnDrop(Arc<CaptureCommitGate>);
55
56#[derive(Default)]
57struct CaptureSourceGate {
58    cancelled: bool,
59    active: Vec<Weak<CaptureCommitGate>>,
60}
61
62struct HostCapturePolicy<'a> {
63    baseline: Option<&'a NativeViewBaseline>,
64    budget: WorkBudget,
65    request: Option<&'a CaptureCommitGate>,
66}
67
68impl Drop for CancelCaptureOnDrop {
69    fn drop(&mut self) {
70        if self
71            .0
72            .state
73            .compare_exchange(
74                CAPTURE_PENDING,
75                CAPTURE_CANCELLED,
76                Ordering::AcqRel,
77                Ordering::Acquire,
78            )
79            .is_ok()
80        {
81            self.0.cancellation.cancel();
82        }
83    }
84}
85
86/// Blocking native callback adapter over one async canonical checkout.
87pub struct CheckoutMountSource<A, O> {
88    checkout: Arc<SharedCheckout<A, O>>,
89    root: NamespacePath,
90    limits: VolumeLimits,
91    profile: FilesystemProfile,
92    cancellation: CancellationToken,
93    capture_gate: StdMutex<CaptureSourceGate>,
94    runtime: CallbackRuntime,
95}
96
97#[derive(Clone)]
98pub(super) struct CallbackRuntime {
99    handle: tokio::runtime::Handle,
100}
101
102static CALLBACK_RUNTIME: OnceLock<Result<tokio::runtime::Runtime, String>> = OnceLock::new();
103
104impl CallbackRuntime {
105    pub(super) fn create() -> Result<Self, NativeMountError> {
106        let runtime = CALLBACK_RUNTIME.get_or_init(|| {
107            tokio::runtime::Builder::new_multi_thread()
108                .worker_threads(2)
109                .enable_all()
110                .thread_name("acyclic-fs-mount")
111                .build()
112                .map_err(|error| error.to_string())
113        });
114        match runtime {
115            Ok(runtime) => Ok(Self {
116                handle: runtime.handle().clone(),
117            }),
118            Err(error) => Err(NativeMountError::Driver(error.clone())),
119        }
120    }
121
122    fn block_on<F>(&self, create: impl FnOnce() -> F + Send) -> Result<F::Output, MountSourceError>
123    where
124        F: Future,
125        F::Output: Send,
126    {
127        match tokio::runtime::Handle::try_current() {
128            Ok(current)
129                if current.runtime_flavor() == tokio::runtime::RuntimeFlavor::MultiThread =>
130            {
131                Ok(tokio::task::block_in_place(|| {
132                    self.handle.block_on(create())
133                }))
134            }
135            Ok(_) => self.block_on_worker(create),
136            // Linux FUSE and macOS NFS callbacks enter from driver-owned
137            // threads; concurrent callbacks can use the shared runtime
138            // directly without allocating a helper thread per request.
139            #[cfg(any(target_os = "linux", target_os = "macos"))]
140            Err(_) => Ok(self.handle.block_on(create())),
141            // ProjFS can synchronously re-enter while a callback is being
142            // serviced. Preserve the scoped boundary to bound that recursion.
143            #[cfg(not(any(target_os = "linux", target_os = "macos")))]
144            Err(_) => self.block_on_worker(create),
145        }
146    }
147
148    fn block_on_worker<F>(
149        &self,
150        create: impl FnOnce() -> F + Send,
151    ) -> Result<F::Output, MountSourceError>
152    where
153        F: Future,
154        F::Output: Send,
155    {
156        std::thread::scope(|scope| {
157            let worker = std::thread::Builder::new()
158                .name("acyclic-fs-callback".to_owned())
159                .spawn_scoped(scope, || self.handle.block_on(create()))
160                .map_err(|error| MountSourceError::Engine(error.to_string()))?;
161            match worker.join() {
162                Ok(value) => Ok(value),
163                Err(payload) => std::panic::resume_unwind(payload),
164            }
165        })
166    }
167
168    pub(super) fn wait<T: Send, F: Future<Output = Result<T, MountSourceError>>>(
169        &self,
170        create: impl FnOnce() -> F + Send,
171    ) -> Result<T, MountSourceError> {
172        self.block_on(|| async {
173            tokio::time::timeout(CALLBACK_TIMEOUT, create())
174                .await
175                .map_err(|_| MountSourceError::Stale)?
176        })?
177    }
178}
179
180struct DetachedMountState<A, O> {
181    file: DetachedFile<A, O>,
182    metadata: FileMetadata,
183    mutation_epoch: u64,
184}
185
186pub(super) struct CheckoutDetachedFile<A, O> {
187    state: tokio::sync::Mutex<DetachedMountState<A, O>>,
188    runtime: CallbackRuntime,
189    cancellation: CancellationToken,
190    profile: FilesystemProfile,
191    limits: VolumeLimits,
192}
193
194struct CheckoutAttachedFile<A, O> {
195    checkout: Arc<SharedCheckout<A, O>>,
196    file_id: FileId,
197    runtime: CallbackRuntime,
198    cancellation: CancellationToken,
199    profile: FilesystemProfile,
200    limits: VolumeLimits,
201}
202
203/// One process-local serialization and publication-fencing boundary for every
204/// adapter, watcher, and transport view of the same checkout.
205pub struct SharedCheckout<A, O> {
206    state: tokio::sync::Mutex<SharedCheckoutState<A, O>>,
207    view_epoch: Arc<AtomicU64>,
208    view_gate: Arc<ViewGate>,
209}
210
211/// Exclusive access to a shared checkout. The retained view lease prevents a
212/// native callback from observing a partial external SDK operation.
213pub struct SharedCheckoutGuard<'a, A, O> {
214    _view: ViewWriteLease,
215    state: tokio::sync::MutexGuard<'a, SharedCheckoutState<A, O>>,
216}
217
218struct SharedCheckoutReadGuard<'a, A, O> {
219    _view: ViewReadLease,
220    state: tokio::sync::MutexGuard<'a, SharedCheckoutState<A, O>>,
221}
222
223/// Locked checkout state. Dereferencing reaches the canonical checkout while
224/// publication helpers retain an indeterminate operation in the same lock.
225pub struct SharedCheckoutState<A, O> {
226    checkout: Checkout<A, O>,
227    publication_operation: Option<OperationId>,
228    publication: MountPublication,
229    view_epoch: Arc<AtomicU64>,
230}
231
232impl<A, O> SharedCheckout<A, O> {
233    /// Creates the sole shared adapter boundary for one canonical checkout.
234    #[must_use]
235    pub fn new(checkout: Checkout<A, O>) -> Self {
236        Self::with_publication(checkout, MountPublication::CloseAndSync)
237    }
238
239    /// Creates a shared checkout with one explicit native publication policy.
240    #[must_use]
241    pub fn with_publication(checkout: Checkout<A, O>, publication: MountPublication) -> Self {
242        let view_epoch = Arc::new(AtomicU64::new(1));
243        Self {
244            state: tokio::sync::Mutex::new(SharedCheckoutState {
245                checkout,
246                publication_operation: None,
247                publication,
248                view_epoch: Arc::clone(&view_epoch),
249            }),
250            view_epoch,
251            view_gate: Arc::new(ViewGate::new()),
252        }
253    }
254
255    /// Serializes external checkout access and excludes native callbacks.
256    pub async fn lock(&self) -> SharedCheckoutGuard<'_, A, O> {
257        let view = self.view_gate.write().await;
258        let state = self.state.lock().await;
259        SharedCheckoutGuard { _view: view, state }
260    }
261
262    async fn read_lock(
263        &self,
264        owner: std::thread::ThreadId,
265    ) -> Result<SharedCheckoutReadGuard<'_, A, O>, MountSourceError> {
266        let view = self.view_gate.read_for_callback(owner, None).await?;
267        let state = self.state.lock().await;
268        Ok(SharedCheckoutReadGuard { _view: view, state })
269    }
270
271    pub(super) fn view_epoch(&self) -> u64 {
272        self.view_epoch.load(Ordering::Acquire)
273    }
274}
275
276impl<A, O> Deref for SharedCheckoutGuard<'_, A, O> {
277    type Target = SharedCheckoutState<A, O>;
278
279    fn deref(&self) -> &Self::Target {
280        &self.state
281    }
282}
283
284impl<A, O> DerefMut for SharedCheckoutGuard<'_, A, O> {
285    fn deref_mut(&mut self) -> &mut Self::Target {
286        &mut self.state
287    }
288}
289
290impl<A, O> Deref for SharedCheckoutReadGuard<'_, A, O> {
291    type Target = SharedCheckoutState<A, O>;
292
293    fn deref(&self) -> &Self::Target {
294        &self.state
295    }
296}
297
298impl<A, O> DerefMut for SharedCheckoutReadGuard<'_, A, O> {
299    fn deref_mut(&mut self) -> &mut Self::Target {
300        &mut self.state
301    }
302}
303
304impl<A, O> SharedCheckoutState<A, O> {
305    async fn seal(&mut self, cancellation: &CancellationToken) -> Result<(), MountSourceError>
306    where
307        A: AsyncAuthorityStore,
308        O: AsyncObjectStore,
309    {
310        let operation_id = self.retained_operation_id();
311        let result = seal_checkout(
312            &mut self.checkout,
313            operation_id,
314            boundary_budget(),
315            cancellation,
316        )
317        .await;
318        if result.is_ok()
319            || (matches!(result, Err(MountSourceError::Stale))
320                && self.checkout.mode().mutations == crate::model::MutationMode::PrivateOverlay)
321        {
322            self.clear_retained_operation(operation_id);
323        }
324        if result.is_ok() {
325            self.view_epoch.fetch_add(1, Ordering::AcqRel);
326        }
327        result
328    }
329
330    async fn seal_with_permit(
331        &mut self,
332        permit: crate::PublicationPermit,
333        force: bool,
334        cancellation: &CancellationToken,
335    ) -> Result<(), MountSourceError>
336    where
337        A: AsyncAuthorityStore,
338        O: AsyncObjectStore,
339    {
340        let operation_id = self.retained_operation_id();
341        let result = if force {
342            super::publication::seal_checkout_with_permit_force(
343                &mut self.checkout,
344                operation_id,
345                permit,
346                boundary_budget(),
347                cancellation,
348            )
349            .await
350        } else {
351            super::seal_checkout_with_permit(
352                &mut self.checkout,
353                operation_id,
354                permit,
355                boundary_budget(),
356                cancellation,
357            )
358            .await
359        };
360        if result.is_ok()
361            || (matches!(result, Err(MountSourceError::Stale))
362                && self.checkout.mode().mutations == crate::model::MutationMode::PrivateOverlay)
363        {
364            self.clear_retained_operation(operation_id);
365        }
366        if result.is_ok() {
367            self.view_epoch.fetch_add(1, Ordering::AcqRel);
368        }
369        result
370    }
371
372    pub(super) async fn publish_after_mutation(
373        &mut self,
374        cancellation: &CancellationToken,
375    ) -> Result<(), MountSourceError>
376    where
377        A: AsyncAuthorityStore,
378        O: AsyncObjectStore,
379    {
380        self.view_epoch.fetch_add(1, Ordering::AcqRel);
381        if self.publication == MountPublication::PerMutation {
382            self.seal(cancellation).await?;
383        }
384        Ok(())
385    }
386
387    async fn publish_at_native_boundary(
388        &mut self,
389        cancellation: &CancellationToken,
390    ) -> Result<(), MountSourceError>
391    where
392        A: AsyncAuthorityStore,
393        O: AsyncObjectStore,
394    {
395        if self.publication != MountPublication::Manual {
396            self.seal(cancellation).await?;
397        }
398        Ok(())
399    }
400
401    /// Rejects later mutation while one publication has an unresolved result.
402    ///
403    /// # Errors
404    ///
405    /// Returns [`MountSourceError::Stale`] while publication is indeterminate.
406    pub fn ensure_publication_resolved(&self) -> Result<(), MountSourceError> {
407        if self.publication_operation.is_some() {
408            return Err(MountSourceError::Stale);
409        }
410        Ok(())
411    }
412
413    /// Returns the exact operation retained until publication resolves.
414    pub fn retained_operation_id(&mut self) -> OperationId {
415        *self
416            .publication_operation
417            .get_or_insert_with(OperationId::new)
418    }
419
420    pub(super) fn has_retained_operation(&self) -> bool {
421        self.publication_operation.is_some()
422    }
423
424    /// Retains a caller-selected operation identity, or rejects a conflicting
425    /// unresolved publication from another surface sharing this checkout.
426    ///
427    /// # Errors
428    ///
429    /// Returns [`MountSourceError::Stale`] for a different retained operation.
430    pub fn retain_operation_id(
431        &mut self,
432        operation_id: OperationId,
433    ) -> Result<(), MountSourceError> {
434        match self.publication_operation {
435            None => {
436                self.publication_operation = Some(operation_id);
437                Ok(())
438            }
439            Some(retained) if retained == operation_id => Ok(()),
440            Some(_) => Err(MountSourceError::Stale),
441        }
442    }
443
444    /// Clears only an operation with a known terminal publication outcome.
445    pub fn clear_retained_operation(&mut self, operation_id: OperationId) {
446        if self.publication_operation == Some(operation_id) {
447            self.publication_operation = None;
448        }
449    }
450
451    pub(super) fn view_epoch(&self) -> u64 {
452        self.view_epoch.load(Ordering::Acquire)
453    }
454
455    pub(super) fn install_candidate(&mut self, candidate: Checkout<A, O>) {
456        self.checkout = candidate;
457    }
458}
459
460impl<A, O> Deref for SharedCheckoutState<A, O> {
461    type Target = Checkout<A, O>;
462
463    fn deref(&self) -> &Self::Target {
464        &self.checkout
465    }
466}
467
468impl<A, O> DerefMut for SharedCheckoutState<A, O> {
469    fn deref_mut(&mut self) -> &mut Self::Target {
470        &mut self.checkout
471    }
472}
473
474impl<A, O> CheckoutDetachedFile<A, O> {
475    fn attribute_name(&self, bytes: &[u8]) -> Result<AttributeName, MountSourceError> {
476        if self.profile != FilesystemProfile::Posix {
477            return Err(MountSourceError::Unsupported(
478                "native POSIX attributes require a POSIX volume profile".to_owned(),
479            ));
480        }
481        AttributeName::new(
482            AttributeClass::PosixXattr,
483            bytes.to_vec(),
484            self.limits.maximum_component_bytes,
485        )
486        .map_err(engine_error)
487    }
488
489    pub(super) fn snapshot(&self) -> Result<(FileRecord, FileMetadata, u64), MountSourceError>
490    where
491        A: AsyncAuthorityStore + Send + Sync,
492        O: AsyncObjectStore + Send + Sync,
493    {
494        self.runtime.wait(|| async {
495            let state = self.state.lock().await;
496            Ok((state.file.record(), state.metadata, state.mutation_epoch))
497        })
498    }
499}
500
501impl<A, O> MountOpenFile for CheckoutDetachedFile<A, O>
502where
503    A: AsyncAuthorityStore + Send + Sync + 'static,
504    O: AsyncObjectStore + Send + Sync + 'static,
505{
506    fn lookup(&self) -> Result<MountLookup, MountSourceError> {
507        self.runtime.wait(|| async {
508            let state = self.state.lock().await;
509            Ok(MountLookup {
510                node: MountNode {
511                    file_id: state.file.file_id(),
512                    kind: MountNodeKind::Regular,
513                    logical_bytes: state.file.logical_bytes(),
514                    link_count: 0,
515                    device: None,
516                },
517                metadata: state.metadata,
518            })
519        })
520    }
521
522    fn read_range(&self, offset: u64, length: u32) -> Result<Bytes, MountSourceError> {
523        self.runtime.wait(|| async {
524            let state = self.state.lock().await;
525            state
526                .file
527                .read_range(
528                    ByteRange {
529                        offset,
530                        length: u64::from(length),
531                    },
532                    boundary_budget(),
533                    &self.cancellation,
534                )
535                .await
536                .map(|receipt| receipt.value.bytes)
537                .map_err(engine_error)
538        })
539    }
540
541    fn read_up_to(&self, offset: u64, maximum_bytes: u32) -> Result<Bytes, MountSourceError> {
542        self.runtime.wait(|| async {
543            let state = self.state.lock().await;
544            let length = state
545                .file
546                .logical_bytes()
547                .saturating_sub(offset)
548                .min(u64::from(maximum_bytes));
549            if length == 0 {
550                return Ok(Bytes::new());
551            }
552            state
553                .file
554                .read_range(
555                    ByteRange { offset, length },
556                    boundary_budget(),
557                    &self.cancellation,
558                )
559                .await
560                .map(|receipt| receipt.value.bytes)
561                .map_err(engine_error)
562        })
563    }
564
565    fn seek(&self, offset: u64, target: MountSeekTarget) -> Result<Option<u64>, MountSourceError> {
566        self.runtime.wait(|| async {
567            let state = self.state.lock().await;
568            state
569                .file
570                .seek(
571                    offset,
572                    match target {
573                        MountSeekTarget::Data => ExtentSeekTarget::Data,
574                        MountSeekTarget::Hole => ExtentSeekTarget::Hole,
575                    },
576                    boundary_budget(),
577                    &self.cancellation,
578                )
579                .await
580                .map(|receipt| receipt.value)
581                .map_err(engine_error)
582        })
583    }
584
585    fn write_range(&self, offset: u64, bytes: Bytes) -> Result<(), MountSourceError> {
586        self.runtime.wait(|| async {
587            let mut state = self.state.lock().await;
588            state
589                .file
590                .write_range(offset, bytes, boundary_budget(), &self.cancellation)
591                .await
592                .map(|_| ())
593                .map_err(engine_error)?;
594            state.mutation_epoch = state
595                .mutation_epoch
596                .checked_add(1)
597                .ok_or(MountSourceError::Stale)?;
598            Ok(())
599        })
600    }
601
602    fn resize(&self, logical_bytes: u64) -> Result<(), MountSourceError> {
603        self.runtime.wait(|| async {
604            let mut state = self.state.lock().await;
605            state
606                .file
607                .resize(logical_bytes, boundary_budget(), &self.cancellation)
608                .await
609                .map(|_| ())
610                .map_err(engine_error)?;
611            state.mutation_epoch = state
612                .mutation_epoch
613                .checked_add(1)
614                .ok_or(MountSourceError::Stale)?;
615            Ok(())
616        })
617    }
618
619    fn allocate_range(
620        &self,
621        offset: u64,
622        length: u64,
623        operation: MountRangeAllocation,
624    ) -> Result<(), MountSourceError> {
625        self.runtime.wait(|| async {
626            let mut state = self.state.lock().await;
627            let range = ByteRange { offset, length };
628            match operation {
629                MountRangeAllocation::PunchHole => {
630                    state
631                        .file
632                        .zero_range(range, false, false, boundary_budget(), &self.cancellation)
633                        .await
634                }
635                MountRangeAllocation::ZeroRange { extend } => {
636                    state
637                        .file
638                        .zero_range(range, true, extend, boundary_budget(), &self.cancellation)
639                        .await
640                }
641                MountRangeAllocation::Preallocate { keep_size } => {
642                    state
643                        .file
644                        .preallocate(range, keep_size, boundary_budget(), &self.cancellation)
645                        .await
646                }
647            }
648            .map(|_| ())
649            .map_err(engine_error)?;
650            state.mutation_epoch = state
651                .mutation_epoch
652                .checked_add(1)
653                .ok_or(MountSourceError::Stale)?;
654            Ok(())
655        })
656    }
657
658    fn set_attributes(
659        &self,
660        metadata: FileMetadata,
661        logical_bytes: Option<u64>,
662    ) -> Result<(), MountSourceError> {
663        self.runtime.wait(|| async {
664            let mut state = self.state.lock().await;
665            state
666                .file
667                .set_attributes(
668                    metadata,
669                    logical_bytes,
670                    boundary_budget(),
671                    &self.cancellation,
672                )
673                .await
674                .map_err(engine_error)?;
675            state.metadata = metadata;
676            state.mutation_epoch = state
677                .mutation_epoch
678                .checked_add(1)
679                .ok_or(MountSourceError::Stale)?;
680            Ok(())
681        })
682    }
683
684    fn read_attribute(&self, name: &[u8]) -> Result<Option<Bytes>, MountSourceError> {
685        let name = self.attribute_name(name)?;
686        self.runtime.wait(|| async {
687            let state = self.state.lock().await;
688            state
689                .file
690                .read_named_attribute(&name, boundary_budget(), &self.cancellation)
691                .await
692                .map(|receipt| receipt.value)
693                .map_err(engine_error)
694        })
695    }
696
697    fn list_attributes(
698        &self,
699        cursor: Option<&[u8]>,
700        maximum_entries: u32,
701    ) -> Result<MountAttributePage, MountSourceError> {
702        let cursor = cursor.map(|name| self.attribute_name(name)).transpose()?;
703        self.runtime.wait(|| async {
704            let state = self.state.lock().await;
705            let receipt = state
706                .file
707                .list_named_attributes(
708                    cursor.as_ref(),
709                    maximum_entries,
710                    boundary_budget(),
711                    &self.cancellation,
712                )
713                .await
714                .map_err(engine_error)?;
715            let has_more = receipt.value.has_more;
716            let names = receipt
717                .value
718                .entries
719                .into_iter()
720                .map(|entry| {
721                    if entry.name.class() != AttributeClass::PosixXattr {
722                        return Err(MountSourceError::Unsupported(
723                            "POSIX mount encountered a non-POSIX named attribute".to_owned(),
724                        ));
725                    }
726                    Ok(entry.name.as_bytes().to_vec())
727                })
728                .collect::<Result<Vec<_>, _>>()?;
729            let next_cursor = has_more.then(|| names.last().cloned()).flatten();
730            Ok(MountAttributePage { names, next_cursor })
731        })
732    }
733
734    fn write_attribute(
735        &self,
736        name: &[u8],
737        value: Bytes,
738        mode: MountAttributeWriteMode,
739    ) -> Result<(), MountSourceError> {
740        let name = self.attribute_name(name)?;
741        let mode = match mode {
742            MountAttributeWriteMode::Upsert => NamedAttributeWriteMode::Upsert,
743            MountAttributeWriteMode::Create => NamedAttributeWriteMode::Create,
744            MountAttributeWriteMode::Replace => NamedAttributeWriteMode::Replace,
745        };
746        self.runtime.wait(|| async {
747            let mut state = self.state.lock().await;
748            let receipt = state
749                .file
750                .write_named_attribute(name, value, mode, boundary_budget(), &self.cancellation)
751                .await
752                .map_err(facade_error)?;
753            state.metadata = receipt.value;
754            state.mutation_epoch = state
755                .mutation_epoch
756                .checked_add(1)
757                .ok_or(MountSourceError::Stale)?;
758            Ok(())
759        })
760    }
761
762    fn remove_attribute(&self, name: &[u8]) -> Result<(), MountSourceError> {
763        let name = self.attribute_name(name)?;
764        self.runtime.wait(|| async {
765            let mut state = self.state.lock().await;
766            let receipt = state
767                .file
768                .remove_named_attribute(name, boundary_budget(), &self.cancellation)
769                .await
770                .map_err(facade_error)?;
771            state.metadata = receipt.value;
772            state.mutation_epoch = state
773                .mutation_epoch
774                .checked_add(1)
775                .ok_or(MountSourceError::Stale)?;
776            Ok(())
777        })
778    }
779}
780
781impl<A, O> CheckoutAttachedFile<A, O> {
782    fn attribute_name(&self, bytes: &[u8]) -> Result<AttributeName, MountSourceError> {
783        if self.profile != FilesystemProfile::Posix {
784            return Err(MountSourceError::Unsupported(
785                "native POSIX attributes require a POSIX volume profile".to_owned(),
786            ));
787        }
788        AttributeName::new(
789            AttributeClass::PosixXattr,
790            bytes.to_vec(),
791            self.limits.maximum_component_bytes,
792        )
793        .map_err(engine_error)
794    }
795}
796
797impl<A, O> MountOpenFile for CheckoutAttachedFile<A, O>
798where
799    A: AsyncAuthorityStore + Send + Sync + 'static,
800    O: AsyncObjectStore + Send + Sync + 'static,
801{
802    fn lookup(&self) -> Result<MountLookup, MountSourceError> {
803        let owner = ViewGate::callback_owner();
804        self.runtime.wait(|| async {
805            let mut checkout = self.checkout.read_lock(owner).await?;
806            let record = checkout
807                .read_file_record_by_id(self.file_id, boundary_budget(), &self.cancellation)
808                .await
809                .map_err(engine_error)?
810                .value;
811            let metadata = checkout
812                .read_metadata_by_id(self.file_id, boundary_budget(), &self.cancellation)
813                .await
814                .map_err(engine_error)?
815                .value;
816            Ok(MountLookup {
817                node: mount_node(record),
818                metadata,
819            })
820        })
821    }
822
823    fn read_range(&self, offset: u64, length: u32) -> Result<Bytes, MountSourceError> {
824        let owner = ViewGate::callback_owner();
825        self.runtime.wait(|| async {
826            let mut checkout = self.checkout.read_lock(owner).await?;
827            checkout
828                .read_file_range_by_id(
829                    self.file_id,
830                    ByteRange {
831                        offset,
832                        length: u64::from(length),
833                    },
834                    boundary_budget(),
835                    &self.cancellation,
836                )
837                .await
838                .map(|receipt| receipt.value.bytes)
839                .map_err(engine_error)
840        })
841    }
842
843    fn read_up_to(&self, offset: u64, maximum_bytes: u32) -> Result<Bytes, MountSourceError> {
844        let owner = ViewGate::callback_owner();
845        self.runtime.wait(|| async {
846            let mut checkout = self.checkout.read_lock(owner).await?;
847            checkout
848                .read_file_up_to_by_id(
849                    self.file_id,
850                    offset,
851                    maximum_bytes,
852                    boundary_budget(),
853                    &self.cancellation,
854                )
855                .await
856                .map(|receipt| receipt.value.bytes)
857                .map_err(engine_error)
858        })
859    }
860
861    fn seek(&self, offset: u64, target: MountSeekTarget) -> Result<Option<u64>, MountSourceError> {
862        let owner = ViewGate::callback_owner();
863        self.runtime.wait(|| async {
864            let mut checkout = self.checkout.read_lock(owner).await?;
865            checkout
866                .seek_file_extent_by_id(
867                    self.file_id,
868                    offset,
869                    match target {
870                        MountSeekTarget::Data => ExtentSeekTarget::Data,
871                        MountSeekTarget::Hole => ExtentSeekTarget::Hole,
872                    },
873                    boundary_budget(),
874                    &self.cancellation,
875                )
876                .await
877                .map(|receipt| receipt.value)
878                .map_err(engine_error)
879        })
880    }
881
882    fn write_range(&self, offset: u64, bytes: Bytes) -> Result<(), MountSourceError> {
883        self.runtime.wait(|| async {
884            let mut checkout = self.checkout.lock().await;
885            checkout.ensure_publication_resolved()?;
886            checkout
887                .write_file_by_id(
888                    self.file_id,
889                    offset,
890                    bytes,
891                    boundary_budget(),
892                    &self.cancellation,
893                )
894                .await
895                .map_err(engine_error)?;
896            checkout.publish_after_mutation(&self.cancellation).await
897        })
898    }
899
900    fn resize(&self, logical_bytes: u64) -> Result<(), MountSourceError> {
901        self.runtime.wait(|| async {
902            let mut checkout = self.checkout.lock().await;
903            checkout.ensure_publication_resolved()?;
904            checkout
905                .resize_file_by_id(
906                    self.file_id,
907                    logical_bytes,
908                    boundary_budget(),
909                    &self.cancellation,
910                )
911                .await
912                .map_err(engine_error)?;
913            checkout.publish_after_mutation(&self.cancellation).await
914        })
915    }
916
917    fn allocate_range(
918        &self,
919        offset: u64,
920        length: u64,
921        operation: MountRangeAllocation,
922    ) -> Result<(), MountSourceError> {
923        self.runtime.wait(|| async {
924            let mut checkout = self.checkout.lock().await;
925            checkout.ensure_publication_resolved()?;
926            let range = ByteRange { offset, length };
927            match operation {
928                MountRangeAllocation::PunchHole => {
929                    checkout
930                        .zero_file_range_by_id(
931                            self.file_id,
932                            range,
933                            false,
934                            false,
935                            boundary_budget(),
936                            &self.cancellation,
937                        )
938                        .await
939                }
940                MountRangeAllocation::ZeroRange { extend } => {
941                    checkout
942                        .zero_file_range_by_id(
943                            self.file_id,
944                            range,
945                            true,
946                            extend,
947                            boundary_budget(),
948                            &self.cancellation,
949                        )
950                        .await
951                }
952                MountRangeAllocation::Preallocate { keep_size } => {
953                    checkout
954                        .preallocate_file_by_id(
955                            self.file_id,
956                            range,
957                            keep_size,
958                            boundary_budget(),
959                            &self.cancellation,
960                        )
961                        .await
962                }
963            }
964            .map_err(engine_error)?;
965            checkout.publish_after_mutation(&self.cancellation).await
966        })
967    }
968
969    fn set_attributes(
970        &self,
971        metadata: FileMetadata,
972        logical_bytes: Option<u64>,
973    ) -> Result<(), MountSourceError> {
974        self.runtime.wait(|| async {
975            let mut checkout = self.checkout.lock().await;
976            checkout.ensure_publication_resolved()?;
977            checkout
978                .set_attributes_by_id(
979                    self.file_id,
980                    metadata,
981                    logical_bytes,
982                    boundary_budget(),
983                    &self.cancellation,
984                )
985                .await
986                .map_err(engine_error)?;
987            checkout.publish_after_mutation(&self.cancellation).await
988        })
989    }
990
991    fn read_attribute(&self, name: &[u8]) -> Result<Option<Bytes>, MountSourceError> {
992        let name = self.attribute_name(name)?;
993        let owner = ViewGate::callback_owner();
994        self.runtime.wait(|| async {
995            let mut checkout = self.checkout.read_lock(owner).await?;
996            checkout
997                .read_named_attribute_by_id(
998                    self.file_id,
999                    &name,
1000                    boundary_budget(),
1001                    &self.cancellation,
1002                )
1003                .await
1004                .map(|receipt| receipt.value)
1005                .map_err(engine_error)
1006        })
1007    }
1008
1009    fn list_attributes(
1010        &self,
1011        cursor: Option<&[u8]>,
1012        maximum_entries: u32,
1013    ) -> Result<MountAttributePage, MountSourceError> {
1014        let cursor = cursor.map(|name| self.attribute_name(name)).transpose()?;
1015        let owner = ViewGate::callback_owner();
1016        self.runtime.wait(|| async {
1017            let mut checkout = self.checkout.read_lock(owner).await?;
1018            let receipt = checkout
1019                .list_named_attributes_by_id(
1020                    self.file_id,
1021                    cursor.as_ref(),
1022                    maximum_entries,
1023                    boundary_budget(),
1024                    &self.cancellation,
1025                )
1026                .await
1027                .map_err(engine_error)?;
1028            let has_more = receipt.value.has_more;
1029            let names = receipt
1030                .value
1031                .entries
1032                .into_iter()
1033                .map(|entry| {
1034                    if entry.name.class() != AttributeClass::PosixXattr {
1035                        return Err(MountSourceError::Unsupported(
1036                            "POSIX mount encountered a non-POSIX named attribute".to_owned(),
1037                        ));
1038                    }
1039                    Ok(entry.name.as_bytes().to_vec())
1040                })
1041                .collect::<Result<Vec<_>, _>>()?;
1042            let next_cursor = has_more.then(|| names.last().cloned()).flatten();
1043            Ok(MountAttributePage { names, next_cursor })
1044        })
1045    }
1046
1047    fn write_attribute(
1048        &self,
1049        name: &[u8],
1050        value: Bytes,
1051        mode: MountAttributeWriteMode,
1052    ) -> Result<(), MountSourceError> {
1053        let name = self.attribute_name(name)?;
1054        let mode = match mode {
1055            MountAttributeWriteMode::Upsert => NamedAttributeWriteMode::Upsert,
1056            MountAttributeWriteMode::Create => NamedAttributeWriteMode::Create,
1057            MountAttributeWriteMode::Replace => NamedAttributeWriteMode::Replace,
1058        };
1059        self.runtime.wait(|| async {
1060            let mut checkout = self.checkout.lock().await;
1061            checkout.ensure_publication_resolved()?;
1062            checkout
1063                .write_named_attribute_by_id(
1064                    self.file_id,
1065                    name,
1066                    value,
1067                    mode,
1068                    boundary_budget(),
1069                    &self.cancellation,
1070                )
1071                .await
1072                .map_err(facade_error)?;
1073            checkout.publish_after_mutation(&self.cancellation).await
1074        })
1075    }
1076
1077    fn remove_attribute(&self, name: &[u8]) -> Result<(), MountSourceError> {
1078        let name = self.attribute_name(name)?;
1079        self.runtime.wait(|| async {
1080            let mut checkout = self.checkout.lock().await;
1081            checkout.ensure_publication_resolved()?;
1082            checkout
1083                .remove_named_attribute_by_id(
1084                    self.file_id,
1085                    name,
1086                    boundary_budget(),
1087                    &self.cancellation,
1088                )
1089                .await
1090                .map_err(facade_error)?;
1091            checkout.publish_after_mutation(&self.cancellation).await
1092        })
1093    }
1094}
1095
1096impl<A, O> CheckoutMountSource<A, O> {
1097    pub(super) fn shared_checkout(&self) -> &SharedCheckout<A, O> {
1098        &self.checkout
1099    }
1100
1101    pub(super) fn cancellation(&self) -> &CancellationToken {
1102        &self.cancellation
1103    }
1104
1105    /// Binds an existing source handle to the same stable identity once a
1106    /// peer stages that identity in the mounted checkout candidate.
1107    pub(super) fn attached_file_by_id(
1108        &self,
1109        file_id: FileId,
1110    ) -> Result<Option<Arc<dyn MountOpenFile>>, MountSourceError>
1111    where
1112        A: AsyncAuthorityStore + Send + Sync + 'static,
1113        O: AsyncObjectStore + Send + Sync + 'static,
1114    {
1115        let owner = ViewGate::callback_owner();
1116        let present = self.runtime.wait(|| async {
1117            let mut checkout = self.checkout.read_lock(owner).await?;
1118            match checkout
1119                .read_file_record_by_id(file_id, boundary_budget(), &self.cancellation)
1120                .await
1121            {
1122                Ok(record) if record.value.kind == FileKind::Regular => Ok(true),
1123                Ok(_) => Err(MountSourceError::Stale),
1124                Err(failure) if matches!(failure.error, FsError::NotFound) => Ok(false),
1125                Err(failure) => Err(engine_error(failure)),
1126            }
1127        })?;
1128        Ok(present.then(|| {
1129            Arc::new(CheckoutAttachedFile {
1130                checkout: Arc::clone(&self.checkout),
1131                file_id,
1132                runtime: self.runtime.clone(),
1133                cancellation: self.cancellation.clone(),
1134                profile: self.profile,
1135                limits: self.limits,
1136            }) as Arc<dyn MountOpenFile>
1137        }))
1138    }
1139
1140    pub(super) fn record_by_id(&self, file_id: FileId) -> Result<FileRecord, MountSourceError>
1141    where
1142        A: AsyncAuthorityStore + Send + Sync,
1143        O: AsyncObjectStore + Send + Sync,
1144    {
1145        self.runtime.wait(|| async {
1146            let mut checkout = self.checkout.lock().await;
1147            checkout.ensure_publication_resolved()?;
1148            checkout
1149                .read_file_record_by_id(file_id, boundary_budget(), &self.cancellation)
1150                .await
1151                .map(|receipt| receipt.value)
1152                .map_err(engine_error)
1153        })
1154    }
1155
1156    pub(super) fn detached_file_from_record(
1157        &self,
1158        record: FileRecord,
1159        metadata: FileMetadata,
1160    ) -> Result<Arc<CheckoutDetachedFile<A, O>>, MountSourceError>
1161    where
1162        A: AsyncAuthorityStore + Send + Sync,
1163        O: AsyncObjectStore + Send + Sync,
1164    {
1165        let file = self.runtime.wait(|| async {
1166            let checkout = self.checkout.lock().await;
1167            Ok(checkout.detached_from_record(record))
1168        })?;
1169        Ok(Arc::new(CheckoutDetachedFile {
1170            state: tokio::sync::Mutex::new(DetachedMountState {
1171                file,
1172                metadata,
1173                mutation_epoch: 0,
1174            }),
1175            runtime: self.runtime.clone(),
1176            cancellation: self.cancellation.clone(),
1177            profile: self.profile,
1178            limits: self.limits,
1179        }))
1180    }
1181
1182    /// Creates an independently cancellable adapter using the shared callback runtime.
1183    ///
1184    /// # Errors
1185    ///
1186    /// Returns a driver failure when the bounded callback runtime cannot start.
1187    pub fn new(
1188        checkout: Arc<SharedCheckout<A, O>>,
1189        config: VolumeConfig,
1190    ) -> Result<Self, NativeMountError> {
1191        let root = NamespacePath::new(Vec::new(), config.limits)
1192            .map_err(|error| NativeMountError::Driver(error.to_string()))?;
1193        Self::new_at(checkout, config, root)
1194    }
1195
1196    /// Creates an adapter rooted at an authenticated workspace subtree.
1197    ///
1198    /// # Errors
1199    ///
1200    /// Returns a profile or runtime failure before any native mount starts.
1201    pub fn new_at(
1202        checkout: Arc<SharedCheckout<A, O>>,
1203        config: VolumeConfig,
1204        root: NamespacePath,
1205    ) -> Result<Self, NativeMountError> {
1206        if !profile_is_native(config.profile) {
1207            return Err(NativeMountError::ProfileUnavailable {
1208                profile: profile_name(config.profile),
1209                platform: std::env::consts::OS,
1210            });
1211        }
1212        let runtime = CallbackRuntime::create()?;
1213        Ok(Self {
1214            checkout,
1215            root,
1216            limits: config.limits,
1217            profile: config.profile,
1218            cancellation: CancellationToken::new(),
1219            capture_gate: StdMutex::new(CaptureSourceGate::default()),
1220            runtime,
1221        })
1222    }
1223
1224    /// Returns the checkout's owning volume without filesystem I/O.
1225    ///
1226    /// # Errors
1227    ///
1228    /// Returns stale when callback admission exceeds the finite deadline.
1229    pub fn volume_id(&self) -> Result<VolumeId, MountSourceError>
1230    where
1231        A: Send + Sync,
1232        O: Send + Sync,
1233    {
1234        let owner = ViewGate::callback_owner();
1235        self.runtime
1236            .wait(|| async { Ok(self.checkout.read_lock(owner).await?.volume_id()) })
1237    }
1238
1239    /// Cancels future and in-flight canonical operations owned by this mount.
1240    pub fn cancel(&self) {
1241        let requests = {
1242            let mut gate = self
1243                .capture_gate
1244                .lock()
1245                .unwrap_or_else(std::sync::PoisonError::into_inner);
1246            gate.cancelled = true;
1247            let requests = gate
1248                .active
1249                .iter()
1250                .filter_map(Weak::upgrade)
1251                .collect::<Vec<_>>();
1252            for request in &requests {
1253                let _ = request.state.compare_exchange(
1254                    CAPTURE_PENDING,
1255                    CAPTURE_CANCELLED,
1256                    Ordering::AcqRel,
1257                    Ordering::Acquire,
1258                );
1259            }
1260            gate.active.clear();
1261            requests
1262        };
1263        self.cancellation.cancel();
1264        for request in requests {
1265            request.cancellation.cancel();
1266        }
1267    }
1268
1269    /// Publishes every pending mount mutation regardless of automatic policy.
1270    ///
1271    /// # Errors
1272    ///
1273    /// Returns the exact fenced publication failure and retains retry identity.
1274    pub fn sync(&self) -> Result<(), MountSourceError>
1275    where
1276        A: AsyncAuthorityStore + Send + Sync,
1277        O: AsyncObjectStore + Send + Sync,
1278    {
1279        self.runtime.wait(|| async {
1280            let mut checkout = self.checkout.lock().await;
1281            checkout.seal(&self.cancellation).await
1282        })
1283    }
1284
1285    /// Asynchronously publishes every pending mount mutation.
1286    ///
1287    /// This is the native customer handle path and does not enter the blocking
1288    /// foreign-thread callback bridge.
1289    ///
1290    /// # Errors
1291    ///
1292    /// Returns the exact fenced publication failure and retains retry identity.
1293    pub async fn sync_async(&self) -> Result<(), MountSourceError>
1294    where
1295        A: AsyncAuthorityStore,
1296        O: AsyncObjectStore,
1297    {
1298        let mut checkout = self.checkout.lock().await;
1299        checkout.seal(&self.cancellation).await
1300    }
1301
1302    /// Publishes pending mutations only while the supplied operation lease is
1303    /// still active in the shared authority provider.
1304    pub async fn sync_async_with_permit(
1305        &self,
1306        permit: crate::PublicationPermit,
1307    ) -> Result<(), MountSourceError>
1308    where
1309        A: AsyncAuthorityStore,
1310        O: AsyncObjectStore,
1311    {
1312        let mut checkout = self.checkout.lock().await;
1313        checkout
1314            .seal_with_permit(permit, false, &self.cancellation)
1315            .await
1316    }
1317
1318    pub(super) async fn sync_async_with_permit_force(
1319        &self,
1320        permit: crate::PublicationPermit,
1321    ) -> Result<(), MountSourceError>
1322    where
1323        A: AsyncAuthorityStore,
1324        O: AsyncObjectStore,
1325    {
1326        let mut checkout = self.checkout.lock().await;
1327        checkout
1328            .seal_with_permit(permit, true, &self.cancellation)
1329            .await
1330    }
1331
1332    /// Safely advances a clean mounted checkout to the workspace head.
1333    ///
1334    /// Callers must publish pending native mutations first. Exact read
1335    /// dependencies are retained and checked against the candidate head;
1336    /// conflict leaves the mounted generation unchanged.
1337    pub async fn refresh_async(&self) -> Result<(), MountSourceError>
1338    where
1339        A: AsyncAuthorityStore,
1340        O: AsyncObjectStore,
1341    {
1342        let mut checkout = self.checkout.lock().await;
1343        checkout.ensure_publication_resolved()?;
1344        let decision = checkout
1345            .rebase_head(
1346                self.limits.maximum_checkout_dependencies,
1347                WorkBudget::UNBOUNDED,
1348                &self.cancellation,
1349            )
1350            .await
1351            .map_err(engine_error)?;
1352        match decision.value {
1353            RebaseDecision::Safe { .. } => {
1354                self.checkout.view_gate.begin_transition();
1355                self.checkout.view_gate.finish_transition();
1356                checkout.view_epoch.fetch_add(1, Ordering::AcqRel);
1357                Ok(())
1358            }
1359            RebaseDecision::Conflicted { .. } => Err(MountSourceError::Stale),
1360        }
1361    }
1362
1363    /// Adopts the workspace head after this checkout's mutations were
1364    /// synchronized and a separate fenced operation published that head.
1365    pub async fn advance_to_head_async(&self) -> Result<(), MountSourceError>
1366    where
1367        A: AsyncAuthorityStore,
1368        O: AsyncObjectStore,
1369    {
1370        let mut checkout = self.checkout.lock().await;
1371        checkout.ensure_publication_resolved()?;
1372        checkout
1373            .refresh_head(WorkBudget::UNBOUNDED, &self.cancellation)
1374            .await
1375            .map_err(engine_error)?;
1376        self.checkout.view_gate.begin_transition();
1377        self.checkout.view_gate.finish_transition();
1378        checkout.view_epoch.fetch_add(1, Ordering::AcqRel);
1379        Ok(())
1380    }
1381
1382    fn path(&self, path: &MountPath) -> Result<NamespacePath, MountSourceError> {
1383        let mut components = self.root.components().to_vec();
1384        components.extend(
1385            path.components()
1386                .iter()
1387                .map(|component| self.logical_name(component))
1388                .collect::<Result<Vec<_>, _>>()?,
1389        );
1390        NamespacePath::new(components, self.limits).map_err(engine_error)
1391    }
1392
1393    fn logical_name(&self, bytes: &[u8]) -> Result<LogicalName, MountSourceError> {
1394        let (encoding, logical_bytes) = match (self.profile, std::env::consts::OS) {
1395            (FilesystemProfile::Posix, "linux" | "macos") => {
1396                (NameEncoding::PosixBytes, bytes.to_vec())
1397            }
1398            (FilesystemProfile::Portable | FilesystemProfile::Browser, "linux" | "macos") => {
1399                std::str::from_utf8(bytes).map_err(|_| {
1400                    MountSourceError::Unsupported(
1401                        "portable volume cannot represent a non-UTF-8 native name".to_owned(),
1402                    )
1403                })?;
1404                (NameEncoding::Utf8, bytes.to_vec())
1405            }
1406            (FilesystemProfile::Windows, "windows") => {
1407                (NameEncoding::WindowsUtf16Le, bytes.to_vec())
1408            }
1409            (FilesystemProfile::Portable | FilesystemProfile::Browser, "windows") => {
1410                if !bytes.len().is_multiple_of(2) {
1411                    return Err(MountSourceError::Invalid(
1412                        "native UTF-16 name has an odd byte length".to_owned(),
1413                    ));
1414                }
1415                let units = bytes
1416                    .chunks_exact(2)
1417                    .map(|unit| {
1418                        let bytes: [u8; 2] = unit.try_into().unwrap_or([0, 0]);
1419                        u16::from_le_bytes(bytes)
1420                    })
1421                    .collect::<Vec<_>>();
1422                let value = String::from_utf16(&units).map_err(|_| {
1423                    MountSourceError::Unsupported(
1424                        "portable volume cannot represent an unpaired UTF-16 native name"
1425                            .to_owned(),
1426                    )
1427                })?;
1428                (NameEncoding::Utf8, value.into_bytes())
1429            }
1430            _ => {
1431                return Err(MountSourceError::Unsupported(
1432                    "volume name profile is incompatible with this native mount".to_owned(),
1433                ));
1434            }
1435        };
1436        LogicalName::new(encoding, logical_bytes, self.limits.maximum_component_bytes)
1437            .map_err(engine_error)
1438    }
1439
1440    fn posix_attribute_name(&self, bytes: &[u8]) -> Result<AttributeName, MountSourceError> {
1441        if self.profile != FilesystemProfile::Posix {
1442            return Err(MountSourceError::Unsupported(
1443                "native POSIX attributes require a POSIX volume profile".to_owned(),
1444            ));
1445        }
1446        AttributeName::new(
1447            AttributeClass::PosixXattr,
1448            bytes.to_vec(),
1449            self.limits.maximum_component_bytes,
1450        )
1451        .map_err(engine_error)
1452    }
1453
1454    fn native_name(name: &LogicalName) -> Result<Vec<u8>, MountSourceError> {
1455        match (name.encoding(), std::env::consts::OS) {
1456            (NameEncoding::Utf8 | NameEncoding::PosixBytes, "linux" | "macos") => {
1457                Ok(name.as_bytes().to_vec())
1458            }
1459            (NameEncoding::WindowsUtf16Le, "windows") => Ok(name.as_bytes().to_vec()),
1460            (NameEncoding::Utf8, "windows") => {
1461                let text = std::str::from_utf8(name.as_bytes()).map_err(|_| {
1462                    MountSourceError::Invalid("authenticated UTF-8 name is malformed".to_owned())
1463                })?;
1464                Ok(text.encode_utf16().flat_map(u16::to_le_bytes).collect())
1465            }
1466            _ => Err(MountSourceError::Unsupported(
1467                "authenticated name encoding is incompatible with this native mount".to_owned(),
1468            )),
1469        }
1470    }
1471
1472    pub(crate) fn capture_host_paths_with_identity(
1473        &self,
1474        source_root: &Path,
1475        paths: &[MountPath],
1476        expected_root_identity: NativeRootIdentity,
1477    ) -> Result<(), MountSourceError>
1478    where
1479        A: AsyncAuthorityStore + Send + Sync + 'static,
1480        O: AsyncObjectStore + Send + Sync + 'static,
1481    {
1482        self.capture_host_with_identity(
1483            source_root,
1484            paths,
1485            expected_root_identity,
1486            HostCaptureScope::ExactPaths,
1487            None,
1488        )
1489    }
1490
1491    fn capture_host_with_identity(
1492        &self,
1493        source_root: &Path,
1494        paths: &[MountPath],
1495        expected_root_identity: NativeRootIdentity,
1496        scope: HostCaptureScope,
1497        baseline: Option<&NativeViewBaseline>,
1498    ) -> Result<(), MountSourceError>
1499    where
1500        A: AsyncAuthorityStore + Send + Sync + 'static,
1501        O: AsyncObjectStore + Send + Sync + 'static,
1502    {
1503        self.runtime.wait(|| {
1504            self.capture_host_with_identity_async(
1505                source_root,
1506                paths,
1507                expected_root_identity,
1508                scope,
1509                HostCapturePolicy {
1510                    baseline,
1511                    budget: boundary_budget(),
1512                    request: None,
1513                },
1514            )
1515        })
1516    }
1517
1518    /// The SDK lifecycle path uses the same capture transaction as callbacks,
1519    /// but its admitted budget must not inherit the callback deadline or cap.
1520    pub(super) async fn capture_host_with_identity_budgeted(
1521        self: &Arc<Self>,
1522        source_root: &Path,
1523        paths: &[MountPath],
1524        expected_root_identity: NativeRootIdentity,
1525        scope: HostCaptureScope,
1526        baseline: Arc<NativeViewBaseline>,
1527        budget: WorkBudget,
1528    ) -> Result<(), MountSourceError>
1529    where
1530        A: AsyncAuthorityStore + Send + Sync + 'static,
1531        O: AsyncObjectStore + Send + Sync + 'static,
1532    {
1533        let source = Arc::clone(self);
1534        let source_root = source_root.to_path_buf();
1535        let paths = paths.to_vec();
1536        let runtime = tokio::runtime::Handle::current();
1537        let request = Arc::new(CaptureCommitGate {
1538            cancellation: CancellationToken::new(),
1539            state: AtomicU8::new(CAPTURE_PENDING),
1540        });
1541        {
1542            let mut gate = self
1543                .capture_gate
1544                .lock()
1545                .unwrap_or_else(std::sync::PoisonError::into_inner);
1546            if gate.cancelled {
1547                return Err(MountSourceError::Stale);
1548            }
1549            self.cancellation
1550                .check()
1551                .map_err(|error| MountSourceError::Engine(error.to_string()))?;
1552            gate.active.retain(|request| request.strong_count() > 0);
1553            gate.active.push(Arc::downgrade(&request));
1554        }
1555        let _cancel_on_drop = CancelCaptureOnDrop(Arc::clone(&request));
1556        tokio::task::spawn_blocking(move || {
1557            runtime.block_on(source.capture_host_with_identity_async(
1558                &source_root,
1559                &paths,
1560                expected_root_identity,
1561                scope,
1562                HostCapturePolicy {
1563                    baseline: Some(&baseline),
1564                    budget,
1565                    request: Some(&request),
1566                },
1567            ))
1568        })
1569        .await
1570        .map_err(|error| MountSourceError::Engine(error.to_string()))?
1571    }
1572
1573    async fn capture_host_with_identity_async(
1574        &self,
1575        source_root: &Path,
1576        paths: &[MountPath],
1577        expected_root_identity: NativeRootIdentity,
1578        scope: HostCaptureScope,
1579        policy: HostCapturePolicy<'_>,
1580    ) -> Result<(), MountSourceError>
1581    where
1582        A: AsyncAuthorityStore + Send + Sync + 'static,
1583        O: AsyncObjectStore + Send + Sync + 'static,
1584    {
1585        let cancellation = policy
1586            .request
1587            .map_or(&self.cancellation, |request| &request.cancellation);
1588        cancellation
1589            .check()
1590            .map_err(|error| MountSourceError::Engine(error.to_string()))?;
1591        self.cancellation
1592            .check()
1593            .map_err(|error| MountSourceError::Engine(error.to_string()))?;
1594        if paths.is_empty() {
1595            return Ok(());
1596        }
1597        if matches!(scope, HostCaptureScope::ExactPaths)
1598            && paths.len() > MAX_NATIVE_EXACT_CAPTURE_PATHS
1599        {
1600            return Err(MountSourceError::Invalid(
1601                "capture path set exceeds the native per-operation limit".to_owned(),
1602            ));
1603        }
1604        let paths = paths
1605            .iter()
1606            .map(|path| self.path(path))
1607            .collect::<Result<Vec<_>, _>>()?;
1608        let maximum_paths = match scope {
1609            HostCaptureScope::ExactPaths => u32::try_from(paths.len())
1610                .map_err(|_| MountSourceError::Invalid("too many capture paths".to_owned()))?,
1611            HostCaptureScope::Subtree => MAX_NATIVE_SUBTREE_CAPTURE_PATHS,
1612        };
1613        {
1614            let options = CaptureOptions {
1615                source_root: source_root.to_path_buf(),
1616                expected_root_identity,
1617                maximum_paths,
1618                maximum_extent_spans: 65_536,
1619            };
1620            for _ in 0..3 {
1621                let (mut candidate, epoch) = {
1622                    let checkout = self.checkout.lock().await;
1623                    checkout.ensure_publication_resolved()?;
1624                    (
1625                        checkout.private_candidate(),
1626                        checkout.view_epoch.load(Ordering::Acquire),
1627                    )
1628                };
1629                // Projected host reads may call back into this checkout. Never
1630                // hold its view gate while observing the virtualization root.
1631                match scope {
1632                    HostCaptureScope::ExactPaths => {
1633                        capture_paths_batched_with_baseline(
1634                            &mut candidate,
1635                            &paths,
1636                            &options,
1637                            64,
1638                            policy.budget,
1639                            cancellation,
1640                            policy.baseline,
1641                        )
1642                        .await
1643                        .map_err(engine_error)?;
1644                    }
1645                    HostCaptureScope::Subtree => {
1646                        let root = paths.first().cloned().ok_or_else(|| {
1647                            MountSourceError::Invalid("missing capture subtree".to_owned())
1648                        })?;
1649                        capture_subtrees_with_policy_and_baseline(
1650                            &mut candidate,
1651                            &[root],
1652                            &options,
1653                            &crate::CapturePolicy::allow_all(),
1654                            policy.budget,
1655                            cancellation,
1656                            policy.baseline,
1657                        )
1658                        .await
1659                        .map_err(engine_error)?;
1660                    }
1661                }
1662                if self
1663                    .commit_host_capture(candidate, epoch, policy.request)
1664                    .await?
1665                {
1666                    return Ok(());
1667                }
1668            }
1669            Err(MountSourceError::Stale)
1670        }
1671    }
1672
1673    async fn commit_host_capture(
1674        &self,
1675        candidate: Checkout<A, O>,
1676        epoch: u64,
1677        request: Option<&CaptureCommitGate>,
1678    ) -> Result<bool, MountSourceError>
1679    where
1680        A: AsyncAuthorityStore + Send + Sync + 'static,
1681        O: AsyncObjectStore + Send + Sync + 'static,
1682    {
1683        let mut checkout = self.checkout.lock().await;
1684        checkout.ensure_publication_resolved()?;
1685        if checkout.view_epoch.load(Ordering::Acquire) != epoch {
1686            return Ok(false);
1687        }
1688        {
1689            let source_gate = self
1690                .capture_gate
1691                .lock()
1692                .unwrap_or_else(std::sync::PoisonError::into_inner);
1693            if source_gate.cancelled {
1694                return Err(MountSourceError::Stale);
1695            }
1696            self.cancellation
1697                .check()
1698                .map_err(|error| MountSourceError::Engine(error.to_string()))?;
1699            if let Some(request) = request {
1700                request
1701                    .cancellation
1702                    .check()
1703                    .map_err(|error| MountSourceError::Engine(error.to_string()))?;
1704                if checkout.publication != MountPublication::Manual {
1705                    return Err(MountSourceError::Invalid(
1706                        "lifecycle capture requires manual publication".to_owned(),
1707                    ));
1708                }
1709                request
1710                    .state
1711                    .compare_exchange(
1712                        CAPTURE_PENDING,
1713                        CAPTURE_COMMITTED,
1714                        Ordering::AcqRel,
1715                        Ordering::Acquire,
1716                    )
1717                    .map_err(|_| MountSourceError::Stale)?;
1718                checkout.checkout = candidate;
1719                checkout.view_epoch.fetch_add(1, Ordering::AcqRel);
1720                return Ok(true);
1721            }
1722            checkout.checkout = candidate;
1723            checkout.view_epoch.fetch_add(1, Ordering::AcqRel);
1724        }
1725        checkout
1726            .publish_at_native_boundary(&self.cancellation)
1727            .await?;
1728        Ok(true)
1729    }
1730}
1731
1732fn profile_is_native(profile: FilesystemProfile) -> bool {
1733    match std::env::consts::OS {
1734        "linux" | "macos" => !matches!(profile, FilesystemProfile::Windows),
1735        "windows" => !matches!(profile, FilesystemProfile::Posix),
1736        _ => false,
1737    }
1738}
1739
1740fn profile_name(profile: FilesystemProfile) -> &'static str {
1741    match profile {
1742        FilesystemProfile::Portable => "portable",
1743        FilesystemProfile::Posix => "posix",
1744        FilesystemProfile::Windows => "windows",
1745        FilesystemProfile::Browser => "browser",
1746    }
1747}
1748
1749impl<A, O> MountFilesystem for CheckoutMountSource<A, O>
1750where
1751    A: AsyncAuthorityStore + Send + Sync + 'static,
1752    O: AsyncObjectStore + Send + Sync + 'static,
1753{
1754    fn supports_posix_named_attributes(&self) -> bool {
1755        self.profile == FilesystemProfile::Posix
1756    }
1757
1758    fn view_epoch(&self) -> Option<u64> {
1759        Some(self.checkout.view_epoch())
1760    }
1761
1762    fn binding_epoch(&self) -> Option<u64> {
1763        Some(self.checkout.view_gate.generation())
1764    }
1765
1766    fn view_is_stable(&self) -> bool {
1767        self.checkout.view_gate.is_stable()
1768    }
1769
1770    fn acquire_view_lease(
1771        &self,
1772        expected_epoch: Option<u64>,
1773    ) -> Result<Box<dyn MountViewLease>, MountSourceError> {
1774        let owner = ViewGate::callback_owner();
1775        let lease = self.runtime.wait(|| {
1776            let gate = Arc::clone(&self.checkout.view_gate);
1777            async move { gate.read_for_callback(owner, None).await }
1778        })?;
1779        if self.view_epoch() != expected_epoch {
1780            return Err(MountSourceError::Stale);
1781        }
1782        Ok(Box::new(lease))
1783    }
1784
1785    fn acquire_binding_lease(
1786        &self,
1787        expected_epoch: Option<u64>,
1788    ) -> Result<Box<dyn MountViewLease>, MountSourceError> {
1789        let owner = ViewGate::callback_owner();
1790        let lease = self.runtime.wait(|| {
1791            let gate = Arc::clone(&self.checkout.view_gate);
1792            async move { gate.read_for_callback(owner, expected_epoch).await }
1793        })?;
1794        Ok(Box::new(lease))
1795    }
1796
1797    fn lookup(&self, path: &MountPath) -> Result<Option<MountLookup>, MountSourceError> {
1798        let path = self.path(path)?;
1799        let owner = ViewGate::callback_owner();
1800        self.runtime.wait(|| async {
1801            let mut checkout = self.checkout.read_lock(owner).await?;
1802            let receipt = checkout
1803                .lookup_no_follow_with_metadata(&path, boundary_budget(), &self.cancellation)
1804                .await
1805                .map_err(engine_error)?;
1806            Ok(receipt.value.map(|value| MountLookup {
1807                node: mount_node(value.record),
1808                metadata: value.metadata,
1809            }))
1810        })
1811    }
1812
1813    fn open_file(&self, path: &MountPath) -> Result<Arc<dyn MountOpenFile>, MountSourceError> {
1814        let path = self.path(path)?;
1815        let owner = ViewGate::callback_owner();
1816        let file_id = self.runtime.wait(|| async {
1817            let mut checkout = self.checkout.read_lock(owner).await?;
1818            let record = checkout
1819                .lookup_no_follow(&path, boundary_budget(), &self.cancellation)
1820                .await
1821                .map_err(engine_error)?
1822                .value
1823                .record
1824                .ok_or(MountSourceError::NotFound)?;
1825            if record.kind != FileKind::Regular {
1826                return Err(MountSourceError::Invalid(
1827                    "only regular files can own native file handles".to_owned(),
1828                ));
1829            }
1830            Ok(record.file_id)
1831        })?;
1832        Ok(Arc::new(CheckoutAttachedFile {
1833            checkout: Arc::clone(&self.checkout),
1834            file_id,
1835            runtime: self.runtime.clone(),
1836            cancellation: self.cancellation.clone(),
1837            profile: self.profile,
1838            limits: self.limits,
1839        }))
1840    }
1841
1842    fn detach_file(&self, path: &MountPath) -> Result<Arc<dyn MountOpenFile>, MountSourceError> {
1843        let path = self.path(path)?;
1844        let owner = ViewGate::callback_owner();
1845        let (file, metadata) = self.runtime.wait(|| async {
1846            let mut checkout = self.checkout.read_lock(owner).await?;
1847            checkout.ensure_publication_resolved()?;
1848            let lookup = checkout
1849                .lookup_no_follow_with_metadata(&path, boundary_budget(), &self.cancellation)
1850                .await
1851                .map_err(engine_error)?
1852                .value
1853                .ok_or(MountSourceError::NotFound)?;
1854            if lookup.record.kind != FileKind::Regular {
1855                return Err(MountSourceError::Invalid(
1856                    "only regular files can own detached native handles".to_owned(),
1857                ));
1858            }
1859            let file = checkout
1860                .detach_regular_file(&path, boundary_budget(), &self.cancellation)
1861                .await
1862                .map_err(engine_error)?
1863                .value;
1864            Ok((file, lookup.metadata))
1865        })?;
1866        Ok(Arc::new(CheckoutDetachedFile {
1867            state: tokio::sync::Mutex::new(DetachedMountState {
1868                file,
1869                metadata,
1870                mutation_epoch: 0,
1871            }),
1872            runtime: self.runtime.clone(),
1873            cancellation: self.cancellation.clone(),
1874            profile: self.profile,
1875            limits: self.limits,
1876        }))
1877    }
1878
1879    fn read_link(&self, path: &MountPath) -> Result<Bytes, MountSourceError> {
1880        let path = self.path(path)?;
1881        let owner = ViewGate::callback_owner();
1882        self.runtime.wait(|| async {
1883            let mut checkout = self.checkout.read_lock(owner).await?;
1884            checkout
1885                .read_symbolic_link(&path, boundary_budget(), &self.cancellation)
1886                .await
1887                .map(|receipt| receipt.value)
1888                .map_err(engine_error)
1889        })
1890    }
1891
1892    fn read_range(
1893        &self,
1894        path: &MountPath,
1895        offset: u64,
1896        length: u32,
1897    ) -> Result<Bytes, MountSourceError> {
1898        let path = self.path(path)?;
1899        let owner = ViewGate::callback_owner();
1900        self.runtime.wait(|| async {
1901            let mut checkout = self.checkout.read_lock(owner).await?;
1902            checkout
1903                .read_file_range(
1904                    &path,
1905                    ByteRange {
1906                        offset,
1907                        length: u64::from(length),
1908                    },
1909                    boundary_budget(),
1910                    &self.cancellation,
1911                )
1912                .await
1913                .map(|receipt| receipt.value.bytes)
1914                .map_err(engine_error)
1915        })
1916    }
1917
1918    fn seek(
1919        &self,
1920        path: &MountPath,
1921        offset: u64,
1922        target: MountSeekTarget,
1923    ) -> Result<Option<u64>, MountSourceError> {
1924        let path = self.path(path)?;
1925        let owner = ViewGate::callback_owner();
1926        self.runtime.wait(|| async {
1927            let mut checkout = self.checkout.read_lock(owner).await?;
1928            checkout
1929                .seek_file_extent(
1930                    &path,
1931                    offset,
1932                    match target {
1933                        MountSeekTarget::Data => ExtentSeekTarget::Data,
1934                        MountSeekTarget::Hole => ExtentSeekTarget::Hole,
1935                    },
1936                    boundary_budget(),
1937                    &self.cancellation,
1938                )
1939                .await
1940                .map(|receipt| receipt.value)
1941                .map_err(engine_error)
1942        })
1943    }
1944
1945    fn read_directory(
1946        &self,
1947        path: &MountPath,
1948        cursor: Option<&[u8]>,
1949        maximum_entries: u32,
1950    ) -> Result<MountDirectoryPage, MountSourceError> {
1951        let path = self.path(path)?;
1952        let cursor = cursor.map(|bytes| self.logical_name(bytes)).transpose()?;
1953        let owner = ViewGate::callback_owner();
1954        self.runtime.wait(|| async {
1955            let mut checkout = self.checkout.read_lock(owner).await?;
1956            let receipt = checkout
1957                .list_directory_records(
1958                    &path,
1959                    cursor.as_ref(),
1960                    maximum_entries,
1961                    boundary_budget(),
1962                    &self.cancellation,
1963                )
1964                .await
1965                .map_err(engine_error)?;
1966            let has_more = receipt.value.has_more;
1967            let entries = receipt
1968                .value
1969                .entries
1970                .into_iter()
1971                .map(|entry| {
1972                    Ok(MountDirectoryEntry {
1973                        name: Self::native_name(&entry.name)?,
1974                        node: mount_node(entry.record),
1975                        metadata: entry.metadata,
1976                    })
1977                })
1978                .collect::<Result<Vec<_>, MountSourceError>>()?;
1979            let next_cursor = has_more
1980                .then(|| entries.last().map(|entry| entry.name.clone()))
1981                .flatten();
1982            Ok(MountDirectoryPage {
1983                entries,
1984                next_cursor,
1985            })
1986        })
1987    }
1988
1989    fn create_file(
1990        &self,
1991        path: &MountPath,
1992        metadata: FileMetadata,
1993    ) -> Result<MountLookup, MountSourceError> {
1994        let path = self.path(path)?;
1995        self.runtime.wait(|| async {
1996            let mut checkout = self.checkout.lock().await;
1997            checkout.ensure_publication_resolved()?;
1998            let receipt = checkout
1999                .apply_authored_transaction(
2000                    vec![AuthoredMutation::CreateFile {
2001                        path,
2002                        bytes: Bytes::new(),
2003                        metadata,
2004                    }],
2005                    boundary_budget(),
2006                    &self.cancellation,
2007                )
2008                .await
2009                .map_err(facade_error)?;
2010            let file_id = receipt
2011                .value
2012                .created_file_ids
2013                .first()
2014                .copied()
2015                .flatten()
2016                .ok_or_else(|| {
2017                    MountSourceError::Engine("create omitted file identity".to_owned())
2018                })?;
2019            checkout.publish_after_mutation(&self.cancellation).await?;
2020            Ok(MountLookup {
2021                node: MountNode {
2022                    file_id,
2023                    kind: MountNodeKind::Regular,
2024                    logical_bytes: 0,
2025                    link_count: 1,
2026                    device: None,
2027                },
2028                metadata,
2029            })
2030        })
2031    }
2032
2033    fn create_directory(
2034        &self,
2035        path: &MountPath,
2036        metadata: FileMetadata,
2037    ) -> Result<MountLookup, MountSourceError> {
2038        let path = self.path(path)?;
2039        self.runtime.wait(|| async {
2040            let mut checkout = self.checkout.lock().await;
2041            checkout.ensure_publication_resolved()?;
2042            let receipt = checkout
2043                .apply_authored_transaction(
2044                    vec![AuthoredMutation::CreateDirectory { path, metadata }],
2045                    boundary_budget(),
2046                    &self.cancellation,
2047                )
2048                .await
2049                .map_err(facade_error)?;
2050            let file_id = receipt
2051                .value
2052                .created_file_ids
2053                .first()
2054                .copied()
2055                .flatten()
2056                .ok_or_else(|| {
2057                    MountSourceError::Engine("create omitted file identity".to_owned())
2058                })?;
2059            checkout.publish_after_mutation(&self.cancellation).await?;
2060            Ok(MountLookup {
2061                node: MountNode {
2062                    file_id,
2063                    kind: MountNodeKind::Directory,
2064                    logical_bytes: 0,
2065                    link_count: 1,
2066                    device: None,
2067                },
2068                metadata,
2069            })
2070        })
2071    }
2072
2073    fn create_symbolic_link(
2074        &self,
2075        path: &MountPath,
2076        target: Bytes,
2077        metadata: FileMetadata,
2078    ) -> Result<MountLookup, MountSourceError> {
2079        let path = self.path(path)?;
2080        let logical_bytes = u64::try_from(target.len()).map_err(|_| {
2081            MountSourceError::Unsupported("symbolic-link target is too large".to_owned())
2082        })?;
2083        self.runtime.wait(|| async {
2084            let mut checkout = self.checkout.lock().await;
2085            checkout.ensure_publication_resolved()?;
2086            let receipt = checkout
2087                .apply_authored_transaction(
2088                    vec![AuthoredMutation::CreateSymbolicLink {
2089                        path,
2090                        target,
2091                        metadata,
2092                    }],
2093                    boundary_budget(),
2094                    &self.cancellation,
2095                )
2096                .await
2097                .map_err(facade_error)?;
2098            let file_id = receipt
2099                .value
2100                .created_file_ids
2101                .first()
2102                .copied()
2103                .flatten()
2104                .ok_or_else(|| {
2105                    MountSourceError::Engine("create omitted file identity".to_owned())
2106                })?;
2107            checkout.publish_after_mutation(&self.cancellation).await?;
2108            Ok(MountLookup {
2109                node: MountNode {
2110                    file_id,
2111                    kind: MountNodeKind::SymbolicLink,
2112                    logical_bytes,
2113                    link_count: 1,
2114                    device: None,
2115                },
2116                metadata,
2117            })
2118        })
2119    }
2120
2121    fn create_special(
2122        &self,
2123        path: &MountPath,
2124        kind: MountNodeKind,
2125        device: Option<(u32, u32)>,
2126        metadata: FileMetadata,
2127    ) -> Result<MountLookup, MountSourceError> {
2128        let path = self.path(path)?;
2129        let (file_kind, mutation) = match (kind, device) {
2130            (MountNodeKind::Fifo, None) => (
2131                FileKind::Fifo,
2132                AuthoredMutation::CreateEmptySpecial {
2133                    path,
2134                    kind: FileKind::Fifo,
2135                    metadata,
2136                },
2137            ),
2138            (MountNodeKind::Socket, None) => (
2139                FileKind::Socket,
2140                AuthoredMutation::CreateEmptySpecial {
2141                    path,
2142                    kind: FileKind::Socket,
2143                    metadata,
2144                },
2145            ),
2146            (MountNodeKind::CharacterDevice, Some((major, minor))) => (
2147                FileKind::CharacterDevice,
2148                AuthoredMutation::CreateDevice {
2149                    path,
2150                    kind: FileKind::CharacterDevice,
2151                    major,
2152                    minor,
2153                    metadata,
2154                },
2155            ),
2156            (MountNodeKind::BlockDevice, Some((major, minor))) => (
2157                FileKind::BlockDevice,
2158                AuthoredMutation::CreateDevice {
2159                    path,
2160                    kind: FileKind::BlockDevice,
2161                    major,
2162                    minor,
2163                    metadata,
2164                },
2165            ),
2166            _ => {
2167                return Err(MountSourceError::Unsupported(
2168                    "special-file kind and device identity disagree".to_owned(),
2169                ));
2170            }
2171        };
2172        self.runtime.wait(|| async {
2173            let mut checkout = self.checkout.lock().await;
2174            checkout.ensure_publication_resolved()?;
2175            let receipt = checkout
2176                .apply_authored_transaction(vec![mutation], boundary_budget(), &self.cancellation)
2177                .await
2178                .map_err(facade_error)?;
2179            let file_id = receipt
2180                .value
2181                .created_file_ids
2182                .first()
2183                .copied()
2184                .flatten()
2185                .ok_or_else(|| {
2186                    MountSourceError::Engine("create omitted file identity".to_owned())
2187                })?;
2188            checkout.publish_after_mutation(&self.cancellation).await?;
2189            Ok(MountLookup {
2190                node: MountNode {
2191                    file_id,
2192                    kind,
2193                    logical_bytes: 0,
2194                    link_count: 1,
2195                    device: match file_kind {
2196                        FileKind::CharacterDevice | FileKind::BlockDevice => device,
2197                        _ => None,
2198                    },
2199                },
2200                metadata,
2201            })
2202        })
2203    }
2204
2205    fn set_attributes(
2206        &self,
2207        path: &MountPath,
2208        metadata: FileMetadata,
2209        logical_bytes: Option<u64>,
2210    ) -> Result<(), MountSourceError> {
2211        let path = self.path(path)?;
2212        self.runtime.wait(|| async {
2213            let mut checkout = self.checkout.lock().await;
2214            checkout.ensure_publication_resolved()?;
2215            let mut authored = vec![AuthoredMutation::SetMetadata {
2216                path: path.clone(),
2217                metadata,
2218            }];
2219            if let Some(logical_bytes) = logical_bytes {
2220                authored.push(AuthoredMutation::Resize {
2221                    path,
2222                    logical_bytes,
2223                });
2224            }
2225            checkout
2226                .apply_authored_transaction(authored, boundary_budget(), &self.cancellation)
2227                .await
2228                .map_err(facade_error)?;
2229            checkout.publish_after_mutation(&self.cancellation).await
2230        })
2231    }
2232
2233    fn read_attribute(
2234        &self,
2235        path: &MountPath,
2236        name: &[u8],
2237    ) -> Result<Option<Bytes>, MountSourceError> {
2238        let path = self.path(path)?;
2239        let name = self.posix_attribute_name(name)?;
2240        let owner = ViewGate::callback_owner();
2241        self.runtime.wait(|| async {
2242            let mut checkout = self.checkout.read_lock(owner).await?;
2243            checkout
2244                .read_named_attribute(&path, &name, boundary_budget(), &self.cancellation)
2245                .await
2246                .map(|receipt| receipt.value)
2247                .map_err(engine_error)
2248        })
2249    }
2250
2251    fn list_attributes(
2252        &self,
2253        path: &MountPath,
2254        cursor: Option<&[u8]>,
2255        maximum_entries: u32,
2256    ) -> Result<MountAttributePage, MountSourceError> {
2257        let path = self.path(path)?;
2258        let cursor = cursor
2259            .map(|name| self.posix_attribute_name(name))
2260            .transpose()?;
2261        let owner = ViewGate::callback_owner();
2262        self.runtime.wait(|| async {
2263            let mut checkout = self.checkout.read_lock(owner).await?;
2264            let receipt = checkout
2265                .list_named_attributes(
2266                    &path,
2267                    cursor.as_ref(),
2268                    maximum_entries,
2269                    boundary_budget(),
2270                    &self.cancellation,
2271                )
2272                .await
2273                .map_err(engine_error)?;
2274            let has_more = receipt.value.has_more;
2275            let names = receipt
2276                .value
2277                .entries
2278                .into_iter()
2279                .map(|entry| {
2280                    if entry.name.class() != AttributeClass::PosixXattr {
2281                        return Err(MountSourceError::Unsupported(
2282                            "POSIX mount encountered a non-POSIX named attribute".to_owned(),
2283                        ));
2284                    }
2285                    Ok(entry.name.as_bytes().to_vec())
2286                })
2287                .collect::<Result<Vec<_>, _>>()?;
2288            let next_cursor = has_more.then(|| names.last().cloned()).flatten();
2289            Ok(MountAttributePage { names, next_cursor })
2290        })
2291    }
2292
2293    fn write_attribute(
2294        &self,
2295        path: &MountPath,
2296        name: &[u8],
2297        value: Bytes,
2298        mode: MountAttributeWriteMode,
2299    ) -> Result<(), MountSourceError> {
2300        let path = self.path(path)?;
2301        let name = self.posix_attribute_name(name)?;
2302        let mode = match mode {
2303            MountAttributeWriteMode::Upsert => NamedAttributeWriteMode::Upsert,
2304            MountAttributeWriteMode::Create => NamedAttributeWriteMode::Create,
2305            MountAttributeWriteMode::Replace => NamedAttributeWriteMode::Replace,
2306        };
2307        self.runtime.wait(|| async {
2308            let mut checkout = self.checkout.lock().await;
2309            checkout.ensure_publication_resolved()?;
2310            checkout
2311                .write_named_attribute(
2312                    path,
2313                    name,
2314                    value,
2315                    mode,
2316                    boundary_budget(),
2317                    &self.cancellation,
2318                )
2319                .await
2320                .map_err(facade_error)?;
2321            checkout.publish_after_mutation(&self.cancellation).await
2322        })
2323    }
2324
2325    fn remove_attribute(&self, path: &MountPath, name: &[u8]) -> Result<(), MountSourceError> {
2326        let path = self.path(path)?;
2327        let name = self.posix_attribute_name(name)?;
2328        self.runtime.wait(|| async {
2329            let mut checkout = self.checkout.lock().await;
2330            checkout.ensure_publication_resolved()?;
2331            checkout
2332                .remove_named_attribute(path, name, boundary_budget(), &self.cancellation)
2333                .await
2334                .map_err(facade_error)?;
2335            checkout.publish_after_mutation(&self.cancellation).await
2336        })
2337    }
2338
2339    fn write_range(
2340        &self,
2341        path: &MountPath,
2342        offset: u64,
2343        bytes: Bytes,
2344    ) -> Result<(), MountSourceError> {
2345        let path = self.path(path)?;
2346        self.runtime.wait(|| async {
2347            let mut checkout = self.checkout.lock().await;
2348            checkout.ensure_publication_resolved()?;
2349            checkout
2350                .write_file(path, offset, bytes, boundary_budget(), &self.cancellation)
2351                .await
2352                .map_err(engine_error)?;
2353            checkout.publish_after_mutation(&self.cancellation).await
2354        })
2355    }
2356
2357    fn resize(&self, path: &MountPath, logical_bytes: u64) -> Result<(), MountSourceError> {
2358        let path = self.path(path)?;
2359        self.runtime.wait(|| async {
2360            let mut checkout = self.checkout.lock().await;
2361            checkout.ensure_publication_resolved()?;
2362            checkout
2363                .resize_file(path, logical_bytes, boundary_budget(), &self.cancellation)
2364                .await
2365                .map_err(engine_error)?;
2366            checkout.publish_after_mutation(&self.cancellation).await
2367        })
2368    }
2369
2370    fn allocate_range(
2371        &self,
2372        path: &MountPath,
2373        offset: u64,
2374        length: u64,
2375        operation: MountRangeAllocation,
2376    ) -> Result<(), MountSourceError> {
2377        let path = self.path(path)?;
2378        self.runtime.wait(|| async {
2379            let mut checkout = self.checkout.lock().await;
2380            checkout.ensure_publication_resolved()?;
2381            match operation {
2382                MountRangeAllocation::PunchHole => {
2383                    checkout
2384                        .zero_file_range(
2385                            path,
2386                            ByteRange { offset, length },
2387                            false,
2388                            false,
2389                            boundary_budget(),
2390                            &self.cancellation,
2391                        )
2392                        .await
2393                }
2394                MountRangeAllocation::ZeroRange { extend } => {
2395                    checkout
2396                        .zero_file_range(
2397                            path,
2398                            ByteRange { offset, length },
2399                            true,
2400                            extend,
2401                            boundary_budget(),
2402                            &self.cancellation,
2403                        )
2404                        .await
2405                }
2406                MountRangeAllocation::Preallocate { keep_size } => {
2407                    checkout
2408                        .preallocate_file(
2409                            path,
2410                            ByteRange { offset, length },
2411                            keep_size,
2412                            boundary_budget(),
2413                            &self.cancellation,
2414                        )
2415                        .await
2416                }
2417            }
2418            .map_err(engine_error)?;
2419            checkout.publish_after_mutation(&self.cancellation).await
2420        })
2421    }
2422
2423    fn clone_range(
2424        &self,
2425        source: &MountPath,
2426        source_offset: u64,
2427        destination: &MountPath,
2428        destination_offset: u64,
2429        length: u64,
2430    ) -> Result<(), MountSourceError> {
2431        let source = self.path(source)?;
2432        let destination = self.path(destination)?;
2433        self.runtime.wait(|| async {
2434            let mut checkout = self.checkout.lock().await;
2435            checkout.ensure_publication_resolved()?;
2436            checkout
2437                .clone_file_range(
2438                    crate::FileCloneRequest {
2439                        source,
2440                        source_offset,
2441                        destination,
2442                        destination_offset,
2443                        length,
2444                    },
2445                    boundary_budget(),
2446                    &self.cancellation,
2447                )
2448                .await
2449                .map_err(engine_error)?;
2450            checkout.publish_after_mutation(&self.cancellation).await
2451        })
2452    }
2453
2454    fn clone_range_by_id(
2455        &self,
2456        source_file_id: FileId,
2457        source_offset: u64,
2458        destination_file_id: FileId,
2459        destination_offset: u64,
2460        length: u64,
2461    ) -> Result<(), MountSourceError> {
2462        self.runtime.wait(|| async {
2463            let mut checkout = self.checkout.lock().await;
2464            checkout.ensure_publication_resolved()?;
2465            checkout
2466                .clone_file_range_by_id(
2467                    source_file_id,
2468                    source_offset,
2469                    destination_file_id,
2470                    destination_offset,
2471                    length,
2472                    boundary_budget(),
2473                    &self.cancellation,
2474                )
2475                .await
2476                .map_err(engine_error)?;
2477            checkout.publish_after_mutation(&self.cancellation).await
2478        })
2479    }
2480
2481    fn remove(&self, path: &MountPath, expected: Option<FileId>) -> Result<(), MountSourceError> {
2482        let path = self.path(path)?;
2483        self.runtime.wait(|| async {
2484            let mut checkout = self.checkout.lock().await;
2485            checkout.ensure_publication_resolved()?;
2486            checkout
2487                .remove(path, expected, boundary_budget(), &self.cancellation)
2488                .await
2489                .map_err(engine_error)?;
2490            checkout.publish_after_mutation(&self.cancellation).await
2491        })
2492    }
2493
2494    fn rename(
2495        &self,
2496        source: &MountPath,
2497        destination: &MountPath,
2498        replace: bool,
2499    ) -> Result<(), MountSourceError> {
2500        let source = self.path(source)?;
2501        let destination = self.path(destination)?;
2502        self.runtime.wait(|| async {
2503            let mut checkout = self.checkout.lock().await;
2504            checkout.ensure_publication_resolved()?;
2505            checkout
2506                .rename(
2507                    source,
2508                    destination,
2509                    replace,
2510                    boundary_budget(),
2511                    &self.cancellation,
2512                )
2513                .await
2514                .map_err(engine_error)?;
2515            checkout.publish_after_mutation(&self.cancellation).await
2516        })
2517    }
2518
2519    fn hard_link(
2520        &self,
2521        source: &MountPath,
2522        destination: &MountPath,
2523    ) -> Result<(), MountSourceError> {
2524        let source = self.path(source)?;
2525        let destination = self.path(destination)?;
2526        self.runtime.wait(|| async {
2527            let mut checkout = self.checkout.lock().await;
2528            checkout.ensure_publication_resolved()?;
2529            checkout
2530                .hard_link(source, destination, boundary_budget(), &self.cancellation)
2531                .await
2532                .map_err(engine_error)?;
2533            checkout.publish_after_mutation(&self.cancellation).await
2534        })
2535    }
2536
2537    fn flush(&self) -> Result<(), MountSourceError> {
2538        self.runtime.wait(|| async {
2539            let mut checkout = self.checkout.lock().await;
2540            checkout
2541                .publish_at_native_boundary(&self.cancellation)
2542                .await
2543        })
2544    }
2545
2546    fn capture_host_path(
2547        &self,
2548        source_root: &Path,
2549        path: &MountPath,
2550    ) -> Result<(), MountSourceError> {
2551        self.capture_host_paths(source_root, std::slice::from_ref(path))
2552    }
2553
2554    fn capture_host_paths(
2555        &self,
2556        source_root: &Path,
2557        paths: &[MountPath],
2558    ) -> Result<(), MountSourceError> {
2559        if paths.is_empty() {
2560            return Ok(());
2561        }
2562        let expected_root_identity = capture_root_identity(source_root).map_err(engine_error)?;
2563        self.capture_host_paths_with_identity(source_root, paths, expected_root_identity)
2564    }
2565
2566    fn capture_host_subtree(
2567        &self,
2568        source_root: &Path,
2569        path: &MountPath,
2570    ) -> Result<(), MountSourceError> {
2571        let path = self.path(path)?;
2572        self.runtime.wait(|| async {
2573            let expected_root_identity =
2574                capture_root_identity(source_root).map_err(engine_error)?;
2575            let options = CaptureOptions {
2576                source_root: source_root.to_path_buf(),
2577                expected_root_identity,
2578                maximum_paths: MAX_NATIVE_SUBTREE_CAPTURE_PATHS,
2579                maximum_extent_spans: 65_536,
2580            };
2581            for _ in 0..3 {
2582                let (mut candidate, epoch) = {
2583                    let checkout = self.checkout.lock().await;
2584                    checkout.ensure_publication_resolved()?;
2585                    (
2586                        checkout.private_candidate(),
2587                        checkout.view_epoch.load(Ordering::Acquire),
2588                    )
2589                };
2590                capture_subtree(
2591                    &mut candidate,
2592                    path.clone(),
2593                    &options,
2594                    boundary_budget(),
2595                    &self.cancellation,
2596                )
2597                .await
2598                .map_err(engine_error)?;
2599                let mut checkout = self.checkout.lock().await;
2600                checkout.ensure_publication_resolved()?;
2601                if checkout.view_epoch.load(Ordering::Acquire) != epoch {
2602                    continue;
2603                }
2604                checkout.checkout = candidate;
2605                checkout.view_epoch.fetch_add(1, Ordering::AcqRel);
2606                return checkout
2607                    .publish_at_native_boundary(&self.cancellation)
2608                    .await;
2609            }
2610            Err(MountSourceError::Stale)
2611        })
2612    }
2613}
2614
2615fn mount_node(record: FileRecord) -> MountNode {
2616    let kind = match record.kind {
2617        FileKind::Regular => MountNodeKind::Regular,
2618        FileKind::Directory => MountNodeKind::Directory,
2619        FileKind::SymbolicLink => MountNodeKind::SymbolicLink,
2620        FileKind::Fifo => MountNodeKind::Fifo,
2621        FileKind::Socket => MountNodeKind::Socket,
2622        FileKind::CharacterDevice => MountNodeKind::CharacterDevice,
2623        FileKind::BlockDevice => MountNodeKind::BlockDevice,
2624        FileKind::ReparsePoint | FileKind::MountBoundary => MountNodeKind::Unsupported,
2625    };
2626    let logical_bytes = match &record.payload {
2627        FilePayload::InlineRegular(data) => {
2628            u64::try_from(data.as_bytes().len()).unwrap_or(u64::MAX)
2629        }
2630        FilePayload::Regular { logical_bytes, .. } => *logical_bytes,
2631        FilePayload::SymbolicLink { target_bytes, .. } => *target_bytes,
2632        FilePayload::Directory { .. }
2633        | FilePayload::Empty
2634        | FilePayload::Device { .. }
2635        | FilePayload::ReparsePoint { .. } => 0,
2636    };
2637    let device = match &record.payload {
2638        FilePayload::Device { major, minor } => Some((*major, *minor)),
2639        _ => None,
2640    };
2641    MountNode {
2642        file_id: record.file_id,
2643        kind,
2644        logical_bytes,
2645        link_count: record.link_count,
2646        device,
2647    }
2648}
2649
2650fn engine_error(error: impl std::fmt::Display) -> MountSourceError {
2651    MountSourceError::Engine(error.to_string())
2652}
2653
2654fn facade_error(error: OperationFailure<FsError>) -> MountSourceError {
2655    match error.error {
2656        FsError::CreationRejected | FsError::Mutation(GenerationMutationError::AlreadyExists) => {
2657            MountSourceError::AlreadyExists
2658        }
2659        FsError::NotFound
2660        | FsError::Mutation(
2661            GenerationMutationError::MissingParent | GenerationMutationError::MissingSource,
2662        ) => MountSourceError::NotFound,
2663        other => MountSourceError::Engine(other.to_string()),
2664    }
2665}
2666
2667fn boundary_budget() -> WorkBudget {
2668    const OPERATIONS: u64 = 1_000_000;
2669    const BYTES: u64 = 256 * 1024 * 1024;
2670    WorkBudget {
2671        authority_records_read: OPERATIONS,
2672        authority_records_appended: OPERATIONS,
2673        authority_bytes_read: BYTES,
2674        authority_bytes_written: BYTES,
2675        object_probes: OPERATIONS,
2676        backend_read_operations: OPERATIONS,
2677        backend_write_operations: OPERATIONS,
2678        durability_operations: OPERATIONS,
2679        page_reads: OPERATIONS,
2680        page_writes: OPERATIONS,
2681        object_bytes_read: BYTES,
2682        object_bytes_written: BYTES,
2683        bytes_hashed: BYTES,
2684        bytes_copied: BYTES,
2685        bytes_encoded: BYTES,
2686        source_bytes_read: BYTES,
2687        source_path_components: OPERATIONS,
2688        source_entries_visited: OPERATIONS,
2689        output_bytes: BYTES,
2690        items_examined: OPERATIONS,
2691        items_returned: OPERATIONS,
2692        allocation_operations: OPERATIONS,
2693        peak_allocation_bytes: BYTES,
2694        materializations: OPERATIONS,
2695    }
2696}
2697
2698#[cfg(test)]
2699mod tests {
2700    use super::*;
2701    use crate::Fs;
2702    use crate::model::{
2703        AccessMode, CheckoutMode, ConsistencyMode, GenerationSelector, Lifecycle, MutationMode,
2704    };
2705    #[cfg(target_os = "linux")]
2706    use std::os::unix::ffi::OsStringExt;
2707
2708    #[cfg(target_os = "macos")]
2709    #[allow(unsafe_code)]
2710    unsafe extern "C" {
2711        fn nfs4_test_exclusive_replay_identity() -> std::ffi::c_int;
2712        fn nfs4_test_namedattr_exclusive_replay_identity() -> std::ffi::c_int;
2713        fn nfs4_test_readdir_cookie_verifier() -> std::ffi::c_int;
2714        fn nfs4_test_sync_acknowledgement() -> std::ffi::c_int;
2715    }
2716
2717    #[cfg(target_os = "macos")]
2718    #[test]
2719    #[allow(unsafe_code)]
2720    fn exclusive_replay_is_bound_to_the_opened_file_handle() {
2721        // SAFETY: the test hook has no arguments and owns all callback state.
2722        assert_eq!(unsafe { nfs4_test_exclusive_replay_identity() }, 0);
2723    }
2724
2725    #[cfg(target_os = "macos")]
2726    #[test]
2727    #[allow(unsafe_code)]
2728    fn named_attribute_exclusive_replay_is_bound_to_owner_and_name() {
2729        // SAFETY: the test hook has no arguments and owns all callback state.
2730        assert_eq!(
2731            unsafe { nfs4_test_namedattr_exclusive_replay_identity() },
2732            0
2733        );
2734    }
2735
2736    #[cfg(target_os = "macos")]
2737    #[test]
2738    #[allow(unsafe_code)]
2739    fn readdir_continuations_are_bound_to_the_namespace_revision() {
2740        // SAFETY: the test hook has no arguments and mutates no shared state.
2741        assert_eq!(unsafe { nfs4_test_readdir_cookie_verifier() }, 0);
2742    }
2743
2744    #[cfg(target_os = "macos")]
2745    #[test]
2746    #[allow(unsafe_code)]
2747    fn nfs_sync_acknowledgement_requires_successful_callback() {
2748        // SAFETY: the test hook owns its callback state.
2749        assert_eq!(unsafe { nfs4_test_sync_acknowledgement() }, 0);
2750    }
2751
2752    #[cfg(target_os = "macos")]
2753    #[test]
2754    #[ignore = "requires a live macOS NFS mount"]
2755    fn macos_explicit_file_fsync_reaches_mount_source() -> Result<(), Box<dyn std::error::Error>> {
2756        use std::io::Write as _;
2757
2758        let source = Arc::new(source(FilesystemProfile::Posix)?);
2759        let temporary = tempfile::tempdir()?;
2760        let mut mount = crate::mount_native(
2761            crate::NativeMountRequest {
2762                mount_id: crate::MountId::new(),
2763                volume_id: source.volume_id()?,
2764                destination: temporary.path().to_path_buf(),
2765                writable: true,
2766            },
2767            Arc::clone(&source) as Arc<dyn MountFilesystem>,
2768        )?;
2769        let path = temporary.path().join("explicit-fsync");
2770        let mut file = std::fs::File::create(&path)?;
2771        file.write_all(b"durable")?;
2772        file.sync_all()?;
2773        assert_eq!(std::fs::read(&path)?, b"durable");
2774        drop(file);
2775        assert!(mount.stop()?);
2776        Ok(())
2777    }
2778
2779    type MemorySource = CheckoutMountSource<
2780        crate::facade::MemoryAuthorityBackend,
2781        crate::facade::MemoryObjectBackend,
2782    >;
2783
2784    #[test]
2785    fn cancelled_lifecycle_capture_cannot_commit_late() -> Result<(), Box<dyn std::error::Error>> {
2786        let (source, _) =
2787            shared_sources_with_publication(FilesystemProfile::Portable, MountPublication::Manual)?;
2788        let epoch = source
2789            .view_epoch()
2790            .ok_or_else(|| std::io::Error::other("missing checkout view epoch"))?;
2791        let candidate = source
2792            .runtime
2793            .block_on(|| async { source.checkout.lock().await.private_candidate() })?;
2794        let request = Arc::new(CaptureCommitGate {
2795            cancellation: CancellationToken::new(),
2796            state: AtomicU8::new(CAPTURE_PENDING),
2797        });
2798        drop(CancelCaptureOnDrop(Arc::clone(&request)));
2799        let result = source
2800            .runtime
2801            .block_on(|| source.commit_host_capture(candidate, epoch, Some(&request)))?;
2802        assert!(result.is_err());
2803        assert_eq!(source.view_epoch(), Some(epoch));
2804        let request = Arc::new(CaptureCommitGate {
2805            cancellation: CancellationToken::new(),
2806            state: AtomicU8::new(CAPTURE_PENDING),
2807        });
2808        source
2809            .capture_gate
2810            .lock()
2811            .unwrap_or_else(std::sync::PoisonError::into_inner)
2812            .active
2813            .push(Arc::downgrade(&request));
2814        source.cancel();
2815        assert!(request.cancellation.is_cancelled());
2816        let candidate = source
2817            .runtime
2818            .block_on(|| async { source.checkout.lock().await.private_candidate() })?;
2819        let result = source
2820            .runtime
2821            .block_on(|| source.commit_host_capture(candidate, epoch, Some(&request)))?;
2822        assert!(result.is_err());
2823        assert_eq!(source.view_epoch(), Some(epoch));
2824        Ok(())
2825    }
2826
2827    #[test]
2828    fn dropping_active_capture_fences_its_detached_worker() -> Result<(), Box<dyn std::error::Error>>
2829    {
2830        let (source, _) =
2831            shared_sources_with_publication(FilesystemProfile::Portable, MountPublication::Manual)?;
2832        let source = Arc::new(source);
2833        let directory = tempfile::tempdir()?;
2834        std::fs::write(directory.path().join("file"), b"child")?;
2835        let identity = capture_root_identity(directory.path())?;
2836        #[cfg(unix)]
2837        let baseline = Arc::new(NativeViewBaseline::new(identity));
2838        #[cfg(windows)]
2839        let baseline = Arc::new(NativeViewBaseline);
2840        let runtime = tokio::runtime::Builder::new_multi_thread()
2841            .worker_threads(2)
2842            .enable_all()
2843            .build()?;
2844        runtime.block_on(async {
2845            let held = source.checkout.lock().await;
2846            let epoch = source.checkout.view_epoch();
2847            let worker = Arc::clone(&source);
2848            let root = directory.path().to_path_buf();
2849            let capture = tokio::spawn(async move {
2850                worker
2851                    .capture_host_with_identity_budgeted(
2852                        &root,
2853                        &[native_test_path("file")],
2854                        identity,
2855                        HostCaptureScope::ExactPaths,
2856                        baseline,
2857                        WorkBudget::UNBOUNDED,
2858                    )
2859                    .await
2860            });
2861            let request = tokio::time::timeout(Duration::from_secs(5), async {
2862                loop {
2863                    if let Some(request) = source
2864                        .capture_gate
2865                        .lock()
2866                        .unwrap_or_else(std::sync::PoisonError::into_inner)
2867                        .active
2868                        .last()
2869                        .cloned()
2870                    {
2871                        break request;
2872                    }
2873                    tokio::task::yield_now().await;
2874                }
2875            })
2876            .await?;
2877            capture.abort();
2878            let _ = capture.await;
2879            drop(held);
2880            tokio::time::timeout(Duration::from_secs(5), async {
2881                while request.strong_count() > 0 {
2882                    tokio::task::yield_now().await;
2883                }
2884            })
2885            .await?;
2886            assert_eq!(source.checkout.view_epoch(), epoch);
2887            assert!(!source.checkout.lock().await.has_pending_mutations());
2888            Ok::<_, Box<dyn std::error::Error>>(())
2889        })?;
2890        Ok(())
2891    }
2892
2893    #[test]
2894    fn batched_capture_budget_is_cumulative_across_paths() -> Result<(), Box<dyn std::error::Error>>
2895    {
2896        let source = source(FilesystemProfile::Portable)?;
2897        let directory = tempfile::tempdir()?;
2898        std::fs::write(directory.path().join("first"), b"first")?;
2899        std::fs::write(directory.path().join("second"), b"other")?;
2900        let first = source.path(&native_test_path("first"))?;
2901        let second = source.path(&native_test_path("second"))?;
2902        let options = CaptureOptions {
2903            source_root: directory.path().to_path_buf(),
2904            expected_root_identity: capture_root_identity(directory.path())?,
2905            maximum_paths: 2,
2906            maximum_extent_spans: 64,
2907        };
2908        let cancellation = CancellationToken::new();
2909        let mut zero_candidate = source
2910            .runtime
2911            .block_on(|| async { source.checkout.lock().await.private_candidate() })?;
2912        let zero_result = source.runtime.block_on(|| {
2913            capture_paths_batched_with_baseline(
2914                &mut zero_candidate,
2915                std::slice::from_ref(&first),
2916                &options,
2917                1,
2918                WorkBudget::default(),
2919                &cancellation,
2920                None,
2921            )
2922        })?;
2923        let Err(zero_failure) = zero_result else {
2924            return Err(std::io::Error::other("zero budget admitted host observation").into());
2925        };
2926        assert_eq!(zero_failure.work.source_entries_visited, 0);
2927        assert!(!zero_candidate.has_pending_mutations());
2928        let mut candidate = source
2929            .runtime
2930            .block_on(|| async { source.checkout.lock().await.private_candidate() })?;
2931        let single = source.runtime.block_on(|| {
2932            capture_paths_batched_with_baseline(
2933                &mut candidate,
2934                std::slice::from_ref(&first),
2935                &options,
2936                1,
2937                WorkBudget::UNBOUNDED,
2938                &cancellation,
2939                None,
2940            )
2941        })??;
2942        assert!(single.work.source_bytes_read > 0);
2943        let mut budget = WorkBudget::UNBOUNDED;
2944        budget.source_bytes_read = single.work.source_bytes_read;
2945        let mut candidate = source
2946            .runtime
2947            .block_on(|| async { source.checkout.lock().await.private_candidate() })?;
2948        let paths = [first, second];
2949        let result = source.runtime.block_on(|| {
2950            capture_paths_batched_with_baseline(
2951                &mut candidate,
2952                &paths,
2953                &options,
2954                1,
2955                budget,
2956                &cancellation,
2957                None,
2958            )
2959        })?;
2960        let Err(failure) = result else {
2961            return Err(std::io::Error::other("aggregate source budget was ignored").into());
2962        };
2963        assert!(failure.error.to_string().contains("source_bytes_read"));
2964        assert!(!candidate.has_pending_mutations());
2965        Ok(())
2966    }
2967
2968    #[cfg(target_os = "linux")]
2969    #[test]
2970    fn external_callbacks_enter_the_shared_runtime_without_transient_threads()
2971    -> Result<(), Box<dyn std::error::Error>> {
2972        let runtime = CallbackRuntime::create()?;
2973        let callers = (0..32)
2974            .map(|index| {
2975                let runtime = runtime.clone();
2976                std::thread::Builder::new()
2977                    .name(format!("callback-probe-{index}"))
2978                    .spawn(move || {
2979                        runtime.wait(|| async {
2980                            Ok::<_, MountSourceError>(
2981                                std::thread::current().name().unwrap_or_default().to_owned(),
2982                            )
2983                        })
2984                    })
2985            })
2986            .collect::<Result<Vec<_>, _>>()?;
2987        for (index, caller) in callers.into_iter().enumerate() {
2988            assert_eq!(
2989                caller
2990                    .join()
2991                    .map_err(|_| std::io::Error::other("callback probe thread panicked"))??,
2992                format!("callback-probe-{index}")
2993            );
2994        }
2995        Ok(())
2996    }
2997
2998    fn metadata() -> FileMetadata {
2999        FileMetadata {
3000            posix_mode: crate::kernel::MetadataField::Unavailable,
3001            posix_uid: crate::kernel::MetadataField::Unavailable,
3002            posix_gid: crate::kernel::MetadataField::Unavailable,
3003            posix_flags: crate::kernel::MetadataField::Unavailable,
3004            windows_attributes: crate::kernel::MetadataField::Unavailable,
3005            created_ns: crate::kernel::MetadataField::Unavailable,
3006            modified_ns: crate::kernel::MetadataField::Unavailable,
3007            accessed_ns: crate::kernel::MetadataField::Unavailable,
3008            changed_ns: crate::kernel::MetadataField::Unavailable,
3009            named_attributes: crate::kernel::MetadataField::Unavailable,
3010            acl: crate::kernel::MetadataField::Unavailable,
3011            security_descriptor: crate::kernel::MetadataField::Unavailable,
3012        }
3013    }
3014
3015    fn checkout_state(
3016        source: &MemorySource,
3017    ) -> Result<(crate::GenerationId, bool), MountSourceError> {
3018        source.runtime.block_on(|| async {
3019            let checkout = source.checkout.lock().await;
3020            (checkout.generation_id(), checkout.has_pending_mutations())
3021        })
3022    }
3023
3024    fn source(profile: FilesystemProfile) -> Result<MemorySource, Box<dyn std::error::Error>> {
3025        Ok(shared_sources(profile)?.0)
3026    }
3027
3028    fn shared_sources(
3029        profile: FilesystemProfile,
3030    ) -> Result<(MemorySource, MemorySource), Box<dyn std::error::Error>> {
3031        shared_sources_with_publication(profile, MountPublication::CloseAndSync)
3032    }
3033
3034    #[test]
3035    fn direct_checkout_view_lease_excludes_concurrent_mutation()
3036    -> Result<(), Box<dyn std::error::Error>> {
3037        let source = Arc::new(source(FilesystemProfile::Portable)?);
3038        let lease = source.acquire_view_lease(source.view_epoch())?;
3039        let writer = Arc::clone(&source);
3040        let (completed, receive) = std::sync::mpsc::sync_channel(1);
3041        let task = std::thread::spawn(move || {
3042            let result =
3043                writer.create_file(&MountPath::root().child(b"leased".to_vec()), metadata());
3044            assert!(completed.send(result).is_ok(), "mutation result receiver");
3045        });
3046        assert!(
3047            receive.recv_timeout(Duration::from_millis(50)).is_err(),
3048            "mutation crossed a retained native callback view"
3049        );
3050        drop(lease);
3051        receive.recv_timeout(Duration::from_secs(5))??;
3052        task.join()
3053            .map_err(|_| std::io::Error::other("mutation thread panicked"))?;
3054        Ok(())
3055    }
3056
3057    #[test]
3058    fn native_mutation_invalidates_cache_without_rebinding_the_mount()
3059    -> Result<(), Box<dyn std::error::Error>> {
3060        let source = source(FilesystemProfile::Portable)?;
3061        let binding = source.binding_epoch();
3062        let cache = source.view_epoch();
3063        source.create_file(&native_test_path("new-file"), metadata())?;
3064        assert_eq!(source.binding_epoch(), binding);
3065        assert_ne!(source.view_epoch(), cache);
3066        let _lease = source.acquire_binding_lease(binding)?;
3067        Ok(())
3068    }
3069
3070    #[test]
3071    fn open_handle_read_up_to_clips_at_eof_without_a_separate_lookup()
3072    -> Result<(), Box<dyn std::error::Error>> {
3073        let source = source(FilesystemProfile::Portable)?;
3074        let path = native_test_path("clipped-read");
3075        source.create_file(&path, metadata())?;
3076        source.write_range(&path, 0, Bytes::from_static(b"abc"))?;
3077        let open = source.open_file(&path)?;
3078        assert_eq!(open.read_up_to(0, 128)?.as_ref(), b"abc");
3079        assert_eq!(open.read_up_to(1, 1)?.as_ref(), b"b");
3080        assert!(open.read_up_to(3, 128)?.is_empty());
3081        assert!(open.read_up_to(u64::MAX, 128)?.is_empty());
3082        source.write_range(&path, 0, Bytes::from_static(b"xyz"))?;
3083        assert_eq!(open.read_up_to(0, 128)?.as_ref(), b"xyz");
3084        Ok(())
3085    }
3086
3087    fn shared_sources_with_publication(
3088        profile: FilesystemProfile,
3089        publication: MountPublication,
3090    ) -> Result<(MemorySource, MemorySource), Box<dyn std::error::Error>> {
3091        let mut config = VolumeConfig::portable(Lifecycle::Ephemeral);
3092        config.profile = profile;
3093        let fs = Fs::memory();
3094        let runtime = tokio::runtime::Builder::new_current_thread().build()?;
3095        let checkout = runtime.block_on(async {
3096            let cancellation = CancellationToken::new();
3097            let volume = fs
3098                .create_volume(config, WorkBudget::UNBOUNDED, &cancellation)
3099                .await?
3100                .value;
3101            volume
3102                .checkout(
3103                    GenerationSelector::Head,
3104                    CheckoutMode {
3105                        access: AccessMode::ReadWrite,
3106                        consistency: ConsistencyMode::Pinned,
3107                        mutations: MutationMode::PrivateOverlay,
3108                    },
3109                    WorkBudget::UNBOUNDED,
3110                    &cancellation,
3111                )
3112                .await
3113                .map(|receipt| receipt.value)
3114        })?;
3115        let checkout = Arc::new(SharedCheckout::with_publication(checkout, publication));
3116        Ok((
3117            CheckoutMountSource::new(Arc::clone(&checkout), config)?,
3118            CheckoutMountSource::new(checkout, config)?,
3119        ))
3120    }
3121
3122    fn tracking_sources(
3123        profile: FilesystemProfile,
3124    ) -> Result<(MemorySource, MemorySource), Box<dyn std::error::Error>> {
3125        let mut config = VolumeConfig::portable(Lifecycle::Ephemeral);
3126        config.profile = profile;
3127        let fs = Fs::memory();
3128        let runtime = tokio::runtime::Builder::new_current_thread().build()?;
3129        let (first, second) = runtime.block_on(async {
3130            let cancellation = CancellationToken::new();
3131            let volume = fs
3132                .create_volume(config, WorkBudget::UNBOUNDED, &cancellation)
3133                .await?
3134                .value;
3135            let mode = CheckoutMode {
3136                access: AccessMode::ReadWrite,
3137                consistency: ConsistencyMode::TrackingSafe,
3138                mutations: MutationMode::PrivateOverlay,
3139            };
3140            let first = volume
3141                .checkout(
3142                    GenerationSelector::Head,
3143                    mode,
3144                    WorkBudget::UNBOUNDED,
3145                    &cancellation,
3146                )
3147                .await?
3148                .value;
3149            let second = volume
3150                .checkout(
3151                    GenerationSelector::Head,
3152                    mode,
3153                    WorkBudget::UNBOUNDED,
3154                    &cancellation,
3155                )
3156                .await?
3157                .value;
3158            Ok::<_, OperationFailure<FsError>>((first, second))
3159        })?;
3160        Ok((
3161            CheckoutMountSource::new(Arc::new(SharedCheckout::new(first)), config)?,
3162            CheckoutMountSource::new(Arc::new(SharedCheckout::new(second)), config)?,
3163        ))
3164    }
3165
3166    fn native_test_path(name: &str) -> MountPath {
3167        #[cfg(target_os = "windows")]
3168        let bytes = name.encode_utf16().flat_map(u16::to_le_bytes).collect();
3169        #[cfg(any(target_os = "linux", target_os = "macos"))]
3170        let bytes = name.as_bytes().to_vec();
3171        MountPath::root().child(bytes)
3172    }
3173
3174    #[test]
3175    fn callback_runtime_is_safe_inside_multi_and_current_thread_runtimes()
3176    -> Result<(), Box<dyn std::error::Error>> {
3177        let callback = CallbackRuntime::create()?;
3178        let multi = tokio::runtime::Builder::new_multi_thread()
3179            .worker_threads(2)
3180            .build()?;
3181        let current = tokio::runtime::Builder::new_current_thread().build()?;
3182        let profile = if cfg!(target_os = "windows") {
3183            FilesystemProfile::Windows
3184        } else {
3185            FilesystemProfile::Posix
3186        };
3187        for runtime in [&multi, &current] {
3188            let source = source(profile)?;
3189            runtime.block_on(async {
3190                let value = callback.block_on(|| async {
3191                    // This future is deliberately non-Send across a real reactor wait.
3192                    let local = std::rc::Rc::new(42_u8);
3193                    tokio::time::sleep(Duration::from_millis(1)).await;
3194                    *local
3195                })?;
3196                assert_eq!(value, 42);
3197                let path = native_test_path("runtime.bin");
3198                source.volume_id()?;
3199                let created = source.create_file(&path, metadata())?;
3200                source.write_range(&path, 0, Bytes::from_static(b"base"))?;
3201                let attached = source.open_file(&path)?;
3202                attached.write_range(4, Bytes::from_static(b"-open"))?;
3203                assert_eq!(attached.read_range(0, 9)?.as_ref(), b"base-open");
3204                let detached = source.detach_file(&path)?;
3205                source.remove(&path, Some(created.node.file_id))?;
3206                detached.write_range(0, Bytes::from_static(b"live"))?;
3207                assert_eq!(detached.read_range(0, 9)?.as_ref(), b"live-open");
3208                source.sync()?;
3209                Ok::<(), MountSourceError>(())
3210            })?;
3211        }
3212        Ok(())
3213    }
3214
3215    #[cfg(any(target_os = "linux", target_os = "macos"))]
3216    #[test]
3217    fn callback_runtime_enters_directly_from_native_driver_threads()
3218    -> Result<(), Box<dyn std::error::Error>> {
3219        let callback = CallbackRuntime::create()?;
3220        let observed = std::thread::Builder::new()
3221            .name("native-driver".to_owned())
3222            .spawn(move || {
3223                callback.block_on(|| async {
3224                    std::thread::current()
3225                        .name()
3226                        .unwrap_or("unnamed")
3227                        .to_owned()
3228                })
3229            })?
3230            .join()
3231            .map_err(|_| "native driver callback panicked")??;
3232
3233        assert_eq!(observed, "native-driver");
3234        Ok(())
3235    }
3236
3237    #[cfg(not(any(target_os = "linux", target_os = "macos")))]
3238    #[test]
3239    fn native_driver_callback_uses_a_fresh_stack() -> Result<(), Box<dyn std::error::Error>> {
3240        let callback = CallbackRuntime::create()?;
3241        let observed = std::thread::Builder::new()
3242            .name("native-driver".to_owned())
3243            .spawn(move || {
3244                callback.block_on(|| async {
3245                    std::thread::current()
3246                        .name()
3247                        .unwrap_or("unnamed")
3248                        .to_owned()
3249                })
3250            })?
3251            .join()
3252            .map_err(|_| "native callback panicked")??;
3253        assert_eq!(observed, "acyclic-fs-callback");
3254        Ok(())
3255    }
3256
3257    #[test]
3258    fn publication_policies_are_exact_and_subtree_roots_do_not_scan_or_escape()
3259    -> Result<(), Box<dyn std::error::Error>> {
3260        let profile = if cfg!(target_os = "windows") {
3261            FilesystemProfile::Windows
3262        } else {
3263            FilesystemProfile::Posix
3264        };
3265        let (manual, _) = shared_sources_with_publication(profile, MountPublication::Manual)?;
3266        let initial = checkout_state(&manual)?.0;
3267        manual.create_directory(&native_test_path("src"), metadata())?;
3268        manual.flush()?;
3269        assert_eq!(checkout_state(&manual)?, (initial, true));
3270        manual.sync()?;
3271        let published = checkout_state(&manual)?.0;
3272        assert_ne!(published, initial);
3273
3274        let config = {
3275            let mut config = VolumeConfig::portable(Lifecycle::Ephemeral);
3276            config.profile = profile;
3277            config
3278        };
3279        let root = NamespacePath::new(
3280            vec![manual.logical_name(native_test_path("src").components()[0].as_slice())?],
3281            config.limits,
3282        )?;
3283        let subtree = CheckoutMountSource::new_at(Arc::clone(&manual.checkout), config, root)?;
3284        let nested = native_test_path("nested.bin");
3285        subtree.create_file(&nested, metadata())?;
3286        assert!(manual.lookup(&native_test_path("nested.bin"))?.is_none());
3287        assert!(
3288            manual
3289                .lookup(&native_test_path("src"))?
3290                .is_some_and(|entry| entry.node.kind == MountNodeKind::Directory)
3291        );
3292        assert!(subtree.lookup(&nested)?.is_some());
3293
3294        let (per_mutation, _) =
3295            shared_sources_with_publication(profile, MountPublication::PerMutation)?;
3296        let before = checkout_state(&per_mutation)?.0;
3297        per_mutation.create_file(&native_test_path("published.bin"), metadata())?;
3298        let published_directory = native_test_path("published-directory");
3299        let created_directory = per_mutation.create_directory(&published_directory, metadata())?;
3300        assert_eq!(
3301            per_mutation.lookup(&published_directory)?,
3302            Some(created_directory)
3303        );
3304        let (after, pending) = checkout_state(&per_mutation)?;
3305        assert_ne!(after, before);
3306        assert!(!pending);
3307
3308        let (portable, _) = shared_sources_with_publication(
3309            FilesystemProfile::Portable,
3310            MountPublication::PerMutation,
3311        )?;
3312        let portable_directory = native_test_path("portable-directory");
3313        let created_directory = portable.create_directory(&portable_directory, metadata())?;
3314        assert_eq!(
3315            portable.lookup(&portable_directory)?,
3316            Some(created_directory)
3317        );
3318        Ok(())
3319    }
3320
3321    #[test]
3322    fn refresh_preserves_overlapping_read_conflicts_and_generation()
3323    -> Result<(), Box<dyn std::error::Error>> {
3324        let profile = if cfg!(target_os = "windows") {
3325            FilesystemProfile::Windows
3326        } else {
3327            FilesystemProfile::Posix
3328        };
3329        let (reader, writer) = tracking_sources(profile)?;
3330        let path = native_test_path("observed.bin");
3331        writer.create_file(&path, metadata())?;
3332        writer.write_range(&path, 0, Bytes::from_static(b"first"))?;
3333        writer.sync()?;
3334        let binding_before = reader.binding_epoch();
3335        reader.runtime.block_on(|| reader.refresh_async())??;
3336        let binding_after = reader.binding_epoch();
3337        assert_ne!(binding_after, binding_before);
3338        assert_eq!(reader.read_range(&path, 0, 5)?.as_ref(), b"first");
3339        let observed_generation = checkout_state(&reader)?.0;
3340
3341        writer.write_range(&path, 0, Bytes::from_static(b"other"))?;
3342        writer.sync()?;
3343        assert!(matches!(
3344            reader.runtime.block_on(|| reader.refresh_async())?,
3345            Err(MountSourceError::Stale)
3346        ));
3347        assert_eq!(checkout_state(&reader)?.0, observed_generation);
3348        assert_eq!(reader.binding_epoch(), binding_after);
3349        assert_eq!(reader.read_range(&path, 0, 5)?.as_ref(), b"first");
3350        Ok(())
3351    }
3352
3353    #[test]
3354    fn detached_mount_file_survives_last_binding_with_sparse_mutation_and_metadata()
3355    -> Result<(), Box<dyn std::error::Error>> {
3356        let profile = if cfg!(target_os = "windows") {
3357            FilesystemProfile::Windows
3358        } else {
3359            FilesystemProfile::Posix
3360        };
3361        let source = source(profile)?;
3362        let file = native_test_path("detached.bin");
3363        source.create_file(&file, metadata())?;
3364        source.resize(&file, 64)?;
3365        source.write_range(&file, 8, Bytes::from_static(b"visible"))?;
3366
3367        let detached = source.detach_file(&file)?;
3368        let expected = source.lookup(&file)?.map(|lookup| lookup.node.file_id);
3369        source.remove(&file, expected)?;
3370        assert_eq!(source.lookup(&file)?, None);
3371        assert_eq!(detached.lookup()?.node.link_count, 0);
3372        assert_eq!(detached.read_range(8, 7)?.as_ref(), b"visible");
3373
3374        detached.write_range(32, Bytes::from_static(b"open"))?;
3375        detached.allocate_range(16, 8, MountRangeAllocation::Preallocate { keep_size: true })?;
3376        let changed = FileMetadata {
3377            modified_ns: crate::kernel::MetadataField::Value(42),
3378            ..metadata()
3379        };
3380        detached.set_attributes(changed, Some(40))?;
3381        let lookup = detached.lookup()?;
3382        assert_eq!(lookup.node.logical_bytes, 40);
3383        assert_eq!(lookup.metadata, changed);
3384        assert_eq!(detached.read_range(32, 4)?.as_ref(), b"open");
3385        assert_eq!(detached.seek(0, MountSeekTarget::Data)?, Some(8));
3386        #[cfg(any(target_os = "linux", target_os = "macos"))]
3387        {
3388            let attribute = b"user.acyclic.detached";
3389            detached.write_attribute(
3390                attribute,
3391                Bytes::from_static(b"retained"),
3392                MountAttributeWriteMode::Create,
3393            )?;
3394            assert_eq!(
3395                detached.read_attribute(attribute)?.as_deref(),
3396                Some(b"retained".as_slice())
3397            );
3398            let attributes = detached.list_attributes(None, 8)?;
3399            assert_eq!(attributes.names, vec![attribute.to_vec()]);
3400            detached.remove_attribute(attribute)?;
3401            assert_eq!(detached.read_attribute(attribute)?, None);
3402        }
3403        Ok(())
3404    }
3405
3406    #[test]
3407    fn shared_checkout_fence_blocks_every_adapter_until_exact_resolution()
3408    -> Result<(), Box<dyn std::error::Error>> {
3409        let profile = if cfg!(target_os = "windows") {
3410            FilesystemProfile::Windows
3411        } else {
3412            FilesystemProfile::Posix
3413        };
3414        let (first, second) = shared_sources(profile)?;
3415        let operation_id = OperationId::new();
3416        first.runtime.block_on(|| async {
3417            first
3418                .checkout
3419                .lock()
3420                .await
3421                .retain_operation_id(operation_id)
3422        })??;
3423        let path = native_test_path("fenced.bin");
3424        assert!(matches!(
3425            second.create_file(&path, metadata()),
3426            Err(MountSourceError::Stale)
3427        ));
3428        first.runtime.block_on(|| async {
3429            first
3430                .checkout
3431                .lock()
3432                .await
3433                .clear_retained_operation(operation_id);
3434        })?;
3435        second.create_file(&path, metadata())?;
3436        Ok(())
3437    }
3438
3439    #[test]
3440    fn detached_sparse_transfer_preserves_holes_and_allocated_spans()
3441    -> Result<(), Box<dyn std::error::Error>> {
3442        let profile = if cfg!(target_os = "windows") {
3443            FilesystemProfile::Windows
3444        } else {
3445            FilesystemProfile::Posix
3446        };
3447        let source = source(profile)?;
3448        let from = native_test_path("sparse-source.bin");
3449        let to = native_test_path("sparse-destination.bin");
3450        source.create_file(&from, metadata())?;
3451        source.resize(&from, 128)?;
3452        source.write_range(&from, 64, Bytes::from_static(b"payload"))?;
3453        source.create_file(&to, metadata())?;
3454        source.write_range(&to, 0, Bytes::from_static(b"dense"))?;
3455        let from = source.detach_file(&from)?;
3456        let to = source.detach_file(&to)?;
3457        let sparse = from.read_sparse_range(0, 128)?;
3458        to.write_sparse_range(0, &sparse)?;
3459        assert_eq!(to.lookup()?.node.logical_bytes, 128);
3460        assert_eq!(to.seek(0, MountSeekTarget::Data)?, Some(64));
3461        assert_eq!(to.seek(64, MountSeekTarget::Hole)?, Some(71));
3462        assert_eq!(to.read_range(64, 7)?.as_ref(), b"payload");
3463        Ok(())
3464    }
3465
3466    #[cfg(any(target_os = "linux", target_os = "macos"))]
3467    #[test]
3468    fn host_capture_preserves_unobservable_named_metadata() -> Result<(), Box<dyn std::error::Error>>
3469    {
3470        let source = source(FilesystemProfile::Posix)?;
3471        let path = native_test_path("metadata.bin");
3472        source.create_file(&path, metadata())?;
3473        source.write_attribute(
3474            &path,
3475            b"user.acyclic.retained",
3476            Bytes::from_static(b"opaque"),
3477            MountAttributeWriteMode::Create,
3478        )?;
3479        let root = tempfile::tempdir()?;
3480        std::fs::write(root.path().join("metadata.bin"), b"host")?;
3481        source.capture_host_path(root.path(), &path)?;
3482        assert_eq!(
3483            source
3484                .read_attribute(&path, b"user.acyclic.retained")?
3485                .as_deref(),
3486            Some(b"opaque".as_slice())
3487        );
3488        Ok(())
3489    }
3490
3491    #[test]
3492    fn attached_open_file_tracks_identity_across_unlink_hard_link_and_rename()
3493    -> Result<(), Box<dyn std::error::Error>> {
3494        let profile = if cfg!(target_os = "windows") {
3495            FilesystemProfile::Windows
3496        } else {
3497            FilesystemProfile::Posix
3498        };
3499        let source = source(profile)?;
3500        let original = native_test_path("original.bin");
3501        let alias = native_test_path("alias.bin");
3502        let renamed = native_test_path("renamed.bin");
3503        let created = source.create_file(&original, metadata())?;
3504        source.write_range(&original, 0, Bytes::from_static(b"base"))?;
3505        let open = source.open_file(&original)?;
3506        source.hard_link(&original, &alias)?;
3507        source.remove(&original, Some(created.node.file_id))?;
3508
3509        assert_eq!(source.lookup(&original)?, None);
3510        assert_eq!(open.lookup()?.node.link_count, 1);
3511        open.write_range(4, Bytes::from_static(b"-open"))?;
3512        assert_eq!(source.read_range(&alias, 0, 9)?.as_ref(), b"base-open");
3513
3514        source.rename(&alias, &renamed, false)?;
3515        open.write_range(0, Bytes::from_static(b"live"))?;
3516        assert_eq!(source.lookup(&alias)?, None);
3517        assert_eq!(source.read_range(&renamed, 0, 9)?.as_ref(), b"live-open");
3518        assert_eq!(open.lookup()?.node.file_id, created.node.file_id);
3519
3520        #[cfg(any(target_os = "linux", target_os = "macos"))]
3521        {
3522            let name = b"user.acyclic.attached";
3523            open.write_attribute(
3524                name,
3525                Bytes::from_static(b"identity"),
3526                MountAttributeWriteMode::Create,
3527            )?;
3528            assert_eq!(
3529                open.read_attribute(name)?.as_deref(),
3530                Some(b"identity".as_slice())
3531            );
3532            assert_eq!(
3533                source.read_attribute(&renamed, name)?.as_deref(),
3534                Some(b"identity".as_slice())
3535            );
3536        }
3537        Ok(())
3538    }
3539
3540    #[cfg(any(target_os = "linux", target_os = "macos"))]
3541    #[test]
3542    fn sparse_seek_and_special_nodes_preserve_canonical_native_semantics()
3543    -> Result<(), Box<dyn std::error::Error>> {
3544        let source = source(FilesystemProfile::Posix)?;
3545        let file = MountPath::root().child(b"sparse".to_vec());
3546        source.create_file(&file, metadata())?;
3547        source.resize(&file, 100)?;
3548        source.write_range(&file, 20, Bytes::from_static(b"data"))?;
3549        assert_eq!(source.seek(&file, 0, MountSeekTarget::Data)?, Some(20));
3550        assert_eq!(source.seek(&file, 20, MountSeekTarget::Hole)?, Some(24));
3551        assert_eq!(source.seek(&file, 24, MountSeekTarget::Data)?, None);
3552        assert_eq!(source.seek(&file, 100, MountSeekTarget::Hole)?, Some(100));
3553
3554        let fifo = MountPath::root().child(b"fifo".to_vec());
3555        let created = source.create_special(&fifo, MountNodeKind::Fifo, None, metadata())?;
3556        assert_eq!(created.node.kind, MountNodeKind::Fifo);
3557        let device = MountPath::root().child(b"device".to_vec());
3558        let created = source.create_special(
3559            &device,
3560            MountNodeKind::CharacterDevice,
3561            Some((12, 34)),
3562            metadata(),
3563        )?;
3564        assert_eq!(created.node.device, Some((12, 34)));
3565        assert_eq!(source.lookup(&device)?, Some(created));
3566        Ok(())
3567    }
3568
3569    #[cfg(target_os = "windows")]
3570    #[test]
3571    fn windows_mount_path_preserves_exact_utf16_components()
3572    -> Result<(), Box<dyn std::error::Error>> {
3573        let source = source(FilesystemProfile::Windows)?;
3574        let (initial, pending) = checkout_state(&source)?;
3575        assert!(!pending);
3576        let name = "exact-α-😀"
3577            .encode_utf16()
3578            .flat_map(u16::to_le_bytes)
3579            .collect::<Vec<_>>();
3580        let path = MountPath::root().child(name.clone());
3581        let exact_metadata = FileMetadata {
3582            windows_attributes: crate::kernel::MetadataField::Value(0x20),
3583            created_ns: crate::kernel::MetadataField::Value(1_234_500),
3584            modified_ns: crate::kernel::MetadataField::Value(2_345_600),
3585            ..metadata()
3586        };
3587        let created = source.create_file(&path, exact_metadata)?;
3588        assert_eq!(source.lookup(&path)?, Some(created));
3589        let page = source.read_directory(&MountPath::root(), None, 8)?;
3590        assert_eq!(page.entries.len(), 1);
3591        assert_eq!(page.entries[0].name, name);
3592        assert_eq!(page.entries[0].metadata, exact_metadata);
3593        assert!(checkout_state(&source)?.1);
3594        source.flush()?;
3595        source.flush()?;
3596        let (published, pending) = checkout_state(&source)?;
3597        assert_ne!(published, initial);
3598        assert!(!pending);
3599
3600        let host = tempfile::tempdir()?;
3601        let captured_text = "captured-β.bin";
3602        std::fs::write(host.path().join(captured_text), b"captured")?;
3603        let captured_name = captured_text
3604            .encode_utf16()
3605            .flat_map(u16::to_le_bytes)
3606            .collect::<Vec<_>>();
3607        let captured_path = MountPath::root().child(captured_name);
3608        source.capture_host_path(host.path(), &captured_path)?;
3609        assert_eq!(
3610            source.read_range(&captured_path, 0, 8)?.as_ref(),
3611            b"captured"
3612        );
3613        std::fs::remove_file(host.path().join(captured_text))?;
3614        source.capture_host_path(host.path(), &captured_path)?;
3615        assert_eq!(source.lookup(&captured_path)?, None);
3616        Ok(())
3617    }
3618
3619    #[cfg(target_os = "windows")]
3620    #[test]
3621    #[ignore = "requires the live Windows ProjFS optional feature"]
3622    fn writable_projfs_captures_closes_renames_links_and_deletes()
3623    -> Result<(), Box<dyn std::error::Error>> {
3624        let source = Arc::new(source(FilesystemProfile::Windows)?);
3625        let temporary = tempfile::tempdir()?;
3626        let destination = temporary.path().join("projection");
3627        std::fs::create_dir(&destination)?;
3628        let mut session = crate::mount_native(
3629            crate::NativeMountRequest {
3630                mount_id: crate::MountId::new(),
3631                volume_id: source.volume_id()?,
3632                destination: destination.clone(),
3633                writable: true,
3634            },
3635            Arc::clone(&source) as Arc<dyn MountFilesystem>,
3636        )?;
3637
3638        let first_name = "written-α.bin";
3639        let second_name = "renamed-β.bin";
3640        let linked_name = "linked-γ.bin";
3641        let write_status = std::process::Command::new("powershell.exe")
3642            .args([
3643                "-NoProfile",
3644                "-NonInteractive",
3645                "-Command",
3646                "[IO.File]::WriteAllBytes($env:ACYCLIC_FS_TEST_PATH, [Text.Encoding]::UTF8.GetBytes('projected'))",
3647            ])
3648            .env("ACYCLIC_FS_TEST_PATH", destination.join(first_name))
3649            .status()?;
3650        assert!(write_status.success());
3651        let first = windows_path(first_name);
3652        assert_eq!(source.read_range(&first, 0, 9)?.as_ref(), b"projected");
3653
3654        let rename_status = std::process::Command::new("powershell.exe")
3655            .args([
3656                "-NoProfile",
3657                "-NonInteractive",
3658                "-Command",
3659                "[IO.File]::Move($env:ACYCLIC_FS_TEST_SOURCE, $env:ACYCLIC_FS_TEST_DESTINATION)",
3660            ])
3661            .env("ACYCLIC_FS_TEST_SOURCE", destination.join(first_name))
3662            .env("ACYCLIC_FS_TEST_DESTINATION", destination.join(second_name))
3663            .status()?;
3664        assert!(rename_status.success());
3665        let second = windows_path(second_name);
3666        assert_eq!(source.lookup(&first)?, None);
3667        assert!(source.lookup(&second)?.is_some());
3668
3669        let link_status = std::process::Command::new("powershell.exe")
3670            .args([
3671                "-NoProfile",
3672                "-NonInteractive",
3673                "-Command",
3674                "New-Item -ItemType HardLink -Path $env:ACYCLIC_FS_TEST_DESTINATION -Target $env:ACYCLIC_FS_TEST_SOURCE | Out-Null",
3675            ])
3676            .env("ACYCLIC_FS_TEST_SOURCE", destination.join(second_name))
3677            .env("ACYCLIC_FS_TEST_DESTINATION", destination.join(linked_name))
3678            .status()?;
3679        assert!(link_status.success());
3680        let linked = windows_path(linked_name);
3681        let second_id = source
3682            .lookup(&second)?
3683            .ok_or("renamed source absent")?
3684            .node
3685            .file_id;
3686        assert_eq!(
3687            source
3688                .lookup(&linked)?
3689                .ok_or("hard link absent")?
3690                .node
3691                .file_id,
3692            second_id
3693        );
3694
3695        let delete_status = std::process::Command::new("powershell.exe")
3696            .args([
3697                "-NoProfile",
3698                "-NonInteractive",
3699                "-Command",
3700                "[IO.File]::Delete($env:ACYCLIC_FS_TEST_PATH)",
3701            ])
3702            .env("ACYCLIC_FS_TEST_PATH", destination.join(second_name))
3703            .status()?;
3704        assert!(delete_status.success());
3705        assert_eq!(source.lookup(&second)?, None);
3706        assert!(source.lookup(&linked)?.is_some());
3707        assert!(session.stop()?);
3708        Ok(())
3709    }
3710
3711    #[cfg(target_os = "windows")]
3712    fn windows_path(value: &str) -> MountPath {
3713        MountPath::root().child(value.encode_utf16().flat_map(u16::to_le_bytes).collect())
3714    }
3715
3716    // Every Unix backend must support independent sequential sessions. On
3717    // macOS this also proves that each loopback NFS server receives its own
3718    // kernel-assigned port; on Linux it exercises two distinct FUSE sessions.
3719    #[cfg(any(target_os = "linux", target_os = "macos"))]
3720    #[test]
3721    #[ignore = "mounts two live Unix sessions; requires the host's native mount capability"]
3722    fn two_sequential_unix_mounts_in_one_process_both_stay_independently_writable()
3723    -> Result<(), Box<dyn std::error::Error>> {
3724        let source_a = Arc::new(source(FilesystemProfile::Posix)?);
3725        let source_b = Arc::new(source(FilesystemProfile::Posix)?);
3726        let temporary_a = tempfile::tempdir()?;
3727        let temporary_b = tempfile::tempdir()?;
3728
3729        let mut mount_a = crate::mount_native(
3730            crate::NativeMountRequest {
3731                mount_id: crate::MountId::new(),
3732                volume_id: source_a.volume_id()?,
3733                destination: temporary_a.path().to_path_buf(),
3734                writable: true,
3735            },
3736            Arc::clone(&source_a) as Arc<dyn MountFilesystem>,
3737        )?;
3738        let mut mount_b = crate::mount_native(
3739            crate::NativeMountRequest {
3740                mount_id: crate::MountId::new(),
3741                volume_id: source_b.volume_id()?,
3742                destination: temporary_b.path().to_path_buf(),
3743                writable: true,
3744            },
3745            Arc::clone(&source_b) as Arc<dyn MountFilesystem>,
3746        )?;
3747
3748        std::fs::write(temporary_a.path().join("a.txt"), b"from-a")?;
3749        std::fs::write(temporary_b.path().join("b.txt"), b"from-b")?;
3750        assert_eq!(std::fs::read(temporary_a.path().join("a.txt"))?, b"from-a");
3751        assert_eq!(std::fs::read(temporary_b.path().join("b.txt"))?, b"from-b");
3752
3753        assert!(mount_a.stop()?);
3754        assert!(mount_b.stop()?);
3755        Ok(())
3756    }
3757
3758    // `O_CREAT|O_EXCL` is what mkstemp, git's index.lock and editors' atomic
3759    // saves use. On macOS the NFS client sends it as an EXCLUSIVE4 create,
3760    // whose verifier the server parks in the new file's timestamps until the
3761    // client's follow-up SETATTR replaces them (RFC 7530 s16.16.5). The
3762    // second create must see the existing file, and the timestamps a caller
3763    // observes must be real ones, not the verifier. Removing and replacing
3764    // the path must not let a remembered verifier reopen the replacement.
3765    #[cfg(any(target_os = "linux", target_os = "macos"))]
3766    #[test]
3767    #[ignore = "mounts a live Unix session; requires the host's native mount capability"]
3768    fn exclusive_create_on_a_unix_mount_creates_once_with_real_timestamps()
3769    -> Result<(), Box<dyn std::error::Error>> {
3770        let source = Arc::new(source(FilesystemProfile::Posix)?);
3771        let temporary = tempfile::tempdir()?;
3772        let mut mount = crate::mount_native(
3773            crate::NativeMountRequest {
3774                mount_id: crate::MountId::new(),
3775                volume_id: source.volume_id()?,
3776                destination: temporary.path().to_path_buf(),
3777                writable: true,
3778            },
3779            Arc::clone(&source) as Arc<dyn MountFilesystem>,
3780        )?;
3781
3782        let path = temporary.path().join("index.lock");
3783        let before = std::time::SystemTime::now() - std::time::Duration::from_secs(60);
3784        let mut first = std::fs::OpenOptions::new()
3785            .write(true)
3786            .create_new(true)
3787            .open(&path)?;
3788        std::io::Write::write_all(&mut first, b"locked")?;
3789        drop(first);
3790        let second = std::fs::OpenOptions::new()
3791            .write(true)
3792            .create_new(true)
3793            .open(&path);
3794        assert_eq!(
3795            second.err().map(|error| error.kind()),
3796            Some(std::io::ErrorKind::AlreadyExists)
3797        );
3798        let metadata = std::fs::metadata(&path)?;
3799        assert!(
3800            metadata.modified()? >= before && metadata.accessed()? >= before,
3801            "timestamps still carry the create verifier: {metadata:?}"
3802        );
3803        assert_eq!(std::fs::read(&path)?, b"locked");
3804
3805        std::fs::remove_file(&path)?;
3806        std::fs::write(&path, b"replacement")?;
3807        let after_replacement = std::fs::OpenOptions::new()
3808            .write(true)
3809            .create_new(true)
3810            .open(&path);
3811        assert_eq!(
3812            after_replacement.err().map(|error| error.kind()),
3813            Some(std::io::ErrorKind::AlreadyExists)
3814        );
3815        assert_eq!(std::fs::read(&path)?, b"replacement");
3816
3817        assert!(mount.stop()?);
3818        Ok(())
3819    }
3820
3821    #[cfg(target_os = "linux")]
3822    #[test]
3823    #[ignore = "mounts a live FUSE session; requires the host's native mount capability"]
3824    fn linux_negative_lookup_clears_on_create_and_invalidation()
3825    -> Result<(), Box<dyn std::error::Error>> {
3826        use std::io::{Read as _, Seek as _, SeekFrom};
3827
3828        let source = Arc::new(source(FilesystemProfile::Posix)?);
3829        let temporary = tempfile::tempdir()?;
3830        let mut mount = crate::mount_native(
3831            crate::NativeMountRequest {
3832                mount_id: crate::MountId::new(),
3833                volume_id: source.volume_id()?,
3834                destination: temporary.path().to_path_buf(),
3835                writable: true,
3836            },
3837            Arc::clone(&source) as Arc<dyn MountFilesystem>,
3838        )?;
3839
3840        let local = temporary.path().join("created-locally");
3841        assert_eq!(
3842            std::fs::metadata(&local).err().map(|error| error.kind()),
3843            Some(std::io::ErrorKind::NotFound)
3844        );
3845        std::fs::write(&local, b"local")?;
3846        assert_eq!(std::fs::read(&local)?, b"local");
3847
3848        let external = temporary.path().join("created-by-source");
3849        assert_eq!(
3850            std::fs::metadata(&external).err().map(|error| error.kind()),
3851            Some(std::io::ErrorKind::NotFound)
3852        );
3853        source.create_file(
3854            &MountPath::root().child(b"created-by-source".to_vec()),
3855            metadata(),
3856        )?;
3857        mount.invalidate(b"/created-by-source")?;
3858        assert!(
3859            external.is_file(),
3860            "invalidation must evict a negative dentry"
3861        );
3862
3863        let nested = temporary.path().join("nested");
3864        std::fs::create_dir(&nested)?;
3865        let nested_external = nested.join("created-by-source");
3866        assert_eq!(
3867            std::fs::metadata(&nested_external)
3868                .err()
3869                .map(|error| error.kind()),
3870            Some(std::io::ErrorKind::NotFound)
3871        );
3872        source.create_file(
3873            &MountPath::root()
3874                .child(b"nested".to_vec())
3875                .child(b"created-by-source".to_vec()),
3876            metadata(),
3877        )?;
3878        mount.invalidate(b"/nested/created-by-source")?;
3879        assert!(
3880            nested_external.is_file(),
3881            "nested invalidation must evict a negative dentry"
3882        );
3883
3884        let data_path = MountPath::root().child(b"changed-by-source".to_vec());
3885        source.create_file(&data_path, metadata())?;
3886        source.write_range(&data_path, 0, Bytes::from_static(b"before"))?;
3887        mount.invalidate(b"/changed-by-source")?;
3888        let mut opened = std::fs::File::open(temporary.path().join("changed-by-source"))?;
3889        let mut body = [0_u8; 6];
3890        opened.read_exact(&mut body)?;
3891        assert_eq!(&body, b"before");
3892        source.write_range(&data_path, 0, Bytes::from_static(b"after!"))?;
3893        mount.invalidate(b"/changed-by-source")?;
3894        opened.seek(SeekFrom::Start(0))?;
3895        opened.read_exact(&mut body)?;
3896        assert_eq!(&body, b"after!");
3897        drop(opened);
3898
3899        assert!(mount.stop()?);
3900        Ok(())
3901    }
3902
3903    #[cfg(target_os = "linux")]
3904    #[test]
3905    #[ignore = "opens a live FUSE file with O_TRUNC; requires the host's native mount capability"]
3906    fn linux_truncate_open_keeps_its_own_handle() -> Result<(), Box<dyn std::error::Error>> {
3907        use std::io::Write as _;
3908
3909        let source = Arc::new(source(FilesystemProfile::Posix)?);
3910        let temporary = tempfile::tempdir()?;
3911        let mut mount = crate::mount_native(
3912            crate::NativeMountRequest {
3913                mount_id: crate::MountId::new(),
3914                volume_id: source.volume_id()?,
3915                destination: temporary.path().to_path_buf(),
3916                writable: true,
3917            },
3918            Arc::clone(&source) as Arc<dyn MountFilesystem>,
3919        )?;
3920        let path = temporary.path().join("incremental-output");
3921        std::fs::write(&path, b"old")?;
3922        for iteration in 0..8 {
3923            let mut file = std::fs::OpenOptions::new()
3924                .write(true)
3925                .truncate(true)
3926                .open(&path)?;
3927            write!(file, "new-{iteration}")?;
3928            file.sync_all()?;
3929            assert_eq!(std::fs::read_to_string(&path)?, format!("new-{iteration}"));
3930        }
3931        assert!(mount.stop()?);
3932        Ok(())
3933    }
3934
3935    #[cfg(target_os = "linux")]
3936    #[test]
3937    #[ignore = "requires a live Linux FUSE mount"]
3938    fn linux_hard_link_alias_survives_mounted_write_and_rename()
3939    -> Result<(), Box<dyn std::error::Error>> {
3940        use std::os::unix::fs::MetadataExt as _;
3941
3942        let source = Arc::new(source(FilesystemProfile::Posix)?);
3943        let temporary = tempfile::tempdir()?;
3944        let mut mount = crate::mount_native(
3945            crate::NativeMountRequest {
3946                mount_id: crate::MountId::new(),
3947                volume_id: source.volume_id()?,
3948                destination: temporary.path().to_path_buf(),
3949                writable: true,
3950            },
3951            Arc::clone(&source) as Arc<dyn MountFilesystem>,
3952        )?;
3953        let original = temporary.path().join("original");
3954        let alias = temporary.path().join("alias");
3955        let renamed = temporary.path().join("renamed");
3956        std::fs::write(&original, b"before")?;
3957        std::fs::hard_link(&original, &alias)?;
3958        assert_eq!(
3959            std::fs::metadata(&original)?.ino(),
3960            std::fs::metadata(&alias)?.ino()
3961        );
3962        assert_eq!(std::fs::metadata(&alias)?.nlink(), 2);
3963
3964        std::fs::write(&alias, b"after")?;
3965        assert_eq!(std::fs::read(&original)?, b"after");
3966        std::fs::rename(&original, &renamed)?;
3967        assert_eq!(
3968            std::fs::metadata(&renamed)?.ino(),
3969            std::fs::metadata(&alias)?.ino()
3970        );
3971        assert_eq!(std::fs::read(&renamed)?, b"after");
3972        std::fs::remove_file(&alias)?;
3973        assert_eq!(std::fs::metadata(&renamed)?.nlink(), 1);
3974        assert_eq!(std::fs::read(&renamed)?, b"after");
3975        assert!(mount.stop()?);
3976        Ok(())
3977    }
3978
3979    #[cfg(target_os = "linux")]
3980    #[test]
3981    #[ignore = "requires a live Linux FUSE mount"]
3982    fn linux_fsync_waits_for_an_admitted_write() -> Result<(), Box<dyn std::error::Error>> {
3983        use std::io::Write as _;
3984
3985        let source = Arc::new(source(FilesystemProfile::Posix)?);
3986        let temporary = tempfile::tempdir()?;
3987        let mut mount = crate::mount_native(
3988            crate::NativeMountRequest {
3989                mount_id: crate::MountId::new(),
3990                volume_id: source.volume_id()?,
3991                destination: temporary.path().to_path_buf(),
3992                writable: true,
3993            },
3994            Arc::clone(&source) as Arc<dyn MountFilesystem>,
3995        )?;
3996        let path = temporary.path().join("ordered-write");
3997        std::fs::write(&path, b"old")?;
3998        let file = std::fs::OpenOptions::new()
3999            .read(true)
4000            .write(true)
4001            .open(&path)?;
4002        let mut writer = file.try_clone()?;
4003        let syncer = file.try_clone()?;
4004        let gate = super::super::fuse::pause_next_write_after_admission();
4005        let writer_thread = std::thread::spawn(move || writer.write_all(b"new"));
4006        assert!(
4007            gate.wait_until_admitted(Duration::from_secs(5)),
4008            "write must reach the per-handle gate"
4009        );
4010
4011        let (started, started_rx) = std::sync::mpsc::sync_channel(1);
4012        let (finished, finished_rx) = std::sync::mpsc::sync_channel(1);
4013        let sync_thread = std::thread::spawn(move || {
4014            let _ = started.send(());
4015            let result = syncer.sync_all();
4016            let _ = finished.send(result);
4017        });
4018        started_rx.recv_timeout(Duration::from_secs(5))?;
4019        assert!(
4020            finished_rx
4021                .recv_timeout(Duration::from_millis(100))
4022                .is_err(),
4023            "fsync must not overtake a write admitted on the same handle"
4024        );
4025        assert!(gate.release(), "release paused write");
4026        writer_thread
4027            .join()
4028            .map_err(|_| std::io::Error::other("mounted writer panicked"))??;
4029        sync_thread
4030            .join()
4031            .map_err(|_| std::io::Error::other("mounted fsync panicked"))?;
4032        finished_rx.recv_timeout(Duration::from_secs(5))??;
4033        drop(file);
4034        assert_eq!(std::fs::read(&path)?, b"new");
4035        assert!(
4036            !checkout_state(&source)?.1,
4037            "fsync published the admitted write"
4038        );
4039        assert!(mount.stop()?);
4040        Ok(())
4041    }
4042
4043    #[cfg(target_os = "linux")]
4044    #[test]
4045    #[ignore = "requires a live Linux FUSE mount"]
4046    fn linux_fsync_on_clean_descriptor_publishes_other_descriptor_write()
4047    -> Result<(), Box<dyn std::error::Error>> {
4048        use std::io::Write as _;
4049
4050        let source = Arc::new(source(FilesystemProfile::Posix)?);
4051        let temporary = tempfile::tempdir()?;
4052        let mut mount = crate::mount_native(
4053            crate::NativeMountRequest {
4054                mount_id: crate::MountId::new(),
4055                volume_id: source.volume_id()?,
4056                destination: temporary.path().to_path_buf(),
4057                writable: true,
4058            },
4059            Arc::clone(&source) as Arc<dyn MountFilesystem>,
4060        )?;
4061        let path = temporary.path().join("cross-descriptor-fsync");
4062        std::fs::write(&path, b"old")?;
4063        let syncer = std::fs::OpenOptions::new().read(true).open(&path)?;
4064        let mut writer = std::fs::OpenOptions::new().write(true).open(&path)?;
4065        writer.write_all(b"new")?;
4066        assert!(checkout_state(&source)?.1, "write remains pending");
4067        syncer.sync_all()?;
4068        assert!(
4069            !checkout_state(&source)?.1,
4070            "explicit fsync published the write"
4071        );
4072        assert_eq!(std::fs::read(&path)?, b"new");
4073        writer.write_all(b"2")?;
4074        assert!(checkout_state(&source)?.1, "second write remains pending");
4075        std::fs::File::open(temporary.path())?.sync_all()?;
4076        assert!(
4077            !checkout_state(&source)?.1,
4078            "directory fsync published the write"
4079        );
4080        drop(writer);
4081        drop(syncer);
4082        assert!(mount.stop()?);
4083        Ok(())
4084    }
4085
4086    #[cfg(target_os = "linux")]
4087    #[test]
4088    #[ignore = "requires a live Linux FUSE mount"]
4089    fn linux_concurrent_truncates_do_not_fence_unrelated_files()
4090    -> Result<(), Box<dyn std::error::Error>> {
4091        use std::io::Write as _;
4092
4093        let source = Arc::new(source(FilesystemProfile::Posix)?);
4094        let temporary = tempfile::tempdir()?;
4095        let mut mount = crate::mount_native(
4096            crate::NativeMountRequest {
4097                mount_id: crate::MountId::new(),
4098                volume_id: source.volume_id()?,
4099                destination: temporary.path().to_path_buf(),
4100                writable: true,
4101            },
4102            Arc::clone(&source) as Arc<dyn MountFilesystem>,
4103        )?;
4104
4105        // Unrelated mounted mutations may advance the global cache epoch
4106        // while another file is opening. They must not fence its attached
4107        // handle or make O_TRUNC report ESTALE.
4108        let barrier = Arc::new(std::sync::Barrier::new(9));
4109        let mut workers = Vec::new();
4110        for worker in 0..8 {
4111            let path = temporary.path().join(format!("concurrent-{worker}"));
4112            std::fs::write(&path, b"old")?;
4113            let barrier = Arc::clone(&barrier);
4114            workers.push(std::thread::spawn(move || -> std::io::Result<()> {
4115                barrier.wait();
4116                for iteration in 0..8 {
4117                    let mut file = std::fs::OpenOptions::new()
4118                        .write(true)
4119                        .truncate(true)
4120                        .open(&path)
4121                        .map_err(|error| {
4122                            std::io::Error::other(format!(
4123                                "concurrent open {worker}/{iteration}: {error}"
4124                            ))
4125                        })?;
4126                    write!(file, "new-{worker}-{iteration}")?;
4127                    file.sync_all()?;
4128                }
4129                assert_eq!(std::fs::read_to_string(&path)?, format!("new-{worker}-7"));
4130                Ok(())
4131            }));
4132        }
4133        barrier.wait();
4134        for worker in workers {
4135            worker
4136                .join()
4137                .map_err(|_| std::io::Error::other("mounted writer panicked"))??;
4138        }
4139        assert!(mount.stop()?);
4140        Ok(())
4141    }
4142
4143    #[cfg(target_os = "linux")]
4144    #[test]
4145    #[ignore = "mounts a live FUSE session and invokes the installed Rust compiler"]
4146    fn rustc_compiles_inside_a_linux_mount() -> Result<(), Box<dyn std::error::Error>> {
4147        let source = Arc::new(source(FilesystemProfile::Posix)?);
4148        let temporary = tempfile::tempdir()?;
4149        let mut mount = crate::mount_native(
4150            crate::NativeMountRequest {
4151                mount_id: crate::MountId::new(),
4152                volume_id: source.volume_id()?,
4153                destination: temporary.path().to_path_buf(),
4154                writable: true,
4155            },
4156            Arc::clone(&source) as Arc<dyn MountFilesystem>,
4157        )?;
4158
4159        std::fs::write(temporary.path().join("main.rs"), b"fn main() {}\n")?;
4160        let output = std::process::Command::new("rustc")
4161            .current_dir(temporary.path())
4162            .args(["--edition", "2021", "main.rs", "-o", "main"])
4163            .output()?;
4164        assert!(
4165            output.status.success(),
4166            "rustc failed\nstdout:\n{}\nstderr:\n{}",
4167            String::from_utf8_lossy(&output.stdout),
4168            String::from_utf8_lossy(&output.stderr)
4169        );
4170        assert!(temporary.path().join("main").is_file());
4171        assert!(
4172            std::process::Command::new(temporary.path().join("main"))
4173                .status()?
4174                .success(),
4175            "compiled executable did not run from the mount"
4176        );
4177
4178        assert!(mount.stop()?);
4179        Ok(())
4180    }
4181
4182    // Simultaneous callers must receive independent servers, callback state,
4183    // kernel mounts, and teardown lifecycles.
4184    #[cfg(any(target_os = "linux", target_os = "macos"))]
4185    #[test]
4186    #[ignore = "mounts two live Unix sessions concurrently; requires the host's native mount capability"]
4187    fn two_concurrent_unix_mounts_in_one_process_remain_independent()
4188    -> Result<(), Box<dyn std::error::Error>> {
4189        let source_a = Arc::new(source(FilesystemProfile::Posix)?);
4190        let source_b = Arc::new(source(FilesystemProfile::Posix)?);
4191        let temporary_a = tempfile::tempdir()?;
4192        let temporary_b = tempfile::tempdir()?;
4193        let request_a = crate::NativeMountRequest {
4194            mount_id: crate::MountId::new(),
4195            volume_id: source_a.volume_id()?,
4196            destination: temporary_a.path().to_path_buf(),
4197            writable: true,
4198        };
4199        let request_b = crate::NativeMountRequest {
4200            mount_id: crate::MountId::new(),
4201            volume_id: source_b.volume_id()?,
4202            destination: temporary_b.path().to_path_buf(),
4203            writable: true,
4204        };
4205        let dyn_source_a = Arc::clone(&source_a) as Arc<dyn MountFilesystem>;
4206        let dyn_source_b = Arc::clone(&source_b) as Arc<dyn MountFilesystem>;
4207
4208        let (result_a, result_b) = std::thread::scope(|scope| {
4209            let handle_a = scope.spawn(|| crate::mount_native(request_a, dyn_source_a));
4210            let handle_b = scope.spawn(|| crate::mount_native(request_b, dyn_source_b));
4211            Ok::<_, &'static str>((
4212                handle_a.join().map_err(|_| "mount a thread panicked")?,
4213                handle_b.join().map_err(|_| "mount b thread panicked")?,
4214            ))
4215        })?;
4216
4217        let mut mount_a = result_a?;
4218        let mut mount_b = result_b?;
4219        std::fs::write(temporary_a.path().join("a.txt"), b"concurrent-a")?;
4220        std::fs::write(temporary_b.path().join("b.txt"), b"concurrent-b")?;
4221        assert_eq!(
4222            std::fs::read(temporary_a.path().join("a.txt"))?,
4223            b"concurrent-a"
4224        );
4225        assert_eq!(
4226            std::fs::read(temporary_b.path().join("b.txt"))?,
4227            b"concurrent-b"
4228        );
4229        assert!(!temporary_a.path().join("b.txt").exists());
4230        assert!(!temporary_b.path().join("a.txt").exists());
4231        assert!(mount_a.stop()?);
4232        assert!(mount_b.stop()?);
4233        Ok(())
4234    }
4235
4236    #[cfg(any(target_os = "linux", target_os = "macos"))]
4237    fn run_unix_mount_cross_process_worker() -> Result<(), Box<dyn std::error::Error>> {
4238        let destination =
4239            std::env::var_os("ACYCLIC_FS_UNIX_CHILD_MOUNT").ok_or("child mount path is absent")?;
4240        let ready =
4241            std::env::var_os("ACYCLIC_FS_UNIX_CHILD_READY").ok_or("child ready path is absent")?;
4242        let release = std::env::var_os("ACYCLIC_FS_UNIX_CHILD_RELEASE")
4243            .ok_or("child release path is absent")?;
4244        let identity = std::env::var("ACYCLIC_FS_UNIX_CHILD_ID")?;
4245        let source = Arc::new(source(FilesystemProfile::Posix)?);
4246        let mut mount = crate::mount_native(
4247            crate::NativeMountRequest {
4248                mount_id: crate::MountId::new(),
4249                volume_id: source.volume_id()?,
4250                destination: destination.clone().into(),
4251                writable: true,
4252            },
4253            source as Arc<dyn MountFilesystem>,
4254        )?;
4255        let filename = format!("{identity}.txt");
4256        std::fs::write(
4257            std::path::Path::new(&destination).join(filename),
4258            identity.as_bytes(),
4259        )?;
4260        std::fs::write(&ready, identity.as_bytes())?;
4261        wait_for_path(
4262            std::path::Path::new(&release),
4263            std::time::Duration::from_secs(10),
4264        )?;
4265        assert!(mount.stop()?);
4266        Ok(())
4267    }
4268
4269    #[cfg(any(target_os = "linux", target_os = "macos"))]
4270    #[test]
4271    #[ignore = "starts two real mount-owning processes; requires the host's native mount capability"]
4272    fn unix_mounts_in_distinct_processes_remain_independent()
4273    -> Result<(), Box<dyn std::error::Error>> {
4274        if std::env::var_os("ACYCLIC_FS_UNIX_CHILD_MOUNT").is_some() {
4275            return run_unix_mount_cross_process_worker();
4276        }
4277        let temporary = tempfile::tempdir()?;
4278        let mount_a = temporary.path().join("mount-a");
4279        let mount_b = temporary.path().join("mount-b");
4280        std::fs::create_dir(&mount_a)?;
4281        std::fs::create_dir(&mount_b)?;
4282        let ready_a = temporary.path().join("ready-a");
4283        let ready_b = temporary.path().join("ready-b");
4284        let release = temporary.path().join("release");
4285        let executable = std::env::current_exe()?;
4286        let worker =
4287            "native_mount::adapter::tests::unix_mounts_in_distinct_processes_remain_independent";
4288        let mut child_a = std::process::Command::new(&executable)
4289            .args(["--ignored", "--exact", worker, "--test-threads=1"])
4290            .env("ACYCLIC_FS_UNIX_CHILD_MOUNT", &mount_a)
4291            .env("ACYCLIC_FS_UNIX_CHILD_READY", &ready_a)
4292            .env("ACYCLIC_FS_UNIX_CHILD_RELEASE", &release)
4293            .env("ACYCLIC_FS_UNIX_CHILD_ID", "a")
4294            .spawn()?;
4295        let mut child_b = std::process::Command::new(&executable)
4296            .args(["--ignored", "--exact", worker, "--test-threads=1"])
4297            .env("ACYCLIC_FS_UNIX_CHILD_MOUNT", &mount_b)
4298            .env("ACYCLIC_FS_UNIX_CHILD_READY", &ready_b)
4299            .env("ACYCLIC_FS_UNIX_CHILD_RELEASE", &release)
4300            .env("ACYCLIC_FS_UNIX_CHILD_ID", "b")
4301            .spawn()?;
4302
4303        let validation = (|| -> Result<(), Box<dyn std::error::Error>> {
4304            wait_for_path(&ready_a, std::time::Duration::from_secs(10))?;
4305            wait_for_path(&ready_b, std::time::Duration::from_secs(10))?;
4306            assert_eq!(std::fs::read(mount_a.join("a.txt"))?, b"a");
4307            assert_eq!(std::fs::read(mount_b.join("b.txt"))?, b"b");
4308            assert!(!mount_a.join("b.txt").exists());
4309            assert!(!mount_b.join("a.txt").exists());
4310            Ok(())
4311        })();
4312        std::fs::write(&release, b"release")?;
4313        let status_a = wait_for_child(&mut child_a, std::time::Duration::from_secs(10))?;
4314        let status_b = wait_for_child(&mut child_b, std::time::Duration::from_secs(10))?;
4315        validation?;
4316        assert!(status_a.success(), "first mount process failed: {status_a}");
4317        assert!(
4318            status_b.success(),
4319            "second mount process failed: {status_b}"
4320        );
4321        Ok(())
4322    }
4323
4324    #[cfg(any(target_os = "linux", target_os = "macos"))]
4325    fn wait_for_path(
4326        path: &std::path::Path,
4327        timeout: std::time::Duration,
4328    ) -> Result<(), Box<dyn std::error::Error>> {
4329        let deadline = std::time::Instant::now() + timeout;
4330        while !path.exists() {
4331            if std::time::Instant::now() >= deadline {
4332                return Err(format!("timed out waiting for {}", path.display()).into());
4333            }
4334            std::thread::sleep(std::time::Duration::from_millis(10));
4335        }
4336        Ok(())
4337    }
4338
4339    #[cfg(any(target_os = "linux", target_os = "macos"))]
4340    fn wait_for_child(
4341        child: &mut std::process::Child,
4342        timeout: std::time::Duration,
4343    ) -> Result<std::process::ExitStatus, Box<dyn std::error::Error>> {
4344        let deadline = std::time::Instant::now() + timeout;
4345        loop {
4346            if let Some(status) = child.try_wait()? {
4347                return Ok(status);
4348            }
4349            if std::time::Instant::now() >= deadline {
4350                child.kill()?;
4351                let _ = child.wait();
4352                return Err("mount child did not terminate within the bounded deadline".into());
4353            }
4354            std::thread::sleep(std::time::Duration::from_millis(10));
4355        }
4356    }
4357
4358    #[cfg(any(target_os = "linux", target_os = "macos"))]
4359    #[test]
4360    fn posix_mount_path_preserves_non_utf8_components() -> Result<(), Box<dyn std::error::Error>> {
4361        let source = source(FilesystemProfile::Posix)?;
4362        let (initial, pending) = checkout_state(&source)?;
4363        assert!(!pending);
4364        let name = vec![b'r', 0xff, b'w'];
4365        let path = MountPath::root().child(name.clone());
4366        let created = source.create_file(&path, metadata())?;
4367        assert_eq!(source.lookup(&path)?, Some(created));
4368        source.write_attribute(
4369            &path,
4370            b"user.acyclic",
4371            Bytes::from_static(b"native-xattr"),
4372            MountAttributeWriteMode::Create,
4373        )?;
4374        assert_eq!(
4375            source.read_attribute(&path, b"user.acyclic")?.as_deref(),
4376            Some(b"native-xattr".as_slice())
4377        );
4378        let attributes = source.list_attributes(&path, None, 8)?;
4379        assert_eq!(attributes.names, vec![b"user.acyclic".to_vec()]);
4380        assert_eq!(attributes.next_cursor, None);
4381        source.remove_attribute(&path, b"user.acyclic")?;
4382        assert_eq!(source.read_attribute(&path, b"user.acyclic")?, None);
4383        let page = source.read_directory(&MountPath::root(), None, 8)?;
4384        assert_eq!(page.entries.len(), 1);
4385        assert_eq!(page.entries[0].name, name);
4386        assert!(checkout_state(&source)?.1);
4387        source.flush()?;
4388        source.flush()?;
4389        let (published, pending) = checkout_state(&source)?;
4390        assert_ne!(published, initial);
4391        assert!(!pending);
4392        Ok(())
4393    }
4394
4395    // The FUSE release callback flushes (seals) the checkout before the
4396    // kernel's next namespace mutation arrives, so mutations must stay
4397    // valid across a publication boundary.
4398    #[test]
4399    fn rename_after_flush_preserves_namespace() -> Result<(), Box<dyn std::error::Error>> {
4400        let profile = if cfg!(target_os = "windows") {
4401            FilesystemProfile::Windows
4402        } else {
4403            FilesystemProfile::Posix
4404        };
4405        let source = source(profile)?;
4406        let path = native_test_path("payload.bin");
4407        source.create_file(&path, metadata())?;
4408        let open = source.open_file(&path)?;
4409        open.write_range(0, Bytes::from_static(b"BRAVO"))?;
4410        drop(open);
4411        source.flush()?;
4412        source.flush()?;
4413        let renamed = native_test_path("renamed.bin");
4414        source.rename(&path, &renamed, true)?;
4415        assert!(source.lookup(&path)?.is_none());
4416        let landed = source.lookup(&renamed)?;
4417        assert!(landed.is_some());
4418        Ok(())
4419    }
4420
4421    #[test]
4422    fn preexisting_entry_survives_published_link_unlink_and_recreate_cycles()
4423    -> Result<(), Box<dyn std::error::Error>> {
4424        let profile = if cfg!(target_os = "windows") {
4425            FilesystemProfile::Windows
4426        } else {
4427            FilesystemProfile::Posix
4428        };
4429        let source = source(profile)?;
4430        let seed = native_test_path("shared-seed.bin");
4431        source.create_file(&seed, metadata())?;
4432        source.write_range(&seed, 0, Bytes::from_static(b"seed"))?;
4433        source.flush()?;
4434
4435        let payload = native_test_path("payload.bin");
4436        source.create_file(&payload, metadata())?;
4437        source.write_range(&payload, 0, Bytes::from_static(b"BRAVO"))?;
4438        source.flush()?;
4439        let renamed = native_test_path("renamed.bin");
4440        source.rename(&payload, &renamed, true)?;
4441        let link = native_test_path("link.bin");
4442        source.hard_link(&renamed, &link)?;
4443        let renamed_id = source
4444            .lookup(&renamed)?
4445            .ok_or("renamed file is absent")?
4446            .node
4447            .file_id;
4448        source.remove(&renamed, Some(renamed_id))?;
4449        let link_id = source
4450            .lookup(&link)?
4451            .ok_or("linked file is absent")?
4452            .node
4453            .file_id;
4454        source.remove(&link, Some(link_id))?;
4455        source.create_file(&link, metadata())?;
4456        source.write_range(&link, 0, Bytes::from_static(b"replacement"))?;
4457        source.flush()?;
4458
4459        assert_eq!(source.read_range(&seed, 0, 4)?.as_ref(), b"seed");
4460        let page = source.read_directory(&MountPath::root(), None, 16)?;
4461        let mut names = page
4462            .entries
4463            .into_iter()
4464            .map(|entry| entry.name)
4465            .collect::<Vec<_>>();
4466        names.sort();
4467        assert_eq!(
4468            names,
4469            vec![
4470                native_test_path("link.bin").components()[0].clone(),
4471                native_test_path("shared-seed.bin").components()[0].clone(),
4472            ]
4473        );
4474        assert_eq!(page.next_cursor, None);
4475        Ok(())
4476    }
4477
4478    // Host-side capture of raw-byte names requires a host filesystem that
4479    // admits them; APFS enforces valid UTF-8 and rejects such names with
4480    // EILSEQ before this crate is involved, so the host half runs on Linux
4481    // only.
4482    #[cfg(target_os = "linux")]
4483    #[test]
4484    fn host_capture_preserves_non_utf8_names() -> Result<(), Box<dyn std::error::Error>> {
4485        let source = source(FilesystemProfile::Posix)?;
4486        let host = tempfile::tempdir()?;
4487        let captured_name = vec![b'c', 0xfe, b'p'];
4488        let host_path = host
4489            .path()
4490            .join(std::ffi::OsString::from_vec(captured_name.clone()));
4491        std::fs::write(&host_path, b"captured")?;
4492        let captured_path = MountPath::root().child(captured_name);
4493        source.capture_host_path(host.path(), &captured_path)?;
4494        assert_eq!(
4495            source.read_range(&captured_path, 0, 8)?.as_ref(),
4496            b"captured"
4497        );
4498        std::fs::remove_file(host_path)?;
4499        source.capture_host_path(host.path(), &captured_path)?;
4500        assert_eq!(source.lookup(&captured_path)?, None);
4501        Ok(())
4502    }
4503}