mod.rs 82.8%

173 of 209 lines run, 417 of 1136 functions · run #539 on main · run missed

//! Repositories on disk (hub spec §4): naming, the registry of open
//! handles and their locks, and the page-ready views built from `oxy-core`.
//!
//! Every `oxy-core` call is synchronous and runs on the blocking pool under
//! the repository's lock (§16.2): pushes and other writes take the write
//! half, page reads the read half.

pub mod archive;
pub mod cache;
pub mod edits;
pub mod functions;
pub mod names;
pub mod proposals;
pub mod search;
pub mod view;

pub use cache::{Loaded, StateCache};
pub use names::RepoName;

use oxy_core::repo::Repo;
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use tokio::sync::{OwnedMutexGuard, OwnedRwLockReadGuard, OwnedRwLockWriteGuard, RwLock};

/// A repository's locks (oxygenhub#249). Readers never wait for a push or
/// a gc: the store gives every handle one head and finds what a repack
/// moved (oxy-core, "readers never wait"). Writers take `writer` in turn;
/// everyone holds `life` shared, and only deleting or moving the
/// repository holds it alone.
#[derive(Default)]
struct Locks {
    life: Arc<RwLock<()>>,
    writer: Arc<tokio::sync::Mutex<()>>,
}

/// A reader's hold: the repository is not deleted or moved under it.
pub type ReadGuard = OwnedRwLockReadGuard<()>;

/// A writer's hold: no other writer runs, and the repository stays put.
/// Readers carry on.
pub struct WriteGuard {
    _writer: OwnedMutexGuard<()>,
    _life: OwnedRwLockReadGuard<()>,
}

/// Deleting or moving a repository: nothing else runs on it.
pub type ExclusiveGuard = OwnedRwLockWriteGuard<()>;

/// Where repositories live and how their access is serialised.
pub struct Repos {
    root: PathBuf,
    locks: Mutex<HashMap<String, Arc<Locks>>>,
    cache: Arc<StateCache>,
    pub searches: search::SearchCache,
}

#[derive(Debug, thiserror::Error)]
pub enum RepoError {
    #[error("no such repository")]
    NotFound,
    #[error("{0} already exists")]
    Exists(String),
    #[error("{0}")]
    Oxy(#[from] oxy_core::repo::RepoError),
    #[error("{0}")]
    Store(#[from] oxy_core::store::StoreError),
    #[error("{0}")]
    Io(#[from] std::io::Error),
}

impl Repos {
    pub fn new(root: PathBuf) -> Repos {
        // States are bounded by their estimated bytes (`cache::budget`); the
        // count only stops many tiny states piling up. At 64 it was always
        // full and evicting long before the bytes were (#145).
        let briefs = root.parent().unwrap_or(&root).join("cache").join("summaries").join(root.file_name().unwrap_or_default());
        let cache = Arc::new(StateCache::new(256).with_briefs(briefs));
        Repos { root, locks: Mutex::new(HashMap::new()), cache, searches: search::SearchCache::default() }
    }

    pub fn cache(&self) -> &Arc<StateCache> {
        &self.cache
    }

    pub fn root(&self) -> &Path {
        &self.root
    }

    /// `<root>/<owner>/<name>`, holding `.oxy/` and nothing else.
    pub fn dir(&self, name: &RepoName) -> PathBuf {
        self.root.join(&name.owner).join(&name.name)
    }

    pub fn exists(&self, name: &RepoName) -> bool {
        self.dir(name).join(oxy_core::repo::OXY_DIR).is_dir()
    }

    /// Open a hosted repository; it is always bare.
    pub fn open(&self, name: &RepoName) -> Result<Repo, RepoError> {
        if !self.exists(name) {
            return Err(RepoError::NotFound);
        }
        Ok(Repo::open(&self.dir(name))?)
    }

    /// Create an empty bare repository. With no `id` it is **unbound**: the
    /// first push binds it to the pusher's repository id, so an existing
    /// repository keeps its identity when it moves here (hub spec §4.1).
    pub fn create(&self, name: &RepoName, id: Option<[u8; 16]>) -> Result<Repo, RepoError> {
        let dir = self.dir(name);
        if dir.exists() {
            return Err(RepoError::Exists(name.to_string()));
        }
        Ok(match id {
            Some(id) => Repo::init_bare_with_id(&dir, id)?,
            None => Repo::init_bare_unbound(&dir)?,
        })
    }

    /// `create` off the async workers (#335): a bare init makes about
    /// sixteen fsyncs, and on a busy disk each waits behind every other
    /// write, so a create once held a worker for 26 s and stalled the
    /// requests queued behind it.
    pub async fn create_async(self: &Arc<Self>, name: &RepoName, id: Option<[u8; 16]>) -> Result<(), RepoError> {
        let (repos, name) = (self.clone(), name.clone());
        tokio::task::spawn_blocking(move || repos.create(&name, id).map(drop))
            .await
            .map_err(|e| RepoError::Io(std::io::Error::other(e.to_string())))?
    }

    /// Create an empty, unbound bare repository whose objects and
    /// operation log are kept in the store at `url` (oxygenhub#96); the
    /// store's credentials must already be known to the opener.
    pub fn create_on(&self, name: &RepoName, url: &str) -> Result<Repo, RepoError> {
        let dir = self.dir(name);
        if dir.exists() {
            return Err(RepoError::Exists(name.to_string()));
        }
        let made = (|| -> Result<Repo, RepoError> {
            drop(Repo::init_bare_on(&dir, [0; 16], Some(url))?);
            // Unbound as `create` makes them: oxy-core's marker file in
            // place of the id, so the first push gives the id.
            let oxy = dir.join(oxy_core::repo::OXY_DIR);
            std::fs::remove_file(oxy.join("id"))?;
            std::fs::write(oxy.join("unbound"), b"")?;
            Ok(Repo::open(&dir)?)
        })();
        if made.is_err() {
            let _ = std::fs::remove_dir_all(&dir);
        }
        made
    }

    /// Fork `src` into `dest` (hub spec §4): a new bare repository with its
    /// own id, the source's objects hard-linked where the file system allows
    /// (copied otherwise; objects are immutable, so sharing them is safe),
    /// and the source's bookmarks. The trust policy is not copied: the
    /// fork's owner sets their own. Runs under the source's writer lock, so
    /// no push lands while its files are linked.
    pub async fn fork(&self, src: &RepoName, dest: &RepoName) -> Result<(), RepoError> {
        let guard = self.write_guard(src).await;
        let (from, to) = (self.dir(src), self.dir(dest));
        if !self.exists(src) {
            return Err(RepoError::NotFound);
        }
        if to.exists() {
            return Err(RepoError::Exists(dest.to_string()));
        }
        if Repo::open(&from).is_ok_and(|r| r.config.storage_url.is_some()) {
            return Err(RepoError::Io(std::io::Error::other(
                "a repository kept in a bucket cannot be forked yet: clone it and push the clone to a new repository",
            )));
        }
        tokio::task::spawn_blocking(move || {
            let _guard = guard;
            let made = (|| -> Result<(), RepoError> {
                drop(Repo::init_bare(&to)?);
                let objects = |d: &Path| d.join(oxy_core::repo::OXY_DIR).join("objects");
                link_tree(&objects(&from), &objects(&to))?;
                let source = Repo::open(&from)?;
                let fork = Repo::open(&to)?;
                let existing: Vec<String> = fork.bookmarks()?.into_iter().map(|b| b.name).collect();
                for b in source.bookmarks()? {
                    if existing.contains(&b.name) {
                        fork.bookmark_set(&b.name, &b.heads)?;
                    } else {
                        fork.bookmark_create(&b.name, Some(&b.heads))?;
                    }
                }
                Ok(())
            })();
            if made.is_err() {
                let _ = std::fs::remove_dir_all(&to);
            }
            made
        })
        .await
        .map_err(|e| RepoError::Io(std::io::Error::other(e.to_string())))?
    }

    /// Move a repository to another name (a transfer, #108): the directory,
    /// and the lock the old name held.
    pub fn rename(&self, from: &RepoName, to: &RepoName) -> Result<(), RepoError> {
        let (src, dest) = (self.dir(from), self.dir(to));
        if !src.exists() {
            return Err(RepoError::NotFound);
        }
        if dest.exists() {
            return Err(RepoError::Exists(to.to_string()));
        }
        if let Some(parent) = dest.parent() {
            std::fs::create_dir_all(parent)?;
        }
        std::fs::rename(&src, &dest)?;
        let mut locks = self.locks.lock().unwrap();
        if let Some(l) = locks.remove(&from.to_string()) {
            locks.insert(to.to_string(), l);
        }
        Ok(())
    }

    /// Move a repository to the trash (hub spec §4.1); a job empties it later.
    pub fn trash(&self, name: &RepoName) -> Result<PathBuf, RepoError> {
        let dir = self.dir(name);
        if !dir.exists() {
            return Err(RepoError::NotFound);
        }
        let trash = self.root.parent().unwrap_or(&self.root).join("trash").join(&name.owner);
        // A drop's name is `drops/ID`: its own directory under the owner's.
        let dest = trash.join(format!("{}-{}", name.name, crate::db::now_ms()));
        std::fs::create_dir_all(dest.parent().unwrap_or(&trash))?;
        std::fs::rename(&dir, &dest)?;
        Ok(dest)
    }

    fn lock(&self, name: &RepoName) -> Arc<Locks> {
        self.locks.lock().unwrap().entry(name.to_string()).or_default().clone()
    }

    /// Read while `f` runs on the blocking pool: never waiting for a
    /// writer, only for the repository being deleted or moved.
    pub async fn read<T: Send + 'static>(
        &self,
        name: &RepoName,
        f: impl FnOnce(&Repo, &StateCache) -> Result<T, RepoError> + Send + 'static,
    ) -> Result<T, RepoError> {
        let guard = self.read_guard(name).await;
        let dir = self.dir(name);
        let exists = self.exists(name);
        let cache = self.cache.clone();
        tokio::task::spawn_blocking(move || {
            let _guard = guard;
            if !exists {
                return Err(RepoError::NotFound);
            }
            let repo = Repo::open(&dir)?;
            f(&repo, &cache)
        })
        .await
        .map_err(|e| RepoError::Io(std::io::Error::other(e)))?
    }

    /// Write while `f` runs on the blocking pool, after the writers ahead.
    pub async fn write<T: Send + 'static>(
        &self,
        name: &RepoName,
        f: impl FnOnce(&Repo, &StateCache) -> Result<T, RepoError> + Send + 'static,
    ) -> Result<T, RepoError> {
        let guard = self.write_guard(name).await;
        let dir = self.dir(name);
        let exists = self.exists(name);
        let cache = self.cache.clone();
        tokio::task::spawn_blocking(move || {
            let _guard = guard;
            if !exists {
                return Err(RepoError::NotFound);
            }
            let repo = Repo::open(&dir)?;
            f(&repo, &cache)
        })
        .await
        .map_err(|e| RepoError::Io(std::io::Error::other(e)))?
    }

    /// The writer's hold alone, for callers that open the repository
    /// themselves (the protocol server's push, a backup's last pass). The
    /// writer lock first, so a writer queued behind others holds nothing a
    /// delete would wait for.
    pub async fn write_guard(&self, name: &RepoName) -> WriteGuard {
        let locks = self.lock(name);
        let writer = locks.writer.clone().lock_owned().await;
        WriteGuard { _writer: writer, _life: locks.life.clone().read_owned().await }
    }

    pub async fn read_guard(&self, name: &RepoName) -> ReadGuard {
        self.lock(name).life.clone().read_owned().await
    }

    /// Deleting or moving the repository: waits for every reader and
    /// writer, and holds off new ones.
    pub async fn exclusive_guard(&self, name: &RepoName) -> ExclusiveGuard {
        self.lock(name).life.clone().write_owned().await
    }
}

/// Hard-link every file under `from` into the same place under `to`,
/// copying when a link is not possible (another file system). Temporary
/// files of writers in progress are skipped.
fn link_tree(from: &Path, to: &Path) -> std::io::Result<()> {
    std::fs::create_dir_all(to)?;
    for e in std::fs::read_dir(from)? {
        let e = e?;
        let name = e.file_name();
        let n = name.to_string_lossy();
        if n.starts_with("tmp-") || n.starts_with("sigs-tmp-") || n.contains(".tmp") {
            continue;
        }
        let (src, dst) = (e.path(), to.join(&name));
        if e.file_type()?.is_dir() {
            link_tree(&src, &dst)?;
        } else if !dst.exists() && std::fs::hard_link(&src, &dst).is_err() {
            std::fs::copy(&src, &dst)?;
        }
    }
    Ok(())
}

`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet

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	414×	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	414×	let briefs = root.parent().unwrap_or(&root).join("cache").join("summaries").join(root.file_name().unwrap_or_default());
78	414×	let cache = Arc::new(StateCache::new(256).with_briefs(briefs));
79	414×	Repos { root, locks: Mutex::new(HashMap::new()), cache, searches: search::SearchCache::default() }
80	414×	}
81
82	9×	pub fn cache(&self) -> &Arc<StateCache> {
83	9×	&self.cache
84	9×	}
85
86	299×	pub fn root(&self) -> &Path {
87	299×	&self.root
88	299×	}
89
90		/// `<root>/<owner>/<name>`, holding `.oxy/` and nothing else.
91	11313×	pub fn dir(&self, name: &RepoName) -> PathBuf {
92	11313×	self.root.join(&name.owner).join(&name.name)
93	11313×	}
94
95	5589×	pub fn exists(&self, name: &RepoName) -> bool {
96	5589×	self.dir(name).join(oxy_core::repo::OXY_DIR).is_dir()
97	5589×	}
98
99		/// Open a hosted repository; it is always bare.
100	1910×	pub fn open(&self, name: &RepoName) -> Result<Repo, RepoError> {
101	1910×	if !self.exists(name) {
102	1×	return Err(RepoError::NotFound);
103	1909×	}
104	1909×	Ok(Repo::open(&self.dir(name))?)
105	1910×	}
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	210×	pub fn create(&self, name: &RepoName, id: Option<[u8; 16]>) -> Result<Repo, RepoError> {
111	210×	let dir = self.dir(name);
112	210×	if dir.exists() {
113	0×	return Err(RepoError::Exists(name.to_string()));
114	210×	}
115	210×	Ok(match id {
116	9×	Some(id) => Repo::init_bare_with_id(&dir, id)?,
117	201×	None => Repo::init_bare_unbound(&dir)?,
118		})
119	210×	}
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	31×	pub async fn create_async(self: &Arc<Self>, name: &RepoName, id: Option<[u8; 16]>) -> Result<(), RepoError> {
126	31×	let (repos, name) = (self.clone(), name.clone());
127	31×	tokio::task::spawn_blocking(move \|\| repos.create(&name, id).map(drop))
128	31×	.await
129	31×	.map_err(\|e\| RepoError::Io(std::io::Error::other(e.to_string())))?
130	31×	}
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	0×	pub fn create_on(&self, name: &RepoName, url: &str) -> Result<Repo, RepoError> {
136	0×	let dir = self.dir(name);
137	0×	if dir.exists() {
138	0×	return Err(RepoError::Exists(name.to_string()));
139	0×	}
140	0×	let made = (\|\| -> Result<Repo, RepoError> {
141	0×	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	0×	let oxy = dir.join(oxy_core::repo::OXY_DIR);
145	0×	std::fs::remove_file(oxy.join("id"))?;
146	0×	std::fs::write(oxy.join("unbound"), b"")?;
147	0×	Ok(Repo::open(&dir)?)
148		})();
149	0×	if made.is_err() {
150	0×	let _ = std::fs::remove_dir_all(&dir);
151	0×	}
152	0×	made
153	0×	}
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	3×	pub async fn fork(&self, src: &RepoName, dest: &RepoName) -> Result<(), RepoError> {
162	3×	let guard = self.write_guard(src).await;
163	3×	let (from, to) = (self.dir(src), self.dir(dest));
164	3×	if !self.exists(src) {
165	0×	return Err(RepoError::NotFound);
166	3×	}
167	3×	if to.exists() {
168	0×	return Err(RepoError::Exists(dest.to_string()));
169	3×	}
170	3×	if Repo::open(&from).is_ok_and(\|r\| r.config.storage_url.is_some()) {
171	0×	return Err(RepoError::Io(std::io::Error::other(
172	0×	"a repository kept in a bucket cannot be forked yet: clone it and push the clone to a new repository",
173	0×	)));
174	3×	}
175	3×	tokio::task::spawn_blocking(move \|\| {
176	3×	let _guard = guard;
177	3×	let made = (\|\| -> Result<(), RepoError> {
178	3×	drop(Repo::init_bare(&to)?);
179	6×	let objects = \|d: &Path\| d.join(oxy_core::repo::OXY_DIR).join("objects");
180	3×	link_tree(&objects(&from), &objects(&to))?;
181	3×	let source = Repo::open(&from)?;
182	3×	let fork = Repo::open(&to)?;
183	3×	let existing: Vec<String> = fork.bookmarks()?.into_iter().map(\|b\| b.name).collect();
184	3×	for b in source.bookmarks()? {
185	3×	if existing.contains(&b.name) {
186	3×	fork.bookmark_set(&b.name, &b.heads)?;
187		} else {
188	0×	fork.bookmark_create(&b.name, Some(&b.heads))?;
189		}
190		}
191	3×	Ok(())
192		})();
193	3×	if made.is_err() {
194	0×	let _ = std::fs::remove_dir_all(&to);
195	3×	}
196	3×	made
197	3×	})
198	3×	.await
199	3×	.map_err(\|e\| RepoError::Io(std::io::Error::other(e.to_string())))?
200	3×	}
201
202		/// Move a repository to another name (a transfer, #108): the directory,
203		/// and the lock the old name held.
204	1×	pub fn rename(&self, from: &RepoName, to: &RepoName) -> Result<(), RepoError> {
205	1×	let (src, dest) = (self.dir(from), self.dir(to));
206	1×	if !src.exists() {
207	0×	return Err(RepoError::NotFound);
208	1×	}
209	1×	if dest.exists() {
210	0×	return Err(RepoError::Exists(to.to_string()));
211	1×	}
212	1×	if let Some(parent) = dest.parent() {
213	1×	std::fs::create_dir_all(parent)?;
214	0×	}
215	1×	std::fs::rename(&src, &dest)?;
216	1×	let mut locks = self.locks.lock().unwrap();
217	1×	if let Some(l) = locks.remove(&from.to_string()) {
218	1×	locks.insert(to.to_string(), l);
219	1×	}
220	1×	Ok(())
221	1×	}
222
223		/// Move a repository to the trash (hub spec §4.1); a job empties it later.
224	6×	pub fn trash(&self, name: &RepoName) -> Result<PathBuf, RepoError> {
225	6×	let dir = self.dir(name);
226	6×	if !dir.exists() {
227	0×	return Err(RepoError::NotFound);
228	6×	}
229	6×	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	6×	let dest = trash.join(format!("{}-{}", name.name, crate::db::now_ms()));
232	6×	std::fs::create_dir_all(dest.parent().unwrap_or(&trash))?;
233	6×	std::fs::rename(&dir, &dest)?;
234	6×	Ok(dest)
235	6×	}
236
237	4135×	fn lock(&self, name: &RepoName) -> Arc<Locks> {
238	4135×	self.locks.lock().unwrap().entry(name.to_string()).or_default().clone()
239	4135×	}
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	2872×	pub async fn read<T: Send + 'static>(
244	2872×	&self,
245	2872×	name: &RepoName,
246	2872×	f: impl FnOnce(&Repo, &StateCache) -> Result<T, RepoError> + Send + 'static,
247	2872×	) -> Result<T, RepoError> {
248	2872×	let guard = self.read_guard(name).await;
249	2872×	let dir = self.dir(name);
250	2872×	let exists = self.exists(name);
251	2872×	let cache = self.cache.clone();
252	2872×	tokio::task::spawn_blocking(move \|\| {
253	2870×	let _guard = guard;
254	2870×	if !exists {
255	1×	return Err(RepoError::NotFound);
256	2869×	}
257	2869×	let repo = Repo::open(&dir)?;
258	2869×	f(&repo, &cache)
259	2870×	})
260	2872×	.await
261	2871×	.map_err(\|e\| RepoError::Io(std::io::Error::other(e)))?
262	2871×	}
263
264		/// Write while `f` runs on the blocking pool, after the writers ahead.
265	112×	pub async fn write<T: Send + 'static>(
266	112×	&self,
267	112×	name: &RepoName,
268	112×	f: impl FnOnce(&Repo, &StateCache) -> Result<T, RepoError> + Send + 'static,
269	112×	) -> Result<T, RepoError> {
270	112×	let guard = self.write_guard(name).await;
271	112×	let dir = self.dir(name);
272	112×	let exists = self.exists(name);
273	112×	let cache = self.cache.clone();
274	112×	tokio::task::spawn_blocking(move \|\| {
275	112×	let _guard = guard;
276	112×	if !exists {
277	0×	return Err(RepoError::NotFound);
278	112×	}
279	112×	let repo = Repo::open(&dir)?;
280	112×	f(&repo, &cache)
281	112×	})
282	112×	.await
283	112×	.map_err(\|e\| RepoError::Io(std::io::Error::other(e)))?
284	112×	}
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	408×	pub async fn write_guard(&self, name: &RepoName) -> WriteGuard {
291	408×	let locks = self.lock(name);
292	408×	let writer = locks.writer.clone().lock_owned().await;
293	407×	WriteGuard { _writer: writer, _life: locks.life.clone().read_owned().await }
294	407×	}
295
296	3718×	pub async fn read_guard(&self, name: &RepoName) -> ReadGuard {
297	3718×	self.lock(name).life.clone().read_owned().await
298	3717×	}
299
300		/// Deleting or moving the repository: waits for every reader and
301		/// writer, and holds off new ones.
302	9×	pub async fn exclusive_guard(&self, name: &RepoName) -> ExclusiveGuard {
303	9×	self.lock(name).life.clone().write_owned().await
304	8×	}
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	37×	fn link_tree(from: &Path, to: &Path) -> std::io::Result<()> {
311	37×	std::fs::create_dir_all(to)?;
312	62×	for e in std::fs::read_dir(from)? {
313	62×	let e = e?;
314	62×	let name = e.file_name();
315	62×	let n = name.to_string_lossy();
316	62×	if n.starts_with("tmp-") \|\| n.starts_with("sigs-tmp-") \|\| n.contains(".tmp") {
317	0×	continue;
318	62×	}
319	62×	let (src, dst) = (e.path(), to.join(&name));
320	62×	if e.file_type()?.is_dir() {
321	34×	link_tree(&src, &dst)?;
322	28×	} else if !dst.exists() && std::fs::hard_link(&src, &dst).is_err() {
323	0×	std::fs::copy(&src, &dst)?;
324	28×	}
325		}
326	37×	Ok(())
327	37×	}

Coverage / crates/hub/src/repos/mod.rs 82.8%

Keyboard