//! Repositories on disk (hub spec §4): naming, the registry of open //! handles and their locks, and the page-ready views built from `oxy-core`. //! //! Every `oxy-core` call is synchronous and runs on the blocking pool under //! the repository's lock (§16.2): pushes and other writes take the write //! half, page reads the read half. pub mod archive; pub mod cache; pub mod edits; pub mod functions; pub mod names; pub mod proposals; pub mod search; pub mod view; pub use cache::{Loaded, StateCache}; pub use names::RepoName; use oxy_core::repo::Repo; use std::collections::HashMap; use std::path::{Path, PathBuf}; use std::sync::{Arc, Mutex}; use tokio::sync::{OwnedMutexGuard, OwnedRwLockReadGuard, OwnedRwLockWriteGuard, RwLock}; /// A repository's locks (oxygenhub#249). Readers never wait for a push or /// a gc: the store gives every handle one head and finds what a repack /// moved (oxy-core, "readers never wait"). Writers take `writer` in turn; /// everyone holds `life` shared, and only deleting or moving the /// repository holds it alone. #[derive(Default)] struct Locks { life: Arc>, writer: Arc>, } /// A reader's hold: the repository is not deleted or moved under it. pub type ReadGuard = OwnedRwLockReadGuard<()>; /// A writer's hold: no other writer runs, and the repository stays put. /// Readers carry on. pub struct WriteGuard { _writer: OwnedMutexGuard<()>, _life: OwnedRwLockReadGuard<()>, } /// Deleting or moving a repository: nothing else runs on it. pub type ExclusiveGuard = OwnedRwLockWriteGuard<()>; /// Where repositories live and how their access is serialised. pub struct Repos { root: PathBuf, locks: Mutex>>, cache: Arc, pub searches: search::SearchCache, } #[derive(Debug, thiserror::Error)] pub enum RepoError { #[error("no such repository")] NotFound, #[error("{0} already exists")] Exists(String), #[error("{0}")] Oxy(#[from] oxy_core::repo::RepoError), #[error("{0}")] Store(#[from] oxy_core::store::StoreError), #[error("{0}")] Io(#[from] std::io::Error), } impl Repos { pub fn new(root: PathBuf) -> Repos { // States are bounded by their estimated bytes (`cache::budget`); the // count only stops many tiny states piling up. At 64 it was always // full and evicting long before the bytes were (#145). let briefs = root.parent().unwrap_or(&root).join("cache").join("summaries").join(root.file_name().unwrap_or_default()); let cache = Arc::new(StateCache::new(256).with_briefs(briefs)); Repos { root, locks: Mutex::new(HashMap::new()), cache, searches: search::SearchCache::default() } } pub fn cache(&self) -> &Arc { &self.cache } pub fn root(&self) -> &Path { &self.root } /// `//`, holding `.oxy/` and nothing else. pub fn dir(&self, name: &RepoName) -> PathBuf { self.root.join(&name.owner).join(&name.name) } pub fn exists(&self, name: &RepoName) -> bool { self.dir(name).join(oxy_core::repo::OXY_DIR).is_dir() } /// Open a hosted repository; it is always bare. pub fn open(&self, name: &RepoName) -> Result { if !self.exists(name) { return Err(RepoError::NotFound); } Ok(Repo::open(&self.dir(name))?) } /// Create an empty bare repository. With no `id` it is **unbound**: the /// first push binds it to the pusher's repository id, so an existing /// repository keeps its identity when it moves here (hub spec §4.1). pub fn create(&self, name: &RepoName, id: Option<[u8; 16]>) -> Result { let dir = self.dir(name); if dir.exists() { return Err(RepoError::Exists(name.to_string())); } Ok(match id { Some(id) => Repo::init_bare_with_id(&dir, id)?, None => Repo::init_bare_unbound(&dir)?, }) } /// `create` off the async workers (#335): a bare init makes about /// sixteen fsyncs, and on a busy disk each waits behind every other /// write, so a create once held a worker for 26 s and stalled the /// requests queued behind it. pub async fn create_async(self: &Arc, name: &RepoName, id: Option<[u8; 16]>) -> Result<(), RepoError> { let (repos, name) = (self.clone(), name.clone()); tokio::task::spawn_blocking(move || repos.create(&name, id).map(drop)) .await .map_err(|e| RepoError::Io(std::io::Error::other(e.to_string())))? } /// Create an empty, unbound bare repository whose objects and /// operation log are kept in the store at `url` (oxygenhub#96); the /// store's credentials must already be known to the opener. pub fn create_on(&self, name: &RepoName, url: &str) -> Result { let dir = self.dir(name); if dir.exists() { return Err(RepoError::Exists(name.to_string())); } let made = (|| -> Result { drop(Repo::init_bare_on(&dir, [0; 16], Some(url))?); // Unbound as `create` makes them: oxy-core's marker file in // place of the id, so the first push gives the id. let oxy = dir.join(oxy_core::repo::OXY_DIR); std::fs::remove_file(oxy.join("id"))?; std::fs::write(oxy.join("unbound"), b"")?; Ok(Repo::open(&dir)?) })(); if made.is_err() { let _ = std::fs::remove_dir_all(&dir); } made } /// Fork `src` into `dest` (hub spec §4): a new bare repository with its /// own id, the source's objects hard-linked where the file system allows /// (copied otherwise; objects are immutable, so sharing them is safe), /// and the source's bookmarks. The trust policy is not copied: the /// fork's owner sets their own. Runs under the source's writer lock, so /// no push lands while its files are linked. pub async fn fork(&self, src: &RepoName, dest: &RepoName) -> Result<(), RepoError> { let guard = self.write_guard(src).await; let (from, to) = (self.dir(src), self.dir(dest)); if !self.exists(src) { return Err(RepoError::NotFound); } if to.exists() { return Err(RepoError::Exists(dest.to_string())); } if Repo::open(&from).is_ok_and(|r| r.config.storage_url.is_some()) { return Err(RepoError::Io(std::io::Error::other( "a repository kept in a bucket cannot be forked yet: clone it and push the clone to a new repository", ))); } tokio::task::spawn_blocking(move || { let _guard = guard; let made = (|| -> Result<(), RepoError> { drop(Repo::init_bare(&to)?); let objects = |d: &Path| d.join(oxy_core::repo::OXY_DIR).join("objects"); link_tree(&objects(&from), &objects(&to))?; let source = Repo::open(&from)?; let fork = Repo::open(&to)?; let existing: Vec = fork.bookmarks()?.into_iter().map(|b| b.name).collect(); for b in source.bookmarks()? { if existing.contains(&b.name) { fork.bookmark_set(&b.name, &b.heads)?; } else { fork.bookmark_create(&b.name, Some(&b.heads))?; } } Ok(()) })(); if made.is_err() { let _ = std::fs::remove_dir_all(&to); } made }) .await .map_err(|e| RepoError::Io(std::io::Error::other(e.to_string())))? } /// Move a repository to another name (a transfer, #108): the directory, /// and the lock the old name held. pub fn rename(&self, from: &RepoName, to: &RepoName) -> Result<(), RepoError> { let (src, dest) = (self.dir(from), self.dir(to)); if !src.exists() { return Err(RepoError::NotFound); } if dest.exists() { return Err(RepoError::Exists(to.to_string())); } if let Some(parent) = dest.parent() { std::fs::create_dir_all(parent)?; } std::fs::rename(&src, &dest)?; let mut locks = self.locks.lock().unwrap(); if let Some(l) = locks.remove(&from.to_string()) { locks.insert(to.to_string(), l); } Ok(()) } /// Move a repository to the trash (hub spec §4.1); a job empties it later. pub fn trash(&self, name: &RepoName) -> Result { let dir = self.dir(name); if !dir.exists() { return Err(RepoError::NotFound); } let trash = self.root.parent().unwrap_or(&self.root).join("trash").join(&name.owner); // A drop's name is `drops/ID`: its own directory under the owner's. let dest = trash.join(format!("{}-{}", name.name, crate::db::now_ms())); std::fs::create_dir_all(dest.parent().unwrap_or(&trash))?; std::fs::rename(&dir, &dest)?; Ok(dest) } fn lock(&self, name: &RepoName) -> Arc { self.locks.lock().unwrap().entry(name.to_string()).or_default().clone() } /// Read while `f` runs on the blocking pool: never waiting for a /// writer, only for the repository being deleted or moved. pub async fn read( &self, name: &RepoName, f: impl FnOnce(&Repo, &StateCache) -> Result + Send + 'static, ) -> Result { let guard = self.read_guard(name).await; let dir = self.dir(name); let exists = self.exists(name); let cache = self.cache.clone(); tokio::task::spawn_blocking(move || { let _guard = guard; if !exists { return Err(RepoError::NotFound); } let repo = Repo::open(&dir)?; f(&repo, &cache) }) .await .map_err(|e| RepoError::Io(std::io::Error::other(e)))? } /// Write while `f` runs on the blocking pool, after the writers ahead. pub async fn write( &self, name: &RepoName, f: impl FnOnce(&Repo, &StateCache) -> Result + Send + 'static, ) -> Result { let guard = self.write_guard(name).await; let dir = self.dir(name); let exists = self.exists(name); let cache = self.cache.clone(); tokio::task::spawn_blocking(move || { let _guard = guard; if !exists { return Err(RepoError::NotFound); } let repo = Repo::open(&dir)?; f(&repo, &cache) }) .await .map_err(|e| RepoError::Io(std::io::Error::other(e)))? } /// The writer's hold alone, for callers that open the repository /// themselves (the protocol server's push, a backup's last pass). The /// writer lock first, so a writer queued behind others holds nothing a /// delete would wait for. pub async fn write_guard(&self, name: &RepoName) -> WriteGuard { let locks = self.lock(name); let writer = locks.writer.clone().lock_owned().await; WriteGuard { _writer: writer, _life: locks.life.clone().read_owned().await } } pub async fn read_guard(&self, name: &RepoName) -> ReadGuard { self.lock(name).life.clone().read_owned().await } /// Deleting or moving the repository: waits for every reader and /// writer, and holds off new ones. pub async fn exclusive_guard(&self, name: &RepoName) -> ExclusiveGuard { self.lock(name).life.clone().write_owned().await } } /// Hard-link every file under `from` into the same place under `to`, /// copying when a link is not possible (another file system). Temporary /// files of writers in progress are skipped. fn link_tree(from: &Path, to: &Path) -> std::io::Result<()> { std::fs::create_dir_all(to)?; for e in std::fs::read_dir(from)? { let e = e?; let name = e.file_name(); let n = name.to_string_lossy(); if n.starts_with("tmp-") || n.starts_with("sigs-tmp-") || n.contains(".tmp") { continue; } let (src, dst) = (e.path(), to.join(&name)); if e.file_type()?.is_dir() { link_tree(&src, &dst)?; } else if !dst.exists() && std::fs::hard_link(&src, &dst).is_err() { std::fs::copy(&src, &dst)?; } } Ok(()) }