OXYGENHUB
# 2fb70d019851 oxygenhub / crates / hub / src / repos / mod.rs
9.9 KiB · Rust · last changed A repository can live in its owner's bucket 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 archive;
9 
pub mod cache;
10 
pub mod edits;
11 
pub mod functions;
12 
pub mod names;
13 
pub mod proposals;
14 
pub mod search;
15 
pub mod view;
16
17 
pub use cache::{Loaded, StateCache};
18 
pub use names::RepoName;
19
20 
use oxy_core::repo::Repo;
21 
use std::collections::HashMap;
22 
use std::path::{Path, PathBuf};
23 
use std::sync::{Arc, Mutex};
24 
use tokio::sync::RwLock;
25
26 
/// Where repositories live and how their access is serialised.
27 
pub struct Repos {
28 
    root: PathBuf,
29 
    locks: Mutex<HashMap<String, Arc<RwLock<()>>>>,
30 
    cache: Arc<StateCache>,
31 
    pub searches: search::SearchCache,
32 
}
33
34 
#[derive(Debug, thiserror::Error)]
35 
pub enum RepoError {
36 
    #[error("no such repository")]
37 
    NotFound,
38 
    #[error("{0} already exists")]
39 
    Exists(String),
40 
    #[error("{0}")]
41 
    Oxy(#[from] oxy_core::repo::RepoError),
42 
    #[error("{0}")]
43 
    Store(#[from] oxy_core::store::StoreError),
44 
    #[error("{0}")]
45 
    Io(#[from] std::io::Error),
46 
}
47
48 
impl Repos {
49 
    pub fn new(root: PathBuf) -> Repos {
50 
        Repos { root, locks: Mutex::new(HashMap::new()), cache: Arc::new(StateCache::new(64)), searches: search::SearchCache::default() }
51 
    }
52
53 
    pub fn cache(&self) -> &Arc<StateCache> {
54 
        &self.cache
55 
    }
56
57 
    pub fn root(&self) -> &Path {
58 
        &self.root
59 
    }
60
61 
    /// `<root>/<owner>/<name>`, holding `.oxy/` and nothing else.
62 
    pub fn dir(&self, name: &RepoName) -> PathBuf {
63 
        self.root.join(&name.owner).join(&name.name)
64 
    }
65
66 
    pub fn exists(&self, name: &RepoName) -> bool {
67 
        self.dir(name).join(oxy_core::repo::OXY_DIR).is_dir()
68 
    }
69
70 
    /// Open a hosted repository; it is always bare.
71 
    pub fn open(&self, name: &RepoName) -> Result<Repo, RepoError> {
72 
        if !self.exists(name) {
73 
            return Err(RepoError::NotFound);
74 
        }
75 
        Ok(Repo::open(&self.dir(name))?)
76 
    }
77
78 
    /// Create an empty bare repository. With no `id` it is **unbound**: the
79 
    /// first push binds it to the pusher's repository id, so an existing
80 
    /// repository keeps its identity when it moves here (hub spec §4.1).
81 
    pub fn create(&self, name: &RepoName, id: Option<[u8; 16]>) -> Result<Repo, RepoError> {
82 
        let dir = self.dir(name);
83 
        if dir.exists() {
84 
            return Err(RepoError::Exists(name.to_string()));
85 
        }
86 
        Ok(match id {
87 
            Some(id) => Repo::init_bare_with_id(&dir, id)?,
88 
            None => Repo::init_bare_unbound(&dir)?,
89 
        })
90 
    }
91
92 
    /// Create an empty, unbound bare repository whose objects and
93 
    /// operation log are kept in the store at `url` (oxygenhub#96); the
94 
    /// store's credentials must already be known to the opener.
95 
    pub fn create_on(&self, name: &RepoName, url: &str) -> Result<Repo, RepoError> {
96 
        let dir = self.dir(name);
97 
        if dir.exists() {
98 
            return Err(RepoError::Exists(name.to_string()));
99 
        }
100 
        let made = (|| -> Result<Repo, RepoError> {
101 
            drop(Repo::init_bare_on(&dir, [0; 16], Some(url))?);
102 
            // Unbound as `create` makes them: oxy-core's marker file in
103 
            // place of the id, so the first push gives the id.
104 
            let oxy = dir.join(oxy_core::repo::OXY_DIR);
105 
            std::fs::remove_file(oxy.join("id"))?;
106 
            std::fs::write(oxy.join("unbound"), b"")?;
107 
            Ok(Repo::open(&dir)?)
108 
        })();
109 
        if made.is_err() {
110 
            let _ = std::fs::remove_dir_all(&dir);
111 
        }
112 
        made
113 
    }
114
115 
    /// Fork `src` into `dest` (hub spec §4): a new bare repository with its
116 
    /// own id, the source's objects hard-linked where the file system allows
117 
    /// (copied otherwise; objects are immutable, so sharing them is safe),
118 
    /// and the source's bookmarks. The trust policy is not copied: the
119 
    /// fork's owner sets their own. Runs under the source's read lock.
120 
    pub async fn fork(&self, src: &RepoName, dest: &RepoName) -> Result<(), RepoError> {
121 
        let guard = self.lock(src).read_owned().await;
122 
        let (from, to) = (self.dir(src), self.dir(dest));
123 
        if !self.exists(src) {
124 
            return Err(RepoError::NotFound);
125 
        }
126 
        if to.exists() {
127 
            return Err(RepoError::Exists(dest.to_string()));
128 
        }
129 
        if Repo::open(&from).is_ok_and(|r| r.config.storage_url.is_some()) {
130 
            return Err(RepoError::Io(std::io::Error::other(
131 
                "a repository kept in a bucket cannot be forked yet: clone it and push the clone to a new repository",
132 
            )));
133 
        }
134 
        tokio::task::spawn_blocking(move || {
135 
            let _guard = guard;
136 
            let made = (|| -> Result<(), RepoError> {
137 
                drop(Repo::init_bare(&to)?);
138 
                let objects = |d: &Path| d.join(oxy_core::repo::OXY_DIR).join("objects");
139 
                link_tree(&objects(&from), &objects(&to))?;
140 
                let source = Repo::open(&from)?;
141 
                let fork = Repo::open(&to)?;
142 
                let existing: Vec<String> = fork.bookmarks()?.into_iter().map(|b| b.name).collect();
143 
                for b in source.bookmarks()? {
144 
                    if existing.contains(&b.name) {
145 
                        fork.bookmark_set(&b.name, &b.heads)?;
146 
                    } else {
147 
                        fork.bookmark_create(&b.name, Some(&b.heads))?;
148 
                    }
149 
                }
150 
                Ok(())
151 
            })();
152 
            if made.is_err() {
153 
                let _ = std::fs::remove_dir_all(&to);
154 
            }
155 
            made
156 
        })
157 
        .await
158 
        .map_err(|e| RepoError::Io(std::io::Error::other(e.to_string())))?
159 
    }
160
161 
    /// Move a repository to another name (a transfer, #108): the directory,
162 
    /// and the lock the old name held.
163 
    pub fn rename(&self, from: &RepoName, to: &RepoName) -> Result<(), RepoError> {
164 
        let (src, dest) = (self.dir(from), self.dir(to));
165 
        if !src.exists() {
166 
            return Err(RepoError::NotFound);
167 
        }
168 
        if dest.exists() {
169 
            return Err(RepoError::Exists(to.to_string()));
170 
        }
171 
        if let Some(parent) = dest.parent() {
172 
            std::fs::create_dir_all(parent)?;
173 
        }
174 
        std::fs::rename(&src, &dest)?;
175 
        let mut locks = self.locks.lock().unwrap();
176 
        if let Some(l) = locks.remove(&from.to_string()) {
177 
            locks.insert(to.to_string(), l);
178 
        }
179 
        Ok(())
180 
    }
181
182 
    /// Move a repository to the trash (hub spec §4.1); a job empties it later.
183 
    pub fn trash(&self, name: &RepoName) -> Result<PathBuf, RepoError> {
184 
        let dir = self.dir(name);
185 
        if !dir.exists() {
186 
            return Err(RepoError::NotFound);
187 
        }
188 
        let trash = self.root.parent().unwrap_or(&self.root).join("trash").join(&name.owner);
189 
        std::fs::create_dir_all(&trash)?;
190 
        let dest = trash.join(format!("{}-{}", name.name, crate::db::now_ms()));
191 
        std::fs::rename(&dir, &dest)?;
192 
        Ok(dest)
193 
    }
194
195 
    fn lock(&self, name: &RepoName) -> Arc<RwLock<()>> {
196 
        self.locks.lock().unwrap().entry(name.to_string()).or_default().clone()
197 
    }
198
199 
    /// Hold the read lock while `f` runs on the blocking pool.
200 
    pub async fn read<T: Send + 'static>(
201 
        &self,
202 
        name: &RepoName,
203 
        f: impl FnOnce(&Repo, &StateCache) -> Result<T, RepoError> + Send + 'static,
204 
    ) -> Result<T, RepoError> {
205 
        let lock = self.lock(name);
206 
        let guard = lock.read_owned().await;
207 
        let dir = self.dir(name);
208 
        let exists = self.exists(name);
209 
        let cache = self.cache.clone();
210 
        tokio::task::spawn_blocking(move || {
211 
            let _guard = guard;
212 
            if !exists {
213 
                return Err(RepoError::NotFound);
214 
            }
215 
            let repo = Repo::open(&dir)?;
216 
            f(&repo, &cache)
217 
        })
218 
        .await
219 
        .map_err(|e| RepoError::Io(std::io::Error::other(e)))?
220 
    }
221
222 
    /// Hold the write lock while `f` runs on the blocking pool.
223 
    pub async fn write<T: Send + 'static>(
224 
        &self,
225 
        name: &RepoName,
226 
        f: impl FnOnce(&Repo, &StateCache) -> Result<T, RepoError> + Send + 'static,
227 
    ) -> Result<T, RepoError> {
228 
        let lock = self.lock(name);
229 
        let guard = lock.write_owned().await;
230 
        let dir = self.dir(name);
231 
        let exists = self.exists(name);
232 
        let cache = self.cache.clone();
233 
        tokio::task::spawn_blocking(move || {
234 
            let _guard = guard;
235 
            if !exists {
236 
                return Err(RepoError::NotFound);
237 
            }
238 
            let repo = Repo::open(&dir)?;
239 
            f(&repo, &cache)
240 
        })
241 
        .await
242 
        .map_err(|e| RepoError::Io(std::io::Error::other(e)))?
243 
    }
244
245 
    /// The write lock alone, for the protocol server, whose handler opens
246 
    /// the repository itself.
247 
    pub async fn write_guard(&self, name: &RepoName) -> tokio::sync::OwnedRwLockWriteGuard<()> {
248 
        self.lock(name).write_owned().await
249 
    }
250
251 
    pub async fn read_guard(&self, name: &RepoName) -> tokio::sync::OwnedRwLockReadGuard<()> {
252 
        self.lock(name).read_owned().await
253 
    }
254 
}
255
256 
/// Hard-link every file under `from` into the same place under `to`,
257 
/// copying when a link is not possible (another file system). Temporary
258 
/// files of writers in progress are skipped.
259 
fn link_tree(from: &Path, to: &Path) -> std::io::Result<()> {
260 
    std::fs::create_dir_all(to)?;
261 
    for e in std::fs::read_dir(from)? {
262 
        let e = e?;
263 
        let name = e.file_name();
264 
        let n = name.to_string_lossy();
265 
        if n.starts_with("tmp-") || n.starts_with("sigs-tmp-") || n.contains(".tmp") {
266 
            continue;
267 
        }
268 
        let (src, dst) = (e.path(), to.join(&name));
269 
        if e.file_type()?.is_dir() {
270 
            link_tree(&src, &dst)?;
271 
        } else if !dst.exists() && std::fs::hard_link(&src, &dst).is_err() {
272 
            std::fs::copy(&src, &dst)?;
273 
        }
274 
    }
275 
    Ok(())
276 
}