gunnar-lfs 1.6.0

//! The on-disk backend: LFS objects **inside the repository's own git dir**.
//!
//! ```text
//! <git_dir>/
//!   objects/            ← gunnar_store::fs::FsStore, git objects
//!   refs/               ← gunnar_store::fs::FsRefStore
//!   lfs/
//!     objects/ab/cd/abcd…   ← here. raw bytes, no header, no compression
//!     incomplete/<tmp>      ← uploads in flight; renamed on verification
//! ```
//!
//! # Why inside the git dir and not a second root
//!
//! Because a repository has to stay **one directory**. `gunnar mirror`, the
//! `repack()` atomic-swap protocol and every backup story in the tree take a
//! repository to be its git dir; a second data root would be a second thing to
//! configure, secure, replicate and forget, and the first time it was forgotten
//! the symptom would be a clone that succeeds and then cannot check out.
//!
//! It is also the same fan-out layout git-lfs uses in a working copy
//! (`.git/lfs/objects/<oid[0..2]>/<oid[2..4]>/<oid>`), so the bytes are readable
//! by anything that already knows LFS.
//!
//! # The rename is the whole safety property
//!
//! An upload is written to `lfs/incomplete/` and `rename`d into `lfs/objects/`
//! **only after** [`crate::session`] has verified the SHA-256 and the byte
//! count. `rename(2)` within one filesystem is atomic, so an object is either
//! absent or correct: a truncated upload, a lying client and a severed
//! connection all leave a file in `incomplete/` that no oid resolves to. There
//! is no window in which a wrong-content object is visible under a valid oid.
//! The rename (and a delete's unlink) is reported done only after the
//! directory holding the name is fsynced, so it also survives a power loss.
//!
//! # No compression, and it is measured
//!
//! Level-6 DEFLATE on large high-entropy content: **ratio 1.000 at 48.2 MB/s**.
//! SHA-1 over the same bytes: **2027 MB/s**. LFS content is video, images,
//! archives and shipped binaries, all compressed already, so compressing it
//! again is 42× slower than the hash that must run anyway and saves nothing.
//! Files here are
//! byte-identical to what was uploaded.

use std::fs::File;
use std::io::Write;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};

use crate::error::{Error, Result};
use crate::oid::Oid;
use crate::store::{LfsObject, LfsStore, PendingWrite};

/// The subdirectory of a git dir that LFS owns.
pub const LFS_DIR: &str = "lfs";

/// LFS objects on disk, under a repository's git dir.
#[derive(Debug, Clone)]
pub struct FsLfsStore {
    root: PathBuf,
}

/// Names temporary files. Process id plus a counter, so two concurrent uploads
/// of the same oid, which is legal and happens, cannot collide on a temp name
/// and truncate each other mid-write.
static TEMP_COUNTER: AtomicU64 = AtomicU64::new(0);

impl FsLfsStore {
    /// Open (creating if needed) the LFS store under `git_dir`.
    ///
    /// Takes the git dir rather than the lfs dir, so callers cannot disagree
    /// about where inside a repository LFS lives.
    pub fn open_or_init(git_dir: impl AsRef<Path>) -> Result<Self> {
        let root = git_dir.as_ref().join(LFS_DIR);
        for dir in ["objects", "incomplete"] {
            std::fs::create_dir_all(root.join(dir)).map_err(|err| {
                Error::store_io(
                    format!("could not create {}", root.join(dir).display()),
                    err,
                )
            })?;
        }
        Ok(FsLfsStore { root })
    }

    /// The `lfs/` directory this store owns.
    pub fn root(&self) -> &Path {
        &self.root
    }

    /// Where an object lives once committed.
    ///
    /// The oid is used verbatim as three path components, which is safe because
    /// [`Oid`] cannot hold anything but 64 lowercase hex characters. The check
    /// is in the type, not repeated here (one implementation).
    pub fn path_of(&self, oid: &Oid) -> PathBuf {
        let hex = oid.to_hex();
        self.root
            .join("objects")
            .join(oid.hex_prefix(2))
            .join(&hex[2..4])
            .join(&hex)
    }
}

impl LfsStore for FsLfsStore {
    fn delete(&self, oid: &Oid) -> Result<bool> {
        let path = self.path_of(oid);
        match std::fs::remove_file(&path) {
            Ok(()) => {
                // The unlink is durable only once its directory is: a crash before that
                // brings a deleted object back.
                if let Some(parent) = path.parent() {
                    sync_dir(parent)?;
                }
                Ok(true)
            }
            Err(err) if err.kind() == std::io::ErrorKind::NotFound => Ok(false),
            Err(err) => Err(Error::store_io(format!("remove {oid}"), err)),
        }
    }

    fn size_of(&self, oid: &Oid) -> Result<Option<u64>> {
        match std::fs::metadata(self.path_of(oid)) {
            Ok(meta) => Ok(Some(meta.len())),
            Err(err) if err.kind() == std::io::ErrorKind::NotFound => Ok(None),
            Err(err) => Err(Error::store_io(format!("stat {oid}"), err)),
        }
    }

    fn open(&self, oid: &Oid) -> Result<Option<LfsObject>> {
        let path = self.path_of(oid);
        let file = match File::open(&path) {
            Ok(file) => file,
            Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(None),
            Err(err) => return Err(Error::store_io(format!("open {oid}"), err)),
        };
        let size = file
            .metadata()
            .map_err(|err| Error::store_io(format!("stat {oid}"), err))?
            .len();
        // A plain `File`, not a `BufReader`: the caller copies in
        // `CHUNK_BYTES` slabs, so a second buffer under it would only add a
        // memcpy per chunk.
        Ok(Some(LfsObject {
            size,
            reader: Box::new(file),
        }))
    }

    /// A file seeks: a range deep into a large object costs one `lseek`, not a
    /// read of everything before it.
    fn open_at(&self, oid: &Oid, offset: u64) -> Result<Option<LfsObject>> {
        use std::io::Seek as _;
        let path = self.path_of(oid);
        let mut file = match File::open(&path) {
            Ok(file) => file,
            Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(None),
            Err(err) => return Err(Error::store_io(format!("open {oid}"), err)),
        };
        let size = file
            .metadata()
            .map_err(|err| Error::store_io(format!("stat {oid}"), err))?
            .len();
        file.seek(std::io::SeekFrom::Start(offset.min(size)))
            .map_err(|err| Error::store_io(format!("seek to {offset} in {oid}"), err))?;
        Ok(Some(LfsObject {
            size,
            reader: Box::new(file),
        }))
    }

    fn begin_write(&self, oid: &Oid, _declared_size: u64) -> Result<Box<dyn PendingWrite>> {
        let seq = TEMP_COUNTER.fetch_add(1, Ordering::Relaxed);
        let temp = self
            .root
            .join("incomplete")
            .join(format!("{}-{seq}.tmp", std::process::id()));
        let file = File::create(&temp)
            .map_err(|err| Error::store_io(format!("create {}", temp.display()), err))?;
        Ok(Box::new(FsPendingWrite {
            final_path: self.path_of(oid),
            temp: Some(temp),
            file: Some(file),
        }))
    }
}

struct FsPendingWrite {
    final_path: PathBuf,
    temp: Option<PathBuf>,
    file: Option<File>,
}

impl Write for FsPendingWrite {
    fn write(&mut self, data: &[u8]) -> std::io::Result<usize> {
        match self.file.as_mut() {
            Some(file) => file.write(data),
            None => Err(std::io::Error::other("write after commit")),
        }
    }

    fn flush(&mut self) -> std::io::Result<()> {
        match self.file.as_mut() {
            Some(file) => file.flush(),
            None => Ok(()),
        }
    }
}

impl PendingWrite for FsPendingWrite {
    fn commit(mut self: Box<Self>) -> Result<()> {
        let mut file = self
            .file
            .take()
            .ok_or_else(|| Error::Store("commit called twice".to_string()))?;
        file.flush().map_err(|err| Error::store_io("flush", err))?;
        // The data must be on the medium before the name that promises it is.
        // Without this, a crash between rename and writeback leaves a
        // correctly-named file with a hole in it, and the oid says it is good.
        file.sync_all()
            .map_err(|err| Error::store_io("fsync", err))?;
        drop(file);

        let temp = self
            .temp
            .take()
            .ok_or_else(|| Error::Store("commit called twice".to_string()))?;
        // Every directory that gets a new entry: the object's own, and the
        // parent of each fan-out directory created for it.
        let mut to_sync: Vec<PathBuf> = Vec::new();
        if let Some(parent) = self.final_path.parent() {
            to_sync.push(parent.to_path_buf());
            let mut dir = parent;
            while !dir.exists() {
                match dir.parent() {
                    Some(up) => {
                        to_sync.push(up.to_path_buf());
                        dir = up;
                    }
                    None => break,
                }
            }
            std::fs::create_dir_all(parent)
                .map_err(|err| Error::store_io(format!("create {}", parent.display()), err))?;
        }
        // Atomic within a filesystem, and `incomplete/` is a sibling of
        // `objects/` under the same root precisely so it always is one.
        std::fs::rename(&temp, &self.final_path).map_err(|err| {
            Error::store_io(
                format!("rename {} → {}", temp.display(), self.final_path.display()),
                err,
            )
        })?;
        // The rename is durable only once the directory holding the new name
        // is: before this, a power loss after a reported commit could lose the
        // object (stinky row 5).
        for dir in &to_sync {
            sync_dir(dir)?;
        }
        Ok(())
    }
}

impl Drop for FsPendingWrite {
    /// An upload that was not committed leaves nothing behind.
    ///
    /// This is what makes "verified or absent" true for a severed connection as
    /// well as for a refused one: the session never gets to call `commit`, the
    /// value is dropped as the stack unwinds, and the partial file goes with it.
    fn drop(&mut self) {
        self.file.take();
        if let Some(temp) = self.temp.take() {
            drop_incomplete(&temp);
        }
    }
}

/// Remove an abandoned upload, and **say so** if it will not go.
///
/// # Why this one is not a `let _ =`
///
/// Nothing else in the tree ever looks at `lfs/incomplete/`. `repack()` sweeps
/// its own scratch by prefix and `PackFiles::all` collects a leaked `.keep`, so
/// those swallows cost a stale marker; this one costs **the bytes**. A rejected
/// or severed upload that cannot be unlinked stays on disk at full size, under a
/// name no oid resolves to, until somebody notices the disk is full. At the
/// stated load — a forge taking LFS pushes — that is monotonic, and a silent
/// monotonic leak is the thing an operator cannot find.
///
/// Still best-effort rather than fallible: this runs in `Drop`, which has
/// nowhere to return to, and the upload has already failed. What was missing
/// was not a `Result`, it was a line in the journal.
///
/// Silent when the file is already gone, because that is the outcome it wanted.
///
/// # Not `gunnar_store::drop_keep`, which is the same six lines
///
/// Reaching it would make `gunnar-lfs` depend on `gunnar-store`, and this
/// crate's dependency budget (`Cargo.toml`'s own argument) is
/// `gunnar-pktline` + `gix-hash` + `thiserror` + `tracing` and nothing else. A
/// whole storage crate is a poor trade for six lines, and the two functions
/// answer to different modules' failures.
fn drop_incomplete(temp: &Path) {
    match std::fs::remove_file(temp) {
        Ok(()) => {}
        Err(err) if err.kind() == std::io::ErrorKind::NotFound => {}
        Err(err) => tracing::error!(
            target: crate::LFS_TARGET,
            path = %temp.display(),
            %err,
            "an abandoned LFS upload could not be removed; it is still occupying its full size \
             under a name no oid resolves to, and nothing else sweeps this directory"
        ),
    }
}

/// The test seam over directory fsyncs: which directories were made durable,
/// and an injected failure. Power loss itself is not unit-testable; what is,
/// is that the fsync is ASKED for and that its failure is not swallowed.
/// **fsync a directory**, so a rename into it or an unlink from it survives a
/// power loss. Its failure is the operation's failure.
fn sync_dir(dir: &Path) -> Result<()> {
    #[cfg(test)]
    if let Some(err) = dir_sync_seam::on_sync(dir) {
        return Err(Error::store_io(format!("fsync {}", dir.display()), err));
    }
    File::open(dir)
        .and_then(|d| d.sync_all())
        .map_err(|err| Error::store_io(format!("fsync {}", dir.display()), err))
}

#[cfg(test)]
pub(crate) mod dir_sync_seam {
    use std::cell::RefCell;
    use std::path::{Path, PathBuf};

    thread_local! {
        static FAIL: RefCell<bool> = const { RefCell::new(false) };
        static SYNCED: RefCell<Vec<PathBuf>> = const { RefCell::new(Vec::new()) };
    }

    /// Make every later directory fsync on this thread fail.
    pub(crate) fn fail_from_now() {
        FAIL.with(|f| *f.borrow_mut() = true);
    }

    /// The directories fsynced on this thread so far.
    pub(crate) fn synced() -> Vec<PathBuf> {
        SYNCED.with(|s| s.borrow().clone())
    }

    /// Called by the directory fsync before it touches the disk.
    pub(crate) fn on_sync(dir: &Path) -> Option<std::io::Error> {
        SYNCED.with(|s| s.borrow_mut().push(dir.to_path_buf()));
        FAIL.with(|f| *f.borrow())
            .then(|| std::io::Error::other("injected directory fsync failure"))
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn store() -> (tempfile::TempDir, FsLfsStore) {
        let dir = tempfile::tempdir().expect("tempdir");
        let store = FsLfsStore::open_or_init(dir.path().join("repo.git")).expect("store");
        (dir, store)
    }

    #[test]
    fn an_object_lands_under_the_git_dir_in_the_fan_out_layout() {
        let (dir, store) = store();
        let oid =
            Oid::parse("e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855").unwrap();
        let path = store.path_of(&oid);
        assert!(path.starts_with(dir.path().join("repo.git").join("lfs").join("objects")));
        assert!(path.to_string_lossy().contains("/e3/b0/e3b0c442"));
    }

    #[test]
    fn bytes_are_stored_raw_and_uncompressed() {
        let (_dir, store) = store();
        let payload = b"the quick brown fox".repeat(100);
        let oid = {
            let mut hasher = crate::oid::Sha256Stream::new();
            hasher.update(&payload);
            hasher.finish()
        };
        let mut pending = store.begin_write(&oid, payload.len() as u64).unwrap();
        pending.write_all(&payload).unwrap();
        pending.commit().unwrap();

        // Byte-identical on disk. Highly compressible input, deliberately: if
        // anything ever started deflating, this is the assertion that goes red.
        let on_disk = std::fs::read(store.path_of(&oid)).unwrap();
        assert_eq!(on_disk, payload);
        assert_eq!(store.size_of(&oid).unwrap(), Some(payload.len() as u64));
    }

    #[test]
    fn a_dropped_upload_leaves_nothing_visible_and_nothing_behind() {
        let (_dir, store) = store();
        let oid = Oid::parse(&"d".repeat(64)).unwrap();
        {
            let mut pending = store.begin_write(&oid, 5).unwrap();
            pending.write_all(b"partl").unwrap();
        }
        assert!(store.open(&oid).unwrap().is_none());
        let incomplete: Vec<_> = std::fs::read_dir(store.root().join("incomplete"))
            .unwrap()
            .map(|e| e.unwrap().file_name())
            .collect();
        assert!(
            incomplete.is_empty(),
            "a dropped upload left {incomplete:?} behind"
        );
    }

    #[test]
    fn two_concurrent_uploads_of_the_same_oid_do_not_collide() {
        // Two temp files, both committed. The second rename replaces the first,
        // which is correct, same oid meaning same bytes, but they must not have
        // been writing to the same temp path, which would interleave two
        // streams into one file.
        let (_dir, store) = store();
        let payload = b"same bytes".to_vec();
        let oid = {
            let mut hasher = crate::oid::Sha256Stream::new();
            hasher.update(&payload);
            hasher.finish()
        };
        let mut a = store.begin_write(&oid, payload.len() as u64).unwrap();
        let mut b = store.begin_write(&oid, payload.len() as u64).unwrap();
        a.write_all(&payload).unwrap();
        b.write_all(&payload).unwrap();
        a.commit().unwrap();
        b.commit().unwrap();
        assert_eq!(std::fs::read(store.path_of(&oid)).unwrap(), payload);
    }

    // ── a rename or an unlink is durable only once its directory is (stinky row 5) ──

    fn committed(store: &FsLfsStore, payload: &[u8]) -> Oid {
        let oid = {
            let mut hasher = crate::oid::Sha256Stream::new();
            hasher.update(payload);
            hasher.finish()
        };
        let mut pending = store.begin_write(&oid, payload.len() as u64).unwrap();
        pending.write_all(payload).unwrap();
        pending.commit().unwrap();
        oid
    }

    /// ★ The rename that publishes an object is followed by an fsync of the
    /// directory it landed in: until then, power loss can take the name away
    /// while the client was already told the upload is stored.
    #[test]
    fn a_commit_makes_the_new_name_durable() {
        let (_dir, store) = store();
        let oid = committed(&store, b"durable bytes");
        let parent = store.path_of(&oid).parent().unwrap().to_path_buf();
        let synced = dir_sync_seam::synced();
        assert!(
            synced.contains(&parent),
            "the object's directory was never fsynced: {synced:?}"
        );
    }

    /// …and a failing directory fsync fails the commit instead of reporting
    /// the object stored.
    #[test]
    fn a_failed_directory_fsync_fails_the_commit() {
        let (_dir, store) = store();
        let payload = b"not yet durable".to_vec();
        let oid = {
            let mut hasher = crate::oid::Sha256Stream::new();
            hasher.update(&payload);
            hasher.finish()
        };
        let mut pending = store.begin_write(&oid, payload.len() as u64).unwrap();
        pending.write_all(&payload).unwrap();
        dir_sync_seam::fail_from_now();
        assert!(
            pending.commit().is_err(),
            "a commit whose directory fsync failed reported success"
        );
    }

    /// ★ The unlink that deletes an object is made durable too, and its
    /// directory fsync's failure is the delete's failure.
    #[test]
    fn a_delete_makes_the_unlink_durable_and_says_so_when_it_cannot() {
        let (_dir, store) = store();
        let oid = committed(&store, b"to be deleted");
        let parent = store.path_of(&oid).parent().unwrap().to_path_buf();
        let before = dir_sync_seam::synced()
            .iter()
            .filter(|d| **d == parent)
            .count();
        assert!(store.delete(&oid).unwrap());
        let after = dir_sync_seam::synced()
            .iter()
            .filter(|d| **d == parent)
            .count();
        assert!(
            after > before,
            "the delete never fsynced the object's directory"
        );

        let other = committed(&store, b"also to be deleted");
        dir_sync_seam::fail_from_now();
        assert!(
            store.delete(&other).is_err(),
            "a delete whose directory fsync failed reported success"
        );
    }
}