OXYGENHUB
◆ v0.9.23 oxygenhub / crates / hub / src / repos / mod.rs
12.4 KiB · Rust · last changed Repository creates run off the async workers (closes #335) 9 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 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::{OwnedMutexGuard, OwnedRwLockReadGuard, OwnedRwLockWriteGuard, RwLock};
25
26
/// A repository's locks (oxygenhub#249). Readers never wait for a push or
27
/// a gc: the store gives every handle one head and finds what a repack
28
/// moved (oxy-core, "readers never wait"). Writers take `writer` in turn;
29
/// everyone holds `life` shared, and only deleting or moving the
30
/// repository holds it alone.
31
#[derive(Default)]
32
struct Locks {
33
    life: Arc<RwLock<()>>,
34
    writer: Arc<tokio::sync::Mutex<()>>,
35
}
36
37
/// A reader's hold: the repository is not deleted or moved under it.
38
pub type ReadGuard = OwnedRwLockReadGuard<()>;
39
40
/// A writer's hold: no other writer runs, and the repository stays put.
41
/// Readers carry on.
42
pub struct WriteGuard {
43
    _writer: OwnedMutexGuard<()>,
44
    _life: OwnedRwLockReadGuard<()>,
45
}
46
47
/// Deleting or moving a repository: nothing else runs on it.
48
pub type ExclusiveGuard = OwnedRwLockWriteGuard<()>;
49
50
/// Where repositories live and how their access is serialised.
51
pub struct Repos {
52
    root: PathBuf,
53
    locks: Mutex<HashMap<String, Arc<Locks>>>,
54
    cache: Arc<StateCache>,
55
    pub searches: search::SearchCache,
56
}
57
58
#[derive(Debug, thiserror::Error)]
59
pub enum RepoError {
60
    #[error("no such repository")]
61
    NotFound,
62
    #[error("{0} already exists")]
63
    Exists(String),
64
    #[error("{0}")]
65
    Oxy(#[from] oxy_core::repo::RepoError),
66
    #[error("{0}")]
67
    Store(#[from] oxy_core::store::StoreError),
68
    #[error("{0}")]
69
    Io(#[from] std::io::Error),
70
}
71
72
impl Repos {
73
    pub fn new(root: PathBuf) -> Repos {
74
        // States are bounded by their estimated bytes (`cache::budget`); the
75
        // count only stops many tiny states piling up. At 64 it was always
76
        // full and evicting long before the bytes were (#145).
77
        let briefs = root.parent().unwrap_or(&root).join("cache").join("summaries").join(root.file_name().unwrap_or_default());
78
        let cache = Arc::new(StateCache::new(256).with_briefs(briefs));
79
        Repos { root, locks: Mutex::new(HashMap::new()), cache, searches: search::SearchCache::default() }
80
    }
81
82
    pub fn cache(&self) -> &Arc<StateCache> {
83
        &self.cache
84
    }
85
86
    pub fn root(&self) -> &Path {
87
        &self.root
88
    }
89
90
    /// `<root>/<owner>/<name>`, holding `.oxy/` and nothing else.
91
    pub fn dir(&self, name: &RepoName) -> PathBuf {
92
        self.root.join(&name.owner).join(&name.name)
93
    }
94
95
    pub fn exists(&self, name: &RepoName) -> bool {
96
        self.dir(name).join(oxy_core::repo::OXY_DIR).is_dir()
97
    }
98
99
    /// Open a hosted repository; it is always bare.
100
    pub fn open(&self, name: &RepoName) -> Result<Repo, RepoError> {
101
        if !self.exists(name) {
102
            return Err(RepoError::NotFound);
103
        }
104
        Ok(Repo::open(&self.dir(name))?)
105
    }
106
107
    /// Create an empty bare repository. With no `id` it is **unbound**: the
108
    /// first push binds it to the pusher's repository id, so an existing
109
    /// repository keeps its identity when it moves here (hub spec §4.1).
110
    pub fn create(&self, name: &RepoName, id: Option<[u8; 16]>) -> Result<Repo, RepoError> {
111
        let dir = self.dir(name);
112
        if dir.exists() {
113
            return Err(RepoError::Exists(name.to_string()));
114
        }
115
        Ok(match id {
116
            Some(id) => Repo::init_bare_with_id(&dir, id)?,
117
            None => Repo::init_bare_unbound(&dir)?,
118
        })
119
    }
120
121
    /// `create` off the async workers (#335): a bare init makes about
122
    /// sixteen fsyncs, and on a busy disk each waits behind every other
123
    /// write, so a create once held a worker for 26 s and stalled the
124
    /// requests queued behind it.
125
    pub async fn create_async(self: &Arc<Self>, name: &RepoName, id: Option<[u8; 16]>) -> Result<(), RepoError> {
126
        let (repos, name) = (self.clone(), name.clone());
127
        tokio::task::spawn_blocking(move || repos.create(&name, id).map(drop))
128
            .await
129
            .map_err(|e| RepoError::Io(std::io::Error::other(e.to_string())))?
130
    }
131
132
    /// Create an empty, unbound bare repository whose objects and
133
    /// operation log are kept in the store at `url` (oxygenhub#96); the
134
    /// store's credentials must already be known to the opener.
135
    pub fn create_on(&self, name: &RepoName, url: &str) -> Result<Repo, RepoError> {
136
        let dir = self.dir(name);
137
        if dir.exists() {
138
            return Err(RepoError::Exists(name.to_string()));
139
        }
140
        let made = (|| -> Result<Repo, RepoError> {
141
            drop(Repo::init_bare_on(&dir, [0; 16], Some(url))?);
142
            // Unbound as `create` makes them: oxy-core's marker file in
143
            // place of the id, so the first push gives the id.
144
            let oxy = dir.join(oxy_core::repo::OXY_DIR);
145
            std::fs::remove_file(oxy.join("id"))?;
146
            std::fs::write(oxy.join("unbound"), b"")?;
147
            Ok(Repo::open(&dir)?)
148
        })();
149
        if made.is_err() {
150
            let _ = std::fs::remove_dir_all(&dir);
151
        }
152
        made
153
    }
154
155
    /// Fork `src` into `dest` (hub spec §4): a new bare repository with its
156
    /// own id, the source's objects hard-linked where the file system allows
157
    /// (copied otherwise; objects are immutable, so sharing them is safe),
158
    /// and the source's bookmarks. The trust policy is not copied: the
159
    /// fork's owner sets their own. Runs under the source's writer lock, so
160
    /// no push lands while its files are linked.
161
    pub async fn fork(&self, src: &RepoName, dest: &RepoName) -> Result<(), RepoError> {
162
        let guard = self.write_guard(src).await;
163
        let (from, to) = (self.dir(src), self.dir(dest));
164
        if !self.exists(src) {
165
            return Err(RepoError::NotFound);
166
        }
167
        if to.exists() {
168
            return Err(RepoError::Exists(dest.to_string()));
169
        }
170
        if Repo::open(&from).is_ok_and(|r| r.config.storage_url.is_some()) {
171
            return Err(RepoError::Io(std::io::Error::other(
172
                "a repository kept in a bucket cannot be forked yet: clone it and push the clone to a new repository",
173
            )));
174
        }
175
        tokio::task::spawn_blocking(move || {
176
            let _guard = guard;
177
            let made = (|| -> Result<(), RepoError> {
178
                drop(Repo::init_bare(&to)?);
179
                let objects = |d: &Path| d.join(oxy_core::repo::OXY_DIR).join("objects");
180
                link_tree(&objects(&from), &objects(&to))?;
181
                let source = Repo::open(&from)?;
182
                let fork = Repo::open(&to)?;
183
                let existing: Vec<String> = fork.bookmarks()?.into_iter().map(|b| b.name).collect();
184
                for b in source.bookmarks()? {
185
                    if existing.contains(&b.name) {
186
                        fork.bookmark_set(&b.name, &b.heads)?;
187
                    } else {
188
                        fork.bookmark_create(&b.name, Some(&b.heads))?;
189
                    }
190
                }
191
                Ok(())
192
            })();
193
            if made.is_err() {
194
                let _ = std::fs::remove_dir_all(&to);
195
            }
196
            made
197
        })
198
        .await
199
        .map_err(|e| RepoError::Io(std::io::Error::other(e.to_string())))?
200
    }
201
202
    /// Move a repository to another name (a transfer, #108): the directory,
203
    /// and the lock the old name held.
204
    pub fn rename(&self, from: &RepoName, to: &RepoName) -> Result<(), RepoError> {
205
        let (src, dest) = (self.dir(from), self.dir(to));
206
        if !src.exists() {
207
            return Err(RepoError::NotFound);
208
        }
209
        if dest.exists() {
210
            return Err(RepoError::Exists(to.to_string()));
211
        }
212
        if let Some(parent) = dest.parent() {
213
            std::fs::create_dir_all(parent)?;
214
        }
215
        std::fs::rename(&src, &dest)?;
216
        let mut locks = self.locks.lock().unwrap();
217
        if let Some(l) = locks.remove(&from.to_string()) {
218
            locks.insert(to.to_string(), l);
219
        }
220
        Ok(())
221
    }
222
223
    /// Move a repository to the trash (hub spec §4.1); a job empties it later.
224
    pub fn trash(&self, name: &RepoName) -> Result<PathBuf, RepoError> {
225
        let dir = self.dir(name);
226
        if !dir.exists() {
227
            return Err(RepoError::NotFound);
228
        }
229
        let trash = self.root.parent().unwrap_or(&self.root).join("trash").join(&name.owner);
230
        // A drop's name is `drops/ID`: its own directory under the owner's.
231
        let dest = trash.join(format!("{}-{}", name.name, crate::db::now_ms()));
232
        std::fs::create_dir_all(dest.parent().unwrap_or(&trash))?;
233
        std::fs::rename(&dir, &dest)?;
234
        Ok(dest)
235
    }
236
237
    fn lock(&self, name: &RepoName) -> Arc<Locks> {
238
        self.locks.lock().unwrap().entry(name.to_string()).or_default().clone()
239
    }
240
241
    /// Read while `f` runs on the blocking pool: never waiting for a
242
    /// writer, only for the repository being deleted or moved.
243
    pub async fn read<T: Send + 'static>(
244
        &self,
245
        name: &RepoName,
246
        f: impl FnOnce(&Repo, &StateCache) -> Result<T, RepoError> + Send + 'static,
247
    ) -> Result<T, RepoError> {
248
        let guard = self.read_guard(name).await;
249
        let dir = self.dir(name);
250
        let exists = self.exists(name);
251
        let cache = self.cache.clone();
252
        tokio::task::spawn_blocking(move || {
253
            let _guard = guard;
254
            if !exists {
255
                return Err(RepoError::NotFound);
256
            }
257
            let repo = Repo::open(&dir)?;
258
            f(&repo, &cache)
259
        })
260
        .await
261
        .map_err(|e| RepoError::Io(std::io::Error::other(e)))?
262
    }
263
264
    /// Write while `f` runs on the blocking pool, after the writers ahead.
265
    pub async fn write<T: Send + 'static>(
266
        &self,
267
        name: &RepoName,
268
        f: impl FnOnce(&Repo, &StateCache) -> Result<T, RepoError> + Send + 'static,
269
    ) -> Result<T, RepoError> {
270
        let guard = self.write_guard(name).await;
271
        let dir = self.dir(name);
272
        let exists = self.exists(name);
273
        let cache = self.cache.clone();
274
        tokio::task::spawn_blocking(move || {
275
            let _guard = guard;
276
            if !exists {
277
                return Err(RepoError::NotFound);
278
            }
279
            let repo = Repo::open(&dir)?;
280
            f(&repo, &cache)
281
        })
282
        .await
283
        .map_err(|e| RepoError::Io(std::io::Error::other(e)))?
284
    }
285
286
    /// The writer's hold alone, for callers that open the repository
287
    /// themselves (the protocol server's push, a backup's last pass). The
288
    /// writer lock first, so a writer queued behind others holds nothing a
289
    /// delete would wait for.
290
    pub async fn write_guard(&self, name: &RepoName) -> WriteGuard {
291
        let locks = self.lock(name);
292
        let writer = locks.writer.clone().lock_owned().await;
293
        WriteGuard { _writer: writer, _life: locks.life.clone().read_owned().await }
294
    }
295
296
    pub async fn read_guard(&self, name: &RepoName) -> ReadGuard {
297
        self.lock(name).life.clone().read_owned().await
298
    }
299
300
    /// Deleting or moving the repository: waits for every reader and
301
    /// writer, and holds off new ones.
302
    pub async fn exclusive_guard(&self, name: &RepoName) -> ExclusiveGuard {
303
        self.lock(name).life.clone().write_owned().await
304
    }
305
}
306
307
/// Hard-link every file under `from` into the same place under `to`,
308
/// copying when a link is not possible (another file system). Temporary
309
/// files of writers in progress are skipped.
310
fn link_tree(from: &Path, to: &Path) -> std::io::Result<()> {
311
    std::fs::create_dir_all(to)?;
312
    for e in std::fs::read_dir(from)? {
313
        let e = e?;
314
        let name = e.file_name();
315
        let n = name.to_string_lossy();
316
        if n.starts_with("tmp-") || n.starts_with("sigs-tmp-") || n.contains(".tmp") {
317
            continue;
318
        }
319
        let (src, dst) = (e.path(), to.join(&name));
320
        if e.file_type()?.is_dir() {
321
            link_tree(&src, &dst)?;
322
        } else if !dst.exists() && std::fs::hard_link(&src, &dst).is_err() {
323
            std::fs::copy(&src, &dst)?;
324
        }
325
    }
326
    Ok(())
327
}