//! 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::RwLock; /// 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 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 read lock. pub async fn fork(&self, src: &RepoName, dest: &RepoName) -> Result<(), RepoError> { let guard = self.lock(src).read_owned().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); std::fs::create_dir_all(&trash)?; let dest = trash.join(format!("{}-{}", name.name, crate::db::now_ms())); std::fs::rename(&dir, &dest)?; Ok(dest) } fn lock(&self, name: &RepoName) -> Arc> { self.locks.lock().unwrap().entry(name.to_string()).or_default().clone() } /// Hold the read lock while `f` runs on the blocking pool. pub async fn read( &self, name: &RepoName, f: impl FnOnce(&Repo, &StateCache) -> Result + Send + 'static, ) -> Result { let lock = self.lock(name); let guard = lock.read_owned().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)))? } /// Hold the write lock while `f` runs on the blocking pool. pub async fn write( &self, name: &RepoName, f: impl FnOnce(&Repo, &StateCache) -> Result + Send + 'static, ) -> Result { let lock = self.lock(name); let guard = lock.write_owned().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 write lock alone, for the protocol server, whose handler opens /// the repository itself. pub async fn write_guard(&self, name: &RepoName) -> tokio::sync::OwnedRwLockWriteGuard<()> { self.lock(name).write_owned().await } pub async fn read_guard(&self, name: &RepoName) -> tokio::sync::OwnedRwLockReadGuard<()> { self.lock(name).read_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(()) }