strata.rs 85.1%

808 of 950 lines run, 151 of 262 functions · run #327 on main · run missed
//! Oxygen Strata (docs/stores.md): the storage a repository can be kept
//! on, as a trait, so that a bucket, a directory or a hub's disk all look
//! the same to the code that pushes to and fetches from them.
//!
//! A store has two halves. **Objects** are written once and never
//! change: packs and their indexes, named by content, which any store
//! can hold safely. **References** are the one mutable record (bookmarks,
//! tags, the policy), updated only by compare-and-swap on a version the
//! store hands out: an ETag on S3, a generation on GCS, a content hash on
//! a directory. Two writers can therefore share a store with no lock
//! server: the loser of a race reads again, merges, and tries again.
//!
//! A repository's own objects (`.oxy/objects`, see [`crate::store`]) are
//! kept on a store too, a directory one, with the same keys a bucket
//! would hold; a remote is a store as well. This module holds the trait,
//! a directory store, an in-memory store for tests, and the conformance
//! check every store must pass. Cloud stores live in a crate of their
//! own, because of what they pull in.

use std::collections::BTreeMap;
use std::fs;
use std::io::{self, Read, Seek, SeekFrom, Write};
use std::ops::Range;
use std::path::{Path, PathBuf};
use std::sync::Mutex;
use std::time::{Duration, Instant};

/// The version of a reference record as the store reports it: opaque,
/// compared only for equality, and only ever valid for the store that
/// issued it. Writing the same bytes again may keep the version (S3's
/// ETag and a directory's content hash are both of the content): a
/// compare-and-swap guards against losing a *different* record, which is
/// all it needs to.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Version(pub String);

/// One key in a listing.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Entry {
    pub key: String,
    pub size: u64,
    /// When the object was written, in wall milliseconds, where the store
    /// knows (garbage collection's grace period reads it).
    pub modified_ms: Option<u64>,
}

#[derive(Debug, thiserror::Error)]
pub enum Error {
    #[error("{0}: not in the store")]
    NotFound(String),
    #[error("{0}: already in the store")]
    Exists(String),
    #[error("{0}: changed since it was read")]
    Conflict(String),
    #[error("bad store key {0:?}: {1}")]
    Key(String, &'static str),
    #[error("{0}: busy (another writer holds it)")]
    Busy(String),
    #[error(transparent)]
    Io(#[from] io::Error),
    #[error("{0}")]
    Backend(String),
}

pub type Result<T> = std::result::Result<T, Error>;

/// Longest key a store accepts; S3 allows 1024 bytes.
pub const MAX_KEY_LEN: usize = 1024;

/// Refuse a key that is not a plain relative path: `/`-separated
/// segments of `[A-Za-z0-9._-]`, none empty, none `.` or `..`. The same
/// rule on every store, so a key written to one can be read from another,
/// and so a key can never escape a directory store's root.
pub fn check_key(key: &str) -> Result<()> {
    let bad = |why| Error::Key(key.to_string(), why);
    if key.is_empty() {
        return Err(bad("empty"));
    }
    if key.len() > MAX_KEY_LEN {
        return Err(bad("longer than 1024 bytes"));
    }
    for seg in key.split('/') {
        if seg.is_empty() {
            return Err(bad(
                "an empty segment (a leading, trailing or doubled slash)",
            ));
        }
        if seg == "." || seg == ".." {
            return Err(bad("a . or .. segment"));
        }
        if !seg
            .bytes()
            .all(|b| b.is_ascii_alphanumeric() || b == b'.' || b == b'_' || b == b'-')
        {
            return Err(bad("a character outside A-Z a-z 0-9 . _ -"));
        }
    }
    Ok(())
}

/// Where a repository's packs and references are kept. Every method is
/// synchronous and may be called from several threads at once.
pub trait Store: Send + Sync {
    /// A name for messages: the URL, or the directory.
    fn describe(&self) -> String;

    // ---- objects: written once, never changed

    fn has(&self, key: &str) -> Result<bool>;
    /// The whole object; `NotFound` when there is none.
    fn get(&self, key: &str) -> Result<Vec<u8>>;
    /// Bytes `range` of an object; an error when the range runs past its end.
    fn get_range(&self, key: &str, range: Range<u64>) -> Result<Vec<u8>>;
    /// Write an object that does not exist yet; `Exists` when it does, and
    /// then nothing is changed. Readers never see a partial object.
    fn put_new(&self, key: &str, bytes: &[u8]) -> Result<()>;
    /// Every key under `prefix` (a key prefix, not necessarily a whole
    /// segment), sorted, with sizes.
    fn list(&self, prefix: &str) -> Result<Vec<Entry>>;
    /// Remove an object; fine when it is already gone.
    fn delete(&self, key: &str) -> Result<()>;
    /// One object's listing entry without listing its neighbours; `None`
    /// when there is none.
    fn stat(&self, key: &str) -> Result<Option<Entry>> {
        Ok(self.list(key)?.into_iter().find(|e| e.key == key))
    }
    /// Write several objects that do not exist yet; one that does is left
    /// as it is, which is not an error (objects are named by content). A
    /// store that can batch its syncs does; this does them one at a time.
    fn put_many_new(&self, objects: &[(String, Vec<u8>)]) -> Result<()> {
        for (key, bytes) in objects {
            match self.put_new(key, bytes) {
                Ok(()) | Err(Error::Exists(_)) => {}
                Err(e) => return Err(e),
            }
        }
        Ok(())
    }
    /// `put_new` of a file's contents: a store on the same disk can move
    /// the file into place instead of copying it.
    fn put_new_file(&self, key: &str, path: &Path) -> Result<()> {
        self.put_new(key, &fs::read(path)?)
    }

    // ---- references: one small record, updated by compare-and-swap

    /// The record and its current version, or `None` when it was never
    /// written.
    fn read_ref(&self, key: &str) -> Result<Option<(Vec<u8>, Version)>>;
    /// Replace the record only if its version is still `expected`
    /// (`None`: only if it does not exist yet). `Conflict` otherwise, and
    /// then nothing is changed. Returns the new version.
    fn write_ref(&self, key: &str, bytes: &[u8], expected: Option<&Version>) -> Result<Version>;
}

/// A reference record's version on stores that have no native one: the
/// hash of its bytes, which two different records never share.
fn content_version(bytes: &[u8]) -> Version {
    Version(blake3::hash(bytes).to_hex().to_string())
}

fn check_expected(key: &str, current: Option<&Version>, expected: Option<&Version>) -> Result<()> {
    if current == expected {
        Ok(())
    } else {
        Err(Error::Conflict(key.to_string()))
    }
}

// ---------------------------------------------------------------- directory

/// A store in a directory: each key a file under the root. Create-only
/// writes go through a hard link, which the file system refuses when the
/// name is taken, and a reference is updated under a lock file beside it.
/// Several processes on one machine can share it; it is also what the
/// tests and `oxy remote check` use.
pub struct LocalStore {
    root: PathBuf,
}

/// How long a reference's lock may be held before it counts as abandoned
/// by a process that died, and how long a writer waits for it.
const LOCK_STALE: Duration = Duration::from_secs(60);
const LOCK_WAIT: Duration = Duration::from_secs(10);

impl LocalStore {
    /// Open the store at `root`, creating the directory if needed.
    pub fn open(root: impl Into<PathBuf>) -> Result<LocalStore> {
        let root = root.into();
        fs::create_dir_all(&root)?;
        Ok(LocalStore { root })
    }

    pub fn root(&self) -> &Path {
        &self.root
    }

    fn path(&self, key: &str) -> Result<PathBuf> {
        check_key(key)?;
        Ok(self.root.join(key))
    }

    fn temp_beside(&self, path: &Path) -> PathBuf {
        let mut nonce = [0u8; 6];
        rand::RngCore::fill_bytes(&mut rand::rngs::OsRng, &mut nonce);
        let name = path
            .file_name()
            .map(|n| n.to_string_lossy().into_owned())
            .unwrap_or_default();
        path.with_file_name(format!(".{name}.tmp-{}", hex::encode(nonce)))
    }

    fn list_dir(&self, dir: &Path, rel: &str, prefix: &str, out: &mut Vec<Entry>) -> Result<()> {
        let entries = match fs::read_dir(dir) {
            Ok(e) => e,
            Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(()),
            Err(e) => return Err(e.into()),
        };
        for entry in entries {
            let entry = entry?;
            let name = entry.file_name().to_string_lossy().into_owned();
            // Temporary and lock files are not objects.
            if name.starts_with('.') {
                continue;
            }
            let key = if rel.is_empty() {
                name.clone()
            } else {
                format!("{rel}/{name}")
            };
            let ty = entry.file_type()?;
            if ty.is_dir() {
                // Descend only where the prefix could still match.
                let dir_prefix = format!("{key}/");
                if prefix.starts_with(&dir_prefix) || dir_prefix.starts_with(prefix) {
                    self.list_dir(&entry.path(), &key, prefix, out)?;
                }
            } else if ty.is_file() && key.starts_with(prefix) {
                let meta = entry.metadata()?;
                out.push(Entry {
                    key,
                    size: meta.len(),
                    modified_ms: modified_ms(&meta),
                });
            }
        }
        Ok(())
    }

    /// Take `path.lock`, waiting for another writer, and removing a lock
    /// a dead process left behind.
    fn lock(&self, key: &str, path: &Path) -> Result<PathBuf> {
        let lock = path.with_file_name(format!(
            ".{}.lock",
            path.file_name()
                .map(|n| n.to_string_lossy().into_owned())
                .unwrap_or_default()
        ));
        let started = Instant::now();
        loop {
            match fs::OpenOptions::new()
                .write(true)
                .create_new(true)
                .open(&lock)
            {
                Ok(mut f) => {
                    let _ = writeln!(f, "{}", std::process::id());
                    return Ok(lock);
                }
                Err(e) if e.kind() == io::ErrorKind::AlreadyExists => {
                    if let Ok(m) = fs::metadata(&lock) {
                        let age = m.modified().ok().and_then(|t| t.elapsed().ok());
                        if age.is_some_and(|a| a > LOCK_STALE) {
                            let _ = fs::remove_file(&lock);
                            continue;
                        }
                    }
                    if started.elapsed() > LOCK_WAIT {
                        return Err(Error::Busy(key.to_string()));
                    }
                    std::thread::sleep(Duration::from_millis(10));
                }
                Err(e) => return Err(e.into()),
            }
        }
    }
}

/// A file's modification time as wall milliseconds.
fn modified_ms(meta: &fs::Metadata) -> Option<u64> {
    meta.modified()
        .ok()
        .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
        .map(|d| d.as_millis() as u64)
}

impl LocalStore {
    /// Link `tmp` into `path`, create-only; the directory is synced and
    /// `tmp` removed either way. `Exists` when the name is taken.
    fn link_into_place(&self, key: &str, tmp: &Path, path: &Path) -> Result<()> {
        let linked = fs::hard_link(tmp, path);
        let _ = fs::remove_file(tmp);
        match linked {
            Ok(()) => {
                if let Some(dir) = path.parent() {
                    crate::durable::sync_dir(dir)?;
                }
                Ok(())
            }
            Err(e) if e.kind() == io::ErrorKind::AlreadyExists => {
                Err(Error::Exists(key.to_string()))
            }
            Err(e) => Err(e.into()),
        }
    }

    /// Create `path`'s directory, syncing the first directory that was
    /// made so its entry is durable too.
    fn ensure_dir(&self, path: &Path) -> Result<()> {
        let Some(dir) = path.parent() else {
            return Ok(());
        };
        if dir.is_dir() {
            return Ok(());
        }
        let mut made = dir.to_path_buf();
        while let Some(parent) = made.parent() {
            if parent.is_dir() || parent == self.root {
                break;
            }
            made = parent.to_path_buf();
        }
        fs::create_dir_all(dir)?;
        if let Some(parent) = made.parent() {
            crate::durable::sync_dir(parent)?;
        }
        Ok(())
    }
}

impl Store for LocalStore {
    fn describe(&self) -> String {
        self.root.display().to_string()
    }

    fn stat(&self, key: &str) -> Result<Option<Entry>> {
        match fs::metadata(self.path(key)?) {
            Ok(m) if m.is_file() => Ok(Some(Entry {
                key: key.to_string(),
                size: m.len(),
                modified_ms: modified_ms(&m),
            })),
            Ok(_) => Ok(None),
            Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(None),
            Err(e) => Err(e.into()),
        }
    }

    /// Every file written and synced (on every core), then each linked
    /// into place and each directory synced once: durable when this
    /// returns, as one at a time would be, with far fewer waits.
    fn put_many_new(&self, objects: &[(String, Vec<u8>)]) -> Result<()> {
        use rayon::prelude::*;
        let mut todo: Vec<(PathBuf, PathBuf, &[u8])> = Vec::new();
        for (key, bytes) in objects {
            let path = self.path(key)?;
            if path.exists() || todo.iter().any(|(p, ..)| *p == path) {
                continue;
            }
            self.ensure_dir(&path)?;
            todo.push((path.clone(), self.temp_beside(&path), bytes));
        }
        if todo.is_empty() {
            return Ok(());
        }
        crate::durable::step("write loose objects")?;
        todo.par_iter()
            .try_for_each(|(_, tmp, bytes)| -> io::Result<()> {
                let mut f = fs::File::create(tmp)?;
                f.write_all(bytes)?;
                f.sync_all()
            })?;
        crate::durable::step("rename loose objects")?;
        let mut dirs: std::collections::BTreeSet<PathBuf> = std::collections::BTreeSet::new();
        for (path, tmp, _) in &todo {
            let linked = fs::hard_link(tmp, path);
            let _ = fs::remove_file(tmp);
            match linked {
                Ok(()) => {
                    if let Some(d) = path.parent() {
                        dirs.insert(d.to_path_buf());
                    }
                }
                Err(e) if e.kind() == io::ErrorKind::AlreadyExists => {}
                Err(e) => return Err(e.into()),
            }
        }
        dirs.par_iter()
            .try_for_each(|d| crate::durable::sync_dir(d))?;
        Ok(())
    }

    /// The file is linked into place when it is on this file system,
    /// else copied; it is removed either way.
    fn put_new_file(&self, key: &str, file: &Path) -> Result<()> {
        let path = self.path(key)?;
        if path.exists() {
            let _ = fs::remove_file(file);
            return Err(Error::Exists(key.to_string()));
        }
        self.ensure_dir(&path)?;
        match fs::hard_link(file, &path) {
            Ok(()) => {
                let _ = fs::remove_file(file);
                if let Some(dir) = path.parent() {
                    crate::durable::sync_dir(dir)?;
                }
                Ok(())
            }
            Err(e) if e.kind() == io::ErrorKind::AlreadyExists => {
                let _ = fs::remove_file(file);
                Err(Error::Exists(key.to_string()))
            }
            Err(_) => {
                // Another file system: copy beside the name, then link.
                let tmp = self.temp_beside(&path);
                let copied = (|| {
                    let mut src = fs::File::open(file)?;
                    let mut dst = fs::File::create(&tmp)?;
                    io::copy(&mut src, &mut dst)?;
                    dst.sync_all()
                })();
                let _ = fs::remove_file(file);
                copied?;
                self.link_into_place(key, &tmp, &path)
            }
        }
    }

    fn has(&self, key: &str) -> Result<bool> {
        Ok(self.path(key)?.is_file())
    }

    fn get(&self, key: &str) -> Result<Vec<u8>> {
        let path = self.path(key)?;
        match fs::read(&path) {
            Ok(b) => Ok(b),
            Err(e) if e.kind() == io::ErrorKind::NotFound => Err(Error::NotFound(key.to_string())),
            Err(e) => Err(e.into()),
        }
    }

    fn get_range(&self, key: &str, range: Range<u64>) -> Result<Vec<u8>> {
        let path = self.path(key)?;
        let mut f = match fs::File::open(&path) {
            Ok(f) => f,
            Err(e) if e.kind() == io::ErrorKind::NotFound => {
                return Err(Error::NotFound(key.to_string()))
            }
            Err(e) => return Err(e.into()),
        };
        let len = f.metadata()?.len();
        if range.end > len || range.start > range.end {
            return Err(Error::Backend(format!(
                "{key}: bytes {}..{} asked of an object of {len} bytes",
                range.start, range.end
            )));
        }
        f.seek(SeekFrom::Start(range.start))?;
        let mut out = vec![0u8; (range.end - range.start) as usize];
        f.read_exact(&mut out)?;
        Ok(out)
    }

    fn put_new(&self, key: &str, bytes: &[u8]) -> Result<()> {
        let path = self.path(key)?;
        if path.exists() {
            return Err(Error::Exists(key.to_string()));
        }
        self.ensure_dir(&path)?;
        // Written beside its name, then linked into it: the link fails if
        // the name was taken meanwhile, and a reader never sees a part.
        let tmp = self.temp_beside(&path);
        let written = (|| {
            let mut f = fs::File::create(&tmp)?;
            f.write_all(bytes)?;
            f.sync_all()
        })();
        if let Err(e) = written {
            let _ = fs::remove_file(&tmp);
            return Err(e.into());
        }
        self.link_into_place(key, &tmp, &path)
    }

    fn list(&self, prefix: &str) -> Result<Vec<Entry>> {
        let mut out = Vec::new();
        self.list_dir(&self.root, "", prefix, &mut out)?;
        out.sort_by(|a, b| a.key.cmp(&b.key));
        Ok(out)
    }

    fn delete(&self, key: &str) -> Result<()> {
        let path = self.path(key)?;
        match fs::remove_file(&path) {
            Ok(()) => {}
            Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(()),
            Err(e) => return Err(e.into()),
        }
        // A directory left empty goes too, up to the root, so the store
        // holds nothing but its objects.
        let mut dir = path.parent();
        while let Some(d) = dir {
            if d == self.root || fs::remove_dir(d).is_err() {
                break;
            }
            dir = d.parent();
        }
        Ok(())
    }

    fn read_ref(&self, key: &str) -> Result<Option<(Vec<u8>, Version)>> {
        let path = self.path(key)?;
        match fs::read(&path) {
            Ok(b) => {
                let v = content_version(&b);
                Ok(Some((b, v)))
            }
            Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(None),
            Err(e) => Err(e.into()),
        }
    }

    fn write_ref(&self, key: &str, bytes: &[u8], expected: Option<&Version>) -> Result<Version> {
        let path = self.path(key)?;
        if let Some(dir) = path.parent() {
            fs::create_dir_all(dir)?;
        }
        let lock = self.lock(key, &path)?;
        let result = (|| {
            let current = self.read_ref(key)?.map(|(_, v)| v);
            check_expected(key, current.as_ref(), expected)?;
            crate::durable::write_atomic(&path, bytes)?;
            Ok(content_version(bytes))
        })();
        let _ = fs::remove_file(&lock);
        result
    }
}

// ---------------------------------------------------------------- memory

/// A store in memory, for tests: the conditional writes behave as the
/// cloud stores' do.
#[derive(Default)]
pub struct MemoryStore {
    objects: Mutex<BTreeMap<String, (Vec<u8>, u64)>>,
}

fn now_ms() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_millis() as u64)
        .unwrap_or(0)
}

impl MemoryStore {
    pub fn new() -> MemoryStore {
        MemoryStore::default()
    }

    fn map(&self) -> std::sync::MutexGuard<'_, BTreeMap<String, (Vec<u8>, u64)>> {
        self.objects.lock().unwrap_or_else(|p| p.into_inner())
    }

    /// Pretend `key` was written `age` ago (for grace-period tests).
    pub fn age(&self, key: &str, age: Duration) {
        if let Some(entry) = self.map().get_mut(key) {
            entry.1 = now_ms().saturating_sub(age.as_millis() as u64);
        }
    }
}

impl Store for MemoryStore {
    fn describe(&self) -> String {
        "memory".into()
    }

    fn has(&self, key: &str) -> Result<bool> {
        check_key(key)?;
        Ok(self.map().contains_key(key))
    }

    fn get(&self, key: &str) -> Result<Vec<u8>> {
        check_key(key)?;
        self.map()
            .get(key)
            .map(|(b, _)| b.clone())
            .ok_or_else(|| Error::NotFound(key.to_string()))
    }

    fn get_range(&self, key: &str, range: Range<u64>) -> Result<Vec<u8>> {
        let all = self.get(key)?;
        let len = all.len() as u64;
        if range.end > len || range.start > range.end {
            return Err(Error::Backend(format!(
                "{key}: bytes {}..{} asked of an object of {len} bytes",
                range.start, range.end
            )));
        }
        Ok(all[range.start as usize..range.end as usize].to_vec())
    }

    fn put_new(&self, key: &str, bytes: &[u8]) -> Result<()> {
        check_key(key)?;
        let mut map = self.map();
        if map.contains_key(key) {
            return Err(Error::Exists(key.to_string()));
        }
        map.insert(key.to_string(), (bytes.to_vec(), now_ms()));
        Ok(())
    }

    fn list(&self, prefix: &str) -> Result<Vec<Entry>> {
        Ok(self
            .map()
            .range(prefix.to_string()..)
            .take_while(|(k, _)| k.starts_with(prefix))
            .map(|(k, (v, t))| Entry {
                key: k.clone(),
                size: v.len() as u64,
                modified_ms: Some(*t),
            })
            .collect())
    }

    fn delete(&self, key: &str) -> Result<()> {
        check_key(key)?;
        self.map().remove(key);
        Ok(())
    }

    fn read_ref(&self, key: &str) -> Result<Option<(Vec<u8>, Version)>> {
        check_key(key)?;
        Ok(self
            .map()
            .get(key)
            .map(|(b, _)| (b.clone(), content_version(b))))
    }

    fn write_ref(&self, key: &str, bytes: &[u8], expected: Option<&Version>) -> Result<Version> {
        check_key(key)?;
        let mut map = self.map();
        let current = map.get(key).map(|(b, _)| content_version(b));
        check_expected(key, current.as_ref(), expected)?;
        map.insert(key.to_string(), (bytes.to_vec(), now_ms()));
        Ok(content_version(bytes))
    }
}

// ---------------------------------------------------------------- caching

/// A store in front of another (oxygen#68): objects are read through a
/// local directory and written through it, to the remote first, so the
/// remote always holds everything and the cache can be dropped at any
/// time. Objects never change, so a cached one is always right (and every
/// read is hashed again above this layer). The cache is bounded: past its
/// limit the objects used least recently are removed. An object larger
/// than an eighth of the limit is not cached; it is read by range from
/// the remote. References are never cached: they are read from and
/// written to the remote every time, so compare-and-swap is the remote's.
pub struct CachingStore {
    cache: LocalStore,
    remote: std::sync::Arc<dyn Store>,
    limit: u64,
    /// Bytes in the cache, counted on first need and kept up to date.
    used: Mutex<Option<u64>>,
}

/// Where a caching store keeps the ranges it read of objects too large to
/// keep whole, and those objects' sizes: apart from every object key.
const RANGES: &str = "cached-range";
const SIZES: &str = "cached-size";

/// How stale a cached object's use time may get before a read renews it.
const TOUCH_AFTER: Duration = Duration::from_secs(60);

impl CachingStore {
    /// `remote` behind a cache in the directory `dir` of at most `limit`
    /// bytes.
    pub fn open(
        remote: std::sync::Arc<dyn Store>,
        dir: impl Into<PathBuf>,
        limit: u64,
    ) -> Result<CachingStore> {
        Ok(CachingStore {
            cache: LocalStore::open(dir)?,
            remote,
            limit,
            used: Mutex::new(None),
        })
    }

    pub fn remote(&self) -> &std::sync::Arc<dyn Store> {
        &self.remote
    }

    /// Bytes the cache holds now.
    pub fn cached_bytes(&self) -> Result<u64> {
        let mut used = self.used.lock().unwrap_or_else(|p| p.into_inner());
        if used.is_none() {
            *used = Some(self.cache.list("")?.iter().map(|e| e.size).sum());
        }
        Ok(used.unwrap_or(0))
    }

    /// Whether an object of `size` bytes is kept in the cache at all.
    fn cacheable(&self, size: u64) -> bool {
        size <= self.limit / 8
    }

    /// Keep a copy of an object the remote holds.
    fn keep(&self, key: &str, bytes: &[u8]) -> Result<()> {
        if !self.cacheable(bytes.len() as u64) {
            return Ok(());
        }
        match self.cache.put_new(key, bytes) {
            Ok(()) => self.added(bytes.len() as u64),
            Err(Error::Exists(_)) => Ok(()),
            Err(e) => Err(e),
        }
    }

    /// Count `n` new bytes, and remove the least recently used objects
    /// once the cache is past its limit, down to nine tenths of it.
    fn added(&self, n: u64) -> Result<()> {
        let total = {
            let mut used = self.used.lock().unwrap_or_else(|p| p.into_inner());
            let total = match *used {
                Some(u) => u + n,
                // Counted now, with the new object already in place.
                None => self.cache.list("")?.iter().map(|e| e.size).sum(),
            };
            *used = Some(total);
            total
        };
        if total <= self.limit {
            return Ok(());
        }
        let mut entries = self.cache.list("")?;
        entries.sort_by_key(|e| e.modified_ms.unwrap_or(0));
        let mut left: u64 = entries.iter().map(|e| e.size).sum();
        let target = self.limit / 10 * 9;
        for e in entries {
            if left <= target {
                break;
            }
            self.cache.delete(&e.key)?;
            left -= e.size;
        }
        *self.used.lock().unwrap_or_else(|p| p.into_inner()) = Some(left);
        Ok(())
    }

    /// An object's size: kept on disk once the remote has said it (objects
    /// never change), so it is asked once. `None` when there is no object.
    fn size_of(&self, key: &str) -> Result<Option<u64>> {
        let skey = format!("{SIZES}/{key}");
        if let Some(n) = self
            .cache
            .get(&skey)
            .ok()
            .and_then(|b| String::from_utf8(b).ok())
            .and_then(|t| t.parse().ok())
        {
            return Ok(Some(n));
        }
        let size = self.remote.stat(key)?.map(|e| e.size);
        if let Some(n) = size {
            self.keep(&skey, n.to_string().as_bytes())?;
        }
        Ok(size)
    }

    /// Mark a cached object used now, at most once a minute.
    fn touch(&self, key: &str) {
        let Ok(path) = self.cache.path(key) else {
            return;
        };
        let stale = fs::metadata(&path)
            .and_then(|m| m.modified())
            .ok()
            .and_then(|t| t.elapsed().ok())
            .is_some_and(|age| age > TOUCH_AFTER);
        if stale {
            if let Ok(f) = fs::File::options().append(true).open(&path) {
                let _ = f.set_modified(std::time::SystemTime::now());
            }
        }
    }
}

impl Store for CachingStore {
    fn describe(&self) -> String {
        format!(
            "{} (cached in {})",
            self.remote.describe(),
            self.cache.root().display()
        )
    }

    fn has(&self, key: &str) -> Result<bool> {
        Ok(self.cache.has(key)? || self.remote.has(key)?)
    }

    fn get(&self, key: &str) -> Result<Vec<u8>> {
        match self.cache.get(key) {
            Ok(bytes) => {
                self.touch(key);
                return Ok(bytes);
            }
            // Not cached, or removed by another process's eviction meanwhile.
            Err(Error::NotFound(_)) => {}
            Err(e) => return Err(e),
        }
        let bytes = self.remote.get(key)?;
        self.keep(key, &bytes)?;
        Ok(bytes)
    }

    fn get_range(&self, key: &str, range: Range<u64>) -> Result<Vec<u8>> {
        match self.cache.get_range(key, range.clone()) {
            Ok(bytes) => {
                self.touch(key);
                return Ok(bytes);
            }
            Err(Error::NotFound(_)) => {}
            Err(e) if self.cache.has(key)? => return Err(e),
            Err(_) => {}
        }
        // A large object's ranges are kept as they are read (packs never
        // change, oxygenhub#234): a server opens the store again for every
        // request, so only what is on disk carries over.
        let rkey = format!("{RANGES}/{key}/{}-{}", range.start, range.end);
        if let Ok(bytes) = self.cache.get(&rkey) {
            self.touch(&rkey);
            return Ok(bytes);
        }
        // A small object is fetched whole and kept; a large one is read by
        // range and the range kept. Its size is kept too, so the remote is
        // asked it once.
        match self.size_of(key)? {
            Some(size) if self.cacheable(size) => {
                let bytes = self.get(key)?;
                let len = bytes.len() as u64;
                if range.end > len || range.start > range.end {
                    return Err(Error::Backend(format!(
                        "{key}: bytes {}..{} asked of an object of {len} bytes",
                        range.start, range.end
                    )));
                }
                Ok(bytes[range.start as usize..range.end as usize].to_vec())
            }
            Some(_) => {
                let bytes = self.remote.get_range(key, range)?;
                self.keep(&rkey, &bytes)?;
                Ok(bytes)
            }
            None => Err(Error::NotFound(key.to_string())),
        }
    }

    fn put_new(&self, key: &str, bytes: &[u8]) -> Result<()> {
        self.remote.put_new(key, bytes)?;
        self.keep(key, bytes)
    }

    fn put_many_new(&self, objects: &[(String, Vec<u8>)]) -> Result<()> {
        self.remote.put_many_new(objects)?;
        for (key, bytes) in objects {
            self.keep(key, bytes)?;
        }
        Ok(())
    }

    fn put_new_file(&self, key: &str, path: &Path) -> Result<()> {
        let size = fs::metadata(path)?.len();
        let copy = if self.cacheable(size) {
            Some(fs::read(path)?)
        } else {
            None
        };
        self.remote.put_new_file(key, path)?;
        match copy {
            Some(bytes) => self.keep(key, &bytes),
            None => Ok(()),
        }
    }

    fn list(&self, prefix: &str) -> Result<Vec<Entry>> {
        self.remote.list(prefix)
    }

    fn stat(&self, key: &str) -> Result<Option<Entry>> {
        self.remote.stat(key)
    }

    fn delete(&self, key: &str) -> Result<()> {
        self.remote.delete(key)?;
        // Its kept ranges and size go with it.
        let mut extra = self.cache.list(&format!("{RANGES}/{key}/"))?;
        extra.extend(self.cache.stat(&format!("{SIZES}/{key}"))?);
        for e in extra {
            self.cache.delete(&e.key)?;
            let mut used = self.used.lock().unwrap_or_else(|p| p.into_inner());
            if let Some(u) = used.as_mut() {
                *u = u.saturating_sub(e.size);
            }
        }
        let size = self.cache.stat(key)?.map(|e| e.size);
        self.cache.delete(key)?;
        if let Some(n) = size {
            let mut used = self.used.lock().unwrap_or_else(|p| p.into_inner());
            if let Some(u) = used.as_mut() {
                *u = u.saturating_sub(n);
            }
        }
        Ok(())
    }

    fn read_ref(&self, key: &str) -> Result<Option<(Vec<u8>, Version)>> {
        self.remote.read_ref(key)
    }

    fn write_ref(&self, key: &str, bytes: &[u8], expected: Option<&Version>) -> Result<Version> {
        self.remote.write_ref(key, bytes, expected)
    }
}

// ---------------------------------------------------------------- conformance

/// Exercise a store under `prefix` (which must be free, and is left
/// empty) and report what passed: object round trips, ranges, create-only
/// refusal, listing, deletion, and compare-and-swap on a reference. Every
/// store runs this in its tests, and `oxy remote check` runs it against a
/// remote before trusting it with a repository.
pub fn conformance(store: &dyn Store, prefix: &str) -> Result<Vec<&'static str>> {
    let mut passed = Vec::new();
    let key = |name: &str| format!("{prefix}/{name}");
    let fail = |what: &str| Error::Backend(format!("{}: {what}", store.describe()));
    let expect = |ok: bool, what: &str| if ok { Ok(()) } else { Err(fail(what)) };

    expect(
        store.list(prefix)?.is_empty(),
        "the probe prefix is not empty",
    )?;
    expect(
        !store.has(&key("a"))?,
        "has() of a missing object says true",
    )?;
    expect(
        matches!(store.get(&key("a")), Err(Error::NotFound(_))),
        "get() of a missing object is not NotFound",
    )?;
    passed.push("missing objects are reported missing");

    let a: Vec<u8> = (0..256u32 * 1024).map(|i| (i % 251) as u8).collect();
    store.put_new(&key("a"), &a)?;
    expect(store.has(&key("a"))?, "has() after put_new says false")?;
    expect(
        store.get(&key("a"))? == a,
        "get() returns other bytes than put_new wrote",
    )?;
    passed.push("an object round-trips");

    expect(
        store.get_range(&key("a"), 1000..1500)? == a[1000..1500],
        "get_range() returns the wrong bytes",
    )?;
    expect(
        store.get_range(&key("a"), 0..0)?.is_empty(),
        "an empty range is not empty",
    )?;
    expect(
        store.get_range(&key("a"), 0..a.len() as u64 + 1).is_err(),
        "a range past the end is not refused",
    )?;
    passed.push("ranges read correctly and are bounded");

    expect(
        matches!(store.put_new(&key("a"), b"other"), Err(Error::Exists(_))),
        "put_new() over an existing object is not refused",
    )?;
    expect(
        store.get(&key("a"))? == a,
        "a refused put_new() changed the object",
    )?;
    passed.push("objects are written once");

    store.put_new(&key("dir/b"), b"bb")?;
    let listed: Vec<(String, u64)> = store
        .list(prefix)?
        .into_iter()
        .map(|e| (e.key, e.size))
        .collect();
    expect(
        listed == vec![(key("a"), a.len() as u64), (key("dir/b"), 2)],
        "list() does not return the two objects, sorted, with their sizes",
    )?;
    expect(
        store
            .stat(&key("dir/b"))?
            .is_some_and(|e| e.key == key("dir/b") && e.size == 2)
            && store.stat(&key("dir/c"))?.is_none(),
        "stat() does not describe one object, or describes a missing one",
    )?;
    expect(
        store
            .list(&key("dir/"))?
            .into_iter()
            .map(|e| (e.key, e.size))
            .collect::<Vec<_>>()
            == vec![(key("dir/b"), 2)],
        "list() of a sub-prefix is wrong",
    )?;
    expect(
        store.list(&key("nothing/"))?.is_empty(),
        "list() of an unused prefix is not empty",
    )?;
    passed.push("listing by prefix");

    store.delete(&key("dir/b"))?;
    expect(
        !store.has(&key("dir/b"))?,
        "an object is still there after delete()",
    )?;
    store.delete(&key("dir/b"))?;
    passed.push("deletion, and deleting again, succeed");

    let r = key("refs");
    expect(
        store.read_ref(&r)?.is_none(),
        "an unwritten reference reads as something",
    )?;
    let v1 = store.write_ref(&r, b"one", None)?;
    expect(
        matches!(store.write_ref(&r, b"again", None), Err(Error::Conflict(_))),
        "a second create-only write of the reference is not refused",
    )?;
    let (bytes, seen) = store
        .read_ref(&r)?
        .ok_or_else(|| fail("the reference vanished"))?;
    expect(
        bytes == b"one" && seen == v1,
        "the reference reads back wrong",
    )?;
    let v2 = store.write_ref(&r, b"two", Some(&v1))?;
    expect(v2 != v1, "the version did not change on write")?;
    expect(
        matches!(
            store.write_ref(&r, b"three", Some(&v1)),
            Err(Error::Conflict(_))
        ),
        "a write with a stale version is not refused",
    )?;
    let (bytes, seen) = store
        .read_ref(&r)?
        .ok_or_else(|| fail("the reference vanished"))?;
    expect(
        bytes == b"two" && seen == v2,
        "a refused write changed the reference",
    )?;
    passed.push("references update by compare-and-swap");

    store.delete(&key("a"))?;
    store.delete(&r)?;
    expect(
        store.list(prefix)?.is_empty(),
        "the probe prefix is not empty after cleanup",
    )?;
    passed.push("cleanup");
    Ok(passed)
}

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

    #[test]
    fn keys_are_plain_relative_paths() {
        for ok in ["a", "packs/ab.pack", "a-b_c.d/e"] {
            check_key(ok).unwrap();
        }
        for bad in [
            "", "/a", "a/", "a//b", "..", "a/../b", "./a", "a b", "a\\b", "ä",
        ] {
            assert!(check_key(bad).is_err(), "{bad:?}");
        }
        assert!(check_key(&"x".repeat(MAX_KEY_LEN + 1)).is_err());
    }

    #[test]
    fn a_caching_store_conforms_and_reads_and_writes_through() {
        let dir = tempfile::tempdir().unwrap();
        let remote: Arc<dyn Store> = Arc::new(MemoryStore::new());
        let store = CachingStore::open(remote.clone(), dir.path().join("cache"), 8 << 20).unwrap();
        assert_eq!(conformance(&store, "probe").unwrap().len(), 8);
        assert_eq!(
            store.cached_bytes().unwrap(),
            0,
            "the probe left nothing cached"
        );

        // Written through: the remote has it, and so does the cache.
        store.put_new("loose/aa", b"written").unwrap();
        assert_eq!(remote.get("loose/aa").unwrap(), b"written");
        assert!(store.cache.has("loose/aa").unwrap());
        // Read through: an object only the remote has is kept on first read.
        remote.put_new("loose/bb", b"remote only").unwrap();
        assert!(!store.cache.has("loose/bb").unwrap());
        assert_eq!(store.get_range("loose/bb", 0..6).unwrap(), b"remote");
        assert!(store.cache.has("loose/bb").unwrap());
        // A cached object read after the remote lost it is still served.
        remote.delete("loose/bb").unwrap();
        assert_eq!(store.get("loose/bb").unwrap(), b"remote only");
        // References are always the remote's.
        let v = remote.write_ref("refs", b"one", None).unwrap();
        assert_eq!(
            store.read_ref("refs").unwrap(),
            Some((b"one".to_vec(), v.clone()))
        );
        remote.write_ref("refs", b"two", Some(&v)).unwrap();
        assert_eq!(store.read_ref("refs").unwrap().unwrap().0, b"two");
        assert!(matches!(
            store.write_ref("refs", b"three", Some(&v)),
            Err(Error::Conflict(_))
        ));
        assert!(!store.cache.has("refs").unwrap());
    }

    /// A remote that answers nothing: what is read through it was cached.
    struct Unreachable;

    impl Store for Unreachable {
        fn describe(&self) -> String {
            "unreachable".into()
        }
        fn has(&self, _: &str) -> Result<bool> {
            Err(Error::Backend("asked the remote".into()))
        }
        fn get(&self, _: &str) -> Result<Vec<u8>> {
            Err(Error::Backend("asked the remote".into()))
        }
        fn get_range(&self, _: &str, _: Range<u64>) -> Result<Vec<u8>> {
            Err(Error::Backend("asked the remote".into()))
        }
        fn put_new(&self, _: &str, _: &[u8]) -> Result<()> {
            Err(Error::Backend("asked the remote".into()))
        }
        fn list(&self, _: &str) -> Result<Vec<Entry>> {
            Err(Error::Backend("asked the remote".into()))
        }
        fn delete(&self, _: &str) -> Result<()> {
            Err(Error::Backend("asked the remote".into()))
        }
        fn stat(&self, _: &str) -> Result<Option<Entry>> {
            Err(Error::Backend("asked the remote".into()))
        }
        fn read_ref(&self, _: &str) -> Result<Option<(Vec<u8>, Version)>> {
            Err(Error::Backend("asked the remote".into()))
        }
        fn write_ref(&self, _: &str, _: &[u8], _: Option<&Version>) -> Result<Version> {
            Err(Error::Backend("asked the remote".into()))
        }
    }

    #[test]
    fn a_caching_store_stays_within_its_limit() {
        let dir = tempfile::tempdir().unwrap();
        let remote: Arc<dyn Store> = Arc::new(MemoryStore::new());
        let store = CachingStore::open(remote.clone(), dir.path().join("cache"), 8000).unwrap();
        for i in 0..20 {
            store
                .put_new(&format!("loose/{i:02}"), &[i as u8; 900])
                .unwrap();
            assert!(store.cached_bytes().unwrap() <= 8000);
        }
        assert_eq!(
            remote.list("loose/").unwrap().len(),
            20,
            "the remote holds everything"
        );
        let cached = store.cache.list("loose/").unwrap();
        assert!(cached.len() < 20 && cached.len() >= 7, "{}", cached.len());
        assert!(
            cached.iter().any(|e| e.key == "loose/19"),
            "the newest is kept"
        );
        for i in 0..20 {
            assert_eq!(
                store.get(&format!("loose/{i:02}")).unwrap(),
                vec![i as u8; 900]
            );
        }
        // An object over an eighth of the limit is read by range, not kept.
        remote.put_new("pack-big", &[7u8; 2000]).unwrap();
        assert_eq!(store.get_range("pack-big", 10..20).unwrap(), vec![7u8; 10]);
        assert_eq!(store.get("pack-big").unwrap().len(), 2000);
        assert!(!store.cache.has("pack-big").unwrap());
        // Its ranges are kept as read, and its size, so a store opened
        // again (a server's next request) asks the remote nothing.
        let again =
            CachingStore::open(Arc::new(Unreachable), dir.path().join("cache"), 8000).unwrap();
        assert_eq!(again.get_range("pack-big", 10..20).unwrap(), vec![7u8; 10]);
        // Counting from scratch agrees with what was kept track of.
        let fresh = CachingStore::open(remote, dir.path().join("cache"), 8000).unwrap();
        assert_eq!(fresh.cached_bytes().unwrap(), store.cached_bytes().unwrap());
        // Deleting it takes its ranges and size too.
        store.delete("pack-big").unwrap();
        assert!(
            store.cache.list("cached-range/").unwrap().is_empty()
                && store.cache.list("cached-size/").unwrap().is_empty()
        );
    }

    #[test]
    fn the_memory_store_conforms() {
        let store = MemoryStore::new();
        let passed = conformance(&store, "probe").unwrap();
        assert_eq!(passed.len(), 8, "{passed:?}");
    }

    #[test]
    fn the_directory_store_conforms() {
        let dir = tempfile::tempdir().unwrap();
        let store = LocalStore::open(dir.path().join("bucket")).unwrap();
        let passed = conformance(&store, "repos/one").unwrap();
        assert_eq!(passed.len(), 8, "{passed:?}");
        // Nothing but the (empty) directories and no stray temporaries.
        let stray: Vec<_> = walkdir(dir.path())
            .into_iter()
            .filter(|p| p.is_file())
            .collect();
        assert!(stray.is_empty(), "{stray:?}");
        assert_eq!(
            fs::read_dir(dir.path().join("bucket")).unwrap().count(),
            0,
            "no empty directories left"
        );
    }

    fn walkdir(dir: &Path) -> Vec<PathBuf> {
        let mut out = vec![];
        if let Ok(rd) = fs::read_dir(dir) {
            for e in rd.flatten() {
                let p = e.path();
                if p.is_dir() {
                    out.extend(walkdir(&p));
                } else {
                    out.push(p);
                }
            }
        }
        out
    }

    #[test]
    fn a_directory_store_refuses_keys_that_leave_its_root() {
        let dir = tempfile::tempdir().unwrap();
        let store = LocalStore::open(dir.path().join("bucket")).unwrap();
        assert!(matches!(
            store.put_new("../escape", b"x"),
            Err(Error::Key(..))
        ));
        assert!(matches!(store.get("/etc/passwd"), Err(Error::Key(..))));
        assert!(!dir.path().join("escape").exists());
    }

    /// Many threads race to move a reference from the same version: exactly
    /// one wins each round, on both stores.
    #[test]
    fn compare_and_swap_admits_one_writer_per_round() {
        let dir = tempfile::tempdir().unwrap();
        let local: Arc<dyn Store> = Arc::new(LocalStore::open(dir.path()).unwrap());
        let memory: Arc<dyn Store> = Arc::new(MemoryStore::new());
        for store in [local, memory] {
            let v0 = store.write_ref("refs", b"0", None).unwrap();
            let wins: Vec<bool> = (0..8)
                .map(|i| {
                    let store = store.clone();
                    let v0 = v0.clone();
                    std::thread::spawn(move || {
                        store
                            .write_ref("refs", format!("{}", i + 1).as_bytes(), Some(&v0))
                            .is_ok()
                    })
                })
                .collect::<Vec<_>>()
                .into_iter()
                .map(|h| h.join().unwrap())
                .collect();
            assert_eq!(
                wins.iter().filter(|w| **w).count(),
                1,
                "{}: {wins:?}",
                store.describe()
            );
            let (bytes, _) = store.read_ref("refs").unwrap().unwrap();
            assert_ne!(bytes, b"0");
        }
    }

    #[test]
    fn a_stale_lock_is_broken_and_a_live_one_waited_for() {
        let dir = tempfile::tempdir().unwrap();
        let store = LocalStore::open(dir.path()).unwrap();
        let lock = dir.path().join(".refs.lock");
        fs::write(&lock, "dead").unwrap();
        let old = std::time::SystemTime::now() - LOCK_STALE - Duration::from_secs(5);
        fs::File::open(&lock).unwrap().set_modified(old).unwrap();
        store.write_ref("refs", b"after a crash", None).unwrap();
        assert!(!lock.exists());
    }

    #[test]
    fn temporaries_and_locks_are_not_listed() {
        let dir = tempfile::tempdir().unwrap();
        let store = LocalStore::open(dir.path()).unwrap();
        store.put_new("packs/a", b"a").unwrap();
        fs::write(dir.path().join("packs/.a.tmp-ffffff"), "junk").unwrap();
        fs::write(dir.path().join(".refs.lock"), "1").unwrap();
        let keys: Vec<String> = store.list("").unwrap().into_iter().map(|e| e.key).collect();
        assert_eq!(keys, ["packs/a"]);
    }
}
`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet
Coverage / crates/oxy-core/src/strata.rs 85.1%

Keyboard
