use std::collections::HashSet;
use std::path::{Path, PathBuf};
use anyhow::Context as _;
use gix::hash::ObjectId;
use crate::datastore::write::Durability;
use crate::store::RepoStore;
pub(crate) const PARTITIONS: &str = "partitions";
/// each pack is scanned once, the answer filed under its checksum.
/// gc keeps the answers of packs that kept exports link.
pub(crate) const REF_DELTAS_KIND: &str = "ref-deltas";
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct Partitions {
public: Vec<PartitionPack>,
cob: Vec<PartitionPack>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct PartitionPack(String);
impl PartitionPack {
pub(crate) fn new(export_dir: &Path, stem: &str) -> anyhow::Result<Self> {
let hash = stem.strip_prefix("pack-").unwrap_or_default();
anyhow::ensure!(
matches!(hash.len(), 40 | 64) && hash.bytes().all(|b| b.is_ascii_hexdigit()),
"{stem:?} isn't the name of a pack"
);
for ext in ["idx", "pack"] {
let path = export_dir
.join("objects/pack")
.join(format!("{stem}.{ext}"));
anyhow::ensure!(path.is_file(), "{} isn't there", path.display());
}
Ok(Self(stem.to_string()))
}
pub fn path(&self, export_dir: &Path, ext: &str) -> PathBuf {
export_dir
.join("objects/pack")
.join(format!("{}.{ext}", self.0))
}
}
/// no object is in two packs here, a reply framed from them can't send one twice.
/// `grow` can leave one in both, when a cob commit gets merged into a public branch.
pub(crate) struct Fresh(Partitions);
impl Fresh {
pub fn public(&self) -> &[PartitionPack] {
&self.0.public
}
pub fn cob(&self) -> &[PartitionPack] {
&self.0.cob
}
}
pub(crate) struct Added {
public: Option<PartitionPack>,
v2: Option<Vec<u8>>,
}
impl Added {
pub fn public(&self) -> Option<&PartitionPack> {
self.public.as_ref()
}
pub fn v2(&self) -> Option<&[u8]> {
self.v2.as_deref()
}
}
impl Partitions {
pub fn read(export_dir: &Path) -> anyhow::Result<Option<Self>> {
let text = match std::fs::read_to_string(export_dir.join(PARTITIONS)) {
Ok(text) => text,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
Err(error) => return Err(error).context("read the partitions"),
};
let mut partitions = Self {
public: Vec::new(),
cob: Vec::new(),
};
for line in text.lines() {
let (partition, stem) = match line.split_once(' ') {
Some(("public", stem)) => (&mut partitions.public, stem),
Some(("cob", stem)) => (&mut partitions.cob, stem),
_ => anyhow::bail!("a partitions line {line:?} names no partition"),
};
partition.push(PartitionPack::new(export_dir, stem)?);
}
anyhow::ensure!(
!partitions.public.is_empty(),
"the partitions have no public pack"
);
Ok(Some(partitions))
}
pub fn public(&self) -> &[PartitionPack] {
&self.public
}
pub fn cob(&self) -> &[PartitionPack] {
&self.cob
}
pub fn pack(
source: &Path,
export_dir: &Path,
public: &[ObjectId],
cob: &[ObjectId],
format: knot_types::ObjectFormat,
store: Option<&RepoStore>,
) -> anyhow::Result<Fresh> {
let repo = knot_git::Repo::open(source).context("open the source repo")?;
let marked = Marked::of(source, public, format, store)?;
let mut packs = Vec::new();
for pack in &marked.packs {
if pack.linkable() {
packs.push(link_pack(&pack.path, export_dir, format)?);
} else {
let keep = |at: u32| pack.has(at);
packs.extend(crate::packfile::filter_copy(
repo.git(),
&pack.path,
&keep,
export_dir,
format,
)?);
}
}
packs.extend(pack_into(source, export_dir, &marked.unpacked)?);
if packs.is_empty() {
packs.push(write_pack(source, export_dir, &[], b"")?);
}
let held = PackIndexes::open(export_dir, &packs, format)?;
let cob = unknown_reached(repo.git(), cob, &held)?;
let partitions = Self {
public: packs,
cob: pack_into(source, export_dir, &cob)?.into_iter().collect(),
};
partitions.write(export_dir)?;
Ok(Fresh(partitions))
}
pub fn grow(
mut self,
source: &Path,
export_dir: &Path,
public: &[ObjectId],
exported: &[ObjectId],
format: knot_types::ObjectFormat,
) -> anyhow::Result<Added> {
let repo = knot_git::Repo::open(source).context("open the source repo")?;
let held_public = PackIndexes::open(export_dir, &self.public, format)?;
let new_public = unknown_reached(repo.git(), public, &held_public)?;
let new_exported = if self.cob.is_empty() && public == exported {
new_public.clone()
} else {
let held = PackIndexes::open(export_dir, self.public.iter().chain(&self.cob), format)?;
unknown_reached(repo.git(), exported, &held)?
};
let new_cob = new_exported.without(&new_public);
let public_pack = pack_into(source, export_dir, &new_public)?;
self.public.extend(public_pack.clone());
self.cob.extend(pack_into(source, export_dir, &new_cob)?);
let v2 = match &public_pack {
_ if new_exported.is_empty() => None,
Some(pack) if new_cob.is_empty() && new_exported.len() == new_public.len() => {
Some(std::fs::read(pack.path(export_dir, "pack")).context("read the public pack")?)
}
_ => Some(pack_objects(
source,
&["--stdout"],
new_exported.object_list(),
)?),
};
self.write(export_dir)?;
Ok(Added {
public: public_pack,
v2,
})
}
/// the packs past what the replies frame stay geometric, each over twice the next, so an export
/// holds O(log writes) of them. a merged pack is built from its packs' own indexes, which keeps
/// every object in it once; the old files stay with the entries that link them.
pub fn merge_tails(
export_dir: &Path,
framed: [usize; 2],
format: knot_types::ObjectFormat,
) -> anyhow::Result<()> {
let Some(mut partitions) = Self::read(export_dir)? else {
return Ok(());
};
let mut merged = Vec::new();
merged.extend(merge_tail(
export_dir,
&mut partitions.public,
framed[0],
format,
)?);
merged.extend(merge_tail(
export_dir,
&mut partitions.cob,
framed[1],
format,
)?);
if merged.is_empty() {
return Ok(());
}
partitions.write(export_dir)?;
let dir = export_dir.join("objects/pack");
for pack in merged {
for entry in std::fs::read_dir(&dir).context("list the export's packs")? {
let path = entry?.path();
if path.file_stem().is_some_and(|stem| *stem == *pack.0) {
std::fs::remove_file(&path)
.with_context(|| format!("remove {}", path.display()))?;
}
}
}
Ok(())
}
fn write(&self, export_dir: &Path) -> anyhow::Result<()> {
let lines: String = (self
.public
.iter()
.map(|pack| format!("public {}\n", pack.0)))
.chain(self.cob.iter().map(|pack| format!("cob {}\n", pack.0)))
.collect();
let path = export_dir.join(PARTITIONS);
crate::datastore::write::publish(&path, lines.as_bytes(), Durability::Rename)
.with_context(|| format!("write {}", path.display()))
}
}
fn merge_tail(
export_dir: &Path,
packs: &mut Vec<PartitionPack>,
framed: usize,
format: knot_types::ObjectFormat,
) -> anyhow::Result<Vec<PartitionPack>> {
let size = |pack: &PartitionPack| {
std::fs::metadata(pack.path(export_dir, "pack")).map(|meta| meta.len())
};
let mut merged = Vec::new();
while packs.len() >= framed + 2 {
let newer = &packs[packs.len() - 1];
let older = &packs[packs.len() - 2];
if size(older)? > 2 * size(newer)? {
break;
}
let mut ids = HashSet::new();
for pack in [older, newer] {
let idx = pack.path(export_dir, "idx");
let index = gix::odb::pack::index::File::at(&idx, format.kind())
.with_context(|| format!("open {}", idx.display()))?;
ids.extend(index.iter().map(|entry| entry.oid));
}
let list: String = ids.iter().map(|oid| format!("{oid}\n")).collect();
let pack = write_pack(export_dir, export_dir, &[], list.as_bytes())?;
merged.extend(packs.drain(packs.len() - 2..));
packs.push(pack);
}
Ok(merged)
}
pub(crate) fn object_ids<'a>(
raws: impl IntoIterator<Item = &'a str>,
) -> anyhow::Result<Vec<ObjectId>> {
raws.into_iter()
.map(|raw| ObjectId::from_hex(raw.as_bytes()).with_context(|| format!("oid {raw:?}")))
.collect()
}
pub(crate) trait Holds {
fn holds(&self, oid: &gix::hash::oid) -> bool;
}
pub(crate) struct PackIndexes(Vec<gix::odb::pack::index::File>);
impl PackIndexes {
pub fn open<'a>(
export_dir: &Path,
packs: impl IntoIterator<Item = &'a PartitionPack>,
format: knot_types::ObjectFormat,
) -> anyhow::Result<Self> {
let dir = export_dir.join("objects/pack");
packs
.into_iter()
.map(|pack| {
let idx = dir.join(format!("{}.idx", pack.0));
gix::odb::pack::index::File::at(&idx, format.kind())
.with_context(|| format!("open {}", idx.display()))
})
.collect::<anyhow::Result<_>>()
.map(Self)
}
}
impl Holds for PackIndexes {
fn holds(&self, oid: &gix::hash::oid) -> bool {
self.0.iter().any(|idx| idx.lookup(oid).is_some())
}
}
/// each object keeps its tree path, since git's delta search looks for a base by it.
/// a partition only takes this walk, since `git rev-list new ^old` lists a restored blob again.
#[derive(Debug, Clone, Default)]
pub(crate) struct Segment {
list: Vec<u8>,
len: usize,
}
impl Segment {
pub fn is_empty(&self) -> bool {
self.len == 0
}
pub fn len(&self) -> usize {
self.len
}
pub(super) fn push(&mut self, oid: &gix::hash::oid, path: Option<&[u8]>) {
let mut hex = [0; gix::hash::Kind::longest().len_in_hex()];
self.list
.extend_from_slice(oid.hex_to_buf(&mut hex).as_bytes());
if let Some(path) = path {
self.list.push(b' ');
self.list.extend_from_slice(path);
}
self.list.push(b'\n');
self.len += 1;
}
fn lines(&self) -> impl Iterator<Item = (ObjectId, &[u8])> {
self.list
.split(|&byte| byte == b'\n')
.filter(|line| !line.is_empty())
.map(|line| {
let hex = line.split(|&byte| byte == b' ').next().unwrap_or(line);
(ObjectId::from_hex(hex).expect("a segment's own line"), line)
})
}
pub fn oids(&self) -> impl Iterator<Item = ObjectId> + '_ {
self.lines().map(|(oid, _)| oid)
}
pub fn without(&self, other: &Segment) -> Segment {
let other: HashSet<ObjectId> = other.oids().collect();
let mut without = Segment::default();
for (oid, line) in self.lines() {
if !other.contains(&oid) {
without.list.extend_from_slice(line);
without.list.push(b'\n');
without.len += 1;
}
}
without
}
pub fn object_list(&self) -> &[u8] {
&self.list
}
}
/// stopping at a known object is fine, `known` holds everything below it too.
pub(crate) fn unknown_reached(
repo: &gix::Repository,
tips: &[ObjectId],
known: &dyn Holds,
) -> anyhow::Result<Segment> {
let kind = repo.object_hash();
// paths grows and shrinks with the stack, each pop cuts it back to its object's path.
let mut stack: Vec<(ObjectId, usize, usize)> = tips.iter().map(|tip| (*tip, 0, 0)).collect();
let mut paths: Vec<u8> = Vec::new();
let mut seen = HashSet::new();
let mut reached = Segment::default();
while let Some((oid, start, end)) = stack.pop() {
paths.truncate(end);
if known.holds(&oid) || !seen.insert(oid) {
continue;
}
let object = repo
.find_object(oid)
.with_context(|| format!("read object {oid}"))?;
let named = (start < end).then_some(start..end);
match object.kind {
gix::object::Kind::Blob => {}
gix::object::Kind::Tree => {
for entry in gix::objs::TreeRefIter::from_bytes(&object.data, kind) {
let entry = entry.with_context(|| format!("parse tree {oid}"))?;
// a gitlink's commit is in another repo, nothing here can walk it.
if !entry.mode.is_commit() {
let at = paths.len();
if let Some(dir) = named.clone() {
paths.extend_from_within(dir);
paths.push(b'/');
}
paths.extend_from_slice(entry.filename);
stack.push((entry.oid.to_owned(), at, paths.len()));
}
}
}
gix::object::Kind::Commit => {
let mut commit = gix::objs::CommitRefIter::from_bytes(&object.data, kind);
let tree = commit
.tree_id()
.with_context(|| format!("parse commit {oid}"))?;
let none = paths.len();
stack.extend(commit.parent_ids().map(|parent| (parent, none, none)));
stack.push((tree, none, none));
}
gix::object::Kind::Tag => {
let target = gix::objs::TagRefIter::from_bytes(&object.data, kind)
.target_id()
.with_context(|| format!("parse tag {oid}"))?;
stack.push((target, paths.len(), paths.len()));
}
}
reached.push(&oid, named.map(|range| &paths[range]));
}
Ok(reached)
}
#[cfg(test)]
use super::reach::bitmap_reach;
pub(crate) use super::reach::{MARK_WALK, filed_ref_deltas, pack_objects, reach};
use super::reach::{Marked, link_pack, pack_into, write_pack};
#[cfg(any(test, feature = "bench"))]
use super::reach::{MarkedPack, mark_held, read_ids};
#[cfg(any(test, feature = "bench"))]
mod benches {
use super::*;
use crate::bench_fixture::{git, repo};
struct Nothing;
impl Holds for Nothing {
fn holds(&self, _: &gix::hash::oid) -> bool {
false
}
}
#[divan::bench(args = [1_000, 100_000])]
fn mark_held(bencher: divan::Bencher, files: usize) {
let dir = tempfile::tempdir().unwrap();
let (_, main) = repo(dir.path(), files, 16, 0);
let listed = git(
dir.path(),
&[
"rev-list",
"--objects",
"--no-object-names",
&main.to_string(),
],
b"",
);
let ids: Vec<[u8; 20]> = listed
.lines()
.map(|line| {
ObjectId::from_hex(line.as_bytes())
.unwrap()
.as_bytes()
.try_into()
.unwrap()
})
.collect();
let marked = Marked::of(dir.path(), &[main], knot_types::ObjectFormat::SHA1, None).unwrap();
let indexes: Vec<gix::odb::pack::index::File> = (marked.packs.iter())
.map(|pack| {
gix::odb::pack::index::File::at(
pack.path.with_extension("idx"),
gix::hash::Kind::Sha1,
)
.unwrap()
})
.collect();
bencher
.with_inputs(|| {
let packs: Vec<MarkedPack> = (marked.packs.iter())
.map(|pack| MarkedPack {
path: pack.path.clone(),
len: pack.len,
bits: vec![0; pack.bits.len()],
ref_deltas: false,
})
.collect();
(ids.clone(), packs)
})
.bench_local_values(|(ids, mut packs)| {
divan::black_box(super::mark_held(ids, &indexes, &mut packs));
packs
});
}
#[divan::bench(args = [1_000, 100_000])]
fn read_ids(bencher: divan::Bencher, files: usize) {
let dir = tempfile::tempdir().unwrap();
let (_, main) = repo(dir.path(), files, 16, 0);
let listed = git(
dir.path(),
&[
"rev-list",
"--objects",
"--no-object-names",
&main.to_string(),
],
b"",
);
let count = listed.lines().count();
let listed = format!("{listed}\n");
bencher.bench_local(|| super::read_ids::<20>(listed.as_bytes(), count).unwrap());
}
#[divan::bench(args = [100, 10_000])]
fn unknown_reached(bencher: divan::Bencher, files: usize) {
let dir = tempfile::tempdir().unwrap();
let (_, main) = repo(dir.path(), files, 16, 0);
let opened = knot_git::Repo::open(dir.path()).unwrap();
bencher.bench_local(|| {
let segment = super::unknown_reached(opened.git(), &[main], &Nothing).unwrap();
divan::black_box(segment.object_list().len())
});
}
}
#[cfg(test)]
#[path = "tests/partitions.rs"]
mod tests;