use super::partitions::{self, Added, PartitionPack};
use super::refs::export_head_state;
use super::replies::REPLY_KEYS;
use super::wire::{v0_advertisement, v2_command};
use super::{ExportTrace, exported_refs};
use gix::hash::ObjectId;
use std::collections::BTreeMap;
use std::path::Path;
use anyhow::Context as _;
use super::atomic::{write_unlinked, write_unlinked_with};
use super::replies::Replies;
use super::wire::{ensure_advertised, pkt_line, starts_with, write_v2_artifacts};
use crate::consolidate::Tail;
use crate::reply_pack::{Checksum, Grown, Rewrite};
use crate::snapshot::RepoSnapshot;
use crate::spare::Spare;
pub(crate) fn write_smart_artifacts(
snapshot: &RepoSnapshot,
export_dir: &Path,
prior_smart: Option<&Path>,
prior_refs: Option<&BTreeMap<String, String>>,
full: Full,
candidates: &[Vec<String>],
) -> anyhow::Result<()> {
let smart_dir = export_dir.join("smart");
// a leftover smart dir holds packs cut against other states. those can't stay.
if smart_dir.exists() {
std::fs::remove_dir_all(&smart_dir)
.with_context(|| format!("clear {}", smart_dir.display()))?;
}
std::fs::create_dir_all(&smart_dir)
.with_context(|| format!("create {}", smart_dir.display()))?;
let write = |name: &str, bytes: &[u8]| {
write_unlinked(&smart_dir.join(name), bytes)
.with_context(|| format!("write {}/{name}", smart_dir.display()))
};
let replies = Replies::new(&smart_dir, prior_smart, prior_refs)?;
let head = export_head_state(export_dir, snapshot)?;
let adv = v0_advertisement(snapshot, &head)?;
let mut advertisement = pkt_line(b"# service=git-upload-pack\n");
advertisement.extend_from_slice(b"0000");
advertisement.extend_from_slice(&adv);
ensure_advertised(&adv, &snapshot.public_refs)?;
write("advertisement", &advertisement)?;
let mut v0: Vec<u8> = snapshot
.public_refs
.iter()
.enumerate()
.flat_map(|(idx, r)| {
let line = if idx == 0 {
format!(
"want {} ofs-delta side-band-64k no-progress agent=knot-static\n",
r.raw
)
} else {
format!("want {}\n", r.raw)
};
pkt_line(line.as_bytes())
})
.collect();
v0.extend_from_slice(b"0000");
v0.extend_from_slice(&pkt_line(b"done\n"));
let wants: Vec<String> = exported_refs(snapshot)
.map(|(_, raw)| format!("want {raw}"))
.collect();
let mut args = vec!["ofs-delta", "no-progress"];
args.extend(wants.iter().map(String::as_str));
args.push("done");
let mut v2 = v2_command(snapshot, "fetch", "knotmirror/0.0.1", &args);
v2.extend_from_slice(b"0000");
// a v0 request without `done` is a negotiation round, answered with a bare NAK.
write("v0-nodone", &pkt_line(b"NAK\n"))?;
let (tx, rx) = std::sync::mpsc::sync_channel::<u64>(1);
std::thread::scope(|scope| -> anyhow::Result<()> {
let replies_ref = &replies;
let v0_ref = &v0;
let v2_ref = &v2;
let full_job = scope.spawn(move || -> anyhow::Result<()> {
let lengths = full_replies(
replies_ref,
export_dir,
snapshot.object_format,
snapshot.repo_did.as_str(),
[v0_ref, v2_ref],
full,
)?;
anyhow::ensure!(lengths.v0 > 0, "empty smart upload-pack response");
anyhow::ensure!(lengths.v2 > 0, "empty protocol v2 fetch response");
let _ = tx.send(lengths.v2);
Ok(())
});
let full_pack = || {
rx.recv()
.map_err(|_| anyhow::anyhow!("full_replies did not produce a v2 pack"))
};
let v2_res = write_v2_artifacts(
snapshot, export_dir, &head, &replies, &full_pack, candidates, &write,
);
let full_res = full_job.join().expect("full_replies panicked");
full_res.and(v2_res)
})?;
write(REPLY_KEYS, replies.keys().as_bytes())
}
pub(crate) enum Full<'a> {
Frame(&'a partitions::Fresh),
Grow {
prior: &'a Path,
added: &'a Added,
tips: &'a [Vec<ObjectId>; 2],
},
}
pub const FULL_REPLIES: [&str; 2] = ["upload-pack", "fetch-done"];
/// v0's NAK and an empty band-2 packet are as long as v2's `packfile` line.
/// both replies put the pack at one offset, which lets v2 share v0's extents.
pub(crate) fn full_prefixes() -> [Vec<u8>; 2] {
let mut nak = pkt_line(b"NAK\n");
nak.extend_from_slice(b"0005\x02");
[nak, pkt_line(b"packfile\n")]
}
pub(crate) const SAME_PACK: &str = "same-pack";
struct FullLengths {
v0: u64,
v2: u64,
}
fn full_replies(
replies: &Replies,
export_dir: &Path,
format: knot_types::ObjectFormat,
did: &str,
requests: [&[u8]; 2],
full: Full,
) -> anyhow::Result<FullLengths> {
let keys = requests.map(|request| replies.key(request));
let paths = FULL_REPLIES.map(|name| replies.smart_dir.join(name));
let same = replies.smart_dir.join(SAME_PACK);
let prior_same = (replies.prior_smart).filter(|dir| dir.join(SAME_PACK).exists());
let mut lengths = [0, 1].map(|at| replies.link(&keys[at], &paths[at]));
let linked = lengths.map(|length| length.is_some());
let framed = match &full {
Full::Frame(fresh) => {
let packs = [
(fresh.public().len(), 0),
(fresh.public().len(), fresh.cob().len()),
];
[0, 1].map(|at| (!linked[at]).then_some(packs[at]))
}
Full::Grow { .. } => [None, None],
};
let prior_tails = replies.prior_smart.and_then(crate::consolidate::tails);
let mut consolidated: [Option<Tail>; 2] = [None, None];
let prior_spares: [Option<Spare>; 2] = replies
.prior_smart
.map(crate::spare::read)
.unwrap_or_default();
let mut spares: [Option<Spare>; 2] = [0, 1].map(|at| {
let fitting = crate::spare::fitting(prior_spares[at].clone(), &paths[at], format);
linked[at].then(|| fitting.ok().flatten()).flatten()
});
let carry_same = match full {
Full::Frame(_) | Full::Grow { .. } if linked == [true, true] => prior_same.is_some(),
Full::Frame(fresh) => {
// v2 wants every tip v0 does, which means a linked v2 always has a linked v0.
let packs = |packs: &[PartitionPack]| -> Vec<std::path::PathBuf> {
packs
.iter()
.map(|pack| pack.path(export_dir, "pack"))
.collect()
};
let (public, cob) = (packs(fresh.public()), packs(fresh.cob()));
let since = prior_spares.iter().flatten().map(|spare| spare.since).max();
let reservation =
crate::spare::reserve_framed(export_dir, did, format, &public, &cob, since)?;
let fillers = (reservation.as_ref())
.map(|(room, nonce)| (*room, crate::spare::entries(nonce, 0..*room)));
let marks = std::cell::Cell::new([None, None]);
let frame = |v0: Option<&mut std::fs::File>| {
let [nak, packfile] = full_prefixes();
write_unlinked_with(&paths[1], |v2| {
let v0 = v0.map(|file| (nak.as_slice(), file));
let fillers = fillers
.as_ref()
.map(|(room, entries)| (*room, &entries[..]));
let v2 = (packfile.as_slice(), v2);
let framed = crate::reply_pack::frame_spared(
format, &public, &cob, v0, v2, None, fillers,
)?;
marks.set(framed);
Ok(())
})
};
if linked[0] {
lengths[1] = Some(frame(None)?);
} else {
lengths[0] = Some(write_unlinked_with(&paths[0], |v0| {
lengths[1] = Some(frame(Some(v0))?);
Ok(())
})?);
}
if let Some((room, nonce)) = reservation {
for (at, mark) in marks.get().into_iter().enumerate() {
if let (false, Some((start, state))) = (linked[at], mark) {
spares[at] = Some(Spare {
trailer: crate::spare::trailer_of(&paths[at], format)?,
nonce,
first: 0,
room,
at: start,
state,
tail_state: Some(state),
since: 0,
});
}
}
}
!linked[0] && fresh.cob().is_empty()
}
Full::Grow { prior, added, tips } => {
let segments = [
(added.public())
.map(|pack| std::fs::read(pack.path(export_dir, "pack")))
.transpose()
.context("read the new public pack")?,
added.v2().map(<[u8]>::to_vec),
];
// one linked reply still leaves the check holding, since it adds nothing.
let same = prior_same.is_some() && segments[0] == segments[1];
// v2 can share v0's bytes when v0 starts with the same NAK prefix.
let [nak, packfile] = full_prefixes();
let shares = same
&& linked == [false, false]
&& starts_with(&prior.join(FULL_REPLIES[0]), &nak)?;
let indexes = crate::spare::Indexes::open(
export_dir,
prior.parent().context("a prior outside an export")?,
format,
)?;
let mut fitting = [None, None];
for (at, spare) in fitting.iter_mut().enumerate() {
let prior_reply = prior.join(FULL_REPLIES[at]);
*spare = crate::spare::fitting(prior_spares[at].clone(), &prior_reply, format)?;
}
// with one pack in both, v0 only consolidates when v2 shares its bytes.
let consolidations = [0, 1].map(|at| {
let tail = prior_tails.as_ref()?.get(at)?;
let alone = !same || (at == 0 && shares);
(alone && !linked[at]).then_some(())?;
let consolidation = consolidation(
export_dir,
&prior.join(FULL_REPLIES[at]),
tail,
segments[at].as_deref()?,
&tips[at],
fitting[at].as_ref(),
format,
);
consolidation.unwrap_or_else(|error| {
ExportTrace::new().note(format_args!("export: tail kept: {error:#}"));
None
})
});
let mut plans = [None, None];
for (at, plan) in plans.iter_mut().enumerate() {
let Some(segment) = segments[at].as_deref().filter(|_| !linked[at]) else {
continue;
};
let tail = prior_tails.as_ref().map(|tails| tails[at].start());
let consolidation = (consolidations[at].as_ref()).map(|(kept, _, consolidated)| {
crate::spare::Consolidation {
kept: *kept,
entries: &consolidated.entries,
}
});
*plan = crate::spare::plan(
&prior.join(FULL_REPLIES[at]),
fitting[at].clone(),
segment,
tail,
consolidation,
did,
format,
&indexes,
)?;
}
let clone: CloneFile = crate::reply_pack::reflink_or_copy;
let grow = |at: usize, checksum: Checksum| -> anyhow::Result<_> {
let prior = prior.join(FULL_REPLIES[at]);
if let Some(plan) = &plans[at] {
let rewrite = Growth::Rewrite(plan.rewrite());
let grown =
grow_reply(&prior, Some(rewrite), &paths[at], format, checksum, clone)?;
return Ok((grown, plan.consolidates()));
}
let segment = segments[at].as_deref().map(Growth::Segment);
let Some((kept, added, consolidated)) = &consolidations[at] else {
let grown = grow_reply(&prior, segment, &paths[at], format, checksum, clone)?;
return Ok((grown, false));
};
let tail = Growth::Tail {
kept: *kept,
added: *added,
entries: &consolidated.entries,
};
match grow_reply(&prior, Some(tail), &paths[at], format, checksum, clone) {
Ok(grown) => Ok((grown, true)),
Err(error) => {
ExportTrace::new().note(format_args!("export: tail kept: {error:#}"));
let grown =
grow_reply(&prior, segment, &paths[at], format, Checksum::Hash, clone)?;
Ok((grown, false))
}
}
};
// with one pack in both and no spare to resume, v2 waits on v0's checksum.
let (send, receive) = std::sync::mpsc::sync_channel(1);
let taken = |at: usize| plans[at].as_ref().and_then(crate::spare::Plan::resumed);
let planned = |at: usize| plans[at].as_ref().map(|_| Checksum::Hash);
let v2_checksum = taken(1).unwrap_or(match same && !linked[0] {
true => Checksum::Given(receive),
false => Checksum::Hash,
});
let v0_checksum = taken(0).or_else(|| planned(0)).unwrap_or(Checksum::Hash);
let grown = std::thread::scope(|scope| {
let v2 = (!linked[1] && !shares).then(|| scope.spawn(|| grow(1, v2_checksum)));
let v0 = (!linked[0]).then(|| grow(0, v0_checksum)).transpose();
if let Ok(Some(((_, Some(grown)), _))) = &v0 {
let _ = send.send(grown.clone());
}
drop(send);
let v2 = v2.map(|v2| v2.join().expect("the v2 grow panicked"));
anyhow::Ok([v0?, v2.transpose()?])
})?;
for (at, grown) in grown.into_iter().enumerate() {
let Some(((length, grown), tail)) = grown else {
continue;
};
lengths[at] = Some(length);
spares[at] = match (&plans[at], grown) {
(Some(plan), Some(grown)) => plan.spare(&grown),
(None, Some(_)) => None,
(_, None) => fitting[at].clone(),
};
if let (true, Some((_, _, consolidation))) = (tail, &consolidations[at]) {
consolidation.write_region(replies.smart_dir)?;
consolidated[at] = Some(consolidation.tail.clone());
}
}
if shares {
consolidated[1] = consolidated[0].clone();
spares[1] = spares[0].clone();
lengths[1] = Some(write_unlinked_with(&paths[1], |v2| {
let v0 = std::fs::File::open(&paths[0])
.with_context(|| format!("open {}", paths[0].display()))?;
crate::reply_pack::reflink_or_copy(&v0, v2).context("share v0's reply")?;
use std::os::unix::fs::FileExt as _;
Ok(v2.write_all_at(&packfile, 0)?)
})?);
}
same
}
};
if carry_same {
write_unlinked(&same, b"")?;
}
// a framed reply's tail starts where its objects end. a grown one keeps the prior's
// tail unless it consolidated.
let mut tails = [None, None];
for (at, tail) in tails.iter_mut().enumerate() {
*tail = match framed[at] {
Some((public, cob)) => {
let file = std::fs::File::open(&paths[at])?;
let framed = crate::reply_pack::Framed::of(&file, format)?;
let header = framed.read(&file, 8..12)?;
let count = u32::from_be_bytes(header.try_into().expect("four bytes"));
Some(match &spares[at] {
Some(spare) => Tail::framed(spare.at, count - spare.room, public, cob),
None => Tail::framed(framed.body(), count, public, cob),
})
}
None => (consolidated[at].take())
.or_else(|| prior_tails.as_ref().map(|tails| tails[at].clone())),
};
}
if let [Some(v0), Some(v2)] = tails {
crate::consolidate::write_tails(replies.smart_dir, replies.prior_smart, &[v0, v2])?;
}
crate::spare::write(replies.smart_dir, &spares)?;
for at in [0, 1] {
replies.record(&keys[at], FULL_REPLIES[at]);
}
let [Some(v0), Some(v2)] = lengths else {
unreachable!("each full reply is linked or made");
};
Ok(FullLengths { v0, v2 })
}
fn consolidation(
export_dir: &Path,
prior: &Path,
tail: &Tail,
segment: &[u8],
tips: &[ObjectId],
spare: Option<&Spare>,
format: knot_types::ObjectFormat,
) -> anyhow::Result<Option<(u64, u32, crate::consolidate::Consolidated)>> {
let file = std::fs::File::open(prior).with_context(|| format!("open {}", prior.display()))?;
let framed = crate::reply_pack::Framed::of(&file, format)?;
// the objects end where the fillers start, and the count drops the unused fillers.
let ends = spare.map_or(framed.body(), |spare| spare.at);
if !tail.due(ends, segment.len() as u64) {
return Ok(None);
}
let header = framed.read(&file, 8..12)?;
let count =
u32::from_be_bytes(header.try_into().expect("four bytes")) - spare.map_or(0, Spare::left);
let added = crate::reply_pack::objects_of(segment);
let consolidated = crate::consolidate::consolidate(
export_dir,
prior.parent().context("a reply outside smart")?,
tail,
tips,
format,
)?;
anyhow::ensure!(
count.checked_add(added) == Some(consolidated.tail.count()),
"a consolidated tail of {} objects for {count} and {added} more",
consolidated.tail.count()
);
Ok(Some((tail.start(), added, consolidated)))
}
pub(crate) type CloneFile = fn(&std::fs::File, &std::fs::File) -> std::io::Result<()>;
pub(crate) enum Growth<'a> {
Segment(&'a [u8]),
Tail {
kept: u64,
added: u32,
entries: &'a [u8],
},
Rewrite(crate::reply_pack::Rewrite<'a>),
}
pub(crate) fn grow_reply(
prior: &Path,
growth: Option<Growth>,
path: &Path,
format: knot_types::ObjectFormat,
checksum: Checksum,
clone: CloneFile,
) -> anyhow::Result<(u64, Option<Grown>)> {
let mut grown = None;
let length = write_unlinked_with(path, |reply| {
let source =
std::fs::File::open(prior).with_context(|| format!("open {}", prior.display()))?;
clone(&source, reply).context("clone the prior reply")?;
grown = match growth {
None => None,
Some(Growth::Segment(segment)) => Some(Grown {
checksum: crate::reply_pack::extend(&source, reply, segment, format, checksum)?,
at_mark: None,
at_tail: None,
}),
Some(Growth::Tail {
kept,
added,
entries,
}) => {
let rewrite = Rewrite {
kept,
dropped: 0,
added,
entries,
mark: None,
tail_mark: None,
};
Some(crate::reply_pack::replace_tail(
&source, reply, rewrite, format, checksum,
)?)
}
Some(Growth::Rewrite(rewrite)) => Some(crate::reply_pack::replace_tail(
&source, reply, rewrite, format, checksum,
)?),
};
Ok(())
})?;
Ok((length, grown))
}