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