use std::collections::HashSet;
use mkit_core::hash::{Hash, to_hex};
use mkit_core::object::{Object, ObjectType};
use mkit_core::ops::graph::{ClosureMode, MAX_REACHABLE, children, reachable_closure_checked};
use mkit_core::ops::merge::is_ancestor;
use mkit_core::pack;
use mkit_core::protocol::UploadLimits;
use mkit_core::refs::RefWriteCondition;
use mkit_core::store::{ObjectStore, StoreError};
use mkit_core::transfer::{self, PackPlan};
use super::{
DispatchError, PackSink, build_and_upload_packs, effective_payload_cap,
encode_delta_candidates_batch, estimate_pack_sizes,
};
pub const MAX_SPLIT_STEPS: usize = 1_000;
pub const MAX_CHAIN_COMMITS: usize = 1_000_000;
pub trait StepAuthority: Send + Sync {
fn authorize(
&self,
branch: &str,
first: RefWriteCondition,
later_steps: bool,
) -> Result<(), String>;
}
pub struct PushControl<'a> {
pub authority: Option<&'a dyn StepAuthority>,
pub max_steps: usize,
pub max_chain: usize,
pub shallow: HashSet<Hash>,
}
impl std::fmt::Debug for PushControl<'_> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("PushControl")
.field("checks_grants", &self.authority.is_some())
.field("max_steps", &self.max_steps)
.field("max_chain", &self.max_chain)
.field("shallow", &self.shallow.len())
.finish()
}
}
impl Default for PushControl<'_> {
fn default() -> Self {
Self {
authority: None,
max_steps: MAX_SPLIT_STEPS,
max_chain: MAX_CHAIN_COMMITS,
shallow: HashSet::new(),
}
}
}
pub(super) fn data_pack_budget(limits: UploadLimits) -> usize {
limits
.tickets_per_advance
.map_or(6, |tickets| tickets.min(7).saturating_sub(1))
}
struct Chain {
commits: Vec<Hash>,
seed: HashSet<Hash>,
}
fn first_parent(obj: &Object) -> Option<Hash> {
match obj {
Object::Commit(c) => c.parents.first().copied(),
Object::Remix(r) => r.parents.first().copied(),
_ => None,
}
}
fn held_closure(store: &ObjectStore, root: Option<Hash>) -> Result<HashSet<Hash>, DispatchError> {
let Some(root) = root.filter(|root| store.contains(root)) else {
return Ok(HashSet::new());
};
match reachable_closure_checked(store, [&root]) {
Ok((set, false)) => Ok(set.into_iter().collect()),
Ok((_, true)) => Err(too_large_to_split()),
Err(StoreError::ObjectNotFound(_)) => Ok(HashSet::new()),
Err(e) => Err(e.into()),
}
}
fn too_large_to_split() -> DispatchError {
DispatchError::PushSplitLimit(format!(
"the history is too large to split ({MAX_REACHABLE} objects or more); no advance of this branch was published"
))
}
fn walk_chain(
store: &ObjectStore,
tip: Hash,
remote_tip: Option<Hash>,
ctl: &PushControl<'_>,
) -> Result<Chain, DispatchError> {
let held = held_closure(store, remote_tip)?;
let mut commits = Vec::new();
let mut base = None;
let mut cursor = tip;
loop {
if held.contains(&cursor) {
base = Some(cursor);
break;
}
if commits.len() >= ctl.max_chain {
return Err(DispatchError::PushSplitLimit(format!(
"the first-parent history between the remote and the local tip is longer than {} commits, too long to split; no advance of this branch was published",
ctl.max_chain
)));
}
if crate::signal::is_shutdown() {
return Err(DispatchError::Interrupted);
}
commits.push(cursor);
if ctl.shallow.contains(&cursor) {
break;
}
match first_parent(&store.read_object(&cursor)?) {
Some(parent) => cursor = parent,
None => break,
}
}
commits.reverse();
let seed = match base {
Some(base) if Some(base) == remote_tip => held,
base => held_closure(store, base)?,
};
Ok(Chain { commits, seed })
}
fn cumulative_weights(
store: &ObjectStore,
chain: &Chain,
ctl: &PushControl<'_>,
) -> Result<Vec<u64>, DispatchError> {
let mut seen = chain.seed.clone();
let mut cumulative = Vec::with_capacity(chain.commits.len() + 1);
let mut total = 0_u64;
cumulative.push(0);
for commit in &chain.commits {
if crate::signal::is_shutdown() {
return Err(DispatchError::Interrupted);
}
let mut stack = vec![*commit];
while let Some(hash) = stack.pop() {
if !seen.insert(hash) {
continue;
}
if seen.len() > MAX_REACHABLE {
return Err(too_large_to_split());
}
total = total
.saturating_add(store.object_metadata(&hash)?.len())
.saturating_add(pack::ENTRY_FRAME_LEN as u64);
if matches!(
store.object_type(&hash)?,
ObjectType::Blob | ObjectType::Delta
) {
continue;
}
let mode = if ctl.shallow.contains(&hash) {
ClosureMode::Snapshot
} else {
ClosureMode::History
};
stack.extend(children(&store.read_object(&hash)?, mode));
}
cumulative.push(total);
}
Ok(cumulative)
}
fn pack_weight_cap(payload_cap: u64, max_pack_bytes: Option<u64>) -> u64 {
let serialized = max_pack_bytes.map_or(u64::MAX, |limit| {
limit.saturating_sub((pack::HEADER_LEN + pack::TRAILER_LEN) as u64)
});
payload_cap.min(serialized).max(1)
}
struct Cutter<'a> {
store: &'a ObjectStore,
chain: &'a Chain,
weights: &'a [u64],
remote_tip: Option<Hash>,
limits: UploadLimits,
cap: u64,
budget: usize,
must_reach: usize,
}
impl Cutter<'_> {
fn probe(&self, from: usize, to: usize) -> Result<(PackPlan, usize), DispatchError> {
if crate::signal::is_shutdown() {
return Err(DispatchError::Interrupted);
}
let base = match from {
0 => self.remote_tip,
from => Some(self.chain.commits[from - 1]),
};
let plan = transfer::plan_pack_with(
self.store,
self.chain.commits[to - 1],
base,
encode_delta_candidates_batch,
)?;
let packs =
estimate_pack_sizes(self.store, &plan, self.cap, self.limits.max_pack_bytes)?.len();
Ok((plan, packs))
}
fn furthest(&self, from: usize, limit: u64) -> usize {
let bound = self.weights[from].saturating_add(limit);
let count = self.weights.partition_point(|&weight| weight <= bound);
count
.saturating_sub(1)
.max(from + 1)
.min(self.weights.len() - 1)
}
fn cut(&self, from: usize) -> Result<usize, DispatchError> {
let wcap = pack_weight_cap(self.cap, self.limits.max_pack_bytes);
let floor_weight = wcap.saturating_mul((self.budget / 2) as u64);
let natural = self.furthest(from, floor_weight);
let floor = if from == 0 {
natural.max(self.must_reach)
} else {
natural
};
let ceiling = self
.furthest(from, wcap.saturating_mul(self.budget as u64))
.max(floor);
let (mut fits, mut over) = (floor, ceiling);
while fits < over {
let mid = fits + (over - fits).div_ceil(2);
if self.probe(from, mid)?.1 <= self.budget {
fits = mid;
} else {
over = mid - 1;
}
}
let heavy = self.weights[fits] - self.weights[from] > floor_weight;
if fits == floor && heavy {
self.verify_forced(from, fits, from == 0 && floor > natural)?;
}
Ok(fits)
}
fn verify_forced(
&self,
from: usize,
to: usize,
holds_remote_head: bool,
) -> Result<(), DispatchError> {
let (plan, packs) = self.probe(from, to)?;
if packs <= self.budget {
return Ok(());
}
build_and_upload_packs(PackSink::Count, self.store, plan, self.cap, self.limits)
.map(drop)
.map_err(|error| match error {
DispatchError::PushTooLarge { packs, limit, .. } => DispatchError::PushTooLarge {
packs,
limit,
commit: Some(self.describe(self.chain.commits[to - 1])),
holds_remote_head,
},
other => other,
})
}
fn describe(&self, commit: Hash) -> String {
match self.store.read_object(&commit) {
Ok(Object::Commit(c)) if c.parents.len() > 1 => format!(
"merge commit {} (second parent {})",
to_hex(&commit),
to_hex(&c.parents[1])
),
_ => format!("commit {}", to_hex(&commit)),
}
}
}
fn positions_to_reach(
store: &ObjectStore,
chain: &Chain,
remote_tip: Option<Hash>,
) -> Result<usize, DispatchError> {
let Some(remote) = remote_tip.filter(|remote| store.contains(remote)) else {
return Ok(0);
};
let (mut low, mut high) = (1, chain.commits.len());
if !is_ancestor(store, remote, chain.commits[high - 1])? {
return Ok(0);
}
while low < high {
let mid = low + (high - low) / 2;
if is_ancestor(store, remote, chain.commits[mid - 1])? {
high = mid;
} else {
low = mid + 1;
}
}
Ok(low)
}
#[allow(clippy::too_many_arguments)]
pub fn plan_push_steps(
store: &ObjectStore,
tip: Hash,
remote_tip: Option<Hash>,
limits: UploadLimits,
pack_payload_cap: u64,
ctl: &PushControl<'_>,
condition: RefWriteCondition,
branch: &str,
) -> Result<Vec<Hash>, DispatchError> {
let cap = effective_payload_cap(pack_payload_cap, limits.max_pack_bytes)?;
let chain = walk_chain(store, tip, remote_tip, ctl)?;
if chain.commits.is_empty() {
return Ok(vec![tip]);
}
if let Some(authority) = ctl.authority {
authority
.authorize(branch, condition, chain.commits.len() > 1)
.map_err(DispatchError::PushNotAuthorized)?;
}
let weights = cumulative_weights(store, &chain, ctl)?;
let cutter = Cutter {
store,
chain: &chain,
weights: &weights,
remote_tip,
limits,
cap,
budget: data_pack_budget(limits),
must_reach: positions_to_reach(store, &chain, remote_tip)?,
};
let mut steps = Vec::new();
let mut from = 0;
while from < chain.commits.len() {
if steps.len() >= ctl.max_steps {
return Err(DispatchError::PushSplitLimit(format!(
"this push would need more than {} advances; no advance of this branch was published. Ask the operator to raise max_pack_bytes, or push an ancestor commit first",
ctl.max_steps
)));
}
from = cutter.cut(from)?;
steps.push(chain.commits[from - 1]);
}
Ok(steps)
}
#[cfg(test)]
mod tests {
use super::*;
use mkit_core::layout::RepoLayout;
use mkit_core::object::{Blob, Commit, EntryMode, Identity, Tree, TreeEntry};
use mkit_core::ops::graph::reachable_objects;
use mkit_core::serialize;
use proptest::prelude::*;
use tempfile::TempDir;
fn store() -> (TempDir, ObjectStore) {
let dir = TempDir::new().unwrap();
let store = ObjectStore::init(&RepoLayout::single(dir.path())).unwrap();
(dir, store)
}
fn put(store: &ObjectStore, obj: &Object) -> Hash {
store.write(&serialize::serialize(obj).unwrap()).unwrap()
}
fn bytes(seed: u64, len: usize, compressible: bool) -> Vec<u8> {
if compressible {
return vec![seed.to_le_bytes()[0]; len];
}
let mut out = vec![0; len];
let mut state = seed | 1;
for chunk in out.chunks_mut(8) {
state = state
.wrapping_mul(6_364_136_223_846_793_005)
.wrapping_add(1_442_695_040_888_963_407);
chunk.copy_from_slice(&state.to_le_bytes()[..chunk.len()]);
}
out
}
fn commit(store: &ObjectStore, files: &[Hash], parents: Vec<Hash>, seed: u8) -> Hash {
let mut entries: Vec<TreeEntry> = files
.iter()
.enumerate()
.map(|(i, hash)| TreeEntry {
name: format!("f{i:04}").into_bytes(),
mode: EntryMode::Blob,
object_hash: *hash,
})
.collect();
entries.sort_by(|a, b| a.name.cmp(&b.name));
let tree = put(store, &Object::Tree(Tree { entries }));
put(
store,
&Object::Commit(Commit::new_unannotated(
tree,
parents,
Identity::ed25519([7; 32]),
[0; 32],
vec![seed],
u64::from(seed),
[0; 64],
)),
)
}
fn blob(store: &ObjectStore, seed: u64, len: usize) -> Hash {
put(
store,
&Object::Blob(Blob {
data: bytes(seed, len, false),
}),
)
}
fn line(store: &ObjectStore, n: u8) -> Vec<Hash> {
let mut files = Vec::new();
let mut commits: Vec<Hash> = Vec::new();
for i in 0..n {
files.push(blob(store, u64::from(i) + 1, 100));
let parents = commits.last().copied().into_iter().collect();
commits.push(commit(store, &files, parents, i));
}
commits
}
#[test]
fn the_walk_stops_at_the_commit_the_remote_holds() {
let (_dir, store) = store();
let c = line(&store, 6);
let chain = walk_chain(&store, c[5], Some(c[2]), &PushControl::default()).unwrap();
assert_eq!(chain.commits, [c[3], c[4], c[5]]);
assert_eq!(chain.seed, held_closure(&store, Some(c[2])).unwrap());
let chain = walk_chain(&store, c[5], None, &PushControl::default()).unwrap();
assert_eq!(chain.commits, c);
assert!(chain.seed.is_empty());
}
#[test]
fn a_diverged_remote_seeds_the_shared_ancestor_not_its_own_tip() {
let (_dir, store) = store();
let c = line(&store, 6);
let other = blob(&store, 900, 100);
let theirs = commit(&store, &[other], vec![c[1]], 90);
let chain = walk_chain(&store, c[5], Some(theirs), &PushControl::default()).unwrap();
assert_eq!(chain.commits, [c[2], c[3], c[4], c[5]]);
assert_eq!(chain.seed, held_closure(&store, Some(c[1])).unwrap());
}
#[test]
fn the_walk_never_enters_a_second_parent_side() {
let (_dir, store) = store();
let c = line(&store, 4);
let side_file = blob(&store, 500, 100);
let side = commit(&store, &[side_file], vec![c[0]], 50);
let merge = commit(&store, &[side_file], vec![c[3], side], 60);
let chain = walk_chain(&store, merge, None, &PushControl::default()).unwrap();
assert_eq!(chain.commits, [c[0], c[1], c[2], c[3], merge]);
let weights = cumulative_weights(&store, &chain, &PushControl::default()).unwrap();
let brought = weights[5] - weights[4];
let lone = weights[4] - weights[3];
assert!(brought > lone, "{weights:?}");
}
#[test]
fn a_shallow_boundary_ends_the_walk_and_its_parents_are_not_followed() {
let (_dir, store) = store();
let c = line(&store, 6);
let control = PushControl {
shallow: std::iter::once(c[3]).collect(),
..PushControl::default()
};
let chain = walk_chain(&store, c[5], None, &control).unwrap();
assert_eq!(chain.commits, [c[3], c[4], c[5]]);
let weights = cumulative_weights(&store, &chain, &control).unwrap();
let everything = cumulative_weights(&store, &chain, &PushControl::default()).unwrap();
assert!(weights[1] < everything[1]);
}
#[test]
fn a_too_long_walk_is_refused() {
let (_dir, store) = store();
let c = line(&store, 6);
let control = PushControl {
max_chain: 4,
..PushControl::default()
};
let error = walk_chain(&store, c[5], None, &control).err().unwrap();
assert!(
matches!(error, DispatchError::PushSplitLimit(_)),
"{error:?}"
);
}
#[test]
fn cumulative_weights_add_up_to_the_new_objects() {
let (_dir, store) = store();
let c = line(&store, 6);
let chain = walk_chain(&store, c[5], Some(c[1]), &PushControl::default()).unwrap();
let weights = cumulative_weights(&store, &chain, &PushControl::default()).unwrap();
assert_eq!(weights.len(), chain.commits.len() + 1);
assert!(weights.windows(2).all(|w| w[0] < w[1]));
let held = reachable_objects(&store, &c[1]).unwrap();
let expected: u64 = reachable_objects(&store, &c[5])
.unwrap()
.difference(&held)
.map(|h| store.object_metadata(h).unwrap().len() + pack::ENTRY_FRAME_LEN as u64)
.sum();
assert_eq!(*weights.last().unwrap(), expected);
}
#[test]
fn a_step_forced_to_reach_the_remote_head_says_how_to_avoid_it() {
let (_dir, store) = store();
let a = line(&store, 4);
let their_file = blob(&store, 900, 100);
let theirs = commit(&store, &[their_file], Vec::new(), 90);
let big: Vec<Hash> = (0..20).map(|i| blob(&store, 300 + i, 2048)).collect();
let merge = commit(&store, &big, vec![a[3], theirs], 91);
let limits = UploadLimits {
max_pack_bytes: Some(8192),
tickets_per_advance: Some(7),
ticket_threshold_bytes: Some(0),
};
let error = plan_push_steps(
&store,
merge,
Some(theirs),
limits,
4096,
&PushControl::default(),
RefWriteCondition::Match(theirs),
"main",
)
.unwrap_err();
assert!(
matches!(
&error,
DispatchError::PushTooLarge {
holds_remote_head: true,
..
}
),
"{error:?}"
);
assert!(
error.to_string().contains("rebase onto the remote head"),
"{error}"
);
}
proptest! {
#![proptest_config(ProptestConfig::with_cases(64))]
#[test]
fn estimate_bounds_the_real_pack_count_and_three_caps_always_fit(
sizes in prop::collection::vec((16_usize..1500, any::<bool>()), 1..12),
payload_cap in 512_u64..16_384,
max_pack in prop::option::of(2048_u64..12_000),
with_deltas in any::<bool>(),
) {
let (_dir, store) = store();
let files: Vec<Hash> = sizes
.iter()
.enumerate()
.map(|(i, (len, compressible))| {
put(&store, &Object::Blob(Blob { data: bytes(i as u64 + 1, *len, *compressible) }))
})
.collect();
let (tip, base, seed) = if with_deltas {
let base = commit(&store, &files, Vec::new(), 1);
let edited: Vec<Hash> = sizes
.iter()
.enumerate()
.map(|(i, (len, compressible))| {
let mut data = bytes(i as u64 + 1, *len, *compressible);
let middle = data.len() / 2;
data[middle] ^= 0xff;
put(&store, &Object::Blob(Blob { data }))
})
.collect();
let tip = commit(&store, &edited, vec![base], 2);
(tip, Some(base), held_closure(&store, Some(base)).unwrap())
} else {
(commit(&store, &files, Vec::new(), 1), None, HashSet::new())
};
let plan = if with_deltas {
transfer::plan_pack_with(&store, tip, base, encode_delta_candidates_batch).unwrap()
} else {
transfer::plan_pack_with(&store, tip, None, |_, candidates| {
Ok(vec![None; candidates.len()])
})
.unwrap()
};
let limits = UploadLimits {
max_pack_bytes: max_pack,
tickets_per_advance: None,
ticket_threshold_bytes: None,
};
let cap = effective_payload_cap(payload_cap, max_pack).unwrap();
let estimate = estimate_pack_sizes(&store, &plan, cap, max_pack).unwrap().len();
let keys = build_and_upload_packs(PackSink::Count, &store, plan, cap, limits)
.map_err(|e| TestCaseError::fail(format!("{e:?}")))?;
prop_assert!(keys.len() <= estimate, "{} > {estimate}", keys.len());
let chain = Chain { commits: vec![tip], seed };
let weight = *cumulative_weights(&store, &chain, &PushControl::default())
.unwrap()
.last()
.unwrap();
if weight <= 3 * pack_weight_cap(cap, max_pack) {
prop_assert!(estimate <= 6, "{estimate} packs for weight {weight}");
}
}
}
}