flow-casfs: a read-only FUSE mount of an Oxygen state (refs #21) f849c7889471

[package]

name = "flow-casfs"

version.workspace = true

edition.workspace = true

license.workspace = true

description = "A read-only FUSE mount of an Oxygen state (OxygenFlow spec 5.1)"

[[bin]]

name = "flow-casfs"

path = "src/main.rs"

[dependencies]

oxy-core.workspace = true

anyhow.workspace = true

# Pure Rust: mounts through the fusermount3 binary, no libfuse headers.

fuser = { version = "0.16", default-features = false }

libc = "0.2"

[dev-dependencies]

tempfile.workspace = true

//! flow-casfs: an Oxygen state mounted read-only (OxygenFlow spec 5.1).

//!

//! The tree comes from the repository's state index; an opaque file's bytes

//! are its stored blob; a text file's bytes are a rendering of its history,

//! made on first read or stat and cached on disk by graph key (equal keys

//! mean equal renderings), so every state in which a file has the same

//! history finds it at once.

//!

//!   flow-casfs mount REPO MOUNTPOINT [--bookmark NAME]

//!

//! It mounts in the foreground and prints how long the mount took; unmount

//! with `fusermount3 -u MOUNTPOINT`.

use anyhow::{bail, Context, Result};

use fuser::{FileAttr, FileType, Filesystem, MountOption, ReplyAttr, ReplyData, ReplyDirectory, ReplyEntry, Request};

use oxy_core::ids::NodeId;

use oxy_core::objects::NodeKind;

use oxy_core::repo::Repo;

use oxy_core::state::State;

use std::collections::HashMap;

use std::ffi::{OsStr, OsString};

use std::path::{Path, PathBuf};

use std::sync::Arc;

use std::time::{Duration, Instant, SystemTime};

/// How long the kernel may keep attributes and names: the mount never

/// changes.

const TTL: Duration = Duration::from_secs(3600);

struct Inode {

    parent: u64,

    kind: NodeKind,

    node: NodeId,

    exec: bool,

    opaque: bool,

    /// For a directory: its children by name.

    children: Vec<(OsString, u64)>,

}

struct CasFs {

    repo: Repo,

    state: State,

    inodes: Vec<Inode>,

    /// Bytes already read, by inode.

    bytes: HashMap<u64, Arc<Vec<u8>>>,

    cache_dir: PathBuf,

    uid: u32,

    gid: u32,

    mounted: SystemTime,

}

impl CasFs {

    fn new(repo: Repo, state: State) -> CasFs {

        let view = state.tree().view();

        let root = Inode { parent: 1, kind: NodeKind::Dir, node: NodeId::ROOT, exec: false, opaque: false, children: Vec::new() };

        let mut inodes = vec![root];

        let mut by_path: HashMap<&str, u64> = HashMap::with_capacity(view.entries.len());

        by_path.insert("", 1);

        // In path order, so a directory comes before what it holds.

        for (path, e) in view.entries.iter() {

            let (dir, name) = path.rsplit_once('/').unwrap_or(("", path.as_str()));

            let Some(&parent) = by_path.get(dir) else { continue };

            let ino = inodes.len() as u64 + 1;

            inodes.push(Inode { parent, kind: e.kind, node: e.node, exec: e.mode.exec, opaque: e.mode.opaque, children: Vec::new() });

            inodes[parent as usize - 1].children.push((OsString::from(name), ino));

            if e.kind == NodeKind::Dir || e.kind == NodeKind::Subrepo {

                by_path.insert(path.as_str(), ino);

            }

        }

        for i in inodes.iter_mut() {

            i.children.sort();

        }

        let cache_dir = repo.root().join(".oxy").join("cache").join("render");

        CasFs {

            repo,

            state,

            inodes,

            bytes: HashMap::new(),

            cache_dir,

            uid: unsafe { libc::getuid() },

            gid: unsafe { libc::getgid() },

            mounted: SystemTime::now(),

        }

    }

    fn inode(&self, ino: u64) -> Option<&Inode> {

        self.inodes.get(ino.checked_sub(1)? as usize)

    }

    /// A file's or symlink's bytes: its blob, or its rendering (cached on

    /// disk by graph key).

    fn content(&mut self, ino: u64) -> Result<Arc<Vec<u8>>> {

        if let Some(b) = self.bytes.get(&ino) {

            return Ok(b.clone());

        }

        let i = self.inode(ino).context("no such inode")?;

        let (node, kind, opaque, parent) = (i.node, i.kind, i.opaque, i.parent);

        // A file added once and never edited: read it with every such file

        // of its directory not read yet, in one batch (a directory is

        // usually read whole), on every core.

        if kind == NodeKind::File && !opaque && self.state.single_text(&node).is_some() {

            let siblings: Vec<(u64, oxy_core::objects::TextRef)> = self.inodes[parent as usize - 1]

                .children

                .iter()

                .map(|(_, c)| *c)

                .filter(|c| !self.bytes.contains_key(c))

                .filter_map(|c| {

                    let i = &self.inodes[c as usize - 1];

                    (i.kind == NodeKind::File && !i.opaque).then_some(())?;

                    Some((c, self.state.single_text(&i.node)?))

                })

                .collect();

            let texts: Vec<oxy_core::objects::TextRef> = siblings.iter().map(|(_, t)| *t).collect();

            for ((c, _), b) in siblings.iter().zip(self.repo.single_texts(&texts)?) {

                self.bytes.insert(*c, Arc::new(b));

            }

            if let Some(b) = self.bytes.get(&ino) {

                return Ok(b.clone());

            }

        }

        let bytes = if kind == NodeKind::File && !opaque {

            self.rendering(&node)?

        } else {

            match self.state.live_blobs(&node).first() {

                Some(b) => self.repo.read_blob(b)?,

                None => Vec::new(),

            }

        };

        let bytes = Arc::new(bytes);

        self.bytes.insert(ino, bytes.clone());

        Ok(bytes)

    }

    fn rendering(&self, node: &NodeId) -> Result<Vec<u8>> {

        // A file added once and never edited is its own text: no graph, and

        // nothing to cache.

        if let Some(t) = self.state.single_text(node) {

            return Ok(self.repo.single_texts(&[t])?.pop().unwrap_or_default());

        }

        let key = self.state.graph_key(node);

        let cached = key.map(|k| self.cache_dir.join(hex(&k)));

        if let Some(p) = &cached {

            if let Ok(b) = std::fs::read(p) {

                return Ok(b);

            }

        }

        let graphs = self.repo.build_graphs_detached(&self.state, &[*node])?;

        let bytes = match graphs.get(node) {

            Some(g) => self.state.render_graph(g).0,

            None => Vec::new(),

        };

        if let Some(p) = cached {

            // A cache: written whole or not at all, without waiting for

            // the disk.

            let _ = std::fs::create_dir_all(&self.cache_dir);

            let tmp = p.with_extension(format!("tmp-{}", std::process::id()));

            if std::fs::write(&tmp, &bytes).is_ok() {

                let _ = std::fs::rename(&tmp, &p);

            }

        }

        Ok(bytes)

    }

    fn attr(&mut self, ino: u64) -> Result<FileAttr> {

        let i = self.inode(ino).context("no such inode")?;

        let (kind, perm) = match i.kind {

            NodeKind::Dir | NodeKind::Subrepo => (FileType::Directory, 0o555),

            NodeKind::Symlink => (FileType::Symlink, 0o777),

            _ if i.exec => (FileType::RegularFile, 0o555),

            _ => (FileType::RegularFile, 0o444),

        };

        let nlink = if kind == FileType::Directory { 2 } else { 1 };

        let size = match kind {

            FileType::Directory => 0,

            _ => self.content(ino)?.len() as u64,

        };

        Ok(FileAttr {

            ino,

            size,

            blocks: size.div_ceil(512),

            atime: self.mounted,

            mtime: self.mounted,

            ctime: self.mounted,

            crtime: self.mounted,

            kind,

            perm,

            nlink,

            uid: self.uid,

            gid: self.gid,

            rdev: 0,

            blksize: 4096,

            flags: 0,

        })

    }

}

fn hex(b: &[u8]) -> String {

    b.iter().map(|x| format!("{x:02x}")).collect()

}

impl Filesystem for CasFs {

    fn lookup(&mut self, _req: &Request<'_>, parent: u64, name: &OsStr, reply: ReplyEntry) {

        let found =

            self.inode(parent).and_then(|p| p.children.binary_search_by(|(n, _)| n.as_os_str().cmp(name)).ok().map(|i| p.children[i].1));

        match found.map(|ino| self.attr(ino)) {

            Some(Ok(a)) => reply.entry(&TTL, &a, 0),

            Some(Err(_)) => reply.error(libc::EIO),

            None => reply.error(libc::ENOENT),

        }

    }

    fn getattr(&mut self, _req: &Request<'_>, ino: u64, _fh: Option<u64>, reply: ReplyAttr) {

        match self.attr(ino) {

            Ok(a) => reply.attr(&TTL, &a),

            Err(_) => reply.error(libc::ENOENT),

        }

    }

    fn readlink(&mut self, _req: &Request<'_>, ino: u64, reply: ReplyData) {

        match self.content(ino) {

            Ok(b) => reply.data(&b),

            Err(_) => reply.error(libc::EIO),

        }

    }

    #[allow(clippy::too_many_arguments)]

    fn read(&mut self, _req: &Request<'_>, ino: u64, _fh: u64, offset: i64, size: u32, _flags: i32, _lock: Option<u64>, reply: ReplyData) {

        match self.content(ino) {

            Ok(b) => {

                let start = (offset.max(0) as usize).min(b.len());

                let end = (start + size as usize).min(b.len());

                reply.data(&b[start..end]);

            }

            Err(_) => reply.error(libc::EIO),

        }

    }

    fn readdir(&mut self, _req: &Request<'_>, ino: u64, _fh: u64, offset: i64, mut reply: ReplyDirectory) {

        let Some(dir) = self.inode(ino) else {

            reply.error(libc::ENOENT);

            return;

        };

        let mut entries: Vec<(u64, FileType, OsString)> =

            vec![(ino, FileType::Directory, ".".into()), (dir.parent, FileType::Directory, "..".into())];

        for (name, child) in &dir.children {

            let kind = match self.inodes[*child as usize - 1].kind {

                NodeKind::Dir | NodeKind::Subrepo => FileType::Directory,

                NodeKind::Symlink => FileType::Symlink,

                _ => FileType::RegularFile,

            };

            entries.push((*child, kind, name.clone()));

        }

        for (i, (child, kind, name)) in entries.into_iter().enumerate().skip(offset.max(0) as usize) {

            if reply.add(child, (i + 1) as i64, kind, name) {

                break;

            }

        }

        reply.ok();

    }

}

fn main() -> Result<()> {

    let args: Vec<String> = std::env::args().collect();

    let usage = "usage: flow-casfs mount REPO MOUNTPOINT [--bookmark NAME]";

    if args.len() < 4 || args[1] != "mount" {

        bail!("{usage}");

    }

    let (repo_dir, mountpoint) = (Path::new(&args[2]), Path::new(&args[3]));

    let bookmark = args.iter().position(|a| a == "--bookmark").and_then(|i| args.get(i + 1)).cloned();

    let started = Instant::now();

    let mut repo = Repo::open(repo_dir).with_context(|| format!("open {}", repo_dir.display()))?;

    if let Some(b) = &bookmark {

        repo.config.bookmark = b.clone();

    }

    let heads = repo.bookmark_heads()?;

    let state = repo.load_state_indexed(&heads)?;

    let loaded = started.elapsed();

    let fs = CasFs::new(repo, state);

    eprintln!(

        "flow-casfs: {} entries; state loaded in {:.1} ms, tree built in {:.1} ms",

        fs.inodes.len(),

        loaded.as_secs_f64() * 1000.0,

        (started.elapsed() - loaded).as_secs_f64() * 1000.0

    );

    let options = [MountOption::RO, MountOption::FSName("flow-casfs".into()), MountOption::DefaultPermissions];

    let session = fuser::spawn_mount2(fs, mountpoint, &options).context("mount")?;

    eprintln!("flow-casfs: mounted at {} in {:.1} ms", mountpoint.display(), started.elapsed().as_secs_f64() * 1000.0);

    // Until the mount goes away (fusermount3 -u) or the process is stopped.

    session.guard.join().ok();

    Ok(())

}

#[cfg(test)]

mod tests {

    use super::*;

    /// The inode of a path, walking from the root.

    fn ino_of(fs: &CasFs, path: &str) -> Option<u64> {

        let mut ino = 1u64;

        for part in path.split('/') {

            let dir = fs.inode(ino)?;

            let i = dir.children.binary_search_by(|(n, _)| n.as_os_str().cmp(OsStr::new(part))).ok()?;

            ino = dir.children[i].1;

        }

        Some(ino)

    }

    #[test]

    fn a_mounted_state_holds_the_committed_files() {

        let dir = tempfile::tempdir().unwrap();

        std::env::set_var("OXY_HOME", dir.path().join("home"));

        let root = dir.path().join("repo");

        let repo = Repo::init(&root).unwrap();

        let big: String = (0..3000).map(|i| format!("line {i}\n")).collect();

        let files = [("a.txt", "one\ntwo\n".to_string()), ("src/lib.rs", "pub fn f() {}\n".to_string()), ("src/deep/big.txt", big)];

        for (p, c) in &files {

            let full = root.join(p);

            std::fs::create_dir_all(full.parent().unwrap()).unwrap();

            std::fs::write(full, c).unwrap();

        }

        repo.commit("first", &[]).unwrap();

        // An edit, so one file's rendering is a history, not one text.

        std::fs::write(root.join("a.txt"), "one\n2\ntwo\n").unwrap();

        repo.commit("second", &[]).unwrap();

        let heads = repo.bookmark_heads().unwrap();

        let state = repo.load_state_indexed(&heads).unwrap();

        let mut fs = CasFs::new(Repo::open(&root).unwrap(), state);

        for (p, c) in [("a.txt", "one\n2\ntwo\n"), ("src/lib.rs", "pub fn f() {}\n")] {

            let ino = ino_of(&fs, p).unwrap();

            assert_eq!(fs.content(ino).unwrap().as_slice(), c.as_bytes(), "{p}");

        }

        let big = ino_of(&fs, "src/deep/big.txt").unwrap();

        assert_eq!(fs.content(big).unwrap().as_slice(), files[2].1.as_bytes());

        assert_eq!(fs.attr(big).unwrap().size, files[2].1.len() as u64);

        assert_eq!(fs.attr(ino_of(&fs, "src/deep").unwrap()).unwrap().kind, FileType::Directory);

        assert!(ino_of(&fs, "missing.txt").is_none());

        // Only the edited file needed rendering, and its rendering is cached

        // by graph key, so a new mount reads it back; the others are their

        // own text blobs.

        let cached = std::fs::read_dir(&fs.cache_dir).unwrap().count();

        assert_eq!(cached, 1, "renderings cached");

        let state = repo.load_state_indexed(&heads).unwrap();

        let mut again = CasFs::new(Repo::open(&root).unwrap(), state);

        let ino = ino_of(&again, "a.txt").unwrap();

        assert_eq!(again.content(ino).unwrap().as_slice(), b"one\n2\ntwo\n");

    }

}

    rootfs "cas://blob/a3b49f36a2ad1838219ea0b7985591e44ac0e6d7f92e9d993d3a3738263fe32e"

    rootfs "cas://blob/a3b49f36a2ad1838219ea0b7985591e44ac0e6d7f92e9d993d3a3738263fe32e"

    exec "cargo" "fmt" "--check" "-p" "flow" "-p" "flow-core" "-p" "flow-init" "-p" "flow-casfs"

name = "cfg_aliases"

version = "0.2.2"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "f079e83a288787bcd14a6aea84cee5c87a67c5a3e660c30f557a3d24761b3527"

[[package]]

]

[[package]]

name = "flow-casfs"

version = "0.1.0"

dependencies = [

 "anyhow",

 "fuser",

 "libc",

 "oxy-core",

 "tempfile",

]

[[package]]

name = "fuser"

version = "0.16.0"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "0bb29a3ae32279fe3e79a958fe01899f5fb23eadccee919cf88e145b54ed9367"

dependencies = [

 "libc",

 "log",

 "memchr",

 "nix",

 "page_size",

 "smallvec",

 "zerocopy",

]

[[package]]

name = "nix"

version = "0.29.0"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "71e2746dc3a24dd78b3cfcb7be93368c6de9963d30f43a6a73998a9cf4b17b46"

dependencies = [

 "bitflags",

 "cfg-if",

 "cfg_aliases",

 "libc",

]

[[package]]

name = "page_size"

version = "0.6.0"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "30d5b2194ed13191c1999ae0704b7839fb18384fa22e49b57eeaa97d79ce40da"

dependencies = [

 "libc",

 "winapi",

]

[[package]]

name = "winapi"

version = "0.3.9"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "5c839a674fcd7a98952e593242ea400abe93992746761e38641405d28b00f419"

dependencies = [

 "winapi-i686-pc-windows-gnu",

 "winapi-x86_64-pc-windows-gnu",

name = "winapi-i686-pc-windows-gnu"

version = "0.4.0"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "ac3b87c63620426dd9b991e5ce0329eff545bccbbb34f3be09ff6fb6ab51b7b6"

[[package]]

[[package]]

name = "winapi-x86_64-pc-windows-gnu"

version = "0.4.0"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f"

members = ["crates/flow-core", "crates/flow-init", "crates/flow", "crates/flow-casfs"]