Readers never wait for a push or gc 7c06f5b28ade
Pages stalled for up to 43 s while a repository's gc or push held its lock, because reads shared one read-write lock with writers. With the store now safe to read alongside writers (oxygen 89ffcd8d: one head per handle, a retry on a miss, repacks retiring what they replace for an hour), each repository has two locks: - a writer lock that pushes, gc, landings, web edits, Diffusion, Flow's blobs, mirror exports, forks and a backup's last pass take in turn; - a lifecycle lock everyone holds shared, and only deleting or moving the repository (or its wiki) holds alone. Page and API reads and fetches take only the shared lifecycle lock. A test holds a writer and reads three pages and a fetch's refs within the timeout, sees a second writer and a delete wait, and nothing read while a delete runs. Two tests follow the new core: the storage page's packing is checked through the store (retired loose files stay an hour), and a test that commits through a second handle refreshes the first before pushing. Pins oxygenflow e0c81b87 (oxygen 89ffcd8d). Closes #249
- Author
- Andy McCall @andymccall <[email protected]>
- via
- agent
agent:claude@andymccall <[email protected]> · keyless, under your grant for changes,refs,seal · key 51e59d8d35c8
- When
- · 2026-10-18 14:38 UTC
- Signature
- ◈ sealed by agent:claude@andymccall
- Hash
7c06f5b28ade52fa5ab62f2e221621ce5dc1033f17c1992e9e6e23265e62c253- Depends on
- 07b1a5bbf481 145306044896 19bb31ff1f5c 380996db84e5 42fa95540468 66fd9d6b1452 6a1780558353 8933faff2043 8a48dd830178 8fb35df7d66e 93be9ce4653e 9ff27da517fe a10446c68043 d0fbea2eb46e d23464c0ac60 d7d182467382 e97500b70f9a
- Depended on by
- 668b2ad7495c 13b27cb2200c
- In bookmarks
- api-unknown-fields cancel-landed-runs diffusion-pin-bookmark draft-ready main oh-comment-body-file proposal-list-fast
- In tags
- v0.6.0 v0.6.1 v0.7.0 v0.8.0 v0.9.0 v0.9.1 v0.9.11 v0.9.15 v0.9.16 v0.9.19 v0.9.2 v0.9.20 v0.9.21 v0.9.22 v0.9.23 v0.9.24 v0.9.3 v0.9.4 v0.9.5 v0.9.6 v0.9.7 v0.9.8 v0.9.9
- Edits
- 31 atoms, 0 tree operations, 9 files
Browse the tree at this change
modified crates/hub/src/backups.rs +3 −1
| … | 1524 unchanged lines |
||
| 1525 | 1525 | k2.insert(f.path.clone(), f.clone()); |
|
| 1526 | 1526 | } |
|
| 1527 | 1527 | } |
|
| 1528 | - | let guard = store.read_guard(&name).await; |
|
| 1528 | + | // Writers wait while the last pass copies (readers carry on), |
|
| 1529 | + | // so the refs copied name only objects copied (#249). |
|
| 1530 | + | let guard = store.write_guard(&name).await; |
|
| 1529 | 1531 | let (d2, dir2) = (dest.clone(), dir.clone()); |
|
| 1530 | 1532 | let second = tokio::task::spawn_blocking(move || { |
|
| 1531 | 1533 | let _guard = guard; |
|
| … | 429 unchanged lines |
modified crates/hub/src/protocol.rs +2 −2
| … | 461 unchanged lines |
||
| 462 | 462 | } |
|
| 463 | 463 | ||
| 464 | 464 | enum Guard { |
|
| 465 | - | Read(#[allow(dead_code)] tokio::sync::OwnedRwLockReadGuard<()>), |
|
| 466 | - | Write(#[allow(dead_code)] tokio::sync::OwnedRwLockWriteGuard<()>), |
|
| 465 | + | Read(#[allow(dead_code)] crate::repos::ReadGuard), |
|
| 466 | + | Write(#[allow(dead_code)] crate::repos::WriteGuard), |
|
| 467 | 467 | } |
modified crates/hub/src/repos/mod.rs +51 −17
| … | 20 unchanged lines |
||
| 21 | 21 | use std::collections::HashMap; |
|
| 22 | 22 | use std::path::{Path, PathBuf}; |
|
| 23 | 23 | use std::sync::{Arc, Mutex}; |
|
| 24 | - | use tokio::sync::RwLock; |
|
| 24 | + | use tokio::sync::{OwnedMutexGuard, OwnedRwLockReadGuard, OwnedRwLockWriteGuard, RwLock}; |
|
| 25 | 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 | + | ||
| 26 | 50 | /// Where repositories live and how their access is serialised. |
|
| 27 | 51 | pub struct Repos { |
|
| 28 | 52 | root: PathBuf, |
|
| 29 | - | locks: Mutex<HashMap<String, Arc<RwLock<()>>>>, |
|
| 53 | + | locks: Mutex<HashMap<String, Arc<Locks>>>, |
|
| 30 | 54 | cache: Arc<StateCache>, |
|
| 31 | 55 | pub searches: search::SearchCache, |
|
| 32 | 56 | } |
|
| … | 88 unchanged lines |
||
| 121 | 145 | /// own id, the source's objects hard-linked where the file system allows |
|
| 122 | 146 | /// (copied otherwise; objects are immutable, so sharing them is safe), |
|
| 123 | 147 | /// and the source's bookmarks. The trust policy is not copied: the |
|
| 124 | - | /// fork's owner sets their own. Runs under the source's read lock. |
|
| 148 | + | /// fork's owner sets their own. Runs under the source's writer lock, so |
|
| 149 | + | /// no push lands while its files are linked. |
|
| 125 | 150 | pub async fn fork(&self, src: &RepoName, dest: &RepoName) -> Result<(), RepoError> { |
|
| 126 | - | let guard = self.lock(src).read_owned().await; |
|
| 151 | + | let guard = self.write_guard(src).await; |
|
| 127 | 152 | let (from, to) = (self.dir(src), self.dir(dest)); |
|
| 128 | 153 | if !self.exists(src) { |
|
| 129 | 154 | return Err(RepoError::NotFound); |
|
| … | 67 unchanged lines |
||
| 197 | 222 | Ok(dest) |
|
| 198 | 223 | } |
|
| 199 | 224 | ||
| 200 | - | fn lock(&self, name: &RepoName) -> Arc<RwLock<()>> { |
|
| 225 | + | fn lock(&self, name: &RepoName) -> Arc<Locks> { |
|
| 201 | 226 | self.locks.lock().unwrap().entry(name.to_string()).or_default().clone() |
|
| 202 | 227 | } |
|
| 203 | 228 | ||
| 204 | - | /// Hold the read lock while `f` runs on the blocking pool. |
|
| 229 | + | /// Read while `f` runs on the blocking pool: never waiting for a |
|
| 230 | + | /// writer, only for the repository being deleted or moved. |
|
| 205 | 231 | pub async fn read<T: Send + 'static>( |
|
| 206 | 232 | &self, |
|
| 207 | 233 | name: &RepoName, |
|
| 208 | 234 | f: impl FnOnce(&Repo, &StateCache) -> Result<T, RepoError> + Send + 'static, |
|
| 209 | 235 | ) -> Result<T, RepoError> { |
|
| 210 | - | let lock = self.lock(name); |
|
| 211 | - | let guard = lock.read_owned().await; |
|
| 236 | + | let guard = self.read_guard(name).await; |
|
| 212 | 237 | let dir = self.dir(name); |
|
| 213 | 238 | let exists = self.exists(name); |
|
| 214 | 239 | let cache = self.cache.clone(); |
|
| … | 9 unchanged lines |
||
| 224 | 249 | .map_err(|e| RepoError::Io(std::io::Error::other(e)))? |
|
| 225 | 250 | } |
|
| 226 | 251 | ||
| 227 | - | /// Hold the write lock while `f` runs on the blocking pool. |
|
| 252 | + | /// Write while `f` runs on the blocking pool, after the writers ahead. |
|
| 228 | 253 | pub async fn write<T: Send + 'static>( |
|
| 229 | 254 | &self, |
|
| 230 | 255 | name: &RepoName, |
|
| 231 | 256 | f: impl FnOnce(&Repo, &StateCache) -> Result<T, RepoError> + Send + 'static, |
|
| 232 | 257 | ) -> Result<T, RepoError> { |
|
| 233 | - | let lock = self.lock(name); |
|
| 234 | - | let guard = lock.write_owned().await; |
|
| 258 | + | let guard = self.write_guard(name).await; |
|
| 235 | 259 | let dir = self.dir(name); |
|
| 236 | 260 | let exists = self.exists(name); |
|
| 237 | 261 | let cache = self.cache.clone(); |
|
| … | 7 unchanged lines |
||
| 245 | 269 | }) |
|
| 246 | 270 | .await |
|
| 247 | 271 | .map_err(|e| RepoError::Io(std::io::Error::other(e)))? |
|
| 272 | + | } |
|
| 273 | + | ||
| 274 | + | /// The writer's hold alone, for callers that open the repository |
|
| 275 | + | /// themselves (the protocol server's push, a backup's last pass). The |
|
| 276 | + | /// writer lock first, so a writer queued behind others holds nothing a |
|
| 277 | + | /// delete would wait for. |
|
| 278 | + | pub async fn write_guard(&self, name: &RepoName) -> WriteGuard { |
|
| 279 | + | let locks = self.lock(name); |
|
| 280 | + | let writer = locks.writer.clone().lock_owned().await; |
|
| 281 | + | WriteGuard { _writer: writer, _life: locks.life.clone().read_owned().await } |
|
| 248 | 282 | } |
|
| 249 | 283 | ||
| 250 | - | /// The write lock alone, for the protocol server, whose handler opens |
|
| 251 | - | /// the repository itself. |
|
| 252 | - | pub async fn write_guard(&self, name: &RepoName) -> tokio::sync::OwnedRwLockWriteGuard<()> { |
|
| 253 | - | self.lock(name).write_owned().await |
|
| 284 | + | pub async fn read_guard(&self, name: &RepoName) -> ReadGuard { |
|
| 285 | + | self.lock(name).life.clone().read_owned().await |
|
| 254 | 286 | } |
|
| 255 | 287 | ||
| 256 | - | pub async fn read_guard(&self, name: &RepoName) -> tokio::sync::OwnedRwLockReadGuard<()> { |
|
| 257 | - | self.lock(name).read_owned().await |
|
| 288 | + | /// Deleting or moving the repository: waits for every reader and |
|
| 289 | + | /// writer, and holds off new ones. |
|
| 290 | + | pub async fn exclusive_guard(&self, name: &RepoName) -> ExclusiveGuard { |
|
| 291 | + | self.lock(name).life.clone().write_owned().await |
|
| 258 | 292 | } |
|
| 259 | 293 | } |
|
| 260 | 294 | ||
| … | 21 unchanged lines |
modified crates/hub/src/web/repo_settings.rs +3 −3
| … | 434 unchanged lines |
||
| 435 | 435 | return Ok(fail("That owner already has a repository of this name.").await?.into_response()); |
|
| 436 | 436 | } |
|
| 437 | 437 | { |
|
| 438 | - | let _guard = app.repos.write_guard(&repo).await; |
|
| 438 | + | let _guard = app.repos.exclusive_guard(&repo).await; |
|
| 439 | 439 | app.repos.rename(&repo, &dest).map_err(|e| Error::from_repo(shell.clone(), e))?; |
|
| 440 | 440 | crate::storage::renamed(&app, &repo, &dest); |
|
| 441 | 441 | if app.wikis.exists(&repo) { |
|
| … | 71 unchanged lines |
||
| 513 | 513 | if !crate::wiki::exists(&app, &repo) { |
|
| 514 | 514 | return settings_page(&app, shell, row, role, None, Some("This repository has no wiki.".into())).await; |
|
| 515 | 515 | } |
|
| 516 | - | let _guard = app.wikis.write_guard(&repo).await; |
|
| 516 | + | let _guard = app.wikis.exclusive_guard(&repo).await; |
|
| 517 | 517 | crate::wiki::trash(&app, &repo).map_err(|e| Error::from_repo(shell.clone(), e))?; |
|
| 518 | 518 | drop(_guard); |
|
| 519 | 519 | accounts::audit(&app.db, viewer.account(), "wiki.data_deleted", &repo.to_string(), "moved to trash").await.ok(); |
|
| … | 21 unchanged lines |
||
| 541 | 541 | .await? |
|
| 542 | 542 | .into_response()); |
|
| 543 | 543 | } |
|
| 544 | - | let _guard = app.repos.write_guard(&repo).await; |
|
| 544 | + | let _guard = app.repos.exclusive_guard(&repo).await; |
|
| 545 | 545 | app.repos.trash(&repo).map_err(|e| Error::from_repo(shell.clone(), e))?; |
|
| 546 | 546 | // A repository kept in a bucket leaves its data there: the owner's. |
|
| 547 | 547 | crate::storage::removed(&app, &repo); |
|
| … | 557 unchanged lines |
modified crates/hub/tests/accounts.rs +3 −6
| … | 520 unchanged lines |
||
| 521 | 521 | assert!(loose_before > 0, "the push left loose objects"); |
|
| 522 | 522 | let r = post(app, "/admin/storage", Some(&andy), &format!("_csrf={}&repo=andy%2Fdemo", csrf_of(&storage.body))).await; |
|
| 523 | 523 | assert!(r.body.contains("Maintained 1 repository") && r.body.contains("loose objects packed"), "{}", r.body); |
|
| 524 | - | let loose_after: usize = walk_files(&app.repos.dir(&repo).join(".oxy/objects/loose")); |
|
| 524 | + | // Packed: the store has no loose objects; their files stay retired for |
|
| 525 | + | // an hour, for readers that listed them before the repack (#249). |
|
| 526 | + | let loose_after = Repo::open(&app.repos.dir(&repo)).unwrap().store().loose_count().unwrap(); |
|
| 525 | 527 | assert_eq!(loose_after, 0, "packed"); |
|
| 526 | 528 | let audit = db::accounts::audit_rows(&app.db, 20).await.unwrap(); |
|
| 527 | 529 | assert!(audit.iter().any(|a| a.action == "admin.gc" && a.target == "andy/demo")); |
|
| … | 1 unchanged lines |
||
| 529 | 531 | assert!(r.body.contains("No repository nobody/none")); |
|
| 530 | 532 | } |
|
| 531 | 533 | ||
| 532 | - | fn walk_files(dir: &Path) -> usize { |
|
| 533 | - | let Ok(rd) = std::fs::read_dir(dir) else { return 0 }; |
|
| 534 | - | rd.flatten().map(|e| if e.path().is_dir() { walk_files(&e.path()) } else { 1 }).sum() |
|
| 535 | - | } |
|
| 536 | - | ||
| 537 | 534 | /// oxygenhub#81: an admin disables an account (sign-in says so, its session |
|
| 538 | 535 | /// and pushes stop) and enables it again; deletes one, refused while it |
|
| 539 | 536 | /// owns a repository, after which its name stays retired. |
|
| … | 1047 unchanged lines |
modified crates/hub/tests/edits.rs +3 −1
| … | 544 unchanged lines |
||
| 545 | 545 | write(work.path(), "b.txt", b"by the agent\n"); |
|
| 546 | 546 | as_agent.snapshot("t").unwrap(); |
|
| 547 | 547 | let by_agent = as_agent.commit("third, by the agent", &[]).unwrap(); |
|
| 548 | - | // Andy pushes both. |
|
| 548 | + | // Andy pushes both: his handle reads the head again first, to see |
|
| 549 | + | // the agent's commit from the other handle (oxygenhub#249). |
|
| 550 | + | client.refresh().unwrap(); |
|
| 549 | 551 | assert!(client.push("origin", false).unwrap().iter().all(|r| r.ok)); |
|
| 550 | 552 | tokio::time::sleep(std::time::Duration::from_millis(300)).await; |
|
| 551 | 553 | ||
| … | 312 unchanged lines |
modified crates/hub/tests/flow.rs +34 −0
| … | 5499 unchanged lines |
||
| 5500 | 5500 | } |
|
| 5501 | 5501 | } |
|
| 5502 | 5502 | ||
| 5503 | + | /// Readers never wait (oxygenhub#249): a page, the API and a fetch's refs |
|
| 5504 | + | /// answer while a writer (a push, a gc) holds the repository; writers take |
|
| 5505 | + | /// turns; deleting or moving it waits for everyone and holds off newcomers. |
|
| 5506 | + | #[tokio::test(flavor = "multi_thread")] |
|
| 5507 | + | async fn reads_do_not_wait_for_a_writer() { |
|
| 5508 | + | use std::time::Duration; |
|
| 5509 | + | let h = hub().await; |
|
| 5510 | + | create(&h.app, "andy", "demo").await; |
|
| 5511 | + | let work = tempfile::tempdir().unwrap(); |
|
| 5512 | + | let c = client(&format!("{}/andy/demo", h.base), work.path()); |
|
| 5513 | + | write(work.path(), "a.txt", "a\n"); |
|
| 5514 | + | c.commit("A change", &[]).unwrap(); |
|
| 5515 | + | assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok)); |
|
| 5516 | + | let name = RepoName::parse("andy/demo").unwrap(); |
|
| 5517 | + | let writer = h.app.repos.write_guard(&name).await; |
|
| 5518 | + | for path in ["/andy/demo", "/andy/demo/changes/main", "/api/v1/repos/andy/demo/bookmarks"] { |
|
| 5519 | + | let (st, body) = tokio::time::timeout(Duration::from_secs(10), call(&h.app, "GET", path, None, None)) |
|
| 5520 | + | .await |
|
| 5521 | + | .unwrap_or_else(|_| panic!("{path} waited for the writer")); |
|
| 5522 | + | assert_eq!(st, StatusCode::OK, "{path}: {body}"); |
|
| 5523 | + | } |
|
| 5524 | + | let refs = tokio::time::timeout(Duration::from_secs(10), call(&h.app, "POST", "/andy/demo/oxy/v1/refs", Some("site"), None)).await; |
|
| 5525 | + | assert!(refs.is_ok(), "a fetch's refs waited for the writer"); |
|
| 5526 | + | assert!(tokio::time::timeout(Duration::from_millis(300), h.app.repos.write_guard(&name)).await.is_err(), "writers take turns"); |
|
| 5527 | + | assert!(tokio::time::timeout(Duration::from_millis(300), h.app.repos.exclusive_guard(&name)).await.is_err(), "a delete waits"); |
|
| 5528 | + | drop(writer); |
|
| 5529 | + | let all = tokio::time::timeout(Duration::from_secs(5), h.app.repos.exclusive_guard(&name)).await.expect("then it runs"); |
|
| 5530 | + | assert!( |
|
| 5531 | + | tokio::time::timeout(Duration::from_millis(300), h.app.repos.read_guard(&name)).await.is_err(), |
|
| 5532 | + | "nothing reads a repository being deleted" |
|
| 5533 | + | ); |
|
| 5534 | + | drop(all); |
|
| 5535 | + | } |
|
| 5536 | + | ||
| 5503 | 5537 | /// Live L5 (docs/live.md, oxygenhub#207): the issue page's timeline |
|
| 5504 | 5538 | /// listens on the issue's topic and the repository's, the overview's latest |
|
| 5505 | 5539 | /// change and bookmarks on the repository's, and its Diffusion card and the |
|
| … | 209 unchanged lines |
replaced deps/oxygenflow +0 −0
No line changes.
modified docs/spec.md +2 −2
| … | 106 unchanged lines |
||
| 107 | 107 | ||
| 108 | 108 | 1. Middleware: request id, tracing span, body limits, session cookie decode, CSRF check on unsafe methods (§15.2), rate limits (§15.6). |
|
| 109 | 109 | 2. Wire protocol routes `/{owner}/{name}/oxy/v1/*` go to the embedded `oxy-server` handlers with the hub's repository resolver and authorizer (§18). |
|
| 110 | - | 3. Page and API handlers resolve `{owner}/{name}` to a `RepoHandle`, check the viewer's role, and run every `oxy-core` call inside `spawn_blocking` under the repository's read-write lock (§16.2). |
|
| 110 | + | 3. Page and API handlers resolve `{owner}/{name}` to a `RepoHandle`, check the viewer's role, and run every `oxy-core` call inside `spawn_blocking` under the repository's locks (§16.2: readers never wait for writers). |
|
| 111 | 111 | 4. Templates receive plain structs; no template touches `oxy-core` directly. |
|
| 112 | 112 | ||
| 113 | 113 | ### 3.4 Crate layout |
|
| … | 612 unchanged lines |
||
| 726 | 726 | ||
| 727 | 727 | ### 16.2 Locks and blocking |
|
| 728 | 728 | ||
| 729 | - | Every `oxy-core` call runs on the blocking pool. Each repository has a `tokio::sync::RwLock`: pushes, landings, web edits and mirror exports take the write lock; page reads take the read lock. `oxy-core` itself is single-threaded per `Repo`, so a `Repo` is opened per request and dropped after; store handles are cheap to open because packs are memory mapped. |
|
| 729 | + | Every `oxy-core` call runs on the blocking pool. **Readers never wait** (oxygenhub#249): each repository has a writer lock that pushes, gc, landings, web edits, Diffusion bumps, Flow's blobs, mirror exports, forks and a backup's last pass take in turn, and a lifecycle lock that everyone holds shared and only deleting or moving the repository holds alone. Page and API reads and fetches take only the shared lifecycle lock, so a push or a gc never holds them up: the store gives each handle one head for its life and finds what a repack moved (a reload and a retry on a miss; repacks retire what they replace for an hour before it is deleted; Oxygen spec §13.4). `oxy-core` itself is single-threaded per `Repo`, so a `Repo` is opened per request and dropped after; store handles are cheap to open because packs are memory mapped. |
|
| 730 | 730 | ||
| 731 | 731 | ### 16.3 Caches |
|
| 732 | 732 | ||
| … | 103 unchanged lines |