crates/gunnar-lfs/src/fs_store.rs
//! 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"
);
}
}