Blocktree/crates/btlib/src/blocktree.rs

pub use private::{Blocktree, ModeAuthorizer};

mod private {
    use btserde::{read_from, write_to};
    use fuse_backend_rs::{
        abi::fuse_abi::{stat64, statvfs64, CreateIn},
        api::filesystem::{
            Context, DirEntry as FuseDirEntry, Entry, FileSystem, FsOptions, OpenOptions,
            SetattrValid,
        },
    };
    use log::{debug, error, warn};
    use serde::{Deserialize, Serialize};
    use std::{
        collections::hash_map::{self, HashMap},
        ffi::CStr,
        fmt::{Display, Formatter},
        fs::File,
        io::{self, SeekFrom, Write},
        path::{Path, PathBuf},
        sync::{
            atomic::{AtomicU64, Ordering},
            RwLock,
        },
        time::Duration,
    };

    use crate::{
        bterr,
        crypto::Creds,
        error::{DisplayErr, IoErr},
        Block, BlockError, BlockMeta, BlockOpenOptions, BlockPath, BlockRecord, BoxInIoErr,
        DirEntry, DirEntryKind, Directory, Epoch, Result,
    };

    type Inode = u64;
    type Handle = u64;

    #[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
    pub enum Error {
        NotOpen(Inode),
        InvalidHandle { handle: u64, inode: u64 },
        NoHandlesAvailable(Inode),
        InodeNotFound(Inode),
    }

    impl Display for Error {
        fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
            match self {
                Error::NotOpen(inode) => write!(f, "inode {inode} is not open"),
                Error::InvalidHandle { handle, inode } => {
                    write!(f, "invalid handle {handle} for inode {inode}")
                }
                Error::NoHandlesAvailable(inode) => {
                    write!(f, "no handles are available for inode {inode}")
                }
                Error::InodeNotFound(inode) => write!(f, "inode {inode} could not be found"),
            }
        }
    }

    impl std::error::Error for Error {}

    #[repr(u64)]
    pub enum SpecInodes {
        RootDir = 1,
        Sb = 2,
        FirstFree = 11,
    }

    impl From<SpecInodes> for Inode {
        fn from(special: SpecInodes) -> Self {
            special as Inode
        }
    }

    #[derive(Debug, Clone, Hash, PartialEq, Eq, Serialize, Deserialize)]
    #[repr(u32)] // This type needs to match `libc::mode_t`.
    /// The type of a file (regular, directory, etc).
    enum FileType {
        /// Directory.
        Dir = libc::S_IFDIR,
        /// Regular file.
        Reg = libc::S_IFREG,
    }

    impl FileType {
        /// Returns the underlying mode bits for this file type.
        fn value(self) -> libc::mode_t {
            self as libc::mode_t
        }

        /// Attempts to convert the given mode bits into a `FileType` enum value.
        fn from_value(value: libc::mode_t) -> Result<Self> {
            if (value & libc::S_IFDIR) != 0 {
                return Ok(FileType::Dir);
            }
            if (value & libc::S_IFREG) != 0 {
                return Ok(FileType::Reg);
            }
            Err(bterr!("unknown file type: 0o{value:0o}"))
        }

        fn dir_entry_kind(self) -> DirEntryKind {
            match self {
                Self::Dir => DirEntryKind::Directory,
                Self::Reg => DirEntryKind::File,
            }
        }
    }

    impl From<FileType> for libc::mode_t {
        fn from(file_type: FileType) -> Self {
            file_type.value()
        }
    }

    impl TryFrom<libc::mode_t> for FileType {
        type Error = crate::Error;
        fn try_from(value: libc::mode_t) -> Result<Self> {
            Self::from_value(value)
        }
    }

    impl From<FileType> for DirEntryKind {
        fn from(value: FileType) -> Self {
            value.dir_entry_kind()
        }
    }

    trait SeekFromExt {
        /// Converts a C-style `(whence, offset)` pair into a [SeekFrom] enum value.
        /// See the POSIX man page of `lseek` for more details.
        fn whence_offset(whence: u32, offset: u64) -> io::Result<SeekFrom> {
            let whence = whence as i32;
            match whence {
                libc::SEEK_SET => Ok(SeekFrom::Start(offset)),
                libc::SEEK_CUR => Ok(SeekFrom::Current(offset as i64)),
                libc::SEEK_END => Ok(SeekFrom::End(offset as i64)),
                _ => Err(io::Error::new(
                    io::ErrorKind::InvalidInput,
                    "`whence` was not one of `libc::{SEEK_SET, SEEK_CUR, SEEK_END}`",
                )),
            }
        }
    }

    impl SeekFromExt for SeekFrom {}

    /// This type provides context for an authorization decision as to whether a given process will
    /// be allowed to access a block.
    pub struct AuthzContext<'a> {
        /// The user ID of the process being authorized.
        pub uid: u32,
        /// The group ID of the process being authorized.
        pub gid: u32,
        /// The process ID of the process being authorized.
        pub pid: libc::pid_t,
        /// A reference to the metadata of a block, the access to which is being authorized.
        pub meta: &'a BlockMeta,
    }

    impl<'a> AuthzContext<'a> {
        pub fn new(ctx: &Context, meta: &'a BlockMeta) -> AuthzContext<'a> {
            AuthzContext {
                uid: ctx.uid,
                gid: ctx.gid,
                pid: ctx.pid,
                meta,
            }
        }
    }

    /// A trait for types which can render authorization decisions.
    pub trait Authorizer {
        /// Returns [Ok] if read authorization is granted, and [Err] otherwise.
        fn can_read<'a>(&self, ctx: &AuthzContext<'a>) -> io::Result<()>;
        /// Returns [Ok] if write authorization is granted, and [Err] otherwise.
        fn can_write<'a>(&self, ctx: &AuthzContext<'a>) -> io::Result<()>;
        /// Returns [Ok] if execute authorization is granted, and [Err] otherwise.
        fn can_exec<'a>(&self, ctx: &AuthzContext<'a>) -> io::Result<()>;
    }

    trait AuthorizerExt: Authorizer {
        fn read_allowed(&self, ctx: &Context, meta: &BlockMeta) -> io::Result<()> {
            Authorizer::can_read(self, &AuthzContext::new(ctx, meta))
        }

        fn write_allowed(&self, ctx: &Context, meta: &BlockMeta) -> io::Result<()> {
            Authorizer::can_write(self, &AuthzContext::new(ctx, meta))
        }

        fn exec_allowed(&self, ctx: &Context, meta: &BlockMeta) -> io::Result<()> {
            Authorizer::can_exec(self, &AuthzContext::new(ctx, meta))
        }
    }

    impl<T: Authorizer> AuthorizerExt for T {}

    /// A particularly simple authorizer that just looks at the mode bits in the block metadata
    /// to make authorization decisions.
    pub struct ModeAuthorizer {}

    impl ModeAuthorizer {
        fn authorize(mode: u32, mask: u32, denied_msg: &str) -> io::Result<()> {
            if (mode & mask) != 0 {
                Ok(())
            } else {
                Err(io::Error::new(io::ErrorKind::PermissionDenied, denied_msg))
            }
        }

        fn user_is_root(ctx: &AuthzContext<'_>) -> bool {
            ctx.uid == 0
        }
    }

    impl Authorizer for ModeAuthorizer {
        fn can_read<'a>(&self, ctx: &AuthzContext<'a>) -> io::Result<()> {
            if Self::user_is_root(ctx) {
                return Ok(());
            }
            let secrets = ctx.meta.body.secrets()?;
            let mask = (libc::S_IRUSR * (secrets.uid == ctx.uid) as u32)
                | (libc::S_IRGRP * (secrets.gid == ctx.gid) as u32)
                | libc::S_IROTH;
            Self::authorize(secrets.mode, mask, "read access denied")
        }

        fn can_write<'a>(&self, ctx: &AuthzContext<'a>) -> io::Result<()> {
            if Self::user_is_root(ctx) {
                return Ok(());
            }
            let secrets = ctx.meta.body.secrets()?;
            let mask = (libc::S_IWUSR * (secrets.uid == ctx.uid) as u32)
                | (libc::S_IWGRP * (secrets.gid == ctx.gid) as u32)
                | libc::S_IWOTH;
            Self::authorize(secrets.mode, mask, "write access denied")
        }

        fn can_exec<'a>(&self, ctx: &AuthzContext<'a>) -> io::Result<()> {
            if Self::user_is_root(ctx) {
                return Ok(());
            }
            let secrets = ctx.meta.body.secrets()?;
            let mask = (libc::S_IXUSR * (secrets.uid == ctx.uid) as u32)
                | (libc::S_IXGRP * (secrets.gid == ctx.gid) as u32)
                | libc::S_IXOTH;
            Self::authorize(secrets.mode, mask, "exec access denied")
        }
    }

    enum HandleValue {
        File {
            block: RwLock<Box<dyn Block>>,
        },
        Directory {
            dir: Directory,
            block: RwLock<Box<dyn Block>>,
        },
    }

    impl HandleValue {
        fn new(block: Box<dyn Block>) -> HandleValue {
            HandleValue::File {
                block: RwLock::new(block),
            }
        }

        fn convert_to_dir(self) -> io::Result<HandleValue> {
            let lock = self.take_block();
            let dir = {
                let mut guard = lock.write().display_err()?;
                guard.seek(SeekFrom::Start(0))?;
                read_from(&mut *guard)?
            };
            Ok(HandleValue::Directory { dir, block: lock })
        }

        fn take_block(self) -> RwLock<Box<dyn Block>> {
            match self {
                Self::File { block, .. } => block,
                Self::Directory { block, .. } => block,
            }
        }

        fn block(&self) -> &RwLock<Box<dyn Block>> {
            match self {
                Self::File { block, .. } => block,
                Self::Directory { block, .. } => block,
            }
        }

        fn block_mut(&mut self) -> io::Result<&mut Box<dyn Block>> {
            let lock = match self {
                Self::File { block, .. } => block,
                Self::Directory { block, .. } => block,
            };
            lock.get_mut().display_err().io_err()
        }

        fn access_block<T, F: FnOnce(&Box<dyn Block>) -> io::Result<T>>(
            &self,
            cb: F,
        ) -> io::Result<T> {
            let guard = self.block().read().display_err()?;
            cb(&guard)
        }

        fn access_block_mut<T, F: FnOnce(&mut Box<dyn Block>) -> io::Result<T>>(
            &self,
            cb: F,
        ) -> io::Result<T> {
            let mut guard = self.block().write().display_err()?;
            cb(&mut guard)
        }

        fn directory(&self) -> io::Result<&Directory> {
            match self {
                Self::Directory { dir, .. } => Ok(dir),
                _ => Err(io::Error::new(
                    io::ErrorKind::Other,
                    "handle is not for a directory",
                )),
            }
        }
    }

    struct InodeTableValue {
        handle_values: HashMap<Handle, HandleValue>,
        next_handle: Handle,
        unclaimed_handles: Vec<Handle>,
        lookup_count: u64,
        delete: bool,
    }

    impl InodeTableValue {
        /// If more than this number of unclaimed blocks are open, then blocks are closed until
        /// only this number remain open.
        const UNCLAIMED_HANDLE_LIMIT: usize = 3;

        fn new(block: Box<dyn Block>) -> InodeTableValue {
            const FIRST_HANDLE: Handle = 1;
            let mut handles = HashMap::with_capacity(1);
            handles.insert(FIRST_HANDLE, HandleValue::new(block));
            Self {
                handle_values: handles,
                next_handle: FIRST_HANDLE + 1,
                lookup_count: 1,
                unclaimed_handles: vec![FIRST_HANDLE],
                delete: false,
            }
        }

        fn invalid_handle_err(handle: Handle) -> io::Error {
            io::Error::new(io::ErrorKind::Other, format!("invalid handle {handle}"))
        }

        fn value(&self, handle: Handle) -> io::Result<&HandleValue> {
            self.handle_values
                .get(&handle)
                .ok_or_else(|| Self::invalid_handle_err(handle))
        }

        fn value_mut(&mut self, handle: Handle) -> io::Result<&mut HandleValue> {
            self.handle_values
                .get_mut(&handle)
                .ok_or_else(|| Self::invalid_handle_err(handle))
        }

        fn block_mut(&mut self, handle: Handle) -> io::Result<&mut Box<dyn Block>> {
            self.value_mut(handle)?.block_mut()
        }

        fn convert_to_dir(&mut self, handle: Handle) -> io::Result<()> {
            let value = self
                .handle_values
                .remove(&handle)
                .ok_or_else(|| Self::invalid_handle_err(handle))?;
            let value = value.convert_to_dir()?;
            self.handle_values.insert(handle, value);
            Ok(())
        }

        fn access_block_mut<T, F: FnOnce(&mut Box<dyn Block>) -> io::Result<T>>(
            &self,
            handle: Handle,
            cb: F,
        ) -> io::Result<T> {
            self.value(handle)?.access_block_mut(cb)
        }

        fn borrow_block<T, F: FnOnce(&mut Box<dyn Block>) -> io::Result<T>>(
            &mut self,
            cb: F,
        ) -> io::Result<T> {
            let handle = self
                .unclaimed_handles
                .last()
                .ok_or_else(|| bterr!("no handles available"))?;
            let value = self.value_mut(*handle)?;
            let block = value.block_mut()?;
            cb(block)
        }

        fn insert(&mut self, block: Box<dyn Block>) {
            let handle = self.next_handle;
            self.next_handle += 1;
            self.handle_values.insert(handle, HandleValue::new(block));
            self.unclaimed_handles.push(handle);
        }

        fn insert_then_get_mut(&mut self, block: Box<dyn Block>) -> &mut Box<dyn Block> {
            self.insert(block);
            let handle = self.unclaimed_handles.last().unwrap();
            self.block_mut(*handle).unwrap()
        }

        fn take_handle(&mut self) -> io::Result<Handle> {
            self.unclaimed_handles
                .pop()
                .ok_or_else(|| io::Error::new(io::ErrorKind::Other, "no unclaimed handles"))
        }

        fn give_handle(&mut self, handle: Handle) {
            self.unclaimed_handles.push(handle);
            while self.unclaimed_handles.len() > Self::UNCLAIMED_HANDLE_LIMIT {
                let handle = self.unclaimed_handles.pop().unwrap();
                self.handle_values.remove(&handle);
            }
        }

        /// Increments `lookup_count` by 1 and returns its current value.
        fn incr_lookup_count(&mut self) -> u64 {
            self.lookup_count += 1;
            self.lookup_count
        }

        /// Decrements `lookup_count` by `count` and returns its current value.
        fn decr_lookup_count(&mut self, count: u64) -> u64 {
            self.lookup_count -= count;
            self.lookup_count
        }
    }

    type InodeTable = HashMap<Inode, RwLock<InodeTableValue>>;
    type InodeTableEntry<'a> = hash_map::Entry<'a, Inode, RwLock<InodeTableValue>>;

    /// Structure for metadata about a blocktree.
    #[derive(Debug, PartialEq, Eq, Serialize, Deserialize)]
    struct Superblock {
        /// The generation number of the cluster this part of the blocktree is stored on.
        generation: u64,
        /// The next free inode available to the cluster.
        next_inode: u64,
    }

    /// Structure for managing the part of a blocktree which is stored in the local filesystem.
    pub struct Blocktree<C, A> {
        /// The path to the directory in the local filesystem where this blocktree is located.
        path: PathBuf,
        /// A map from inode numbers to their reference counts.
        inodes: RwLock<InodeTable>,
        /// The next inode that will be assigned to a new block.
        next_inode: AtomicU64,
        /// The generation number of this filesystem. This is the same for every other server in
        /// the same cluster.
        generation: u64,
        /// The credentials this blocktree instance will use for all cryptographic operations.
        creds: C,
        authorizer: A,
    }

    impl<C: Creds + 'static, A> Blocktree<C, A> {
        /// Creates a new empty blocktree at the given path.
        pub fn new_empty(
            btdir: PathBuf,
            generation: u64,
            creds: C,
            authorizer: A,
        ) -> Result<Blocktree<C, A>> {
            let root_block_path = creds
                .writecap()
                .ok_or(BlockError::MissingWritecap)?
                .root_block_path();

            // Initialize the superblock.
            let mut sb_block = Self::open_block(
                &btdir,
                SpecInodes::Sb.into(),
                creds.clone(),
                root_block_path.to_owned(),
            )?;
            let sb = Superblock {
                generation,
                next_inode: SpecInodes::FirstFree.into(),
            };
            write_to(&sb, &mut sb_block)?;
            sb_block.mut_meta_body().access_secrets(|secrets| {
                secrets.block_id.generation = generation;
                secrets.block_id.inode = SpecInodes::Sb.into();
                secrets.mode = FileType::Reg.value() | 0o666;
                secrets.uid = 0;
                secrets.gid = 0;
                secrets.nlink = 1;
                Ok(())
            })?;
            sb_block.flush()?;

            // Initialize the root directory.
            let mut root_block = Self::open_block(
                &btdir,
                SpecInodes::RootDir.into(),
                creds.clone(),
                root_block_path,
            )?;
            write_to(&Directory::new(), &mut root_block)?;
            root_block.mut_meta_body().access_secrets(|secrets| {
                secrets.block_id.generation = generation;
                secrets.block_id.inode = SpecInodes::RootDir.into();
                secrets.mode = FileType::Dir.value() | 0o777;
                secrets.uid = 0;
                secrets.gid = 0;
                secrets.nlink = 1;
                Ok(())
            })?;
            root_block.flush()?;

            Self::new(btdir, sb, sb_block, root_block, creds, authorizer)
        }

        /// Opens an existing blocktree stored at the given path.
        pub fn new_existing(btdir: PathBuf, creds: C, authorizer: A) -> Result<Blocktree<C, A>> {
            let root_block_path = creds
                .writecap()
                .ok_or(BlockError::MissingWritecap)?
                .root_block_path();
            let mut sb_block = Self::open_block(
                &btdir,
                SpecInodes::Sb.into(),
                creds.clone(),
                root_block_path.to_owned(),
            )?;
            let sb = read_from(&mut sb_block)?;
            let root_block = Self::open_block(
                &btdir,
                SpecInodes::RootDir.into(),
                creds.clone(),
                root_block_path,
            )?;
            Self::new(btdir, sb, sb_block, root_block, creds, authorizer)
        }

        fn new(
            btdir: PathBuf,
            sb: Superblock,
            sb_block: Box<dyn Block>,
            root_block: Box<dyn Block>,
            creds: C,
            authorizer: A,
        ) -> Result<Blocktree<C, A>> {
            let mut inodes = HashMap::with_capacity(1);
            inodes.insert(
                SpecInodes::Sb.into(),
                RwLock::new(InodeTableValue::new(sb_block)),
            );
            inodes.insert(
                SpecInodes::RootDir.into(),
                RwLock::new(InodeTableValue::new(root_block)),
            );
            Ok(Blocktree {
                path: btdir,
                inodes: RwLock::new(inodes),
                next_inode: AtomicU64::new(sb.next_inode),
                generation: sb.generation,
                creds,
                authorizer,
            })
        }

        /// Returns the path to the file storing the given inode's data.
        fn block_path<P: AsRef<Path>>(parent: P, inode: Inode) -> PathBuf {
            let group = inode / 0xFF;
            let mut path = PathBuf::new();
            path.push(parent);
            path.push(format!("{group:02x}"));
            path.push(format!("{inode:x}.blk"));
            path
        }

        fn open_block<P: AsRef<Path>>(
            btdir: P,
            inode: Inode,
            creds: C,
            block_path: BlockPath,
        ) -> io::Result<Box<dyn Block>> {
            let path = Self::block_path(&btdir, inode);
            let dir = path.ancestors().nth(1).unwrap();
            if let Err(err) = std::fs::create_dir(dir) {
                match err.kind() {
                    io::ErrorKind::AlreadyExists => (),
                    _ => return Err(err),
                }
            }
            let file = std::fs::OpenOptions::new()
                .read(true)
                .write(true)
                .create(true)
                .open(path)?;
            Self::open_block_file(file, creds, block_path)
        }

        fn open_block_file(
            file: File,
            creds: C,
            block_path: BlockPath,
        ) -> io::Result<Box<dyn Block>> {
            BlockOpenOptions::new()
                .with_creds(creds)
                .with_compress(false)
                .with_encrypt(true)
                .with_inner(file)
                .with_block_path(block_path)
                .open()
                .io_err()
        }

        /// Returns the [Err] variant containing the [io::Error] corresponding to [libc::ENOSYS].
        fn not_supported<T>() -> io::Result<T> {
            let err = io::Error::from_raw_os_error(libc::ENOSYS);
            debug!("{err}");
            Err(err)
        }

        fn access_entry<T, F: FnOnce(InodeTableEntry) -> io::Result<T>>(
            &self,
            inode: Inode,
            cb: F,
        ) -> io::Result<T> {
            let mut inodes = self.inodes.write().display_err()?;
            let entry = inodes.entry(inode);
            cb(entry)
        }

        fn access_value<T, F: FnOnce(&InodeTableValue) -> io::Result<T>>(
            &self,
            inode: Inode,
            cb: F,
        ) -> io::Result<T> {
            let inodes = self.inodes.read().display_err()?;
            let guard = inodes
                .get(&inode)
                .ok_or_else(|| bterr!(Error::NotOpen(inode)))?
                .read()
                .display_err()?;
            cb(&guard)
        }

        fn access_value_mut<T, F: FnOnce(&mut InodeTableValue) -> io::Result<T>>(
            &self,
            inode: Inode,
            cb: F,
        ) -> io::Result<T> {
            let inodes = self.inodes.read().display_err()?;
            let mut guard = inodes
                .get(&inode)
                .ok_or_else(|| bterr!(Error::NotOpen(inode)))?
                .write()
                .display_err()?;
            cb(&mut guard)
        }

        fn access_block_mut<T, F: FnOnce(&mut Box<dyn Block>) -> io::Result<T>>(
            &self,
            inode: Inode,
            handle: Handle,
            cb: F,
        ) -> io::Result<T> {
            self.access_value(inode, |value| value.access_block_mut(handle, cb))
        }

        fn access_meta<T, F: FnOnce(&BlockMeta) -> io::Result<T>>(
            &self,
            inode: Inode,
            cb: F,
        ) -> io::Result<T> {
            self.access_value(inode, |value| {
                let handle_value = value
                    .handle_values
                    .values()
                    .next()
                    .ok_or_else(|| bterr!(Error::NotOpen(inode)))?;
                // Because we're using any of the meta data structs we need to ensure that any
                // modification of meta data is performed on all open blocks.
                handle_value.access_block(|block| cb(block.meta()))
            })
        }

        fn borrow_block<T, F: FnOnce(&mut Box<dyn Block>) -> io::Result<T>>(
            &self,
            inode: Inode,
            cb: F,
        ) -> io::Result<T> {
            self.access_value_mut(inode, |value| {
                let block = match value.unclaimed_handles.last() {
                    Some(handle) => value.handle_values.get_mut(handle).unwrap().block_mut()?,
                    None => {
                        let block_path = value
                            .handle_values
                            .values()
                            .next()
                            .ok_or_else(|| bterr!(Error::NotOpen(inode)))?
                            .access_block(|block| Ok(block.meta_body().path.clone()))?;
                        let block =
                            Self::open_block(&self.path, inode, self.creds.clone(), block_path)?;
                        value.insert_then_get_mut(block)
                    }
                };
                cb(block)
            })
        }

        fn take_handle_if_ok<T, F: FnOnce(Handle, &mut Box<dyn Block>) -> io::Result<T>>(
            &self,
            inode: Inode,
            cb: F,
        ) -> io::Result<T> {
            self.access_value_mut(inode, |value| {
                let handle = value
                    .take_handle()
                    .map_err(|_| bterr!(Error::NoHandlesAvailable(inode)))?;
                let block = value.block_mut(handle)?;
                let result = cb(handle, block);
                if result.is_err() {
                    value.give_handle(handle);
                }
                result
            })
        }

        fn open_value<T, F: FnOnce(&mut InodeTableValue) -> io::Result<T>>(
            &self,
            inode: Inode,
            block_path: BlockPath,
            cb: F,
        ) -> io::Result<T> {
            self.access_entry(inode, |entry| match entry {
                InodeTableEntry::Vacant(entry) => {
                    let block =
                        Self::open_block(&self.path, inode, self.creds.clone(), block_path)?;
                    let mut value = InodeTableValue::new(block);
                    let result = cb(&mut value);
                    entry.insert(RwLock::new(value));
                    result
                }
                InodeTableEntry::Occupied(mut entry) => {
                    let value = entry.get_mut().get_mut().display_err()?;
                    if value.unclaimed_handles.is_empty() {
                        let block =
                            Self::open_block(&self.path, inode, self.creds.clone(), block_path)?;
                        value.insert(block);
                    }
                    cb(value)
                }
            })
        }

        fn open_then_take_handle<T, F: FnOnce(Handle, &mut InodeTableValue) -> io::Result<T>>(
            &self,
            inode: Inode,
            block_path: BlockPath,
            cb: F,
        ) -> io::Result<T> {
            self.open_value(inode, block_path, |value| {
                let handle = value.take_handle().unwrap();
                cb(handle, value)
            })
        }

        fn inode_forget(&self, inode: Inode, count: u64) -> io::Result<()> {
            let mut inodes = self.inodes.write().display_err()?;
            let lookup_count = {
                let inode_lock = match inodes.get_mut(&inode) {
                    Some(inode_lock) => inode_lock,
                    None => {
                        warn!("an attempt was made to forget non-existent inode {inode}");
                        return Ok(());
                    }
                };
                let mut value = inode_lock.write().display_err()?;
                value.decr_lookup_count(count)
            };
            if 0 == lookup_count {
                let delete = inodes
                    .remove(&inode)
                    .unwrap()
                    .into_inner()
                    .display_err()?
                    .delete;
                if delete {
                    let path = Self::block_path(&self.path, inode);
                    std::fs::remove_file(path)?;
                }
            }
            Ok(())
        }

        /// Returns the next available inode and updates the superblock in one atomic operation.
        /// TODO: Obviously this strategy won't work when there are multiple servers in this
        /// generation.
        fn next_inode(&self) -> io::Result<Inode> {
            self.borrow_block(SpecInodes::Sb.into(), |block| {
                // We don't need strict ordering because the lock the inode table is already
                // serializing access.
                let inode = self.next_inode.fetch_add(1, Ordering::Relaxed);
                let sb = Superblock {
                    generation: self.generation,
                    next_inode: inode + 1,
                };
                block.seek(SeekFrom::Start(0))?;
                write_to(&sb, block)?;
                Ok(inode)
            })
        }

        fn attr_timeout(&self) -> Duration {
            Duration::from_secs(5)
        }

        fn entry_timeout(&self) -> Duration {
            Duration::from_secs(5)
        }

        fn unsupported_flag<T>(flag: &str) -> io::Result<T> {
            Err(io::Error::new(
                io::ErrorKind::Unsupported,
                format!("unsupported flag: {flag}"),
            ))
        }

        fn fuse_entry(&self, inode: Inode, stat: stat64) -> Entry {
            Entry {
                generation: self.generation,
                inode,
                attr: stat,
                attr_flags: 0,
                attr_timeout: self.attr_timeout(),
                entry_timeout: self.entry_timeout(),
            }
        }
    }

    unsafe impl<C: Sync, A: Sync> Sync for Blocktree<C, A> {}

    impl<C: Creds + Clone + 'static, A: Authorizer> FileSystem for Blocktree<C, A> {
        type Inode = Inode;
        type Handle = u64;

        fn init(&self, _capable: FsOptions) -> io::Result<FsOptions> {
            debug!("Blocktree::init called");
            Ok(FsOptions::empty())
        }

        fn destroy(&self) {
            debug!("Blocktree::destroy called");
        }

        fn lookup(&self, ctx: &Context, parent: Self::Inode, name: &CStr) -> io::Result<Entry> {
            debug!("Blocktree::lookup called on parent {parent}");
            let name = name.to_str().box_err()?;
            let (dir, block_path) = self.borrow_block(parent, |block| {
                self.authorizer.exec_allowed(ctx, block.meta())?;
                let dir = block.read_dir()?;
                let path = block.meta_body().path.to_owned();
                Ok((dir, path))
            })?;
            let entry = dir
                .entries
                .get(name)
                .ok_or_else(|| io::Error::from_raw_os_error(libc::ENOENT))?;
            let inode = entry.inode().ok_or_else(|| {
                io::Error::new(io::ErrorKind::Unsupported, "can't lookup server entry")
            })?;
            let stat = self.open_value(inode, block_path, |value| {
                let stat = value.borrow_block(|block| Ok(block.meta_body().secrets()?.stat()))?;
                value.incr_lookup_count();
                Ok(stat)
            })?;
            Ok(self.fuse_entry(inode, stat))
        }

        fn open(
            &self,
            ctx: &Context,
            inode: Self::Inode,
            flags: u32,
            // This is the second field of the `fuse_open_in` struct, which is currently unused
            // by the kernel. (https://man7.org/linux/man-pages/man4/fuse.4.html)
            _fuse_flags: u32,
        ) -> io::Result<(Option<Self::Handle>, OpenOptions)> {
            debug!("Blocktree::open called on inode {inode}");
            let flags: i32 = flags.try_into().box_err()?;
            if flags & libc::O_APPEND != 0 {
                return Self::unsupported_flag("O_APPEND");
            }
            if flags & libc::O_CLOEXEC != 0 {
                return Self::unsupported_flag("O_CLOEXEC");
            }
            if flags & libc::O_DIRECTORY != 0 {
                return Self::unsupported_flag("O_DIRECTORY");
            }
            let handle = self.take_handle_if_ok(inode, |handle, block| {
                let ctx = AuthzContext::new(ctx, block.meta());
                if flags == libc::O_RDONLY || (flags & libc::O_RDWR) != 0 {
                    self.authorizer.can_read(&ctx)?;
                }
                let write_mask = libc::O_WRONLY | libc::O_RDWR;
                if write_mask & flags != 0 {
                    self.authorizer.can_write(&ctx)?;
                }
                Ok(handle)
            })?;
            Ok((Some(handle), OpenOptions::empty()))
        }

        fn release(
            &self,
            _ctx: &Context,
            inode: Self::Inode,
            _flags: u32,
            handle: Self::Handle,
            flush: bool,
            _flock_release: bool,
            _lock_owner: Option<u64>,
        ) -> io::Result<()> {
            debug!("Blocktree::release called on inode {inode}");
            self.access_value_mut(inode, |value| {
                value.access_block_mut(handle, |block| {
                    if flush {
                        block.flush()?;
                    }
                    // Be kind, rewind.
                    block.seek(SeekFrom::Start(0))
                })?;
                value.give_handle(handle);
                Ok(())
            })
        }

        fn opendir(
            &self,
            ctx: &Context,
            inode: Self::Inode,
            _flags: u32,
        ) -> io::Result<(Option<Self::Handle>, OpenOptions)> {
            debug!("Blocktree::opendir called on inode {inode}");
            let handle = self.access_value_mut(inode, |value| {
                value.borrow_block(|block| {
                    self.authorizer.exec_allowed(ctx, block.meta())?;
                    Ok(())
                })?;
                let handle = value.take_handle()?;
                value.convert_to_dir(handle)?;
                Ok(handle)
            })?;
            Ok((Some(handle), OpenOptions::empty()))
        }

        fn releasedir(
            &self,
            _ctx: &Context,
            inode: Self::Inode,
            _flags: u32,
            handle: Self::Handle,
        ) -> io::Result<()> {
            debug!("Blocktree::releasedir called for inode {inode}");
            self.access_value_mut(inode, |value| {
                value.give_handle(handle);
                Ok(())
            })
        }

        fn create(
            &self,
            ctx: &Context,
            parent: Self::Inode,
            name: &CStr,
            args: CreateIn,
        ) -> std::io::Result<(Entry, Option<Self::Handle>, OpenOptions)> {
            debug!("Blocktree::create called on parent {parent}");

            let name = name
                .to_str()
                .map_err(|err| io::Error::new(io::ErrorKind::InvalidInput, err))?
                .to_owned();

            // Reserve a free inode.
            let inode = self.next_inode()?;

            // Add a directory entry to the parent for the new inode.
            let mut block_path = self.borrow_block(parent, |block| {
                self.authorizer.write_allowed(ctx, block.meta())?;

                let mut dir = block.read_dir()?;
                dir.add_file(name.clone(), inode)?;
                block.write_dir(&dir)?;

                Ok(block.meta_body().path.clone())
            })?;
            block_path.push_component(name);

            let (handle, stat) =
                self.open_then_take_handle(inode, block_path, |handle, value| {
                    let block = value.block_mut(handle)?;
                    Ok(block.mut_meta_body().access_secrets(|secrets| {
                        secrets.block_id.generation = self.generation;
                        secrets.block_id.inode = inode;
                        secrets.mode = args.mode & !args.umask;
                        secrets.uid = ctx.uid;
                        secrets.gid = ctx.gid;
                        let now = Epoch::now();
                        secrets.atime = now;
                        secrets.ctime = now;
                        secrets.mtime = now;
                        secrets.nlink = 1;
                        Ok((handle, secrets.stat()))
                    })?)
                })?;

            Ok((
                self.fuse_entry(inode, stat),
                Some(handle),
                OpenOptions::empty(),
            ))
        }

        fn write(
            &self,
            _ctx: &Context,
            inode: Self::Inode,
            handle: Self::Handle,
            r: &mut dyn fuse_backend_rs::api::filesystem::ZeroCopyReader,
            size: u32,
            _offset: u64,
            _lock_owner: Option<u64>,
            _delayed_write: bool,
            // `flags` and `fuse_flags` are the arguments that were passed to `open` when this
            // handle was returned.
            flags: u32,
            _fuse_flags: u32,
        ) -> io::Result<usize> {
            debug!("Blocktree::write called called on inode {inode}");
            if flags as libc::c_int == libc::O_RDONLY {
                return Err(io::Error::new(
                    io::ErrorKind::PermissionDenied,
                    "file is readonly",
                ));
            }
            let mut size: usize = size.try_into().box_err()?;
            self.access_block_mut(inode, handle, |block| {
                let mut buf = [0u8; crate::SECTOR_SZ_DEFAULT];
                let mut written = 0;
                while size > 0 {
                    let read = match r.read(&mut buf) {
                        Ok(size) => size,
                        Err(err) => {
                            if written > 0 {
                                error!("error while reading: {err}");
                                return Ok(written);
                            } else {
                                return Err(err);
                            }
                        }
                    };
                    if 0 == read {
                        break;
                    }
                    let filled = &buf[..read];
                    size -= read;
                    if let Err(err) = block.write_all(filled) {
                        if written > 0 {
                            error!("error while writing: {err}");
                            return Ok(written);
                        } else {
                            return Err(err);
                        }
                    }
                    written += filled.len();
                }
                Ok(written)
            })
        }

        fn flush(
            &self,
            _ctx: &Context,
            inode: Self::Inode,
            handle: Self::Handle,
            _lock_owner: u64,
        ) -> io::Result<()> {
            debug!("Blocktree::flush called for inode {inode}");
            self.access_block_mut(inode, handle, |block| block.flush())
        }

        fn read(
            &self,
            _ctx: &Context,
            inode: Self::Inode,
            handle: Self::Handle,
            w: &mut dyn fuse_backend_rs::api::filesystem::ZeroCopyWriter,
            size: u32,
            _offset: u64,
            _lock_owner: Option<u64>,
            flags: u32,
        ) -> io::Result<usize> {
            debug!("Blocktree::read called on inode {inode}");
            if (flags as libc::c_int & libc::O_WRONLY) != 0 {
                return Err(io::Error::new(
                    io::ErrorKind::PermissionDenied,
                    "file is write only",
                ));
            }
            let mut size: usize = size.try_into().box_err()?;
            self.access_block_mut(inode, handle, |block| {
                let mut buf = [0u8; crate::SECTOR_SZ_DEFAULT];
                let mut read = 0;
                while size > 0 {
                    let just_read = match block.read(&mut buf) {
                        Ok(just_read) => just_read,
                        Err(err) => {
                            if read > 0 {
                                error!("error while reading from block: {err}");
                                return Ok(read);
                            } else {
                                return Err(err);
                            }
                        }
                    };
                    if 0 == just_read {
                        break;
                    }
                    read += just_read;
                    let filled = &buf[..just_read];
                    if let Err(err) = w.write_all(filled) {
                        if read > 0 {
                            error!("error while writing: {err}");
                            return Ok(read);
                        } else {
                            return Err(err);
                        }
                    }
                    size -= filled.len();
                }
                Ok(read)
            })
        }

        fn readdir(
            &self,
            _ctx: &Context,
            inode: Self::Inode,
            handle: Self::Handle,
            size: u32,
            offset: u64,
            add_entry: &mut dyn FnMut(
                fuse_backend_rs::api::filesystem::DirEntry,
            ) -> io::Result<usize>,
        ) -> io::Result<()> {
            debug!("Blocktree::readdir called on inode {inode}");
            let mut size: usize = size.try_into().box_err()?;
            self.access_value(inode, |value| {
                let dir = value
                    .value(handle)
                    .map_err(|_| bterr!(Error::InvalidHandle { handle, inode }))?
                    .directory()?;
                let mut index: u64 = 0;
                for (name, entry) in dir.entries() {
                    index += 1;
                    if index <= offset {
                        continue;
                    }
                    let inode = match entry.inode() {
                        Some(inode) => inode,
                        None => continue,
                    };
                    let dir_entry = FuseDirEntry {
                        ino: inode,
                        offset: index,
                        type_: entry.kind() as u32,
                        name: name.as_bytes(),
                    };
                    size = size.saturating_sub(add_entry(dir_entry)?);
                    if size == 0 {
                        break;
                    }
                }
                Ok(())
            })
        }

        fn getattr(
            &self,
            _ctx: &Context,
            inode: Self::Inode,
            _handle: Option<Self::Handle>,
        ) -> io::Result<(stat64, Duration)> {
            debug!("Blocktree::getattr called for inode {inode}");
            let stat = self.access_meta(inode, |meta| Ok(meta.body.secrets()?.stat()))?;
            Ok((stat, self.attr_timeout()))
        }

        fn forget(&self, _ctx: &Context, inode: Self::Inode, count: u64) {
            debug!("Blocktree::forget called for inode {inode}");
            if let Err(err) = self.inode_forget(inode, count) {
                error!("Blocktree::forget failed for inode {inode}: {err}");
            }
        }

        fn lseek(
            &self,
            _ctx: &Context,
            inode: Self::Inode,
            handle: Self::Handle,
            offset: u64,
            whence: u32,
        ) -> io::Result<u64> {
            debug!("Blocktree::lseek called for inode {inode}");
            let seek_from = SeekFrom::whence_offset(whence, offset)?;
            self.access_block_mut(inode, handle, |block| block.seek(seek_from))
        }

        fn unlink(&self, ctx: &Context, parent: Self::Inode, name: &CStr) -> io::Result<()> {
            debug!("Blocktree::unlink called on parent {parent}");
            let name = name.to_str().box_err()?;
            let (block_path, inode) = self.borrow_block(parent, |block| {
                self.authorizer.write_allowed(ctx, block.meta())?;

                let mut dir = block.read_dir()?;
                let entry = match dir.entries.remove(name) {
                    None => return Err(io::Error::from_raw_os_error(libc::ENOENT)),
                    Some(entry) => entry,
                };
                let inode = entry.inode().ok_or_else(|| {
                    io::Error::new(
                        io::ErrorKind::Other,
                        "no inode associated with the given name",
                    )
                })?;
                block.write_dir(&dir)?;

                let mut block_path = block.meta_body().path.clone();
                block_path.push_component(name.to_owned());
                Ok((block_path, inode))
            })?;
            self.open_value(inode, block_path, |value| {
                // We mark the block for deletion if `nlink` drops to zero.
                value.delete = value.borrow_block(|block| {
                    let nlink = block.mut_meta_body().access_secrets(|secrets| {
                        secrets.nlink -= 1;
                        Ok(secrets.nlink)
                    })?;
                    block.flush_meta()?;
                    Ok(0 == nlink)
                })?;
                Ok(())
            })
        }

        fn link(
            &self,
            ctx: &Context,
            inode: Self::Inode,
            newparent: Self::Inode,
            newname: &CStr,
        ) -> io::Result<Entry> {
            debug!("Blocktree::link called for inode {inode}");
            let newname = newname.to_str().box_err()?;
            self.borrow_block(newparent, |block| {
                self.authorizer.write_allowed(ctx, block.meta())?;

                let mut dir = block.read_dir()?;
                if dir.entries.contains_key(newname) {
                    return Err(io::Error::from_raw_os_error(libc::EEXIST));
                }

                let (file_type, stat) = self.access_value_mut(inode, |value| {
                    let (file_type, stat) = value.borrow_block(|block| {
                        let meta = block.mut_meta_body();
                        let (mode, stat) = meta.access_secrets(|secrets| {
                            secrets.nlink += 1;
                            Ok((secrets.mode, secrets.stat()))
                        })?;
                        let file_type = FileType::from_value(mode)?;
                        block.flush_meta()?;
                        Ok((file_type, stat))
                    })?;
                    value.incr_lookup_count();
                    Ok((file_type, stat))
                })?;
                let entry = match file_type {
                    FileType::Reg => DirEntry::File(BlockRecord::new(inode)),
                    FileType::Dir => DirEntry::Directory(BlockRecord::new(inode)),
                };
                dir.entries.insert(newname.to_owned(), entry);
                block.write_dir(&dir)?;
                Ok(self.fuse_entry(inode, stat))
            })
        }

        fn setattr(
            &self,
            ctx: &Context,
            inode: Self::Inode,
            attr: stat64,
            _handle: Option<Self::Handle>,
            valid: SetattrValid,
        ) -> io::Result<(stat64, Duration)> {
            debug!("Blocktree::setattr called for inode {inode}");
            let stat = self.borrow_block(inode, |block| {
                self.authorizer.write_allowed(ctx, block.meta())?;

                let stat = block.mut_meta_body().access_secrets(|secrets| {
                    if valid.intersects(SetattrValid::MODE) {
                        secrets.mode = attr.st_mode;
                    }
                    if valid.intersects(SetattrValid::UID) {
                        secrets.uid = attr.st_uid;
                    }
                    if valid.intersects(SetattrValid::GID) {
                        secrets.gid = attr.st_gid;
                    }
                    if valid.intersects(SetattrValid::SIZE) {
                        warn!("modifying file size with setattr is not supported");
                        return Err(io::Error::from_raw_os_error(libc::EINVAL).into());
                    }
                    if valid.intersects(SetattrValid::ATIME) {
                        secrets.atime = (attr.st_atime as u64).into();
                    }
                    if valid.intersects(SetattrValid::MTIME) {
                        secrets.mtime = (attr.st_mtime as u64).into();
                    }
                    if valid.intersects(SetattrValid::CTIME) {
                        secrets.ctime = (attr.st_ctime as u64).into();
                    }
                    let atime_now = valid.intersects(SetattrValid::ATIME_NOW);
                    let mtime_now = valid.intersects(SetattrValid::MTIME_NOW);
                    if atime_now || mtime_now {
                        let now = Epoch::now();
                        if atime_now {
                            secrets.atime = now;
                        }
                        if mtime_now {
                            secrets.mtime = now;
                        }
                    }
                    if valid.intersects(SetattrValid::KILL_SUIDGID) {
                        secrets.mode &= !(libc::S_ISUID | libc::S_ISGID)
                    }
                    Ok(secrets.stat())
                })?;
                block.flush_meta()?;
                Ok(stat)
            })?;
            Ok((stat, self.attr_timeout()))
        }

        fn mkdir(
            &self,
            ctx: &Context,
            parent: Self::Inode,
            name: &CStr,
            mode: u32,
            umask: u32,
        ) -> io::Result<Entry> {
            debug!("Blocktree::mkdir called");
            let name = name.to_str().box_err()?.to_owned();
            let (inode, mut block_path) = self.borrow_block(parent, |block| {
                self.authorizer.write_allowed(ctx, block.meta())?;

                let mut dir = block.read_dir()?;
                if dir.entries.contains_key(&name) {
                    return Err(io::Error::from_raw_os_error(libc::EEXIST));
                }

                let inode = self.next_inode()?;
                dir.add_file(name.clone(), inode)?;
                block.write_dir(&dir)?;

                Ok((inode, block.meta_body().path.clone()))
            })?;
            block_path.push_component(name);
            let stat = self.open_value(inode, block_path, |value| {
                let stat = value.borrow_block(|block| {
                    let stat = block.mut_meta_body().access_secrets(|secrets| {
                        secrets.block_id.generation = self.generation;
                        secrets.block_id.inode = inode;
                        secrets.mode = libc::S_IFDIR | mode & !umask;
                        secrets.uid = ctx.uid;
                        secrets.gid = ctx.gid;
                        Ok(secrets.stat())
                    })?;
                    block.write_dir(&Directory::new())?;
                    block.flush_meta()?;
                    Ok(stat)
                })?;
                value.incr_lookup_count();
                Ok(stat)
            })?;
            Ok(self.fuse_entry(inode, stat))
        }

        fn rmdir(&self, ctx: &Context, parent: Self::Inode, name: &CStr) -> io::Result<()> {
            debug!("Blocktree::rmdir called on parent {parent}");
            let name = name.to_str().box_err()?.to_owned();
            self.borrow_block(parent, |block| {
                self.authorizer.write_allowed(ctx, block.meta())?;

                let mut dir = block.read_dir()?;
                if dir.entries.remove(&name).is_none() {
                    return Err(io::Error::from_raw_os_error(libc::ENOENT));
                }

                block.write_dir(&dir)?;
                Ok(())
            })
        }

        //////////////////////////////////
        // METHODS WHICH ARE NOT SUPPORTED
        //////////////////////////////////

        fn getxattr(
            &self,
            _ctx: &Context,
            inode: Self::Inode,
            _name: &CStr,
            _size: u32,
        ) -> io::Result<fuse_backend_rs::api::filesystem::GetxattrReply> {
            debug!("Blocktree::getxattr called for inode {inode}");
            Self::not_supported()
        }

        fn ioctl(
            &self,
            _ctx: &Context,
            inode: Self::Inode,
            _handle: Self::Handle,
            _flags: u32,
            _cmd: u32,
            _data: fuse_backend_rs::api::filesystem::IoctlData,
            _out_size: u32,
        ) -> io::Result<fuse_backend_rs::api::filesystem::IoctlData> {
            debug!("Blocktree::ioctl called for inode {inode}");
            Self::not_supported()
        }

        fn access(&self, _ctx: &Context, inode: Self::Inode, _mask: u32) -> io::Result<()> {
            debug!("Blocktree::access called for inode {inode}");
            Self::not_supported()
        }

        fn batch_forget(&self, _ctx: &Context, _requests: Vec<(Self::Inode, u64)>) {
            debug!("Blocktree::batch_forget called");
            Self::not_supported().unwrap()
        }

        fn bmap(
            &self,
            _ctx: &Context,
            inode: Self::Inode,
            _block: u64,
            _blocksize: u32,
        ) -> io::Result<u64> {
            debug!("Blocktree::bmap called for inode {inode}");
            Self::not_supported()
        }

        fn fallocate(
            &self,
            _ctx: &Context,
            inode: Self::Inode,
            _handle: Self::Handle,
            _mode: u32,
            _offset: u64,
            _length: u64,
        ) -> io::Result<()> {
            debug!("Blocktree::fallocate called for inode {inode}");
            Self::not_supported()
        }

        fn fsync(
            &self,
            _ctx: &Context,
            inode: Self::Inode,
            _datasync: bool,
            _handle: Self::Handle,
        ) -> io::Result<()> {
            debug!("Blocktree::fsync called for inode {inode}");
            Self::not_supported()
        }

        fn fsyncdir(
            &self,
            _ctx: &Context,
            inode: Self::Inode,
            _datasync: bool,
            _handle: Self::Handle,
        ) -> io::Result<()> {
            debug!("Blocktree::fsyncdir called for inode {inode}");
            Self::not_supported()
        }

        fn getlk(
            &self,
            _ctx: &Context,
            inode: Self::Inode,
            _handle: Self::Handle,
            _owner: u64,
            _lock: fuse_backend_rs::api::filesystem::FileLock,
            _flags: u32,
        ) -> io::Result<fuse_backend_rs::api::filesystem::FileLock> {
            debug!("Blocktree::getlk called for inode {inode}");
            Self::not_supported()
        }

        fn listxattr(
            &self,
            _ctx: &Context,
            inode: Self::Inode,
            _size: u32,
        ) -> io::Result<fuse_backend_rs::api::filesystem::ListxattrReply> {
            debug!("Blocktree::listxattr called for inode {inode}");
            Self::not_supported()
        }

        fn mknod(
            &self,
            _ctx: &Context,
            inode: Self::Inode,
            _name: &CStr,
            _mode: u32,
            _rdev: u32,
            _umask: u32,
        ) -> io::Result<Entry> {
            debug!("Blocktree::mknod called for inode {inode}");
            Self::not_supported()
        }

        fn notify_reply(&self) -> io::Result<()> {
            debug!("Blocktree::notify_reply called");
            Self::not_supported()
        }

        fn poll(
            &self,
            _ctx: &Context,
            inode: Self::Inode,
            _handle: Self::Handle,
            _khandle: Self::Handle,
            _flags: u32,
            _events: u32,
        ) -> io::Result<u32> {
            debug!("Blocktree::poll called for inode {inode}");
            Self::not_supported()
        }

        fn readdirplus(
            &self,
            _ctx: &Context,
            _inode: Self::Inode,
            _handle: Self::Handle,
            _size: u32,
            _offset: u64,
            _add_entry: &mut dyn FnMut(
                fuse_backend_rs::api::filesystem::DirEntry,
                Entry,
            ) -> io::Result<usize>,
        ) -> io::Result<()> {
            debug!("Blocktree::readdirplus called");
            Self::not_supported()
        }

        fn readlink(&self, _ctx: &Context, inode: Self::Inode) -> io::Result<Vec<u8>> {
            debug!("Blocktree::readlink called for inode {inode}");
            Self::not_supported()
        }

        fn removexattr(&self, _ctx: &Context, inode: Self::Inode, _name: &CStr) -> io::Result<()> {
            debug!("Blocktree::removexattr called for inode {inode}");
            Self::not_supported()
        }

        fn rename(
            &self,
            _ctx: &Context,
            _olddir: Self::Inode,
            _oldname: &CStr,
            _newdir: Self::Inode,
            _newname: &CStr,
            _flags: u32,
        ) -> io::Result<()> {
            debug!("Blocktree::rename called");
            Self::not_supported()
        }

        fn setlk(
            &self,
            _ctx: &Context,
            inode: Self::Inode,
            _handle: Self::Handle,
            _owner: u64,
            _lock: fuse_backend_rs::api::filesystem::FileLock,
            _flags: u32,
        ) -> io::Result<()> {
            debug!("Blocktree::setlk called for inode {inode}");
            Self::not_supported()
        }

        fn setlkw(
            &self,
            _ctx: &Context,
            inode: Self::Inode,
            _handle: Self::Handle,
            _owner: u64,
            _lock: fuse_backend_rs::api::filesystem::FileLock,
            _flags: u32,
        ) -> io::Result<()> {
            debug!("Blocktree::setlkw called for inode {inode}");
            Self::not_supported()
        }

        fn setxattr(
            &self,
            _ctx: &Context,
            inode: Self::Inode,
            _name: &CStr,
            _value: &[u8],
            _flags: u32,
        ) -> io::Result<()> {
            debug!("Blocktree::setxattr called for inode {inode}");
            Self::not_supported()
        }

        fn statfs(&self, _ctx: &Context, inode: Self::Inode) -> io::Result<statvfs64> {
            debug!("Blocktree::statfs called for inode {inode}");
            Self::not_supported()
        }

        fn symlink(
            &self,
            _ctx: &Context,
            _linkname: &CStr,
            _parent: Self::Inode,
            _name: &CStr,
        ) -> io::Result<Entry> {
            debug!("Blocktree::symlink called");
            Self::not_supported()
        }
    }
}

#[cfg(test)]
mod tests {
    use fuse_backend_rs::{
        abi::fuse_abi::CreateIn,
        api::filesystem::{Context, FileSystem, FsOptions},
    };
    use std::{ffi::CString, io};
    use tempdir::TempDir;

    use test_helpers::*;

    use crate::{crypto::ConcreteCreds, test_helpers, BlockMeta, Decompose};

    use super::{private::SpecInodes, *};

    /// Tests for the [ModeAuthorizer] struct.
    mod mode_authorizer_tests {
        use super::{
            super::private::{Authorizer, AuthzContext},
            *,
        };

        struct TestCase {
            ctx_uid: u32,
            ctx_gid: u32,
            meta: BlockMeta,
        }

        impl TestCase {
            const BLOCK_UID: u32 = 1000;
            const BLOCK_GID: u32 = 1000;
            const CTX_PID: libc::pid_t = 100;

            fn new(ctx_uid: u32, ctx_gid: u32, mode: u32) -> TestCase {
                let mut meta = BlockMeta::new(&*test_helpers::NODE_CREDS)
                    .expect("failed to create block metadata");
                meta.body
                    .access_secrets(|secrets| {
                        secrets.uid = Self::BLOCK_UID;
                        secrets.gid = Self::BLOCK_GID;
                        secrets.mode = mode;
                        Ok(())
                    })
                    .expect("failed to update secrets");
                TestCase {
                    ctx_uid,
                    ctx_gid,
                    meta,
                }
            }

            fn context(&self) -> AuthzContext<'_> {
                AuthzContext {
                    uid: self.ctx_uid,
                    gid: self.ctx_gid,
                    pid: Self::CTX_PID,
                    meta: &self.meta,
                }
            }
        }

        #[test]
        fn cant_read_when_no_bits_set() {
            let case = TestCase::new(TestCase::BLOCK_UID, TestCase::BLOCK_GID, 0);
            let result = ModeAuthorizer {}.can_read(&case.context());
            assert!(result.is_err())
        }

        #[test]
        fn cant_write_when_no_bits_set() {
            let case = TestCase::new(TestCase::BLOCK_UID, TestCase::BLOCK_GID, 0);
            let result = ModeAuthorizer {}.can_write(&case.context());
            assert!(result.is_err())
        }

        #[test]
        fn cant_exec_when_no_bits_set() {
            let case = TestCase::new(TestCase::BLOCK_UID, TestCase::BLOCK_GID, 0);
            let result = ModeAuthorizer {}.can_exec(&case.context());
            assert!(result.is_err())
        }

        #[test]
        fn user_can_read_when_bit_set() {
            let case = TestCase::new(TestCase::BLOCK_UID, TestCase::BLOCK_GID, libc::S_IRUSR);
            let result = ModeAuthorizer {}.can_read(&case.context());
            assert!(result.is_ok())
        }

        #[test]
        fn user_can_write_when_bit_set() {
            let case = TestCase::new(TestCase::BLOCK_UID, TestCase::BLOCK_GID, libc::S_IWUSR);
            let result = ModeAuthorizer {}.can_write(&case.context());
            assert!(result.is_ok())
        }

        #[test]
        fn user_can_exec_when_bit_set() {
            let case = TestCase::new(TestCase::BLOCK_UID, TestCase::BLOCK_GID, libc::S_IXUSR);
            let result = ModeAuthorizer {}.can_exec(&case.context());
            assert!(result.is_ok())
        }

        #[test]
        fn group_can_read_when_bit_set() {
            let case = TestCase::new(TestCase::BLOCK_UID, TestCase::BLOCK_GID, libc::S_IRGRP);
            let result = ModeAuthorizer {}.can_read(&case.context());
            assert!(result.is_ok())
        }

        #[test]
        fn group_can_write_when_bit_set() {
            let case = TestCase::new(TestCase::BLOCK_UID, TestCase::BLOCK_GID, libc::S_IWGRP);
            let result = ModeAuthorizer {}.can_write(&case.context());
            assert!(result.is_ok())
        }

        #[test]
        fn group_can_exec_when_bit_set() {
            let case = TestCase::new(TestCase::BLOCK_UID, TestCase::BLOCK_GID, libc::S_IXGRP);
            let result = ModeAuthorizer {}.can_exec(&case.context());
            assert!(result.is_ok())
        }

        #[test]
        fn other_can_read_when_bit_set() {
            let case = TestCase::new(TestCase::BLOCK_UID, TestCase::BLOCK_GID, libc::S_IROTH);
            let result = ModeAuthorizer {}.can_read(&case.context());
            assert!(result.is_ok())
        }

        #[test]
        fn other_can_write_when_bit_set() {
            let case = TestCase::new(TestCase::BLOCK_UID, TestCase::BLOCK_GID, libc::S_IWOTH);
            let result = ModeAuthorizer {}.can_write(&case.context());
            assert!(result.is_ok())
        }

        #[test]
        fn other_can_exec_when_bit_set() {
            let case = TestCase::new(TestCase::BLOCK_UID, TestCase::BLOCK_GID, libc::S_IXOTH);
            let result = ModeAuthorizer {}.can_exec(&case.context());
            assert!(result.is_ok())
        }

        #[test]
        fn other_cant_write_even_if_user_can() {
            let case = TestCase::new(
                TestCase::BLOCK_UID + 1,
                TestCase::BLOCK_GID + 1,
                libc::S_IWUSR,
            );
            let result = ModeAuthorizer {}.can_write(&case.context());
            assert!(result.is_err())
        }

        #[test]
        fn other_cant_write_even_if_group_can() {
            let case = TestCase::new(
                TestCase::BLOCK_UID + 1,
                TestCase::BLOCK_GID + 1,
                libc::S_IWGRP,
            );
            let result = ModeAuthorizer {}.can_write(&case.context());
            assert!(result.is_err())
        }

        #[test]
        fn user_allowed_read_when_only_other_bit_set() {
            let case = TestCase::new(TestCase::BLOCK_UID, TestCase::BLOCK_GID, libc::S_IROTH);
            let result = ModeAuthorizer {}.can_read(&case.context());
            assert!(result.is_ok())
        }

        #[test]
        fn root_always_allowed() {
            let case = TestCase::new(0, 0, 0);
            let ctx = case.context();
            let authorizer = ModeAuthorizer {};
            assert!(authorizer.can_read(&ctx).is_ok());
            assert!(authorizer.can_write(&ctx).is_ok());
            assert!(authorizer.can_exec(&ctx).is_ok());
        }
    }

    struct BtTestCase {
        dir: TempDir,
        bt: Blocktree<ConcreteCreds, ModeAuthorizer>,
    }

    impl BtTestCase {
        fn new_empty() -> BtTestCase {
            let dir = TempDir::new("fuse").expect("failed to create temp dir");
            let bt =
                Blocktree::new_empty(dir.path().to_owned(), 0, Self::creds(), ModeAuthorizer {})
                    .expect("failed to create empty blocktree");
            bt.init(FsOptions::empty()).expect("init failed");
            BtTestCase { dir, bt }
        }

        fn new_existing(dir: TempDir) -> BtTestCase {
            let bt =
                Blocktree::new_existing(dir.path().to_owned(), Self::creds(), ModeAuthorizer {})
                    .expect("failed to create blocktree from existing directory");
            bt.init(FsOptions::empty()).expect("init failed");
            BtTestCase { dir, bt }
        }

        fn creds() -> ConcreteCreds {
            test_helpers::NODE_CREDS.clone()
        }

        fn context(&self) -> Context {
            let (stat, ..) = self
                .bt
                .getattr(&Default::default(), SpecInodes::RootDir.into(), None)
                .expect("getattr failed");
            Context {
                uid: stat.st_uid,
                gid: stat.st_gid,
                pid: 1,
            }
        }
    }

    /// Tests that a new file can be created, written to and the written data can be read from it.
    #[test]
    fn create_write_lseek_read() {
        let case = BtTestCase::new_empty();
        let bt = &case.bt;
        let ctx = case.context();
        let name = CString::new("README.md").unwrap();
        let flags = libc::O_RDWR as u32;

        let (entry, handle, ..) = bt
            .create(
                &ctx,
                SpecInodes::RootDir.into(),
                name.as_c_str(),
                CreateIn {
                    mode: libc::S_IFREG | 0o644,
                    umask: 0,
                    flags,
                    fuse_flags: 0,
                },
            )
            .expect("failed to create file");
        let inode = entry.inode;
        let handle = handle.unwrap();

        const LEN: usize = 32;
        let mut expected = BtCursor::new([1u8; LEN]);
        let written = bt
            .write(
                &ctx,
                inode,
                handle,
                &mut expected,
                LEN as u32,
                0,
                None,
                false,
                flags,
                0,
            )
            .expect("write failed");
        assert_eq!(LEN, written);

        bt.lseek(&ctx, inode, handle, 0, 0).expect("lseek failed");

        let mut actual = BtCursor::new([0u8; LEN]);
        let read = bt
            .read(&ctx, inode, handle, &mut actual, LEN as u32, 0, None, flags)
            .expect("failed to read");
        assert_eq!(LEN, read);

        assert_eq!(expected, actual)
    }

    #[test]
    fn lookup() {
        let case = BtTestCase::new_empty();
        let bt = &case.bt;
        let ctx = case.context();

        let name = CString::new("README.md").unwrap();
        let (expected, ..) = bt
            .create(
                &ctx,
                SpecInodes::RootDir.into(),
                name.as_c_str(),
                Default::default(),
            )
            .expect("failed to create file");

        let actual = bt
            .lookup(&Default::default(), SpecInodes::RootDir.into(), &name)
            .expect("lookup failed");

        assert_eq!(expected.generation, actual.generation);
        assert_eq!(expected.inode, actual.inode);
    }

    /// Tests that data written by one instance of [Blocktree] can be read by a subsequent
    /// instance.
    #[test]
    fn new_existing() {
        const EXPECTED: &[u8] = b"cool as cucumbers";
        let name = CString::new("RESIGNATION.docx").unwrap();
        let case = BtTestCase::new_empty();
        let bt = &case.bt;
        let ctx = case.context();
        let flags = libc::O_RDWR as u32;

        {
            let (entry, handle, ..) = bt
                .create(
                    &ctx,
                    SpecInodes::RootDir.into(),
                    name.as_c_str(),
                    CreateIn {
                        mode: libc::S_IFREG | 0o644,
                        umask: 0,
                        flags,
                        fuse_flags: 0,
                    },
                )
                .expect("failed to create file");
            let inode = entry.inode;
            let handle = handle.unwrap();

            let mut vec = Vec::with_capacity(EXPECTED.len());
            vec.extend_from_slice(EXPECTED);
            let mut cursor = BtCursor::new(vec);
            let written = bt
                .write(
                    &Default::default(),
                    inode,
                    handle,
                    &mut cursor,
                    EXPECTED.len() as u32,
                    0,
                    None,
                    false,
                    flags,
                    0,
                )
                .expect("write failed");
            assert_eq!(EXPECTED.len(), written);

            bt.flush(&Default::default(), inode, handle, 0)
                .expect("flush failed");
        }

        let case = BtTestCase::new_existing(case.dir);
        let bt = &case.bt;

        let entry = bt
            .lookup(&Default::default(), SpecInodes::RootDir.into(), &name)
            .expect("lookup failed");
        let inode = entry.inode;

        let (handle, ..) = bt
            .open(&Default::default(), entry.inode, 0, 0)
            .expect("open failed");
        let handle = handle.unwrap();

        let mut actual = BtCursor::new([0u8; EXPECTED.len()]);
        let _ = bt
            .read(
                &Default::default(),
                inode,
                handle,
                &mut actual,
                EXPECTED.len() as u32,
                0,
                None,
                flags,
            )
            .expect("read failed");

        assert_eq!(EXPECTED, actual.into_inner().as_slice())
    }

    /// Tests that an error is returned by the `Blocktree::write` method if the file was opened
    /// read-only.
    #[test]
    fn open_read_only_write_is_error() {
        let name = CString::new("books.ods").unwrap();
        let case = BtTestCase::new_empty();
        let bt = &case.bt;
        let ctx = case.context();

        let (entry, handle, ..) = bt
            .create(
                &ctx,
                SpecInodes::RootDir.into(),
                name.as_c_str(),
                CreateIn {
                    mode: libc::S_IFREG | 0o644,
                    umask: 0,
                    flags: 0,
                    fuse_flags: 0,
                },
            )
            .expect("failed to create file");
        let inode = entry.inode;
        let handle = handle.unwrap();

        bt.release(&ctx, inode, 0, handle, false, false, None)
            .expect("release failed");

        let flags = libc::O_RDONLY as u32;
        let (handle, ..) = bt.open(&ctx, inode, flags, 0).expect("open failed");
        let handle = handle.unwrap();

        const LEN: usize = 32;
        let mut reader = BtCursor::new([1u8; LEN]);
        let result = bt.write(
            &ctx,
            inode,
            handle,
            &mut reader,
            LEN.try_into().unwrap(),
            0,
            None,
            false,
            flags,
            0,
        );

        let err = result.err().unwrap();
        assert_eq!(io::ErrorKind::PermissionDenied, err.kind());
    }

    /// Tests that a call to `read` fails when a file is opened write only.
    #[test]
    fn open_write_only_read_is_error() {
        let name = CString::new("books.ods").unwrap();
        let case = BtTestCase::new_empty();
        let bt = &case.bt;
        let ctx = case.context();

        let (entry, handle, ..) = bt
            .create(
                &ctx,
                SpecInodes::RootDir.into(),
                name.as_c_str(),
                CreateIn {
                    mode: libc::S_IFREG | 0o644,
                    umask: 0,
                    flags: 0,
                    fuse_flags: 0,
                },
            )
            .expect("failed to create file");
        let inode = entry.inode;
        let handle = handle.unwrap();

        bt.release(&ctx, inode, 0, handle, false, false, None)
            .expect("release failed");

        let flags = libc::O_WRONLY as u32;
        let (handle, ..) = bt.open(&ctx, inode, flags, 0).expect("open failed");
        let handle = handle.unwrap();

        const LEN: usize = 32;
        let mut reader = BtCursor::new([1u8; LEN]);
        let result = bt.read(
            &ctx,
            inode,
            handle,
            &mut reader,
            LEN.try_into().unwrap(),
            0,
            None,
            flags,
        );

        let err = result.err().unwrap();
        assert_eq!(io::ErrorKind::PermissionDenied, err.kind());
    }
}