use std::path::Path;
use anyhow::Context as _;
use knot_types::ObjectFormat;
use crate::git_export::write_unlinked;
use crate::partitions::{Holds, PackIndexes, Partitions};
use crate::reply_pack::{Checksum, Framed, Grown, Hasher, Midstate, Rewrite};
/// a whole hash under this size takes about a tenth of a second.
pub(crate) const THRESHOLD: u64 = if cfg!(test) { 4 << 20 } else { 256 << 20 };
const FLOOR: u64 = 1024;
const CAP_SHARE: u64 = 100;
pub(crate) const SPARE: &str = "spare";
const FILLER: usize = 20;
#[derive(Clone, Debug, PartialEq, Eq)]
pub(crate) struct Spare {
pub trailer: Vec<u8>,
pub nonce: [u8; 16],
pub first: u32,
/// `room` is where the unused fillers end, not how many there are.
pub room: u32,
/// `state` hashes the pack up to `at`, where the fillers start.
pub at: u64,
pub state: Midstate,
pub tail_state: Option<Midstate>,
pub since: u64,
}
impl Spare {
pub fn left(&self) -> u32 {
self.room - self.first
}
fn line(&self) -> String {
format!(
"{} {} {} {} {} {} {} {}\n",
hex(&self.trailer),
hex(&self.nonce),
self.first,
self.room,
self.at,
self.state.to_hex(),
self.tail_state
.map_or_else(|| "-".to_string(), Midstate::to_hex),
self.since
)
}
fn parse(line: &str) -> Option<Self> {
let mut fields = line.split(' ');
let spare = Self {
trailer: unhex(fields.next()?)?,
nonce: unhex(fields.next()?)?.try_into().ok()?,
first: fields.next()?.parse().ok()?,
room: fields.next()?.parse().ok()?,
at: fields.next()?.parse().ok()?,
state: Midstate::from_hex(fields.next()?)?,
tail_state: match fields.next()? {
"-" => None,
hex => Some(Midstate::from_hex(hex)?),
},
since: fields.next()?.parse().ok()?,
};
(fields.next().is_none() && spare.first <= spare.room).then_some(spare)
}
pub fn ids(&self, format: ObjectFormat) -> Vec<Vec<u8>> {
(self.first..self.room)
.map(|n| filler_id(format, &self.nonce, n))
.collect()
}
}
pub(crate) fn read(dir: &Path) -> [Option<Spare>; 2] {
let text = std::fs::read_to_string(dir.join(SPARE)).unwrap_or_default();
let mut lines = text.lines();
[0, 1].map(|_| lines.next().and_then(Spare::parse))
}
pub(crate) fn write(dir: &Path, spares: &[Option<Spare>; 2]) -> anyhow::Result<()> {
let text: String = (spares.iter())
.map(|spare| {
spare
.as_ref()
.map_or_else(|| "-\n".to_string(), Spare::line)
})
.collect();
write_unlinked(&dir.join(SPARE), text.as_bytes())
}
pub(crate) fn room(objects: u64, since: u64) -> u32 {
let most = (objects / CAP_SHARE).max(FLOOR);
(2 * since).clamp(FLOOR, most).min(u32::MAX as u64 / 2) as u32
}
/// a nonce with a filler id the repo holds is rerolled with the next attempt.
pub(crate) fn nonce(
did: &str,
seed: &[u8],
room: u32,
format: ObjectFormat,
holds: &dyn Holds,
) -> anyhow::Result<[u8; 16]> {
for attempt in 0u32..16 {
let mut hasher = Hasher::new(ObjectFormat::SHA256);
hasher.update(did.as_bytes());
hasher.update(&[0]);
hasher.update(seed);
hasher.update(&attempt.to_be_bytes());
let nonce: [u8; 16] = hasher.finish()[..16].try_into().expect("16 bytes");
let held = (0..room).any(|n| {
let id = filler_id(format, &nonce, n);
gix::hash::oid::try_from_bytes(&id).is_ok_and(|id| holds.holds(id))
});
if !held {
return Ok(nonce);
}
}
anyhow::bail!("every nonce tried names an object the repo holds")
}
fn filler(nonce: &[u8; 16], n: u32) -> [u8; FILLER] {
let mut bytes = [0; FILLER];
bytes[..16].copy_from_slice(nonce);
bytes[16..].copy_from_slice(&n.to_be_bytes());
bytes
}
fn filler_id(format: ObjectFormat, nonce: &[u8; 16], n: u32) -> Vec<u8> {
let mut hasher = Hasher::new(format);
hasher.update(format!("blob {FILLER}\0").as_bytes());
hasher.update(&filler(nonce, n));
hasher.finish()
}
/// uncompressed stored blocks keep a filler's bytes fixed.
pub(crate) fn entries(nonce: &[u8; 16], range: std::ops::Range<u32>) -> Vec<u8> {
let mut entries = Vec::with_capacity(range.len() * (FILLER + 13));
for n in range {
let data = filler(nonce, n);
// a varint size: 4 in the low bits, then 1 << 4, which is 20.
entries.extend_from_slice(&[0xb4, 0x01, 0x78, 0x01, 0x01]);
entries.extend_from_slice(&(FILLER as u16).to_le_bytes());
entries.extend_from_slice(&(!(FILLER as u16)).to_le_bytes());
entries.extend_from_slice(&data);
entries.extend_from_slice(&adler32(&data).to_be_bytes());
}
entries
}
fn adler32(data: &[u8]) -> u32 {
let (mut a, mut b) = (1u32, 0u32);
for &byte in data {
a = (a + byte as u32) % 65521;
b = (b + a) % 65521;
}
(b << 16) | a
}
fn hex(bytes: &[u8]) -> String {
knot_types::lowercase_hex(bytes)
}
fn unhex(text: &str) -> Option<Vec<u8>> {
knot_types::decode_hex(text)
}
pub(crate) fn fitting(
spare: Option<Spare>,
path: &Path,
format: ObjectFormat,
) -> anyhow::Result<Option<Spare>> {
let Some(spare) = spare else {
return Ok(None);
};
let file = std::fs::File::open(path).with_context(|| format!("open {}", path.display()))?;
let framed = Framed::of(&file, format)?;
let fits = trailer(&file, &framed)? == spare.trailer && spare.at <= framed.body();
Ok(fits.then_some(spare))
}
fn trailer(file: &std::fs::File, framed: &Framed) -> anyhow::Result<Vec<u8>> {
let len = framed.format().kind().len_in_bytes() as u64;
framed.read(file, framed.body()..framed.body() + len)
}
pub(crate) fn trailer_of(path: &Path, format: ObjectFormat) -> anyhow::Result<Vec<u8>> {
let file = std::fs::File::open(path).with_context(|| format!("open {}", path.display()))?;
trailer(&file, &Framed::of(&file, format)?)
}
pub(crate) fn reserve_framed(
export_dir: &Path,
did: &str,
format: ObjectFormat,
public: &[std::path::PathBuf],
cob: &[std::path::PathBuf],
since: Option<u64>,
) -> anyhow::Result<Option<(u32, [u8; 16])>> {
let mut size = 0;
let mut objects = 0u64;
let mut seed = Vec::new();
for pack in public.iter().chain(cob) {
size += std::fs::metadata(pack)
.with_context(|| format!("stat {}", pack.display()))?
.len();
objects += crate::reply_pack::objects_in(pack)? as u64;
seed.extend_from_slice(pack.file_name().unwrap_or_default().as_encoded_bytes());
}
if size < THRESHOLD {
return Ok(None);
}
let room = room(objects, since.unwrap_or(0));
let indexes = Indexes::open(export_dir, export_dir, format)?;
Ok(Some((room, nonce(did, &seed, room, format, &indexes.all)?)))
}
pub(crate) struct Indexes {
all: PackIndexes,
new: PackIndexes,
}
impl Indexes {
pub fn open(export_dir: &Path, prior: &Path, format: ObjectFormat) -> anyhow::Result<Self> {
let partitions = Partitions::read(export_dir)?.context("an export without partitions")?;
let packs: Vec<_> = partitions.public().iter().chain(partitions.cob()).collect();
let before: Vec<_> = (Partitions::read(prior)?.iter())
.flat_map(|prior| {
prior
.public()
.iter()
.chain(prior.cob())
.cloned()
.collect::<Vec<_>>()
})
.collect();
let new = packs.iter().copied().filter(|pack| !before.contains(pack));
Ok(Self {
all: PackIndexes::open(export_dir, packs.iter().copied(), format)?,
new: PackIndexes::open(export_dir, new, format)?,
})
}
}
fn holds_any(holds: &dyn Holds, ids: &[Vec<u8>]) -> bool {
(ids.iter()).any(|id| gix::hash::oid::try_from_bytes(id).is_ok_and(|id| holds.holds(id)))
}
pub(crate) struct Plan {
kept: u64,
dropped: u32,
added: u32,
entries: Vec<u8>,
mark: u64,
tail_mark: Option<u64>,
resume: Option<Midstate>,
consolidates: bool,
slots: Option<([u8; 16], u32, u32, u64)>,
tail_state: Option<Midstate>,
}
impl Plan {
pub fn rewrite(&self) -> Rewrite<'_> {
Rewrite {
kept: self.kept,
dropped: self.dropped,
added: self.added,
entries: &self.entries,
mark: Some(self.mark),
tail_mark: self.tail_mark,
}
}
pub fn resumed(&self) -> Option<Checksum> {
self.resume.map(Checksum::Spare)
}
pub fn consolidates(&self) -> bool {
self.consolidates
}
pub fn spare(&self, grown: &Grown) -> Option<Spare> {
let ((nonce, first, room, since), state) = self.slots.zip(grown.at_mark)?;
let tail_state = match self.consolidates {
true => Some(state),
false => grown.at_tail.or(self.tail_state),
};
Some(Spare {
trailer: grown.checksum.clone(),
nonce,
first,
room,
at: self.mark,
state,
tail_state,
since,
})
}
}
pub(crate) struct Consolidation<'a> {
pub kept: u64,
pub entries: &'a [u8],
}
#[allow(clippy::too_many_arguments)]
pub(crate) fn plan(
prior: &Path,
spare: Option<Spare>,
segment: &[u8],
tail: Option<u64>,
consolidation: Option<Consolidation>,
did: &str,
format: ObjectFormat,
indexes: &Indexes,
) -> anyhow::Result<Option<Plan>> {
let (added, segment) = crate::reply_pack::pack_entries(segment, format.kind().len_in_bytes())?;
let consolidates = consolidation.is_some();
let file = std::fs::File::open(prior).with_context(|| format!("open {}", prior.display()))?;
let framed = Framed::of(&file, format)?;
let ends = spare.as_ref().map_or(framed.body(), |spare| spare.at);
let (kept, entries) = match consolidation {
Some(Consolidation { kept, entries }) => (kept, entries),
None => (ends, segment),
};
let mark = kept + entries.len() as u64;
let resumed = |spare: &Spare| match consolidates {
true => spare.tail_state,
false => Some(spare.state),
};
if let Some(spare) = &spare
&& let Some(first) = (spare.first.checked_add(added)).filter(|first| *first <= spare.room)
&& let Some(state) = resumed(spare)
{
let left = Spare {
first,
..spare.clone()
};
if !holds_any(&indexes.new, &left.ids(format)) {
let mut grown = entries.to_vec();
grown.extend(self::entries(&spare.nonce, first..spare.room));
return Ok(Some(Plan {
kept,
dropped: 0,
added: 0,
entries: grown,
mark,
tail_mark: None,
resume: Some(state),
consolidates,
slots: Some((spare.nonce, first, spare.room, spare.since + added as u64)),
tail_state: spare.tail_state,
}));
}
}
if spare.is_none() && mark < THRESHOLD {
return Ok(None);
}
let count = u32::from_be_bytes(framed.read(&file, 8..12)?.try_into().expect("four bytes"));
let left = spare.as_ref().map_or(0, Spare::left);
let objects = (count - left + added) as u64;
let mut grown = entries.to_vec();
let mut slots = None;
if mark >= THRESHOLD {
let since = spare.as_ref().map_or(0, |spare| spare.since) + added as u64;
let room = room(objects, since);
let nonce = nonce(did, &trailer(&file, &framed)?, room, format, &indexes.all)?;
grown.extend(self::entries(&nonce, 0..room));
slots = Some((nonce, 0, room, 0));
}
let room = slots.map_or(0, |(_, _, room, _)| room);
Ok(Some(Plan {
kept,
dropped: left,
added: added.checked_add(room).context("too many objects")?,
entries: grown,
mark,
tail_mark: tail.filter(|tail| (12..kept).contains(tail)),
resume: None,
consolidates,
slots,
tail_state: None,
}))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn room_is_twice_what_pushes_added_within_bounds() {
assert_eq!(room(10_000_000, 30_000), 60_000);
assert_eq!(room(10_000_000, 10), 1024);
assert_eq!(room(10_000_000, 10_000_000), 100_000);
assert_eq!(room(1000, 1_000_000), 1024);
}
struct These(Vec<Vec<u8>>);
impl Holds for These {
fn holds(&self, oid: &gix::hash::oid) -> bool {
self.0.iter().any(|id| id.as_slice() == oid.as_bytes())
}
}
#[test]
fn a_nonce_names_no_object_the_repo_holds() -> anyhow::Result<()> {
let format = ObjectFormat::SHA1;
let none = These(Vec::new());
let first = nonce("did:plc:a", b"seed", 8, format, &none)?;
assert_eq!(first, nonce("did:plc:a", b"seed", 8, format, &none)?);
assert_ne!(first, nonce("did:plc:b", b"seed", 8, format, &none)?);
assert_ne!(first, nonce("did:plc:a", b"seeds", 8, format, &none)?);
let held = These(vec![filler_id(format, &first, 7)]);
let other = nonce("did:plc:a", b"seed", 8, format, &held)?;
assert_ne!(other, first);
assert!(!holds_any(
&held,
&(0..8)
.map(|n| filler_id(format, &other, n))
.collect::<Vec<_>>()
));
Ok(())
}
#[test]
fn spare_slots_read_back_as_written() -> anyhow::Result<()> {
let dir = tempfile::tempdir()?;
let sha1 = Spare {
trailer: vec![7; 20],
nonce: [3; 16],
first: 5,
room: 1024,
at: 123_456,
state: Midstate::from_hex(&"0123abcd".repeat(5)).unwrap(),
tail_state: None,
since: 9,
};
let sha256 = Spare {
trailer: vec![8; 32],
state: Midstate::from_hex(&"89abcdef".repeat(8)).unwrap(),
tail_state: Midstate::from_hex(&"01234567".repeat(8)),
..sha1.clone()
};
for spares in [[Some(sha1.clone()), None], [None, Some(sha256)]] {
write(dir.path(), &spares)?;
assert_eq!(read(dir.path()), spares);
}
Ok(())
}
#[test]
fn fillers_are_blobs_with_their_ids() {
let nonce = [5; 16];
let mut pack = crate::reply_pack::pack_header(4);
pack.extend(entries(&nonce, 2..6));
let mut hasher = Hasher::new(ObjectFormat::SHA1);
hasher.update(&pack);
pack.extend(hasher.finish());
let repo = tempfile::tempdir().unwrap();
let mut init = crate::git_export::git_command(repo.path());
init.args(["init", "-q", "--bare"]);
crate::git_export::run_cmd(init, "git init").unwrap();
let mut index = crate::git_export::git_command(repo.path());
index.args(["index-pack", "--strict", "--stdin"]);
index.stdin(crate::git_export::stdin_bytes(&pack).unwrap());
crate::git_export::run_cmd(index, "git index-pack").unwrap();
let mut all = crate::git_export::git_command(repo.path());
all.args([
"cat-file",
"--batch-all-objects",
"--batch-check=%(objectname)",
]);
let listed = crate::git_export::run_cmd(all, "git cat-file")
.unwrap()
.stdout;
let mut listed: Vec<String> = String::from_utf8(listed)
.unwrap()
.lines()
.map(String::from)
.collect();
listed.sort();
let spare = Spare {
trailer: Vec::new(),
nonce,
first: 2,
room: 6,
at: 0,
state: Midstate::from_hex(&"0".repeat(40)).unwrap(),
tail_state: None,
since: 0,
};
let mut ids: Vec<String> = (spare.ids(ObjectFormat::SHA1).iter())
.map(|id| hex(id))
.collect();
ids.sort();
assert_eq!(listed, ids);
}
}