chadtmiller.com/knotless
git that lives in your Atmosphere account
1
10
Refs name their packs, so pushes and fetches never list them
.changeset/refs-name-packs.mdnew+7−0
| @@ -0,0 +1,7 @@ | ||
--- | ||
default: patch | ||
--- | ||
| ||
# Pushes and fetches no longer wait on a listing of the packs | ||
| ||
Each ref now names the pack holding its commit, so `git fetch` reads just those packs, and `git push`, `git ls-remote` and a fetch with nothing new no longer list the packs at all. On a repository with many recent small pushes that listing ran to megabytes, and about half of each call's time. The owner's `git push` now starts `knotless compact --due` after every push, which counts the packs itself and compacts only at `KNOTLESS_COMPACT_THRESHOLD`. | ||
crates/knotless/examples/bench.rs+6−1
| @@ -180,7 +180,12 @@ fn main() { | ||
knotless::remote::Remote::connect(&ctx, &knotless::url::RepoUrl::parse(&url).unwrap()) | ||
.await | ||
.unwrap(); | ||
r.snapshot(&ctx).await.unwrap().packs.items.len() | ||
let snap = r.snapshot(&ctx).await.unwrap(); | ||
r.packs(snap.key().unwrap(), None) | ||
.await | ||
.unwrap() | ||
.items | ||
.len() | ||
}); | ||
println!("git clone, {packs:3} packs {}", Stats(clones)); | ||
let (ok, _, err) = s.knotless(&["compact", &url], b""); | ||
crates/knotless/src/cli.rs+6−1
| @@ -2419,7 +2419,12 @@ async fn command( | ||
delete_data(ctx, handle, yes, stdin, out).await?; | ||
} | ||
"compact" => { | ||
let c = compact::compact_repo(ctx, &args.url()?).await?; | ||
let url = args.url()?; | ||
let c = match args.rest.next().map(String::as_str) { | ||
Some("--due") => compact::compact_due_repo(ctx, &url).await?, | ||
Some(other) => bail!("unexpected {other}"), | ||
None => compact::compact_repo(ctx, &url).await?, | ||
}; | ||
let (packs, folded, claims, refs) = (c.before, c.removed, c.claims, c.refs); | ||
let line = format!( | ||
"{packs} packs, {folded} folded, {claims} old claims dropped, \ | ||
crates/knotless/src/compact.rs+31−2
| @@ -20,7 +20,7 @@ use crate::mirror::Mirror; | ||
use crate::push; | ||
use crate::records::{self, nsid}; | ||
use crate::remote::{self, Remote, Snapshot}; | ||
use crate::session::Session; | ||
use crate::session::{self, Session}; | ||
use crate::url::RepoUrl; | ||
use anyhow::{Context, Result}; | ||
use serde::Serialize; | ||
| @@ -60,6 +60,34 @@ pub async fn compact_repo(ctx: &Ctx, url: &RepoUrl) -> Result<Compacted> { | ||
compact(ctx, &remote, &auth).await | ||
} | ||
| ||
/// [`compact`], once the repository holds [`compact_threshold`] packs, as | ||
/// the owner's push asks for; under that, only the count, as `before`. | ||
/// | ||
/// [`compact_threshold`]: crate::env::Env::compact_threshold | ||
pub async fn compact_due(ctx: &Ctx, remote: &Remote, session: &Session) -> Result<Compacted> { | ||
let snap = remote.snapshot(ctx).await?; | ||
let Some(key) = snap.key() else { | ||
return Ok(Compacted::default()); | ||
}; | ||
let page = remote.packs(key, None).await?; | ||
let before = page.items.len() + usize::from(page.next.is_some()) * remote::PAGE; | ||
if before < ctx.env.compact_threshold { | ||
return Ok(Compacted { | ||
before, | ||
..Compacted::default() | ||
}); | ||
} | ||
compact(ctx, remote, session).await | ||
} | ||
| ||
/// [`compact_due`] of the repository at `url`, as its owner logged in here: | ||
/// what the owner's push starts aside. | ||
pub async fn compact_due_repo(ctx: &Ctx, url: &RepoUrl) -> Result<Compacted> { | ||
let remote = Remote::open(ctx, url).await?; | ||
let auth = session::ensure(ctx, &remote.identity).await?; | ||
compact_due(ctx, &remote, &auth).await | ||
} | ||
| ||
pub async fn compact(ctx: &Ctx, remote: &Remote, session: &Session) -> Result<Compacted> { | ||
let snap = remote.snapshot(ctx).await?; | ||
let Some(key) = snap.key().map(str::to_string) else { | ||
| @@ -172,7 +200,7 @@ async fn fold( | ||
// had stops the fold rather than being left out: the packs holding it | ||
// would go, and the pack naming it be left naming what no pack holds. | ||
// It folds once that pack is old enough to go too. | ||
let fetched = fetch::fetch(remote, key, &mirror.git, &kept, None).await; | ||
let fetched = fetch::fetch(remote, key, &mirror.git, &kept, snap).await; | ||
fetched.context("compacting would lose a base a younger pack names")?; | ||
tips.extend(kept); | ||
let mut writes = Vec::new(); | ||
| @@ -221,6 +249,7 @@ mod tests { | ||
updated_at: moved.into(), | ||
created_at: None, | ||
pusher: None, | ||
pack: None, | ||
}, | ||
}; | ||
let snap = Snapshot { | ||
crates/knotless/src/fetch.rs+98−19
| @@ -1,36 +1,57 @@ | ||
//! A fetch: walk the pack records newest first, download the ones whose tips | ||
//! are missing, and index them, until every wanted object is here. | ||
//! A fetch: read the packs the wanted refs name, and index them; walk the | ||
//! pack records newest first for whatever is still missing, downloading the | ||
//! ones whose tips are missing, until every wanted object is here. | ||
| ||
use crate::git::Git; | ||
use crate::records::PackRecord; | ||
use crate::remote::{Page, Remote, Stored}; | ||
use crate::records::{PackRecord, RefRecord}; | ||
use crate::remote::{Remote, Snapshot, Stored}; | ||
use anyhow::{Result, bail}; | ||
use futures::{StreamExt, TryStreamExt}; | ||
use std::collections::BTreeSet; | ||
| ||
/// Downloads parallelism. Inline packs arrive with the listing and cost | ||
/// Downloads parallelism. Inline packs arrive with their record and cost | ||
/// nothing here. | ||
const PARALLEL: usize = 8; | ||
| ||
/// Makes every object `wants` reaches present in `git`. `first` is the | ||
/// newest page of packs when the caller already has it. Returns the number | ||
/// Makes every object `wants` reaches present in `git`, from the | ||
/// repository `snap` read. The newest packs it read, if any, come first, at | ||
/// no cost; then the packs its refs name as holding their targets, read by | ||
/// key; then a walk of the packs for what those leave missing, as a pointer | ||
/// a compaction left stale, or a base no ref points at. Returns the number | ||
/// of packs indexed. | ||
pub async fn fetch( | ||
remote: &Remote, | ||
key: &str, | ||
git: &Git, | ||
wants: &[String], | ||
first: Option<Page<PackRecord>>, | ||
snap: &Snapshot, | ||
) -> Result<usize> { | ||
let mut missing: BTreeSet<String> = git.missing(wants).await?.into_iter().collect(); | ||
if missing.is_empty() { | ||
return Ok(0); | ||
} | ||
let mut page = match first { | ||
Some(page) => page, | ||
None => remote.packs(key, None).await?, | ||
// The packs indexed so far, by key. | ||
let mut seen = BTreeSet::new(); | ||
// The packs the refs name are read once: before the walk lists any, | ||
// unless a page of the newest is in hand, and then after it. | ||
let mut named = false; | ||
let mut page = match &snap.recent { | ||
Some(page) => page.clone(), | ||
None => { | ||
named = true; | ||
by_name(remote, git, &snap.refs, &mut missing, &mut seen).await?; | ||
if missing.is_empty() { | ||
return Ok(seen.len()); | ||
} | ||
remote.packs(key, None).await? | ||
} | ||
}; | ||
let mut indexed = 0; | ||
// A pack's tips being here says only that some pack brought them: one | ||
// pushed on top of a commit no ref held, as a branch cut from where a | ||
// force push left main, carries that commit too, and the pack it came in | ||
// first may still hold its parents. So the walk takes a pack with a tip | ||
// missing, and once it has run out of pages with something still | ||
// missing, walks again taking every pack it has not. | ||
let mut restarted = false; | ||
loop { | ||
let tips: Vec<String> = page | ||
| @@ -42,14 +63,12 @@ pub async fn fetch( | ||
let needed: Vec<&Stored<PackRecord>> = page | ||
.items | ||
.iter() | ||
.filter(|p| p.value.tips.iter().any(|t| absent.contains(t))) | ||
.filter(|p| !seen.contains(&p.rkey)) | ||
.filter(|p| restarted || p.value.tips.iter().any(|t| absent.contains(t))) | ||
.collect(); | ||
let records: Vec<PackRecord> = needed.iter().map(|p| p.value.clone()).collect(); | ||
let got = futures::stream::iter(records) | ||
.map(|record| async move { | ||
let bytes = download(remote, &record).await?; | ||
git.index_pack(&bytes).await | ||
}) | ||
.map(|record| async move { index(remote, git, &record).await }) | ||
.buffer_unordered(PARALLEL) | ||
.try_collect::<Vec<()>>() | ||
.await; | ||
| @@ -61,12 +80,16 @@ pub async fn fetch( | ||
continue; | ||
} | ||
got?; | ||
indexed += needed.len(); | ||
seen.extend(needed.iter().map(|p| p.rkey.clone())); | ||
missing.extend(needed.iter().flat_map(|p| p.value.bases.iter().cloned())); | ||
let check: Vec<String> = missing.into_iter().collect(); | ||
missing = git.missing(&check).await?.into_iter().collect(); | ||
if !named && !missing.is_empty() { | ||
named = true; | ||
by_name(remote, git, &snap.refs, &mut missing, &mut seen).await?; | ||
} | ||
if missing.is_empty() { | ||
return Ok(indexed); | ||
return Ok(seen.len()); | ||
} | ||
page = match page.next { | ||
Some(cursor) => remote.packs(key, Some(&cursor)).await?, | ||
| @@ -85,6 +108,62 @@ pub async fn fetch( | ||
} | ||
} | ||
| ||
/// Reads and indexes the packs `refs` name for the targets in `missing`, | ||
/// all at once, taking from `missing` what they bring and adding each to | ||
/// `seen`. A pack's bases are other refs' targets as it was pushed, so each | ||
/// round may name more, until no ref missing names one not read. A pack | ||
/// gone, as one a compaction folded, or failing to download, is left to the | ||
/// walk. | ||
async fn by_name( | ||
remote: &Remote, | ||
git: &Git, | ||
refs: &[Stored<RefRecord>], | ||
missing: &mut BTreeSet<String>, | ||
seen: &mut BTreeSet<String>, | ||
) -> Result<()> { | ||
let mut tried = BTreeSet::new(); | ||
loop { | ||
let names: Vec<String> = refs | ||
.iter() | ||
.filter(|r| r.value.target.as_ref().is_some_and(|t| missing.contains(t))) | ||
.filter_map(|r| r.value.pack.clone()) | ||
.filter(|pack| tried.insert(pack.clone())) | ||
.collect(); | ||
if names.is_empty() { | ||
return Ok(()); | ||
} | ||
let got: Vec<(String, PackRecord)> = futures::stream::iter(names) | ||
.map(|rkey| async move { | ||
let pack = named(remote, git, &rkey).await.ok().flatten(); | ||
pack.map(|pack| (rkey, pack)) | ||
}) | ||
.buffer_unordered(PARALLEL) | ||
.filter_map(std::future::ready) | ||
.collect() | ||
.await; | ||
seen.extend(got.iter().map(|(rkey, _)| rkey.clone())); | ||
missing.extend(got.iter().flat_map(|(_, p)| p.bases.iter().cloned())); | ||
let check: Vec<String> = std::mem::take(missing).into_iter().collect(); | ||
*missing = git.missing(&check).await?.into_iter().collect(); | ||
} | ||
} | ||
| ||
/// The pack record `rkey`, indexed into `git`; `None` once a compaction has | ||
/// folded it. | ||
async fn named(remote: &Remote, git: &Git, rkey: &str) -> Result<Option<PackRecord>> { | ||
let Some(pack) = remote.pack(rkey).await? else { | ||
return Ok(None); | ||
}; | ||
index(remote, git, &pack.value).await?; | ||
Ok(Some(pack.value)) | ||
} | ||
| ||
/// Downloads `pack` and indexes it into `git`. | ||
async fn index(remote: &Remote, git: &Git, pack: &PackRecord) -> Result<()> { | ||
let bytes = download(remote, pack).await?; | ||
git.index_pack(&bytes).await | ||
} | ||
| ||
/// A pack's bytes, from the record itself or its blobs. | ||
pub async fn download(remote: &Remote, pack: &PackRecord) -> Result<Vec<u8>> { | ||
let bytes = match &pack.data { | ||
crates/knotless/src/forge.rs+1−8
| @@ -2642,14 +2642,7 @@ pub async fn head_at( | ||
}; | ||
let source_key = source.snap.key().expect("an opened repository has a key"); | ||
let wants = std::slice::from_ref(&sha); | ||
fetch::fetch( | ||
&source.remote, | ||
source_key, | ||
git, | ||
wants, | ||
Some(source.snap.packs.clone()), | ||
) | ||
.await?; | ||
fetch::fetch(&source.remote, source_key, git, wants, &source.snap).await?; | ||
Ok(Ok((sha, generation))) | ||
} | ||
| ||
crates/knotless/src/helper.rs+38−27
| @@ -9,7 +9,7 @@ use crate::forge; | ||
use crate::gateway; | ||
use crate::git::Git; | ||
use crate::push::{self, Report, Update}; | ||
use crate::remote::{Remote, Snapshot}; | ||
use crate::remote::{PAGE, RECENT, Remote, Snapshot}; | ||
use crate::session::{self, Session}; | ||
use crate::url::RepoUrl; | ||
use anyhow::{Result, bail}; | ||
| @@ -23,7 +23,8 @@ pub struct Listing { | ||
} | ||
| ||
pub trait Backend { | ||
fn list(&mut self) -> impl Future<Output = Result<Listing>>; | ||
/// The refs, for a push when `for_push`, as git says, else a fetch. | ||
fn list(&mut self, for_push: bool) -> impl Future<Output = Result<Listing>>; | ||
fn fetch(&mut self, wants: Vec<String>) -> impl Future<Output = Result<()>>; | ||
fn push(&mut self, updates: Vec<Update>) -> impl Future<Output = Result<Report>>; | ||
} | ||
| @@ -41,7 +42,7 @@ where | ||
} else if line.starts_with("option ") { | ||
"unsupported\n".to_string() | ||
} else if line == "list" || line == "list for-push" { | ||
let listing = backend.list().await?; | ||
let listing = backend.list(line == "list for-push").await?; | ||
let mut out = String::new(); | ||
for (sha, name) in &listing.refs { | ||
out.push_str(&format!("{sha} {name}\n")); | ||
| @@ -153,11 +154,28 @@ impl Knotless { | ||
}) | ||
} | ||
| ||
async fn snapshot(&mut self) -> Result<Snapshot> { | ||
/// The snapshot this session read, or a new one: for a fetch, when | ||
/// `to_fetch`, with the newest packs beside the refs, which a push does | ||
/// without. | ||
async fn snapshot(&mut self, to_fetch: bool) -> Result<Snapshot> { | ||
if let Some(snap) = &self.snap { | ||
return Ok(snap.clone()); | ||
} | ||
let snap = self.remote.snapshot(&self.ctx).await?; | ||
let read = match to_fetch { | ||
true => { | ||
// A clone, into a repository with no refs yet, wants every | ||
// pack, so takes a page of them. An `ls-remote` run outside | ||
// any repository fetches nothing, so takes the few newest. | ||
let cloning = self.git.refs().await.is_ok_and(|r| r.is_empty()); | ||
let newest = match cloning { | ||
true => PAGE, | ||
false => RECENT, | ||
}; | ||
self.remote.snapshot_to_fetch(&self.ctx, newest).await | ||
} | ||
false => self.remote.snapshot(&self.ctx).await, | ||
}; | ||
let snap = read?; | ||
let (remote, snap) = self.remote.clone().follow(&self.ctx, snap).await?; | ||
self.remote = remote; | ||
if let Some(repo) = &snap.repo { | ||
| @@ -176,17 +194,16 @@ impl Knotless { | ||
Ok(snap) | ||
} | ||
| ||
async fn after_push(&self, snap: &Snapshot, auth: &session::Session) -> Result<()> { | ||
let count = snap.packs.items.len() | ||
+ 1 | ||
+ usize::from(snap.packs.next.is_some()) * crate::remote::PAGE; | ||
if count < self.ctx.env.compact_threshold { | ||
return Ok(()); | ||
} | ||
/// Compacts once the repository holds [`compact_threshold`] packs. The | ||
/// count needs a read of the packs, which a push no longer makes, so the | ||
/// compaction makes it: by default aside, where git does not wait. | ||
/// | ||
/// [`compact_threshold`]: crate::env::Env::compact_threshold | ||
async fn after_push(&self, auth: &session::Session) -> Result<()> { | ||
match self.ctx.env.compact_mode { | ||
CompactMode::Off => {} | ||
CompactMode::Inline => { | ||
let done = compact::compact(&self.ctx, &self.remote, auth).await; | ||
let done = compact::compact_due(&self.ctx, &self.remote, auth).await; | ||
said(done.map(drop)); | ||
} | ||
CompactMode::Background => { | ||
| @@ -195,6 +212,7 @@ impl Knotless { | ||
let _ = std::process::Command::new(exe) | ||
.arg("compact") | ||
.arg(self.remote.url.to_string()) | ||
.arg("--due") | ||
.stdin(std::process::Stdio::null()) | ||
.stdout(std::process::Stdio::null()) | ||
.stderr(std::process::Stdio::null()) | ||
| @@ -247,9 +265,9 @@ impl Knotless { | ||
} | ||
| ||
impl Backend for Knotless { | ||
async fn list(&mut self) -> Result<Listing> { | ||
async fn list(&mut self, for_push: bool) -> Result<Listing> { | ||
self.snap = None; | ||
let snap = self.snapshot().await?; | ||
let snap = self.snapshot(!for_push).await?; | ||
Ok(Listing { | ||
refs: snap | ||
.live_refs() | ||
| @@ -265,23 +283,16 @@ impl Backend for Knotless { | ||
} | ||
| ||
async fn fetch(&mut self, wants: Vec<String>) -> Result<()> { | ||
let snap = self.snapshot().await?; | ||
let snap = self.snapshot(true).await?; | ||
let Some(key) = snap.key() else { | ||
bail!("{}: no such repository", self.remote.url); | ||
}; | ||
fetch::fetch( | ||
&self.remote, | ||
key, | ||
&self.git, | ||
&wants, | ||
Some(snap.packs.clone()), | ||
) | ||
.await?; | ||
fetch::fetch(&self.remote, key, &self.git, &wants, &snap).await?; | ||
Ok(()) | ||
} | ||
| ||
async fn push(&mut self, updates: Vec<Update>) -> Result<Report> { | ||
let snap = self.snapshot().await?; | ||
let snap = self.snapshot(false).await?; | ||
if let Some(me) = session::reader(&self.ctx, &self.remote.identity).await | ||
&& me.did != self.remote.did() | ||
{ | ||
| @@ -304,7 +315,7 @@ impl Backend for Knotless { | ||
.await?; | ||
self.snap = None; | ||
if report.iter().any(|(_, r)| r.is_ok()) { | ||
self.after_push(&snap, &auth).await?; | ||
self.after_push(&auth).await?; | ||
} | ||
self.claim_email(&auth, &report).await; | ||
self.landed(&snap, &updates, &report).await; | ||
| @@ -329,7 +340,7 @@ mod tests { | ||
} | ||
| ||
impl Backend for Fake { | ||
async fn list(&mut self) -> Result<Listing> { | ||
async fn list(&mut self, _for_push: bool) -> Result<Listing> { | ||
Ok(Listing { | ||
refs: vec![("a".repeat(40), "refs/heads/main".into())], | ||
head: Some("refs/heads/main".into()), | ||
crates/knotless/src/mirror.rs+4−5
| @@ -124,11 +124,10 @@ impl Mirror { | ||
}; | ||
let live = snap.live_refs(); | ||
let wants: Vec<String> = live.iter().map(|(_, sha)| sha.clone()).collect(); | ||
// Indexing packs needs no lock: each lands under its own name. A | ||
// snapshot read without its packs leaves the fetch to read them, and | ||
// only if an object is missing. | ||
let first = (!snap.packs.items.is_empty()).then(|| snap.packs.clone()); | ||
fetch::fetch(remote, key, &self.git, &wants, first).await?; | ||
// Indexing packs needs no lock: each lands under its own name. The | ||
// fetch reads packs only if an object is missing, and then the ones | ||
// the refs name. | ||
fetch::fetch(remote, key, &self.git, &wants, snap).await?; | ||
let branch = snap.repo.as_ref(); | ||
let branch = branch.and_then(|r| r.value.default_branch.clone()); | ||
// Set in a task of its own, which holds the lock until git is done: | ||
crates/knotless/src/push.rs+15−0
| @@ -218,6 +218,7 @@ pub async fn push_by( | ||
None => Some(now.clone()), | ||
}, | ||
pusher: pusher.clone(), | ||
pack: a.to.as_deref().and_then(|to| holding(snap, to, &pack_rkey)), | ||
}; | ||
let rkey = keys::ref_rkey(&key, &a.name); | ||
let value = records::typed(nsid::REF, &record); | ||
| @@ -290,6 +291,20 @@ pub async fn push_by( | ||
Ok(report) | ||
} | ||
| ||
/// The pack a ref moved to `to` names as holding it. A commit some ref | ||
/// already points at was a base of the push's pack, so is not in it: it is | ||
/// in whatever pack that ref names, if any. Any other is in the push's. | ||
fn holding(snap: &Snapshot, to: &str, pushed: &Option<String>) -> Option<String> { | ||
let at = snap | ||
.refs | ||
.iter() | ||
.find(|r| r.value.target.as_deref() == Some(to)); | ||
match at { | ||
Some(r) => r.value.pack.clone(), | ||
None => pushed.clone(), | ||
} | ||
} | ||
| ||
/// Why a push stopped at a rate limit, and when to push again: `reset` is | ||
/// when the limit lets writes in again, in seconds since the epoch. | ||
fn out_of_writes(reset: Option<u64>) -> String { | ||
crates/knotless/src/records.rs+6−0
| @@ -889,6 +889,12 @@ pub struct RefRecord { | ||
/// runner writing through the gateway. | ||
#[serde(default, skip_serializing_if = "Option::is_none")] | ||
pub pusher: Option<String>, | ||
/// The record key of the pack holding `target`, as the move that set it | ||
/// knew: a fetch reads it rather than walking the packs. Gone once a | ||
/// compaction folds that pack, and absent from records a client before | ||
/// it wrote; a fetch then walks. | ||
#[serde(default, skip_serializing_if = "Option::is_none")] | ||
pub pack: Option<String>, | ||
} | ||
| ||
/// The move of a ref off one generation. Its key is the lock; its body is | ||