snix_castore/fs/
mod.rs

1mod file_attr;
2mod inode_tracker;
3mod inodes;
4pub mod root_nodes;
5
6#[cfg(feature = "fuse")]
7pub mod fuse;
8
9#[cfg(feature = "virtiofs")]
10pub mod virtiofs;
11
12pub use self::root_nodes::RootNodes;
13use self::{
14    file_attr::ROOT_FILE_ATTR,
15    inode_tracker::InodeTracker,
16    inodes::{DirectoryInodeData, InodeData},
17};
18use crate::{
19    B3Digest, Node,
20    blobservice::{BlobReader, BlobService},
21    directoryservice::DirectoryService,
22    path::PathComponent,
23};
24use bstr::ByteVec;
25use fuse_backend_rs::api::filesystem::{
26    Context, FileSystem, FsOptions, GetxattrReply, ListxattrReply, ROOT_ID,
27};
28use fuse_backend_rs::{
29    abi::fuse_abi::{Attr, OpenOptions, stat64},
30    api::filesystem::Entry,
31};
32use futures::{StreamExt, stream::BoxStream};
33use parking_lot::RwLock;
34use std::sync::Mutex;
35use std::{
36    collections::HashMap,
37    io,
38    sync::atomic::AtomicU64,
39    sync::{Arc, atomic::Ordering},
40    time::Duration,
41};
42use std::{ffi::CStr, io::Cursor};
43use tokio::io::{AsyncReadExt, AsyncSeekExt};
44use tracing::{Span, debug, error, instrument, warn};
45
46/// This implements a read-only [FileSystem] for a snix-castore
47/// with the passed [BlobService], [DirectoryService] and [RootNodes].
48///
49/// Linux uses inodes in filesystems. When implementing the trait, most calls
50/// *are for* a given inode.
51///
52/// This means, we need to have a stable mapping of inode numbers to the
53/// corresponding store nodes.
54///
55/// We internally delegate all inode allocation and state keeping to the
56/// inode tracker.
57/// We store a mapping from currently "explored" names in the root to their
58/// inode.
59///
60/// There's some places where inodes are allocated / data inserted into
61/// the inode tracker, if not allocated before already:
62///  - Processing a `lookup` request, either in the mount root, or somewhere
63///    deeper.
64///  - Processing a `readdir` request
65///
66///  Things pointing to the same contents get the same inodes, irrespective of
67///  their own location.
68///  This means:
69///  - Symlinks with the same target will get the same inode.
70///  - Regular/executable files with the same contents will get the same inode
71///  - Directories with the same contents will get the same inode.
72///
73/// Due to the above being valid across the whole store, and considering the
74/// merkle structure is a DAG, not a tree, this also means we can't do "bucketed
75/// allocation", aka reserve Directory.size inodes for each directory node we
76/// explore.
77/// Tests for this live in the snix-store crate.
78pub struct SnixStoreFs<BS, DS, RN: RootNodes> {
79    blob_service: BS,
80    directory_service: DS,
81    root_nodes_provider: Arc<RN>,
82    settings: FSSettings,
83
84    /// This maps a given basename in the root to the inode we allocated for the node.
85    root_nodes: RwLock<HashMap<PathComponent, u64>>,
86
87    /// This keeps track of inodes and data alongside them.
88    inode_tracker: RwLock<InodeTracker>,
89
90    // FUTUREWORK: have a generic container type for dir/file handles and handle
91    // allocation.
92    /// This holds all opendir handles (for the root inode), keyed by the handle
93    /// returned from the opendir call.
94    /// For each handle, we store an enumerated Result<(PathComponent, Node), crate::Error>.
95    /// The index is needed as we need to send offset information.
96    #[allow(clippy::type_complexity)]
97    dir_handles: RwLock<
98        HashMap<
99            u64,
100            (
101                Span,
102                Arc<Mutex<BoxStream<'static, (usize, Result<(PathComponent, Node), RN::Error>)>>>,
103            ),
104        >,
105    >,
106
107    next_dir_handle: AtomicU64,
108
109    /// This holds all open file handles
110    #[allow(clippy::type_complexity)]
111    file_handles: RwLock<HashMap<u64, (Span, Arc<Mutex<Box<dyn BlobReader>>>)>>,
112
113    next_file_handle: AtomicU64,
114
115    tokio_handle: tokio::runtime::Handle,
116}
117
118/// Configures some filesystem settings
119#[derive(Debug, Default)]
120pub struct FSSettings {
121    /// Whether to (try) listing elements in the root.
122    pub list_root: bool,
123
124    /// If uid/gid should be overridden, their values
125    pub uid_gid_override: Option<(u32, u32)>,
126
127    /// Whether to expose blob and directory digests as extended attributes.
128    pub show_xattr: bool,
129}
130
131impl<BS, DS, RN> SnixStoreFs<BS, DS, RN>
132where
133    BS: BlobService,
134    DS: DirectoryService,
135    RN: RootNodes,
136{
137    pub fn new(
138        blob_service: BS,
139        directory_service: DS,
140        root_nodes_provider: RN,
141
142        settings: FSSettings,
143        tokio_handle: tokio::runtime::Handle,
144    ) -> Self {
145        Self {
146            blob_service,
147            directory_service,
148            root_nodes_provider: Arc::new(root_nodes_provider),
149
150            settings,
151
152            root_nodes: RwLock::new(HashMap::default()),
153            inode_tracker: RwLock::new(Default::default()),
154
155            dir_handles: RwLock::new(Default::default()),
156            next_dir_handle: AtomicU64::new(1),
157
158            file_handles: RwLock::new(Default::default()),
159            next_file_handle: AtomicU64::new(1),
160            tokio_handle,
161        }
162    }
163
164    /// Retrieves the inode for a given root node basename, if present.
165    /// This obtains a read lock on self.root_nodes.
166    fn get_inode_for_root_name(&self, name: &PathComponent) -> Option<u64> {
167        self.root_nodes.read().get(name).cloned()
168    }
169
170    /// For a given inode, look up the given directory behind it (from
171    /// self.inode_tracker), and return its children.
172    /// The inode_tracker MUST know about this inode already, and it MUST point
173    /// to a [InodeData::Directory].
174    /// It is ok if it's a [DirectoryInodeData::Sparse] - in that case, a lookup
175    /// in self.directory_service is performed, and self.inode_tracker is updated with the
176    /// [DirectoryInodeData::Populated].
177    #[allow(clippy::type_complexity)]
178    #[instrument(skip(self), err)]
179    fn get_directory_children(
180        &self,
181        ino: u64,
182    ) -> io::Result<(B3Digest, Vec<(u64, PathComponent, Node)>)> {
183        let data = self.inode_tracker.read().get(ino).unwrap();
184        match *data {
185            // if it's populated already, return children.
186            InodeData::Directory(DirectoryInodeData::Populated(parent_digest, ref children)) => {
187                Ok((parent_digest, children.clone()))
188            }
189            // if it's sparse, fetch data using directory_service, populate child nodes
190            // and update it in [self.inode_tracker].
191            InodeData::Directory(DirectoryInodeData::Sparse(parent_digest, _)) => {
192                let directory = self
193                    .tokio_handle
194                    .block_on(async { self.directory_service.get(&parent_digest).await })
195                    .map_err(|err| {
196                        warn!(%err, "error from directory service");
197                        io::Error::other(err)
198                    })?
199                    .ok_or_else(|| {
200                        warn!(directory.digest=%parent_digest, "directory not found");
201                        // If the Directory can't be found, this is a hole, bail out.
202                        io::Error::from_raw_os_error(libc::EIO)
203                    })?;
204
205                // Turn the retrieved directory into a InodeData::Directory(DirectoryInodeData::Populated(..)),
206                // allocating inodes for the children on the way.
207                // FUTUREWORK: there's a bunch of cloning going on here, which we can probably avoid.
208                let children = {
209                    let mut inode_tracker = self.inode_tracker.write();
210
211                    let children: Vec<(u64, PathComponent, Node)> = directory
212                        .into_nodes()
213                        .map(|(child_name, child_node)| {
214                            let inode_data = InodeData::from_node(&child_node);
215
216                            let child_ino = inode_tracker.put(inode_data);
217                            (child_ino, child_name, child_node)
218                        })
219                        .collect();
220
221                    // replace.
222                    inode_tracker.replace(
223                        ino,
224                        Arc::new(InodeData::Directory(DirectoryInodeData::Populated(
225                            parent_digest,
226                            children.clone(),
227                        ))),
228                    );
229
230                    children
231                };
232
233                Ok((parent_digest, children))
234            }
235            // if the parent inode was not a directory, this doesn't make sense
236            InodeData::Regular(..) | InodeData::Symlink(_) => {
237                Err(io::Error::from_raw_os_error(libc::ENOTDIR))
238            }
239        }
240    }
241
242    /// This will turn a lookup request for a name in the root to a ino and
243    /// [InodeData].
244    /// It will peek in [self.root_nodes], and then either look it up from
245    /// [self.inode_tracker],
246    /// or otherwise fetch from [self.root_nodes], and then insert into
247    /// [self.inode_tracker].
248    /// In the case the name can't be found, a libc::ENOENT is returned.
249    fn name_in_root_to_ino_and_data(
250        &self,
251        name: &PathComponent,
252    ) -> io::Result<(u64, Arc<InodeData>)> {
253        // Look up the inode for that root node.
254        // If there's one, [self.inode_tracker] MUST also contain the data,
255        // which we can then return.
256        if let Some(inode) = self.get_inode_for_root_name(name) {
257            return Ok((
258                inode,
259                self.inode_tracker
260                    .read()
261                    .get(inode)
262                    .expect("must exist")
263                    .to_owned(),
264            ));
265        }
266
267        // We don't have it yet, look it up in [self.root_nodes].
268        match self
269            .tokio_handle
270            .block_on(async { self.root_nodes_provider.get_by_basename(name).await })
271        {
272            // if there was an error looking up the root node, propagate up an IO error.
273            Err(_e) => Err(io::Error::from_raw_os_error(libc::EIO)),
274            // the root node doesn't exist, so the file doesn't exist.
275            Ok(None) => Err(io::Error::from_raw_os_error(libc::ENOENT)),
276            // The root node does exist
277            Ok(Some(root_node)) => {
278                // Let's check if someone else beat us to updating the inode tracker and
279                // root_nodes map. This avoids locking inode_tracker for writing.
280                if let Some(ino) = self.root_nodes.read().get(name) {
281                    return Ok((
282                        *ino,
283                        self.inode_tracker.read().get(*ino).expect("must exist"),
284                    ));
285                }
286
287                // Only in case it doesn't, lock [self.root_nodes] and
288                // [self.inode_tracker] for writing.
289                let mut root_nodes = self.root_nodes.write();
290                let mut inode_tracker = self.inode_tracker.write();
291
292                // insert the (sparse) inode data and register in
293                // self.root_nodes.
294                let inode_data = InodeData::from_node(&root_node);
295                let ino = inode_tracker.put(inode_data.clone());
296                root_nodes.insert(name.to_owned(), ino);
297
298                Ok((ino, Arc::new(inode_data)))
299            }
300        }
301    }
302
303    /// Helper function, converting a [InodeData] to [Attr],
304    /// applying uid/gid override if configured.
305    fn inode_data_to_attr(&self, inode_data: &InodeData, ino: u64) -> Attr {
306        let mode = match inode_data {
307            InodeData::Regular(_, _, false) => libc::S_IFREG | 0o444,
308            // executable
309            InodeData::Regular(_, _, true) => libc::S_IFREG | 0o555,
310            InodeData::Symlink(_) => libc::S_IFLNK | 0o444,
311            InodeData::Directory(_) => libc::S_IFDIR | 0o555,
312        };
313        // libc::S_IFREG, libc::S_IFLNK & libc::S_IFDIR are u32 on Linux and u16 on MacOS
314        #[cfg(target_os = "macos")]
315        let mode = mode as u32;
316        let mut attr = Attr {
317            ino,
318            // FUTUREWORK: play with this numbers, as it affects read sizes for client applications.
319            blocks: 1024,
320            size: match inode_data {
321                InodeData::Regular(_, size, _) => *size,
322                InodeData::Symlink(target) => target.len() as u64,
323                InodeData::Directory(DirectoryInodeData::Sparse(_, size)) => *size,
324                InodeData::Directory(DirectoryInodeData::Populated(_, children)) => {
325                    children.len() as u64
326                }
327            },
328            mode,
329            nlink: 1, // 0 would prevent lndir from recursing into directories.
330            mtime: 1, // Everything in /nix/store must have timestamp "1".
331            ..Default::default()
332        };
333
334        if let Some((uid, gid)) = self.settings.uid_gid_override {
335            attr.uid = uid;
336            attr.gid = gid;
337        }
338
339        attr
340    }
341}
342fn attr_to_fuse_entry(attr: Attr) -> Entry {
343    Entry {
344        inode: attr.ino,
345        attr: attr.into(),
346        attr_timeout: Duration::MAX,
347        entry_timeout: Duration::MAX,
348        ..Default::default()
349    }
350}
351
352/// Returns the u32 fuse type
353fn node_to_fuse_type(node: &Node) -> u32 {
354    #[allow(clippy::let_and_return)]
355    let ty = match node {
356        Node::Directory { .. } => libc::S_IFDIR,
357        Node::File { .. } => libc::S_IFREG,
358        Node::Symlink { .. } => libc::S_IFLNK,
359    };
360    // libc::S_IFDIR is u32 on Linux and u16 on MacOS
361    #[cfg(target_os = "macos")]
362    let ty = ty as u32;
363
364    ty
365}
366
367const XATTR_NAME_DIRECTORY_DIGEST: &[u8] = b"user.snix.castore.directory.digest";
368const XATTR_NAME_BLOB_DIGEST: &[u8] = b"user.snix.castore.blob.digest";
369
370#[cfg(all(feature = "virtiofs", target_os = "linux"))]
371impl<BS, DS, RN> fuse_backend_rs::api::filesystem::Layer for SnixStoreFs<BS, DS, RN>
372where
373    BS: BlobService,
374    DS: DirectoryService,
375    RN: RootNodes,
376{
377    fn root_inode(&self) -> Self::Inode {
378        ROOT_ID
379    }
380}
381
382impl<BS, DS, RN> FileSystem for SnixStoreFs<BS, DS, RN>
383where
384    BS: BlobService,
385    DS: DirectoryService,
386    RN: RootNodes,
387{
388    type Handle = u64;
389    type Inode = u64;
390
391    fn init(&self, _capable: FsOptions) -> io::Result<FsOptions> {
392        let mut opts = FsOptions::empty();
393
394        // allow more than one pending read request per file-handle at any time
395        opts |= FsOptions::ASYNC_READ;
396
397        #[cfg(target_os = "linux")]
398        {
399            // the filesystem supports readdirplus
400            opts |= FsOptions::DO_READDIRPLUS;
401            // issue both readdir and readdirplus depending on the information expected to be required
402            opts |= FsOptions::READDIRPLUS_AUTO;
403            // allow concurrent lookup() and readdir() requests for the same directory
404            opts |= FsOptions::PARALLEL_DIROPS;
405            // have the kernel cache symlink contents
406            opts |= FsOptions::CACHE_SYMLINKS;
407        }
408        // TODO: figure out what dawrin options make sense.
409
410        Ok(opts)
411    }
412
413    #[tracing::instrument(skip_all, fields(rq.inode = inode))]
414    fn getattr(
415        &self,
416        _ctx: &Context,
417        inode: Self::Inode,
418        _handle: Option<Self::Handle>,
419    ) -> io::Result<(stat64, Duration)> {
420        let attr = if inode == ROOT_ID {
421            ROOT_FILE_ATTR
422        } else {
423            self.inode_data_to_attr(
424                self.inode_tracker
425                    .read()
426                    .get(inode)
427                    .ok_or_else(|| io::Error::from_raw_os_error(libc::ENOENT))?
428                    .as_ref(),
429                inode,
430            )
431        };
432
433        Ok((attr.into(), Duration::MAX))
434    }
435
436    #[tracing::instrument(skip_all, fields(rq.parent_inode = parent, rq.name = ?name))]
437    fn lookup(
438        &self,
439        _ctx: &Context,
440        parent: Self::Inode,
441        name: &std::ffi::CStr,
442    ) -> io::Result<Entry> {
443        debug!("lookup");
444
445        // convert the CStr to a PathComponent
446        // If it can't be converted, we definitely don't have anything here.
447        let name: PathComponent = name.try_into().map_err(|_| std::io::ErrorKind::NotFound)?;
448
449        // This goes from a parent inode to a node.
450        let (ino, inode_data) = if parent == ROOT_ID {
451            // If the parent is [ROOT_ID], we need to check [self.root_nodes] (fetching from a [RootNode] provider if needed)
452            self.name_in_root_to_ino_and_data(&name)?
453        } else {
454            // else the parent must be a directory, otherwise we would never come up with this request.
455            // Lookup the parent in [self.inode_tracker] (which must be a [InodeData::Directory]), and find the child with that name.
456            let (parent_digest, children) = self.get_directory_children(parent)?;
457
458            Span::current().record("directory.digest", parent_digest.to_string());
459            // Search for that name in the list of children (which we know are sorted)
460            // and return the FileAttrs.
461            let idx = children
462                .binary_search_by_key(&&name, |(_, n, _)| n)
463                .map_err(|_| {
464                    // Child not found, return ENOENT.
465                    io::Error::from_raw_os_error(libc::ENOENT)
466                })?;
467
468            let (child_ino, _, child_node) = &children[idx];
469
470            // Reply with the file attributes for the child,
471            (*child_ino, Arc::new(InodeData::from_node(child_node)))
472        };
473
474        debug!(inode_data=?&inode_data, ino=ino, "Some");
475
476        let attr = self.inode_data_to_attr(
477            self.inode_tracker
478                .read()
479                .get(ino)
480                .ok_or_else(|| io::Error::from_raw_os_error(libc::ENOENT))?
481                .as_ref(),
482            ino,
483        );
484        Ok(attr_to_fuse_entry(attr))
485    }
486
487    #[tracing::instrument(skip_all, fields(rq.inode = inode))]
488    fn opendir(
489        &self,
490        _ctx: &Context,
491        inode: Self::Inode,
492        _flags: u32,
493    ) -> io::Result<(Option<Self::Handle>, OpenOptions)> {
494        // In case opendir on the root is called, we provide the handle, as re-entering that listing is expensive.
495        // For all other directory inodes we just let readdir take care of it.
496        if inode == ROOT_ID {
497            if !self.settings.list_root {
498                return Err(io::Error::from_raw_os_error(libc::EPERM)); // same error code as ipfs/kubo
499            }
500
501            let root_nodes_provider = self.root_nodes_provider.clone();
502            let stream = self
503                .tokio_handle
504                .block_on(async move { root_nodes_provider.list().enumerate().boxed() });
505
506            // Put the stream into [self.dir_handles].
507            // TODO: this will overflow after 2**64 operations,
508            // which is fine for now.
509            // See https://cl.tvl.fyi/c/depot/+/8834/comment/a6684ce0_d72469d1
510            // for the discussion on alternatives.
511            let dh = self.next_dir_handle.fetch_add(1, Ordering::SeqCst);
512
513            self.dir_handles
514                .write()
515                .insert(dh, (Span::current(), Arc::new(Mutex::new(stream))));
516
517            return Ok((Some(dh), OpenOptions::NONSEEKABLE));
518        }
519
520        let mut opts = OpenOptions::empty();
521
522        opts |= OpenOptions::KEEP_CACHE;
523        #[cfg(target_os = "linux")]
524        {
525            opts |= OpenOptions::CACHE_DIR;
526        }
527        // allow caching this directory contents, don't invalidate on open
528        Ok((None, opts))
529    }
530
531    #[tracing::instrument(skip_all, fields(rq.inode = inode, rq.handle = handle, rq.offset = offset), parent = self.dir_handles.read().get(&handle).and_then(|x| x.0.id()))]
532    fn readdir(
533        &self,
534        _ctx: &Context,
535        inode: Self::Inode,
536        handle: Self::Handle,
537        _size: u32,
538        offset: u64,
539        add_entry: &mut dyn FnMut(fuse_backend_rs::api::filesystem::DirEntry) -> io::Result<usize>,
540    ) -> io::Result<()> {
541        debug!("readdir");
542
543        if inode == ROOT_ID {
544            if !self.settings.list_root {
545                return Err(io::Error::from_raw_os_error(libc::EPERM)); // same error code as ipfs/kubo
546            }
547
548            // get the stream from [self.dir_handles]
549            let dir_handles = self.dir_handles.read();
550            let (_span, stream) = dir_handles.get(&handle).ok_or_else(|| {
551                warn!("dir handle {} unknown", handle);
552                io::Error::from_raw_os_error(libc::EIO)
553            })?;
554
555            let mut stream = stream
556                .lock()
557                .map_err(|_| io::Error::other("mutex poisoned"))?;
558
559            while let Some((i, n)) = self.tokio_handle.block_on(async { stream.next().await }) {
560                let (name, node) = n.map_err(|e| {
561                    warn!("failed to retrieve root node: {}", e);
562                    io::Error::from_raw_os_error(libc::EIO)
563                })?;
564
565                // obtain the inode, or allocate a new one.
566                let ino = self.get_inode_for_root_name(&name).unwrap_or_else(|| {
567                    // insert the (sparse) inode data and register in
568                    // self.root_nodes.
569                    let ino = self.inode_tracker.write().put(InodeData::from_node(&node));
570                    self.root_nodes.write().insert(name.clone(), ino);
571                    ino
572                });
573
574                let written = add_entry(fuse_backend_rs::api::filesystem::DirEntry {
575                    ino,
576                    offset: offset + (i as u64) + 1,
577                    type_: node_to_fuse_type(&node),
578                    name: name.as_ref(),
579                })?;
580                // If the buffer is full, add_entry will return `Ok(0)`.
581                if written == 0 {
582                    break;
583                }
584            }
585            return Ok(());
586        }
587
588        // Non root-node case: lookup the children, or return an error if it's not a directory.
589        let (parent_digest, children) = self.get_directory_children(inode)?;
590        Span::current().record("directory.digest", parent_digest.to_string());
591
592        for (i, (ino, child_name, child_node)) in
593            children.into_iter().skip(offset as usize).enumerate()
594        {
595            // the second parameter will become the "offset" parameter on the next call.
596            let written = add_entry(fuse_backend_rs::api::filesystem::DirEntry {
597                ino,
598                offset: offset + (i as u64) + 1,
599                type_: node_to_fuse_type(&child_node),
600                name: child_name.as_ref(),
601            })?;
602            // If the buffer is full, add_entry will return `Ok(0)`.
603            if written == 0 {
604                break;
605            }
606        }
607
608        Ok(())
609    }
610
611    #[tracing::instrument(skip_all, fields(rq.inode = inode, rq.handle = handle), parent = self.dir_handles.read().get(&handle).and_then(|x| x.0.id()))]
612    fn readdirplus(
613        &self,
614        _ctx: &Context,
615        inode: Self::Inode,
616        handle: Self::Handle,
617        _size: u32,
618        offset: u64,
619        add_entry: &mut dyn FnMut(
620            fuse_backend_rs::api::filesystem::DirEntry,
621            Entry,
622        ) -> io::Result<usize>,
623    ) -> io::Result<()> {
624        debug!("readdirplus");
625
626        if inode == ROOT_ID {
627            if !self.settings.list_root {
628                return Err(io::Error::from_raw_os_error(libc::EPERM)); // same error code as ipfs/kubo
629            }
630
631            // get the stream from [self.dir_handles]
632            let dir_handles = self.dir_handles.read();
633            let (_span, stream) = dir_handles.get(&handle).ok_or_else(|| {
634                warn!("dir handle {} unknown", handle);
635                io::Error::from_raw_os_error(libc::EIO)
636            })?;
637
638            let mut stream = stream
639                .lock()
640                .map_err(|_| io::Error::other("mutex poisoned"))?;
641
642            while let Some((i, n)) = self.tokio_handle.block_on(async { stream.next().await }) {
643                let (name, node) = n.map_err(|e| {
644                    warn!("failed to retrieve root node: {}", e);
645                    io::Error::from_raw_os_error(libc::EPERM)
646                })?;
647
648                let inode_data = InodeData::from_node(&node);
649
650                // obtain the inode, or allocate a new one.
651                let ino = self.get_inode_for_root_name(&name).unwrap_or_else(|| {
652                    // insert the (sparse) inode data and register in
653                    // self.root_nodes.
654                    let ino = self.inode_tracker.write().put(inode_data.clone());
655                    self.root_nodes.write().insert(name.clone(), ino);
656                    ino
657                });
658
659                let written = add_entry(
660                    fuse_backend_rs::api::filesystem::DirEntry {
661                        ino,
662                        offset: offset + (i as u64) + 1,
663                        type_: node_to_fuse_type(&node),
664                        name: name.as_ref(),
665                    },
666                    attr_to_fuse_entry(self.inode_data_to_attr(&inode_data, ino)),
667                )?;
668                // If the buffer is full, add_entry will return `Ok(0)`.
669                if written == 0 {
670                    break;
671                }
672            }
673            return Ok(());
674        }
675
676        // Non root-node case: lookup the children, or return an error if it's not a directory.
677        let (parent_digest, children) = self.get_directory_children(inode)?;
678        Span::current().record("directory.digest", parent_digest.to_string());
679
680        for (i, (ino, name, child_node)) in children.into_iter().skip(offset as usize).enumerate() {
681            let inode_data = InodeData::from_node(&child_node);
682
683            // the second parameter will become the "offset" parameter on the next call.
684            let written = add_entry(
685                fuse_backend_rs::api::filesystem::DirEntry {
686                    ino,
687                    offset: offset + (i as u64) + 1,
688                    type_: node_to_fuse_type(&child_node),
689                    name: name.as_ref(),
690                },
691                attr_to_fuse_entry(self.inode_data_to_attr(&inode_data, ino)),
692            )?;
693            // If the buffer is full, add_entry will return `Ok(0)`.
694            if written == 0 {
695                break;
696            }
697        }
698
699        Ok(())
700    }
701
702    #[tracing::instrument(skip_all, fields(rq.inode = inode, rq.handle = handle), parent = self.dir_handles.read().get(&handle).and_then(|x| x.0.id()))]
703    fn releasedir(
704        &self,
705        _ctx: &Context,
706        inode: Self::Inode,
707        _flags: u32,
708        handle: Self::Handle,
709    ) -> io::Result<()> {
710        if inode == ROOT_ID {
711            // drop the stream.
712            if let Some(stream) = self.dir_handles.write().remove(&handle) {
713                // drop it, which will close it.
714                drop(stream)
715            } else {
716                warn!("dir handle not found");
717            }
718        }
719
720        Ok(())
721    }
722
723    #[tracing::instrument(skip_all, fields(rq.inode = inode))]
724    fn open(
725        &self,
726        _ctx: &Context,
727        inode: Self::Inode,
728        _flags: u32,
729        _fuse_flags: u32,
730    ) -> io::Result<(Option<Self::Handle>, OpenOptions, Option<u32>)> {
731        if inode == ROOT_ID {
732            return Err(io::Error::from_raw_os_error(libc::ENOSYS));
733        }
734
735        // lookup the inode
736        match *self.inode_tracker.read().get(inode).unwrap() {
737            // read is invalid on non-files.
738            InodeData::Directory(..) | InodeData::Symlink(_) => {
739                warn!("is directory");
740                Err(io::Error::from_raw_os_error(libc::EISDIR))
741            }
742            InodeData::Regular(ref blob_digest, _blob_size, _) => {
743                Span::current().record("blob.digest", blob_digest.to_string());
744
745                match self
746                    .tokio_handle
747                    .block_on(async { self.blob_service.open_read(blob_digest).await })
748                {
749                    Ok(None) => {
750                        warn!("blob not found");
751                        Err(io::Error::from_raw_os_error(libc::EIO))
752                    }
753                    Err(e) => {
754                        warn!(e=?e, "error opening blob");
755                        Err(io::Error::from_raw_os_error(libc::EIO))
756                    }
757                    Ok(Some(blob_reader)) => {
758                        // get a new file handle
759                        // TODO: this will overflow after 2**64 operations,
760                        // which is fine for now.
761                        // See https://cl.tvl.fyi/c/depot/+/8834/comment/a6684ce0_d72469d1
762                        // for the discussion on alternatives.
763                        let fh = self.next_file_handle.fetch_add(1, Ordering::SeqCst);
764
765                        self.file_handles
766                            .write()
767                            .insert(fh, (Span::current(), Arc::new(Mutex::new(blob_reader))));
768
769                        Ok((
770                            Some(fh),
771                            // Don't invalidate the data cache on open.
772                            OpenOptions::KEEP_CACHE,
773                            None,
774                        ))
775                    }
776                }
777            }
778        }
779    }
780
781    #[tracing::instrument(skip_all, fields(rq.inode = inode, rq.handle = handle), parent = self.file_handles.read().get(&handle).and_then(|x| x.0.id()))]
782    fn release(
783        &self,
784        _ctx: &Context,
785        inode: Self::Inode,
786        _flags: u32,
787        handle: Self::Handle,
788        _flush: bool,
789        _flock_release: bool,
790        _lock_owner: Option<u64>,
791    ) -> io::Result<()> {
792        match self.file_handles.write().remove(&handle) {
793            // drop the blob reader, which will close it.
794            Some(blob_reader) => drop(blob_reader),
795            None => {
796                // These might already be dropped if a read error occurred.
797                warn!("file handle not found");
798            }
799        }
800
801        Ok(())
802    }
803
804    #[tracing::instrument(skip_all, fields(rq.inode = inode, rq.handle = handle, rq.offset = offset, rq.size = size), parent = self.file_handles.read().get(&handle).and_then(|x| x.0.id()))]
805    fn read(
806        &self,
807        _ctx: &Context,
808        inode: Self::Inode,
809        handle: Self::Handle,
810        w: &mut dyn fuse_backend_rs::api::filesystem::ZeroCopyWriter,
811        size: u32,
812        offset: u64,
813        _lock_owner: Option<u64>,
814        _flags: u32,
815    ) -> io::Result<usize> {
816        debug!("read");
817
818        // We need to take out the blob reader from self.file_handles, so we can
819        // interact with it in the separate task.
820        // On success, we pass it back out of the task, so we can put it back in self.file_handles.
821        let (_span, blob_reader) = self
822            .file_handles
823            .read()
824            .get(&handle)
825            .ok_or_else(|| {
826                warn!("file handle {} unknown", handle);
827                io::Error::from_raw_os_error(libc::EIO)
828            })
829            .cloned()?;
830
831        let mut blob_reader = blob_reader
832            .lock()
833            .map_err(|_| io::Error::other("mutex poisoned"))?;
834
835        let buf = self.tokio_handle.block_on(async move {
836            // seek to the offset specified, which is relative to the start of the file.
837            let pos = blob_reader
838                .seek(io::SeekFrom::Start(offset))
839                .await
840                .map_err(|e| {
841                    warn!("failed to seek to offset {}: {}", offset, e);
842                    io::Error::from_raw_os_error(libc::EIO)
843                })?;
844
845            debug_assert_eq!(offset, pos);
846
847            // As written in the fuse docs, read should send exactly the number
848            // of bytes requested except on EOF or error.
849
850            let mut buf: Vec<u8> = Vec::with_capacity(size as usize);
851
852            // copy things from the internal buffer into buf to fill it till up until size
853            tokio::io::copy(&mut blob_reader.as_mut().take(size as u64), &mut buf).await?;
854
855            Ok::<_, std::io::Error>(buf)
856        })?;
857
858        // We cannot use w.write() here, we're required to call write multiple
859        // times until we wrote the entirety of the buffer (which is `size`, except on EOF).
860        let buf_len = buf.len();
861        let bytes_written = io::copy(&mut Cursor::new(buf), w)?;
862        if bytes_written != buf_len as u64 {
863            error!(bytes_written=%bytes_written, "unable to write all of buf to kernel");
864            return Err(io::Error::from_raw_os_error(libc::EIO));
865        }
866
867        Ok(bytes_written as usize)
868    }
869
870    #[tracing::instrument(skip_all, fields(rq.inode = inode))]
871    fn readlink(&self, _ctx: &Context, inode: Self::Inode) -> io::Result<Vec<u8>> {
872        if inode == ROOT_ID {
873            return Err(io::Error::from_raw_os_error(libc::ENOSYS));
874        }
875
876        // lookup the inode
877        match *self.inode_tracker.read().get(inode).unwrap() {
878            InodeData::Directory(..) | InodeData::Regular(..) => {
879                Err(io::Error::from_raw_os_error(libc::EINVAL))
880            }
881            InodeData::Symlink(ref target) => Ok(target.to_vec()),
882        }
883    }
884
885    #[tracing::instrument(skip_all, fields(rq.inode = inode, name=?name))]
886    fn getxattr(
887        &self,
888        _ctx: &Context,
889        inode: Self::Inode,
890        name: &CStr,
891        size: u32,
892    ) -> io::Result<GetxattrReply> {
893        if !self.settings.show_xattr {
894            return Err(io::Error::from_raw_os_error(libc::ENOSYS));
895        }
896
897        // Peek at the inode requested, and construct the response.
898        let digest_str = match *self
899            .inode_tracker
900            .read()
901            .get(inode)
902            .ok_or_else(|| io::Error::from_raw_os_error(libc::ENODATA))?
903        {
904            InodeData::Directory(DirectoryInodeData::Sparse(ref digest, _))
905            | InodeData::Directory(DirectoryInodeData::Populated(ref digest, _))
906                if name.to_bytes() == XATTR_NAME_DIRECTORY_DIGEST =>
907            {
908                digest.to_string()
909            }
910            InodeData::Regular(ref digest, _, _) if name.to_bytes() == XATTR_NAME_BLOB_DIGEST => {
911                digest.to_string()
912            }
913            _ => {
914                return Err(io::Error::from_raw_os_error(libc::ENODATA));
915            }
916        };
917
918        if size == 0 {
919            Ok(GetxattrReply::Count(digest_str.len() as u32))
920        } else if size < digest_str.len() as u32 {
921            Err(io::Error::from_raw_os_error(libc::ERANGE))
922        } else {
923            Ok(GetxattrReply::Value(digest_str.into_bytes()))
924        }
925    }
926
927    #[tracing::instrument(skip_all, fields(rq.inode = inode))]
928    fn listxattr(
929        &self,
930        _ctx: &Context,
931        inode: Self::Inode,
932        size: u32,
933    ) -> io::Result<ListxattrReply> {
934        if !self.settings.show_xattr {
935            return Err(io::Error::from_raw_os_error(libc::ENOSYS));
936        }
937
938        // determine the (\0-terminated list) to of xattr keys present, depending on the type of the inode.
939        let xattrs_names = {
940            let mut out = Vec::new();
941            if let Some(inode_data) = self.inode_tracker.read().get(inode) {
942                match *inode_data {
943                    InodeData::Directory(_) => {
944                        out.extend_from_slice(XATTR_NAME_DIRECTORY_DIGEST);
945                        out.push_byte(b'\x00');
946                    }
947                    InodeData::Regular(..) => {
948                        out.extend_from_slice(XATTR_NAME_BLOB_DIGEST);
949                        out.push_byte(b'\x00');
950                    }
951                    _ => {}
952                }
953            }
954            out
955        };
956
957        if size == 0 {
958            Ok(ListxattrReply::Count(xattrs_names.len() as u32))
959        } else if size < xattrs_names.len() as u32 {
960            Err(io::Error::from_raw_os_error(libc::ERANGE))
961        } else {
962            Ok(ListxattrReply::Names(xattrs_names.to_vec()))
963        }
964    }
965}
snix_castore/fs/mod.rs

snix_castore/fs/
mod.rs
