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