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