1use 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
47struct 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
86pub 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 #[cfg(any(target_os = "linux", target_os = "macos"))]
140 Err(_) => Ok(self.handle.block_on(create())),
141 #[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
203pub struct SharedCheckout<A, O> {
206 state: tokio::sync::Mutex<SharedCheckoutState<A, O>>,
207 view_epoch: Arc<AtomicU64>,
208 view_gate: Arc<ViewGate>,
209}
210
211pub 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
223pub 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 #[must_use]
235 pub fn new(checkout: Checkout<A, O>) -> Self {
236 Self::with_publication(checkout, MountPublication::CloseAndSync)
237 }
238
239 #[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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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, ¤t] {
3188 let source = source(profile)?;
3189 runtime.block_on(async {
3190 let value = callback.block_on(|| async {
3191 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 #[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 #[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 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 #[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 #[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 #[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}