use super::atomic::hardlink_tree;
use super::cmd::{stdin_bytes, upload_pack_into};
use super::refs::head_tip;
use super::{FETCH_BASES, exported_refs};
use gix::hash::ObjectId;
use std::collections::BTreeSet;
use std::collections::HashSet;
use std::path::Path;
use anyhow::Context as _;
use super::atomic::{write_unlinked, write_unlinked_with};
use super::cmd::{
command_into, git_command, objects_before_alternates, pack_objects_command, run_cmd,
};
use super::replies::Replies;
use super::wire::pkt_line;
use crate::snapshot::RepoSnapshot;
/// nginx picks one reply by repo and request. the bare names ack the round, the `-done` ones carry the pack.
/// a depth-1 fetch of tags or several tips gets the full pack.
pub(crate) fn write_fetch_replies(
snapshot: &RepoSnapshot,
export_dir: &Path,
replies: &Replies,
candidates: &[Vec<String>],
full_pack: &dyn Fn() -> anyhow::Result<u64>,
fetch: &dyn Fn(&[&str]) -> Vec<u8>,
write: &dyn Fn(&str, &[u8]) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
let repo = knot_git::Repo::open(export_dir).context("open export for fetch replies")?;
let shallow = ShallowTips::of(snapshot, head_tip(export_dir).as_deref())?;
let branches: Vec<String> = shallow.branches.iter().map(ObjectId::to_string).collect();
// `git clone --depth=1` is single-branch. it asks for HEAD alone.
let depth_one: Vec<(String, ObjectId, Vec<u8>)> = (shallow
.head
.iter()
.map(|tip| ("shallow-head".into(), *tip)))
.chain((shallow.branches.iter().enumerate()).map(|(n, tip)| (format!("shallow-{n}"), *tip)))
.map(|(name, tip)| {
let want = format!("want {tip}");
let request = fetch(&["ofs-delta", "no-progress", "deepen 1", &want, "done"]);
(name, tip, request)
})
.collect();
let depth_one = &depth_one;
std::thread::scope(|scope| {
let shallow = scope.spawn(move || -> anyhow::Result<()> {
let prior_head = replies.prior_smart.map(|dir| dir.join("shallow-head"));
for (name, tip, request) in depth_one {
// a branch is built from its prior reply and HEAD's, since most of its tree is HEAD's.
let mut like = Vec::new();
if name != "shallow-head" {
like = replies.prior_depth_one(snapshot, tip);
like.push(replies.smart_dir.join("shallow-head"));
}
like.extend(prior_head.clone().filter(|head| !like.contains(head)));
replies.reply(name, request, |path| {
crate::depth_one::reply(export_dir, snapshot.object_format, tip, path, &like)
})?;
}
Ok(())
});
let rest = write_have_replies(
snapshot, &repo, export_dir, replies, candidates, full_pack, fetch, write,
);
let shallow = shallow.join().expect("the depth-1 cuts panicked");
rest.and(shallow)
})?;
write(
SHALLOW_BRANCHES,
(branches.join("\n") + "\n").trim_start().as_bytes(),
)?;
write(
"err-unshallow",
&pkt_line(b"ERR fetch --unshallow isn't served here: clone again without --depth\n"),
)?;
write(
"err-v0-shallow",
&pkt_line(b"ERR shallow fetches need git protocol v2: git -c protocol.version=2 ...\n"),
)
}
/// an older base gets the full pack instead, since copying the aside to every base is too much.
const ASIDE_BASES: usize = 1;
pub fn have_replies(n: usize, moved: bool) -> bool {
!moved || n < ASIDE_BASES
}
#[allow(clippy::too_many_arguments)]
fn write_have_replies(
snapshot: &RepoSnapshot,
repo: &knot_git::Repo,
export_dir: &Path,
replies: &Replies,
candidates: &[Vec<String>],
full_pack: &dyn Fn() -> anyhow::Result<u64>,
fetch: &dyn Fn(&[&str]) -> Vec<u8>,
write: &dyn Fn(&str, &[u8]) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
let wants: Vec<String> = exported_refs(snapshot)
.map(|(_, raw)| format!("want {raw}"))
.collect();
// a cold export packs the aside since its heads. the first push links it instead.
let heads: Vec<String> = (snapshot.public_refs.iter())
.filter(|r| r.name.starts_with("refs/heads/"))
.map(|r| r.raw.clone())
.collect();
let oldest = candidates.last().unwrap_or(&heads);
let since = Some(independent_tips(repo, export_dir, oldest)).filter(|since| !since.is_empty());
let moved = match &since {
Some(since) if !candidates.is_empty() => moved_wants(snapshot, export_dir, since)?,
_ => Vec::new(),
};
let aside = match since {
Some(since) => Some(TagAside::of(snapshot, repo, export_dir, replies, since)?),
None => None,
};
let mut picked: Vec<Vec<String>> = Vec::new();
for candidate in candidates {
if picked.len() == FETCH_BASES {
break;
}
let base = independent_tips(repo, export_dir, candidate);
if !base.is_empty()
&& !picked.contains(&base)
&& crate::nginx::base_fits_row(snapshot.repo_did.as_str(), &base)
{
picked.push(base);
}
}
let moved_oids: Vec<String> = moved
.iter()
.map(|want| want["want ".len()..].into())
.collect();
let requests = |base: &[String], wants: &[String]| {
let haves: Vec<String> = base.iter().map(|tip| format!("have {tip}")).collect();
let mut args = vec!["ofs-delta", "no-progress"];
args.extend(wants.iter().map(String::as_str));
args.extend(haves.iter().map(String::as_str));
let round = fetch(&args);
args.push("done");
(round, fetch(&args))
};
let bases: Vec<_> = (picked.iter().enumerate())
.map(|(n, base)| {
let moved_row =
crate::nginx::moved_fits_row(snapshot.repo_did.as_str(), base, &moved_oids);
let have = have_replies(n, moved_row).then(|| requests(base, &wants));
(base, have, moved_row.then(|| requests(base, &moved)))
})
.collect();
let made = in_parallel(bases.iter().enumerate(), |(n, (base, have, moved))| {
let mut len = 0;
if let Some((_, done)) = have {
len += replies.reply(&format!("have-{n}-done"), done, |path| match &aside {
Some(aside) if aside.covers(export_dir, base)? => {
let prior = (replies.prior_smart).map(|dir| dir.join(format!("have-{n}-done")));
aside.reply(snapshot, export_dir, base, path, prior.as_deref())
}
_ => upload_pack_into(export_dir, done, true, path),
})?;
}
if let Some((_, done)) = moved {
len += replies.reply(&format!("moved-{n}-done"), done, |path| {
upload_pack_into(export_dir, done, true, path)
})?;
}
Ok(len)
})?;
let full_pack = full_pack()?;
let kept = (made.iter())
.scan(full_pack, |left, len| {
*left = left.checked_sub(*len)?;
Some(())
})
.count();
for (n, (_, have, moved)) in bases.iter().enumerate().skip(kept) {
if have.is_some() {
replies.forget(&format!("have-{n}-done"))?;
}
if moved.is_some() {
replies.forget(&format!("moved-{n}-done"))?;
}
}
let bases = &bases[..kept];
in_parallel(bases.iter().enumerate(), |(n, (base, have, moved))| {
for (kind, request) in [("have", have), ("moved", moved)] {
if let Some((round, _)) = request {
let done_name = format!("{kind}-{n}-done");
round_reply(replies, base, &format!("{kind}-{n}"), round, &done_name)?;
}
}
Ok(())
})?;
if bases.iter().any(|(_, _, moved)| moved.is_some()) {
write(MOVED_WANTS, (moved_oids.join(" ") + "\n").as_bytes())?;
}
let index: String = (bases.iter())
.map(|(base, _, _)| base.join(" ") + "\n")
.collect();
write("fetch-bases", index.as_bytes())
}
fn round_reply(
replies: &Replies,
base: &[String],
name: &str,
round: &[u8],
done_name: &str,
) -> anyhow::Result<u64> {
replies.reply(name, round, |path| {
let mut acks = pkt_line(b"acknowledgments\n");
for tip in base {
acks.extend(pkt_line(format!("ACK {tip}\n").as_bytes()));
}
acks.extend(pkt_line(b"ready\n"));
acks.extend_from_slice(b"0001");
let done = std::fs::File::open(replies.smart_dir.join(done_name))?;
write_unlinked_with(path, |file| crate::reply_pack::behind(&acks, &done, file))
})
}
fn in_parallel<T: Send, R: Send>(
items: impl Iterator<Item = T>,
make: impl Fn(T) -> anyhow::Result<R> + Sync,
) -> anyhow::Result<Vec<R>> {
std::thread::scope(|scope| {
let made: Vec<_> = items.map(|item| scope.spawn(|| make(item))).collect();
made.into_iter()
.map(|thread| thread.join().expect("a reply thread panicked"))
.collect()
})
}
pub(crate) const MOVED_WANTS: &str = "moved-wants";
/// none when every tag is new, since the moved set would then be every tip.
fn moved_wants(
snapshot: &RepoSnapshot,
export_dir: &Path,
since: &[String],
) -> anyhow::Result<Vec<String>> {
let is_tag = |name: &str| name.starts_with("refs/tags/");
let tips: BTreeSet<&str> = (exported_refs(snapshot))
.filter(|(name, _)| !is_tag(name))
.map(|(_, raw)| raw)
.collect();
let mut command = git_command(export_dir);
command.args(["rev-list", "--stdin"]);
let revs: String = (tips.iter().map(|tip| format!("{tip}\n")))
.chain(since.iter().map(|tip| format!("^{tip}\n")))
.collect();
command.stdin(stdin_bytes(revs.as_bytes())?);
let made = String::from_utf8(run_cmd(command, "git rev-list")?.stdout)?;
let made: HashSet<&str> = made.lines().collect();
let (new, old): (Vec<_>, Vec<_>) = (snapshot.public_refs.iter())
.filter(|r| is_tag(&r.name))
.partition(|tag| made.contains(tag.peeled.as_deref().unwrap_or(&tag.raw)));
if old.is_empty() {
return Ok(Vec::new());
}
let mut wants = tips;
wants.extend(new.iter().map(|tag| tag.raw.as_str()));
Ok(wants.iter().map(|tip| format!("want {tip}")).collect())
}
/// the `tag` and `since` lines the aside was packed for, checked before a later push reuses it.
const TAG_ASIDE: &str = "tag-aside";
/// git's `^` exclusion misses trees and blobs both sides reach, which is why the cut reads this objects dir first.
const TAG_ASIDE_OBJECTS: &str = "tag-aside-objects";
/// packed once, and each cut walks only what the other refs add, not all history down to the oldest tag.
struct TagAside {
since: Vec<String>,
tags: Vec<String>,
objects: std::path::PathBuf,
pack: std::path::PathBuf,
}
impl TagAside {
fn of(
snapshot: &RepoSnapshot,
repo: &knot_git::Repo,
export_dir: &Path,
replies: &Replies,
since: Vec<String>,
) -> anyhow::Result<Self> {
let mut tags: Vec<String> = (snapshot.public_refs.iter())
.filter(|r| r.name.starts_with("refs/tags/"))
.map(|r| r.raw.clone())
.collect();
tags.sort();
tags.dedup();
let objects = replies.smart_dir.join(TAG_ASIDE_OBJECTS);
let index_path = replies.smart_dir.join(TAG_ASIDE);
let lines = |since: &[String]| -> String {
let tags = tags.iter().map(|tag| format!("tag {tag}\n"));
tags.chain(since.iter().map(|tip| format!("since {tip}\n")))
.collect()
};
// reuse the prior's aside only while the tags match and the export still has its `since` tips.
let prior = replies.prior_smart.and_then(|dir| {
let index = std::fs::read_to_string(dir.join(TAG_ASIDE)).ok()?;
let since: Vec<String> = (index.lines())
.filter_map(|line| line.strip_prefix("since "))
.map(str::to_string)
.collect();
let held = independent_tips(repo, export_dir, &since) == since;
(held && index == lines(&since)).then_some((dir, since))
});
if let Some((dir, since)) = prior
&& hardlink_tree(&dir.join(TAG_ASIDE_OBJECTS), &objects).is_ok()
&& let Some(pack) = kept_pack(&objects)?
{
std::fs::hard_link(dir.join(TAG_ASIDE), &index_path)?;
return Ok(Self {
since,
tags,
objects,
pack,
});
}
if objects.exists() {
std::fs::remove_dir_all(&objects)?;
}
let packs = objects.join("pack");
std::fs::create_dir_all(&packs)?;
let revs: String = (tags.iter().map(|tag| format!("{tag}\n")))
.chain(since.iter().map(|tip| format!("^{tip}\n")))
.collect();
let mut command = git_command(export_dir);
command.args(["pack-objects", "--revs", "-q", "--delta-base-offset"]);
command.arg(std::path::absolute(packs.join("pack"))?);
command.stdin(stdin_bytes(revs.as_bytes())?);
let name = String::from_utf8(run_cmd(command, "git pack-objects")?.stdout)?;
let pack = packs.join(format!("pack-{}.pack", name.trim()));
write_unlinked(&pack.with_extension("keep"), b"")?;
write_unlinked(&index_path, lines(&since).as_bytes())?;
Ok(Self {
since,
tags,
objects,
pack,
})
}
fn covers(&self, export_dir: &Path, base: &[String]) -> anyhow::Result<bool> {
let mut command = git_command(export_dir);
command
.args(["rev-list", "--count"])
.args(&self.since)
.arg("--not")
.args(base);
let output = run_cmd(command, "git rev-list --count")?;
Ok(String::from_utf8_lossy(&output.stdout).trim() == "0")
}
fn reply(
&self,
snapshot: &RepoSnapshot,
export_dir: &Path,
base: &[String],
path: &Path,
like: Option<&Path>,
) -> anyhow::Result<u64> {
let revs: String = exported_refs(snapshot)
.filter(|(name, _)| !name.starts_with("refs/tags/"))
.map(|(_, raw)| format!("{raw}\n"))
.chain(base.iter().chain(&self.tags).map(|tip| format!("^{tip}\n")))
.collect();
let cut = path.with_file_name(format!(".{}.cut", path.file_name().unwrap().display()));
let mut command = pack_objects_command(export_dir, &revs)?;
objects_before_alternates(&mut command, &self.objects, export_dir)?;
command.arg("--honor-pack-keep");
command_into(command, &cut)?;
// the aside goes first because that keeps the reply's bytes equal to the prior's up to the cut.
let packs = [self.pack.clone(), cut.clone()];
let like = like.and_then(|like| std::fs::File::open(like).ok());
let framed = write_unlinked_with(path, |file| {
crate::reply_pack::frame(
snapshot.object_format,
&packs,
&[],
None,
(&pkt_line(b"packfile\n"), file),
like.as_ref().map(|like| (like, 1)),
)
});
std::fs::remove_file(&cut).with_context(|| format!("remove {}", cut.display()))?;
framed
}
}
pub(crate) fn kept_pack(objects: &Path) -> anyhow::Result<Option<std::path::PathBuf>> {
let Ok(entries) = std::fs::read_dir(objects.join("pack")) else {
return Ok(None);
};
for entry in entries {
let path = entry?.path();
if path.extension().is_some_and(|ext| ext == "keep") {
return Ok(Some(path.with_extension("pack")));
}
}
Ok(None)
}
struct ShallowTips {
head: Option<ObjectId>,
branches: Vec<ObjectId>,
}
impl ShallowTips {
fn of(snapshot: &RepoSnapshot, head: Option<&str>) -> anyhow::Result<Self> {
let parse =
|tip: &str| ObjectId::from_hex(tip.as_bytes()).with_context(|| format!("tip {tip}"));
let head = head.map(parse).transpose()?;
let mut branches = snapshot
.public_refs
.iter()
.filter(|r| r.name.starts_with("refs/heads/"))
.map(|r| parse(&r.raw))
.collect::<anyhow::Result<Vec<_>>>()?;
branches.sort_unstable();
branches.dedup();
if branches.len() > DEPTH_ONE_BRANCHES {
branches.clear();
}
branches.retain(|tip| Some(*tip) != head);
Ok(Self { head, branches })
}
}
/// a push moves about one branch, but every reply weighs a whole tree.
/// past this many branches a depth-1 fetch of a branch gets the full pack.
pub const DEPTH_ONE_BRANCHES: usize = 64;
pub const SHALLOW_BRANCHES: &str = "shallow-branches";
pub(crate) fn independent_tips(
repo: &knot_git::Repo,
export_dir: &Path,
candidate: &[String],
) -> Vec<String> {
let present: Vec<&str> = candidate
.iter()
.map(String::as_str)
.filter(|tip| knot_types::Oid::from_hex(tip).is_ok_and(|oid| repo.contains(oid)))
.collect();
if present.is_empty() {
return Vec::new();
}
let mut command = git_command(export_dir);
command.args(["merge-base", "--independent"]).args(&present);
let Ok(output) = run_cmd(command, "git merge-base --independent") else {
return Vec::new();
};
let mut tips: Vec<String> = String::from_utf8_lossy(&output.stdout)
.split_whitespace()
.map(str::to_string)
.collect();
tips.sort();
tips
}