aster_forge_cloud_files_linux/
native.rs

1//! Native `fuser` mapping for read-only, writable, and durable-create Linux engines.
2
3use std::{
4    ffi::OsStr,
5    io,
6    path::Path,
7    time::{Duration, SystemTime},
8};
9
10use aster_forge_cloud_files_core::CloudFilesBackend;
11use fuser::{
12    BsdFileFlags, Config, Errno, FileAttr, FileHandle, FileType, Filesystem, FopenFlags,
13    Generation, INodeNo, LockOwner, MountOption, OpenAccMode, OpenFlags, RenameFlags, ReplyAttr,
14    ReplyCreate, ReplyData, ReplyDirectory, ReplyEmpty, ReplyEntry, ReplyOpen, ReplyWrite, Request,
15    TimeOrNow, WriteFlags,
16};
17use tokio::runtime::Handle;
18
19use crate::{
20    LinuxDirectoryHandle, LinuxDispatchRejection, LinuxErrorCode, LinuxFileAccess,
21    LinuxFileAttributes, LinuxFileHandle, LinuxInode, LinuxInvalidation, LinuxMountConfig,
22    LinuxNode, LinuxNodeKind, LinuxReadOnlyEngine, LinuxRequestDispatcher, LinuxWritableEngine,
23    LinuxWritebackStore, Result,
24};
25
26/// Native kernel-cache invalidation port owned by one mounted FUSE session.
27#[derive(Debug, Clone)]
28pub struct LinuxKernelNotifier {
29    inner: fuser::Notifier,
30}
31
32impl LinuxKernelNotifier {
33    fn new(inner: fuser::Notifier) -> Self {
34        Self { inner }
35    }
36
37    /// Applies one engine-produced invalidation to the mounted kernel namespace.
38    ///
39    /// # Errors
40    ///
41    /// Returns an error when the kernel rejects the invalidation or the mounted session is no
42    /// longer available.
43    pub fn apply(&self, invalidation: &LinuxInvalidation) -> io::Result<()> {
44        match invalidation {
45            LinuxInvalidation::Entry { parent, name } => self
46                .inner
47                .inval_entry(INodeNo(parent.get()), OsStr::new(name)),
48            LinuxInvalidation::Inode { inode } => {
49                self.inner.inval_inode(INodeNo(inode.get()), 0, 0)
50            }
51            LinuxInvalidation::Delete {
52                parent,
53                child,
54                name,
55            } => self.inner.delete(
56                INodeNo(parent.get()),
57                INodeNo(child.get()),
58                OsStr::new(name),
59            ),
60        }
61    }
62
63    /// Applies an ordered engine plan, stopping at the first kernel error.
64    ///
65    /// # Errors
66    ///
67    /// Returns the first error produced while applying an invalidation to the kernel.
68    pub fn apply_all<'a>(
69        &self,
70        invalidations: impl IntoIterator<Item = &'a LinuxInvalidation>,
71    ) -> io::Result<()> {
72        for invalidation in invalidations {
73            self.apply(invalidation)?;
74        }
75        Ok(())
76    }
77}
78
79/// Background FUSE session that exposes notifier and explicit join/unmount boundaries.
80#[derive(Debug)]
81pub struct LinuxBackgroundSession {
82    inner: fuser::BackgroundSession,
83}
84
85impl LinuxBackgroundSession {
86    /// Returns a cloneable kernel notification port for remote-change workers.
87    #[must_use]
88    pub fn notifier(&self) -> LinuxKernelNotifier {
89        LinuxKernelNotifier::new(self.inner.notifier())
90    }
91
92    /// Waits for the mounted session thread to finish.
93    ///
94    /// # Errors
95    ///
96    /// Returns an error when the background session thread terminates with an I/O failure.
97    pub fn join(self) -> io::Result<()> {
98        self.inner.join()
99    }
100
101    /// Unmounts the filesystem and then joins its session thread.
102    ///
103    /// # Errors
104    ///
105    /// Returns an error when unmounting or joining the background session fails.
106    pub fn unmount_and_join(self) -> io::Result<()> {
107        self.inner.umount_and_join()
108    }
109}
110
111fn errno(code: LinuxErrorCode) -> Errno {
112    match code {
113        LinuxErrorCode::NotFound => Errno::ENOENT,
114        LinuxErrorCode::AlreadyExists => Errno::EEXIST,
115        LinuxErrorCode::DirectoryNotEmpty => Errno::ENOTEMPTY,
116        LinuxErrorCode::IsDirectory => Errno::EISDIR,
117        LinuxErrorCode::NotDirectory => Errno::ENOTDIR,
118        LinuxErrorCode::AccessDenied => Errno::EACCES,
119        LinuxErrorCode::ReadOnlyFilesystem => Errno::EROFS,
120        LinuxErrorCode::TryAgain => Errno::EAGAIN,
121        LinuxErrorCode::Stale => Errno::ESTALE,
122        LinuxErrorCode::NotSupported => Errno::ENOSYS,
123        LinuxErrorCode::InvalidArgument => Errno::EINVAL,
124        LinuxErrorCode::Io => Errno::EIO,
125    }
126}
127
128fn file_access(mode: OpenAccMode) -> LinuxFileAccess {
129    match mode {
130        OpenAccMode::O_RDONLY => LinuxFileAccess::Read,
131        OpenAccMode::O_WRONLY => LinuxFileAccess::Write,
132        OpenAccMode::O_RDWR => LinuxFileAccess::ReadWrite,
133    }
134}
135
136fn request_inode(inode: INodeNo) -> Result<LinuxInode> {
137    LinuxInode::new(u64::from(inode))
138}
139
140fn file_type(kind: LinuxNodeKind) -> FileType {
141    match kind {
142        LinuxNodeKind::File => FileType::RegularFile,
143        LinuxNodeKind::Directory => FileType::Directory,
144    }
145}
146
147fn file_attr(attributes: &LinuxFileAttributes) -> FileAttr {
148    FileAttr {
149        ino: INodeNo(attributes.inode().get()),
150        size: attributes.size(),
151        blocks: attributes.blocks(),
152        atime: attributes.time(),
153        mtime: attributes.time(),
154        ctime: attributes.time(),
155        crtime: attributes.time(),
156        kind: file_type(attributes.kind()),
157        perm: attributes.permissions(),
158        nlink: attributes.links(),
159        uid: attributes.uid(),
160        gid: attributes.gid(),
161        rdev: 0,
162        blksize: attributes.block_size(),
163        flags: 0,
164    }
165}
166
167fn reply_entry(reply: ReplyEntry, ttl: Duration, node: &LinuxNode) {
168    reply.entry(
169        &ttl,
170        &file_attr(node.attributes()),
171        Generation(node.generation().get()),
172    );
173}
174
175fn rejection_errno(rejection: LinuxDispatchRejection) -> Errno {
176    errno(rejection.error_code())
177}
178
179/// Native FUSE filesystem that delegates all remote work to a bounded explicit runtime.
180pub struct LinuxReadOnlyFilesystem<B> {
181    engine: LinuxReadOnlyEngine<B>,
182    dispatcher: LinuxRequestDispatcher,
183}
184
185/// Native writable FUSE filesystem with direct-I/O writeback and optional durable create.
186pub struct LinuxWritableFilesystem<B, S> {
187    engine: LinuxWritableEngine<B, S>,
188    dispatcher: LinuxRequestDispatcher,
189}
190
191impl<B, S> LinuxWritableFilesystem<B, S>
192where
193    B: CloudFilesBackend + 'static,
194    S: LinuxWritebackStore + 'static,
195{
196    /// Creates a native writable filesystem without mounting it.
197    ///
198    /// # Errors
199    ///
200    /// Returns an error when `max_in_flight` is zero and the bounded dispatcher cannot be
201    /// constructed.
202    pub fn new(
203        engine: LinuxWritableEngine<B, S>,
204        runtime: Handle,
205        max_in_flight: usize,
206    ) -> Result<Self> {
207        Ok(Self {
208            engine,
209            dispatcher: LinuxRequestDispatcher::new(runtime, max_in_flight)?,
210        })
211    }
212
213    /// Returns the shared dispatcher for metrics and controlled shutdown.
214    #[must_use]
215    pub const fn dispatcher(&self) -> &LinuxRequestDispatcher {
216        &self.dispatcher
217    }
218}
219
220impl<B> LinuxReadOnlyFilesystem<B>
221where
222    B: CloudFilesBackend + 'static,
223{
224    /// Creates a native filesystem without mounting it.
225    ///
226    /// # Errors
227    ///
228    /// Returns an error when `max_in_flight` is zero and the bounded dispatcher cannot be
229    /// constructed.
230    pub fn new(
231        engine: LinuxReadOnlyEngine<B>,
232        runtime: Handle,
233        max_in_flight: usize,
234    ) -> Result<Self> {
235        Ok(Self {
236            engine,
237            dispatcher: LinuxRequestDispatcher::new(runtime, max_in_flight)?,
238        })
239    }
240
241    /// Returns the shared dispatcher for metrics and controlled shutdown.
242    #[must_use]
243    pub const fn dispatcher(&self) -> &LinuxRequestDispatcher {
244        &self.dispatcher
245    }
246}
247
248impl<B> Filesystem for LinuxReadOnlyFilesystem<B>
249where
250    B: CloudFilesBackend + 'static,
251{
252    fn destroy(&mut self) {
253        self.dispatcher.close();
254    }
255
256    fn lookup(&self, _request: &Request, parent: INodeNo, name: &OsStr, reply: ReplyEntry) {
257        let parent = match request_inode(parent) {
258            Ok(parent) => parent,
259            Err(error) => {
260                reply.error(errno(error.error_code()));
261                return;
262            }
263        };
264        let Some(name) = name.to_str().map(str::to_owned) else {
265            reply.error(Errno::EINVAL);
266            return;
267        };
268        let reservation = match self.dispatcher.reserve() {
269            Ok(reservation) => reservation,
270            Err(rejection) => {
271                reply.error(rejection_errno(rejection));
272                return;
273            }
274        };
275        let engine = self.engine.clone();
276        let ttl = engine.attribute_policy().cache_ttl();
277        reservation.spawn(async move {
278            match engine.lookup(parent, &name).await {
279                Ok(node) => reply_entry(reply, ttl, &node),
280                Err(error) => reply.error(errno(error.error_code())),
281            }
282        });
283    }
284
285    fn getattr(
286        &self,
287        _request: &Request,
288        inode: INodeNo,
289        _handle: Option<FileHandle>,
290        reply: ReplyAttr,
291    ) {
292        let inode = match request_inode(inode) {
293            Ok(inode) => inode,
294            Err(error) => {
295                reply.error(errno(error.error_code()));
296                return;
297            }
298        };
299        let reservation = match self.dispatcher.reserve() {
300            Ok(reservation) => reservation,
301            Err(rejection) => {
302                reply.error(rejection_errno(rejection));
303                return;
304            }
305        };
306        let engine = self.engine.clone();
307        let ttl = engine.attribute_policy().cache_ttl();
308        reservation.spawn(async move {
309            match engine.getattr(inode).await {
310                Ok(node) => reply.attr(&ttl, &file_attr(node.attributes())),
311                Err(error) => reply.error(errno(error.error_code())),
312            }
313        });
314    }
315
316    fn open(&self, _request: &Request, inode: INodeNo, flags: OpenFlags, reply: ReplyOpen) {
317        if flags.acc_mode() != OpenAccMode::O_RDONLY {
318            reply.error(Errno::EROFS);
319            return;
320        }
321        let inode = match request_inode(inode) {
322            Ok(inode) => inode,
323            Err(error) => {
324                reply.error(errno(error.error_code()));
325                return;
326            }
327        };
328        let reservation = match self.dispatcher.reserve() {
329            Ok(reservation) => reservation,
330            Err(rejection) => {
331                reply.error(rejection_errno(rejection));
332                return;
333            }
334        };
335        let engine = self.engine.clone();
336        reservation.spawn(async move {
337            match engine.open_file(inode).await {
338                Ok(handle) => reply.opened(
339                    FileHandle(handle.get()),
340                    FopenFlags::FOPEN_DIRECT_IO | FopenFlags::FOPEN_NOFLUSH,
341                ),
342                Err(error) => reply.error(errno(error.error_code())),
343            }
344        });
345    }
346
347    fn read(
348        &self,
349        _request: &Request,
350        inode: INodeNo,
351        handle: FileHandle,
352        offset: u64,
353        size: u32,
354        _flags: OpenFlags,
355        _lock_owner: Option<LockOwner>,
356        reply: ReplyData,
357    ) {
358        let inode = match request_inode(inode) {
359            Ok(inode) => inode,
360            Err(error) => {
361                reply.error(errno(error.error_code()));
362                return;
363            }
364        };
365        let handle = match LinuxFileHandle::new(u64::from(handle)) {
366            Ok(handle) => handle,
367            Err(error) => {
368                reply.error(errno(error.error_code()));
369                return;
370            }
371        };
372        let reservation = match self.dispatcher.reserve() {
373            Ok(reservation) => reservation,
374            Err(rejection) => {
375                reply.error(rejection_errno(rejection));
376                return;
377            }
378        };
379        let engine = self.engine.clone();
380        reservation.spawn(async move {
381            match engine.read_file(inode, handle, offset, size).await {
382                Ok(bytes) => reply.data(&bytes),
383                Err(error) => reply.error(errno(error.error_code())),
384            }
385        });
386    }
387
388    fn write(
389        &self,
390        _request: &Request,
391        _inode: INodeNo,
392        _handle: FileHandle,
393        _offset: u64,
394        _data: &[u8],
395        _write_flags: WriteFlags,
396        _flags: OpenFlags,
397        _lock_owner: Option<LockOwner>,
398        reply: ReplyWrite,
399    ) {
400        reply.error(Errno::EROFS);
401    }
402
403    fn release(
404        &self,
405        _request: &Request,
406        _inode: INodeNo,
407        handle: FileHandle,
408        _flags: OpenFlags,
409        _lock_owner: Option<LockOwner>,
410        _flush: bool,
411        reply: ReplyEmpty,
412    ) {
413        if let Ok(handle) = LinuxFileHandle::new(u64::from(handle)) {
414            let _result = self.engine.release_file(handle);
415        }
416        reply.ok();
417    }
418
419    fn opendir(&self, _request: &Request, inode: INodeNo, _flags: OpenFlags, reply: ReplyOpen) {
420        let inode = match request_inode(inode) {
421            Ok(inode) => inode,
422            Err(error) => {
423                reply.error(errno(error.error_code()));
424                return;
425            }
426        };
427        let reservation = match self.dispatcher.reserve() {
428            Ok(reservation) => reservation,
429            Err(rejection) => {
430                reply.error(rejection_errno(rejection));
431                return;
432            }
433        };
434        let engine = self.engine.clone();
435        reservation.spawn(async move {
436            match engine.open_directory(inode).await {
437                Ok(handle) => reply.opened(FileHandle(handle.get()), FopenFlags::FOPEN_CACHE_DIR),
438                Err(error) => reply.error(errno(error.error_code())),
439            }
440        });
441    }
442
443    fn readdir(
444        &self,
445        _request: &Request,
446        inode: INodeNo,
447        handle: FileHandle,
448        offset: u64,
449        mut reply: ReplyDirectory,
450    ) {
451        let result = request_inode(inode).and_then(|inode| {
452            let handle = LinuxDirectoryHandle::new(u64::from(handle))?;
453            self.engine
454                .directory_snapshot(inode, handle)
455                .map(|snapshot| (inode, snapshot))
456        });
457        let (inode, snapshot) = match result {
458            Ok(result) => result,
459            Err(error) => {
460                reply.error(errno(error.error_code()));
461                return;
462            }
463        };
464        let Ok(start) = usize::try_from(offset) else {
465            reply.ok();
466            return;
467        };
468        let mut index = 0usize;
469        let mut full = false;
470        if index >= start {
471            full = reply.add(INodeNo(inode.get()), 1, FileType::Directory, ".");
472        }
473        index += 1;
474        if !full && index >= start {
475            full = reply.add(
476                INodeNo(snapshot.parent().get()),
477                2,
478                FileType::Directory,
479                "..",
480            );
481        }
482        index += 1;
483        for entry in snapshot.entries() {
484            if full {
485                break;
486            }
487            let Ok(next) = u64::try_from(index + 1) else {
488                break;
489            };
490            if index >= start {
491                full = reply.add(
492                    INodeNo(entry.inode().get()),
493                    next,
494                    file_type(entry.kind()),
495                    entry.name(),
496                );
497            }
498            index += 1;
499        }
500        reply.ok();
501    }
502
503    fn releasedir(
504        &self,
505        _request: &Request,
506        _inode: INodeNo,
507        handle: FileHandle,
508        _flags: OpenFlags,
509        reply: ReplyEmpty,
510    ) {
511        if let Ok(handle) = LinuxDirectoryHandle::new(u64::from(handle)) {
512            let _result = self.engine.release_directory(handle);
513        }
514        reply.ok();
515    }
516}
517
518impl<B, S> Filesystem for LinuxWritableFilesystem<B, S>
519where
520    B: CloudFilesBackend + 'static,
521    S: LinuxWritebackStore + 'static,
522{
523    fn destroy(&mut self) {
524        self.dispatcher.close();
525    }
526
527    fn lookup(&self, _request: &Request, parent: INodeNo, name: &OsStr, reply: ReplyEntry) {
528        let parent = match request_inode(parent) {
529            Ok(parent) => parent,
530            Err(error) => {
531                reply.error(errno(error.error_code()));
532                return;
533            }
534        };
535        let Some(name) = name.to_str().map(str::to_owned) else {
536            reply.error(Errno::EINVAL);
537            return;
538        };
539        let reservation = match self.dispatcher.reserve() {
540            Ok(reservation) => reservation,
541            Err(rejection) => {
542                reply.error(rejection_errno(rejection));
543                return;
544            }
545        };
546        let engine = self.engine.clone();
547        let ttl = engine.readonly().attribute_policy().cache_ttl();
548        reservation.spawn(async move {
549            match engine.lookup(parent, &name).await {
550                Ok(node) => reply_entry(reply, ttl, &node),
551                Err(error) => reply.error(errno(error.error_code())),
552            }
553        });
554    }
555
556    fn getattr(
557        &self,
558        _request: &Request,
559        inode: INodeNo,
560        handle: Option<FileHandle>,
561        reply: ReplyAttr,
562    ) {
563        let inode = match request_inode(inode) {
564            Ok(inode) => inode,
565            Err(error) => {
566                reply.error(errno(error.error_code()));
567                return;
568            }
569        };
570        let handle = match handle
571            .map(|handle| LinuxFileHandle::new(u64::from(handle)))
572            .transpose()
573        {
574            Ok(handle) => handle,
575            Err(error) => {
576                reply.error(errno(error.error_code()));
577                return;
578            }
579        };
580        let reservation = match self.dispatcher.reserve() {
581            Ok(reservation) => reservation,
582            Err(rejection) => {
583                reply.error(rejection_errno(rejection));
584                return;
585            }
586        };
587        let engine = self.engine.clone();
588        let ttl = engine.readonly().attribute_policy().cache_ttl();
589        reservation.spawn(async move {
590            match engine.getattr(inode, handle).await {
591                Ok(node) => reply.attr(&ttl, &file_attr(node.attributes())),
592                Err(error) => reply.error(errno(error.error_code())),
593            }
594        });
595    }
596
597    fn setattr(
598        &self,
599        _request: &Request,
600        inode: INodeNo,
601        mode: Option<u32>,
602        uid: Option<u32>,
603        gid: Option<u32>,
604        size: Option<u64>,
605        atime: Option<TimeOrNow>,
606        mtime: Option<TimeOrNow>,
607        status_change_time: Option<SystemTime>,
608        handle: Option<FileHandle>,
609        creation_time: Option<SystemTime>,
610        attribute_change_time: Option<SystemTime>,
611        backup_time: Option<SystemTime>,
612        flags: Option<BsdFileFlags>,
613        reply: ReplyAttr,
614    ) {
615        if mode.is_some()
616            || uid.is_some()
617            || gid.is_some()
618            || atime.is_some()
619            || mtime.is_some()
620            || status_change_time.is_some()
621            || creation_time.is_some()
622            || attribute_change_time.is_some()
623            || backup_time.is_some()
624            || flags.is_some()
625        {
626            reply.error(Errno::ENOSYS);
627            return;
628        }
629        let Some(size) = size else {
630            reply.error(Errno::EINVAL);
631            return;
632        };
633        let inode = match request_inode(inode) {
634            Ok(inode) => inode,
635            Err(error) => {
636                reply.error(errno(error.error_code()));
637                return;
638            }
639        };
640        let handle = match handle
641            .map(|handle| LinuxFileHandle::new(u64::from(handle)))
642            .transpose()
643        {
644            Ok(handle) => handle,
645            Err(error) => {
646                reply.error(errno(error.error_code()));
647                return;
648            }
649        };
650        let reservation = match self.dispatcher.reserve() {
651            Ok(reservation) => reservation,
652            Err(rejection) => {
653                reply.error(rejection_errno(rejection));
654                return;
655            }
656        };
657        let engine = self.engine.clone();
658        let ttl = engine.readonly().attribute_policy().cache_ttl();
659        reservation.spawn(async move {
660            let result = match handle {
661                Some(handle) => engine.truncate_file(inode, handle, size).await,
662                None => engine.truncate_once(inode, size).await,
663            };
664            match result {
665                Ok(node) => reply.attr(&ttl, &file_attr(node.attributes())),
666                Err(error) => reply.error(errno(error.error_code())),
667            }
668        });
669    }
670
671    fn open(&self, _request: &Request, inode: INodeNo, flags: OpenFlags, reply: ReplyOpen) {
672        let inode = match request_inode(inode) {
673            Ok(inode) => inode,
674            Err(error) => {
675                reply.error(errno(error.error_code()));
676                return;
677            }
678        };
679        let reservation = match self.dispatcher.reserve() {
680            Ok(reservation) => reservation,
681            Err(rejection) => {
682                reply.error(rejection_errno(rejection));
683                return;
684            }
685        };
686        let engine = self.engine.clone();
687        let access = file_access(flags.acc_mode());
688        reservation.spawn(async move {
689            match engine.open_file(inode, access).await {
690                Ok(handle) => reply.opened(FileHandle(handle.get()), FopenFlags::FOPEN_DIRECT_IO),
691                Err(error) => reply.error(errno(error.error_code())),
692            }
693        });
694    }
695
696    fn create(
697        &self,
698        _request: &Request,
699        parent: INodeNo,
700        name: &OsStr,
701        mode: u32,
702        umask: u32,
703        flags: i32,
704        reply: ReplyCreate,
705    ) {
706        let parent = match request_inode(parent) {
707            Ok(parent) => parent,
708            Err(error) => {
709                reply.error(errno(error.error_code()));
710                return;
711            }
712        };
713        let Some(name) = name.to_str().map(str::to_owned) else {
714            reply.error(Errno::EINVAL);
715            return;
716        };
717        let reservation = match self.dispatcher.reserve() {
718            Ok(reservation) => reservation,
719            Err(rejection) => {
720                reply.error(rejection_errno(rejection));
721                return;
722            }
723        };
724        let engine = self.engine.clone();
725        let access = file_access(OpenFlags(flags).acc_mode());
726        let ttl = engine.readonly().attribute_policy().cache_ttl();
727        reservation.spawn(async move {
728            match engine.create_file(parent, &name, mode, umask, access).await {
729                Ok((node, handle)) => reply.created(
730                    &ttl,
731                    &file_attr(node.attributes()),
732                    Generation(node.generation().get()),
733                    FileHandle(handle.get()),
734                    FopenFlags::FOPEN_DIRECT_IO,
735                ),
736                Err(error) => reply.error(errno(error.error_code())),
737            }
738        });
739    }
740
741    fn mkdir(
742        &self,
743        _request: &Request,
744        parent: INodeNo,
745        name: &OsStr,
746        mode: u32,
747        umask: u32,
748        reply: ReplyEntry,
749    ) {
750        let parent = match request_inode(parent) {
751            Ok(parent) => parent,
752            Err(error) => {
753                reply.error(errno(error.error_code()));
754                return;
755            }
756        };
757        let Some(name) = name.to_str().map(str::to_owned) else {
758            reply.error(Errno::EINVAL);
759            return;
760        };
761        let reservation = match self.dispatcher.reserve() {
762            Ok(reservation) => reservation,
763            Err(rejection) => {
764                reply.error(rejection_errno(rejection));
765                return;
766            }
767        };
768        let engine = self.engine.clone();
769        let ttl = engine.readonly().attribute_policy().cache_ttl();
770        reservation.spawn(async move {
771            match engine.create_directory(parent, &name, mode, umask).await {
772                Ok(node) => reply_entry(reply, ttl, &node),
773                Err(error) => reply.error(errno(error.error_code())),
774            }
775        });
776    }
777
778    fn unlink(&self, _request: &Request, parent: INodeNo, name: &OsStr, reply: ReplyEmpty) {
779        self.remove_request(parent, name, LinuxNodeKind::File, reply);
780    }
781
782    fn rmdir(&self, _request: &Request, parent: INodeNo, name: &OsStr, reply: ReplyEmpty) {
783        self.remove_request(parent, name, LinuxNodeKind::Directory, reply);
784    }
785
786    fn rename(
787        &self,
788        _request: &Request,
789        parent: INodeNo,
790        name: &OsStr,
791        new_parent: INodeNo,
792        new_name: &OsStr,
793        flags: RenameFlags,
794        reply: ReplyEmpty,
795    ) {
796        if flags.intersects(RenameFlags::RENAME_EXCHANGE | RenameFlags::RENAME_WHITEOUT) {
797            reply.error(Errno::ENOSYS);
798            return;
799        }
800        let result = request_inode(parent)
801            .and_then(|parent| request_inode(new_parent).map(|new_parent| (parent, new_parent)));
802        let (parent, new_parent) = match result {
803            Ok(result) => result,
804            Err(error) => {
805                reply.error(errno(error.error_code()));
806                return;
807            }
808        };
809        let (Some(name), Some(new_name)) = (
810            name.to_str().map(str::to_owned),
811            new_name.to_str().map(str::to_owned),
812        ) else {
813            reply.error(Errno::EINVAL);
814            return;
815        };
816        let reservation = match self.dispatcher.reserve() {
817            Ok(reservation) => reservation,
818            Err(rejection) => {
819                reply.error(rejection_errno(rejection));
820                return;
821            }
822        };
823        let engine = self.engine.clone();
824        let no_replace = flags.contains(RenameFlags::RENAME_NOREPLACE);
825        reservation.spawn(async move {
826            match engine
827                .rename(parent, &name, new_parent, &new_name, no_replace)
828                .await
829            {
830                Ok(()) => reply.ok(),
831                Err(error) => reply.error(errno(error.error_code())),
832            }
833        });
834    }
835
836    fn read(
837        &self,
838        _request: &Request,
839        inode: INodeNo,
840        handle: FileHandle,
841        offset: u64,
842        size: u32,
843        _flags: OpenFlags,
844        _lock_owner: Option<LockOwner>,
845        reply: ReplyData,
846    ) {
847        let result = request_inode(inode).and_then(|inode| {
848            LinuxFileHandle::new(u64::from(handle)).map(|handle| (inode, handle))
849        });
850        let (inode, handle) = match result {
851            Ok(result) => result,
852            Err(error) => {
853                reply.error(errno(error.error_code()));
854                return;
855            }
856        };
857        let reservation = match self.dispatcher.reserve() {
858            Ok(reservation) => reservation,
859            Err(rejection) => {
860                reply.error(rejection_errno(rejection));
861                return;
862            }
863        };
864        let engine = self.engine.clone();
865        reservation.spawn(async move {
866            match engine.read_file(inode, handle, offset, size).await {
867                Ok(bytes) => reply.data(&bytes),
868                Err(error) => reply.error(errno(error.error_code())),
869            }
870        });
871    }
872
873    fn write(
874        &self,
875        _request: &Request,
876        inode: INodeNo,
877        handle: FileHandle,
878        offset: u64,
879        data: &[u8],
880        _write_flags: WriteFlags,
881        _flags: OpenFlags,
882        _lock_owner: Option<LockOwner>,
883        reply: ReplyWrite,
884    ) {
885        let result = request_inode(inode).and_then(|inode| {
886            LinuxFileHandle::new(u64::from(handle)).map(|handle| (inode, handle))
887        });
888        let (inode, handle) = match result {
889            Ok(result) => result,
890            Err(error) => {
891                reply.error(errno(error.error_code()));
892                return;
893            }
894        };
895        let reservation = match self.dispatcher.reserve() {
896            Ok(reservation) => reservation,
897            Err(rejection) => {
898                reply.error(rejection_errno(rejection));
899                return;
900            }
901        };
902        let engine = self.engine.clone();
903        let bytes = bytes::Bytes::copy_from_slice(data);
904        reservation.spawn(async move {
905            match engine.write_file(inode, handle, offset, bytes).await {
906                Ok(written) => reply.written(written),
907                Err(error) => reply.error(errno(error.error_code())),
908            }
909        });
910    }
911
912    fn flush(
913        &self,
914        _request: &Request,
915        inode: INodeNo,
916        handle: FileHandle,
917        _lock_owner: LockOwner,
918        reply: ReplyEmpty,
919    ) {
920        self.sync_request(inode, handle, false, reply);
921    }
922
923    fn fsync(
924        &self,
925        _request: &Request,
926        inode: INodeNo,
927        handle: FileHandle,
928        data_only: bool,
929        reply: ReplyEmpty,
930    ) {
931        self.sync_request(inode, handle, data_only, reply);
932    }
933
934    fn release(
935        &self,
936        _request: &Request,
937        _inode: INodeNo,
938        handle: FileHandle,
939        _flags: OpenFlags,
940        _lock_owner: Option<LockOwner>,
941        _flush: bool,
942        reply: ReplyEmpty,
943    ) {
944        let Ok(handle) = LinuxFileHandle::new(u64::from(handle)) else {
945            reply.ok();
946            return;
947        };
948        let engine = self.engine.clone();
949        self.dispatcher.spawn_cleanup(async move {
950            let _result = engine.release_file(handle).await;
951            reply.ok();
952        });
953    }
954
955    fn opendir(&self, _request: &Request, inode: INodeNo, _flags: OpenFlags, reply: ReplyOpen) {
956        let inode = match request_inode(inode) {
957            Ok(inode) => inode,
958            Err(error) => {
959                reply.error(errno(error.error_code()));
960                return;
961            }
962        };
963        let reservation = match self.dispatcher.reserve() {
964            Ok(reservation) => reservation,
965            Err(rejection) => {
966                reply.error(rejection_errno(rejection));
967                return;
968            }
969        };
970        let engine = self.engine.clone();
971        reservation.spawn(async move {
972            match engine.open_directory(inode).await {
973                Ok(handle) => reply.opened(FileHandle(handle.get()), FopenFlags::FOPEN_CACHE_DIR),
974                Err(error) => reply.error(errno(error.error_code())),
975            }
976        });
977    }
978
979    fn readdir(
980        &self,
981        _request: &Request,
982        inode: INodeNo,
983        handle: FileHandle,
984        offset: u64,
985        mut reply: ReplyDirectory,
986    ) {
987        let result = request_inode(inode).and_then(|inode| {
988            let handle = LinuxDirectoryHandle::new(u64::from(handle))?;
989            self.engine
990                .directory_snapshot(inode, handle)
991                .map(|snapshot| (inode, snapshot))
992        });
993        let (inode, snapshot) = match result {
994            Ok(result) => result,
995            Err(error) => {
996                reply.error(errno(error.error_code()));
997                return;
998            }
999        };
1000        let Ok(start) = usize::try_from(offset) else {
1001            reply.ok();
1002            return;
1003        };
1004        let mut index = 0usize;
1005        let mut full = false;
1006        if index >= start {
1007            full = reply.add(INodeNo(inode.get()), 1, FileType::Directory, ".");
1008        }
1009        index += 1;
1010        if !full && index >= start {
1011            full = reply.add(
1012                INodeNo(snapshot.parent().get()),
1013                2,
1014                FileType::Directory,
1015                "..",
1016            );
1017        }
1018        index += 1;
1019        for entry in snapshot.entries() {
1020            if full {
1021                break;
1022            }
1023            let Ok(next) = u64::try_from(index + 1) else {
1024                break;
1025            };
1026            if index >= start {
1027                full = reply.add(
1028                    INodeNo(entry.inode().get()),
1029                    next,
1030                    file_type(entry.kind()),
1031                    entry.name(),
1032                );
1033            }
1034            index += 1;
1035        }
1036        reply.ok();
1037    }
1038
1039    fn releasedir(
1040        &self,
1041        _request: &Request,
1042        _inode: INodeNo,
1043        handle: FileHandle,
1044        _flags: OpenFlags,
1045        reply: ReplyEmpty,
1046    ) {
1047        if let Ok(handle) = LinuxDirectoryHandle::new(u64::from(handle)) {
1048            let _result = self.engine.release_directory(handle);
1049        }
1050        reply.ok();
1051    }
1052}
1053
1054impl<B, S> LinuxWritableFilesystem<B, S>
1055where
1056    B: CloudFilesBackend + 'static,
1057    S: LinuxWritebackStore + 'static,
1058{
1059    fn sync_request(&self, inode: INodeNo, handle: FileHandle, data_only: bool, reply: ReplyEmpty) {
1060        let result = request_inode(inode).and_then(|inode| {
1061            LinuxFileHandle::new(u64::from(handle)).map(|handle| (inode, handle))
1062        });
1063        let (inode, handle) = match result {
1064            Ok(result) => result,
1065            Err(error) => {
1066                reply.error(errno(error.error_code()));
1067                return;
1068            }
1069        };
1070        let reservation = match self.dispatcher.reserve() {
1071            Ok(reservation) => reservation,
1072            Err(rejection) => {
1073                reply.error(rejection_errno(rejection));
1074                return;
1075            }
1076        };
1077        let engine = self.engine.clone();
1078        reservation.spawn(async move {
1079            match engine.sync_file(inode, handle, data_only).await {
1080                Ok(()) => reply.ok(),
1081                Err(error) => reply.error(errno(error.error_code())),
1082            }
1083        });
1084    }
1085
1086    fn remove_request(
1087        &self,
1088        parent: INodeNo,
1089        name: &OsStr,
1090        kind: LinuxNodeKind,
1091        reply: ReplyEmpty,
1092    ) {
1093        let parent = match request_inode(parent) {
1094            Ok(parent) => parent,
1095            Err(error) => {
1096                reply.error(errno(error.error_code()));
1097                return;
1098            }
1099        };
1100        let Some(name) = name.to_str().map(str::to_owned) else {
1101            reply.error(Errno::EINVAL);
1102            return;
1103        };
1104        let reservation = match self.dispatcher.reserve() {
1105            Ok(reservation) => reservation,
1106            Err(rejection) => {
1107                reply.error(rejection_errno(rejection));
1108                return;
1109            }
1110        };
1111        let engine = self.engine.clone();
1112        reservation.spawn(async move {
1113            match engine.remove(parent, &name, kind).await {
1114                Ok(()) => reply.ok(),
1115                Err(error) => reply.error(errno(error.error_code())),
1116            }
1117        });
1118    }
1119}
1120
1121/// Mounts a blocking read-only FUSE session using product-neutral hardened defaults.
1122///
1123/// # Errors
1124///
1125/// Returns an error when the mountpoint cannot be opened or the FUSE session cannot be mounted.
1126pub fn mount_read_only<B>(
1127    filesystem: LinuxReadOnlyFilesystem<B>,
1128    mountpoint: impl AsRef<Path>,
1129    config: &LinuxMountConfig,
1130) -> io::Result<()>
1131where
1132    B: CloudFilesBackend + 'static,
1133{
1134    let mut options = Config::default();
1135    options.n_threads = config.kernel_threads();
1136    options.clone_fd = config.clone_fd();
1137    options.mount_options.extend([
1138        MountOption::RO,
1139        MountOption::DefaultPermissions,
1140        MountOption::NoDev,
1141        MountOption::NoSuid,
1142        MountOption::NoExec,
1143        MountOption::FSName(config.filesystem_name().to_owned()),
1144    ]);
1145    if let Some(subtype) = config.subtype() {
1146        options
1147            .mount_options
1148            .push(MountOption::Subtype(subtype.to_owned()));
1149    }
1150    fuser::mount(filesystem, mountpoint, &options)
1151}
1152
1153/// Spawns a read-only FUSE session and returns its kernel invalidation lifecycle handle.
1154///
1155/// # Errors
1156///
1157/// Returns an error when the mountpoint cannot be opened or the background FUSE session cannot be
1158/// started.
1159pub fn spawn_mount_read_only<B>(
1160    filesystem: LinuxReadOnlyFilesystem<B>,
1161    mountpoint: impl AsRef<Path>,
1162    config: &LinuxMountConfig,
1163) -> io::Result<LinuxBackgroundSession>
1164where
1165    B: CloudFilesBackend + 'static,
1166{
1167    let mut options = Config::default();
1168    options.n_threads = config.kernel_threads();
1169    options.clone_fd = config.clone_fd();
1170    options.mount_options.extend([
1171        MountOption::RO,
1172        MountOption::DefaultPermissions,
1173        MountOption::NoDev,
1174        MountOption::NoSuid,
1175        MountOption::NoExec,
1176        MountOption::FSName(config.filesystem_name().to_owned()),
1177    ]);
1178    if let Some(subtype) = config.subtype() {
1179        options
1180            .mount_options
1181            .push(MountOption::Subtype(subtype.to_owned()));
1182    }
1183    fuser::spawn_mount(filesystem, mountpoint, &options)
1184        .map(|inner| LinuxBackgroundSession { inner })
1185}
1186
1187/// Mounts a blocking writable FUSE session without kernel writeback-cache semantics.
1188///
1189/// # Errors
1190///
1191/// Returns an error when the mountpoint cannot be opened or the FUSE session cannot be mounted.
1192pub fn mount_writable<B, S>(
1193    filesystem: LinuxWritableFilesystem<B, S>,
1194    mountpoint: impl AsRef<Path>,
1195    config: &LinuxMountConfig,
1196) -> io::Result<()>
1197where
1198    B: CloudFilesBackend + 'static,
1199    S: LinuxWritebackStore + 'static,
1200{
1201    let mut options = Config::default();
1202    options.n_threads = config.kernel_threads();
1203    options.clone_fd = config.clone_fd();
1204    options.mount_options.extend([
1205        MountOption::DefaultPermissions,
1206        MountOption::NoDev,
1207        MountOption::NoSuid,
1208        MountOption::NoExec,
1209        MountOption::FSName(config.filesystem_name().to_owned()),
1210    ]);
1211    if let Some(subtype) = config.subtype() {
1212        options
1213            .mount_options
1214            .push(MountOption::Subtype(subtype.to_owned()));
1215    }
1216    fuser::mount(filesystem, mountpoint, &options)
1217}
1218
1219/// Spawns a writable direct-I/O FUSE session with a kernel invalidation handle.
1220///
1221/// # Errors
1222///
1223/// Returns an error when the mountpoint cannot be opened or the background FUSE session cannot be
1224/// started.
1225pub fn spawn_mount_writable<B, S>(
1226    filesystem: LinuxWritableFilesystem<B, S>,
1227    mountpoint: impl AsRef<Path>,
1228    config: &LinuxMountConfig,
1229) -> io::Result<LinuxBackgroundSession>
1230where
1231    B: CloudFilesBackend + 'static,
1232    S: LinuxWritebackStore + 'static,
1233{
1234    let mut options = Config::default();
1235    options.n_threads = config.kernel_threads();
1236    options.clone_fd = config.clone_fd();
1237    options.mount_options.extend([
1238        MountOption::DefaultPermissions,
1239        MountOption::NoDev,
1240        MountOption::NoSuid,
1241        MountOption::NoExec,
1242        MountOption::FSName(config.filesystem_name().to_owned()),
1243    ]);
1244    if let Some(subtype) = config.subtype() {
1245        options
1246            .mount_options
1247            .push(MountOption::Subtype(subtype.to_owned()));
1248    }
1249    fuser::spawn_mount(filesystem, mountpoint, &options)
1250        .map(|inner| LinuxBackgroundSession { inner })
1251}
aster_forge_cloud_files_linux/native.rs

aster_forge_cloud_files_linux/
native.rs
