Skip to main content

acyclic_fs/native_mount/
customer.rs

1//! Customer-facing native mount composition over the canonical checkout.
2
3use super::adapter::HostCaptureScope;
4use super::{
5    CheckoutMountSource, LazyMountSource, MountPath, NativeMountError, NativeMountRequest,
6    NativeMountSession, SharedCheckout, capture_root_identity, mount_native,
7};
8use crate::demand::{DemandSource, SourceNode, SourceNodeKind};
9use crate::kernel::FileKind;
10use crate::kernel::NamespacePath;
11use crate::model::{CheckoutMode, GenerationSelector};
12use crate::native_capture::NativeViewBaseline;
13use crate::workspace::{Workspace, WorkspaceError, customer_path};
14use crate::{
15    AsyncAuthorityStore, AsyncObjectStore, GenerationId, IdempotencyKey, LazyLookup, LazyWorkspace,
16    LazyWorkspaceError, LazyWorkspaceStore, MountId, MountSourceError, NativeRootIdentity,
17    VolumeId,
18};
19use std::path::{Path, PathBuf};
20use std::sync::{Arc, Mutex};
21use thiserror::Error;
22
23/// When authored native mutations become a durable workspace generation.
24#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
25pub enum MountPublication {
26    /// Publish on native close/fsync/rename boundaries and explicit sync.
27    #[default]
28    CloseAndSync,
29    /// Publish every independently admitted native mutation.
30    PerMutation,
31    /// Publish only through [`Mount::sync`] or orderly [`Mount::unmount`].
32    Manual,
33}
34
35/// Explicit customer mount configuration.
36#[derive(Clone, Debug, Eq, PartialEq)]
37pub struct MountOptions {
38    writable: bool,
39    subdirectory: String,
40    publication: MountPublication,
41}
42
43impl MountOptions {
44    /// Creates a writable tracking-safe sparse mount with close/fsync publication.
45    #[must_use]
46    pub fn read_write() -> Self {
47        Self {
48            writable: true,
49            subdirectory: "/".to_owned(),
50            publication: MountPublication::CloseAndSync,
51        }
52    }
53
54    /// Creates a pinned immutable mount.
55    #[must_use]
56    pub fn read_only() -> Self {
57        Self {
58            writable: false,
59            subdirectory: "/".to_owned(),
60            publication: MountPublication::Manual,
61        }
62    }
63
64    /// Projects one exact workspace directory as the native mount root.
65    #[must_use]
66    pub fn subdirectory(mut self, path: impl Into<String>) -> Self {
67        self.subdirectory = path.into();
68        self
69    }
70
71    /// Selects the exact publication policy for writable authored effects.
72    #[must_use]
73    pub const fn publication(mut self, publication: MountPublication) -> Self {
74        self.publication = publication;
75        self
76    }
77}
78
79/// One process-owned customer mount over a canonical workspace checkout.
80pub struct Mount<A, O> {
81    source: Arc<CheckoutMountSource<A, O>>,
82    session: Mutex<Option<NativeMountSession>>,
83    destination: PathBuf,
84}
85
86/// One process-owned native mount over a demand-backed sparse workspace.
87pub struct LazyMount<A, O, D, S> {
88    source: Arc<LazyMountSource<A, O, D, S>>,
89    session: Mutex<Option<NativeMountSession>>,
90    destination: PathBuf,
91}
92
93/// SDK-owned capture and publication boundary for a pinned native directory.
94///
95/// The caller owns the physical volume-root directory and supplies exact
96/// changed paths. A selected subdirectory retains its namespace path below
97/// that physical root; its contents are not remapped to the root itself.
98/// Filesystem semantics and fenced publication use an exact-generation
99/// checkout. This handle never rebinds to the live lazy source or starts a
100/// mount driver.
101pub struct LazyWorkingSet<A, O> {
102    source: Arc<CheckoutMountSource<A, O>>,
103    workspace: Workspace<A, O>,
104    source_root: PathBuf,
105    source_identity: NativeRootIdentity,
106    selected_root: PathBuf,
107    expected_generation: GenerationId,
108    baseline: Arc<NativeViewBaseline>,
109    capture_budget: crate::WorkBudget,
110}
111
112struct AbortTaskOnDrop<T>(tokio::task::JoinHandle<T>);
113
114impl<T> Drop for AbortTaskOnDrop<T> {
115    fn drop(&mut self) {
116        self.0.abort();
117    }
118}
119
120impl<A, O> LazyWorkingSet<A, O>
121where
122    A: AsyncAuthorityStore + Send + Sync + 'static,
123    O: AsyncObjectStore + Send + Sync + 'static,
124{
125    /// Revalidates a prepared native view immediately before a quiescent
126    /// presentation switch. This is optimistic admission, not a lock on the
127    /// workspace head: publication remains fenced by the operation permit.
128    /// It reads only the selected directory metadata and the current head.
129    pub async fn validate_for_presentation(&self) -> Result<(), MountLifecycleError> {
130        let expected = self.expected_generation;
131        if self.workspace.head().await?.id() != expected {
132            return Err(MountLifecycleError::Workspace(
133                WorkspaceError::StaleGeneration,
134            ));
135        }
136        let source_root = self.source_root.clone();
137        let selected_root = self.selected_root.clone();
138        let identity = self.source_identity;
139        tokio::task::spawn_blocking(move || {
140            validate_native_working_set_root(&source_root, &selected_root, identity)
141        })
142        .await
143        .map_err(|error| MountSourceError::Engine(error.to_string()))?
144        .map_err(MountLifecycleError::Source)?;
145        if self.workspace.head().await?.id() != expected {
146            return Err(MountLifecycleError::Workspace(
147                WorkspaceError::StaleGeneration,
148            ));
149        }
150        Ok(())
151    }
152
153    /// Captures only the supplied changed paths from a native working directory.
154    ///
155    /// The source root is authenticated by the core capture path. This is a
156    /// path hint, not proof that other native writes did not occur. Call this
157    /// at an operation boundary, then publish with [`Self::sync_with_permit`].
158    pub async fn capture_host_paths(&self, paths: &[MountPath]) -> Result<(), MountLifecycleError> {
159        self.source
160            .capture_host_with_identity_budgeted(
161                &self.source_root,
162                paths,
163                self.source_identity,
164                HostCaptureScope::ExactPaths,
165                Arc::clone(&self.baseline),
166                self.capture_budget,
167            )
168            .await
169            .map_err(MountLifecycleError::Source)
170    }
171
172    /// Reconciles a complete selected subtree after a watcher overflow or
173    /// missed event. Both host and checkout descendants participate, so
174    /// creations and deletions are captured without scanning sibling trees.
175    pub async fn capture_host_subtree(&self, root: &MountPath) -> Result<(), MountLifecycleError> {
176        self.source
177            .capture_host_with_identity_budgeted(
178                &self.source_root,
179                std::slice::from_ref(root),
180                self.source_identity,
181                HostCaptureScope::Subtree,
182                Arc::clone(&self.baseline),
183                self.capture_budget,
184            )
185            .await
186            .map_err(MountLifecycleError::Source)
187    }
188
189    /// Reconciles the selected subtree and publishes under an active, fenced
190    /// operation permit. Path hints alone cannot establish completeness: a
191    /// missing or delayed notification must not silently omit a native write.
192    pub async fn sync_with_permit(
193        &self,
194        permit: crate::PublicationPermit,
195    ) -> Result<(), MountLifecycleError> {
196        self.capture_host_subtree(&MountPath::root()).await?;
197        self.source
198            .sync_async_with_permit(permit)
199            .await
200            .map_err(MountLifecycleError::Source)
201    }
202}
203
204fn flush_session_callbacks(
205    session: &Mutex<Option<NativeMountSession>>,
206) -> Result<(), MountLifecycleError> {
207    let owner = match session.lock() {
208        Ok(owner) => owner,
209        Err(poisoned) => poisoned.into_inner(),
210    };
211    if let Some(session) = owner.as_ref() {
212        session
213            .flush_callbacks()
214            .map_err(MountLifecycleError::Native)?;
215    }
216    Ok(())
217}
218
219impl<A, O, D, S> LazyMount<A, O, D, S> {
220    /// Exact mounted host path.
221    #[must_use]
222    pub fn path(&self) -> &Path {
223        &self.destination
224    }
225}
226
227impl<A, O> Mount<A, O> {
228    /// Exact mounted host path.
229    #[must_use]
230    pub fn path(&self) -> &Path {
231        &self.destination
232    }
233}
234
235impl<A, O> Mount<A, O>
236where
237    A: AsyncAuthorityStore + Send + Sync + 'static,
238    O: AsyncObjectStore + Send + Sync + 'static,
239{
240    /// Publishes all pending authored effects with one fenced generation.
241    ///
242    /// # Errors
243    ///
244    /// Returns a typed publication failure without discarding pending state.
245    pub async fn sync(&self) -> Result<(), MountLifecycleError> {
246        flush_session_callbacks(&self.session)?;
247        self.source
248            .sync_async()
249            .await
250            .map_err(MountLifecycleError::Source)?;
251        Ok(())
252    }
253
254    /// Publishes all pending effects under one active operation lease.
255    pub async fn sync_with_permit(
256        &self,
257        permit: crate::PublicationPermit,
258    ) -> Result<(), MountLifecycleError> {
259        flush_session_callbacks(&self.session)?;
260        self.source
261            .sync_async_with_permit(permit)
262            .await
263            .map_err(MountLifecycleError::Source)
264    }
265
266    /// Publishes pending writes and advances the live checkout to workspace head.
267    ///
268    /// This keeps a direct-parent agent mount coherent after a child join
269    /// without exposing remount orchestration to adapters.
270    pub async fn refresh(&self) -> Result<(), MountLifecycleError> {
271        self.sync().await?;
272        self.advance_to_head().await
273    }
274
275    /// Advances an already-clean live checkout to the workspace head.
276    ///
277    /// Use this after an external, fenced workspace publication when the
278    /// caller already synchronized the mount before that publication. Unlike
279    /// [`Self::refresh`], this does not try to republish against the newer head.
280    ///
281    /// # Errors
282    ///
283    /// Returns a typed conflict or storage failure without changing the
284    /// mounted generation.
285    pub async fn advance_to_head(&self) -> Result<(), MountLifecycleError> {
286        self.source
287            .advance_to_head_async()
288            .await
289            .map_err(MountLifecycleError::Source)
290    }
291
292    /// Publishes all pending effects on the source's dedicated callback runtime.
293    ///
294    /// # Errors
295    ///
296    /// Returns a typed publication failure without discarding pending state.
297    pub fn sync_blocking(&self) -> Result<(), MountLifecycleError> {
298        flush_session_callbacks(&self.session)?;
299        self.source.sync().map_err(MountLifecycleError::Source)
300    }
301
302    /// Publishes pending effects and detaches the native namespace exactly once.
303    ///
304    /// # Errors
305    ///
306    /// Publication failure leaves the live mount owned by this handle. A detach
307    /// failure retains driver state and its destination fence for drop retry.
308    pub async fn unmount(&self) -> Result<(), MountLifecycleError> {
309        self.sync().await?;
310        self.detach()
311    }
312
313    /// Detaches and cancels this mount without publishing pending callbacks.
314    ///
315    /// This is the fail-closed recovery path after a fenced publication. The
316    /// caller may remount from the last durable generation.
317    pub fn abandon(&self) -> Result<(), MountLifecycleError> {
318        self.detach()
319    }
320
321    /// Synchronously publishes and detaches for foreign-runtime worker threads.
322    ///
323    /// # Errors
324    ///
325    /// Publication or detach failure retains the complete owner for retry.
326    pub fn unmount_blocking(&self) -> Result<(), MountLifecycleError> {
327        self.sync_blocking()?;
328        self.detach()
329    }
330
331    fn detach(&self) -> Result<(), MountLifecycleError> {
332        let mut owner = match self.session.lock() {
333            Ok(owner) => owner,
334            Err(poisoned) => poisoned.into_inner(),
335        };
336        let Some(session) = owner.as_mut() else {
337            return Ok(());
338        };
339        session.stop().map_err(MountLifecycleError::Native)?;
340        owner.take();
341        self.source.cancel();
342        Ok(())
343    }
344}
345
346impl<A, O, D, S> LazyMount<A, O, D, S>
347where
348    A: AsyncAuthorityStore + Send + Sync + 'static,
349    O: AsyncObjectStore + Send + Sync + 'static,
350    D: DemandSource + 'static,
351    S: LazyWorkspaceStore,
352{
353    /// Publishes all pending authored effects with one fenced generation.
354    pub async fn sync(&self) -> Result<(), MountLifecycleError> {
355        flush_session_callbacks(&self.session)?;
356        self.source
357            .sync_async()
358            .await
359            .map_err(MountLifecycleError::Source)
360    }
361
362    /// Publishes all pending effects under one active operation lease.
363    pub async fn sync_with_permit(
364        &self,
365        permit: crate::PublicationPermit,
366    ) -> Result<(), MountLifecycleError> {
367        flush_session_callbacks(&self.session)?;
368        self.source
369            .sync_async_with_permit(permit)
370            .await
371            .map_err(MountLifecycleError::Source)
372    }
373
374    /// Publishes pending writes and advances the live checkout to workspace head.
375    pub async fn refresh(&self) -> Result<(), MountLifecycleError> {
376        self.sync().await?;
377        self.advance_to_head().await
378    }
379
380    /// Advances an already-clean checkout to workspace head.
381    pub async fn advance_to_head(&self) -> Result<(), MountLifecycleError> {
382        self.source
383            .advance_to_head_async()
384            .await
385            .map_err(MountLifecycleError::Source)
386    }
387
388    /// Publishes pending effects on the source callback runtime.
389    pub fn sync_blocking(&self) -> Result<(), MountLifecycleError> {
390        flush_session_callbacks(&self.session)?;
391        self.source.sync().map_err(MountLifecycleError::Source)
392    }
393
394    /// Publishes pending effects and detaches the native namespace exactly once.
395    pub async fn unmount(&self) -> Result<(), MountLifecycleError> {
396        self.sync().await?;
397        self.detach()
398    }
399
400    /// Detaches and cancels this lazy mount without publishing pending state.
401    pub fn abandon(&self) -> Result<(), MountLifecycleError> {
402        self.detach()
403    }
404
405    /// Synchronously publishes and detaches for foreign-runtime worker threads.
406    pub fn unmount_blocking(&self) -> Result<(), MountLifecycleError> {
407        self.sync_blocking()?;
408        self.detach()
409    }
410
411    fn detach(&self) -> Result<(), MountLifecycleError> {
412        let mut owner = match self.session.lock() {
413            Ok(owner) => owner,
414            Err(poisoned) => poisoned.into_inner(),
415        };
416        let Some(session) = owner.as_mut() else {
417            return Ok(());
418        };
419        session.stop().map_err(MountLifecycleError::Native)?;
420        owner.take();
421        self.source.cancel();
422        Ok(())
423    }
424}
425
426/// Customer mount admission, publication, and lifecycle failures.
427#[derive(Debug, Error)]
428pub enum MountLifecycleError {
429    /// Workspace path, state, or storage failed admission.
430    #[error(transparent)]
431    Workspace(#[from] WorkspaceError),
432    /// Lazy source or sparse-overlay admission failed.
433    #[error(transparent)]
434    Lazy(#[from] LazyWorkspaceError),
435    /// Native driver admission or teardown failed.
436    #[error(transparent)]
437    Native(#[from] NativeMountError),
438    /// Native callback publication failed.
439    #[error(transparent)]
440    Source(#[from] MountSourceError),
441    /// Combined exactification and materialization exceeded its work budget.
442    #[error(transparent)]
443    Work(#[from] crate::WorkError),
444}
445
446impl<A, O, D, S> LazyWorkspace<A, O, D, S>
447where
448    A: AsyncAuthorityStore + Send + Sync + 'static,
449    O: AsyncObjectStore + Send + Sync + 'static,
450    D: DemandSource + 'static,
451    S: LazyWorkspaceStore,
452{
453    /// Prepares an authenticated native working set from one exact generation.
454    ///
455    /// The destination must be an existing empty real directory. The SDK first
456    /// exactifies only the selected subtree, materializes that exact generation,
457    /// then admits the resulting directory. Preparation never starts a mount
458    /// driver and fails if the workspace head advances before admission. The
459    /// supplied work budget also caps each later capture or sync on the handle;
460    /// those operations do not inherit the kernel-callback deadline or budget.
461    #[allow(clippy::too_many_lines)]
462    pub async fn prepare_native_working_set(
463        &self,
464        subdirectory: &str,
465        options: &crate::MaterializeOptions,
466        budget: crate::WorkBudget,
467        cancellation: &crate::CancellationToken,
468        permit: crate::PublicationPermit,
469    ) -> Result<LazyWorkingSet<A, O>, MountLifecycleError> {
470        if !options.destination.is_absolute() {
471            return Err(MountLifecycleError::Source(MountSourceError::Invalid(
472                "native working-set root must be absolute".to_owned(),
473            )));
474        }
475        // Keep the cold exactification frame independent of materialization:
476        // both are large state machines, and nesting them exhausts the small
477        // default stacks used by some host test/runtime threads.
478        let lazy = self.clone();
479        let selected = subdirectory.to_owned();
480        let token = cancellation.clone();
481        let mut task = AbortTaskOnDrop(tokio::spawn(async move {
482            lazy.exactify_subtree_with_permit(&selected, budget, &token, permit)
483                .await
484        }));
485        let exact = (&mut task.0)
486            .await
487            .map_err(|error| MountSourceError::Engine(error.to_string()))??;
488        let remaining = exact.work.remaining(budget)?;
489        let materialized = if subdirectory == "/" {
490            exact
491                .value
492                .materialize_with_mode(
493                    options,
494                    remaining,
495                    cancellation,
496                    super::MaterializeMode::ReconstructibleView,
497                )
498                .await?
499        } else {
500            exact
501                .value
502                .materialize_path_with_mode(
503                    subdirectory,
504                    options,
505                    remaining,
506                    cancellation,
507                    super::MaterializeMode::ReconstructibleView,
508                )
509                .await?
510        };
511        // Admission stays inside this operation: an arbitrary directory must
512        // never be rebound to a caller-claimed generation without having been
513        // materialized from that exact generation first.
514        let expected_generation = exact.value.id();
515        let source_root = options.destination.clone();
516        let identity_root = source_root.clone();
517        let source_identity =
518            tokio::task::spawn_blocking(move || capture_root_identity(&identity_root))
519                .await
520                .map_err(|error| MountSourceError::Engine(error.to_string()))?
521                .map_err(|error| MountSourceError::Engine(error.to_string()))?;
522        let options = MountOptions::read_write()
523            .subdirectory(subdirectory)
524            .publication(MountPublication::Manual);
525        let (source, _, root, _) = self
526            .prepare_authored_mount_source(&options, Some(expected_generation))
527            .await?;
528        let relative = crate::namespace_to_host_path(&root)
529            .map_err(|error| MountSourceError::Engine(error.to_string()))?;
530        let validated_root = source_root.clone();
531        let selected_root = relative.clone();
532        tokio::task::spawn_blocking(move || {
533            validate_native_working_set_root(&validated_root, &selected_root, source_identity)
534        })
535        .await
536        .map_err(|error| MountSourceError::Engine(error.to_string()))?
537        .map_err(MountLifecycleError::Source)?;
538        #[cfg(unix)]
539        let baseline = NativeViewBaseline::new(source_identity);
540        #[cfg(windows)]
541        let baseline = NativeViewBaseline;
542        let _ = materialized;
543        let final_root = source_root.clone();
544        let final_selected = relative.clone();
545        tokio::task::spawn_blocking(move || {
546            validate_native_working_set_root(&final_root, &final_selected, source_identity)
547        })
548        .await
549        .map_err(|error| MountSourceError::Engine(error.to_string()))?
550        .map_err(MountLifecycleError::Source)?;
551        // Directory validation may wait for host I/O while another writer
552        // advances the workspace. Do not return an already-stale preparation.
553        if self.workspace().head().await?.id() != expected_generation {
554            return Err(MountLifecycleError::Workspace(
555                WorkspaceError::StaleGeneration,
556            ));
557        }
558        Ok(LazyWorkingSet {
559            source,
560            workspace: self.workspace().clone(),
561            source_root,
562            source_identity,
563            selected_root: relative,
564            expected_generation,
565            baseline: Arc::new(baseline),
566            capture_budget: budget,
567        })
568    }
569
570    async fn prepare_authored_mount_source(
571        &self,
572        options: &MountOptions,
573        expected_generation: Option<GenerationId>,
574    ) -> Result<
575        (
576            Arc<CheckoutMountSource<A, O>>,
577            VolumeId,
578            NamespacePath,
579            String,
580        ),
581        MountLifecycleError,
582    > {
583        if let Some(expected) = expected_generation
584            && self.workspace().head().await?.id() != expected
585        {
586            return Err(MountLifecycleError::Workspace(
587                WorkspaceError::StaleGeneration,
588            ));
589        }
590        let probe = self
591            .workspace()
592            .engine_checkout(GenerationSelector::Head, CheckoutMode::read_only_pinned())
593            .await?;
594        let config = probe.volume_config();
595        drop(probe);
596        let root = customer_path(&options.subdirectory, config)?;
597        let root_text = portable_namespace_path(&root)?;
598        let selected = self.lookup(&root_text).await?;
599        if !matches!(
600            selected,
601            LazyLookup::Authored {
602                stat: crate::WorkspaceStat {
603                    kind: FileKind::Directory,
604                    ..
605                },
606                ..
607            } | LazyLookup::Source(SourceNode {
608                kind: SourceNodeKind::Directory,
609                ..
610            }) | LazyLookup::Shadow {
611                record: crate::kernel::FileRecord {
612                    kind: FileKind::Directory,
613                    ..
614                },
615                ..
616            }
617        ) {
618            return Err(MountLifecycleError::Workspace(WorkspaceError::NotDirectory));
619        }
620        if let LazyLookup::Source(node) = selected {
621            if expected_generation.is_some() {
622                return Err(MountLifecycleError::Workspace(
623                    WorkspaceError::StaleGeneration,
624                ));
625            }
626            let expected_source = self.source_file_id(&node);
627            let mut hasher = blake3::Hasher::new();
628            hasher.update(b"acyclic-fs-lazy-mount-root-v1\0");
629            hasher.update(&self.workspace().id().into_bytes());
630            hasher.update(root_text.as_bytes());
631            hasher.update(&node.version.0);
632            let mut key = [0_u8; 16];
633            key.copy_from_slice(&hasher.finalize().as_bytes()[..16]);
634            self.promote_exact(
635                &root_text,
636                expected_source,
637                0,
638                IdempotencyKey::from_bytes(key),
639            )
640            .await?;
641        }
642
643        let mode = if options.writable {
644            CheckoutMode::tracking_transaction()
645        } else {
646            CheckoutMode::read_only_pinned()
647        };
648        let checkout = self
649            .workspace()
650            .engine_checkout(
651                expected_generation.map_or(GenerationSelector::Head, GenerationSelector::Exact),
652                mode,
653            )
654            .await?;
655        let shared = Arc::new(SharedCheckout::with_publication(
656            checkout,
657            options.publication,
658        ));
659        let authored = Arc::new(CheckoutMountSource::new_at(shared, config, root.clone())?);
660        if let Some(expected) = expected_generation
661            && self.workspace().head().await?.id() != expected
662        {
663            return Err(MountLifecycleError::Workspace(
664                WorkspaceError::StaleGeneration,
665            ));
666        }
667        Ok((authored, self.workspace().id().volume_id(), root, root_text))
668    }
669
670    /// Mounts a demand-backed workspace without scanning its source tree.
671    ///
672    /// Admission observes only the selected root node. Unresolved descendants
673    /// remain source-backed until a native callback addresses them.
674    pub async fn mount(
675        &self,
676        destination: impl Into<PathBuf>,
677        options: MountOptions,
678    ) -> Result<LazyMount<A, O, D, S>, MountLifecycleError> {
679        if options.writable && options.publication == MountPublication::PerMutation {
680            return Err(MountLifecycleError::Source(MountSourceError::Unsupported(
681                "lazy source whiteouts publish with a checkout boundary, not per callback"
682                    .to_owned(),
683            )));
684        }
685        let (authored, volume_id, _, root_text) =
686            self.prepare_authored_mount_source(&options, None).await?;
687        let source = Arc::new(LazyMountSource::new(
688            Arc::new(self.clone()),
689            authored,
690            root_text,
691        )?);
692        let destination = destination.into();
693        let session = mount_native(
694            NativeMountRequest {
695                mount_id: MountId::new(),
696                volume_id,
697                destination: destination.clone(),
698                writable: options.writable,
699            },
700            Arc::clone(&source) as Arc<dyn super::MountFilesystem>,
701        )?;
702        Ok(LazyMount {
703            source,
704            session: Mutex::new(Some(session)),
705            destination,
706        })
707    }
708}
709
710fn validate_native_working_set_root(
711    source_root: &Path,
712    selected_root: &Path,
713    expected_identity: NativeRootIdentity,
714) -> Result<(), MountSourceError> {
715    let root = crate::native_host::HostRoot::open(source_root)
716        .map_err(|error| MountSourceError::Engine(error.to_string()))?;
717    if root.identity() != expected_identity {
718        return Err(MountSourceError::Stale);
719    }
720    let metadata = root
721        .symlink_metadata_held(selected_root)
722        .map_err(|error| MountSourceError::Engine(error.to_string()))?;
723    if !metadata.is_dir() {
724        return Err(MountSourceError::Invalid(
725            "selected native working-set directory is absent".to_owned(),
726        ));
727    }
728    Ok(())
729}
730
731fn portable_namespace_path(
732    path: &crate::kernel::NamespacePath,
733) -> Result<String, MountLifecycleError> {
734    let mut value = String::new();
735    for component in path.components() {
736        value.push('/');
737        value.push_str(std::str::from_utf8(component.as_bytes()).map_err(|_| {
738            MountLifecycleError::Source(MountSourceError::Unsupported(
739                "lazy portable mounts cannot root at a non-UTF-8 path".to_owned(),
740            ))
741        })?);
742    }
743    if value.is_empty() {
744        value.push('/');
745    }
746    Ok(value)
747}
748
749impl<A, O> Workspace<A, O>
750where
751    A: AsyncAuthorityStore + Send + Sync + 'static,
752    O: AsyncObjectStore + Send + Sync + 'static,
753{
754    /// Mounts the complete workspace or one exact subtree as a native directory.
755    ///
756    /// Admission authenticates only the selected root frontier; it never scans
757    /// the workspace or materializes file bodies.
758    ///
759    /// # Errors
760    ///
761    /// Rejects invalid or non-directory roots, unavailable host capabilities,
762    /// busy/nonempty destinations, and storage/authentication failures.
763    pub async fn mount(
764        &self,
765        destination: impl Into<PathBuf>,
766        options: MountOptions,
767    ) -> Result<Mount<A, O>, MountLifecycleError> {
768        let mode = if options.writable {
769            CheckoutMode::tracking_transaction()
770        } else {
771            CheckoutMode::read_only_pinned()
772        };
773        let mut checkout = self.engine_checkout(GenerationSelector::Head, mode).await?;
774        let config = checkout.volume_config();
775        let root = customer_path(&options.subdirectory, config)?;
776        let selected = checkout
777            .lookup_no_follow(
778                &root,
779                crate::WorkBudget::UNBOUNDED,
780                &crate::CancellationToken::new(),
781            )
782            .await
783            .map_err(WorkspaceError::engine)?
784            .value
785            .record
786            .ok_or(WorkspaceError::NotFound)?;
787        if selected.kind != FileKind::Directory {
788            return Err(MountLifecycleError::Workspace(WorkspaceError::NotDirectory));
789        }
790        let shared = Arc::new(SharedCheckout::with_publication(
791            checkout,
792            options.publication,
793        ));
794        let source = Arc::new(CheckoutMountSource::new_at(shared, config, root)?);
795        let destination = destination.into();
796        let session = mount_native(
797            NativeMountRequest {
798                mount_id: MountId::new(),
799                volume_id: self.id().volume_id(),
800                destination: destination.clone(),
801                writable: options.writable,
802            },
803            Arc::clone(&source) as Arc<dyn super::MountFilesystem>,
804        )?;
805        Ok(Mount {
806            source,
807            session: Mutex::new(Some(session)),
808            destination,
809        })
810    }
811}
812
813#[cfg(test)]
814mod tests {
815    use super::*;
816    use crate::demand::native::NativeDemandSource;
817    use crate::model::{FilesystemProfile, VolumeLimits};
818    use crate::{Fs, MemoryLazyWorkspaceStore, PublicationPermit};
819
820    #[tokio::test]
821    async fn prepared_subtree_captures_native_edit_and_rejects_stale_head()
822    -> Result<(), Box<dyn std::error::Error>> {
823        let source = tempfile::tempdir()?;
824        std::fs::create_dir(source.path().join("hot"))?;
825        std::fs::write(source.path().join("hot/file.txt"), b"base")?;
826        std::fs::create_dir(source.path().join("hot/sub"))?;
827        std::fs::write(source.path().join("hot/sub/child.txt"), b"old child")?;
828        let native_view = tempfile::tempdir()?;
829        let fs = Fs::memory();
830        let demand = Arc::new(
831            NativeDemandSource::open(
832                source.path(),
833                FilesystemProfile::Portable,
834                VolumeLimits::default(),
835            )
836            .await?,
837        );
838        let lazy = LazyWorkspace::attach(
839            &fs,
840            "prepared-native-working-set",
841            demand,
842            MemoryLazyWorkspaceStore::default(),
843        )
844        .await?;
845        let before_invalid = lazy.workspace().head().await?.id();
846        assert!(matches!(
847            lazy.prepare_native_working_set(
848                "/hot",
849                &crate::MaterializeOptions::native("relative-native-view"),
850                crate::WorkBudget::UNBOUNDED,
851                &crate::CancellationToken::new(),
852                PublicationPermit::Unrestricted,
853            )
854            .await,
855            Err(MountLifecycleError::Source(MountSourceError::Invalid(_)))
856        ));
857        assert_eq!(lazy.workspace().head().await?.id(), before_invalid);
858        let mut working_set = lazy
859            .prepare_native_working_set(
860                "/hot",
861                &crate::MaterializeOptions::native(native_view.path()),
862                crate::WorkBudget::UNBOUNDED,
863                &crate::CancellationToken::new(),
864                PublicationPermit::Unrestricted,
865            )
866            .await?;
867        let generation = working_set.expected_generation;
868        let working = native_view.path();
869        working_set.validate_for_presentation().await?;
870        working_set.capture_budget = crate::WorkBudget::default();
871        assert!(
872            working_set
873                .capture_host_subtree(&MountPath::root())
874                .await
875                .is_err()
876        );
877        assert_eq!(lazy.workspace().head().await?.id(), generation);
878        working_set.capture_budget = crate::WorkBudget::UNBOUNDED;
879        let name = if cfg!(windows) {
880            "file.txt"
881                .encode_utf16()
882                .flat_map(u16::to_le_bytes)
883                .collect()
884        } else {
885            b"file.txt".to_vec()
886        };
887        let unchanged_head = lazy.workspace().head().await?.id();
888        working_set
889            .sync_with_permit(PublicationPermit::Unrestricted)
890            .await?;
891        assert_eq!(lazy.workspace().head().await?.id(), unchanged_head);
892        #[cfg(unix)]
893        let canonical_before = lazy
894            .workspace()
895            .head()
896            .await?
897            .stat("/hot/file.txt")
898            .await?
899            .metadata;
900        #[cfg(unix)]
901        {
902            working_set
903                .capture_host_paths(&[MountPath::root().child(name.clone())])
904                .await?;
905            working_set
906                .sync_with_permit(PublicationPermit::Unrestricted)
907                .await?;
908            let canonical_after = lazy
909                .workspace()
910                .head()
911                .await?
912                .stat("/hot/file.txt")
913                .await?
914                .metadata;
915            assert_eq!(canonical_after.changed_ns, canonical_before.changed_ns);
916            #[cfg(target_os = "linux")]
917            assert_eq!(canonical_after.created_ns, canonical_before.created_ns);
918        }
919        #[cfg(target_os = "linux")]
920        {
921            let accessed = std::time::UNIX_EPOCH + std::time::Duration::from_secs(1_700_000_000);
922            std::fs::File::open(working.join("hot/file.txt"))?
923                .set_times(std::fs::FileTimes::new().set_accessed(accessed))?;
924            working_set
925                .sync_with_permit(PublicationPermit::Unrestricted)
926                .await?;
927            let canonical = lazy
928                .workspace()
929                .head()
930                .await?
931                .stat("/hot/file.txt")
932                .await?
933                .metadata;
934            assert_eq!(canonical.accessed_ns, Some(1_700_000_000_000_000_000));
935        }
936        std::fs::write(working.join("hot/file.txt"), b"native-edit")?;
937        working_set
938            .capture_host_paths(&[MountPath::root().child(name)])
939            .await?;
940        working_set
941            .sync_with_permit(PublicationPermit::Unrestricted)
942            .await?;
943        assert_eq!(
944            lazy.workspace().read("/hot/file.txt", 32).await?.as_ref(),
945            b"native-edit"
946        );
947        std::fs::write(working.join("hot/created.txt"), b"new")?;
948        working_set.capture_host_subtree(&MountPath::root()).await?;
949        working_set
950            .sync_with_permit(PublicationPermit::Unrestricted)
951            .await?;
952        assert_eq!(
953            lazy.workspace()
954                .read("/hot/created.txt", 16)
955                .await?
956                .as_ref(),
957            b"new"
958        );
959        // The publication boundary must capture writes even when the caller
960        // receives no change notification and supplies no path hint.
961        std::fs::write(working.join("hot/unreported.txt"), b"unreported")?;
962        working_set
963            .sync_with_permit(PublicationPermit::Unrestricted)
964            .await?;
965        assert_eq!(
966            lazy.workspace()
967                .read("/hot/unreported.txt", 32)
968                .await?
969                .as_ref(),
970            b"unreported"
971        );
972        std::fs::remove_file(working.join("hot/unreported.txt"))?;
973        working_set
974            .sync_with_permit(PublicationPermit::Unrestricted)
975            .await?;
976        assert!(
977            lazy.workspace()
978                .read("/hot/unreported.txt", 32)
979                .await
980                .is_err()
981        );
982        std::fs::remove_dir_all(working.join("hot/sub"))?;
983        std::fs::write(working.join("hot/sub"), b"replacement")?;
984        working_set.capture_host_subtree(&MountPath::root()).await?;
985        working_set
986            .sync_with_permit(PublicationPermit::Unrestricted)
987            .await?;
988        assert_eq!(
989            lazy.workspace().read("/hot/sub", 32).await?.as_ref(),
990            b"replacement"
991        );
992        assert!(
993            lazy.workspace()
994                .read("/hot/sub/child.txt", 32)
995                .await
996                .is_err()
997        );
998        lazy.write("/later.txt", bytes::Bytes::from_static(b"later"))
999            .await?;
1000        lazy.write("/hot/conflict.txt", bytes::Bytes::from_static(b"parent"))
1001            .await?;
1002        std::fs::write(working.join("hot/conflict.txt"), b"child")?;
1003        working_set.capture_host_subtree(&MountPath::root()).await?;
1004        assert!(matches!(
1005            working_set
1006                .sync_with_permit(PublicationPermit::Unrestricted)
1007                .await,
1008            Err(MountLifecycleError::Source(MountSourceError::Stale))
1009        ));
1010        assert_eq!(
1011            lazy.workspace()
1012                .read("/hot/conflict.txt", 16)
1013                .await?
1014                .as_ref(),
1015            b"parent"
1016        );
1017        assert!(matches!(
1018            working_set.validate_for_presentation().await,
1019            Err(MountLifecycleError::Workspace(
1020                WorkspaceError::StaleGeneration
1021            ))
1022        ));
1023        assert_ne!(lazy.workspace().head().await?.id(), generation);
1024        let latest_view = tempfile::tempdir()?;
1025        let ready = lazy
1026            .prepare_native_working_set(
1027                "/hot",
1028                &crate::MaterializeOptions::native(latest_view.path()),
1029                crate::WorkBudget::UNBOUNDED,
1030                &crate::CancellationToken::new(),
1031                PublicationPermit::Unrestricted,
1032            )
1033            .await?;
1034        ready.validate_for_presentation().await?;
1035        assert_eq!(
1036            std::fs::read(latest_view.path().join("hot/file.txt"))?,
1037            b"native-edit"
1038        );
1039        std::fs::rename(
1040            latest_view.path(),
1041            source.path().join("retired-native-view"),
1042        )?;
1043        std::fs::create_dir(latest_view.path())?;
1044        std::fs::create_dir(latest_view.path().join("hot"))?;
1045        assert!(matches!(
1046            ready.validate_for_presentation().await,
1047            Err(MountLifecycleError::Source(MountSourceError::Stale))
1048        ));
1049        Ok(())
1050    }
1051}