use std::collections::{HashMap, HashSet};
use std::io::{Cursor, ErrorKind, Read};
use std::path::Path;
use crate::branch_tracking;
use crate::environment::RepositoryOptions;
use crate::error::{Error, Result};
use crate::fetch::Progress;
use crate::objects::{HashAlgo, ObjectId};
use crate::pkt_line::{self, Packet};
use crate::progress::{poll_cancel, Cancel};
use crate::push_report::{PushRefResult, PushRefStatus};
use crate::transfer::{
build_pack, open_odb, PackBuildOptions, PushOptions, PushOutcome, PushRefSpec,
};
use crate::transport::Connection;
fn finish_push_outcome(
local_git_dir: &Path,
opts: &PushOptions,
results: Vec<PushRefResult>,
) -> Result<PushOutcome> {
if !opts.dry_run {
if let Some(remote) = opts.tracking_remote.as_deref() {
branch_tracking::apply_push_remote_tracking_updates(local_git_dir, remote, &results)?;
}
}
Ok(PushOutcome { results })
}
const PUSH_CAPS_BASE: &str = "report-status report-status-v2 quiet";
pub fn push_remote(
local_git_dir: &Path,
conn: &mut dyn Connection,
refs: &[PushRefSpec],
opts: &PushOptions,
progress: &mut dyn Progress,
cancel: &dyn Cancel,
) -> Result<PushOutcome> {
crate::trace::trace_span!("grit.push");
poll_cancel(progress, cancel)?;
use crate::net_trace::net_trace;
net_trace!(
opts.network_trace,
opts.diagnostics.as_ref(),
"push_remote: begin — {} ref update(s), protocol v{}, {} push-option(s)",
refs.len(),
conn.protocol_version(),
opts.push_options.len()
);
if conn.protocol_version() >= 2 {
return Err(Error::Message(
"push_remote: protocol v2 not supported in this phase (use v0/v1)".to_owned(),
));
}
let local_odb = open_odb(local_git_dir);
let algo = local_odb.hash_algo();
let adv = AdvertisedState::from_connection(conn);
net_trace!(
opts.network_trace,
opts.diagnostics.as_ref(),
"push_remote: remote advertised {} ref(s)",
adv.remote_refs.len()
);
require_push_options_supported(&adv, opts)?;
let mut plan = match plan_push(refs, &local_odb, local_git_dir, &adv, opts)? {
PlanOutcome::Send(plan) => plan,
PlanOutcome::Done(results) => {
if !opts.dry_run {
crate::transport::send_client_flush(conn)?;
}
return finish_push_outcome(local_git_dir, opts, results);
}
};
let commands = build_command_block(&plan, &adv, algo, &opts.push_options)?;
net_trace!(
opts.network_trace,
opts.diagnostics.as_ref(),
"push_remote: sending {} command(s)…",
plan.decisions.len()
);
conn.writer().write_all(&commands)?;
conn.writer().flush()?;
if let Some(pack) = build_push_pack(&plan, local_git_dir, &local_odb, &adv)? {
poll_cancel(progress, cancel)?;
net_trace!(
opts.network_trace,
opts.diagnostics.as_ref(),
"push_remote: sending pack ({} bytes)…",
pack.len()
);
conn.writer().write_all(&pack)?;
conn.writer().flush()?;
} else {
net_trace!(
opts.network_trace,
opts.diagnostics.as_ref(),
"push_remote: no pack (deletion-only / up-to-date)"
);
}
let sent_refs: Vec<&str> = plan
.to_send
.iter()
.map(|&i| plan.decisions[i].result.remote_ref.as_str())
.collect();
let raw = read_receive_pack_response(conn.reader(), &sent_refs, adv.server_sideband)?;
let report = if adv.server_sideband {
demux_report_and_remote_messages(&raw, progress)?
} else {
raw
};
apply_report_status(&report, &mut plan.decisions);
let results: Vec<_> = plan.decisions.into_iter().map(|d| d.result).collect();
net_trace!(
opts.network_trace,
opts.diagnostics.as_ref(),
"push_remote: done — {} result(s)",
results.len()
);
finish_push_outcome(local_git_dir, opts, results)
}
pub fn push_http(
client: std::sync::Arc<dyn crate::transport::http::HttpClient>,
local_git_dir: &Path,
repo_url: &str,
refs: &[PushRefSpec],
opts: &PushOptions,
progress: &mut dyn Progress,
cancel: &dyn Cancel,
) -> Result<PushOutcome> {
use crate::transport::http::SmartHttpTransport;
use crate::transport::{ConnectOptions, Service, Transport};
let connect = ConnectOptions {
protocol_version: 0,
diagnostics: opts.diagnostics.clone(),
network_trace: opts.network_trace,
..Default::default()
};
let transport = SmartHttpTransport::new(client);
let mut conn = transport.connect(repo_url, Service::ReceivePack, &connect)?;
push_remote(local_git_dir, &mut *conn, refs, opts, progress, cancel)
}
struct AdvertisedState {
remote_refs: HashMap<String, ObjectId>,
advertised_haves: Vec<ObjectId>,
server_sideband: bool,
server_ofs_delta: bool,
server_push_options: bool,
}
impl AdvertisedState {
fn from_connection(conn: &mut dyn Connection) -> Self {
let mut remote_refs: HashMap<String, ObjectId> = HashMap::new();
let mut advertised_haves: Vec<ObjectId> = Vec::new();
for (name, oid) in conn.advertised_refs() {
if name == ".have" {
advertised_haves.push(*oid);
} else {
remote_refs.insert(name.clone(), *oid);
}
}
let caps = conn.capabilities();
Self {
remote_refs,
advertised_haves,
server_sideband: caps
.iter()
.any(|c| c == "side-band-64k" || c == "side-band"),
server_ofs_delta: caps.iter().any(|c| c == "ofs-delta"),
server_push_options: caps.iter().any(|c| c == "push-options"),
}
}
}
struct PushPlan {
decisions: Vec<PushDecision>,
to_send: Vec<usize>,
}
enum PlanOutcome {
Send(PushPlan),
Done(Vec<PushRefResult>),
}
fn plan_push(
refs: &[PushRefSpec],
local_odb: &crate::odb::Odb,
local_git_dir: &Path,
adv: &AdvertisedState,
opts: &PushOptions,
) -> Result<PlanOutcome> {
let mut decisions: Vec<PushDecision> = Vec::with_capacity(refs.len());
for spec in refs {
decisions.push(decide_push_wire(
spec,
local_odb,
local_git_dir,
&adv.remote_refs,
)?);
}
let any_rejected = decisions.iter().any(|d| d.result.status.is_error());
if opts.atomic && any_rejected {
for d in &mut decisions {
if matches!(d.result.status, PushRefStatus::Ok) {
d.result.status = PushRefStatus::AtomicPushFailed;
d.send = false;
}
}
return Ok(PlanOutcome::Done(
decisions.into_iter().map(|d| d.result).collect(),
));
}
let to_send: Vec<usize> = decisions
.iter()
.enumerate()
.filter_map(|(i, d)| if d.send { Some(i) } else { None })
.collect();
if to_send.is_empty() || opts.dry_run {
return Ok(PlanOutcome::Done(
decisions.into_iter().map(|d| d.result).collect(),
));
}
Ok(PlanOutcome::Send(PushPlan { decisions, to_send }))
}
fn require_push_options_supported(adv: &AdvertisedState, opts: &PushOptions) -> Result<()> {
if !opts.push_options.is_empty() && !adv.server_push_options {
return Err(Error::PushOptionsUnsupported);
}
Ok(())
}
fn build_command_block(
plan: &PushPlan,
adv: &AdvertisedState,
algo: HashAlgo,
push_options: &[String],
) -> Result<Vec<u8>> {
let zero_hex = "0".repeat(algo.hex_len());
let mut command_caps = PUSH_CAPS_BASE.to_owned();
if adv.server_sideband {
command_caps.push_str(" side-band-64k");
}
if !push_options.is_empty() {
command_caps.push_str(" push-options");
}
command_caps.push_str(&format!(" object-format={}", algo.name()));
let mut commands: Vec<u8> = Vec::new();
let mut first = true;
for &i in &plan.to_send {
let d = &plan.decisions[i];
let old_hex = d
.result
.old_oid
.map(|o| o.to_hex())
.unwrap_or_else(|| zero_hex.clone());
let new_hex = d
.result
.new_oid
.map(|o| o.to_hex())
.unwrap_or_else(|| zero_hex.clone());
let line = if first {
first = false;
format!(
"{old_hex} {new_hex} {}\0{command_caps}",
d.result.remote_ref
)
} else {
format!("{old_hex} {new_hex} {}", d.result.remote_ref)
};
pkt_line::write_line_to_vec(&mut commands, &line)?;
}
commands.extend_from_slice(b"0000");
if !push_options.is_empty() {
for opt in push_options {
pkt_line::write_line_to_vec(&mut commands, opt)?;
}
commands.extend_from_slice(b"0000");
}
Ok(commands)
}
fn build_push_pack(
plan: &PushPlan,
local_git_dir: &Path,
local_odb: &crate::odb::Odb,
adv: &AdvertisedState,
) -> Result<Option<Vec<u8>>> {
let wants: Vec<ObjectId> = plan
.to_send
.iter()
.filter_map(|&i| plan.decisions[i].new_tip)
.collect();
if wants.is_empty() {
return Ok(None);
}
let haves = local_odb.present_oids(
adv.remote_refs
.values()
.chain(&adv.advertised_haves)
.copied(),
)?;
let local_cfg = crate::config::ConfigSet::load(
&crate::environment::Environment::empty(),
Some(local_git_dir),
true,
)
.ok();
let mut pack_opts = if haves.is_empty() {
PackBuildOptions::for_local_copy(local_cfg.as_ref())
} else {
PackBuildOptions::for_local_push(local_cfg.as_ref())
};
pack_opts.use_ofs_delta = adv.server_ofs_delta;
build_pack(local_odb, &wants, &haves, &pack_opts).map(Some)
}
struct PushDecision {
result: PushRefResult,
new_tip: Option<ObjectId>,
send: bool,
}
fn decide_push_wire(
spec: &PushRefSpec,
local_odb: &crate::odb::Odb,
local_git_dir: &Path,
remote_refs: &HashMap<String, ObjectId>,
) -> Result<PushDecision> {
let remote_current = remote_refs.get(&spec.dst).copied();
let no_op = |status: PushRefStatus,
old: Option<ObjectId>,
new: Option<ObjectId>,
deletion: bool,
message: Option<String>| {
PushDecision {
result: PushRefResult {
local_ref: crate::transfer::push_spec_local_ref(spec),
remote_ref: spec.dst.clone(),
old_oid: old,
new_oid: new,
forced: false,
deletion,
status,
message,
},
new_tip: None,
send: false,
}
};
if !spec.delete {
if let Some(src) = spec.src {
if remote_current == Some(src) {
return Ok(no_op(
PushRefStatus::UpToDate,
remote_current,
Some(src),
false,
None,
));
}
}
}
if spec.expect_absent && remote_current.is_some() {
return Ok(no_op(
PushRefStatus::RejectStale,
remote_current,
spec.src,
spec.delete,
Some("stale info".to_owned()),
));
}
if let Some(expected) = spec.expected_old {
if remote_current != Some(expected) {
return Ok(no_op(
PushRefStatus::RejectStale,
remote_current,
spec.src,
spec.delete,
Some("stale info".to_owned()),
));
}
}
if spec.delete {
return Ok(match remote_current {
Some(_) => PushDecision {
result: PushRefResult {
local_ref: crate::transfer::push_spec_local_ref(spec),
remote_ref: spec.dst.clone(),
old_oid: remote_current,
new_oid: None,
forced: false,
deletion: true,
status: PushRefStatus::Ok,
message: None,
},
new_tip: None,
send: true,
},
None => no_op(
PushRefStatus::RejectRemoteRefMissing,
None,
None,
true,
Some("remote ref does not exist".to_owned()),
),
});
}
let Some(src) = spec.src else {
return Err(Error::Message(format!(
"push to '{}' has no source object and is not a deletion",
spec.dst
)));
};
if !local_odb.contains(&src)? {
return Err(Error::Message(format!(
"source object {src} for '{}' is missing from the local object store",
spec.dst
)));
}
let Some(old) = remote_current else {
return Ok(PushDecision {
result: PushRefResult {
local_ref: crate::transfer::push_spec_local_ref(spec),
remote_ref: spec.dst.clone(),
old_oid: None,
new_oid: Some(src),
forced: false,
deletion: false,
status: PushRefStatus::Ok,
message: None,
},
new_tip: Some(src),
send: true,
});
};
if !spec.force && !local_odb.contains(&old)? {
return Ok(no_op(
PushRefStatus::RejectFetchFirst,
Some(old),
Some(src),
false,
Some("fetch first".to_owned()),
));
}
let is_ff = if spec.force {
false
} else if crate::merge_base::is_ancestor_odb_walk(local_odb, old, src)? {
true
} else if let Ok(repo) =
crate::repo::Repository::open(local_git_dir, &RepositoryOptions::empty())
{
crate::merge_base::is_ancestor(&repo, old, src).unwrap_or(false)
} else {
false
};
if is_ff {
Ok(PushDecision {
result: PushRefResult {
local_ref: crate::transfer::push_spec_local_ref(spec),
remote_ref: spec.dst.clone(),
old_oid: Some(old),
new_oid: Some(src),
forced: false,
deletion: false,
status: PushRefStatus::Ok,
message: None,
},
new_tip: Some(src),
send: true,
})
} else if spec.force {
Ok(PushDecision {
result: PushRefResult {
local_ref: crate::transfer::push_spec_local_ref(spec),
remote_ref: spec.dst.clone(),
old_oid: Some(old),
new_oid: Some(src),
forced: true,
deletion: false,
status: PushRefStatus::Ok,
message: None,
},
new_tip: Some(src),
send: true,
})
} else {
Ok(PushDecision {
result: PushRefResult {
local_ref: crate::transfer::push_spec_local_ref(spec),
remote_ref: spec.dst.clone(),
old_oid: Some(old),
new_oid: Some(src),
forced: false,
deletion: false,
status: PushRefStatus::RejectNonFastForward,
message: Some("non-fast-forward".to_owned()),
},
new_tip: None,
send: false,
})
}
}
fn read_receive_pack_response(
reader: &mut dyn Read,
sent_refs: &[&str],
server_sideband: bool,
) -> Result<Vec<u8>> {
let mut raw = Vec::new();
let mut buf = [0u8; 8192];
loop {
match reader.read(&mut buf) {
Ok(0) => break,
Ok(n) => raw.extend_from_slice(&buf[..n]),
Err(e) if receive_pack_read_recoverable(&e) => {
if receive_pack_response_complete(&raw, sent_refs, server_sideband)? {
break;
}
return Err(Error::Io(e));
}
Err(e) => return Err(Error::Io(e)),
}
}
if !receive_pack_response_complete(&raw, sent_refs, server_sideband)? {
return Err(Error::Message(
"receive-pack response ended before a complete report-status".to_owned(),
));
}
Ok(raw)
}
fn receive_pack_read_recoverable(e: &std::io::Error) -> bool {
matches!(
e.kind(),
ErrorKind::ConnectionReset | ErrorKind::BrokenPipe | ErrorKind::UnexpectedEof
)
}
fn receive_pack_response_complete(
raw: &[u8],
sent_refs: &[&str],
server_sideband: bool,
) -> Result<bool> {
let report = if server_sideband {
demux_sideband_report(raw, None)?
} else {
raw.to_vec()
};
Ok(report_lines_complete(&report, sent_refs))
}
fn report_lines_complete(report: &[u8], sent_refs: &[&str]) -> bool {
let mut saw_unpack = false;
let mut per_ref = HashSet::<String>::new();
let mut cursor = Cursor::new(report);
while let Ok(Some(pkt)) = pkt_line::read_packet(&mut cursor) {
let Packet::Data(line) = pkt else {
continue;
};
let line = line.trim_end();
if line.starts_with("unpack ") {
saw_unpack = true;
} else if let Some(refname) = line.strip_prefix("ok ") {
per_ref.insert(refname.trim().to_owned());
} else if let Some(rest) = line.strip_prefix("ng ") {
let (refname, _) = rest.split_once(' ').unwrap_or((rest, ""));
per_ref.insert(refname.trim().to_owned());
}
}
saw_unpack && sent_refs.iter().all(|r| per_ref.contains(*r))
}
fn apply_report_status(report: &[u8], decisions: &mut [PushDecision]) {
let mut by_ref: HashMap<&str, usize> = HashMap::new();
for (i, d) in decisions.iter().enumerate() {
if d.send {
by_ref.insert(d.result.remote_ref.as_str(), i);
}
}
let mut unpack_error: Option<String> = None;
let mut updates: Vec<(usize, Option<String>)> = Vec::new();
let mut cursor = Cursor::new(report);
while let Ok(Some(pkt)) = pkt_line::read_packet(&mut cursor) {
let Packet::Data(line) = pkt else {
continue;
};
let line = line.trim_end();
if let Some(rest) = line.strip_prefix("unpack ") {
if rest.trim() != "ok" {
unpack_error = Some(rest.trim().to_owned());
}
} else if let Some(refname) = line.strip_prefix("ok ") {
let _ = by_ref.get(refname.trim());
} else if let Some(rest) = line.strip_prefix("ng ") {
let (refname, reason) = rest.split_once(' ').unwrap_or((rest, ""));
if let Some(&idx) = by_ref.get(refname.trim()) {
let msg = if reason.trim().is_empty() {
None
} else {
Some(reason.trim().to_owned())
};
updates.push((idx, msg));
}
}
}
for (idx, msg) in updates {
decisions[idx].result.status = PushRefStatus::RemoteRejected;
decisions[idx].result.message = msg;
}
if let Some(reason) = unpack_error {
for d in decisions.iter_mut() {
if d.send && !matches!(d.result.status, PushRefStatus::RemoteRejected) {
d.result.status = PushRefStatus::RemoteRejected;
d.result.message = Some(format!("unpack failed: {reason}"));
}
}
}
}
fn demux_sideband_report(input: &[u8], mut progress: Option<&mut dyn Progress>) -> Result<Vec<u8>> {
let mut report = Vec::new();
let mut i = 0usize;
while i + 4 <= input.len() {
let len = match pkt_line::parse_hex_len(&input[i..i + 4]) {
Ok(l) => l,
Err(_) => break,
};
i += 4;
if len == 0 {
continue;
}
if len < 4 || i + (len - 4) > input.len() {
break;
}
let payload = &input[i..i + (len - 4)];
i += len - 4;
if payload.is_empty() {
continue;
}
let band = payload[0];
let data = &payload[1..];
match band {
1 => report.extend_from_slice(data),
2 | 3 => {
if let Some(p) = progress.as_deref_mut() {
p.message(data);
}
}
_ => {}
}
}
Ok(report)
}
fn demux_report_and_remote_messages(input: &[u8], progress: &mut dyn Progress) -> Result<Vec<u8>> {
demux_sideband_report(input, Some(progress))
}
#[cfg(test)]
mod tests {
use super::*;
use crate::objects::ObjectKind;
fn make_decision(refname: &str, send: bool) -> PushDecision {
PushDecision {
result: PushRefResult {
local_ref: None,
remote_ref: refname.to_owned(),
old_oid: None,
new_oid: None,
forced: false,
deletion: false,
status: PushRefStatus::Ok,
message: None,
},
new_tip: None,
send,
}
}
fn report_bytes(lines: &[&str]) -> Vec<u8> {
let mut buf = Vec::new();
for l in lines {
pkt_line::write_line_to_vec(&mut buf, l).unwrap();
}
buf.extend_from_slice(b"0000");
buf
}
fn adv_state(sideband: bool, ofs_delta: bool, push_options: bool) -> AdvertisedState {
AdvertisedState {
remote_refs: HashMap::new(),
advertised_haves: Vec::new(),
server_sideband: sideband,
server_ofs_delta: ofs_delta,
server_push_options: push_options,
}
}
fn decode_block(block: &[u8]) -> Vec<Option<String>> {
let mut cur = Cursor::new(block);
let mut out = Vec::new();
while let Ok(pkt) = pkt_line::read_packet(&mut cur) {
match pkt {
Some(Packet::Data(s)) => out.push(Some(s.trim_end_matches('\n').to_owned())),
Some(Packet::Flush) => out.push(None),
_ => break,
}
}
out
}
fn flaky_push_odb(
dir: &Path,
) -> (
crate::odb::Odb,
std::sync::Arc<crate::odb::store::flaky::FlakyStore>,
) {
let flaky = crate::odb::store::flaky::FlakyStore::new();
let odb = crate::odb::OdbBuilder::files(dir.join("objects"))
.primary(flaky.clone())
.alternates(false)
.build();
(odb, flaky)
}
#[test]
fn push_pack_haves_fail_on_backend_error_instead_of_being_dropped() {
let dir = tempfile::tempdir().unwrap();
let (odb, flaky) = flaky_push_odb(dir.path());
let blob = odb.write(ObjectKind::Blob, b"to send").unwrap();
let remote_tip = odb.write(ObjectKind::Blob, b"remote has this").unwrap();
let plan = PushPlan {
decisions: vec![send_decision("refs/heads/main", blob)],
to_send: vec![0],
};
let mut adv = adv_state(false, true, false);
adv.remote_refs
.insert("refs/heads/main".to_owned(), remote_tip);
assert!(build_push_pack(&plan, dir.path(), &odb, &adv)
.unwrap()
.is_some());
flaky.fail_contains(true);
let err = build_push_pack(&plan, dir.path(), &odb, &adv).unwrap_err();
assert!(crate::odb::store::flaky::is_injected(&err), "{err:?}");
}
#[test]
fn push_source_check_fails_on_backend_error_instead_of_missing() {
let dir = tempfile::tempdir().unwrap();
let (odb, flaky) = flaky_push_odb(dir.path());
let src = odb.write(ObjectKind::Blob, b"source").unwrap();
let spec = PushRefSpec {
src: Some(src),
src_ref: None,
dst: "refs/heads/main".to_owned(),
force: false,
delete: false,
expected_old: None,
expect_absent: false,
};
let decision = decide_push_wire(&spec, &odb, dir.path(), &HashMap::new()).unwrap();
assert_eq!(decision.new_tip, Some(src));
flaky.fail_contains(true);
let err = decide_push_wire(&spec, &odb, dir.path(), &HashMap::new())
.err()
.unwrap();
assert!(crate::odb::store::flaky::is_injected(&err), "{err:?}");
}
fn send_decision(refname: &str, new_oid: ObjectId) -> PushDecision {
PushDecision {
result: PushRefResult {
local_ref: None,
remote_ref: refname.to_owned(),
old_oid: None,
new_oid: Some(new_oid),
forced: false,
deletion: false,
status: PushRefStatus::Ok,
message: None,
},
new_tip: Some(new_oid),
send: true,
}
}
#[test]
fn command_block_without_push_options_has_no_capability_or_lines() {
let new = ObjectId::from_hex(&"1".repeat(40)).unwrap();
let plan = PushPlan {
decisions: vec![send_decision("refs/heads/main", new)],
to_send: vec![0],
};
let block = build_command_block(&plan, &adv_state(false, false, true), HashAlgo::Sha1, &[])
.unwrap();
let pkts = decode_block(&block);
assert_eq!(pkts.len(), 2);
let cmd = pkts[0].as_deref().unwrap();
assert!(
cmd.contains("refs/heads/main"),
"first line is the ref command, got {cmd:?}"
);
assert!(
!cmd.contains("push-options"),
"no push-options capability without options, got {cmd:?}"
);
assert_eq!(pkts[1], None, "single trailing flush");
}
#[test]
fn command_block_with_push_options_negotiates_cap_and_emits_lines() {
let new = ObjectId::from_hex(&"1".repeat(40)).unwrap();
let plan = PushPlan {
decisions: vec![send_decision("refs/heads/main", new)],
to_send: vec![0],
};
let opts = vec!["ci.skip".to_owned(), "reviewer=alice".to_owned()];
let block = build_command_block(&plan, &adv_state(true, true, true), HashAlgo::Sha1, &opts)
.unwrap();
let pkts = decode_block(&block);
assert_eq!(
pkts,
vec![
pkts[0].clone(),
None,
Some("ci.skip".to_owned()),
Some("reviewer=alice".to_owned()),
None,
],
"push-option lines must follow the command-list flush, then a flush"
);
let cmd = pkts[0].as_deref().unwrap();
assert!(
cmd.contains("push-options"),
"capability list must advertise push-options, got {cmd:?}"
);
assert!(cmd.contains("report-status"));
assert!(cmd.contains("side-band-64k"));
assert!(cmd.contains("object-format=sha1"));
}
#[test]
fn require_push_options_errors_typed_when_server_lacks_capability() {
let opts = PushOptions {
push_options: vec!["x".to_owned()],
..PushOptions::default()
};
let err = require_push_options_supported(&adv_state(true, true, false), &opts).unwrap_err();
assert!(
matches!(err, Error::PushOptionsUnsupported),
"expected PushOptionsUnsupported, got {err:?}"
);
assert_eq!(
err.to_string(),
"the receiving end does not support push options"
);
require_push_options_supported(&adv_state(true, true, true), &opts).unwrap();
require_push_options_supported(&adv_state(true, true, false), &PushOptions::default())
.unwrap();
}
#[test]
fn report_ng_demotes_to_remote_rejected() {
let mut decisions = vec![
make_decision("refs/heads/main", true),
make_decision("refs/heads/topic", true),
];
let report = report_bytes(&[
"unpack ok",
"ok refs/heads/main",
"ng refs/heads/topic non-fast-forward",
]);
apply_report_status(&report, &mut decisions);
assert_eq!(decisions[0].result.status, PushRefStatus::Ok);
assert_eq!(decisions[1].result.status, PushRefStatus::RemoteRejected);
assert_eq!(
decisions[1].result.message.as_deref(),
Some("non-fast-forward")
);
}
#[test]
fn report_unpack_failure_rejects_all_sent() {
let mut decisions = vec![make_decision("refs/heads/main", true)];
let report = report_bytes(&["unpack index-pack abort"]);
apply_report_status(&report, &mut decisions);
assert_eq!(decisions[0].result.status, PushRefStatus::RemoteRejected);
assert!(decisions[0]
.result
.message
.as_deref()
.unwrap()
.starts_with("unpack failed:"));
}
#[test]
fn demux_separates_report_and_progress() {
struct Cap(Vec<u8>);
impl crate::progress::ProgressSink for Cap {}
impl Progress for Cap {
fn message(&mut self, bytes: &[u8]) {
self.0.extend_from_slice(bytes);
}
}
let mut wire = Vec::new();
let mut band1 = vec![1u8];
band1.extend_from_slice(b"unpack ok\n");
pkt_line::write_packet_raw(&mut wire, &band1).unwrap();
let mut band2 = vec![2u8];
band2.extend_from_slice(b"hello from hook\n");
pkt_line::write_packet_raw(&mut wire, &band2).unwrap();
wire.extend_from_slice(b"0000");
let mut cap = Cap(Vec::new());
let report = demux_report_and_remote_messages(&wire, &mut cap).unwrap();
assert_eq!(report, b"unpack ok\n");
assert_eq!(cap.0, b"hello from hook\n");
}
#[test]
fn report_lines_complete_requires_unpack_and_every_sent_ref() {
let sent = ["refs/heads/main"];
let full = report_bytes(&["unpack ok", "ok refs/heads/main"]);
assert!(report_lines_complete(&full, &sent));
let missing_ok = report_bytes(&["unpack ok"]);
assert!(!report_lines_complete(&missing_ok, &sent));
let missing_unpack = report_bytes(&["ok refs/heads/main"]);
assert!(!report_lines_complete(&missing_unpack, &sent));
let ng_ok = report_bytes(&["unpack ok", "ng refs/heads/main denied"]);
assert!(report_lines_complete(&ng_ok, &sent));
}
struct AbruptAfterRead {
payload: Vec<u8>,
pos: usize,
err_kind: ErrorKind,
}
impl Read for AbruptAfterRead {
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
if self.pos < self.payload.len() {
let n = std::cmp::min(buf.len(), self.payload.len() - self.pos);
buf[..n].copy_from_slice(&self.payload[self.pos..self.pos + n]);
self.pos += n;
return Ok(n);
}
Err(std::io::Error::new(self.err_kind, "forcibly closed"))
}
}
fn sideband_receive_pack_wire(hook_line: &[u8]) -> Vec<u8> {
let report_body = report_bytes(&["unpack ok", "ok refs/heads/main"]);
let mut band1 = vec![1u8];
band1.extend_from_slice(&report_body);
let mut wire = Vec::new();
pkt_line::write_packet_raw(&mut wire, &band1).unwrap();
let mut band2 = vec![2u8];
band2.extend_from_slice(hook_line);
pkt_line::write_packet_raw(&mut wire, &band2).unwrap();
wire.extend_from_slice(b"0000");
wire
}
struct EofAfterRead {
payload: Vec<u8>,
pos: usize,
}
impl Read for EofAfterRead {
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
if self.pos < self.payload.len() {
let n = std::cmp::min(buf.len(), self.payload.len() - self.pos);
buf[..n].copy_from_slice(&self.payload[self.pos..self.pos + n]);
self.pos += n;
Ok(n)
} else {
Ok(0)
}
}
}
fn read_push_report(
reader: &mut dyn Read,
sent_refs: &[&str],
sideband: bool,
progress: &mut dyn Progress,
) -> Result<Vec<u8>> {
let raw = read_receive_pack_response(reader, sent_refs, sideband)?;
if sideband {
demux_report_and_remote_messages(&raw, progress)
} else {
Ok(raw)
}
}
#[test]
fn read_receive_pack_response_tolerates_reset_after_full_report() {
let report = report_bytes(&["unpack ok", "ok refs/heads/main"]);
let mut reader = AbruptAfterRead {
payload: report.clone(),
pos: 0,
err_kind: ErrorKind::ConnectionReset,
};
let out =
read_receive_pack_response(&mut reader, &["refs/heads/main"], false).expect("reset");
assert_eq!(out, report);
}
#[test]
fn read_receive_pack_response_fails_reset_on_incomplete_report() {
let partial = report_bytes(&["unpack ok"]);
let mut reader = AbruptAfterRead {
payload: partial,
pos: 0,
err_kind: ErrorKind::ConnectionReset,
};
let err = read_receive_pack_response(&mut reader, &["refs/heads/main"], false).unwrap_err();
assert!(
matches!(err, Error::Io(_)),
"incomplete report must stay an I/O error, got {err:?}"
);
}
#[test]
fn sideband_diagnostics_delivered_once_on_clean_eof() {
struct Cap(Vec<u8>);
impl crate::progress::ProgressSink for Cap {}
impl Progress for Cap {
fn message(&mut self, bytes: &[u8]) {
self.0.extend_from_slice(bytes);
}
}
let hook = b"hook output\n";
let wire = sideband_receive_pack_wire(hook);
let mut reader = EofAfterRead {
payload: wire,
pos: 0,
};
let mut cap = Cap(Vec::new());
read_push_report(&mut reader, &["refs/heads/main"], true, &mut cap)
.expect("clean eof side-band push report");
assert_eq!(cap.0.as_slice(), hook);
}
#[test]
fn sideband_diagnostics_delivered_once_on_connection_reset() {
struct Cap(Vec<u8>);
impl crate::progress::ProgressSink for Cap {}
impl Progress for Cap {
fn message(&mut self, bytes: &[u8]) {
self.0.extend_from_slice(bytes);
}
}
let hook = b"hook output\n";
let wire = sideband_receive_pack_wire(hook);
let mut reader = AbruptAfterRead {
payload: wire,
pos: 0,
err_kind: ErrorKind::ConnectionReset,
};
let mut cap = Cap(Vec::new());
read_push_report(&mut reader, &["refs/heads/main"], true, &mut cap)
.expect("reset after complete side-band report");
assert_eq!(cap.0.as_slice(), hook);
}
#[test]
fn push_pack_initial_branch_push_uses_deltas() {
use std::process::Command;
fn pack_delta_count(bytes: &[u8], algo_len: usize) -> usize {
fn read_type_size(b: &[u8]) -> (u8, usize, usize) {
let type_code = (b[0] >> 4) & 7;
let mut size = (b[0] & 0x0f) as usize;
let mut shift = 4;
let mut i = 1;
while b[i - 1] & 0x80 != 0 {
size |= ((b[i] & 0x7f) as usize) << shift;
shift += 7;
i += 1;
}
(type_code, size, i)
}
fn zlib_consume(b: &[u8]) -> usize {
use std::io::Read as _;
let mut dec = flate2::bufread::ZlibDecoder::new(b);
let mut sink = Vec::new();
dec.read_to_end(&mut sink).expect("zlib");
dec.total_in() as usize
}
let nr = u32::from_be_bytes(bytes[8..12].try_into().unwrap()) as usize;
let mut pos = 12usize;
let mut deltas = 0usize;
for _ in 0..nr {
let (type_code, _, consumed) = read_type_size(&bytes[pos..]);
pos += consumed;
match type_code {
6 | 7 => {
deltas += 1;
if type_code == 7 {
pos += algo_len;
} else {
while bytes[pos] & 0x80 != 0 {
pos += 1;
}
pos += 1;
}
}
_ => {}
}
pos += zlib_consume(&bytes[pos..]);
}
deltas
}
let tmp = tempfile::tempdir().unwrap();
let dir = tmp.path();
for i in 0..40 {
if i == 0 {
Command::new("git")
.current_dir(dir)
.args(["init", "-q", "-b", "main", "."])
.env("GIT_CONFIG_GLOBAL", "/dev/null")
.env("GIT_CONFIG_SYSTEM", "/dev/null")
.status()
.unwrap();
}
std::fs::write(dir.join(format!("f{i:03}.txt")), format!("content {i}\n")).unwrap();
std::fs::write(dir.join("shared.txt"), format!("shared rev {i}\n")).unwrap();
Command::new("git")
.current_dir(dir)
.args(["add", "-A"])
.env("GIT_CONFIG_GLOBAL", "/dev/null")
.env("GIT_CONFIG_SYSTEM", "/dev/null")
.status()
.unwrap();
Command::new("git")
.current_dir(dir)
.args(["commit", "-q", "-m", &format!("commit {i}")])
.env("GIT_AUTHOR_NAME", "T")
.env("GIT_AUTHOR_EMAIL", "t@example.com")
.env("GIT_AUTHOR_DATE", "2005-04-07T22:13:13 +0200")
.env("GIT_COMMITTER_NAME", "T")
.env("GIT_COMMITTER_EMAIL", "t@example.com")
.env("GIT_COMMITTER_DATE", "2005-04-07T22:13:13 +0200")
.env("GIT_CONFIG_GLOBAL", "/dev/null")
.env("GIT_CONFIG_SYSTEM", "/dev/null")
.status()
.unwrap();
}
Command::new("git")
.current_dir(dir)
.args(["repack", "-adf"])
.env("GIT_CONFIG_GLOBAL", "/dev/null")
.env("GIT_CONFIG_SYSTEM", "/dev/null")
.status()
.unwrap();
let git_dir = dir.join(".git");
let odb = open_odb(&git_dir);
let tip = crate::refs::resolve_ref(&git_dir, "HEAD").expect("head");
let plan = PushPlan {
decisions: vec![send_decision("refs/heads/main", tip)],
to_send: vec![0],
};
let adv = adv_state(false, true, false);
let pack = build_push_pack(&plan, &git_dir, &odb, &adv)
.unwrap()
.expect("pack bytes");
let deltas = pack_delta_count(&pack, odb.hash_algo().len());
assert!(
deltas > 0,
"initial push pack must contain delta objects (got {deltas})"
);
}
}