chadtmiller.com/knotless
git that lives in your Atmosphere account
1
10
Storage: the page as the code is; a lost pack restarts a fetch, and mirror locks are the system's
crates/knotless/src/fetch.rs+10−2
| @@ -45,14 +45,22 @@ pub async fn fetch( | ||
.filter(|p| p.value.tips.iter().any(|t| absent.contains(t))) | ||
.collect(); | ||
let records: Vec<PackRecord> = needed.iter().map(|p| p.value.clone()).collect(); | ||
futures::stream::iter(records) | ||
let got = futures::stream::iter(records) | ||
.map(|record| async move { | ||
let bytes = download(remote, &record).await?; | ||
git.index_pack(&bytes).await | ||
}) | ||
.buffer_unordered(PARALLEL) | ||
.try_collect::<Vec<()>>() | ||
.await?; | ||
.await; | ||
// A compaction can delete a listed pack before it downloads. Its | ||
// replacement is on the newest page, so one fresh walk finds it. | ||
if got.is_err() && !restarted { | ||
restarted = true; | ||
page = remote.packs(key, None).await?; | ||
continue; | ||
} | ||
got?; | ||
indexed += needed.len(); | ||
missing.extend(needed.iter().flat_map(|p| p.value.bases.iter().cloned())); | ||
let check: Vec<String> = missing.into_iter().collect(); | ||
crates/knotless/src/mirror.rs+27−33
| @@ -10,11 +10,12 @@ use anyhow::Result; | ||
use std::path::PathBuf; | ||
use std::time::Duration; | ||
| ||
/// A lock file. One left behind by a process that died is taken over once | ||
/// it is older than any sync could be. | ||
pub(crate) struct Lock(PathBuf); | ||
| ||
const STALE: Duration = Duration::from_secs(120); | ||
/// A lock file, held as the operating system's lock on it. The system | ||
/// lets go when the holder ends, crashed or not, so no lock is left behind | ||
/// to take over, and the file itself stays. | ||
pub(crate) struct Lock { | ||
_file: std::fs::File, | ||
} | ||
| ||
/// The packs a mirror holds before they are made one, as git's own | ||
/// `gc.autoPackLimit` would: mirrors turn its gc off. | ||
| @@ -22,37 +23,22 @@ const PACKS: usize = 50; | ||
| ||
impl Lock { | ||
pub(crate) async fn acquire(path: PathBuf) -> Result<Lock> { | ||
let file = std::fs::OpenOptions::new() | ||
.write(true) | ||
.create(true) | ||
.truncate(false) | ||
.open(&path)?; | ||
loop { | ||
match std::fs::OpenOptions::new() | ||
.write(true) | ||
.create_new(true) | ||
.open(&path) | ||
{ | ||
Ok(_) => return Ok(Lock(path)), | ||
Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => { | ||
let age = std::fs::metadata(&path) | ||
.and_then(|m| m.modified()) | ||
.ok() | ||
.and_then(|t| t.elapsed().ok()) | ||
.unwrap_or_default(); | ||
if age > STALE { | ||
let _ = std::fs::remove_file(&path); | ||
} else { | ||
tokio::time::sleep(Duration::from_millis(10)).await; | ||
} | ||
match file.try_lock() { | ||
Err(std::fs::TryLockError::WouldBlock) => { | ||
tokio::time::sleep(Duration::from_millis(10)).await; | ||
} | ||
Err(e) => return Err(e.into()), | ||
done => return Ok(done.map(|()| Lock { _file: file })?), | ||
} | ||
} | ||
} | ||
} | ||
| ||
impl Drop for Lock { | ||
fn drop(&mut self) { | ||
let _ = std::fs::remove_file(&self.0); | ||
} | ||
} | ||
| ||
pub struct Mirror { | ||
pub git: Git, | ||
} | ||
| @@ -136,11 +122,19 @@ mod tests { | ||
assert!(!waiter.is_finished()); | ||
drop(held); | ||
drop(waiter.await.unwrap().unwrap()); | ||
// Abandoned long ago: taken over. | ||
let file = std::fs::File::create(&path).unwrap(); | ||
file.set_modified(std::time::SystemTime::now() - STALE * 2) | ||
// A lock only looks old: it is still held, and a waiter waits. | ||
let held = Lock::acquire(path.clone()).await.unwrap(); | ||
std::fs::File::options() | ||
.write(true) | ||
.open(&path) | ||
.unwrap() | ||
.set_modified(std::time::SystemTime::now() - Duration::from_secs(3600)) | ||
.unwrap(); | ||
drop(Lock::acquire(path.clone()).await.unwrap()); | ||
let waiter = tokio::spawn(Lock::acquire(path.clone())); | ||
tokio::time::sleep(Duration::from_millis(50)).await; | ||
assert!(!waiter.is_finished()); | ||
drop(held); | ||
drop(waiter.await.unwrap().unwrap()); | ||
assert!( | ||
Lock::acquire(dir.path().join("no/such/dir.lock")) | ||
.await | ||
crates/knotless/tests/library.rs+30−1
| @@ -327,7 +327,8 @@ fn many_refs_many_packs_many_repos() { | ||
| ||
// 105 more packs, each one commit, written straight as records. | ||
let git = Git::new(work.join(".git")); | ||
let mut parent = git.rev_parse("HEAD").await.unwrap(); | ||
let head = git.rev_parse("HEAD").await.unwrap(); | ||
let mut parent = head.clone(); | ||
let mut writes = Vec::new(); | ||
for i in 0..105 { | ||
let tree = git.text(&["rev-parse", "HEAD^{tree}"], None).await.unwrap(); | ||
| @@ -408,6 +409,34 @@ fn many_refs_many_packs_many_repos() { | ||
assert_eq!(c.removed, 106); | ||
assert_eq!(r.packs(&key, None).await.unwrap().items.len(), 1); | ||
| ||
// A walk that listed a pack the compaction then deleted: its blob | ||
// is gone, and one fresh walk finds the new pack instead. | ||
let gone = remote::Page { | ||
items: vec![remote::Stored { | ||
value: PackRecord { | ||
data: None, | ||
parts: vec![records::BlobRef { | ||
kind: "blob".into(), | ||
link: records::Link { | ||
cid: page.items[0].cid.clone(), | ||
}, | ||
mime_type: "application/x-git-packed-objects".into(), | ||
size: 1, | ||
}], | ||
size: 1, | ||
tips: vec![head.clone()], | ||
..page.items[0].value.clone() | ||
}, | ||
..page.items[0].clone() | ||
}], | ||
next: None, | ||
}; | ||
let after = Git::init_bare(&s.path("after.git")).await.unwrap(); | ||
let n = fetch::fetch(&r, &key, &after, std::slice::from_ref(&head), Some(gone)) | ||
.await | ||
.unwrap(); | ||
assert_eq!(n, 1); | ||
| ||
// More than a page of repositories in one account. | ||
let mut repos = Vec::new(); | ||
for i in 0..101 { | ||
docs/design/storage.md+37−19
| @@ -74,7 +74,8 @@ A pack record holds one git pack: every object reachable from its `tips` and | ||
not from its `bases`, the objects its history needs and does not hold. A pack | ||
is self-contained, with no thin deltas, so it can be indexed alone. It is | ||
also immutable. Exactly one of `data` (the pack inline) and `parts` (the | ||
blobs it was split over, in order) is present. | ||
blobs it was split over, in order) is present. A pack written by compaction | ||
also lists, in `supersedes`, the keys of the packs it replaced. | ||
| ||
A rename leaves a redirect under the old name, pointing at the repo record by | ||
AT-URI and key, and repoints every older redirect, so an old name is one hop | ||
| @@ -214,7 +215,8 @@ A larger pack is cut into consecutive parts and uploaded with `uploadBlob`. | ||
Parts are 4,000,000 bytes, never above the blob limit the PDS advertises. A | ||
PDS's advertised limit is not always what is in front of it, a proxy for | ||
instance. So a part refused as too large halves the part size and the pack | ||
is uploaded again, down to 64 KiB. | ||
is uploaded again, until parts are no larger than 64 KiB. From the default | ||
size, the last try uses parts of 62,500 bytes. | ||
| ||
Details: | ||
| ||
| @@ -237,8 +239,11 @@ present. | ||
A compaction can delete packs between two pages, after the walk has passed | ||
the newest page where their replacement lands. So when the pages run out | ||
with something still missing, the fetch walks once more from the newest | ||
page. If something is still missing after that, the fetch fails: "the remote | ||
has no pack holding <sha>". | ||
page. A compaction can also delete a pack the walk has listed but not yet | ||
downloaded, and its blobs with it. A download that fails restarts the walk | ||
from the newest page the same way. The two cases share the one restart. If | ||
something is still missing after it, the fetch fails: "the remote has no | ||
pack holding <sha>"; a download that fails again fails the fetch. | ||
| ||
A fetch of a small push is one round trip, because its inline pack arrives | ||
with the listing. Larger packs add parallel blob reads. | ||
| @@ -252,10 +257,18 @@ Details: | ||
## Compaction | ||
| ||
Every push adds a pack, and a clone reads them all. Compaction folds old | ||
packs into one. The owner's `git push` through the helper starts it in the | ||
background once the repository holds 32 packs, and `knotless compact` runs | ||
it on demand. It writes and deletes the owner's records, so only the owner | ||
runs it. | ||
packs into one. `knotless compact` runs it on demand, and the owner's `git | ||
push` through the helper starts it once the repository holds 32 packs | ||
(`KNOTLESS_COMPACT_THRESHOLD`). The count comes from the snapshot the push | ||
read: its newest page of packs, plus the push's own, with a further page | ||
counted as 100 more. `KNOTLESS_COMPACT` runs the compaction in the | ||
`background` (the default, a detached `knotless compact`), `inline` before | ||
the push returns, or `off`. It writes and deletes the owner's records, so | ||
only the owner runs it. | ||
| ||
No other write starts it: not a push through the gateway, nor `knotless | ||
commit`, nor a commit over MCP. A repository written only those ways gains a | ||
pack with every write until its owner runs `knotless compact`. | ||
| ||
Compaction must not pull packs out from under a fetch in progress. Only | ||
packs older than ten minutes, aged by the TID in their key, are folded. A | ||
| @@ -265,7 +278,8 @@ least two old packs. | ||
| ||
The fold syncs the local mirror, then packs every live ref target with no | ||
bases, so the new pack holds everything the live refs reach. It is placed | ||
like a push's pack, keyed with a fresh TID. The same writes delete the old | ||
like a push's pack, keyed with a fresh TID, its `supersedes` naming the | ||
packs it replaces. The same writes delete the old | ||
packs and the repository's claims older than a day, in chunks of 200 with | ||
the new pack in the first. | ||
| ||
| @@ -330,16 +344,20 @@ temporary file and renamed, and a reader never sees half a file. A missing | ||
or unreadable file reads as empty. | ||
| ||
A mirror is a bare copy of a repository at `mirrors/<did>/<key>.git` in the | ||
cache, kept in step with the records. Compaction, checkout-less commits, | ||
branches, merges, the gateway and knotless-serve all work from one, shared by | ||
every process using the cache. A sync fetches what the snapshot's live refs | ||
need, then sets the mirror's refs exactly to the snapshot's in one `git | ||
update-ref --stdin` transaction. Packs are indexed without a lock, because | ||
each lands under its own name. The ref transaction runs under a lock file | ||
created exclusively beside the mirror. A lock file older than two minutes is | ||
treated as abandoned, so a crashed process cannot block the mirror. A new | ||
mirror is created under a temporary name and renamed into place, so racing | ||
processes end up with the same one. | ||
cache, with each `:` of the DID written `_`, kept in step with the records. | ||
Compaction, checkout-less commits, branches, merges, the gateway and | ||
knotless-serve all work from one, shared by every process using the cache. A | ||
sync fetches what the snapshot's live refs need, then sets the mirror's refs | ||
exactly to the snapshot's in one `git update-ref --stdin` transaction. Packs | ||
are indexed without a lock, because each lands under its own name. The ref | ||
transaction runs under the operating system's lock on a file beside the | ||
mirror. The system lets go of it when its process ends, so a crashed process | ||
cannot block the mirror. A new mirror is created under a temporary name and | ||
renamed into place, so racing processes end up with the same one. | ||
| ||
A mirror is created with git's own gc off (`gc.auto=0`). Instead, a sync | ||
that leaves it holding more than 50 packs repacks them into one, keeping | ||
objects no ref reaches. | ||
| ||
## The git remote helper | ||
Cargo.toml+1−1
| @@ -6,7 +6,7 @@ resolver = "3" | ||
version = "0.1.0" | ||
edition = "2024" | ||
license = "MIT OR Apache-2.0" | ||
rust-version = "1.88" | ||
rust-version = "1.89" | ||
| ||
[workspace.dependencies] | ||
knotless = { path = "crates/knotless" } | ||