OXYGENHUB
# abb98fa9964e oxygenhub / crates / hub / src / repos / mod.rs
4.8 KiB · Rust · last changed Milestone 0.1: browse 1 month ago source blame history raw
1 
//! Repositories on disk (hub spec §4): naming, the registry of open
2 
//! handles and their locks, and the page-ready views built from `oxy-core`.
3 
//!
4 
//! Every `oxy-core` call is synchronous and runs on the blocking pool under
5 
//! the repository's lock (§16.2): pushes and other writes take the write
6 
//! half, page reads the read half.
7
8 
pub mod names;
9 
pub mod view;
10
11 
pub use names::RepoName;
12
13 
use oxy_core::repo::Repo;
14 
use std::collections::HashMap;
15 
use std::path::{Path, PathBuf};
16 
use std::sync::{Arc, Mutex};
17 
use tokio::sync::RwLock;
18
19 
/// Where repositories live and how their access is serialised.
20 
pub struct Repos {
21 
    root: PathBuf,
22 
    locks: Mutex<HashMap<String, Arc<RwLock<()>>>>,
23 
}
24
25 
#[derive(Debug, thiserror::Error)]
26 
pub enum RepoError {
27 
    #[error("no such repository")]
28 
    NotFound,
29 
    #[error("{0} already exists")]
30 
    Exists(String),
31 
    #[error("{0}")]
32 
    Oxy(#[from] oxy_core::repo::RepoError),
33 
    #[error("{0}")]
34 
    Store(#[from] oxy_core::store::StoreError),
35 
    #[error("{0}")]
36 
    Io(#[from] std::io::Error),
37 
}
38
39 
impl Repos {
40 
    pub fn new(root: PathBuf) -> Repos {
41 
        Repos { root, locks: Mutex::new(HashMap::new()) }
42 
    }
43
44 
    pub fn root(&self) -> &Path {
45 
        &self.root
46 
    }
47
48 
    /// `<root>/<owner>/<name>`, holding `.oxy/` and nothing else.
49 
    pub fn dir(&self, name: &RepoName) -> PathBuf {
50 
        self.root.join(&name.owner).join(&name.name)
51 
    }
52
53 
    pub fn exists(&self, name: &RepoName) -> bool {
54 
        self.dir(name).join(oxy_core::repo::OXY_DIR).is_dir()
55 
    }
56
57 
    /// Open a hosted repository; it is always bare.
58 
    pub fn open(&self, name: &RepoName) -> Result<Repo, RepoError> {
59 
        if !self.exists(name) {
60 
            return Err(RepoError::NotFound);
61 
        }
62 
        Ok(Repo::open(&self.dir(name))?)
63 
    }
64
65 
    /// Create an empty bare repository. With no `id` it is **unbound**: the
66 
    /// first push binds it to the pusher's repository id, so an existing
67 
    /// repository keeps its identity when it moves here (hub spec §4.1).
68 
    pub fn create(&self, name: &RepoName, id: Option<[u8; 16]>) -> Result<Repo, RepoError> {
69 
        let dir = self.dir(name);
70 
        if dir.exists() {
71 
            return Err(RepoError::Exists(name.to_string()));
72 
        }
73 
        Ok(match id {
74 
            Some(id) => Repo::init_bare_with_id(&dir, id)?,
75 
            None => Repo::init_bare_unbound(&dir)?,
76 
        })
77 
    }
78
79 
    /// Move a repository to the trash (hub spec §4.1); a job empties it later.
80 
    pub fn trash(&self, name: &RepoName) -> Result<PathBuf, RepoError> {
81 
        let dir = self.dir(name);
82 
        if !dir.exists() {
83 
            return Err(RepoError::NotFound);
84 
        }
85 
        let trash = self.root.parent().unwrap_or(&self.root).join("trash").join(&name.owner);
86 
        std::fs::create_dir_all(&trash)?;
87 
        let dest = trash.join(format!("{}-{}", name.name, crate::db::now_ms()));
88 
        std::fs::rename(&dir, &dest)?;
89 
        Ok(dest)
90 
    }
91
92 
    fn lock(&self, name: &RepoName) -> Arc<RwLock<()>> {
93 
        self.locks.lock().unwrap().entry(name.to_string()).or_default().clone()
94 
    }
95
96 
    /// Hold the read lock while `f` runs on the blocking pool.
97 
    pub async fn read<T: Send + 'static>(
98 
        &self,
99 
        name: &RepoName,
100 
        f: impl FnOnce(&Repo) -> Result<T, RepoError> + Send + 'static,
101 
    ) -> Result<T, RepoError> {
102 
        let lock = self.lock(name);
103 
        let guard = lock.read_owned().await;
104 
        let dir = self.dir(name);
105 
        let exists = self.exists(name);
106 
        tokio::task::spawn_blocking(move || {
107 
            let _guard = guard;
108 
            if !exists {
109 
                return Err(RepoError::NotFound);
110 
            }
111 
            let repo = Repo::open(&dir)?;
112 
            f(&repo)
113 
        })
114 
        .await
115 
        .map_err(|e| RepoError::Io(std::io::Error::other(e)))?
116 
    }
117
118 
    /// Hold the write lock while `f` runs on the blocking pool.
119 
    pub async fn write<T: Send + 'static>(
120 
        &self,
121 
        name: &RepoName,
122 
        f: impl FnOnce(&Repo) -> Result<T, RepoError> + Send + 'static,
123 
    ) -> Result<T, RepoError> {
124 
        let lock = self.lock(name);
125 
        let guard = lock.write_owned().await;
126 
        let dir = self.dir(name);
127 
        let exists = self.exists(name);
128 
        tokio::task::spawn_blocking(move || {
129 
            let _guard = guard;
130 
            if !exists {
131 
                return Err(RepoError::NotFound);
132 
            }
133 
            let repo = Repo::open(&dir)?;
134 
            f(&repo)
135 
        })
136 
        .await
137 
        .map_err(|e| RepoError::Io(std::io::Error::other(e)))?
138 
    }
139
140 
    /// The write lock alone, for the protocol server, whose handler opens
141 
    /// the repository itself.
142 
    pub async fn write_guard(&self, name: &RepoName) -> tokio::sync::OwnedRwLockWriteGuard<()> {
143 
        self.lock(name).write_owned().await
144 
    }
145
146 
    pub async fn read_guard(&self, name: &RepoName) -> tokio::sync::OwnedRwLockReadGuard<()> {
147 
        self.lock(name).read_owned().await
148 
    }
149 
}